From 142f1978f2e5a13ae2abcd9e9bb9082c0516982d Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Fri, 2 Oct 2026 02:23:20 +0300 Subject: [PATCH] reattach round: attach-first handshake, honest Reattaching placeholder --- components/chat.tsx | 40 ++++++++++++++--------------- components/session-core.ts | 52 +++++++++++++++++++++++++++----------- components/session.ts | 8 +++++- tests/session-core.test.ts | 48 +++++++++++++++++++++++++++++++++++ 4 files changed, 112 insertions(+), 36 deletions(-) diff --git a/components/chat.tsx b/components/chat.tsx index cf74ec1..d893766 100644 --- a/components/chat.tsx +++ b/components/chat.tsx @@ -15,6 +15,7 @@ import { import { Message, MessageContent, MessageResponse } from "@/components/ai-elements/message"; import { Reasoning, ReasoningContent, ReasoningTrigger } from "@/components/ai-elements/reasoning"; import { Shimmer } from "@/components/ai-elements/shimmer"; +import { BrailleSpinner } from "@/components/braille-spinner"; import { LiveToolCard } from "@/components/artifacts/live-tool-card"; import { Separator } from "@/components/ui/separator"; import { AnsweredCard } from "@/components/answered-card"; @@ -185,7 +186,7 @@ export function Chat({ // (core/view.ts projectView); sends are ONE explicit input event each // ({chatId, input} — the stash indirection died with the SDK), and the // turn stream is the VERBATIM AgentEvent passthrough the server emits. - const { messages, nativeMessages, send, submitAnswer, setMessages, adoptRecord, status, error, stop, clearError, resumeStream, setReattachSettled } = + const { messages, nativeMessages, send, submitAnswer, setMessages, adoptRecord, status, error, reattaching, stop, clearError, resumeStream, setReattachSettled } = useChatSession({ chatId, initialMessages, initiallyInFlight }); // The reattach settled terminal → the sync-and-continue seam (below). @@ -200,19 +201,6 @@ export function Chat({ } }, [initialPrompt, initialMessages.length, send]); - // Reattach on mount (stream-reattach wave): a turn that outlived its - // client (navigate away mid-turn, back while in flight) is re-entered - // through the SAME rendering path as a live one — resumeStream() feeds - // the replay burst (instant catch-up) then the live deltas into the - // SDK's own chunk pipeline. Raced settle is the transport's turn-settled - // terminal: onSettled refetches the record. No once-ref guard: every - // mount resumes (the deps are stable — this fires once per mount), and - // React StrictMode's dev double-mount ABORTS the first attempt at - // dispose — the retry must be allowed to run. - useEffect(() => { - if (initiallyInFlight) void resumeStream(); - }, [resumeStream, initiallyInFlight]); - // A turn running server-side, attached to no live view of ours (navigate // away mid-turn, or a wake turn discovered mid-session): the reattach // stream is the primary channel (resumeStream — the same rendering path @@ -1193,12 +1181,24 @@ export function Chat({ 2026-10-01): once a tool call starts writing, its LIVE card carries the activity inline — the old bottom tool-activity line retired rather than double-render. Same glimmer recipe - as the thinking row: duration 1. */} - {(status === "submitted" || turnInFlight) && ( - - Thinking… - - )} + as the thinking row: duration 1. The REATTACH variant + (reattach round 2026-10-02): while a fresh mount reattaches + to a server-side turn, the placeholder says what is actually + happening — Reattaching… + the working-state spinner — + instead of a bare Thinking shimmer. */} + {(status === "submitted" || turnInFlight) && + (reattaching ? ( + + + + Reattaching… + + + ) : ( + + Thinking… + + ))} {error && (
Something went wrong: {error.message} diff --git a/components/session-core.ts b/components/session-core.ts index 5a456dd..b2c7e35 100644 --- a/components/session-core.ts +++ b/components/session-core.ts @@ -140,6 +140,10 @@ export interface SessionState { nativeMessages: LhMessage[]; status: SessionStatus; error: { message: string } | null; + /** A reattach handshake is in flight (resumeStream, no events yet): + * the view shows the honest “Reattaching…” placeholder instead of a + * bare Thinking shimmer (owner, reattach round 2026-10-02). */ + reattaching: boolean; } export interface SessionModelOptions extends SessionModelDeps { @@ -174,6 +178,9 @@ export class SessionModel { private streamingMessageId: string | null = null; private statusValue: SessionStatus = "ready"; private errorValue: { message: string } | null = null; + /** True from resumeStream() entry until its first event / settle / + * error — the handshake window (see SessionState.reattaching). */ + private reattachingValue = false; // The single live consume (POST response or reattach stream): a newer // consume cancels the older — one reader at a time. private consumeAbort: AbortController | null = null; @@ -209,6 +216,7 @@ export class SessionModel { nativeMessages: [...this.base, ...this.tail], status: this.statusValue, error: this.errorValue, + reattaching: this.reattachingValue, }; } @@ -385,6 +393,7 @@ export class SessionModel { } if (!sawEvent) { sawEvent = true; + this.reattachingValue = false; // first event: content, not a handshake if (this.statusValue !== "error") this.setStatus("streaming"); } this.applyWireEvent(event); @@ -392,33 +401,43 @@ export class SessionModel { } finally { this.streamingIds = new Set(); this.executions = new Map(); + this.reattachingValue = false; // settled/aborted: no handshake anymore if (this.statusValue !== "error") this.setStatus("ready"); } } /** * Reattach (the wake/handoff/mount path): a server-side turn with no - * live view of ours. The record is refetched FIRST (write-ahead - * adoption — the tail's optimistic entries are absorbed, its partial - * stream content is about to be fully replayed), then the stream - * replays the turn from its first event through the SAME consume path. + * live view of ours. ATTACH-FIRST (owner, reattach round 2026-10-02): + * a fresh-mount reattach passes refetchRecordFirst=false — the page + * render just read the record, so the blocking record round-trip only + * added handshake latency before the instant replay; the running + * turn's partial lives solely in the server event log, and the replay + * rebuilds it regardless of the refetch. Mid-session callers (the + * handoff watch) keep the default: their base may trail the record, + * and the refetch absorbs it (write-ahead adoption) before the replay + * lands on top. The stream then replays the turn from its first event + * through the SAME consume path. */ - async resumeStream(): Promise { + async resumeStream(refetchRecordFirst = true): Promise { this.consumeAbort?.abort(); const controller = new AbortController(); this.consumeAbort = controller; + this.reattachingValue = true; this.setStatus("submitted"); try { - const recordRes = await this.fetch(`/api/chats/${encodeURIComponent(this.chatId)}`, { - signal: controller.signal, - }); - if (recordRes.ok) { - const record = (await recordRes.json()) as { messages?: LhMessage[] }; - if (Array.isArray(record.messages)) { - this.mutate(() => { - this.base = record.messages!; - this.tail = []; - }); + if (refetchRecordFirst) { + const recordRes = await this.fetch(`/api/chats/${encodeURIComponent(this.chatId)}`, { + signal: controller.signal, + }); + if (recordRes.ok) { + const record = (await recordRes.json()) as { messages?: LhMessage[] }; + if (Array.isArray(record.messages)) { + this.mutate(() => { + this.base = record.messages!; + this.tail = []; + }); + } } } const res = await this.fetch(`/api/chats/${encodeURIComponent(this.chatId)}/stream`, { @@ -428,6 +447,7 @@ export class SessionModel { await this.consumeStream(res); } catch (err) { if ((err as Error).name === "AbortError") return; // a newer consume won + this.reattachingValue = false; this.setError((err as Error).message); this.setStatus("error"); } @@ -445,6 +465,7 @@ export class SessionModel { const controller = new AbortController(); this.consumeAbort = controller; this.setError(null); + this.reattachingValue = false; // an aborted handshake is not a placeholder this.setStatus("submitted"); // The optimistic appends — exactly the messages the server's // write-ahead will persist (inputAppends), with LOCAL ids so @@ -492,6 +513,7 @@ export class SessionModel { stop(): void { this.consumeAbort?.abort(); this.consumeAbort = null; + this.reattachingValue = false; this.streamingIds = new Set(); this.streamingMessageId = null; this.executions = new Map(); diff --git a/components/session.ts b/components/session.ts index 4e2e2c4..0a90192 100644 --- a/components/session.ts +++ b/components/session.ts @@ -49,8 +49,13 @@ export function useChatSession({ chatId, initialMessages, initiallyInFlight }: U }, [model, bump]); // Reattach on mount (StrictMode-safe: a newer resume cancels the older). + // ATTACH-FIRST (owner, reattach round 2026-10-02): the page render just + // read the record — the blocking record refetch only added handshake + // latency before the instant replay, so a fresh mount attaches + // directly. This hook's effect is the SINGLE mount-reattach owner + // (chat.tsx's duplicate was deleted with the same round). useEffect(() => { - if (initiallyInFlight) void model.resumeStream(); + if (initiallyInFlight) void model.resumeStream(false); }, [initiallyInFlight, model]); // The view + state, derived per model change from the model. @@ -81,6 +86,7 @@ export function useChatSession({ chatId, initialMessages, initiallyInFlight }: U nativeMessages: state.nativeMessages, status: state.status, error: state.error, + reattaching: state.reattaching, send, submitAnswer, setMessages, diff --git a/tests/session-core.test.ts b/tests/session-core.test.ts index 890862e..f2f7101 100644 --- a/tests/session-core.test.ts +++ b/tests/session-core.test.ts @@ -468,6 +468,54 @@ describe("resumeStream (the reattach path)", () => { assert.equal(m.getState().status, "ready"); }); + it("ATTACH-FIRST (owner, reattach round 2026-10-02): a fresh-mount reattach skips the blocking record refetch", async () => { + const urls: string[] = []; + globalThis.fetch = (async (input: RequestInfo | URL) => { + const url = String(input); + urls.push(url); + if (url === "/api/chats/chat_x/stream") + return sseResponse( + { type: "message_start", message: { role: "assistant", content: [{ type: "text", text: "hi" }] } }, + { type: "message_end", message: { role: "assistant", content: [{ type: "text", text: "hi" }], stopReason: "stop" } }, + { type: "agent_end", messages: [] }, + "[DONE]", + ); + throw new Error(`unexpected fetch: ${url}`); + }) as typeof globalThis.fetch; + const m = model(); + await m.resumeStream(false); + // one round-trip: straight to the stream — the page render already + // read the record; the replay rebuilds the partial regardless. + assert.deepEqual(urls, ["/api/chats/chat_x/stream"]); + assert.equal(m.getState().status, "ready"); + // the mid-session caller (handoff watch) keeps the default refetch + // (the earlier test in this block pins that order). + }); + + it("reattaching is honest: true during the handshake, false once events flow", async () => { + let release!: (r: Response) => void; + const gate = new Promise((res) => (release = res)); + globalThis.fetch = (async (input: RequestInfo | URL) => { + const url = String(input); + if (url === "/api/chats/chat_x") return gate; // the record fetch hangs + if (url === "/api/chats/chat_x/stream") + return sseResponse( + { type: "message_start", message: { role: "assistant", content: [{ type: "text", text: "hi" }] } }, + { type: "agent_end", messages: [] }, + "[DONE]", + ); + throw new Error(`unexpected fetch: ${url}`); + }) as typeof globalThis.fetch; + const m = model(); + const p = m.resumeStream(); // runs sync to the first await (the gated fetch) + assert.equal(m.getState().reattaching, true, "mid-handshake: the placeholder is Reattaching…"); + assert.equal(m.getState().status, "submitted"); + release(Response.json({ messages: [] })); + await p; + assert.equal(m.getState().reattaching, false, "first event landed: content, not a handshake"); + assert.equal(m.getState().status, "ready"); + }); + it("turn-settled fires the settle hook and ends cleanly", async () => { let settled = 0; globalThis.fetch = (async (input: RequestInfo | URL) => {