Keep a run live, and its questions asked, when its machine resumes after a server restart

apps/server/src/chat/service.ts

@@ -2124,7 +2124,7 @@ const storedStages = new WeakMap<Chat, Map<string, { ids: string[]; json: Map<st
  * than its card carries. A report that only adds and changes stages writes those; one that drops
  * or reorders stages replaces the run's.
  */
-function storeStages(context: Room, cards: Wire.Run[]): Promise<void> | undefined {
+function storeStages(context: RunsRoom, cards: Wire.Run[]): Promise<void> | undefined {
 	let known = storedStages.get(context.chat)
 		?? new Map<string, { ids: string[]; json: Map<string, string> }>();
 	storedStages.set(context.chat, known);
@@ -2173,7 +2173,7 @@ function cardWindow(run: Wire.Run): Wire.Run {
 }
 
 /** Forgets the stored stages of runs Chat no longer shows. */
-async function forgetStages(context: Room, shown: Wire.Run[] | undefined): Promise<void> {
+async function forgetStages(context: RunsRoom, shown: Wire.Run[] | undefined): Promise<void> {
 	let ids = (shown ?? []).map(run => run.id);
 	let known = storedStages.get(context.chat);
 	for (let id of known?.keys() ?? []) if (!ids.includes(id)) known!.delete(id);
@@ -2185,26 +2185,65 @@ async function forgetStages(context: Room, shown: Wire.Run[] | undefined): Promi
 }
 
 const publishing = new WeakMap<Chat, Promise<void>>();
-const owners = new WeakMap<Chat, Map<string, PlannerSession>>();
-const unsaved = new WeakMap<PlannerSession, Runs>();
-const gone = new WeakSet<PlannerSession>();
+const owners = new WeakMap<Chat, Map<string, RunsOwner>>();
+const unsaved = new WeakMap<RunsOwner, Runs>();
+const gone = new WeakSet<RunsOwner>();
 
 type Runs = { active: string[]; paused: string[]; cards?: Wire.Run[] };
 
+/** The session whose runs a card shows: a retained Planner session, or one a restart rebuilt. */
+type RunsOwner = Pick<PlannerSession, "suspended">;
+
+/** What showing and storing a session's run cards needs of the room. */
+type RunsRoom = Pick<Room, "chat" | "room" | "server" | "persist"> & {
+	auth: { storage: Pick<Room["auth"]["storage"], "runStages"> };
+};
+
+/** Stand-ins, by session id, for the sessions a restart rebuilt whose clients report runs. */
+const rebuiltOwners = new WeakMap<Chat, Map<string, { owner: RunsOwner; down: boolean }>>();
+
+/**
+ * Shows the runs reported by a Planner session a restart rebuilt and its client took back, `down`
+ * while its client's link has dropped again. No turn of this process opened that session, so
+ * nothing here retains it and its reports reach Chat only this way, as a retained session's do.
+ */
+export function rebuiltRuns(
+	context: RunsRoom,
+	session: string,
+	runs: Runs,
+	down = false,
+): Promise<void> {
+	if (context.chat.closed) return Promise.resolve();
+	let known = rebuiltOwners.get(context.chat)
+		?? new Map<string, { owner: RunsOwner; down: boolean }>();
+	rebuiltOwners.set(context.chat, known);
+	let entry = known.get(session);
+	if (!entry) {
+		let made: { owner: RunsOwner; down: boolean } = {
+			owner: { suspended: () => made.down },
+			down,
+		};
+		entry = made;
+		known.set(session, entry);
+	}
+	entry.down = down;
+	return publishRuns(context, entry.owner, runs);
+}
+
 /**
  * Show the session's runs, keeping ended ones as summary cards until the next
  * run starts, store them with the document so a reload keeps them, and say once
  * in the transcript when one ends. A card belongs to the session that reported
  * it: a report or release only changes its own session's cards.
  */
 function publishRuns(
-	context: Room,
-	session: PlannerSession,
+	context: RunsRoom,
+	session: RunsOwner,
 	runs: Runs | undefined,
 	releasing = false,
-	followUp = false,
+	followUp?: (run: Wire.Run, said: string) => Promise<void>,
 ): Promise<void> {
-	let known = owners.get(context.chat) ?? new Map<string, PlannerSession>();
+	let known = owners.get(context.chat) ?? new Map<string, RunsOwner>();
 	owners.set(context.chat, known);
 	for (let run of runs?.cards ?? []) known.set(run.id, session);
 	let turn = (publishing.get(context.chat) ?? Promise.resolve()).then(() =>
@@ -2215,11 +2254,11 @@ function publishRuns(
 }
 
 async function publishReport(
-	context: Room,
-	session: PlannerSession,
+	context: RunsRoom,
+	session: RunsOwner,
 	runs: Runs | undefined,
 	releasing: boolean,
-	followUp: boolean,
+	followUp: ((run: Wire.Run, said: string) => Promise<void>) | undefined,
 ): Promise<void> {
 	let { chat, room, server } = context;
 	let at = now();
@@ -2301,7 +2340,7 @@ async function publishReport(
 	for (let entry of finished) announce(server, room, entry);
 	state(chat, server, room);
 	if (prior.some(run => !shown.has(run.id))) await forgetStages(context, next);
-	for (let { run, said } of attention) await followUpEndedRun(context, run, said);
+	for (let { run, said } of attention) await followUp?.(run, said);
 }
 
 /**
@@ -2371,7 +2410,13 @@ function retain(
 			if (chat.retained?.session !== opened.session) return;
 			let planning = chat.busy && !chat.job && chat.agent === opened.session;
 			let followUp = !planning && opened.session.followsUpRuns !== false;
-			void publishRuns(context, opened.session, next, false, followUp);
+			void publishRuns(
+				context,
+				opened.session,
+				next,
+				false,
+				followUp ? (run, said) => followUpEndedRun(context, run, said) : undefined,
+			);
 			if (next.active.length || next.paused.length) return;
 			release();
 		});

apps/server/src/harness/atomic/human-input.ts

@@ -227,7 +227,7 @@ export type HumanInput = {
  * cannot tell apart.
  */
 export function createHumanInput(
-	room: DocumentRoom,
+	room: Pick<DocumentRoom, "id" | "plan" | "server" | "anchors" | "focusedChild">,
 	hold?: (workflowRunId: string, signal: AbortSignal) => Promise<void>,
 	report?: (status: HumanInputStatus) => void,
 	pending?: PendingInputs,

apps/server/src/harness/remote/attachments.ts

@@ -110,6 +110,11 @@ export type PlannerAttachmentOptions = {
 	restoredRuns?: (channelId: string, outcome: RestoredRuns) => void;
 	/** Heard when a hand-over window ends without the client back, with what was missing. */
 	missed?: (channelId: string, cause: "login" | "client") => void;
+	/**
+	 * Opens a document, and keeps it open until the returned function is called, for a question
+	 * the client of a hold a restart rebuilt asks while nobody has the document open.
+	 */
+	openDocument?: (channelId: string) => Promise<() => void>;
 };
 
 /** What a document's members are told about its Planner: only the holder's login and state. */
@@ -574,10 +579,12 @@ export class PlannerAttachments {
 				continue;
 			}
 			if (row.sessions.length > 0) {
+				let open = this.#options.openDocument;
 				hold.parked = AgentLink.rebuilt(row.channelId, row.sessions, {
 					lapse: () => this.#lapsed(hold, hold.parked!),
 					ended: () => this.#runs(hold, "ended"),
 					resumed: () => this.#runs(hold, "up"),
+					...(open ? { open: () => open(row.channelId) } : {}),
 				});
 			}
 			let length = row.handoverUntil
@@ -621,14 +628,21 @@ export class PlannerAttachments {
 	}
 
 	/**
-	 * Whether a hold a restart rebuilt still keeps sessions its client has not taken back, so the
-	 * workflow runs Chat stored for the document are live and not stopped when its Chat opens:
-	 * `live` through the hold's window, `down` once its client is known to be away.
+	 * Whether a hold a restart rebuilt still keeps sessions with work, so the workflow runs Chat
+	 * stored for the document are live and not stopped when its Chat opens: `live` through the
+	 * hold's window and once its client has taken them back, `down` while its client is known to be
+	 * away. A client that took them back keeps them on its link, where no turn of this process
+	 * retains them, so a document that opens after it resumed still shows its runs live.
 	 */
 	restoredWork(channelId: string): "live" | "down" | undefined {
 		let hold = this.#holds.get(channelId);
-		if (!hold?.restored || !hold.parked || hold.parked.snapshot().length === 0) return undefined;
-		return this.#inWindow(hold) ? "live" : "down";
+		if (!hold) return undefined;
+		if (hold.restored && hold.parked && hold.parked.snapshot().length > 0) {
+			return this.#inWindow(hold) ? "live" : "down";
+		}
+		if (hold.link?.keepsRebuilt) return "live";
+		if (hold.parked?.keepsRebuilt) return "down";
+		return undefined;
 	}
 
 	/**

apps/server/src/harness/remote/handover-window.test.ts

@@ -1,6 +1,8 @@
 import { afterEach, describe, expect, it, spyOn } from "bun:test";
 
 import { Sessions } from "../../auth/session";
+import * as Chat from "../../chat/service";
+import * as Plan from "../../plan/service";
 import * as Store from "../../questions/store";
 import { hostInputRoom } from "../../testing/decisions";
 import { expectPlainStatus } from "../../testing/plain-words";
@@ -16,8 +18,9 @@ import { untilUnpaused } from "../atomic/full";
 import { createRemoteAdapter } from "./adapter";
 import { PlannerAttachments } from "./attachments";
 
+import type { DocumentRoom } from "../../agent/tools";
 import type { Lease } from "../../storage/model";
-import type { PlannerAttachment } from "./attachments";
+import type { PlannerAttachment, PlannerAttachmentOptions } from "./attachments";
 import type { RemotePlanner } from "./registry";
 
 const OFFER = "resumable-sessions, restart-notice";
@@ -119,7 +122,7 @@ async function deploy(input: Record<string, unknown> = ASK_CONFIRM, asked = 1) {
 	async function cut(
 		windowMs = WINDOW_MS,
 		spent: { beforeWrite?: number; beforeRelease?: number; beforeTakeover?: number } = {},
-		takeover: { failWrites?: boolean } = {},
+		takeover: { failWrites?: boolean; options?: PlannerAttachmentOptions } = {},
 	) {
 		first.attachments.shutdown();
 		let rows = first.attachments.freeze(windowMs);
@@ -144,6 +147,7 @@ async function deploy(input: Record<string, unknown> = ASK_CONFIRM, asked = 1) {
 			},
 			restoredRuns: (id, outcome) => events.push([`runs:${outcome}`, id]),
 			missed: (id, cause) => events.push([`missed:${cause}`, id]),
+			...takeover.options,
 		});
 		let changes: PlannerAttachment[] = [];
 		attachments.onChange((_id, status) => changes.push(status));
@@ -189,24 +193,47 @@ async function deploy(input: Record<string, unknown> = ASK_CONFIRM, asked = 1) {
 
 /**
  * The second process opens the document again from what the first stored, as the first request
- * for it does, so the questions its rebuilt hold's client left open are asked of the members again.
+ * for it does, so the questions its rebuilt hold's client left open are asked of the members again,
+ * and Chat shows what the sessions it rebuilt report and ask, as `main.ts` wires them.
  */
 async function reopen(
 	d: Awaited<ReturnType<typeof deploy>>,
 	next: Awaited<ReturnType<Awaited<ReturnType<typeof deploy>>["cut"]>>,
 ) {
-	await d.f.close();
+	if (!d.closed.first) await d.f.close();
 	d.closed.first = true;
 	let unsent: string[] = [];
 	let g = await hostInputRoom(undefined, {
 		storage: d.f.storage,
 		channelId: d.f.channelId,
 		lease: d.f.lease,
 		backend: {
+			// The fixture's document and the link's have their own ids: this is the link's.
+			plannerKeepsRuns: () => next.attachments.restoredWork(d.channel.id),
 			rejoinPlannerQuestions: (plan, server) =>
 				next.attachments.rejoinInputs(d.channel.id, {
 					join: (asked, signal) => rejoinInput({ id: plan.id, plan, server }, asked, signal),
 					unsent: decisions => unsent.push(...decisions),
+					live: () => !plan.persistence.closing,
+					humanInput: (hold, report) =>
+						createHumanInput(
+							{ id: plan.id, plan, server, anchors() {} } as DocumentRoom,
+							hold,
+							report,
+						),
+					runs: (session, runs, down) =>
+						void Chat.rebuiltRuns(
+							{
+								chat: plan.chat,
+								room: plan.id,
+								server,
+								persist: chat => Plan.persist(plan, chat),
+								auth: { storage: d.f.storage },
+							},
+							session,
+							runs,
+							down,
+						),
 				}),
 		},
 	});
@@ -783,3 +810,115 @@ describe("a remote Planner whose link a cut closed", () => {
 		expect(rows).toEqual([]);
 	});
 });
+
+describe("a workflow run its client carries on through a restart", () => {
+	const TRUST = {
+		method: "confirm",
+		title: "Trust the toolchain?",
+		message: "Its checks pass.",
+	} as const;
+
+	/** The first process's Chat shows the run as its retained session reported it, and stores it. */
+	async function shown(d: Awaited<ReturnType<typeof deploy>>) {
+		d.f.plan.chat.runs = [run];
+		await Plan.persist(d.f.plan, () => ({
+			entries: d.f.plan.chat.entries,
+			runs: [run],
+			initiativePaused: false,
+		}));
+	}
+
+	/** The client is back a second after the restart, before any browser opens the document. */
+	async function resumed(
+		d: Awaited<ReturnType<typeof deploy>>,
+		next: Awaited<ReturnType<Awaited<ReturnType<typeof deploy>>["cut"]>>,
+	) {
+		let client = await returns(d, next);
+		client.send({ type: "resume", sessions: [{ session: d.sessionId, requests: ["ask-1"] }] });
+		await client.next("resumed");
+		client.send({
+			type: "runs",
+			session: d.sessionId,
+			active: ["run"],
+			paused: [],
+			cards: [{
+				...run,
+				updated: 2,
+				stages: [{ id: "verify", name: "verify", status: "running" as const }],
+			}],
+		});
+		return client;
+	}
+
+	it("is shown running, not stopped, when its document opens after the client is back, with the stages it reports", async () => {
+		let d = await deploy();
+		await shown(d);
+		let next = await d.cut();
+		await resumed(d, next);
+		await Bun.sleep(50);
+		let { g } = await reopen(d, next);
+		expect(g.plan.chat.runs?.map(card => card.status)).toEqual(["running"]);
+		await until(() => g.plan.chat.runs?.[0]?.stages.some(stage => stage.id === "verify") === true);
+		expect(g.plan.chat.runs).toMatchObject([{ id: "run", status: "running", updated: 2 }]);
+		expect(g.plan.chat.runs?.[0]?.disconnected).toBeUndefined();
+	});
+
+	it("asks the question it asks after the restart as a Decision, and gives the client the answer", async () => {
+		let d = await deploy();
+		await shown(d);
+		let next = await d.cut();
+		let client = await resumed(d, next);
+		let { g } = await reopen(d, next);
+		client.send({
+			type: "input-request",
+			request: "ask-2",
+			session: d.sessionId,
+			workflowRunId: "run",
+			input: TRUST,
+		});
+		let asked = (await g.cards(2)).find(card => card.id !== d.card.id);
+		expect(asked!.definition.questions[0]!.question).toContain("Trust the toolchain?");
+		await g.answer(asked!.id, [0]);
+		expect(await client.next("input-result")).toMatchObject({
+			request: "ask-2",
+			status: "answered",
+			value: true,
+		});
+	});
+
+	it("opens its document to ask the question when nobody has it open, and keeps it open until the answer", async () => {
+		let d = await deploy();
+		await shown(d);
+		await d.f.close();
+		d.closed.first = true;
+		let opened: Array<Awaited<ReturnType<typeof reopen>>> = [];
+		let released = 0;
+		let next: Awaited<ReturnType<typeof d.cut>> = await d.cut(WINDOW_MS, {}, {
+			options: {
+				openDocument: async () => {
+					opened.push(await reopen(d, next));
+					return () => void released++;
+				},
+			},
+		});
+		let client = await resumed(d, next);
+		client.send({
+			type: "input-request",
+			request: "ask-2",
+			session: d.sessionId,
+			workflowRunId: "run",
+			input: TRUST,
+		});
+		await until(() => opened.length === 1);
+		let { g } = opened[0]!;
+		let asked = (await g.cards(2)).find(card => card.id !== d.card.id);
+		expect(released).toBe(0);
+		await g.answer(asked!.id, [0]);
+		expect(await client.next("input-result")).toMatchObject({
+			request: "ask-2",
+			status: "answered",
+			value: true,
+		});
+		await until(() => released === 1);
+	});
+});