diff --git a/apps/client/src/features/ai-chat/components/chat-thread.test.tsx b/apps/client/src/features/ai-chat/components/chat-thread.test.tsx index 9aaaed60..137c993f 100644 --- a/apps/client/src/features/ai-chat/components/chat-thread.test.tsx +++ b/apps/client/src/features/ai-chat/components/chat-thread.test.tsx @@ -193,9 +193,9 @@ describe("ChatThread — send now", () => { expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "queued text" }); }); - it("#488 commit 5: autonomous live sendNow with a known runId POSTs a supersede body", () => { - // A streaming assistant message carrying the start-metadata runId -> the FSM - // adopts the run-fact; sendNow then interrupts via the server CAS. + // A streaming assistant message carrying the start-metadata runId -> the FSM + // adopts the run-fact so sendNow can interrupt via the server CAS. + const withRunId = () => { h.state.status = "streaming"; h.state.messages = [ { @@ -205,19 +205,58 @@ describe("ChatThread — send now", () => { metadata: { runId: "run-1" }, }, ]; + }; + + it("#488 commit 5 / F1: sendNow ABORTS stream A first, then sends B (CAS) only after A finalizes", async () => { + withRunId(); renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() }); fireEvent.click(screen.getByTestId("queue-btn")); fireEvent.click(screen.getByLabelText("Send now")); - // The message is sent... + // F1: A is ABORTED, and B is NOT sent yet (no overlap). + expect(h.state.stop).toHaveBeenCalledTimes(1); + expect(h.state.sendMessage).not.toHaveBeenCalled(); + // A's onFinish fires -> B is scheduled on a microtask (after A finalizes). + await act(async () => { + h.state.onFinish?.({ + message: { id: "a1", role: "assistant", parts: [] }, + isAbort: true, + isDisconnect: false, + isError: false, + }); + await Promise.resolve(); + }); expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "queued text" }); - // ...and the NEXT POST carries supersede:{runId} (read-and-cleared once). + // B's POST carries supersede:{runId} (read-and-cleared once). const prep = h.state.transport!.prepareSendMessagesRequest!; - const body = prep({ messages: [], body: {} }).body as Record; - expect(body.supersede).toEqual({ runId: "run-1" }); - // One-shot: a subsequent request has no supersede. + expect(prep({ messages: [], body: {} }).body.supersede).toEqual({ + runId: "run-1", + }); expect(prep({ messages: [], body: {} }).body.supersede).toBeUndefined(); }); + it("#488 F1: a superseded stream's LATE disconnect does NOT falsely reconnect (epoch stamp drops it)", async () => { + // MUTATION-VERIFY: drop `epoch: stampEpoch` from the supersede-branch + // FINISH_DISCONNECT dispatch and this goes red (A's disconnect -> reconnecting). + withRunId(); + renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() }); + fireEvent.click(screen.getByTestId("queue-btn")); + fireEvent.click(screen.getByLabelText("Send now")); // -> superseding, aborts A + // A ends via isDisconnect (the server CAS closed it). The OLD overlap bug would + // route the LIVE new run into a false reconnect banner. + await act(async () => { + h.state.onFinish?.({ + message: { id: "a1", role: "assistant", parts: [{ type: "text", text: "x" }] }, + isAbort: false, + isDisconnect: true, + isError: false, + }); + await Promise.resolve(); + }); + expect(screen.queryByText(/reconnecting/i)).toBeNull(); + // ...and B was still sent. + expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "queued text" }); + }); + it("#488 commit 5: a supersede POST that 409s SUPERSEDE_TIMEOUT is returned as-is (NO retry ladder)", async () => { h.state.status = "streaming"; h.state.messages = [ @@ -494,6 +533,31 @@ describe("ChatThread — resume (attach) machinery", () => { expect(onResumeFallback).not.toHaveBeenCalledWith(true); }); + it("#488 F2: a local send DURING the mount getRun round-trip is NOT hijacked by the late attach", async () => { + // getRun stays pending while the user sends locally; its late resolve would + // otherwise ATTACH_START and flip the local turn into an observer-attach. + let resolveGetRun!: (v: { + run: { id: string; status: string } | null; + message: unknown; + }) => void; + h.state.getRun.mockReturnValue( + new Promise((r) => { + resolveGetRun = r; + }), + ); + renderThread({ autonomousRunsEnabled: true, initialRows: userTail() }); + // Local send in flight of getRun -> SEND_LOCAL (phase sending, ownership local). + fireEvent.click(screen.getByTestId("send-btn")); + expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "typed text" }); + // getRun resolves with an ACTIVE run -> the F2 guard ignores ATTACH_START. + await act(async () => { + resolveGetRun({ run: { id: "run-1", status: "running" }, message: null }); + await Promise.resolve(); + }); + // The local turn was NOT hijacked into a resume/attach. + expect(h.state.resumeStream).not.toHaveBeenCalled(); + }); + it("strips the streaming tail from the seed, keeps a user tail whole", () => { renderThread({ autonomousRunsEnabled: true, initialRows: streamingTail() }); expect(h.state.seededMessages).toHaveLength(1); diff --git a/apps/client/src/features/ai-chat/components/chat-thread.tsx b/apps/client/src/features/ai-chat/components/chat-thread.tsx index 8a4d72cb..946e466c 100644 --- a/apps/client/src/features/ai-chat/components/chat-thread.tsx +++ b/apps/client/src/features/ai-chat/components/chat-thread.tsx @@ -313,6 +313,18 @@ export default function ChatThread({ // (like chatIdRef/roleIdRef), the single replacement for the three REMOVED // one-shot flags flushOnAbortRef/interruptNextSendRef/supersedeRetryRef. const pendingSupersedeRef = useRef(null); + // #488 F1: the interrupt-and-send text, held until the superseded stream A has + // FULLY finalized (its SDK `finally` cleared `activeResponse`) — only THEN is + // stream B started, so A cannot corrupt B (ai@6 AbstractChat.makeRequest reads + // `this.activeResponse.state.message` in its finally and then nulls it, so an + // overlapping B is clobbered). Send-plumbing DATA, not a lifecycle flag. + const pendingSupersedeTextRef = useRef(null); + // #488 F1: the epoch under which the CURRENTLY-OWNED stream started, used to + // STAMP its onFinish (I1). A superseded/dead stream's late finish carries an OLD + // generation and is dropped by the reducer, so it cannot drive the live machine + // into a false reconnect or reset its run-fact. Set at each honored stream start + // (local send, resume/reconnect attach, the supersede B-send). + const turnEpochRef = useRef(0); // --- Effect runner: executes the reducer's command effects --------------- const runEffect = useCallback( @@ -324,6 +336,9 @@ export default function ChatThread({ // seed already stripped it), so the live replay's text-start rebuilds it // without duplicating parts (#430). pendingAttachEpochRef.current = epoch; + // The resumed stream's onFinish is stamped with THIS attach generation + // (F1), so a superseded attempt's late finish is dropped. + turnEpochRef.current = epoch; if (machineRef.current.phase.name === "reconnecting") { const anchor = strippedRowRef.current; if (anchor) @@ -415,6 +430,8 @@ export default function ChatThread({ // FIFO dequeue + local send of the next queued message (no-op when empty). const localSend = useCallback((text: string) => { dispatchRef.current({ type: "SEND_LOCAL" }); + // F1: this local stream's onFinish is stamped with the just-bumped generation. + turnEpochRef.current = epochRef.current; sendMessageRef.current?.({ text }); }, []); const flushNext = useCallback(() => { @@ -581,6 +598,10 @@ export default function ChatThread({ transport, experimental_throttle: STREAM_THROTTLE_MS, onFinish: ({ message, isAbort, isDisconnect, isError }) => { + // #488 F1: STAMP this finish with the generation the stream STARTED under + // (I1). A superseded/dead stream's late finish carries an OLD generation and + // is dropped by the reducer, so it cannot drive the live machine. + const stampEpoch = turnEpochRef.current; // Ownership (I2) is the FSM ctx: a resumed/attached/reconnected turn is an // OBSERVER; a local send is the owner. The queue flushes ONLY under local // ownership; an observer never flushes (invariant 7). @@ -589,22 +610,50 @@ export default function ChatThread({ // for #161); fires even while unmounting. onTurnFinished(extractServerChatId(message), threadKey); + // A missing message (a pre-first-frame break) has no visible content. + const msgHasVisible = message + ? assistantMessageHasVisibleContent(message) + : false; + + // #488 F1: the SUPERSEDED stream A just finalized (we are still `superseding` + // and stream B has not been sent yet). Record A's terminal outcome STAMPED — + // I1 drops it (superseding bumped the epoch), so A cannot drive a false + // reconnect / reset the run-fact — then start B NOW that A is fully done. B is + // deferred to a microtask so A's SDK `finally` (`activeResponse = void 0`) + // runs BEFORE B's makeRequest sets `activeResponse`, else the dying A clobbers + // B (ai@6). This is the no-overlap guarantee the CAS supersede needs. + if ( + machineRef.current.phase.name === "superseding" && + pendingSupersedeTextRef.current !== null + ) { + if (isError) dispatch({ type: "FINISH_ERROR", kind: "stream", epoch: stampEpoch }); + else if (isAbort) dispatch({ type: "FINISH_ABORT", epoch: stampEpoch }); + else if (isDisconnect) + dispatch({ type: "FINISH_DISCONNECT", hasVisibleContent: msgHasVisible, epoch: stampEpoch }); + else dispatch({ type: "FINISH_CLEAN", epoch: stampEpoch }); + const text = pendingSupersedeTextRef.current; + pendingSupersedeTextRef.current = null; + setStopNotice(null); + queueMicrotask(() => { + if (!mountedRef.current) return; + turnEpochRef.current = epochRef.current; // B's (superseding) generation + sendMessageRef.current?.({ text }); + }); + return; + } + if (isError) { - dispatch({ type: "FINISH_ERROR", kind: "stream" }); + dispatch({ type: "FINISH_ERROR", kind: "stream", epoch: stampEpoch }); setStopNotice(null); return; } if (isAbort) { // A user Stop / an interrupt abort finished. The FSM stopping/idle exit is // by DATA (this terminal outcome, I4). - dispatch({ type: "FINISH_ABORT" }); + dispatch({ type: "FINISH_ABORT", epoch: stampEpoch }); setStopNotice("manual"); return; } - // A missing message (a pre-first-frame break) has no visible content. - const msgHasVisible = message - ? assistantMessageHasVisibleContent(message) - : false; if (isDisconnect) { if (wasObserver) { // A resumed/attached OBSERVER stream dropped. Recover via the degraded @@ -624,6 +673,7 @@ export default function ChatThread({ dispatch({ type: "FINISH_DISCONNECT", hasVisibleContent: hasVisible, + epoch: stampEpoch, }); } setStopNotice(null); @@ -659,10 +709,15 @@ export default function ChatThread({ dispatch({ type: "FINISH_DISCONNECT", hasVisibleContent: msgHasVisible, + epoch: stampEpoch, }); setStopNotice(null); } else { - dispatch({ type: "FINISH_DISCONNECT", hasVisibleContent: false }); + dispatch({ + type: "FINISH_DISCONNECT", + hasVisibleContent: false, + epoch: stampEpoch, + }); setStopNotice("disconnect"); } return; @@ -680,17 +735,17 @@ export default function ChatThread({ queryClient.invalidateQueries({ queryKey: AI_CHAT_MESSAGES_RQ_KEY(chatIdRef.current), }); - dispatch({ type: "STREAM_INCOMPLETE", reason: "starved" }); + dispatch({ type: "STREAM_INCOMPLETE", reason: "starved", epoch: stampEpoch }); } else { // Healthy resumed finish — nothing to restore/arm, just settle. - dispatch({ type: "FINISH_CLEAN" }); + dispatch({ type: "FINISH_CLEAN", epoch: stampEpoch }); } } setStopNotice(null); return; } // Local clean finish: settle + flush the queue (gated on liveness, #486). - dispatch({ type: "FINISH_CLEAN" }); + dispatch({ type: "FINISH_CLEAN", epoch: stampEpoch }); setStopNotice(null); if (mountedRef.current) flushNext(); }, @@ -730,9 +785,16 @@ export default function ChatThread({ if (machineRef.current.phase.name === "stopping") onServerStopRef.current?.(serverChatId); } + // #488 F4: the FIRST assistant frame of a LOCAL turn moves the FSM + // `sending -> streaming` (and adopts the runId), matching the spec's + // STREAM_START transition. Later frames / observer turns (already streaming) + // just refresh the run-fact. const runId = extractRunId(tail); - if (runId && machineRef.current.ctx.runFact?.runId !== runId) + if (machineRef.current.phase.name === "sending") { + dispatch({ type: "STREAM_START", runId, epoch: epochRef.current }); + } else if (runId && machineRef.current.ctx.runFact?.runId !== runId) { dispatch({ type: "RUN_FACT", runFact: { runId } }); + } }, [messages, onServerChatId, dispatch]); // Live "turn was interrupted" marker (a manual Stop vs a dropped connection — a @@ -839,9 +901,16 @@ export default function ChatThread({ runId !== "pending" ) { // CAS supersede: the server stops the old run + starts this one atomically. + // #488 F1: do NOT start stream B synchronously here — the local stream A is + // still live, and overlapping streams corrupt each other in ai@6 (A's + // `finally` reads/ nulls the shared `activeResponse`). Instead ABORT A now + // and stash the text; A's onFinish (dropped by the epoch stamp) starts B in + // a microtask, AFTER A is fully finalized (no overlap). The POST then + // carries the supersede body (pendingSupersedeRef, armed by the effect). setQueue(removeQueuedById(queuedRef.current, id)); + pendingSupersedeTextRef.current = msg.text; dispatch({ type: "SUPERSEDE_REQUESTED", targetRunId: runId }); - sendMessageRef.current?.({ text: msg.text }); // POST carries supersede body + stopFnRef.current?.(); // abort A -> its onFinish sends B return; } if (liveStreaming) { diff --git a/apps/client/src/features/ai-chat/state/run-fsm.spec.md b/apps/client/src/features/ai-chat/state/run-fsm.spec.md index f6a84e01..c4302e07 100644 --- a/apps/client/src/features/ai-chat/state/run-fsm.spec.md +++ b/apps/client/src/features/ai-chat/state/run-fsm.spec.md @@ -36,36 +36,47 @@ Legend: **†** = command-transition (bumps `epoch`, I1). Effects in `[…]`. | `FINISH_DISCONNECT` (no runFact, not liveFollow) | streaming | idle | runFact←null (plain terminal "connection lost") | | `STREAM_INCOMPLETE{reason}` (observer starved/torn clean finish) | streaming(observer) | polling(reason) | `[armPoll(reason)]` | | `FINISH_ERROR{kind}` (onFinish isError) | any | error(kind) | `[disarmPoll, cancelReconnect]`, runFact←null | -| `ATTACH_START{runId}` (mount resume) | idle | attaching **†** | `[resumeStream]`, ownership=observer, runFact←runId | +| `STREAM_START{runId}` (first assistant frame of a local turn) | sending | streaming | runFact←runId, `[cancelReconnect, disarmPoll]` | +| `ATTACH_START{runId}` (mount resume) | **idle only** (F2) | attaching **†** | `[resumeStream]`, ownership=observer, runFact←runId; ignored from any non-idle phase | | `ATTACH_LIVE` (attach GET 2xx) | attaching | streaming | — | | `ATTACH_NONE` (attach GET 204/err/throw) | attaching | polling(attach-none) | `[armPoll(attach-none)]` | -| `RECONNECT_BEGIN` | streaming, idle | reconnecting(1) **†** *iff runFact* | `[scheduleReconnect(1)]`, ownership=observer | | `RECONNECT_ATTEMPT{n}` (backoff timer) | reconnecting | reconnecting(n) **†** | `[resumeStream]` | | `RECONNECT_ATTACHED` (reconnect GET 2xx) | reconnecting | streaming | `[cancelReconnect, disarmPoll]` — **counter reset** (commit 3) | | `RECONNECT_NONE` (reconnect GET 204/err), attempt { expect(m.phase).toEqual({ name: "error", kind: "run-already-active" }); expect(m.effects).toEqual([]); }); +}); - it("RUN_SUPERSEDED (observer's run killed) → attaching + postRun(observer-follow)", () => { - const ctx = { epoch: 0, ownership: "observer" as const, runFact: { runId: "run-dead" }, liveFollow: false }; - const streaming: Machine = { phase: { name: "streaming" }, ctx, effects: [] }; - const m = reduce(streaming, { type: "RUN_SUPERSEDED" }); +// #488 F2 — a late mount `getRun → ATTACH_START` must not hijack a local turn. +describe("run-fsm — F2: ATTACH_START only from idle", () => { + it("ATTACH_START from a local `sending` turn is ignored (no observer hijack)", () => { + const sending = reduce(initialMachine(), { type: "SEND_LOCAL" }); // idle -> sending, local + const m = reduce(sending, { type: "ATTACH_START", runId: "r" }); + expect(m.phase.name).toBe("sending"); + expect(m.ctx.ownership).toBe("local"); // NOT flipped to observer + expect(m.effects).toEqual([]); // no resumeStream + }); + + it("ATTACH_START from idle attaches as normal", () => { + const m = reduce(initialMachine(), { type: "ATTACH_START", runId: "r" }); expect(m.phase.name).toBe("attaching"); - expect(m.effects.find((e) => e.type === "postRun")).toEqual({ - type: "postRun", - reason: "observer-follow", - }); + expect(m.ctx.ownership).toBe("observer"); + expect(hasEffect(m, "resumeStream")).toBe(true); }); }); diff --git a/apps/client/src/features/ai-chat/state/run-fsm.ts b/apps/client/src/features/ai-chat/state/run-fsm.ts index a9ff9cca..6239b450 100644 --- a/apps/client/src/features/ai-chat/state/run-fsm.ts +++ b/apps/client/src/features/ai-chat/state/run-fsm.ts @@ -113,7 +113,7 @@ export interface Machine { export type Effect = /** POST /run to (re)establish or verify the run-fact. `reason` is diagnostic. */ - | { type: "postRun"; reason: "mount" | "verify" | "observer-follow" } + | { type: "postRun"; reason: "mount" | "verify" } /** Trigger the SDK `resumeStream()` (attach GET via prepareReconnectToStream). */ | { type: "resumeStream" } /** Schedule a reconnect attempt after a backoff, then dispatch RECONNECT_ATTEMPT. */ @@ -153,15 +153,12 @@ export type Event = | { type: "ATTACH_START"; runId?: string } | { type: "ATTACH_LIVE"; epoch?: number } | { type: "ATTACH_NONE"; epoch?: number } - // -- reconnect after a live disconnect -- - /** #488 commit 2: entered by the RUN-FACT, not by an assistant message. */ - | { type: "RECONNECT_BEGIN" } + // -- reconnect after a live disconnect (entered by FINISH_DISCONNECT, #488 c2) -- | { type: "RECONNECT_ATTEMPT"; attempt: number; epoch?: number } | { type: "RECONNECT_ATTACHED"; epoch?: number } | { type: "RECONNECT_NONE"; epoch?: number } | { type: "RETRY" } // -- degraded poll -- - | { type: "POLL_ACTIVITY" } | { type: "POLL_TERMINAL" } | { type: "POLL_IDLE_CAP" } // -- run-fact (server-confirmed active run) -- @@ -175,10 +172,6 @@ export type Event = | { type: "SUPERSEDE_TIMEOUT"; epoch?: number } | { type: "SUPERSEDE_INVALID"; epoch?: number } | { type: "RUN_ALREADY_ACTIVE" } - /** An observer's attached run was aborted because it was superseded server-side: - * ask for the latest run and follow it (else an observer tab freezes on the - * killed run). */ - | { type: "RUN_SUPERSEDED" } // -- lifecycle -- | { type: "DISPOSE" }; @@ -332,6 +325,12 @@ export function reduce(m: Machine, event: Event): Machine { // ---- mount attach (resume) ---------------------------------------- case "ATTACH_START": // A reopened tab attaches to a still-running run: observer ownership. + // #488 F2: ONLY from idle. The mount `getRun` round-trip resolves async, and + // a local send may have started meanwhile (phase `sending`, ownership local); + // a late ATTACH_START must NOT hijack that local turn into an observer-attach + // (queue would stop flushing, "Send now" would hide). Guarding in the reducer + // covers every dispatch source. + if (m.phase.name !== "idle") return stay(m); return command(m, { name: "attaching" }, [{ type: "resumeStream" }], { ownership: "observer", runFact: event.runId ? { runId: event.runId } : m.ctx.runFact, @@ -351,17 +350,6 @@ export function reduce(m: Machine, event: Event): Machine { }); // ---- reconnect after a live disconnect ---------------------------- - case "RECONNECT_BEGIN": - // Explicit begin (used when the runtime decides to reconnect from a signal - // other than FINISH_DISCONNECT). Requires a run-fact (I3). - if (!m.ctx.runFact) return stay(m); - return command( - m, - { name: "reconnecting", attempt: 1, failed: false }, - [{ type: "scheduleReconnect", attempt: 1, delayMs: reconnectDelayMs(1) }], - { ownership: "observer" }, - ); - case "RECONNECT_ATTEMPT": // A scheduled backoff fired — fire the attach GET. epoch++ so the previous // attempt's late outcome cannot drive this one. @@ -420,11 +408,6 @@ export function reduce(m: Machine, event: Event): Machine { return stay(m); // ---- degraded poll ------------------------------------------------- - case "POLL_ACTIVITY": - // The polled rows changed (progress). No phase change — the runtime resets - // its inactivity clock. Only meaningful while a poll-bearing phase is active. - return stay(m); - case "POLL_TERMINAL": // The run reached a terminal row via the poll (or the reconcile merge). Go // idle and disarm everything (I4: this is a DATA-driven exit, incl. exit @@ -527,13 +510,6 @@ export function reduce(m: Machine, event: Event): Machine { // offers "interrupt and send" (supersede) instead. return to(m, { name: "error", kind: "run-already-active" }); - case "RUN_SUPERSEDED": - // An observer's attached run was killed by a server-side supersede. Ask for - // the latest run and follow it, instead of freezing on the dead run. - return command(m, { name: "attaching" }, [{ type: "postRun", reason: "observer-follow" }], { - ownership: "observer", - }); - // ---- lifecycle ----------------------------------------------------- case "DISPOSE": // Unmount: abort in-flight controllers, drop timers, and bump the epoch so