diff --git a/apps/hub/src/mailbox-send.test.ts b/apps/hub/src/mailbox-send.test.ts new file mode 100644 index 000000000..697b77bd6 --- /dev/null +++ b/apps/hub/src/mailbox-send.test.ts @@ -0,0 +1,44 @@ +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 }) }, + 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, 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(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 5dbd959b3..eb7ca95d8 100644 --- a/apps/hub/src/mailbox-send.ts +++ b/apps/hub/src/mailbox-send.ts @@ -6,6 +6,8 @@ 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"; @@ -26,6 +28,16 @@ function frameBody(raw: Uint8Array): string { return split < 0 ? "" : text.slice(split + 4).trimEnd(); } +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; + const parsed = TriggerError(detail); + return !(parsed instanceof type.errors) && parsed.error.code === "workflow_run_terminal"; +} + export type MailboxDeliverOpts = { /** The hub app itself — the trigger runs through its real route stack, * middleware and grants included. */ @@ -71,6 +83,15 @@ export function createMailboxDeliver( const err = new Error( `trigger for ${runId} answered ${String(response.status)}: ${detail}`, ); + 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", extra: { runId, tenantId }, 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..717827515 --- /dev/null +++ b/apps/web/src/chat/threads-api.test.ts @@ -0,0 +1,44 @@ +import { afterEach, describe, expect, spyOn, test } from "bun:test"; +import { ChatApiError, sendToWorkbench } from "./threads-api"; + +const fetchSpy = spyOn(globalThis, "fetch"); +afterEach(() => { + fetchSpy.mockReset(); +}); + +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" }], + content: "hello", + }).then( + () => undefined, + (error: unknown) => error, + ); +} + +describe("sendToWorkbench failures", () => { + 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 agent has finished 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)."); + }); + + 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 f81272231..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: { @@ -502,6 +508,9 @@ export async function sendToWorkbench(input: { throw new ChatApiError(cause instanceof Error ? cause.message : String(cause)); } 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); + }); +});