diff --git a/CHANGELOG.md b/CHANGELOG.md index 055032298..1fb8b5808 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- A session's `session_shutdown` now disposes only the children that session launched. Before, the extension's shutdown handler disposed every live in-process child through the process-wide factory, so in a multi-session host such as pi-web, one idle session being reaped aborted another session's running subagents mid-flight: the child died with `Request was aborted` while its parent session kept running. The detached runner's whole-process exit path still disposes everything. Thanks to [@Jarcis-cy](https://github.com/Jarcis-cy) for [#3204](https://github.com/nicobailon/pi-subagents/pull/3204). + ## [0.86.0] - 2026-11-05 ### Highlights diff --git a/src/extension/index.ts b/src/extension/index.ts index a6f04c741..f2345e4fa 100644 --- a/src/extension/index.ts +++ b/src/extension/index.ts @@ -60,7 +60,7 @@ import registerSubagentNotify from "../runs/background/notify.ts"; import { createParentWake } from "../shared/parent-wake.ts"; import { handleSubagentSteeringNotice, type SubagentSteeringMessageDetails } from "./steering-notices.ts"; import { SUBAGENT_CHILD_ENV, SUBAGENT_PARENT_SESSION_ENV } from "../runs/shared/child-runtime-config.ts"; -import { disposeChildSessions } from "../runs/shared/child-session.ts"; +import { disposeChildSessionsForParent } from "../runs/shared/child-session.ts"; import { resolveCurrentSubagentCapabilityCeiling } from "../runs/shared/capability-ceiling.ts"; import { loadConfig, resolveAsyncByDefault, resolveScheduledStoreRoot } from "./config.ts"; import { buildSubagentToolDescription, buildSubagentToolPromptMetadata } from "./tool-description.ts"; @@ -1258,13 +1258,18 @@ export default function registerSubagentExtension(pi: ExtensionAPI): void { }); pi.on("session_shutdown", async (event) => { + // Captured first: runtimeEntry.cleanup() below clears state.parentSessionFile, and the + // disposal must stay scoped to this session even while its own state is being torn down. + const parentSessionFile = state.parentSessionFile ?? undefined; sessionSwitchGuard.sessionShutdown(event); cancelCompactionResume(); completionNotifier.sessionShutdown(event?.reason); parentWake.sessionShutdown(event?.reason); runtimeEntry.cleanup(event?.reason); try { - await disposeChildSessions(); + // Only this session's children: a multi-session host (pi-web) shuts sessions down + // one at a time, and one session's shutdown must not abort another session's runs. + await disposeChildSessionsForParent(parentSessionFile); } catch (error) { console.error("Failed to dispose in-process child sessions:", error); } diff --git a/src/runs/shared/child-session.ts b/src/runs/shared/child-session.ts index 42f46a7e0..197b68418 100644 --- a/src/runs/shared/child-session.ts +++ b/src/runs/shared/child-session.ts @@ -153,6 +153,13 @@ export interface ChildSessionFactory { create(launch: ChildSessionLaunch): Promise; /** Abort and dispose every live attached child; detached children keep running and hold the shared runtime. */ dispose(): Promise; + /** + * Abort and dispose only the attached children that `parentSessionFile`'s session launched. + * A multi-session host such as pi-web shuts sessions down one at a time; one session's + * shutdown must not destroy another session's running children. Children of sessions + * without a session file are disposed when `parentSessionFile` is undefined. + */ + disposeForParent(parentSessionFile: string | undefined): Promise; } export type PiCodingAgentModule = typeof import("@earendil-works/pi-coding-agent"); @@ -489,12 +496,17 @@ export function createDefaultChildSessionFactory(options: DefaultChildSessionFac const loadPiCodingAgent = options.loadPiCodingAgent ?? loadHostPiCodingAgent; const shutdownTimeoutMs = options.shutdownTimeoutMs ?? 5_000; let runtime: ReturnType | undefined; - const live = new Set(); + /** Live children keyed by the session file of the parent that launched them; undefined for a parent without one. */ + const live = new Map(); /** Extension shutdowns still running for disposed children; `dispose()` waits for them. */ const shutdowns = new Set>(); /** Launches still waiting for their MCP tools; `dispose()` ends the wait and waits for their shutdown. */ - const mcpWaits = new Map>(); - let disposals = 0; + const mcpWaits = new Map; parent: string | undefined }>(); + let globalDisposals = 0; + /** Per-parent disposal epochs: a child's launch aborts only when its own parent's scope (or the whole process) was disposed while it started. */ + const parentDisposals = new Map(); + const parentDisposalEpoch = (parent: string | undefined) => parentDisposals.get(parent) ?? 0; + const bumpParentDisposal = (parent: string | undefined) => parentDisposals.set(parent, parentDisposalEpoch(parent) + 1); const sharedRuntime = async (pi: PiCodingAgentModule) => { runtime ??= pi.ModelRuntime.create().catch((error: unknown) => { runtime = undefined; @@ -502,9 +514,27 @@ export function createDefaultChildSessionFactory(options: DefaultChildSessionFac }); return runtime; }; + /** Abort and dispose the attached children `matches` selects, waiting for their shutdowns. */ + const disposeMatching = async (matches: (parent: string | undefined) => boolean, bump: () => void) => { + bump(); + for (const [cancel, wait] of mcpWaits) if (matches(wait.parent)) cancel.abort(); + const children = [...live.keys()].filter((child) => !child.detached && matches(live.get(child))); + for (const child of children) child.shutDown = true; + await Promise.allSettled(children.map((child) => child.abort())); + for (const child of children) { + try { void child.dispose(); } catch { /* best effort */ } + } + await Promise.allSettled([ + ...shutdowns, + ...[...mcpWaits.values()].filter((wait) => matches(wait.parent)).map((wait) => wait.ready), + ]); + if (live.size === 0) runtime = undefined; + }; return { async create(launch) { - const disposalsAtStart = disposals; + const globalDisposalsAtStart = globalDisposals; + const parentEpochAtStart = parentDisposalEpoch(launch.parentSessionFile); + const disposedSinceStart = () => globalDisposals !== globalDisposalsAtStart || parentDisposalEpoch(launch.parentSessionFile) !== parentEpochAtStart; const pi = await loadPiCodingAgent(); const builtinMcpTools = launch.builtinMcpTools ?? []; let readinessTool: ToolDefinition | undefined; @@ -688,11 +718,11 @@ export function createDefaultChildSessionFactory(options: DefaultChildSessionFac if (builtinMcp) { const cancel = new AbortController(); const signal = launch.signal ? AbortSignal.any([cancel.signal, launch.signal]) : cancel.signal; - if (disposals !== disposalsAtStart) cancel.abort(); + if (disposedSinceStart()) cancel.abort(); const ready = (async () => { try { await builtinMcp.waitForTools(signal, () => loader.getExtensions().errors.find(({ path }) => path === "builtin:mcp")?.error); - if (disposals !== disposalsAtStart) throw new Error("The child session factory was disposed while the child waited for its MCP tools."); + if (disposedSinceStart()) throw new Error("The child session factory was disposed while the child waited for its MCP tools."); if (signal.aborted) throw new Error("The child launch was cancelled while waiting for its MCP tools."); const missing = missingBuiltinMcpTools(session, builtinMcpTools.map(({ name }) => name)); if (missing.length) throw new Error(`Selected built-in MCP tools did not register in the child session: ${missing.join(", ")}. The MCP server may have failed to connect, or the name belongs to a different tool than the selector names; check it with /mcp.`); @@ -701,7 +731,7 @@ export function createDefaultChildSessionFactory(options: DefaultChildSessionFac throw error; } })(); - mcpWaits.set(cancel, ready); + mcpWaits.set(cancel, { ready, parent: launch.parentSessionFile }); try { await ready; } finally { mcpWaits.delete(cancel); } } const child: ChildSession = { @@ -731,20 +761,14 @@ export function createDefaultChildSessionFactory(options: DefaultChildSessionFac get virtualModelId() { return session.model?.api === "pi-virtual" ? `${session.model.provider}/${session.model.id}` : undefined; }, get contextWindow() { return session.model?.contextWindow; }, }; - live.add(child); + live.set(child, launch.parentSessionFile); return child; }, async dispose() { - disposals += 1; - for (const cancel of mcpWaits.keys()) cancel.abort(); - const children = [...live].filter((child) => !child.detached); - for (const child of children) child.shutDown = true; - await Promise.allSettled(children.map((child) => child.abort())); - for (const child of children) { - try { void child.dispose(); } catch { /* best effort */ } - } - await Promise.allSettled([...shutdowns, ...mcpWaits.values()]); - if (live.size === 0) runtime = undefined; + await disposeMatching(() => true, () => { globalDisposals += 1; }); + }, + async disposeForParent(parentSessionFile: string | undefined) { + await disposeMatching((parent) => parent === parentSessionFile, () => bumpParentDisposal(parentSessionFile)); }, }; } @@ -764,6 +788,7 @@ function createLazyPlacementFactory(local: ChildSessionFactory): ChildSessionFac return { async create(launch) { return launch.machine ? (await factory()).create(launch) : local.create(launch); }, async dispose() { if (placed) await placed.dispose(); else await local.dispose(); }, + async disposeForParent(parentSessionFile: string | undefined) { if (placed) await placed.disposeForParent(parentSessionFile); else await local.disposeForParent(parentSessionFile); }, }; } @@ -793,9 +818,27 @@ export function setChildSessionFactoryModule(modulePath: string | undefined): vo activeFactoryModule = modulePath; } -/** Abort and dispose every live in-process child and release the shared runtime. */ +/** + * Abort and dispose live in-process children and release the shared runtime. + * + * Without `parentSessionFile`, every live attached child is disposed: the whole-process + * path the detached runner uses at exit. With it, only the children that session + * launched are disposed: a multi-session host such as pi-web shuts sessions down one + * at a time, and one session's shutdown must not destroy another session's children. + */ export async function disposeChildSessions(): Promise { const factory = activeFactory; if (!factory) return; await factory.dispose(); } + +/** + * Abort and dispose only the children `parentSessionFile`'s session launched. + * `undefined` names the children of a parent without a session file; it stays scoped, + * unlike disposeChildSessions(), which is the detached runner's whole-process exit path. + */ +export async function disposeChildSessionsForParent(parentSessionFile: string | undefined): Promise { + const factory = activeFactory; + if (!factory) return; + await factory.disposeForParent(parentSessionFile); +} diff --git a/src/runs/shared/herdr-placed-run.ts b/src/runs/shared/herdr-placed-run.ts index 6014c62e8..b1330b248 100644 --- a/src/runs/shared/herdr-placed-run.ts +++ b/src/runs/shared/herdr-placed-run.ts @@ -172,5 +172,5 @@ async function startPlacedPiSession(launch: ChildSes function createPlacedPiSession(launch: ChildSessionLaunch): Promise { return launch.machine?.provider === "rex" ? createRexPiSession(launch) : createHerdrPiSession(launch); } export function createPlacementAwareChildSessionFactory(local: ChildSessionFactory, createRemote: (launch: ChildSessionLaunch) => Promise = createPlacedPiSession): ChildSessionFactory { - const remote = new Set(); return { async create(launch) { if (!launch.machine) return local.create(launch); serializeHerdrPiLaunch(launch); const child = await createRemote(launch); remote.add(child); return child; }, async dispose() { await Promise.allSettled([...remote].map((child) => child.dispose())); remote.clear(); await local.dispose(); } }; + const remote = new Map(); return { async create(launch) { if (!launch.machine) return local.create(launch); serializeHerdrPiLaunch(launch); const child = await createRemote(launch); remote.set(child, launch.parentSessionFile); return child; }, async dispose() { await Promise.allSettled([...remote.keys()].map((child) => child.dispose())); remote.clear(); await local.dispose(); }, async disposeForParent(parentSessionFile: string | undefined) { const matching = [...remote].filter(([, parent]) => parent === parentSessionFile).map(([child]) => child); await Promise.allSettled(matching.map((child) => child.dispose())); for (const child of matching) remote.delete(child); await local.disposeForParent(parentSessionFile); } }; } diff --git a/test/unit/child-session-scoped-dispose.test.ts b/test/unit/child-session-scoped-dispose.test.ts new file mode 100644 index 000000000..433b0c881 --- /dev/null +++ b/test/unit/child-session-scoped-dispose.test.ts @@ -0,0 +1,120 @@ +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; +import { + createDefaultChildSessionFactory, + type ChildSessionLaunch, + type PiCodingAgentModule, +} from "../../src/runs/shared/child-session.ts"; + +/** + * A multi-session host such as pi-web runs several parent sessions in one process. + * Its idle reaper shuts wrappers down one at a time, each shutdown emitting + * `session_shutdown` to that session's extension instance. The extension must dispose + * only that session's children: one session's shutdown must not abort another + * session's running children. + */ + +interface SessionLog { + parent: string | undefined; + events: string[]; +} + +function stubPiModule(logs: SessionLog[]): () => PiCodingAgentModule { + return () => { + const log: SessionLog = { parent: undefined, events: [] }; + logs.push(log); + return { + ModelRuntime: { create: async () => ({ refresh: async () => {} }) }, + SettingsManager: { create: () => ({}) }, + DefaultResourceLoader: class { async reload() {} getExtensions() { return { extensions: [], errors: [], runtime: { pendingProviderRegistrations: [], pendingNativeProviderRegistrations: [] } }; } }, + SessionManager: { inMemory: () => ({}) }, + resolveCliModel: () => ({}), + createAgentSession: async () => ({ + session: { + bindExtensions: async () => {}, + dispose() { log.events.push("dispose"); }, + extensionRunner: { hasHandlers: () => false }, + subscribe: () => () => {}, + prompt: async () => {}, + abort: async () => { log.events.push("abort"); }, + steer: async () => {}, + followUp: async () => {}, + messages: [], + sessionId: `child-${logs.length}`, + }, + }), + } as unknown as PiCodingAgentModule; + }; +} + +function launchFor(parentSessionFile: string | undefined, cwd: string): ChildSessionLaunch { + return { + cwd, + storage: { kind: "memory" }, + parentSessionFile, + extensionPaths: [], + ambientExtensions: false, + hooks: [], + noSkills: true, + noContextFiles: true, + runtime: { fanoutChild: false, depth: 1, waitTool: { enabled: false }, fast: false }, + }; +} + +describe("disposeForParent scopes child disposal to one session", () => { + it("disposes only the parent's own children, leaving other sessions' children running", async () => { + const logs: SessionLog[] = []; + const factory = createDefaultChildSessionFactory({ loadPiCodingAgent: stubPiModule(logs) }); + const parentA = "/tmp/sessions/parent-a.jsonl"; + const parentB = "/tmp/sessions/parent-b.jsonl"; + const childA1 = await factory.create(launchFor(parentA, process.cwd())); + await factory.create(launchFor(parentB, process.cwd())); + const childA1Abort = childA1.abort.bind(childA1); + + await factory.disposeForParent(parentA); + + assert.deepEqual(logs[0]?.events, ["abort", "dispose"], "parent A's child is aborted and disposed"); + assert.deepEqual(logs[1]?.events, [], "parent B's child is untouched"); + // The surviving child still works and can be disposed on its own session's shutdown. + await assert.doesNotReject(childA1Abort); + }); + + it("disposes only children of file-less parents when the parent session has no file", async () => { + const logs: SessionLog[] = []; + const factory = createDefaultChildSessionFactory({ loadPiCodingAgent: stubPiModule(logs) }); + await factory.create(launchFor("/tmp/sessions/parent-a.jsonl", process.cwd())); + await factory.create(launchFor(undefined, process.cwd())); + + await factory.disposeForParent(undefined); + + assert.deepEqual(logs[0]?.events, [], "the file-backed parent's child is untouched"); + assert.deepEqual(logs[1]?.events, ["abort", "dispose"], "the file-less parent's child is disposed"); + }); + + it("dispose() still disposes every live child", async () => { + const logs: SessionLog[] = []; + const factory = createDefaultChildSessionFactory({ loadPiCodingAgent: stubPiModule(logs) }); + await factory.create(launchFor("/tmp/sessions/parent-a.jsonl", process.cwd())); + await factory.create(launchFor("/tmp/sessions/parent-b.jsonl", process.cwd())); + await factory.create(launchFor(undefined, process.cwd())); + + await factory.dispose(); + + for (const log of logs) { + assert.deepEqual(log.events, ["abort", "dispose"], "every child is aborted and disposed"); + } + }); + + it("detached children keep running across both disposal scopes", async () => { + const logs: SessionLog[] = []; + const factory = createDefaultChildSessionFactory({ loadPiCodingAgent: stubPiModule(logs) }); + const parentA = "/tmp/sessions/parent-a.jsonl"; + const child = await factory.create(launchFor(parentA, process.cwd())); + child.detached = true; + + await factory.disposeForParent(parentA); + await factory.dispose(); + + assert.deepEqual(logs[0]?.events, [], "a detached child is never aborted by disposal"); + }); +}); diff --git a/test/unit/extension-session-shutdown-scope.test.ts b/test/unit/extension-session-shutdown-scope.test.ts new file mode 100644 index 000000000..a39cebcec --- /dev/null +++ b/test/unit/extension-session-shutdown-scope.test.ts @@ -0,0 +1,89 @@ +import assert from "node:assert/strict"; +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import { spawnSync } from "node:child_process"; +import { describe, it } from "node:test"; + +/** + * A session's session_shutdown must dispose only that session's children, and it must + * still know its own parent session file when it does: runtimeEntry.cleanup() clears + * state.parentSessionFile inside the same handler, so the file has to be captured + * before cleanup runs. Drives the real extension's session_start and session_shutdown + * handlers against a scripted factory and asserts the scoped call. + */ + +const script = String.raw` + const handlers = new Map(); + const calls = []; + const fs = await import("node:fs"); + const os = await import("node:os"); + const path = await import("node:path"); + const projectRoot = fs.mkdtempSync(path.join(os.tmpdir(), "pi-shutdown-scope-")); + console.error = () => {}; + const { default: registerSubagentExtension } = await import("./src/extension/index.ts"); + const { setChildSessionFactory } = await import("./src/runs/shared/child-session.ts"); + const { createEventBus } = await import("@earendil-works/pi-coding-agent"); + const events = createEventBus(); + const parentSessionFile = path.join(projectRoot, "parent-session.jsonl"); + const scripted = { + async create() { throw new Error("scripted factory: create must not run in this test"); }, + async dispose() { calls.push(["dispose"]); }, + async disposeForParent(parent) { calls.push(["disposeForParent", parent]); }, + }; + setChildSessionFactory(scripted); + try { + const pi = new Proxy({ + events, + on(name, handler) { const list = handlers.get(name) ?? []; list.push(handler); handlers.set(name, list); }, + registerTool() {}, registerCommand() {}, registerShortcut() {}, registerMessageRenderer() {}, getSessionName() {}, + sendMessage() {}, + }, { get(target, property) { return property in target ? target[property] : () => undefined; } }); + const ctx = { + cwd: projectRoot, hasUI: false, model: undefined, isIdle() { return false; }, + ui: { setWidget() {}, requestRender() {}, theme: { fg(_name, text) { return text; }, bg(_name, text) { return text; }, bold(text) { return text; } } }, + sessionManager: { + getSessionId() { return "shutdown-scope-session"; }, + getSessionFile() { return parentSessionFile; }, + getEntries() { return []; }, + }, + modelRegistry: { getAvailable() { return []; } }, + }; + registerSubagentExtension(pi); + for (const handler of handlers.get("session_start") ?? []) await handler({ reason: "startup" }, ctx); + for (const handler of handlers.get("session_shutdown") ?? []) await handler({ reason: "quit" }, ctx); + console.log(JSON.stringify(calls)); + } finally { + setChildSessionFactory(undefined); + } +`; + +function runScript(root: string): Array<[string, unknown?]> { + const result = spawnSync(process.execPath, ["--experimental-strip-types", "--input-type=module", "--eval", script], { + cwd: process.cwd(), + encoding: "utf-8", + env: { + ...process.env, + HOME: path.join(root, "home"), USERPROFILE: path.join(root, "home"), + PI_CODING_AGENT_DIR: path.join(root, "agent"), + TMPDIR: root, TMP: root, TEMP: root, + PI_SUBAGENT_CHILD: undefined, + PI_SUBAGENTS_TEST_NO_OWNER: "1", + }, + timeout: 10_000, + }); + assert.equal(result.status, 0, result.stderr); + return JSON.parse(result.stdout.trim().split("\n").pop() ?? "[]"); +} + +describe("session_shutdown disposal scope", () => { + it("calls disposeForParent with this session's file, not the whole-process dispose", () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), "pi-shutdown-scope-fixture-")); + const calls = runScript(root); + const scoped = calls.filter(([name]) => name === "disposeForParent"); + const global = calls.filter(([name]) => name === "dispose"); + assert.equal(scoped.length, 1, `exactly one scoped dispose, got: ${JSON.stringify(calls)}`); + assert.ok(String(scoped[0]?.[1]).endsWith("parent-session.jsonl"), `scoped to this session's file, got: ${scoped[0]?.[1]}`); + assert.equal(global.length, 0, "the whole-process dispose must not run on session_shutdown"); + }); +});