Merge pull request #2 from clawdbot/feat/email-triage-v1

src/commands/registry.ts

@@ -10,6 +10,9 @@ import { stateGetCommand, stateSetCommand } from "./stdlib/state.js";
 import { diffLastCommand } from "./stdlib/diff_last.js";
 import { workflowsListCommand } from "./workflows/workflows_list.js";
 import { workflowsRunCommand } from "./workflows/workflows_run.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();
@@ -28,6 +31,9 @@ export function createDefaultRegistry() {
     diffLastCommand,
     workflowsListCommand,
     workflowsRunCommand,
+    gogGmailSearchCommand,
+    gogGmailSendCommand,
+    emailTriageCommand,
   ]) {
     commands.set(cmd.name, cmd);
   }

src/commands/stdlib/email_triage.ts

@@ -0,0 +1,105 @@
+type EmailLike = {
+  id?: string;
+  threadId?: string;
+  from?: string;
+  subject?: string;
+  date?: string;
+  snippet?: string;
+  labels?: string[];
+};
+
+function normalizeEmail(raw: any): Required<Pick<EmailLike, "id" | "threadId" | "from" | "subject" | "date" | "snippet">> & {
+  labels: string[];
+} {
+  const id = String(raw?.id ?? raw?.messageId ?? "").trim();
+  const threadId = String(raw?.threadId ?? raw?.thread_id ?? id).trim();
+  const from = String(raw?.from ?? raw?.sender ?? "").trim();
+  const subject = String(raw?.subject ?? "").trim();
+  const date = String(raw?.date ?? raw?.internalDate ?? raw?.timestamp ?? "").trim();
+  const snippet = String(raw?.snippet ?? raw?.bodyPreview ?? "").trim();
+  const labels = Array.isArray(raw?.labels) ? raw.labels.map((x: any) => String(x)) : [];
+
+  return {
+    id,
+    threadId: threadId || id,
+    from,
+    subject,
+    date,
+    snippet,
+    labels,
+  };
+}
+
+function isLikelyNoReply(from: string) {
+  const f = from.toLowerCase();
+  return (
+    f.includes("no-reply") ||
+    f.includes("noreply") ||
+    f.includes("do-not-reply") ||
+    f.includes("donotreply")
+  );
+}
+
+export const emailTriageCommand = {
+  name: "email.triage",
+  help() {
+    return (
+      `email.triage — deterministic bucketing + summary for email messages\n\n` +
+      `Usage:\n` +
+      `  gog.gmail.search --query 'newer_than:1d' --max 20 | email.triage\n\n` +
+      `Output:\n` +
+      `  Single object: { summary, buckets, emails }\n\n` +
+      `Notes:\n` +
+      `  - Read-only: does not send anything.\n` +
+      `  - Combine with an agent step to draft replies, then approve + gog.gmail.send.\n`
+    );
+  },
+  async run({ input, args }) {
+    const limit = Number(args.limit ?? 20);
+
+    const emails: ReturnType<typeof normalizeEmail>[] = [];
+    for await (const item of input) {
+      emails.push(normalizeEmail(item));
+      if (emails.length >= limit) break;
+    }
+
+    const buckets = {
+      needsReply: [] as any[],
+      needsAction: [] as any[],
+      fyi: [] as any[],
+    };
+
+    for (const e of emails) {
+      const subjLower = e.subject.toLowerCase();
+      const unread = e.labels.some((l) => l.toUpperCase() === "UNREAD");
+
+      if (subjLower.includes("action required") || subjLower.includes("urgent")) {
+        buckets.needsAction.push(e);
+        continue;
+      }
+
+      if (unread && !isLikelyNoReply(e.from)) {
+        buckets.needsReply.push(e);
+        continue;
+      }
+
+      buckets.fyi.push(e);
+    }
+
+    const summary = `${buckets.needsReply.length} need replies, ${buckets.needsAction.length} need action, ${buckets.fyi.length} FYI`;
+
+    return {
+      output: (async function* () {
+        yield {
+          summary,
+          buckets: {
+            needsReply: buckets.needsReply.map((x) => x.id),
+            needsAction: buckets.needsAction.map((x) => x.id),
+            fyi: buckets.fyi.map((x) => x.id),
+          },
+          emails,
+        };
+      })(),
+    };
+  },
+};

