Add Build plan and live sync status to the document toolbar

apps/server/src/auth/routes.ts

@@ -22,6 +22,7 @@ type Dependencies = {
 	github?: GitHub;
 	clock?: Clock;
 	agent?: boolean;
+	liveBuild?: boolean;
 	device?: DeviceAuthorization;
 	credentials?: LocalCredentials;
 	delay?: (ms: number, signal: AbortSignal) => Promise<void>;
@@ -343,6 +344,7 @@ export function registerAuthRoutes(
 			let reply = authenticated
 				? json({
 					agent: dependencies.agent ?? true,
+					...(dependencies.liveBuild ? { liveBuild: true } : {}),
 					...(local ? { auth: "local" } : {}),
 					installUrl: "/auth/github/install",
 					user: {
@@ -355,6 +357,7 @@ export function registerAuthRoutes(
 				: json({
 					user: null,
 					agent: dependencies.agent ?? true,
+					...(dependencies.liveBuild ? { liveBuild: true } : {}),
 					...(local ? { auth: "local" } : {}),
 				});
 			if (restoredCookie) reply.headers.append("set-cookie", restoredCookie);

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

@@ -266,6 +266,11 @@ test("the draft instruction comes from the plan and refuses an unsettled one", a
 	expect(prepare.text).toContain("Prepare an implementation graph");
 	expect(prepare.text).toContain("Do not approve or start implementation.");
 	expect(prepare.said).toBe("@ana asked Chopin to break the plan into tasks.");
+	let built = draftInstruction(plan, "ana", true);
+	expect(built.text).toContain("Chopin approves and starts these tasks automatically");
+	expect(built.text).toContain("do not describe them as unapproved or unstarted");
+	expect(built.text).not.toContain("Do not approve or start implementation.");
+	expect(built.said).toBe(prepare.said);
 	expect(draftRefusal(plan)).toBeUndefined();
 	let revised = await implementationGraphs().revise(plan, {
 		planRevision: plan.revision,
@@ -287,6 +292,9 @@ test("the draft instruction comes from the plan and refuses an unsettled one", a
 	let revise = draftInstruction(plan, "ana");
 	expect(revise.text).toContain("Revise the implementation graph");
 	expect(revise.said).toBe("@ana asked Chopin to update the tasks.");
+	expect(revise.text).toContain("Do not approve or start implementation.");
+	expect(draftInstruction(plan, "ana", true).text)
+		.toContain("Chopin approves and starts these tasks automatically");
 	plan.builds = [{ state: "running" } as never];
 	expect(draftRefusal(plan)).toBe("implementation is already active");
 });

apps/server/src/experiments/connections.ts

@@ -48,6 +48,8 @@ export class Connections {
 	#runTokens = new Map<string, string>();
 	#waiters = new Map<string, Set<() => void>>();
 	#builds = new Map<string, string>();
+	/** Documents whose builds wait behind a connection's current one, first queued first. */
+	#queued = new Map<string, string[]>();
 	#used = new Map<string, string>();
 	#locks = new Map<string, Promise<unknown>>();
 	constructor(private now = () => Date.now()) {}
@@ -135,9 +137,40 @@ export class Connections {
 	assigned(connectionId: string) {
 		return this.#builds.get(connectionId);
 	}
-	/** Forget a finished, deleted, or unreachable build so it cannot strand the connection. */
+	/** Queue a document's build behind the other document's build a connection holds. */
+	enqueue(connectionId: string, documentId: string) {
+		if (!this.#connections.has(connectionId) || this.#builds.get(connectionId) === documentId) {
+			return;
+		}
+		let queue = this.#queued.get(connectionId) ?? [];
+		if (!queue.includes(documentId)) this.#queued.set(connectionId, [...queue, documentId]);
+	}
+	/** Documents whose builds wait for this connection, in the order they will start. */
+	queued(connectionId: string): string[] {
+		return [...(this.#queued.get(connectionId) ?? [])];
+	}
+	/**
+	 * Forget a finished, deleted, or unreachable build so it cannot strand the connection. The
+	 * next document queued behind it takes its place, and the connector is woken to claim it.
+	 */
 	release(connectionId: string, documentId: string) {
-		if (this.#builds.get(connectionId) === documentId) this.#builds.delete(connectionId);
+		let queue = this.#queued.get(connectionId) ?? [];
+		if (this.#builds.get(connectionId) === documentId) {
+			let [next, ...rest] = queue;
+			if (next === undefined) {
+				this.#builds.delete(connectionId);
+				return;
+			}
+			this.#builds.set(connectionId, next);
+			if (rest.length) this.#queued.set(connectionId, rest);
+			else this.#queued.delete(connectionId);
+			let connection = this.#connections.get(connectionId);
+			if (connection) this.wake(connection.source.repositoryId);
+			return;
+		}
+		let rest = queue.filter(id => id !== documentId);
+		if (rest.length) this.#queued.set(connectionId, rest);
+		else this.#queued.delete(connectionId);
 	}
 	/** Remember the connection that last ran work for a document, to prefer it next time. */
 	use(documentId: string, connectionId: string) {
@@ -172,6 +205,7 @@ export class Connections {
 		let connection = this.#connections.get(id);
 		this.#connections.delete(id);
 		this.#builds.delete(id);
+		this.#queued.delete(id);
 		for (let [document, used] of this.#used) if (used === id) this.#used.delete(document);
 		for (let key of this.#runTokens.keys()) {
 			if (key.startsWith(`${id}:`)) this.#runTokens.delete(key);

apps/server/src/experiments/routes.test.ts

@@ -451,3 +451,34 @@ test("a stale Ready to run cannot cancel another person's queued run", async ()
 	expect((await refused.json()).error).toBe("connection-forbidden");
 	expect((await get(id))?.state).toBe("queued");
 });
+
+test("builds queued behind a connection's build take its place in the order they queued", () => {
+	let connections = new Connections(() => 0);
+	let pending = connections.create({
+		repository: "org/repo",
+		commit: "a".repeat(40),
+		label: "Laptop",
+	});
+	let { id } = connections.approve(
+		pending.id,
+		{ id: "alice", login: "alice", sessionId: "s" },
+		"repo",
+	);
+	connections.assign(id, "one");
+	connections.enqueue(id, "two");
+	connections.enqueue(id, "three");
+	connections.enqueue(id, "two");
+	connections.enqueue(id, "one");
+	expect(connections.queued(id)).toEqual(["two", "three"]);
+	// A queued document that gives up leaves its place without moving the current build.
+	connections.release(id, "three");
+	connections.enqueue(id, "three");
+	connections.release(id, "one");
+	expect(connections.assigned(id)).toBe("two");
+	expect(connections.queued(id)).toEqual(["three"]);
+	connections.release(id, "two");
+	expect(connections.assigned(id)).toBe("three");
+	connections.release(id, "three");
+	expect(connections.assigned(id)).toBeUndefined();
+	expect(connections.queued(id)).toEqual([]);
+});

apps/server/src/main.ts

@@ -60,8 +60,10 @@ import { DocumentSummaryCoordinator } from "./jobs/summary-coordinator";
 import { invokeRouted } from "./mcp/invoke";
 import { liveClassifier } from "./tasks/live-gate";
 import { LiveSyncCoordinator } from "./tasks/live-sync";
-import { locksEditing } from "./tasks/builds";
+import { locksEditing, requestBuild } from "./tasks/builds";
 import { pendingLinks, relinkInstruction } from "./tasks/relink";
+import { BuildReadiness } from "./tasks/readiness";
+import { askJev } from "./conversation-plan/jev";
 import { registerMcpRoutes } from "./mcp/routes";
 import { registerNavigationRoutes } from "./navigation/routes";
 import * as Service from "./plan/service";
@@ -152,6 +154,7 @@ let recoveringResearch: Promise<void> | undefined;
 let referenceService: ReferenceService | undefined;
 let summaryCoordinator: DocumentSummaryCoordinator | undefined;
 let liveSync: LiveSyncCoordinator | undefined;
+let buildReadiness: BuildReadiness | undefined;
 let descriptionProjector: DocumentDescriptionProjector | undefined;
 let researchBriefJobs: JobService | undefined;
 let documentLocks = new Map<string, Promise<void>>();
@@ -235,6 +238,7 @@ function documentBackend(): Service.Backend {
 		onDocumentPersisted: target => {
 			summaryCoordinator?.schedule(target);
 			liveSync?.schedule(target);
+			buildReadiness?.schedule(target);
 		},
 		liveBuild: !!config.liveBuild,
 		onBuildStopped: id => liveSync?.stopped(id),
@@ -651,7 +655,10 @@ async function closeRoom(room: Rooms.Room, force = false): Promise<void> {
 			await stopped;
 			await Service.close(held);
 		}
-		if (force || room.members.size === 0) Rooms.forget(room);
+		if (force || room.members.size === 0) {
+			Rooms.forget(room);
+			buildReadiness?.forget(room.id);
+		}
 	});
 	room.closing = closing;
 	try {
@@ -800,6 +807,7 @@ async function receive(ws: Socket, raw: string): Promise<void> {
 				if (
 					typeof frame.requestId !== "string" || !REQUEST_ID.test(frame.requestId)
 					|| !Number.isInteger(frame.planRevision) || frame.planRevision < 0
+					|| frame.build !== undefined && frame.build !== true
 				) {
 					fail(ws, frame.rid, "invalid task draft request");
 					return;
@@ -816,7 +824,7 @@ async function receive(ws: Socket, raw: string): Promise<void> {
 					fail(ws, frame.rid, refusal);
 					return;
 				}
-				let { text, said } = draftInstruction(opened, ws.data.handle);
+				let { text, said } = draftInstruction(opened, ws.data.handle, frame.build);
 				let result = await Chat.draftTasks(
 					chat(room, ws),
 					ws.data.handle,
@@ -828,6 +836,7 @@ async function receive(ws: Socket, raw: string): Promise<void> {
 					fail(ws, frame.rid, result.reason);
 					return;
 				}
+				if (frame.build) requestBuild(opened, ws.data.principalId);
 				reply(ws, frame.rid, {
 					kind: "implementation:draft",
 					ts: 0,
@@ -1475,6 +1484,7 @@ function drain(): Promise<void> {
 		if (summaryCoordinator) await attempt(() => summaryCoordinator!.flush());
 		summaryCoordinator?.close();
 		liveSync?.close();
+		buildReadiness?.close();
 		let stoppingJobs = jobRunner?.shutdown();
 		ownerBindings?.revokeAll();
 		plannerAttachments?.close();
@@ -1604,7 +1614,9 @@ async function archiveChannelLocked(channelId: string, now: Date) {
 }
 
 function archiveChannel(channelId: string, now: Date) {
-	return withDocumentTransition(channelId, () => archiveChannelLocked(channelId, now));
+	return withDocumentTransition(channelId, () => archiveChannelLocked(channelId, now)).finally(() =>
+		buildReadiness?.forget(channelId)
+	);
 }
 
 async function restoreChannelLocked(channelId: string, now: Date) {
@@ -1666,6 +1678,7 @@ async function deleteChannelLocked(channelId: string): Promise<boolean> {
 			ws.close(4404, "document deleted");
 		}
 		if (room) Rooms.forget(room);
+		buildReadiness?.forget(channelId);
 		void Service.announceDeletedUnanswered(server, storage, channel);
 		summaryCoordinator?.resume(channelId);
 		return true;
@@ -1862,6 +1875,7 @@ let hostedAuth = registerAuthRoutes(router, {
 	config: config.auth,
 	storage,
 	agent: config.agent,
+	liveBuild: !!config.liveBuild,
 	onSessionRevoked: sessionRevoked,
 	onCredentialsWillRotate: credentialsWillRotate,
 });
@@ -1979,6 +1993,28 @@ referenceService = new ReferenceService({
 	research: researchService,
 	id: ulid,
 });
+// A living document or one with build history is past the question of readiness.
+function pastReadiness(channelId: string): boolean {
+	let held = Rooms.get(channelId)?.plan;
+	return !!held && (!!held.live || held.builds.length > 0 || held.lifecycle.history.length > 0);
+}
+if (config.liveBuild) {
+	buildReadiness = new BuildReadiness({
+		current: currentDocumentTarget,
+		skip: pastReadiness,
+		ask: process.env.JEV_API_KEY || process.env.TYPESAFE_API_KEY
+			? request => askJev(request)
+			: undefined,
+		changed: (channelId, planRevision, ready) =>
+			broadcast(server, channelId, {
+				kind: "implementation:readiness",
+				ts: 0,
+				planRevision,
+				ready,
+			}),
+		error: err => console.error("chopin: build readiness failed -", err),
+	});
+}
 if (config.agent && config.backgroundJobs) {
 	summaryCoordinator = new DocumentSummaryCoordinator({
 		service: jobService,
@@ -2067,12 +2103,15 @@ registerMcpRoutes(router, hostedAuth, {
 	onDocumentPersisted: target => {
 		if (summaryCoordinator) void summaryCoordinator.enqueueNow(target).catch(() => {});
 		liveSync?.schedule(target);
+		buildReadiness?.schedule(target);
 	},
 	restoreChannel,
 	serializeDocument: (channelId, action) =>
 		withDocumentTransition(channelId, () => withDocumentLock(channelId, action)),
 });
 implementations = registerImplementationRoutes(router, hostedAuth, {
+	buildable: (id, planRevision) => buildReadiness?.ready(id, planRevision) ?? false,
+	liveBuild: !!config.liveBuild,
 	connections: experiments.connections,
 	busy: async connectionId =>
 		(await experiments!.service.store.active()).some(item =>
@@ -2192,6 +2231,7 @@ try {
 	await stoppingLease;
 	summaryCoordinator?.close();
 	liveSync?.close();
+	buildReadiness?.close();
 	let stoppingJobs = jobRunner.shutdown();
 	ownerBindings.revokeAll();
 	await stoppingJobs.catch(() => {});

apps/server/src/plan/service.ts

@@ -1,5 +1,5 @@
 import { locksEditing, restoreBuilds, restoreLive } from "../tasks/builds";
-import { implementationStatus } from "../tasks/notifications";
+import { announceImplementation, implementationStatus } from "../tasks/notifications";
 import type { BuildRequest } from "@chopin/protocol/implementation";
 /**
  * The plan, as a room offers it.
@@ -230,6 +230,11 @@ export type Plan = {
 	lifecycle: Lifecycle;
 	/** The finished build this living document was last delivered from. */
 	live?: LiveBuild;
+	/**
+	 * A one-click Build plan waiting for its tasks, kept until the build starts or is cancelled so
+	 * the click outlives a decision the draft raised and a reload. In memory only (prototype).
+	 */
+	buildRequested?: { by: string; revision: number };
 	/** A claim has closed mutation ingress while accepted work drains. */
 	claiming: boolean;
 	/**
@@ -886,6 +891,7 @@ async function commitHosted(
 	}
 	try {
 		let sourceChanged = captured.source !== durable.committedSource;
+		let wasInSync = durable.committedSource === plan.live?.baseSource;
 		let result = await durable.storage.collaboration.commit({
 			channelId: durable.channelId,
 			lease: durable.lease(),
@@ -924,6 +930,10 @@ async function commitHosted(
 		durable.committedDocument = captured.document;
 		durable.committedSidecar = captured.sidecar;
 		announceUnanswered(plan);
+		// A living document's sync status flips only here, between rebuilds.
+		if (plan.live && wasInSync !== (captured.source === plan.live.baseSource)) {
+			announceImplementation(plan);
+		}
 		if (plan.document.epoch === captured.epoch) {
 			plan.document.checkpoint = new Uint8Array(captured.document);
 		}

apps/server/src/research/placement-close.test-fixtures.ts

@@ -44,8 +44,9 @@ let { closeRoom, withDocumentLock } = new Function(
 	"conversationRuntime",
 	"documentBackend",
 	"server",
+	"buildReadiness",
 	transpiled + "\nreturn {closeRoom,withDocumentLock};",
-)(new Map(), Rooms, Plan, runtime, () => opened.backend, opened.server);
+)(new Map(), Rooms, Plan, runtime, () => opened.backend, opened.server, undefined);
 let { research, jobs } = researchStorage({ opened, plan } as Parameters<typeof researchStorage>[0]);
 let entered = Promise.withResolvers<void>();
 let release = Promise.withResolvers<void>();