Evict expired turn claims and stabilize turn listing order

packages/chat/src/agent-turns.test.ts

@@ -111,6 +111,28 @@ describe("createInMemoryAgentTurnStore", () => {
     expect(listed.map((turn) => turn.id)).toEqual([second.id, first.id]);
   });
 
+  // CL-7200: listing used to sort by startedAt string alone, while
+  // findRunningTurn sorted by occurrence. Two turns sharing a timestamp
+  // (same-millisecond ISO strings compare equal) then disagreed — listing
+  // kept Map insertion order, the running-turn resolver picked the later
+  // occurrence. Both orderings share startedAt then occurrence as the
+  // deterministic tiebreak so they name the same newest turn.
+  test("listing and findRunningTurn agree when two turns share a timestamp", async () => {
+    const clock = 1_000;
+    const store = createInMemoryAgentTurnStore({ now: () => clock });
+    const first = await store.startTurn(BASE);
+    const second = await store.startTurn(BASE);
+    expect(first.startedAt).toBe(second.startedAt);
+    expect(second.occurrence).toBeGreaterThan(first.occurrence);
+
+    expect((await store.findRunningTurn(BASE))?.id).toBe(second.id);
+    const listed = await store.listTurns({
+      tenantId: BASE.tenantId,
+      workbenchId: BASE.workbenchId,
+    });
+    expect(listed.map((turn) => turn.id)).toEqual([second.id, first.id]);
+  });
+
   // CL-6451: a dispatch the supervisor failed (or a hub that died
   // mid-turn) never sends the event that closes its row, so a turn
   // still `running` past the occurrence timeout is dead by construction

packages/chat/src/agent-turns.ts

@@ -187,6 +187,19 @@ function sectionKey(input: {
   return `${input.tenantId} ${input.workbenchId} ${input.agentAddress}`;
 }
 
+/**
+ * Newest-first order shared by `listTurns` and `findRunningTurn`
+ * (CL-7200): `startedAt` descending, then `occurrence` descending as
+ * the deterministic tiebreak when two turns share a timestamp. Listing
+ * used to sort by the started-at string alone and the running-turn
+ * resolver by occurrence alone, so same-millisecond rows could disagree
+ * on which turn was newest.
+ */
+function compareTurnsNewestFirst(left: AgentTurn, right: AgentTurn): number {
+  const byStarted = right.startedAt.localeCompare(left.startedAt);
+  return byStarted !== 0 ? byStarted : right.occurrence - left.occurrence;
+}
+
 /**
  * The process-local pubsub `waitUntilFree` is built on: a (workbench,
  * agent) key's waiters all resolve the moment `notify` fires for that
@@ -298,7 +311,7 @@ export function createInMemoryAgentTurnStore(
           turn.agentAddress === input.agentAddress &&
           turn.status === "running",
       )
-      .sort((a, b) => b.occurrence - a.occurrence)[0];
+      .sort(compareTurnsNewestFirst)[0];
   }
 
   return {
@@ -391,7 +404,7 @@ export function createInMemoryAgentTurnStore(
             turn.tenantId === input.tenantId &&
             turn.workbenchId === input.workbenchId,
         )
-        .sort((a, b) => b.startedAt.localeCompare(a.startedAt))
+        .sort(compareTurnsNewestFirst)
         .slice(0, input.limit ?? AGENT_TURNS_PAGE_SIZE);
     },
 
@@ -505,7 +518,7 @@ export function createDrizzleAgentTurnStore<
           eq(agentTurns.status, "running"),
         ),
       )
-      .orderBy(desc(agentTurns.occurrence))
+      .orderBy(desc(agentTurns.startedAt), desc(agentTurns.occurrence))
       .limit(1);
     return row === undefined ? undefined : toAgentTurn(row as AgentTurnRow);
   }
@@ -614,7 +627,7 @@ export function createDrizzleAgentTurnStore<
             eq(agentTurns.workbenchId, input.workbenchId),
           ),
         )
-        .orderBy(desc(agentTurns.startedAt))
+        .orderBy(desc(agentTurns.startedAt), desc(agentTurns.occurrence))
         .limit(input.limit ?? AGENT_TURNS_PAGE_SIZE);
       return rows.map((row) => toAgentTurn(row as AgentTurnRow));
     },

packages/chat/src/turn-claims.test.ts

@@ -64,6 +64,30 @@ describe("createInMemoryTurnClaimStore", () => {
     ).resolves.toBe(true);
   });
 
+  test("a mismatched release evicts an already-expired entry, so the true original token then finds nothing left to release", async () => {
+    let now = 0;
+    const store = createInMemoryTurnClaimStore({
+      ttlMs: 1_000,
+      now: () => now,
+    });
+    const token = await store.tryClaim({ workbenchId: "wb_1" });
+    now += 1_000; // expired, but nobody has claimed it since
+
+    // A release with the wrong token still can't claim credit for
+    // freeing this workbench, but it notices the stale entry it found
+    // along the way and cleans it up rather than leaving it in place.
+    await expect(
+      store.release({ workbenchId: "wb_1" }, "not_the_real_token"),
+    ).resolves.toBe(false);
+
+    // Proof the entry is actually gone, not merely ignored: the true
+    // original token — still technically correct — now finds nothing
+    // left to release.
+    await expect(
+      store.release({ workbenchId: "wb_1" }, token as string),
+    ).resolves.toBe(false);
+  });
+
   test("a claim older than the TTL is reclaimable even without release — the crash/hang backstop", async () => {
     let now = 0;
     const store = createInMemoryTurnClaimStore({
@@ -107,5 +131,21 @@ describe("createInMemoryTurnClaimStore", () => {
         store.holds({ workbenchId: "wb_1" }, token as string),
       ).resolves.toBe(false);
     });
+
+    test("observing an expired claim deletes it, so the original token then finds nothing to release", async () => {
+      let now = 0;
+      const store = createInMemoryTurnClaimStore({
+        ttlMs: 1_000,
+        now: () => now,
+      });
+      const token = await store.tryClaim({ workbenchId: "wb_1" });
+      now += 1_000;
+      await expect(
+        store.holds({ workbenchId: "wb_1" }, token as string),
+      ).resolves.toBe(false);
+      await expect(
+        store.release({ workbenchId: "wb_1" }, token as string),
+      ).resolves.toBe(false);
+    });
   });
 });

packages/chat/src/turn-claims.ts

@@ -41,20 +41,23 @@ export type TurnClaimStore = {
   /**
    * Un-claims `workbenchId` — called once the turn that won the claim
    * has settled, success or failure alike, so the next queued batch (or
-   * a fresh message) can run. A no-op (returns `false`) if `token` is
-   * not the current holder — e.g. the TTL already reassigned the claim
-   * to a second winner, in which case releasing here would free a
-   * claim this caller no longer owns. Never left un-called on a path
-   * that can still complete; the TTL below is only the backstop for a
-   * dispatch that never settles at all.
+   * a fresh message) can run. Always returns `false` if `token` is not
+   * the current live holder — e.g. the TTL already reassigned the claim
+   * to a second winner, in which case releasing here would free a claim
+   * this caller no longer owns — but may still evict the entry it
+   * found: an already-expired entry carries no live claim for anyone
+   * to lose, so finding one is incidental cleanup, never this call's
+   * own claim to release. Never left un-called on a path that can
+   * still complete; the TTL below is only the backstop for a dispatch
+   * that never settles at all.
    */
   release(claim: TurnClaim, token: TurnClaimToken): Promise<boolean>;
   /**
    * Whether `token` is still the current, unexpired holder of
    * `workbenchId`'s claim — how a long-running holder notices the TTL
    * reassigned its claim to someone else without itself calling
    * `tryClaim` again (which would attempt to *take* the claim, not just
-   * check it).
+   * check it). Observing an expired entry deletes it (CL-7200).
    */
   holds(claim: TurnClaim, token: TurnClaimToken): Promise<boolean>;
 };
