From 2a1bac89edcf0b83c5cb7ed5dabdfae7eaea339a Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Fri, 2 Oct 2026 03:00:56 +0300 Subject: [PATCH] optimistic round 3: action-gated placeholder, first-applied-event flip, clean-slate replay, empty message renders the thinking row --- components/chat.tsx | 31 ++++---- components/session-core.ts | 70 ++++++++---------- components/session.ts | 1 - src/core/view.ts | 21 +++++- tests/chat-flow.test.ts | 24 +++++++ tests/session-core.test.ts | 143 ++++++++++++++++++++++--------------- 6 files changed, 176 insertions(+), 114 deletions(-) diff --git a/components/chat.tsx b/components/chat.tsx index 48a2fe3..10619d9 100644 --- a/components/chat.tsx +++ b/components/chat.tsx @@ -186,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, awaitingFirstChunk, stop, clearError, resumeStream, setReattachSettled } = + const { messages, nativeMessages, send, submitAnswer, setMessages, adoptRecord, status, error, stop, clearError, resumeStream, setReattachSettled } = useChatSession({ chatId, initialMessages, initiallyInFlight }); // The reattach settled terminal → the sync-and-continue seam (below). @@ -1177,19 +1177,22 @@ export function Chat({ ))} - {/* THE pre-first-CONTENT placeholder, ONE grammar everywhere - (owner, optimistic round 2, 2026-10-02): spinner + - shimmered Thinking… from the instant of any action that - produces tokens (send, answer, session start, reattach, - wake) — status flips synchronously — and it stays past - agent_start (protocol events release nothing; the old - status gate vanished the line into an empty wait) until - the turn's first VISIBLE content: a reasoning/text delta - or a tool call being written, where the stream takes over - with the SAME look (the collapsible reasoning row / live - tool card). The release is sticky — no flash between - assistant steps. Never collapsible itself. */} - {(awaitingFirstChunk || turnInFlight) && ( + {/* THE placeholder, ACTION-gated (owner, optimistic round + 3, 2026-10-02): spinner + shimmered Thinking… from the + instant of any action that produces tokens — send, + answer, session start, reattach (status flips to + submitted SYNCHRONOUSLY with the action) — until the + turn's first APPLIED event, where the stream takes over + with the same look: an arrived-but-empty message renders + the thinking ROW (view.ts projects one empty reasoning + part), so submitted ⟺ placeholder and streaming ⟺ a + rendered row, mutually exclusive by construction. + Protocol events (agent_start) apply nothing and release + nothing — the placeholder covers the whole + provider-connect gap. turnInFlight: the same state + entered by external discovery (wake) / the poll + fallback. Never collapsible itself. */} + {(status === "submitted" || turnInFlight) && ( diff --git a/components/session-core.ts b/components/session-core.ts index 65bbb7d..382aef0 100644 --- a/components/session-core.ts +++ b/components/session-core.ts @@ -135,30 +135,11 @@ export function absorbed(m: LhMessage, record: readonly LhMessage[]): boolean { return false; } -/** Does this content array carry anything VISIBLE (owner, optimistic - * round 2, 2026-10-02)? A text/thinking block with text, or a tool call - * being written (the live tool card renders from its first streamed - * block). Empty leading parts (headers, empty message starts) do NOT - * count — the placeholder lives until real content, past agent_start. */ -function contentHasVisible(content: unknown): boolean { - if (!Array.isArray(content)) return false; - return content.some((b) => { - const block = b as { type?: string; text?: string }; - return block?.type === "toolCall" || (typeof block?.text === "string" && block.text.length > 0); - }); -} - export interface SessionState { messages: LhViewMessage[]; nativeMessages: LhMessage[]; status: SessionStatus; error: { message: string } | null; - /** A turn is running (this client submitted or reattached) and has - * produced NO visible content yet — the Thinking… placeholder's - * honest gate: protocol events (agent_start) do not release it, the - * turn's first content does, and the release is sticky for the rest - * of the turn (owner, optimistic round 2, 2026-10-02). */ - awaitingFirstChunk: boolean; } export interface SessionModelOptions extends SessionModelDeps { @@ -193,9 +174,6 @@ export class SessionModel { private streamingMessageId: string | null = null; private statusValue: SessionStatus = "ready"; private errorValue: { message: string } | null = null; - /** Sticky per-consume: this turn has shown visible content (see - * SessionState.awaitingFirstChunk). Reset at every consume start. */ - private sawContentValue = 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; @@ -231,8 +209,6 @@ export class SessionModel { nativeMessages: [...this.base, ...this.tail], status: this.statusValue, error: this.errorValue, - awaitingFirstChunk: - (this.statusValue === "submitted" || this.statusValue === "streaming") && !this.sawContentValue, }; } @@ -285,14 +261,18 @@ export class SessionModel { /** * ONE event application for both the POST stream and the reattach * replay (pi's events carry the FULL partial message — a view REPLACES, - * never assembles deltas). + * never assembles deltas). Returns whether the event did VIEW WORK: + * the consume flips submitted→streaming on the first ACTED event, not + * the first wire byte — agent_start/turn_start release nothing, so + * the Thinking… placeholder (gated on submitted) covers the whole + * provider-connect gap (owner, optimistic round 3, 2026-10-02). */ - applyWireEvent(event: Record): void { + applyWireEvent(event: Record): boolean { if (event.type === "error") { // Loop-level failure (agent.continue threw outside the stream) — // raw text (owner ruling, Wave D): the banner shows exactly this. this.setError(String(event.errorText ?? "An error occurred.")); - return; + return false; } const agent = event as unknown as AgentEvent; switch (agent.type) { @@ -311,14 +291,13 @@ export class SessionModel { // start+end only, no streaming partials). Without this the // live stream dropped the marker until the settle sync. const mid = message.id; - if (mid === undefined) return; + if (mid === undefined) return false; this.mutate(() => { - this.sawContentValue = true; // a rendered marker is content const idx = this.tail.findIndex((m) => m.id === mid); this.tail = idx === -1 ? [...this.tail, message as LhMessage] : this.tail.with(idx, message as LhMessage); }); - return; + return true; } // pi streams AssistantMessages with NO id — the id exists only // once the server's persistTurn stamps the record (the live @@ -334,23 +313,21 @@ export class SessionModel { // the wire) still wins. const id = message.id ?? this.streamingMessageId ?? localId(); const withId = { ...message, id } as LhMessage; - const hasContent = contentHasVisible((message as { content?: unknown }).content); if (agent.type === "message_start") { this.streamingMessageId = id; this.streamingIds.add(id); this.mutate(() => { - this.sawContentValue ||= hasContent; this.tail = [...this.tail, withId]; }); } else { this.mutate(() => { - this.sawContentValue ||= hasContent; const idx = this.tail.findIndex((m) => m.id === id); this.tail = idx === -1 ? [...this.tail, withId] : this.tail.with(idx, withId); }); } if (agent.type === "message_end") { this.streamingMessageId = null; // slot done — next start mints fresh + // (fallthrough: returns true below — the message landed) const stop = (message as { stopReason?: string }).stopReason; if (stop === "error") { // Provider error, verbatim (owner ruling) — the record @@ -360,7 +337,7 @@ export class SessionModel { ); } } - return; + return true; // a message event is view work, even empty content } case "tool_execution_end": { // The execution layer: clean question placeholders are IGNORED @@ -376,13 +353,12 @@ export class SessionModel { errorText: text, }); }); - return; + return true; // the execution layer updated the tool part's card } case "turn_end": { // The results batch: append as native messages (client-stamped // ids — the settle sync lands the server canon). this.mutate(() => { - this.sawContentValue = true; // rendered results are content this.tail = [ ...this.tail, ...(agent.toolResults as ToolResultMessage[]).map( @@ -390,10 +366,10 @@ export class SessionModel { ), ]; }); - return; + return true; } default: - return; // agent_start/turn_start/tool_execution_start|update: no view work + return false; // agent_start/turn_start/tool_execution_start|update: no view work } } @@ -403,7 +379,6 @@ export class SessionModel { if (!res.body) throw new Error("The turn stream body is empty."); const reader = res.body.getReader(); let sawEvent = false; - this.sawContentValue = false; // fresh consume: nothing visible yet try { for await (const event of sseEvents(reader)) { if (event.type === "turn-settled") { @@ -413,11 +388,11 @@ export class SessionModel { this.onReattachSettled?.(); return; } - if (!sawEvent) { + const acted = this.applyWireEvent(event); + if (!sawEvent && acted) { sawEvent = true; if (this.statusValue !== "error") this.setStatus("streaming"); } - this.applyWireEvent(event); } } finally { this.streamingIds = new Set(); @@ -443,6 +418,19 @@ export class SessionModel { this.consumeAbort?.abort(); const controller = new AbortController(); this.consumeAbort = controller; + // REPLAY ONTO A CLEAN SLATE (owner, optimistic round 3, 2026-10-02): + // the reattach stream replays the turn from its FIRST event — an + // aborted earlier attempt (React StrictMode's simulated remount) + // must not leave partials the replay would DOUBLE onto. The old + // record refetch provided this reset accidentally; attach-first + // made it explicit. Safe on both paths: a fresh mount's tail is + // empty or aborted-replay leftovers; the mid-session handoff path + // still refetches first (its reset is redundant, not harmful — + // the settle sync has already canonized the view). + this.mutate(() => { + this.tail = []; + this.streamingMessageId = null; + }); this.setStatus("submitted"); try { if (refetchRecordFirst) { diff --git a/components/session.ts b/components/session.ts index c7686be..7ce0431 100644 --- a/components/session.ts +++ b/components/session.ts @@ -86,7 +86,6 @@ export function useChatSession({ chatId, initialMessages, initiallyInFlight }: U nativeMessages: state.nativeMessages, status: state.status, error: state.error, - awaitingFirstChunk: state.awaitingFirstChunk, send, submitAnswer, setMessages, diff --git a/src/core/view.ts b/src/core/view.ts index 5eb313d..640e189 100644 --- a/src/core/view.ts +++ b/src/core/view.ts @@ -83,6 +83,11 @@ export interface LiveTurnState { * input-streaming (the shimmer); settled dangling calls render * input-available (the dock mounts from the record). */ inFlight: boolean; + /** Ids of the messages THIS consume is streaming (optimistic round 3, + * 2026-10-02): an arrived-but-empty streaming message projects one + * empty reasoning part — the thinking row — so the placeholder's + * handoff is seamless; settled empty messages stay hidden. */ + streamingIds: ReadonlySet; } function textPart(text: string): LhViewPart { @@ -178,7 +183,21 @@ export function projectView(messages: readonly LhMessage[], live?: LiveTurnState } } const parts: LhViewPart[] = [...reasoning, ...rest]; - if (parts.length === 0) continue; // presentation-only (nothing renderable) + if (parts.length === 0) { + // An arrived-but-silent STREAMING message (owner, optimistic + // round 3, 2026-10-02): the turn's first applied event can + // carry empty content — project ONE empty reasoning part so + // the message's ARRIVAL renders the thinking row (chat.tsx's + // Reasoning isStreamings the last part: the same spinner + + // shimmer as the placeholder it replaces — the handoff is + // seamless and the two can never double). Settled empty + // messages stay hidden (presentation-only). + if (live?.streamingIds.has(m.id)) { + parts.push({ type: "reasoning", id: `${m.id}-r0`, text: "", state: "done" }); + } else { + continue; + } + } const group = lastAssistant(); if (group) { // Same group: the step separator renders between the steps. diff --git a/tests/chat-flow.test.ts b/tests/chat-flow.test.ts index 0618e61..c636e9b 100644 --- a/tests/chat-flow.test.ts +++ b/tests/chat-flow.test.ts @@ -540,6 +540,30 @@ describe("teacher chat flow (scripted model)", () => { assert.deepEqual(view2[1]!.parts.filter((p) => p.type !== "step-start").map((p) => p.type), ["reasoning", "text"], "think-first messages project identically"); }); + it("an arrived-but-empty STREAMING message projects the thinking row; a settled empty one stays hidden (optimistic round 3)", () => { + const emptyArrived: LhMessage = { + id: "a_empty", + role: "assistant", + content: [], + stopReason: "stop", + } as never; + const live = { + executions: new Map(), + inFlight: true, + streamingIds: new Set(["a_empty"]), + } as never; + // streaming: the message's ARRIVAL renders — one empty reasoning part + // (chat.tsx Reasoning isStreamings the last part: same spinner + + // shimmer as the placeholder it replaces — seamless handoff). + const view = projectView([{ id: "u0", role: "user", content: "hi", timestamp: 1 } as never, emptyArrived], live); + const parts = view[1]!.parts.filter((p) => p.type !== "step-start"); + assert.deepEqual(parts.map((p) => p.type), ["reasoning"], "empty streaming message = the thinking row"); + assert.equal(parts[0]!.text, "", "empty until the first token lands"); + // settled (no live window): presentation-only, hidden as always. + const settled = projectView([{ id: "u0", role: "user", content: "hi", timestamp: 1 } as never, emptyArrived]); + assert.equal(settled.length, 1, "a settled empty message renders nothing"); + }); + it("a cancelled view NEVER aborts the turn — it runs to completion (owner ruling 2026-09-25)", async () => { let release!: () => void; const gate = new Promise((r) => (release = r)); diff --git a/tests/session-core.test.ts b/tests/session-core.test.ts index 459afe2..74d24c6 100644 --- a/tests/session-core.test.ts +++ b/tests/session-core.test.ts @@ -513,20 +513,28 @@ describe("resumeStream (the reattach path)", () => { // submitted at entry, before its fetches). }); - it("CONTENT-GATED, not status-gated (owner, optimistic round 2): agent_start does NOT release the placeholder — the first visible content does", async () => { - // The old bug: the server emits agent_start instantly (pre-provider), - // the first wire event flipped status to streaming, and the line - // VANISHED into an empty wait until the provider's first token. - let release!: () => void; - const gate = new Promise((res) => (release = res)); + it("PROTOCOL EVENTS RELEASE NOTHING (owner, optimistic round 3): agent_start keeps the placeholder; the first APPLIED event flips streaming", async () => { + // The round-2 bug chain, fixed at the machine: status used to flip + // streaming at the first wire BYTE (agent_start, emitted before the + // provider has produced anything), vanishing the placeholder into an + // empty wait. Now the flip waits for an event applyWireEvent ACTS + // on — the placeholder (gated on submitted) covers the whole + // provider-connect gap, and the applied event's message renders the + // thinking row even when its content is empty (view.ts projection). + let releaseProtocol!: () => void; + let releaseRest!: () => void; + const protocolGate = new Promise((res) => (releaseProtocol = res)); + const restGate = new Promise((res) => (releaseRest = res)); const body = new ReadableStream({ async start(c) { const enc = new TextEncoder(); c.enqueue(enc.encode(sse({ type: "agent_start" }, { type: "turn_start" }))); - await gate; // the provider is slow — nothing but protocol so far + await protocolGate; // the provider is slow — nothing but protocol so far + c.enqueue(enc.encode(sse({ type: "message_start", message: { role: "assistant", content: [] } }))); + await restGate; // the message arrived; its first token is still pending c.enqueue( enc.encode( - sse({ type: "message_update", message: { role: "assistant", content: [{ type: "text", text: "hi" }] } }), + sse({ type: "message_update", message: { role: "assistant", content: [{ type: "thinking", thinking: "pondering" }] } }), ), ); c.enqueue(enc.encode(sse({ type: "agent_end", messages: [] }, "[DONE]"))); @@ -537,61 +545,82 @@ describe("resumeStream (the reattach path)", () => { new Response(body, { headers: { "Content-Type": "text/event-stream" } })) as typeof globalThis.fetch; const m = model(); const p = m.send({ type: "text", text: "q" }); - assert.equal(m.getState().awaitingFirstChunk, true, "submitted: placeholder up"); - await new Promise((r) => setTimeout(r, 25)); // let agent_start arrive + consume - assert.equal(m.getState().status, "streaming", "agent_start arrived — status flipped"); - assert.equal(m.getState().awaitingFirstChunk, true, "protocol events release NOTHING: no vanish into the empty wait"); - release(); + assert.equal(m.getState().status, "submitted", "the action is optimistic: placeholder up pre-everything"); + await new Promise((r) => setTimeout(r, 25)); // agent_start + turn_start consumed + assert.equal( + m.getState().status, + "submitted", + "protocol events apply NOTHING — the placeholder holds through the connect gap", + ); + releaseProtocol(); + await new Promise((r) => setTimeout(r, 25)); // message_start consumed + assert.equal(m.getState().status, "streaming", "the first APPLIED event flips streaming — the thinking row renders"); + releaseRest(); await p; assert.equal(m.getState().status, "ready"); - assert.equal(m.getState().awaitingFirstChunk, false, "the turn ended with content — nothing awaited"); }); - it("EMPTY leading parts do not release; the release is STICKY for the rest of the turn", async () => { - let release!: () => void; - const gate = new Promise((res) => (release = res)); - const body = new ReadableStream({ - async start(c) { - const enc = new TextEncoder(); - // message_start with EMPTY content: an empty thinking header, - // nothing renderable — the placeholder must stay. - c.enqueue(enc.encode(sse({ type: "message_start", message: { role: "assistant", content: [] } }))); - await gate; - c.enqueue( - enc.encode( - sse({ type: "message_update", message: { role: "assistant", content: [{ type: "thinking", text: "pondering" }] } }), - ), + it("REATTACH REPLAYS ONTO A CLEAN SLATE (owner, optimistic round 3): an aborted attempt's partials never double", async () => { + // StrictMode's simulated remount aborts the first resume mid-replay; + // its partial tail survived into the retry and the replay DOUBLED + // every message (the record refetch's accidental tail reset died + // with attach-first). resumeStream now wipes the tail itself. + let hold!: () => void; + const holdGate = new Promise((res) => (hold = res)); + let calls = 0; + const enc = new TextEncoder(); + const replay = () => { + calls++; + if (calls === 1) + return new Response( + new ReadableStream({ + async start(c) { + c.enqueue( + enc.encode( + sse({ type: "message_update", message: { role: "assistant", content: [{ type: "text", text: "part one" }] } }), + ), + ); + await holdGate; // aborted here (the partial stays in the tail) + c.enqueue(enc.encode(sse({ type: "agent_end", messages: [] }, "[DONE]"))); + c.close(); + }, + }), + { headers: { "Content-Type": "text/event-stream" } }, ); - c.enqueue( - enc.encode( - sse({ type: "message_end", message: { role: "assistant", content: [{ type: "thinking", text: "pondering" }], stopReason: "stop" } }), - ), - ); - // a SECOND assistant message opens with empty content — the - // release must NOT flash the placeholder back mid-turn. - c.enqueue(enc.encode(sse({ type: "message_start", message: { role: "assistant", content: [] } }))); - await new Promise((r) => setTimeout(r, 10)); - c.enqueue( - enc.encode( - sse({ type: "message_end", message: { role: "assistant", content: [{ type: "text", text: "done" }], stopReason: "stop" } }), - ), - ); - c.enqueue(enc.encode(sse({ type: "agent_end", messages: [] }, "[DONE]"))); - c.close(); - }, - }); - globalThis.fetch = (async () => - new Response(body, { headers: { "Content-Type": "text/event-stream" } })) as typeof globalThis.fetch; + return new Response( + new ReadableStream({ + async start(c) { + c.enqueue( + enc.encode( + sse( + { type: "message_start", message: { role: "assistant", content: [{ type: "text", text: "part one" }] } }, + { type: "agent_end", messages: [] }, + "[DONE]", + ), + ), + ); + c.close(); + }, + }), + { headers: { "Content-Type": "text/event-stream" } }, + ); + }; + globalThis.fetch = (async (input: RequestInfo | URL) => { + const url = String(input); + if (url === "/api/chats/chat_x/stream") return replay(); + throw new Error(`unexpected fetch: ${url}`); + }) as typeof globalThis.fetch; const m = model(); - const p = m.send({ type: "text", text: "q" }); - await new Promise((r) => setTimeout(r, 25)); // message_start (empty) consumed - assert.equal(m.getState().status, "streaming"); - assert.equal(m.getState().awaitingFirstChunk, true, "empty leading parts render nothing — placeholder stays"); - release(); - await new Promise((r) => setTimeout(r, 25)); // first thinking delta + second empty message_start - assert.equal(m.getState().awaitingFirstChunk, false, "first content released — sticky through the empty second start"); - await p; - assert.equal(m.getState().awaitingFirstChunk, false); + const attempt1 = m.resumeStream(false); + await new Promise((r) => setTimeout(r, 25)); // attempt 1 consumed its partial + // the simulated remount: attempt 2 aborts attempt 1 mid-replay + const attempt2 = m.resumeStream(false); + hold(); // attempt 1's gate resolves after it was already aborted + await Promise.allSettled([attempt1, attempt2]); + const texts = m + .getState() + .messages.flatMap((msg) => msg.parts.filter((p) => p.type === "text").map((p) => (p as { text?: string }).text)); + assert.deepEqual(texts, ["part one"], "the replayed turn appears exactly ONCE — no doubles"); }); it("turn-settled fires the settle hook and ends cleanly", async () => {