diff --git a/src/server/chat-handler.ts b/src/server/chat-handler.ts index ffb545a..e2fb63d 100644 --- a/src/server/chat-handler.ts +++ b/src/server/chat-handler.ts @@ -129,7 +129,10 @@ export async function handleChatRequest( // appended after the learner's answer (conversion-safe: the answer carries // its output). Claiming happens after beginTurn, so this turn owns the // registry and the write-ahead persist below makes the claim durable. - const steers = claimSteers(chatId); + // The incoming body is passed too: a pump-fired wake turn ALREADY CARRIES + // its steer, so claimSteers finalizes that entry instead of appending a + // second copy (duplicate-key bug, 2026-09-18). + const steers = claimSteers(chatId, body.messages as UIMessage[]); const outgoing = ( steers.length > 0 ? [...(body.messages as UIMessage[]), ...steers] : body.messages ) as UIMessage[]; diff --git a/src/server/jobs.ts b/src/server/jobs.ts index aab8518..2465e61 100644 --- a/src/server/jobs.ts +++ b/src/server/jobs.ts @@ -173,8 +173,19 @@ export function deliverJobResult(steer: SteerMessage): Promise { * are finalized, not returned. The pump re-checks steerDelivered right * before firing, so a claim racing the pump converges to exactly one * delivery either way. + * + * incoming (the turn's request body — the wake pump fires turns that + * ALREADY CARRY the steer it dequeues only after the turn returns 200, so + * the entry is still queued here): a steer already in the incoming list is + * being delivered by THIS VERY TURN — finalized and resolved, but never + * appended again. Without this the write-ahead persist would hold two + * copies of the same steer id (a live React duplicate-key warning the + * owner hit 2026-09-18; withMessageIds only heals it at onFinish). */ -export function claimSteers(chatId: string): SteerMessage[] { +export function claimSteers( + chatId: string, + incoming?: ReadonlyArray, +): SteerMessage[] { const s = state(); const queue = s.queues.get(chatId); if (!queue || queue.length === 0) return []; @@ -182,8 +193,13 @@ export function claimSteers(chatId: string): SteerMessage[] { if (!record) return []; // the handler validated the chat; defensive only const claimed: SteerMessage[] = []; for (const entry of queue) { - if (steerDelivered(record.messages, entry.steer.id)) { - // A wake already landed it — just finish the bookkeeping. + if ( + steerDelivered(record.messages, entry.steer.id) || + incoming?.some((m) => m.id === entry.steer.id) + ) { + // A wake already landed it — or this turn's own request already + // carries it (the pump path): the delivery is done either way, + // just finish the bookkeeping. NEVER append twice. finalizeSteer(entry.steer); } else { claimed.push(entry.steer); diff --git a/tests/jobs.test.ts b/tests/jobs.test.ts index 1c3db76..eb6839f 100644 --- a/tests/jobs.test.ts +++ b/tests/jobs.test.ts @@ -778,6 +778,40 @@ describe("claimSteers + pump duplicate-guard (continuation coalescing)", () => { assert.deepEqual(claimSteers("chat_never_existed"), []); }); + it("pump path: a steer already carried by the incoming turn body is finalized, never appended again (duplicate-key bug, 2026-09-18)", async () => { + const { chatId, jobId, steer } = setup(); + // Own runner: this chat's gates are OPEN, so the pump would otherwise + // fire whatever runner the NEXT test registers (leaked counter). + let fired = 0; + setTurnRunner(async () => { + fired++; + return 200; + }); + // The wake pump's request ALREADY carries the steer (it dequeues only + // after the turn returns), and the record does not — the exact window + // the chat handler claims in. + const delivery = deliverJobResult(steer); + await delay(300); // the pump is inside waitForIdle's grace (750ms) + const incoming = [...(chatStore.getChat(chatId)!.messages as never[]), steer]; + const claimed = claimSteers(chatId, incoming); + assert.equal(claimed.length, 0, "the already-carried steer is not re-claimed"); + assert.equal(jobStore.getJob(jobId)!.status, "done", "job still finalizes — the turn delivers it"); + await delivery; + assert.deepEqual(claimSteers(chatId), [], "queue is drained"); + // The handler persists the outgoing write-ahead immediately after the + // claim — the pump's post-idle re-check reads the RECORD (not the + // queue), so without this the wake would double-fire at grace-end. + const record2 = chatStore.getChat(chatId)!; + chatStore.saveMessages(chatId, [...record2.messages, ...incoming] as never); + await delay(800); // the pump's grace elapses — the delivered steer is skipped + assert.equal(fired, 0, "the claimed-and-carried steer never double-fires a wake"); + // The handler's outgoing = incoming + claimed: with claimed empty the + // write-ahead holds exactly ONE copy of the steer id. + const outgoing = [...incoming, ...claimed] as never[]; + const ids = outgoing.map((m: { id: string }) => m.id); + assert.equal(ids.filter((id) => id === steer.id).length, 1, "exactly one steer copy persists"); + }); + it("pump duplicate-guard: a steer coalesced into the record while the gates were closed is never double-fired", async () => { const { chatId, jobId, steer } = setup(); let fired = 0;