@@ -99,31 +102,45 @@ export function createInMemoryTurnClaimStore(options: {
     return now() - holder.claimedAt >= options.ttlMs;
   }
 
+  /**
+   * The live holder of `workbenchId`, or `undefined` if nobody holds it.
+   * An expired entry is deleted here so the map agrees with what
+   * `tryClaim`/`holds`/`release` already conclude — rather than lingering
+   * until a future `tryClaim` for this workbench happens to overwrite it
+   * (CL-7200).
+   */
+  function liveHolder(
+    workbenchId: string,
+  ): { token: string; claimedAt: number } | undefined {
+    const existing = holders.get(workbenchId);
+    if (existing === undefined) return undefined;
+    if (isExpired(existing)) {
+      holders.delete(workbenchId);
+      return undefined;
+    }
+    return existing;
+  }
+
   return {
     async tryClaim(claim) {
-      const existing = holders.get(claim.workbenchId);
-      if (existing !== undefined && !isExpired(existing)) {
+      if (liveHolder(claim.workbenchId) !== undefined) {
         return false;
       }
       const token = String(++nextToken);
       holders.set(claim.workbenchId, { token, claimedAt: now() });
       return token;
     },
     async release(claim, token) {
-      const existing = holders.get(claim.workbenchId);
+      const existing = liveHolder(claim.workbenchId);
       if (existing === undefined || existing.token !== token) {
         return false;
       }
       holders.delete(claim.workbenchId);
       return true;
     },
     async holds(claim, token) {
-      const existing = holders.get(claim.workbenchId);
-      return (
-        existing !== undefined &&
-        existing.token === token &&
-        !isExpired(existing)
-      );
+      const existing = liveHolder(claim.workbenchId);
+      return existing !== undefined && existing.token === token;
     },
   };
 }