From 885a086cb2ab5971f3dd8f712012fa3564e43017 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 4 Oct 2026 15:37:29 -0700 Subject: [PATCH 1/4] Add tests for sending to a finished worker --- apps/hub/src/mailbox-send.test.ts | 34 ++++++++++++++++++++++++ apps/web/src/chat/threads-api.test.ts | 37 +++++++++++++++++++++++++++ 2 files changed, 71 insertions(+) create mode 100644 apps/hub/src/mailbox-send.test.ts create mode 100644 apps/web/src/chat/threads-api.test.ts diff --git a/apps/hub/src/mailbox-send.test.ts b/apps/hub/src/mailbox-send.test.ts new file mode 100644 index 000000000..68828c996 --- /dev/null +++ b/apps/hub/src/mailbox-send.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, test } from "bun:test"; +import { Hono } from "hono"; +import { captureMailboxRequest, createMailboxDeliver } from "./mailbox-send"; + +function sendTo(triggerStatus: number, triggerBody: string) { + const deliver = createMailboxDeliver({ + app: { request: () => new Response(triggerBody, { status: triggerStatus }) }, + persistMail: () => Promise.resolve(), + }); + const app = new Hono(); + app.use("*", captureMailboxRequest()); + app.post("/api/tenants/:tenantId/send", async (c) => { + await deliver({ + raw: new TextEncoder().encode("Subject: hi\r\n\r\nhello"), + from: "me@example.test", + to: ["run_abc@example.test"], + messageId: "m1", + }); + return c.json({ ok: true }); + }); + return app.request("/api/tenants/t1/send", { method: "POST" }); +} + +describe("mailbox send to a run", () => { + test("a terminal run surfaces as 409, not a 500", async () => { + const response = await sendTo(409, JSON.stringify({ code: "workflow_run_terminal" })); + expect(response.status).toBe(409); + }); + + test("other trigger failures stay a server error", async () => { + const response = await sendTo(500, "boom"); + expect(response.status).toBe(500); + }); +}); diff --git a/apps/web/src/chat/threads-api.test.ts b/apps/web/src/chat/threads-api.test.ts new file mode 100644 index 000000000..a6616d85d --- /dev/null +++ b/apps/web/src/chat/threads-api.test.ts @@ -0,0 +1,37 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { ChatApiError, sendToWorkbench } from "./threads-api"; + +const realFetch = globalThis.fetch; +afterEach(() => { + globalThis.fetch = realFetch; +}); + +function send(status: number) { + globalThis.fetch = (() => Promise.resolve(new Response("{}", { status }))) as typeof fetch; + return sendToWorkbench({ + workbenchTenantId: "t1", + participants: [ + { id: "a1", kind: "agent", name: "Scout", address: "run_abc@example.test" }, + ], + content: "hello", + }).then( + () => undefined, + (error: unknown) => error, + ); +} + +describe("sendToWorkbench failures", () => { + test("409 says the worker has stopped", async () => { + const error = await send(409); + expect(error).toBeInstanceOf(ChatApiError); + expect((error as ChatApiError).message).toBe( + "This worker has finished its work and can't take new messages.", + ); + expect((error as ChatApiError).status).toBe(409); + }); + + test("other statuses keep the unreachable message", async () => { + const error = await send(500); + expect((error as ChatApiError).message).toBe("The workbench could not be reached (500)."); + }); +}); From 0ef01f92debf8c16f40f29953a7b087518082c17 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 4 Oct 2026 15:39:50 -0700 Subject: [PATCH 2/4] mailbox send: a finished worker answers 409 with a plain message (CL-9908) --- apps/hub/src/mailbox-send.ts | 4 ++++ apps/web/src/chat/threads-api.test.ts | 12 +++++------- apps/web/src/chat/threads-api.ts | 6 ++++++ 3 files changed, 15 insertions(+), 7 deletions(-) diff --git a/apps/hub/src/mailbox-send.ts b/apps/hub/src/mailbox-send.ts index 5dbd959b3..d21eeb0b4 100644 --- a/apps/hub/src/mailbox-send.ts +++ b/apps/hub/src/mailbox-send.ts @@ -6,6 +6,7 @@ import { AsyncLocalStorage } from "node:async_hooks"; import type { Context, MiddlewareHandler } from "hono"; +import { HTTPException } from "hono/http-exception"; import { parseRunAddress } from "@intx/types"; import type { OutgoingMailboxMessage } from "@corbits/mailbox"; import { reportError } from "@corbits/error-sink"; @@ -71,6 +72,9 @@ export function createMailboxDeliver( const err = new Error( `trigger for ${runId} answered ${String(response.status)}: ${detail}`, ); + if (response.status === 409) { + throw new HTTPException(409, { message: detail, cause: err }); + } reportError(err, { operation: "hub.mailboxDeliver.trigger", extra: { runId, tenantId }, diff --git a/apps/web/src/chat/threads-api.test.ts b/apps/web/src/chat/threads-api.test.ts index a6616d85d..014da325c 100644 --- a/apps/web/src/chat/threads-api.test.ts +++ b/apps/web/src/chat/threads-api.test.ts @@ -1,18 +1,16 @@ -import { afterEach, describe, expect, test } from "bun:test"; +import { afterEach, describe, expect, spyOn, test } from "bun:test"; import { ChatApiError, sendToWorkbench } from "./threads-api"; -const realFetch = globalThis.fetch; +const fetchSpy = spyOn(globalThis, "fetch"); afterEach(() => { - globalThis.fetch = realFetch; + fetchSpy.mockReset(); }); function send(status: number) { - globalThis.fetch = (() => Promise.resolve(new Response("{}", { status }))) as typeof fetch; + fetchSpy.mockResolvedValue(new Response("{}", { status })); return sendToWorkbench({ workbenchTenantId: "t1", - participants: [ - { id: "a1", kind: "agent", name: "Scout", address: "run_abc@example.test" }, - ], + participants: [{ id: "a1", kind: "agent", name: "Scout", address: "run_abc@example.test" }], content: "hello", }).then( () => undefined, diff --git a/apps/web/src/chat/threads-api.ts b/apps/web/src/chat/threads-api.ts index f81272231..9bb0394df 100644 --- a/apps/web/src/chat/threads-api.ts +++ b/apps/web/src/chat/threads-api.ts @@ -501,6 +501,12 @@ export async function sendToWorkbench(input: { } catch (cause) { throw new ChatApiError(cause instanceof Error ? cause.message : String(cause)); } + if (response.status === 409) { + throw new ChatApiError( + "This worker has finished its work and can't take new messages.", + response.status, + ); + } if (!response.ok) { throw new ChatApiError( `The workbench could not be reached (${String(response.status)}).`, From 07778b36f7a3e6056af21c7319cb480aa92a4342 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 4 Oct 2026 16:00:52 -0700 Subject: [PATCH 3/4] mailbox send: only workflow_run_terminal reads as a finished agent (CL-9908) --- apps/hub/src/mailbox-send.test.ts | 12 +++- apps/hub/src/mailbox-send.ts | 25 ++++++- apps/web/src/chat/threads-api.test.ts | 19 ++++-- apps/web/src/chat/threads-api.ts | 15 +++-- e2e/hub/mailbox-send-terminal.test.ts | 93 +++++++++++++++++++++++++++ 5 files changed, 150 insertions(+), 14 deletions(-) create mode 100644 e2e/hub/mailbox-send-terminal.test.ts diff --git a/apps/hub/src/mailbox-send.test.ts b/apps/hub/src/mailbox-send.test.ts index 68828c996..1d6863991 100644 --- a/apps/hub/src/mailbox-send.test.ts +++ b/apps/hub/src/mailbox-send.test.ts @@ -2,6 +2,8 @@ import { describe, expect, test } from "bun:test"; import { Hono } from "hono"; import { captureMailboxRequest, createMailboxDeliver } from "./mailbox-send"; +const body = (code: string) => JSON.stringify({ error: { code } }); + function sendTo(triggerStatus: number, triggerBody: string) { const deliver = createMailboxDeliver({ app: { request: () => new Response(triggerBody, { status: triggerStatus }) }, @@ -23,10 +25,18 @@ function sendTo(triggerStatus: number, triggerBody: string) { describe("mailbox send to a run", () => { test("a terminal run surfaces as 409, not a 500", async () => { - const response = await sendTo(409, JSON.stringify({ code: "workflow_run_terminal" })); + const response = await sendTo(409, body("workflow_run_terminal")); expect(response.status).toBe(409); + expect(await response.json()).toMatchObject({ error: { code: "workflow_run_terminal" } }); }); + for (const code of ["deployment_unreachable", "invalid_workflow"]) { + test(`409 ${code} is not reported as finished`, async () => { + const response = await sendTo(409, body(code)); + expect(response.status).toBe(500); + }); + } + test("other trigger failures stay a server error", async () => { const response = await sendTo(500, "boom"); expect(response.status).toBe(500); diff --git a/apps/hub/src/mailbox-send.ts b/apps/hub/src/mailbox-send.ts index d21eeb0b4..491c44b46 100644 --- a/apps/hub/src/mailbox-send.ts +++ b/apps/hub/src/mailbox-send.ts @@ -7,6 +7,7 @@ import { AsyncLocalStorage } from "node:async_hooks"; import type { Context, MiddlewareHandler } from "hono"; import { HTTPException } from "hono/http-exception"; +import { type } from "arktype"; import { parseRunAddress } from "@intx/types"; import type { OutgoingMailboxMessage } from "@corbits/mailbox"; import { reportError } from "@corbits/error-sink"; @@ -27,6 +28,20 @@ function frameBody(raw: Uint8Array): string { return split < 0 ? "" : text.slice(split + 4).trimEnd(); } +const TriggerError = type({ error: { code: "string" } }); + +/** `workflow_run_terminal` is the one 409 that means the run is finished; + * the trigger's other 409s are real failures. */ +function isTerminalRun(status: number, detail: string): boolean { + if (status !== 409) return false; + try { + const parsed = TriggerError(JSON.parse(detail)); + return !(parsed instanceof type.errors) && parsed.error.code === "workflow_run_terminal"; + } catch { + return false; + } +} + export type MailboxDeliverOpts = { /** The hub app itself — the trigger runs through its real route stack, * middleware and grants included. */ @@ -72,8 +87,14 @@ export function createMailboxDeliver( const err = new Error( `trigger for ${runId} answered ${String(response.status)}: ${detail}`, ); - if (response.status === 409) { - throw new HTTPException(409, { message: detail, cause: err }); + if (isTerminalRun(response.status, detail)) { + throw new HTTPException(409, { + res: Response.json( + { error: { code: "workflow_run_terminal", message: "The run has finished." } }, + { status: 409 }, + ), + cause: err, + }); } reportError(err, { operation: "hub.mailboxDeliver.trigger", diff --git a/apps/web/src/chat/threads-api.test.ts b/apps/web/src/chat/threads-api.test.ts index 014da325c..717827515 100644 --- a/apps/web/src/chat/threads-api.test.ts +++ b/apps/web/src/chat/threads-api.test.ts @@ -6,8 +6,10 @@ afterEach(() => { fetchSpy.mockReset(); }); -function send(status: number) { - fetchSpy.mockResolvedValue(new Response("{}", { status })); +function send(status: number, code?: string) { + fetchSpy.mockResolvedValue( + new Response(JSON.stringify(code === undefined ? {} : { error: { code } }), { status }), + ); return sendToWorkbench({ workbenchTenantId: "t1", participants: [{ id: "a1", kind: "agent", name: "Scout", address: "run_abc@example.test" }], @@ -19,11 +21,11 @@ function send(status: number) { } describe("sendToWorkbench failures", () => { - test("409 says the worker has stopped", async () => { - const error = await send(409); + test("a terminal run says the agent has finished", async () => { + const error = await send(409, "workflow_run_terminal"); expect(error).toBeInstanceOf(ChatApiError); expect((error as ChatApiError).message).toBe( - "This worker has finished its work and can't take new messages.", + "This agent has finished and can't take new messages.", ); expect((error as ChatApiError).status).toBe(409); }); @@ -32,4 +34,11 @@ describe("sendToWorkbench failures", () => { const error = await send(500); expect((error as ChatApiError).message).toBe("The workbench could not be reached (500)."); }); + + for (const code of ["deployment_unreachable", "invalid_workflow"]) { + test(`409 ${code} keeps the unreachable message`, async () => { + const error = await send(409, code); + expect((error as ChatApiError).message).toBe("The workbench could not be reached (409)."); + }); + } }); diff --git a/apps/web/src/chat/threads-api.ts b/apps/web/src/chat/threads-api.ts index 9bb0394df..8dd06dc35 100644 --- a/apps/web/src/chat/threads-api.ts +++ b/apps/web/src/chat/threads-api.ts @@ -452,6 +452,12 @@ export function ancestorChain( return chain; } +const TerminalRunBody = type({ error: { code: "'workflow_run_terminal'" } }); + +async function isTerminalRunBody(response: Response): Promise { + return !(TerminalRunBody(await response.json().catch(() => undefined)) instanceof type.errors); +} + // The one send seam for a workbench (see docs/chat-mail-threading.md for // the roster mechanism). export async function sendToWorkbench(input: { @@ -501,13 +507,10 @@ export async function sendToWorkbench(input: { } catch (cause) { throw new ChatApiError(cause instanceof Error ? cause.message : String(cause)); } - if (response.status === 409) { - throw new ChatApiError( - "This worker has finished its work and can't take new messages.", - response.status, - ); - } if (!response.ok) { + if (response.status === 409 && (await isTerminalRunBody(response))) { + throw new ChatApiError("This agent has finished and can't take new messages.", 409); + } throw new ChatApiError( `The workbench could not be reached (${String(response.status)}).`, response.status, diff --git a/e2e/hub/mailbox-send-terminal.test.ts b/e2e/hub/mailbox-send-terminal.test.ts new file mode 100644 index 000000000..dc5dce57f --- /dev/null +++ b/e2e/hub/mailbox-send-terminal.test.ts @@ -0,0 +1,93 @@ +// DB-gated: a terminal run's 409 must survive the real @corbits/mailbox +// mount, which awaits `deliver` with no catch of its own. +import { afterAll, expect, test } from "bun:test"; +import { Hono } from "hono"; +import { createDB } from "@intx/db"; +import { principal, tenant } from "@intx/db/schema"; +import { + createInMemoryMailboxEventBus, + createMailboxDb, + mountMailbox, + runMailboxMigrations, + type MailboxDb, +} from "@corbits/mailbox"; +import { captureMailboxRequest, createMailboxDeliver } from "../../apps/hub/src/mailbox-send"; +import { dbGate } from "../lib/db-gate"; + +const databaseUrl = process.env["DATABASE_URL"] ?? ""; +const describeIfDb = dbGate(databaseUrl, import.meta.path); + +const closers: (() => Promise)[] = []; +afterAll(async () => { + for (const close of closers) await close(); +}); + +function dbConfigFromUrl(url: string) { + const parsed = new URL(url); + return { + host: parsed.hostname, + port: parsed.port === "" ? 5432 : Number(parsed.port), + user: decodeURIComponent(parsed.username), + password: decodeURIComponent(parsed.password), + database: parsed.pathname.replace(/^\//, ""), + }; +} + +describeIfDb("mailbox send through the real mount", () => { + async function sendToRun(triggerStatus: number, code: string) { + const { db, close } = createMailboxDb(databaseUrl) as { + db: MailboxDb; + close: () => Promise; + }; + closers.push(close); + await runMailboxMigrations(db); + const { db: hubDb, close: closeHub } = createDB(dbConfigFromUrl(databaseUrl)); + closers.push(closeHub); + const suffix = crypto.randomUUID().slice(0, 8); + const tenantId = `tnt_term_${suffix}`; + const principalId = `prn_term_${suffix}`; + await hubDb.insert(tenant).values({ + id: tenantId, + name: "Terminal Run Tenant", + slug: `term-${suffix}`, + domain: `term-${suffix}.test`, + }); + await hubDb + .insert(principal) + .values({ id: principalId, tenantId, kind: "user", refId: principalId, status: "active" }); + const app = new Hono(); + app.use("/tenants/:tenantId/mailbox/me/inbox/send", captureMailboxRequest()); + const mailboxApp = new Hono(); + mailboxApp.use("/me/inbox/send", captureMailboxRequest()); + mountMailbox(mailboxApp, { + db, + bus: createInMemoryMailboxEventBus(), + resolvePrincipal: () => ({ tenantId, principalId }), + senderAddressFor: () => "person@term.test", + deliver: createMailboxDeliver({ + app: { + request: () => + Response.json({ error: { code, message: "x" } }, { status: triggerStatus }), + }, + persistMail: () => Promise.resolve(), + }), + }); + app.route("/tenants/:tenantId/mailbox", mailboxApp); + return app.request(`/tenants/${tenantId}/mailbox/me/inbox/send`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ to: ["run_abc@term.test"], subject: "hi", body: "hello" }), + }); + } + + test("workflow_run_terminal answers 409 with its code", async () => { + const response = await sendToRun(409, "workflow_run_terminal"); + expect(response.status).toBe(409); + expect(await response.json()).toMatchObject({ error: { code: "workflow_run_terminal" } }); + }); + + test("deployment_unreachable stays a 500", async () => { + const response = await sendToRun(409, "deployment_unreachable"); + expect(response.status).toBe(500); + }); +}); From 625aa6a4bc55a671d0070584596b9adabfbfb6dd Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 4 Oct 2026 16:03:12 -0700 Subject: [PATCH 4/4] mailbox send: parse the trigger error body without a bare catch (CL-9908) --- apps/hub/src/mailbox-send.test.ts | 2 +- apps/hub/src/mailbox-send.ts | 10 +++------- 2 files changed, 4 insertions(+), 8 deletions(-) diff --git a/apps/hub/src/mailbox-send.test.ts b/apps/hub/src/mailbox-send.test.ts index 1d6863991..697b77bd6 100644 --- a/apps/hub/src/mailbox-send.test.ts +++ b/apps/hub/src/mailbox-send.test.ts @@ -38,7 +38,7 @@ describe("mailbox send to a run", () => { } test("other trigger failures stay a server error", async () => { - const response = await sendTo(500, "boom"); + const response = await sendTo(409, "not json"); expect(response.status).toBe(500); }); }); diff --git a/apps/hub/src/mailbox-send.ts b/apps/hub/src/mailbox-send.ts index 491c44b46..eb7ca95d8 100644 --- a/apps/hub/src/mailbox-send.ts +++ b/apps/hub/src/mailbox-send.ts @@ -28,18 +28,14 @@ function frameBody(raw: Uint8Array): string { return split < 0 ? "" : text.slice(split + 4).trimEnd(); } -const TriggerError = type({ error: { code: "string" } }); +const TriggerError = type("string.json.parse").pipe(type({ error: { code: "string" } })); /** `workflow_run_terminal` is the one 409 that means the run is finished; * the trigger's other 409s are real failures. */ function isTerminalRun(status: number, detail: string): boolean { if (status !== 409) return false; - try { - const parsed = TriggerError(JSON.parse(detail)); - return !(parsed instanceof type.errors) && parsed.error.code === "workflow_run_terminal"; - } catch { - return false; - } + const parsed = TriggerError(detail); + return !(parsed instanceof type.errors) && parsed.error.code === "workflow_run_terminal"; } export type MailboxDeliverOpts = {