From 8255789eb58c68dbcbb01e8851811ffd1ae06489 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 00:49:45 -0700 Subject: [PATCH 1/2] Add tests for expired turn claim eviction and listing order (CL-7200) --- packages/chat/src/agent-turns.test.ts | 22 +++++++++++++++ packages/chat/src/turn-claims.test.ts | 40 +++++++++++++++++++++++++++ 2 files changed, 62 insertions(+) diff --git a/packages/chat/src/agent-turns.test.ts b/packages/chat/src/agent-turns.test.ts index 7fb0bdc4d..2adf0be95 100644 --- a/packages/chat/src/agent-turns.test.ts +++ b/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 diff --git a/packages/chat/src/turn-claims.test.ts b/packages/chat/src/turn-claims.test.ts index c2b01e952..4b2553faf 100644 --- a/packages/chat/src/turn-claims.test.ts +++ b/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); + }); }); }); From 586160715c494c89f7da37d697baa5aef3e13627 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 00:49:49 -0700 Subject: [PATCH 2/2] Evict expired turn claims and order listings with occurrence tiebreak (CL-7200) --- packages/chat/src/agent-turns.ts | 21 +++++++++++--- packages/chat/src/turn-claims.ts | 49 +++++++++++++++++++++----------- 2 files changed, 50 insertions(+), 20 deletions(-) diff --git a/packages/chat/src/agent-turns.ts b/packages/chat/src/agent-turns.ts index ec3880acf..74be717d9 100644 --- a/packages/chat/src/agent-turns.ts +++ b/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)); }, diff --git a/packages/chat/src/turn-claims.ts b/packages/chat/src/turn-claims.ts index 9dabfacc8..20883bbb4 100644 --- a/packages/chat/src/turn-claims.ts +++ b/packages/chat/src/turn-claims.ts @@ -41,12 +41,15 @@ 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; /** @@ -54,7 +57,7 @@ export type TurnClaimStore = { * `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; }; @@ -99,10 +102,28 @@ 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); @@ -110,7 +131,7 @@ export function createInMemoryTurnClaimStore(options: { 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; } @@ -118,12 +139,8 @@ export function createInMemoryTurnClaimStore(options: { 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; }, }; }