Finish the living document flow's rough edges

.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.

.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.

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: <T>(channelId: string, action: () => Promise<T>) => Promise<T>;
+	/** Attach the document to a held room, reopening it if it was evicted. */
+	open: (room: Rooms.Room) => Promise<Service.Plan>;
+	conversation: (
+		room: Rooms.Room,
+		claimantSessionId: string | undefined,
+		repository: Chat.Room["repository"],
+	) => Chat.Room;
+	instruct: (context: Chat.Room, instruction: Instruction) => void | Promise<void>;
+	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<boolean> {
+	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();
+		}
+	});
+}

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) {

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. */

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()) {

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<ReturnType<typeof hosted>>) {
+	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);
+	}
+});