mailbox send: a finished worker answers 409 with a plain message (CL-9908)

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);
+  });
+});

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 },

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).");
+    });
+  }
+});

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<boolean> {
+  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,

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<void>)[] = [];
+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<void>;
+    };
+    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);
+  });
+});