CL-6644: replace per-hop wake/mail bounds with one turn-level dispatch deadline

packages/chat/src/index.ts

@@ -217,6 +217,8 @@ export type {
 } from "./run-participant";
 export {
   dispatchTurn,
+  DEFAULT_TURN_DISPATCH_TIMEOUT_MS,
+  turnDispatchTimeoutMessage,
   launchAndJoinAgent,
   postCannedGreeting,
   cannedGreeting,

packages/chat/src/platform-adapter.ts

@@ -52,6 +52,7 @@ import { getLogger } from "@intx/log";
 import { extractPartByPath } from "@intx/mime";
 import { workbenchLaunch } from "./schema";
 import { isWorkbenchHostDefinitionName } from "./workbench-host-naming";
+import { withTimeout } from "./with-timeout";
 import type { EventCollectorRegistry, SidecarRouter } from "@intx/hub-sessions";
 import type { InferencePreference } from "@intx/agent";
 import { formatRunAddress } from "@intx/types";
@@ -268,26 +269,6 @@ export function createHubChatPlatform(
     return new Promise((resolveSleep) => setTimeout(resolveSleep, ms));
   }
 
-  function withTimeout<T>(
-    promise: Promise<T>,
-    ms: number,
-    message: string,
-  ): Promise<T> {
-    return new Promise((resolve, reject) => {
-      const timer = setTimeout(() => reject(new Error(message)), ms);
-      promise.then(
-        (value) => {
-          clearTimeout(timer);
-          resolve(value);
-        },
-        (cause: unknown) => {
-          clearTimeout(timer);
-          reject(cause);
-        },
-      );
-    });
-  }
-
   function isAgentUnreachable(err: unknown): boolean {
     return err instanceof Error && err.message.includes("agent is unreachable");
   }

packages/chat/src/routes.ts

@@ -147,6 +147,12 @@ export type CreateChatRoutesDeps = {
   isInvitableDefinition: (definition: InvitableDefinitionRecord) => boolean;
   /** Per-turn timeout, the default write-claim TTL. */
   turnTimeoutMs: number;
+  /**
+   * CL-6644's turn-level deadline: see `SendWorkbenchMessageDeps`'s field
+   * of the same name in `./workbench-service.ts`. Omitted,
+   * `dispatchTurnBatch` uses `DEFAULT_TURN_DISPATCH_TIMEOUT_MS`.
+   */
+  turnDispatchTimeoutMs?: number;
   /**
    * Resolves a principal to the display name a greeting can use. The
    * hub wires this to its user table; omitted, the canned greeting
@@ -2163,6 +2169,9 @@ export function createChatRoutes(deps: CreateChatRoutesDeps): Hono<TenantEnv> {
             ? { agentTurns: deps.agentTurns }
             : {}),
           ...(deps.threads !== undefined ? { threads: deps.threads } : {}),
+          ...(deps.turnDispatchTimeoutMs !== undefined
+            ? { turnDispatchTimeoutMs: deps.turnDispatchTimeoutMs }
+            : {}),
         },
         {
           tenantId: ownerTenantId,
@@ -2335,6 +2344,9 @@ export function createChatRoutes(deps: CreateChatRoutesDeps): Hono<TenantEnv> {
               ? { agentTurns: deps.agentTurns }
               : {}),
             ...(deps.threads !== undefined ? { threads: deps.threads } : {}),
+            ...(deps.turnDispatchTimeoutMs !== undefined
+              ? { turnDispatchTimeoutMs: deps.turnDispatchTimeoutMs }
+              : {}),
           },
           {
             tenantId: ownerTenantId,

packages/chat/src/with-timeout.ts

@@ -0,0 +1,25 @@
+// A wall-clock bound on a promise that may never settle at all — the
+// shape #316 first wrote inline in `platform-adapter.ts` to catch a
+// wedged post-deploy mail ack, and CL-6644's turn-level deadline in
+// `workbench-service.ts` reuses rather than duplicates. `promise`
+// rejecting or resolving on its own always wins the race; the timer
+// only fires when neither ever happens.
+export function withTimeout<T>(
+  promise: Promise<T>,
+  ms: number,
+  message: string,
+): Promise<T> {
+  return new Promise((resolve, reject) => {
+    const timer = setTimeout(() => reject(new Error(message)), ms);
+    promise.then(
+      (value) => {
+        clearTimeout(timer);
+        resolve(value);
+      },
+      (cause: unknown) => {
+        clearTimeout(timer);
+        reject(cause);
+      },
+    );
+  });
+}

packages/chat/src/workbench-service.ts

@@ -47,6 +47,7 @@ import type { WorkbenchSubscriberRegistry } from "./workbench-events";
 import type { QueuedTurn, WorkbenchTurnQueue } from "./turn-queue";
 import type { WorkbenchTenancyStore } from "./workbench-tenancy";
 import type { ChatStore } from "./store";
+import { withTimeout } from "./with-timeout";
 
 const provisionLog = getLogger(["chat", "provision-space"]);
 const removeLog = getLogger(["chat", "remove-participant"]);
@@ -923,8 +924,40 @@ export type SendWorkbenchMessageDeps = {
    * with the whole room. See `./turn-context.ts`.
    */
   readonly threads?: Pick<ThreadStore, "listThreadAssignments">;
+  /**
+   * The turn-level deadline (CL-6644): `dispatchTurnBatch` wraps every
+   * recipient's `dispatchTurn` call in this single wall-clock budget,
+   * defaulting to `DEFAULT_TURN_DISPATCH_TIMEOUT_MS`. This is the
+   * structural fix three rounds of per-hop timeouts (#312's wake bound,
+   * #314's bypassed-wake bound, #316's mail-delivery bound) kept falling
+   * short of: each closed one stalling hop and a new one appeared next
+   * (see the CL-6644 comment isolating a fourth hang — a direct send to
+   * an already-live run — that needed neither #312's nor #316's bound).
+   * No agent turn may hang past this budget regardless of which
+   * internal hop stalls; the per-hop bounds stay in place as
+   * diagnostics that produce a sharper cause when they fire first.
+   * Injectable so tests exercise the bound in milliseconds instead of
+   * the production default.
+   */
+  readonly turnDispatchTimeoutMs?: number;
 };
 
+/** CL-6644's default turn-level deadline: generous enough to cover a
+ * cold wake plus a remote inference round-trip, the same reasoning
+ * `DEFAULT_WAKE_TIMEOUT_MS` (30s) uses for the wake step alone. */
+export const DEFAULT_TURN_DISPATCH_TIMEOUT_MS = 120_000;
+
+/** The turn-level deadline's own rejection message: names the turn's
+ * run address (the recipient every dispatch failure is already reported
+ * and notified against) and the budget it exceeded, so a person reading
+ * the `reportError` refId's logged cause sees exactly what expired. */
+export function turnDispatchTimeoutMessage(
+  agentAddress: string,
+  timeoutMs: number,
+): string {
+  return `turn for "${agentAddress}" did not settle within ${String(timeoutMs)}ms`;
+}
+
 export type SendWorkbenchMessageInput = {
   readonly tenantId: string;
   readonly principalId: string;
@@ -1206,11 +1239,16 @@ async function routeToRecipients(
  * exactly as a single, unqueued message's fan-out always has.
  */
 async function dispatchTurnBatch(
-  deps: Pick<SendWorkbenchMessageDeps, "platform" | "roomMessages" | "publish">,
+  deps: Pick<
+    SendWorkbenchMessageDeps,
+    "platform" | "roomMessages" | "publish" | "turnDispatchTimeoutMs"
+  >,
   tenantId: string,
   workbenchId: string,
   batch: readonly QueuedTurn[],
 ): Promise<void> {
+  const turnDispatchTimeoutMs =
+    deps.turnDispatchTimeoutMs ?? DEFAULT_TURN_DISPATCH_TIMEOUT_MS;
   const recipientSet = new Set<string>();
   for (const turn of batch) {
     for (const agentAddress of turn.recipients) recipientSet.add(agentAddress);
@@ -1238,14 +1276,26 @@ async function dispatchTurnBatch(
   await Promise.all(
     recipients.map(async (agentAddress) => {
       try {
-        await dispatchTurn(deps, {
-          tenantId,
-          workbenchId,
-          principalId: last.principalId,
-          agentAddress,
-          parts,
-          requestMessageIds: messageIds,
-        });
+        // CL-6644: one deadline around the whole turn, not another
+        // per-hop bound. `dispatchTurn` only ever reaches "the mail was
+        // handed to the agent's mailbox" (see `./turn-queue.ts`'s own
+        // note) — the agent's actual streaming reply is produced and
+        // posted onto the timeline later, off this call stack, through
+        // `chat-orchestrator.ts`'s independent sidecar-event
+        // subscription. This deadline therefore can never cut off a
+        // reply in progress: nothing it awaits is the reply.
+        await withTimeout(
+          dispatchTurn(deps, {
+            tenantId,
+            workbenchId,
+            principalId: last.principalId,
+            agentAddress,
+            parts,
+            requestMessageIds: messageIds,
+          }),
+          turnDispatchTimeoutMs,
+          turnDispatchTimeoutMessage(agentAddress, turnDispatchTimeoutMs),
+        );
       } catch (err) {
         const refId = reportError(err, {
           operation: "chat.dispatchTurn",

packages/chat/test/turn-dispatch-deadline.test.ts

@@ -0,0 +1,178 @@
+// CL-6644's structural close: three rounds of per-hop timeouts (#312
+// wake, #314 bypasses, #316 reclaim mail) each fixed one stalling hop
+// and a fourth kept appearing — a fourth hang variant (a direct send to
+// an already-live run) needed neither #312's wake bound nor #316's
+// reclaim-retry bound, and still hung silently. The fix is one
+// turn-level deadline: `dispatchTurnBatch` wraps each recipient's
+// `dispatchTurn` call in a single wall-clock budget, so no agent turn
+// may hang past it regardless of which internal hop stalls. Per-hop
+// bounds stay as diagnostics; this is the backstop.
+import { describe, expect, test } from "bun:test";
+
+import { createChatRoutes } from "../src/routes";
+import {
+  buildDeps,
+  createWorkbench,
+  fakePlatform,
+  mountAs,
+  settleFanout,
+  timelineOf,
+} from "./test-support";
+
+describe("dispatchTurnBatch's turn-level deadline (CL-6644)", () => {
+  test("a dispatchTurn that never settles still posts an undelivered notice within the injected budget", async () => {
+    const platform = fakePlatform({
+      invitable: [{ id: "wfd_echo", name: "echo" }],
+    });
+    // Models a stall anywhere inside `dispatchTurn`'s own call chain that
+    // never throws and never resolves -- the exact shape #316 fixed for
+    // the mail-delivery hop specifically, and CL-6644's last comment
+    // says stop bounding hops one at a time and put one deadline around
+    // the whole turn instead.
+    platform.sendMail = () => new Promise<never>(() => {});
+
+    const deps = buildDeps({ platform, turnDispatchTimeoutMs: 20 });
+    const app = mountAs(createChatRoutes(deps), "prn_alice");
+    const { body: workbench } = await createWorkbench(app, {
+      kind: "chat",
+      definitionId: "wfd_echo",
+    });
+
+    const started = Date.now();
+    await app.request(`/workbenches/${workbench.id}/messages`, {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({ parts: [{ kind: "text", text: "hello" }] }),
+    });
+    await settleFanout();
+    const elapsedMs = Date.now() - started;
+
+    // The whole HTTP round trip -- including the notice post -- settles
+    // well inside a budget generous enough to also cover CI jitter,
+    // proving the stall didn't wedge the batch.
+    expect(elapsedMs).toBeLessThan(2000);
+
+    const timeline = await timelineOf(deps, workbench.id);
+    const notice = timeline.find(
+      (message) =>
+        message.sender.address === "ins_invited1@acme.example" &&
+        message.parts.some(
+          (part) => part.kind === "text" && part.turnFailed === true,
+        ),
+    );
+    const noticePart = notice?.parts.find((part) => part.kind === "text");
+    expect(noticePart).toMatchObject({ kind: "text", turnFailed: true });
+    const text = noticePart?.kind === "text" ? noticePart.text : "";
+    expect(text).toMatch(/\(ref [^)]+\)$/);
+  });
+
+  test("a timed-out turn releases the workbench's claim instead of wedging it for the TTL", async () => {
+    const platform = fakePlatform({
+      invitable: [{ id: "wfd_echo", name: "echo" }],
+    });
+    platform.sendMail = () => new Promise<never>(() => {});
+
+    const deps = buildDeps({
+      platform,
+      turnDispatchTimeoutMs: 20,
+      turnTimeoutMs: 60_000, // the claim TTL backstop -- must never be needed
+    });
+    const app = mountAs(createChatRoutes(deps), "prn_alice");
+    const { body: workbench } = await createWorkbench(app, {
+      kind: "chat",
+      definitionId: "wfd_echo",
+    });
+
+    await app.request(`/workbenches/${workbench.id}/messages`, {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({ parts: [{ kind: "text", text: "first" }] }),
+    });
+    await settleFanout();
+
+    // A second message right after the first's deadline fired must
+    // dispatch immediately rather than queue behind a claim the TTL
+    // (60s) hasn't released yet -- proving the deadline's rejection
+    // released the claim itself, not just posted a notice.
+    const response = await app.request(
+      `/workbenches/${workbench.id}/messages`,
+      {
+        method: "POST",
+        headers: { "content-type": "application/json" },
+        body: JSON.stringify({ parts: [{ kind: "text", text: "second" }] }),
+      },
+    );
+    await settleFanout();
+    expect(response.status).toBe(201);
+
+    const timeline = await timelineOf(deps, workbench.id);
+    const notices = timeline.filter(
+      (message) =>
+        message.sender.address === "ins_invited1@acme.example" &&
+        message.parts.some(
+          (part) => part.kind === "text" && part.turnFailed === true,
+        ),
+    );
+    // Both messages hit the same never-settling `sendMail`, so both
+    // should have failed loud on their own -- neither queued silently
+    // behind a wedged claim.
+    expect(notices).toHaveLength(2);
+  });
+
+  test("a slow but eventually-settling dispatch is not killed by the deadline", async () => {
+    const platform = fakePlatform({
+      invitable: [{ id: "wfd_echo", name: "echo" }],
+    });
+    const realSendMail = platform.sendMail.bind(platform);
+    platform.sendMail = async (input) => {
+      await Bun.sleep(30);
+      return realSendMail(input);
+    };
+
+    // `dispatchTurn`'s own promise only ever covers "mail handed to the
+    // agent's mailbox" (see `./turn-queue.ts`'s note) -- the agent's
+    // real, possibly long, streaming reply is produced and posted
+    // entirely off this call stack by the orchestrator, so a deadline
+    // around `dispatchTurn` can never cut off a reply in progress. This
+    // proves the mail-handoff hop itself, once it settles inside the
+    // budget, is never mistaken for a stall.
+    const deps = buildDeps({ platform, turnDispatchTimeoutMs: 500 });
+    const app = mountAs(createChatRoutes(deps), "prn_alice");
+    const { body: workbench } = await createWorkbench(app, {
+      kind: "chat",
+      definitionId: "wfd_echo",
+    });
+
+    const response = await app.request(
+      `/workbenches/${workbench.id}/messages`,
+      {
+        method: "POST",
+        headers: { "content-type": "application/json" },
+        body: JSON.stringify({ parts: [{ kind: "text", text: "hello" }] }),
+      },
+    );
+    await settleFanout();
+    expect(response.status).toBe(201);
+
+    const timeline = await timelineOf(deps, workbench.id);
+    const notice = timeline.find(
+      (message) =>
+        message.sender.address === "ins_invited1@acme.example" &&
+        message.parts.some(
+          (part) => part.kind === "text" && part.turnFailed === true,
+        ),
+    );
+    expect(notice).toBeUndefined();
+    expect((platform as ReturnType<typeof fakePlatform>).sentMail).toHaveLength(
+      1,
+    );
+  });
+
+  test("the timeout message names the turn's run address and its elapsed budget", async () => {
+    const { turnDispatchTimeoutMessage } =
+      await import("../src/workbench-service");
+    expect(turnDispatchTimeoutMessage("ins_echo1@acme.example", 30_000)).toBe(
+      'turn for "ins_echo1@acme.example" did not settle within 30000ms',
+    );
+  });
+});