diff --git a/components/session-core.ts b/components/session-core.ts index 2d765fa..657d0da 100644 --- a/components/session-core.ts +++ b/components/session-core.ts @@ -227,7 +227,10 @@ export class SessionModel { /** The in-flight assistant message's client id (see applyWireEvent). */ private streamingMessageId: string | null = null; private statusValue: SessionStatus = "ready"; + /** Request/loop failures are transient; provider failures derive from messages. */ private errorValue: { message: string } | null = null; + /** A completed failure from THIS consume, not the previous attempt's record. */ + private liveFailureId: string | null = null; // 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; @@ -258,12 +261,22 @@ export class SessionModel { this.statusValue === "submitted" || this.statusValue === "streaming" ? { executions: this.executions, inFlight: true, streamingIds: this.streamingIds } : undefined; + const messages = [...this.base, ...this.tail]; + const latest = messages.at(-1); + // One rule for live and restored failures: only the final completed + // assistant can own the banner. While trying again, hide the old + // failure from the first Thinking frame, but show a new failure at end. + const providerError = + latest?.role === "assistant" && latest.stopReason === "error" && + !this.incompleteUsageIds.has(latest.id) && (!live || latest.id === this.liveFailureId) + ? { message: latest.errorMessage ?? "An error occurred." } + : null; return { - messages: projectView([...this.base, ...this.tail], live), - nativeMessages: [...this.base, ...this.tail], - contextTokens: measuredContextTokens([...this.base, ...this.tail], this.incompleteUsageIds), + messages: projectView(messages, live), + nativeMessages: messages, + contextTokens: measuredContextTokens(messages, this.incompleteUsageIds), status: this.statusValue, - error: this.errorValue, + error: this.errorValue ?? providerError, steerQueue: this.steerQueueValue, }; } @@ -279,6 +292,7 @@ export class SessionModel { } private setStatus(status: SessionStatus): void { + if (status === "submitted") this.liveFailureId = null; this.statusValue = status; this.readyRef = status === "ready"; this.onChange?.(); @@ -369,8 +383,15 @@ export class SessionModel { // the wire) still wins. const id = message.id ?? this.streamingMessageId ?? localId(); const withId = { ...message, id } as LhMessage; - if (agent.type === "message_end") this.incompleteUsageIds.delete(id); - else this.incompleteUsageIds.add(id); + if (agent.type === "message_end") { + this.incompleteUsageIds.delete(id); + if (message.role === "assistant" && message.stopReason === "error") { + this.liveFailureId = id; + // A newer provider failure replaces an earlier connection/loop + // error, just as the live message_end did before restoration. + this.errorValue = null; + } + } else this.incompleteUsageIds.add(id); if (agent.type === "message_start") { this.streamingMessageId = id; this.streamingIds.add(id); @@ -385,15 +406,6 @@ export class SessionModel { } 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 - // persists the same message; no masking anywhere. - this.setError( - String((message as { errorMessage?: string }).errorMessage ?? "An error occurred."), - ); - } } return true; // a message event is view work, even empty content } @@ -433,17 +445,18 @@ export class SessionModel { /** Consume one SSE body to its [DONE] terminal. The turn-settled * terminal (stream-route only) fires the settle hook and ends. */ - async consumeStream(res: Response): Promise { + async consumeStream(res: Response, controller = this.consumeAbort): Promise { if (!res.body) throw new Error("The turn stream body is empty."); const reader = res.body.getReader(); let sawEvent = false; try { for await (const event of sseEvents(reader)) { + if (this.consumeAbort !== controller) return; if (event.type === "turn-settled") { // The raced-settle terminal: no turn to attach to — the // record is current; the settle hook refetches. await reader.cancel().catch(() => {}); - this.onReattachSettled?.(); + if (this.consumeAbort === controller) this.onReattachSettled?.(); return; } const acted = this.applyWireEvent(event); @@ -453,9 +466,13 @@ export class SessionModel { } } } finally { - this.streamingIds = new Set(); - this.executions = new Map(); - if (this.statusValue !== "error") this.setStatus("ready"); + // A cancelled consume may finish after its replacement has entered + // Thinking. Only the current owner can settle shared state. + if (this.consumeAbort === controller) { + this.streamingIds = new Set(); + this.executions = new Map(); + if (this.statusValue !== "error") this.setStatus("ready"); + } } } @@ -505,8 +522,10 @@ export class SessionModel { const recordRes = await this.fetch(`/api/chats/${encodeURIComponent(this.chatId)}`, { signal: controller.signal, }); + if (this.consumeAbort !== controller) return; if (recordRes.ok) { const record = (await recordRes.json()) as { messages?: LhMessage[] }; + if (this.consumeAbort !== controller) return; if (Array.isArray(record.messages)) { this.mutate(() => { this.base = record.messages!; @@ -527,10 +546,11 @@ export class SessionModel { const res = await this.fetch(`/api/chats/${encodeURIComponent(this.chatId)}/stream`, { signal: controller.signal, }); + if (this.consumeAbort !== controller) return; if (!res.ok) throw new Error((await res.text()) || "Failed to attach to the running turn."); - await this.consumeStream(res); + await this.consumeStream(res, controller); } catch (err) { - if ((err as Error).name === "AbortError") return; // a newer consume won + if (this.consumeAbort !== controller || (err as Error).name === "AbortError") return; // a newer consume won this.setError((err as Error).message); this.setStatus("error"); } diff --git a/tests/session-core.test.ts b/tests/session-core.test.ts index f13e614..7e5ecd7 100644 --- a/tests/session-core.test.ts +++ b/tests/session-core.test.ts @@ -25,7 +25,7 @@ import { absorbed, pendingCallTool, SessionModel } from "../components/session-c import type { ChatInput } from "../src/core/input-events.ts"; import type { LhMessage } from "../src/core/native-messages.ts"; import { newUserMessage, newAnswerResult, newThinkingMessage, needsContinuation, endsInDismissal } from "../src/core/native-messages.ts"; -import { projectView } from "../src/core/index.ts"; +import { newCompactionMarker, newPlanBlockMessage, projectView } from "../src/core/index.ts"; import { dismissedAnswer } from "../src/core/question-core.ts"; import { buildAnswer, type QuestionData } from "../src/core/question-core.ts"; @@ -483,6 +483,207 @@ it("Wave C: markers injected mid-run (custom role, follow-up delivery) append to }); }); +// ─── provider failures: one final-entry rule for live and saved records ────── + +describe("provider-error banner (live and saved)", () => { + function failed(id = "failed", errorMessage: string | undefined = "Google 500: fixture failure"): LhMessage { + return { ...assistantMsg(id, ""), content: [], stopReason: "error", errorMessage } as LhMessage; + } + + it("restores an empty failed response on load and full sync without adding a transcript card", () => { + const record = [newUserMessage("topic"), failed()]; + const m = model({ initialMessages: record }); + assert.equal(m.getState().error?.message, "Google 500: fixture failure"); + assert.equal(m.getState().status, "ready"); + assert.equal(m.readyRef, true, "a restored provider failure does not block retry"); + assert.deepEqual(m.getState().nativeMessages, record, "diagnostic history stays intact"); + assert.equal(m.getState().messages.length, 1, "only the user bubble, no permanent error card"); + const synced = model(); + synced.setMessages(record); + assert.equal(synced.getState().error?.message, m.getState().error?.message); + const adopted = model({ initialMessages: record.slice(0, 1) }); + adopted.adoptRecord(record); + assert.equal(adopted.getState().error?.message, m.getState().error?.message); + }); + + it("uses the existing fallback wording when the saved failure has no error text", () => { + const withoutText = { ...failed(), errorMessage: undefined } as LhMessage; + assert.equal(model({ initialMessages: [withoutText] }).getState().error?.message, "An error occurred."); + assert.equal(model().getState().error, null); + }); + + it("any later entry supersedes the failure, including plan/compaction markers and aborts", () => { + const failure = failed(); + const later: LhMessage[] = [ + newUserMessage("retry"), + assistantMsg("success", "Hello"), + { ...assistantMsg("aborted", ""), stopReason: "aborted", errorMessage: "Aborted" } as LhMessage, + newThinkingMessage("problem", "my attempt"), + resultMsg("call"), + newPlanBlockMessage("plan", { kind: "full", nodes: [], edges: [], current: null }), + newCompactionMarker("summary", { firstKeptMessageId: failure.id, tokensBefore: 100, model: "fixture", at: "2026-01-01T00:00:00Z" }), + ]; + for (const entry of later) { + const m = model({ initialMessages: [failure] }); + m.applyWireEvent({ type: "message_end", message: failure }); + assert.ok(m.getState().error, "the live failure is initially visible"); + m.setMessages([failure, entry]); + assert.equal(m.getState().error, null, `live banner superseded by ${entry.role}`); + assert.equal(model({ initialMessages: [failure, entry] }).getState().error, null, "reopened view agrees"); + } + }); + + it("a newer live message supersedes the failure before its response finishes", () => { + const m = model(); + m.applyWireEvent({ type: "message_end", message: failed() }); + assert.ok(m.getState().error); + m.applyWireEvent({ type: "message_start", message: assistantMsg("next", "") }); + assert.equal(m.getState().error, null); + m.applyWireEvent({ type: "message_end", message: assistantMsg("next", "Hello") }); + assert.equal(m.getState().error, null); + }); + + it("does not treat an unfinished assistant's error-shaped partial as a completed failure", () => { + const m = model(); + const failure = failed(); + m.applyWireEvent({ type: "message_start", message: failure }); + assert.equal(m.getState().error, null); + m.applyWireEvent({ type: "message_update", message: failure }); + assert.equal(m.getState().error, null); + m.applyWireEvent({ type: "message_end", message: failure }); + assert.equal(m.getState().error?.message, "Google 500: fixture failure"); + }); + + for (const input of [{ type: "text", text: "retry" }, { type: "continue" }] as const) { + it(`hides the old failure during the optimistic ${input.type} wait, then shows a new live failure`, async () => { + let acknowledge!: (response: Response) => void; + const ack = new Promise((resolve) => { acknowledge = resolve; }); + const old = failed("old", "old failure"); + const fresh = failed("fresh", "new failure"); + let finish!: () => void; + const stream = new ReadableStream({ start(controller) { + finish = () => { + controller.enqueue(new TextEncoder().encode(sse("[DONE]"))); + controller.close(); + }; + } }); + const m = model({ initialMessages: [old], fetch: async (url) => { + if (url === "/api/chat") return ack; + if (url === "/api/chats/chat_x") return Response.json({ messages: [old] }); + if (url === "/api/chats/chat_x/stream") return new Response(stream); + throw new Error(`unexpected fetch: ${url}`); + } }); + assert.equal(m.getState().error?.message, "old failure"); + const sending = m.send(input); + assert.equal(m.getState().status, "submitted"); + assert.equal(m.getState().error, null, "hidden even before the acknowledgement"); + acknowledge(Response.json({ ok: true })); + await new Promise((resolve) => setTimeout(resolve, 0)); + assert.equal(m.getState().error, null, "refetching the old failed record does not resurrect it"); + m.applyWireEvent({ type: "message_start", message: fresh }); + assert.equal(m.getState().error, null, "no error from an unfinished partial"); + m.applyWireEvent({ type: "message_end", message: fresh }); + assert.equal(m.getState().status, "submitted"); + assert.equal(m.getState().error?.message, "new failure", "new failure appears before stream closure"); + finish(); + await sending; + assert.equal(m.getState().error?.message, "new failure"); + }); + } + + it("hides the old failure immediately on fresh-mount reattach, and restores final-entry truth at idle", async () => { + let settle!: (response: Response) => void; + const response = new Promise((resolve) => { settle = resolve; }); + const m = model({ initialMessages: [failed()], fetch: async () => response }); + const attaching = m.resumeStream(false); + assert.equal(m.getState().status, "submitted"); + assert.equal(m.getState().error, null); + settle(sseResponse({ type: "turn-settled" }, "[DONE]")); + await attaching; + assert.equal(m.getState().error?.message, "Google 500: fixture failure", "no newer entry superseded it"); + }); + + it("recovers a failure that settled before the submitter could attach to its stream", async () => { + const record = [newUserMessage("topic"), failed()]; + const m = model({ fetch: async (url) => { + if (url === "/api/chat") return Response.json({ ok: true }); + if (url === "/api/chats/chat_x") return Response.json({ messages: record }); + if (url === "/api/chats/chat_x/stream") return sseResponse({ type: "turn-settled" }, "[DONE]"); + throw new Error(`unexpected fetch: ${url}`); + } }); + m.setReattachSettled(() => m.setMessages(record)); + await m.send({ type: "text", text: "topic" }); + assert.equal(m.getState().status, "ready"); + assert.equal(m.getState().error?.message, "Google 500: fixture failure"); + }); + + it("a newer live provider failure replaces an earlier attachment error immediately", async () => { + let attempts = 0; + const seen: Array<{ status: string; error: string | undefined }> = []; + const m = model({ fetch: async () => { + if (++attempts === 1) return new Response("earlier attach failure", { status: 500 }); + return sseResponse( + { type: "message_start", message: failed("new", "new provider failure") }, + { type: "message_end", message: failed("new", "new provider failure") }, + "[DONE]", + ); + } }); + await m.resumeStream(false); + assert.equal(m.getState().error?.message, "earlier attach failure"); + m.onChange = () => { seen.push({ status: m.status, error: m.getState().error?.message }); }; + await m.resumeStream(false); + assert.ok(seen.some((s) => s.status === "streaming" && s.error === "new provider failure"), "new failure wins before stream closure"); + assert.equal(m.getState().error?.message, "new provider failure"); + assert.equal(model({ initialMessages: m.getState().nativeMessages }).getState().error?.message, "new provider failure", "live and reopened views agree"); + }); + + it("an aborted older consume cannot restore the old banner during a replacement reattach", async () => { + let attempts = 0; + let firstRead!: () => void; + const reading = new Promise((resolve) => { firstRead = resolve; }); + let settleReplacement!: (response: Response) => void; + const replacement = new Promise((resolve) => { settleReplacement = resolve; }); + const m = model({ initialMessages: [failed()], fetch: async (_url, init) => { + if (++attempts > 1) return replacement; + return new Response(new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(sse({ type: "agent_start" }))); + init?.signal?.addEventListener("abort", () => controller.error(new DOMException("Aborted", "AbortError")), { once: true }); + }, + pull() { firstRead(); }, + })); + } }); + const first = m.resumeStream(false); + await reading; + const second = m.resumeStream(false); + await first; + assert.equal(m.getState().status, "submitted", "old cleanup must not settle the replacement attempt"); + assert.equal(m.getState().error, null, "old saved failure stays hidden during the replacement's Thinking wait"); + settleReplacement(sseResponse( + { type: "message_start", message: failed("new", "replacement failure") }, + { type: "message_end", message: failed("new", "replacement failure") }, + "[DONE]", + )); + await second; + assert.equal(m.getState().error?.message, "replacement failure"); + }); + + it("keeps loop/request errors transient, with the existing precedence and clearError behavior", async () => { + const m = model({ initialMessages: [failed()] }); + m.applyWireEvent({ type: "error", errorText: "loop failure" }); + assert.equal(m.getState().error?.message, "loop failure"); + assert.equal(model({ initialMessages: m.getState().nativeMessages }).getState().error?.message, "Google 500: fixture failure"); + m.clearError(); + assert.equal(m.getState().error?.message, "Google 500: fixture failure"); + m.fetch = async () => new Response("request refused", { status: 500 }); + await m.send({ type: "text", text: "retry" }); + assert.equal(m.getState().error?.message, "request refused"); + assert.equal(model({ initialMessages: m.getState().nativeMessages }).getState().error, null, "request errors are not fabricated as saved messages"); + m.clearError(); + assert.equal(m.getState().error, null); + }); +}); + // ─── reattach (resumeStream) ──────────────────────────────────────────────── describe("resumeStream (the reattach path)", () => {