src/commands/stdlib/gog_gmail_search.ts

@@ -0,0 +1,92 @@
+import { spawn } from "node:child_process";
+
+function run(cmd: string, argv: string[], env: Record<string, string | undefined>, cwd?: string) {
+  return new Promise<{ stdout: string; stderr: string; code: number | null }>((resolve, reject) => {
+    const child = spawn(cmd, argv, {
+      env: { ...process.env, ...env },
+      cwd,
+      stdio: ["ignore", "pipe", "pipe"],
+    });
+
+    let stdout = "";
+    let stderr = "";
+    child.stdout?.on("data", (d) => (stdout += String(d)));
+    child.stderr?.on("data", (d) => (stderr += String(d)));
+
+    child.on("error", (err: any) => {
+      if (err?.code === "ENOENT") {
+        reject(new Error("gog not found on PATH (install: https://github.com/steipete/gog)"));
+        return;
+      }
+      reject(err);
+    });
+
+    child.on("close", (code) => {
+      resolve({ stdout, stderr, code });
+    });
+  });
+}
+
+export const gogGmailSearchCommand = {
+  name: "gog.gmail.search",
+  help() {
+    return (
+      `gog.gmail.search — fetch Gmail messages via gog (JSON)\n\n` +
+      `Usage:\n` +
+      `  gog.gmail.search --query 'newer_than:1d' --max 20\n\n` +
+      `Notes:\n` +
+      `  - Requires the gog CLI: https://github.com/steipete/gog\n` +
+      `  - Set GOG_BIN to override the executable used (default: gog).\n` +
+      `  - This command outputs an array of message objects (as a stream).\n`
+    );
+  },
+  async run({ input, args, ctx }) {
+    // Drain input
+    for await (const _item of input) {
+      // no-op
+    }
+
+    const query = String(args.query ?? "newer_than:1d");
+    const max = Number(args.max ?? args.limit ?? 20);
+
+    const gogBinRaw = String(ctx.env.GOG_BIN ?? "gog");
+
+    // gog CLI (v0.9.x) expects the query as a positional argument:
+    //   gog gmail search <query> ... --json
+    // Earlier draft versions used --query; keep Lobster's --query flag but translate it.
+    const argvBase = ["gmail", "search", query, "--json", "--max", String(max)];
+
+    // Test-friendly: allow pointing GOG_BIN at a node script.
+    const isScript = /\.(mjs|cjs|js|ts)$/i.test(gogBinRaw);
+    const gogBin = isScript ? process.execPath : gogBinRaw;
+    const argv = isScript ? [gogBinRaw, ...argvBase] : argvBase;
+
+    const res = await run(gogBin, argv, ctx.env, process.cwd());
+    if (res.code !== 0) {
+      throw new Error(`gog.gmail.search failed (${res.code ?? "?"}): ${res.stderr.slice(0, 400)}`);
+    }
+
+    let parsed: any;
+    try {
+      parsed = JSON.parse(res.stdout);
+    } catch (_err) {
+      throw new Error("gog.gmail.search expected JSON output");
+    }
+
+    // gog gmail search --json returns either:
+    // - an array of message/thread objects (older versions / some commands), or
+    // - an object like { nextPageToken, threads: [...] } (gog v0.9.x).
+    const items = Array.isArray(parsed)
+      ? // Some gog versions return: [ { nextPageToken, threads: [...] } ]
+        (parsed as any[]).flatMap((x) => (Array.isArray(x?.threads) ? x.threads : [x]))
+      : Array.isArray((parsed as any)?.threads)
+        ? (parsed as any).threads
+        : [parsed];
+
+    return {
+      output: (async function* () {
+        for (const item of items) yield item;
+      })(),
+    };
+  },
+};

src/commands/stdlib/gog_gmail_send.ts

