From 646c75de41d0dc95cea7c356590f4c5eaf66cc3c Mon Sep 17 00:00:00 2001 From: Joshua Gilbert <54961107+Joshua-Gilbert@users.noreply.github.com> Date: Fri, 3 Jul 2026 09:58:19 -0700 Subject: [PATCH 1/3] feat(worker): --webhook to POST task transitions to Slack/Discord/generic Add a --webhook flag that POSTs notable worker transitions (a task done/blocked/needs-input/retrying, and the worker stopping/erroring) to a URL, hung off the existing emit() transition point beside --emit-json. One application/json payload serves three receivers with no per-target config: text (Slack), content (Discord), and a structured { event, seq, data, worker, ts } for a generic consumer. Best-effort by construction: a POST never blocks or crashes the drain, a per-request timeout plus an in-flight cap bound resource use against a slow/dead endpoint, and flush() drains pending POSTs before every process exit so the final event still lands. seq is monotonic so a receiver can reorder the concurrently-delivered POSTs. The URL is validated at startup (http/https only) and redacted in logs, since a Slack/Discord URL carries its secret in the path. --webhook-secret HMAC-signs the body (X-Bob-Signature) for a generic receiver to verify; the payload transmits cwd, task titles, and question text, so point it only at a trusted endpoint. --- CHANGELOG.md | 7 ++ README.md | 15 ++- src/webhook.test.ts | 241 ++++++++++++++++++++++++++++++++++++++++++++ src/webhook.ts | 226 +++++++++++++++++++++++++++++++++++++++++ src/worker.ts | 51 +++++++++- 5 files changed, 536 insertions(+), 4 deletions(-) create mode 100644 src/webhook.test.ts create mode 100644 src/webhook.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 5d23a5e..54aff57 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,13 @@ All notable changes to this project are documented here. Format loosely follows - **Published to npm as `@pounceai/bob-control`.** `npx -y @pounceai/bob-control` runs the MCP server standalone via a new `bob-control` bin. The `files` allowlist ships only runtime `dist` (no tests or fixtures), and a `check:shebang` publish gate guards both bin shebangs. +- **`--webhook ` on the worker.** POSTs notable transitions (a task done / blocked / needs-input / + retrying, and the worker stopping / erroring) to a URL as `application/json`. One payload serves Slack + (`text`), Discord (`content`), and a generic receiver (`{ event, seq, data, worker, ts }`, `seq` + monotonic for reordering). Best-effort: a POST never blocks or crashes the drain, an in-flight cap drops + bursts to a slow endpoint, and pending POSTs flush before every exit so the final event lands. The URL + is validated at startup (http/https only) and redacted in logs; `--webhook-secret ` HMAC-signs the + body (`X-Bob-Signature`) for a generic receiver to verify. ### Fixed diff --git a/README.md b/README.md index 0708281..564f218 100644 --- a/README.md +++ b/README.md @@ -312,7 +312,8 @@ Flags: `--once` `--tag ` `--dry-run` `--pipe` `--poll` `--timeout` `--assigne `--no-notify` `--no-defer` `--answer-followups` `--escalate-all` `--review-plans` `--allow-commands ` `--deny-commands ` `--no-command-gate` `--allow-all-commands` `--no-checkpoint` `--no-idle-watchdog` `--idle-timeout ` -`--no-budget` `--budget-cap ` `--max-turns ` (plus the verify-and-continue flags below). +`--no-budget` `--budget-cap ` `--max-turns ` `--webhook ` `--webhook-secret ` +(plus the verify-and-continue flags below). Needs Bob running with IPC enabled (see below). Aborted or timed-out tasks are parked as `blocked`; with `--retry ` the worker first re-dispatches a transient failure (timeout/abort) up to N times before parking it. @@ -328,6 +329,18 @@ Each mode has a risk level, and the worker only dispatches tasks at or below dispatch. On finish the worker pops a tray toast (`--no-notify` to silence; the system sound and terminal bell are off by default). +`--webhook ` POSTs the notable transitions — a task finishing, blocking, needing input, or +retrying, and the worker itself stopping or erroring — to a URL as `application/json`. One payload +serves three consumers: it carries `text` (a Slack incoming webhook renders it), `content` (a Discord +webhook renders it), and the structured `{ event, seq, data, worker, ts }` for a generic receiver +(`seq` monotonic, so concurrent POSTs can be reordered). Delivery is +best-effort: a POST never blocks or crashes the drain, bursts past an in-flight cap to a slow endpoint +are dropped, and the worker flushes pending POSTs before it exits so the final event still lands. The +URL is validated at startup (http/https only) and only its host is logged, since a Slack/Discord webhook +URL carries its secret in the path. The payload transmits the repo path (`cwd`), task titles, and Bob's +question text to that URL — point it only at an endpoint you trust; `--webhook-secret ` HMAC-signs the +body (`X-Bob-Signature: sha256=…`) so a generic receiver can verify authenticity. + **Run it standing (hands-off loop).** Creating a task doesn't start Bob — a worker has to pull it. Keep one **always draining** and the plugin's dispatch skills become end-to-end hands-off: they `create_task`, then `await_task` hooks back with Bob's result in the same turn (no manual dispatch, diff --git a/src/webhook.test.ts b/src/webhook.test.ts new file mode 100644 index 0000000..89a609a --- /dev/null +++ b/src/webhook.test.ts @@ -0,0 +1,241 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { createServer } from "node:http"; +import { once } from "node:events"; +import { createHmac } from "node:crypto"; +import type { AddressInfo } from "node:net"; +import { + buildPayload, + isNotable, + createWebhookSink, + validateWebhookUrl, + redactUrl, + type WebhookMeta, +} from "./webhook.js"; + +const META: WebhookMeta = { cwd: "/repo", assignee: "bob", tag: "rpg" }; + +// A recording fake fetch: captures calls, returns a controllable Response. +function fakeFetch(response: { ok: boolean; status: number } = { ok: true, status: 200 }) { + const calls: Array<{ url: string; init: RequestInit }> = []; + const impl = (async (url: string | URL, init?: RequestInit) => { + calls.push({ url: String(url), init: init ?? {} }); + return response as unknown as Response; + }) as typeof fetch; + return { impl, calls }; +} + +// A fake fetch whose calls stay pending until the test resolves them — for backpressure / flush races. +function gatedFetch() { + const calls: Array<{ url: string; init: RequestInit }> = []; + const resolvers: Array<(r: Response) => void> = []; + const impl = (async (url: string | URL, init?: RequestInit) => { + calls.push({ url: String(url), init: init ?? {} }); + return new Promise((r) => resolvers.push(r)); + }) as typeof fetch; + const releaseAll = () => resolvers.forEach((r) => r({ ok: true, status: 200 } as unknown as Response)); + return { impl, calls, resolvers, releaseAll }; +} + +test("isNotable: notable transitions vs chatty heartbeats", () => { + for (const t of ["taskDone", "taskFail", "taskRetry", "question", "stopped", "error"]) { + assert.equal(isNotable(t), true, t); + } + for (const t of ["idle", "deferred", "resumed", "connected", "taskStart"]) { + assert.equal(isNotable(t), false, t); + } +}); + +test("validateWebhookUrl: accepts http(s), rejects junk + non-http schemes", () => { + assert.equal(validateWebhookUrl("https://hooks.slack.com/services/T/B/X"), null); + assert.equal(validateWebhookUrl("http://127.0.0.1:9000/hook"), null); + assert.match(validateWebhookUrl("not a url") ?? "", /not a valid URL/); + assert.match(validateWebhookUrl("file:///etc/passwd") ?? "", /unsupported scheme/); + assert.match(validateWebhookUrl("ftp://host/x") ?? "", /unsupported scheme/); +}); + +test("redactUrl: drops the credential-bearing path/query/userinfo", () => { + assert.equal(redactUrl("https://hooks.slack.com/services/T00/B00/SECRET"), "https://hooks.slack.com/…"); + assert.equal(redactUrl("https://u:p@example.com/hook?token=abc"), "https://example.com/…"); + assert.equal(redactUrl("https://example.com/"), "https://example.com"); + assert.equal(redactUrl("garbage"), "(unparseable url)"); +}); + +test("buildPayload: same summary in text + content, structured event, seq", () => { + const p = buildPayload("taskDone", { id: 42, title: "Fix it", status: "done", filesChanged: 3 }, META, "T0", 7); + assert.equal(p.text, p.content); // Slack reads text, Discord reads content — identical + assert.ok(p.text.includes("Bob finished #42 (done)")); + assert.ok(p.text.includes("Fix it")); + assert.ok(p.text.includes("3 file(s) changed")); + assert.ok(p.text.includes("[rpg]")); // tag surfaced + assert.equal(p.event, "taskDone"); + assert.equal(p.seq, 7); + assert.equal(p.ts, "T0"); + assert.deepEqual(p.worker, { cwd: "/repo", assignee: "bob", tag: "rpg" }); + assert.equal(p.data.filesChanged, 3); // raw event preserved for generic consumers +}); + +test("buildPayload: status-specific failure phrasing so an operator can tell them apart", () => { + const s = (data: Record) => buildPayload("taskFail", data, META, "T", 0).text; + assert.ok(s({ id: 1, status: "verify-failed" }).includes("failed verification")); + assert.ok(s({ id: 1, status: "timeout" }).includes("timed out")); + assert.ok(s({ id: 1, status: "idle" }).includes("stalled (idle)")); + assert.ok(s({ id: 1, status: "error", message: "boom" }).includes("errored")); + assert.ok(s({ id: 1, status: "error", message: "boom" }).includes("boom")); + assert.ok(s({ id: 1, status: "weird-new" }).includes("failed (weird-new)")); // graceful fallback +}); + +test("buildPayload: other event summaries", () => { + const s = (type: Parameters[0], data: Record) => + buildPayload(type, data, META, "T", 0).text; + assert.ok(s("taskRetry", { id: 2, attempt: 2 }).includes("retrying #2 (attempt 2)")); + assert.ok(s("question", { id: 3, question: "which branch?" }).includes("needs input on #3")); + assert.ok(s("question", { id: 3, question: "which branch?" }).includes("which branch?")); + assert.ok(s("stopped", {}).includes("worker stopped")); + assert.ok(s("error", { message: "lease held" }).includes("worker error")); +}); + +test("sink posts notable events, skips heartbeats", async () => { + const { impl, calls } = fakeFetch(); + const sink = createWebhookSink("http://x/hook", META, { fetchImpl: impl, now: () => "T" }); + sink.post("idle", { gated: 0 }); // skipped + sink.post("connected", { pipe: "p" }); // skipped + sink.post("taskDone", { id: 1, status: "done" }); // posted + await sink.flush(); + assert.equal(calls.length, 1); + assert.equal(calls[0].url, "http://x/hook"); + assert.equal(calls[0].init.method, "POST"); + assert.equal((calls[0].init.headers as Record)["content-type"], "application/json"); + const body = JSON.parse(calls[0].init.body as string); + assert.equal(body.event, "taskDone"); + assert.equal(body.data.id, 1); +}); + +test("seq increments monotonically across posts", async () => { + const { impl, calls } = fakeFetch(); + const sink = createWebhookSink("http://x", META, { fetchImpl: impl }); + sink.post("taskDone", { id: 1, status: "done" }); + sink.post("taskDone", { id: 2, status: "done" }); + sink.post("stopped", {}); + await sink.flush(); + const seqs = calls.map((c) => JSON.parse(c.init.body as string).seq); + assert.deepEqual(seqs, [0, 1, 2]); +}); + +test("HMAC: X-Bob-Signature present + correct when a secret is set, absent otherwise", async () => { + const secret = "s3cr3t"; + const { impl, calls } = fakeFetch(); + const sink = createWebhookSink("http://x", META, { fetchImpl: impl, secret, now: () => "T" }); + sink.post("taskDone", { id: 1, status: "done" }); + await sink.flush(); + const body = calls[0].init.body as string; + const sig = (calls[0].init.headers as Record)["x-bob-signature"]; + const expected = `sha256=${createHmac("sha256", secret).update(body).digest("hex")}`; + assert.equal(sig, expected); + + const plain = fakeFetch(); + const sink2 = createWebhookSink("http://x", META, { fetchImpl: plain.impl }); + sink2.post("taskDone", { id: 1, status: "done" }); + await sink2.flush(); + assert.equal((plain.calls[0].init.headers as Record)["x-bob-signature"], undefined); +}); + +test("backpressure: over the cap, excess POSTs are dropped and warned once", async () => { + const { impl, calls } = gatedFetch(); + const logs: string[] = []; + const sink = createWebhookSink("http://x", META, { fetchImpl: impl, maxInFlight: 2, log: (m) => logs.push(m) }); + sink.post("taskDone", { id: 1, status: "done" }); // in flight + sink.post("taskDone", { id: 2, status: "done" }); // in flight (now at cap) + sink.post("taskDone", { id: 3, status: "done" }); // dropped + sink.post("taskDone", { id: 4, status: "done" }); // dropped, no second warning + assert.equal(calls.length, 2, "only up-to-cap POSTs reach fetch"); + assert.equal(logs.filter((l) => l.includes("dropping")).length, 1, "warned exactly once"); +}); + +test("flush awaits in-flight POSTs", async () => { + let release!: (r: Response) => void; + const gate = new Promise((r) => (release = r)); + const impl = (async () => gate) as typeof fetch; + const sink = createWebhookSink("http://x", META, { fetchImpl: impl }); + sink.post("taskDone", { id: 1, status: "done" }); + let settled = false; + const flushed = sink.flush().then(() => { + settled = true; + }); + await Promise.resolve(); // let microtasks drain; the gate is still blocking (manual release) + assert.equal(settled, false, "flush resolved before the POST settled"); + release({ ok: true, status: 200 } as unknown as Response); + await flushed; + assert.equal(settled, true); +}); + +test("flush drains a POST that arrives while flush is already in progress", async () => { + const g = gatedFetch(); + const sink = createWebhookSink("http://x", META, { fetchImpl: g.impl }); + sink.post("taskDone", { id: 1, status: "done" }); // A: in flight + const flushed = sink.flush(); // snapshots [A], then awaits + sink.post("stopped", {}); // B: arrives DURING the flush — must still be drained + await Promise.resolve(); + g.resolvers[0]({ ok: true, status: 200 } as unknown as Response); // settle A + await Promise.resolve(); + g.resolvers[1]({ ok: true, status: 200 } as unknown as Response); // settle B + await flushed; + assert.equal(g.calls.length, 2, "both A and the during-flush B were delivered"); +}); + +test("flush is wall-clock bounded — a stuck endpoint can't hang exit", async () => { + const impl = (async () => new Promise(() => {})) as typeof fetch; // never settles + const sink = createWebhookSink("http://x", META, { fetchImpl: impl, timeoutMs: 50 }); + sink.post("taskDone", { id: 1, status: "done" }); + await sink.flush(); // resolves via the deadline (timeoutMs + 1000), not the request +}); + +test("a failing POST is swallowed (best-effort), logged not thrown", async () => { + const logs: string[] = []; + const impl = (async () => { + throw new Error("network down"); + }) as typeof fetch; + const sink = createWebhookSink("http://x", META, { fetchImpl: impl, log: (m) => logs.push(m) }); + assert.doesNotThrow(() => sink.post("taskFail", { id: 2, status: "error" })); + await sink.flush(); // must not reject + assert.ok(logs.some((l) => l.includes("network down"))); +}); + +test("a non-2xx response is logged (redacted url), not thrown", async () => { + const logs: string[] = []; + const { impl } = fakeFetch({ ok: false, status: 500 }); + const sink = createWebhookSink("https://hooks.slack.com/services/T/B/SECRET", META, { + fetchImpl: impl, + log: (m) => logs.push(m), + }); + sink.post("stopped", {}); + await sink.flush(); + assert.ok(logs.some((l) => l.includes("HTTP 500"))); + assert.ok(!logs.some((l) => l.includes("SECRET")), "the credential path must not appear in logs"); +}); + +test("delivers over the network to a real endpoint (real global fetch)", async () => { + const received: Array<{ contentType?: string; body: Record }> = []; + const server = createServer((req, res) => { + let buf = ""; + req.on("data", (c) => (buf += c)); + req.on("end", () => { + received.push({ contentType: req.headers["content-type"], body: JSON.parse(buf) }); + res.writeHead(200).end("ok"); + }); + }); + server.listen(0); + await once(server, "listening"); + const { port } = server.address() as AddressInfo; + const sink = createWebhookSink(`http://127.0.0.1:${port}/hook`, META, { now: () => "T" }); + sink.post("taskDone", { id: 7, title: "Ship it", status: "done", filesChanged: 2 }); + await sink.flush(); + server.close(); + assert.equal(received.length, 1); + assert.equal(received[0].contentType, "application/json"); + const body = received[0].body; + assert.equal(body.event, "taskDone"); + assert.equal(body.seq, 0); + assert.equal((body.data as Record).id, 7); + assert.ok(String(body.text).includes("Ship it")); +}); diff --git a/src/webhook.ts b/src/webhook.ts new file mode 100644 index 0000000..d613c52 --- /dev/null +++ b/src/webhook.ts @@ -0,0 +1,226 @@ +import { createHmac } from "node:crypto"; + +// Best-effort webhook sink: fire-and-forget POSTs of notable worker transitions to a URL. One payload +// serves three consumers with no per-target config — `text` (Slack), `content` (Discord), and the +// structured `event`+`data` (a generic receiver) — each ignoring the fields it doesn't use. + +/** The full worker event taxonomy — a closed union so a typo at a call site is a compile error, not a + * silently-dropped POST. NOTABLE_EVENTS is the subset the webhook delivers. */ +export type WorkerEvent = + | "taskStart" + | "taskDone" + | "taskFail" + | "taskRetry" + | "question" + | "idle" + | "deferred" + | "resumed" + | "connected" + | "stopped" + | "error"; + +export interface WebhookMeta { + cwd: string; + assignee: string; + tag?: string; +} + +export interface WebhookSink { + /** Queue a POST for a notable event; a no-op for events not in NOTABLE_EVENTS. Never throws. */ + post(type: WorkerEvent, data: Record): void; + /** Await in-flight POSTs (bounded by the request timeout, then a wall clock). Call before exit. */ + flush(): Promise; +} + +// Events that warrant a push: a task reached a terminal / attention state, or the worker itself +// stopped or errored. The chatty between-poll states (idle / deferred / resumed / connected / taskStart) +// are intentionally excluded — a webhook is for "something happened", not a heartbeat. +export const NOTABLE_EVENTS = new Set([ + "taskDone", + "taskFail", + "taskRetry", + "question", + "stopped", + "error", +]); + +export function isNotable(type: string): boolean { + return NOTABLE_EVENTS.has(type as WorkerEvent); +} + +/** Validate a --webhook value up front so a typo fails loud at startup, not silently at first event. + * Returns an error string, or null if the URL is a well-formed http(s) URL. Rejecting non-http(s) + * schemes (file:, data:, …) also closes the obvious SSRF-by-scheme door. */ +export function validateWebhookUrl(url: string): string | null { + let u: URL; + try { + u = new URL(url); + } catch { + return `not a valid URL: ${url}`; + } + if (u.protocol !== "http:" && u.protocol !== "https:") { + return `unsupported scheme '${u.protocol}' (use http: or https:): ${url}`; + } + return null; +} + +/** Log/display form of a webhook URL with the path, query, and any userinfo stripped — a Slack/Discord + * incoming-webhook URL carries its secret in the path, so the full URL must never reach a log. */ +export function redactUrl(url: string): string { + try { + const u = new URL(url); + const tail = u.pathname && u.pathname !== "/" ? "/…" : ""; + return `${u.protocol}//${u.host}${tail}`; + } catch { + return "(unparseable url)"; + } +} + +export interface WebhookPayload { + text: string; // Slack renders this + content: string; // Discord renders this — intentionally identical to `text` + event: WorkerEvent; + seq: number; // monotonic per-worker; concurrent POSTs can land out of order, so receivers can reorder + data: Record; + worker: { cwd: string; assignee: string; tag?: string }; + ts: string; +} + +export function buildPayload( + type: WorkerEvent, + data: Record, + meta: WebhookMeta, + ts: string, + seq: number, +): WebhookPayload { + const summary = summarize(type, data, meta); + return { + text: summary, + content: summary, + event: type, + seq, + data, + worker: { cwd: meta.cwd, assignee: meta.assignee, tag: meta.tag }, + ts, + }; +} + +// A failure `status` → human phrase, so an operator watching Slack can tell a stall from a timeout from +// a verify miss without opening the board. The structured `data.status` still carries the raw value. +const FAIL_PHRASE: Record = { + "verify-failed": "failed verification", + idle: "stalled (idle)", + timeout: "timed out", + aborted: "was aborted", + error: "errored", + blocked: "blocked", +}; + +/** One-line human summary — the Slack `text` / Discord `content`. */ +function summarize(type: WorkerEvent, data: Record, meta: WebhookMeta): string { + const id = data.id !== undefined ? `#${data.id}` : ""; + const title = typeof data.title === "string" && data.title ? `: ${data.title}` : ""; + const where = meta.tag ? ` [${meta.tag}]` : ""; + switch (type) { + case "taskDone": { + const files = typeof data.filesChanged === "number" ? ` — ${data.filesChanged} file(s) changed` : ""; + return `✓ Bob finished ${id} (${data.status})${title}${files}${where}`; + } + case "taskFail": { + const phrase = FAIL_PHRASE[String(data.status)] ?? `failed (${data.status})`; + const msg = data.message ? ` — ${data.message}` : ""; + return `✗ Bob ${id} ${phrase}${title}${msg}${where}`; + } + case "taskRetry": + return `↻ Bob retrying ${id} (attempt ${data.attempt})${title}${where}`; + case "question": + return `❓ Bob needs input on ${id}${title} — ${data.question ?? ""}${where}`; + case "stopped": + return `■ Bob worker stopped${where}`; + case "error": + return `⚠ Bob worker error${where}: ${data.message ?? ""}`; + default: + return `Bob ${type}${id ? ` ${id}` : ""}`; + } +} + +export interface WebhookOptions { + /** Injectable for tests; defaults to the global fetch. */ + fetchImpl?: typeof fetch; + /** Per-request timeout, ms — long enough for a slow Slack, short enough not to stall exit. Default 5000. */ + timeoutMs?: number; + /** Injectable clock (ISO string); defaults to wall clock. */ + now?: () => string; + /** Diagnostic sink for delivery failures; defaults to stderr. */ + log?: (msg: string) => void; + /** Shared secret: when set, sign the body with HMAC-SHA256 and send it as `X-Bob-Signature: sha256=…` + * so a generic receiver can verify authenticity (Slack/Discord URLs are already pre-authenticated). */ + secret?: string; + /** Backpressure cap on concurrent in-flight POSTs; excess is dropped (best-effort). Default 16. */ + maxInFlight?: number; +} + +const DEFAULT_TIMEOUT_MS = 5000; +const DEFAULT_MAX_IN_FLIGHT = 16; + +export function createWebhookSink(url: string, meta: WebhookMeta, opts: WebhookOptions = {}): WebhookSink { + const doFetch = opts.fetchImpl ?? fetch; + const timeoutMs = opts.timeoutMs ?? DEFAULT_TIMEOUT_MS; + const now = opts.now ?? (() => new Date().toISOString()); + const log = opts.log ?? ((m: string) => console.error(m)); + const maxInFlight = opts.maxInFlight ?? DEFAULT_MAX_IN_FLIGHT; + const safeUrl = redactUrl(url); // never log the credential-bearing path + const inFlight = new Set>(); + let seq = 0; + let warnedOverflow = false; + + return { + post(type, data) { + if (!isNotable(type)) return; + // Backpressure: a slow/dead endpoint must not accumulate unbounded connections + abort timers. + // Dropping under overload is correct for a best-effort sink; warn once per overload episode. + if (inFlight.size >= maxInFlight) { + if (!warnedOverflow) { + log( + `[bob-control] webhook: ${maxInFlight} POSTs in flight to ${safeUrl} (endpoint slow/down?) — dropping until it drains`, + ); + warnedOverflow = true; + } + return; + } + let body: string; + try { + body = JSON.stringify(buildPayload(type, data, meta, now(), seq++)); + } catch (e) { + log(`[bob-control] webhook: could not build payload for ${type}: ${(e as Error).message}`); + return; + } + const headers: Record = { "content-type": "application/json" }; + if (opts.secret) + headers["x-bob-signature"] = `sha256=${createHmac("sha256", opts.secret).update(body).digest("hex")}`; + const p = doFetch(url, { method: "POST", headers, body, signal: AbortSignal.timeout(timeoutMs) }) + .then((res) => { + if (!res.ok) log(`[bob-control] webhook ${safeUrl} → HTTP ${res.status} for ${type}`); + }) + .catch((e) => log(`[bob-control] webhook ${safeUrl} POST failed for ${type}: ${(e as Error).message}`)) + .finally(() => { + inFlight.delete(p); + if (inFlight.size === 0) warnedOverflow = false; // re-arm the overload warning for the next episode + }); + inFlight.add(p); + }, + async flush() { + // Drain until the set is empty — a post() that arrives while we're awaiting is picked up on the next + // loop, so a late final event isn't dropped at the flush boundary. Bounded by a wall clock (a hair + // over the per-request timeout) so a stuck request can never hold the process exit open. + const drain = (async () => { + while (inFlight.size > 0) await Promise.allSettled([...inFlight]); + })(); + const deadline = new Promise((r) => { + const t = setTimeout(r, timeoutMs + 1000); + t.unref?.(); + }); + await Promise.race([drain, deadline]); + }, + }; +} diff --git a/src/worker.ts b/src/worker.ts index 009862d..f23ceb4 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -30,6 +30,7 @@ import { createPollLoop, defaultCaptureSnapshot } from "./bob-polls.js"; import { ExternalActivity } from "./defer.js"; import { PollStatusLatch } from "./worker-status.js"; import { notify } from "./notify.js"; +import { createWebhookSink, validateWebhookUrl, redactUrl, type WebhookSink, type WorkerEvent } from "./webhook.js"; import { shouldRetry, executeRetry } from "./retry-policy.js"; import { buildJudgeVerifier, captureGitBaseline, captureChangedFiles, type GitBaseline } from "./judge.js"; import { captureCheckpoint, preserveWipToBranch, releaseCheckpoint } from "./checkpoint.js"; @@ -82,6 +83,8 @@ function buttonPatchPresent(): boolean | null { * node dist/worker.js --verify-and-continue verify result and loop with Bob until it passes * node dist/worker.js --detect-plan-stop catch plan-only completions (no code written) and auto-continue * node dist/worker.js --emit-json also print @@WORKER {json} event lines (for the extension) + * node dist/worker.js --webhook POST notable transitions (done/blocked/needs-input/…) to a URL (Slack/Discord/generic) + * node dist/worker.js --webhook-secret HMAC-sign the webhook body (X-Bob-Signature) for a generic receiver to verify * node dist/worker.js --retry 3 auto-retry transient failures (timeout/abort) up to 3 total attempts * node dist/worker.js --no-checkpoint don't preserve partial work to a branch on a failed dispatch * node dist/worker.js --no-idle-watchdog disable the idle / blocked-on-ask watchdog (wall-clock only) @@ -171,6 +174,10 @@ export interface Opts { denyCommands: string[]; /** Sandbox escape hatch: auto-run ALL commands (Bob commandPolicy 'auto'); disables the gate. */ allowAllCommands: boolean; + /** POST notable transitions (done/blocked/needs-input/retry/stop/error) to this URL. Off when unset. */ + webhookUrl?: string; + /** Shared secret to HMAC-sign the webhook body (X-Bob-Signature). Off when unset. */ + webhookSecret?: string; } // Watchdog / budget defaults. The blocked-ask grace is the high-value, low-false-positive guard @@ -219,6 +226,15 @@ export function parseOpts(argv: string[]): Opts { .filter((s) => s.length > 0) : []; const allowCommands = csv(val("--allow-commands")); + // Fail loud on a malformed --webhook rather than silently delivering nothing all session. + const webhookUrl = val("--webhook"); + if (webhookUrl !== undefined) { + const err = validateWebhookUrl(webhookUrl); + if (err) { + console.error(`invalid --webhook ${err}`); + process.exit(1); + } + } return { once: has("--once"), newTab, @@ -261,6 +277,8 @@ export function parseOpts(argv: string[]): Opts { permissionGate: !has("--no-command-gate"), denyCommands: csv(val("--deny-commands")), allowAllCommands: has("--allow-all-commands"), + webhookUrl, + webhookSecret: val("--webhook-secret"), }; } @@ -334,9 +352,13 @@ function buildPrompt(task: Task): string { return header + body; } -/** Structured event for the extension (parsed from stdout lines). */ -function emit(opts: Opts, type: string, data: Record = {}): void { +let webhookSink: WebhookSink | null = null; // set at startup when --webhook is passed + +/** Structured event: to the extension over stdout (--emit-json) and/or to a webhook (--webhook). + * `type` is the closed WorkerEvent union, so a mistyped event name is a compile error, not a drop. */ +function emit(opts: Opts, type: WorkerEvent, data: Record = {}): void { if (opts.emitJson) console.log(`@@WORKER ${JSON.stringify({ type, ...data })}`); + webhookSink?.post(type, data); } /** Is a process alive? `process.kill(pid, 0)` sends no signal but throws ESRCH if the pid is gone @@ -972,6 +994,19 @@ export async function main(): Promise { const detail = reason instanceof Error ? (reason.stack ?? reason.message) : String(reason); console.error(`bob-worker: unhandled rejection (ignored): ${detail}`); }); + // Wire the webhook sink before the first emit so even a startup error (lease/connect) is delivered. + if (opts.webhookUrl) { + webhookSink = createWebhookSink( + opts.webhookUrl, + { cwd: process.cwd(), assignee: opts.assignee, tag: opts.tag }, + { secret: opts.webhookSecret }, + ); + // Redact the URL: a Slack/Discord webhook carries its secret in the path. + const signed = opts.webhookSecret ? ", HMAC-signed" : ""; + console.log( + `bob-worker: webhook = on → ${redactUrl(opts.webhookUrl)}${signed} (POSTs done/blocked/needs-input/retry/stop/error).`, + ); + } repo.getDb(); // surface schema errors up front // Worktree lease + heartbeat (T7): at most one live worker per checkout. Keyed on the normalized cwd, @@ -1004,6 +1039,7 @@ export async function main(): Promise { `worker, or (after a hard kill) wait ${Math.ceil(repo.WORKER_HEARTBEAT_WINDOW_MS / 1000)}s for the lease to lapse.`, ); emit(opts, "error", { message: `worktree lease held by pid ${h.pid ?? "?"} on ${process.cwd()}` }); + await webhookSink?.flush(); process.exit(1); } // Lease held (the claim recorded the first beat). startHeartbeat refreshes on a timer; its stop() is @@ -1041,6 +1077,7 @@ export async function main(): Promise { console.error(`bob-worker: could not connect — ${(err as Error).message}`); console.error("Is Bob running, launched WITH ROO_CODE_IPC_SOCKET_PATH set? Try bob-control.mjs --list-pipes"); emit(opts, "error", { message: (err as Error).message }); + await webhookSink?.flush(); process.exit(1); } } @@ -1269,6 +1306,7 @@ export async function main(): Promise { // clean stop, so a supervisor/extension treats it as one (startup error + parked task explain it). if (workspaceMismatch) { await parkWorkspaceMismatch(task, opts, workspaceMismatch); + await webhookSink?.flush(); client.close(); process.exit(1); } @@ -1297,6 +1335,7 @@ export async function main(): Promise { } emit(opts, "stopped", {}); + await webhookSink?.flush(); // let the final POSTs land before the hard exit client.close(); process.exit(0); } @@ -1304,8 +1343,14 @@ export async function main(): Promise { // Auto-run only as a CLI. Importing worker.ts — to reuse parseOpts / pickEligible / main from the 2.0 // in-process driver (or a test) — must have no side effects. Matches cli.ts's is-main guard. if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { - main().catch((err) => { + main().catch(async (err) => { console.error("bob-worker fatal:", err); + // A crash escaping main() is the event an operator most wants pinged — deliver any queued error POST. + try { + await webhookSink?.flush(); + } catch { + /* best-effort on the way out */ + } process.exit(1); }); } From 7874c15bc8488b8e92c8f2a3b257c677145e2425 Mon Sep 17 00:00:00 2001 From: Joshua Gilbert <54961107+Joshua-Gilbert@users.noreply.github.com> Date: Fri, 3 Jul 2026 10:19:50 -0700 Subject: [PATCH 2/3] fix(webhook): keep seq honest on serialize failure; harden flush + exit paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - seq advances only after JSON.stringify succeeds, so a serialization failure (e.g. a BigInt in the event data) no longer burns a sequence number and leaves a phantom gap a receiver would read as a dropped delivery. - validateWebhookUrl no longer echoes the URL in its error, and the parse error is redacted at the call site, so a rejected but secret-bearing URL can't leak to stderr — consistent with the redaction applied everywhere else. - flush() is a single self-bounded loop: it drains in-flight POSTs but its own wall clock (started at flush time) caps the total wait, so a slow endpoint or a steady trickle of new POSTs can't hold process exit open. Each round's timer is cleared rather than leaked, and there's no longer a dangling drain coroutine. - the stdin-end shutdown path flushes before exit; the second-Ctrl-C force-quit deliberately does not (a force path must not block on network I/O). --- src/webhook.test.ts | 13 +++++++++++++ src/webhook.ts | 37 +++++++++++++++++++++++-------------- src/worker.ts | 8 ++++++-- 3 files changed, 42 insertions(+), 16 deletions(-) diff --git a/src/webhook.test.ts b/src/webhook.test.ts index 89a609a..416e504 100644 --- a/src/webhook.test.ts +++ b/src/webhook.test.ts @@ -52,6 +52,9 @@ test("validateWebhookUrl: accepts http(s), rejects junk + non-http schemes", () assert.match(validateWebhookUrl("not a url") ?? "", /not a valid URL/); assert.match(validateWebhookUrl("file:///etc/passwd") ?? "", /unsupported scheme/); assert.match(validateWebhookUrl("ftp://host/x") ?? "", /unsupported scheme/); + // The error must never echo the URL — a rejected-but-secret-bearing value would leak to stderr. + const err = validateWebhookUrl("htp://hooks.slack.com/services/T/B/SECRET") ?? ""; + assert.ok(!err.includes("SECRET") && !err.includes("hooks.slack.com"), "error leaked the URL"); }); test("redactUrl: drops the credential-bearing path/query/userinfo", () => { @@ -122,6 +125,16 @@ test("seq increments monotonically across posts", async () => { assert.deepEqual(seqs, [0, 1, 2]); }); +test("seq is not burned when payload serialization fails", async () => { + const { impl, calls } = fakeFetch(); + const sink = createWebhookSink("http://x", META, { fetchImpl: impl, log: () => {} }); + sink.post("taskDone", { id: 1n }); // BigInt → JSON.stringify throws → dropped; seq must NOT advance + sink.post("taskDone", { id: 2, status: "done" }); // first *sent* POST → seq 0, not 1 + await sink.flush(); + assert.equal(calls.length, 1, "the unserializable event is dropped, not sent"); + assert.equal(JSON.parse(calls[0].init.body as string).seq, 0, "seq was not burned by the failed post"); +}); + test("HMAC: X-Bob-Signature present + correct when a secret is set, absent otherwise", async () => { const secret = "s3cr3t"; const { impl, calls } = fakeFetch(); diff --git a/src/webhook.ts b/src/webhook.ts index d613c52..da9eb98 100644 --- a/src/webhook.ts +++ b/src/webhook.ts @@ -56,10 +56,10 @@ export function validateWebhookUrl(url: string): string | null { try { u = new URL(url); } catch { - return `not a valid URL: ${url}`; + return "not a valid URL"; } if (u.protocol !== "http:" && u.protocol !== "https:") { - return `unsupported scheme '${u.protocol}' (use http: or https:): ${url}`; + return `unsupported scheme '${u.protocol}' (use http: or https:)`; } return null; } @@ -190,14 +190,19 @@ export function createWebhookSink(url: string, meta: WebhookMeta, opts: WebhookO } let body: string; try { - body = JSON.stringify(buildPayload(type, data, meta, now(), seq++)); + // Assign seq only on successful serialization — a stringify failure (e.g. a BigInt in data) must + // not burn a sequence number, or a receiver tracking seq would infer a phantom dropped delivery. + body = JSON.stringify(buildPayload(type, data, meta, now(), seq)); } catch (e) { log(`[bob-control] webhook: could not build payload for ${type}: ${(e as Error).message}`); return; } + seq++; const headers: Record = { "content-type": "application/json" }; if (opts.secret) headers["x-bob-signature"] = `sha256=${createHmac("sha256", opts.secret).update(body).digest("hex")}`; + // p closes over itself in .finally; safe because .finally runs as a microtask, strictly after the + // synchronous inFlight.add(p) below — delete never precedes add. (Don't "simplify" by splitting.) const p = doFetch(url, { method: "POST", headers, body, signal: AbortSignal.timeout(timeoutMs) }) .then((res) => { if (!res.ok) log(`[bob-control] webhook ${safeUrl} → HTTP ${res.status} for ${type}`); @@ -210,17 +215,21 @@ export function createWebhookSink(url: string, meta: WebhookMeta, opts: WebhookO inFlight.add(p); }, async flush() { - // Drain until the set is empty — a post() that arrives while we're awaiting is picked up on the next - // loop, so a late final event isn't dropped at the flush boundary. Bounded by a wall clock (a hair - // over the per-request timeout) so a stuck request can never hold the process exit open. - const drain = (async () => { - while (inFlight.size > 0) await Promise.allSettled([...inFlight]); - })(); - const deadline = new Promise((r) => { - const t = setTimeout(r, timeoutMs + 1000); - t.unref?.(); - }); - await Promise.race([drain, deadline]); + // Drain in-flight POSTs before exit; a post() arriving mid-drain is picked up on the next loop, so a + // late final event isn't dropped at the boundary. Hard-bounded by its own wall clock (started here, + // at flush time) so a slow endpoint — or a steady trickle of new POSTs — can't hold exit open. + const hardDeadline = Date.now() + timeoutMs + 1000; + while (inFlight.size > 0 && Date.now() < hardDeadline) { + let timer: ReturnType | undefined; + await Promise.race([ + Promise.allSettled([...inFlight]), + new Promise((r) => { + timer = setTimeout(r, hardDeadline - Date.now()); + timer.unref?.(); + }), + ]); + clearTimeout(timer); + } }, }; } diff --git a/src/worker.ts b/src/worker.ts index f23ceb4..403bd5d 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -231,7 +231,7 @@ export function parseOpts(argv: string[]): Opts { if (webhookUrl !== undefined) { const err = validateWebhookUrl(webhookUrl); if (err) { - console.error(`invalid --webhook ${err}`); + console.error(`invalid --webhook: ${err}`); // err omits the URL — it may carry a secret path process.exit(1); } } @@ -1109,6 +1109,9 @@ export async function main(): Promise { let stopping = false; const stop = () => { + // Second Ctrl-C = force-quit now: exit immediately, no flush — a force path must not block on network + // I/O (the endpoint may be as wedged as whatever we're killing). The graceful first press drains via + // the normal-stop flush at the end of main(). if (stopping) process.exit(0); stopping = true; console.log("\nbob-worker: finishing current task, then stopping… (Ctrl-C again to force)"); @@ -1133,7 +1136,8 @@ export async function main(): Promise { }); process.stdin.on("end", () => { console.log("bob-worker: parent closed stdin — exiting."); - process.exit(0); + // Deliver any queued webhook POST before exiting (sync handler → flush, then exit). + void webhookSink?.flush().finally(() => process.exit(0)); }); process.stdin.resume(); } From 377b1235816f6f522b93fbab4541f5aad9bcff97 Mon Sep 17 00:00:00 2001 From: Joshua Gilbert <54961107+Joshua-Gilbert@users.noreply.github.com> Date: Fri, 3 Jul 2026 10:32:51 -0700 Subject: [PATCH 3/3] fix(webhook): keep the flush deadline timer ref'd so it fires on Node 22 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An unref'd deadline timer is abandoned when every in-flight POST is a pure pending promise with no socket to hold the event loop: Node 22 tears the loop down and flush() never resolves ("Promise resolution is still pending but the event loop has already resolved"). The timer is cleared right after the race, so it never outlives the call — unref was both unnecessary and the bug. --- src/webhook.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/webhook.ts b/src/webhook.ts index da9eb98..9d3f010 100644 --- a/src/webhook.ts +++ b/src/webhook.ts @@ -223,9 +223,11 @@ export function createWebhookSink(url: string, meta: WebhookMeta, opts: WebhookO let timer: ReturnType | undefined; await Promise.race([ Promise.allSettled([...inFlight]), + // Keep this timer ref'd: it both bounds the wait and, when every in-flight POST is a pure + // pending promise (no socket to hold the loop), is the only thing keeping the loop alive long + // enough for the race to settle. clearTimeout right after means it never outlives the call. new Promise((r) => { timer = setTimeout(r, hardDeadline - Date.now()); - timer.unref?.(); }), ]); clearTimeout(timer);