fix: defer the jobs settle-sync until the turn is over (coalescing follow-up 2)
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.
This commit is contained in:
+29
-6
@@ -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<JobRecord[]>([]);
|
||||
const jobPartCount = useMemo(
|
||||
() =>
|
||||
@@ -201,6 +205,8 @@ export function Chat({
|
||||
const localStreamRef = useRef(false);
|
||||
localStreamRef.current = localStream;
|
||||
const jobStatusRef = useRef<Map<string, string>>(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
|
||||
|
||||
+3
-2
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user