fix: keep current provider failures authoritative across reattach
This commit is contained in:
@@ -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<void> {
|
||||
async consumeStream(res: Response, controller = this.consumeAbort): Promise<void> {
|
||||
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");
|
||||
}
|
||||
|
||||
@@ -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<void>((resolve) => { firstRead = resolve; });
|
||||
let settleReplacement!: (response: Response) => void;
|
||||
const replacement = new Promise<Response>((resolve) => { settleReplacement = resolve; });
|
||||
const m = model({ initialMessages: [failed()], fetch: async (_url, init) => {
|
||||
if (++attempts > 1) return replacement;
|
||||
return new Response(new ReadableStream<Uint8Array>({
|
||||
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" });
|
||||
|
||||
Reference in New Issue
Block a user