From fe7a645efbb5cf88e38e80ff704d2d2953428112 Mon Sep 17 00:00:00 2001 From: Maggie Appleton <5599295+MaggieAppleton@users.noreply.github.com> Date: Sun, 11 Oct 2026 00:59:33 +0100 Subject: [PATCH 1/3] Finish the living document flow's rough edges - Spike results fold: a result callout shows its headline and recommendation; findings and screenshots open behind "Show more". Callout gains an optional `fold` attribute (leading blocks kept visible), held as node state and folded per reader in the editor. The spike prompt asks for at most three one-sentence findings, and over-escaped quotes are cleaned up. - When a result lands, the Planner rewrites the passage it answered as a decision and keeps the callout as evidence, under the existing Planner owner or the spike requester's open session. - A local agent that reconnects rescans its owner's documents whose last scout found no agent, so missed passages spike without another edit. - Build view task titles wrap to two lines, with the dot and pull request pinned to the first line. Co-Authored-By: Claude Opus 5.5 --- ...SD0MSTZ12Y-landed-fires-once-when-a-com.md | 8 ++ ...D4JWW04C6P-fold-is-nodestate-outside-co.md | 8 ++ .../server/src/experiments/spike-host.test.ts | 38 +++++- apps/server/src/experiments/spike-host.ts | 8 +- .../src/experiments/spike-scout.test.ts | 39 ++++++ apps/server/src/experiments/spike-scout.ts | 24 +++- apps/server/src/experiments/spikes.test.ts | 30 ++++- apps/server/src/experiments/spikes.ts | 60 +++++++-- apps/server/src/main.ts | 85 ++++++++++--- apps/web/src/build-view.css | 23 +++- packages/dialect/src/dialect.ts | 2 + packages/dialect/src/index.ts | 1 + packages/dialect/src/limits.ts | 2 + packages/dialect/src/nodes/callout-fold.ts | 17 +++ .../dialect/src/nodes/container-visitors.ts | 21 +++- packages/dialect/src/nodes/containers.test.ts | 9 ++ packages/editor/src/callout.css | 37 +++++- .../editor/src/widgets/callout-fold.test.ts | 15 +++ packages/editor/src/widgets/callout-fold.ts | 25 ++++ packages/editor/src/widgets/callout.tsx | 114 ++++++++++++++++-- 20 files changed, 518 insertions(+), 48 deletions(-) create mode 100644 .lane/memory/apps/server/src/experiments/spike-host.ts/01M4M43PSSVR2QCZSD0MSTZ12Y-landed-fires-once-when-a-com.md create mode 100644 .lane/memory/packages/dialect/src/nodes/callout-fold.ts/01M4M43PSAZGBJBTD4JWW04C6P-fold-is-nodestate-outside-co.md create mode 100644 packages/dialect/src/nodes/callout-fold.ts create mode 100644 packages/editor/src/widgets/callout-fold.test.ts create mode 100644 packages/editor/src/widgets/callout-fold.ts 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/experiments/spike-host.test.ts b/apps/server/src/experiments/spike-host.test.ts index 139cf1665..5320db25f 100644 --- a/apps/server/src/experiments/spike-host.test.ts +++ b/apps/server/src/experiments/spike-host.test.ts @@ -22,12 +22,14 @@ 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 host = spikeHost({ service, connections: new Connections(), withPlan: (_id, action) => action(context.plan), + landed: (_channelId, value) => landed.push(value.id), }); - return { ...context, service, host }; + return { ...context, service, host, landed }; } test("a spike is durably placing before its callout is published", async () => { @@ -290,3 +292,37 @@ 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]); + 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..34e16650a 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. */ + 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) => { @@ -135,6 +138,9 @@ export function spikeHost({ service, connections, withPlan }: Options): SpikeHos item.spike!.callout = callout; item.spike!.calloutDigest = digest; }); + if (placed.status === "placed" && key === "completed" && spike.rendered !== key) { + landed?.(channelId, value); + } } } }, diff --git a/apps/server/src/experiments/spike-scout.test.ts b/apps/server/src/experiments/spike-scout.test.ts index b6e7bf93f..3142f1f9d 100644 --- a/apps/server/src/experiments/spike-scout.test.ts +++ b/apps/server/src/experiments/spike-scout.test.ts @@ -70,6 +70,7 @@ function harness(snapshot: Partial = {}, verdict?: SpikeJudge) { timers, delays, disconnect: () => connected = false, + reconnect: () => connected = true, }; } @@ -281,3 +282,41 @@ 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); +}); diff --git a/apps/server/src/experiments/spike-scout.ts b/apps/server/src/experiments/spike-scout.ts index 1c4592078..f587c70a6 100644 --- a/apps/server/src/experiments/spike-scout.ts +++ b/apps/server/src/experiments/spike-scout.ts @@ -115,6 +115,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, by repository, to rescan on reconnect. */ + #offline = new Map(); #closed = false; constructor(options: SpikeScoutOptions) { @@ -167,6 +169,19 @@ export class SpikeScout { return spikes; } + /** + * 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, repository] of this.#offline) { + if (repository !== repositoryId || this.#editors.get(channelId) !== 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. */ refresh(channelId: string): Promise { return this.#serial(channelId, async () => { @@ -227,9 +242,13 @@ 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.#offline.set(channelId, snapshot.repositoryId); + return; + } + this.#offline.delete(channelId); let verdicts = await (this.#options.judge ?? heuristicJudge)(candidates); if (!verdicts) return; // Approved passages beyond capacity stay unseen so a later scan can start them. @@ -265,5 +284,6 @@ export class SpikeScout { this.#seen.clear(); this.#editors.clear(); this.#waiting.clear(); + this.#offline.clear(); } } diff --git a/apps/server/src/experiments/spikes.test.ts b/apps/server/src/experiments/spikes.test.ts index d8bac2bbf..7d024c78b 100644 --- a/apps/server/src/experiments/spikes.test.ts +++ b/apps/server/src/experiments/spikes.test.ts @@ -6,7 +6,7 @@ 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 { renderKey, resolveInstruction, spikeCallout, spikeReport, uncertain } from "./spikes"; import type { Investigation } from "@chopin/experiment/records"; @@ -63,7 +63,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(` { 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..a18d9f3ac 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) => @@ -88,7 +94,13 @@ export function renderKey(value: Investigation): string { 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 +108,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,7 +123,15 @@ 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 === "stopped") { return callout(spike.callout, "warning", "Prototype stopped", [ @@ -137,3 +158,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..7a60e7937 100644 --- a/apps/server/src/main.ts +++ b/apps/server/src/main.ts @@ -18,6 +18,7 @@ import { registerAuthRoutes } from "./auth/routes"; import { registerExperimentRoutes } from "./experiments/routes"; import { registerImageRoutes } from "./images/routes"; import type { ExperimentRuntime } from "./experiments/routes"; +import type { Investigation } from "@chopin/experiment/records"; import { fingerprint } from "./experiments/service"; import * as Chat from "./chat/service"; import { CHAT_CAPABILITIES, incomingFrame, sidebarFrame } from "./chat/incoming"; @@ -52,6 +53,7 @@ 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 { resolveInstruction } from "./experiments/spikes"; import { askJev } from "./conversation-plan/jev"; import { locksEditing, requestBuild } from "./tasks/builds"; import { pendingLinks, relinkInstruction } from "./tasks/relink"; @@ -214,14 +216,62 @@ function documentBackend(): Service.Backend { * Re-attempt decision links a first build's lock refused, under the existing Planner owner. * Never claims ownership: without an owner the cards stay pending until the next Planner turn. */ -async function relinkDecisions(channelId: string): Promise { +function relinkDecisions(channelId: string): Promise { + return 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", + }; + }); +} + +/** + * 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. + */ +function settleSpikePassage(channelId: string, value: Investigation): Promise { + return serverInstruction(channelId, current => { + if (Service.implementationActive(current)) return; + let { text, said } = resolveInstruction(value); + return { handle: value.spike!.login, text, said }; + }, value.requester); +} + +/** + * Start a Planner turn nobody typed. `claimant` is a user whose open, writable socket in the + * room may claim an unowned Planner; without one, only an existing owner will do. + */ +async function serverInstruction( + channelId: string, + compose: (plan: Service.Plan) => { handle: string; text: string; said: string } | undefined, + claimant?: 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(); + if (!opened || !compose(opened)) return; + 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 ownerBindings?.resolve(channelId); + if (!binding) return; + repository = binding.repository; + binding.release(); + } await withDocumentTransition(channelId, async () => { await Rooms.get(channelId)?.closing; let held = Rooms.hold(channelId); @@ -232,15 +282,10 @@ async function relinkDecisions(channelId: string): Promise { 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", - ); + let instruction = compose(current); + if (!instruction) return; + let context = conversation(held.room, socket?.data.sessionId, repository); + await Chat.instruct(context, instruction.handle, instruction.text, instruction.said); if (context.chat.running) { running = true; void context.chat.running.finally(release).catch(() => {}); @@ -1712,7 +1757,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 +2036,11 @@ if (config.liveSpikes) { service: experiments.service, connections: experiments.connections, withPlan: withImplementationPlan, + landed: (channelId, value) => { + void settleSpikePassage(channelId, value).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 })) 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`, + ), + ]; })} ); From d1a8a552654a2a9ae6351c760c621f446768859e Mon Sep 17 00:00:00 2001 From: Maggie Appleton <5599295+MaggieAppleton@users.noreply.github.com> Date: Sun, 11 Oct 2026 01:24:27 +0100 Subject: [PATCH 2/3] Keep a landed spike's settle pending until the Planner accepts it A completed spike persisted rendered=completed before its settle turn ran, so a room evicted after placement skipped the settle forever. The spike record now carries a settle flag set in the same commit and cleared only after Chat accepts the turn. Without an open room, an existing Planner owner reopens it through the normal hold path; pending settles retry when the room opens or an owner becomes available. Unowned channels are still claimed only for the requester's open writable socket. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/chat/server-instruction.ts | 86 +++++++++++ .../server/src/experiments/spike-host.test.ts | 1 + apps/server/src/experiments/spike-host.ts | 11 +- .../src/experiments/spike-settle.test.ts | 137 ++++++++++++++++++ apps/server/src/experiments/spike-settle.ts | 40 +++++ apps/server/src/main.ts | 121 ++++++++-------- packages/experiment/src/records.ts | 5 + 7 files changed, 332 insertions(+), 69 deletions(-) create mode 100644 apps/server/src/chat/server-instruction.ts create mode 100644 apps/server/src/experiments/spike-settle.test.ts create mode 100644 apps/server/src/experiments/spike-settle.ts 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/spike-host.test.ts b/apps/server/src/experiments/spike-host.test.ts index 5320db25f..0365466f4 100644 --- a/apps/server/src/experiments/spike-host.test.ts +++ b/apps/server/src/experiments/spike-host.test.ts @@ -318,6 +318,7 @@ test("a landed result asks once for its passage to be settled", async () => { 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); diff --git a/apps/server/src/experiments/spike-host.ts b/apps/server/src/experiments/spike-host.ts index 34e16650a..1de0ac2bb 100644 --- a/apps/server/src/experiments/spike-host.ts +++ b/apps/server/src/experiments/spike-host.ts @@ -17,7 +17,7 @@ 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. */ + /** A spike's result was just written into its callout; `spike.settle` is persisted. */ landed?: (channelId: string, value: Investigation) => void; }; @@ -131,16 +131,17 @@ export function spikeHost({ service, connections, withPlan, landed }: Options): 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 (placed.status === "placed" && key === "completed" && spike.rendered !== key) { - landed?.(channelId, value); - } + if (landing) landed?.(channelId, updated); } } }, 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/main.ts b/apps/server/src/main.ts index 7a60e7937..97d5f5d8e 100644 --- a/apps/server/src/main.ts +++ b/apps/server/src/main.ts @@ -18,9 +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 { Investigation } from "@chopin/experiment/records"; +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"; @@ -53,6 +54,7 @@ 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"; @@ -216,8 +218,8 @@ function documentBackend(): Service.Backend { * Re-attempt decision links a first build's lock refused, under the existing Planner owner. * Never claims ownership: without an owner the cards stay pending until the next Planner turn. */ -function relinkDecisions(channelId: string): Promise { - return serverInstruction(channelId, current => { +async function relinkDecisions(channelId: string): Promise { + await serverInstruction(channelId, current => { let pending = pendingLinks(current); if (pending.length === 0) return; return { @@ -231,69 +233,59 @@ function relinkDecisions(channelId: string): Promise { /** * 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. + * 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. */ -function settleSpikePassage(channelId: string, value: Investigation): Promise { - return serverInstruction(channelId, current => { - if (Service.implementationActive(current)) return; - let { text, said } = resolveInstruction(value); - return { handle: value.spike!.login, text, said }; - }, value.requester); +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. `claimant` is a user whose open, writable socket in the - * room may claim an unowned Planner; without one, only an existing owner will do. + * 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. */ -async function serverInstruction( +function serverInstruction( channelId: string, - compose: (plan: Service.Plan) => { handle: string; text: string; said: string } | undefined, + compose: (plan: Service.Plan) => Instruction | undefined, claimant?: string, -): Promise { - if (!config.agent || deletingChannels.has(channelId)) return; - let opened = Rooms.get(channelId)?.plan; - if (!opened || !compose(opened)) return; - 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 ownerBindings?.resolve(channelId); - if (!binding) return; - 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); - }; - let running = false; - try { - let current = await plan(held.room, server); - let instruction = compose(current); - if (!instruction) return; - let context = conversation(held.room, socket?.data.sessionId, repository); - await Chat.instruct(context, instruction.handle, instruction.text, instruction.said); - if (context.chat.running) { - running = true; - void context.chat.running.finally(release).catch(() => {}); - } - } finally { - if (!running) release(); - } - }); +): 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( @@ -377,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 @@ -510,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) { @@ -781,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, }), @@ -2037,7 +2030,7 @@ if (config.liveSpikes) { connections: experiments.connections, withPlan: withImplementationPlan, landed: (channelId, value) => { - void settleSpikePassage(channelId, value).catch(err => + void spikeSettlement.settle(channelId, value.id).catch(err => console.error("chopin: could not settle a prototyped passage -", err) ); }, @@ -2072,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/packages/experiment/src/records.ts b/packages/experiment/src/records.ts index 3daf6762e..868c22710 100644 --- a/packages/experiment/src/records.ts +++ b/packages/experiment/src/records.ts @@ -47,6 +47,11 @@ 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(), }).strict(); export type Spike = z.infer; From 51de746bd6e7043bfe1c57cf777c5b5a5c6d9d18 Mon Sep 17 00:00:00 2001 From: Maggie Appleton <5599295+MaggieAppleton@users.noreply.github.com> Date: Sun, 11 Oct 2026 02:11:59 +0100 Subject: [PATCH 3/3] Resume a spike its local agent's lost connection interrupted A hard-killed connector stayed live until its 90-second expiry, so a spike could be queued on it, then interrupted with investigation copy and never retried after reconnect. Interrupted spikes now wait calmly for the owner's coding agent and are handed back on reconnect up to twice; new spike work skips a connection silent for two heartbeat intervals. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/experiments/connections.ts | 16 +++- apps/server/src/experiments/routes.ts | 5 + apps/server/src/experiments/service.ts | 15 +++ .../server/src/experiments/spike-host.test.ts | 92 ++++++++++++++++++- apps/server/src/experiments/spike-host.ts | 62 ++++++++----- .../src/experiments/spike-scout.test.ts | 58 ++++++++++++ apps/server/src/experiments/spike-scout.ts | 59 +++++++++--- apps/server/src/experiments/spikes.test.ts | 32 ++++++- apps/server/src/experiments/spikes.ts | 28 +++++- packages/experiment/src/records.ts | 2 + 10 files changed, 328 insertions(+), 41 deletions(-) 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 0365466f4..c0533c525 100644 --- a/apps/server/src/experiments/spike-host.test.ts +++ b/apps/server/src/experiments/spike-host.test.ts @@ -15,7 +15,7 @@ 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."; @@ -23,15 +23,95 @@ 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, landed }; + 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(); @@ -173,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); diff --git a/apps/server/src/experiments/spike-host.ts b/apps/server/src/experiments/spike-host.ts index 1de0ac2bb..b82a1c6a4 100644 --- a/apps/server/src/experiments/spike-host.ts +++ b/apps/server/src/experiments/spike-host.ts @@ -31,6 +31,38 @@ export function spikeHost({ service, connections, withPlan, landed }: Options): }); 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 => { @@ -54,7 +86,10 @@ export function spikeHost({ service, connections, withPlan, landed }: Options): }), 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 }) { @@ -88,26 +123,11 @@ export function spikeHost({ service, connections, withPlan, landed }: Options): 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) { diff --git a/apps/server/src/experiments/spike-scout.test.ts b/apps/server/src/experiments/spike-scout.test.ts index 3142f1f9d..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,6 +71,7 @@ function harness(snapshot: Partial = {}, verdict?: SpikeJudge) { scout, state, started, + retried, dismissed, judged, records, @@ -320,3 +328,53 @@ test("a reconnect rescans only documents its owner's scan could not serve", asyn 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 f587c70a6..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,8 +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, by repository, to rescan on reconnect. */ - #offline = new Map(); + /** Documents whose last scan found no local agent for these owners, to rescan on reconnect. */ + #offline = new Map }>(); #closed = false; constructor(options: SpikeScoutOptions) { @@ -175,18 +181,24 @@ export class SpikeScout { */ connected(repositoryId: string, owner: string): void { if (this.#closed) return; - for (let [channelId, repository] of this.#offline) { - if (repository !== repositoryId || this.#editors.get(channelId) !== owner) continue; + 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. */ + /** + * 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); + } }); } @@ -219,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)); @@ -245,10 +277,9 @@ export class SpikeScout { // 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) { - this.#offline.set(channelId, snapshot.repositoryId); + this.#away(channelId, snapshot.repositoryId, editor); return; } - this.#offline.delete(channelId); let verdicts = await (this.#options.judge ?? heuristicJudge)(candidates); if (!verdicts) return; // Approved passages beyond capacity stay unseen so a later scan can start them. @@ -275,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(); diff --git a/apps/server/src/experiments/spikes.test.ts b/apps/server/src/experiments/spikes.test.ts index 7d024c78b..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, resolveInstruction, spikeCallout, spikeReport, uncertain } from "./spikes"; +import { + MAX_SPIKE_RETRIES, + renderKey, + resolveInstruction, + retryable, + spikeCallout, + spikeReport, + uncertain, +} from "./spikes"; import type { Investigation } from "@chopin/experiment/records"; @@ -91,7 +99,27 @@ test("running and stopped spikes render their own callouts without naming a mach expect(queued).toContain(`title="Prototype queued"`); expect(queued).not.toContain("Prototyping"); room.validate(queued); - expect(renderKey(spike("interrupted"))).toBe("stopped"); + expect(renderKey(spike("interrupted"))).toBe("waiting"); +}); + +test("a spike its connection lost waits calmly, then stops once out of retries", () => { + 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 () => { diff --git a/apps/server/src/experiments/spikes.ts b/apps/server/src/experiments/spikes.ts index a18d9f3ac..f9c65e901 100644 --- a/apps/server/src/experiments/spikes.ts +++ b/apps/server/src/experiments/spikes.ts @@ -85,9 +85,22 @@ 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"; @@ -133,11 +146,24 @@ export function spikeCallout(value: Investigation): RootContent { 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.", }), ]); } diff --git a/packages/experiment/src/records.ts b/packages/experiment/src/records.ts index 868c22710..c661a31df 100644 --- a/packages/experiment/src/records.ts +++ b/packages/experiment/src/records.ts @@ -52,6 +52,8 @@ export const spikeSchema = z.object({ * 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;