From 6adb21ecf6ca0abe60cf101892422b5032b57c58 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sat, 22 Aug 2026 06:23:05 -0700 Subject: [PATCH 1/2] Add tests for CL-6644's turn-level dispatch deadline Covers the structural fix the ticket's last comment mandates: a never-settling dispatchTurn internals must post an undelivered notice within the injected budget and release the workbench's turn claim rather than wedging it for the claim TTL, while a dispatch that settles inside the budget (however slowly) must never be killed. --- .../chat/test/turn-dispatch-deadline.test.ts | 178 ++++++++++++++++++ 1 file changed, 178 insertions(+) create mode 100644 packages/chat/test/turn-dispatch-deadline.test.ts diff --git a/packages/chat/test/turn-dispatch-deadline.test.ts b/packages/chat/test/turn-dispatch-deadline.test.ts new file mode 100644 index 00000000..290f2af0 --- /dev/null +++ b/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(() => {}); + + 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(() => {}); + + 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).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', + ); + }); +}); From 72ba7f3235a14f713aff01c3f1514581162768fe Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sat, 22 Aug 2026 06:23:19 -0700 Subject: [PATCH 2/2] Wrap each dispatched turn in one turn-level deadline instead of another per-hop bound Three rounds of per-hop timeouts (#312's wake bound, #314's bypassed- wake bound, #316's mail-delivery bound) each closed one stalling hop and a fourth kept appearing -- most recently a direct send to an already-live run that needed neither #312's nor #316's bound and still hung silently with no notice posted. dispatchTurnBatch now wraps each recipient's dispatchTurn call in one wall-clock deadline (DEFAULT_TURN_DISPATCH_TIMEOUT_MS, 120s, injectable via SendWorkbenchMessageDeps.turnDispatchTimeoutMs) so no agent turn may hang past its budget regardless of which internal hop stalls. A timeout rejects with a message naming the turn's run address and its elapsed budget, which flows into dispatchTurnBatch's existing catch (#313): a reportError refId is logged and an undelivered notice is posted. The workbench's turn claim releases in createWorkbenchTurnQueue's existing finally block the moment dispatchTurnBatch's per-recipient Promise.all settles, so a timed-out turn never wedges the room for the claim TTL. dispatchTurn's own promise only ever covers "the mail was handed to the agent's mailbox" (see turn-queue.ts's 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. The deadline therefore can never cut off a reply in progress: nothing it awaits is the reply itself. Lifted the withTimeout helper #316 wrote inline in platform-adapter.ts into a shared with-timeout.ts so the wake bound, the mail-delivery bound, and this new turn-level deadline share one implementation. Per-hop bounds (#312/#314/#316) stay in place: they still produce a sharper cause when they fire first, and the turn-level deadline is only the backstop for whatever they don't yet cover. Not fixed here, and not this PR's job: why a direct send to a live, registered run stalls at all -- that root cause continues under CL-6648. This deadline makes the stall loud instead of silent; the reportError refId will name it. --- packages/chat/src/index.ts | 2 + packages/chat/src/platform-adapter.ts | 21 +------- packages/chat/src/routes.ts | 12 +++++ packages/chat/src/with-timeout.ts | 25 ++++++++++ packages/chat/src/workbench-service.ts | 68 ++++++++++++++++++++++---- 5 files changed, 99 insertions(+), 29 deletions(-) create mode 100644 packages/chat/src/with-timeout.ts diff --git a/packages/chat/src/index.ts b/packages/chat/src/index.ts index 01380b8f..128c15ad 100644 --- a/packages/chat/src/index.ts +++ b/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, diff --git a/packages/chat/src/platform-adapter.ts b/packages/chat/src/platform-adapter.ts index 6fc113a6..eed9a1d8 100644 --- a/packages/chat/src/platform-adapter.ts +++ b/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( - promise: Promise, - ms: number, - message: string, - ): Promise { - 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"); } diff --git a/packages/chat/src/routes.ts b/packages/chat/src/routes.ts index 99527db0..a3354a1a 100644 --- a/packages/chat/src/routes.ts +++ b/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 { ? { 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 { ? { agentTurns: deps.agentTurns } : {}), ...(deps.threads !== undefined ? { threads: deps.threads } : {}), + ...(deps.turnDispatchTimeoutMs !== undefined + ? { turnDispatchTimeoutMs: deps.turnDispatchTimeoutMs } + : {}), }, { tenantId: ownerTenantId, diff --git a/packages/chat/src/with-timeout.ts b/packages/chat/src/with-timeout.ts new file mode 100644 index 00000000..1a3f4b4f --- /dev/null +++ b/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( + promise: Promise, + ms: number, + message: string, +): Promise { + 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); + }, + ); + }); +} diff --git a/packages/chat/src/workbench-service.ts b/packages/chat/src/workbench-service.ts index 2644f0bb..4ab60bc1 100644 --- a/packages/chat/src/workbench-service.ts +++ b/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; + /** + * 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, + deps: Pick< + SendWorkbenchMessageDeps, + "platform" | "roomMessages" | "publish" | "turnDispatchTimeoutMs" + >, tenantId: string, workbenchId: string, batch: readonly QueuedTurn[], ): Promise { + const turnDispatchTimeoutMs = + deps.turnDispatchTimeoutMs ?? DEFAULT_TURN_DISPATCH_TIMEOUT_MS; const recipientSet = new Set(); 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",