fix(harness): record a stop before interrupting, derive recovery in the drain

apps/server/src/db.ts

@@ -172,6 +172,7 @@ const migrations = {
 				foundation TEXT,
 				placement_space_id TEXT,
 				placement_dir TEXT,
+				parked INTEGER NOT NULL DEFAULT 0,
 				restarts INTEGER NOT NULL DEFAULT 0,
 				attempt INTEGER,
 				run_state TEXT NOT NULL DEFAULT 'idle',

apps/server/src/mirror.ts

@@ -98,6 +98,7 @@ const threadCore = (thread: Codework.ThreadSummary) => {
 		foundation: thread.foundation ?? null,
 		placementSpaceId: thread.placement?.spaceId ?? null,
 		placementDir: thread.placement?.directory ?? null,
+		parked: thread.parked ? 1 : 0,
 		restarts: thread.restarts,
 		attempt: thread.attempt ?? null,
 		runState: thread.runState,

apps/server/src/reads.ts

@@ -299,6 +299,7 @@ const head = (sql: SqlClient.SqlClient, { sessionId, threadId }: LineRequest) =>
 		runEndedAt: number | null;
 		runOutcome: string | null;
 		runError: string | null;
+		parked: number;
 		updatedAt: number;
 		floor: number | null;
 		dir: string;
@@ -309,7 +310,7 @@ const head = (sql: SqlClient.SqlClient, { sessionId, threadId }: LineRequest) =>
 	}>`
 		SELECT thread.id, thread.session_id, session.machine_id, thread.title, thread.model_provider, thread.model_id,
 			thread.thinking_level, thread.run_state, thread.run_started_at, thread.run_ended_at, thread.run_outcome,
-			thread.run_error, thread.updated_at, thread.floor, session.dir, session.hostdir,
+			thread.run_error, thread.parked, thread.updated_at, thread.floor, session.dir, session.hostdir,
 			project.id AS project_id, project.name AS project_name,
 			thread.deleted_at IS NOT NULL OR session.deleted_at IS NOT NULL AS removed
 		FROM thread
@@ -395,6 +396,7 @@ export const line = Effect.fn("Reads.line")(function* (request: LineRequest) {
 			...(row.thinkingLevel === null ? {} : { thinkingLevel: row.thinkingLevel }),
 			runState: row.runState,
 			...lastRun,
+			parked: row.parked === 1,
 			updatedAt: row.updatedAt,
 			directory: row.dir,
 			...(row.hostdir === null ? {} : { hostDir: row.hostdir }),

apps/server/test/__artifacts__/reads.line.json

@@ -6,6 +6,7 @@
 			"machineId": "<id 1>",
 			"title": "Fork prompt",
 			"runState": "idle",
+			"parked": false,
 			"updatedAt": "<time>",
 			"directory": "<home>",
 			"project": {
@@ -58,6 +59,7 @@
 			"machineId": "<id 1>",
 			"title": "Fork prompt",
 			"runState": "idle",
+			"parked": false,
 			"updatedAt": "<time>",
 			"directory": "<home>",
 			"project": {
@@ -110,6 +112,7 @@
 			"machineId": "<id 1>",
 			"title": "Fork prompt",
 			"runState": "idle",
+			"parked": false,
 			"updatedAt": "<time>",
 			"directory": "<home>",
 			"project": {
@@ -159,6 +162,7 @@
 				"machineId": "<id 1>",
 				"title": "Fork prompt",
 				"runState": "idle",
+				"parked": false,
 				"updatedAt": "<time>",
 				"directory": "<home>",
 				"project": {
@@ -215,6 +219,7 @@
 			"lastRun": {
 				"startedAt": "<time>"
 			},
+			"parked": false,
 			"updatedAt": "<time>",
 			"directory": "<home>",
 			"project": {
@@ -267,5 +272,39 @@
 		],
 		"more": true,
 		"floor": "<seq 0>"
+	},
+	"stopped": {
+		"thread": {
+			"id": "<thr 1>",
+			"sessionId": "<ses 1>",
+			"machineId": "<id 1>",
+			"title": "gate",
+			"runState": "idle",
+			"lastRun": {
+				"startedAt": "<time>",
+				"endedAt": "<time>",
+				"outcome": "interrupted",
+				"error": {
+					"_tag": "interrupted",
+					"reason": "user"
+				}
+			},
+			"parked": true,
+			"updatedAt": "<time>",
+			"directory": "<home>",
+			"project": {
+				"id": "<id 2>",
+				"name": "Line"
+			}
+		},
+		"inbox": [
+			{
+				"id": "<id 6>",
+				"delivery": "followUp",
+				"text": "Queued",
+				"admittedSeq": 30,
+				"createdAt": "<time>"
+			}
+		]
 	}
 }

apps/server/test/__artifacts__/recovery.machines.crashloop.json

@@ -27,9 +27,8 @@
 		"session.run.failed crashloop",
 		"session.prompt.admitted steer \"next\"",
 		"session.execution.started resumed=false",
-		"session.prompt.promoted steer \"next\"",
-		"session.llm.started",
 		"session.prompt.promoted followUp \"parked by the loop\"",
+		"session.prompt.promoted steer \"next\"",
 		"session.llm.started",
 		"session.execution.succeeded"
 	],

apps/server/test/__artifacts__/recovery.machines.stop.json

@@ -5,13 +5,13 @@
 		"session.prompt.promoted steer \"hold\"",
 		"session.llm.started",
 		"session.prompt.admitted followUp \"after the stop\"",
+		"session.execution.stopping",
 		"session.llm.failed",
 		"session.execution.interrupted user",
 		"session.prompt.admitted steer \"next\"",
 		"session.execution.started resumed=false",
-		"session.prompt.promoted steer \"next\"",
-		"session.llm.started",
 		"session.prompt.promoted followUp \"after the stop\"",
+		"session.prompt.promoted steer \"next\"",
 		"session.llm.started",
 		"session.execution.succeeded"
 	],

apps/server/test/__artifacts__/schema.json

@@ -148,7 +148,7 @@
 			"type": "table",
 			"name": "thread",
 			"tbl_name": "thread",
-			"sql": "CREATE TABLE thread ( id TEXT PRIMARY KEY, organization_id TEXT NOT NULL REFERENCES organization(id), session_id TEXT NOT NULL REFERENCES session(id) ON DELETE CASCADE, origin_kind TEXT NOT NULL, origin_thread_id TEXT, origin_entry_id TEXT, origin_call_id TEXT, origin_committed_seq INTEGER, title TEXT, model_provider TEXT, model_id TEXT, thinking_level TEXT, foundation TEXT, placement_space_id TEXT, placement_dir TEXT, restarts INTEGER NOT NULL DEFAULT 0, attempt INTEGER, run_state TEXT NOT NULL DEFAULT 'idle', run_state_at INTEGER, run_started_at INTEGER, run_ended_at INTEGER, run_outcome TEXT, run_error TEXT, cost REAL NOT NULL DEFAULT 0, tokens_input INTEGER NOT NULL DEFAULT 0, tokens_output INTEGER NOT NULL DEFAULT 0, tokens_cache_read INTEGER NOT NULL DEFAULT 0, tokens_cache_write INTEGER NOT NULL DEFAULT 0, status TEXT NOT NULL, status_at INTEGER NOT NULL, priority INTEGER NOT NULL DEFAULT 0, grouping_id TEXT REFERENCES grouping(id) ON DELETE SET NULL, ordering REAL NOT NULL, color TEXT, cursor INTEGER, floor INTEGER, created_seq INTEGER NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, deleted_at INTEGER )"
+			"sql": "CREATE TABLE thread ( id TEXT PRIMARY KEY, organization_id TEXT NOT NULL REFERENCES organization(id), session_id TEXT NOT NULL REFERENCES session(id) ON DELETE CASCADE, origin_kind TEXT NOT NULL, origin_thread_id TEXT, origin_entry_id TEXT, origin_call_id TEXT, origin_committed_seq INTEGER, title TEXT, model_provider TEXT, model_id TEXT, thinking_level TEXT, foundation TEXT, placement_space_id TEXT, placement_dir TEXT, parked INTEGER NOT NULL DEFAULT 0, restarts INTEGER NOT NULL DEFAULT 0, attempt INTEGER, run_state TEXT NOT NULL DEFAULT 'idle', run_state_at INTEGER, run_started_at INTEGER, run_ended_at INTEGER, run_outcome TEXT, run_error TEXT, cost REAL NOT NULL DEFAULT 0, tokens_input INTEGER NOT NULL DEFAULT 0, tokens_output INTEGER NOT NULL DEFAULT 0, tokens_cache_read INTEGER NOT NULL DEFAULT 0, tokens_cache_write INTEGER NOT NULL DEFAULT 0, status TEXT NOT NULL, status_at INTEGER NOT NULL, priority INTEGER NOT NULL DEFAULT 0, grouping_id TEXT REFERENCES grouping(id) ON DELETE SET NULL, ordering REAL NOT NULL, color TEXT, cursor INTEGER, floor INTEGER, created_seq INTEGER NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, deleted_at INTEGER )"
 		},
 		{
 			"type": "index",

apps/server/test/__artifacts__/sync.first.json

@@ -73,6 +73,7 @@
 			"foundation": null,
 			"placement_space_id": null,
 			"placement_dir": null,
+			"parked": 0,
 			"restarts": 0,
 			"attempt": null,
 			"run_state": "idle",
@@ -115,6 +116,7 @@
 			"foundation": null,
 			"placement_space_id": null,
 			"placement_dir": null,
+			"parked": 0,
 			"restarts": 0,
 			"attempt": null,
 			"run_state": "idle",

apps/server/test/__artifacts__/sync.removed.json

@@ -64,6 +64,7 @@
 			"foundation": null,
 			"placement_space_id": null,
 			"placement_dir": null,
+			"parked": 0,
 			"restarts": 0,
 			"attempt": null,
 			"run_state": "idle",
@@ -106,6 +107,7 @@
 			"foundation": null,
 			"placement_space_id": null,
 			"placement_dir": null,
+			"parked": 0,
 			"restarts": 0,
 			"attempt": null,
 			"run_state": "idle",

apps/server/test/__artifacts__/sync.restored.json

@@ -94,6 +94,7 @@
 			"foundation": null,
 			"placement_space_id": null,
 			"placement_dir": null,
+			"parked": 0,
 			"restarts": 0,
 			"attempt": null,
 			"run_state": "idle",
@@ -136,6 +137,7 @@
 			"foundation": null,
 			"placement_space_id": null,
 			"placement_dir": null,
+			"parked": 0,
 			"restarts": 0,
 			"attempt": null,
 			"run_state": "idle",
@@ -238,6 +240,7 @@
 				"foundation": null,
 				"placement_space_id": null,
 				"placement_dir": null,
+				"parked": 0,
 				"restarts": 0,
 				"attempt": null,
 				"run_state": "idle",
@@ -280,6 +283,7 @@
 				"foundation": null,
 				"placement_space_id": null,
 				"placement_dir": null,
+				"parked": 0,
 				"restarts": 0,
 				"attempt": null,
 				"run_state": "idle",

apps/server/test/__artifacts__/sync.resume.json

@@ -55,6 +55,7 @@
 			"foundation": null,
 			"placement_space_id": null,
 			"placement_dir": null,
+			"parked": 0,
 			"restarts": 0,
 			"attempt": null,
 			"run_state": "idle",

apps/server/test/__artifacts__/sync.store.json

@@ -86,6 +86,7 @@
 			"foundation": null,
 			"placement_space_id": null,
 			"placement_dir": null,
+			"parked": 0,
 			"restarts": 0,
 			"attempt": null,
 			"run_state": "idle",
@@ -128,6 +129,7 @@
 			"foundation": null,
 			"placement_space_id": null,
 			"placement_dir": null,
+			"parked": 0,
 			"restarts": 0,
 			"attempt": null,
 			"run_state": "idle",

apps/server/test/reads.test.ts

@@ -272,6 +272,17 @@ it("a fork's line pages through inherited entries by keyset, takes its ancestor'
 	expect(queued.thread.title).toBe("gate");
 	expect(queued.inbox.map((prompt) => [prompt.text, prompt.delivery])).toEqual([["Queued", "followUp"]]);
 	expect(queued.entries.map((entry) => entry.id)).toEqual(newest.entries.map((entry) => entry.id));
+	expect(queued.thread.parked).toBe(false);
+
+	// A stop leaves the follow-up waiting for the next prompt: the thread reads parked, the prompt still queued.
+	await server.run((rpc) => rpc["thread.interrupt"](ref));
+	await waitFor(
+		() => (query(appHome, "SELECT parked FROM thread WHERE id = ?", fork.id)[0]?.parked === 1 ? true : undefined),
+		"the parked thread",
+	);
+	const parked = await line({ limit: 0 });
+	expect(parked.thread).toMatchObject({ runState: "idle", parked: true });
+	expect(parked.inbox.map((prompt) => prompt.text)).toEqual(["Queued"]);
 
 	const { storeId } = await synced(appHome, machineId, server);
 	await snapshot(
@@ -287,6 +298,7 @@ it("a fork's line pages through inherited entries by keyset, takes its ancestor'
 				budgeted: { entries: budgeted.entries.map((entry) => entry.id), more: budgeted.more },
 				paged: { ...paged, below: brief(below) },
 				queued: brief(queued),
+				stopped: { thread: parked.thread, inbox: parked.inbox },
 			}),
 		),
 	);

apps/server/test/recovery.machines.live.test.ts

@@ -96,6 +96,7 @@ const shown = new Set([
 	"session.execution.started",
 	"session.execution.succeeded",
 	"session.execution.failed",
+	"session.execution.stopping",
 	"session.execution.interrupted",
 	"session.run.failed",
 	"session.llm.started",
@@ -312,10 +313,11 @@ describeMachines("durable runs across real machine failures", () => {
 		await idle(ids);
 		await record("next prompt");
 		const ran = (await lifecycle(ids)).filter((line) => line.startsWith("session.prompt.promoted"));
+		// The parked follow-up and the new steer go in together, in the order they were sent.
 		expect(ran).toEqual([
 			'session.prompt.promoted steer "hold"',
-			'session.prompt.promoted steer "next"',
 			'session.prompt.promoted followUp "after the stop"',
+			'session.prompt.promoted steer "next"',
 		]);
 		await snapshot("stop", { log: await lifecycle(ids), steps });
 	}, 300_000);
@@ -356,7 +358,8 @@ describeMachines("durable runs across real machine failures", () => {
 		await record("next prompt");
 		expect(steps.at(-1)).toMatchObject({ restarts: 0, outcome: "succeeded" });
 		const log = await lifecycle(ids);
-		expect(count(await events(ids), "session.llm.started")).toBe(Control.RESTART_LIMIT + 2);
+		// One request per run that died, then one for the parked follow-up and the next prompt together.
+		expect(count(await events(ids), "session.llm.started")).toBe(Control.RESTART_LIMIT + 1);
 		await snapshot("crashloop", { log, steps });
 	}, 900_000);
 

apps/web/src/widgets/session/backend.ts

@@ -1,6 +1,6 @@
 import { readTopics, SyncChanged, syncTopics } from "@codeworksh/contract";
 import { Bus, Codework, defineBackend, handle, type LineThread, Sync } from "@codeworksh/widget/backend";
-import { LiveDelta, QueuedPrompt, UNTITLED } from "@codeworksh/widget/codework";
+import { LiveDelta, QueuedPrompt, stopping, UNTITLED } from "@codeworksh/widget/codework";
 import type { Event } from "@codeworksh/widget/event";
 import { Deferred, Effect, Fiber, Option, Result, Schema, Stream } from "effect";
 
@@ -43,6 +43,8 @@ const toInfo = (thread: LineThread): Info => {
 		title: thread.title ?? UNTITLED,
 		updated: thread.updatedAt,
 		runState: thread.runState,
+		stopping: stopping(thread),
+		parked: thread.parked,
 		project: thread.project,
 		...(thread.hostDir === undefined ? {} : { hostDir: thread.hostDir }),
 		...(thread.thinkingLevel === undefined ? {} : { thinkingLevel: thread.thinkingLevel }),

apps/web/src/widgets/session/queue.tsx

@@ -32,12 +32,15 @@ export const Queue = memo(function Queue({
 	thread,
 	queued,
 	running,
+	parked,
 	onError,
 	onRestore,
 }: {
 	readonly thread: ThreadRef;
 	readonly queued: readonly QueuedPrompt[];
 	readonly running: boolean;
+	/** Left waiting by a stop or a failure: they run with the next message, not by themselves. */
+	readonly parked: boolean;
 	readonly onError: (message: string | undefined) => void;
 	/** Puts a withdrawn prompt's text back in the composer. */
 	readonly onRestore: (text: string) => void;
@@ -100,7 +103,14 @@ export const Queue = memo(function Queue({
 	const many = queued.length > 4;
 	return (
 		<div className="mx-2 rounded-t-lg border border-b-0 border-ink/10 bg-(--cw-composer) px-2.5 py-0.5 text-ui leading-snug text-ink-muted backdrop-blur-xl">
-			{queued.length > 3 && <p className="py-1 font-medium text-ink-muted tabular-nums">{queued.length} queued</p>}
+			{parked ? (
+				<p className="py-1 tabular-nums">
+					<span className="font-medium">{queued.length > 3 ? `${queued.length} parked` : "Parked"}</span>
+					<span> · Runs with your next message</span>
+				</p>
+			) : (
+				queued.length > 3 && <p className="py-1 font-medium text-ink-muted tabular-nums">{queued.length} queued</p>
+			)}
 			<ol
 				aria-label="Queued messages"
 				className={`flex flex-col ${many ? "max-h-[97px] overflow-y-auto mask-[linear-gradient(to_bottom,transparent,black_6px,black_calc(100%-10px),transparent)]" : ""}`}

apps/web/src/widgets/session/schema.ts

@@ -138,6 +138,10 @@ export const Info = Schema.Struct({
 	updated: Schema.Number,
 	/** Where the thread's run is, as the harness stores it. */
 	runState: RunState,
+	/** The run is still going, but CodeWork has accepted a stop for it. */
+	stopping: Schema.Boolean,
+	/** Its queued prompts wait for the next message instead of running by themselves. */
+	parked: Schema.Boolean,
 	/**
 	 * Why the last run failed: CodeWork's failure category (`sandbox-mount-error` and the like) and
 	 * its message. Absent once another run starts, or when it did not fail.

apps/web/src/widgets/session/view.tsx

@@ -456,7 +456,12 @@ function Conversation({
 	// A prompt this window sent is pending until it lands in app.db, as a message or a queued prompt.
 	const key = threadKey(thread);
 	const inFlight = usePendingPrompts(key);
-	const { landed, forget } = useSent();
+	const { landed, forget, stop } = useSent();
+	// A stop CodeWork accepted, from any window or before a reload, shows as this window's own until the run ends.
+	const stopAccepted = loaded.status === "ready" && loaded.info.stopping;
+	useEffect(() => {
+		if (stopAccepted) stop(key);
+	}, [stopAccepted, key, stop]);
 	useEffect(() => {
 		if (loaded.status !== "ready" || inFlight.size === 0) return;
 		const known = new Set([...loaded.messages, ...loaded.queued].map((item) => item.id));
@@ -593,6 +598,7 @@ function Conversation({
 						thread={thread}
 						queued={view.queued}
 						running={running}
+						parked={view.info.parked}
 						onError={setSendError}
 						onRestore={restore}
 					/>

apps/widget/src/backend.ts

@@ -81,6 +81,8 @@ export const LineThread = Schema.Struct({
 	thinkingLevel: Schema.optionalKey(ThinkingLevel),
 	runState: RunState,
 	lastRun: Schema.optionalKey(ThreadRun),
+	/** Its queued prompts wait for its next prompt instead of running by themselves. */
+	parked: Schema.Boolean,
 	/** Epoch milliseconds. */
 	updatedAt: Schema.Finite,
 	directory: Schema.String,