Keep the Planner while its workflows run, with Stop/Resume and a run card

apps/server/src/agent/planner.ts

@@ -249,11 +249,18 @@ and cannot change GitHub. Ground the plan in what those reading tools return.`;
 		? `Your working directory, ${workspace.cwd}, is a local checkout of ${repository}
 verified against its origin. Its branch and working tree may differ from what the
 repository tools read.`
-		: `Your working directory, ${workspace.cwd}, is a scratch directory Chopin created
-empty for this document. It is not a checkout and holds no repository files, so read
+		: `Your working directory, ${workspace.cwd}, is a scratch directory Chopin keeps for
+this document. It is not a checkout and holds no repository files, so read
 ${repository} through the repository tools.`;
 	let questions =
 		`\`ask_user_question\` and \`workflow\` questions appear to the document's members as
 Decisions. If one expires unanswered, proceed on your best judgement and say what you assumed.`;
-	return [PROMPT, reading, place, questions, bootstrap].filter(Boolean).join("\n\n");
+	let surface = `The document's members use Chopin in a browser. They cannot run slash commands,
+terminal commands, or Atomic CLI commands, so never tell them to use \`/workflow connect\`,
+\`/workflow status\`, \`/tasks\`, \`atomic\`, or anything else typed into a terminal, even when a
+tool result suggests it. When you start a workflow, say in plain words what it will do and
+where to follow it: Chat shows a card for each run with its stages and status, its questions
+appear under Decisions, **Stop Planner** pauses it, and **Resume Planner** resumes it. Offer to
+check on or steer a run yourself with your \`workflow\` and \`intercom\` tools when someone asks.`;
+	return [PROMPT, reading, place, questions, surface, bootstrap].filter(Boolean).join("\n\n");
 }

apps/server/src/chat/invoke.test.ts

@@ -1,4 +1,4 @@
-import { afterEach, expect, test } from "bun:test";
+import { afterEach, beforeEach, expect, test } from "bun:test";
 import { mkdtemp, readdir, realpath, rm } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
@@ -7,14 +7,16 @@ import { ActiveOwnerBindings } from "../agent/active-owner";
 import { Admission } from "../auth/admission";
 import { Sessions } from "../auth/session";
 import { fullPlanner } from "../harness/atomic/full";
-import { removeWorkspaces } from "../harness/atomic/workspace";
+import { forgetWorkspaces } from "../harness/atomic/workspace";
 import { openPlannerSession } from "../harness/session";
 import * as Plan from "../plan/service";
 import { MemoryStorage } from "../storage/memory/adapter";
 import { configured, LOCAL } from "../testing/config";
 import * as Chat from "./service";
 
 import type { Server } from "bun";
+import type { ActiveOwnerBinding } from "../agent/active-owner";
+import type { Chat as Wire } from "@chopin/protocol";
 import type { HostedAuth } from "../auth/routes";
 import type { Config } from "../config";
 import type { GitHub } from "../github/client";
@@ -23,9 +25,17 @@ import type { Socket, SocketData } from "../wire";
 const ATOMIC = { HARNESS: "atomic", HARNESS_AUTH: "ai-gateway", MODEL: "stub/model" };
 
 let cleanups: Array<() => Promise<void>> = [];
+let previousState = process.env.XDG_STATE_HOME;
+beforeEach(async () => {
+	let state = await mkdtemp(join(tmpdir(), "chopin-invoke-state-"));
+	process.env.XDG_STATE_HOME = state;
+	cleanups.push(() => rm(state, { recursive: true, force: true }));
+});
 afterEach(async () => {
 	for (let cleanup of cleanups.splice(0)) await cleanup();
-	await removeWorkspaces();
+	forgetWorkspaces();
+	if (previousState === undefined) delete process.env.XDG_STATE_HOME;
+	else process.env.XDG_STATE_HOME = previousState;
 });
 
 function grant(accessToken: string) {
@@ -409,3 +419,208 @@ test("HARNESS=atomic runs every Planner session full in hosted and local configu
 		}
 	}
 });
