diff --git a/README.md b/README.md
new file mode 100644
index 0000000..95fe30f
--- /dev/null
+++ b/README.md
@@ -0,0 +1,39 @@
+# Lobster
+
+A Clawdbot-native workflow shell: typed (JSON-first) pipelines, jobs, and approval gates.
+
+This repo is an MVP scaffold focused on the core shell runtime and a first Gmail integration via the `steipete/gog` skill/CLI.
+
+## Goals
+
+- Typed pipelines (objects/arrays), not text pipes.
+- Local-first execution.
+- No new auth surface: Lobster must not own OAuth/tokens.
+- Composable macros that Clawdbot can invoke in one step to save tokens.
+
+## Quick start
+
+From this folder:
+
+- `node ./bin/lobster.js --help`
+- `node ./bin/lobster.js "exec --json 'echo [1,2,3]' | where '0>=0' | json"`
+
+If you have `gog` installed:
+
+- `node ./bin/lobster.js "gog.gmail.search --query 'newer_than:7d' --max 5 | table"`
+
+## Commands
+
+- `exec`: run OS commands
+- `gog.gmail.search`: fetch Gmail search results via `gog`
+- `gog.gmail.send`: send email via `gog` (use approval gates)
+- `email.triage`: deterministic triage report (rule-based)
+- `where`, `pick`, `head`: data shaping
+- `json`, `table`: renderers
+- `approve`: approval gate (TTY prompt or `--emit` for Clawdbot integration)
+
+## Next steps
+
+- Canonical `EmailMessage` schema (normalize gog output predictably).
+- `email.draft` + `email.send` macros (compose approvals cleanly).
+- Clawdbot integration: expose Lobster as a first-class tool (`lobster.run`).
diff --git a/bin/lobster.js b/bin/lobster.js
new file mode 100755
index 0000000..dcbb842
--- /dev/null
+++ b/bin/lobster.js
@@ -0,0 +1,4 @@
+#!/usr/bin/env node
+import { runCli } from '../src/cli.js';
+
+await runCli(process.argv.slice(2));
diff --git a/docs/DESIGN.md b/docs/DESIGN.md
new file mode 100644
index 0000000..be7d0cc
--- /dev/null
+++ b/docs/DESIGN.md
@@ -0,0 +1,57 @@
+# Lobster design (draft)
+
+## What Lobster is
+
+Lobster is a Clawdbot-native workflow shell.
+
+- JSON-first typed pipelines (records/arrays) instead of byte streams.
+- Deterministic composition of tools/skills into reusable macros.
+- Human-in-the-loop approval gates as language primitives.
+- Local-first: Lobster should talk to a local Clawdbot runtime and/or local CLIs.
+
+## What Lobster is not
+
+- Not a terminal emulator.
+- Not a POSIX-compatible shell (at least initially).
+- Not an auth broker: Lobster must not store OAuth tokens.
+
+## Why it exists
+
+- Turns multi-step tool orchestration into a single `lobster.run(...)` call.
+- Saves tokens by moving deterministic orchestration out of the LLM.
+- Makes automation auditable and safe-by-default.
+
+## Data model
+
+- A pipeline is a list of stages.
+- Each stage consumes an async stream of items and produces an async stream.
+- Items are arbitrary JSON values, but common shapes should be standardized (e.g. EmailMessage).
+
+## Safety model
+
+- Commands declare capabilities (e.g. `email.read`, `email.send`, `fs.write`).
+- Approval gates must fail closed in non-interactive mode.
+- When integrated into Clawdbot, approvals are surfaced to the user by Clawdbot.
+
+## Clawdbot integration (target)
+
+Preferred: Lobster does not run `gog` directly. Instead it calls Clawdbot tools.
+
+- Clawdbot exposes `gog` (or `google`) as a tool.
+- Lobster calls `tools.invoke({ tool: 'gog', action: 'gmail.search', ...})`.
+
+In the MVP, `gog.gmail.search` shells out to `gog` for fast iteration.
+
+## MVP scope
+
+- Parser for `cmd ... | cmd ...` pipelines.
+- A minimal standard library (where/pick/head/json/table).
+- An interactive `approve` primitive.
+- A first Gmail read primitive via `gog.gmail.search`.
+
+## Next milestones
+
+1. `email.normalize`: normalize gog output to `EmailMessage`.
+2. `email.triage`: classify + draft + propose label actions.
+3. Non-interactive approvals: emit `requiresApproval` objects instead of prompting.
+4. Clawdbot tool bridge: replace `gog` exec with tool invocation.
diff --git a/package.json b/package.json
new file mode 100644
index 0000000..ebcd60a
--- /dev/null
+++ b/package.json
@@ -0,0 +1,15 @@
+{
+  "name": "lobster-shell",
+  "version": "0.1.0",
+  "private": true,
+  "type": "module",
+  "bin": {
+    "lobster": "./bin/lobster.js"
+  },
+  "scripts": {
+    "test": "node --test"
+  },
+  "engines": {
+    "node": ">=20"
+  }
+}
diff --git a/src/cli.js b/src/cli.js
new file mode 100644
index 0000000..86513e1
--- /dev/null
+++ b/src/cli.js
@@ -0,0 +1,74 @@
+import { parsePipeline } from './parser.js';
+import { createDefaultRegistry } from './commands/registry.js';
+import { runPipeline } from './runtime.js';
+
+export async function runCli(argv) {
+  const registry = createDefaultRegistry();
+
+  if (argv.length === 0 || argv.includes('-h') || argv.includes('--help')) {
+    process.stdout.write(helpText());
+    return;
+  }
+
+  if (argv[0] === 'help') {
+    const topic = argv[1];
+    if (!topic) {
+      process.stdout.write(helpText());
+      return;
+    }
+    const cmd = registry.get(topic);
+    if (!cmd) {
+      process.stderr.write(`Unknown command: ${topic}\n`);
+      process.exitCode = 2;
+      return;
+    }
+    process.stdout.write(cmd.help());
+    return;
+  }
+
+  const inputLine = argv.join(' ');
+  let pipeline;
+  try {
+    pipeline = parsePipeline(inputLine);
+  } catch (err) {
+    process.stderr.write(`Parse error: ${err?.message ?? String(err)}\n`);
+    process.exitCode = 2;
+    return;
+  }
+
+  try {
+    const output = await runPipeline({
+      pipeline,
+      registry,
+      stdin: process.stdin,
+      stdout: process.stdout,
+      stderr: process.stderr,
+      env: process.env,
+    });
+
+    // Default rendering: if the last command didn't render, print JSON.
+    if (!output.rendered) {
+      process.stdout.write(JSON.stringify(output.items, null, 2));
+      process.stdout.write('\n');
+    }
+  } catch (err) {
+    process.stderr.write(`Error: ${err?.message ?? String(err)}\n`);
+    process.exitCode = 1;
+  }
+}
+
+function helpText() {
+  return `lobster (v0.1) — Clawdbot-native typed shell\n\n` +
+    `Usage:\n` +
+    `  lobster '<pipeline>'\n` +
+    `  lobster help <command>\n\n` +
+    `Pipeline basics:\n` +
+    `  - Commands are piped with |\n` +
+    `  - Data is JSON-first (arrays/objects), not text-first\n` +
+    `  - Most commands accept --flag value or --flag=value\n\n` +
+    `Examples:\n` +
+    `  lobster 'exec --json "echo [1,2,3]" | json'\n` +
+    `  lobster 'gog.gmail.search --query "newer_than:7d" --max 10 | pick id,subject,from | table'\n` +
+    `\nCommands:\n` +
+    `  exec, head, json, pick, table, where, approve, gog.gmail.search, gog.gmail.send, email.triage\n`;
+}
diff --git a/src/commands/registry.js b/src/commands/registry.js
new file mode 100644
index 0000000..7767e7b
--- /dev/null
+++ b/src/commands/registry.js
@@ -0,0 +1,38 @@
+import { execCommand } from './stdlib/exec.js';
+import { headCommand } from './stdlib/head.js';
+import { jsonCommand } from './stdlib/json.js';
+import { pickCommand } from './stdlib/pick.js';
+import { tableCommand } from './stdlib/table.js';
+import { whereCommand } from './stdlib/where.js';
+import { approveCommand } from './stdlib/approve.js';
+import { gogGmailSearchCommand } from './stdlib/gog_gmail_search.js';
+import { gogGmailSendCommand } from './stdlib/gog_gmail_send.js';
+import { emailTriageCommand } from './stdlib/email_triage.js';
+
+export function createDefaultRegistry() {
+  const commands = new Map();
+
+  for (const cmd of [
+    execCommand,
+    headCommand,
+    jsonCommand,
+    pickCommand,
+    tableCommand,
+    whereCommand,
+    approveCommand,
+    gogGmailSearchCommand,
+    gogGmailSendCommand,
+    emailTriageCommand,
+  ]) {
+    commands.set(cmd.name, cmd);
+  }
+
+  return {
+    get(name) {
+      return commands.get(name);
+    },
+    list() {
+      return [...commands.keys()].sort();
+    },
+  };
+}
diff --git a/src/commands/stdlib/approve.js b/src/commands/stdlib/approve.js
new file mode 100644
index 0000000..e3d2495
--- /dev/null
+++ b/src/commands/stdlib/approve.js
@@ -0,0 +1,62 @@
+function isInteractive(stdin) {
+  return Boolean(stdin.isTTY);
+}
+
+export const approveCommand = {
+  name: 'approve',
+  help() {
+    return `approve — require confirmation to continue\n\nUsage:\n  ... | approve --prompt "Send these emails?"\n  ... | approve --emit --prompt "Send these emails?"\n\nModes:\n  - Interactive (default): prompts on TTY and passes items through if approved.\n  - Emit (--emit): returns an approval request object and stops the pipeline.\n\nNotes:\n  - Default behavior fails closed in non-interactive mode.\n`;
+  },
+  async run({ input, args, ctx }) {
+    const prompt = args.prompt ?? 'Approve?';
+
+    const items = [];
+    for await (const item of input) items.push(item);
+
+    if (args.emit) {
+      return {
+        output: (async function* () {
+          yield {
+            type: 'approval_request',
+            prompt,
+            items,
+          };
+        })(),
+      };
+    }
+
+    if (!isInteractive(ctx.stdin)) {
+      throw new Error('approve requires an interactive TTY (or pass --emit)');
+    }
+
+    ctx.stdout.write(`${prompt} [y/N] `);
+    const answer = await readLine(ctx.stdin);
+
+    if (!/^y(es)?$/i.test(String(answer).trim())) {
+      throw new Error('Not approved');
+    }
+
+    return { output: asStream(items) };
+  },
+};
+
+function readLine(stdin) {
+  return new Promise((resolve) => {
+    let buf = '';
+
+    const onData = (chunk) => {
+      buf += chunk.toString('utf8');
+      const idx = buf.indexOf('\n');
+      if (idx !== -1) {
+        stdin.off('data', onData);
+        resolve(buf.slice(0, idx));
+      }
+    };
+
+    stdin.on('data', onData);
+  });
+}
+
+async function* asStream(items) {
+  for (const item of items) yield item;
+}
diff --git a/src/commands/stdlib/email_triage.js b/src/commands/stdlib/email_triage.js
new file mode 100644
index 0000000..9054c74
--- /dev/null
+++ b/src/commands/stdlib/email_triage.js
@@ -0,0 +1,93 @@
+function getField(obj, path, fallback = undefined) {
+  if (!obj || typeof obj !== 'object') return fallback;
+  if (!path) return fallback;
+  const parts = String(path).split('.');
+  let cur = obj;
+  for (const p of parts) {
+    if (!cur || typeof cur !== 'object') return fallback;
+    cur = cur[p];
+  }
+  return cur ?? fallback;
+}
+
+function normalizeString(v) {
+  if (v === null || v === undefined) return '';
+  return String(v);
+}
+
+function parseEmailAddress(from) {
+  // Handles "Name <email@x.com>" or "email@x.com".
+  const s = String(from ?? '').trim();
+  const m = s.match(/<([^>]+)>/);
+  return (m ? m[1] : s).trim();
+}
+
+function classifyEmail({ subject, snippet }) {
+  const text = `${subject} ${snippet}`.toLowerCase();
+  if (/(unsubscribe|newsletter|promo|sale|discount)/.test(text)) return { bucket: 'fyi', reason: 'newsletter/promo-ish' };
+  if (/(invoice|receipt|payment|charged|billing)/.test(text)) return { bucket: 'needs_action', reason: 'finance keyword' };
+  if (/(asap|urgent|action required|deadline|due)/.test(text)) return { bucket: 'needs_action', reason: 'urgency keyword' };
+  if (/[?]/.test(text)) return { bucket: 'needs_reply', reason: 'question mark' };
+  return { bucket: 'fyi', reason: 'default' };
+}
+
+export const emailTriageCommand = {
+  name: 'email.triage',
+  help() {
+    return `email.triage — deterministic email triage report\n\n` +
+      `Usage:\n` +
+      `  <emails> | email.triage [--subject-field subject] [--from-field from] [--snippet-field snippet] [--id-field id]\n\n` +
+      `Output:\n` +
+      `  Single object: { summary, items, buckets }\n\n` +
+      `Notes:\n` +
+      `  - This is intentionally non-LLM: rule-based classification (fast, predictable).\n` +
+      `  - Use --*-field flags to map provider-specific JSON into the triage schema.\n`;
+  },
+  async run({ input, args }) {
+    const idField = args['id-field'] ?? 'id';
+    const threadField = args['thread-field'] ?? 'threadId';
+    const subjectField = args['subject-field'] ?? 'subject';
+    const fromField = args['from-field'] ?? 'from';
+    const snippetField = args['snippet-field'] ?? 'snippet';
+    const dateField = args['date-field'] ?? 'date';
+
+    const items = [];
+    for await (const raw of input) {
+      const subject = normalizeString(getField(raw, subjectField, getField(raw, 'Subject')));
+      const from = normalizeString(getField(raw, fromField, getField(raw, 'From')));
+      const snippet = normalizeString(getField(raw, snippetField, getField(raw, 'Snippet')));
+
+      const classification = classifyEmail({ subject, snippet });
+
+      items.push({
+        id: getField(raw, idField),
+        threadId: getField(raw, threadField),
+        from,
+        fromEmail: parseEmailAddress(from),
+        subject,
+        snippet,
+        date: getField(raw, dateField),
+        bucket: classification.bucket,
+        reason: classification.reason,
+        raw,
+      });
+    }
+
+    const buckets = {
+      needs_reply: items.filter((x) => x.bucket === 'needs_reply'),
+      needs_action: items.filter((x) => x.bucket === 'needs_action'),
+      fyi: items.filter((x) => x.bucket === 'fyi'),
+    };
+
+    const summary = {
+      total: items.length,
+      needs_reply: buckets.needs_reply.length,
+      needs_action: buckets.needs_action.length,
+      fyi: buckets.fyi.length,
+    };
+
+    const report = { summary, items, buckets };
+
+    return { output: (async function* () { yield report; })() };
+  },
+};
diff --git a/src/commands/stdlib/exec.js b/src/commands/stdlib/exec.js
new file mode 100644
index 0000000..20d1335
--- /dev/null
+++ b/src/commands/stdlib/exec.js
@@ -0,0 +1,71 @@
+import { spawn } from 'node:child_process';
+
+export const execCommand = {
+  name: 'exec',
+  help() {
+    return `exec — run an OS command\n\n` +
+      `Usage:\n` +
+      `  exec <command...>\n` +
+      `  exec --json <command...>\n` +
+      `  exec --shell "<command line>"\n\n` +
+      `Notes:\n` +
+      `  - With --json, parses stdout as JSON (single value).\n` +
+      `  - With --shell (or a single arg containing spaces), runs via /bin/sh -lc.\n`;
+  },
+  async run({ args, ctx }) {
+    const cmd = args._;
+    if (!cmd.length) throw new Error('exec requires a command');
+
+    const shellLine = typeof args.shell === 'string' ? args.shell : null;
+    const useShell = Boolean(args.shell) || (cmd.length === 1 && /\s/.test(cmd[0]));
+
+    const result = useShell
+      ? await runProcess('/bin/sh', ['-lc', shellLine ?? cmd[0] ?? ''], { env: ctx.env, cwd: process.cwd() })
+      : await runProcess(cmd[0], cmd.slice(1), { env: ctx.env, cwd: process.cwd() });
+
+    if (args.json) {
+      let parsed;
+      try {
+        parsed = JSON.parse(result.stdout.trim() || 'null');
+      } catch (err) {
+        throw new Error(`exec --json could not parse stdout as JSON: ${err?.message ?? String(err)}`);
+      }
+
+      return {
+        output: asStream(Array.isArray(parsed) ? parsed : [parsed]),
+      };
+    }
+
+    const lines = result.stdout.split(/\r?\n/).filter(Boolean);
+    return { output: asStream(lines) };
+  },
+};
+
+function runProcess(command, argv, { env, cwd }) {
+  return new Promise((resolve, reject) => {
+    const child = spawn(command, argv, {
+      env,
+      cwd,
+      stdio: ['ignore', 'pipe', 'pipe'],
+    });
+
+    let stdout = '';
+    let stderr = '';
+
+    child.stdout.setEncoding('utf8');
+    child.stderr.setEncoding('utf8');
+
+    child.stdout.on('data', (d) => { stdout += d; });
+    child.stderr.on('data', (d) => { stderr += d; });
+
+    child.on('error', reject);
+    child.on('close', (code) => {
+      if (code === 0) return resolve({ stdout, stderr });
+      reject(new Error(`exec failed (${code}): ${stderr.trim() || stdout.trim() || command}`));
+    });
+  });
+}
+
+async function* asStream(items) {
+  for (const item of items) yield item;
+}
diff --git a/src/commands/stdlib/gog_gmail_search.js b/src/commands/stdlib/gog_gmail_search.js
new file mode 100644
index 0000000..b44d4ac
--- /dev/null
+++ b/src/commands/stdlib/gog_gmail_search.js
@@ -0,0 +1,70 @@
+import { spawn } from 'node:child_process';
+
+export const gogGmailSearchCommand = {
+  name: 'gog.gmail.search',
+  help() {
+    return `gog.gmail.search — fetch Gmail messages via steipete/gog\n\n` +
+      `Usage:\n` +
+      `  gog.gmail.search --query "newer_than:7d" --max 10 [--account you@gmail.com]\n\n` +
+      `Behavior:\n` +
+      `  - Runs: gog gmail search <query> --max <n> --json --no-input\n` +
+      `  - Does not handle auth; relies on existing gog auth/credentials locally.\n`;
+  },
+  async run({ args, ctx }) {
+    const query = args.query ?? args._[0];
+    const max = args.max ?? 10;
+    const account = args.account;
+
+    if (!query) throw new Error('gog.gmail.search requires --query');
+
+    const env = { ...ctx.env };
+    if (account) env.GOG_ACCOUNT = String(account);
+
+    const argv = ['gmail', 'search', String(query), '--max', String(max), '--json', '--no-input'];
+    const { stdout } = await runProcess('gog', argv, { env, cwd: process.cwd() });
+
+    let parsed;
+    try {
+      parsed = JSON.parse(stdout.trim() || '[]');
+    } catch (err) {
+      throw new Error(`gog gmail search returned non-JSON output`);
+    }
+
+    // Keep it permissive: pass through what gog returns.
+    // Later we can normalize into a canonical EmailMessage schema.
+    const items = Array.isArray(parsed) ? parsed : [parsed];
+
+    return { output: asStream(items) };
+  },
+};
+
+function runProcess(command, argv, { env, cwd }) {
+  return new Promise((resolve, reject) => {
+    const child = spawn(command, argv, { env, cwd, stdio: ['ignore', 'pipe', 'pipe'] });
+
+    let stdout = '';
+    let stderr = '';
+
+    child.stdout.setEncoding('utf8');
+    child.stderr.setEncoding('utf8');
+    child.stdout.on('data', (d) => { stdout += d; });
+    child.stderr.on('data', (d) => { stderr += d; });
+
+    child.on('error', (err) => {
+      if (err?.code === 'ENOENT') {
+        reject(new Error('gog not found on PATH (install steipete/gog from ClawdHub)'));
+        return;
+      }
+      reject(err);
+    });
+
+    child.on('close', (code) => {
+      if (code === 0) return resolve({ stdout, stderr });
+      reject(new Error(`gog failed (${code}): ${stderr.trim() || stdout.trim()}`));
+    });
+  });
+}
+
+async function* asStream(items) {
+  for (const item of items) yield item;
+}
diff --git a/src/commands/stdlib/gog_gmail_send.js b/src/commands/stdlib/gog_gmail_send.js
new file mode 100644
index 0000000..1f5f36d
--- /dev/null
+++ b/src/commands/stdlib/gog_gmail_send.js
@@ -0,0 +1,63 @@
+import { spawn } from 'node:child_process';
+
+export const gogGmailSendCommand = {
+  name: 'gog.gmail.send',
+  help() {
+    return `gog.gmail.send — send an email via steipete/gog\n\n` +
+      `Usage:\n` +
+      `  gog.gmail.send --to a@b.com --subject "Hi" --body "Hello" [--account you@gmail.com]\n\n` +
+      `Notes:\n` +
+      `  - Does not handle auth; relies on existing gog auth/credentials locally.\n` +
+      `  - Prefer running behind an approval gate (e.g. | approve --prompt "Send?").\n`;
+  },
+  async run({ args, ctx }) {
+    const to = args.to;
+    const subject = args.subject;
+    const body = args.body;
+    const account = args.account;
+
+    if (!to || !subject || body === undefined) {
+      throw new Error('gog.gmail.send requires --to, --subject, and --body');
+    }
+
+    const env = { ...ctx.env };
+    if (account) env.GOG_ACCOUNT = String(account);
+
+    const argv = ['gmail', 'send', '--to', String(to), '--subject', String(subject), '--body', String(body), '--no-input'];
+
+    const { stdout } = await runProcess('gog', argv, { env, cwd: process.cwd() });
+
+    return {
+      output: (async function* () {
+        yield { ok: true, to, subject, result: stdout.trim() };
+      })(),
+    };
+  },
+};
+
+function runProcess(command, argv, { env, cwd }) {
+  return new Promise((resolve, reject) => {
+    const child = spawn(command, argv, { env, cwd, stdio: ['ignore', 'pipe', 'pipe'] });
+
+    let stdout = '';
+    let stderr = '';
+
+    child.stdout.setEncoding('utf8');
+    child.stderr.setEncoding('utf8');
+    child.stdout.on('data', (d) => { stdout += d; });
+    child.stderr.on('data', (d) => { stderr += d; });
+
+    child.on('error', (err) => {
+      if (err?.code === 'ENOENT') {
+        reject(new Error('gog not found on PATH (install steipete/gog from ClawdHub)'));
+        return;
+      }
+      reject(err);
+    });
+
+    child.on('close', (code) => {
+      if (code === 0) return resolve({ stdout, stderr });
+      reject(new Error(`gog failed (${code}): ${stderr.trim() || stdout.trim()}`));
+    });
+  });
+}
diff --git a/src/commands/stdlib/head.js b/src/commands/stdlib/head.js
new file mode 100644
index 0000000..2ae1710
--- /dev/null
+++ b/src/commands/stdlib/head.js
@@ -0,0 +1,20 @@
+export const headCommand = {
+  name: 'head',
+  help() {
+    return `head — take first N items\n\nUsage:\n  head --n 10\n`;
+  },
+  async run({ input, args }) {
+    const n = args.n === undefined ? 10 : Number(args.n);
+    if (!Number.isFinite(n) || n < 0) throw new Error('head --n must be a non-negative number');
+
+    return {
+      output: (async function* () {
+        let i = 0;
+        for await (const item of input) {
+          if (i++ >= n) break;
+          yield item;
+        }
+      })(),
+    };
+  },
+};
diff --git a/src/commands/stdlib/json.js b/src/commands/stdlib/json.js
new file mode 100644
index 0000000..57f1802
--- /dev/null
+++ b/src/commands/stdlib/json.js
@@ -0,0 +1,14 @@
+export const jsonCommand = {
+  name: 'json',
+  help() {
+    return `json — render pipeline output as JSON\n\nUsage:\n  ... | json\n`;
+  },
+  async run({ input, ctx }) {
+    const items = [];
+    for await (const item of input) items.push(item);
+    ctx.render.json(items);
+    return { output: emptyStream(), rendered: true };
+  },
+};
+
+async function* emptyStream() {}
diff --git a/src/commands/stdlib/pick.js b/src/commands/stdlib/pick.js
new file mode 100644
index 0000000..5cba1e1
--- /dev/null
+++ b/src/commands/stdlib/pick.js
@@ -0,0 +1,25 @@
+export const pickCommand = {
+  name: 'pick',
+  help() {
+    return `pick — project fields from objects\n\nUsage:\n  ... | pick id,subject,from\n`;
+  },
+  async run({ input, args }) {
+    const spec = args._[0];
+    if (!spec) throw new Error('pick requires a comma-separated field list');
+    const fields = spec.split(',').map((s) => s.trim()).filter(Boolean);
+
+    return {
+      output: (async function* () {
+        for await (const item of input) {
+          if (item === null || typeof item !== 'object') {
+            yield item;
+            continue;
+          }
+          const out = {};
+          for (const f of fields) out[f] = item[f];
+          yield out;
+        }
+      })(),
+    };
+  },
+};
diff --git a/src/commands/stdlib/table.js b/src/commands/stdlib/table.js
new file mode 100644
index 0000000..17c0d36
--- /dev/null
+++ b/src/commands/stdlib/table.js
@@ -0,0 +1,56 @@
+function stringifyCell(v) {
+  if (v === null || v === undefined) return '';
+  if (typeof v === 'string') return v;
+  if (typeof v === 'number' || typeof v === 'boolean') return String(v);
+  return JSON.stringify(v);
+}
+
+export const tableCommand = {
+  name: 'table',
+  help() {
+    return `table — render items as a simple table\n\nUsage:\n  ... | table\n\nNotes:\n  - If items are objects, columns are union of keys (first 20 items).\n`;
+  },
+  async run({ input, ctx }) {
+    const items = [];
+    for await (const item of input) items.push(item);
+
+    if (items.length === 0) {
+      ctx.stdout.write('(no results)\n');
+      return { output: emptyStream(), rendered: true };
+    }
+
+    const sample = items.slice(0, 20);
+    const objectItems = sample.filter((x) => x && typeof x === 'object' && !Array.isArray(x));
+
+    if (objectItems.length === sample.length) {
+      const cols = [];
+      const seen = new Set();
+      for (const obj of objectItems) {
+        for (const k of Object.keys(obj)) {
+          if (!seen.has(k)) {
+            seen.add(k);
+            cols.push(k);
+          }
+        }
+      }
+
+      const rows = [cols, ...items.map((it) => cols.map((c) => stringifyCell(it?.[c])))]
+        .map((row) => row.map((cell) => cell.replace(/\n/g, ' ')));
+
+      const widths = cols.map((_, i) => Math.max(...rows.map((r) => r[i].length), 3));
+
+      const renderRow = (row) => row.map((cell, i) => cell.padEnd(widths[i])).join('  ');
+      ctx.stdout.write(renderRow(rows[0]) + '\n');
+      ctx.stdout.write(widths.map((w) => '-'.repeat(w)).join('  ') + '\n');
+      for (const row of rows.slice(1)) ctx.stdout.write(renderRow(row) + '\n');
+
+      return { output: emptyStream(), rendered: true };
+    }
+
+    // Fallback: render each item on a line.
+    for (const item of items) ctx.stdout.write(stringifyCell(item) + '\n');
+    return { output: emptyStream(), rendered: true };
+  },
+};
+
+async function* emptyStream() {}
diff --git a/src/commands/stdlib/where.js b/src/commands/stdlib/where.js
new file mode 100644
index 0000000..559ca6d
--- /dev/null
+++ b/src/commands/stdlib/where.js
@@ -0,0 +1,56 @@
+function parsePredicate(expr) {
+  const m = expr.match(/^([a-zA-Z0-9_\.]+)\s*(==|=|!=|<=|>=|<|>)\s*(.+)$/);
+  if (!m) throw new Error(`Invalid where expression: ${expr}`);
+  const [, path, op, rawValue] = m;
+
+  let value = rawValue;
+  if (rawValue === 'true') value = true;
+  else if (rawValue === 'false') value = false;
+  else if (rawValue === 'null') value = null;
+  else if (!Number.isNaN(Number(rawValue)) && rawValue.trim() !== '') value = Number(rawValue);
+
+  return { path, op: op === '=' ? '==' : op, value };
+}
+
+function getPath(obj, path) {
+  const parts = path.split('.');
+  let cur = obj;
+  for (const p of parts) {
+    if (cur === null || typeof cur !== 'object') return undefined;
+    cur = cur[p];
+  }
+  return cur;
+}
+
+function compare(left, op, right) {
+  switch (op) {
+    case '==': return left == right; // intentional loose equality for convenience
+    case '!=': return left != right;
+    case '<': return left < right;
+    case '<=': return left <= right;
+    case '>': return left > right;
+    case '>=': return left >= right;
+    default: throw new Error(`Unsupported operator: ${op}`);
+  }
+}
+
+export const whereCommand = {
+  name: 'where',
+  help() {
+    return `where — filter objects by a simple predicate\n\nUsage:\n  ... | where unread=true\n  ... | where minutes>=30\n  ... | where sender.domain==example.com\n`;
+  },
+  async run({ input, args }) {
+    const expr = args._[0];
+    if (!expr) throw new Error('where requires an expression (e.g. field=value)');
+    const pred = parsePredicate(expr);
+
+    return {
+      output: (async function* () {
+        for await (const item of input) {
+          const left = getPath(item, pred.path);
+          if (compare(left, pred.op, pred.value)) yield item;
+        }
+      })(),
+    };
+  },
+};
diff --git a/src/parser.js b/src/parser.js
new file mode 100644
index 0000000..ee74bc9
--- /dev/null
+++ b/src/parser.js
@@ -0,0 +1,140 @@
+function isWhitespace(ch) {
+  return ch === ' ' || ch === '\t' || ch === '\n' || ch === '\r';
+}
+
+function splitPipes(input) {
+  const parts = [];
+  let current = '';
+  let quote = null;
+
+  for (let i = 0; i < input.length; i++) {
+    const ch = input[i];
+
+    if (quote) {
+      if (ch === '\\') {
+        const next = input[i + 1];
+        if (next) {
+          current += next;
+          i++;
+          continue;
+        }
+      }
+      if (ch === quote) {
+        quote = null;
+        continue;
+      }
+      current += ch;
+      continue;
+    }
+
+    if (ch === '"' || ch === "'") {
+      quote = ch;
+      continue;
+    }
+
+    if (ch === '|') {
+      parts.push(current.trim());
+      current = '';
+      continue;
+    }
+
+    current += ch;
+  }
+
+  if (quote) throw new Error('Unclosed quote');
+  if (current.trim().length > 0) parts.push(current.trim());
+  return parts;
+}
+
+function tokenizeCommand(input) {
+  const tokens = [];
+  let current = '';
+  let quote = null;
+
+  const push = () => {
+    if (current.length > 0) tokens.push(current);
+    current = '';
+  };
+
+  for (let i = 0; i < input.length; i++) {
+    const ch = input[i];
+
+    if (quote) {
+      if (ch === '\\') {
+        const next = input[i + 1];
+        if (next) {
+          current += next;
+          i++;
+          continue;
+        }
+      }
+      if (ch === quote) {
+        quote = null;
+        continue;
+      }
+      current += ch;
+      continue;
+    }
+
+    if (ch === '"' || ch === "'") {
+      quote = ch;
+      continue;
+    }
+
+    if (isWhitespace(ch)) {
+      push();
+      continue;
+    }
+
+    current += ch;
+  }
+
+  if (quote) throw new Error('Unclosed quote');
+  push();
+  return tokens;
+}
+
+function parseArgs(tokens) {
+  const args = { _: [] };
+
+  for (let i = 0; i < tokens.length; i++) {
+    const tok = tokens[i];
+
+    if (tok.startsWith('--')) {
+      const eq = tok.indexOf('=');
+      if (eq !== -1) {
+        const key = tok.slice(2, eq);
+        const value = tok.slice(eq + 1);
+        args[key] = value;
+        continue;
+      }
+
+      const key = tok.slice(2);
+      const next = tokens[i + 1];
+      if (!next || next.startsWith('--')) {
+        args[key] = true;
+        continue;
+      }
+      args[key] = next;
+      i++;
+      continue;
+    }
+
+    args._.push(tok);
+  }
+
+  return args;
+}
+
+export function parsePipeline(input) {
+  const stages = splitPipes(input);
+  if (stages.length === 0) throw new Error('Empty pipeline');
+
+  return stages.map((stage) => {
+    const tokens = tokenizeCommand(stage);
+    if (tokens.length === 0) throw new Error('Empty command stage');
+    const name = tokens[0];
+    const args = parseArgs(tokens.slice(1));
+    return { name, args, raw: stage };
+  });
+}
diff --git a/src/renderers/json.js b/src/renderers/json.js
new file mode 100644
index 0000000..a98189a
--- /dev/null
+++ b/src/renderers/json.js
@@ -0,0 +1,11 @@
+export function createJsonRenderer(stdout) {
+  return {
+    json(items) {
+      stdout.write(JSON.stringify(items, null, 2));
+      stdout.write('\n');
+    },
+    lines(lines) {
+      for (const line of lines) stdout.write(String(line) + '\n');
+    },
+  };
+}
diff --git a/src/runtime.js b/src/runtime.js
new file mode 100644
index 0000000..bba867a
--- /dev/null
+++ b/src/runtime.js
@@ -0,0 +1,37 @@
+import { createJsonRenderer } from './renderers/json.js';
+
+export async function runPipeline({ pipeline, registry, stdin, stdout, stderr, env }) {
+  let stream = emptyStream();
+  let rendered = false;
+
+  const ctx = {
+    stdin,
+    stdout,
+    stderr,
+    env,
+    registry,
+    render: createJsonRenderer(stdout),
+  };
+
+  for (const stage of pipeline) {
+    const command = registry.get(stage.name);
+    if (!command) {
+      throw new Error(`Unknown command: ${stage.name}`);
+    }
+
+    const result = await command.run({ input: stream, args: stage.args, ctx });
+
+    if (result && result.rendered) {
+      rendered = true;
+    }
+
+    stream = result?.output ?? emptyStream();
+  }
+
+  const items = [];
+  for await (const item of stream) items.push(item);
+
+  return { items, rendered };
+}
+
+async function* emptyStream() {}
diff --git a/test/email_triage.test.js b/test/email_triage.test.js
new file mode 100644
index 0000000..1ca33cd
--- /dev/null
+++ b/test/email_triage.test.js
@@ -0,0 +1,65 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { createDefaultRegistry } from '../src/commands/registry.js';
+import { runPipeline } from '../src/runtime.js';
+
+function streamOf(items) {
+  return (async function* () {
+    for (const item of items) yield item;
+  })();
+}
+
+test('email.triage classifies by keywords deterministically', async () => {
+  const registry = createDefaultRegistry();
+  const pipeline = [{ name: 'email.triage', args: { _: [] }, raw: 'email.triage' }];
+
+  const output = await runPipeline({
+    pipeline,
+    registry,
+    stdin: process.stdin,
+    stdout: process.stdout,
+    stderr: process.stderr,
+    env: process.env,
+  });
+
+  // Default input is empty stream => report total=0.
+  assert.equal(output.items.length, 1);
+  assert.equal(output.items[0].summary.total, 0);
+});
+
+test('email.triage maps custom fields', async () => {
+  const registry = createDefaultRegistry();
+  const pipeline = [
+    {
+      name: 'email.triage',
+      args: {
+        _: [],
+        'subject-field': 'meta.title',
+        'from-field': 'meta.sender',
+        'snippet-field': 'meta.snip',
+        'id-field': 'meta.id',
+      },
+      raw: 'email.triage',
+    },
+  ];
+
+  const output = await (async () => {
+    // Inject custom input by monkey-patching runtime: call command directly.
+    const cmd = registry.get('email.triage');
+    const result = await cmd.run({
+      input: streamOf([
+        { meta: { id: '1', title: 'Invoice due', sender: 'Billing <b@example.com>', snip: 'Pay now' } },
+      ]),
+      args: pipeline[0].args,
+      ctx: { stdin: process.stdin, stdout: process.stdout, stderr: process.stderr, env: process.env, registry, render: { json() {}, lines() {} } },
+    });
+    const items = [];
+    for await (const it of result.output) items.push(it);
+    return items;
+  })();
+
+  assert.equal(output.length, 1);
+  assert.equal(output[0].summary.total, 1);
+  assert.equal(output[0].items[0].bucket, 'needs_action');
+  assert.equal(output[0].items[0].fromEmail, 'b@example.com');
+});
diff --git a/test/parser.test.js b/test/parser.test.js
new file mode 100644
index 0000000..2db7cbf
--- /dev/null
+++ b/test/parser.test.js
@@ -0,0 +1,20 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { parsePipeline } from '../src/parser.js';
+
+test('parsePipeline splits stages and args', () => {
+  const p = parsePipeline("exec echo hi | where a=1 | pick id,subject");
+  assert.equal(p.length, 3);
+  assert.equal(p[0].name, 'exec');
+  assert.deepEqual(p[0].args._, ['echo', 'hi']);
+  assert.equal(p[1].name, 'where');
+  assert.equal(p[1].args._[0], 'a=1');
+  assert.equal(p[2].name, 'pick');
+  assert.equal(p[2].args._[0], 'id,subject');
+});
+
+test('parsePipeline keeps quoted pipes', () => {
+  const p = parsePipeline("exec echo 'a|b' | json");
+  assert.equal(p.length, 2);
+  assert.deepEqual(p[0].args._, ['echo', 'a|b']);
+});