fix(chat): dispatch claim-less on opening claim failure instead of dropping the turn

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<string, string>();
+    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"]);
+  });
 });

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;

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<string, string>();
+  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");
+  });
+});