diff --git a/.lane/memory/apps/server/src/experiments/spike-host.ts/01M4M43PSSVR2QCZSD0MSTZ12Y-landed-fires-once-when-a-com.md b/.lane/memory/apps/server/src/experiments/spike-host.ts/01M4M43PSSVR2QCZSD0MSTZ12Y-landed-fires-once-when-a-com.md new file mode 100644 index 000000000..b3bcfe6ec --- /dev/null +++ b/.lane/memory/apps/server/src/experiments/spike-host.ts/01M4M43PSSVR2QCZSD0MSTZ12Y-landed-fires-once-when-a-com.md @@ -0,0 +1,8 @@ +--- +id: 01M4M43PSSVR2QCZSD0MSTZ12Y +anchor: landed +created: 2026-10-10T23:59:27Z +norm: '1' +--- + +landed fires once, when a completed result is first placed (rendered changes to completed); main.ts uses it to start a Planner turn that settles the passage, claiming the Planner for the spike requester's open socket if unowned. diff --git a/.lane/memory/packages/dialect/src/nodes/callout-fold.ts/01M4M43PSAZGBJBTD4JWW04C6P-fold-is-nodestate-outside-co.md b/.lane/memory/packages/dialect/src/nodes/callout-fold.ts/01M4M43PSAZGBJBTD4JWW04C6P-fold-is-nodestate-outside-co.md new file mode 100644 index 000000000..d45d3993a --- /dev/null +++ b/.lane/memory/packages/dialect/src/nodes/callout-fold.ts/01M4M43PSAZGBJBTD4JWW04C6P-fold-is-nodestate-outside-co.md @@ -0,0 +1,8 @@ +--- +id: 01M4M43PSAZGBJBTD4JWW04C6P +anchor: const calloutFoldState +created: 2026-10-10T23:59:27Z +norm: '1' +--- + +Fold is NodeState outside containers.ts on purpose: containers.ts carries whole-file sourceHash design-contract exceptions, so any edit there needs exception renewal. The editor plugin owns data-fold-state/data-fold-lead and the unmanaged disclosure host. diff --git a/apps/server/src/chat/server-instruction.ts b/apps/server/src/chat/server-instruction.ts new file mode 100644 index 000000000..ac535bb2c --- /dev/null +++ b/apps/server/src/chat/server-instruction.ts @@ -0,0 +1,86 @@ +import * as Rooms from "../rooms"; + +import type * as Service from "../plan/service"; +import type * as Chat from "./service"; + +export type Instruction = { handle: string; text: string; said: string }; + +export type ServerInstructionDeps = { + agent: boolean; + unavailable: (channelId: string) => boolean; + /** The channel's existing Planner owner, if one is bound. */ + owner: ( + channelId: string, + ) => Promise<{ repository: Chat.Room["repository"]; release: () => void } | undefined>; + transition: (channelId: string, action: () => Promise) => Promise; + /** Attach the document to a held room, reopening it if it was evicted. */ + open: (room: Rooms.Room) => Promise; + conversation: ( + room: Rooms.Room, + claimantSessionId: string | undefined, + repository: Chat.Room["repository"], + ) => Chat.Room; + instruct: (context: Chat.Room, instruction: Instruction) => void | Promise; + evict: (room: Rooms.Room) => void; +}; + +/** + * Start a Planner turn nobody typed, and say whether Chat accepted it (started or queued). + * `claimant` is a user whose open, writable socket in the room may claim an unowned Planner; + * without one, only an existing owner will do, and a room evicted since is reopened for it. + */ +export async function serverInstruction( + deps: ServerInstructionDeps, + channelId: string, + compose: (plan: Service.Plan) => Instruction | undefined, + claimant?: string, +): Promise { + if (!deps.agent || deps.unavailable(channelId)) return false; + let opened = Rooms.get(channelId)?.plan; + if (opened && !compose(opened)) return false; + let socket = claimant + ? [...(Rooms.get(channelId)?.members.values() ?? [])].find(ws => + ws.data.principalId === claimant && ws.data.canEdit && !ws.data.closed + ) + : undefined; + let repository: Chat.Room["repository"] | undefined; + if (socket) { + repository = { + id: socket.data.repositoryId, + owner: socket.data.repositoryOwner, + name: socket.data.repositoryName, + defaultBranch: socket.data.repositoryDefaultBranch, + }; + } else { + let binding = await deps.owner(channelId); + if (!binding) return false; + repository = binding.repository; + binding.release(); + } + return deps.transition(channelId, async () => { + await Rooms.get(channelId)?.closing; + let held = Rooms.hold(channelId); + let release = () => { + held.release(); + deps.evict(held.room); + }; + let running = false; + try { + let current = await deps.open(held.room); + let instruction = compose(current); + if (!instruction) return false; + let context = deps.conversation(held.room, socket?.data.sessionId, repository); + let before = context.chat.running; + let waiting = context.chat.waiting.length; + await deps.instruct(context, instruction); + if (context.chat.running) { + running = true; + void context.chat.running.finally(release).catch(() => {}); + } + return (!!context.chat.running && context.chat.running !== before) + || context.chat.waiting.length > waiting; + } finally { + if (!running) release(); + } + }); +} diff --git a/apps/server/src/experiments/connections.ts b/apps/server/src/experiments/connections.ts index bd52e88c1..f46d5a9c5 100644 --- a/apps/server/src/experiments/connections.ts +++ b/apps/server/src/experiments/connections.ts @@ -28,6 +28,11 @@ export type Grant = { run?: { id: string; documentId: string; generation: number; kind?: "implementation" | "rebuild" }; }; +/** A connector that stops polling is forgotten this long after it was last heard from. */ +const CONNECTION_MS = 90_000; +/** Connectors poll at least every 20 seconds while idle and heartbeat every 10 while working. */ +const FRESH_MS = 45_000; + const CODE_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789"; function confirmation() { let bytes = randomBytes(8); @@ -89,7 +94,7 @@ export class Connections { sessionId: owner.sessionId, label, source: { ...source, repositoryId }, - expiresAt: this.now() + 90_000, + expiresAt: this.now() + CONNECTION_MS, }; value.connection = connection; value.token = randomBytes(32).toString("base64url"); @@ -113,7 +118,14 @@ export class Connections { return value && value.expiresAt > this.now() ? value : undefined; } touch(connection: Connection) { - connection.expiresAt = this.now() + 90_000; + connection.expiresAt = this.now() + CONNECTION_MS; + } + /** + * Heard from within about two of its long-poll or heartbeat intervals. A silent connection + * stays usable until it expires, but new work should not be queued on it. + */ + fresh(connection: Connection) { + return connection.expiresAt - CONNECTION_MS > this.now() - FRESH_MS; } /** Live connections for a repository, most recently heard from first. */ list(repositoryId: string, owner?: string) { diff --git a/apps/server/src/experiments/routes.ts b/apps/server/src/experiments/routes.ts index 73962eb85..a42129c56 100644 --- a/apps/server/src/experiments/routes.ts +++ b/apps/server/src/experiments/routes.ts @@ -278,7 +278,12 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio if (repository.id !== found.connection.source.repositoryId) fail("repository-forbidden"); if (found.grant.run) await target(session, found.connection, found.grant.run.documentId); if (!connections.get(found.connection.id)) fail("connection-unavailable"); + let stale = !connections.fresh(found.connection); connections.touch(found.connection); + // Work held back from a silent connection can go to it again now it is heard from. + if (stale) { + options.connected?.(found.connection.source.repositoryId, found.connection.owner); + } return { ...found, session: session! }; } /** Write access to a run's document, which must belong to the connection's repository. */ diff --git a/apps/server/src/experiments/service.ts b/apps/server/src/experiments/service.ts index cd6d18782..018a48c9a 100644 --- a/apps/server/src/experiments/service.ts +++ b/apps/server/src/experiments/service.ts @@ -164,6 +164,21 @@ export class Experiments { delete value.candidate; }); } + /** Return a spike its lost connection interrupted to `requested`, for another local agent. */ + retry(id: string) { + return this.mutate(id, value => { + if (value.state !== "interrupted" || !value.spike || value.spike.dismissed) { + fail("invalid-state"); + } + value.state = "requested"; + value.progress = ""; + value.expiresAt = 0; + value.spike.retries = (value.spike.retries ?? 0) + 1; + delete value.connectionId; + delete value.input; + delete value.candidate; + }); + } async recover(all = false) { for (let value of await this.store.active()) { if (all || value.state !== "queued" && value.expiresAt <= this.now()) { diff --git a/apps/server/src/experiments/spike-host.test.ts b/apps/server/src/experiments/spike-host.test.ts index 139cf1665..c0533c525 100644 --- a/apps/server/src/experiments/spike-host.test.ts +++ b/apps/server/src/experiments/spike-host.test.ts @@ -15,21 +15,103 @@ 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"; +import { MAX_SPIKE_RETRIES, 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 landed: string[] = []; + let clock = { now: Date.now() }; + let connections = new Connections(() => clock.now); let host = spikeHost({ service, - connections: new Connections(), + connections, withPlan: (_id, action) => action(context.plan), + landed: (_channelId, value) => landed.push(value.id), }); - return { ...context, service, host }; + return { ...context, service, host, landed, connections, clock }; } +/** Pair a local agent for the plan's repository, as the connector and pairing page do. */ +function pair(context: Awaited>) { + let pending = context.connections.create({ + repository: "org/repo", + commit: "a".repeat(40), + label: "laptop", + }); + return context.connections.approve( + pending.id, + { id: "U_test", login: "maggie", sessionId: "S_1" }, + context.plan.persistence.repositoryId, + ); +} + +test("a spike its connection lost waits, then runs on reconnect a bounded number of times", async () => { + let errors = spyOn(console, "error").mockImplementation(() => {}); + let context = await hosted(); + let scout = new SpikeScout({ host: context.host }); + let { plan, service } = 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 service.store.list(plan.id); + expect(value.state).toBe("interrupted"); + // Nothing to run on yet: the callout waits for the owner's agent, without "Propose" copy. + await scout.refresh(plan.id); + expect(room.project(plan.document)).toContain("Waiting for @maggie’s coding agent"); + expect(room.project(plan.document)).not.toContain("Propose"); + + for (let attempt = 1; attempt <= MAX_SPIKE_RETRIES; attempt++) { + let connection = pair(context); + scout.connected(plan.persistence.repositoryId, "U_test"); + await scout.refresh(plan.id); + value = (await service.store.get(value.id))!; + expect(value).toMatchObject({ state: "queued", connectionId: connection.id }); + expect(value.spike?.retries).toBe(attempt); + expect(room.project(plan.document)).toContain(`title="Prototype queued"`); + // The connector is killed again; the sweep's expiry interrupts it. + context.connections.revoke(connection.id); + await service.stop(value.id, "interrupted", "Workspace connection expired."); + await scout.refresh(plan.id); + } + // Out of retries: a fresh connection no longer restarts it, and it reads as stopped. + pair(context); + scout.connected(plan.persistence.repositoryId, "U_test"); + await scout.refresh(plan.id); + expect((await service.store.get(value.id))?.state).toBe("interrupted"); + expect(room.project(plan.document)).toContain('type="warning"'); + expect(room.project(plan.document)).not.toContain("Waiting for"); + } finally { + scout.close(); + errors.mockRestore(); + await Service.close(plan); + } +}); + +test("new spike work skips a connection silent for two heartbeats", async () => { + let context = await hosted(); + try { + let connection = pair(context); + let repositoryId = context.plan.persistence.repositoryId; + expect(await context.host.connection(repositoryId, "U_test", context.plan.id)).toMatchObject({ + id: connection.id, + }); + context.clock.now += 50_000; + expect(context.connections.get(connection.id)).toBeDefined(); + expect(await context.host.connection(repositoryId, "U_test", context.plan.id)) + .toBeUndefined(); + context.connections.touch(connection); + expect(await context.host.connection(repositoryId, "U_test", context.plan.id)).toBeDefined(); + } finally { + await Service.close(context.plan); + } +}); + test("a spike is durably placing before its callout is published", async () => { let errors = spyOn(console, "error").mockImplementation(() => {}); let context = await hosted(); @@ -171,8 +253,12 @@ test("a spike result deferred by a first build's lock lands once the lock releas 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.state).toBe("interrupted"); expect(value.spike?.rendered).toBe("queued"); + // Out of automatic retries, so it renders as stopped rather than waiting. + await context.service.mutate(value.id, item => { + item.spike!.retries = MAX_SPIKE_RETRIES; + }); plan.builds = [{ id: crypto.randomUUID(), state: "running" } as never]; announceImplementation(plan); @@ -290,3 +376,38 @@ test("a spike that finishes while a first build is queued lands before the build await Service.close(plan); } }); + +test("a landed result asks once for its passage to be settled", async () => { + let errors = spyOn(console, "error").mockImplementation(() => {}); + let context = await hosted(); + 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 context.service.mutate(value.id, item => { + item.state = "completed"; + item.result = { + schemaVersion: 1, + report: "**Handles work**\n\n**Recommendation:** Keep them.\n\n- Fine on iOS.\n", + datasets: [], + views: [], + evidence: [], + provenance: { environment: "test", checks: [], limitations: [] }, + }; + }); + await context.host.refresh(plan.id); + await context.host.refresh(plan.id); + expect(context.landed).toEqual([value.id]); + expect((await context.service.store.get(value.id))?.spike?.settle).toBe(true); + let source = room.project(plan.document); + expect(source).toContain(`title="Handles work" fold="1"`); + room.validate(source); + } finally { + errors.mockRestore(); + await Service.close(plan); + } +}); diff --git a/apps/server/src/experiments/spike-host.ts b/apps/server/src/experiments/spike-host.ts index 0d73ff69e..b82a1c6a4 100644 --- a/apps/server/src/experiments/spike-host.ts +++ b/apps/server/src/experiments/spike-host.ts @@ -10,16 +10,19 @@ import { eligible, renderKey, spikeBrief, spikeCallout, text } from "./spikes"; import type { Connections } from "./connections"; import type { Experiments } from "./service"; +import type { Investigation } from "@chopin/experiment/records"; import type { SpikeHost } from "./spike-scout"; type Options = { service: Experiments; connections: Connections; withPlan: (channelId: string, action: (plan: Service.Plan) => Promise) => Promise; + /** A spike's result was just written into its callout; `spike.settle` is persisted. */ + landed?: (channelId: string, value: Investigation) => void; }; /** Spike records live with investigations; their callouts are projections placed after commit. */ -export function spikeHost({ service, connections, withPlan }: Options): SpikeHost { +export function spikeHost({ service, connections, withPlan, landed }: Options): SpikeHost { let spikes = async (channelId: string) => (await service.store.list(channelId)).filter(value => value.spike); let dismiss = async (_channelId: string, id: string) => { @@ -28,6 +31,38 @@ export function spikeHost({ service, connections, withPlan }: Options): SpikeHos }); await service.stop(id, "cancelled", "Dismissed."); }; + /** Hand a requested spike to a local agent; a vanished connection leaves it retryable. */ + let dispatch = async ( + channelId: string, + value: Investigation, + owner: string, + connectionId: string, + ) => { + let live = connections.get(connectionId); + try { + if (!live) throw new Error("connection gone"); + let context = await withPlan(channelId, plan => Service.readCurrentDocument(plan)); + await service.authorize(value.id, live.id, { + id: value.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( + value.id, + live ? "failed" : "interrupted", + "The local agent disconnected before it could start.", + ); + } + }; return { snapshot: channelId => withPlan(channelId, async plan => { @@ -51,7 +86,10 @@ export function spikeHost({ service, connections, withPlan }: Options): SpikeHos }), spikes, async connection(repositoryId, owner, channelId) { - let found = connections.candidates(repositoryId, owner, channelId)[0]; + // A connection silent past two heartbeats is probably gone; let it reconnect first. + let found = connections.candidates(repositoryId, owner, channelId).find(value => + connections.fresh(value) + ); return found && { id: found.id, login: found.login }; }, async start(channelId, { owner, connection, block }) { @@ -85,26 +123,11 @@ export function spikeHost({ service, connections, withPlan }: Options): SpikeHos 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."); - } + await dispatch(channelId, value, owner, connection.id); + }, + async retry(channelId, id, { owner, connection }) { + let value = await service.retry(id); + await dispatch(channelId, value, owner, connection.id); }, dismiss, async refresh(channelId) { @@ -128,13 +151,17 @@ export function spikeHost({ service, connections, withPlan }: Options): SpikeHos if (placed.status === "missing") await dismiss(channelId, value.id); else if (placed.status !== "deferred") { let { callout, digest } = placed; - await service.mutate(value.id, item => { + let landing = placed.status === "placed" && key === "completed" + && spike.rendered !== key; + let updated = 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; + if (landing) item.spike!.settle = true; }); + if (landing) landed?.(channelId, updated); } } }, diff --git a/apps/server/src/experiments/spike-scout.test.ts b/apps/server/src/experiments/spike-scout.test.ts index b6e7bf93f..e6d4cfd42 100644 --- a/apps/server/src/experiments/spike-scout.test.ts +++ b/apps/server/src/experiments/spike-scout.test.ts @@ -14,6 +14,7 @@ function block( function harness(snapshot: Partial = {}, verdict?: SpikeJudge) { let started: string[] = []; + let retried: string[] = []; let dismissed: string[] = []; let judged: string[][] = []; let records: Investigation[] = []; @@ -43,6 +44,12 @@ function harness(snapshot: Partial = {}, verdict?: SpikeJudge) { }, } as Investigation); }, + retry: async (_channelId, id, input) => { + retried.push(`${id}:${input.owner}`); + let value = records.find(item => item.id === id)!; + value.state = "queued"; + value.spike!.retries = (value.spike!.retries ?? 0) + 1; + }, dismiss: async (_id, id) => { dismissed.push(id); }, @@ -64,12 +71,14 @@ function harness(snapshot: Partial = {}, verdict?: SpikeJudge) { scout, state, started, + retried, dismissed, judged, records, timers, delays, disconnect: () => connected = false, + reconnect: () => connected = true, }; } @@ -281,3 +290,91 @@ test("passages waiting at the active-spike cap are judged when a spike frees cap expect(h.started).toEqual(["a", "b", "c", "d", "e"]); expect(asked).toHaveLength(3); }); + +test("a reconnecting local agent rescans its owner's documents missed while it was away", async () => { + let h = harness({ blocks: [block("a")] }); + h.disconnect(); + h.scout.schedule({ channelId: "D", editor: "U_1" }); + h.timers[0](); + await h.scout.check("D"); + expect(h.started).toEqual([]); + h.reconnect(); + h.scout.connected("R_2", "U_1"); + h.scout.connected("R_1", "U_2"); + await new Promise(resolve => setTimeout(resolve, 0)); + expect(h.started).toEqual([]); + h.scout.connected("R_1", "U_1"); + await new Promise(resolve => setTimeout(resolve, 0)); + expect(h.started).toEqual(["a"]); +}); + +test("a reconnect rescans only documents its owner's scan could not serve", async () => { + let h = harness({ blocks: [block("a")] }); + h.disconnect(); + h.scout.schedule({ channelId: "D", editor: "U_1" }); + h.timers[0](); + await h.scout.check("D"); + h.reconnect(); + let checks = 0; + let original = h.scout.check.bind(h.scout); + h.scout.check = id => { + checks++; + return original(id); + }; + h.scout.connected("R_1", "U_1"); + await new Promise(resolve => setTimeout(resolve, 0)); + expect(checks).toBe(1); + expect(h.started).toEqual(["a"]); + h.scout.connected("R_1", "U_1"); + expect(checks).toBe(1); +}); + +function interrupted(id: string, retries = 0): Investigation { + return { + id, + requester: "U_1", + state: "interrupted", + spike: { digest: id, callout: `CALLOUT${id}`, placed: true, retries }, + } as Investigation; +} + +test("a spike its connection lost is handed back on reconnect, with no edit", async () => { + let h = harness({ blocks: [block("a")], callouts: new Set(["CALLOUTa"]) }); + h.records.push(interrupted("a")); + h.disconnect(); + await h.scout.refresh("D"); + expect(h.retried).toEqual([]); + h.reconnect(); + h.scout.connected("R_1", "U_2"); + await new Promise(resolve => setTimeout(resolve, 0)); + expect(h.retried).toEqual([]); + h.scout.connected("R_1", "U_1"); + await new Promise(resolve => setTimeout(resolve, 0)); + expect(h.retried).toEqual(["a:U_1"]); + // The passage is never judged or started afresh. + expect(h.judged).toEqual([]); + expect(h.started).toEqual([]); +}); + +test("automatic retries stop at the cap", async () => { + let h = harness({ blocks: [block("a")], callouts: new Set(["CALLOUTa"]) }); + h.records.push(interrupted("a", 2)); + await h.scout.refresh("D"); + h.scout.connected("R_1", "U_1"); + await new Promise(resolve => setTimeout(resolve, 0)); + expect(h.retried).toEqual([]); + expect(h.started).toEqual([]); +}); + +test("a dismissed spike stays dismissed when its owner's agent reconnects", async () => { + let h = harness({ blocks: [block("a")] }); + h.records.push(interrupted("a")); + h.disconnect(); + // Its callout is gone, so the scan dismisses it rather than waiting to retry it. + await h.scout.refresh("D"); + expect(h.dismissed).toEqual(["a"]); + h.reconnect(); + h.scout.connected("R_1", "U_1"); + await h.scout.refresh("D"); + expect(h.retried).toEqual([]); +}); diff --git a/apps/server/src/experiments/spike-scout.ts b/apps/server/src/experiments/spike-scout.ts index 1c4592078..dcdcc5bdf 100644 --- a/apps/server/src/experiments/spike-scout.ts +++ b/apps/server/src/experiments/spike-scout.ts @@ -1,4 +1,4 @@ -import { uncertain } from "./spikes"; +import { retryable, uncertain } from "./spikes"; import type { JevRequest, JevResult } from "../conversation-plan/jev"; import type { Investigation } from "@chopin/experiment/records"; @@ -33,6 +33,12 @@ export type SpikeHost = { channelId: string, input: { owner: string; connection: { id: string; login: string }; block: SpikeBlock }, ): Promise; + /** Hand a spike its lost connection interrupted to its owner's reconnected local agent. */ + retry( + channelId: string, + id: string, + input: { owner: string; connection: { id: string; login: string } }, + ): 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. */ @@ -115,6 +121,8 @@ export class SpikeScout { #waiting = new Set(); /** Documents with a dismissal check already queued, so a burst of edits shares one. */ #dismissing = new Set(); + /** Documents whose last scan found no local agent for these owners, to rescan on reconnect. */ + #offline = new Map }>(); #closed = false; constructor(options: SpikeScoutOptions) { @@ -167,11 +175,30 @@ export class SpikeScout { return spikes; } - /** A spike record changed; project it into its callout, and scan again if it freed capacity. */ + /** + * A local agent (re)connected to a repository. Rescan its owner's documents whose last scan + * stopped for want of one, so a passage written while it was away needs no further edit. + */ + connected(repositoryId: string, owner: string): void { + if (this.#closed) return; + for (let [channelId, away] of this.#offline) { + if (away.repositoryId !== repositoryId || !away.owners.has(owner)) continue; + this.#offline.delete(channelId); + if (!this.#pending.has(channelId)) void this.check(channelId); + } + } + + /** + * A spike record changed; project it into its callout, and scan again if it freed capacity + * or a lost connection left a spike to hand to its owner's next local agent. + */ refresh(channelId: string): Promise { return this.#serial(channelId, async () => { await this.#options.host.refresh(channelId); - if (this.#waiting.delete(channelId)) await this.#check(channelId); + let waiting = this.#waiting.delete(channelId); + if (waiting || (await this.#options.host.spikes(channelId)).some(retryable)) { + await this.#check(channelId); + } }); } @@ -204,12 +231,32 @@ export class SpikeScout { let snapshot = await host.snapshot(channelId); if (!snapshot) return; let spikes = await this.#dismiss(channelId, snapshot); + this.#offline.delete(channelId); + let retried = 0; + let active = () => + retried + + spikes.filter(value => ACTIVE.includes(value.state) && !value.spike!.dismissed).length; + // A spike its connection lost goes back to its owner's local agent once one is heard from. + for (let value of spikes.filter(retryable)) { + if (this.#closed) return; + if (active() >= MAX_ACTIVE) { + this.#waiting.add(channelId); + break; + } + let owner = value.requester; + let connection = await host.connection(snapshot.repositoryId, owner, channelId); + if (!connection) { + this.#away(channelId, snapshot.repositoryId, owner); + continue; + } + await host.retry(channelId, value.id, { owner, connection }); + retried++; + } // 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 capacity = MAX_ACTIVE - active(); let seen = this.#seen.get(channelId) ?? new Set(); this.#seen.set(channelId, seen); let started = new Set(spikes.map(value => value.spike!.digest)); @@ -227,9 +274,12 @@ export class SpikeScout { 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. + // Without the editor's local agent there is nowhere to run; judge these again on reconnect. let connection = await host.connection(snapshot.repositoryId, editor, channelId); - if (!connection) return; + if (!connection) { + this.#away(channelId, snapshot.repositoryId, editor); + 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. @@ -256,6 +306,12 @@ export class SpikeScout { } } + #away(channelId: string, repositoryId: string, owner: string) { + let away = this.#offline.get(channelId) ?? { repositoryId, owners: new Set() }; + away.owners.add(owner); + this.#offline.set(channelId, away); + } + close(): void { this.#closed = true; for (let pending of this.#pending.values()) pending.cancel(); @@ -265,5 +321,6 @@ export class SpikeScout { this.#seen.clear(); this.#editors.clear(); this.#waiting.clear(); + this.#offline.clear(); } } diff --git a/apps/server/src/experiments/spike-settle.test.ts b/apps/server/src/experiments/spike-settle.test.ts new file mode 100644 index 000000000..913a72bb5 --- /dev/null +++ b/apps/server/src/experiments/spike-settle.test.ts @@ -0,0 +1,137 @@ +import { expect, test } from "bun:test"; + +import { serverInstruction } from "../chat/server-instruction"; +import * as Rooms from "../rooms"; +import { spikeSettler } from "./spike-settle"; + +import type { ServerInstructionDeps } from "../chat/server-instruction"; +import type * as Chat from "../chat/service"; +import type * as Service from "../plan/service"; +import type { Socket } from "../wire"; +import type { Investigation } from "@chopin/experiment/records"; + +const REPOSITORY = { id: "R_1", owner: "acme", name: "app", defaultBranch: "main" }; + +/** The settle path exactly as main.ts wires it, over real Rooms and fake Planner plumbing. */ +function harness(options: { agent?: boolean; owner?: boolean } = {}) { + let channelId = crypto.randomUUID(); + let record = { + id: crypto.randomUUID(), + documentId: channelId, + requester: "U_requester", + spike: { login: "maggie", settle: true }, + } as unknown as Investigation; + let opened: Rooms.Room[] = []; + let turns: { sessionId: string | undefined; repository: unknown; handle: string }[] = []; + let deps: ServerInstructionDeps = { + agent: options.agent ?? true, + unavailable: () => false, + owner: async () => options.owner ? { repository: REPOSITORY, release: () => {} } : undefined, + transition: (_id, action) => action(), + open: async room => { + opened.push(room); + return room.plan ??= {} as Service.Plan; + }, + conversation: (_room, sessionId, repository) => { + let chat = { waiting: [] as unknown[], running: undefined as Promise | undefined }; + return { chat, sessionId, repository } as unknown as Chat.Room; + }, + instruct: (context, instruction) => { + let { sessionId, repository } = context as unknown as { + sessionId?: string; + repository: unknown; + }; + turns.push({ sessionId, repository, handle: instruction.handle }); + context.chat.running = Promise.resolve(); + }, + evict: () => {}, + }; + let settler = spikeSettler({ + get: async () => record, + list: async () => [record], + mutate: async (_id, action) => action(record), + start: (id, value) => + serverInstruction(deps, id, () => ({ + handle: value.spike!.login, + text: "Settle the passage", + said: "Settling", + }), value.requester), + }); + let cleanup = () => { + let room = Rooms.get(channelId); + if (room) Rooms.forget(room); + }; + return { channelId, record, opened, turns, settler, cleanup }; +} + +function socket(channelId: string, principalId: string): Socket { + return { + data: { + room: channelId, + client: crypto.randomUUID(), + handle: "requester", + principalId, + sessionId: "S_requester", + canEdit: true, + repositoryId: REPOSITORY.id, + repositoryOwner: REPOSITORY.owner, + repositoryName: REPOSITORY.name, + repositoryDefaultBranch: REPOSITORY.defaultBranch, + }, + } as unknown as Socket; +} + +test("a requester with an open writable socket claims the Planner and the settle clears", async () => { + let context = harness(); + try { + let room = Rooms.join(socket(context.channelId, "U_requester")); + room.plan = {} as Service.Plan; + await context.settler.settle(context.channelId, context.record.id); + expect(context.turns).toEqual([ + { sessionId: "S_requester", repository: REPOSITORY, handle: "maggie" }, + ]); + expect(context.record.spike?.settle).toBeUndefined(); + } finally { + context.cleanup(); + } +}); + +test("without the requester, an existing owner settles it in a reopened room", async () => { + let context = harness({ owner: true }); + try { + expect(Rooms.get(context.channelId)).toBeUndefined(); + await context.settler.pending(context.channelId); + expect(context.opened.map(room => room.id)).toEqual([context.channelId]); + expect(context.turns).toEqual([ + { sessionId: undefined, repository: REPOSITORY, handle: "maggie" }, + ]); + expect(context.record.spike?.settle).toBeUndefined(); + } finally { + context.cleanup(); + } +}); + +test("without the requester or an owner, nothing is claimed and the settle stays pending", async () => { + let context = harness(); + try { + // Someone else's socket never claims for the requester. + Rooms.join(socket(context.channelId, "U_other")).plan = {} as Service.Plan; + await context.settler.settle(context.channelId, context.record.id); + expect(context.turns).toEqual([]); + expect(context.opened).toEqual([]); + expect(context.record.spike?.settle).toBe(true); + } finally { + context.cleanup(); + } +}); + +test("with the agent off, no turn starts and the settle stays pending", async () => { + let context = harness({ agent: false, owner: true }); + try { + await context.settler.pending(context.channelId); + expect(context.turns).toEqual([]); + expect(context.record.spike?.settle).toBe(true); + } finally { + context.cleanup(); + } +}); diff --git a/apps/server/src/experiments/spike-settle.ts b/apps/server/src/experiments/spike-settle.ts new file mode 100644 index 000000000..55b1ffefd --- /dev/null +++ b/apps/server/src/experiments/spike-settle.ts @@ -0,0 +1,40 @@ +import type { Investigation } from "@chopin/experiment/records"; + +type Deps = { + get: (id: string) => Promise; + list: (channelId: string) => Promise; + mutate: (id: string, action: (value: Investigation) => void) => Promise; + /** Start the Planner turn that settles the passage; true once Chat accepted it. */ + start: (channelId: string, value: Investigation) => Promise; +}; + +/** + * Settles landed spikes. `spike.settle` is persisted with `rendered = completed` and cleared + * only after Chat accepts the turn, so a landing with no room or no Planner owner stays + * pending and is retried when the room opens or an owner becomes available. + */ +export function spikeSettler({ get, list, mutate, start }: Deps) { + let settling = new Set(); + async function settle(channelId: string, id: string): Promise { + if (settling.has(id)) return; + settling.add(id); + try { + let value = await get(id); + if (!value?.spike?.settle || value.spike.dismissed || value.documentId !== channelId) return; + if (!await start(channelId, value)) return; + await mutate(id, item => { + delete item.spike!.settle; + }); + } finally { + settling.delete(id); + } + } + return { + settle, + async pending(channelId: string): Promise { + for (let value of await list(channelId)) { + if (value.spike?.settle && !value.spike.dismissed) await settle(channelId, value.id); + } + }, + }; +} diff --git a/apps/server/src/experiments/spikes.test.ts b/apps/server/src/experiments/spikes.test.ts index d8bac2bbf..de843e573 100644 --- a/apps/server/src/experiments/spikes.test.ts +++ b/apps/server/src/experiments/spikes.test.ts @@ -6,7 +6,15 @@ 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 { + MAX_SPIKE_RETRIES, + renderKey, + resolveInstruction, + retryable, + spikeCallout, + spikeReport, + uncertain, +} from "./spikes"; import type { Investigation } from "@chopin/experiment/records"; @@ -63,7 +71,7 @@ test("a submitted spike renders as a titled callout with findings and screenshot expect(renderKey(value)).toBe("completed"); let source = serialize({ type: "root", children: [spikeCallout(value)] }); expect(source).toContain(` { + let waiting = serialize({ type: "root", children: [spikeCallout(spike("interrupted"))] }); + expect(waiting).toContain("Waiting for @maggie’s coding agent to reconnect"); + expect(waiting).toContain(`type="note"`); + expect(waiting).not.toContain("Propose"); + room.validate(waiting); + let spent = spike("interrupted"); + spent.spike!.retries = MAX_SPIKE_RETRIES; + spent.progress = "Workspace connection expired. Propose a new attempt to retry."; + expect(renderKey(spent)).toBe("stopped"); + let stopped = serialize({ type: "root", children: [spikeCallout(spent)] }); + expect(stopped).toContain(`type="warning"`); + expect(stopped).not.toContain("Propose"); + room.validate(stopped); + let dismissed = spike("interrupted"); + dismissed.spike!.dismissed = true; + expect(retryable(dismissed)).toBe(false); + expect(retryable(spike("cancelled"))).toBe(false); }); test("a callout lands directly under its passage and is later replaced in place", async () => { @@ -185,3 +213,29 @@ test("explicit uncertainty is the fallback signal", () => { expect(uncertain("This is an open decision for the team.")).toBe(true); expect(uncertain("The page lists every document in the repository.")).toBe(false); }); + +test("a one-block result does not fold, and a landed result names its passage and callout", () => { + let value = spike( + "completed", + spikeReport({ headline: "It works", findings: ["Yes."], recommendation: "Ship.", images: [] }), + ); + value.result!.report = "**It works**\n\n**Recommendation:** Ship.\n"; + let source = serialize({ type: "root", children: [spikeCallout(value)] }); + expect(source).not.toContain("fold="); + let { text, said } = resolveInstruction(value); + expect(text).toContain("> We are unsure whether drag handles work on touch."); + expect(text).toContain(value.spike!.callout); + expect(text).toContain("Keep the callout"); + expect(said).toBe("Updating the passage the prototype answered"); +}); + +test("over-escaped quotes from an agent's tool call read as plain quotes", () => { + let report = spikeReport({ + headline: 'Use \\"No notes yet\\"', + findings: ["It\\'s clear."], + recommendation: 'Show \\"No notes yet\\" in a card.', + images: [], + }); + expect(report).not.toContain("\\"); + expect(report).toContain('Show "No notes yet" in a card.'); +}); diff --git a/apps/server/src/experiments/spikes.ts b/apps/server/src/experiments/spikes.ts index 876968b3e..f9c65e901 100644 --- a/apps/server/src/experiments/spikes.ts +++ b/apps/server/src/experiments/spikes.ts @@ -46,8 +46,9 @@ export function spikeBrief(passage: string): string { "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.", + + "Then call submit_spike_result with a headline of at most 8 words, at most 3 findings of " + + "one short sentence each, a one- or two-sentence recommendation that states the decision, " + + "and the uploaded image paths.", ].join("\n"); } @@ -55,20 +56,25 @@ function paragraph(...children: PhrasingContent[]): BlockContent { return { type: "paragraph", children }; } +/** Agents often over-escape quotes in tool arguments; prose never needs a backslash before one. */ +function unescaped(value: string): string { + return value.replace(/\\+(["'“”‘’])/g, "$1"); +} + /** Canonical report MDX: bold headline, the recommendation first, findings, 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 })], + children: [paragraph({ type: "text", value: unescaped(finding) })], })); let root: Root = { type: "root", children: [ - paragraph({ type: "strong", children: [{ type: "text", value: input.headline }] }), + paragraph({ type: "strong", children: [{ type: "text", value: unescaped(input.headline) }] }), paragraph( { type: "strong", children: [{ type: "text", value: "Recommendation:" }] }, - { type: "text", value: ` ${input.recommendation}` }, + { type: "text", value: ` ${unescaped(input.recommendation)}` }, ), { type: "list", ordered: false, spread: false, children: items }, ...input.images.map((url, index) => @@ -79,16 +85,35 @@ export function spikeReport(input: SpikeSubmission): string { return serialize(root); } +/** Automatic re-dispatches of a spike whose local agent lost its connection. */ +export const MAX_SPIKE_RETRIES = 2; + +/** + * Only a lost connection interrupts a spike; dismissal cancels it. An interrupted spike waits + * for its owner's coding agent to reconnect, a bounded number of times. + */ +export function retryable(value: Investigation): boolean { + return value.state === "interrupted" && !!value.spike && !value.spike.dismissed + && (value.spike.retries ?? 0) < MAX_SPIKE_RETRIES; +} + /** 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 (retryable(value)) return "waiting"; 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 { +function callout( + id: string, + type: string, + title: string, + children: RootContent[], + fold?: number, +): RootContent { return { type: "mdxJsxFlowElement", name: "Callout", @@ -96,6 +121,7 @@ function callout(id: string, type: string, title: string, children: RootContent[ { type: "mdxJsxAttribute", name: "id", value: id }, { type: "mdxJsxAttribute", name: "type", value: type }, { type: "mdxJsxAttribute", name: "title", value: title.slice(0, 100) }, + ...(fold ? [{ type: "mdxJsxAttribute" as const, name: "fold", value: String(fold) }] : []), ], children: children as BlockContent[], }; @@ -110,13 +136,34 @@ export function spikeCallout(value: Investigation): RootContent { 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); + let body = headline ? rest : blocks; + // Only the recommendation shows; findings and screenshots fold behind a disclosure. + return callout( + spike.callout, + "tip", + headline || "Prototype result", + body, + body.length > 1 ? 1 : undefined, + ); + } + if (key === "waiting") { + return callout(spike.callout, "note", "Prototype paused", [ + paragraph({ + type: "text", + value: `Waiting for @${spike.login}’s coding agent to reconnect. It will pick the ` + + "prototype back up then. Delete this callout to cancel.", + }), + ]); } if (key === "stopped") { return callout(spike.callout, "warning", "Prototype stopped", [ paragraph({ type: "text", - value: value.progress.trim() || "The coding agent stopped before reporting.", + // Interruption copy is written for investigations; a spike says what happened. + value: value.state === "interrupted" + ? `@${spike.login}’s coding agent kept losing its connection, so this prototype ` + + "stopped." + : value.progress.trim() || "The coding agent stopped before reporting.", }), ]); } @@ -137,3 +184,28 @@ export function spikeCallout(value: Investigation): RootContent { }), ]); } + +/** + * The Planner turn that settles a spike's passage once its result lands, so the document stops + * asking a question its own callout answers. The callout stays as the evidence. + */ +export function resolveInstruction(value: Investigation): { text: string; said: string } { + let spike = value.spike; + if (!spike) throw new Error("not a spike"); + let quoted = spike.passage.trim().split("\n").map(line => `> ${line}`).join("\n"); + return { + text: [ + "A quick prototype answered the open question in this passage:", + "", + quoted, + "", + `Its result is the tip Callout with id ${spike.callout} directly below the passage. ` + + "Call read_plan, then rewrite only that passage so it states the decision the " + + "callout's recommendation supports, as settled prose of about the same length, with no " + + "remaining doubt about the question the prototype answered. Keep the callout where it " + + "is and do not edit it. If someone has already settled the passage, or it now says " + + "something the result does not answer, change nothing. Do not change anything else.", + ].join("\n"), + said: "Updating the passage the prototype answered", + }; +} diff --git a/apps/server/src/main.ts b/apps/server/src/main.ts index 0411c8789..97d5f5d8e 100644 --- a/apps/server/src/main.ts +++ b/apps/server/src/main.ts @@ -18,8 +18,10 @@ import { registerAuthRoutes } from "./auth/routes"; import { registerExperimentRoutes } from "./experiments/routes"; import { registerImageRoutes } from "./images/routes"; import type { ExperimentRuntime } from "./experiments/routes"; +import type { Instruction } from "./chat/server-instruction"; import { fingerprint } from "./experiments/service"; import * as Chat from "./chat/service"; +import { serverInstruction as runServerInstruction } from "./chat/server-instruction"; import { CHAT_CAPABILITIES, incomingFrame, sidebarFrame } from "./chat/incoming"; import { ReferenceService } from "./chat/references"; import { createConversationRuntime } from "./conversation-plan/runtime"; @@ -52,6 +54,8 @@ 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 { spikeSettler } from "./experiments/spike-settle"; +import { resolveInstruction } from "./experiments/spikes"; import { askJev } from "./conversation-plan/jev"; import { locksEditing, requestBuild } from "./tasks/builds"; import { pendingLinks, relinkInstruction } from "./tasks/relink"; @@ -215,42 +219,75 @@ function documentBackend(): Service.Backend { * Never claims ownership: without an owner the cards stay pending until the next Planner turn. */ async function relinkDecisions(channelId: string): Promise { - if (!config.agent || deletingChannels.has(channelId)) return; - let opened = Rooms.get(channelId)?.plan; - if (!opened || pendingLinks(opened).length === 0) return; - let binding = await ownerBindings?.resolve(channelId); - if (!binding) return; - let repository = binding.repository; - binding.release(); - await withDocumentTransition(channelId, async () => { - await Rooms.get(channelId)?.closing; - let held = Rooms.hold(channelId); - let release = () => { - held.release(); - evict(held.room); + await serverInstruction(channelId, current => { + let pending = pendingLinks(current); + if (pending.length === 0) return; + return { + handle: "Chopin", + text: relinkInstruction(pending), + said: "Linking decisions to the document after the build", }; - let running = false; - try { - let current = await plan(held.room, server); - let pending = pendingLinks(current); - if (pending.length === 0) return; - let context = conversation(held.room, undefined, repository); - await Chat.instruct( - context, - "Chopin", - relinkInstruction(pending), - "Linking decisions to the document after the build", - ); - if (context.chat.running) { - running = true; - void context.chat.running.finally(release).catch(() => {}); - } - } finally { - if (!running) release(); - } }); } +/** + * A spike's result landed: have the Planner settle the passage it answered. Runs under the + * existing Planner owner, or claims one for the person whose edit started the spike while they + * have the document open. A build that has since locked the document skips it. Until Chat + * accepts the turn, `spike.settle` stays set and the room opening or an owner arriving retries. + */ +const spikeSettlement = spikeSettler({ + get: id => experiments!.service.store.get(id), + list: id => experiments!.service.store.list(id), + mutate: (id, action) => experiments!.service.mutate(id, action), + start: (channelId, value) => + serverInstruction(channelId, current => { + if (Service.implementationActive(current)) return; + let { text, said } = resolveInstruction(value); + return { handle: value.spike!.login, text, said }; + }, value.requester), +}); + +function settlePendingSpikes(channelId: string): void { + if (!config.agent || !config.liveSpikes || !experiments) return; + void spikeSettlement.pending(channelId).catch(err => + console.error("chopin: could not settle a prototyped passage -", err) + ); +} + +/** Wake owner-bound work: background jobs and spike settlements that waited for an owner. */ +function ownerAvailable(channelId: string): Promise { + settlePendingSpikes(channelId); + return jobRunner?.ownerAvailable(channelId) ?? Promise.resolve(); +} + +/** + * Start a Planner turn nobody typed; true once Chat accepted it. `claimant` is a user whose + * open, writable socket in the room may claim an unowned Planner; without one, only an + * existing owner will do. + */ +function serverInstruction( + channelId: string, + compose: (plan: Service.Plan) => Instruction | undefined, + claimant?: string, +): Promise { + return runServerInstruction( + { + agent: !!config.agent, + unavailable: id => deletingChannels.has(id), + owner: async id => await ownerBindings?.resolve(id), + transition: withDocumentTransition, + open: room => plan(room, server), + conversation, + instruct: (context, { handle, text, said }) => Chat.instruct(context, handle, text, said), + evict, + }, + channelId, + compose, + claimant, + ); +} + function placeResearchReference( channelId: string, workspaceId: string, @@ -332,6 +369,7 @@ async function plan(room: Rooms.Room, server: Server): Promise Service.persist(opened, chat), activeOwner: () => ownerBindings!.resolve(room.id), - ownerAvailable: () => jobRunner?.ownerAvailable(room.id) ?? Promise.resolve(), + ownerAvailable: () => ownerAvailable(room.id), jobs: config.backgroundJobs ? jobService : undefined, references: referenceService, investigations: experiments @@ -465,7 +503,7 @@ function conversation( originMessageId: request.entryId, requestedBy: request.userId, requestedByHandle: request.handle, - beforeStart: () => jobRunner?.ownerAvailable(room.id) ?? Promise.resolve(), + beforeStart: () => ownerAvailable(room.id), placeReference: id => placeResearchReference(room.id, id), }); } catch (err) { @@ -736,7 +774,7 @@ async function receive(ws: Socket, raw: string): Promise { auth: hostedAuth, refreshAccess, unavailable: id => archivingChannels.has(id) || deletingChannels.has(id), - ownerAvailable: id => jobRunner?.ownerAvailable(id) ?? Promise.resolve(), + ownerAvailable, placeReference: placeResearchReference, scheduleRecovery: scheduleResearchRecovery, }), @@ -1712,7 +1750,10 @@ experiments = registerExperimentRoutes(router, hostedAuth, { return heldLease; }, context: currentDocumentTarget, - connected: (repositoryId, owner) => liveSync?.connected(repositoryId, owner), + connected: (repositoryId, owner) => { + liveSync?.connected(repositoryId, owner); + spikeScout?.connected(repositoryId, owner); + }, place: (id, experiment, view, decision, remove) => withDocumentTransition(id, async () => { await Rooms.get(id)?.closing; @@ -1988,6 +2029,11 @@ if (config.liveSpikes) { service: experiments.service, connections: experiments.connections, withPlan: withImplementationPlan, + landed: (channelId, value) => { + void spikeSettlement.settle(channelId, value.id).catch(err => + console.error("chopin: could not settle a prototyped passage -", err) + ); + }, }), judge: process.env.JEV_API_KEY || process.env.TYPESAFE_API_KEY ? jevJudge(request => askJev(request, { model: jev, timeoutMs })) @@ -2019,7 +2065,7 @@ registerResearchWorkspaceRoutes(router, hostedAuth, { channel.id, session.session.id, ); - await jobRunner?.ownerAvailable(channel.id); + await ownerAvailable(channel.id); }, }); registerNavigationRoutes(router, hostedAuth, { storage }); diff --git a/apps/web/src/build-view.css b/apps/web/src/build-view.css index 9e878e31b..964df515e 100644 --- a/apps/web/src/build-view.css +++ b/apps/web/src/build-view.css @@ -198,7 +198,7 @@ .build-task-head { display: flex; - align-items: center; + align-items: flex-start; gap: calc(var(--spacing) * 2); } @@ -207,9 +207,10 @@ min-width: 0; min-height: calc(var(--spacing) * 10); flex: 1; - align-items: center; + align-items: flex-start; gap: calc(var(--spacing) * 3); - padding: 0; + /* One line sits centred in the 40px row; a wrapped title grows it from the top. */ + padding: calc((var(--spacing) * 10 - 1lh) / 2) 0; border: 0; background: none; color: var(--color-text-primary); @@ -220,10 +221,19 @@ } .build-task-title { + display: -webkit-box; min-width: 0; overflow: hidden; - text-overflow: ellipsis; - white-space: nowrap; + -webkit-box-orient: vertical; + -webkit-line-clamp: 2; + line-clamp: 2; + overflow-wrap: anywhere; + text-wrap: pretty; +} + +/* A title that wraps keeps its dot and pull request on its first line. */ +.build-task-toggle > .build-task-dot { + margin-block-start: calc((1lh - var(--spacing) * 3) / 2); } /* Hollow queued, petrol partial ring in progress, success done, warning blocked. */ @@ -259,6 +269,9 @@ --build-pr-surface: var(--color-success-surface); --build-pr-icon: var(--color-success-icon); flex: none; + /* A fixed 22px capsule, centred on the 40px first line. */ + margin-block-start: calc(var(--spacing) * 2.25); + line-height: calc(var(--spacing) * 4); padding: calc(var(--spacing) / 2) calc(var(--spacing) * 2); border: var(--edge-width) solid color-mix(in srgb, var(--build-pr-icon) 12%, transparent); border-radius: var(--radius-full); diff --git a/packages/dialect/src/dialect.ts b/packages/dialect/src/dialect.ts index e9a7ea9d1..c7029c738 100644 --- a/packages/dialect/src/dialect.ts +++ b/packages/dialect/src/dialect.ts @@ -282,6 +282,8 @@ export const COMPONENTS: Readonly> = Object.freeze({ attributes: { type: { type: "enum", required: true, values: CALLOUT_TYPES }, title: { type: "text", required: false, max: limits.MAX_CALLOUT_TITLE }, + /** How many leading blocks stay visible; the rest fold behind a disclosure. */ + fold: { type: "integer", required: false, min: 1, max: limits.MAX_CALLOUT_FOLD }, }, }), diff --git a/packages/dialect/src/index.ts b/packages/dialect/src/index.ts index 581eb6064..de67936d7 100644 --- a/packages/dialect/src/index.ts +++ b/packages/dialect/src/index.ts @@ -44,6 +44,7 @@ export type { ConvertOptions } from "./convert"; export { plugins, registry, tablePlugin } from "./registry"; export type { Registry } from "./registry"; +export { calloutFoldState } from "./nodes/callout-fold"; export { $createColumnNode, $createColumnsNode, diff --git a/packages/dialect/src/limits.ts b/packages/dialect/src/limits.ts index 87f3e5eb1..2451f7c62 100644 --- a/packages/dialect/src/limits.ts +++ b/packages/dialect/src/limits.ts @@ -31,6 +31,8 @@ export const MAX_IMAGE_WIDTH = 4096; export const MAX_TAB_LABEL = 60; export const MAX_CALLOUT_TITLE = 100; +/** A folded callout keeps at most this many leading blocks visible. */ +export const MAX_CALLOUT_FOLD = 3; /** Questionnaire shape, matching the `ask` tool's contract. */ export const MAX_QUESTIONS = 10; diff --git a/packages/dialect/src/nodes/callout-fold.ts b/packages/dialect/src/nodes/callout-fold.ts new file mode 100644 index 000000000..21b50c9ea --- /dev/null +++ b/packages/dialect/src/nodes/callout-fold.ts @@ -0,0 +1,17 @@ +/** + * How many leading blocks a callout keeps visible before the rest fold away. + * + * Node state rather than a callout property, so it collaborates and serializes + * like the callout's title without changing the node class. 0 never folds. + */ + +import { createState } from "lexical"; + +import { MAX_CALLOUT_FOLD } from "../limits"; + +export const calloutFoldState = createState("plan-fold", { + parse: (value: unknown): number => + Number.isInteger(value) && (value as number) >= 0 && (value as number) <= MAX_CALLOUT_FOLD + ? value as number + : 0, +}); diff --git a/packages/dialect/src/nodes/container-visitors.ts b/packages/dialect/src/nodes/container-visitors.ts index b28ff4006..3d4216c31 100644 --- a/packages/dialect/src/nodes/container-visitors.ts +++ b/packages/dialect/src/nodes/container-visitors.ts @@ -7,6 +7,8 @@ * working. */ +import { $getState, $setState } from "lexical"; + import { $createCalloutNode, $createTabNode, @@ -17,6 +19,7 @@ import { } from "./containers"; import { $createColumnNode, $createColumnsNode, $isColumnNode, $isColumnsNode } from "./columns"; import { attribute, identity, isFlow, PRIORITY } from "./shared"; +import { calloutFoldState } from "./callout-fold"; import type { LexicalExportVisitor, MdastImportVisitor } from "@mdxeditor/editor"; import type { LexicalNode } from "lexical"; @@ -63,12 +66,15 @@ export const MdastTabVisitor = importer( node => $createTabNode(attribute(node, "id") ?? "", attribute(node, "label") ?? ""), ); -export const MdastCalloutVisitor = importer("Callout", node => - $createCalloutNode( +export const MdastCalloutVisitor = importer("Callout", node => { + let callout = $createCalloutNode( attribute(node, "id") ?? "", (attribute(node, "type") ?? "note") as CalloutType, attribute(node, "title") ?? "", - )); + ); + let fold = calloutFoldState.parse(Number(attribute(node, "fold") ?? 0)); + return fold ? $setState(callout, calloutFoldState, fold) : callout; +}); export const MdastColumnsVisitor = importer( "Columns", @@ -95,7 +101,14 @@ export const LexicalTabVisitor: Exporter = exporter( export const LexicalCalloutVisitor: Exporter = exporter( "Callout", $isCalloutNode, - node => identity(node.getId(), { type: node.getCalloutType(), title: node.getTitle() }), + node => + identity(node.getId(), { + type: node.getCalloutType(), + title: node.getTitle(), + fold: $getState(node, calloutFoldState) + ? String($getState(node, calloutFoldState)) + : undefined, + }), ); export const LexicalColumnsVisitor: Exporter = exporter( diff --git a/packages/dialect/src/nodes/containers.test.ts b/packages/dialect/src/nodes/containers.test.ts index 4fdd2e827..30bc6d670 100644 --- a/packages/dialect/src/nodes/containers.test.ts +++ b/packages/dialect/src/nodes/containers.test.ts @@ -135,6 +135,15 @@ describe("structural components", () => { expect(canonical(deep)).toContain("| a | b |"); }); + it("round-trips a callout's fold and rejects one out of range", () => { + let folded = + `\n\nLead.\n\n- detail\n\n\n`; + expect(canonical(folded)).toContain(``); + expect(validate(parse(folded))).toEqual({ ok: true }); + expect(validate(parse(folded.replace(`fold="1"`, `fold="4"`))).ok).toBe(false); + expect(validate(parse(folded.replace(`fold="1"`, `fold="0"`))).ok).toBe(false); + }); + it("omits optional attributes that are unset", () => { let out = canonical(`\n\nx\n\n\n`); expect(out).not.toContain("title="); diff --git a/packages/editor/src/callout.css b/packages/editor/src/callout.css index 73515bea7..2d127e9a6 100644 --- a/packages/editor/src/callout.css +++ b/packages/editor/src/callout.css @@ -160,6 +160,33 @@ margin-block-end: 0; } +/* A folded callout keeps its leading blocks; each reader opens the rest for themselves. */ +.plan-content [data-fold-state="folded"][data-fold-lead="1"] > [data-plan-body] > :nth-child(n+2), +.plan-content [data-fold-state="folded"][data-fold-lead="2"] > [data-plan-body] > :nth-child(n+3), +.plan-content [data-fold-state="folded"][data-fold-lead="3"] > [data-plan-body] > :nth-child(n+4) { + display: none; +} + +.plan-content [data-fold-state="folded"] > [data-plan-body] > * { + margin-block-end: 0; +} + +.plan-content [data-fold-state="open"][data-fold-lead="1"] > [data-plan-body] > :nth-child(n+2), +.plan-content [data-fold-state="open"][data-fold-lead="2"] > [data-plan-body] > :nth-child(n+3), +.plan-content [data-fold-state="open"][data-fold-lead="3"] > [data-plan-body] > :nth-child(n+4) { + animation: plan-callout-reveal var(--duration-base) var(--ease-out); +} + +.plan-callout-fold { + padding-inline-start: 2.125rem; + margin-block-start: 0.25rem; +} + +.plan-callout-fold > .plan-research-more { + padding-block: 0.25rem; + margin-block: -0.25rem; +} + .plan-callout-menu { z-index: 20; width: 10rem; @@ -218,8 +245,16 @@ } } +@keyframes plan-callout-reveal { + from { + opacity: 0; + transform: translateY(-0.25rem); + } +} + @media (prefers-reduced-motion: reduce) { - .plan-callout-menu { + .plan-callout-menu, + .plan-content [data-fold-state="open"] > [data-plan-body] > * { animation: none; } } diff --git a/packages/editor/src/widgets/callout-fold.test.ts b/packages/editor/src/widgets/callout-fold.test.ts new file mode 100644 index 000000000..cc1448812 --- /dev/null +++ b/packages/editor/src/widgets/callout-fold.test.ts @@ -0,0 +1,15 @@ +import { expect, test } from "bun:test"; + +import { foldOf } from "./callout-fold"; + +test("a callout folds only when it has more blocks than it keeps visible", () => { + expect(foldOf({ fold: 0, blocks: 4, opened: false })).toBe("none"); + expect(foldOf({ fold: 1, blocks: 1, opened: false })).toBe("none"); + expect(foldOf({ fold: 1, blocks: 3, opened: false })).toBe("folded"); + expect(foldOf({ fold: 2, blocks: 3, opened: true })).toBe("open"); +}); + +test("a caret in a hidden block opens the fold, one in a leading block does not", () => { + expect(foldOf({ fold: 1, blocks: 3, opened: false, caret: 0 })).toBe("folded"); + expect(foldOf({ fold: 1, blocks: 3, opened: false, caret: 2 })).toBe("open"); +}); diff --git a/packages/editor/src/widgets/callout-fold.ts b/packages/editor/src/widgets/callout-fold.ts new file mode 100644 index 000000000..a426e57f0 --- /dev/null +++ b/packages/editor/src/widgets/callout-fold.ts @@ -0,0 +1,25 @@ +/** + * Whether a callout shows only its leading blocks. + * + * Folding is a view preference: the document's `fold` says how many blocks lead, + * and each reader opens the rest for themselves. A caret inside a hidden block + * always opens it, so nobody types into something they cannot see. + */ +export type Fold = "none" | "folded" | "open"; + +export function foldOf( + { fold, blocks, opened, caret }: { + /** Leading blocks the document keeps visible; 0 never folds. */ + fold: number; + /** Blocks in the callout's body. */ + blocks: number; + /** The reader opened it. */ + opened: boolean; + /** Index of the body block holding the caret, if it is in this callout. */ + caret?: number; + }, +): Fold { + if (fold <= 0 || blocks <= fold) return "none"; + if (opened || (caret !== undefined && caret >= fold)) return "open"; + return "folded"; +} diff --git a/packages/editor/src/widgets/callout.tsx b/packages/editor/src/widgets/callout.tsx index aa6539323..d8c84d927 100644 --- a/packages/editor/src/widgets/callout.tsx +++ b/packages/editor/src/widgets/callout.tsx @@ -21,16 +21,33 @@ import { useLexicalComposerContext } from "@lexical/react/LexicalComposerContext import { readOnly$ } from "@mdxeditor/editor"; import { useCellValue } from "@mdxeditor/gurx"; import * as Select from "@radix-ui/react-select"; -import { $getNodeByKey, $getRoot, $isElementNode } from "lexical"; -import { $isCalloutNode, CALLOUT_TYPES, limits } from "@chopin/dialect"; - +import { + $getNodeByKey, + $getRoot, + $getSelection, + $getState, + $isElementNode, + $isRangeSelection, + setDOMUnmanaged, +} from "lexical"; +import { $isCalloutNode, CALLOUT_TYPES, calloutFoldState, limits } from "@chopin/dialect"; + +import { foldOf } from "./callout-fold"; import { registerCalloutNormalization } from "./callout-shape"; import type { CalloutType } from "@chopin/dialect"; import type { IconSize } from "@chopin/icons"; import type { ElementNode, LexicalEditor } from "lexical"; -type Callout = { key: string; type: CalloutType; title: string }; +type Callout = { + key: string; + type: CalloutType; + title: string; + fold: number; + blocks: number; + /** Index of the body block holding the caret, when it is inside this callout. */ + caret?: number; +}; const LABELS: Record = { note: "Note", @@ -60,13 +77,25 @@ function collect(editor: LexicalEditor): Callout[] { let out: Callout[] = []; editor.getEditorState().read(() => { + let selection = $getSelection(); + let anchor = $isRangeSelection(selection) ? selection.anchor.getNode() : undefined; let walk = (node: ElementNode) => { for (let child of node.getChildren()) { if ($isCalloutNode(child)) { + let fold = $getState(child, calloutFoldState); + let caret: number | undefined; + if (fold && anchor) { + let block = anchor.getParents().find(parent => parent.getParent()?.is(child)) + ?? (anchor.getParent()?.is(child) ? anchor : undefined); + if (block) caret = block.getIndexWithinParent(); + } out.push({ key: child.getKey(), type: child.getCalloutType(), title: child.getTitle(), + fold, + blocks: child.getChildrenSize(), + ...(caret === undefined ? {} : { caret }), }); } if ($isElementNode(child)) walk(child); @@ -309,10 +338,68 @@ function Heading( ); } +/** + * Where a folded callout's disclosure renders: after the body, outside the slot Lexical + * reconciles, and marked unmanaged so its mutation observer leaves it in place. + */ +function disclosureHost(editor: LexicalEditor, key: string): HTMLElement | null { + let element = editor.getElementByKey(key); + if (!element) return null; + let host = element.querySelector(":scope > [data-plan-chrome='callout-fold']"); + if (host) return host; + host = document.createElement("div"); + host.dataset.planChrome = "callout-fold"; + setDOMUnmanaged(host); + element.append(host); + return host; +} + +/** Opens or closes the blocks a callout folds; a reader's own view, never the document. */ +function FoldToggle({ open, onToggle }: { open: boolean; onToggle: () => void }) { + return ( +
+ +
+ ); +} + export function CalloutPlugin() { let [editor] = useLexicalComposerContext(); let disabled = useCellValue(readOnly$); let [callouts, setCallouts] = useState([]); + let [opened, setOpened] = useState>(() => new Set()); + let folds = new Map(callouts.map(callout => [ + callout.key, + foldOf({ ...callout, opened: opened.has(callout.key) }), + ])); + + // Mirrored onto the callout's own element; Lexical leaves attributes it does not own alone. + useLayoutEffect(() => { + for (let callout of callouts) { + let element = editor.getElementByKey(callout.key); + if (!element) continue; + let fold = folds.get(callout.key) ?? "none"; + let lead = fold === "none" ? undefined : String(callout.fold); + if (element.dataset.foldState !== (fold === "none" ? undefined : fold)) { + if (fold === "none") delete element.dataset.foldState; + else element.dataset.foldState = fold; + } + if (element.dataset.foldLead !== lead) { + if (lead) element.dataset.foldLead = lead; + else delete element.dataset.foldLead; + } + } + }); + + let toggle = (key: string) => + setOpened(previous => { + let next = new Set(previous); + if (next.has(key)) next.delete(key); + else next.add(key); + return next; + }); useEffect(() => registerCalloutNormalization(editor), [editor]); @@ -328,11 +415,20 @@ export function CalloutPlugin() { let host = editor.getElementByKey(callout.key) ?.querySelector("[data-plan-chrome='callout']"); if (!host) return null; - return createPortal( - , - host, - callout.key, - ); + let fold = folds.get(callout.key); + let foldHost = fold === "none" ? null : disclosureHost(editor, callout.key); + return [ + createPortal( + , + host, + callout.key, + ), + foldHost && createPortal( + toggle(callout.key)} />, + foldHost, + `${callout.key}-fold`, + ), + ]; })} ); diff --git a/packages/experiment/src/records.ts b/packages/experiment/src/records.ts index 3daf6762e..c661a31df 100644 --- a/packages/experiment/src/records.ts +++ b/packages/experiment/src/records.ts @@ -47,6 +47,13 @@ export const spikeSchema = z.object({ /** Digest of the callout block as last rendered; a different block holds human edits. */ calloutDigest: z.string().max(100).optional(), dismissed: z.boolean().optional(), + /** + * Set with `rendered = completed` and cleared once Chat accepts the Planner turn that + * settles the passage, so a landing nobody could act on yet is retried. + */ + settle: z.boolean().optional(), + /** Automatic re-dispatches after its local agent's connection was lost; capped. */ + retries: z.number().int().nonnegative().max(10).optional(), }).strict(); export type Spike = z.infer;