@@ -0,0 +1,109 @@
+import { spawn } from "node:child_process";
+
+function run(cmd: string, argv: string[], env: Record<string, string | undefined>, cwd?: string) {
+  return new Promise<{ stdout: string; stderr: string; code: number | null }>((resolve, reject) => {
+    const child = spawn(cmd, argv, {
+      env: { ...process.env, ...env },
+      cwd,
+      stdio: ["ignore", "pipe", "pipe"],
+    });
+
+    let stdout = "";
+    let stderr = "";
+    child.stdout?.on("data", (d) => (stdout += String(d)));
+    child.stderr?.on("data", (d) => (stderr += String(d)));
+
+    child.on("error", (err: any) => {
+      if (err?.code === "ENOENT") {
+        reject(new Error("gog not found on PATH (install: https://github.com/steipete/gog)"));
+        return;
+      }
+      reject(err);
+    });
+
+    child.on("close", (code) => {
+      resolve({ stdout, stderr, code });
+    });
+  });
+}
+
+type Draft = {
+  to: string;
+  subject: string;
+  body: string;
+};
+
+function parseDraft(item: any): Draft {
+  if (!item || typeof item !== "object") {
+    throw new Error("gog.gmail.send expects draft objects");
+  }
+  const to = String((item as any).to ?? "").trim();
+  const subject = String((item as any).subject ?? "").trim();
+  const body = String((item as any).body ?? "").trim();
+  if (!to) throw new Error("gog.gmail.send draft missing to");
+  return { to, subject, body };
+}
+
+export const gogGmailSendCommand = {
+  name: "gog.gmail.send",
+  help() {
+    return (
+      `gog.gmail.send — send Gmail messages via gog\n\n` +
+      `Usage:\n` +
+      `  ... | approve --prompt 'Send replies?' | gog.gmail.send\n\n` +
+      `Input:\n` +
+      `  Stream of draft objects: { to, subject, body }\n\n` +
+      `Notes:\n` +
+      `  - Requires the gog CLI: https://github.com/steipete/gog\n` +
+      `  - Set GOG_BIN to override the executable used (default: gog).\n`
+    );
+  },
+  async run({ input, args, ctx }) {
+    const dryRun = Boolean(args.dryRun ?? args["dry-run"] ?? false);
+    const gogBinRaw = String(ctx.env.GOG_BIN ?? "gog");
+    const isScript = /\.(mjs|cjs|js|ts)$/i.test(gogBinRaw);
+    const gogBin = isScript ? process.execPath : gogBinRaw;
+
+    const results: any[] = [];
+
+    for await (const item of input) {
+      const draft = parseDraft(item);
+
+      if (dryRun) {
+        results.push({ ok: true, dryRun: true, ...draft });
+        continue;
+      }
+
+      const argvBase = [
+        "gmail",
+        "send",
+        "--to",
+        draft.to,
+        ...(draft.subject ? ["--subject", draft.subject] : []),
+        ...(draft.body ? ["--body", draft.body] : []),
+        "--json",
+      ];
+
+      const argv = isScript ? [gogBinRaw, ...argvBase] : argvBase;
+      const res = await run(gogBin, argv, ctx.env, process.cwd());
+      if (res.code !== 0) {
+        throw new Error(`gog.gmail.send failed (${res.code ?? "?"}): ${res.stderr.slice(0, 400)}`);
+      }
+
+      let parsed: any;
+      try {
+        parsed = res.stdout ? JSON.parse(res.stdout) : { ok: true };
+      } catch (_err) {
+        parsed = { ok: true, raw: res.stdout };
+      }
+
+      results.push(parsed);
+    }
+
+    return {
+      output: (async function* () {
+        for (const r of results) yield r;
+      })(),
+    };
+  },
+};

test/email_triage.test.ts