+
+type Runs = { active: string[]; paused: string[]; cards: Wire.Run[] };
+
+function card(status: Wire.Run["status"]): Wire.Run {
+	let ended = status === "finished" || status === "failed" || status === "stopped";
+	return {
+		id: "run-1",
+		name: "plan-review",
+		status,
+		started: 1_000,
+		updated: 1_600,
+		...(ended ? { ended: 1_600 } : {}),
+		stages: [{
+			id: "run-1:draft",
+			name: "draft-1",
+			status: status === "waiting" ? "awaiting_input" : ended ? "completed" : "running",
+			started: 1_000,
+		}],
+		waiting: status === "waiting" ? 1 : 0,
+	};
+}
+
+/** A Planner session that owns workflow runs, driven by the test. */
+function runningSession() {
+	let runs: Runs = { active: ["run-1"], paused: [], cards: [card("running")] };
+	let listeners = new Set<(runs: Runs) => void>();
+	let calls: string[] = [];
+	let session = {
+		async stream(prompt: string) {
+			calls.push(`stream ${prompt}`);
+			return {
+				fullStream: (async function*() {
+					yield { type: "finish" };
+				})(),
+			} as never;
+		},
+		async destroy() {
+			calls.push("destroy");
+		},
+		runs: () => runs,
+		watchRuns(listener: (runs: Runs) => void) {
+			listeners.add(listener);
+			return () => listeners.delete(listener);
+		},
+		async pauseRuns() {
+			calls.push("pause");
+		},
+		async resumeRuns() {
+			calls.push("resume");
+		},
+		async pauseRun(runId: string) {
+			calls.push(`pause ${runId}`);
+		},
+		async resumeRun(runId: string) {
+			calls.push(`resume ${runId}`);
+		},
+	};
+	return {
+		session,
+		calls,
+		set(next: Runs) {
+			runs = next;
+			for (let listener of listeners) listener(next);
+		},
+	};
+}
+
+test("a Planner that still owns workflow runs outlives its turn, pauses, resumes, and is let go when they finish", async () => {
+	let { context, events, user } = await setup(configured(ATOMIC));
+	let planner = runningSession();
+	let opened = 0;
+	context.openPlannerSession = async () => {
+		opened++;
+		return { ok: true, value: planner.session };
+	};
+	let holds = 0;
+	context.hold = () => {
+		holds++;
+		return () => holds--;
+	};
+	let ws = { data: { handle: "ana" } } as unknown as Socket;
+	let said = () => context.chat.entries.at(-1)?.text;
+	let runs = () => (events.filter(event => event.kind === "chat:state").at(-1)?.runs as
+		| Wire.Run[]
+		| undefined);
+
+	expect(await Chat.invoke(context, user, "Run the workflow")).toBeUndefined();
+	await context.chat.running;
+	expect(planner.calls).toEqual(["stream @ana: Run the workflow"]);
+	expect(context.chat.busy).toBe(false);
+	expect(context.chat.runs).toEqual([card("running")]);
+	expect(runs()).toEqual([card("running")]);
+	expect(holds).toBe(1);
+
+	await Chat.abort(context, ws);
+	expect(planner.calls.at(-1)).toBe("pause");
+	expect(said()).toBe("@ana stopped the Planner and paused its workflows.");
+	planner.set({ active: [], paused: ["run-1"], cards: [card("paused")] });
+	expect(runs()?.map(run => run.status)).toEqual(["paused"]);
+	expect(planner.calls).not.toContain("destroy");
+
+	await Chat.resume(context, ws);
+	expect(planner.calls.at(-1)).toBe("resume");
+	expect(said()).toBe("@ana resumed the Planner's workflows.");
+	planner.set({ active: ["run-1"], paused: [], cards: [card("waiting")] });
+	expect(runs()?.[0]).toMatchObject({
+		status: "waiting",
+		waiting: 1,
+		stages: [{ status: "awaiting_input" }],
+	});
+
+	expect(await Chat.invoke(context, user, "How is it going?")).toBeUndefined();
+	await context.chat.running;
+	expect(opened).toBe(1);
+	expect(planner.calls.filter(call => call.startsWith("stream"))).toHaveLength(2);
+	expect(planner.calls).not.toContain("destroy");
+
+	planner.set({ active: [], paused: [], cards: [card("finished")] });
+	expect(context.chat.entries.some(entry => entry.text === "plan-review finished after 10 min."))
+		.toBe(true);
+	await new Promise(resolve => setTimeout(resolve, 0));
+	expect(planner.calls.at(-1)).toBe("destroy");
+	expect(context.chat.retained).toBeUndefined();
+	expect(context.chat.runs).toEqual([card("finished")]);
+	expect(runs()?.map(run => run.status)).toEqual(["finished"]);
+	expect(holds).toBe(0);
+
+	await Chat.resume(context, ws);
+	await Chat.abort(context, ws);
+	expect(planner.calls.at(-1)).toBe("destroy");
+});
+
+test("a retained Planner is let go when its owner's binding ends, and sessions without runs still end with their turn", async () => {
+	let { context, user } = await setup(configured(ATOMIC));
+	let planner = runningSession();
+	let bindings: ActiveOwnerBinding[] = [];
+	let resolve = context.activeOwner!;
+	context.activeOwner = async () => {
+		let binding = await resolve();
+		if (binding) bindings.push(binding);
+		return binding;
+	};
+	context.openPlannerSession = async () => ({ ok: true, value: planner.session });
+	expect(await Chat.invoke(context, user, "Run the workflow")).toBeUndefined();
+	await context.chat.running;
+	expect(context.chat.retained).toBeDefined();
+	bindings[0]!.release();
+	await new Promise(resolve => setTimeout(resolve, 0));
+	expect(context.chat.retained).toBeUndefined();
+	expect(planner.calls.at(-1)).toBe("destroy");
+
+	let quiet = {
+		...runningSession().session,
+		runs: (): Runs => ({ active: [], paused: [], cards: [] }),
+	};
+	let destroyed = 0;
+	quiet.destroy = async () => {
+		destroyed++;
+	};
+	context.openPlannerSession = async () => ({ ok: true, value: quiet });
+	expect(await Chat.invoke(context, user, "Just answer")).toBeUndefined();
+	await context.chat.running;
+	expect(destroyed).toBe(1);
+	expect(context.chat.retained).toBeUndefined();
+});
+
+test("run cards keep ended runs until a new run starts, and a released session's live runs become stopped", () => {
+	let live = card("running");
+	let done = { ...card("finished"), id: "run-0" };
+	expect(Chat.mergeRuns([done], [live], 2_000)).toEqual([live]);
+	expect(Chat.mergeRuns([done, live], [card("waiting")], 2_000)).toEqual([done, card("waiting")]);
+	let released = Chat.mergeRuns([done, card("waiting")], undefined, 2_000);
+	expect(released.map(run => [run.id, run.status, run.waiting])).toEqual([
+		["run-0", "finished", 0],
+		["run-1", "stopped", 0],
+	]);
+	expect(released[1]?.ended).toBe(2_000);
+	expect(Chat.mergeRuns([{ ...card("paused"), ended: undefined }], undefined, 2_000)[0])
+		.toMatchObject({ status: "stopped", ended: 2_000 });
+	expect(Chat.restoreRuns([])).toBeUndefined();
+});
+
+test("one run can be paused and resumed by name while others keep going", async () => {
+	let { context, user } = await setup(configured(ATOMIC));
+	let planner = runningSession();
+	context.openPlannerSession = async () => ({ ok: true, value: planner.session });
+	let ws = { data: { handle: "ana" } } as unknown as Socket;
+	expect(await Chat.invoke(context, user, "Run the workflow")).toBeUndefined();
+	await context.chat.running;
+
+	await Chat.controlRun(context, ws, { kind: "chat:resume-run", runId: "run-1" });
+	await Chat.controlRun(context, ws, { kind: "chat:pause-run", runId: "unknown" });
+	await Chat.controlRun(context, ws, { kind: "chat:pause-run", runId: 7 });
+	expect(planner.calls.filter(call => call.includes("run-1") || call.includes("unknown"))).toEqual(
+		[],
+	);
+
+	await Chat.controlRun(context, ws, { kind: "chat:pause-run", runId: "run-1" });
+	expect(planner.calls.at(-1)).toBe("pause run-1");
+	expect(context.chat.entries.at(-1)?.text).toBe("@ana paused plan-review.");
+	planner.set({ active: [], paused: ["run-1"], cards: [card("paused")] });
+	await Chat.controlRun(context, ws, { kind: "chat:resume-run", runId: "run-1" });
+	expect(planner.calls.at(-1)).toBe("resume run-1");
+	expect(context.chat.entries.at(-1)?.text).toBe("@ana resumed plan-review.");
+});