sorokin.oleg.edu/learnharness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity

fix: recover session submissions and failed turn settlement

Browse Source
This commit is contained in:
amosblomqvist
2026-10-25 19:45:17 +02:00
parent b277ee3e0f
commit 7d65fa8754
21 changed files with 944 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
components/session-core.ts
+173 -28
View File
@@ -17,6 +17,8 @@ import type { AgentEvent } from "@earendil-works/pi-agent-core";
import type { ToolResultMessage } from "@earendil-works/pi-ai";
import {
needsContinuation,
danglingQuestionOf,
answeredCallIds,
measuredContextTokens,
newAnswerResult,
newNextMessage,
@@ -30,6 +32,7 @@ import {
type LhViewMessage,
type LiveExecution,
type QuestionAnswerData,
type QuestionDraft,
} from "../src/core";
export type SessionStatus = "ready" | "submitted" | "streaming" | "error";
@@ -78,13 +81,25 @@ async function turnInFlight(model: SessionModel): Promise<boolean | null> {
}
}
/** One wire event off the SSE body: parsed JSON, or the [DONE] terminal. */
async function* sseEvents(reader: ReadableStreamDefaultReader<Uint8Array>): AsyncGenerator<Record<string, unknown>> {
const STREAM_DROPPED = "The connection dropped before the turn completed.";
/** Mark transport failures without confusing JSON/application bugs with drops. */
class StreamReadError extends Error {
constructor(cause: unknown) {
super(cause instanceof Error ? cause.message : String(cause), { cause });
}
}
/** Wire events; return true ONLY for the explicit completion marker. */
async function* sseEvents(reader: ReadableStreamDefaultReader<Uint8Array>): AsyncGenerator<Record<string, unknown>, boolean> {
const decoder = new TextDecoder();
let buf = "";
for (;;) {
const { done, value } = await reader.read();
if (done) return;
let chunk: ReadableStreamReadResult<Uint8Array>;
try { chunk = await reader.read(); }
catch (err) { throw new StreamReadError(err); }
const { done, value } = chunk;
if (done) return false;
buf += decoder.decode(value, { stream: true });
for (;;) {
const end = buf.indexOf("\n\n");
@@ -94,7 +109,7 @@ async function* sseEvents(reader: ReadableStreamDefaultReader<Uint8Array>): Asyn
const dataLine = event.split("\n").find((l) => l.startsWith("data: "));
if (!dataLine) continue;
const data = dataLine.slice("data: ".length);
if (data === "[DONE]") return;
if (data === "[DONE]") return true;
yield JSON.parse(data) as Record<string, unknown>;
}
}
@@ -196,6 +211,8 @@ export function visibleQueuedSteers(
export interface SessionModelOptions extends SessionModelDeps {
chatId: string;
initialMessages: LhMessage[];
/** Retained, memory-only server failure when reopening a session. */
initialError?: string;
/** Fires on every state change (the hook re-renders). */
onChange?: () => void;
/** The reattach settled terminal (turn-settled): no turn running, the
@@ -220,6 +237,9 @@ export class SessionModel {
// at every full sync; the write-ahead absorbs the optimistic entries.
private base: LhMessage[];
private tail: LhMessage[] = [];
private questionDrafts = new Map<string, QuestionDraft>();
/** Only learner answers submitted through send(), never silent dismissals. */
private unconfirmedAnswers = new Map<string, string>();
// The live turn's execution layer (view model mid-turn states):
private executions = new Map<string, LiveExecution>();
private streamingIds = new Set<string>();
@@ -229,6 +249,8 @@ export class SessionModel {
private statusValue: SessionStatus = "ready";
/** Request/loop failures are transient; provider failures derive from messages. */
private errorValue: { message: string } | null = null;
/** Dismissing a banner must not authorize an automatic replacement turn. */
private automaticContinuationBlocked = false;
/** 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
@@ -240,6 +262,8 @@ export class SessionModel {
constructor(opts: SessionModelOptions) {
this.chatId = opts.chatId;
this.base = opts.initialMessages;
this.errorValue = opts.initialError === undefined ? null : { message: opts.initialError };
this.automaticContinuationBlocked = opts.initialError !== undefined;
// BOUND: the browser's native fetch is a Window method — storing it
// unbound makes every call throw "Illegal invocation" in the browser
// (Node's undici doesn't care, so the tests never saw it). Always go
@@ -300,6 +324,7 @@ export class SessionModel {
private setError(message: string | null): void {
this.errorValue = message === null ? null : { message };
if (message !== null) this.automaticContinuationBlocked = true;
this.onChange?.();
}
@@ -346,6 +371,12 @@ export class SessionModel {
}
const agent = event as unknown as AgentEvent;
switch (agent.type) {
case "agent_start":
// Positive evidence of a new/replayed server-owned turn. Any
// later failure in that replay blocks continuation again.
this.automaticContinuationBlocked = false;
if (this.errorValue !== null) this.setError(null);
return false;
case "message_start":
case "message_update":
case "message_end": {
@@ -445,18 +476,28 @@ 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, controller = this.consumeAbort): Promise<void> {
async consumeStream(res: Response, controller = this.consumeAbort, allowRecovery = true): Promise<void> {
if (!res.body) throw new Error("The turn stream body is empty.");
const reader = res.body.getReader();
let sawEvent = false;
let sawDone = false;
try {
for await (const event of sseEvents(reader)) {
const events = sseEvents(reader);
for (;;) {
let next: IteratorResult<Record<string, unknown>, boolean>;
try { next = await events.next(); }
catch (err) {
if (err instanceof StreamReadError) break;
throw err;
}
if (this.consumeAbort !== controller) return;
if (next.done) { sawDone = next.value; break; }
const event = next.value;
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(() => {});
if (this.consumeAbort === controller) this.onReattachSettled?.();
sawDone = true;
this.onReattachSettled?.();
return;
}
const acted = this.applyWireEvent(event);
@@ -466,14 +507,32 @@ export class SessionModel {
}
}
} finally {
await reader.cancel().catch(() => {});
reader.releaseLock();
// 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");
if (sawDone && this.statusValue !== "error") this.setStatus("ready");
}
}
if (this.consumeAbort !== controller || sawDone) return;
if (!allowRecovery) throw new StreamReadError(new Error(STREAM_DROPPED));
// One read-only recovery: refresh the canonical base, then replay the
// SAME server-owned turn. Never resend the learner's input.
await this.resumeStream(true, false);
}
private async fetchView(url: string, signal: AbortSignal, recoveringDrop: boolean): Promise<Response> {
try {
const res = await this.fetch(url, { signal });
if (recoveringDrop && !res.ok) throw new Error((await res.text()) || STREAM_DROPPED);
return res;
} catch (err) {
if (recoveringDrop && !signal.aborted) throw new StreamReadError(err);
throw err;
}
}
/**
@@ -489,7 +548,7 @@ export class SessionModel {
* lands on top. The stream then replays the turn from its first event
* through the SAME consume path.
*/
async resumeStream(refetchRecordFirst = true): Promise<void> {
async resumeStream(refetchRecordFirst = true, allowRecovery = true): Promise<void> {
this.consumeAbort?.abort();
const controller = new AbortController();
this.consumeAbort = controller;
@@ -519,16 +578,27 @@ export class SessionModel {
this.setStatus("submitted");
try {
if (refetchRecordFirst) {
const recordRes = await this.fetch(`/api/chats/${encodeURIComponent(this.chatId)}`, {
signal: controller.signal,
});
const recordRes = await this.fetchView(`/api/chats/${encodeURIComponent(this.chatId)}`, controller.signal, !allowRecovery);
if (this.consumeAbort !== controller) return;
if (recordRes.ok) {
const record = (await recordRes.json()) as { messages?: LhMessage[] };
let text: string;
try { text = await recordRes.text(); }
catch (err) {
if (!allowRecovery && !controller.signal.aborted) throw new StreamReadError(err);
throw err;
}
// Parse outside the transport catch: malformed JSON is a
// contract bug, not a connection-drop diagnostic.
const record = JSON.parse(text) as { messages?: LhMessage[]; turnError?: string };
if (this.consumeAbort !== controller) return;
if (Array.isArray(record.messages)) {
this.mutate(() => {
this.base = record.messages!;
if (record.turnError !== undefined) {
this.errorValue = { message: record.turnError };
this.automaticContinuationBlocked = true;
}
this.forgetResolvedDrafts(record.messages!);
this.incompleteUsageIds.clear();
this.tail = [];
});
@@ -543,15 +613,19 @@ export class SessionModel {
});
}
}
const res = await this.fetch(`/api/chats/${encodeURIComponent(this.chatId)}/stream`, {
signal: controller.signal,
});
const res = await this.fetchView(`/api/chats/${encodeURIComponent(this.chatId)}/stream`, controller.signal, !allowRecovery);
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, controller);
await this.consumeStream(res, controller, allowRecovery);
} catch (err) {
if (this.consumeAbort !== controller || (err as Error).name === "AbortError") return; // a newer consume won
this.setError((err as Error).message);
if (this.consumeAbort !== controller || controller.signal.aborted) return; // intentionally cancelled
// A failed record read can leave the optimistic answer in the tail.
// Restore its Submit affordance; the next attempt reconciles first.
this.mutate(() => {
this.tail = this.tail.filter(m => !this.unconfirmedAnswers.has(m.id));
this.unconfirmedAnswers.clear();
});
this.setError(!allowRecovery && err instanceof StreamReadError ? STREAM_DROPPED : (err as Error).message);
this.setStatus("error");
}
}
@@ -563,12 +637,42 @@ export class SessionModel {
* server's 409 + identity retry covers the wake-race window).
*/
async send(input: ChatInput): Promise<void> {
if (!this.readyRef) return;
const recovering = this.statusValue === "error";
if (!this.readyRef && !recovering) return;
this.consumeAbort?.abort();
const controller = new AbortController();
this.consumeAbort = controller;
this.setError(null);
this.setStatus("submitted");
if (recovering && input.type === "answer") {
try {
const res = await this.fetch(`/api/chats/${encodeURIComponent(this.chatId)}`, { signal: controller.signal });
if (this.consumeAbort !== controller) return;
if (!res.ok) throw new Error((await res.text()) || "An error occurred.");
const record = await res.json() as { messages?: LhMessage[]; turnInFlight?: boolean; turnError?: string };
if (this.consumeAbort !== controller) return;
// No verified record/liveness: retain the draft and never guess
// whether a lost acknowledgement meant the input was accepted.
if (!Array.isArray(record.messages) || typeof record.turnInFlight !== "boolean") throw new Error("An error occurred.");
this.setMessages(record.messages, record.turnError);
if (record.turnInFlight) {
await this.resumeStream(false);
return;
}
if (input.type === "answer" && (answeredCallIds(record.messages).has(input.toolCallId) || danglingQuestionOf(record.messages)?.toolCallId !== input.toolCallId)) {
// The call already resolved, or no longer owns the dock. Adopt
// truth, never POST that answer a second time.
this.setStatus("ready");
return;
}
} catch (err) {
if (this.consumeAbort !== controller || controller.signal.aborted) return;
this.setError((err as Error).message);
this.setStatus("error");
return;
}
}
if (input.type !== "continue") this.automaticContinuationBlocked = false;
// The optimistic appends — exactly the messages the server's
// write-ahead will persist (inputAppends), with LOCAL ids so
// adoption can tell them from the record's absorbed copies.
@@ -585,7 +689,9 @@ export class SessionModel {
optimistic.push({ ...newNextMessage(), id: localId() });
} else if (input.type === "answer") {
const tool = pendingCallTool([...this.base, ...this.tail], input.toolCallId) ?? "quiz";
optimistic.push({ ...newAnswerResult(input.toolCallId, tool, input.answer), id: localId() });
const result = { ...newAnswerResult(input.toolCallId, tool, input.answer), id: localId() };
optimistic.push(result);
this.unconfirmedAnswers.set(result.id, input.toolCallId);
}
if (optimistic.length > 0) this.mutate(() => {
this.tail = [...this.tail, ...optimistic];
@@ -605,9 +711,7 @@ export class SessionModel {
// connection is Diet A's whole job.
const res = await this.postInputEvent(input);
if (res.status === 409) {
this.setError("A turn is already running — try again in a moment.");
this.setStatus("error");
return;
throw new Error("A turn is already running — try again in a moment.");
}
if (!res.ok) {
// Refusal envelopes (503 BYOK/key gates) and 400s: the RAW body
@@ -629,7 +733,17 @@ export class SessionModel {
if (this.consumeAbort !== controller) return;
await this.resumeStream(true);
} catch (err) {
if ((err as Error).name === "AbortError") return; // stop() or a newer consume won
if (this.consumeAbort !== controller || controller.signal.aborted) return;
// Restore the question when acceptance failed or is unknown. Remove
// only THIS optimistic answer, never saved copies/newer live content.
// Other input kinds retain their existing optimistic presentation.
if (input.type === "answer") {
const ids = new Set(optimistic.map(m => m.id));
this.mutate(() => {
this.tail = this.tail.filter(m => !ids.has(m.id));
for (const id of ids) this.unconfirmedAnswers.delete(id);
});
}
this.setError((err as Error).message);
this.setStatus("error");
}
@@ -740,10 +854,32 @@ export class SessionModel {
this.setStatus("ready");
}
/** Submission snapshots outlive the dock, not the session mount. */
questionDraft(toolCallId: string): QuestionDraft | undefined {
return this.questionDrafts.get(toolCallId);
}
saveQuestionDraft(toolCallId: string, draft: Partial<QuestionDraft>): void {
this.questionDrafts.set(toolCallId, { ...this.questionDrafts.get(toolCallId), ...draft });
}
private forgetResolvedDrafts(record: readonly LhMessage[]): void {
const answered = answeredCallIds(record);
for (const id of answered) this.questionDrafts.delete(id);
for (const [id, callId] of this.unconfirmedAnswers) {
if (answered.has(callId)) this.unconfirmedAnswers.delete(id);
}
}
/** Full snapshot swap (the idle settle sync — server canon). */
setMessages(record: LhMessage[]): void {
setMessages(record: LhMessage[], turnError?: string): void {
this.mutate(() => {
this.base = record;
if (turnError !== undefined) {
this.errorValue = { message: turnError };
this.automaticContinuationBlocked = true;
}
this.forgetResolvedDrafts(record);
this.incompleteUsageIds.clear();
this.tail = [];
this.streamingMessageId = null; // the tail (and its slot) is gone
@@ -763,10 +899,19 @@ export class SessionModel {
if (baseAbsorbed) return; // not a pure append — the idle full sync is the net
this.mutate(() => {
this.base = record;
this.forgetResolvedDrafts(record);
this.tail = this.tail.filter((m) => !absorbed(m, record));
});
}
/** Automatic mount/sync continuation is never permission to retry a
* failed turn. Read current model truth, not a stale React closure. */
async continueIfOwed(record: readonly LhMessage[]): Promise<void> {
if (this.statusValue === "ready" && !this.automaticContinuationBlocked && this.getState().error === null && needsContinuation(record)) {
await this.send({ type: "continue" });
}
}
/** The mount/sync continuation fold (native-messages.ts). */
static needsContinuation = needsContinuation;