diff --git a/apps/server/src/experiments/routes.ts b/apps/server/src/experiments/routes.ts index 8d9a355a2..aea94993b 100644 --- a/apps/server/src/experiments/routes.ts +++ b/apps/server/src/experiments/routes.ts @@ -170,6 +170,11 @@ export const PLANNER_IDLE_MS = 30 * 60_000; */ export const SHARED_REACH_MS = 10 * 60_000; const SHARED_RETRY_MS = 60_000; +/** Why a machine that is no longer shared stopped a teammate's Planner, told to its connector. */ +function stoppedSharing(login: string) { + return (label: string) => + `${login} stopped sharing ${label} with teammates, so it stopped running this document's Planner.`; +} /** * The connector protocols this server speaks; a connector outside them must update first. * Protocol 2 pairs a machine once and connects checkouts with its stored token. Protocol 4 polls @@ -439,6 +444,13 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio */ let reach = new Map(); let reachKey = (machineId: string, repositoryId: string) => `${machineId}\n${repositoryId}`; + /** + * Machines whose owner stopped sharing them while teammates' Planners on them were mid-work, by + * machine id: who stopped, and those documents. Each keeps its Planner there until its work + * finishes (a reply, an open question, a live run), then stops; no new teammate document + * comes to the machine meanwhile. Forgotten with a restart, which ends every Planner anyway. + */ + let finishing = new Map }>(); /** The creator's own machine that can run the document's Planner now: the one builds would use. */ function ownPlannerMachine(channel: ChannelRecord) { return connections.candidates(channel.repositoryId, channel.createdBy, channel.id, null) @@ -450,14 +462,34 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio runsPlanners(value) && reach.get(reachKey(value.machineId, channel.repositoryId))?.ok )?.machineId; } - /** Whether a machine can run a document's Planner: its creator's own, or one shared with them. */ + /** + * Whether a machine can run a document's Planner: its creator's own, one shared with them, or + * one that stopped being shared while finishing this document's work. + */ function readyFor(machineId: string, channel: ChannelRecord) { + let finishes = !!finishing.get(machineId)?.documents.has(channel.id); return connections.machine(machineId).some(value => serves(value, channel.repositoryId) && runsPlanners(value) - && (value.owner === channel.createdBy + && (value.owner === channel.createdBy || finishes || !!value.shared && !!reach.get(reachKey(machineId, channel.repositoryId))?.ok) ); } + /** + * Stop the teammates' Planners a machine that is no longer shared kept while they were mid-work, + * once each has finished or no longer runs there. + */ + async function finishShared() { + for (let [machineId, { login, documents }] of finishing) { + for (let documentId of documents) { + let runsThere = defaults.get(documentId) === machineId; + if (runsThere && options.attachments?.()?.busy(documentId)) continue; + documents.delete(documentId); + let channel = runsThere ? await auth.storage.channels.get(documentId) : undefined; + if (channel) await redefault(channel, stoppedSharing(login)); + } + if (!documents.size) finishing.delete(machineId); + } + } /** * The machine Default machine runs a document's Planner on, while someone uses it and its * creator chose no machine for it: the machine it already runs on while that one can, else the @@ -1153,6 +1185,8 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio ], autoRun: machine.autoRun, shared: !!machine.shared, + // Teammates' documents it still runs the Planner of until their work finishes. + finishing: finishing.get(machine.id)?.documents.size ?? 0, createdAt: machine.createdAt.getTime(), lastSeenAt: machine.lastSeenAt.getTime(), connected: connections.machine(machine.id).length > 0, @@ -1207,7 +1241,8 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio /** * The owner shares a machine with teammates, or stops. Shared, it may run the Planner of a * document on Default machine whose creator has no machine of their own that can, in a - * repository its owner can push to. Stopping ends the Planners it runs for teammates. + * repository its owner can push to. Stopping takes no new teammate documents and ends the + * Planners it runs for teammates: at once where idle, else once their current work finishes. */ route("POST", "/api/connector/machines/:id/shared", async (request, params) => { let session = await auth.sessions.authenticate(request); @@ -1216,19 +1251,28 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio let userId = session.user.id; await readChoices(); if (!await auth.machines.share(params.id, userId, shared)) fail("not-found"); - connections.share(params.id, shared); + finishing.delete(params.id); + let teammates: ChannelRecord[] = []; if (!shared) { - for (let key of reach.keys()) if (key.startsWith(`${params.id}\n`)) reach.delete(key); for (let [documentId, machineId] of defaults) { if (machineId !== params.id) continue; let channel = await auth.storage.channels.get(documentId); - if (!channel || channel.createdBy === userId) continue; - await redefault( - channel, - label => - `${session.user.login} stopped sharing ${label} with teammates, so it stopped running ` - + "this document's Planner.", - ); + if (channel && channel.createdBy !== userId) teammates.push(channel); + } + // A teammate mid-reply, mid-question or mid-run keeps the Planner until it finishes; + // recorded before sharing ends, so no poll in between moves it. + let busy = new Set( + teammates.filter(channel => options.attachments?.()?.busy(channel.id)).map(value => + value.id + ), + ); + if (busy.size) finishing.set(params.id, { login: session.user.login, documents: busy }); + for (let key of reach.keys()) if (key.startsWith(`${params.id}\n`)) reach.delete(key); + } + connections.share(params.id, shared); + for (let channel of teammates) { + if (!finishing.get(params.id)?.documents.has(channel.id)) { + await redefault(channel, stoppedSharing(session.user.login)); } } await plannersMoved(params.id, userId, shared); @@ -2034,6 +2078,9 @@ export function registerExperimentRoutes(router: Router, auth: HostedAuth, optio }); async function sweep() { connections.sweep(); + await finishShared().catch(error => + console.error("chopin: could not stop a machine's finished teammate Planners -", error) + ); await plannersIdle().catch(error => console.error("chopin: could not stop an unused document's Planner -", error) ); diff --git a/apps/server/src/harness/remote/attachments.ts b/apps/server/src/harness/remote/attachments.ts index 1205d5b0a..8dc64ba08 100644 --- a/apps/server/src/harness/remote/attachments.ts +++ b/apps/server/src/harness/remote/attachments.ts @@ -132,6 +132,11 @@ export class PlannerAttachments { return this.#holds.get(channelId)?.link?.waiting ?? false; } + /** Whether the client attached to the document is mid-work: replying, asking, or running. */ + busy(channelId: string): boolean { + return this.#holds.get(channelId)?.link?.busy ?? false; + } + /** Waits at most `ms` for a client to attach to the document; whether one is attached then. */ attachedWithin(channelId: string, ms: number): Promise { if (this.attached(channelId)) return Promise.resolve(true); diff --git a/apps/server/src/harness/remote/link.ts b/apps/server/src/harness/remote/link.ts index 8af750dcc..db5591851 100644 --- a/apps/server/src/harness/remote/link.ts +++ b/apps/server/src/harness/remote/link.ts @@ -182,6 +182,11 @@ class RemoteSession { void this.#started.promise.catch(() => {}); } + /** Whether a turn is under way: the Planner is replying. */ + get replying(): boolean { + return this.#current !== undefined; + } + async start(): Promise { let mode = this.planner.isolated ? "isolated" as const : "planner" as const; if (!this.#port.send({ type: "start", session: this.id, mode })) { @@ -639,6 +644,19 @@ export class AgentLink { return false; } + /** + * Whether the client is in the middle of work for the document's members: a reply under way, + * a question still open, or a live workflow run. A paused run alone is not; it waits on people. + */ + get busy(): boolean { + if (this.#state === "closed") return false; + if (this.#inputs.size > 0) return true; + for (let session of this.#sessions.values()) { + if (session.replying || (session.planner.runs?.active.length ?? 0) > 0) return true; + } + return false; + } + /** Opens a Planner session on the client; throws when it cannot. */ async startSession(sessionId: string, planner: RemotePlanner): Promise { if (this.#state !== "attached") throw new Error("The agent connection is not attached."); diff --git a/apps/server/src/harness/remote/planner-dispatch.test.ts b/apps/server/src/harness/remote/planner-dispatch.test.ts index 2843361cb..eea7e97fb 100644 --- a/apps/server/src/harness/remote/planner-dispatch.test.ts +++ b/apps/server/src/harness/remote/planner-dispatch.test.ts @@ -855,6 +855,107 @@ describe("a teammate's machine shared with teammates", () => { expect((await choice("member")).planner).toEqual({ state: "none", login: null, machine: null }); }); + /** rival's shared build-box runs the member's document's Planner, attached with a live session. */ + async function sharedAndAttached() { + let dispatch = await dispatchServer(); + let { machine, share, visit, channel, server } = dispatch; + let box = await machine("rival", "U_rival", "build-box"); + await share("rival", box.id); + await visit(); + await box.poll([]); + let link = await box.backend(); + await link.next("attached"); + let room = await hostInputRoom(); + cleanups.push(room.close); + let planner: RemotePlanner = { + channelId: channel.id, + isolated: false, + humanInput: report => createHumanInput(room.room, undefined, report), + }; + let started = link.next("start").then(start => { + link.send({ type: "started", session: start.session }); + }); + let session = await server.links.at(-1)!.startSession("planner-session", planner); + cleanups.push(() => session.doDestroy?.()); + await started; + /** What build-box is told about the document's Planner after a sweep. */ + let afterSweep = async () => { + await dispatch.runtime.sweep(); + return box.poll([channel.id]); + }; + let stillRuns = async () => "waiting" in await afterSweep(); + let finishing = async () => + (await (await dispatch.send("/api/connector/machines", { login: "rival" })).json()) + .machines[0].finishing as number; + return { ...dispatch, box, link, room, session, afterSweep, stillRuns, finishing }; + } + + it("keeps a teammate's Planner through an open question when sharing stops, then stops it", async () => { + let { share, box, link, room, afterSweep, stillRuns, finishing, choice, channel } = + await sharedAndAttached(); + link.send({ + type: "input-request", + request: "ship", + session: "planner-session", + input: { method: "confirm", title: "Ship?", message: "Now." }, + }); + let [card] = await room.cards(1); + expect(await share("rival", box.id, false)).toBe(200); + expect(await stillRuns()).toBe(true); + expect(await finishing()).toBe(1); + expect((await choice("member")).planner) + .toEqual({ state: "held", login: "rival", machine: null }); + + await room.answer(card!.id, [0]); + expect(await link.next("input-result")).toMatchObject({ status: "answered", value: true }); + expect(await afterSweep()).toEqual({ + kind: "planners", + planners: [], + stopped: { [channel.id]: expect.stringContaining("stopped sharing build-box") }, + }); + expect(await link.next("refused")).toMatchObject({ code: "access-revoked" }); + expect(await finishing()).toBe(0); + }); + + it("keeps a teammate's Planner through its reply when sharing stops, then stops it", async () => { + let { share, box, link, session, stillRuns } = await sharedAndAttached(); + let turn = session.doPromptTurn({ + prompt: "outline the plan", + skills: [], + tools: [], + emit: () => {}, + }); + let { turn: id } = await link.next("turn"); + await turn; + expect(await share("rival", box.id, false)).toBe(200); + expect(await stillRuns()).toBe(true); + link.send({ type: "turn-end", session: "planner-session", turn: id, status: "finished" }); + await Bun.sleep(50); + expect(await stillRuns()).toBe(false); + expect(await link.next("refused")).toMatchObject({ code: "access-revoked" }); + }); + + it("keeps it through a live run, not a paused one, and keeps it when shared again meanwhile", async () => { + let { share, box, link, stillRuns, finishing } = await sharedAndAttached(); + let runs = async (active: string[], paused: string[]) => { + link.send({ type: "runs", session: "planner-session", active, paused, cards: [] }); + await Bun.sleep(50); + }; + await runs(["root"], []); + expect(await share("rival", box.id, false)).toBe(200); + expect(await stillRuns()).toBe(true); + // Shared again before it finished: nothing is stopped later. + expect(await share("rival", box.id)).toBe(200); + expect(await finishing()).toBe(0); + await runs([], ["root"]); + expect(await stillRuns()).toBe(true); + + // A paused run waits on people, not on the machine: sharing off stops the Planner at once. + expect(await share("rival", box.id, false)).toBe(200); + expect(await link.next("refused")).toMatchObject({ code: "access-revoked" }); + expect(await stillRuns()).toBe(false); + }); + it("lets only the owner share a machine", async () => { let { machine, share } = await dispatchServer(); let box = await machine("rival", "U_rival", "build-box"); diff --git a/apps/web/src/experiments/machine-list.test.ts b/apps/web/src/experiments/machine-list.test.ts index 8c3095ea0..cb74de19e 100644 --- a/apps/web/src/experiments/machine-list.test.ts +++ b/apps/web/src/experiments/machine-list.test.ts @@ -1,5 +1,6 @@ import { expect, test } from "bun:test"; import { + finishingWords, groupMachines, machineMatches, machineState, @@ -79,3 +80,10 @@ test("machines group by kind, Local then Cloud, leaving out an empty group", () expect(groupMachines([box]).map(group => group.kind)).toEqual(["cloud"]); expect(groupMachines([])).toEqual([]); }); + +test("a machine that stopped being shared says how many teammate documents it is finishing", () => { + expect(finishingWords({})).toBeUndefined(); + expect(finishingWords({ finishing: 0 })).toBeUndefined(); + expect(finishingWords({ finishing: 1 })).toBe("Finishing 1 teammate document"); + expect(finishingWords({ finishing: 2 })).toBe("Finishing 2 teammate documents"); +}); diff --git a/apps/web/src/experiments/machine-list.tsx b/apps/web/src/experiments/machine-list.tsx index c7d2359f3..b45d9e7dd 100644 --- a/apps/web/src/experiments/machine-list.tsx +++ b/apps/web/src/experiments/machine-list.tsx @@ -155,7 +155,8 @@ function Dash({ label }: { label: string }) { /** What sharing a machine with teammates does, as the switch's tooltip and the phone row say. */ export const SHARE_DETAIL = "A teammate's document with no machine of its own can run its Planner here, as you, " - + "in repositories you can push to. Builds stay yours."; + + "in repositories you can push to. Builds stay yours. Turned off, a teammate's Planner " + + "finishes the reply, question or run it is on first."; /** * Whether teammates' documents may plan on a machine: a switch, named for the machine, that @@ -164,21 +165,36 @@ export const SHARE_DETAIL = function ShareSwitch( { machine, onChange }: { machine: Machine; onChange: (shared: boolean) => void }, ) { + let finishingId = useId(); + let finishing = !machine.shared && finishingWords(machine); return ( - + + + {finishing && {finishing}} + ); } +/** + * What a machine that stopped being shared still does for teammates: their Planners keep going + * until their current work finishes. Undefined once none does. + */ +export function finishingWords(machine: Pick): string | undefined { + let count = machine.finishing ?? 0; + return count ? `Finishing ${count} teammate document${count === 1 ? "" : "s"}` : undefined; +} + function copiesFor(machine: Machine, recommended: string | null): CopyItem[] { return [ ...(updateAvailable(machine.version, recommended) @@ -722,7 +738,11 @@ export function MachineDetail(