@@ -0,0 +1,91 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import { fileURLToPath } from "node:url";
+import { dirname, join } from "node:path";
+
+import { createDefaultRegistry } from "../src/commands/registry.js";
+import { runPipeline } from "../src/runtime.js";
+
+const __filename = fileURLToPath(import.meta.url);
+const __dirname = dirname(__filename);
+
+test("gog.gmail.search | email.triage works end-to-end (mock gog)", async () => {
+  const registry = createDefaultRegistry();
+
+  // Tests run from dist/, but fixtures live in source tree.
+  const repoRoot = join(__dirname, "..", "..");
+  const mockGog = join(repoRoot, "test", "fixtures", "mock-gog.mjs");
+
+  const result = await runPipeline({
+    pipeline: [
+      { name: "gog.gmail.search", args: { query: "newer_than:1d", max: 20 }, raw: "" },
+      { name: "email.triage", args: { limit: 20 }, raw: "" },
+    ],
+    registry,
+    input: [],
+    stdin: process.stdin,
+    stdout: process.stdout,
+    stderr: process.stderr,
+    env: { ...process.env, GOG_BIN: mockGog },
+    mode: "tool",
+  } as any);
+
+  assert.equal(result.items.length, 1);
+  assert.equal(result.items[0].summary, "1 need replies, 1 need action, 1 FYI");
+});
+
+test("email.triage buckets based on subject/from/labels", async () => {
+  const registry = createDefaultRegistry();
+
+  const emails = [
+    {
+      id: "m1",
+      threadId: "t1",
+      from: "Alice <alice@example.com>",
+      subject: "Quick question",
+      date: "2026-01-22T07:00:00Z",
+      snippet: "Hey, can you take a look?",
+      labels: ["INBOX", "UNREAD"],
+    },
+    {
+      id: "m2",
+      threadId: "t2",
+      from: "no-reply@service.com",
+      subject: "Your receipt",
+      date: "2026-01-22T06:00:00Z",
+      snippet: "Thanks",
+      labels: ["INBOX", "UNREAD"],
+    },
+    {
+      id: "m3",
+      threadId: "t3",
+      from: "Bob <bob@example.com>",
+      subject: "Action required: NDA",
+      date: "2026-01-21T23:00:00Z",
+      snippet: "Please sign",
+      labels: ["INBOX"],
+    },
+  ];
+
+  const input = (async function* () {
+    for (const e of emails) yield e;
+  })();
+
+  const result = await runPipeline({
+    pipeline: [{ name: "email.triage", args: { limit: 20 }, raw: "" }],
+    registry,
+    input,
+    stdin: process.stdin,
+    stdout: process.stdout,
+    stderr: process.stderr,
+    env: process.env,
+    mode: "tool",
+  } as any);
+
+  assert.equal(result.items.length, 1);
+  const out = result.items[0];
+  assert.equal(out.summary, "1 need replies, 1 need action, 1 FYI");
+  assert.deepEqual(out.buckets.needsReply, ["m1"]);
+  assert.deepEqual(out.buckets.needsAction, ["m3"]);
+  assert.deepEqual(out.buckets.fyi, ["m2"]);
+});

test/fixtures/gog_gmail_search.json

@@ -0,0 +1,29 @@
+[
+  {
+    "id": "m1",
+    "threadId": "t1",
+    "from": "Alice <alice@example.com>",
+    "subject": "Quick question",
+    "date": "2026-01-22T07:00:00Z",
+    "snippet": "Hey, can you take a look?",
+    "labels": ["INBOX", "UNREAD"]
+  },
+  {
+    "id": "m2",
+    "threadId": "t2",
+    "from": "no-reply@service.com",
+    "subject": "Your receipt",
+    "date": "2026-01-22T06:00:00Z",
+    "snippet": "Thanks for your purchase",
+    "labels": ["INBOX", "UNREAD"]
+  },
+  {
+    "id": "m3",
+    "threadId": "t3",
+    "from": "Bob <bob@example.com>",
+    "subject": "Action required: NDA",
+    "date": "2026-01-21T23:00:00Z",
+    "snippet": "Please sign",
+    "labels": ["INBOX"]
+  }
+]

test/fixtures/mock-gog.mjs

@@ -0,0 +1,25 @@
+#!/usr/bin/env node
+import { readFileSync } from 'node:fs';
+import { fileURLToPath } from 'node:url';
+import { dirname, join } from 'node:path';
+
+const __filename = fileURLToPath(import.meta.url);
+const __dirname = dirname(__filename);
+
+const argv = process.argv.slice(2);
+
+// Minimal mock for `gog gmail search` and `gog gmail send`.
+if (argv[0] === 'gmail' && argv[1] === 'search') {
+  const data = readFileSync(join(__dirname, 'gog_gmail_search.json'), 'utf8');
+  process.stdout.write(data);
+  process.exit(0);
+}
+
+if (argv[0] === 'gmail' && argv[1] === 'send') {
+  // Echo a json success object.
+  process.stdout.write(JSON.stringify({ ok: true }));
+  process.exit(0);
+}
+
+process.stderr.write('mock-gog: unsupported args: ' + argv.join(' ') + '\n');
+process.exit(2);