diff --git a/apps/connector/src/image-file.test.ts b/apps/connector/src/image-file.test.ts new file mode 100644 index 000000000..ec5a22121 --- /dev/null +++ b/apps/connector/src/image-file.test.ts @@ -0,0 +1,114 @@ +import { afterEach, beforeEach, expect, test } from "bun:test"; +import { mkdir, mkdtemp, rm, symlink, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { Client } from "@modelcontextprotocol/sdk/client/index.js"; +import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js"; + +import { imageFileArguments, MAX_IMAGE_FILE_BYTES } from "./image-file"; +import { implementationBridge } from "./implementation"; + +type BridgeApi = Parameters[0]; + +const PNG = new Uint8Array([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 1, 2, 3]); + +let base: string; +let root: string; +beforeEach(async () => { + base = await mkdtemp(join(tmpdir(), "chopin-image-")); + root = join(base, "worktree"); + await mkdir(join(root, "shots"), { recursive: true }); +}); +afterEach(() => rm(base, { recursive: true, force: true })); + +test("reads an image inside the worktree as upload arguments", async () => { + await writeFile(join(root, "shots", "a.png"), PNG); + expect(await imageFileArguments(root, "shots/a.png")).toEqual({ + data: Buffer.from(PNG).toString("base64"), + mimeType: "image/png", + }); + expect((await imageFileArguments(root, join(root, "shots", "a.png"))).mimeType).toBe( + "image/png", + ); +}); + +test("refuses paths outside the worktree, including through a symlink", async () => { + await writeFile(join(base, "outside.png"), PNG); + await symlink(join(base, "outside.png"), join(root, "link.png")); + await expect(imageFileArguments(root, "../outside.png")).rejects.toThrow("inside this run"); + await expect(imageFileArguments(root, join(base, "outside.png"))).rejects.toThrow( + "inside this run", + ); + await expect(imageFileArguments(root, "link.png")).rejects.toThrow("inside this run"); + await expect(imageFileArguments(root, "missing.png")).rejects.toThrow("No file"); +}); + +test("refuses non-images and images over 1 MiB with a smaller-screenshot hint", async () => { + await writeFile(join(root, "notes.txt"), "hello"); + await expect(imageFileArguments(root, "notes.txt")).rejects.toThrow("PNG, JPEG or WebP"); + let large = new Uint8Array(MAX_IMAGE_FILE_BYTES + 1); + large.set(PNG); + await writeFile(join(root, "large.png"), large); + await expect(imageFileArguments(root, "large.png")).rejects.toThrow("1280x800"); +}); + +test("the spike bridge serves upload_image_file by forwarding the file's bytes", async () => { + await writeFile(join(root, "shot.png"), PNG); + let invoked: Array<{ name: string; args: Record }> = []; + let api = { + tools: async () => ({ + tools: [{ + name: "upload_investigation_image", + inputSchema: { type: "object" as const }, + }], + }), + invoke: async (name: string, args: Record = {}) => { + invoked.push({ name, args }); + return { content: [{ type: "text", text: JSON.stringify({ path: "/images/x.png" }) }] }; + }, + }; + let server = await implementationBridge(api as unknown as BridgeApi, root); + let [client, transport] = InMemoryTransport.createLinkedPair(); + await server.connect(transport); + let mcp = new Client({ name: "test", version: "1" }); + await mcp.connect(client); + try { + let names = (await mcp.listTools()).tools.map(tool => tool.name); + expect(names).toEqual(["upload_investigation_image", "upload_image_file"]); + let result = await mcp.callTool({ name: "upload_image_file", arguments: { path: "shot.png" } }); + expect(result.isError).toBeFalsy(); + expect(invoked).toEqual([{ + name: "upload_investigation_image", + args: { data: Buffer.from(PNG).toString("base64"), mimeType: "image/png" }, + }]); + let refused = await mcp.callTool({ + name: "upload_image_file", + arguments: { path: "../escape.png" }, + }); + expect(refused.isError).toBe(true); + expect(invoked).toHaveLength(1); + } finally { + await mcp.close(); + await server.close(); + } +}); + +test("bridges without a worktree or image upload offer no local tool", async () => { + let api = { + tools: async () => ({ + tools: [{ name: "read_rebuild", inputSchema: { type: "object" as const } }], + }), + invoke: async () => ({ content: [] }), + }; + let server = await implementationBridge(api as unknown as BridgeApi, root); + let [client, transport] = InMemoryTransport.createLinkedPair(); + await server.connect(transport); + let mcp = new Client({ name: "test", version: "1" }); + await mcp.connect(client); + try { + expect((await mcp.listTools()).tools.map(tool => tool.name)).toEqual(["read_rebuild"]); + } finally { + await mcp.close(); + await server.close(); + } +}); diff --git a/apps/connector/src/image-file.ts b/apps/connector/src/image-file.ts new file mode 100644 index 000000000..e35a85330 --- /dev/null +++ b/apps/connector/src/image-file.ts @@ -0,0 +1,44 @@ +import { readFile, realpath } from "node:fs/promises"; +import { isAbsolute, relative, resolve } from "node:path"; + +/** Chopin hosts images of at most 1 MiB. */ +export const MAX_IMAGE_FILE_BYTES = 1024 * 1024; + +const TOO_LARGE = + "Take a smaller screenshot (a 1280x800 viewport PNG of the running prototype) and upload that."; + +function mimeType(bytes: Uint8Array): string | undefined { + let ascii = (start: number, end: number) => String.fromCharCode(...bytes.subarray(start, end)); + if (bytes[0] === 0x89 && ascii(1, 4) === "PNG") return "image/png"; + if (bytes[0] === 0xff && bytes[1] === 0xd8 && bytes[2] === 0xff) return "image/jpeg"; + if (ascii(0, 4) === "RIFF" && ascii(8, 12) === "WEBP") return "image/webp"; +} + +/** + * Read a PNG, JPEG or WebP inside the run's worktree as upload_investigation_image arguments, so + * the agent passes a path instead of emitting the image's base64 as tokens. + */ +export async function imageFileArguments( + root: string, + path: string, +): Promise<{ data: string; mimeType: string }> { + let base = await realpath(root); + let target: string; + try { + target = await realpath(resolve(base, path)); + } catch { + throw new Error(`No file at ${path} in this worktree.`); + } + let inside = relative(base, target); + if (!inside || inside.startsWith("..") || isAbsolute(inside)) { + throw new Error("Upload only image files inside this run's worktree."); + } + let bytes = new Uint8Array(await readFile(target)); + let type = mimeType(bytes); + if (!type) throw new Error("Upload a PNG, JPEG or WebP image."); + if (bytes.byteLength > MAX_IMAGE_FILE_BYTES) { + let size = Math.ceil(bytes.byteLength / 1024); + throw new Error(`${path} is ${size} KiB; images are limited to 1 MiB. ${TOO_LARGE}`); + } + return { data: Buffer.from(bytes).toString("base64"), mimeType: type }; +} diff --git a/apps/connector/src/implementation.ts b/apps/connector/src/implementation.ts index 3628c09b5..e77d668ea 100644 --- a/apps/connector/src/implementation.ts +++ b/apps/connector/src/implementation.ts @@ -1,21 +1,78 @@ import { Server } from "@modelcontextprotocol/sdk/server/index.js"; import { CallToolRequestSchema, ListToolsRequestSchema } from "@modelcontextprotocol/sdk/types.js"; +import { imageFileArguments } from "./image-file"; import type { remote } from "./mcp"; -export async function implementationBridge(api: Awaited>) { - let tools = await api.tools(); - let names = new Set(tools.tools.map(tool => tool.name)); - let server = new Server({ name: "chopin-implementation", version: "0.1.0" }, { - capabilities: { tools: {} }, - }); - server.setRequestHandler(ListToolsRequestSchema, () => tools); - server.setRequestHandler(CallToolRequestSchema, async request => { - if (!names.has(request.params.name)) { - return { isError: true, content: [{ type: "text", text: "Tool unavailable for this run." }] }; +/** + * Relay a run's server tools. With a worktree `root`, a run that may upload images also gets the + * connector-local upload_image_file, which reads the file here and forwards its bytes. + */ +export async function implementationBridge( + api: Pick>, "tools" | "invoke">, + root?: string, +) { + return (await relay(api, root))(); +} + +/** List the run's tools once and return a factory for relaying servers, one per transport. */ +export async function relay( + api: Pick>, "tools" | "invoke">, + root?: string, +) { + let listed = await api.tools(); + let names = new Set(listed.tools.map(tool => tool.name)); + let local = !!root && names.has("upload_investigation_image"); + let tools = local + ? { + ...listed, + tools: [...listed.tools, { + name: "upload_image_file", + description: + "Upload a PNG, JPEG or WebP screenshot (at most 1 MiB) saved in this worktree and " + + "return its image path for submit_spike_result.", + inputSchema: { + type: "object" as const, + properties: { + path: { + type: "string", + description: "The image file, relative to the worktree root.", + }, + }, + required: ["path"], + additionalProperties: false, + }, + }], } - return api.invoke(request.params.name, request.params.arguments ?? {}); - }); - return server; + : listed; + return () => { + let server = new Server({ name: "chopin-implementation", version: "0.1.0" }, { + capabilities: { tools: {} }, + }); + server.setRequestHandler(ListToolsRequestSchema, () => tools); + server.setRequestHandler(CallToolRequestSchema, async request => { + let args = request.params.arguments ?? {}; + if (local && request.params.name === "upload_image_file") { + try { + if (typeof args.path !== "string" || !args.path) throw new Error("Give an image path."); + return await api.invoke( + "upload_investigation_image", + await imageFileArguments(root!, args.path), + ); + } catch (error) { + let text = error instanceof Error ? error.message : "Image upload failed."; + return { isError: true, content: [{ type: "text", text }] }; + } + } + if (!names.has(request.params.name)) { + return { + isError: true, + content: [{ type: "text", text: "Tool unavailable for this run." }], + }; + } + return api.invoke(request.params.name, args); + }); + return server; + }; } export let implementationPrompt = [ diff --git a/apps/connector/src/local-bridge.ts b/apps/connector/src/local-bridge.ts new file mode 100644 index 000000000..f4291eb1b --- /dev/null +++ b/apps/connector/src/local-bridge.ts @@ -0,0 +1,42 @@ +import { randomBytes, timingSafeEqual } from "node:crypto"; +import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js"; +import type { Server } from "@modelcontextprotocol/sdk/server/index.js"; + +/** + * Serve run-scoped MCP tools to a local ACP agent over loopback streamable HTTP. Some agents (Copilot + * CLI in ACP mode) reject client-provided stdio servers, so connector-local tools need an HTTP URL. + * Each request gets a fresh stateless server; a per-run bearer token keeps other local processes + * and pages out. + */ +export function serveLocalBridge(open: () => Promise) { + let token = randomBytes(32).toString("base64url"); + let expected = Buffer.from(`Bearer ${token}`); + let http = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + idleTimeout: 255, + async fetch(request) { + let given = Buffer.from(request.headers.get("authorization") ?? ""); + if (given.length !== expected.length || !timingSafeEqual(given, expected)) { + return new Response(null, { status: 401 }); + } + if (new URL(request.url).pathname !== "/mcp") return new Response(null, { status: 404 }); + let server = await open(); + let transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: undefined, + enableJsonResponse: true, + }); + await server.connect(transport); + try { + return await transport.handleRequest(request); + } finally { + void server.close(); + } + }, + }); + return { + url: `http://127.0.0.1:${http.port}/mcp`, + token, + close: () => http.stop(true), + }; +} diff --git a/apps/connector/src/main.ts b/apps/connector/src/main.ts index 6f7af116f..193f3b591 100644 --- a/apps/connector/src/main.ts +++ b/apps/connector/src/main.ts @@ -25,8 +25,10 @@ async function bridge() { process.env.CHOPIN_BRIDGE_TOKEN!, ); let server; - if (process.env.CHOPIN_BRIDGE_KIND === "implementation") server = await implementationBridge(api); - else { + // A spike's tools are the server's run-scoped list, relayed as they are. + if (["implementation", "spike"].includes(process.env.CHOPIN_BRIDGE_KIND ?? "")) { + server = await implementationBridge(api, process.env.CHOPIN_BRIDGE_ROOT); + } else { let context = await api.call("read_experiment") as { input: unknown }; server = createBridge(requestSchema.parse(context.input), async result => { await api.call("submit_experiment_result", { result }); diff --git a/apps/connector/src/run.test.ts b/apps/connector/src/run.test.ts index f50ab4ddf..a73862591 100644 --- a/apps/connector/src/run.test.ts +++ b/apps/connector/src/run.test.ts @@ -5,6 +5,9 @@ import { join } from "node:path"; import { runWork } from "./run"; import { git } from "./workspace"; import type { remote } from "./mcp"; +import { Server } from "@modelcontextprotocol/sdk/server/index.js"; +import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js"; +import { CallToolRequestSchema, ListToolsRequestSchema } from "@modelcontextprotocol/sdk/types.js"; /** Run one implementation claim against a fixture repository and record connector calls. */ async function run( @@ -89,3 +92,96 @@ test("a living document's first build gets a prompt without verification or revi { id, state: "stopped" }, ]); }); + +test("a spike run serves connector-local tools to an HTTP agent over an authenticated loopback bridge", async () => { + let root = await mkdtemp(join(tmpdir(), "chopin-spike-")); + let state = await mkdtemp(join(tmpdir(), "chopin-spike-state-")); + let received: Array<[string, Record]> = []; + let chopin = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + async fetch(request) { + if (request.headers.get("authorization") !== "Bearer run-token") { + return new Response(null, { status: 401 }); + } + let server = new Server({ name: "fake-chopin", version: "1" }, { + capabilities: { tools: {} }, + }); + server.setRequestHandler(ListToolsRequestSchema, () => ({ + tools: ["upload_investigation_image", "submit_spike_result"].map(name => ({ + name, + inputSchema: { type: "object" as const }, + })), + })); + server.setRequestHandler(CallToolRequestSchema, request => { + received.push([request.params.name, request.params.arguments ?? {}]); + return { content: [{ type: "text", text: "{}" }] }; + }); + let transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: undefined, + enableJsonResponse: true, + }); + await server.connect(transport); + return transport.handleRequest(request); + }, + }); + try { + git(root, "init"); + git(root, "config", "user.name", "Fixture"); + git(root, "config", "user.email", "fixture@example.test"); + git(root, "remote", "add", "origin", "git@github.com:org/repo.git"); + await writeFile(join(root, "source.txt"), "committed"); + git(root, "add", "."); + git(root, "commit", "-m", "fixture"); + let calls: string[] = []; + let api = { + call: async (name: string) => { + calls.push(name); + return {}; + }, + } as unknown as Awaited>; + let id = crypto.randomUUID(); + await runWork(api, "experiment", { + spike: true, + generation: 1, + runToken: "run-token", + input: { + id, + documentId: crypto.randomUUID(), + brief: "Prototype the passage", + source: { + repositoryId: "R_repo", + repository: "org/repo", + commit: git(root, "rev-parse", "HEAD"), + }, + context: "", + requester: "a", + authorizer: "a", + }, + }, { + root, + directory: state, + command: [ + process.execPath, + new URL("./testing/fake-agent.ts", import.meta.url).pathname, + "--http", + "--spike", + ], + url: `http://127.0.0.1:${chopin.port}`, + script: "unused", + signal: new AbortController().signal, + }); + expect(calls.filter(name => name !== "renew_experiment")).toEqual(["complete_experiment"]); + expect(received.map(([name]) => name)).toEqual([ + "upload_investigation_image", + "submit_spike_result", + ]); + expect(received[0][1]).toEqual({ data: "iVBORw0KGgo=", mimeType: "image/png" }); + // The bridge closes with the run. + await expect(fetch(received[1][1].url as string, { method: "POST" })).rejects.toThrow(); + } finally { + await chopin.stop(true); + await rm(root, { recursive: true, force: true }); + await rm(state, { recursive: true, force: true }); + } +}); diff --git a/apps/connector/src/run.ts b/apps/connector/src/run.ts index 67ee9700e..469b0e13a 100644 --- a/apps/connector/src/run.ts +++ b/apps/connector/src/run.ts @@ -1,10 +1,16 @@ import { createInterface } from "node:readline/promises"; import { requestSchema } from "@chopin/experiment"; import type { BuildRequest } from "@chopin/protocol/implementation"; -import type { remote } from "./mcp"; +import { remote } from "./mcp"; import { runAgent } from "./acp"; +import { serveLocalBridge } from "./local-bridge"; import { git, prepareWorkspace } from "./workspace"; -import { implementationPrompt, liveImplementationPrompt, rebuildPrompt } from "./implementation"; +import { + implementationPrompt, + liveImplementationPrompt, + rebuildPrompt, + relay, +} from "./implementation"; type Options = { root: string; @@ -26,6 +32,8 @@ export async function runWork( build: BuildRequest; /** A living document's first build, whose run ends at its last complete_task. */ live?: boolean; + /** An unprompted prototype under a passage, run on a throwaway branch. */ + spike?: boolean; input: unknown; generation: number; runToken: string; @@ -55,12 +63,15 @@ export async function runWork( if ((result as { state?: string } | undefined)?.state === "stopped") clearInterval(heartbeat); }).catch(() => abort.abort()).finally(() => renewing = false); }, options.heartbeatMs ?? 10_000); + let local: ReturnType | undefined; + let bridgeApi: Promise>> | undefined; try { let prepared = await prepareWorkspace(options.root, options.directory, input); // A rebuild commits onto the existing pull request branches the agent checks out itself. if (implementation && !rebuild) { git(prepared.path, "checkout", "-b", `chopin/implement-${input.id.slice(0, 8)}`); } + if (claim.spike) git(prepared.path, "checkout", "-b", `chopin/spike-${input.id.slice(0, 8)}`); console.error(`Running ${input.id} in ${prepared.path}`); let stop = await runAgent({ command: options.command, @@ -69,28 +80,58 @@ export async function runWork( ? rebuildPrompt : implementation ? claim.live ? liveImplementationPrompt : implementationPrompt + : claim.spike + ? "Use read_investigation from the Chopin MCP server to read the spike brief and " + + "follow it: build the smallest prototype on this throwaway branch, take 1-3 " + + "1280x800 PNG screenshots of the running prototype with Playwright, save them in " + + "this worktree, upload each with upload_image_file({path}), and report with " + + "submit_spike_result, then finish. Never push." : "Use read_investigation from the Chopin MCP server to read the authorized request. " + "Perform that investigation using your normal project instructions and tools. " + "Submit a bounded result with submit_investigation_result, then finish. " + "Do not commit or push unless the authorized brief specifically requests it.", - mcpServers: http => - http - ? [{ - name: implementation ? "chopin-implementation" : "chopin-investigation", + // A spike needs the connector-local upload_image_file, so its tools come from this process: + // a loopback HTTP bridge when the agent accepts HTTP servers, the stdio bridge otherwise. + mcpServers(http) { + let name = implementation ? "chopin-implementation" : "chopin-investigation"; + if (http && claim.spike) { + let tools: ReturnType | undefined; + local = serveLocalBridge(() => { + tools ??= (bridgeApi ??= remote(options.url, claim.runToken)) + .then(api => relay(api, prepared.path)) + .catch(error => { + tools = bridgeApi = undefined; + throw error; + }); + return tools.then(make => make()); + }); + return [{ + name, + type: "http", + url: local.url, + headers: [{ name: "Authorization", value: `Bearer ${local.token}` }], + }]; + } + if (http) { + return [{ + name, type: "http", url: options.url + "/connector/mcp", headers: [{ name: "Authorization", value: `Bearer ${claim.runToken}` }], - }] - : [{ - name: implementation ? "chopin-implementation" : "chopin-investigation", - command: process.execPath, - args: [options.script, "bridge"], - env: [ - { name: "CHOPIN_BRIDGE_ORIGIN", value: options.url }, - { name: "CHOPIN_BRIDGE_TOKEN", value: claim.runToken }, - { name: "CHOPIN_BRIDGE_KIND", value: kind }, - ], - }], + }]; + } + return [{ + name, + command: process.execPath, + args: [options.script, "bridge"], + env: [ + { name: "CHOPIN_BRIDGE_ORIGIN", value: options.url }, + { name: "CHOPIN_BRIDGE_TOKEN", value: claim.runToken }, + { name: "CHOPIN_BRIDGE_KIND", value: claim.spike ? "spike" : kind }, + ...(claim.spike ? [{ name: "CHOPIN_BRIDGE_ROOT", value: prepared.path }] : []), + ], + }]; + }, signal: abort.signal, onSession: implementation ? session => @@ -139,6 +180,8 @@ export async function runWork( : message.slice(0, 2000), }).catch(() => {}); } finally { + local?.close(); + await bridgeApi?.then(api => api.close()).catch(() => {}); clearInterval(heartbeat); options.signal.removeEventListener("abort", cancel); } diff --git a/apps/connector/src/testing/fake-agent.ts b/apps/connector/src/testing/fake-agent.ts index c1db0b933..0000fdf2f 100644 --- a/apps/connector/src/testing/fake-agent.ts +++ b/apps/connector/src/testing/fake-agent.ts @@ -1,4 +1,6 @@ import { AgentApp, ndJsonStream, PROTOCOL_VERSION } from "@agentclientprotocol/sdk"; +import { writeFile } from "node:fs/promises"; +import { join } from "node:path"; import { Readable, Writable } from "node:stream"; import { Client } from "@modelcontextprotocol/sdk/client/index.js"; import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; @@ -7,6 +9,7 @@ import { performance } from "@chopin/experiment/fixtures"; import type { McpServer } from "@agentclientprotocol/sdk"; let servers: McpServer[] = []; +let cwd = ""; let cancelled = false; let release: (() => void) | undefined; @@ -20,6 +23,7 @@ let app = new AgentApp() ) .onRequest("session/new", ({ params: input }) => { servers = input.mcpServers; + cwd = input.cwd; if (!input.cwd.startsWith("/")) throw new Error("Expected absolute cwd"); return { sessionId: "test-session" }; }) @@ -39,6 +43,40 @@ let app = new AgentApp() return { stopReason: "cancelled" as const }; } let config = servers.find(server => server.name === "chopin-investigation"); + if (process.argv.includes("--spike")) { + if (!config || !("url" in config) || !config.url.startsWith("http://127.0.0.1:")) { + throw new Error("Expected a loopback HTTP bridge"); + } + let anonymous = await fetch(config.url, { method: "POST", body: "{}" }); + if (anonymous.status !== 401) throw new Error(`Unauthenticated bridge: ${anonymous.status}`); + let png = Buffer.from("89504e470d0a1a0a", "hex"); + await writeFile(join(cwd, "shot.png"), png); + await writeFile(join(cwd, "..", "outside.png"), png); + let bridge = new Client({ name: "fake-spiker", version: "1" }); + await bridge.connect( + new StreamableHTTPClientTransport(new URL(config.url), { + requestInit: { + headers: Object.fromEntries(config.headers.map(item => [item.name, item.value])), + }, + }), + ); + try { + let shot = await bridge.callTool({ + name: "upload_image_file", + arguments: { path: "shot.png" }, + }); + if (shot.isError) throw new Error(JSON.stringify(shot.content)); + let escape = await bridge.callTool({ + name: "upload_image_file", + arguments: { path: "../outside.png" }, + }); + if (!escape.isError) throw new Error("Uploaded a file outside the worktree"); + await bridge.callTool({ name: "submit_spike_result", arguments: { url: config.url } }); + } finally { + await bridge.close(); + } + return { stopReason: "end_turn" as const }; + } if (config && ("command" in config || "url" in config)) { let bridge = new Client({ name: "fake-investigator", version: "1" }); await bridge.connect( diff --git a/apps/connector/src/workspace.test.ts b/apps/connector/src/workspace.test.ts index 39e8b92c0..97836563e 100644 --- a/apps/connector/src/workspace.test.ts +++ b/apps/connector/src/workspace.test.ts @@ -1,5 +1,5 @@ import { expect, test } from "bun:test"; -import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { mkdtemp, readdir, readFile, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { git, lockWorkspace, prepareWorkspace, workspace } from "./workspace"; @@ -40,3 +40,32 @@ test("isolates exact source without stashing dirty work and never executes an ac await rm(state, { recursive: true, force: true }); } }); + +test("replaces a lock left by a connector whose process is gone, and keeps a live one", async () => { + let state = await mkdtemp(join(tmpdir(), "chopin-state-")); + try { + let root = "/fixture/checkout"; + let release = await lockWorkspace(root, state); + let [lock] = (await readdir(state)).filter(name => name.endsWith(".lock")); + let path = join(state, lock); + await release(); + + // A live process, even when it is not this one, still holds the workspace. + await writeFile(path, JSON.stringify({ pid: process.ppid, root })); + await expect(lockWorkspace(root, state)).rejects.toThrow("already locked"); + + // An exited process's lock is stale. + let child = Bun.spawn(["true"]); + await child.exited; + await writeFile(path, JSON.stringify({ pid: child.pid, root })); + let replaced = await lockWorkspace(root, state); + expect(JSON.parse(await readFile(path, "utf8")).pid).toBe(process.pid); + await replaced(); + + // A lock that does not name a process is left for a person to check. + await writeFile(path, ""); + await expect(lockWorkspace(root, state)).rejects.toThrow("already locked"); + } finally { + await rm(state, { recursive: true, force: true }); + } +}); diff --git a/apps/connector/src/workspace.ts b/apps/connector/src/workspace.ts index be11705a4..4e5097bbf 100644 --- a/apps/connector/src/workspace.ts +++ b/apps/connector/src/workspace.ts @@ -1,5 +1,5 @@ import { createHash } from "node:crypto"; -import { mkdir, open, realpath, rm } from "node:fs/promises"; +import { mkdir, open, readFile, realpath, rm } from "node:fs/promises"; import { homedir } from "node:os"; import { join } from "node:path"; import { execFile, spawnSync } from "node:child_process"; @@ -44,9 +44,18 @@ export async function lockWorkspace(root: string, directory: string) { try { file = await open(path, "wx", 0o600); } catch { - throw new Error( - `Workspace already locked. After confirming no connector is running, remove ${path}`, - ); + // A connector that crashed leaves its lock behind; replace it once its process is gone. + if (!await staleLock(path)) { + throw new Error( + `Workspace already locked. After confirming no connector is running, remove ${path}`, + ); + } + await rm(path, { force: true }); + try { + file = await open(path, "wx", 0o600); + } catch { + throw new Error(`Workspace already locked by a connector that just started. ${path}`); + } } await file.writeFile(JSON.stringify({ pid: process.pid, root })); return async () => { @@ -55,6 +64,24 @@ export async function lockWorkspace(root: string, directory: string) { }; } +/** Only a lock naming a process that no longer exists is stale; unreadable locks are kept. */ +async function staleLock(path: string): Promise { + let pid: unknown; + try { + pid = JSON.parse(await readFile(path, "utf8")).pid; + } catch { + return false; + } + if (!Number.isSafeInteger(pid) || (pid as number) <= 0) return false; + try { + process.kill(pid as number, 0); + return false; + } catch (err) { + // EPERM means the process exists but belongs to someone else. + return (err as NodeJS.ErrnoException).code === "ESRCH"; + } +} + export async function prepareWorkspace( root: string, directory: string, diff --git a/apps/server/src/config.test.ts b/apps/server/src/config.test.ts index 5a0242d86..8700f92a3 100644 --- a/apps/server/src/config.test.ts +++ b/apps/server/src/config.test.ts @@ -39,6 +39,9 @@ describe("configuration", () => { expect(configured().liveBuild).toBe(false); expect(configured({ LIVE_BUILD: "yes" }).liveBuild).toBe(false); expect(configured({ LIVE_BUILD: "on" }).liveBuild).toBe(true); + expect(configured().liveSpikes).toBe(false); + expect(configured({ LIVE_SPIKES: "on" }).liveSpikes).toBe(true); + expect(configured({ LIVE_BUILD: "on" }).liveSpikes).toBe(false); expect(description(configured({ LIVE_BUILD: "on" }))).toContain("live build: on"); }); diff --git a/apps/server/src/config.ts b/apps/server/src/config.ts index cdb27c40c..8034d7b96 100644 --- a/apps/server/src/config.ts +++ b/apps/server/src/config.ts @@ -45,6 +45,8 @@ export type Config = { plannerVisuals?: boolean; /** Keep a finished build's document as the source for later rebuilds. */ liveBuild?: boolean; + /** Let the editor's local agent prototype uncertain passages before the first build. */ + liveSpikes?: boolean; /** * Origin of a running Vite, when developing. * @@ -119,6 +121,7 @@ export function load(): Config { let conversationPlan = process.env.CONVERSATION_PLAN === "on"; let plannerVisuals = process.env.PLANNER_VISUALS === "on" && agent; let liveBuild = process.env.LIVE_BUILD === "on"; + let liveSpikes = process.env.LIVE_SPIKES === "on"; let conversationPlanModel = process.env.JEV_MODEL || "jev-latest"; let timeoutRaw = process.env.JEV_TIMEOUT_MS; let conversationPlanTimeoutMs = timeoutRaw === undefined ? 30_000 : Number(timeoutRaw); @@ -154,6 +157,7 @@ export function load(): Config { conversationPlanTimeoutMs, plannerVisuals, liveBuild, + liveSpikes, devClient: process.env.DEV_CLIENT || undefined, storage: storage(), auth: loadAuth(serverPort, selection.host), @@ -196,6 +200,7 @@ export function describe(config: Config): string { ? `Planner visuals: ${config.conversationPlanModel}` : "Planner visuals: off", config.liveBuild ? "live build: on" : "live build: off", + config.liveSpikes ? "live spikes: on" : "live spikes: off", admission, `storage: ${config.storage.driver}`, ]; diff --git a/apps/server/src/experiments/mcp-schema.ts b/apps/server/src/experiments/mcp-schema.ts index 0291951d3..f07946595 100644 --- a/apps/server/src/experiments/mcp-schema.ts +++ b/apps/server/src/experiments/mcp-schema.ts @@ -1,8 +1,10 @@ import { z } from "zod"; import { resultSchema } from "@chopin/experiment"; +import { spikeSubmissionSchema } from "./spikes"; export function connectorSchemas( runScoped: boolean, + spike = false, ): Record>> { let claim = { id: z.string().uuid(), generation: z.number().int().positive() }; if (runScoped) { @@ -11,6 +13,15 @@ export function connectorSchemas( read_investigation: z.object({}).strict(), submit_experiment_result: z.object({ result: resultSchema }).strict(), submit_investigation_result: z.object({ result: resultSchema }).strict(), + ...(spike + ? { + upload_investigation_image: z.object({ + data: z.string().min(1), + mimeType: z.string().max(40), + }).strict(), + submit_spike_result: spikeSubmissionSchema, + } + : {}), }; } return { diff --git a/apps/server/src/experiments/routes.test.ts b/apps/server/src/experiments/routes.test.ts index 258929381..56ad58a1b 100644 --- a/apps/server/src/experiments/routes.test.ts +++ b/apps/server/src/experiments/routes.test.ts @@ -369,3 +369,167 @@ test("builds queued behind a connection's build take its place in the order they expect(connections.assigned(id)).toBeUndefined(); expect(connections.queued(id)).toEqual([]); }); + +test("a spike runs with an image upload and a structured report, and stays out of the list", async () => { + let { alice, documentId, request, tool, pair, runtime } = await setup(); + let paired = await pair(alice.cookie); + let id = crypto.randomUUID(); + await runtime.service.create(documentId, "alice", "Spike brief", id, undefined, { + digest: "sha256:x", + passage: "Unsure whether this works.", + callout: "01JAAAAAAAAAAAAAAAAAAAAAAA", + login: "alice", + placed: true, + }); + await runtime.service.authorize(id, paired.connection.id, { + id, + documentId, + requester: "alice", + authorizer: "alice", + brief: "Spike brief", + source: paired.connection.source, + context: "# Context", + }); + let listed = + await (await request(`/api/documents/${documentId}/experiments`, undefined, alice.cookie)) + .json(); + expect(listed.experiments).toEqual([]); + let claim = text(await tool(paired.token, "claim_experiment", { id })); + expect(claim.spike).toBe(true); + let tools = await (await request( + "/connector/mcp", + { jsonrpc: "2.0", id: 1, method: "tools/list" }, + undefined, + claim.runToken, + )).json(); + expect(tools.result.tools.map((item: { name: string }) => item.name)).toContain( + "submit_spike_result", + ); + let png = Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 0, 0, 0, 0]); + let uploaded = text( + await tool(claim.runToken, "upload_investigation_image", { + data: png.toString("base64"), + mimeType: "image/png", + }), + ); + expect(uploaded.path).toMatch(/^\/images\/[0-9a-f]{64}\.png$/); + let missing = `/images/${"b".repeat(64)}.png`; + expect( + error( + await tool(claim.runToken, "submit_spike_result", { + headline: "It works", + findings: ["Yes."], + recommendation: "Ship.", + images: [missing], + }), + ), + ).toContain("missing-image"); + expect( + error( + await tool(claim.runToken, "submit_spike_result", { + headline: "It works", + findings: ["Pointer events fire.", "Latency is fine."], + recommendation: "Ship it.", + images: [uploaded.path], + }), + ), + ).toBe(""); + await tool(paired.token, "complete_experiment", { id, generation: claim.generation }); + let value = await runtime.service.store.get(id); + expect(value?.state).toBe("completed"); + expect(value?.result?.report).toContain("**It works**"); + expect(value?.result?.report).toContain(uploaded.path); +}); + +test("an ordinary run cannot upload images or send an image-sized body", async () => { + let { alice, documentId, request, tool, pair } = await setup(); + let paired = await pair(alice.cookie); + let id = crypto.randomUUID(); + await request(`/api/documents/${documentId}/experiments`, { id, brief: "Measure" }, alice.cookie); + await request(`/api/documents/${documentId}/experiments/${id}/run`, {}, alice.cookie); + let claim = text(await tool(paired.token, "claim_experiment", { id })); + let tools = await (await request( + "/connector/mcp", + { jsonrpc: "2.0", id: 1, method: "tools/list" }, + undefined, + claim.runToken, + )).json(); + let names = tools.result.tools.map((item: { name: string }) => item.name); + expect(names).not.toContain("upload_investigation_image"); + expect(names).not.toContain("submit_spike_result"); + let png = Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 0, 0, 0, 0]); + expect( + error( + await tool(claim.runToken, "upload_investigation_image", { + data: png.toString("base64"), + mimeType: "image/png", + }), + ), + ).toContain("tool-forbidden"); + let large = await request( + "/connector/mcp", + { + jsonrpc: "2.0", + id: 1, + method: "tools/call", + params: { name: "read_experiment", arguments: { padding: "x".repeat(1_200_000) } }, + }, + undefined, + claim.runToken, + ); + expect((await large.json()).error).toBe("request-too-large"); +}); + +test("a spike uploads at most three images and cites only its own", async () => { + let { alice, documentId, tool, pair, runtime, storage } = await setup(); + let paired = await pair(alice.cookie); + let id = crypto.randomUUID(); + await runtime.service.create(documentId, "alice", "Spike brief", id, undefined, { + digest: "sha256:x", + passage: "Unsure whether this works.", + callout: "01JAAAAAAAAAAAAAAAAAAAAAAA", + login: "alice", + placed: true, + }); + await runtime.service.authorize(id, paired.connection.id, { + id, + documentId, + requester: "alice", + authorizer: "alice", + brief: "Spike brief", + source: paired.connection.source, + context: "# Context", + }); + let claim = text(await tool(paired.token, "claim_experiment", { id })); + let upload = (seed: number) => + tool(claim.runToken, "upload_investigation_image", { + data: Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, seed]).toString("base64"), + mimeType: "image/png", + }); + // Parallel uploads in one run must all count, and the cap must hold across them. + let results = await Promise.all([1, 2, 3, 4].map(upload)); + expect(results.filter(result => error(result).includes("image-limit"))).toHaveLength(1); + let seeds = [1, 2, 3, 4].filter((_seed, index) => !error(results[index])); + let paths: string[] = seeds.map(seed => text(results[seed - 1]).path); + expect(paths).toHaveLength(3); + expect(text(await upload(seeds[0])).path).toBe(paths[0]); + expect(error(await upload(5))).toContain("image-limit"); + let foreign = "c".repeat(64); + await storage.images.put({ + channelId: documentId, + sha256: foreign, + mimeType: "image/png", + bytes: new Uint8Array([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]), + uploadedBy: "bob", + now: new Date(), + }); + let submit = (images: string[]) => + tool(claim.runToken, "submit_spike_result", { + headline: "It works", + findings: ["Yes."], + recommendation: "Ship.", + images, + }); + expect(error(await submit([`/images/${foreign}.png`]))).toContain("missing-image"); + expect(error(await submit(paths))).toBe(""); +}); diff --git a/apps/server/src/experiments/routes.ts b/apps/server/src/experiments/routes.ts index f0cb207ad..73962eb85 100644 --- a/apps/server/src/experiments/routes.ts +++ b/apps/server/src/experiments/routes.ts @@ -15,6 +15,10 @@ import { implementationSchemas } from "../tasks/connector"; import type { ImplementationConnector } from "../tasks/connector"; import { connectorSchemas } from "./mcp-schema"; import type { Lease } from "../storage/model"; +import { createHash } from "node:crypto"; +import { imagePath } from "../images/format"; +import { MAX_IMAGE_REQUEST_BYTES, prepareImage } from "../mcp/image"; +import { spikeReport, spikeSubmissionSchema } from "./spikes"; export type ExperimentRuntime = ReturnType; type Options = { @@ -32,21 +36,30 @@ type Options = { decision: string, remove: boolean, ) => Promise; + /** Land a document's finished spike results before a first build claims it. */ + settleSpikes?: (documentId: string) => Promise; }; const pairingSchema = sourceSchema.omit({ repositoryId: true }).extend({ label: z.string().trim().min(1).max(100), }).strict(); -async function body(request: Request) { +const MAX_BODY_BYTES = limits.resultBytes + 64 * 1024; +/** Screenshots a spike run may upload; distinct images, so a retried upload is free. */ +export const MAX_SPIKE_IMAGES = 3; + +async function raw(request: Request, max: number) { let size = 0; let chunks: Uint8Array[] = []; if (!request.body) fail("invalid-request"); for await (let chunk of request.body) { size += chunk.length; - if (size > limits.resultBytes + 64 * 1024) fail("request-too-large"); + if (size > max) fail("request-too-large"); chunks.push(chunk); } - return JSON.parse(Buffer.concat(chunks).toString("utf8")); + return Buffer.concat(chunks); +} +async function body(request: Request, max = MAX_BODY_BYTES) { + return JSON.parse((await raw(request, max)).toString("utf8")); } function json(value: unknown, status = 200) { return Response.json(value, { @@ -65,6 +78,8 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio options.changed(id); }; let service = new Experiments(auth.storage.experiments, options.lease, changed); + /** Image hashes each spike run uploaded; a run may only cite its own screenshots. */ + let uploads = new Map>(); async function access(session: AuthenticatedSession | undefined, id: string, write: boolean) { if (!session) fail("authentication-required"); let channel = await auth.storage.channels.get(id); @@ -182,16 +197,19 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio route("GET", "/api/documents/:id/experiments", async (request, params) => { await access(await auth.sessions.authenticate(request), params.id, false); return json({ - experiments: (await service.store.list(params.id)).map(value => ({ - id: value.id, - brief: value.brief, - state: value.state, - revision: value.revision, - requester: value.requester, - decisionCount: value.decisions.length, - progress: value.progress, - createdAt: value.createdAt, - })), + // A spike's callout under its passage is its record in the document. + experiments: (await service.store.list(params.id)).filter(value => !value.spike).map( + value => ({ + id: value.id, + brief: value.brief, + state: value.state, + revision: value.revision, + requester: value.requester, + decisionCount: value.decisions.length, + progress: value.progress, + createdAt: value.createdAt, + }), + ), }); }); route("POST", "/api/documents/:id/experiments", async (request, params) => { @@ -273,10 +291,11 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio if (channel.repositoryId !== connection.source.repositoryId) fail("repository-forbidden"); } /** The connection's queued, running, or publishing investigations across its repository. */ + /** A connection's work in creation order, so queued runs are dispatched first come first served. */ async function work(connection: Connection, states: string[]) { return (await service.store.active()).filter(item => item.connectionId === connection.id && states.includes(item.state) - ); + ).sort((a, b) => a.createdAt - b.createdAt); } route( "GET", @@ -359,16 +378,27 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio let { connection, grant, session } = await connector(token); let implementation = options.implementations?.(); let implementationRun = grant.run?.kind === "implementation" || grant.run?.kind === "rebuild"; + let spikeRun = !!grant.run && !implementationRun + && !!(await service.store.get(grant.run.id))?.spike; let schemas = implementationRun ? implementationSchemas(true, grant.run?.kind === "rebuild") : { - ...connectorSchemas(!!grant.run), + ...connectorSchemas(!!grant.run, spikeRun), ...(!grant.run && implementation ? implementationSchemas(false) : {}), }; + // Only a spike run's image upload may exceed the ordinary connector body limit. + let bytes = await raw( + request, + spikeRun ? Math.max(MAX_BODY_BYTES, MAX_IMAGE_REQUEST_BYTES) : MAX_BODY_BYTES, + ); let call = z.object({ jsonrpc: z.literal("2.0"), id: z.union([z.string(), z.number()]).optional(), method: z.string(), params: z.record(z.string(), z.unknown()).optional(), - }).parse(await body(request)); + }).parse(JSON.parse(bytes.toString("utf8"))); + if ( + bytes.length > MAX_BODY_BYTES + && (call.method !== "tools/call" || call.params?.name !== "upload_investigation_image") + ) fail("request-too-large"); let respond = (result: unknown) => json({ jsonrpc: "2.0", id: call.id, result }); if (call.method.startsWith("notifications/")) return new Response(null, { status: 202 }); if (call.method === "initialize") { @@ -438,6 +468,8 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio let documentId = grant.run?.documentId ?? connections.assigned(connection.id); if (documentId) await target(session, connection, documentId); if (name === "claim_implementation_build") { + // A build queued behind a spike on this connection must start with its findings. + if (documentId) await options.settleSpikes?.(documentId); value = await connections.locked(connection.id, async () => { if ((await work(connection, ["running", "publishing"])).length) { fail("workspace-busy"); @@ -483,6 +515,7 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio return { input: claimed.input, generation: claimed.generation, + ...(claimed.spike ? { spike: true } : {}), runToken: connections.runToken( connection.id, claimed.documentId, @@ -515,6 +548,62 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio ); } else if (name === "fail_experiment") { value = await service.stop(id, "failed", z.string().max(2000).parse(args.error)); + } else if (name === "upload_investigation_image") { + if (!current.spike) fail("tool-forbidden"); + let prepared = prepareImage({ id: current.documentId, ...args }); + if (!prepared) fail("invalid-image"); + if ("refusal" in prepared) fail(prepared.refusal); + let { bytes, mimeType } = prepared.input; + let sha256 = createHash("sha256").update(bytes).digest("hex"); + // Parallel uploads share one set and reserve their slot before awaiting storage, + // so neither overwrites the other's record nor slips past the cap. + let uploaded = uploads.get(id); + if (!uploaded) uploads.set(id, uploaded = new Set()); + let reserved = !uploaded.has(sha256); + if (reserved && uploaded.size >= MAX_SPIKE_IMAGES) { + fail("image-limit", `A spike may upload at most ${MAX_SPIKE_IMAGES} images.`); + } + uploaded.add(sha256); + try { + await auth.storage.images.put({ + channelId: current.documentId, + sha256, + mimeType, + bytes, + uploadedBy: session.user.id, + now: new Date(), + }); + } catch (err) { + if (reserved) uploaded.delete(sha256); + throw err; + } + value = { path: imagePath(sha256, mimeType) }; + } else if (name === "submit_spike_result") { + if (!current.spike) fail("tool-forbidden"); + let input = spikeSubmissionSchema.parse(args); + for (let path of input.images) { + let sha256 = path.slice("/images/".length).split(".")[0]; + if ( + !uploads.get(id)?.has(sha256) + || !await auth.storage.images.get(current.documentId, sha256) + ) fail("missing-image", `Upload ${path} with upload_investigation_image first.`); + } + let report = canonicalReport(spikeReport(input)); + if ("issues" in report) { + fail("invalid-report", report.issues.map(issue => issue.message).join("; ")); + } + value = await service.candidate(id, connection.id, generation, { + schemaVersion: 1, + report: report.source, + datasets: [], + views: [], + evidence: [], + provenance: { + environment: "Throwaway prototype by the editor's local coding agent.", + checks: [], + limitations: [], + }, + }); } else if (name === "submit_experiment_result") { let result = args.result as { report?: unknown }; if (!result || typeof result.report !== "string") fail("invalid-result"); @@ -539,7 +628,9 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio value = { accepted: true, state: (value as { state?: string }).state, - ...(name === "read_experiment" ? value as object : {}), + ...(name === "read_experiment" || name === "upload_investigation_image" + ? value as object + : {}), }; } return respond({ content: [{ type: "text", text: JSON.stringify(value) }] }); @@ -561,7 +652,11 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio connections.sweep(); await service.recover(); await options.implementations?.()?.sweep(); - for (let value of await service.store.active()) { + let active = await service.store.active(); + for (let id of uploads.keys()) { + if (!active.some(value => value.id === id)) uploads.delete(id); + } + for (let value of active) { if (value.connectionId && !connections.get(value.connectionId)) { await service.stop( value.id, diff --git a/apps/server/src/experiments/service.ts b/apps/server/src/experiments/service.ts index 44c067814..cd6d18782 100644 --- a/apps/server/src/experiments/service.ts +++ b/apps/server/src/experiments/service.ts @@ -6,7 +6,7 @@ import { parseResult, requestSchema, } from "@chopin/experiment"; -import type { Investigation, InvestigationState } from "@chopin/experiment/records"; +import type { Investigation, InvestigationState, Spike } from "@chopin/experiment/records"; import type { RunInput } from "@chopin/experiment"; import { applySelection, selectionPatchSchema } from "@chopin/experiment/selection"; import type { ExperimentStore } from "../storage/experiments"; @@ -34,6 +34,7 @@ export class Experiments { brief: string, id: string = crypto.randomUUID(), parentId?: string, + spike?: Spike, ) { let existing = await this.store.get(id); if (existing) { @@ -52,6 +53,7 @@ export class Experiments { requester, brief, parentId, + ...(spike ? { spike } : {}), revision: 0, state: "requested", generation: 0, diff --git a/apps/server/src/experiments/spike-host.test.ts b/apps/server/src/experiments/spike-host.test.ts new file mode 100644 index 000000000..139cf1665 --- /dev/null +++ b/apps/server/src/experiments/spike-host.test.ts @@ -0,0 +1,292 @@ +import { expect, spyOn, test } from "bun:test"; +import { ulid } from "@chopin/dialect/ulid"; + +import * as Comments from "../comments/service"; +import * as room from "../plan/room"; +import * as Service from "../plan/service"; +import { cardWithProse } from "../questions/prose.test-fixtures"; +import * as Questions from "../questions/service"; +import { pickBuild, queueBuild } from "../tasks/builds"; +import { announceImplementation } from "../tasks/notifications"; +import { claimImplementation, implementationGraphs } from "../tasks/plan-graphs"; +import { openPlan } from "../testing/plan"; +import { Connections } from "./connections"; +import { Experiments } from "./service"; +import { spikeHost } from "./spike-host"; +import { calloutDigest, callouts, placeSpikeCallout } from "./spike-placement"; +import { SpikeScout } from "./spike-scout"; +import { spikeCallout } from "./spikes"; + +const PASSAGE = "We are unsure whether drag handles work on touch screens at all."; + +async function hosted(source = `# Plan\n\n${PASSAGE}\n\nAnother paragraph.\n`) { + let context = await openPlan(source); + let service = new Experiments(context.storage.experiments, () => context.lease); + let host = spikeHost({ + service, + connections: new Connections(), + withPlan: (_id, action) => action(context.plan), + }); + return { ...context, service, host }; +} + +test("a spike is durably placing before its callout is published", async () => { + let errors = spyOn(console, "error").mockImplementation(() => {}); + let context = await hosted(); + let states: Array<{ placing?: boolean; placed: boolean } | undefined> = []; + let commit = context.storage.collaboration.commit.bind(context.storage.collaboration); + let spy = spyOn(context.storage.collaboration, "commit").mockImplementation(async input => { + let [value] = await context.service.store.list(context.plan.id); + states.push(value?.spike && { placing: value.spike.placing, placed: value.spike.placed }); + return commit(input); + }); + try { + let digest = room.digests(context.plan.document)[1]; + await context.host.start(context.plan.id, { + owner: "U_test", + connection: { id: "gone", login: "maggie" }, + block: { digest, text: PASSAGE }, + }); + expect(states[0]).toEqual({ placing: true, placed: false }); + let [value] = await context.service.store.list(context.plan.id); + expect(value.spike).toMatchObject({ placed: true, rendered: "queued" }); + expect(value.spike?.placing).toBeUndefined(); + } finally { + spy.mockRestore(); + errors.mockRestore(); + await Service.close(context.plan); + } +}); + +test("recovery finds a published callout in place and never inserts another", async () => { + let errors = spyOn(console, "error").mockImplementation(() => {}); + let context = await hosted(); + try { + let id = crypto.randomUUID(); + let callout = ulid(); + let digest = room.digests(context.plan.document)[1]; + let value = await context.service.create(context.plan.id, "U_test", "brief", id, undefined, { + digest, + passage: PASSAGE, + callout, + login: "maggie", + placed: false, + placing: true, + }); + let node = spikeCallout(value); + await context.service.mutate(id, item => { + item.spike!.calloutDigest = calloutDigest(node); + }); + // The callout was published, then the process died before `placed` persisted. + await placeSpikeCallout(context.plan, { callout, node, after: digest }); + await context.host.refresh(context.plan.id); + let recovered = await context.service.store.get(id); + expect(recovered?.spike).toMatchObject({ placed: true, callout }); + expect(recovered?.spike?.placing).toBeUndefined(); + expect(recovered?.spike?.dismissed).toBeUndefined(); + let source = room.project(context.plan.document); + expect(source.split(" { + let errors = spyOn(console, "error").mockImplementation(() => {}); + let context = await cardWithProse(`# Title\n\n${PASSAGE}\n\nDecision prose.\n`, 2); + let seen: Array<{ records: boolean; threads: boolean; placed: boolean }> = []; + let callout = ulid(); + let records = context.plan.records; + let threads = context.plan.threads; + let questions = spyOn(Questions, "rebase"); + let comments = spyOn(Comments, "rebase").mockImplementation(plan => { + seen.push({ + records: plan.records !== records, + threads: plan.threads !== threads, + placed: callouts(room.project(plan.document)).has(callout), + }); + }); + try { + let before = Questions.prose(context.plan)[0].anchors[0]; + let digest = room.digests(context.plan.document)[1]; + let placed = await placeSpikeCallout(context.plan, { + callout, + node: spikeCallout( + { + state: "running", + progress: "", + spike: { callout, login: "maggie" }, + } as Parameters[0], + ), + after: digest, + }); + expect(placed.status).toBe("placed"); + expect(questions).toHaveBeenCalledTimes(2); + expect(seen).toEqual([ + { records: true, threads: true, placed: false }, + { records: true, threads: true, placed: true }, + ]); + expect(context.plan.records).not.toBe(records); + let after = Questions.prose(context.plan)[0]; + expect(after.orphaned).toBe(false); + expect(after.anchors[0].digest).toBe(before.digest); + } finally { + questions.mockRestore(); + comments.mockRestore(); + errors.mockRestore(); + await Service.close(context.plan); + } +}); + +test("a spike result deferred by a first build's lock lands once the lock releases", async () => { + let errors = spyOn(console, "error").mockImplementation(() => {}); + let context = await hosted(); + let scout = new SpikeScout({ host: context.host }); + let { plan } = context; + // The wiring main.ts gives a released lock. + plan.persistence.liveBuild = true; + let released: Promise | undefined; + plan.persistence.onEditingUnlocked = id => { + released = scout.refresh(id); + }; + try { + await context.host.start(plan.id, { + owner: "U_test", + connection: { id: "gone", login: "maggie" }, + block: { digest: room.digests(plan.document)[1], text: PASSAGE }, + }); + let [value] = await context.service.store.list(plan.id); + expect(value.state).toBe("failed"); + expect(value.spike?.rendered).toBe("queued"); + + plan.builds = [{ id: crypto.randomUUID(), state: "running" } as never]; + announceImplementation(plan); + await scout.refresh(plan.id); + // Queued renders as a note and stopped as a warning; titles are copy and may change. + expect(room.project(plan.document)).toContain('type="note"'); + expect((await context.service.store.get(value.id))?.spike?.rendered).toBe("queued"); + + plan.builds = [{ ...plan.builds[0], state: "stopped" } as never]; + announceImplementation(plan); + expect(released).toBeDefined(); + await released; + expect(room.project(plan.document)).toContain('type="warning"'); + expect(room.project(plan.document)).not.toContain('type="note"'); + expect((await context.service.store.get(value.id))?.spike?.rendered).toBe("stopped"); + } finally { + scout.close(); + errors.mockRestore(); + await Service.close(plan); + } +}); + +test("a spike that finishes while a first build is queued lands before the build claims", async () => { + let errors = spyOn(console, "error").mockImplementation(() => {}); + let context = await hosted(); + let scout = new SpikeScout({ host: context.host }); + let { plan } = context; + try { + await context.host.start(plan.id, { + owner: "U_test", + connection: { id: "gone", login: "maggie" }, + block: { digest: room.digests(plan.document)[1], text: PASSAGE }, + }); + let [value] = await context.service.store.list(plan.id); + await implementationGraphs().revise(plan, { + planRevision: plan.revision, + graphRevision: 0, + operations: [{ + op: "add", + task: { + id: "first", + title: "First task", + context: "Card grid", + goal: "Build the grid", + acceptance: ["Renders", "Matches the spike"], + dependsOn: [], + }, + }], + }); + let checkout = { repository: "owner/repository", branch: "main", commit: "a".repeat(40) }; + let queued = await queueBuild(plan, { + planRevision: plan.revision, + graphVersion: 1, + graphRevision: 1, + user: "U_test", + connectionId: "laptop", + repositoryId: "R_test", + checkout, + }); + expect(Service.implementationActive(plan)).toBe(true); + + // The connection finishes the spike that was ahead of the build. + await context.service.mutate(value.id, item => { + item.state = "completed"; + item.result = { + schemaVersion: 1, + report: "Two columns read better at 1280px.\n", + datasets: [], + views: [], + evidence: [], + provenance: { environment: "test", checks: [], limitations: [] }, + }; + }); + // What the build claim waits for. + await scout.settle(plan.id); + expect(room.project(plan.document)).toContain('type="tip"'); + expect(room.project(plan.document)).toContain("Two columns read better"); + expect((await context.service.store.get(value.id))?.spike?.rendered).toBe("completed"); + expect(plan.builds.at(-1)?.planRevision).toBe(plan.revision); + expect(plan.graph?.versions.at(-1)?.planRevision).toBe(plan.revision); + + let picked = await pickBuild(plan, "U_test", "laptop"); + expect(picked?.planRevision).toBe(plan.revision); + let claim = await claimImplementation(plan, { + planRevision: picked!.planRevision, + graphRevision: picked!.graphRevision, + run: { + id: queued.id, + user: "U_test", + client: { name: "chopin-acp", version: "0.1.0" }, + session: "session", + graphVersion: picked!.graphVersion, + graphRevision: picked!.graphRevision, + planRevision: picked!.planRevision, + repository: checkout.repository, + branch: `chopin/implement-${queued.id.slice(0, 8)}`, + commit: checkout.commit, + startedAt: new Date().toISOString(), + }, + }); + expect(claim.kind).toBe("started"); + + // Once the build runs, results wait for the lock to release again. + await context.service.mutate(value.id, item => { + item.result!.report = "Changed after the build started.\n"; + }); + await context.service.mutate(value.id, item => { + item.state = "failed"; + }); + await scout.settle(plan.id); + expect(room.project(plan.document)).not.toContain('type="warning"'); + } finally { + scout.close(); + errors.mockRestore(); + await Service.close(plan); + } +}); diff --git a/apps/server/src/experiments/spike-host.ts b/apps/server/src/experiments/spike-host.ts new file mode 100644 index 000000000..0d73ff69e --- /dev/null +++ b/apps/server/src/experiments/spike-host.ts @@ -0,0 +1,142 @@ +import { limits } from "@chopin/experiment"; +import { parse } from "@chopin/dialect/parse"; +import { serialize } from "@chopin/dialect/serialize"; +import { ulid } from "@chopin/dialect/ulid"; + +import * as room from "../plan/room"; +import * as Service from "../plan/service"; +import { calloutDigest, calloutId, placeSpikeCallout } from "./spike-placement"; +import { eligible, renderKey, spikeBrief, spikeCallout, text } from "./spikes"; + +import type { Connections } from "./connections"; +import type { Experiments } from "./service"; +import type { SpikeHost } from "./spike-scout"; + +type Options = { + service: Experiments; + connections: Connections; + withPlan: (channelId: string, action: (plan: Service.Plan) => Promise) => Promise; +}; + +/** Spike records live with investigations; their callouts are projections placed after commit. */ +export function spikeHost({ service, connections, withPlan }: Options): SpikeHost { + let spikes = async (channelId: string) => + (await service.store.list(channelId)).filter(value => value.spike); + let dismiss = async (_channelId: string, id: string) => { + await service.mutate(id, value => { + value.spike!.dismissed = true; + }); + await service.stop(id, "cancelled", "Dismissed."); + }; + return { + snapshot: channelId => + withPlan(channelId, async plan => { + let current = await Service.readCurrentDocument(plan); + let children = parse(current.source).children; + return { + repositoryId: plan.persistence.repositoryId, + live: !!plan.live, + blocks: children.flatMap((node, index) => { + if (!eligible(node)) return []; + let next = children[index + 1]; + let after = next && calloutId(next); + return [{ + digest: room.digest(serialize({ type: "root", children: [node] })), + text: text(node).replace(/\s+/g, " ").trim().slice(0, 4000), + ...(after ? { calloutAfter: after } : {}), + }]; + }), + callouts: new Set(children.flatMap(node => calloutId(node) ?? [])), + }; + }), + spikes, + async connection(repositoryId, owner, channelId) { + let found = connections.candidates(repositoryId, owner, channelId)[0]; + return found && { id: found.id, login: found.login }; + }, + async start(channelId, { owner, connection, block }) { + let id = crypto.randomUUID(); + let callout = ulid(); + let created = await service.create(channelId, owner, spikeBrief(block.text), id, undefined, { + digest: block.digest, + passage: block.text, + callout, + login: connection.login, + placed: false, + placing: true, + }); + let node = spikeCallout(created); + // Persist what is about to be published, so recovery recognises it as untouched. + let value = await service.mutate(id, item => { + item.spike!.calloutDigest = calloutDigest(node); + }); + let placed = await withPlan( + channelId, + plan => placeSpikeCallout(plan, { callout, node, after: block.digest }), + ); + if (placed.status !== "placed") { + await dismiss(channelId, id); + return; + } + await service.mutate(id, item => { + item.spike!.placed = true; + delete item.spike!.placing; + item.spike!.rendered = renderKey(created); + item.spike!.callout = placed.callout; + item.spike!.calloutDigest = placed.digest; + }); + let live = connections.get(connection.id); + try { + if (!live) throw new Error("connection gone"); + let context = await withPlan(channelId, plan => Service.readCurrentDocument(plan)); + await service.authorize(id, live.id, { + id, + documentId: channelId, + requester: owner, + authorizer: owner, + brief: value.brief, + source: live.source, + context: `Document revision ${context.revision}\n${context.source}`.slice( + 0, + limits.context, + ), + }); + connections.use(channelId, live.id); + } catch { + await service.stop(id, "failed", "The local agent disconnected before it could start."); + } + }, + dismiss, + async refresh(channelId) { + for (let value of await spikes(channelId)) { + let spike = value.spike!; + if (spike.dismissed) continue; + // A crash after publishing but before `placed` persisted: only ever look in place. + let recovering = !spike.placed && !!spike.placing; + if (!spike.placed && !recovering) continue; + let key = renderKey(value); + if (!recovering && key === spike.rendered) continue; + let placed = await withPlan( + channelId, + plan => + placeSpikeCallout(plan, { + callout: spike.callout, + node: spikeCallout(value), + rendered: spike.calloutDigest, + }, { queued: true }), + ); + if (placed.status === "missing") await dismiss(channelId, value.id); + else if (placed.status !== "deferred") { + let { callout, digest } = placed; + await service.mutate(value.id, item => { + item.spike!.placed = true; + delete item.spike!.placing; + item.spike!.rendered = key; + item.spike!.callout = callout; + item.spike!.calloutDigest = digest; + }); + } + } + }, + }; +} diff --git a/apps/server/src/experiments/spike-placement.ts b/apps/server/src/experiments/spike-placement.ts new file mode 100644 index 000000000..408e587c7 --- /dev/null +++ b/apps/server/src/experiments/spike-placement.ts @@ -0,0 +1,180 @@ +import * as Y from "yjs"; +import { parse } from "@chopin/dialect/parse"; +import { serialize } from "@chopin/dialect/serialize"; +import { ulid } from "@chopin/dialect/ulid"; +import { assert } from "@chopin/dialect/validate"; + +import * as Comments from "../comments/service"; +import * as room from "../plan/room"; +import * as Service from "../plan/service"; +import * as Questions from "../questions/service"; +import { carryQueued, queuedOnly } from "../tasks/builds"; +import { announceImplementation } from "../tasks/notifications"; + +import type { RootContent } from "mdast"; + +export type CalloutInput = { + callout: string; + node: RootContent; + /** Digest of the passage to insert under when the callout is not in the document. */ + after?: string; + /** Digest of the callout as last rendered; a different block holds human edits. */ + rendered?: string; +}; + +/** Where the spike's callout now lives, and the digest of its block. */ +export type CalloutPlacement = + | { status: "placed"; callout: string; digest: string } + | { status: "unchanged"; callout: string; digest: string } + | { status: "missing" } + | { status: "deferred" }; + +export function calloutId(node: RootContent): string | undefined { + if (node.type !== "mdxJsxFlowElement" || node.name !== "Callout") return undefined; + let id = node.attributes.find(item => item.type === "mdxJsxAttribute" && item.name === "id"); + return typeof id?.value === "string" ? id.value : undefined; +} + +/** Callout ids present at the top level of a source. */ +export function callouts(source: string): Set { + return new Set(parse(source).children.flatMap(node => calloutId(node) ?? [])); +} + +function blockSource(node: RootContent): string { + return serialize({ type: "root", children: [node] }); +} + +export function calloutDigest(node: RootContent): string { + return room.digest(blockSource(node)); +} + +function withId(node: RootContent, id: string): RootContent { + if (node.type !== "mdxJsxFlowElement") return node; + return { + ...node, + attributes: node.attributes.map(item => + item.type === "mdxJsxAttribute" && item.name === "id" ? { ...item, value: id } : item + ), + }; +} + +/** + * Replace a spike's callout in place while it is still exactly as last rendered, or insert it + * under the one block whose digest matches `after`. A callout someone edited keeps their edits; + * the new state goes into a fresh sibling callout directly after it. An ambiguous or vanished + * passage is `missing` rather than a guess. + */ +export function withCallout( + children: RootContent[], + input: CalloutInput, +): { children: RootContent[]; callout: string } | "missing" | "unchanged" { + let next = [...children]; + let existing = next.findIndex(node => calloutId(node) === input.callout); + if (existing >= 0) { + if (blockSource(next[existing]) === blockSource(input.node)) return "unchanged"; + if (input.rendered === undefined || calloutDigest(next[existing]) === input.rendered) { + next[existing] = input.node; + return { children: next, callout: input.callout }; + } + let callout = ulid(); + next.splice(existing + 1, 0, withId(input.node, callout)); + return { children: next, callout }; + } + if (!input.after) return "missing"; + let matches = next.flatMap((node, index) => + room.digest(blockSource(node)) === input.after ? [index] : [] + ); + if (matches.length !== 1) return "missing"; + next.splice(matches[0] + 1, 0, input.node); + return { children: next, callout: input.callout }; +} + +/** Validate a callout change and reconcile it into a staged document. */ +export function reconcileCallout( + document: room.Document, + input: CalloutInput, +): { mutation: room.Mutation; callout: string } | "missing" | "unchanged" { + let root = parse(room.project(document)); + let placed = withCallout(root.children, input); + if (typeof placed === "string") return placed; + let next = serialize({ ...root, children: placed.children }); + let parsed = parse(next); + assert(parsed, { bytes: new TextEncoder().encode(next).byteLength }); + if (parsed.children.length !== placed.children.length) { + throw new Error("callout did not round-trip"); + } + room.validate(next); + let mutation = room.reconcile(document, root.children, placed.children); + return mutation ? { mutation, callout: placed.callout } : "unchanged"; +} + +function digestOf(source: string, callout: string): string | undefined { + let found = parse(source).children.find(node => calloutId(node) === callout); + return found && calloutDigest(found); +} + +/** + * Place or update a spike's callout. With `queued`, an update may also land under a first build + * that is only queued, so the build starts with the spike's findings: the callout-only revision + * carries the build and its approved graph in the same commit. Any other lock defers it. + */ +export function placeSpikeCallout( + plan: Service.Plan, + input: CalloutInput, + options: { queued?: boolean } = {}, +): Promise { + return Service.exclusive(plan, async (): Promise => { + let queued = Service.implementationActive(plan) && !!options.queued && !!queuedOnly(plan); + if (Service.implementationActive(plan) && !queued) return { status: "deferred" }; + let document = await room.restore( + plan.document.epoch, + Y.encodeStateAsUpdate(plan.document.doc), + room.project(plan.document), + [], + ); + document.seq = plan.document.seq; + let candidate: Service.Plan = { + ...plan, + document, + records: new Map(plan.records), + threads: new Map(plan.threads), + }; + try { + // Carry anchors onto the old document before the edit, as Service.rewrite does. + Questions.rebase(candidate); + Comments.rebase(candidate); + let placed = reconcileCallout(document, input); + if (placed === "missing") return { status: "missing" }; + if (placed === "unchanged") { + let digest = digestOf(room.project(document), input.callout); + return digest + ? { status: "unchanged", callout: input.callout, digest } + : { status: "missing" }; + } + Questions.rebase(candidate); + Comments.rebase(candidate); + let digest = digestOf(room.project(document), placed.callout); + if (!digest) throw new Error("callout did not land"); + let carried = queued + ? carryQueued( + plan, + room.project(document) === plan.persistence.committedSource + ? plan.revision + : plan.revision + 1, + ) + : undefined; + if (carried) Object.assign(candidate, carried); + await Service.publishStaged(plan, plan.server, plan.id, candidate, placed.mutation, { + agent: true, + queuedBuild: queued, + }); + if (carried) { + Object.assign(plan, carried); + announceImplementation(plan); + } + return { status: "placed", callout: placed.callout, digest }; + } finally { + document.doc.destroy(); + } + }); +} diff --git a/apps/server/src/experiments/spike-scout.test.ts b/apps/server/src/experiments/spike-scout.test.ts new file mode 100644 index 000000000..b6e7bf93f --- /dev/null +++ b/apps/server/src/experiments/spike-scout.test.ts @@ -0,0 +1,283 @@ +import { expect, test } from "bun:test"; + +import { FOLLOW_UP_MS, jevJudge, SpikeScout } from "./spike-scout"; + +import type { Investigation } from "@chopin/experiment/records"; +import type { SpikeBlock, SpikeHost, SpikeJudge, SpikeSnapshot } from "./spike-scout"; + +function block( + digest: string, + text = `Passage ${digest} is unsure whether this works.`, +): SpikeBlock { + return { digest, text }; +} + +function harness(snapshot: Partial = {}, verdict?: SpikeJudge) { + let started: string[] = []; + let dismissed: string[] = []; + let judged: string[][] = []; + let records: Investigation[] = []; + let connected = true; + let timers: Array<() => void> = []; + let delays: number[] = []; + let state: SpikeSnapshot = { + repositoryId: "R_1", + live: false, + blocks: [], + callouts: new Set(), + ...snapshot, + }; + let host: SpikeHost = { + snapshot: async () => state, + spikes: async () => records, + connection: async () => connected ? { id: "C_1", login: "maggie" } : undefined, + start: async (_id, input) => { + started.push(input.block.digest); + state.callouts.add(`CALLOUT${input.block.digest}`); + records.push({ + state: "running", + spike: { + digest: input.block.digest, + callout: `CALLOUT${input.block.digest}`, + placed: true, + }, + } as Investigation); + }, + dismiss: async (_id, id) => { + dismissed.push(id); + }, + refresh: async () => {}, + }; + let scout = new SpikeScout({ + host, + judge: async blocks => { + judged.push(blocks.map(item => item.digest)); + return verdict ? verdict(blocks) : blocks.map(item => item.text.includes("unsure")); + }, + after: (delay, action) => { + delays.push(delay); + timers.push(action); + return () => {}; + }, + }); + return { + scout, + state, + started, + dismissed, + judged, + records, + timers, + delays, + disconnect: () => connected = false, + }; +} + +test("a settled edit starts spikes for uncertain passages and never re-judges them", async () => { + let h = harness({ + blocks: [block("a"), block("b", "A settled descriptive paragraph about lists.")], + }); + h.scout.schedule({ channelId: "D", editor: "U_1" }); + expect(h.timers).toHaveLength(1); + h.timers[0](); + await h.scout.check("D"); + expect(h.started).toEqual(["a"]); + expect(h.judged).toEqual([["a", "b"]]); + await h.scout.check("D"); + expect(h.judged).toHaveLength(1); + expect(h.started).toEqual(["a"]); +}); + +test("edits by the server alone never schedule a scan", () => { + let h = harness({ blocks: [block("a")] }); + h.scout.schedule({ channelId: "D" }); + expect(h.timers).toHaveLength(0); +}); + +test("a scan judges at most five passages and keeps three spikes active", async () => { + let h = harness({ blocks: ["a", "b", "c", "d", "e", "f", "g"].map(digest => block(digest)) }); + h.scout.schedule({ channelId: "D", editor: "U_1" }); + await h.scout.check("D"); + expect(h.judged[0]).toEqual(["a", "b", "c", "d", "e"]); + expect(h.started).toEqual(["a", "b", "c"]); + await h.scout.check("D"); + expect(h.judged).toHaveLength(1); + expect(h.dismissed).toEqual([]); +}); + +test("a living build stops the scout", async () => { + let h = harness({ live: true, blocks: [block("a")] }); + h.scout.schedule({ channelId: "D", editor: "U_1" }); + await h.scout.check("D"); + expect(h.judged).toHaveLength(0); + expect(h.started).toHaveLength(0); +}); + +test("without the editor's local agent nothing starts and nothing is judged", async () => { + let h = harness({ blocks: [block("a")] }); + h.disconnect(); + h.scout.schedule({ channelId: "D", editor: "U_1" }); + await h.scout.check("D"); + expect(h.judged).toHaveLength(0); + expect(h.started).toHaveLength(0); +}); + +test("a removed callout dismisses its spike and the passage never re-triggers", async () => { + let h = harness({ blocks: [block("a")] }); + h.records.push({ + id: "X", + state: "running", + spike: { digest: "a", callout: "GONE", placed: true }, + } as Investigation); + h.scout.schedule({ channelId: "D", editor: "U_1" }); + await h.scout.check("D"); + expect(h.dismissed).toEqual(["X"]); + expect(h.started).toEqual([]); + expect(h.judged).toHaveLength(0); +}); + +test("Jev decides at 0.6 and a failure decides nothing", async () => { + let asked = 0; + let judge = jevJudge(async request => { + asked++; + expect(Object.keys(request.questions)).toEqual(["spike_0", "spike_1"]); + return { + model: "jev", + answers: { + spike_0: { type: "noul", noul: 0.61 }, + spike_1: { type: "noul", noul: 0.59 }, + }, + usage: { input_tokens: 1, output_tokens: 1 }, + latencyMs: 1, + }; + }); + let blocks = [block("a", "Plain text."), block("b")]; + expect(await judge(blocks)).toEqual([true, false]); + expect(asked).toBe(1); + let failing = jevJudge(async () => { + throw new Error("timeout"); + }); + expect(await failing(blocks)).toBeUndefined(); +}); + +test("a judge that cannot decide leaves every passage to be judged again", async () => { + let h = harness({ blocks: [block("a")] }, async () => undefined); + h.scout.schedule({ channelId: "D", editor: "U_1" }); + await h.scout.check("D"); + await h.scout.check("D"); + expect(h.judged).toEqual([["a"], ["a"]]); + expect(h.started).toEqual([]); +}); + +test("approved passages over capacity are started by a later scan", async () => { + let h = harness({ blocks: ["a", "b", "c", "d"].map(digest => block(digest)) }); + h.scout.schedule({ channelId: "D", editor: "U_1" }); + await h.scout.check("D"); + expect(h.started).toEqual(["a", "b", "c"]); + h.records[0].state = "completed"; + await h.scout.check("D"); + expect(h.judged[1]).toEqual(["d"]); + expect(h.started).toEqual(["a", "b", "c", "d"]); +}); + +test("a living build still dismisses a deleted callout's spike", async () => { + let h = harness({ live: true, blocks: [block("a")] }); + h.records.push({ + id: "X", + state: "running", + spike: { digest: "a", callout: "GONE", placed: true }, + } as Investigation); + h.scout.schedule({ channelId: "D", editor: "U_1" }); + await h.scout.check("D"); + expect(h.dismissed).toEqual(["X"]); + expect(h.judged).toHaveLength(0); +}); + +test("a scan that leaves passages unjudged looks at the next batch shortly", async () => { + let digests = ["a", "b", "c", "d", "e", "f", "g"]; + let h = harness( + { blocks: digests.map(digest => block(digest, `Settled passage ${digest} about lists.`)) }, + ); + h.scout.schedule({ channelId: "D", editor: "U_1" }); + h.timers[0](); + await Bun.sleep(1); + expect(h.judged).toEqual([["a", "b", "c", "d", "e"]]); + expect(h.delays.at(-1)).toBe(FOLLOW_UP_MS); + h.timers.at(-1)!(); + await Bun.sleep(1); + expect(h.judged[1]).toEqual(["f", "g"]); + // Everything has been judged, so no further follow-up is scheduled. + let scheduled = h.timers.length; + await h.scout.check("D"); + expect(h.timers).toHaveLength(scheduled); +}); + +test("a rewritten passage directly above a spike callout is not spiked again", async () => { + let h = harness({ + blocks: [ + { ...block("a2"), calloutAfter: "CALLOUTa" }, + { ...block("b"), calloutAfter: "SOMEONE_ELSES" }, + ], + callouts: new Set(["CALLOUTa", "SOMEONE_ELSES"]), + }); + h.records.push({ + id: "X", + state: "completed", + spike: { digest: "a", callout: "CALLOUTa", placed: true }, + } as Investigation); + h.scout.schedule({ channelId: "D", editor: "U_1" }); + await h.scout.check("D"); + expect(h.judged).toEqual([["b"]]); + expect(h.started).toEqual(["b"]); +}); + +test("deleting a callout stops its spike on the edit, before the scan", async () => { + let h = harness({ blocks: [block("a")] }); + h.records.push({ + id: "X", + state: "running", + spike: { digest: "a", callout: "GONE", placed: true }, + } as Investigation); + h.scout.schedule({ channelId: "D", editor: "U_1" }); + h.scout.schedule({ channelId: "D", editor: "U_1" }); + await Bun.sleep(0); + await h.scout.refresh("D"); + expect(h.dismissed).toEqual(["X"]); + expect(h.judged).toHaveLength(0); +}); + +test("passages waiting at the active-spike cap are judged when a spike frees capacity", async () => { + let asked: string[][] = []; + let judge = jevJudge(async request => { + let passages = (request.state as { passages: Array<{ key: string }> }).passages; + asked.push(passages.map(passage => passage.key)); + return { + answers: Object.fromEntries( + passages.map(passage => [passage.key, { type: "noul", noul: 0.9 }]), + ), + } as never; + }); + let h = harness( + { blocks: ["a", "b", "c", "d", "e", "f", "g"].map(digest => block(digest)) }, + judge, + ); + h.scout.schedule({ channelId: "D", editor: "U_1" }); + h.timers[0](); + await Bun.sleep(1); + expect(h.started).toEqual(["a", "b", "c"]); + // The follow-up for f and g finds no capacity and judges nothing. + h.timers.at(-1)!(); + await Bun.sleep(1); + expect(h.judged).toHaveLength(1); + // A refresh while every spike is still active changes nothing. + await h.scout.refresh("D"); + expect(h.judged).toHaveLength(1); + h.records[1].state = "failed"; + await h.scout.refresh("D"); + expect(h.judged[1]).toEqual(["d", "e", "f", "g"]); + expect(h.started).toEqual(["a", "b", "c", "d"]); + h.records[0].state = "completed"; + await h.scout.refresh("D"); + expect(h.started).toEqual(["a", "b", "c", "d", "e"]); + expect(asked).toHaveLength(3); +}); diff --git a/apps/server/src/experiments/spike-scout.ts b/apps/server/src/experiments/spike-scout.ts new file mode 100644 index 000000000..1c4592078 --- /dev/null +++ b/apps/server/src/experiments/spike-scout.ts @@ -0,0 +1,269 @@ +import { uncertain } from "./spikes"; + +import type { JevRequest, JevResult } from "../conversation-plan/jev"; +import type { Investigation } from "@chopin/experiment/records"; + +export type SpikeBlock = { + digest: string; + text: string; + /** Id of the Callout directly after this passage, if one is. */ + calloutAfter?: string; +}; +export type SpikeSnapshot = { + repositoryId: string; + /** The document already has a living build; spikes are for the plan before it. */ + live: boolean; + /** Eligible prose blocks, in document order. */ + blocks: SpikeBlock[]; + /** Callout ids currently in the document. */ + callouts: Set; +}; + +/** The durable side of spikes; the scout only decides. */ +export type SpikeHost = { + snapshot(channelId: string): Promise; + spikes(channelId: string): Promise; + /** The editor's live local-agent connection for the repository, if any. */ + connection( + repositoryId: string, + owner: string, + channelId: string, + ): Promise<{ id: string; login: string } | undefined>; + start( + channelId: string, + input: { owner: string; connection: { id: string; login: string }; block: SpikeBlock }, + ): Promise; + /** A placed callout was removed: stop its spike and never re-trigger the passage. */ + dismiss(channelId: string, id: string): Promise; + /** Bring every placed callout up to its record's state. */ + refresh(channelId: string): Promise; +}; + +/** + * Which candidate passages would materially benefit from a quick prototype. `undefined` means + * the judge could not decide, so every passage stays unjudged for the next scan. + */ +export type SpikeJudge = (blocks: SpikeBlock[]) => Promise; + +export const heuristicJudge: SpikeJudge = async blocks => + blocks.map(block => uncertain(block.text)); + +export const SPIKE_THRESHOLD = 0.6; +export const MAX_CANDIDATES = 5; +export const MAX_ACTIVE = 3; +/** How soon a scan that left passages unjudged looks at the next batch. */ +export const FOLLOW_UP_MS = 5_000; +const ACTIVE = ["requested", "queued", "running", "publishing"]; + +/** + * One Jev noul per passage. A failure decides nothing: explicit-uncertainty matching is only + * for deployments with no Jev key, never a silent substitute for a configured judge. + */ +export function jevJudge(ask: (request: JevRequest) => Promise): SpikeJudge { + return async blocks => { + if (!blocks.length) return []; + try { + let questions = Object.fromEntries(blocks.map((_block, index) => [`spike_${index}`, { + type: "noul" as const, + instructions: + `Would a small, quick prototype or proof of concept (under ~15 min of coding agent time) materially inform or de-risk passage spike_${index}?`, + criteria: { + true: + "The passage hinges on an open technical, interaction, or feasibility question a quick prototype could answer.", + false: + "The passage is settled, descriptive, or a prototype would not change what the team decides.", + }, + }])); + let result = await ask({ + state: { + passages: blocks.map((block, index) => ({ key: `spike_${index}`, text: block.text })), + }, + questions, + }); + return blocks.map((_block, index) => { + let found = result.answers[`spike_${index}`]; + if (found?.type !== "noul" || !Number.isFinite(found.noul)) throw new Error("invalid"); + return found.noul >= SPIKE_THRESHOLD; + }); + } catch { + return undefined; + } + }; +} + +export type SpikeScoutOptions = { + host: SpikeHost; + judge?: SpikeJudge; + debounceMs?: number; + after?: (delayMs: number, action: () => void) => () => void; + error?: (err: unknown) => void; +}; + +/** + * Notices passages a quick prototype would inform, while a document has no living build yet, + * and hands each to the local agent of the person who last edited the document. + */ +export class SpikeScout { + #options: SpikeScoutOptions; + #pending = new Map void }>(); + #chains = new Map>(); + /** Digests already judged per document, so unchanged passages are never asked again. */ + #seen = new Map>(); + #editors = new Map(); + #followUps = new Map void>(); + /** Documents whose last scan left passages waiting for a spike to free capacity. */ + #waiting = new Set(); + /** Documents with a dismissal check already queued, so a burst of edits shares one. */ + #dismissing = new Set(); + #closed = false; + + constructor(options: SpikeScoutOptions) { + this.#options = options; + } + + /** A person's edit persisted; scan once everyone has been quiet for the debounce window. */ + schedule(target: { channelId: string; editor?: string }): void { + if (this.#closed || !target.editor) return; + let id = target.channelId; + this.#editors.set(id, target.editor); + this.#pending.get(id)?.cancel(); + this.#followUps.get(id)?.(); + this.#followUps.delete(id); + // A deleted callout stops its spike now rather than after the scan's debounce. + if (!this.#dismissing.has(id)) { + this.#dismissing.add(id); + void this.#serial(id, async () => { + this.#dismissing.delete(id); + let snapshot = await this.#options.host.snapshot(id); + if (snapshot) await this.#dismiss(id, snapshot); + }); + } + let entry = { cancel: () => {} }; + entry.cancel = this.#after(this.#options.debounceMs ?? 20_000, () => { + if (this.#pending.get(id) !== entry) return; + this.#pending.delete(id); + void this.check(id); + }); + this.#pending.set(id, entry); + } + + #after(delay: number, action: () => void): () => void { + if (this.#options.after) return this.#options.after(delay, action); + let timer = setTimeout(action, delay); + return () => clearTimeout(timer); + } + + /** Stop the spikes whose placed callouts are gone; returns every spike of the document. */ + async #dismiss(channelId: string, snapshot: SpikeSnapshot): Promise { + let host = this.#options.host; + let spikes = await host.spikes(channelId); + for (let value of spikes) { + let spike = value.spike!; + if (spike.placed && !spike.dismissed && !snapshot.callouts.has(spike.callout)) { + await host.dismiss(channelId, value.id); + spike.dismissed = true; + } + } + return spikes; + } + + /** A spike record changed; project it into its callout, and scan again if it freed capacity. */ + refresh(channelId: string): Promise { + return this.#serial(channelId, async () => { + await this.#options.host.refresh(channelId); + if (this.#waiting.delete(channelId)) await this.#check(channelId); + }); + } + + /** + * Land every spike result now, after any refresh already queued, without scouting further. + * A first build waits on this before it claims, so it starts with the findings in place. + */ + settle(channelId: string): Promise { + return this.#serial(channelId, () => this.#options.host.refresh(channelId)); + } + + check(channelId: string): Promise { + return this.#serial(channelId, () => this.#check(channelId)); + } + + #serial(channelId: string, action: () => Promise): Promise { + let previous = this.#chains.get(channelId) ?? Promise.resolve(); + let operation = previous.then(() => this.#closed ? undefined : action()).catch(err => { + this.#options.error?.(err); + }); + this.#chains.set(channelId, operation); + void operation.then(() => { + if (this.#chains.get(channelId) === operation) this.#chains.delete(channelId); + }); + return operation; + } + + async #check(channelId: string): Promise { + let host = this.#options.host; + let snapshot = await host.snapshot(channelId); + if (!snapshot) return; + let spikes = await this.#dismiss(channelId, snapshot); + // A living build ends new scouting, but deleting a callout must still stop its spike. + if (snapshot.live) return; + let editor = this.#editors.get(channelId); + if (!editor) return; + let capacity = MAX_ACTIVE + - spikes.filter(value => ACTIVE.includes(value.state) && !value.spike!.dismissed).length; + let seen = this.#seen.get(channelId) ?? new Set(); + this.#seen.set(channelId, seen); + let started = new Set(spikes.map(value => value.spike!.digest)); + let spikeCallouts = new Set(spikes.map(value => value.spike!.callout)); + // A rewritten passage keeps the spike callout under it, so it is not spiked again. + let unjudged = snapshot.blocks.filter(block => + !seen.has(block.digest) && !started.has(block.digest) + && !(block.calloutAfter && spikeCallouts.has(block.calloutAfter)) + ); + this.#waiting.delete(channelId); + if (!unjudged.length) return; + // A finished, failed or dismissed spike refreshes the document and resumes this scan. + if (capacity <= 0) { + this.#waiting.add(channelId); + return; + } + let candidates = unjudged.slice(0, MAX_CANDIDATES); + // Without the editor's local agent there is nowhere to run; judge these again after an edit. + let connection = await host.connection(snapshot.repositoryId, editor, channelId); + if (!connection) return; + let verdicts = await (this.#options.judge ?? heuristicJudge)(candidates); + if (!verdicts) return; + // Approved passages beyond capacity stay unseen so a later scan can start them. + let approved = candidates.filter((_block, index) => verdicts[index]); + let hits = approved.slice(0, capacity); + if (approved.length > hits.length) this.#waiting.add(channelId); + for (let [index, block] of candidates.entries()) { + if (!verdicts[index]) seen.add(block.digest); + } + for (let block of hits) { + if (this.#closed) return; + seen.add(block.digest); + await host.start(channelId, { owner: editor, connection, block }); + } + // Later passages are judged in the next batch soon, not only after another edit. + if (unjudged.length > candidates.length && !this.#closed && !this.#pending.has(channelId)) { + this.#followUps.get(channelId)?.(); + let cancel = this.#after(FOLLOW_UP_MS, () => { + if (this.#followUps.get(channelId) !== cancel) return; + this.#followUps.delete(channelId); + void this.check(channelId); + }); + this.#followUps.set(channelId, cancel); + } + } + + close(): void { + this.#closed = true; + for (let pending of this.#pending.values()) pending.cancel(); + this.#pending.clear(); + for (let cancel of this.#followUps.values()) cancel(); + this.#followUps.clear(); + this.#seen.clear(); + this.#editors.clear(); + this.#waiting.clear(); + } +} diff --git a/apps/server/src/experiments/spikes.test.ts b/apps/server/src/experiments/spikes.test.ts new file mode 100644 index 000000000..5eb18408d --- /dev/null +++ b/apps/server/src/experiments/spikes.test.ts @@ -0,0 +1,185 @@ +import { expect, spyOn, test } from "bun:test"; +import { parse } from "@chopin/dialect/parse"; +import { serialize } from "@chopin/dialect/serialize"; +import { ulid } from "@chopin/dialect/ulid"; + +import * as room from "../plan/room"; +import { canonical } from "../mcp/create"; +import { calloutDigest, callouts, reconcileCallout, withCallout } from "./spike-placement"; +import { renderKey, spikeCallout, spikeReport, uncertain } from "./spikes"; + +import type { Investigation } from "@chopin/experiment/records"; + +const IMAGE = `/images/${"a".repeat(64)}.png`; + +function spike(state: Investigation["state"], report?: string): Investigation { + return { + id: crypto.randomUUID(), + documentId: crypto.randomUUID(), + requester: "U_1", + brief: "brief", + revision: 1, + state, + generation: 1, + expiresAt: 0, + createdAt: 0, + updatedAt: 0, + progress: "", + views: {}, + decisions: [], + receipts: {}, + spike: { + digest: "sha256:x", + passage: "We are unsure whether drag handles work on touch.", + callout: ulid(), + login: "maggie", + placed: true, + }, + ...(report + ? { + result: { + schemaVersion: 1 as const, + report, + datasets: [], + views: [], + evidence: [], + provenance: { environment: "", checks: [], limitations: [] }, + }, + } + : {}), + }; +} + +test("a submitted spike renders as a titled callout with findings and screenshots", () => { + let report = spikeReport({ + headline: "Drag handles work on touch with a 44px target", + findings: ["Pointer events fire on iOS Safari.", "A
handle {needs} padding."], + recommendation: "Keep drag handles; size them to 44px.", + images: [IMAGE], + }); + let checked = canonical(report); + expect("source" in checked).toBe(true); + let value = spike("completed", "source" in checked ? checked.source : report); + expect(renderKey(value)).toBe("completed"); + let source = serialize({ type: "root", children: [spikeCallout(value)] }); + expect(source).toContain(` { + let running = serialize({ type: "root", children: [spikeCallout(spike("running"))] }); + expect(running).toContain(`title="Prototyping…"`); + expect(running).toContain("maggie's coding agent"); + room.validate(running); + let stopped = spike("failed"); + stopped.progress = "Agent stopped: cancelled"; + let source = serialize({ type: "root", children: [spikeCallout(stopped)] }); + expect(source).toContain(`type="warning"`); + expect(source).toContain("Agent stopped: cancelled"); + expect(renderKey(spike("running"))).toBe("running"); + expect(renderKey(spike("requested"))).toBe("queued"); + expect(renderKey(spike("queued"))).toBe("queued"); + let queued = serialize({ type: "root", children: [spikeCallout(spike("queued"))] }); + expect(queued).toContain(`title="Queued"`); + expect(queued).not.toContain("Prototyping"); + room.validate(queued); + expect(renderKey(spike("interrupted"))).toBe("stopped"); +}); + +test("a callout lands directly under its passage and is later replaced in place", async () => { + let errors = spyOn(console, "error").mockImplementation(() => {}); + let passage = "We are unsure whether drag handles work on touch screens at all."; + let document = await room.create(`# Plan\n\n${passage}\n\nAnother paragraph.\n`); + try { + let value = spike("running"); + let digest = room.digests(document)[1]; + let first = reconcileCallout(document, { + callout: value.spike!.callout, + node: spikeCallout(value), + after: digest, + }); + expect(typeof first).toBe("object"); + let source = room.project(document); + let blocks = parse(source).children; + expect(blocks.map(node => node.type)).toEqual([ + "heading", + "paragraph", + "mdxJsxFlowElement", + "paragraph", + ]); + expect(callouts(source).has(value.spike!.callout)).toBe(true); + expect( + reconcileCallout(document, { callout: value.spike!.callout, node: spikeCallout(value) }), + ).toBe("unchanged"); + let done = spike( + "completed", + spikeReport({ + headline: "It works", + findings: ["Yes."], + recommendation: "Ship.", + images: [], + }), + ); + done.spike = value.spike; + reconcileCallout(document, { callout: value.spike!.callout, node: spikeCallout(done) }); + let after = room.project(document); + expect(after).toContain(`title="It works"`); + expect(after).not.toContain("Prototyping"); + expect(parse(after).children).toHaveLength(4); + expect( + reconcileCallout(document, { + callout: ulid(), + node: spikeCallout(done), + after: "sha256:none", + }), + ).toBe("missing"); + } finally { + document.doc.destroy(); + errors.mockRestore(); + } +}); + +test("an edited callout keeps its edits and the new state lands in a sibling after it", () => { + let value = spike("running"); + let running = spikeCallout(value); + let rendered = calloutDigest(running); + let passage = parse("Passage that started the spike.\n").children[0]; + let edited = parse( + serialize({ type: "root", children: [running] }).replace("quick prototype", "tiny prototype"), + ).children[0]; + let done = spike("failed"); + done.spike = value.spike; + let untouched = withCallout([passage, running], { + callout: value.spike!.callout, + node: spikeCallout(done), + rendered, + }); + if (typeof untouched === "string") throw new Error("expected a replacement"); + expect(untouched.callout).toBe(value.spike!.callout); + expect(untouched.children).toHaveLength(2); + let kept = withCallout([passage, edited], { + callout: value.spike!.callout, + node: spikeCallout(done), + rendered, + }); + if (typeof kept === "string") throw new Error("expected a sibling"); + expect(kept.callout).not.toBe(value.spike!.callout); + expect(kept.children).toHaveLength(3); + let source = serialize({ type: "root", children: kept.children }); + expect(source).toContain("tiny prototype"); + expect(source).toContain(` { + expect(uncertain("We need to know whether WebSockets survive the proxy.")).toBe(true); + expect(uncertain("This is an open decision for the team.")).toBe(true); + expect(uncertain("The page lists every document in the repository.")).toBe(false); +}); diff --git a/apps/server/src/experiments/spikes.ts b/apps/server/src/experiments/spikes.ts new file mode 100644 index 000000000..f9a98802a --- /dev/null +++ b/apps/server/src/experiments/spikes.ts @@ -0,0 +1,139 @@ +import { parse } from "@chopin/dialect/parse"; +import { serialize } from "@chopin/dialect/serialize"; +import { z } from "zod"; + +import type { Investigation } from "@chopin/experiment/records"; +import type { BlockContent, ListItem, Nodes, PhrasingContent, Root, RootContent } from "mdast"; + +/** What a spike's local agent submits; the server turns it into the report and the callout. */ +export const spikeSubmissionSchema = z.object({ + headline: z.string().trim().min(1).max(100), + findings: z.array(z.string().trim().min(1).max(600)).min(1).max(6), + recommendation: z.string().trim().min(1).max(1200), + images: z.array(z.string().regex(/^\/images\/[0-9a-f]{64}\.(png|jpg|jpeg|webp|gif)$/)).max(3), +}).strict(); +export type SpikeSubmission = z.infer; + +/** Explicit uncertainty, used when Jev cannot judge a passage. */ +const UNCERTAIN = + /\b(unsure|not sure|uncertain|whether|open (question|decision)|need to (know|check|find out|test|confirm)|unclear|tbd|don't know|do not know|might not|risk(y)?)\b/i; + +export function uncertain(text: string): boolean { + return UNCERTAIN.test(text); +} + +/** Plain text of an MDAST node. */ +export function text(node: Nodes): string { + if ("value" in node && typeof node.value === "string") return node.value; + if ("children" in node) return (node.children as Nodes[]).map(text).join(" "); + return ""; +} + +/** Prose a spike can start from: paragraphs, lists, quotes. Never headings or components. */ +export function eligible(node: RootContent): boolean { + if (!["paragraph", "list", "blockquote"].includes(node.type)) return false; + return text(node).replace(/\s+/g, " ").trim().length >= 40; +} + +export function spikeBrief(passage: string): string { + let quoted = passage.trim().split("\n").map(line => `> ${line}`).join("\n"); + return [ + "Spike: build the smallest prototype or proof of concept that answers the open question " + + "in this passage. Spend at most about 15 minutes.", + "", + quoted, + "", + "You are on a throwaway branch in a disposable worktree. Do not push or open a pull request.", + "Take 1-3 screenshots of the running prototype with Playwright at a 1280x800 viewport, " + + "saved as PNG files in this worktree, and upload each with upload_image_file({path}). " + + "Then call submit_spike_result with a one-line headline, 3-6 short findings, a " + + "recommendation, and the uploaded image paths.", + ].join("\n"); +} + +function paragraph(...children: PhrasingContent[]): BlockContent { + return { type: "paragraph", children }; +} + +/** Canonical report MDX: bold headline, findings, recommendation, then screenshots. */ +export function spikeReport(input: SpikeSubmission): string { + let items: ListItem[] = input.findings.map(finding => ({ + type: "listItem", + spread: false, + children: [paragraph({ type: "text", value: finding })], + })); + let root: Root = { + type: "root", + children: [ + paragraph({ type: "strong", children: [{ type: "text", value: input.headline }] }), + { type: "list", ordered: false, spread: false, children: items }, + paragraph( + { type: "strong", children: [{ type: "text", value: "Recommendation:" }] }, + { type: "text", value: ` ${input.recommendation}` }, + ), + ...input.images.map((url, index) => + paragraph({ type: "image", url, alt: `Prototype screenshot ${index + 1}` }) + ), + ], + }; + return serialize(root); +} + +/** The callout state a spike's record implies; rewritten only when this changes. */ +export function renderKey(value: Investigation): string { + if (value.state === "completed") return "completed"; + if (["failed", "interrupted", "cancelled"].includes(value.state)) return "stopped"; + // Spikes run one at a time per local agent; until one claims this run it is only waiting. + if (value.state === "requested" || value.state === "queued") return "queued"; + return "running"; +} + +function callout(id: string, type: string, title: string, children: RootContent[]): RootContent { + return { + type: "mdxJsxFlowElement", + name: "Callout", + attributes: [ + { type: "mdxJsxAttribute", name: "id", value: id }, + { type: "mdxJsxAttribute", name: "type", value: type }, + { type: "mdxJsxAttribute", name: "title", value: title.slice(0, 100) }, + ], + children: children as BlockContent[], + }; +} + +/** The Callout projected under a spike's passage for its current state. */ +export function spikeCallout(value: Investigation): RootContent { + let spike = value.spike; + if (!spike) throw new Error("not a spike"); + let key = renderKey(value); + if (key === "completed" && value.result) { + let blocks = parse(value.result.report).children; + let [first, ...rest] = blocks; + let headline = first?.type === "paragraph" ? text(first).trim() : ""; + return callout(spike.callout, "tip", headline || "Prototype result", headline ? rest : blocks); + } + if (key === "stopped") { + return callout(spike.callout, "warning", "Prototype stopped", [ + paragraph({ + type: "text", + value: value.progress.trim() || "The coding agent stopped before reporting.", + }), + ]); + } + if (key === "queued") { + return callout(spike.callout, "note", "Queued", [ + paragraph({ + type: "text", + value: `${spike.login}'s coding agent will build a quick prototype to test the passage ` + + "above once it finishes its current work. Delete this callout to cancel it.", + }), + ]); + } + return callout(spike.callout, "note", "Prototyping…", [ + paragraph({ + type: "text", + value: `${spike.login}'s coding agent is building a quick prototype to test the passage ` + + "above. Delete this callout to stop it.", + }), + ]); +} diff --git a/apps/server/src/main.ts b/apps/server/src/main.ts index 0ba3e43d1..0411c8789 100644 --- a/apps/server/src/main.ts +++ b/apps/server/src/main.ts @@ -50,10 +50,12 @@ import { JobService } from "./jobs/service"; import { DocumentSummaryCoordinator } from "./jobs/summary-coordinator"; import { liveClassifier } from "./tasks/live-gate"; import { LiveSyncCoordinator } from "./tasks/live-sync"; +import { jevJudge, SpikeScout } from "./experiments/spike-scout"; +import { spikeHost } from "./experiments/spike-host"; +import { askJev } from "./conversation-plan/jev"; 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"; @@ -114,6 +116,7 @@ const LEASE_SAFETY_MS = 5_000; const SESSION_CLEANUP_MS = 5 * 60_000; const ACCESS_RECHECK_MS = 60_000; const RESEARCH_RECOVERY_RETRY_MS = 10_000; +const SPIKE_SETTLE_MS = 30_000; let server: Server; let heldLease: Lease | undefined; @@ -134,6 +137,7 @@ let referenceService: ReferenceService | undefined; let summaryCoordinator: DocumentSummaryCoordinator | undefined; let liveSync: LiveSyncCoordinator | undefined; let buildReadiness: BuildReadiness | undefined; +let spikeScout: SpikeScout | undefined; let descriptionProjector: DocumentDescriptionProjector | undefined; let researchBriefJobs: JobService | undefined; let documentLocks = new Map>(); @@ -192,6 +196,7 @@ function documentBackend(): Service.Backend { summaryCoordinator?.schedule(target); liveSync?.schedule(target); buildReadiness?.schedule(target); + spikeScout?.schedule(target); }, liveBuild: !!config.liveBuild, onBuildStopped: id => liveSync?.stopped(id), @@ -199,6 +204,8 @@ function documentBackend(): Service.Backend { void relinkDecisions(id).catch(err => console.error("chopin: could not re-link decisions after a build -", err) ); + // Spike callouts the lock deferred land now. + void spikeScout?.refresh(id); }, }; } @@ -322,6 +329,9 @@ async function plan(room: Rooms.Room, server: Server): Promise {}); if (Inject.enabled(opened)) Inject.ask(opened, server, room.id); if (Marks.enabled(opened)) await Marks.mark(opened); + // A lock that released while the room was closed left spike callouts deferred. + // Retrying only when unlocked keeps the retry's own room open from looping. + if (spikeScout && !Service.implementationActive(opened)) void spikeScout.refresh(room.id); } return opened; }); @@ -1275,6 +1285,7 @@ function drain(): Promise { summaryCoordinator?.close(); liveSync?.close(); buildReadiness?.close(); + spikeScout?.close(); let stoppingJobs = jobRunner?.shutdown(); ownerBindings?.revokeAll(); for (let result of await Promise.allSettled([stoppingJobs])) { @@ -1719,10 +1730,20 @@ experiments = registerExperimentRoutes(router, hostedAuth, { } }); }), + async settleSpikes(documentId) { + if (!spikeScout) return; + let timer: ReturnType | undefined; + // Bounded: a stuck placement must not hold a build forever. + await Promise.race([ + spikeScout.settle(documentId), + new Promise(resolve => timer = setTimeout(resolve, SPIKE_SETTLE_MS)), + ]).finally(() => clearTimeout(timer)); + }, changed(documentId) { if (server && !draining) { broadcast(server, documentId, { kind: "experiment:changed", ts: 0, documentId }); } + void spikeScout?.refresh(documentId); }, async canMutate(id) { let active = Rooms.get(id)?.plan; @@ -1959,6 +1980,21 @@ if (config.liveBuild) { error: err => console.error("chopin: living-document rebuild failed -", err), }); } +if (config.liveSpikes) { + let jev = config.conversationPlanModel ?? "jev-latest"; + let timeoutMs = config.conversationPlanTimeoutMs ?? 30_000; + spikeScout = new SpikeScout({ + host: spikeHost({ + service: experiments.service, + connections: experiments.connections, + withPlan: withImplementationPlan, + }), + judge: process.env.JEV_API_KEY || process.env.TYPESAFE_API_KEY + ? jevJudge(request => askJev(request, { model: jev, timeoutMs })) + : undefined, + error: err => console.error("chopin: living-document spike failed -", err), + }); +} registerChannelRoutes(router, hostedAuth, { onAgentReset: channelOwnerReset, onChannelArchived: archiveChannel, @@ -2045,6 +2081,7 @@ try { summaryCoordinator?.close(); liveSync?.close(); buildReadiness?.close(); + spikeScout?.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 a85cd876f..939a86665 100644 --- a/apps/server/src/plan/service.ts +++ b/apps/server/src/plan/service.ts @@ -1,4 +1,4 @@ -import { locksEditing, restoreBuilds, restoreLive } from "../tasks/builds"; +import { locksEditing, queuedOnly, restoreBuilds, restoreLive } from "../tasks/builds"; import { announceImplementation, implementationStatus } from "../tasks/notifications"; import type { BuildRequest } from "@chopin/protocol/implementation"; /** @@ -127,6 +127,8 @@ export type DocumentTarget = { revision: number; source: string; sourceHash: string; + /** The person whose update this was, when one person's edit caused it. */ + editor?: string; }; /** Durable MCP context for a document created through the hosted surface. */ @@ -163,6 +165,8 @@ type Persistence = Backend & { committedDocument: Uint8Array; committedSidecar: JsonValue; closing: boolean; + /** The member whose batch is committing, for derived work that follows their edit. */ + editor?: string; }; type Captured = { @@ -942,6 +946,7 @@ async function commitHosted( revision: captured.revision, source: captured.source, sourceHash: captured.sourceHash, + ...(durable.editor ? { editor: durable.editor } : {}), }); } catch (err) { console.warn(`[plan] could not schedule derived work for ${durable.channelId}:`, err); @@ -1640,6 +1645,7 @@ async function commit(plan: Plan): Promise { createHash("sha256").update(merged).digest("hex") }`; try { + plan.persistence.editor = batch.at(-1)?.ws.data.principalId; await commitHosted( plan, merged, @@ -1655,6 +1661,8 @@ async function commit(plan: Plan): Promise { plan.threads = previousThreads; await rejectBatch(plan, batch, ["research-reference-conflict"]); return; + } finally { + plan.persistence.editor = undefined; } for (let item of batch) { @@ -1726,9 +1734,16 @@ export async function publishStaged( roomId: string, candidate: Plan, mutation?: room.Mutation, - options?: { notifyDocumentPersisted?: boolean; agent?: boolean }, + options?: { + notifyDocumentPersisted?: boolean; + agent?: boolean; + /** A callout-only edit the caller carries a queued first build across. */ + queuedBuild?: boolean; + }, ): Promise { - if (implementationActive(plan)) throw new ImplementationActiveError(); + if (implementationActive(plan) && !(options?.queuedBuild && queuedOnly(plan))) { + throw new ImplementationActiveError(); + } let source = room.project(candidate.document); if (!mutation && source !== room.project(plan.document)) { throw new Error("staged document changed without a mutation"); diff --git a/apps/server/src/tasks/builds.ts b/apps/server/src/tasks/builds.ts index 3feb10f99..34f53abde 100644 --- a/apps/server/src/tasks/builds.ts +++ b/apps/server/src/tasks/builds.ts @@ -55,6 +55,35 @@ export function locksEditing(build: { state: string; kind?: string }): boolean { return build.kind !== "rebuild" && ["queued", "starting", "running"].includes(build.state); } +/** + * A first build still waiting for its connection, when it is all that locks editing and it is + * bound to the current revision. Only callout-only edits may land under it, carried by + * `carryQueued`. + */ +export function queuedOnly(plan: Plan): BuildRequest | undefined { + if (plan.claiming || plan.execution) return undefined; + let locking = plan.builds.filter(locksEditing); + let build = locking.length === 1 ? locking[0] : undefined; + let version = plan.graph?.versions.at(-1); + return build?.state === "queued" && build === plan.builds.at(-1) + && build.planRevision === plan.revision && version?.state === "approved" + && version.planRevision === plan.revision + ? build + : undefined; +} + +/** Rebind the queued build and its approved graph to the revision a callout-only edit makes. */ +export function carryQueued( + plan: Plan, + revision: number, +): { builds: BuildRequest[]; graph: Graph } { + let build = queuedOnly(plan); + if (!build) throw new Error("no queued build to carry"); + let graph = structuredClone(plan.graph!); + graph.versions[graph.versions.length - 1].planRevision = revision; + return { builds: [...plan.builds.slice(0, -1), { ...build, planRevision: revision }], graph }; +} + export function restoreBuilds(value: unknown, repositoryId?: string): BuildRequest[] { if (value === undefined) return []; let builds = z.array(buildSchema).max(100).parse(value); diff --git a/apps/server/src/tasks/routes.test.ts b/apps/server/src/tasks/routes.test.ts index c948a882a..08c9b58ea 100644 --- a/apps/server/src/tasks/routes.test.ts +++ b/apps/server/src/tasks/routes.test.ts @@ -31,7 +31,9 @@ async function prepare(plan: Plan.Plan) { }); } -async function setup(options: { liveBuild?: boolean } = {}) { +async function setup( + options: { liveBuild?: boolean; settleSpikes?: (documentId: string) => Promise } = {}, +) { let context = await openPlan("# Launcher plan\n"); let github = { async user(token: string) { @@ -94,6 +96,7 @@ async function setup(options: { liveBuild?: boolean } = {}) { changed: () => {}, context: async () => ({ source: "# Launcher plan", revision: 0 }), implementations: () => implementations, + settleSpikes: options.settleSpikes, }); implementations = registerImplementationRoutes(router, auth, { liveBuild: options.liveBuild ?? false, @@ -176,6 +179,26 @@ async function tool( return result.isError ? result : JSON.parse(result.content[0].text); } +test("a build claim waits for the document's spike results to land", async () => { + let seen: string[] = []; + let context: Awaited> = await setup({ + settleSpikes: async documentId => { + await new Promise(resolve => setTimeout(resolve, 10)); + seen.push(`${documentId}:${context.plan.builds.at(-1)?.state}`); + }, + }); + let connection = await paired(context); + let built = await (await context.call(context.path, { + planRevision: 0, + graphVersion: 1, + graphRevision: 1, + })).json(); + let claim = await tool(context, connection.token, "claim_implementation_build", { id: built.id }); + expect(claim.build.id).toBe(built.id); + expect(seen).toEqual([`${context.plan.id}:queued`]); + await Plan.close(context.plan); +}); + test("a paired workspace claims a browser build once and reports through run-scoped MCP", async () => { let context = await setup(); let connection = await paired(context); diff --git a/e2e/jev.ts b/e2e/jev.ts index 475237904..d9aa8cf35 100644 --- a/e2e/jev.ts +++ b/e2e/jev.ts @@ -414,6 +414,21 @@ let fake = async (input: Parameters[0], init?: Parameters key.startsWith("spike_"))) { + let passages = (request.state as { passages?: Array<{ key: string; text: string }> }) + .passages ?? []; + let open = /\b(unsure|whether|open decision|need to know|prototype)\b/i; + return Response.json({ + model: "jev-e2e-fixture", + answers: Object.fromEntries( + passages.map(passage => [passage.key, { + type: "noul", + noul: open.test(passage.text) ? 0.9 : 0.1, + }]), + ), + usage: { input_tokens: 10, output_tokens: 5 }, + }); + } if ("rebuild" in request.questions) { let changes = (request.state as { changes?: string }).changes ?? ""; let lines = changes.split("\n").filter(line => /^[-+] /.test(line)); diff --git a/packages/experiment/src/records.ts b/packages/experiment/src/records.ts index 116429175..3daf6762e 100644 --- a/packages/experiment/src/records.ts +++ b/packages/experiment/src/records.ts @@ -23,6 +23,33 @@ export const decisionSchema = z.object({ }).strict(); export type EvidenceDecision = z.infer; +/** + * A prototype the scout started under a passage. The callout is a projection; this record owns + * whether the spike was dismissed, so a removed callout never re-triggers its passage. + */ +export const spikeSchema = z.object({ + /** Canonical digest of the source passage when the spike started. */ + digest: z.string().min(1).max(100), + /** The passage text the brief was drafted from. */ + passage: z.string().min(1).max(4000), + /** ULID of the Callout projected under the passage. */ + callout: z.string().regex(/^[0-7][0-9A-HJKMNP-TV-Z]{25}$/), + /** GitHub login of the person whose local agent runs it. */ + login: z.string().min(1).max(100), + placed: z.boolean(), + /** + * Set before the callout is first published and cleared once `placed` is persisted, so a + * crash in between recovers by finding the callout in place rather than inserting another. + */ + placing: z.boolean().optional(), + /** The state last written into the callout, so progress renewals never rewrite edits. */ + rendered: z.string().max(40).optional(), + /** Digest of the callout block as last rendered; a different block holds human edits. */ + calloutDigest: z.string().max(100).optional(), + dismissed: z.boolean().optional(), +}).strict(); +export type Spike = z.infer; + export const investigationSchema = z.object({ id: z.string().uuid(), documentId: z.string(), @@ -47,6 +74,7 @@ export const investigationSchema = z.object({ updatedAt: z.number(), progress: z.string().max(2000), parentId: z.string().uuid().optional(), + spike: spikeSchema.optional(), candidate: resultSchema.optional(), result: resultSchema.optional(), views: z.record(z.string(), stateSchema),