Merge current main into PR61 finishing branch

src/server/chat-handler.ts

@@ -17,7 +17,7 @@ import { createExtractionPipeline } from "../agent/subagents/extraction.ts";
 import type { ModelRole } from "../../config.ts";
 import {
 	deliverJobResult,
-	noteSteerVisible,
+	noteSteersDurable,
 	reconcileUserSteers,
 	registerLiveAgent,
 	restoreUncommittedAutomatic,
@@ -42,8 +42,8 @@ import {
 import { modelsNotConfigured, providerKeyMissing as providerKeyMissingEnvelope, type ProviderKeyMissing } from "../core/settings.ts";
 import { beginTurn, endTurn } from "./turn-liveness.ts";
 import { applyParkedDismissal } from "./dismissal-park.ts";
-import { registerTurn, unregisterTurn } from "./turn-runs.ts";
-import { persistTurn, triggerExtraction } from "./turn-service.ts";
+import { registerTurn, rememberTurnFailure, unregisterTurn } from "./turn-runs.ts";
+import { persistTurn, triggerExtraction, TurnPersistenceError } from "./turn-service.ts";
 import { hasProbeDone, danglingQuestionOf } from "../core/native-messages.ts";
 import { composeInstructions } from "../agent/teacher-agent.ts";
 import { ensureCompacted, kickCompaction, effectiveThresholdTokens } from "./compaction.ts";
@@ -55,10 +55,11 @@ import {
 	ChatInputError,
 	inputAppends,
 	parseChatInput,
+	retryDrops,
 	type TurnInput,
 } from "../core/input-events.ts";
 import { steerDelivered } from "../core/contracts.ts";
-import type {CallUsage, SteerMessage} from "../core/contracts.ts";
+import type {CallUsage} from "../core/contracts.ts";
 
 /**
  * One chat turn over the server-owned native record. Framework-free (plain
@@ -79,6 +80,10 @@ import type {CallUsage, SteerMessage} from "../core/contracts.ts";
  * for write-ahead persistence, turn liveness, and the turn services.
  */
 
+function settlementErrorText(error: unknown): string {
+	return error instanceof TurnPersistenceError ? error.message : "An error occurred.";
+}
+
 const jobRunner = createJobRunner({
 	chatStore: stores.chatStore,
 	jobStore: stores.jobStore,
@@ -182,6 +187,17 @@ export async function handleChatInput(
 		return Response.json({ error: "Unknown chat" }, { status: 404 });
 	}
 
+	// Refuse an ineligible Retry before catalog refresh, compaction, or any
+	// turn/write effects. Revalidate after the awaited gates below too.
+	if (input.type === "retry") {
+		try {
+			retryDrops(record.messages);
+		} catch (err) {
+			if (err instanceof ChatInputError) return Response.json({ error: err.message }, { status: 400 });
+			throw err;
+		}
+	}
+
 	// THE UNIVERSAL TURN GATE (BYOK, owner, 2026-09-22): a turn needs the
 	// CONVERSATION chain — the chat's own model override over the strong
 	// slot — and there is no factory floor; when both are empty the turn
@@ -255,7 +271,20 @@ export async function handleChatInput(
 		chatId,
 	);
 	const compactedRecord = stores.chatStore.getChat(chatId);
-	if (compactedRecord) record = compactedRecord;
+	if (!compactedRecord) return Response.json({ error: "Unknown chat" }, { status: 404 });
+	record = compactedRecord;
+	// Recompute Retry eligibility on current truth, still before beginTurn.
+	// A compaction marker or another writer may have landed during awaits.
+	// These omissions affect ONLY the agent context, never the saved record.
+	let retryOmissions = new Set<string>();
+	if (input.type === "retry") {
+		try {
+			retryOmissions = retryDrops(record.messages);
+		} catch (err) {
+			if (err instanceof ChatInputError) return Response.json({ error: err.message }, { status: 400 });
+			throw err;
+		}
+	}
 
 	// The input's native appends (pure; the honest 400 for an answer with no
 	// pending question call). Computed BEFORE beginTurn so a refused input
@@ -289,7 +318,11 @@ export async function handleChatInput(
 	// the composer gates), so adjacency holds by construction.
 	if (input.type === "answer" && danglingQuestionOf([...record.messages, ...appends]) !== null) {
 		const merged = [...record.messages, ...appends];
-		stores.chatStore.saveMessages(chatId, merged);
+		try { stores.chatStore.saveMessages(chatId, merged); }
+		catch (err) {
+			console.error(`[chat ${chatId}] write-ahead persist failed:`, err);
+			return new Response("Your input could not be saved. Please try again.", { status: 500 });
+		}
 		emit("record-changed", { chatId });
 		return settledTurnResponse(); // the cheap terminal — no beginTurn, no turn
 	}
@@ -309,18 +342,26 @@ export async function handleChatInput(
 	// ride this same write-ahead - the pump de-queued them at fire, so this
 	// is the ONLY copy ever appended.)
 	const merged = [...record.messages, ...appends];
+	// Retry preserves every persisted attempt. Pi needs a non-assistant
+	// tail to start continue(), so its temporary history omits the trailing
+	// failures only. Count that SAME temporary prefix at settlement, or the
+	// retained failures would make persistTurn skip genuinely new output.
+	const agentMessages = input.type === "retry" ? merged.filter((m) => !retryOmissions.has(m.id)) : merged;
 	const stagedAutomatic = isSubscriptionProvider(conversation.provider) && input.type === "steers";
-	let persistBaseCount = stagedAutomatic ? record.messages.length : merged.length;
+	let persistBaseCount = stagedAutomatic ? record.messages.length : agentMessages.length;
 	try {
 		if (!stagedAutomatic) stores.chatStore.saveMessages(chatId, merged);
-		// Dock fix (owner ruling 2026-09-26): persisted markers are visible
-		// in the record — flip their jobs now (the wake's steers ride this
-		// write-ahead; the delivering turn may run for minutes).
-		for (const appended of stagedAutomatic ? [] : appends) {
-			if ((appended as { role?: string }).role === "custom") noteSteerVisible(appended as SteerMessage);
-		}
 	} catch (err) {
 		console.error(`[chat ${chatId}] write-ahead persist failed:`, err);
+		endTurn(chatId);
+		return new Response("Your input could not be saved. Please try again.", { status: 500 });
+	}
+	// Visibility bookkeeping is not the canonical save. Its failure must
+	// neither misreport a saved input as lost nor prevent the turn running.
+	try {
+		if (!stagedAutomatic) noteSteersDurable(chatId, merged);
+	} catch (err) {
+		console.error(`[chat ${chatId}] input visibility bookkeeping failed:`, err);
 	}
 
 	// MONEY CHANNEL (wave 4): one collector per request, fed by the agent's
@@ -346,7 +387,7 @@ export async function handleChatInput(
 	// the session is still probing: build the PROBER (quiz/ask + probe_done).
 	// Both share this whole composition (write-ahead persist, liveness,
 	// steers, usage, extraction) — only the agent differs.
-	const teaching = hasProbeDone(merged);
+	const teaching = hasProbeDone(agentMessages);
 	// BYOK (owner, 2026-09-22): the conversation chain was resolved by the
 	// turn gate above (chat.model override → strong slot → refused) — the
 	// engine build is pure from here; thinking levels clamp natively.
@@ -365,19 +406,23 @@ export async function handleChatInput(
 	// sub-agent JOB tools (research/demo/visual/pdf; adjudicated with the
 	// user: quiz/ask/fs/plan/probe_done are always-on).
 	if (isSubscriptionProvider(conversation.provider)) requestCycle = new SubscriptionRequestCycle({
-		initialMessages: asAgentMessages(merged),
+		initialMessages: asAgentMessages(agentMessages),
 		staged: stagedAutomatic ? asAgentMessages(appends) : undefined,
 		foregroundOwed: !callPurpose.background,
 		purpose: callPurpose,
 		scopeCurrent: (scope) => connectionScopeCurrent(conversation.provider, scope),
-		commitStaged: () => { stores.chatStore.saveMessages(chatId, merged); persistBaseCount = merged.length; },
+		commitStaged: () => {
+			stores.chatStore.saveMessages(chatId, merged);
+			persistBaseCount = merged.length;
+			noteSteersDurable(chatId, merged);
+		},
 		restoreAutomatic: (automaticIds, retainedIds) => restoreUncommittedAutomatic(chatId, constructed, automaticIds, retainedIds),
 	});
 	const agentOpts: AgentBuildOptions = {
 		engine,
 		callPurpose,
 		requestCycle,
-		messages: asAgentMessages(merged),
+		messages: asAgentMessages(agentMessages),
 		onUsage: collectUsage,
 		toolToggles: effective.agent.tools,
 		// Debug (owner, 2026-10-04): teacher/prober raw provider events under
@@ -468,11 +513,12 @@ export async function handleChatInput(
 		// run's first LLM call (the native replacement for continuation
 		// coalescing). Settlement still cannot beat the flow below: it needs
 		// a streamed response, which never happens inside the constructor.
-		registerLiveAgent(chatId, constructed);
+		registerLiveAgent(chatId, constructed, input.type !== "steers" && input.type !== "continue",
+			input.type === "steers" ? new Set(appends.map((message) => message.id)) : undefined);
 		// Debug recorder run marker (owner, 2026-10-04): correlates the
 		// JSONL event stream with the record's turns; no-op unless
 		// LH_PROVIDER_EVENTS=1.
-		providerEventNote(chatId, "turn", `run start (base: ${merged.length} messages)`);
+		providerEventNote(chatId, "turn", `run start (base: ${agentMessages.length} messages)`);
 		const handle = runPiTurn({
 			agent: constructed,
 			requestCycle,
@@ -496,10 +542,26 @@ export async function handleChatInput(
 				console.error(`[chat ${chatId}] turn error:`, error);
 				return "An error occurred.";
 			},
+			onSettleError: settlementErrorText,
+			onSettleFailure: (error, reportedErrorText) => {
+				// POST only acknowledges acceptance; the submitter may attach
+				// after this turn has already failed and left the registry.
+				rememberTurnFailure(chatId, reportedErrorText ?? settlementErrorText(error));
+				endTurn(chatId);
+				unregisterTurn(chatId);
+				unregisterLiveAgent(chatId);
+				// The canonical save may have succeeded before later bookkeeping
+				// threw. Reconcile against actual durability, never replay input
+				// merely because usage/extraction/other settlement work failed.
+				const persisted = stores.chatStore.getChat(chatId)?.messages ?? [];
+				reconcileUserSteers(chatId, persisted);
+				settleSteers(chatId, persisted, true, constructed);
+			},
 			onFinish: (turnMessages, isAborted) => {
-				// Wave B: the record write is the agent's own final state —
-				// the new messages are the slice past the merged base.
+				// Wave B: append only messages past the temporary agent prefix.
+				// Retry's saved record still includes its earlier failed attempts.
 				const stamped = persistTurn(stores.chatStore, chatId, persistBaseCount, turnMessages, isAborted);
+				noteSteersDurable(chatId, stores.chatStore.getChat(chatId)?.messages ?? stamped);
 				// usage log append (idempotent per turn id) — the stats bar
 				// refolds this on the client after the turn settles. The
 				// provenance message is the turn's FIRST new assistant message.
@@ -545,7 +607,7 @@ export async function handleChatInput(
 				// resolve; stranded markers return to pending), then let
 				// the pump take any stragglers down the idle wake.
 				unregisterLiveAgent(chatId);
-				settleSteers(chatId, stamped);
+				settleSteers(chatId, stamped, false, constructed);
 				// Composer steers (2026-10-03): delivered ids dropped, stranded
 				// ones stay parked for the chat's next run (the registry is the
 				// stranding safety net — the agent's undrained copy died above).

src/server/chats.ts

@@ -1,10 +1,13 @@
 import { abortExtraction, extractionStatuses } from "../agent/subagents/extraction.ts";
+import { isJobLive } from "../agent/subagents/aborts.ts";
 import { coverageOf } from "../core/coverage.ts";
 import { deleteChatTraces, type ChatDeleteMode, type ChatDeletionStats } from "../core/graph-deletion.ts";
 import { titleFromPrompt } from "../core/contracts.ts";
 import { graphRevision } from "./graph.ts";
 import { stores } from "./stores.ts";
 import { isTurnRunning } from "./turn-liveness.ts";
+import { clearTurnFailure } from "./turn-runs.ts";
+import { discardChatDeliveries } from "./jobs.ts";
 
 /**
  * Server-side chat entries for the transport layer (app/api/chats*,
@@ -65,10 +68,7 @@ export type ChatDeleteOutcome =
  *      the wind-down finds the chat gone at its merge and aborts with
  *      nothing merged.
  */
-export async function deleteChatSession(chatId: string, mode: ChatDeleteMode): Promise<ChatDeleteOutcome> {
-	if (!stores.chatStore.getChat(chatId)) {
-		return { ok: false, status: 404, code: "not_found", error: "No such session" };
-	}
+function runningWorkBlocker(chatId: string): Extract<ChatDeleteOutcome, { status: 409 }> | null {
 	if (isTurnRunning(chatId)) {
 		return {
 			ok: false,
@@ -78,7 +78,11 @@ export async function deleteChatSession(chatId: string, mode: ChatDeleteMode): P
 			jobIds: [],
 		};
 	}
-	const runningJobs = stores.jobStore.listJobs(chatId).filter((job) => job.status === "running");
+	// Legacy records kept completed workers running until delivery. A saved
+	// result without a live loop is pending session data, not computing work;
+	// deletion must not depend on a prior recovery poll to recognize it.
+	const runningJobs = stores.jobStore.listJobs(chatId).filter((job) =>
+		job.status === "running" && (!job.result || isJobLive(job.id)));
 	if (runningJobs.length > 0) {
 		return {
 			ok: false,
@@ -88,6 +92,15 @@ export async function deleteChatSession(chatId: string, mode: ChatDeleteMode): P
 			jobIds: runningJobs.map((job) => job.id),
 		};
 	}
+	return null;
+}
+
+export async function deleteChatSession(chatId: string, mode: ChatDeleteMode): Promise<ChatDeleteOutcome> {
+	if (!stores.chatStore.getChat(chatId)) {
+		return { ok: false, status: 404, code: "not_found", error: "No such session" };
+	}
+	const blocker = runningWorkBlocker(chatId);
+	if (blocker) return blocker;
 	if (extractionStatuses().some((run) => run.chatId === chatId)) {
 		abortExtraction(chatId); // true cancellation — spend stops, nothing merges
 		const deadline = Date.now() + WIND_DOWN_TIMEOUT_MS;
@@ -104,11 +117,19 @@ export async function deleteChatSession(chatId: string, mode: ChatDeleteMode): P
 		}
 	}
 
+	// A result wake or a worker may have started during extraction wind-down.
+	// Recheck immediately before the synchronous purge, with no await gap.
+	const lateBlocker = runningWorkBlocker(chatId);
+	if (lateBlocker) return lateBlocker;
+
 	// THE SYNC BLOCK — no awaits below (see the doc comment). Order:
 	// ledgers → jobs → the chat record.
 	const stats = deleteChatTraces(stores.graphStore, chatId, mode);
 	const jobsRemoved = stores.jobStore.deleteJobsForChat(chatId);
-	stores.chatStore.deleteChat(chatId);
+	if (stores.chatStore.deleteChat(chatId)) {
+		discardChatDeliveries(chatId);
+		clearTurnFailure(chatId);
+	}
 	return { ok: true, mode, stats, jobsRemoved, graphRevision: graphRevision() };
 }
 

src/server/compaction.ts

@@ -69,7 +69,7 @@ async function subagentSummarizer(prompt: string, engine: PiEngine): Promise<{ s
 		totalTokens: 0,
 		cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
 	};
-	const { text } = await runSubagent([], {
+	const { text, newMessages } = await runSubagent([], {
 		// The engine override is always present (the chat's own conversation
 		// engine — owner ruling), so the role label never resolves an
 		// engine; "researcher" is the least-wrong existing ModelRole label
@@ -83,6 +83,12 @@ async function subagentSummarizer(prompt: string, engine: PiEngine): Promise<{ s
 			usage = u;
 		},
 	});
+	// A provider-declared incomplete response is not a usable checkpoint,
+	// even when it contains nonempty text. Match Pi's compaction failure
+	// semantics: leave the existing history and prior summary untouched.
+	if (newMessages.some((message) => message.role === "assistant" && message.stopReason === "length")) {
+		throw new Error("Compaction summary hit the output limit and is incomplete");
+	}
 	return { summary: text, usage };
 }
 
@@ -201,7 +207,14 @@ async function compact(deps: CompactionDeps, chatId: string): Promise<void> {
 		spanStart = prevKept;
 		priorSummary = previous.content;
 	}
-	const spanText = serializeForSummary(messages.slice(spanStart, cut));
+	const span = messages.slice(spanStart, cut);
+	// The separately supplied prior summary replaces all summary markers
+	// in this span. Without a resolved prior boundary, retain the existing
+	// legacy fallback rather than silently discarding summary-only context.
+	const summaryMessages = priorSummary
+		? span.filter((m) => m.role !== "custom" || m.customType !== "compaction")
+		: span;
+	const spanText = serializeForSummary(summaryMessages);
 	const fullSpan = priorSummary ? `[SYSTEM NOTE — prior compaction summary, still authoritative where not superseded]\n${priorSummary}\n\n${spanText}` : spanText;
 
 	const prompt = buildCompactionPrompt({

src/server/index.ts

@@ -43,4 +43,5 @@ export * from "./compaction.ts";
 export { runPiTurn, piTurnResponse, type PiTurnHandle, type PiTurnSpec } from "./pi/ui-stream.ts";
 export * from "./observability.ts";
 export * from "./model-catalog.ts";
+export { subscriptionControlsResponse, subscriptionBrowserResponse } from "./web-subscription-http.ts";
 export { sessionCost } from "./pricing.ts";

src/server/jobs.ts

@@ -1,7 +1,7 @@
 /**
  * THE WAKE PIPELINE (contracts.ts "THE ASYNC PROTOCOL" §4) — server-side
  * steer delivery for finished sub-agent jobs. Wave C (2026-09-26):
- * delivery rides pi's OWN follow-up queue when a run is live, and the
+ * delivery rides pi's OWN steering queue when a run is live, and the
  * idle wake otherwise. One registry, two vehicles.
  *
  *   deliverJobResult(steer) — the runner calls this on job completion.
@@ -40,15 +40,17 @@
  *
  *   settleSteers(chatId, persisted) — the handler calls this at turn
  *     settlement (after unregisterLiveAgent): entries the run DELIVERED
- *     resolve + flip their jobs; entries that were handed to the dying
+ *     resolve + acknowledge delivery; entries handed to the dying
  *     run but never injected (stopper exit, abort, a missed final poll)
  *     return to pending and the pump restarts — no stranding, no loss.
  *
- *   Crash recovery (recoverStaleJobs): a process restart can orphan jobs —
- *     "running" with a result (the wake never fired) or without one (the
- *     sub-agent loop itself died). The sweep re-delivers finished results
- *     (idempotently) and re-spawns unfinished loops, both via injected
- *     seams. It runs lazily off the /api/jobs poll, throttled.
+ *   Worker status ends at computation completion; JobRecord.delivery owns
+ *     the pending result or abort notice independently. Registry entries
+ *     survive in-flight wakes, so duplicate submissions share one promise.
+ *     Failed starts stay pending without timers or worker errors. Recovery
+ *     restores known pending deliveries and re-spawns unfinished workers;
+ *     registration also restores pending delivery before the first poll.
+ *     Legacy terminal jobs without delivery metadata are never replayed.
  *
  * All mutable state lives on globalThis (the turn-liveness lesson: Next dev
  * gives each route entry its own module instances — a module-level Map would
@@ -83,6 +85,14 @@ export type JobRespawn = (jobId: string) => void;
 
 interface QueuedSteer {
 	steer: SteerMessage;
+	promise: Promise<void>;
+	/** Wake cargo remains owned here, but must not steer into its own run. */
+	waking?: boolean;
+	/** The constructed vehicle, once startup claims a turn. Until then only
+	 * the pump owns waking: an unrelated foreground settlement cannot free it. */
+	wakeAgent?: Agent;
+	/** A failed start waits for a learner run or process restart, not a timer. */
+	deferred?: boolean;
 	resolve: () => void;
 	reject: (reason: Error) => void;
 	/** The live agent this marker was steered into (agent.steer) — set by
@@ -166,18 +176,37 @@ function jobIdOf(steer: SteerMessage): string | null {
 	return typeof details?.jobId === "string" ? details.jobId : null;
 }
 
-/** Dock bookkeeping: the marker reached the learner's view — flip its job
- *  NOW instead of waiting for the delivering turn to settle. Called from
- *  BOTH visibility seams: the marker's message_end wire event (live steer
- *  injection — the card is on screen while the teacher keeps teaching) and
- *  the write-ahead persist in handleChatInput (the wake's input appends —
- *  the card is durable in the record). The registry's delivery promise
- *  still settles via settleSteers / the runner-200 path; finalizeJob is
- *  idempotent, so the later flip is a no-op. Owner ruling (2026-09-26): a
- *  delivered subagent leaves the background dock the moment its result is
- *  visible, not when the teacher's turn ends. */
-export function noteSteerVisible(steer: SteerMessage): void {
-	finalizeSteer(steer);
+/** Call only after canonical conversation persistence. Streaming visibility
+ * is not durable acceptance. The saved marker wins over delivery-bookkeeping
+ * failures; acknowledging delivery never changes the worker's outcome. */
+export function noteSteersDurable(chatId: string, persisted: readonly LhMessage[]): void {
+	for (const message of persisted) {
+		if (message.role === "custom") finalizeSteer(message);
+	}
+	for (const entry of [...(state().queues.get(chatId) ?? [])]) {
+		if (persisted.some((message) => message.id === entry.steer.id)) completeEntry(chatId, entry);
+	}
+}
+
+function removeEntry(chatId: string, entry: QueuedSteer): void {
+	const remaining = (state().queues.get(chatId) ?? []).filter((candidate) => candidate !== entry);
+	if (remaining.length > 0) state().queues.set(chatId, remaining);
+	else state().queues.delete(chatId);
+}
+
+function completeEntry(chatId: string, entry: QueuedSteer): void {
+	finalizeSteer(entry.steer);
+	removeEntry(chatId, entry);
+	entry.resolve();
+}
+
+/** Deletion discards the session's obligations, never wakes a deleted chat. */
+export function discardChatDeliveries(chatId: string): void {
+	const s = state();
+	const queue = s.queues.get(chatId) ?? [];
+	s.queues.delete(chatId);
+	s.userSteers.delete(chatId);
+	for (const entry of queue) entry.resolve();
 }
 
 /** Finalize a delivered steer's job bookkeeping — job steers only. */
@@ -186,24 +215,30 @@ function finalizeSteer(steer: SteerMessage): void {
 	if (jobId) finalizeJob(jobId);
 }
 
-/** Fail a steer's job on unrecoverable delivery failure — job steers only. */
-function failSteer(steer: SteerMessage, err: Error): void {
-	const jobId = jobIdOf(steer);
-	if (jobId) failJob(jobId, err);
+/** Queue one logical delivery per ID, including in-flight wake cargo.
+ * Job promises resolve on durable marker acceptance (or session deletion);
+ * failed starts stay pending. Non-job probe handoffs keep their existing
+ * wake-acceptance/error semantics. Workers never await this promise. */
+export function deliverJobResult(steer: SteerMessage): Promise<void> {
+	const chatId = chatIdOf(steer);
+	const entry = queueSteer(chatId, steer);
+	void pump(chatId);
+	return entry.promise;
 }
 
-/** Queue ANY steer (result, abort, probe handoff) through the pipeline.
- *  Resolves when the wake turn has completed; rejects when delivery fails
- *  unrecoverably. Never rejects for queueing itself. */
-export function deliverJobResult(steer: SteerMessage): Promise<void> {
-	return new Promise<void>((resolve, reject) => {
-		const s = state();
-		const chatId = chatIdOf(steer);
-		const queue = s.queues.get(chatId) ?? [];
-		queue.push({ steer, resolve, reject });
-		s.queues.set(chatId, queue);
-		void pump(chatId);
-	});
+function queueSteer(chatId: string, steer: SteerMessage): QueuedSteer {
+	const queue = state().queues.get(chatId) ?? [];
+	const existing = queue.find((entry) => entry.steer.id === steer.id);
+	if (existing) return existing;
+	let resolve!: () => void;
+	let reject!: (error: Error) => void;
+	const promise = new Promise<void>((res, rej) => { resolve = res; reject = rej; });
+	// Recovery/registration may own an entry with no waiting caller.
+	void promise.catch(() => undefined);
+	const entry: QueuedSteer = { steer, promise, resolve, reject };
+	queue.push(entry);
+	state().queues.set(chatId, queue);
+	return entry;
 }
 
 /** Hand every eligible pending marker to a live agent's steering queue
@@ -227,7 +262,7 @@ function handOffEligible(chatId: string, agent: Agent): void {
 	// remain deliverable, including the first poll of an answer run.
 	if (automaticFollowUpAllowed(chatId, agent) && queue && queue.length > 0) {
 		for (const entry of queue) {
-			if (entry.handedTo !== agent && entry.steer.customType !== "probe-handoff") {
+			if (!entry.waking && !entry.deferred && entry.handedTo !== agent && entry.steer.customType !== "probe-handoff") {
 				agent.steer(entry.steer);
 				entry.handedTo = agent;
 			}
@@ -254,8 +289,13 @@ function handOffEligible(chatId: string, agent: Agent): void {
  *  markers are steered into the run right here. The settle seam cannot
  *  beat the registration: settlement needs a streamed response, which
  *  never happens inside the constructor call. */
-export function registerLiveAgent(chatId: string, agent: Agent): void {
+export function registerLiveAgent(chatId: string, agent: Agent, allowDeferred = true, wakeSteerIds?: ReadonlySet<string>): void {
 	state().liveAgents.set(chatId, agent);
+	restorePendingDeliveries(chatId);
+	for (const entry of state().queues.get(chatId) ?? []) {
+		if (allowDeferred) entry.deferred = false;
+		if (entry.waking && wakeSteerIds?.has(entry.steer.id)) entry.wakeAgent = agent;
+	}
 	handOffEligible(chatId, agent);
 }
 
@@ -326,7 +366,7 @@ export function restoreUncommittedAutomatic(chatId: string, agent: Agent, automa
  * unregisterLiveAgent, with the run's persisted messages. Two jobs:
  *
  *   - entries whose marker the run DELIVERED (in-run follow-up injection —
- *     their ids are in the persisted slice) resolve and flip their jobs;
+ *     their ids are in the persisted slice) resolve and acknowledge delivery;
  *   - entries handed to the dying run but never injected (the question
  *     stopper exited before the follow-up poll; an abort; a delivery that
  *     raced the final poll) return to PENDING (handedTo cleared) and the
@@ -335,7 +375,7 @@ export function restoreUncommittedAutomatic(chatId: string, agent: Agent, automa
  * This is what makes the live hand-off safe: pi owns no retry for a marker
  * stranded in a settled agent's queue, so the registry stays the truth and
  * the dead agent's copy is simply abandoned (GC'd with the agent). */
-export function settleSteers(chatId: string, persisted: { id?: string }[]): void {
+export function settleSteers(chatId: string, persisted: { id?: string }[], failed = false, settledAgent?: Agent): void {
 	const queue = state().queues.get(chatId);
 	if (!queue || queue.length === 0) return;
 	const deliveredIds = new Set(persisted.map((m) => m.id).filter(Boolean));
@@ -345,7 +385,11 @@ export function settleSteers(chatId: string, persisted: { id?: string }[]): void
 			finalizeSteer(entry.steer);
 			entry.resolve();
 		} else {
-			entry.handedTo = undefined;
+			const handedHere = entry.handedTo !== undefined && (!settledAgent || entry.handedTo === settledAgent);
+			const wakeHere = settledAgent !== undefined && entry.wakeAgent === settledAgent;
+			if (handedHere) entry.handedTo = undefined;
+			if (wakeHere) { entry.waking = false; entry.wakeAgent = undefined; }
+			if (failed && (handedHere || wakeHere) && jobIdOf(entry.steer)) entry.deferred = true;
 			remaining.push(entry);
 		}
 	}
@@ -373,16 +417,10 @@ export interface AbortOutcome {
  *      runner registered it at loop start) — the in-flight model call
  *      aborts, not just the next loop check. The runner unwinds silently
  *      (its catch checks abortSignal.aborted; finishJob checks status).
- *   2. DROP any queued result steer for this job (the delivering-result
- *      phase): resolve its entry without firing the wake. If the wake
- *      turn is ALREADY in flight (the pump shifted the entry), it's too
- *      late — the result steer lands, and the cancellation steer dedupes
- *      against it via steerDelivered (same jobId metadata).
- *   3. FLIP the record to error "Aborted by learner" immediately — the
- *      poll reflects the user's action at once (the user chose
- *      error-with-reason over a distinct cancelled status). finalizeJob
- *      never overwrites a non-running status, so the flip is terminal no
- *      matter what the wake does.
+ *   2. REFUSE if computation already finished. Queued results belong to
+ *      the session, not to an abortable background worker.
+ *   3. SAVE error "Aborted by learner" and a pending abort delivery together.
+ *      Delivery acknowledgement never overwrites the worker's outcome.
  *   4. STEER the teacher: a one-line abort note (buildAbortSteer) through
  *      the same wake pipeline, so the teacher's promise of a result
  *      doesn't hang silently in the lesson.
@@ -394,28 +432,18 @@ export function abortJob(jobId: string): AbortOutcome {
 	const job = stores.jobStore.getJob(jobId);
 	if (!job) return { ok: false, reason: "not_found" };
 	if (job.status !== "running") return { ok: false, reason: "not_running" };
-
-	fireJobAbort(jobId); // no-op when the loop already returned (delivering)
-
-	// Drop any queued steers for this job (their deliver() resolves unfired).
-	// HANDED entries (already in a live run's follow-up queue) are past
-	// recall — same ruling as a fired wake: the result lands, and the abort
-	// note below dedupes against it via steerDelivered (same jobId).
-	const queue = state().queues.get(job.chatId);
-	if (queue) {
-		const remaining = queue.filter(
-			(e) => jobIdOf(e.steer) !== jobId || e.handedTo !== undefined,
-		);
-		if (remaining.length === 0) state().queues.delete(job.chatId);
-		else state().queues.set(job.chatId, remaining);
-		for (const entry of queue) {
-			if (jobIdOf(entry.steer) === jobId && entry.handedTo === undefined) entry.resolve();
-		}
+	if (job.result) {
+		normalizeLegacyResult(job);
+		return { ok: false, reason: "not_running" };
 	}
 
-	stores.jobStore.updateJob(jobId, { status: "error", error: "Aborted by learner", activity: "aborted" });
+	fireJobAbort(jobId);
+	stores.jobStore.updateJob(jobId, {
+		status: "error", error: "Aborted by learner", activity: "aborted",
+		delivery: { kind: "abort", status: "pending" },
+	});
 	void deliverJobResult(buildAbortSteer(job)).catch(() => {
-		/* deliverJobResult marks the job error on unrecoverable failure */
+		/* The saved pending abort notice remains owned by the session. */
 	});
 	return { ok: true };
 }
@@ -470,18 +498,9 @@ async function pump(chatId: string): Promise<void> {
 			// Idempotent delivery: the transcript already carries these steers
 			// (an in-run injection persisted, a wake completed, or a duplicate
 			// deliver raced in). Resolve and flip without re-firing anything.
-			const stillPending = queue.filter((e) => !steerDelivered(record.messages, e.steer.id));
-			for (const entry of queue) {
-				if (!stillPending.includes(entry)) {
-					finalizeSteer(entry.steer);
-					entry.resolve();
-				}
-			}
-			if (stillPending.length === 0) {
-				s.queues.delete(chatId);
-				return;
-			}
-			s.queues.set(chatId, stillPending);
+			noteSteersDurable(chatId, record.messages);
+			const stillPending = (s.queues.get(chatId) ?? []).filter((entry) => !entry.waking && !entry.deferred);
+			if (stillPending.length === 0) { parked = true; return; }
 			if (!automaticFollowUpAllowed(chatId, s.liveAgents.get(chatId))) { parked = true; return; }
 
 			// VEHICLE 1 — THE LIVE RUN: pi's steering queue. The agent of the
@@ -526,28 +545,17 @@ async function pump(chatId: string): Promise<void> {
 			// The runner is required — chat-handler sets it at init. If some
 			// module instance missed the wiring, fail loudly rather than hang.
 			if (!s.turnRunner) {
-				s.queues.delete(chatId);
 				const err = new Error("wake pipeline: no turn runner wired (setTurnRunner)");
-				for (const entry of stillPending) {
-					failSteer(entry.steer, err);
-					entry.reject(err);
-				}
+				for (const entry of stillPending) deferOrReject(chatId, entry, err);
+				parked = true;
 				return;
 			}
 
 			// A live run may have delivered some markers while the gates were
 			// closed — re-check: only still-missing markers ride the wake.
-			const fire = stillPending.filter((e) => !steerDelivered(fresh.messages, e.steer.id));
-			for (const entry of stillPending) {
-				if (!fire.includes(entry)) {
-					finalizeSteer(entry.steer);
-					entry.resolve();
-				}
-			}
-			if (fire.length === 0) {
-				s.queues.delete(chatId);
-				return;
-			}
+			noteSteersDurable(chatId, fresh.messages);
+			const fire = (s.queues.get(chatId) ?? []).filter((entry) => !entry.waking && !entry.deferred && !entry.handedTo);
+			if (fire.length === 0) return;
 
 			// THE DISMISSAL PAUSE (owner design 2026-10-03, "A2"): the record
 			// ends in a user dismissal — the session waits for the USER. Hold
@@ -561,50 +569,37 @@ async function pump(chatId: string): Promise<void> {
 				return;
 			}
 
-			// DEQUEUE-AT-FIRE (the Wave C lesson from the Wave B regression):
-			// the carried markers leave the registry BEFORE the turn runs —
-			// the wake run's OWN follow-up poll must never see its own cargo,
-			// and a 409 (client raced in) must be able to push them back.
-			s.queues.delete(chatId);
+			// Keep ownership through the wake. The flag excludes cargo from
+			// its own live-agent handoff while duplicate callers share it.
+			for (const entry of fire) entry.waking = true;
 			const steers = fire.map((e) => e.steer);
 
-			// Fire the wake (every marker as the turn's input appends). A 409 =
-			// a client turn raced in between idle and firing — push the cargo
-			// back and retry after backoff (the racing turn's own settlement
-			// re-pumps if it beats us to the re-queue).
+			// Fire the wake with cargo still registered. A 409 releases its
+			// vehicle flag and retries after the existing backoff, without
+			// appending another registry entry.
 			const status = await s.turnRunner(chatId, { type: "steers", steers }).catch((err: unknown) => {
 				console.error(`[steer ${steers.map((x) => x.id).join(",")}] wake turn crashed:`, err);
 				return -1;
 			});
-			if (status === 403) {
-				// Consent may have changed after the last gate. Preserve completed
-				// job/result ownership and wait for an explicit connection change.
-				const back = s.queues.get(chatId) ?? [];
-				back.push(...fire);
-				s.queues.set(chatId, back);
-				parked = true;
-				return;
-			}
-			if (status === 409) {
-				const back = s.queues.get(chatId) ?? [];
-				back.push(...fire);
-				s.queues.set(chatId, back);
+			const saved = stores.chatStore.getChat(chatId);
+			if (saved) noteSteersDurable(chatId, saved.messages);
+			const owned = fire.filter((entry) => s.queues.get(chatId)?.includes(entry));
+			if (status === 403 || status === 409) {
+				for (const entry of owned) { entry.waking = false; entry.wakeAgent = undefined; }
+				if (status === 403) { parked = true; return; }
 				await delay(RETRY_BACKOFF_MS[Math.min(attempt++, RETRY_BACKOFF_MS.length - 1)]!);
 				continue;
 			}
 			attempt = 0;
-
-			if (status === 200) {
-				for (const entry of fire) {
-					finalizeSteer(entry.steer);
-					entry.resolve();
-				}
-			} else {
-				const err = new Error(`wake turn failed with status ${status}`);
-				for (const entry of fire) {
-					failSteer(entry.steer, err);
-					entry.reject(err);
+			for (const entry of owned) {
+				if (status === 200 && !jobIdOf(entry.steer)) {
+					// Preserve the non-job probe handoff's acceptance semantics.
+					completeEntry(chatId, entry);
+				} else if (status !== 200 || !isTurnRunning(chatId)) {
+					deferOrReject(chatId, entry, new Error(`wake turn did not persist steer (status ${status})`));
 				}
+				// An accepted live turn still owns its cargo until canonical save
+				// or settlement. HTTP 200 alone is not a durable marker receipt.
 			}
 		}
 	} finally {
@@ -614,7 +609,7 @@ async function pump(chatId: string): Promise<void> {
 		// a live run are NOT stragglers (settleSteers owns their resurrection)
 		// — restarting for them would just park in waitForIdle forever.
 		const q = state().queues.get(chatId) ?? [];
-		if (!parked && automaticFollowUpAllowed(chatId, state().liveAgents.get(chatId)) && q.some((e) => e.handedTo === undefined)) void pump(chatId);
+		if (!parked && automaticFollowUpAllowed(chatId, state().liveAgents.get(chatId)) && q.some((e) => e.handedTo === undefined && !e.waking && !e.deferred)) void pump(chatId);
 	}
 }
 
@@ -638,25 +633,56 @@ state().stopConnectionWatch = onSubscriptionChange(() => {
 	}
 });
 
-/** Flip the job to its final status now that the lesson has resumed. */
-function finalizeJob(jobId: string): void {
-	const job = stores.jobStore.getJob(jobId);
-	if (!job || job.status !== "running") return;
-	const result = job.result;
-	const failed = result ? isSubagentFailure(job.kind, result as never) : true;
-	stores.jobStore.updateJob(jobId, {
-		status: failed ? "error" : "done",
-		activity: failed ? "failed" : "delivered",
-	});
+function deferOrReject(chatId: string, entry: QueuedSteer, error: Error): void {
+	entry.waking = false;
+	entry.wakeAgent = undefined;
+	if (jobIdOf(entry.steer)) {
+		entry.deferred = true;
+		console.error(`[steer ${entry.steer.id}] delivery remains pending:`, error.message);
+	} else {
+		removeEntry(chatId, entry);
+		entry.reject(error);
+	}
 }
 
-function failJob(jobId: string, err: Error): void {
-	// An aborted job stays aborted even when its (abort) steer
-	// delivery fails — the user's action is the terminal fact, not the
-	// delivery error. Non-running records are never overwritten here.
-	const job = stores.jobStore.getJob(jobId);
-	if (!job || job.status !== "running") return;
-	stores.jobStore.updateJob(jobId, { status: "error", error: err.message, activity: "delivery failed" });
+/** Acknowledge session delivery without changing the worker outcome. */
+function finalizeJob(jobId: string): void {
+	try {
+		let job = stores.jobStore.getJob(jobId);
+		if (!job) return;
+		if (!job.delivery) job = normalizeLegacyResult(job);
+		if (job.delivery?.status !== "pending") return;
+		stores.jobStore.updateJob(jobId, { delivery: { ...job.delivery, status: "delivered" } });
+	} catch (error) {
+		// The conversation is already durable. A bookkeeping failure must
+		// never replay that marker or turn a successful worker into a failure.
+		console.error(`[job ${jobId}] delivery acknowledgement failed:`, error);
+	}
+}
+
+function normalizeLegacyResult(job: JobRecord): JobRecord {
+	if (job.delivery || job.status !== "running" || !job.result) return job;
+	const failed = isSubagentFailure(job.kind, job.result as never);
+	return stores.jobStore.updateJob(job.id, {
+		status: failed ? "error" : "done", activity: failed ? "failed" : "finished",
+		delivery: { kind: "result", status: "pending" },
+	}) ?? job;
+}
+
+/** Restore known obligations only; old terminal records are never replayed. */
+function restorePendingDeliveries(chatId: string): void {
+	const record = stores.chatStore.getChat(chatId);
+	if (!record) return;
+	noteSteersDurable(chatId, record.messages);
+	for (const stored of stores.jobStore.listJobs(chatId)) {
+		const job = isJobLive(stored.id) ? stored : normalizeLegacyResult(stored);
+		if (job.delivery?.status !== "pending") continue;
+		const steer = job.delivery.kind === "abort" ? buildAbortSteer(job)
+			: job.result ? buildSubagentSteer(job, job.result as never) : undefined;
+		if (!steer) continue;
+		if (steerDelivered(record.messages, steer.id)) finalizeSteer(steer);
+		else queueSteer(chatId, steer);
+	}
 }
 
 /**
@@ -669,7 +695,7 @@ function failJob(jobId: string, err: Error): void {
 async function waitForIdle(chatId: string): Promise<void> {
 	for (;;) {
 		const record = stores.chatStore.getChat(chatId);
-		if (!record) return;
+		if (!record || !(state().queues.get(chatId) ?? []).some((entry) => !entry.waking && !entry.deferred && !entry.handedTo)) return;
 		const blocked = isTurnRunning(chatId) || danglingQuestionOf(record.messages) !== null;
 		if (!blocked) {
 			await delay(GRACE_MS);
@@ -699,9 +725,9 @@ const SWEEP_THROTTLE_MS = 30_000;
  * hung-but-alive loops beside themselves — the figure-maker incident,
  * 2026-09-18). A record WITH a live registration is provably executing
  * in THIS process — never touched, however old: a hung first model call
- * (zero usage ever billed) stays honestly "running" in the dock until
- * END kills it. The abort registration spans loop start to delivery
- * completion, so a live registration can never belong to a dead run.
+ * stays honestly running until END kills it. Terminal workers with known
+ * pending delivery are restored independently; old terminal records without
+ * delivery metadata are never inferred to owe a delivery.
  */
 /** Re-run a job's sub-agent loop — exposed for the sweep's tests (the
  *  /api/jobs poll is its only production caller, disabled under NODE_ENV=test). */
@@ -713,18 +739,14 @@ export function recoverStaleJobs(): void {
 	try {
 		for (const chat of stores.chatStore.listChats()) {
 			for (const job of stores.jobStore.listJobs(chat.id)) {
-				if (job.status !== "running") continue;
-				if (isJobLive(job.id)) continue; // provably executing here — never re-run beside it
+				if (isJobLive(job.id)) continue; // never re-run beside live computation
 				if (!stores.chatStore.getChat(job.chatId)) continue;
-				if (isTurnRunning(job.chatId)) continue; // a wake turn may be delivering it right now
-				if (job.result) {
-					// The wake never fired (or crashed mid-flight): re-deliver.
-					const steer = buildSubagentSteer(job, job.result as never);
-					stores.jobStore.updateJob(job.id, { activity: "recovering: re-delivering" });
-					void deliverJobResult(steer).catch(() => {
-						/* deliverJobResult marks the job error on failure */
-					});
-				} else {
+				if (isTurnRunning(job.chatId)) continue;
+				const recovered = normalizeLegacyResult(job);
+				if (recovered.delivery?.status === "pending") {
+					restorePendingDeliveries(job.chatId);
+					void pump(job.chatId);
+				} else if (job.status === "running" && !job.result) {
 					// The sub-agent loop itself died with the process: re-spawn.
 					if (!s.respawn) continue;
 					stores.jobStore.updateJob(job.id, { activity: "recovering: re-running" });

src/server/model-catalog.ts

@@ -248,10 +248,17 @@ export function validateChatSettings(settings: ChatSettings): SettingsValidation
 	return { ok: true };
 }
 
-export function validateModelsSettings(settings: ModelsSettings): SettingsValidation {
+/** Validate introduced/replaced model selections, not unchanged saved pins.
+ * A disconnected subscription loses its available catalog, but its stored
+ * selections must not prevent repairing another slot or editing thinking.
+ * Only the SAME slot's provider+id can pass through; runtime admission stays
+ * separate. Callers without a stored section retain strict catalog checking. */
+export function validateModelsSettings(settings: ModelsSettings, stored?: ModelsSettings): SettingsValidation {
 	for (const name of MODEL_SLOTS) {
 		const slot = settings[name];
 		if (!slot) continue;
+		const previous = stored?.[name]?.model;
+		if (slot.model && previous && slot.model.provider === previous.provider && slot.model.id === previous.id) continue;
 		const verdict = validateSlotSpec(name, slot);
 		if (!verdict.ok) return verdict;
 	}
@@ -283,9 +290,8 @@ function validateSlotSpec(name: ModelSlotName, slot: { model?: { provider: strin
 // runtime — and the clamped value is what lands in the record. The
 // server is the truth: the record shows the clamped level after write.
 
-/** Clamp one level against one model (identity when the model is unknown
- *  or the level already supported — the existence gate rejects unknown
- *  models separately, so this only ever sees real ones in practice). */
+/** Clamp one level against one model. Preserve the preference when metadata
+ *  is unavailable for an unchanged saved pin; runtime clamps after reconnect. */
 function clampLevelFor(provider: string, id: string, level: ThinkingLevel): ThinkingLevel {
 	const model = connectionModel(provider, id);
 	return model ? clampThinkingLevel(model, level) : level;

src/server/pi/ui-stream.ts

@@ -1,6 +1,4 @@
 import type { Agent, AgentEvent, AgentMessage } from "@earendil-works/pi-agent-core";
-import type { SteerMessage } from "../../core/contracts.ts";
-import { noteSteerVisible } from "../jobs.ts";
 import { ToolCallIdUniquifier } from "./call-ids.ts";
 import type { ChatGPTRequestCycle } from "../../agent/chatgpt-request-cycle.ts";
 
@@ -79,6 +77,10 @@ export interface PiTurnSpec {
 	 *  event stream itself is verbatim — provider errors ride their
 	 *  message's stopReason/errorMessage through untouched). */
 	onError?: (error: unknown) => string;
+	/** Report settlement failure while the views are still open. */
+	onSettleError?: (error: unknown) => string;
+	/** Release liveness even if the normal finish seam throws. */
+	onSettleFailure?: (error: unknown, reportedErrorText?: string) => void;
 	/** The canonical end-of-turn seam: persist + usage + extraction + endTurn.
 	 *  Receives the agent's FULL final state (native messages) — the caller
 	 *  persists the slice past its own base count. */
@@ -119,11 +121,12 @@ export interface PiTurnHandle {
 }
 
 export function runPiTurn(spec: PiTurnSpec): PiTurnHandle {
-	const { agent, onError, onFinish, usedToolCallIds, updateFlushMs = 100, requestCycle } = spec;
+	const { agent, onError, onFinish, onSettleError, onSettleFailure, usedToolCallIds, updateFlushMs = 100, requestCycle } = spec;
 	const uniquifier = usedToolCallIds ? new ToolCallIdUniquifier(usedToolCallIds) : undefined;
 
 	let settled = false;
 	let sawAbort = false;
+	let reportedErrorText: string | undefined;
 
 	// THE PER-TURN STATE LOG (memory-only — the settle-only disk writes
 	// are unchanged). COMPACTION LAW (owner, 2026-10-09, supersedes the
@@ -221,11 +224,22 @@ export function runPiTurn(spec: PiTurnSpec): PiTurnHandle {
 	// messages of its own (the prompt is composed per turn), so dropping
 	// the role wholesale is exact; revisit if prompt-updates ever land.
 	const settleWith = () => {
-		const settled_ = [...agent.state.messages].filter((m) => (m as { role?: string }).role !== "system");
-		onFinish((uniquifier?.remapSettled(settled_) as AgentMessage[]) ?? settled_, sawAbort);
+		try {
+			const settled_ = [...agent.state.messages].filter((m) => (m as { role?: string }).role !== "system");
+			onFinish((uniquifier?.remapSettled(settled_) as AgentMessage[]) ?? settled_, sawAbort);
+		} catch (err) {
+			console.error("[turn] settlement failed:", err);
+			try { onSettleFailure?.(err, reportedErrorText); }
+			catch (cleanupError) { console.error("[turn] settlement cleanup failed:", cleanupError); }
+			if (reportedErrorText === undefined) {
+				reportedErrorText = onSettleError?.(err) ?? "An error occurred.";
+				send({ type: "error", errorText: reportedErrorText });
+			}
+		}
 	};
 
 	const handle = (event: AgentEvent) => {
+		if (settled) return;
 		// THE LACAN FIX: ids uniquify HERE, before any view — the one seam
 		// every consumer downstream shares. No uniquifier → verbatim.
 		const uniqued = uniquifier ? uniquifier.event(event) : event;
@@ -239,15 +253,11 @@ export function runPiTurn(spec: PiTurnSpec): PiTurnHandle {
 		if (event.type === "message_end") {
 			const m = event.message as { role?: string; stopReason?: string } | undefined;
 			if (m?.role === "assistant" && m.stopReason === "aborted") sawAbort = true;
-			// Dock fix (owner ruling 2026-09-26): an injected steer marker's
-			// message_end = the card is visible to the learner — flip the
-			// job's bookkeeping now (the turn may keep teaching for minutes;
-			// settleSteers re-runs the flip as a no-op idempotently).
-			if (m?.role === "custom") noteSteerVisible(event.message as SteerMessage);
 		}
 		if (event.type === "agent_end") {
-			closeAll();
+			// Persist/report before the terminal closes every view to new frames.
 			settleWith();
+			closeAll();
 		}
 	};
 	requestCycle?.bindRelease(handle);
@@ -273,11 +283,15 @@ export function runPiTurn(spec: PiTurnSpec): PiTurnHandle {
 			// stream as stopReason "error"): surface, close, persist. If the
 			// agent already settled (agent_end reached before the throw),
 			// the seam has run — never settle twice.
-			if (settled) return;
+			if (settled) {
+				console.error("[turn] error after settlement:", err);
+				return;
+			}
 			console.error("[turn] agent loop failed:", err);
-			send({ type: "error", errorText: onError?.(err) ?? "An error occurred." });
-			closeAll();
+			reportedErrorText = onError?.(err) ?? "An error occurred.";
+			send({ type: "error", errorText: reportedErrorText });
 			settleWith();
+			closeAll();
 		}
 	})();
 

src/server/settings-body.ts

@@ -19,12 +19,14 @@ import type {
 	BackgroundModelRole,
 	CompactionSettings,
 	GeneralSettings,
+	DictationSettings,
 	GraphSettings,
 	ModelsSettings,
 	RoleModelOverride,
 } from "../core/settings.ts";
 import { BACKGROUND_MODEL_ROLES, COMPACTION_THRESHOLD_FLOOR, EXTRACTION_CHUNK_TOKENS_FLOOR, MODEL_SLOTS } from "../core/settings.ts";
 import { WRITABLE_TOOL_KEYS } from "../core/pending.ts";
+import { isDictationLanguage } from "../core/dictation.ts";
 
 /** Parsed-or-rejected — `ok: false` carries the 400's error message. */
 export type SettingsBodyResult<T> =
@@ -375,6 +377,15 @@ export function parseCompactionSettingsBody(body: unknown): SettingsBodyResult<C
 	return { ok: true, settings };
 }
 
+/** dictation: { language?: one of Nova-3's supported language codes } */
+export function parseDictationSettingsBody(body: unknown): SettingsBodyResult<DictationSettings> {
+	const raw = asRecord(body);
+	if (!raw) return { ok: false, error: "Invalid JSON body" };
+	if (raw.language === undefined) return { ok: true, settings: {} };
+	if (!isDictationLanguage(raw.language)) return { ok: false, error: "Unsupported dictation language" };
+	return { ok: true, settings: { language: raw.language } };
+}
+
 /** The section→parser map for PUT /api/settings/[section] (the route also
  *  carries the special prompts sub-resource). */
 export const SETTINGS_SECTIONS = {
@@ -385,6 +396,7 @@ export const SETTINGS_SECTIONS = {
 	graph: parseGraphSettingsBody,
 	compaction: parseCompactionSettingsBody,
 	models: parseModelsSettingsBody,
+	dictation: parseDictationSettingsBody,
 } as const;
 
 export type SettingsSectionName = keyof typeof SETTINGS_SECTIONS;

src/server/turn-liveness.ts

@@ -21,6 +21,7 @@
  */
 
 import { emit } from "../core/events.ts";
+import { clearTurnFailure } from "./turn-runs.ts";
 
 interface LhGlobal {
 	__lhTurnRegistry?: Map<string, number>;
@@ -43,15 +44,17 @@ const MAX_TURN_MS = 30 * 60_000;
  */
 export function beginTurn(chatId: string): boolean {
 	if (isTurnRunning(chatId)) return false;
+	clearTurnFailure(chatId);
 	inFlight.set(chatId, Date.now());
 	emit("turn-started", { chatId });
 	return true;
 }
 
 /**
- * Mark the chat's turn as finished (runs in onFinish, aborted or not —
- * AFTER persistTurn, so "turn-settled" guarantees a subsequent fetch reads
- * durable truth). Emits only when an entry was actually present: a
+ * Mark the chat's turn as finished after normal settlement, or after the
+ * failure notice has been retained by settlement cleanup. A subsequent
+ * fetch reads durable truth, never a promise that an unsaved reply landed.
+ * Emits only when an entry was actually present: a
  * defensive no-op endTurn must not fire a spurious settle.
  */
 export function endTurn(chatId: string): void {

src/server/turn-runs.ts

@@ -13,7 +13,9 @@
  * Registration lifecycle: chat-handler registers the handle the moment
  * the turn starts (after the 409 guard) and its onFinish unregisters LAST
  * — after persist, usage, extraction, and endTurn — so anything that sees
- * "no running turn" also sees an already-persisted record. The per-chat
+ * "no running turn" sees durable record truth. Failed settlement also
+ * retains its small notice before unregistering, so late attachments do
+ * not mistake an unsaved reply for success. The per-chat
  * 409 guard means a chatId can hold at most one handle, and beginTurn's
  * 30-minute TTL bounds the pathological never-settling case the same way
  * it does for the liveness map: if onFinish never fires, the stale entry
@@ -30,6 +32,7 @@ import type { PiTurnHandle } from "./pi/ui-stream.ts";
 
 interface LhTurnRunsGlobal {
 	__lhTurnRuns?: Map<string, PiTurnHandle>;
+	__lhTurnFailures?: Map<string, string>;
 }
 
 function runs(): Map<string, PiTurnHandle> {
@@ -37,6 +40,25 @@ function runs(): Map<string, PiTurnHandle> {
 	return (g.__lhTurnRuns ??= new Map<string, PiTurnHandle>());
 }
 
+/** One small, transient notice per failed session. Never stores messages.
+ * Cleared on the next accepted begin, session deletion, or process exit. */
+function failures(): Map<string, string> {
+	const g = globalThis as typeof globalThis & LhTurnRunsGlobal;
+	return (g.__lhTurnFailures ??= new Map<string, string>());
+}
+
+export function rememberTurnFailure(chatId: string, errorText: string): void {
+	failures().set(chatId, errorText);
+}
+
+export function turnFailure(chatId: string): string | undefined {
+	return failures().get(chatId);
+}
+
+export function clearTurnFailure(chatId: string): void {
+	failures().delete(chatId);
+}
+
 /** Register the chat's running turn (chat-handler, right after the turn
  *  starts). One per chat by construction — the 409 guard. */
 export function registerTurn(chatId: string, handle: PiTurnHandle): void {
@@ -67,11 +89,14 @@ export function abortTurn(chatId: string): { ok: true } | { ok: false; reason: "
 /** The cheap terminal when GET /api/chats/[id]/stream finds no running
  *  turn (raced settle, or the client checked turnInFlight a beat too
  *  late): a same-dialect SSE response whose single chunk tells the client
- *  "nothing to attach to — the record is current, refetch it". Chosen as
- *  an EVENT (not a status code) so the reattach path consumes ONE wire
+ *  "nothing to attach to — refetch durable truth". A retained settlement
+ *  failure is delivered first, including when no view attached in time.
+ *  Chosen as an EVENT (not a status code), consuming ONE wire
  *  protocol end to end. */
-export function settledTurnResponse(): Response {
-	return new Response('data: {"type":"turn-settled"}\n\ndata: [DONE]\n\n', {
+export function settledTurnResponse(chatId?: string): Response {
+	const errorText = chatId ? turnFailure(chatId) : undefined;
+	const notice = errorText === undefined ? "" : `data: ${JSON.stringify({ type: "error", errorText })}\n\n`;
+	return new Response(`${notice}data: {"type":"turn-settled"}\n\ndata: [DONE]\n\n`, {
 		headers: {
 			"Content-Type": "text/event-stream",
 			"Cache-Control": "no-cache, no-transform",

src/server/turn-service.ts

@@ -1,5 +1,5 @@
 /**
- * The chat-turn services — one named, error-contained seam per turn
+ * The chat-turn services — one named seam per turn
  * responsibility, extracted from the chat-handler so it composes instead of
  * implementing. Each takes its dependencies explicitly (storage ports, the
  * job runner) — no module here imports an adapter.
@@ -26,6 +26,14 @@ import type { ChatStore, GraphStore } from "../core/ports.ts";
 import type { ExtractionPipeline } from "../agent/subagents/extraction.ts";
 import { ModelsNotConfiguredError, ProviderKeyMissingError } from "../core/settings.ts";
 
+/** Distinguish a failed canonical save from later bookkeeping failures. */
+export class TurnPersistenceError extends Error {
+	constructor(cause: unknown) {
+		super("The reply could not be saved.", { cause });
+		this.name = "TurnPersistenceError";
+	}
+}
+
 /**
  * The canonical state write (Wave B, owner-approved 2026-09-26): append the
  * turn's own messages — the agent loop's native history, provider/model/
@@ -45,7 +53,8 @@ import { ModelsNotConfiguredError, ProviderKeyMissingError } from "../core/setti
  * - Plan blocks: one frozen snapshot per successful plan_init call,
  *   idempotent (persist replays are safe).
  *
- * Returns the stamped new messages (usage provenance reads them).
+ * Returns the stamped new messages (usage provenance reads them). Throws
+ * TurnPersistenceError on failure so the runner can report it before DONE.
  */
 export function persistTurn(
 	chatStore: ChatStore,
@@ -56,7 +65,7 @@ export function persistTurn(
 ): LhMessage[] {
 	try {
 		const record = chatStore.getChat(chatId);
-		if (!record) return [];
+		if (!record) throw new Error(`Chat not found: ${chatId}`);
 		const stamped: LhMessage[] = [];
 		for (const m of turnMessages.slice(baseCount)) {
 			const message = m as LhMessage;
@@ -79,7 +88,7 @@ export function persistTurn(
 		return stamped;
 	} catch (err) {
 		console.error(`[chat ${chatId}] persist failed:`, err, { isAborted });
-		return [];
+		throw new TurnPersistenceError(err);
 	}
 }
 

src/server/web-subscription-http.ts

@@ -0,0 +1,68 @@
+import { NATIVE_SUBSCRIPTIONS } from "../core/contracts/subscriptions.ts";
+import { webSubscriptionOwner, webSubscriptionsEnabled, webSubscriptionStatus } from "./web-subscriptions.ts";
+
+const HEADERS = { "Cache-Control": "no-store", "Referrer-Policy": "no-referrer", "X-Content-Type-Options": "nosniff" };
+/** Same-origin browser controls, not user authentication. Web/dev deliberately
+ * retains the existing shared-server/API-key trust model. Never trust forwarded
+ * origins or add CORS grants here. A public deployment needs its own access gate. */
+export function subscriptionRequestOrigin(request: Request): string | undefined {
+	try {
+		const url = new URL(request.url);
+		const host = request.headers.get("host") ?? url.host;
+		const addressed = new URL(`${url.protocol}//${host}`);
+		if (!["http:", "https:"].includes(addressed.protocol) || addressed.username || addressed.password || addressed.pathname !== "/" || addressed.search || addressed.hash) return;
+		return addressed.origin;
+	} catch { return; }
+}
+function authorized(request: Request): boolean {
+	const origin = subscriptionRequestOrigin(request);
+	if (!origin || request.headers.get("x-learnharness-subscriptions") !== "1") return false;
+	if (request.headers.get("sec-fetch-site") === "cross-site") return false;
+	const supplied = request.headers.get("origin");
+	return request.method === "GET" ? !supplied || supplied === origin : supplied === origin;
+}
+function error(message: string, status: number) { return Response.json({ error: message }, { status, headers: HEADERS }); }
+export async function subscriptionControlsResponse(request: Request): Promise<Response> {
+	if (!webSubscriptionsEnabled()) return error("Web subscription controls are unavailable in desktop-owned mode.", 404);
+	if (!authorized(request)) return error("Subscription controls require a same-origin LearnHarness request.", 403);
+	try {
+		if (request.method === "GET") return Response.json(await webSubscriptionStatus(), { headers: HEADERS });
+		if (request.method !== "POST") return error("Unsupported method.", 405);
+		if (request.headers.get("content-type")?.split(";")[0]?.trim() !== "application/json") return error("Expected JSON.", 415);
+		// Bound both advertised and streamed input before parsing commands.
+		if (Number(request.headers.get("content-length")) > 16384) return error("Request too large.", 413);
+		const reader = request.body?.getReader();
+		if (!reader) return error("Expected a command.", 400);
+		const chunks: Uint8Array[] = []; let size = 0;
+		try {
+			for (;;) {
+				const next = await reader.read(); if (next.done) break;
+				size += next.value.byteLength;
+				if (size > 16384) { await reader.cancel(); return error("Request too large.", 413); }
+				chunks.push(next.value);
+			}
+		} finally { reader.releaseLock(); }
+		const command: unknown = JSON.parse(Buffer.concat(chunks).toString("utf8"));
+		if (!command || typeof command !== "object" || Array.isArray(command)) return error("Invalid command.", 400);
+		const fields = command as Record<string, unknown>;
+		const providerId = fields.providerId;
+		if (providerId !== "chatgpt" && !NATIVE_SUBSCRIPTIONS.some((provider) => provider.providerId === providerId)) return error("Unsupported connection.", 400);
+		const allowed = providerId === "chatgpt" ? ["login", "cancel", "select", "disconnect", "logout", "consent"] : ["login", "cancel", "respond", "open", "retry", "disconnect", "consent"];
+		if (typeof fields.op !== "string" || !allowed.includes(fields.op)) return error("Unsupported control.", 400);
+		return Response.json(await webSubscriptionOwner().control(providerId as string, fields, subscriptionRequestOrigin(request)!), { headers: HEADERS });
+	} catch {
+		return error("The subscription connection could not be changed. Retry or reconnect in Providers.", 400);
+	}
+}
+/** One-use same-origin navigation. Full provider authorization URLs go only to
+ * a top-level browser redirect, never status JSON, the SSE bus or application JS. */
+export async function subscriptionBrowserResponse(request: Request): Promise<Response> {
+	if (!webSubscriptionsEnabled()) return error("Unavailable.", 404);
+	const origin = subscriptionRequestOrigin(request);
+	if (!origin || (request.method === "POST" ? !authorized(request) : request.headers.get("sec-fetch-site") !== "same-origin")) return error("Open sign-in from LearnHarness.", 403);
+	const id = new URL(request.url).searchParams.get("id");
+	if (!id || !/^[0-9a-f-]{36}$/.test(id)) return error("This sign-in action is no longer available.", 410);
+	const target = await webSubscriptionOwner().takeBrowserAction(id, origin);
+	if (!target) return error("This sign-in action is no longer available.", 410);
+	return new Response(null, { status: 303, headers: { ...HEADERS, Location: target } });
+}

src/server/web-subscriptions.ts

@@ -0,0 +1,127 @@
+/** Persistent Node web/dev owner. Never imported by client code. Desktop keeps
+ * its existing main-process owner; a broken desktop IPC channel cannot fall
+ * back to these files. One server process must own each configured auth dir. */
+import { AsyncLocalStorage } from "node:async_hooks";
+import { randomUUID } from "node:crypto";
+import path from "node:path";
+import { AUTH_FILE } from "../../config.ts";
+import { createChatGPT, fileStore } from "../../electron/chatgpt.cjs";
+import { createSubscriptions } from "../../electron/subscriptions.cjs";
+import { chatGPTControl } from "../../electron/chatgpt-ipc.cjs";
+import { subscriptionControl } from "../../electron/subscription-ipc.cjs";
+import { emit } from "../core/events.ts";
+import type { ChatGPTStatus } from "../core/contracts/chatgpt.ts";
+import type { NativeSubscriptionSnapshot } from "../core/contracts/subscriptions.ts";
+import { configureWebSubscriptionRuntime, publishWebSubscriptionPacket, webSubscriptionsEnabled, type SubscriptionRuntimeChannel as Channel, type SubscriptionRuntimePacket as Packet } from "../agent/subscription-runtime.ts";
+export { webSubscriptionsEnabled } from "../agent/subscription-runtime.ts";
+interface BrowserAction { target: string; origin: string; scope: string; expires: number; current(): Promise<boolean> }
+const KEY = Symbol.for("lh.webSubscriptions");
+interface Registry {
+	owner?: ReturnType<typeof createWebSubscriptionOwner>;
+}
+function registry(): Registry {
+	const global = globalThis as typeof globalThis & { [KEY]?: Registry };
+	return global[KEY] ??= {};
+}
+function publish(channel: Channel, packet: Packet) {
+	publishWebSubscriptionPacket(channel, packet);
+	emit("subscriptions-changed", {});
+}
+async function loadSDK() {
+	const [sdk, copilot, xai] = await Promise.all([
+		import("@earendil-works/pi-ai"),
+		import("@earendil-works/pi-ai/providers/github-copilot"),
+		import("@earendil-works/pi-ai/providers/xai"),
+	]);
+	return { ...sdk, ...copilot, ...xai };
+}
+
+/** Injectable factories/stores keep all test authentication offline. */
+export function createWebSubscriptionOwner(options: {
+	directory: string;
+	chatgptStore?: Parameters<typeof createChatGPT>[0]["store"];
+	sdk?: unknown;
+	onPacket?: (channel: Channel, packet: Packet) => void;
+} ) {
+	const notify = options.onPacket ?? (() => {});
+	const launches = new AsyncLocalStorage<string[]>();
+	const actions = new Map<string, BrowserAction>();
+	let chatGPTLaunch = 0;
+	const openExternal = async (target: string) => {
+		const captured = launches.getStore();
+		if (!captured || new URL(target).protocol !== "https:") throw new Error("No browser action is pending.");
+		captured.push(target);
+	};
+	const chatgpt = createChatGPT({
+		store: options.chatgptStore ?? fileStore(path.join(options.directory, "chatgpt-credentials.json")),
+		openExternal,
+		onChange: (packet?: Packet) => { if (packet) notify("lh:chatgpt", packet); },
+	});
+	const native = createSubscriptions({
+		directory: options.directory, sdk: options.sdk, loadSDK, openExternal,
+		onChange: (snapshot: NativeSubscriptionSnapshot) => notify("lh:subscriptions", { channel: "lh:subscriptions", op: "changed", snapshot }),
+	});
+	return {
+		chatgpt, native,
+		async control(providerId: string, command: Record<string, unknown>, origin: string) {
+			const targets: string[] = [];
+			const value = await launches.run(targets, () => providerId === "chatgpt"
+				? chatGPTControl(command, chatgpt)
+				: subscriptionControl({ ...command, providerId }, native));
+			let browserAction: string | undefined;
+			const target = targets[0];
+			if (target) {
+				const scope = providerId === "chatgpt" ? "chatgpt" : `${providerId}:${command.attemptId}:${command.actionId}`;
+				for (const [id, action] of actions) if (action.expires <= Date.now() || action.scope === scope) actions.delete(id);
+				const id = randomUUID();
+				const generation = providerId === "chatgpt" ? ++chatGPTLaunch : chatGPTLaunch;
+				const current = providerId === "chatgpt"
+					? async () => generation === chatGPTLaunch && (await chatgpt.status()).pending === true
+					: async () => (await native.status()).some((status: { providerId: string; interaction?: { attemptId: string; browser?: { id: string } } }) => status.providerId === providerId && status.interaction?.attemptId === command.attemptId && status.interaction?.browser?.id === command.actionId);
+				actions.set(id, { target, origin, scope, expires: Date.now() + 180000, current });
+				browserAction = `/api/subscriptions/browser?id=${id}`;
+			}
+			return { value, ...(browserAction ? { browserAction } : {}) };
+		},
+		async takeBrowserAction(id: string, origin: string): Promise<string | undefined> {
+			const action = actions.get(id);
+			if (!action || action.origin !== origin) return;
+			actions.delete(id);
+			if (action.expires <= Date.now() || !await action.current()) return;
+			return action.target;
+		},
+		close() { actions.clear(); chatgpt.close(); native.close(); },
+	};
+}
+export function webSubscriptionOwner() {
+	if (!webSubscriptionsEnabled()) throw new Error("Web subscription controls are unavailable in desktop-owned mode.");
+	const state = registry();
+	if (!state.owner) {
+		state.owner = createWebSubscriptionOwner({ directory: path.dirname(AUTH_FILE), onPacket: publish });
+		process.once("exit", () => state.owner?.close());
+	}
+	return state.owner;
+}
+/** Private runtime calls only: no HTTP lease/token operation exists. */
+export async function webSubscriptionRequest<T>(channel: Channel, op: string, fields: Record<string, unknown>): Promise<T> {
+	const owner = webSubscriptionOwner();
+	let value: unknown;
+	if (channel === "lh:chatgpt") {
+		if (op === "status") value = { status: await owner.chatgpt.status(), ...owner.chatgpt.stamp() };
+		else if (op === "models") value = await owner.chatgpt.models();
+		else if (op === "lease" && typeof fields.background === "boolean") value = await owner.chatgpt.lease(fields.background, fields);
+		else throw new Error("Invalid private operation.");
+	} else {
+		if (op === "snapshot") value = await owner.native.snapshot();
+		else if (op === "lease" && typeof fields.background === "boolean") value = await owner.native.lease(fields.scope, fields.background);
+		else throw new Error("Invalid private operation.");
+	}
+	return value as T;
+}
+configureWebSubscriptionRuntime(webSubscriptionRequest);
+
+export async function webSubscriptionStatus(): Promise<{ chatgpt: ChatGPTStatus; native: NativeSubscriptionSnapshot["statuses"] }> {
+	const owner = webSubscriptionOwner();
+	const [chatgpt, native] = await Promise.all([owner.chatgpt.status(), owner.native.status()]);
+	return { chatgpt, native };
+}