From 2535d7ad35c8fe6a000abb0d70089a20721961cf Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Wed, 9 Sep 2026 05:54:04 +0300 Subject: [PATCH] fix: defer the jobs settle-sync until the turn is over (coalescing follow-up 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With continuation coalescing, a job flips done at CLAIM time — while the learner's own answer turn is still streaming. The poller's settle re-sync fired on the flip, swapping in a mid-turn record snapshot mid-stream; the steer after the assistant message broke the SDK's in-place continuation merge, and the streaming response landed as a SECOND message with the same id — duplicate React keys in the live transcript (the user's live test hit it; the record itself stayed clean). A settle during a busy turn (local stream OR server turnInFlight) now marks a pending sync; the first idle poll tick adopts the finished record. This also fixes the latent multi-job race: steer B settling while steer A's wake turn is in flight would have grabbed the same mid-turn snapshot. Out-of-lane surgical touch (flagged for master review): the wave-1 poller block in components/chat.tsx. --- components/chat.tsx | 35 +++++++++++++++++++++++++++++------ tests/jobs.test.ts | 5 +++-- 2 files changed, 32 insertions(+), 8 deletions(-) diff --git a/components/chat.tsx b/components/chat.tsx index 6f089b8..3ec736f 100644 --- a/components/chat.tsx +++ b/components/chat.tsx @@ -178,11 +178,15 @@ export function Chat({ // ── Central job observability (contracts.ts "THE ASYNC PROTOCOL" §4) ── // ONE poll of /api/jobs drives every job surface (transcript dispatch // cards, the sub-agents panel) and re-syncs the transcript exactly once - // when a job settles: job status flips done/error only AFTER the wake - // turn's messages are persisted, so a completion means the record already - // holds the steer + the teacher's response — one syncFromServer adopts it - // all. Poll fast while jobs run, slow otherwise (new-dispatch discovery), - // and not at all when this session has never dispatched a job. + // when a job settles. A settle no longer implies the record is complete: + // CONTINUATION COALESCING flips the job done at claim time — while the + // learner's own answer turn is still streaming (and a settle can race a + // second job's wake turn). Syncing mid-turn would swap in a stale snapshot + // mid-stream, which breaks the SDK's in-place continuation merge (duplicate + // message ids). So: a settle during a busy turn defers the sync; the next + // idle poll tick adopts the finished record. Poll fast while jobs run, + // slow otherwise (new-dispatch discovery), and not at all when this + // session has never dispatched a job. const [jobs, setJobs] = useState([]); const jobPartCount = useMemo( () => @@ -201,6 +205,8 @@ export function Chat({ const localStreamRef = useRef(false); localStreamRef.current = localStream; const jobStatusRef = useRef>(new Map()); + // A settle seen during a busy turn — the sync is owed, not dropped. + const pendingSyncRef = useRef(false); useEffect(() => { if (pollMs === 0) return; let cancelled = false; @@ -214,9 +220,26 @@ export function Chat({ const list = data.jobs ?? []; for (const j of list) { const prev = jobStatusRef.current.get(j.id); - if (prev === "running" && j.status !== "running") void syncFromServer(); + if (prev === "running" && j.status !== "running") { + // The record is only guaranteed complete when no turn is in + // flight — locally or server-side. A coalesced settle fires + // mid-turn by design; adopt the record on the first idle tick. + if (localStreamRef.current || data.turnInFlight === true) { + pendingSyncRef.current = true; + } else { + void syncFromServer(); + } + } jobStatusRef.current.set(j.id, j.status); } + if ( + pendingSyncRef.current && + !localStreamRef.current && + data.turnInFlight !== true + ) { + pendingSyncRef.current = false; + void syncFromServer(); + } setJobs(list); // Wake-turn adoption: a server-side teacher turn with no client request // of ours (sub-agent delivery) — the same machinery as returning diff --git a/tests/jobs.test.ts b/tests/jobs.test.ts index b2382a1..49590da 100644 --- a/tests/jobs.test.ts +++ b/tests/jobs.test.ts @@ -27,6 +27,7 @@ import assert from "node:assert/strict"; import * as fs from "node:fs"; import * as path from "node:path"; import { after, describe, it } from "node:test"; +import type { UIMessage } from "ai"; import { MockLanguageModelV4, simulateReadableStream } from "ai/test"; import type { LanguageModelV4StreamPart } from "@ai-sdk/provider"; import { @@ -771,8 +772,8 @@ describe("persistTurn id hygiene", () => { // response copy under one id — the record must keep only the merge. const chat = chatStore.createChat("Test: dedupe continuation copies"); trackChat(chat.id); - const short = { id: "msg_dup_1", role: "assistant", parts: [{ type: "text", text: "step 1" }] }; - const long = { id: "msg_dup_1", role: "assistant", parts: [{ type: "text", text: "step 1" }, { type: "text", text: "step 2 (continued)" }] }; + const short: UIMessage = { id: "msg_dup_1", role: "assistant", parts: [{ type: "text", text: "step 1" }] }; + const long: UIMessage = { id: "msg_dup_1", role: "assistant", parts: [{ type: "text", text: "step 1" }, { type: "text", text: "step 2 (continued)" }] }; persistTurn(chatStore, chat.id, [ { id: "u1", role: "user", parts: [{ type: "text", text: "hi" }] }, short,