From 5cf4e589be25165d550e2a6012cc8eb481e6fa74 Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Sun, 25 Oct 2026 00:00:29 +0300 Subject: [PATCH] fix: keep current provider failures authoritative across reattach --- components/session-core.ts | 29 ++++++++++++++++------ tests/session-core.test.ts | 51 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 8 deletions(-) diff --git a/components/session-core.ts b/components/session-core.ts index c0d7247..657d0da 100644 --- a/components/session-core.ts +++ b/components/session-core.ts @@ -385,7 +385,12 @@ export class SessionModel { const withId = { ...message, id } as LhMessage; if (agent.type === "message_end") { this.incompleteUsageIds.delete(id); - if (message.role === "assistant" && message.stopReason === "error") this.liveFailureId = 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; @@ -440,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); @@ -460,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"); + } } } @@ -512,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!; @@ -534,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 631769f..7e5ecd7 100644 --- a/tests/session-core.test.ts +++ b/tests/session-core.test.ts @@ -617,6 +617,57 @@ describe("provider-error banner (live and saved)", () => { 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" });