Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -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.
Original file line number Diff line number Diff line change
@@ -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.
4 changes: 3 additions & 1 deletion AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -126,7 +126,9 @@ collaborative state and external implementation runs are durable.
data for callers with push or administration access.
- **Planner ownership is process-bound.** The first eligible Planner or
model-backed research request supplies its GitHub App token and Copilot
entitlement. The database stores only a token-free owner reference and durable
entitlement. Server-authored turns nobody typed (re-linking decisions after a
build, settling a prototyped passage) never claim an owner; under
`HARNESS=remote` they also wait for an attached Planner. The database stores only a token-free owner reference and durable
context.
- **Planner tools are repository-fixed under `copilot-sdk` and `pi`.** Those
Planners have no shell, checkout, host filesystem, skills, plugins, or arbitrary
Expand Down
70 changes: 70 additions & 0 deletions apps/server/src/chat/server-instruction.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
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).
* It runs only under the document's existing Planner owner and never claims one: nobody asked
* for this turn, so nobody's credentials are taken for it. A room evicted since is reopened.
*/
export async function serverInstruction(
deps: ServerInstructionDeps,
channelId: string,
compose: (plan: Service.Plan) => Instruction | undefined,
): Promise<boolean> {
if (!deps.agent || deps.unavailable(channelId)) return false;
let opened = Rooms.get(channelId)?.plan;
if (opened && !compose(opened)) return false;
let binding = await deps.owner(channelId);
if (!binding) return false;
let 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, undefined, 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();
}
});
}
16 changes: 14 additions & 2 deletions apps/server/src/experiments/connections.ts
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,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);
Expand Down Expand Up @@ -96,7 +101,7 @@ export class Connections {
sessionId: owner.sessionId,
label,
source: { ...source, repositoryId },
expiresAt: this.now() + 90_000,
expiresAt: this.now() + CONNECTION_MS,
autoRun,
};
value.connection = connection;
Expand All @@ -121,7 +126,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) {
Expand Down
5 changes: 5 additions & 0 deletions apps/server/src/experiments/routes.ts
Original file line number Diff line number Diff line change
Expand Up @@ -345,7 +345,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. */
Expand Down
15 changes: 15 additions & 0 deletions apps/server/src/experiments/service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -204,6 +204,21 @@ export class Experiments {
delete value.candidate;
});
}
/** Return a spike its lost connection interrupted to `requested`, for another local agent. */
retrySpike(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()) {
Expand Down
131 changes: 127 additions & 4 deletions apps/server/src/experiments/spike-host.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,21 +15,105 @@ import { Experiments } from "./service";
import { spikeHost } from "./spike-host";
import { calloutDigest, calloutIds, 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, landed, connections, clock };
}

/** Pair a local agent for the plan's repository, allowing automatic work, as the pairing page can. */
function pair(context: Awaited<ReturnType<typeof hosted>>) {
let pending = context.connections.create({
repository: "org/repo",
commit: "a".repeat(40),
label: "laptop",
});
return { ...context, service, host };
return context.connections.approve(
pending.id,
{ id: "U_test", login: "maggie", sessionId: "S_1" },
context.plan.persistence.repositoryId,
// Spikes run only on a laptop paired with consent to automatic work.
true,
);
}

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();
Expand Down Expand Up @@ -171,8 +255,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);
Expand Down Expand Up @@ -339,3 +427,38 @@ test("a spike runs only on a laptop whose owner allowed automatic work at pairin
await Service.close(context.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);
}
});
Loading
Loading