From 9c02c9dea4c3582dca036edf316a5b6a918917c7 Mon Sep 17 00:00:00 2001 From: Alex Lavaee Date: Wed, 28 Oct 2026 22:04:25 -0700 Subject: [PATCH] Let a teammate's Planner finish its current work when its machine stops being shared Turning a machine's sharing off stopped every teammate's Planner on it at once. A teammate whose Planner was replying lost the reply mid-stream, an open question was cancelled, and a live workflow run was paused, all because of someone else's click. Now turning sharing off takes no new teammate documents, and: - a teammate's Planner that is idle, or waits only on a paused run, stops at once, as before; - one that is replying, has a question open, or has a live run keeps going until that work finishes, then stops on the next sweep (every 15 s) with the same reason, " stopped sharing with teammates"; - sharing the machine again meanwhile keeps it, with nothing stopped later. A paused run doesn't hold the Planner: it waits on people for as long as they take, and stays on the machine to resume if its owner shares it again. The busy documents are recorded before sharing ends, so no poll in between moves their Planner. `link.busy` and `attachments.busy` say whether an attached client is replying, asking or running; `waiting` (used by the idle rule) is unchanged. The Machines table shows "Finishing 1 teammate document" under the switch while this lasts, described to screen readers by the switch, and the phone view says the same. GET /api/connector/machines reports `finishing`. Assistant-model: Claude Opus 5.5 (fast) Assistant-workflow: inline Assistant-verification: bun test unit passed: planner-dispatch.test.ts (27 pass) with 3 new cases: an open question, a reply in progress, a live run versus a paused one and sharing again meanwhile; src/harness/remote and src/experiments (219 pass) Assistant-verification: Chromium E2E passed: bun run e2e e2e/planner-picker.e2e.ts e2e/machines.e2e.ts (15 pass), including hubot's reply in progress while mona stops sharing studio Assistant-verification: typecheck and lint passed: bun run types and bun run ci User-preference: When a shared machine stops being shared, a teammate's Planner finishes its current work before it stops Co-authored-by: Alex Lavaee --- apps/server/src/experiments/routes.ts | 71 +++++++++--- apps/server/src/harness/remote/attachments.ts | 5 + apps/server/src/harness/remote/link.ts | 18 ++++ .../harness/remote/planner-dispatch.test.ts | 101 ++++++++++++++++++ apps/web/src/experiments/machine-list.test.ts | 8 ++ apps/web/src/experiments/machine-list.tsx | 46 +++++--- apps/web/src/machine-summary.ts | 2 + docs/chopin-cli.md | 9 +- e2e/planner-picker.e2e.ts | 53 ++++++++- 9 files changed, 284 insertions(+), 29 deletions(-) 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(