From ea20d3ea306f300d70946eb641cd130320874771 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 15 Sep 2026 11:53:32 -0700 Subject: [PATCH] fix(chat): dispatch claim-less on opening claim failure instead of dropping the turn (CL-6167) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The opening tryClaim in turnQueue.run was the one claim-store call run never guarded. A transient rejection propagated into routeMessage's log-and-swallow: message durable, no agent ever asked, no timeline notice — the load-independent sequential drop (turn-2 after healthy turn-1, greeting/turn-1 on a cold store). Now reported loudly (chat.turnQueue.claim with refId) and dispatched claim-less; the turn is deliberately not parked in pending (nothing would drain it while the store is down). --- packages/chat/src/turn-queue.test.ts | 53 ++++++++++++ packages/chat/src/turn-queue.ts | 45 ++++++++++- packages/chat/test/turn-claim-failure.test.ts | 81 +++++++++++++++++++ 3 files changed, 178 insertions(+), 1 deletion(-) create mode 100644 packages/chat/test/turn-claim-failure.test.ts diff --git a/packages/chat/src/turn-queue.test.ts b/packages/chat/src/turn-queue.test.ts index bdf4c6b9f9..75731274bc 100644 --- a/packages/chat/src/turn-queue.test.ts +++ b/packages/chat/src/turn-queue.test.ts @@ -395,4 +395,57 @@ describe("turn queue drains everything that was enqueued", () => { // directly rather than only being queued. expect(holders.has("wb")).toBe(false); }); + + // CL-6167: a transient claim-store failure on a turn's opening tryClaim + // must not drop the turn. Sequential, zero concurrency — turn-1 healthy, + // turn-2's claim rejects (durable I/O can reject outright, not just + // resolve false) — and `run()` documents "never rejects", so the throw + // must neither propagate (routeMessage logs-and-swallows it, leaving the + // message durable but no agent ever asked and no notice on the timeline) + // nor strand the turn in the pending map (nothing would ever drain it + // while the store keeps failing). The turn dispatches claim-less instead. + test("does not reject run() and still dispatches when the opening tryClaim throws", async () => { + const holders = new Map(); + let n = 0; + let tryClaimCalls = 0; + + const claims: TurnClaimStore = { + async tryClaim(c) { + tryClaimCalls += 1; + // Turn-1 claims cleanly; turn-2 arrives after turn-1 released and + // meets a transient store failure instead. + if (tryClaimCalls === 1) { + const t = String(++n); + holders.set(c.workbenchId, t); + return t; + } + throw new Error("claim store connection reset"); + }, + async release(c, t) { + if (holders.get(c.workbenchId) !== t) return false; + holders.delete(c.workbenchId); + return true; + }, + async holds(c, t) { + return holders.get(c.workbenchId) === t; + }, + }; + + const queue = createWorkbenchTurnQueue({ + claims, + publish: () => undefined, + }); + const dispatched: string[] = []; + const dispatch = async (batch: readonly QueuedTurn[]) => { + for (const t of batch) dispatched.push(t.messageId); + }; + + await queue.run("wb", turn("m1", "one"), dispatch); + expect(dispatched).toEqual(["m1"]); + + // Does not reject, despite the opening tryClaim throwing — and the + // turn still reaches dispatch rather than dropping silently. + await queue.run("wb", turn("m2", "two"), dispatch); + expect(dispatched).toEqual(["m1", "m2"]); + }); }); diff --git a/packages/chat/src/turn-queue.ts b/packages/chat/src/turn-queue.ts index 7bb67a8038..c4d5ce439f 100644 --- a/packages/chat/src/turn-queue.ts +++ b/packages/chat/src/turn-queue.ts @@ -207,7 +207,50 @@ export function createWorkbenchTurnQueue( return { async run(workbenchId, turn, dispatch) { - const token = await deps.claims.tryClaim({ workbenchId }); + // CL-6167: the claim store is durable I/O and can reject outright, + // not just resolve `false` — and this opening `tryClaim` was the one + // call `run` never guarded. The throw propagated into `routeMessage`'s + // log-and-swallow, dropping the turn with no notice on the timeline + // (message durable, no agent ever asked): the load-independent + // sequential drop. Report loudly and dispatch the turn claim-less + // instead of dropping it — ordering across concurrent runs degrades + // to the pre-queue behavior while the store is down, and the turn is + // deliberately NOT parked in the process-local pending map (nothing + // would ever drain it while the store keeps failing). + let token: TurnClaimToken | false; + try { + token = await deps.claims.tryClaim({ workbenchId }); + } catch (err) { + const refId = reportError(err, { + operation: "chat.turnQueue.claim", + roomId: workbenchId, + }); + log.error( + "turn queue: opening claim failed for workbench {workbenchId} (ref {refId}); dispatching without the claim rather than dropping the turn: {err}", + { workbenchId, refId, err }, + ); + try { + await dispatch([turn]); + } catch (dispatchErr) { + const dispatchRefId = reportError(dispatchErr, { + operation: "chat.turnQueue.dispatch", + roomId: workbenchId, + extra: { messageIds: [turn.messageId] }, + }); + // Same contract as `drain`'s own guard below: `dispatch` + // documents "must never reject" (see `DispatchTurnBatch`). + log.error( + 'turn queue: dispatch rejected for workbench {workbenchId}, message(s) {messageIds} (ref {refId}), violating its "never reject" contract: {err}', + { + workbenchId, + messageIds: [turn.messageId], + refId: dispatchRefId, + err: dispatchErr, + }, + ); + } + return; + } if (token !== false) { await drain(workbenchId, token, [turn], dispatch); return; diff --git a/packages/chat/test/turn-claim-failure.test.ts b/packages/chat/test/turn-claim-failure.test.ts new file mode 100644 index 0000000000..eb31abd87d --- /dev/null +++ b/packages/chat/test/turn-claim-failure.test.ts @@ -0,0 +1,81 @@ +// CL-6167: load-independent sequential turn drops. A transient +// claim-store failure on a turn's opening `tryClaim` used to propagate out +// of `turnQueue.run` into `routeMessage`'s log-and-swallow — the message +// persisted fine, but no agent was ever asked and no notice ever reached +// the timeline. Sequential, zero concurrency: turn-1 healthy, turn-2 meets +// the failure, and turn-2 must still reach the agent. +import { describe, expect, test } from "bun:test"; + +import { createChatRoutes } from "../src/routes"; +import type { TurnClaimStore } from "../src/turn-claims"; +import { createWorkbenchTurnQueue } from "../src/turn-queue"; +import { + buildDeps, + createWorkbench, + fakePlatform, + mountAs, + sendText, +} from "./test-support"; + +/** Healthy for the first turn, then a transient failure on every later + * opening claim — the durable-I/O hiccup the in-memory store never + * models, arriving on an otherwise idle workbench. */ +function failAfterFirstClaim(): TurnClaimStore { + const holders = new Map(); + let n = 0; + let tryClaimCalls = 0; + return { + async tryClaim(c) { + tryClaimCalls += 1; + if (tryClaimCalls === 1) { + const t = String(++n); + holders.set(c.workbenchId, t); + return t; + } + throw new Error("claim store connection reset"); + }, + async release(c, t) { + if (holders.get(c.workbenchId) !== t) return false; + holders.delete(c.workbenchId); + return true; + }, + async holds(c, t) { + return holders.get(c.workbenchId) === t; + }, + }; +} + +describe("CL-6167 sequential turn drops", () => { + test("a transient claim-store failure on turn-2 still asks the agent", async () => { + const platform = fakePlatform(); + const deps = buildDeps({ + platform, + turnQueue: createWorkbenchTurnQueue({ + claims: failAfterFirstClaim(), + publish: () => undefined, + }), + }); + const app = mountAs(createChatRoutes(deps), "prn_alice"); + const { body: workbench } = await createWorkbench(app, { + kind: "workbench", + name: "demo", + participants: ["ins_echo1@acme.example"], + }); + + const first = await sendText(app, workbench.id, "hi @ins_echo1"); + expect(first.status).toBe(201); + const sentMail = platform.sentMail; + expect(sentMail).toHaveLength(1); + + // Turn-2 meets the transient claim failure. It must still reach the + // agent — pre-fix this send persisted fine but dispatched nothing. + const second = await sendText( + app, + workbench.id, + "are you there @ins_echo1", + ); + expect(second.status).toBe(201); + expect(sentMail).toHaveLength(2); + expect(sentMail[1]?.workbenchId).toBe("ins_echo1"); + }); +});