diff --git a/apps/server/src/auth/routes.ts b/apps/server/src/auth/routes.ts index 898fe88f2..7f1481447 100644 --- a/apps/server/src/auth/routes.ts +++ b/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; @@ -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); diff --git a/apps/server/src/chat/draft.test.ts b/apps/server/src/chat/draft.test.ts index c8e69119b..90d0d503e 100644 --- a/apps/server/src/chat/draft.test.ts +++ b/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"); }); diff --git a/apps/server/src/experiments/connections.ts b/apps/server/src/experiments/connections.ts index ab8764554..bcf3b9e3e 100644 --- a/apps/server/src/experiments/connections.ts +++ b/apps/server/src/experiments/connections.ts @@ -48,6 +48,8 @@ export class Connections { #runTokens = new Map(); #waiters = new Map void>>(); #builds = new Map(); + /** Documents whose builds wait behind a connection's current one, first queued first. */ + #queued = new Map(); #used = new Map(); #locks = new Map>(); 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); diff --git a/apps/server/src/experiments/routes.test.ts b/apps/server/src/experiments/routes.test.ts index de2fa4cb5..9dcfabdfc 100644 --- a/apps/server/src/experiments/routes.test.ts +++ b/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([]); +}); diff --git a/apps/server/src/main.ts b/apps/server/src/main.ts index 22b1f3340..6c3896488 100644 --- a/apps/server/src/main.ts +++ b/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 | 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>(); @@ -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 { 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 { 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 { 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 { 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 { 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 { 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(() => {}); diff --git a/apps/server/src/plan/service.ts b/apps/server/src/plan/service.ts index 9516386f3..07b081f55 100644 --- a/apps/server/src/plan/service.ts +++ b/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); } diff --git a/apps/server/src/research/placement-close.test-fixtures.ts b/apps/server/src/research/placement-close.test-fixtures.ts index 7440aabe1..5b8f497c9 100644 --- a/apps/server/src/research/placement-close.test-fixtures.ts +++ b/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[0]); let entered = Promise.withResolvers(); let release = Promise.withResolvers(); diff --git a/apps/server/src/tasks/builds.ts b/apps/server/src/tasks/builds.ts index 72b0980c4..8f734c9e0 100644 --- a/apps/server/src/tasks/builds.ts +++ b/apps/server/src/tasks/builds.ts @@ -218,6 +218,7 @@ export async function queueBuild(plan: Plan, input: BuildInput): Promise { + return exclusive(plan, async () => { + let build = plan.builds.at(-1); + if (build?.id !== id || build.state !== "queued") return; + let previous = plan.builds; + plan.builds = [...previous.slice(0, -1), { ...build, expiresAt: Date.now() + 90_000 }]; + try { + await persistExclusive(plan, true); + } catch (error) { + plan.builds = previous; + throw error; + } + }); +} + /** Persist pickup before the companion receives permission to spawn a process. */ export function pickBuild( plan: Plan, @@ -868,6 +893,18 @@ export function liveSnapshot( let builder = viewer === live.user; let latest = plan.builds.findLast(build => build.kind === "rebuild"); let rebuild = latest && (builder ? structuredClone(latest) : withoutConnection(latest)); + // The first build's tasks, then each rebuild's appended version, with their last reported state. + let first = plan.builds.find(build => build.id === live.buildId)?.graphVersion; + let history = plan.graph ? historyFor(plan.graph, plan.lifecycle) : []; + let tasks = (plan.graph?.versions ?? []).filter(version => + first !== undefined && version.number >= first + ).flatMap(version => { + let progress = history.findLast(item => item.run.graphVersion === version.number)?.progress; + return version.definition.tasks.map(task => { + let report = progress?.tasks.find(item => item.id === task.id); + return { ...structuredClone(task), ...(report ? { progress: structuredClone(report) } : {}) }; + }); + }); let unfinished = outstandingTasks(plan).map(({ id, title, state, blocker }) => ({ id, title, @@ -877,6 +914,7 @@ export function liveSnapshot( return { ...structuredClone(rest), commits: structuredClone(commits ?? []), + tasks, ...(unfinished.length ? { outstandingTasks: unfinished } : {}), ...(rebuild ? { rebuild } : {}), outOfSync: source(plan) !== baseSource, diff --git a/apps/server/src/tasks/connector.ts b/apps/server/src/tasks/connector.ts index f9f2f96f8..5145fed56 100644 --- a/apps/server/src/tasks/connector.ts +++ b/apps/server/src/tasks/connector.ts @@ -6,7 +6,14 @@ import { LIFECYCLE_TOOLS, lifecycleCall } from "../mcp/lifecycle"; import { exclusive, source } from "../plan/service"; import type { Plan } from "../plan/service"; import { claimImplementation, reportImplementationLifecycle } from "./plan-graphs"; -import { pickBuild, readRebuild, rebuildReportSchema, reportBuild, reportRebuild } from "./builds"; +import { + pickBuild, + readRebuild, + rebuildReportSchema, + renewQueued, + reportBuild, + reportRebuild, +} from "./builds"; import { implementationLifecycle } from "./lifecycle"; export type WithPlan = (id: string, action: (plan: Plan) => Promise) => Promise; @@ -105,6 +112,8 @@ export function implementationConnector( connections: Connections, documentUrl: (id: string) => Promise, documentExists: (id: string) => Promise, + /** Whether a prototype is executing on a connection; a build queued behind it waits. */ + prototyping?: (connectionId: string) => Promise, options: { retryMs?: number; retries?: number } = {}, ) { let tracked = new Set(); @@ -119,24 +128,26 @@ export function implementationConnector( }); return operation; }; + /** The connection's current build; a finished one hands over to the next queued document. */ let pending = async (connection: Connection) => { - let documentId = connections.assigned(connection.id); - if (!documentId) return undefined; - try { - let build = await documentExists(documentId) - ? await withPlan(documentId, async plan => { - let build = plan.builds.at(-1); - return build?.connectionId === connection.id && build.user === connection.owner - ? build - : undefined; - }) - : undefined; - if (!build || !["queued", "starting", "running"].includes(build.state)) { + // Bounded by the queue: each pass releases one document. + for (let pass = 0; pass <= 100; pass++) { + let documentId = connections.assigned(connection.id); + if (!documentId) return undefined; + try { + let build = await documentExists(documentId) + ? await withPlan(documentId, async plan => { + let build = plan.builds.at(-1); + return build?.connectionId === connection.id && build.user === connection.owner + ? build + : undefined; + }) + : undefined; + if (build && ["queued", "starting", "running"].includes(build.state)) return build; connections.release(connection.id, documentId); + } catch { + return undefined; } - return build; - } catch { - return undefined; } }; let api = { @@ -350,7 +361,18 @@ export function implementationConnector( tracked.delete(id); return; } - if (!connections.get(build.connectionId) || build.expiresAt <= Date.now()) { + let connected = !!connections.get(build.connectionId); + // Queued behind another document's build on the same connection. + let behind = connections.assigned(build.connectionId) !== id + && connections.queued(build.connectionId).includes(id); + if ( + connected && build.state === "queued" + && (behind || await prototyping?.(build.connectionId)) + ) { + await renewQueued(plan, build.id); + return; + } + if (!connected || build.expiresAt <= Date.now()) { await reportBuild(plan, build.user, build.connectionId, build.id, { state: "failed", error: "Workspace disconnected or agent stopped responding. No automatic replay.", diff --git a/apps/server/src/tasks/draft.ts b/apps/server/src/tasks/draft.ts index 94389d385..e7075e5c3 100644 --- a/apps/server/src/tasks/draft.ts +++ b/apps/server/src/tasks/draft.ts @@ -19,16 +19,22 @@ export function draftRefusal(plan: Plan): string | undefined { * * Prepared from the plan rather than the asker, so a client cannot choose what * the Planner is told. A graph returned for changes carries its reason along. + * A one-click `build` starts the tasks as soon as they are saved, so the Planner + * must not report them as waiting for approval. */ export function draftInstruction( plan: Plan, handle: string, + build = false, ): { text: string; said: string } { + let close = build + ? "Do not approve or start implementation yourself: Chopin approves and starts these tasks automatically once they are saved, so do not describe them as unapproved or unstarted." + : "Do not approve or start implementation."; let graph = plan.graph?.versions.at(-1); if (!graph) { return { text: - "Prepare an implementation graph for this settled plan. Read the current document and graph, then use edit_implementation_graph to create small reviewable tasks, acceptance criteria and explicit dependencies. Do not approve or start implementation.", + `Prepare an implementation graph for this settled plan. Read the current document and graph, then use edit_implementation_graph to create small reviewable tasks, acceptance criteria and explicit dependencies. ${close}`, said: `@${handle} asked Chopin to break the plan into tasks.`, }; } @@ -46,7 +52,7 @@ export function draftInstruction( : ""; return { text: - `Revise the implementation graph for review. The displayed graph version ${graph.number} is bound to document revision ${graph.planRevision}; the displayed document revision is ${plan.revision}.${returned} Read the latest document and graph, then use edit_implementation_graph to save a draft against the latest document revision. Preserve valid task goals, acceptance criteria and dependencies. If the tasks are still correct, replace an existing task unchanged to refresh the graph's document binding. Do not stop at summarizing an already-approved graph: approval does not make an older document revision current. Do not approve or start implementation.`, + `Revise the implementation graph for review. The displayed graph version ${graph.number} is bound to document revision ${graph.planRevision}; the displayed document revision is ${plan.revision}.${returned} Read the latest document and graph, then use edit_implementation_graph to save a draft against the latest document revision. Preserve valid task goals, acceptance criteria and dependencies. If the tasks are still correct, replace an existing task unchanged to refresh the graph's document binding. Do not stop at summarizing an already-approved graph: approval does not make an older document revision current. ${close}`, said: `@${handle} asked Chopin to update the tasks.`, }; } diff --git a/apps/server/src/tasks/live-sync.test.ts b/apps/server/src/tasks/live-sync.test.ts index e8b6270c3..ceb1fef29 100644 --- a/apps/server/src/tasks/live-sync.test.ts +++ b/apps/server/src/tasks/live-sync.test.ts @@ -165,6 +165,40 @@ async function settle(plan: Plan) { } describe("living-document sync", () => { + it("tells viewers only when an edit flips the document in or out of sync", async () => { + let { plan } = await harness(); + let frames: string[] = []; + spyOn(plan.server, "publish").mockImplementation( + ((_topic: string, data: string) => { + if (data.includes('"plan:implementation"')) frames.push(data); + return 0; + }) as never, + ); + let write = async (text: string, key: string) => { + let result = await rewrite(plan, text, (value, revision) => ({ + idempotencyKey: key, + fingerprint: key, + fromRevision: plan.revision, + client: { name: "test", version: "1" }, + document: { source: value, revision, title: "Living document", url: "https://chopin.test" }, + })); + expect(result.ok).toBe(true); + await settle(plan); + }; + await write("The built paragraph.\n", "built"); + expect(frames).toHaveLength(0); + // As a landed rebuild leaves it: the live base is the current source. + let built = source(plan); + plan.live = { ...plan.live!, baseSource: built }; + await write(`${built}\nOne edit.\n`, "out-1"); + expect(frames).toHaveLength(1); + await write(`${built}\nTwo edits.\n`, "out-2"); + expect(frames).toHaveLength(1); + await write(built, "back"); + expect(frames).toHaveLength(2); + await close(plan); + }); + it("waits for 45 seconds of quiet before rebuilding", async () => { let { plan, queued, edit, advance } = await harness(); edit(); diff --git a/apps/server/src/tasks/readiness.test.ts b/apps/server/src/tasks/readiness.test.ts new file mode 100644 index 000000000..e8e005343 --- /dev/null +++ b/apps/server/src/tasks/readiness.test.ts @@ -0,0 +1,230 @@ +import { describe, expect, test } from "bun:test"; +import { BuildReadiness, judgeBuildable, looksBuildable, withoutCallouts } from "./readiness"; + +import type { JevRequest, JevResult } from "../conversation-plan/jev"; +import type { DocumentTarget } from "../plan/service"; + +let long = `# Title\n\n## Goal\n\n${"word ".repeat(130)}\n\n## Steps\n\nDo it.`; + +function target(source: string, revision = 1): DocumentTarget { + return { channelId: "doc", revision, source, sourceHash: String(revision) }; +} + +function jev(noul: number, seen: JevRequest[] = []) { + return async (request: JevRequest): Promise => { + seen.push(request); + return { + model: "test", + answers: { build_ready: { type: "noul", noul } }, + usage: { input_tokens: 1, output_tokens: 1 }, + latencyMs: 0, + }; + }; +} + +function timers() { + let queue: Array<{ delay: number; action: () => void; cancelled: boolean }> = []; + return { + queue, + after(delay: number, action: () => void) { + let entry = { delay, action, cancelled: false }; + queue.push(entry); + return () => { + entry.cancelled = true; + }; + }, + fire() { + for (let entry of queue.splice(0)) if (!entry.cancelled) entry.action(); + }, + }; +} + +describe("looksBuildable", () => { + test("needs enough words and two headings", () => { + expect(looksBuildable(long)).toBe(true); + expect(looksBuildable(`# One\n\n${"word ".repeat(200)}`)).toBe(false); + expect(looksBuildable("# One\n\n## Two\n\nShort.")).toBe(false); + expect(looksBuildable("")).toBe(false); + }); +}); + +describe("judgeBuildable", () => { + test("asks Jev one noul question about the document", async () => { + let seen: JevRequest[] = []; + expect(await judgeBuildable("tiny", jev(0.9, seen))).toBe(true); + expect(Object.keys(seen[0]!.questions)).toEqual(["build_ready"]); + expect(seen[0]!.questions.build_ready!.type).toBe("noul"); + expect(await judgeBuildable(long, jev(0.2))).toBe(false); + }); + + test("falls back to the heuristic when Jev fails or is absent", async () => { + let failing = async () => { + throw new Error("Jev API key is not configured"); + }; + expect(await judgeBuildable(long, failing)).toBe(true); + expect(await judgeBuildable("tiny", failing)).toBe(false); + expect(await judgeBuildable(long)).toBe(true); + }); +}); + +describe("BuildReadiness", () => { + test("judges once edits settle and reports a change", async () => { + let clock = timers(); + let source = "tiny"; + let revision = 1; + let changes: boolean[] = []; + let asks: JevRequest[] = []; + let readiness = new BuildReadiness({ + current: async () => target(source, revision), + ask: jev(0, asks), + after: clock.after, + changed: (_id, _revision, ready) => changes.push(ready), + }); + readiness.schedule(target(source, 1)); + readiness.schedule(target(source, 1)); + expect(clock.queue.map(entry => entry.delay)).toEqual([10_000, 10_000]); + expect(clock.queue[0]!.cancelled).toBe(true); + clock.fire(); + await Bun.sleep(0); + expect(asks.length).toBe(1); + expect(readiness.ready("doc", 1)).toBe(false); + expect(changes).toEqual([false]); + }); + + test("reads trigger a judgement and keep the last one until the next arrives", async () => { + let clock = timers(); + let noul = 0.9; + let revision = 1; + let source = long; + let changes: boolean[] = []; + let readiness = new BuildReadiness({ + current: async () => target(source, revision), + ask: request => jev(noul)(request), + after: clock.after, + changed: (_id, _revision, ready) => changes.push(ready), + }); + expect(readiness.ready("doc", 1)).toBe(false); + await Bun.sleep(0); + expect(readiness.ready("doc", 1)).toBe(true); + revision = 2; + noul = 0.1; + source = `${long}\n\nMore detail.`; + readiness.schedule(target(source, 2)); + // Pending: the previous judgement stands and no new one starts early. + expect(readiness.ready("doc", 2)).toBe(true); + clock.fire(); + await Bun.sleep(0); + expect(readiness.ready("doc", 2)).toBe(false); + expect(changes).toEqual([true, false]); + }); + + test("callout-only edits keep the last verdict without asking again", async () => { + let asks: JevRequest[] = []; + let noul = 0.9; + let revision = 1; + let source = long; + let changes: boolean[] = []; + let readiness = new BuildReadiness({ + current: async () => target(source, revision), + ask: request => jev(noul, asks)(request), + changed: (_id, _revision, ready) => changes.push(ready), + }); + readiness.ready("doc", 1); + await Bun.sleep(0); + expect(readiness.ready("doc", 1)).toBe(true); + noul = 0.1; + revision = 2; + source = long.replace( + "## Steps", + '\n\tPrototyping…\n\n\n## Steps', + ); + readiness.ready("doc", 2); + await Bun.sleep(0); + expect(readiness.ready("doc", 2)).toBe(true); + expect(asks).toHaveLength(1); + expect(JSON.stringify(asks[0]!.state)).not.toContain("Callout"); + expect(changes).toEqual([true]); + }); + + test("strips top-level callouts from the judged source", () => { + let source = 'Intro.\n\n\n\tA spike.\n\n\nOutro.\n'; + let stripped = withoutCallouts(source); + expect(stripped).toContain("Intro."); + expect(stripped).toContain("Outro."); + expect(stripped).not.toContain("Callout"); + expect(withoutCallouts(source)).toBe(withoutCallouts("Intro.\n\nOutro.\n")); + }); + + test("without Jev, the heuristic decides", async () => { + let readiness = new BuildReadiness({ current: async () => target(long) }); + readiness.ready("doc", 1); + await Bun.sleep(0); + expect(readiness.ready("doc", 1)).toBe(true); + }); + + test("close cancels a pending judgement", () => { + let clock = timers(); + let readiness = new BuildReadiness({ current: async () => target(long), after: clock.after }); + readiness.schedule(target(long)); + readiness.close(); + expect(clock.queue[0]!.cancelled).toBe(true); + }); + + test("skipped documents are never judged", async () => { + let asks: JevRequest[] = []; + let readiness = new BuildReadiness({ + current: async () => target(long), + ask: jev(1, asks), + skip: () => true, + }); + expect(readiness.ready("doc", 1)).toBe(false); + await Bun.sleep(0); + expect(asks.length).toBe(0); + }); + + test("caps remembered judgements", async () => { + let readiness = new BuildReadiness({ + current: async id => ({ ...target(long), channelId: id }), + limit: 2, + }); + for (let id of ["a", "b", "c"]) { + readiness.ready(id, 1); + await Bun.sleep(0); + } + expect(readiness.ready("c", 1)).toBe(true); + expect(readiness.ready("a", 1)).toBe(false); + }); + + test("a judgement requested mid-run runs again afterwards", async () => { + let revision = 1; + let source = long; + let gate: (() => void) | undefined; + let asks = 0; + let readiness = new BuildReadiness({ + current: async () => target(source, revision), + ask: async request => { + asks++; + if (asks === 1) await new Promise(resolve => (gate = resolve)); + return jev(0.9)(request); + }, + }); + readiness.ready("doc", 1); + await Bun.sleep(0); + revision = 2; + source = `${long}\n\nMore detail.`; + readiness.ready("doc", 2); + gate!(); + await Bun.sleep(5); + expect(asks).toBe(2); + expect(readiness.ready("doc", 2)).toBe(true); + }); + + test("forwards Jev errors before falling back", async () => { + let errors: unknown[] = []; + let failing = async (): Promise => { + throw new Error("down"); + }; + expect(await judgeBuildable(long, failing, err => errors.push(err))).toBe(true); + expect(errors.length).toBe(1); + }); +}); diff --git a/apps/server/src/tasks/readiness.ts b/apps/server/src/tasks/readiness.ts new file mode 100644 index 000000000..2f512a865 --- /dev/null +++ b/apps/server/src/tasks/readiness.ts @@ -0,0 +1,192 @@ +import { parse } from "@chopin/dialect/parse"; +import { serialize } from "@chopin/dialect/serialize"; + +import type { JevRequest, JevResult } from "../conversation-plan/jev"; +import type { DocumentTarget } from "../plan/service"; + +export type ReadinessAsk = (request: JevRequest) => Promise; + +export type ReadinessOptions = { + current: (channelId: string) => Promise; + /** Jev, when configured; otherwise, or when it fails, the heuristic decides. */ + ask?: ReadinessAsk; + /** Called after a document's judgement changes. */ + changed?: (channelId: string, revision: number, ready: boolean) => void; + debounceMs?: number; + after?: (delayMs: number, action: () => void) => () => void; + error?: (err: unknown) => void; + /** Documents already past readiness (living, built, or implemented) are never judged. */ + skip?: (channelId: string) => boolean; + /** Most judgements kept in memory. */ + limit?: number; +}; + +const QUESTION = + "Is this document a complete enough plan that a coding agent could build a first version now?"; +/** Jev state is bounded; the opening of a long document carries the judgement. */ +const SOURCE_LIMIT = 20_000; + +/** Enough prose and structure to be worth building, for when Jev cannot say. */ +export function looksBuildable(source: string): boolean { + let lines = source.split("\n"); + let headings = lines.filter(line => /^#{1,6}\s+\S/.test(line)).length; + let words = source.match(/[\p{L}\p{N}][\p{L}\p{N}'’-]*/gu)?.length ?? 0; + return words >= 120 && headings >= 2; +} + +/** + * The document without its top-level callouts. Spike results arrive as callouts, and a verdict + * that changed whenever one was inserted would make Build plan flicker. + */ +export function withoutCallouts(source: string): string { + try { + // Always reserialized, so a source with and without callouts compare equal. + let root = parse(source); + return serialize({ + ...root, + children: root.children.filter(node => + node.type !== "mdxJsxFlowElement" || node.name !== "Callout" + ), + }); + } catch { + return source; + } +} + +export async function judgeBuildable( + source: string, + ask?: ReadinessAsk, + error?: (err: unknown) => void, +): Promise { + if (!ask) return looksBuildable(source); + try { + let result = await ask({ + state: { document: source.slice(0, SOURCE_LIMIT) }, + questions: { + build_ready: { + type: "noul", + instructions: QUESTION, + criteria: { + true: + "The document states what to build and enough of how for a first working version.", + false: "The document is a sketch, a list of open questions, or too thin to build from.", + }, + }, + }, + }); + let answer = result.answers.build_ready; + if (answer?.type !== "noul") throw new Error("invalid Jev readiness answer"); + return answer.noul >= 0.5; + } catch (err) { + error?.(err); + return looksBuildable(source); + } +} + +/** + * Whether each document reads as ready to build, judged once its edits settle. + * + * In memory only: a restart forgets every judgement and the next read asks again. + */ +export class BuildReadiness { + #options: ReadinessOptions; + #debounceMs: number; + /** `key` hashes the judged source without callouts, so callout-only edits keep the verdict. */ + #judged = new Map(); + #pending = new Map void>(); + #running = new Map>(); + #dirty = new Set(); + #closed = false; + + constructor(options: ReadinessOptions) { + this.#options = options; + this.#debounceMs = options.debounceMs ?? 10_000; + } + + /** + * The latest judgement, or `false` before the first. A stale one stands until + * its successor arrives, so a small edit does not hide a ready document. + */ + ready(channelId: string, revision: number): boolean { + if (this.#options.skip?.(channelId)) return false; + let judged = this.#judged.get(channelId); + if (judged) { + this.#judged.delete(channelId); + this.#judged.set(channelId, judged); + } + if (judged?.revision !== revision && !this.#pending.has(channelId)) { + void this.#judge(channelId); + } + return judged?.ready ?? false; + } + + /** A committed edit: judge again once edits stop for the debounce window. */ + schedule(target: DocumentTarget): void { + if (this.#closed || this.#options.skip?.(target.channelId)) return; + this.#pending.get(target.channelId)?.(); + let after = this.#options.after ?? ((delay, action) => { + let timer = setTimeout(action, delay); + return () => clearTimeout(timer); + }); + let cancel = after(this.#debounceMs, () => { + if (this.#pending.get(target.channelId) !== cancel) return; + this.#pending.delete(target.channelId); + void this.#judge(target.channelId); + }); + this.#pending.set(target.channelId, cancel); + } + + /** Judge the document as it stands now; concurrent calls share one judgement. */ + #judge(channelId: string): Promise { + let running = this.#running.get(channelId); + if (running) { + this.#dirty.add(channelId); + return running; + } + let work = (async () => { + let again = false; + try { + this.#dirty.delete(channelId); + let target = await this.#options.current(channelId); + if (!target || this.#closed) return; + let source = withoutCallouts(target.source); + let key = Bun.hash(source); + let known = this.#judged.get(channelId); + let ready = known?.key === key + ? known.ready + : await judgeBuildable(source, this.#options.ask, this.#options.error); + if (this.#closed) return; + let previous = this.#judged.get(channelId); + if (previous && previous.revision > target.revision) return; + this.#judged.delete(channelId); + this.#judged.set(channelId, { revision: target.revision, ready, key }); + for (let key of this.#judged.keys()) { + if (this.#judged.size <= (this.#options.limit ?? 500)) break; + this.#judged.delete(key); + } + again = this.#dirty.has(channelId); + if (previous?.ready !== ready) this.#options.changed?.(channelId, target.revision, ready); + } catch (err) { + this.#options.error?.(err); + } finally { + this.#running.delete(channelId); + if (again && !this.#closed) void this.#judge(channelId); + } + })(); + this.#running.set(channelId, work); + return work; + } + + forget(channelId: string): void { + this.#pending.get(channelId)?.(); + this.#pending.delete(channelId); + this.#judged.delete(channelId); + this.#dirty.delete(channelId); + } + + close(): void { + this.#closed = true; + for (let cancel of this.#pending.values()) cancel(); + this.#pending.clear(); + } +} diff --git a/apps/server/src/tasks/routes.test.ts b/apps/server/src/tasks/routes.test.ts index 5819c0787..e8314d001 100644 --- a/apps/server/src/tasks/routes.test.ts +++ b/apps/server/src/tasks/routes.test.ts @@ -7,7 +7,7 @@ import * as Plan from "../plan/service"; import { registerExperimentRoutes } from "../experiments/routes"; import { implementationGraphs } from "./plan-graphs"; import { registerImplementationRoutes } from "./routes"; -import { queueRebuild } from "./builds"; +import { queueRebuild, requestBuild } from "./builds"; import { implementationStatus } from "./notifications"; import type { GitHub } from "../github/client"; import type { HostedAuth } from "../auth/routes"; @@ -31,7 +31,7 @@ async function prepare(plan: Plan.Plan) { }); } -async function setup() { +async function setup(options: { liveBuild?: boolean } = {}) { let context = await openPlan("# Launcher plan\n"); let github = { async user(token: string) { @@ -96,6 +96,7 @@ async function setup() { implementations: () => implementations, }); implementations = registerImplementationRoutes(router, auth, { + liveBuild: options.liveBuild ?? false, connections: experiments.connections, busy: async connectionId => (await experiments.service.store.active()).some(item => @@ -137,6 +138,7 @@ async function setup() { router, path, call, + cookie: issued.cookie, otherCookie: other.cookie, implementations, experiments, @@ -334,6 +336,38 @@ test("a build runs only on the clicker's own connection and records who started await Plan.close(context.plan); }); +test("a one-click build request outlives a reload until its build starts or is cancelled", async () => { + let context = await setup(); + let cancel = () => + context.router.handle( + new Request(`${context.auth.config.origin}${context.path}/request`, { + method: "DELETE", + headers: { origin: context.auth.config.origin, cookie: context.cookie.split(";")[0] }, + }), + ); + try { + expect((await (await context.call(context.path)).json()).buildRequested).toBeUndefined(); + requestBuild(context.plan, "U_test"); + expect((await (await context.call(context.path)).json()).buildRequested).toEqual({ + by: "U_test", + revision: context.plan.revision, + }); + expect((await cancel())!.status).toBe(200); + expect(context.plan.buildRequested).toBeUndefined(); + requestBuild(context.plan, "U_test"); + await paired(context); + let built = await context.call(context.path, { + planRevision: 0, + graphVersion: 1, + graphRevision: 1, + }); + expect(built.status).toBe(200); + expect((await (await context.call(context.path)).json()).buildRequested).toBeUndefined(); + } finally { + await Plan.close(context.plan); + } +}); + test("investigations and implementations cannot execute on the same workspace together", async () => { let context = await setup(); let connection = await paired(context); @@ -395,8 +429,8 @@ test("a deleted document cannot block implementation recovery sweeps", async () await expect(context.implementations.sweep()).resolves.toBeUndefined(); }); -test("Build waits for an investigation-busy workspace before approving the graph", async () => { - let context = await setup(); +test("Build queues behind a running prototype and starts once it finishes (LIVE_BUILD)", async () => { + let context = await setup({ liveBuild: true }); let connection = await paired(context); let investigation = await (await context.call(`/api/documents/${context.plan.id}/experiments`, { id: crypto.randomUUID(), @@ -404,20 +438,73 @@ test("Build waits for an investigation-busy workspace before approving the graph })).json(); await context.call(`/api/documents/${context.plan.id}/experiments/${investigation.id}/run`, {}); await tool(context, connection.token, "claim_experiment", { id: investigation.id }); + let response = await context.call(context.path, { + planRevision: 0, + graphVersion: 1, + graphRevision: 1, + }); + expect(response.status).toBe(200); + let build = await response.json(); + expect(build.state).toBe("queued"); let snapshot = await (await context.call(context.path)).json(); - expect(snapshot.localAgent).toBe(true); + expect(snapshot).toMatchObject({ build: { id: build.id }, waitingForPrototype: true }); + expect(await tool(context, connection.token, "claim_implementation_build", { id: build.id })) + .toMatchObject({ isError: true }); + // A prototype outlasting the pickup window keeps the build queued rather than failing it. + context.plan.builds = [{ ...context.plan.builds.at(-1)!, expiresAt: Date.now() - 1 }]; + await context.implementations.sweep(); + expect(context.plan.builds.at(-1)).toMatchObject({ id: build.id, state: "queued" }); + expect(context.plan.builds.at(-1)!.expiresAt).toBeGreaterThan(Date.now()); + await context.experiments.service.stop(investigation.id, "interrupted", "Finished."); + expect((await (await context.call(context.path)).json()).waitingForPrototype).toBeUndefined(); + expect(await tool(context, connection.token, "wait_for_work")) + .toEqual({ id: build.id, kind: "implementation" }); expect( - (await context.call(context.path, { - planRevision: 0, - graphVersion: 1, - graphRevision: 1, - })).status, - ).toBe(409); + (await tool(context, connection.token, "claim_implementation_build", { id: build.id })).build, + ).toMatchObject({ id: build.id, state: "starting" }); + await Plan.close(context.plan); +}); + +test("without LIVE_BUILD a build is refused while an investigation holds the workspace", async () => { + let context = await setup(); + let connection = await paired(context); + let investigation = await (await context.call(`/api/documents/${context.plan.id}/experiments`, { + id: crypto.randomUUID(), + brief: "Inspect a fixture", + })).json(); + await context.call(`/api/documents/${context.plan.id}/experiments/${investigation.id}/run`, {}); + await tool(context, connection.token, "claim_experiment", { id: investigation.id }); + let response = await context.call(context.path, { + planRevision: 0, + graphVersion: 1, + graphRevision: 1, + }); + expect(response.status).toBe(409); + expect(await response.text()).toContain("workspace is offline or busy"); expect(context.plan.builds).toEqual([]); - expect(context.plan.graph?.versions[0].state).toBe("draft"); + expect((await (await context.call(context.path)).json()).waitingForPrototype).toBeUndefined(); await Plan.close(context.plan); }); +test("without LIVE_BUILD a second document is refused while the connection builds another", async () => { + let context = await setup(); + let second = await openSiblingPlan(context, "# Second plan\n"); + await prepare(second); + let secondPath = `/api/channels/${second.id}/implementation`; + let review = { planRevision: 0, graphVersion: 1, graphRevision: 1 }; + await paired(context); + try { + expect((await context.call(context.path, review)).status).toBe(200); + let refused = await context.call(secondPath, review); + expect(refused.status).toBe(409); + expect(await refused.text()).toContain("workspace is offline or busy"); + expect(second.builds).toEqual([]); + } finally { + await Plan.close(second); + await Plan.close(context.plan); + } +}); + test("a committed build wakes the waiting connector and a failed pickup can be retried", async () => { let context = await setup(); let connection = await paired(context); @@ -494,8 +581,8 @@ test("returning a stopped implementation for changes durably unlocks it once", a } }); -test("one connection builds two documents and a finished or deleted build never strands it", async () => { - let context = await setup(); +test("a second document queues behind the first on one connection, which a finished or deleted build never strands (LIVE_BUILD)", async () => { + let context = await setup({ liveBuild: true }); let second = await openSiblingPlan(context, "# Second plan\n"); await prepare(second); context.plans.set(second.id, second); @@ -504,21 +591,37 @@ test("one connection builds two documents and a finished or deleted build never let connection = await paired(context); try { let first = await (await context.call(context.path, review)).json(); - expect((await context.call(secondPath, review)).status).toBe(409); + // The agent holds the first document's build, so the second queues behind it. + let waiting = await context.call(secondPath, review); + expect(waiting.status).toBe(200); + let next = await waiting.json(); + expect(next).toMatchObject({ state: "queued", connectionId: connection.connection.id }); + expect((await context.call(secondPath, review)).status).toBe(200); + expect(second.builds).toHaveLength(1); + let title = (await context.storage.channels.get(context.plan.id))!.title; + expect((await (await context.call(secondPath)).json()).waitingForDocument).toEqual({ title }); + // Waiting behind another build renews the queued one instead of expiring it. + second.builds = [{ ...second.builds[0]!, expiresAt: Date.now() - 1 }]; + await context.implementations.sweep(); + expect(second.builds.at(-1)).toMatchObject({ state: "queued" }); + expect(second.builds.at(-1)!.expiresAt).toBeGreaterThan(Date.now()); + expect(await tool(context, connection.token, "wait_for_work")).toEqual({ + id: first.id, + kind: "implementation", + }); await tool(context, connection.token, "claim_implementation_build", { id: first.id }); await tool(context, connection.token, "report_implementation_build", { id: first.id, state: "failed", error: "Startup failed", }); - let next = await (await context.call(secondPath, review)).json(); - expect(next.connectionId).toBe(connection.connection.id); expect(second.builds.at(-1)?.id).toBe(next.id); expect(context.plan.builds).toHaveLength(1); expect(await tool(context, connection.token, "wait_for_work")).toEqual({ id: next.id, kind: "implementation", }); + expect((await (await context.call(secondPath)).json()).waitingForDocument).toBeUndefined(); let claim = await tool(context, connection.token, "claim_implementation_build", { id: next.id, }); diff --git a/apps/server/src/tasks/routes.ts b/apps/server/src/tasks/routes.ts index b540c2f2d..b41d6045b 100644 --- a/apps/server/src/tasks/routes.ts +++ b/apps/server/src/tasks/routes.ts @@ -1,13 +1,13 @@ import { documentPath } from "@chopin/protocol/document-url"; import { z } from "zod"; -import { liveSnapshot, queueBuild } from "./builds"; +import { liveSnapshot, queueBuild, requestBuild } from "./builds"; import { implementationLifecycle } from "./lifecycle"; import { implementationReadiness, reportImplementationLifecycle } from "./plan-graphs"; import { exclusive } from "../plan/service"; import { GitHubError } from "../github/client"; import { AdmissionDenied } from "../auth/admission"; import { implementationConnector } from "./connector"; -import type { Connections } from "../experiments/connections"; +import type { Connection, Connections } from "../experiments/connections"; import type { HostedAuth } from "../auth/routes"; import type { Plan } from "../plan/service"; import type { RouteHandler, Router } from "../http/router"; @@ -18,6 +18,10 @@ type Options = { connections: Connections; /** Whether an investigation is executing on a connection, in any document. */ busy?: (connectionId: string) => Promise; + /** Whether the document reads as ready to build, apart from its blockers. */ + buildable?: (id: string, planRevision: number) => boolean; + /** LIVE_BUILD: builds may queue behind a prototype or another document's build. */ + liveBuild?: boolean; }; let buildSchema = z.object({ retryOf: z.string().uuid().optional(), @@ -72,6 +76,7 @@ export function registerImplementationRoutes(router: Router, auth: HostedAuth, o ).href; }, async id => !!await auth.storage.channels.get(id), + options.liveBuild ? options.busy : undefined, ); let safe = (handler: RouteHandler): RouteHandler => async (request, url, params) => { try { @@ -98,22 +103,41 @@ export function registerImplementationRoutes(router: Router, auth: HostedAuth, o return { session, channel, repository }; } /** - * The caller's live connections for a repository that can take a build for `id`: the one - * last used for this document first, then the most recently heard from. A build already - * waiting on this document does not make its connection busy, so a repeated request can - * return that build. + * The caller's live connections for a repository that can take a build for `id`: idle ones + * first, then those still running a prototype, then those building another document, which + * queue the build until theirs finishes. Within each, the one last used for this document + * first, then the most recently heard from. A build already waiting on this document does not + * make its connection busy, so a repeated request can return that build. */ async function available(owner: string, repositoryId: string, id: string) { - let free = []; - for (let connection of options.connections.candidates(repositoryId, owner, id)) { - let here = options.connections.assigned(connection.id) === id; - if (await options.busy?.(connection.id) || !here && await connector.busy(connection)) { - continue; + if (!options.liveBuild) { + // Without LIVE_BUILD a busy connection is refused, never queued. + let idle: Connection[] = []; + for (let connection of options.connections.candidates(repositoryId, owner, id)) { + let here = options.connections.assigned(connection.id) === id; + if (await options.busy?.(connection.id) || !here && await connector.busy(connection)) { + continue; + } + if (here) idle.unshift(connection); + else idle.push(connection); } - if (here) free.unshift(connection); - else free.push(connection); + return idle; + } + let free: Connection[] = []; + let prototyping: Connection[] = []; + let building: Connection[] = []; + for (let connection of options.connections.candidates(repositoryId, owner, id)) { + let here = options.connections.assigned(connection.id) === id + || options.connections.queued(connection.id).includes(id); + let list = !here && await connector.busy(connection) + ? building + : await options.busy?.(connection.id) + ? prototyping + : free; + if (here) list.unshift(connection); + else list.push(connection); } - return free; + return [...free, ...prototyping, ...building]; } router.on( "GET", @@ -123,6 +147,18 @@ export function registerImplementationRoutes(router: Router, auth: HostedAuth, o connector.track(id); // Cheap on purpose: whether one is free is decided when Build is pressed. let localAgent = options.connections.list(channel.repositoryId, session.user.id).length > 0; + // Read before the plan lock; the experiment store has its own. + let queued = options.liveBuild + ? await options.withPlan(id, async plan => plan.builds.at(-1)) + : undefined; + let ahead = queued?.state === "queued" + && options.connections.queued(queued.connectionId).includes(id) + ? options.connections.assigned(queued.connectionId) + : undefined; + // The other document shares this repository, so its title is readable here. + let aheadTitle = ahead && (await auth.storage.channels.get(ahead))?.title; + let waiting = !ahead && queued?.state === "queued" + && !!await options.busy?.(queued.connectionId); let current = await options.withPlan(id, plan => exclusive(plan, async () => { let ready = implementationReadiness(plan, plan.revision); @@ -141,6 +177,10 @@ export function registerImplementationRoutes(router: Router, auth: HostedAuth, o }), blockers: ready.ok ? [] : ready.blockers, localAgent, + buildReady: ready.ok && (options.buildable?.(id, plan.revision) ?? false), + ...(waiting ? { waitingForPrototype: true as const } : {}), + ...(ahead ? { waitingForDocument: aheadTitle ? { title: aheadTitle } : {} } : {}), + ...(plan.buildRequested ? { buildRequested: { ...plan.buildRequested } } : {}), lifecycle: plan.graph ? implementationLifecycle({ graph: plan.graph, @@ -154,7 +194,12 @@ export function registerImplementationRoutes(router: Router, auth: HostedAuth, o })); // Read outside the plan lock; a missing user only loses the attribution. let login = current.build && (await auth.storage.users.get(current.build.user))?.login; - return json(login ? { ...current, startedBy: login } : current); + let builder = current.live && (await auth.storage.users.get(current.live.user))?.login; + return json({ + ...current, + ...(login ? { startedBy: login } : {}), + ...(builder ? { builtBy: builder } : {}), + }); }), ); router.on( @@ -173,7 +218,12 @@ export function registerImplementationRoutes(router: Router, auth: HostedAuth, o throw new GitHubError("no-workspace", 409); } let [connection] = await available(session.user.id, channel.repositoryId, id); - if (!connection) throw new GitHubError("workspace is offline or busy", 409); + if (!connection) { + throw new GitHubError( + options.liveBuild ? "no-workspace" : "workspace is offline or busy", + 409, + ); + } let checkout = { repository: connection.source.repository, commit: connection.source.commit, @@ -183,9 +233,13 @@ export function registerImplementationRoutes(router: Router, auth: HostedAuth, o return options.connections.locked(connection.id, async () => { if ( !options.connections.get(connection.id) - || !(await available(session.user.id, channel.repositoryId, id)).includes(connection) + || !options.liveBuild + && !(await available(session.user.id, channel.repositoryId, id)).includes(connection) ) { - throw new GitHubError("workspace is offline or busy", 409); + throw new GitHubError( + options.liveBuild ? "no-workspace" : "workspace is offline or busy", + 409, + ); } return options.withPlan(id, async plan => { try { @@ -196,7 +250,11 @@ export function registerImplementationRoutes(router: Router, auth: HostedAuth, o repositoryId: channel.repositoryId, checkout, }); - options.connections.assign(connection.id, id); + let holder = options.connections.assigned(connection.id); + // Another document's build holds the connection: this one starts after it. + if (options.liveBuild && holder && holder !== id) { + options.connections.enqueue(connection.id, id); + } else options.connections.assign(connection.id, id); options.connections.use(id, connection.id); options.connections.wake(channel.repositoryId); return json(build); @@ -207,6 +265,21 @@ export function registerImplementationRoutes(router: Router, auth: HostedAuth, o }); }), ); + router.on( + "DELETE", + "/api/channels/:id/implementation/request", + safe(async (request, _url, { id }) => { + if (request.headers.get("origin") !== auth.config.origin) { + throw new GitHubError("origin is not allowed", 403); + } + let { repository } = await browser(request, id); + if (!repository.permissions.push && !repository.permissions.admin) { + throw new GitHubError("repository write access is required", 403); + } + await options.withPlan(id, async plan => requestBuild(plan, undefined)); + return json({ accepted: true }); + }), + ); router.on( "POST", "/api/channels/:id/implementation/revise", diff --git a/apps/web/src/api.ts b/apps/web/src/api.ts index 51382a7fc..1101d9676 100644 --- a/apps/web/src/api.ts +++ b/apps/web/src/api.ts @@ -7,6 +7,7 @@ export type User = { export type Session = { user: User | null; agent: boolean; + liveBuild?: boolean; installUrl?: string; expiresAt?: string; auth?: "local"; diff --git a/apps/web/src/assets/icons/panel-close.svg b/apps/web/src/assets/icons/panel-close.svg index e12fd20e4..d562da359 100644 --- a/apps/web/src/assets/icons/panel-close.svg +++ b/apps/web/src/assets/icons/panel-close.svg @@ -1,53 +1,16 @@ - - - - - - - + diff --git a/apps/web/src/build-model.test.ts b/apps/web/src/build-model.test.ts index 0144a36d5..1f7645c42 100644 --- a/apps/web/src/build-model.test.ts +++ b/apps/web/src/build-model.test.ts @@ -2,17 +2,31 @@ import { describe, expect, it } from "bun:test"; import { advanceDraft, + advanceFirstBuild, + ago, + attentionHint, buildPhase, draftKey, draftRefusalCopy, elapsed, + firstBuildStep, + pullRequestCommits, pullRequestNumber, shouldAutoDraft, startedBy, + startingHint, + startingLabel, + syncHint, + syncStatus, taskStartsOpen, + waitingLabel, } from "./build-model"; -import type { BuildRequest, ImplementationSnapshot } from "@chopin/protocol/implementation"; +import type { + BuildRequest, + ImplementationSnapshot, + LiveSnapshot, +} from "@chopin/protocol/implementation"; type History = ImplementationSnapshot["lifecycle"]["history"][number]; @@ -38,6 +52,7 @@ function snapshot(change: Partial = {}): ImplementationS }, localAgent: false, blockers: [], + buildReady: true, lifecycle: { execution: { state: "idle" }, history: [] }, ...change, }; @@ -276,3 +291,279 @@ describe("refusals", () => { expect(draftRefusalCopy("could not save the request")).toBeUndefined(); }); }); + +describe("first build", () => { + let idle = { stage: "idle" } as const; + let fresh = snapshot({ graph: undefined }); + + it("offers the button only for a ready document that was never built", () => { + expect(firstBuildStep(fresh, idle)).toEqual({ view: "ready" }); + expect(firstBuildStep(snapshot(), idle)).toEqual({ view: "ready" }); + expect(firstBuildStep(undefined, idle)).toEqual({ view: "hidden" }); + expect(firstBuildStep(snapshot({ graph: undefined, buildReady: false }), idle).view) + .toBe("hidden"); + expect(firstBuildStep(snapshot({ blockers: ["unanswered questionnaires"] }), idle).view) + .toBe("hidden"); + expect(firstBuildStep(snapshot({ build: build("failed") }), idle).view).toBe("hidden"); + expect(firstBuildStep(snapshot({ build: build("running") }), idle).view).toBe("working"); + }); + + it("drafts, then starts the drafted tasks without a review step", () => { + let state = advanceFirstBuild(idle, { type: "press" }); + expect(firstBuildStep(fresh, state)).toEqual({ view: "working", next: "draft" }); + state = advanceFirstBuild(state, { type: "draft-sent" }); + expect(firstBuildStep(fresh, state)).toEqual({ view: "working" }); + // Tasks for an older revision are not the drafted ones. + let stale = snapshot({ planRevision: 5 }); + expect(firstBuildStep(stale, state)).toEqual({ view: "working" }); + expect(firstBuildStep(snapshot(), state)).toEqual({ view: "working", next: "start" }); + state = advanceFirstBuild(state, { type: "start" }); + expect(state).toEqual({ stage: "starting" }); + expect(firstBuildStep(snapshot(), state)).toEqual({ view: "working" }); + state = advanceFirstBuild(state, { type: "started" }); + expect(firstBuildStep(snapshot(), state)).toEqual({ view: "working" }); + expect(firstBuildStep(snapshot({ build: build("queued") }), state)) + .toEqual({ view: "working", next: "reset" }); + }); + + it("starts existing tasks for this revision at once", () => { + let state = advanceFirstBuild(idle, { type: "press" }); + expect(firstBuildStep(snapshot(), state).next).toBe("start"); + }); + + it("returns to the button when drafting or starting fails", () => { + let sent = advanceFirstBuild(advanceFirstBuild(idle, { type: "press" }), { + type: "draft-sent", + }); + expect(advanceFirstBuild(sent, { type: "draft-ended", drafted: true })).toBe(sent); + let failed = advanceFirstBuild(sent, { type: "draft-ended", drafted: false }); + expect(failed.stage).toBe("failed"); + expect(firstBuildStep(fresh, failed)).toEqual({ view: "ready" }); + expect(advanceFirstBuild(failed, { type: "press" })).toEqual({ + stage: "drafting", + sent: false, + }); + + let starting = advanceFirstBuild(sent, { type: "start" }); + let noAgent = advanceFirstBuild(starting, { type: "start-refused", agent: true }); + expect(noAgent).toEqual({ stage: "failed", agent: true, message: undefined }); + expect(firstBuildStep(snapshot(), noAgent)).toEqual({ view: "ready" }); + }); + + it("waits through a decision the draft raised, then redrafts and starts", () => { + let requested = { buildRequested: { by: "me", revision: 4 } }; + let state = advanceFirstBuild(advanceFirstBuild(idle, { type: "press" }), { + type: "draft-sent", + }); + let blocked = snapshot({ blockers: ["unanswered questionnaires"], ...requested }); + expect(firstBuildStep(blocked, state, "me")).toEqual({ view: "waiting", next: "wait" }); + state = advanceFirstBuild(state, { type: "wait" }); + expect(state).toEqual({ stage: "waiting" }); + // The draft turn ending while blocked is not a failure. + expect(advanceFirstBuild(state, { type: "draft-ended", drafted: false })).toBe(state); + expect(firstBuildStep(blocked, state, "me")).toEqual({ view: "waiting" }); + expect(waitingLabel(blocked)).toEqual({ label: "Waiting on a decision", target: "decisions" }); + // Answering moved the document on, so the tasks are stale: draft them again. + let answered = snapshot({ planRevision: 5, ...requested }); + expect(firstBuildStep(answered, state, "me")).toEqual({ view: "working", next: "draft" }); + state = advanceFirstBuild(state, { type: "draft-sent" }); + expect(firstBuildStep(answered, state, "me")).toEqual({ view: "working" }); + let redrafted = snapshot({ ...requested }); + expect(firstBuildStep(redrafted, state, "me")).toEqual({ view: "working", next: "start" }); + }); + + it("resumes a request the server kept across a reload", () => { + let requested = { buildRequested: { by: "me", revision: 4 } }; + let blocked = snapshot({ blockers: ["unanswered questionnaires"], ...requested }); + expect(firstBuildStep(blocked, idle, "me")).toEqual({ view: "waiting", next: "resume" }); + expect(firstBuildStep(snapshot(requested), idle, "me")) + .toEqual({ view: "working", next: "resume" }); + // Someone else's request, or one for a document since built, is not this viewer's to resume. + expect(firstBuildStep(blocked, idle, "you").view).toBe("hidden"); + expect( + firstBuildStep( + { + ...blocked, + lifecycle: { ...blocked.lifecycle, history: [run({ kind: "implemented" })] }, + }, + idle, + "me", + ).next, + ).toBeUndefined(); + let state = advanceFirstBuild(idle, { type: "resume" }); + expect(state).toEqual({ stage: "waiting" }); + expect(firstBuildStep(snapshot(requested), state, "me").next).toBe("start"); + // Cancelled: the request is gone, so the wait ends. + expect(firstBuildStep(snapshot({ blockers: ["unanswered questionnaires"] }), state, "me")) + .toEqual({ view: "hidden", next: "reset" }); + }); +}); + +const PR = "https://github.com/o/r/pull/7"; + +function live(change: Partial = {}): ImplementationSnapshot { + return snapshot({ + live: { + buildId: "b", + user: "u", + repositoryId: "r", + checkout: { repository: "o/r", commit: "0".repeat(40) }, + baseRevision: 4, + pullRequests: [PR], + commits: [], + tasks: [], + outOfSync: false, + builderConnected: true, + ...change, + }, + }); +} + +function rebuild(state: BuildRequest["state"]): BuildRequest { + return { ...build(state), id: "r1", kind: "rebuild", baseRevision: 4, targetRevision: 6 }; +} + +describe("living document sync", () => { + it("is absent until the first build has delivered", () => { + expect(syncStatus(snapshot())).toBeUndefined(); + expect(syncStatus(undefined)).toBeUndefined(); + }); + + it("reads as in sync when the pull requests match the document", () => { + expect(syncStatus(live())).toEqual({ kind: "in-sync" }); + expect(buildPhase(live())).toEqual({ kind: "live", sync: { kind: "in-sync" } }); + }); + + it("reads as building while a rebuild or the first build runs", () => { + for (let state of ["queued", "starting", "running"] as const) { + expect(syncStatus(live({ outOfSync: true, rebuild: rebuild(state) }))) + .toEqual({ kind: "building" }); + } + expect(syncStatus({ ...live(), build: build("running") })).toEqual({ kind: "building" }); + }); + + it("explains a first build queued behind another document's build", () => { + let queued = snapshot({ build: build("queued"), waitingForDocument: { title: "Docs site" } }); + expect(startingLabel(queued)).toEqual({ + label: "Queued", + queued: true, + hint: "Starts when your agent finishes “Docs site”", + }); + expect(startingHint(snapshot({ build: build("queued"), waitingForDocument: {} }))) + .toBe("Starts when your agent finishes another document"); + }); + + it("explains a first build queued behind a prototype", () => { + let queued = snapshot({ build: build("queued"), waitingForPrototype: true }); + expect(firstBuildStep(queued, { stage: "idle" }).view).toBe("working"); + expect(startingHint(queued)).toBe("Starts when the current prototype finishes"); + expect(startingLabel(queued)).toEqual({ + label: "Queued", + queued: true, + hint: "Starts when the current prototype finishes", + }); + expect(startingLabel(snapshot({ build: build("queued") }))) + .toEqual({ label: "Building…", queued: false }); + expect(startingLabel(undefined)).toEqual({ label: "Building…", queued: false }); + expect(startingHint(snapshot({ build: build("queued") }))).toBeUndefined(); + expect(startingHint(snapshot({ build: build("running"), waitingForPrototype: true }))) + .toBeUndefined(); + }); + + it("explains why it is out of sync", () => { + let pending = live({ outOfSync: true, rebuild: rebuild("stopped") }); + expect(syncStatus(pending)).toEqual({ kind: "out-of-sync", reason: "pending", outstanding: 0 }); + expect(syncHint(syncStatus(pending), pending, "me")).toBe("Changes will build shortly"); + let waiting = { ...live({ outOfSync: true, builderConnected: false }), builtBy: "jev" }; + expect(syncStatus(waiting)).toEqual({ kind: "out-of-sync", reason: "waiting", outstanding: 0 }); + expect(syncHint(syncStatus(waiting), waiting, "me")).toBe("Waiting for @jev’s agent"); + expect(syncHint(syncStatus(waiting), waiting, "u")).toBe("Waiting for your agent"); + let failed = live({ outOfSync: true, rebuild: rebuild("failed") }); + expect(syncStatus(failed)).toEqual({ kind: "out-of-sync", reason: "failed", outstanding: 0 }); + expect(syncHint(syncStatus(failed), failed, "me")).toBe("The last rebuild failed"); + }); + + it("needs attention while a blocked task waits for an edit, even in sync", () => { + let outstandingTasks = [ + { id: "a", title: "Store graphs", state: "blocked" as const, blocker: "Which database?" }, + { id: "b", title: "Render graphs", state: "queued" as const }, + ]; + let stuck = live({ outstandingTasks }); + expect(syncStatus(stuck)).toEqual({ kind: "needs-attention", outstanding: 2 }); + expect(syncHint(syncStatus(stuck), stuck, "me")) + .toBe( + "“Store graphs” is blocked: Which database? (and 1 other). Edit the document to retry.", + ); + let edited = live({ outstandingTasks, outOfSync: true }); + expect(syncStatus(edited)).toEqual({ kind: "out-of-sync", reason: "pending", outstanding: 2 }); + expect(syncHint(syncStatus(edited), edited, "me")) + .toBe("Changes will build shortly · will also retry 2 blocked tasks"); + // Syncing beats needing attention. + expect(syncStatus(live({ outstandingTasks, rebuild: rebuild("running") }))) + .toEqual({ kind: "building" }); + }); + + it("ends a blocker's hint with one stop, whatever punctuation it brought", () => { + let hint = (blocker: string) => + attentionHint(live({ + outstandingTasks: [{ id: "a", title: "Ship", state: "blocked", blocker }], + })); + expect(hint("I can't decide this myself.")).toBe( + "“Ship” is blocked: I can't decide this myself. Edit the document to retry.", + ); + expect(hint("Which database?")).toBe( + "“Ship” is blocked: Which database? Edit the document to retry.", + ); + }); + + it("needs attention when a first build stopped on a blocker before any pull request", () => { + let stopped = live({ + pullRequests: [], + outstandingTasks: [{ id: "a", title: "Ship", state: "blocked", blocker: "Pick a host" }], + }); + expect(buildPhase(stopped)).toEqual({ + kind: "live", + sync: { kind: "needs-attention", outstanding: 1 }, + }); + expect(attentionHint(stopped)).toBe( + "“Ship” is blocked: Pick a host. Edit the document to retry.", + ); + }); + + it("does not call a failed rebuild out of sync once later edits match", () => { + expect(syncStatus(live({ rebuild: rebuild("failed") }))).toEqual({ kind: "in-sync" }); + expect(syncHint({ kind: "in-sync" }, live(), "me")).toBeUndefined(); + }); +}); + +describe("living document commits", () => { + let commit = (sha: string, at: string, pullRequest = PR) => ({ + pullRequest, + sha, + message: sha, + revision: 5, + at, + }); + + it("groups one pull request's commits, newest first", () => { + let value = live({ + commits: [ + commit("a", "2026-10-10T10:00:00.000Z"), + commit("b", "2026-10-10T11:00:00.000Z", "https://github.com/o/r/pull/8"), + commit("c", "2026-10-10T12:00:00.000Z"), + commit("d", "2026-10-10T12:00:00.000Z"), + ], + }); + expect(pullRequestCommits(value, PR).map(item => item.sha)).toEqual(["d", "c", "a"]); + expect(pullRequestCommits(snapshot(), PR)).toEqual([]); + }); + + it("says how long ago a commit landed", () => { + let now = Date.parse("2026-10-10T12:00:00.000Z"); + expect(ago("2026-10-10T11:59:40.000Z", now)).toBe("just now"); + expect(ago("2026-10-10T11:55:00.000Z", now)).toBe("5m ago"); + expect(ago("2026-10-10T09:00:00.000Z", now)).toBe("3h ago"); + expect(ago("2026-10-08T12:00:00.000Z", now)).toBe("2d ago"); + expect(ago("nonsense", now)).toBeUndefined(); + }); +}); diff --git a/apps/web/src/build-model.ts b/apps/web/src/build-model.ts index 42083b5be..645da34a6 100644 --- a/apps/web/src/build-model.ts +++ b/apps/web/src/build-model.ts @@ -17,7 +17,132 @@ export type BuildPhase = | { kind: "failed" } /** The build ended while its run still holds the tasks and the document lock. */ | { kind: "stopped" } - | { kind: "done"; pullRequests: number }; + | { kind: "done"; pullRequests: number } + /** A living document: its pull requests follow later edits. */ + | { kind: "live"; sync: SyncStatus }; + +/** + * Whether a living document's pull requests match it. `out-of-sync` says why: + * `pending` waits for the builder's connected agent to pick the edits up, + * `waiting` for that agent to connect, and `failed` that the last rebuild failed. + * `outstanding` counts tasks an earlier build left unfinished; only an edit retries + * them, so a document otherwise in sync with any `needs-attention`. + */ +export type SyncStatus = + | { kind: "building" } + | { kind: "in-sync" } + | { kind: "needs-attention"; outstanding: number } + | { kind: "out-of-sync"; reason: "pending" | "waiting" | "failed"; outstanding: number }; + +const RUNNING = new Set(["queued", "starting", "running"]); + +/** + * The living document's sync status, or `undefined` before its first build has delivered. + * The snapshot's `build` is the first build, or nothing once a rebuild adds a task version, + * so a running rebuild is read from `live.rebuild`. + */ +export function syncStatus(snapshot: Snapshot | undefined): SyncStatus | undefined { + let live = snapshot?.live; + if (!snapshot || !live) return; + if ( + live.rebuild && RUNNING.has(live.rebuild.state) + || snapshot.build && RUNNING.has(snapshot.build.state) + || snapshot.lifecycle.execution.state === "active" + ) return { kind: "building" }; + let outstanding = live.outstandingTasks?.length ?? 0; + if (!live.outOfSync) { + return outstanding ? { kind: "needs-attention", outstanding } : { kind: "in-sync" }; + } + if (live.rebuild?.state === "failed") { + return { kind: "out-of-sync", reason: "failed", outstanding }; + } + return { + kind: "out-of-sync", + reason: live.builderConnected ? "pending" : "waiting", + outstanding, + }; +} + +export const SYNC_LABEL: Record = { + building: "Building…", + "in-sync": "In sync", + "needs-attention": "Needs attention", + "out-of-sync": "Out of sync", +}; + +/** Why the pull requests lag the document, naming the builder whose agent must run them. */ +export function syncHint( + status: SyncStatus | undefined, + snapshot: Snapshot | undefined, + userId: string | undefined, +): string | undefined { + if (status?.kind === "needs-attention") return attentionHint(snapshot); + if (status?.kind !== "out-of-sync") return; + let retry = status.outstanding + ? ` · will also retry ${plural(status.outstanding, "blocked task", "blocked tasks")}` + : ""; + if (status.reason === "pending") return `Changes will build shortly${retry}`; + if (status.reason === "failed") return `The last rebuild failed${retry}`; + if (snapshot?.live?.user === userId) return `Waiting for your agent${retry}`; + return snapshot?.builtBy + ? `Waiting for @${snapshot.builtBy}’s agent${retry}` + : `Waiting for the builder’s agent${retry}`; +} + +/** What an unfinished task is stuck on, and that an edit retries it. */ +export function attentionHint(snapshot: Snapshot | undefined): string | undefined { + let tasks = snapshot?.live?.outstandingTasks ?? []; + if (!tasks.length) return; + let first = tasks.find(task => task.blocker) ?? tasks[0]!; + let reason = first.blocker + ? `“${first.title}” is blocked: ${first.blocker.trim().replace(/\.+$/, "")}` + : `“${first.title}” didn’t finish`; + let more = tasks.length > 1 ? ` (and ${plural(tasks.length - 1, "other", "others")})` : ""; + let said = `${reason}${more}`; + return `${said}${/[?!…]$/.test(said) ? "" : "."} Edit the document to retry.`; +} + +/** Why a first build has not started yet, when the viewer's agent is finishing a prototype. */ +export function startingHint(snapshot: Snapshot | undefined): string | undefined { + if (snapshot?.build?.state !== "queued") return; + let other = snapshot.waitingForDocument; + if (other) { + return `Starts when your agent finishes ${ + other.title ? `“${other.title}”` : "another document" + }`; + } + return snapshot.waitingForPrototype ? "Starts when the current prototype finishes" : undefined; +} + +/** + * The header slot while a first build is under way: `queued` while it waits for + * the agent's prototype, so the wait reads without hovering for the hint. + */ +export function startingLabel( + snapshot: Snapshot | undefined, +): { label: string; queued: boolean; hint?: string } { + let hint = startingHint(snapshot); + return hint ? { label: "Queued", queued: true, hint } : { label: "Building…", queued: false }; +} + +/** One pull request's living-document commits, newest first. */ +export function pullRequestCommits(snapshot: Snapshot | undefined, url: string) { + return (snapshot?.live?.commits ?? []).map((commit, index) => ({ commit, index })) + .filter(item => item.commit.pullRequest === url) + .sort((a, b) => Date.parse(b.commit.at) - Date.parse(a.commit.at) || b.index - a.index) + .map(item => item.commit); +} + +/** "just now", "5m ago", "3h ago" or "2d ago". */ +export function ago(at: string, now: number): string | undefined { + let start = Date.parse(at); + if (!Number.isFinite(start)) return; + let minutes = Math.max(0, Math.floor((now - start) / 60_000)); + if (minutes < 1) return "just now"; + if (minutes < 60) return `${minutes}m ago`; + let hours = Math.floor(minutes / 60); + return hours < 24 ? `${hours}h ago` : `${Math.floor(hours / 24)}d ago`; +} function latestRun(snapshot: Snapshot) { let graph = snapshot.graph; @@ -34,6 +159,8 @@ export function buildProgress(snapshot: Snapshot | undefined): Progress | undefi export function buildPhase(snapshot: Snapshot | undefined): BuildPhase { if (!snapshot) return { kind: "loading" }; + let sync = syncStatus(snapshot); + if (sync) return { kind: "live", sync }; let run = latestRun(snapshot); let active = snapshot.lifecycle.execution.state === "active"; let build = snapshot.build; @@ -207,3 +334,140 @@ export function draftRefusalCopy(message: string): string | undefined { } if (/cannot be built/.test(message)) return "This document can’t be built."; } + +/** Whether this document has ever been handed to a coding agent. */ +export function hasBuilt(snapshot: Snapshot): boolean { + return !!snapshot.build || snapshot.lifecycle.execution.state === "active" + || snapshot.lifecycle.history.length > 0; +} + +/** + * The header's one-click first build: draft the tasks, then start them, with + * no review in between. `failed` keeps the button so a person can try again; + * `agent` says the last start found no local agent to run it. + */ +export type FirstBuild = + | { stage: "idle" } + | { stage: "drafting"; sent: boolean } + /** The draft raised a blocker, or a reload found the request: continue once it clears. */ + | { stage: "waiting" } + | { stage: "starting" } + | { stage: "started" } + | { stage: "failed"; agent?: boolean; message?: string }; + +export type FirstBuildEvent = + | { type: "press" } + /** The server still holds this viewer's one-click request, e.g. after a reload. */ + | { type: "resume" } + /** The draft left the document blocked; keep the intent until the blockers clear. */ + | { type: "wait" } + | { type: "draft-sent" } + /** The draft turn ended; `drafted` when a fresh read then had tasks to start. */ + | { type: "draft-ended"; drafted: boolean } + | { type: "draft-refused"; message?: string } + | { type: "start" } + | { type: "started" } + | { type: "start-refused"; agent: boolean; message?: string } + | { type: "reset" }; + +export function advanceFirstBuild(state: FirstBuild, event: FirstBuildEvent): FirstBuild { + switch (event.type) { + case "press": + return state.stage === "idle" || state.stage === "failed" + ? { stage: "drafting", sent: false } + : state; + case "resume": + return state.stage === "idle" ? { stage: "waiting" } : state; + case "wait": + return state.stage === "drafting" ? { stage: "waiting" } : state; + case "draft-sent": + return state.stage === "drafting" || state.stage === "waiting" + ? { stage: "drafting", sent: true } + : state; + case "draft-ended": + return state.stage === "drafting" && state.sent && !event.drafted + ? { stage: "failed", message: "Chopin couldn’t break the plan into tasks." } + : state; + case "draft-refused": + return state.stage === "drafting" ? { stage: "failed", message: event.message } : state; + case "start": + return state.stage === "drafting" || state.stage === "waiting" || state.stage === "failed" + ? { stage: "starting" } + : state; + case "started": + return state.stage === "starting" ? { stage: "started" } : state; + case "start-refused": + return state.stage === "starting" + ? { stage: "failed", agent: event.agent, message: event.message } + : state; + case "reset": + return { stage: "idle" }; + } +} + +/** + * What the header shows for a snapshot and the first build's stage, and the + * one step the browser should take next, if any. + * + * The button appears only for a document judged ready that has never been + * built. Once pressed it reads as working until the build runs or the attempt + * ends; `reset` hands any later state back to the Build view. A draft that + * raises a decision does not drop the press: the slot reads `waiting` until the + * blockers clear, then drafts again if the tasks are stale and starts them. The + * server keeps the press (`buildRequested`), so a reload `resume`s it. + */ +export function firstBuildStep( + snapshot: Snapshot | undefined, + state: FirstBuild, + userId?: string, +): { + view: "hidden" | "ready" | "working" | "waiting"; + next?: "draft" | "start" | "reset" | "resume" | "wait"; +} { + if (!snapshot) return { view: state.stage === "idle" ? "hidden" : "working" }; + let phase = buildPhase(snapshot); + if (phase.kind === "building") { + return { view: "working", ...(state.stage === "idle" ? {} : { next: "reset" as const }) }; + } + let requested = !!userId && snapshot.buildRequested?.by === userId && !hasBuilt(snapshot); + let pending = phase.kind === "blocked" || phase.kind === "drafting" || phase.kind === "review"; + if (state.stage === "idle" && requested && pending) { + return { view: phase.kind === "blocked" ? "waiting" : "working", next: "resume" }; + } + if (state.stage === "idle" || state.stage === "failed") { + let open = phase.kind === "drafting" || phase.kind === "review"; + return { view: snapshot.buildReady && open && !hasBuilt(snapshot) ? "ready" : "hidden" }; + } + if (state.stage === "starting") return { view: "working" }; + if (state.stage === "started") { + return phase.kind === "review" ? { view: "working" } : { view: "hidden", next: "reset" }; + } + if (state.stage === "waiting") { + // Cancelled, or someone else built it: nothing is left to continue. + if (!requested) return { view: "hidden", next: "reset" }; + if (phase.kind === "blocked") return { view: "waiting" }; + if (phase.kind === "drafting") return { view: "working", next: "draft" }; + if (phase.kind === "review") return { view: "working", next: "start" }; + return { view: "hidden", next: "reset" }; + } + if (phase.kind === "review") return { view: "working", next: "start" }; + if (phase.kind === "drafting") { + return { view: "working", ...(state.sent ? {} : { next: "draft" }) }; + } + if (phase.kind === "blocked") return { view: "waiting", next: "wait" }; + return { view: "hidden", next: "reset" }; +} + +/** The header's wait for a blocked one-click build, and where pressing it leads. */ +export function waitingLabel( + snapshot: Snapshot | undefined, +): { label: string; target: "decisions" | "build" } { + let phase = snapshot && buildPhase(snapshot); + if (phase?.kind === "blocked" && phase.decisions) { + return { label: "Waiting on a decision", target: "decisions" }; + } + if (phase?.kind === "blocked" && phase.comments) { + return { label: "Waiting on comments", target: "build" }; + } + return { label: "Waiting to build", target: "build" }; +} diff --git a/apps/web/src/build-plan-button.tsx b/apps/web/src/build-plan-button.tsx new file mode 100644 index 000000000..7a5846c04 --- /dev/null +++ b/apps/web/src/build-plan-button.tsx @@ -0,0 +1,235 @@ +import { CheckIcon, LoaderIcon } from "@chopin/icons"; +import { useEffect, useReducer, useRef, useState } from "react"; + +import { + advanceFirstBuild, + buildPhase, + draftRefusalCopy, + firstBuildStep, + startingLabel, + SYNC_LABEL, + syncHint, + syncStatus, + waitingLabel, +} from "./build-model"; +import { + cancelBuildRequest, + implementationEndpoint, + implementationResponse, + startBuild, +} from "./build-start"; + +import type { Implementation, ImplementationSnapshot } from "@chopin/protocol/implementation"; +import type { FirstBuild } from "./build-model"; +import type { Wire } from "./wire"; + +const RELOAD_ON = [ + "plan:implementation", + "plan:open", + "implementation:readiness", + "experiment:changed", +]; + +/** + * The document's first build in one press: ask the Planner for tasks, then + * start them on the viewer's local agent. The Build view owns every later build. + * Once that build has delivered, the slot quietly reports whether the pull + * requests still match the living document. + */ +export function BuildPlanButton( + { onNeedsAgent, onShowBuild, onShowDecisions, room, userId, wire }: { + /** No local agent could take the build; the Build view explains how to start one. */ + onNeedsAgent: () => void; + onShowBuild: () => void; + onShowDecisions: () => void; + room: string; + userId?: string; + wire?: Wire; + }, +) { + let [snapshot, setSnapshot] = useState(); + let [refresh, setRefresh] = useState(0); + let [state, dispatch] = useReducer(advanceFirstBuild, { stage: "idle" } as FirstBuild); + let draftId = useRef(undefined); + let endpoint = implementationEndpoint(room); + + let read = async (signal?: AbortSignal) => { + let value = await implementationResponse( + await fetch(endpoint, { signal, cache: "no-store" }), + ); + setSnapshot(current => !current || value.revision >= current.revision ? value : current); + return value; + }; + + useEffect(() => { + let controller = new AbortController(); + read(controller.signal).catch(() => {}); + return () => controller.abort(); + }, [endpoint, refresh]); + + useEffect(() => { + let reload = () => setRefresh(value => value + 1); + let off = RELOAD_ON.map(kind => wire?.on(kind, reload)); + return () => off.forEach(remove => remove?.()); + }, [wire]); + + // Whether the draft turn left tasks to start is only known from a fresh read. + let drafted = async () => { + draftId.current = undefined; + try { + let fresh = await read(); + dispatch({ type: "draft-ended", drafted: buildPhase(fresh).kind !== "drafting" }); + } catch { + dispatch({ type: "draft-ended", drafted: false }); + } + }; + + useEffect( + () => + wire?.on("implementation:drafting", frame => { + if (frame.id === draftId.current && frame.state === "ended") void drafted(); + }), + [wire], + ); + + // A failed attempt gives up the server's request, so a reload does not try it again. + useEffect(() => { + if (state.stage === "failed" && snapshot?.buildRequested?.by === userId) { + void cancelBuildRequest(room).catch(() => {}); + } + }, [state.stage]); + + let step = firstBuildStep(snapshot, state, userId); + useEffect(() => { + if (!step.next) return; + if (step.next === "reset" || step.next === "resume" || step.next === "wait") { + dispatch({ type: step.next }); + return; + } + if (step.next === "draft") { + if (!snapshot || !wire?.connected) { + dispatch({ type: "draft-refused", message: "Chopin isn’t connected." }); + return; + } + dispatch({ type: "draft-sent" }); + void wire.ask("implementation:draft", { + requestId: crypto.randomUUID(), + planRevision: snapshot.planRevision, + build: true, + }).then(answer => { + draftId.current = answer.id; + if (answer.state === "ended") void drafted(); + }, error => { + let copy = error instanceof Error ? draftRefusalCopy(error.message) : undefined; + dispatch({ type: "draft-refused", message: copy }); + }); + return; + } + if (!snapshot) return; + dispatch({ type: "start" }); + startBuild(room, snapshot).then(result => { + if (result === "started") { + dispatch({ type: "started" }); + setRefresh(value => value + 1); + return; + } + dispatch({ type: "start-refused", agent: true }); + onNeedsAgent(); + }, error => { + dispatch({ + type: "start-refused", + agent: false, + message: error instanceof Error + ? error.message.replace(/^[a-z]/, letter => letter.toUpperCase()) + : undefined, + }); + }); + }, [step.next, snapshot?.revision]); + + let sync = syncStatus(snapshot); + if (sync) { + let hint = syncHint(sync, snapshot, userId); + return ( + + ); + } + if (step.view === "hidden") return null; + if (step.view === "waiting") { + let { label, target } = waitingLabel(snapshot); + let hint = "Building starts once this is resolved"; + return ( + + ); + } + if (step.view === "working") { + let { hint, label, queued } = startingLabel(snapshot); + return ( + + ); + } + let failure = state.stage === "failed" && !state.agent + ? state.message ?? "Chopin couldn’t start the build." + : undefined; + return ( + <> + + {failure && {failure}} + + ); +} diff --git a/apps/web/src/build-start.ts b/apps/web/src/build-start.ts new file mode 100644 index 000000000..8b2e62f79 --- /dev/null +++ b/apps/web/src/build-start.ts @@ -0,0 +1,61 @@ +import { ApiError } from "./api"; + +import type { ImplementationSnapshot } from "@chopin/protocol/implementation"; + +export function implementationEndpoint(room: string): string { + return `/api/channels/${encodeURIComponent(room)}/implementation`; +} + +export async function implementationResponse(result: Response): Promise { + let value = await result.json(); + if (!result.ok) throw new ApiError(value.error ?? "Build is unavailable", result.status); + return value; +} + +/** + * The planned `no-workspace` 409: no local agent is connected. A connected agent that is still + * prototyping takes the build once it finishes, so busy is not offline. + */ +function offline(error: unknown): boolean { + return error instanceof ApiError && error.status === 409 && error.message === "no-workspace"; +} + +/** + * Approve the snapshot's tasks and start building them on the viewer's local agent. + * + * The one place the browser starts a build. `needs-agent` means no connected + * agent could take it; any other refusal throws. + */ +export async function startBuild( + room: string, + snapshot: ImplementationSnapshot, +): Promise<"started" | "needs-agent"> { + let graph = snapshot.graph; + if (!graph) throw new ApiError("There are no tasks to build", 409); + // The server picks the clicker's own connection; a snapshot can predate connecting. + try { + await implementationResponse( + await fetch(implementationEndpoint(room), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + ...(snapshot.build ? { retryOf: snapshot.build.id } : {}), + planRevision: snapshot.planRevision, + graphVersion: graph.number, + graphRevision: graph.revision, + }), + }), + ); + return "started"; + } catch (error) { + if (offline(error)) return "needs-agent"; + throw error; + } +} + +/** Give up a one-click Build plan that is still waiting for its tasks. */ +export async function cancelBuildRequest(room: string): Promise { + await implementationResponse( + await fetch(`${implementationEndpoint(room)}/request`, { method: "DELETE" }), + ); +} diff --git a/apps/web/src/build-view.css b/apps/web/src/build-view.css index 396004643..b5d4c6664 100644 --- a/apps/web/src/build-view.css +++ b/apps/web/src/build-view.css @@ -300,6 +300,35 @@ color: var(--color-warning-ink); } +/* A living document's commits on a task's pull request, newest first. */ +.build-task-commits { + display: grid; + gap: var(--spacing); + margin: 0; + padding: 0; + list-style: none; +} + +.build-task-commits > li { + display: flex; + align-items: baseline; + gap: calc(var(--spacing) * 2); + min-width: 0; +} + +.build-task-sha { + flex: none; + color: var(--color-text-tertiary); + font-family: var(--font-mono); + font-size: var(--text-xs); + text-decoration: none; +} + +.build-task-sha:hover { + color: var(--color-brand); + text-decoration: underline; +} + .build-hint { margin: calc(var(--spacing) * 3) 0 0; color: var(--color-text-tertiary); diff --git a/apps/web/src/build-view.tsx b/apps/web/src/build-view.tsx index c4b92ac31..038fd7d4e 100644 --- a/apps/web/src/build-view.tsx +++ b/apps/web/src/build-view.tsx @@ -1,8 +1,8 @@ import { useEffect, useId, useRef, useState } from "react"; -import { ApiError } from "./api"; import { advanceDraft, + ago, buildPhase, buildProgress, draftInFlight, @@ -10,12 +10,22 @@ import { draftRefusalCopy, elapsed, plural, + pullRequestCommits, pullRequestNumber, shouldAutoDraft, startedBy, + startingLabel, + SYNC_LABEL, + syncHint, TASK_STATE_LABEL, taskStartsOpen, } from "./build-model"; +import { + cancelBuildRequest, + implementationEndpoint, + implementationResponse as response, + startBuild, +} from "./build-start"; import type { Implementation, ImplementationSnapshot } from "@chopin/protocol/implementation"; import type { ReactNode } from "react"; @@ -29,28 +39,17 @@ const requests = new Map(); /** How long a request may wait for the server's answer before it counts as failed. */ const ANSWER_WINDOW_MS = 10_000; -async function response(result: Response): Promise { - let value = await result.json(); - if (!result.ok) throw new ApiError(value.error ?? "Build is unavailable", result.status); - return value; -} - function sentence(text: string): string { return text.replace(/^[a-z]/, letter => letter.toUpperCase()); } -/** The server's refusals for a connection it cannot reach mean the local agent is not running. */ -function offline(error: unknown): boolean { - return error instanceof ApiError && error.status === 409 - && /^no-workspace$|offline or busy/.test(error.message); -} - export function BuildView( { active, canEdit, chatLoaded, comments, + needsAgentSignal = 0, onShowDecisions, onShowDocument, planner, @@ -66,6 +65,8 @@ export function BuildView( chatLoaded: boolean; /** Accepted comments the document has not yet taken in. */ comments: number; + /** Increments when the header's Build plan found no local agent, to show how to start one. */ + needsAgentSignal?: number; onShowDecisions: () => void; onShowDocument: () => void; planner: boolean; @@ -105,7 +106,7 @@ export function BuildView( */ let activation = useRef({ id: 0, decided: false, eligible: undefined as string | undefined }); let [autoSent, setAutoSent] = useState(0); - let endpoint = `/api/channels/${encodeURIComponent(room)}/implementation`; + let endpoint = implementationEndpoint(room); let phase = buildPhase(snapshot); let progress = buildProgress(snapshot); let draft = phase.kind === "drafting" ? phase.draft : undefined; @@ -221,7 +222,7 @@ export function BuildView( return () => window.clearTimeout(timer); }, [request?.state, request?.key]); - let pending = phase.kind === "building"; + let pending = phase.kind === "building" || phase.kind === "live"; useEffect(() => { if (!pending || !active) return; setNow(Date.now()); @@ -231,6 +232,9 @@ export function BuildView( // The local-agent note answers one press of one button; a new phase clears it. useEffect(() => setNeedsAgent(false), [phase.kind]); + useEffect(() => { + if (needsAgentSignal > 0) setNeedsAgent(true); + }, [needsAgentSignal]); useEffect(() => { let follow = () => { @@ -255,26 +259,13 @@ export function BuildView( let start = async () => { if (!snapshot?.graph || busy) return; setActionError(undefined); - // The server picks the clicker's own connection; a snapshot can predate connecting. setNeedsAgent(false); setBusy(true); try { - await response( - await fetch(endpoint, { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ - ...(snapshot.build ? { retryOf: snapshot.build.id } : {}), - planRevision: snapshot.planRevision, - graphVersion: snapshot.graph.number, - graphRevision: snapshot.graph.revision, - }), - }), - ); - setRefresh(value => value + 1); + if (await startBuild(room, snapshot) === "needs-agent") setNeedsAgent(true); + else setRefresh(value => value + 1); } catch (error) { - if (offline(error)) setNeedsAgent(true); - else setActionError(sentence(error instanceof Error ? error.message : "Build failed")); + setActionError(sentence(error instanceof Error ? error.message : "Build failed")); } finally { setBusy(false); } @@ -302,7 +293,10 @@ export function BuildView( } }; - let tasks = snapshot?.graph?.definition.tasks ?? []; + // A living document lists its first build's tasks and every rebuild's after them. + let tasks = phase.kind === "live" + ? snapshot?.live?.tasks ?? [] + : snapshot?.graph?.definition.tasks ?? []; // Drafting: our turn is under way, or Build is about to ask, or it failed, or a person decides. let awaitingAuto = !!key && key === activation.current.eligible && autoSent !== activation.current.id && !drafted.has(key); @@ -393,17 +387,42 @@ export function BuildView( ) : "This document can’t be built yet."; + // A one-click Build plan waits here and starts once the blockers clear. + if (canEdit && userId && snapshot?.buildRequested?.by === userId) { + action = ( + + ); + } } else if (phase.kind === "review") { status = plural(tasks.length, "task", "tasks"); if (canEdit) action = primary; } else if (phase.kind === "building") { since = snapshot?.build && elapsed(snapshot.build.createdAt, now); let who = startedBy(snapshot, userId); + let { hint, queued } = startingLabel(snapshot); status = ( <> -