diff --git a/CONTEXT.md b/CONTEXT.md index 635bc9cc3..b6c853963 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -192,7 +192,9 @@ with `volli session answer ` naming the rest), so the child's prose never reads as the parent's user (VC-457). The parent is never parked on it, nothing bounds the child's time, and stopping the parent stops its children. A child a person resumes after that notice reports to its parent again when the -resumed turn ends. The Session +resumed turn ends; a turn a host notice opens in it (its own shell exiting, a +Browser Tab event) does not, unless a person or another Session steers a +message into that turn (VC-809). The Session Presentation Contract projects that notice into the parent's transcript as a quiet host-authored receipt row, not as a Turn in the person's voice. This row records the delegation outcome; it is not a listing of the child. While work is @@ -475,7 +477,10 @@ SIGKILL. The three names ride one port, appended last to the vocabulary. A shell belongs to the Session that started it — a subagent never sees its parent's — is capped per Session, runs inside the workspace, is spawned through the same environment record and the same attachment identity the -`execute` tool gets, and is killed when the attachment ends. Every shell +`execute` tool gets, and is killed when the attachment ends — a subagent's +already when a turn of it completes, silently, so a shell it left running +cannot reopen it after its parent has the answer (VC-809); its exited shells' +output is forgotten at that point too. Every shell result restates the Session's live shells, because the tool calls are the durable record; shells are live resources, not ledger facts, and do not survive a relaunch. The Activity Island's shells cluster reads them through diff --git a/packages/agent-runtime/src/pi/shell-tools.ts b/packages/agent-runtime/src/pi/shell-tools.ts index d47f78f4d..964d3c4b3 100644 --- a/packages/agent-runtime/src/pi/shell-tools.ts +++ b/packages/agent-runtime/src/pi/shell-tools.ts @@ -273,7 +273,8 @@ const DESCRIPTIONS: Record = { "Start a command that runs beside the turn instead of holding it: a dev server, a watch build, a long test run, a log tail.", "Returns the shell's id and whatever it printed in its first second, so a server's listening line comes back in the same call.", "When it exits, Volli sends you a notice with how it ended and the end of its output, by itself: do not poll or sleep for it. Optional notifyOn adds one notice the first time output matches, even before a newline.", - "Volli caps how many a Session may hold and kills every one when the Session's attachment ends. Use execute for a command you want to wait on.", + "Volli caps how many a Session may hold and kills every one when the Session's attachment ends. A Subagent Session's shells end silently when its turn completes, so a subagent reads what it needs or waits inside the turn.", + "Use execute for a command you want to wait on.", ].join(" "), shell_output: [ "Read a background shell's output. By default this returns only what is new since the last read of that shell, so polling costs the same each time;", @@ -281,7 +282,7 @@ const DESCRIPTIONS: Record = { ].join(" "), shell_kill: [ "End a background shell: SIGTERM, then SIGKILL if it lingers.", - "Its output stays readable with shell_output until the Session's attachment ends.", + "Its output stays readable with shell_output until the Session's attachment ends, or for a Subagent Session until its turn completes.", ].join(" "), }; diff --git a/packages/host-core/src/agent-tool-door.test.ts b/packages/host-core/src/agent-tool-door.test.ts index fbc1e9164..c140f5626 100644 --- a/packages/host-core/src/agent-tool-door.test.ts +++ b/packages/host-core/src/agent-tool-door.test.ts @@ -1461,6 +1461,7 @@ describe("session_delegate through the Agent Tool Surface (VC-9)", () => { delegate: () => ({ watching: () => false, rearm: async () => undefined, + rearmJoined: async () => undefined, delegate: async (input) => { delegated.push(input); if (input.task.includes("overflow")) { @@ -1725,6 +1726,7 @@ describe("verb result details match the registry (VC-471)", () => { delegate: () => ({ watching: () => false, rearm: async () => undefined, + rearmJoined: async () => undefined, delegate: async (input) => ({ childSessionId: CHILD, handle: CHILD.slice(0, 8), diff --git a/packages/host-core/src/session-runtime/agents.ts b/packages/host-core/src/session-runtime/agents.ts index da896296b..ca2597ae8 100644 --- a/packages/host-core/src/session-runtime/agents.ts +++ b/packages/host-core/src/session-runtime/agents.ts @@ -9,7 +9,7 @@ import { actorSessionTicketDisplay } from "../agent-dispatch/resolution"; import type { AgentToolDoor } from "../agent-tool-door"; import { createHostAgentToolDoor, createHostAgentWatches } from "../agent-services"; import type { Watches } from "../watches"; -import { createDelegations, type Delegations } from "./delegate-session"; +import { commandOriginOf, createDelegations, type Delegations } from "./delegate-session"; import type { TicketSessionDelegationStore } from "./delegation-store"; import type { RuntimeSessionFacade, RuntimeSessionServices } from "./facade"; import { readRecoveredSessionServices, type RecoveredSessionServices } from "./lifecycle"; @@ -73,17 +73,27 @@ export function createRuntimeSessionAgents(options: RuntimeSessionAgentOptions) const store = sessionDelegation; unsubscribeDelegations = sessionWakeBus.subscribe((wake) => { const payload = wake.event.payload; - if (payload.kind !== "turn.started" || created.watching(wake.event.sessionId)) return; + // A turn's start re-arms a settled child; so does a message steered + // into a turn already running (VC-809), which that start never saw. + const joined = + payload.kind === "command.recorded" && payload.command.intent.kind === "message.submit"; + if (payload.kind !== "turn.started" && !joined) return; + if (created.watching(wake.event.sessionId)) return; const entry = store.subagentDelegation(wake.event.sessionId); if (entry === null) return; - const work = created - .rearm(entry, { turnId: payload.turnId, afterSequence: wake.event.sequence }) - .catch((error: unknown) => { - options.log.error("could not re-arm the notice for a resumed subagent", { - childSessionId: entry.childSessionId, - error, - }); + const work = ( + payload.kind === "turn.started" + ? created.rearm(entry, { turnId: payload.turnId, afterSequence: wake.event.sequence }) + : created.rearmJoined(entry, { + origin: commandOriginOf(wake.event), + afterSequence: wake.event.sequence, + }) + ).catch((error: unknown) => { + options.log.error("could not re-arm the notice for a resumed subagent", { + childSessionId: entry.childSessionId, + error, }); + }); liveHost?.detachedWork.track(work); }); } diff --git a/packages/host-core/src/session-runtime/crash-recovery.integration.test.ts b/packages/host-core/src/session-runtime/crash-recovery.integration.test.ts index 5162ec1ff..29a9a1da9 100644 --- a/packages/host-core/src/session-runtime/crash-recovery.integration.test.ts +++ b/packages/host-core/src/session-runtime/crash-recovery.integration.test.ts @@ -33,6 +33,13 @@ afterEach(async () => { const QUESTION = "Which implementation should continue?"; const OPTIONS = [{ id: "first", label: "First implementation" }]; const provenance = { source: { kind: "system" as const, id: "test", detail: null }, venue: null }; +// The runtime's queue admission bound covers EVERY attach (its workspace and +// executor stages) and every cold binding rebuild, not only queued messages. +// A short wall-clock bound on all of them made a merely slow real Pi attach +// under CI load a false `rejected` (VC-778). Genuine work keeps the production +// bound; only a stage this fixture deliberately stalls gets the short one. +const PRODUCTION_ADMISSION_BOUND_MS = 30_000; +const STALLED_STAGE_BOUND_MS = 50; function fixture() { const database = openTestDb(); @@ -63,7 +70,17 @@ function fixture() { attachFailure: Error | null; attachGate: Promise | null; admissionGate: Promise | null; - } = { attachFailure: null, attachGate: null, admissionGate: null }; + /** Real delay before every attach, to model a slow but healthy executor. */ + attachDelayMs: number; + /** Set synchronously on entering a gate, before the runtime arms its timer. */ + stalled: boolean; + } = { + attachFailure: null, + attachGate: null, + admissionGate: null, + attachDelayMs: 0, + stalled: false, + }; const adapter = createPiNativeAdapter({ sessionDataDir, models: script.models, @@ -90,13 +107,20 @@ function fixture() { ...adapter, attach: async (spec, sink) => { attachSpecs.push(spec); - if (control.attachGate) await control.attachGate; + if (control.attachGate) { + control.stalled = true; + await control.attachGate; + } + if (control.attachDelayMs > 0) + await new Promise((resolve) => setTimeout(resolve, control.attachDelayMs)); if (control.attachFailure) throw control.attachFailure; const binding = await adapter.attach(spec, sink); const dispatch = binding.dispatch.bind(binding); binding.dispatch = async (command) => { - if (command.kind === "message.submit" && control.admissionGate) + if (command.kind === "message.submit" && control.admissionGate) { + control.stalled = true; await control.admissionGate; + } return dispatch(command); }; return binding; @@ -105,7 +129,11 @@ function fixture() { followUps: createSqliteSessionFollowUpLedger(database.db), // Explicit recovery wakes exercise retry; no background backoff races. followUpRetryDelaysMs: [], - followUpAdmissionTimeoutMs: 50, + // Read when each bounded stage arms its timer, which is after the + // wrappers above have entered (and flagged) any gate synchronously. + get followUpAdmissionTimeoutMs() { + return control.stalled ? STALLED_STAGE_BOUND_MS : PRODUCTION_ADMISSION_BOUND_MS; + }, onFollowUpFailure: (error) => { failures.push(error); }, @@ -194,7 +222,7 @@ async function start(f: Host, settledPrefix = false) { commandId: `attach-${clock++}`, command: { kind: "adapter.attach", continuity: "fresh" }, }); - expect(attached.receipt?.status).toBe("accepted"); + expect(attached.receipt?.status, JSON.stringify(attached.receipt)).toBe("accepted"); if (settledPrefix) { const settled = await f.host.command({ sessionId, @@ -208,7 +236,7 @@ async function start(f: Host, settledPrefix = false) { }, }, }); - expect(settled.receipt?.status).toBe("accepted"); + expect(settled.receipt?.status, JSON.stringify(settled.receipt)).toBe("accepted"); } const commandId = `submit-${clock++}`; const result = await f.host.command({ @@ -224,7 +252,7 @@ async function start(f: Host, settledPrefix = false) { }, }, }); - expect(result.receipt?.status).toBe("accepted"); + expect(result.receipt?.status, JSON.stringify(result.receipt)).toBe("accepted"); await expect .poll( async () => (await f.host.projection({ sessionId })).projection.interactions.active.length, @@ -254,7 +282,7 @@ async function queue(current: Launch, sessionId: string, commandId = "queue-next }, }, }); - expect(result.receipt?.status).toBe("accepted"); + expect(result.receipt?.status, JSON.stringify(result.receipt)).toBe("accepted"); await current.host.recoverFollowUps(); } @@ -340,6 +368,24 @@ async function assertNoAnswers(current: Launch, sessionId: string) { // The provider alone is scripted. Session admission, SQLite events/queue, // Pi sidecars, boot recovery and the adapter's parked ask are production code. describe("crash recovery through the durable Pi binding (VC-753)", () => { + it("accepts a slow but healthy attach and recovers its binding without a false admission timeout (VC-778)", async () => { + const f = fixture(); + // Longer than the stalled-stage bound: a fixture that bounded genuine + // work by it rejected the fresh attach and failed the boot rebuild. + const slow = STALLED_STAGE_BOUND_MS * 3; + f.control.attachDelayMs = slow; + const { sessionId } = await start(f); + const current = await f.restart(f, true); + current.control.attachDelayMs = slow; + expect(await f.boot(current)).toEqual({ closed: 0, errors: [] }); + expect(current.attachSpecs.map(({ continuity }) => continuity)).toEqual(["native_resume"]); + const recovered = (await current.host.projection({ sessionId })).projection; + expect(recovered.liveExecutor).not.toBeNull(); + expect( + recovered.attention.active.filter(({ kind }) => kind === "adapter_unrecoverable"), + ).toEqual([]); + }); + it("reconciles an accepted command with a different persisted timestamp without a false recovery failure", async () => { const f = fixture(); const { sessionId, commandId } = await start(f); @@ -655,7 +701,7 @@ describe("crash recovery through the durable Pi binding (VC-753)", () => { sessionId, command: { kind: "adapter.release", attachmentId }, }); - expect(released.receipt?.status).toBe("accepted"); + expect(released.receipt?.status, JSON.stringify(released.receipt)).toBe("accepted"); current.control.attachFailure = new Error("Scripted attach unavailable"); await queue(current, sessionId); const projection = (await current.host.projection({ sessionId })).projection; @@ -678,7 +724,7 @@ describe("crash recovery through the durable Pi binding (VC-753)", () => { sessionId, command: { kind: "adapter.attach", continuity: "context_replay" }, }); - expect(attached.receipt?.status).toBe("accepted"); + expect(attached.receipt?.status, JSON.stringify(attached.receipt)).toBe("accepted"); await assertDeliveredOnce(f, current, sessionId); }); @@ -751,7 +797,10 @@ describe("crash recovery through the durable Pi binding (VC-753)", () => { expect.objectContaining({ id: expect.stringContaining("message-delivery:"), kind: "adapter_unrecoverable", - detail: expect.stringContaining("queue_admission_timeout"), + // The stalled stage, not a slow real rebuild before it, is what timed out. + detail: expect.stringContaining( + `queue_admission_timeout: ${stage === "attach" ? "executor attach" : "message admission"} did not finish`, + ), }), ); expect(f.script.requests).toHaveLength(2); @@ -761,6 +810,7 @@ describe("crash recovery through the durable Pi binding (VC-753)", () => { gate.reject(new Error("Stalled fixture transport ended")); current.control.attachGate = null; current.control.admissionGate = null; + current.control.stalled = false; } const relaunched = await f.restart(current); const projection = (await relaunched.host.projection({ sessionId })).projection; diff --git a/packages/host-core/src/session-runtime/delegate-browser-cleanup.integration.test.ts b/packages/host-core/src/session-runtime/delegate-browser-cleanup.integration.test.ts index 88862ce10..3c5d8bbd0 100644 --- a/packages/host-core/src/session-runtime/delegate-browser-cleanup.integration.test.ts +++ b/packages/host-core/src/session-runtime/delegate-browser-cleanup.integration.test.ts @@ -25,6 +25,8 @@ import { insertProject } from "../db/projects-repo"; import { insertTicket } from "../db/tickets-repo"; import { openTestDb, testProject, testTicket, type TestDb } from "../db/test-helpers"; import { createSqliteSessionLedger } from "../session-control/sqlite-ledger"; +import { createAgentShellPort } from "../shell/agent-port"; +import { BackgroundShellHost, type BackgroundShellNotice } from "../shell/background-shell-host"; import { installHostLog } from "../log/root"; import { createDelegations, DELEGATION_ID_SUFFIXES } from "./delegate-session"; import type { DelegateSessionPorts } from "./delegate-session"; @@ -142,11 +144,14 @@ class MemoryBrowser extends BrowserTabRegistry { let db: TestDb | undefined; let runtime: ReturnType | undefined; let unblockLoad: (() => void) | undefined; +let shellHost: BackgroundShellHost | undefined; afterEach(async () => { unblockLoad?.(); unblockLoad = undefined; await runtime?.close(); runtime = undefined; + await shellHost?.close(); + shellHost = undefined; db?.cleanup(); db = undefined; }); @@ -171,7 +176,14 @@ const openPage = (path: string): ScriptedReply => ({ tool: { name: "browser_navigate", args: { url: `https://example.com/${path}` } }, }); -function fixture(role: "project" | "ticket") { +const SHELL_SURFACE: readonly SessionToolId[] = ["shell_start", "shell_output", "shell_kill"]; + +/** `shells`: wire a real background shell host, and offer its tools (VC-809). */ +function fixture(role: "project" | "ticket", options: { shells?: BackgroundShellHost } = {}) { + const { shells } = options; + // In the canonical order: the shell tools sit before `browser_find`. + const surfaceTools: readonly SessionToolId[] = + shells === undefined ? SURFACE : [...SURFACE.slice(0, -1), ...SHELL_SURFACE, "browser_find"]; const database = (db = openTestDb()); const directory = join(dirname(database.dbPath), "workspace"); mkdirSync(directory); @@ -204,6 +216,23 @@ function fixture(role: "project" | "ticket") { }, now: () => clock++, resolveBrowserPort, + ...(shells === undefined + ? {} + : { + resolveShellPort: ( + scope: Parameters< + NonNullable[0]["resolveShellPort"]> + >[0], + ) => + createAgentShellPort({ + host: shells, + scope, + session: scope, + workspacePath: directory, + identity: { sessionId: scope.sessionId, ticketDisplayId: null }, + pathPrefixes: [], + }), + }), resolveRuntimeContext: async (sessionId) => { const { projection } = await host.projection({ sessionId }); const session = projection.session; @@ -266,7 +295,7 @@ function fixture(role: "project" | "ticket") { const sessionId = created.sessionId; await engine.getOrRecordSessionInput({ sessionId, - input: { kind: "tool-surface", tools: SURFACE }, + input: { kind: "tool-surface", tools: surfaceTools }, provenance: { source: { kind: "system", id: "pi-runtime", detail: null }, venue: null }, }); const attached = await host.command({ @@ -533,3 +562,120 @@ describe("completed delegate browser cleanup (VC-754)", () => { }, ); }); + +const startShell = (command: string): ScriptedReply => ({ + tool: { name: "shell_start", args: { command } }, +}); +/** + * The pids `shell_start` reported, one per call, in the order first seen. The + * spawned shell leads its own group, and the pid is known at the spawn — so + * nothing races the turn's end the way a pid file the shell writes would. + */ +function startedPids(requests: ReturnType["script"]["requests"]): number[] { + const byCall = new Map(); + for (const message of requests.flat()) { + if (message.role !== "toolResult") continue; + const match = /Started background shell \S+ \(pid (\d+)\)/.exec( + JSON.stringify(message.content), + ); + if (match !== null && !byCall.has(message.toolCallId)) + byCall.set(message.toolCallId, Number(match[1])); + } + return [...byCall.values()]; +} +/** Whether any process of the group `pid` leads is still alive. */ +function groupAlive(pid: number): boolean { + try { + process.kill(-pid, 0); + return true; + } catch { + return false; + } +} + +// Real processes: the shell host spawns them in their own groups, as in a build. +describe("completed delegate background shell cleanup (VC-809)", () => { + it("kills a delegate's shells with its attachment open, silently, and keeps a project's", async () => { + const notices: BackgroundShellNotice[] = []; + const shells = (shellHost = new BackgroundShellHost({ + publishState: () => undefined, + publishRemoved: () => undefined, + onNotice: (notice) => notices.push(notice), + exitNoticeGraceMs: 10, + settleMs: 50, + killGraceMs: 200, + })); + const f = fixture("project", { shells }); + const parent = await f.start({ + operationId: "parent", + projectId: "project-1", + ticketId: null, + role: "project", + title: "Parent", + }); + const child = await f.start({ + operationId: "child", + projectId: "project-1", + ticketId: null, + role: "subagent", + parentSessionId: parent.sessionId, + title: "Helper", + }); + f.replies.splice( + 0, + f.replies.length, + startShell("exec sleep 300"), + { text: "Parent ready" }, + startShell("exec sleep 300"), + { text: "Child answer" }, + startShell("exec sleep 300"), + { text: "Child again" }, + ); + await f.submitSessionMessage({ + sessionId: parent.sessionId, + commandId: "parent:first", + messageId: "parent:first-message", + text: "Start a server", + }); + await f.submitSessionMessage({ + sessionId: child.sessionId, + commandId: "child:first", + messageId: "child:first-message", + text: "Start a server and check it", + }); + const { projection } = await f.host.projection({ sessionId: child.sessionId }); + expect(projection.lastTurnOutcome).toBe("completed"); + expect(projection.liveExecutor?.status).toBe("open"); + expect(shells.list(child.sessionId)).toEqual([]); + const [parentPid, first] = startedPids(f.script.requests); + expect(first).toBeTypeOf("number"); + await vi.waitFor(() => expect(groupAlive(first!)).toBe(false), { timeout: 5_000 }); + // A Board or Ticket Session's shells outlive its turn, as before. + expect(shells.list(parent.sessionId)).toMatchObject([{ state: "running" }]); + expect(groupAlive(parentPid!)).toBe(true); + + // A person resuming it gets a working shell port, retired again at the end. + await f.submitSessionMessage({ + sessionId: child.sessionId, + commandId: "child:resume", + messageId: "child:resume-message", + text: "Once more", + }); + expect(shells.list(child.sessionId)).toEqual([]); + const second = startedPids(f.script.requests)[2]; + expect(second).toBeTypeOf("number"); + await vi.waitFor(() => expect(groupAlive(second!)).toBe(false), { timeout: 5_000 }); + const toolResults = f.script.requests.flatMap((request) => + request.filter((message) => message.role === "toolResult"), + ); + // Each request carries its Session's whole history, so count calls once. + expect(new Set(toolResults.map((message) => message.toolCallId)).size).toBe(3); + expect(toolResults.every((message) => !message.isError)).toBe(true); + + // Volli's kill is not news: no exit notice reaches the answered child. + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(notices).toEqual([]); + const events = await f.engine.listEvents({ sessionId: child.sessionId }); + expect(events.filter(({ payload }) => payload.kind === "turn.started")).toHaveLength(2); + }); +}); diff --git a/packages/host-core/src/session-runtime/delegate-session.test.ts b/packages/host-core/src/session-runtime/delegate-session.test.ts index 0c9b013c2..c7b71cd37 100644 --- a/packages/host-core/src/session-runtime/delegate-session.test.ts +++ b/packages/host-core/src/session-runtime/delegate-session.test.ts @@ -44,6 +44,7 @@ import type { import { EMPTY_SESSION_USAGE_SUMMARY } from "@volli/shared"; import { + commandOriginOf, createDelegations, DELEGATION_ID_SUFFIXES, resumedNoticeIds, @@ -1343,6 +1344,149 @@ describe("the notice carries the answer (VC-457)", () => { }); }); +describe("rearmJoined — a message steered into a running turn reports (VC-809)", () => { + const entry = { + operationId: `${PARENT}:tc-1`, + parentSessionId: PARENT, + childSessionId: CHILD, + projectId: "project-1", + title: "Find where the auth token is refreshed", + }; + const parentOrigin = { kind: "session", sessionId: PARENT } as const; + /** A settled child mid-way through a turn its own shell's exit notice opened. */ + async function inShellNoticeTurn(h: ReturnType) { + await h.delegate(); + await completeChildTurn(h); + h.projections.set( + CHILD, + projection(CHILD, { + latestTurnId: "t2", + latestTurnOrigin: { kind: "volli", reason: "shell-notice" }, + turnActive: true, + }), + ); + await h.emit(CHILD, 7, { kind: "turn.started", attachmentId: "a", turnId: "t2" }); + await h.delegations.rearm({ ...entry, answered: true }, { turnId: "t2", afterSequence: 7 }); + expect(h.delegations.watching(CHILD)).toBe(false); + } + async function endTurn(h: ReturnType) { + await h.emit(CHILD, 10, { + kind: "transcript.referenced", + attachmentId: "a", + turnId: "t2", + reference: { id: "y", mediaType: "m", digest: "d" }, + }); + await h.emit(CHILD, 11, { kind: "turn.completed", attachmentId: "a", turnId: "t2" }); + } + + it("reports exactly once when the parent steers into a shell-notice turn", async () => { + const h = harness({ + answers: new Map([ + ["x", "first"], + ["y", "joined answer"], + ]), + }); + await inShellNoticeTurn(h); + const told = h.parentCommands().length; + await h.delegations.rearmJoined( + { ...entry, answered: true }, + { origin: parentOrigin, afterSequence: 8 }, + ); + // A second message joining the same turn, and its start's wake landing + // late, arm nothing more. + await h.delegations.rearmJoined( + { ...entry, answered: true }, + { origin: { kind: "user" }, afterSequence: 9 }, + ); + await h.delegations.rearm({ ...entry, answered: true }, { turnId: "t2", afterSequence: 7 }); + expect(h.activeSubscriptions(CHILD)).toHaveLength(1); + await endTurn(h); + + const ids = resumedNoticeIds(entry.operationId, "t2"); + expect(h.parentCommands()).toHaveLength(told + 1); + const resumed = h.commands.filter((command) => command.commandId === ids.commandId); + expect(resumed).toHaveLength(1); + const command = resumed[0]?.command; + const text = + command?.kind === "message.submit" && command.message.parts[0]?.type === "text" + ? command.message.parts[0].text + : ""; + expect(text).toContain(" | joined answer"); + expect(h.delegations.watching(CHILD)).toBe(false); + }); + + it.each([ + ["a host notice", { kind: "volli", reason: "browser-notice" } as const], + ["anything, into an idle child", parentOrigin], + ])("arms nothing for %s", async (who, origin) => { + const h = harness(); + await inShellNoticeTurn(h); + if (who !== "a host notice") { + h.projections.set(CHILD, projection(CHILD, { latestTurnId: "t2", turnActive: false })); + } + const told = h.parentCommands().length; + await h.delegations.rearmJoined({ ...entry, answered: true }, { origin, afterSequence: 8 }); + expect(h.delegations.watching(CHILD)).toBe(false); + await endTurn(h); + expect(h.parentCommands()).toHaveLength(told); + }); + + it("leaves an unanswered delegation to its turn's start", async () => { + const h = harness(); + h.projections.set(CHILD, projection(CHILD, { latestTurnId: "t1", turnActive: true })); + await h.delegations.rearmJoined( + { ...entry, answered: false }, + { origin: parentOrigin, afterSequence: 3 }, + ); + expect(h.delegations.watching(CHILD)).toBe(false); + }); + + it("reports when the running turn cannot be read, and arms nothing", async () => { + const h = harness(); + await inShellNoticeTurn(h); + h.projections.delete(CHILD); + await h.delegations.rearmJoined( + { ...entry, answered: true }, + { origin: parentOrigin, afterSequence: 8 }, + ); + expect(h.delegations.watching(CHILD)).toBe(false); + expect(h.reports).toEqual([expect.stringContaining("could not read subagent")]); + }); +}); + +describe("commandOriginOf", () => { + const recorded = (overrides: Partial): SessionEvent => ({ + ...event(CHILD, 8, { + kind: "command.recorded", + command: { + id: "c", + sessionId: CHILD, + intent: { kind: "message.submit", reference: { id: "m", mediaType: "m", digest: "d" } }, + createdAt: 8, + route: { adapterId: "pi", attachmentId: null }, + }, + } as SessionEvent["payload"]), + ...overrides, + }); + + it("reads the door's attribution from provenance, and prefers the slim field", () => { + expect( + commandOriginOf( + recorded({ + provenance: { + ...PROVENANCE, + source: { ...PROVENANCE.source, detail: { sessionOrigin: { kind: "user" } } }, + }, + }), + ), + ).toEqual({ kind: "user" }); + expect( + commandOriginOf(recorded({ commandOrigin: { kind: "volli", reason: "shell-notice" } })), + ).toEqual({ kind: "volli", reason: "shell-notice" }); + expect(commandOriginOf(recorded({}))).toBeNull(); + }); +}); + describe("rearm — a child a person resumes reports again (VC-457)", () => { const entry = { operationId: `${PARENT}:tc-1`, @@ -1398,8 +1542,79 @@ describe("rearm — a child a person resumes reports again (VC-457)", () => { expect(h.delegations.watching(CHILD)).toBe(false); }); + it.each(["shell-notice", "browser-notice", "watch-notice", "worktree-notice"] as const)( + "does not re-arm for a turn a %s opened in a settled child (VC-809)", + async (reason) => { + const h = harness({ answers: new Map([["x", "first"]]) }); + await h.delegate(); + await completeChildTurn(h); + const told = h.parentCommands().length; + h.projections.set( + CHILD, + projection(CHILD, { + latestTurnId: "t2", + latestTurnOrigin: { kind: "volli", reason }, + }), + ); + await h.delegations.rearm({ ...entry, answered: true }, { turnId: "t2", afterSequence: 10 }); + expect(h.delegations.watching(CHILD)).toBe(false); + expect(h.activeSubscriptions(CHILD)).toEqual([]); + await completeChildTurn(h, CHILD, { from: 11, turnId: "t2", reference: "x" }); + expect(h.parentCommands()).toHaveLength(told); + expect( + h.commands.some((command) => command.commandId.includes(`${entry.operationId}:resume:`)), + ).toBe(false); + }, + ); + + it.each([ + ["a person", { kind: "user" } as const], + ["the parent", { kind: "session", sessionId: PARENT } as const], + ["an unknown origin", null], + ])("re-arms for a turn %s opened", async (_who, origin) => { + const h = harness(); + await h.delegate(); + await completeChildTurn(h); + h.projections.set(CHILD, projection(CHILD, { latestTurnId: "t2", latestTurnOrigin: origin })); + await h.delegations.rearm({ ...entry, answered: true }, { turnId: "t2", afterSequence: 10 }); + expect(h.delegations.watching(CHILD)).toBe(true); + }); + + it("re-arms when a later turn already replaced the one this wake names", async () => { + const h = harness(); + await h.delegate(); + await completeChildTurn(h); + h.projections.set( + CHILD, + projection(CHILD, { + latestTurnId: "t3", + latestTurnOrigin: { kind: "volli", reason: "shell-notice" }, + }), + ); + await h.delegations.rearm({ ...entry, answered: true }, { turnId: "t2", afterSequence: 10 }); + expect(h.delegations.watching(CHILD)).toBe(true); + }); + + it("re-arms when what opened the turn cannot be read, and says so", async () => { + const h = harness(); + await h.delegate(); + await completeChildTurn(h); + h.projections.delete(CHILD); + await h.delegations.rearm({ ...entry, answered: true }, { turnId: "t2", afterSequence: 10 }); + expect(h.delegations.watching(CHILD)).toBe(true); + expect(h.reports).toEqual([expect.stringContaining("could not read what opened subagent")]); + }); + it("answers under the delegation's own ids when its notice never landed", async () => { const h = harness(); + // Unanswered, so even a host notice's turn reports: the parent was never told. + h.projections.set( + CHILD, + projection(CHILD, { + latestTurnId: "t1", + latestTurnOrigin: { kind: "volli", reason: "shell-notice" }, + }), + ); await h.delegations.rearm({ ...entry, answered: false }, { turnId: "t1", afterSequence: 3 }); // A kickoff receipt from the original watch is irrelevant to resumed work. await h.emit(CHILD, 4, kickoffRejection(`${PARENT}:tc-1:kickoff`)); diff --git a/packages/host-core/src/session-runtime/delegate-session.ts b/packages/host-core/src/session-runtime/delegate-session.ts index 6e9046c90..4d0044a81 100644 --- a/packages/host-core/src/session-runtime/delegate-session.ts +++ b/packages/host-core/src/session-runtime/delegate-session.ts @@ -109,6 +109,7 @@ import { sessionStopSummary, type SessionStopDetail } from "@volli/shared"; import { + readSessionOrigin, sessionHostNoticeMetadata, shortSessionId, SUBAGENT_NOTICE_MESSAGE_ID_SUFFIX, @@ -118,6 +119,7 @@ import type { ModelSelection, RuntimeSessionIdentity, SessionEvent, + SessionOrigin, SessionStopActor, SubagentNoticeReason, SubagentNoticeState, @@ -398,6 +400,40 @@ function outcomeOf( export type { NoticeDelivery } from "./host-notice-delivery"; +/** + * Which of Volli's own origins are a host notice — Volli telling the Session + * something, not anyone resuming it (VC-809). A Record, so a new reason fails + * to compile here until someone decides which it is. + */ +const IS_HOST_NOTICE: Record["reason"], boolean> = { + "watch-notice": true, + "subagent-notice": true, + "browser-notice": true, + "shell-notice": true, + "worktree-notice": true, + "relaunch-recovery": false, + "scheduled-resume": false, + supervision: false, + "auto-title": false, +}; + +/** A turn Volli opened by telling the Session something, not by anyone resuming it. */ +function isHostNoticeOrigin(origin: SessionOrigin | null): boolean { + return origin?.kind === "volli" && IS_HOST_NOTICE[origin.reason]; +} + +/** + * The origin a `command.recorded` event carries: the slim fold's field when + * present, else the door's attribution in its provenance. Null when unknown. + */ +export function commandOriginOf(event: SessionEvent): SessionOrigin | null { + if (event.commandOrigin !== undefined) return event.commandOrigin; + const detail = event.provenance?.source.detail; + return typeof detail === "object" && detail !== null && !Array.isArray(detail) + ? readSessionOrigin((detail as Record).sessionOrigin) + : null; +} + /** * The ids a resumed child's notice spends, keyed on the turn that resumed it * so each resumption lands one notice. DURABLE, not live, like the kickoff's. @@ -425,12 +461,26 @@ export interface Delegations { /** * Watch a child whose delegation already settled, from the turn a person * resumed it with (VC-457), so that turn's end notifies the parent too. - * Idempotent per child while a watch is live. + * A turn a host notice opened in an answered child is not a resumption and + * arms nothing (VC-809). Idempotent per child while a watch is live. */ rearm( entry: DelegationRef & { answered: boolean }, resumed: { turnId: string; afterSequence: number }, ): Promise; + /** + * Watch an answered child that a person or another Session steered a + * message into while a turn of it was already running (VC-809). That turn + * was opened by something else — a host notice, typically, which + * {@link rearm} leaves alone — so its start named no resumption; the joined + * message is one, and the turn's end reports under that turn's resume ids. + * A message whose origin is a host notice, or that arrived while the child + * was idle (its own turn's start re-arms), arms nothing. + */ + rearmJoined( + entry: DelegationRef & { answered: boolean }, + joined: { origin: SessionOrigin | null; afterSequence: number }, + ): Promise; /** * Notify every delegation a relaunch left unanswered, off the children's own * ledgers. Run once at boot, after stale attachments are retired; the notice @@ -471,6 +521,23 @@ export function createDelegations(ports: DelegateSessionPorts): Delegations { } } + /** + * Whether the child's turn `turnId` was opened by a host notice (VC-809): + * true only when that turn is still the latest and its recorded origin is + * one. A projection that cannot be read is reported and answers false. + */ + async function openedByHostNotice(childSessionId: string, turnId: string): Promise { + try { + const { projection } = await ports.runtime.projection({ sessionId: childSessionId }); + return projection.latestTurnId === turnId && isHostNoticeOrigin(projection.latestTurnOrigin); + } catch (error) { + report( + `could not read what opened subagent ${shortSessionId(childSessionId)}'s turn: ${errorText(error)}`, + ); + return false; + } + } + /** * The one delivery. The outcome state rides along so the message can carry * it as metadata for the chat surface (VC-330), and the child's answer rides @@ -740,6 +807,14 @@ export function createDelegations(ports: DelegateSessionPorts): Delegations { return live.has(childSessionId); }, async rearm({ answered, ...entry }, resumed) { + if (live.has(entry.childSessionId)) return; + // Only a turn someone OPENED reports again (VC-809). A host notice the + // child is told — its own shell exiting, a Browser Tab event — opens a + // turn too, but nobody resumed the child, and the parent already has + // its answer. The origin is read only when this wake's turn is still + // the latest; otherwise, or when it cannot be read, it stays unknown + // and the child reports, as before. + if (answered && (await openedByHostNotice(entry.childSessionId, resumed.turnId))) return; if (live.has(entry.childSessionId)) return; // A delegation whose own notice never landed — a child a person retried // after its first attach failed — answers under the delegation's ids, so @@ -753,6 +828,30 @@ export function createDelegations(ports: DelegateSessionPorts): Delegations { true, ); }, + async rearmJoined({ answered, ...entry }, joined) { + // An unanswered delegation re-arms on any turn's start already. + if (!answered || isHostNoticeOrigin(joined.origin)) return; + if (live.has(entry.childSessionId)) return; + let turnId: string | null; + try { + const { projection } = await ports.runtime.projection({ sessionId: entry.childSessionId }); + turnId = projection.turnActive ? projection.latestTurnId : null; + } catch (error) { + report( + `could not read subagent ${shortSessionId(entry.childSessionId)}'s running turn: ${errorText(error)}`, + ); + return; + } + if (turnId === null || live.has(entry.childSessionId)) return; + // Keyed on the joined turn, so a racing `rearm` for the same turn + // spends the same ids and the parent is told once. + await watchChild( + entry, + joined.afterSequence, + resumedNoticeIds(entry.operationId, turnId), + true, + ); + }, async recover(unanswered) { const recovery: DelegationRecovery = { answered: 0, reported: 0, skipped: 0 }; for (const entry of unanswered) { diff --git a/packages/host-core/src/session-runtime/facade.test.ts b/packages/host-core/src/session-runtime/facade.test.ts index c8df0b8db..90ce84355 100644 --- a/packages/host-core/src/session-runtime/facade.test.ts +++ b/packages/host-core/src/session-runtime/facade.test.ts @@ -34,7 +34,10 @@ vi.mock("./sessions", async (original) => ({ })); vi.mock("./auto-title", () => ({ createAutoTitler: vi.fn() })); vi.mock("../session-control/peek-summary", () => ({ createPeekSummarizer: vi.fn() })); -vi.mock("./delegate-session", () => ({ createDelegations: vi.fn() })); +vi.mock("./delegate-session", async (importOriginal) => ({ + ...(await importOriginal()), + createDelegations: vi.fn(), +})); vi.mock("../db/projects-repo", () => ({ getProjectById: vi.fn(), listProjects: vi.fn(() => []), @@ -92,6 +95,7 @@ async function fixture() { recover: vi.fn(), watching: vi.fn(), rearm: vi.fn(async () => {}), + rearmJoined: vi.fn(async () => {}), } as never); const sessions = { create: vi.fn() } as unknown as Sessions; const titler = { refine: vi.fn(async () => {}) } as AutoTitler; @@ -356,6 +360,7 @@ describe("lifted Session facade and recovered agent staging", () => { liveChildren: vi.fn(() => ["child"]), watching: vi.fn(() => false), rearm: vi.fn(async () => {}), + rearmJoined: vi.fn(async () => {}), }; vi.mocked(createDelegations).mockReturnValue( delegated as unknown as ReturnType, @@ -406,6 +411,29 @@ describe("lifted Session facade and recovered agent staging", () => { turnId: "turn", afterSequence: 12, }); + // A message steered into a running turn re-arms too (VC-809), with the + // origin its door recorded; any other command does not. + vi.mocked(f.delegation.subagentDelegation).mockReturnValueOnce(entry); + listener({ + event: { + sessionId: "child", + sequence: 13, + commandOrigin: { kind: "user" }, + payload: { kind: "command.recorded", command: { intent: { kind: "message.submit" } } }, + }, + } as never); + expect(delegated.rearmJoined).toHaveBeenCalledExactlyOnceWith(entry, { + origin: { kind: "user" }, + afterSequence: 13, + }); + listener({ + event: { + sessionId: "child", + sequence: 14, + payload: { kind: "command.recorded", command: { intent: { kind: "context.compact" } } }, + }, + } as never); + expect(delegated.rearmJoined).toHaveBeenCalledOnce(); // The door's mutations reach the staging's own event bus. expect(f.agentServices.createToolDoor.mock.calls[0]![0]).toEqual({ events: f.events }); const toolPorts = f.agentServices.createToolDoor.mock.calls[0]![1]; @@ -668,6 +696,7 @@ describe("agent staging ports", () => { liveChildren: vi.fn(() => ["child"]), watching: vi.fn(() => false), rearm: vi.fn(async () => {}), + rearmJoined: vi.fn(async () => {}), }; vi.mocked(createDelegations).mockReturnValue(delegated as never); const runner = { run: vi.fn(async () => "run") }; diff --git a/packages/host-core/src/session-runtime/pi-adapter.ts b/packages/host-core/src/session-runtime/pi-adapter.ts index 342238d3f..7fd405f73 100644 --- a/packages/host-core/src/session-runtime/pi-adapter.ts +++ b/packages/host-core/src/session-runtime/pi-adapter.ts @@ -136,8 +136,11 @@ import type { UIMessage } from "ai"; import type { SessionWebPorts } from "../web/ports"; import { readWorkspaceEnvironment } from "../session-env"; import type { TurnAttachments } from "../turn-attachments"; +import { hostLogger } from "../log/root"; import { STRUCTURED_ADAPTER_ID } from "./sessions"; +const log = hostLogger("pi-adapter"); + /** * The one adapter id. Pi is the structured product's single target executor. * @@ -268,7 +271,10 @@ export type AttachmentBrowserPort = RuntimeBrowserPort & { * The runtime's shell port with the one lifecycle door the adapter drives * (VC-270): the attachment ending. Required rather than optional, for * {@link AttachmentBrowserPort}'s reason — it is what keeps a Session's shells - * from outliving it. + * from outliving it. A completed Subagent turn drives the same door early + * (VC-809): every shell the child owns, running or exited, is killed and + * forgotten (its removal published), and the port keeps serving the + * attachment for a turn a person resumes it with. */ export type AttachmentShellPort = RuntimeShellPort & { dispose: () => void }; @@ -1978,6 +1984,17 @@ class PiBinding implements BindingHandle { // Capture at this boundary, before stream delivery can resume the child. if (this.#context.role === "subagent" && observation.state === "completed") { this.#browser?.turnEnded({ closeHeadlessOwned: true }); + // Its background shells end here too (VC-809), killed as Volli's so + // no exit notice follows: a dev server left running would otherwise + // reopen the answered child when it exits, long after the parent was + // told. The port stays usable for a turn a person resumes it with. + // Like the tab cleanup, a failure here must not cost the delegate its + // answer: it is logged, and the attachment's end retries the kill. + try { + this.#shell?.dispose(); + } catch (error) { + log.warn("completed delegate background shell cleanup failed", { error }); + } } else { this.#browser?.turnEnded(); } diff --git a/packages/host-core/src/shell/background-shell-host.ts b/packages/host-core/src/shell/background-shell-host.ts index 73d0abeda..e258b4e3d 100644 --- a/packages/host-core/src/shell/background-shell-host.ts +++ b/packages/host-core/src/shell/background-shell-host.ts @@ -21,9 +21,9 @@ * {@link SHELL_OUTPUT_MAX_BYTES}; a read cursor, so a read returns only what * arrived since the last one and a poll loop costs the same context every * time; `startedAt`, `exitedAt`, the exit code and the signal. An exited - * shell's record and tail stay readable until its Session's attachment ends, - * so a model that comes back to a finished build can still read how it - * finished. + * shell's record and tail stay readable until its Session's attachment ends — + * a Subagent Session's only until its turn completes (VC-809) — so a model + * that comes back to a finished build can still read how it finished. * * WHAT IT REFUSES. The per-Session cap, a shell another Session owns (unknown * rather than forbidden — a shell this Session cannot touch is one it was @@ -56,6 +56,8 @@ * {@link SHELL_KILL_GRACE_MS}. {@link disposeSession} is the attachment's end * — stop, done, detach, replace, relaunch, all through the adapter's one * release path — and kills every shell the Session started and forgets them. + * A Subagent Session's completed turn calls it too (VC-809), so a shell it + * left running cannot reopen it after its parent has the answer. * Shells are live resources, not ledger facts: nothing here survives a * relaunch, and nothing is written to the database. The record flips to * `exited` on the child's `close`, not its `exit` — the two are not ordered @@ -1022,8 +1024,8 @@ export class BackgroundShellHost { } /** - * The attachment's end: kill every shell the Session started and forget - * them. Fire-and-forget on the kill, because release must not wait on a + * The attachment's end — or a Subagent Session's completed turn (VC-809): + * kill every shell the Session started and forget them. Fire-and-forget on the kill, because release must not wait on a * process that ignores SIGTERM — the SIGKILL still follows after the grace. * * A Session that has ended is told nothing (VC-495): an exit notice still