Fit wide diagrams, date builds from their finish, follow renames, and place Planner questions

apps/connector/src/planner.test.ts

@@ -1129,6 +1129,87 @@ test.skipIf(process.platform === "win32")(
 	40_000,
 );
 
+test.skipIf(process.platform === "win32")(
+	"a dispatched Planner whose document was renamed goes by its new address, in status, output and its next start",
+	async () => {
+		let chopin = Bun.serve({
+			hostname: "127.0.0.1",
+			port: 0,
+			fetch: async request =>
+				new URL(request.url).pathname === "/api/connector/planner"
+					? Response.json({
+						token: `chopin_p_${"P".repeat(43)}`,
+						expiresAt: Date.now() + 15 * 60_000,
+						document: {
+							id: "D_1",
+							repository: "octo-org/score",
+							url: ((await request.json()) as { document: string }).document,
+						},
+					})
+					: new Response(null, { status: 404 }),
+		});
+		let directory = await mkdtemp(join(scratch, "renamed-"));
+		let script = join(directory, "backend.ts");
+		let log = join(directory, "launches.jsonl");
+		await writeFile(
+			script,
+			`import { appendFileSync } from "node:fs";
+appendFileSync(process.argv[2], JSON.stringify({ pid: process.pid, document: process.env.CHOPIN_DOCUMENT }) + "\\n");
+process.on("SIGTERM", () => process.exit(0));
+setTimeout(() => process.exit(0), 1_000);
+`,
+		);
+		let abort = new AbortController();
+		let said = "";
+		let sink = {
+			write: (text: string) => {
+				said += text;
+				return true;
+			},
+		};
+		let recorded: string[][] = [];
+		let planners = dispatchedPlanners({
+			url: `http://127.0.0.1:${chopin.port}`,
+			machine: MACHINE,
+			documents: [],
+			command: [process.execPath, script, log],
+			checkouts: checkouts({ directory, registered: { "octo-org/score": directory } }),
+			directory,
+			output: createOutput(sink, sink, {}),
+			signal: abort.signal,
+			chosen: {
+				record: async documents => {
+					recorded.push(documents);
+				},
+			},
+		});
+		let launched = async () => (await entries(log)).filter(entry => entry.pid);
+		try {
+			planners.assign([{ id: "D_1", url: "http://x/documents/o/r/old-name" }]);
+			await until(async () => (await launched()).length === 1, 10_000);
+			expect(said).toContain("Planner for http://x/documents/o/r/old-name started.");
+			// Renamed while it runs: Chopin's next answer names the same document at its new address.
+			planners.assign([{ id: "D_1", url: "http://x/documents/o/r/new-name" }]);
+			expect(recorded.at(-1)).toEqual(["http://x/documents/o/r/new-name"]);
+			expect(planners.running()).toEqual(["D_1"]);
+			// It is not restarted for the rename; once it exits by itself, it starts again at the new one.
+			await until(async () => planners.running().length === 0, 10_000);
+			planners.assign([{ id: "D_1", url: "http://x/documents/o/r/new-name" }]);
+			await until(async () => (await launched()).length === 2, 10_000);
+			expect(said).toContain("Planner for http://x/documents/o/r/new-name exited (exit 0)");
+			expect((await launched()).map(entry => entry.document)).toEqual([
+				"http://x/documents/o/r/old-name",
+				"http://x/documents/o/r/new-name",
+			]);
+		} finally {
+			abort.abort();
+			await planners.stop();
+			await chopin.stop(true);
+		}
+	},
+	40_000,
+);
+
 test.skipIf(process.platform === "win32")(
 	"one machine runs the Planners of documents in several repositories, each in its repository's checkout",
 	async () => {

apps/connector/src/planner.ts

@@ -149,8 +149,14 @@ export function dispatchedPlanners(options: {
 		assign(planners: PlannerAssignment[], stopped?: Record<string, string>) {
 			let wanted = new Map(planners.map(value => [value.id, value]));
 			for (let [id, backend] of running) {
-				if (wanted.has(id)) backend.exited?.(true);
-				else withdraw(id, backend, stopped?.[id]);
+				let planner = wanted.get(id);
+				if (!planner) {
+					withdraw(id, backend, stopped?.[id]);
+					continue;
+				}
+				// A renamed document keeps its id; Chopin names its new address.
+				if (!backend.own) backend.url = planner.url;
+				backend.exited?.(true);
 			}
 			for (let [id, planner] of wanted) {
 				if (running.has(id) || options.signal.aborted) continue;
@@ -162,6 +168,7 @@ export function dispatchedPlanners(options: {
 					backend.served = servePlanner({
 						...options,
 						document: planner.url,
+						address: () => backend.url,
 						id: planner.id,
 						signal,
 						again: () =>
@@ -211,6 +218,8 @@ async function servePlanner(options: {
 	document: string;
 	/** The document's id in Chopin, when known before its credential names it. */
 	id?: string;
+	/** The document's current address, when Chopin can name a new one as it is renamed. */
+	address?: () => string;
 	command: string[];
 	checkouts: Checkouts;
 	directory: string;
@@ -229,10 +238,12 @@ async function servePlanner(options: {
 	 */
 	again?: () => Promise<boolean>;
 }) {
-	let { output, signal, document } = options;
+	let { output, signal } = options;
 	let healthyMs = options.healthyMs ?? HEALTHY_MS;
 	let values = Object.values(options.env ?? {});
+	let current = () => options.address?.() ?? options.document;
 	for (let attempt = 0; !signal.aborted;) {
+		let document = current();
 		let started = Date.now();
 		let credential;
 		try {
@@ -294,14 +305,21 @@ async function servePlanner(options: {
 		if (checkout) {
 			output.say(`› Planner for ${document} started.`);
 			let healthy = setTimeout(() => void options.failures?.started(document), healthyMs);
-			outcome = await launch({ ...options, cwd: checkout }, credential.token, worktree, github);
+			outcome = await launch(
+				{ ...options, document, cwd: checkout },
+				credential.token,
+				worktree,
+				github,
+			);
 			clearTimeout(healthy);
 		} else {
 			output.error(`Can't start the Planner for ${document}: ${outcome!.how}`);
 		}
 		let { how, stderr } = outcome!;
 		await github?.stop();
 		if (signal.aborted) return;
+		// Renamed while it ran: say so by its new address.
+		document = current();
 		// What the backend printed may hold its credential, a GitHub token or an --env value.
 		let secrets = [credential.token, ...github?.secrets() ?? [], ...values];
 		let ended = Date.now();
@@ -319,6 +337,7 @@ async function servePlanner(options: {
 			// Status shows a failed start while Chopin is asked whether it still wants this Planner.
 			if (early) await failed(ended);
 			if (!(await options.again())) return;
+			document = current();
 		}
 		if (ended - started >= HEALTHY_MS) attempt = 0;
 		let wait = backoff(attempt++);

apps/server/src/experiments/routes.ts

@@ -393,6 +393,11 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio
 	let dispatched = new Map<string, Map<string, string>>();
 	/** Why each machine stopped running a document's Planner, told to its connector once. */
 	let stopReasons = new Map<string, Map<string, string>>();
+	/**
+	 * The address each connection was last told for each document whose Planner it runs, by
+	 * connection id, so a rename reaches the connector: its Planner poll names documents by id.
+	 */
+	let toldUrls = new Map<string, Map<string, string>>();
 	/**
 	 * Whether a connection can run its owner's Planners on request: it has a Planner command, speaks
 	 * protocol 4, and this server runs the remote Planner with machines' own GitHub authorization.
@@ -635,6 +640,7 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio
 	 * machine's other connections able to run Planners serves, and Default machine's move on.
 	 */
 	connections.onEnded = connection => {
+		toldUrls.delete(connection.id);
 		let others = connections.machine(connection.machineId).filter(runsPlanners);
 		void stopPlanners(
 			connection.machineId,
@@ -1638,9 +1644,20 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio
 				let running = new Set(args.planners as string[]);
 				let changed = async (): Promise<PlannerControl | undefined> => {
 					let planners = await plannerDocuments(connection);
-					if (planners.length === running.size && planners.every(value => running.has(value.id))) {
+					let told = toldUrls.get(connection.id) ?? new Map<string, string>();
+					toldUrls.set(connection.id, told);
+					// A document renamed since this connection was told about it is a change too.
+					let moved = planners.some(value =>
+						told.has(value.id) && told.get(value.id) !== value.url
+					);
+					for (let value of planners) if (!told.has(value.id)) told.set(value.id, value.url);
+					if (
+						!moved && planners.length === running.size
+						&& planners.every(value => running.has(value.id))
+					) {
 						return undefined;
 					}
+					toldUrls.set(connection.id, new Map(planners.map(value => [value.id, value.url])));
 					let listed = new Set(planners.map(value => value.id));
 					let reasons = stopReasons.get(connection.machineId);
 					let stopped: Record<string, string> = {};

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

@@ -359,6 +359,50 @@ test("a workflow card shows its question as asked, without run or stage ids", as
 	);
 });
 
+/** A document of a title and two sections, whose first section is block 1. */
+async function sections() {
+	let f = await fixture();
+	let edited = Edit.apply(f.plan, f.plan.revision, [{
+		op: "replace_root",
+		source: "# Existing document\n\nStorage section.\n\nDeployment section.\n",
+	}]);
+	if (!edited.ok) throw new Error("fixture edit failed");
+	if (edited.mutation) await Plan.publish(f.plan, f.server, f.room.id, edited.mutation);
+	expect(Room.digests(f.plan.document)).toHaveLength(3);
+	return f;
+}
+
+test("a question that names its block is placed beside that block and linked to it", async () => {
+	let f = await sections();
+	let response = f.input.questionnaire({ questions: [params.questions[1]!] }, {
+		...f.options,
+		block: { index: 1, digest: Room.digests(f.plan.document)[1]! },
+	});
+	let [card] = await f.cards(1);
+	let source = Plan.source(f.plan);
+	expect(source.indexOf("Storage section.")).toBeLessThan(source.indexOf("<Questionnaire"));
+	expect(source.indexOf("<Questionnaire")).toBeLessThan(source.indexOf("Deployment section."));
+	let record = f.plan.records.get(card!.id)!;
+	let question = record.definition.questions[0]!.id;
+	expect(record.anchors?.questions[question]?.anchors).toHaveLength(1);
+	await f.answer(card!.id, [0]);
+	expect((await response).cancelled).toBe(false);
+});
+
+test("a question whose named block changed stays unlinked at the end of the document", async () => {
+	let f = await sections();
+	let response = f.input.questionnaire({ questions: [params.questions[1]!] }, {
+		...f.options,
+		block: { index: 1, digest: `sha256:${"0".repeat(64)}` },
+	});
+	let [card] = await f.cards(1);
+	let source = Plan.source(f.plan);
+	expect(source.indexOf("Deployment section.")).toBeLessThan(source.indexOf("<Questionnaire"));
+	expect(f.plan.records.get(card!.id)!.anchors).toBeUndefined();
+	await f.answer(card!.id, [0]);
+	await response;
+});
+
 test("abort before presentation creates no card and cancellation waits for durable storage", async () => {
 	let f = await fixture();
 	let cancelled = new AbortController();

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

@@ -11,6 +11,20 @@ import type { DocumentRoom } from "../../agent/tools";
 
 export type HumanInputStatus = "answered" | "cancelled";
 
+/**
+ * A host input request's options, with the block it concerns: its `index` and `digest` from the
+ * Planner's latest read of the document. Its card goes beside that block and is linked to it; a
+ * changed or repeated block leaves the card unlinked at the end of the document.
+ */
+export type HumanInputOptions = HostInputOptions & { block?: { index: number; digest: string } };
+
+/** Chopin's HostInput, whose methods also take the block a request concerns. */
+export type HumanInput = {
+	[Method in keyof HostInput]: HostInput[Method] extends (...args: infer A) => infer R
+		? (...args: { [K in keyof A]: A[K] extends HostInputOptions ? HumanInputOptions : A[K] }) => R
+		: never;
+};
+
 /**
  * Atomic owns input requests; Chopin owns their shared decision records.
  *
@@ -25,10 +39,10 @@ export function createHumanInput(
 	room: DocumentRoom,
 	hold?: (workflowRunId: string, signal: AbortSignal) => Promise<void>,
 	report?: (status: HumanInputStatus) => void,
-): HostInput {
+): HumanInput {
 	async function questionnaire(
 		params: QuestionParams,
-		options: HostInputOptions,
+		options: HumanInputOptions,
 	): Promise<QuestionnaireResult> {
 		if (options.signal.aborted) {
 			report?.("cancelled");
@@ -54,7 +68,7 @@ export function createHumanInput(
 			room.server,
 			room.id,
 			definition,
-			undefined,
+			options.block ? { near: options.block } : undefined,
 			room.anchors,
 			options.signal,
 		);
@@ -105,7 +119,7 @@ export function createHumanInput(
 		header: string,
 		question: string,
 		choices: string[],
-		options: HostInputOptions,
+		options: HumanInputOptions,
 	) {
 		let result = await questionnaire({
 			questions: [{

apps/server/src/harness/remote/link.test.ts

@@ -14,6 +14,8 @@ import { AgentRefused } from "@chopin/agent-link/client";
 import { GitHubError } from "../../github/client";
 import * as Store from "../../questions/store";
 import * as Plan from "../../plan/service";
+import * as Edit from "../../plan/edit";
+import * as Room from "../../plan/room";
 import { createPlannerAgent } from "../agents";
 import { createHumanInput } from "../atomic/human-input";
 import { openPlannerSession } from "../session";
@@ -1345,6 +1347,27 @@ describe("Agent connection questions", () => {
 		expect(await confirm).toMatchObject({ status: "answered", value: true });
 	});
 
+	it("places a question that names its block beside that block, linked to it", async () => {
+		let { f, client, sessionId } = await questions();
+		let edited = Edit.apply(f.plan, f.plan.revision, [{
+			op: "replace_root",
+			source: "# Existing document\n\nStorage section.\n\nDeployment section.\n",
+		}]);
+		if (!edited.ok) throw new Error("fixture edit failed");
+		if (edited.mutation) await Plan.publish(f.plan, f.server, f.room.id, edited.mutation);
+		let asked = client.ask(sessionId, { method: "questionnaire", params }, {
+			block: { index: 1, digest: Room.digests(f.plan.document)[1]! },
+		});
+		let [card] = await f.cards(1);
+		let source = Plan.source(f.plan);
+		expect(source.indexOf("Storage section.")).toBeLessThan(source.indexOf("<Questionnaire"));
+		expect(source.indexOf("<Questionnaire")).toBeLessThan(source.indexOf("Deployment section."));
+		let record = f.plan.records.get(card!.id)!;
+		expect(record.anchors?.questions[record.definition.questions[0]!.id]?.anchors).toHaveLength(1);
+		await f.answer(card!.id, [1]);
+		expect(await asked).toMatchObject({ status: "answered" });
+	});
+
 	it("keeps a client's question open past half an hour until a member answers it", async () => {
 		let { f, client, sessionId } = await questions();
 		let clock = laterTimers();

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

@@ -842,6 +842,7 @@ export class AgentLink {
 			sessionId: message.session,
 			...(message.workflowRunId ? { workflowRunId: message.workflowRunId } : {}),
 			...(message.workflowStageId ? { workflowStageId: message.workflowStageId } : {}),
+			...(message.block ? { block: message.block } : {}),
 		};
 		let input = message.input;
 		try {

apps/server/src/harness/remote/planner-dispatch.test.ts

@@ -306,6 +306,29 @@ describe("choosing a machine for a document", () => {
 		expect(server.attachments.attached(channel.id)).toBe(false);
 	});
 
+	it("tells the machine a renamed document's current address, once", async () => {
+		let { machine, choose, channel, storage } = await dispatchServer();
+		let laptop = await machine("member", "U_member", "laptop");
+		await choose("member", laptop.id);
+		let told = await laptop.poll([]);
+		expect(told.planners[0].url).toEndWith(`/${channel.slug}`);
+		let renamed = await storage.channels.rename({
+			id: channel.id,
+			title: "Renamed document",
+			now: new Date(),
+		});
+		expect(renamed.channel.slug).not.toBe(channel.slug);
+		// The same document, so a connector running it hears only its new address.
+		expect(await laptop.poll([channel.id])).toEqual({
+			kind: "planners",
+			planners: [{
+				id: channel.id,
+				url: expect.stringMatching(new RegExp(`/${renamed.channel.slug}$`)),
+			}],
+		});
+		expect(await laptop.poll([channel.id])).toEqual({ waiting: true });
+	});
+
 	it("never takes over a Planner another account holds", async () => {
 		let { machine, choose, choice, channel, server, url } = await dispatchServer();
 		let laptop = await machine("member", "U_member", "laptop");