reattach round: attach-first handshake, honest Reattaching placeholder

This commit is contained in:
amosblomqvist
2026-10-02 02:23:20 +03:00
parent 0751fec9d8
commit 142f1978f2
4 changed files with 112 additions and 36 deletions
+20 -20
View File
@@ -15,6 +15,7 @@ import {
import { Message, MessageContent, MessageResponse } from "@/components/ai-elements/message"; import { Message, MessageContent, MessageResponse } from "@/components/ai-elements/message";
import { Reasoning, ReasoningContent, ReasoningTrigger } from "@/components/ai-elements/reasoning"; import { Reasoning, ReasoningContent, ReasoningTrigger } from "@/components/ai-elements/reasoning";
import { Shimmer } from "@/components/ai-elements/shimmer"; import { Shimmer } from "@/components/ai-elements/shimmer";
import { BrailleSpinner } from "@/components/braille-spinner";
import { LiveToolCard } from "@/components/artifacts/live-tool-card"; import { LiveToolCard } from "@/components/artifacts/live-tool-card";
import { Separator } from "@/components/ui/separator"; import { Separator } from "@/components/ui/separator";
import { AnsweredCard } from "@/components/answered-card"; import { AnsweredCard } from "@/components/answered-card";
@@ -185,7 +186,7 @@ export function Chat({
// (core/view.ts projectView); sends are ONE explicit input event each // (core/view.ts projectView); sends are ONE explicit input event each
// ({chatId, input} — the stash indirection died with the SDK), and the // ({chatId, input} — the stash indirection died with the SDK), and the
// turn stream is the VERBATIM AgentEvent passthrough the server emits. // 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 }); useChatSession({ chatId, initialMessages, initiallyInFlight });
// The reattach settled terminal → the sync-and-continue seam (below). // The reattach settled terminal → the sync-and-continue seam (below).
@@ -200,19 +201,6 @@ export function Chat({
} }
}, [initialPrompt, initialMessages.length, send]); }, [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 // 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 // away mid-turn, or a wake turn discovered mid-session): the reattach
// stream is the primary channel (resumeStream — the same rendering path // 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 2026-10-01): once a tool call starts writing, its LIVE card
carries the activity inline — the old bottom tool-activity carries the activity inline — the old bottom tool-activity
line retired rather than double-render. Same glimmer recipe line retired rather than double-render. Same glimmer recipe
as the thinking row: duration 1. */} as the thinking row: duration 1. The REATTACH variant
{(status === "submitted" || turnInFlight) && ( (reattach round 2026-10-02): while a fresh mount reattaches
<Shimmer duration={1} className="text-sm"> to a server-side turn, the placeholder says what is actually
Thinking… happening — Reattaching… + the working-state spinner —
</Shimmer> instead of a bare Thinking shimmer. */}
)} {(status === "submitted" || turnInFlight) &&
(reattaching ? (
<span className="flex items-center gap-2">
<BrailleSpinner label="Reattaching to the running turn" />
<Shimmer duration={1} className="text-sm">
Reattaching…
</Shimmer>
</span>
) : (
<Shimmer duration={1} className="text-sm">
Thinking…
</Shimmer>
))}
{error && ( {error && (
<div className="rounded-lg border border-destructive/50 bg-destructive/10 px-3 py-2 text-sm"> <div className="rounded-lg border border-destructive/50 bg-destructive/10 px-3 py-2 text-sm">
Something went wrong: {error.message} Something went wrong: {error.message}
+37 -15
View File
@@ -140,6 +140,10 @@ export interface SessionState {
nativeMessages: LhMessage[]; nativeMessages: LhMessage[];
status: SessionStatus; status: SessionStatus;
error: { message: string } | null; 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 { export interface SessionModelOptions extends SessionModelDeps {
@@ -174,6 +178,9 @@ export class SessionModel {
private streamingMessageId: string | null = null; private streamingMessageId: string | null = null;
private statusValue: SessionStatus = "ready"; private statusValue: SessionStatus = "ready";
private errorValue: { message: string } | null = null; 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 // The single live consume (POST response or reattach stream): a newer
// consume cancels the older — one reader at a time. // consume cancels the older — one reader at a time.
private consumeAbort: AbortController | null = null; private consumeAbort: AbortController | null = null;
@@ -209,6 +216,7 @@ export class SessionModel {
nativeMessages: [...this.base, ...this.tail], nativeMessages: [...this.base, ...this.tail],
status: this.statusValue, status: this.statusValue,
error: this.errorValue, error: this.errorValue,
reattaching: this.reattachingValue,
}; };
} }
@@ -385,6 +393,7 @@ export class SessionModel {
} }
if (!sawEvent) { if (!sawEvent) {
sawEvent = true; sawEvent = true;
this.reattachingValue = false; // first event: content, not a handshake
if (this.statusValue !== "error") this.setStatus("streaming"); if (this.statusValue !== "error") this.setStatus("streaming");
} }
this.applyWireEvent(event); this.applyWireEvent(event);
@@ -392,33 +401,43 @@ export class SessionModel {
} finally { } finally {
this.streamingIds = new Set(); this.streamingIds = new Set();
this.executions = new Map(); this.executions = new Map();
this.reattachingValue = false; // settled/aborted: no handshake anymore
if (this.statusValue !== "error") this.setStatus("ready"); if (this.statusValue !== "error") this.setStatus("ready");
} }
} }
/** /**
* Reattach (the wake/handoff/mount path): a server-side turn with no * Reattach (the wake/handoff/mount path): a server-side turn with no
* live view of ours. The record is refetched FIRST (write-ahead * live view of ours. ATTACH-FIRST (owner, reattach round 2026-10-02):
* adoption — the tail's optimistic entries are absorbed, its partial * a fresh-mount reattach passes refetchRecordFirst=false — the page
* stream content is about to be fully replayed), then the stream * render just read the record, so the blocking record round-trip only
* replays the turn from its first event through the SAME consume path. * 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<void> { async resumeStream(refetchRecordFirst = true): Promise<void> {
this.consumeAbort?.abort(); this.consumeAbort?.abort();
const controller = new AbortController(); const controller = new AbortController();
this.consumeAbort = controller; this.consumeAbort = controller;
this.reattachingValue = true;
this.setStatus("submitted"); this.setStatus("submitted");
try { try {
const recordRes = await this.fetch(`/api/chats/${encodeURIComponent(this.chatId)}`, { if (refetchRecordFirst) {
signal: controller.signal, 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 (recordRes.ok) {
if (Array.isArray(record.messages)) { const record = (await recordRes.json()) as { messages?: LhMessage[] };
this.mutate(() => { if (Array.isArray(record.messages)) {
this.base = record.messages!; this.mutate(() => {
this.tail = []; this.base = record.messages!;
}); this.tail = [];
});
}
} }
} }
const res = await this.fetch(`/api/chats/${encodeURIComponent(this.chatId)}/stream`, { const res = await this.fetch(`/api/chats/${encodeURIComponent(this.chatId)}/stream`, {
@@ -428,6 +447,7 @@ export class SessionModel {
await this.consumeStream(res); await this.consumeStream(res);
} catch (err) { } catch (err) {
if ((err as Error).name === "AbortError") return; // a newer consume won if ((err as Error).name === "AbortError") return; // a newer consume won
this.reattachingValue = false;
this.setError((err as Error).message); this.setError((err as Error).message);
this.setStatus("error"); this.setStatus("error");
} }
@@ -445,6 +465,7 @@ export class SessionModel {
const controller = new AbortController(); const controller = new AbortController();
this.consumeAbort = controller; this.consumeAbort = controller;
this.setError(null); this.setError(null);
this.reattachingValue = false; // an aborted handshake is not a placeholder
this.setStatus("submitted"); this.setStatus("submitted");
// The optimistic appends — exactly the messages the server's // The optimistic appends — exactly the messages the server's
// write-ahead will persist (inputAppends), with LOCAL ids so // write-ahead will persist (inputAppends), with LOCAL ids so
@@ -492,6 +513,7 @@ export class SessionModel {
stop(): void { stop(): void {
this.consumeAbort?.abort(); this.consumeAbort?.abort();
this.consumeAbort = null; this.consumeAbort = null;
this.reattachingValue = false;
this.streamingIds = new Set(); this.streamingIds = new Set();
this.streamingMessageId = null; this.streamingMessageId = null;
this.executions = new Map(); this.executions = new Map();
+7 -1
View File
@@ -49,8 +49,13 @@ export function useChatSession({ chatId, initialMessages, initiallyInFlight }: U
}, [model, bump]); }, [model, bump]);
// Reattach on mount (StrictMode-safe: a newer resume cancels the older). // 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(() => { useEffect(() => {
if (initiallyInFlight) void model.resumeStream(); if (initiallyInFlight) void model.resumeStream(false);
}, [initiallyInFlight, model]); }, [initiallyInFlight, model]);
// The view + state, derived per model change from the 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, nativeMessages: state.nativeMessages,
status: state.status, status: state.status,
error: state.error, error: state.error,
reattaching: state.reattaching,
send, send,
submitAnswer, submitAnswer,
setMessages, setMessages,
+48
View File
@@ -468,6 +468,54 @@ describe("resumeStream (the reattach path)", () => {
assert.equal(m.getState().status, "ready"); 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<Response>((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 () => { it("turn-settled fires the settle hook and ends cleanly", async () => {
let settled = 0; let settled = 0;
globalThis.fetch = (async (input: RequestInfo | URL) => { globalThis.fetch = (async (input: RequestInfo | URL) => {