From d25c8bca1e1166338e3fdd7e0de9e5f48ae1fec1 Mon Sep 17 00:00:00 2001 From: Alex Lavaee Date: Thu, 12 Nov 2026 17:12:56 -0800 Subject: [PATCH] Keep a workflow run live, and its questions asked, when its machine resumes after a restart After a deploy restarted the server, a run a machine was still working on showed as stopped in Chat, its stages gone, and the question it asked a few minutes later never became a Decision, so the run waited for an answer nobody could give. The machine had resumed its sessions a second after the restart, as designed. Two things then broke. The cards stopped because a document that opens after its client is back restored them as stopped: whether a restart's rebuilt hold still keeps runs was asked only of the hold before its client took it back. A client that resumes moves the sessions to its new link, where nothing in the new process retains them, so the answer was "nothing", and Chat stopped every live card. The sessions a restart rebuilt also had no way to reach the document: their run reports went nowhere, and a question on one was refused with "Decisions are not available to this session." A session a restart rebuilt now reaches its document through the room's host, which the document gives it when it opens, as it already did for the questions left open at the cut. Its run reports are Chat's run cards, its link dropping again marks them disconnected, and a document that opens while a link holds such a session with work shows its cards live, with what the client last reported. A question asked on it becomes Decisions as one asked before the restart did; the document is opened for it if nobody has it open and kept open until the question ends, and the answer goes to the client's request. Assistant-model: Claude Opus 5.5 Assistant-workflow: inline (subagent) Assistant-verification: bun test apps/server/src/harness/remote/handover-window.test.ts (37 pass; the 3 new restart-mid-run tests fail on 74772eec4: card "stopped", question refused "Decisions are not available to this session.") Assistant-verification: bun test apps/server/src/harness/remote apps/server/src/harness/atomic/human-input.test.ts apps/server/src/testing/resume-after-restart.test.ts (342 pass) Assistant-verification: bun test apps/server/src/harness/session.test.ts apps/server/src/harness/atomic/full.test.ts apps/server/src/chat/invoke.test.ts apps/server/src/chat/service.test.ts apps/server/src/chat/steering.test.ts apps/server/src/chat/job-heading-memory.test.ts apps/server/src/chat/remote-planner.test.ts apps/server/src/plan/persistence.test.ts (297 pass) Assistant-verification: bun run types; bun run ci User-preference: Things should be polished like the Slack UX Co-authored-by: Alex Lavaee --- apps/server/src/chat/service.ts | 73 +++++++-- apps/server/src/harness/atomic/human-input.ts | 2 +- apps/server/src/harness/remote/attachments.ts | 24 ++- .../harness/remote/handover-window.test.ts | 147 +++++++++++++++++- apps/server/src/harness/remote/link.ts | 114 +++++++++++++- apps/server/src/harness/remote/registry.ts | 7 +- apps/server/src/main.ts | 47 +++++- docs/agent-connection.md | 10 +- docs/hosted-agent.md | 5 +- 9 files changed, 396 insertions(+), 33 deletions(-) diff --git a/apps/server/src/chat/service.ts b/apps/server/src/chat/service.ts index c726abf09..6fd27d695 100644 --- a/apps/server/src/chat/service.ts +++ b/apps/server/src/chat/service.ts @@ -2124,7 +2124,7 @@ const storedStages = new WeakMap | undefined { +function storeStages(context: RunsRoom, cards: Wire.Run[]): Promise | undefined { let known = storedStages.get(context.chat) ?? new Map }>(); storedStages.set(context.chat, known); @@ -2173,7 +2173,7 @@ function cardWindow(run: Wire.Run): Wire.Run { } /** Forgets the stored stages of runs Chat no longer shows. */ -async function forgetStages(context: Room, shown: Wire.Run[] | undefined): Promise { +async function forgetStages(context: RunsRoom, shown: Wire.Run[] | undefined): Promise { let ids = (shown ?? []).map(run => run.id); let known = storedStages.get(context.chat); for (let id of known?.keys() ?? []) if (!ids.includes(id)) known!.delete(id); @@ -2185,12 +2185,51 @@ async function forgetStages(context: Room, shown: Wire.Run[] | undefined): Promi } const publishing = new WeakMap>(); -const owners = new WeakMap>(); -const unsaved = new WeakMap(); -const gone = new WeakSet(); +const owners = new WeakMap>(); +const unsaved = new WeakMap(); +const gone = new WeakSet(); type Runs = { active: string[]; paused: string[]; cards?: Wire.Run[] }; +/** The session whose runs a card shows: a retained Planner session, or one a restart rebuilt. */ +type RunsOwner = Pick; + +/** What showing and storing a session's run cards needs of the room. */ +type RunsRoom = Pick & { + auth: { storage: Pick }; +}; + +/** Stand-ins, by session id, for the sessions a restart rebuilt whose clients report runs. */ +const rebuiltOwners = new WeakMap>(); + +/** + * Shows the runs reported by a Planner session a restart rebuilt and its client took back, `down` + * while its client's link has dropped again. No turn of this process opened that session, so + * nothing here retains it and its reports reach Chat only this way, as a retained session's do. + */ +export function rebuiltRuns( + context: RunsRoom, + session: string, + runs: Runs, + down = false, +): Promise { + if (context.chat.closed) return Promise.resolve(); + let known = rebuiltOwners.get(context.chat) + ?? new Map(); + rebuiltOwners.set(context.chat, known); + let entry = known.get(session); + if (!entry) { + let made: { owner: RunsOwner; down: boolean } = { + owner: { suspended: () => made.down }, + down, + }; + entry = made; + known.set(session, entry); + } + entry.down = down; + return publishRuns(context, entry.owner, runs); +} + /** * Show the session's runs, keeping ended ones as summary cards until the next * run starts, store them with the document so a reload keeps them, and say once @@ -2198,13 +2237,13 @@ type Runs = { active: string[]; paused: string[]; cards?: Wire.Run[] }; * it: a report or release only changes its own session's cards. */ function publishRuns( - context: Room, - session: PlannerSession, + context: RunsRoom, + session: RunsOwner, runs: Runs | undefined, releasing = false, - followUp = false, + followUp?: (run: Wire.Run, said: string) => Promise, ): Promise { - let known = owners.get(context.chat) ?? new Map(); + let known = owners.get(context.chat) ?? new Map(); owners.set(context.chat, known); for (let run of runs?.cards ?? []) known.set(run.id, session); let turn = (publishing.get(context.chat) ?? Promise.resolve()).then(() => @@ -2215,11 +2254,11 @@ function publishRuns( } async function publishReport( - context: Room, - session: PlannerSession, + context: RunsRoom, + session: RunsOwner, runs: Runs | undefined, releasing: boolean, - followUp: boolean, + followUp: ((run: Wire.Run, said: string) => Promise) | undefined, ): Promise { let { chat, room, server } = context; let at = now(); @@ -2301,7 +2340,7 @@ async function publishReport( for (let entry of finished) announce(server, room, entry); state(chat, server, room); if (prior.some(run => !shown.has(run.id))) await forgetStages(context, next); - for (let { run, said } of attention) await followUpEndedRun(context, run, said); + for (let { run, said } of attention) await followUp?.(run, said); } /** @@ -2371,7 +2410,13 @@ function retain( if (chat.retained?.session !== opened.session) return; let planning = chat.busy && !chat.job && chat.agent === opened.session; let followUp = !planning && opened.session.followsUpRuns !== false; - void publishRuns(context, opened.session, next, false, followUp); + void publishRuns( + context, + opened.session, + next, + false, + followUp ? (run, said) => followUpEndedRun(context, run, said) : undefined, + ); if (next.active.length || next.paused.length) return; release(); }); diff --git a/apps/server/src/harness/atomic/human-input.ts b/apps/server/src/harness/atomic/human-input.ts index 974cd9e18..9801223ba 100644 --- a/apps/server/src/harness/atomic/human-input.ts +++ b/apps/server/src/harness/atomic/human-input.ts @@ -227,7 +227,7 @@ export type HumanInput = { * cannot tell apart. */ export function createHumanInput( - room: DocumentRoom, + room: Pick, hold?: (workflowRunId: string, signal: AbortSignal) => Promise, report?: (status: HumanInputStatus) => void, pending?: PendingInputs, diff --git a/apps/server/src/harness/remote/attachments.ts b/apps/server/src/harness/remote/attachments.ts index b64f3648e..0751cfe25 100644 --- a/apps/server/src/harness/remote/attachments.ts +++ b/apps/server/src/harness/remote/attachments.ts @@ -110,6 +110,11 @@ export type PlannerAttachmentOptions = { restoredRuns?: (channelId: string, outcome: RestoredRuns) => void; /** Heard when a hand-over window ends without the client back, with what was missing. */ missed?: (channelId: string, cause: "login" | "client") => void; + /** + * Opens a document, and keeps it open until the returned function is called, for a question + * the client of a hold a restart rebuilt asks while nobody has the document open. + */ + openDocument?: (channelId: string) => Promise<() => void>; }; /** What a document's members are told about its Planner: only the holder's login and state. */ @@ -574,10 +579,12 @@ export class PlannerAttachments { continue; } if (row.sessions.length > 0) { + let open = this.#options.openDocument; hold.parked = AgentLink.rebuilt(row.channelId, row.sessions, { lapse: () => this.#lapsed(hold, hold.parked!), ended: () => this.#runs(hold, "ended"), resumed: () => this.#runs(hold, "up"), + ...(open ? { open: () => open(row.channelId) } : {}), }); } let length = row.handoverUntil @@ -621,14 +628,21 @@ export class PlannerAttachments { } /** - * Whether a hold a restart rebuilt still keeps sessions its client has not taken back, so the - * workflow runs Chat stored for the document are live and not stopped when its Chat opens: - * `live` through the hold's window, `down` once its client is known to be away. + * Whether a hold a restart rebuilt still keeps sessions with work, so the workflow runs Chat + * stored for the document are live and not stopped when its Chat opens: `live` through the + * hold's window and once its client has taken them back, `down` while its client is known to be + * away. A client that took them back keeps them on its link, where no turn of this process + * retains them, so a document that opens after it resumed still shows its runs live. */ restoredWork(channelId: string): "live" | "down" | undefined { let hold = this.#holds.get(channelId); - if (!hold?.restored || !hold.parked || hold.parked.snapshot().length === 0) return undefined; - return this.#inWindow(hold) ? "live" : "down"; + if (!hold) return undefined; + if (hold.restored && hold.parked && hold.parked.snapshot().length > 0) { + return this.#inWindow(hold) ? "live" : "down"; + } + if (hold.link?.keepsRebuilt) return "live"; + if (hold.parked?.keepsRebuilt) return "down"; + return undefined; } /** diff --git a/apps/server/src/harness/remote/handover-window.test.ts b/apps/server/src/harness/remote/handover-window.test.ts index aebb2c7c1..69e329192 100644 --- a/apps/server/src/harness/remote/handover-window.test.ts +++ b/apps/server/src/harness/remote/handover-window.test.ts @@ -1,6 +1,8 @@ import { afterEach, describe, expect, it, spyOn } from "bun:test"; import { Sessions } from "../../auth/session"; +import * as Chat from "../../chat/service"; +import * as Plan from "../../plan/service"; import * as Store from "../../questions/store"; import { hostInputRoom } from "../../testing/decisions"; import { expectPlainStatus } from "../../testing/plain-words"; @@ -16,8 +18,9 @@ import { untilUnpaused } from "../atomic/full"; import { createRemoteAdapter } from "./adapter"; import { PlannerAttachments } from "./attachments"; +import type { DocumentRoom } from "../../agent/tools"; import type { Lease } from "../../storage/model"; -import type { PlannerAttachment } from "./attachments"; +import type { PlannerAttachment, PlannerAttachmentOptions } from "./attachments"; import type { RemotePlanner } from "./registry"; const OFFER = "resumable-sessions, restart-notice"; @@ -119,7 +122,7 @@ async function deploy(input: Record = ASK_CONFIRM, asked = 1) { async function cut( windowMs = WINDOW_MS, spent: { beforeWrite?: number; beforeRelease?: number; beforeTakeover?: number } = {}, - takeover: { failWrites?: boolean } = {}, + takeover: { failWrites?: boolean; options?: PlannerAttachmentOptions } = {}, ) { first.attachments.shutdown(); let rows = first.attachments.freeze(windowMs); @@ -144,6 +147,7 @@ async function deploy(input: Record = ASK_CONFIRM, asked = 1) { }, restoredRuns: (id, outcome) => events.push([`runs:${outcome}`, id]), missed: (id, cause) => events.push([`missed:${cause}`, id]), + ...takeover.options, }); let changes: PlannerAttachment[] = []; attachments.onChange((_id, status) => changes.push(status)); @@ -189,13 +193,14 @@ async function deploy(input: Record = ASK_CONFIRM, asked = 1) { /** * The second process opens the document again from what the first stored, as the first request - * for it does, so the questions its rebuilt hold's client left open are asked of the members again. + * for it does, so the questions its rebuilt hold's client left open are asked of the members again, + * and Chat shows what the sessions it rebuilt report and ask, as `main.ts` wires them. */ async function reopen( d: Awaited>, next: Awaited>["cut"]>>, ) { - await d.f.close(); + if (!d.closed.first) await d.f.close(); d.closed.first = true; let unsent: string[] = []; let g = await hostInputRoom(undefined, { @@ -203,10 +208,32 @@ async function reopen( channelId: d.f.channelId, lease: d.f.lease, backend: { + // The fixture's document and the link's have their own ids: this is the link's. + plannerKeepsRuns: () => next.attachments.restoredWork(d.channel.id), rejoinPlannerQuestions: (plan, server) => next.attachments.rejoinInputs(d.channel.id, { join: (asked, signal) => rejoinInput({ id: plan.id, plan, server }, asked, signal), unsent: decisions => unsent.push(...decisions), + live: () => !plan.persistence.closing, + humanInput: (hold, report) => + createHumanInput( + { id: plan.id, plan, server, anchors() {} } as DocumentRoom, + hold, + report, + ), + runs: (session, runs, down) => + void Chat.rebuiltRuns( + { + chat: plan.chat, + room: plan.id, + server, + persist: chat => Plan.persist(plan, chat), + auth: { storage: d.f.storage }, + }, + session, + runs, + down, + ), }), }, }); @@ -783,3 +810,115 @@ describe("a remote Planner whose link a cut closed", () => { expect(rows).toEqual([]); }); }); + +describe("a workflow run its client carries on through a restart", () => { + const TRUST = { + method: "confirm", + title: "Trust the toolchain?", + message: "Its checks pass.", + } as const; + + /** The first process's Chat shows the run as its retained session reported it, and stores it. */ + async function shown(d: Awaited>) { + d.f.plan.chat.runs = [run]; + await Plan.persist(d.f.plan, () => ({ + entries: d.f.plan.chat.entries, + runs: [run], + initiativePaused: false, + })); + } + + /** The client is back a second after the restart, before any browser opens the document. */ + async function resumed( + d: Awaited>, + next: Awaited>["cut"]>>, + ) { + let client = await returns(d, next); + client.send({ type: "resume", sessions: [{ session: d.sessionId, requests: ["ask-1"] }] }); + await client.next("resumed"); + client.send({ + type: "runs", + session: d.sessionId, + active: ["run"], + paused: [], + cards: [{ + ...run, + updated: 2, + stages: [{ id: "verify", name: "verify", status: "running" as const }], + }], + }); + return client; + } + + it("is shown running, not stopped, when its document opens after the client is back, with the stages it reports", async () => { + let d = await deploy(); + await shown(d); + let next = await d.cut(); + await resumed(d, next); + await Bun.sleep(50); + let { g } = await reopen(d, next); + expect(g.plan.chat.runs?.map(card => card.status)).toEqual(["running"]); + await until(() => g.plan.chat.runs?.[0]?.stages.some(stage => stage.id === "verify") === true); + expect(g.plan.chat.runs).toMatchObject([{ id: "run", status: "running", updated: 2 }]); + expect(g.plan.chat.runs?.[0]?.disconnected).toBeUndefined(); + }); + + it("asks the question it asks after the restart as a Decision, and gives the client the answer", async () => { + let d = await deploy(); + await shown(d); + let next = await d.cut(); + let client = await resumed(d, next); + let { g } = await reopen(d, next); + client.send({ + type: "input-request", + request: "ask-2", + session: d.sessionId, + workflowRunId: "run", + input: TRUST, + }); + let asked = (await g.cards(2)).find(card => card.id !== d.card.id); + expect(asked!.definition.questions[0]!.question).toContain("Trust the toolchain?"); + await g.answer(asked!.id, [0]); + expect(await client.next("input-result")).toMatchObject({ + request: "ask-2", + status: "answered", + value: true, + }); + }); + + it("opens its document to ask the question when nobody has it open, and keeps it open until the answer", async () => { + let d = await deploy(); + await shown(d); + await d.f.close(); + d.closed.first = true; + let opened: Array>> = []; + let released = 0; + let next: Awaited> = await d.cut(WINDOW_MS, {}, { + options: { + openDocument: async () => { + opened.push(await reopen(d, next)); + return () => void released++; + }, + }, + }); + let client = await resumed(d, next); + client.send({ + type: "input-request", + request: "ask-2", + session: d.sessionId, + workflowRunId: "run", + input: TRUST, + }); + await until(() => opened.length === 1); + let { g } = opened[0]!; + let asked = (await g.cards(2)).find(card => card.id !== d.card.id); + expect(released).toBe(0); + await g.answer(asked!.id, [0]); + expect(await client.next("input-result")).toMatchObject({ + request: "ask-2", + status: "answered", + value: true, + }); + await until(() => released === 1); + }); +}); diff --git a/apps/server/src/harness/remote/link.ts b/apps/server/src/harness/remote/link.ts index ac1aff418..1c1b4371e 100644 --- a/apps/server/src/harness/remote/link.ts +++ b/apps/server/src/harness/remote/link.ts @@ -14,6 +14,7 @@ import { parseClientMessage, usableCatalog, } from "@chopin/agent-link"; +import { untilUnpaused } from "../atomic/full"; import type { HarnessV1PromptControl, @@ -34,7 +35,8 @@ import type { } from "@chopin/agent-link"; import type { Chat as Wire } from "@chopin/protocol"; import type { JsonValue, PlannerHeldSession } from "../../storage/model"; -import type { AskedInput, HumanInputStatus, InputMethod } from "../atomic/human-input"; +import type { PlannerRuns } from "../atomic/full"; +import type { AskedInput, HumanInput, HumanInputStatus, InputMethod } from "../atomic/human-input"; import type { PlannerTurn } from "../turn-level"; import type { RemotePlanner } from "./registry"; @@ -303,8 +305,71 @@ export type RejoinHost = { * it. They are handed to the Planner as a turn, as a decision nobody waits on is. */ unsent(decisions: string[]): void; + /** Whether the document this host asks in is still open; absent, it is. */ + live?(): boolean; + /** + * Asks the document's members, as Decisions, a question a rebuilt session's client asks after + * the restart. No turn of this process opened that session, so nothing else can. + */ + humanInput?( + hold: (workflowRunId: string, signal: AbortSignal) => Promise, + report: (status: HumanInputStatus) => void, + ): HumanInput; + /** + * Shows what a rebuilt session's client reports of its workflow runs as the document's run + * cards, `down` while its link has dropped again. Nothing in this process retains that session, + * so its reports reach Chat only this way. + */ + runs?(session: string, runs: PlannerRuns, down: boolean): void; +}; + +/** + * What a session a restart rebuilt needs to reach its document: the host of the document's room + * while it is open, and a way to open it for as long as a question waits. + */ +export type RebuiltDocument = { + host?: RejoinHost; + /** Opens the document and keeps it open until the returned function is called. */ + open?: () => Promise<() => void>; }; +const METHODS = ["questionnaire", "confirm", "select", "input", "editor"] as const; + +/** + * Decisions for a session a restart rebuilt. Each question opens the document, or keeps the open + * one, for as long as it waits, and is asked through the host of its room, so a question the + * client asks after the restart becomes a Decision as one asked before it did. + */ +function rebuiltInput( + planner: RemotePlanner, + document: RebuiltDocument, + report: (status: HumanInputStatus) => void, +): HumanInput { + let live = () => { + let host = document.host; + return host?.humanInput && host.live?.() !== false ? host : undefined; + }; + let through = async (ask: (input: HumanInput) => Promise): Promise => { + let release = document.open ? await document.open() : undefined; + try { + let host = live(); + if (!host) throw new Error("Decisions are not available to this session."); + return await ask( + host.humanInput!((runId, signal) => untilUnpaused(planner, runId, signal), report), + ); + } finally { + release?.(); + } + }; + return Object.fromEntries( + METHODS.map(method => [ + method, + (...args: unknown[]) => + through(input => (input[method] as (...args: unknown[]) => Promise)(...args)), + ]), + ) as HumanInput; +} + /** One Planner session whose turns run in the attached client. */ class RemoteSession { readonly id: string; @@ -560,6 +625,8 @@ class RemoteSession { restore(stops: JsonValue[]): void { this.#acknowledged = true; this.planner.suspended = true; + // A workflow question names its root, so a paused root holds its answer, as after `start`. + this.planner.rootOf = runId => this.#roots.get(runId) ?? runId; for (let stop of stops) { let run = (stop as { run?: string | null } | null)?.run; this.#hold("stop", typeof run === "string" ? run : undefined); @@ -1077,7 +1144,13 @@ export class AgentLink { static rebuilt( channelId: string, held: PlannerHeldSession[], - hooks: { lapse(): void; ended(): void; resumed(): void }, + hooks: { + lapse(): void; + ended(): void; + resumed(): void; + /** Opens the document for a question its client asks, until the returned function is called. */ + open?: () => Promise<() => void>; + }, ): AgentLink { let link = new AgentLink({ send() {}, close() {} }, { attach: async () => ({ @@ -1092,7 +1165,17 @@ export class AgentLink { link.#lapse = hooks.lapse; link.#rebuilt = hooks; for (let item of held) { - let session = new RemoteSession(item.id, { channelId, isolated: false }, link.#port); + // No turn of this process opens the session, so it reaches its document through the room's + // host: what its client reports and asks after the restart is shown and asked as before. + let document: RebuiltDocument = hooks.open ? { open: hooks.open } : {}; + let planner: RemotePlanner = { channelId, isolated: false, rebuilt: document }; + planner.humanInput = report => rebuiltInput(planner, document, report); + let show = () => { + if (planner.runs) document.host?.runs?.(item.id, planner.runs, planner.suspended === true); + }; + planner.onRuns = show; + planner.linkChanged = show; + let session = new RemoteSession(item.id, planner, link.#port); session.restore(item.stops); link.#sessions.set(item.id, session); for (let owed of item.owed) { @@ -1159,6 +1242,14 @@ export class AgentLink { rejoinInputs(host: RejoinHost): void { if (this.#successor) return this.#successor.rejoinInputs(host); if (this.#state === "closed") return; + // A session the restart rebuilt asks and reports through this room from now on, and its + // cards show what its client last reported. + for (let session of this.#sessions.values()) { + let rebuilt = session.planner.rebuilt; + if (!rebuilt) continue; + rebuilt.host = host; + if (session.planner.runs) session.planner.onRuns?.(session.planner.runs); + } for (let [request, input] of this.#inputs) { if (!input.restored || input.joined === host) continue; input.joined = host; @@ -1198,6 +1289,23 @@ export class AgentLink { } } + /** + * Whether the link holds a session a restart rebuilt that still has work: a live or paused run, + * an open question, or no report yet of what it kept. Nothing in this process retains such a + * session, so the run cards Chat stored are its own, and live, when the document opens. + */ + get keepsRebuilt(): boolean { + if (this.#successor) return this.#successor.keepsRebuilt; + if (this.#state === "closed") return false; + for (let session of this.#sessions.values()) { + if (!session.planner.rebuilt) continue; + let runs = session.planner.runs; + if (!runs || runs.active.length > 0 || runs.paused.length > 0) return true; + if (this.#waitingOn(session.id) > 0) return true; + } + return false; + } + /** Whether the client's work in `session` is kept when this link drops without being ended. */ #keeps(session: RemoteSession): boolean { return this.resumable && session.resumable diff --git a/apps/server/src/harness/remote/registry.ts b/apps/server/src/harness/remote/registry.ts index a7a452c2e..04fd2d0dc 100644 --- a/apps/server/src/harness/remote/registry.ts +++ b/apps/server/src/harness/remote/registry.ts @@ -1,6 +1,6 @@ import type { FullPlanner } from "../atomic/full"; import type { HumanInput, HumanInputStatus } from "../atomic/human-input"; -import type { AgentLink } from "./link"; +import type { AgentLink, RebuiltDocument } from "./link"; /** * What Chopin keeps for one Planner session whose model runs in an attached @@ -55,6 +55,11 @@ export type RemotePlanner = forkedNatively?: boolean; /** Reports `{ session, turn }` as each remote turn starts, so a thread can fork from it. */ onRemoteTurn?: (point: { session: string; turn: string }) => void; + /** + * A session a restart rebuilt, which no turn of this process opened: how what its client + * reports and asks reaches the document. + */ + rebuilt?: RebuiltDocument; }; let planners = new Map(); diff --git a/apps/server/src/main.ts b/apps/server/src/main.ts index 3198c79bc..4ab93f5b8 100644 --- a/apps/server/src/main.ts +++ b/apps/server/src/main.ts @@ -131,7 +131,7 @@ import * as Explore from "./questions/explore"; import { runningFindingRefs } from "./questions/service-findings"; import { answerTurns } from "./questions/answer-turns"; import { decisionGeneration } from "./questions/card-actions"; -import { rejoinInput } from "./harness/atomic/human-input"; +import { createHumanInput, rejoinInput } from "./harness/atomic/human-input"; import type { RejoinHost } from "./harness/remote/link"; import { deterministicChannelId } from "./channels/id"; import type { Research as ResearchWire } from "@chopin/protocol"; @@ -529,8 +529,34 @@ function answerDecision(channelId: string, id: string, generation: number): void * members again, and an answer its client can never be given becomes a turn of its own. */ function plannerQuestions(plan: Service.Plan, server: Server): RejoinHost { + let live = () => !plan.persistence.closing && !plan.chat.closed; return { join: (asked, signal) => rejoinInput({ id: plan.id, plan, server }, asked, signal), + live, + // A session a restart rebuilt asks and reports here as a retained session of this Chat would. + humanInput: (hold, report) => + createHumanInput( + { id: plan.id, plan, server, anchors: () => Service.anchors(plan, server, plan.id) }, + hold, + report, + ), + runs: (session, runs, down) => { + if (!live()) return; + void Chat.rebuiltRuns( + { + chat: plan.chat, + room: plan.id, + server, + persist: chat => Service.persist(plan, chat), + auth: hostedAuth, + }, + session, + runs, + down, + ).catch(err => + console.error("chopin: could not show the workflow runs a restart kept -", err) + ); + }, unsent: decisions => { for (let id of decisions) { let record = plan.records.get(id); @@ -3083,6 +3109,25 @@ if (config.agent && config.harness === "remote") { }, }, }), + // A question a rebuilt hold's client asks keeps its document open until it is answered. + openDocument: async channelId => { + if (deletingChannels.has(channelId) || archivingChannels.has(channelId)) { + throw new Error("document is unavailable"); + } + await Rooms.get(channelId)?.closing; + let held = Rooms.hold(channelId); + let release = () => { + held.release(); + evict(held.room); + }; + try { + await plan(held.room, server); + } catch (err) { + release(); + throw err; + } + return release; + }, restoredRuns: (channelId, outcome) => { let room = Rooms.get(channelId); let opened = room?.plan; diff --git a/docs/agent-connection.md b/docs/agent-connection.md index 5b907f325..7b3ab7c3a 100644 --- a/docs/agent-connection.md +++ b/docs/agent-connection.md @@ -1422,9 +1422,13 @@ A returning client attaches, and sends `resume` as usual: sent, once. An answer given before the client is back waits and is sent after `resumed`; one given after goes straight to it. No turn of its own follows, because the question is waited on. A question a client resumes without, because it no longer waits on it, is withdrawn, as - `input-cancel` withdraws it. Each Stop held for the client is sent first. The client's - `runs` for a resumed session update nothing Chat shows until a Planner turn starts a session - of its own. + `input-cancel` withdraws it. Each Stop held for the client is sent first. No turn of the new + process opened a resumed session, yet it is the document's as before: its `runs` are Chat's + run cards, and a document that opens after the client is back shows them live, never stopped, + with what the client last reported. A question it asks after the restart, an `input-request` + on the resumed session, becomes Decisions as one asked before it did, and its answer goes to + that request; the document is opened for it if nobody has it open, and kept open until the + question ends. - **An answer its client is never given.** When the client that asked does not come back to resume (it attaches with nothing to resume, or resumes without the session), an answer a member had given meanwhile cannot reach its request. It reaches the Planner once, as a turn of diff --git a/docs/hosted-agent.md b/docs/hosted-agent.md index 278933a89..98d85081c 100644 --- a/docs/hosted-agent.md +++ b/docs/hosted-agent.md @@ -633,7 +633,10 @@ kept, the server writes each hold to `planner_link_holds` (see [Storage](storage.md#postgresql-schema)): the sessions the client holds, the questions it waits on, the answers it has not acknowledged and the Stops it has not heard. The next process rebuilds each as a suspended hold, so a client that returns and sends `resume` -takes its sessions back. Startup's reset keeps the owner of a document whose hold names a +takes its sessions back. No turn of the new process opened those sessions, so they reach the +document through its room: their run reports are Chat's run cards, shown live, not stopped, when +the document opens after the client is back, and a question asked on one after the restart is a +Decision, for which the document is opened if nobody has it open. Startup's reset keeps the owner of a document whose hold names a browser sign-in that has stored credentials; a paired machine's hold names the machine and no login, which a restart cannot keep: the machine's backend attaches again with its credential, which is stored by digest and so accepted at once, and its new login takes over