From 3fff3c100d03f882a47e2a3b4c3ce9b1027937d3 Mon Sep 17 00:00:00 2001 From: Marinus Hvelplund <138297722+marinushv@users.noreply.github.com> Date: Mon, 19 Oct 2026 15:22:00 +0000 Subject: [PATCH 1/7] fix(server): keep owned ACP work in agent resource scopes --- apps/server/src/process/agentScope.test.ts | 185 ++++++++++++++++- apps/server/src/process/agentScope.ts | 97 ++++++++- .../provider/acp/AcpJsonRpcConnection.test.ts | 192 ++++++++++++++++++ .../src/server/AcpSessionRuntime.ts | 78 ++++--- packages/shared/src/AgentScope.ts | 14 ++ 5 files changed, 529 insertions(+), 37 deletions(-) diff --git a/apps/server/src/process/agentScope.test.ts b/apps/server/src/process/agentScope.test.ts index 396fc22987d43..3ece9d3d4115b 100644 --- a/apps/server/src/process/agentScope.test.ts +++ b/apps/server/src/process/agentScope.test.ts @@ -7,7 +7,11 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { describe, expect, it } from "@effect/vitest"; import * as HostProcess from "@t3tools/shared/HostProcess"; import * as Effect from "effect/Effect"; +import * as Deferred from "effect/Deferred"; +import * as Fiber from "effect/Fiber"; +import * as TestClock from "effect/testing/TestClock"; import * as Layer from "effect/Layer"; +import * as FileSystem from "effect/FileSystem"; import { ChildProcessSpawner } from "effect/process"; import * as ServerConfig from "../config.ts"; @@ -61,12 +65,26 @@ describe("agentSliceMemoryLimits", () => { }); }); +interface FakeScopeState { + activeState: string; + result: string; + oomKills?: number | undefined; + populated?: boolean; +} + const withFakeSystemd = ( body: (input: { readonly scope: Effect.Success; - readonly units: Map; + readonly units: Map; readonly calls: ReadonlyArray>; readonly stateDir: string; + readonly handlers: Map< + string, + ( + input: ProcessRunner.ProcessRunInput, + ) => Effect.Effect + >; + readonly output: (stdout: string) => ProcessRunner.ProcessRunOutput; }) => Effect.Effect, ) => Effect.gen(function* () { @@ -74,8 +92,14 @@ const withFakeSystemd = ( for (const name of ["systemd-run", "systemctl", "agent"]) { NodeFS.writeFileSync(NodePath.join(bin, name), "", { mode: 0o755 }); } - const units = new Map(); + const units = new Map(); const calls: Array> = []; + const handlers = new Map< + string, + ( + input: ProcessRunner.ProcessRunInput, + ) => Effect.Effect + >(); const output = (stdout: string): ProcessRunner.ProcessRunOutput => ({ stdout, stderr: "", @@ -87,27 +111,46 @@ const withFakeSystemd = ( stderrInvalidUtf8: false, }); const runner = ProcessRunner.ProcessRunner.of({ - run: ({ args }) => - Effect.sync(() => { + run: (input) => + Effect.gen(function* () { + const { args } = input; calls.push(args); - const [, verb, unit = ""] = args; + const [, verb = "", unit = ""] = args; + const handled = yield* handlers.get(verb)?.(input) ?? Effect.succeed(undefined); + if (handled !== undefined) return handled; + if (verb === "stop") units.delete(unit); if (verb === "reset-failed") units.delete(unit); if (verb !== "show") return output(""); const state = units.get(unit); return output( state === undefined ? "LoadState=not-found\nActiveState=inactive\nResult=success\n" - : `LoadState=loaded\nActiveState=${state.activeState}\nResult=${state.result}\n`, + : `LoadState=loaded\nActiveState=${state.activeState}\nResult=${state.result}\nControlGroup=/t3-agent-scope-test/${unit}\n`, ); }), }); + const fileSystem = yield* FileSystem.FileSystem; const scope = yield* make.pipe( + Effect.provideService(FileSystem.FileSystem, { + ...fileSystem, + readFileString: (path, encoding) => { + if (!path.startsWith("/sys/fs/cgroup/t3-agent-scope-test/")) { + return fileSystem.readFileString(path, encoding); + } + const state = units.get(NodePath.basename(NodePath.dirname(path))); + return Effect.succeed( + path.endsWith("memory.events") + ? `oom_kill ${state?.oomKills ?? 0}\n` + : `populated ${state?.populated === false ? 0 : 1}\n`, + ); + }, + }), Effect.provideService(ProcessRunner.ProcessRunner, runner), Effect.provideService(HostProcess.Platform, "linux"), Effect.provideService(HostProcess.Environment, { PATH: bin, XDG_RUNTIME_DIR: "/run/user/1" }), ); const { stateDir } = yield* ServerConfig.ServerConfig; - return yield* body({ scope, units, calls, stateDir }).pipe( + return yield* body({ scope, units, calls, stateDir, handlers, output }).pipe( Effect.ensuring(Effect.sync(() => NodeFS.rmSync(bin, { recursive: true, force: true }))), ); }).pipe( @@ -126,6 +169,134 @@ const sliceOf = (launch: { readonly args: ReadonlyArray }) => launch.args.find((arg) => arg.startsWith("--slice="))?.slice("--slice=".length) ?? ""; describe("AgentScope service", () => { + it.effect("waits for scope shutdown with a timeout longer than TimeoutStopSec", () => + withFakeSystemd(({ scope, units, calls, handlers, output }) => + Effect.gen(function* () { + const launch = yield* scope.wrap({ command: "agent", args: [], name: "acp" }); + const unit = unitOf(launch); + expect(launch.unit).toBe(unit); + expect(launch.args).toContain("--property=TimeoutStopSec=10s"); + units.set(unit, { activeState: "active", result: "success", populated: true }); + const stopping = yield* Deferred.make(); + const stopped = yield* Deferred.make(); + handlers.set("stop", (input) => + Deferred.succeed(stopping, input).pipe( + Effect.andThen(Deferred.await(stopped)), + Effect.andThen(Effect.sync(() => units.delete(unit))), + Effect.as(output("")), + ), + ); + const fiber = yield* scope.stop(unit).pipe(Effect.forkChild); + expect((yield* Deferred.await(stopping)).timeout).toBe("15 seconds"); + expect(fiber.pollUnsafe()).toBeUndefined(); + yield* Deferred.succeed(stopped, undefined); + yield* Fiber.join(fiber); + expect(calls.filter(([, verb]) => verb === "stop")).toEqual([["--user", "stop", unit]]); + }), + ), + ); + + it.effect.each(["kernel OOM", "OOM result", "deactivating"] as const)( + "awaits %s shutdown without replacing its diagnosis", + (pending) => + withFakeSystemd(({ scope, units, calls, handlers }) => + Effect.gen(function* () { + const launch = yield* scope.wrap({ + command: "agent", + args: [], + name: "acp", + threadId: "a", + }); + const unit = unitOf(launch); + units.set(unit, { + activeState: pending === "deactivating" ? "deactivating" : "active", + result: pending === "OOM result" ? "oom-kill" : "success", + oomKills: pending === "kernel OOM" ? 1 : 0, + populated: true, + }); + const waiting = yield* Deferred.make(); + const completed = yield* Deferred.make(); + let reads = 0; + handlers.set("show", () => + ++reads === 2 + ? Deferred.succeed(waiting, undefined).pipe( + Effect.andThen(Deferred.await(completed)), + Effect.as(undefined), + ) + : Effect.succeed(undefined), + ); + const fiber = yield* scope.stop(unit).pipe(Effect.forkChild); + yield* Deferred.await(waiting); + expect(fiber.pollUnsafe()).toBeUndefined(); + expect(calls.filter(([, verb]) => verb === "stop")).toEqual([]); + units.set(unit, { activeState: "failed", result: "oom-kill", populated: false }); + yield* Deferred.succeed(completed, undefined); + yield* Fiber.join(fiber); + expect(yield* scope.oomKilled("a")).toBe(true); + expect(calls.filter(([, verb]) => verb === "stop")).toEqual([]); + }), + ), + ); + + it.effect("fails when pending OOM shutdown never finishes", () => + withFakeSystemd(({ scope, units, handlers }) => + Effect.gen(function* () { + const launch = yield* scope.wrap({ command: "agent", args: [], name: "acp" }); + const unit = unitOf(launch); + units.set(unit, { activeState: "active", result: "success", oomKills: 1, populated: true }); + const waiting = yield* Deferred.make(); + let reads = 0; + handlers.set("show", () => + ++reads === 2 + ? Deferred.succeed(waiting, undefined).pipe(Effect.andThen(Effect.never)) + : Effect.succeed(undefined), + ); + const fiber = yield* scope.stop(unit).pipe(Effect.uninterruptible, Effect.forkChild); + yield* Deferred.await(waiting); + yield* TestClock.adjust("15 seconds"); + expect(yield* Effect.flip(Fiber.join(fiber))).toMatchObject({ + _tag: "AgentScopeStopError", + detail: "Agent scope did not finish stopping", + }); + }), + ), + ); + + it.effect.each(["failure", "timeout"] as const)("reports a scope stop %s", (failure) => + withFakeSystemd(({ scope, units, handlers, output }) => + Effect.gen(function* () { + const launch = yield* scope.wrap({ command: "agent", args: [], name: "acp" }); + const unit = unitOf(launch); + units.set(unit, { activeState: "active", result: "success", populated: true }); + handlers.set("stop", () => + Effect.succeed({ + ...output(""), + code: ChildProcessSpawner.ExitCode(failure === "failure" ? 1 : 0), + timedOut: failure === "timeout", + }), + ); + expect(yield* Effect.flip(scope.stop(unit))).toMatchObject({ + _tag: "AgentScopeStopError", + unit, + }); + }), + ), + ); + + it.effect("does not report an unreadable scope as stopped", () => + withFakeSystemd(({ scope, handlers, output }) => + Effect.gen(function* () { + const launch = yield* scope.wrap({ command: "agent", args: [], name: "acp" }); + handlers.set("show", () => + Effect.succeed({ ...output(""), code: ChildProcessSpawner.ExitCode(1) }), + ); + expect(yield* Effect.flip(scope.stop(unitOf(launch)))).toMatchObject({ + _tag: "AgentScopeStopError", + }); + }), + ), + ); + it.effect("labels a thread from its newest scope until its next session clears it", () => withFakeSystemd(({ scope, units }) => Effect.gen(function* () { diff --git a/apps/server/src/process/agentScope.ts b/apps/server/src/process/agentScope.ts index ed69eb2ad753f..c417f386c6e65 100644 --- a/apps/server/src/process/agentScope.ts +++ b/apps/server/src/process/agentScope.ts @@ -4,6 +4,7 @@ import * as NodeOS from "node:os"; import { AgentScope, + AgentScopeStopError, agentScopeCommand, type AgentScopeShape, type AgentSystemdScope, @@ -44,7 +45,7 @@ export interface ScopeState { readonly loadState: string; readonly activeState: string; readonly result: string; - readonly oomKills: number; + readonly oomKills: number | undefined; readonly populated: boolean | undefined; } @@ -62,7 +63,11 @@ export function classifyScope(state: ScopeState): "oom-killed" | "gone" | "runni ) { return "gone"; } - if (state.activeState === "deactivating" || state.oomKills > 0 || state.populated === false) { + if ( + state.activeState === "deactivating" || + (state.oomKills ?? 0) > 0 || + state.populated === false + ) { return "stopping"; } return "running"; @@ -171,7 +176,8 @@ export const make = Effect.gen(function* () { // older user managers still get scopes, without the OOM label. const stopOptions = [STOP_TIMEOUT_OPTION]; const oomOptions = [...stopOptions, "--property=OOMPolicy=stop"]; - const scopeOptions = (yield* probe(oomOptions)) + const oomPolicy = yield* probe(oomOptions); + const scopeOptions = oomPolicy ? oomOptions : (yield* probe(stopOptions)) ? stopOptions @@ -197,13 +203,14 @@ export const make = Effect.gen(function* () { yield* Effect.logInfo("Agent scopes are on", { slice: AGENT_SLICE, ...limits, - oomPolicy: scopeOptions.length > 0, + oomPolicy, }); return { systemdRun, systemctl, runtimeDir, scopeOptions, + oomPolicy, }; }), ); @@ -228,7 +235,7 @@ export const make = Effect.gen(function* () { loadState: values.get("LoadState") ?? "", activeState: values.get("ActiveState") ?? "", result: values.get("Result") ?? "", - oomKills: parseCgroupCounter(memoryEvents, "oom_kill") ?? 0, + oomKills: parseCgroupCounter(memoryEvents, "oom_kill"), populated: populated === undefined ? undefined : populated === 1, } satisfies ScopeState; }); @@ -303,6 +310,83 @@ export const make = Effect.gen(function* () { return false; }); + const hasStopped = (state: ScopeState) => + state.loadState === "not-found" || + state.populated === false || + (state.populated === undefined && + (state.activeState === "inactive" || state.activeState === "failed")); + + const awaitStopped = (systemctl: string, unit: string) => + Effect.gen(function* () { + while (true) { + const state = yield* readScope(systemctl, unit); + if (state === undefined) { + return yield* new AgentScopeStopError({ + unit, + detail: "Could not read agent scope state", + }); + } + if (hasStopped(state)) return; + yield* Effect.sleep(STOPPING_POLL_INTERVAL); + } + }).pipe( + Effect.interruptible, + Effect.timeoutOption("15 seconds"), + Effect.flatMap((completed) => + completed._tag === "Some" + ? Effect.void + : Effect.fail( + new AgentScopeStopError({ unit, detail: "Agent scope did not finish stopping" }), + ), + ), + ); + + const stop = Effect.fn("AgentScope.stop")(function* (unit: string) { + if (!unit.startsWith("t3code-") || !unit.endsWith(".scope")) { + return yield* new AgentScopeStopError({ unit, detail: "Cannot stop an unowned agent scope" }); + } + const scope = yield* systemd; + if (scope === undefined) { + return yield* new AgentScopeStopError({ unit, detail: "Agent scope manager is unavailable" }); + } + const state = yield* readScope(scope.systemctl, unit); + if (state === undefined) { + return yield* new AgentScopeStopError({ unit, detail: "Could not read agent scope state" }); + } + if (hasStopped(state)) return; + // Let OOMPolicy finish and record a kernel kill before a manual stop overwrites it. + const stopping = + state.activeState === "deactivating" || + state.result === "oom-kill" || + (scope.oomPolicy && (state.oomKills === undefined || state.oomKills > 0)); + if (!stopping) { + const stopped = yield* runner + .run({ + command: scope.systemctl, + args: ["--user", "stop", unit], + timeout: "15 seconds", + }) + .pipe( + Effect.mapError( + (cause) => + new AgentScopeStopError({ + unit, + detail: "Could not stop agent scope", + cause, + }), + ), + ); + if (stopped.code !== 0 || stopped.timedOut) { + return yield* new AgentScopeStopError({ + unit, + detail: "Could not stop agent scope", + cause: stopped, + }); + } + } + yield* awaitStopped(scope.systemctl, unit); + }); + const clear: AgentScopeShape["clear"] = (threadId) => Effect.gen(function* () { if (!scopeByThread.has(threadId)) return; @@ -334,9 +418,10 @@ export const make = Effect.gen(function* () { // the unit even when it fails. options: threadId === undefined ? [...scope.scopeOptions, "--collect"] : scope.scopeOptions, }; - return agentScopeCommand({ command: resolved, args, scope: systemdScope }); + return { ...agentScopeCommand({ command: resolved, args, scope: systemdScope }), unit }; }), oomKilled, + stop, clear, } satisfies AgentScopeShape; }); diff --git a/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts b/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts index 4d1de51cc1b76..661d004e6e593 100644 --- a/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts +++ b/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts @@ -1,4 +1,5 @@ // @effect-diagnostics nodeBuiltinImport:off +import * as NodeChildProcess from "node:child_process"; import * as NodePath from "node:path"; import * as NodeOS from "node:os"; import * as NodeURL from "node:url"; @@ -6,6 +7,7 @@ import * as NodeFS from "node:fs"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { it } from "@effect/vitest"; +import * as Crypto from "effect/Crypto"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -13,13 +15,25 @@ import * as Fiber from "effect/Fiber"; import * as Option from "effect/Option"; import * as Scope from "effect/Scope"; import * as Result from "effect/Result"; +import * as Schema from "effect/Schema"; import * as TestClock from "effect/testing/TestClock"; import * as Stream from "effect/Stream"; import { describe, expect } from "vite-plus/test"; +import { ChildProcessSpawner } from "effect/process"; import * as AcpSessionRuntime from "@t3tools/provider-acp/server/AcpSessionRuntime"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import { + AgentScope, + AgentScopeStopError, + agentScopeCommand, + type AgentScopeShape, +} from "@t3tools/shared/AgentScope"; import type * as EffectAcpProtocol from "effect-acp/protocol"; import * as EffectAcpErrors from "effect-acp/errors"; +import * as AgentScopeLive from "../../process/agentScope.ts"; +import * as ProcessRunner from "../../processRunner.ts"; +import * as ServerConfig from "../../config.ts"; const __dirname = NodePath.dirname(NodeURL.fileURLToPath(import.meta.url)); const mockAgentPath = NodePath.join(__dirname, "../../../scripts/acp-mock-agent.ts"); @@ -33,6 +47,184 @@ const mockRuntimeOptions = { } satisfies AcpSessionRuntime.AcpSessionRuntimeOptions; describe("AcpSessionRuntime", () => { + describe.skipIf( + HostProcess.Platform.defaultValue() !== "linux" || + NodeChildProcess.spawnSync("systemctl", ["--user", "show-environment"]).status !== 0, + )("native agent scope", () => { + it.live.each(["stop", "exit"] as const)( + "keeps owned ACP work in its native scope and reaps detached descendants on %s", + (end) => + Effect.gen(function* () { + const unit = `t3code-acp-test-${yield* (yield* Crypto.Crypto).randomUUIDv4()}.scope`; + const stopped = yield* Deferred.make(); + let stopCount = 0; + let legacyLeaseCount = 0; + // Keep test subprocesses inside the host's validation budget. + const parentSlice = + NodeFS.readFileSync("/proc/self/cgroup", "utf8") + .split("/") + .toReversed() + .find((part) => part.endsWith(".slice")) ?? "app.slice"; + const descendant = ` + process.on("SIGTERM", () => {}); + setInterval(() => {}, 1000); + process.send({ pid: process.pid, cgroup: require("node:fs").readFileSync("/proc/self/cgroup", "utf8") }, () => process.disconnect()); + `; + const fixture = ` + const { spawn } = require("node:child_process"); + const reply = (id, result) => process.stdout.write(JSON.stringify({jsonrpc:"2.0", id, result}) + "\\n"); + require("node:readline").createInterface({ input: process.stdin }).on("line", line => { + const frame = JSON.parse(line); + if (frame.method === "_test/exit") return process.exit(19); + if (!("id" in frame)) return; + if (frame.method === "initialize") return reply(frame.id, {protocolVersion:1, agentCapabilities:{}}); + if (frame.method === "session/new") return reply(frame.id, {sessionId:"scope-test"}); + const child = spawn(process.execPath, ["-e", ${JSON.stringify(descendant)}], { + detached:true, stdio:["ignore", "ignore", "ignore", "ipc"] + }); + child.once("message", work => reply(frame.id, { + ...work, rootPid:process.pid, rootCgroup:require("node:fs").readFileSync("/proc/self/cgroup", "utf8") + })); + }); + `; + const runner = yield* ProcessRunner.make(); + yield* Effect.addFinalizer(() => + runner + .run({ + command: "systemctl", + args: ["--user", "stop", unit], + timeout: "15 seconds", + }) + .pipe(Effect.ignore), + ); + // Probe success avoids changing the live agents slice or its memory limits. + const successfulProbe: ProcessRunner.ProcessRunOutput = { + stdout: "", + stderr: "", + code: ChildProcessSpawner.ExitCode(0), + timedOut: false, + stdoutTruncated: false, + stderrTruncated: false, + stdoutInvalidUtf8: false, + stderrInvalidUtf8: false, + }; + const nativeScope = yield* AgentScopeLive.make.pipe( + Effect.provideService(ProcessRunner.ProcessRunner, { + run: (input) => + NodePath.basename(input.command) === "systemd-run" || + input.args[1] === "set-property" + ? Effect.succeed(successfulProbe) + : runner.run(input), + }), + Effect.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-acp-scope-" })), + ); + const agentScope: AgentScopeShape = { + ...AgentScope.defaultValue(), + wrap: ({ command, args }) => + Effect.succeed({ + ...agentScopeCommand({ + command, + args, + scope: { + systemdRun: "systemd-run", + slice: parentSlice, + unit, + runtimeDir: process.env.XDG_RUNTIME_DIR ?? `/run/user/${process.getuid?.()}`, + options: ["--collect", "--property=TimeoutStopSec=1s"], + }, + }), + unit, + }), + stop: (scopeUnit: string) => + nativeScope.stop(scopeUnit).pipe( + Effect.tap(() => + Effect.sync(() => { + expect(scopeUnit).toBe(unit); + stopCount++; + }), + ), + Effect.andThen(Deferred.succeed(stopped, undefined)), + Effect.asVoid, + ), + }; + yield* Effect.gen(function* () { + const runtime = yield* AcpSessionRuntime.make({ + ...mockRuntimeOptions, + spawn: { command: process.execPath, args: ["-e", fixture], shell: false }, + ownDetachedProcessGroup: true, + ownDescendantProcessGroups: true, + processGroupPlatform: "linux", + linuxCgroupController: { + create: () => { + legacyLeaseCount++; + return undefined; + }, + }, + }); + yield* runtime.start(); + expect(legacyLeaseCount).toBe(0); + const work = yield* runtime.request("_test/work", {}).pipe( + Effect.flatMap( + Schema.decodeUnknownEffect( + Schema.Struct({ + pid: Schema.Number, + cgroup: Schema.String, + rootPid: Schema.Number, + rootCgroup: Schema.String, + }), + ), + ), + ); + expect(work.cgroup.trim()).toContain(`/${unit}`); + expect(work.rootCgroup).toBe(work.cgroup); + expect(runtime.processContainment).toBe("cgroup-v2"); + if (end === "stop") { + expect(runtime.terminateProcessGroup).toBeDefined(); + if (runtime.terminateProcessGroup !== undefined) yield* runtime.terminateProcessGroup; + const descendantState = AcpSessionRuntime.makePosixProcessTreeController( + "linux", + ).identity(work.pid)?.state; + expect(descendantState === undefined || descendantState === "Z").toBe(true); + } else yield* runtime.notify("_test/exit", {}); + yield* Deferred.await(stopped); + const events = yield* runtime.getEvents().pipe(Stream.take(1), Stream.runCollect); + expect(events[0]?._tag).toBe("ConnectionTerminated"); + for (const pid of [work.pid, work.rootPid]) { + const state = + AcpSessionRuntime.makePosixProcessTreeController("linux").identity(pid)?.state; + expect(state === undefined || state === "Z").toBe(true); + } + }).pipe(Effect.scoped, Effect.provideService(AgentScope, agentScope)); + expect(stopCount).toBe(1); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + }); + + it.effect("fails explicit teardown when native scope cleanup fails", () => + Effect.gen(function* () { + const cause = new AgentScopeStopError({ + unit: "t3code-acp-test.scope", + detail: "systemctl stop failed", + }); + const runtime = yield* AcpSessionRuntime.make({ + ...mockRuntimeOptions, + ownDetachedProcessGroup: true, + }).pipe( + Effect.provideService(AgentScope, { + ...AgentScope.defaultValue(), + wrap: ({ command, args }) => Effect.succeed({ command, args, unit: cause.unit }), + stop: () => Effect.fail(cause), + }), + ); + yield* runtime.start(); + if (runtime.terminateProcessGroup === undefined) return expect.fail("Missing ACP teardown"); + expect(yield* Effect.flip(runtime.terminateProcessGroup)).toMatchObject({ + _tag: "AcpProcessGroupTerminationError", + cause, + }); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + it.effect.each(["session/new", "session/resume"] as const)( "buffers root metadata while %s startup is still pending", (setupMethod) => diff --git a/packages/provider-acp/src/server/AcpSessionRuntime.ts b/packages/provider-acp/src/server/AcpSessionRuntime.ts index 8f40efd164fad..9556b166a9967 100644 --- a/packages/provider-acp/src/server/AcpSessionRuntime.ts +++ b/packages/provider-acp/src/server/AcpSessionRuntime.ts @@ -1401,8 +1401,37 @@ export const make = ( ...(options.spawn.env ? { env: options.spawn.env } : {}), extendEnv: options.spawn.extendEnv ?? true, }); + const agentScope = yield* AgentScope; + const scopedLaunch = spawnCommand.shell + ? undefined + : yield* agentScope.wrap({ + command: spawnCommand.command, + args: spawnCommand.args, + name: "acp", + threadId: yield* AgentScopeThreadId, + env: { ...process.env, ...options.spawn.env }, + }); + const scopeUnit = scopedLaunch?.unit; + const stopScope = + scopeUnit === undefined + ? Effect.void + : yield* Effect.cached( + agentScope.stop(scopeUnit).pipe( + Effect.mapError( + (cause) => + new AcpProcessGroupTerminationError({ + detail: `Failed to stop ACP agent scope ${scopeUnit}`, + cause, + }), + ), + Effect.uninterruptible, + ), + ); + yield* Scope.addFinalizer(runtimeScope, stopScope.pipe(Effect.ignoreCause({ log: true }))); const linuxCgroupLease = - options.ownDescendantProcessGroups === true && options.processGroupPlatform === "linux" + scopeUnit === undefined && + options.ownDescendantProcessGroups === true && + options.processGroupPlatform === "linux" ? yield* Effect.sync(() => { try { if (options.linuxCgroupController === null) return undefined; @@ -1438,18 +1467,7 @@ export const make = ( }); }); const scopedSpawnCommand = - linuxCgroupLease !== undefined || spawnCommand.shell - ? spawnCommand - : { - ...(yield* (yield* AgentScope).wrap({ - command: spawnCommand.command, - args: spawnCommand.args, - name: "acp", - threadId: yield* AgentScopeThreadId, - env: { ...process.env, ...options.spawn.env }, - })), - shell: false, - }; + scopedLaunch === undefined ? spawnCommand : { ...scopedLaunch, shell: false }; const containedSpawnCommand = linuxCgroupLease === undefined ? scopedSpawnCommand @@ -1492,12 +1510,21 @@ export const make = ( ), ); + // An inherited stdout pipe can outlive the agent that owned the scope. + if (scopeUnit !== undefined) { + yield* child.exitCode.pipe( + Effect.matchEffect({ onFailure: () => stopScope, onSuccess: () => stopScope }), + Effect.ignoreCause({ log: true }), + Effect.forkIn(runtimeScope), + ); + } + const posixOwnershipLedger = new Map(); const posixOwnershipFrontier = new Map(); const posixOwnershipChildQueues = new Map>(); const posixOwnershipRoot: AcpPosixOwnershipRoot = { value: undefined }; const posixController = - options.ownDescendantProcessGroups === true + scopeUnit === undefined && options.ownDescendantProcessGroups === true ? yield* Effect.try({ try: () => { if (options.posixProcessTreeController !== undefined) { @@ -1590,6 +1617,7 @@ export const make = ( }); const terminateOwnedProcessGroupImpl = Effect.gen(function* () { if (options.ownDetachedProcessGroup !== true) return; + if (scopeUnit !== undefined) return yield* stopScope; if (options.processGroupPlatform === undefined) { return yield* new AcpProcessGroupTerminationError({ detail: "Detached ACP process-group ownership requires a host platform", @@ -1629,15 +1657,17 @@ export const make = ( if (options.ownDetachedProcessGroup === true) { const hostPlatform = yield* HostProcess.Platform; const forceTerminateOwnedProcessGroup = - hostPlatform === "win32" - ? terminateWindowsProcessTreeWithTaskkill(Number(child.pid)).pipe( - Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), - ) - : linuxCgroupLease !== undefined - ? terminateLinuxCgroupLease(linuxCgroupLease) - : options.ownDescendantProcessGroups === true - ? terminatePosixProcessTree(0, hostPlatform) - : signalOwnedProcessGroup("SIGKILL").pipe(Effect.asVoid); + scopeUnit !== undefined + ? stopScope + : hostPlatform === "win32" + ? terminateWindowsProcessTreeWithTaskkill(Number(child.pid)).pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + ) + : linuxCgroupLease !== undefined + ? terminateLinuxCgroupLease(linuxCgroupLease) + : options.ownDescendantProcessGroups === true + ? terminatePosixProcessTree(0, hostPlatform) + : signalOwnedProcessGroup("SIGKILL").pipe(Effect.asVoid); yield* Scope.addFinalizer( runtimeScope, Effect.uninterruptible(forceTerminateOwnedProcessGroup).pipe( @@ -2377,7 +2407,7 @@ export const make = ( return { processContainment: - linuxCgroupLease !== undefined + scopeUnit !== undefined || linuxCgroupLease !== undefined ? "cgroup-v2" : options.ownDescendantProcessGroups === true ? "process-ledger-reduced-guarantee" diff --git a/packages/shared/src/AgentScope.ts b/packages/shared/src/AgentScope.ts index 08af8f0ba23ab..4a16b53d2df87 100644 --- a/packages/shared/src/AgentScope.ts +++ b/packages/shared/src/AgentScope.ts @@ -1,5 +1,15 @@ import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; + +export class AgentScopeStopError extends Schema.TaggedError()( + "AgentScopeStopError", + { unit: Schema.String, detail: Schema.String, cause: Schema.optional(Schema.Defect()) }, +) { + override get message(): string { + return this.detail; + } +} export interface AgentScopeShape { readonly wrap: (input: { @@ -16,6 +26,8 @@ export interface AgentScopeShape { readonly survivesRestart?: boolean | undefined; }) => Effect.Effect; readonly oomKilled: (threadId: string) => Effect.Effect; + /** Waits for scope cleanup, including descendants in other process groups. */ + readonly stop: (unit: string) => Effect.Effect; /** * Forgets the thread's scope. The session manager calls it before it opens * a provider session, so a scope only answers for the session that launched @@ -25,6 +37,7 @@ export interface AgentScopeShape { } export interface AgentScopeCommand { + readonly unit?: string | undefined; readonly command: string; readonly args: ReadonlyArray; } @@ -33,6 +46,7 @@ export const AgentScope = Context.Reference("@t3tools/shared/Ag defaultValue: () => ({ wrap: ({ command, args }) => Effect.succeed({ command, args }), oomKilled: () => Effect.succeed(false), + stop: () => Effect.void, clear: () => Effect.void, }), }); From d2aa1f94339b5a94cbb5ddaa25fe5c5f98d765f2 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 17:17:44 +0000 Subject: [PATCH 2/7] fix(server): handle scope cleanup errors in provider finalizers --- .../provider/acp/AcpJsonRpcConnection.test.ts | 20 ++++++++++++------- packages/provider-muse/src/server/adapter.ts | 6 +++++- packages/provider-pi/src/server/adapter.ts | 6 +++++- packages/shared/src/AgentScope.ts | 2 +- 4 files changed, 24 insertions(+), 10 deletions(-) diff --git a/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts b/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts index 12a62349cce69..5c7527ca262d9 100644 --- a/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts +++ b/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts @@ -1,10 +1,5 @@ -import * as ServerConfig from "../../config.ts"; -import * as ProcessRunner from "../../processRunner.ts"; -import * as AgentScopeLive from "../../process/agentScope.ts"; -import * as Schema from "effect/Schema"; -import * as Crypto from "effect/Crypto"; -import * as NodeChildProcess from "node:child_process"; // @effect-diagnostics nodeBuiltinImport:off +import * as NodeChildProcess from "node:child_process"; import * as NodePath from "node:path"; import * as NodeOS from "node:os"; import * as NodeURL from "node:url"; @@ -13,6 +8,8 @@ import * as NodeFS from "node:fs"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as Crypto from "effect/Crypto"; +import * as Schema from "effect/Schema"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -24,9 +21,18 @@ import { ChildProcessSpawner } from "effect/process"; import { describe, expect } from "vite-plus/test"; import * as AcpSessionRuntime from "@t3tools/provider-acp/server/AcpSessionRuntime"; -import { AgentScope, AgentScopeStopError } from "@t3tools/shared/AgentScope"; +import { + AgentScope, + AgentScopeStopError, + agentScopeCommand, + type AgentScopeShape, +} from "@t3tools/shared/AgentScope"; import * as EffectAcpErrors from "effect-acp/errors"; +import * as ServerConfig from "../../config.ts"; +import * as ProcessRunner from "../../processRunner.ts"; +import * as AgentScopeLive from "../../process/agentScope.ts"; + const __dirname = NodePath.dirname(NodeURL.fileURLToPath(import.meta.url)); const mockAgentPath = NodePath.join(__dirname, "../../../scripts/acp-mock-agent.ts"); const mockAgentCommand = "node"; diff --git a/packages/provider-muse/src/server/adapter.ts b/packages/provider-muse/src/server/adapter.ts index 40785df6bada1..11cd9e019493f 100644 --- a/packages/provider-muse/src/server/adapter.ts +++ b/packages/provider-muse/src/server/adapter.ts @@ -1677,7 +1677,11 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( // Finalizers run in reverse order: reap detached work after closing the host. const scopeUnit = launch.unit; const stopScope = - scopeUnit === undefined ? Effect.void : yield* Effect.cached(agentScope.stop(scopeUnit)); + scopeUnit === undefined + ? Effect.void + : yield* Effect.cached( + agentScope.stop(scopeUnit).pipe(Effect.ignoreCause({ log: true })), + ); yield* Scope.addFinalizer(scope, stopScope); const processExited = yield* Deferred.make(); // Observe raw exit before acquisition: inherited stdout can also block startup cleanup. diff --git a/packages/provider-pi/src/server/adapter.ts b/packages/provider-pi/src/server/adapter.ts index aca13e35643ed..5bbb78e53142c 100644 --- a/packages/provider-pi/src/server/adapter.ts +++ b/packages/provider-pi/src/server/adapter.ts @@ -478,7 +478,11 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( // Finalizers run in reverse order: reap detached work after terminating Pi. const scopeUnit = scopedLaunch.unit; const stopScope = - scopeUnit === undefined ? Effect.void : yield* Effect.cached(agentScope.stop(scopeUnit)); + scopeUnit === undefined + ? Effect.void + : yield* Effect.cached( + agentScope.stop(scopeUnit).pipe(Effect.ignoreCause({ log: true })), + ); yield* Effect.addFinalizer(() => stopScope); const connection: PiRpcConnection = yield* makePiRpcConnection({ command: scopedLaunch.command, diff --git a/packages/shared/src/AgentScope.ts b/packages/shared/src/AgentScope.ts index 47563d4db972c..e20a66a489a00 100644 --- a/packages/shared/src/AgentScope.ts +++ b/packages/shared/src/AgentScope.ts @@ -1,7 +1,7 @@ import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; - import * as Schema from "effect/Schema"; + export class AgentScopeStopError extends Schema.TaggedError()( "AgentScopeStopError", { unit: Schema.String, detail: Schema.String, cause: Schema.optional(Schema.Defect()) }, From 736aeba22eb3de4136d517442e42e95ef2113b2a Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 17:22:22 +0000 Subject: [PATCH 3/7] refactor(server): trace the scope cleanup wait --- apps/server/src/process/agentScope.ts | 27 +++++++++++++++------------ 1 file changed, 15 insertions(+), 12 deletions(-) diff --git a/apps/server/src/process/agentScope.ts b/apps/server/src/process/agentScope.ts index 9ebc5a6445c39..0ff91b6ac36f9 100644 --- a/apps/server/src/process/agentScope.ts +++ b/apps/server/src/process/agentScope.ts @@ -306,8 +306,8 @@ export const make = Effect.gen(function* () { (state.populated === undefined && (state.activeState === "inactive" || state.activeState === "failed")); - const awaitStopped = (systemctl: string, unit: string) => - Effect.gen(function* () { + const awaitStopped = Effect.fn("AgentScope.awaitStopped")( + function* (systemctl: string, unit: string) { while (true) { const state = yield* readScope(systemctl, unit); if (state === undefined) { @@ -319,17 +319,20 @@ export const make = Effect.gen(function* () { if (hasStopped(state)) return; yield* Effect.sleep(STOPPING_POLL_INTERVAL); } - }).pipe( - Effect.interruptible, - Effect.timeoutOption("15 seconds"), - Effect.flatMap((completed) => - completed._tag === "Some" - ? Effect.void - : Effect.fail( - new AgentScopeStopError({ unit, detail: "Agent scope did not finish stopping" }), - ), + }, + (effect, _systemctl, unit) => + effect.pipe( + Effect.interruptible, + Effect.timeoutOption("15 seconds"), + Effect.flatMap((completed) => + completed._tag === "Some" + ? Effect.void + : Effect.fail( + new AgentScopeStopError({ unit, detail: "Agent scope did not finish stopping" }), + ), + ), ), - ); + ); const stop = Effect.fn("AgentScope.stop")(function* (unit: string) { if (!unit.startsWith("t3code-") || !unit.endsWith(".scope")) { From f65d58b39384f2eb769e24d8f67941c62aa050ea Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 17:29:09 +0000 Subject: [PATCH 4/7] fix(server): reap ACP launchers before scope teardown --- apps/server/src/process/agentScope.test.ts | 98 ++++++++++--------- apps/server/src/process/agentScope.ts | 2 +- .../provider/acp/AcpJsonRpcConnection.test.ts | 37 +++++++ .../src/server/AcpSessionRuntime.ts | 19 +++- 4 files changed, 106 insertions(+), 50 deletions(-) diff --git a/apps/server/src/process/agentScope.test.ts b/apps/server/src/process/agentScope.test.ts index 1480b9a58b354..27c5eb4e6383a 100644 --- a/apps/server/src/process/agentScope.test.ts +++ b/apps/server/src/process/agentScope.test.ts @@ -253,7 +253,7 @@ describe("AgentScope service", () => { ), ); - it.effect.each([0, undefined])( + it.effect.each([0])( "stops populated work with OOM counter %s and waits for completion", (oomKills) => withFakeSystemd(({ scope, units, calls, cgroupRoot, handlers }) => @@ -276,54 +276,58 @@ describe("AgentScope service", () => { ), ); - it.effect("waits for pending OOM cleanup without replacing its diagnosis", () => - withFakeSystemd(({ scope, units, calls, cgroupRoot, handlers }) => - Effect.gen(function* () { - const agent = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" }); - const unit = unitOf(agent); - exitedAgent(units, cgroupRoot, unit, { populated: true, oomKills: 1 }); - const waiting = yield* Deferred.make(); - const stopped = yield* Deferred.make(); - let reads = 0; - handlers.set("show", () => - ++reads === 2 - ? Deferred.succeed(waiting, undefined).pipe(Effect.andThen(Deferred.await(stopped))) - : Effect.void, - ); - const fiber = yield* scope.stop(unit).pipe(Effect.forkChild); - yield* Deferred.await(waiting); - expect(fiber.pollUnsafe()).toBeUndefined(); - expect(stopsOf(calls)).toEqual([]); - units.set(unit, { activeState: "failed", result: "oom-kill" }); - yield* Deferred.succeed(stopped, undefined); - yield* Fiber.join(fiber); - expect(yield* scope.oomKilled("a")).toBe(true); - }), - ), + it.effect.each([1, undefined])( + "waits for pending OOM cleanup with counter %s without replacing its diagnosis", + (oomKills) => + withFakeSystemd(({ scope, units, calls, cgroupRoot, handlers }) => + Effect.gen(function* () { + const agent = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" }); + const unit = unitOf(agent); + exitedAgent(units, cgroupRoot, unit, { populated: true, oomKills }); + const waiting = yield* Deferred.make(); + const stopped = yield* Deferred.make(); + let reads = 0; + handlers.set("show", () => + ++reads === 2 + ? Deferred.succeed(waiting, undefined).pipe(Effect.andThen(Deferred.await(stopped))) + : Effect.void, + ); + const fiber = yield* scope.stop(unit).pipe(Effect.forkChild); + yield* Deferred.await(waiting); + expect(fiber.pollUnsafe()).toBeUndefined(); + expect(stopsOf(calls)).toEqual([]); + units.set(unit, { activeState: "failed", result: "oom-kill" }); + yield* Deferred.succeed(stopped, undefined); + yield* Fiber.join(fiber); + expect(yield* scope.oomKilled("a")).toBe(true); + }), + ), ); - it.effect("bounds pending cleanup even for an uninterruptible caller", () => - withFakeSystemd(({ scope, units, cgroupRoot, handlers }) => - Effect.gen(function* () { - const agent = yield* scope.wrap({ command: "agent", args: [], name: "t" }); - const unit = unitOf(agent); - exitedAgent(units, cgroupRoot, unit, { populated: true, oomKills: 1 }); - const waiting = yield* Deferred.make(); - let reads = 0; - handlers.set("show", () => - ++reads === 2 - ? Deferred.succeed(waiting, undefined).pipe(Effect.andThen(Effect.never)) - : Effect.void, - ); - const fiber = yield* scope.stop(unit).pipe(Effect.uninterruptible, Effect.forkChild); - yield* Deferred.await(waiting); - yield* TestClock.adjust("15 seconds"); - expect(yield* Effect.flip(Fiber.join(fiber))).toMatchObject({ - _tag: "AgentScopeStopError", - detail: "Agent scope did not finish stopping", - }); - }), - ), + it.effect.each([1, undefined])( + "bounds pending cleanup with counter %s even for an uninterruptible caller", + (oomKills) => + withFakeSystemd(({ scope, units, cgroupRoot, handlers }) => + Effect.gen(function* () { + const agent = yield* scope.wrap({ command: "agent", args: [], name: "t" }); + const unit = unitOf(agent); + exitedAgent(units, cgroupRoot, unit, { populated: true, oomKills }); + const waiting = yield* Deferred.make(); + let reads = 0; + handlers.set("show", () => + ++reads === 2 + ? Deferred.succeed(waiting, undefined).pipe(Effect.andThen(Effect.never)) + : Effect.void, + ); + const fiber = yield* scope.stop(unit).pipe(Effect.uninterruptible, Effect.forkChild); + yield* Deferred.await(waiting); + yield* TestClock.adjust("15 seconds"); + expect(yield* Effect.flip(Fiber.join(fiber))).toMatchObject({ + _tag: "AgentScopeStopError", + detail: "Agent scope did not finish stopping", + }); + }), + ), ); it.effect("reports failed scope cleanup", () => diff --git a/apps/server/src/process/agentScope.ts b/apps/server/src/process/agentScope.ts index 0ff91b6ac36f9..91f625899d5b3 100644 --- a/apps/server/src/process/agentScope.ts +++ b/apps/server/src/process/agentScope.ts @@ -351,7 +351,7 @@ export const make = Effect.gen(function* () { const stopping = state.activeState === "deactivating" || state.result === "oom-kill" || - (scope.supportsOomPolicy && (state.oomKills ?? 0) > 0); + (scope.supportsOomPolicy && (state.oomKills === undefined || state.oomKills > 0)); if (!stopping) { const stopped = yield* runner .run({ diff --git a/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts b/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts index 5c7527ca262d9..e4b3e6dd93541 100644 --- a/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts +++ b/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts @@ -198,6 +198,43 @@ describe("AcpSessionRuntime", () => { ); }); + it.effect("reaps the scope launcher before accepting an absent scope", () => + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + let childIsRunning = Effect.succeed(true); + let stops = 0; + const runtime = yield* AcpSessionRuntime.make({ + ...mockRuntimeOptions, + ownDetachedProcessGroup: true, + }).pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, { + ...spawner, + spawn: (command) => + spawner.spawn(command).pipe( + Effect.tap((child) => + Effect.sync(() => { + childIsRunning = child.isRunning.pipe(Effect.orDie); + }), + ), + ), + }), + Effect.provideService(AgentScope, { + ...AgentScope.defaultValue(), + wrap: ({ command, args }) => + Effect.succeed({ command, args, unit: "t3code-acp-test.scope" }), + stop: () => + Effect.gen(function* () { + expect(yield* childIsRunning).toBe(false); + stops++; + }), + }), + ); + if (runtime.terminateProcessGroup === undefined) return expect.fail("Missing ACP teardown"); + yield* runtime.terminateProcessGroup; + expect(stops).toBe(1); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + it.effect("uses the agent scope before creating an owned Linux cgroup", () => Effect.gen(function* () { let leases = 0; diff --git a/packages/provider-acp/src/server/AcpSessionRuntime.ts b/packages/provider-acp/src/server/AcpSessionRuntime.ts index 1d926def538be..ad386a4f90fca 100644 --- a/packages/provider-acp/src/server/AcpSessionRuntime.ts +++ b/packages/provider-acp/src/server/AcpSessionRuntime.ts @@ -1477,6 +1477,21 @@ export const make = Effect.fn("AcpSessionRuntime.make")(function* ( ), ); + const terminateScopedProcess = Effect.gen(function* () { + yield* child.isRunning.pipe( + Effect.flatMap((running) => + running ? child.kill({ forceKillAfter: "1 second" }) : Effect.void, + ), + Effect.ignore, + ); + if (yield* child.isRunning.pipe(Effect.catch(() => Effect.succeed(true)))) { + return yield* new AcpProcessGroupTerminationError({ + detail: "ACP scope launcher is still running", + }); + } + yield* stopScope; + }); + // Detached children can keep stdout open, delaying protocol termination. if (scopeUnit !== undefined) { yield* child.exitCode.pipe( @@ -1585,7 +1600,7 @@ export const make = Effect.fn("AcpSessionRuntime.make")(function* ( ); const terminateOwnedProcessGroupImpl = Effect.gen(function* () { if (options.ownDetachedProcessGroup !== true) return; - if (scopeUnit !== undefined) return yield* stopScope; + if (scopeUnit !== undefined) return yield* terminateScopedProcess; if (options.processGroupPlatform === undefined) { return yield* new AcpProcessGroupTerminationError({ detail: "Detached ACP process-group ownership requires a host platform", @@ -1626,7 +1641,7 @@ export const make = Effect.fn("AcpSessionRuntime.make")(function* ( const hostPlatform = yield* HostProcess.Platform; const forceTerminateOwnedProcessGroup = scopeUnit !== undefined - ? stopScope + ? terminateScopedProcess : hostPlatform === "win32" ? terminateWindowsProcessTreeWithTaskkill(Number(child.pid)).pipe( Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), From db03b05935c0e36c07c2317dc6c30fc1e7a5241a Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 17:30:47 +0000 Subject: [PATCH 5/7] refactor(server): use direct scope launcher error recovery --- packages/provider-acp/src/server/AcpSessionRuntime.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/provider-acp/src/server/AcpSessionRuntime.ts b/packages/provider-acp/src/server/AcpSessionRuntime.ts index ad386a4f90fca..54db1bce55f5a 100644 --- a/packages/provider-acp/src/server/AcpSessionRuntime.ts +++ b/packages/provider-acp/src/server/AcpSessionRuntime.ts @@ -1484,7 +1484,7 @@ export const make = Effect.fn("AcpSessionRuntime.make")(function* ( ), Effect.ignore, ); - if (yield* child.isRunning.pipe(Effect.catch(() => Effect.succeed(true)))) { + if (yield* child.isRunning.pipe(Effect.orElseSucceed(() => true))) { return yield* new AcpProcessGroupTerminationError({ detail: "ACP scope launcher is still running", }); From 270c54675db52b2e17ab12d593c358d662994d95 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 17:41:49 +0000 Subject: [PATCH 6/7] fix(acp): refuse automatic replacement after failed cleanup --- .../Adapters/AcpAdapterV2.test.ts | 161 ++++++++++-------- packages/provider-acp/src/server/adapter.ts | 11 ++ 2 files changed, 101 insertions(+), 71 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 1cb40e1fab2ed..6443637fb44fd 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -2425,79 +2425,98 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); - it.live("replaces an unexpectedly terminated ACP runtime before the next turn", () => - Effect.gen(function* () { - const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const path = yield* Path.Path; - const selfInvocation = yield* resolveSelfInvocation(); - const mockAgentPath = yield* path.fromFileUrl( - new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), - ); - const runtimeInputs: AcpAdapterV2RuntimeInput[] = []; - let runtimeOrdinalSeen = 0; - const baseMakeRuntime = makeMockRuntime({ - childProcessSpawner, - mockAgentPath, - environment: (runtimeOrdinal) => { - runtimeOrdinalSeen = runtimeOrdinal; - return {}; - }, - }); - const instanceId = ProviderInstanceId.make("acp-test-unexpected-termination"); - const adapter = yield* makeAcpAdapterV2({ - instanceId, - flavor: { - driver: ACP_TEST_DRIVER, - capabilities: AcpProviderCapabilitiesV2, - makeRuntime: (runtimeInput) => - Effect.sync(() => { - runtimeInputs.push(runtimeInput); - }).pipe(Effect.andThen(baseMakeRuntime(runtimeInput))), - }, - selfInvocation, - }); - const threadId = ThreadId.make("thread-acp-unexpected-termination"); - const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ - runtimeMode: "full-access", - interactionMode: "default", - cwd: process.cwd(), - }); - const modelSelection = { instanceId, model: "default" } as const; - const runtime = yield* adapter.openSession({ - threadId, - providerSessionId: ProviderSessionId.make("provider-session-acp-unexpected-termination"), - modelSelection, - runtimePolicy, - }); - const providerThread = yield* runtime.ensureThread({ - threadId, - modelSelection, - runtimePolicy, - }); - assert.equal(runtimeOrdinalSeen, 1); - yield* runtimeInputs[0]!.onTermination!( - new EffectAcpErrors.AcpTransportError({ - detail: "Injected unexpected writer termination", - cause: "test", - }), - ); - - yield* runtime.startTurn( - makeTurnInput({ - threadId, - providerThread, + it.live.each([false, true])( + "replaces an unexpectedly terminated ACP runtime only after cleanup (failure=%s)", + (cleanupFails) => + Effect.gen(function* () { + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const path = yield* Path.Path; + const selfInvocation = yield* resolveSelfInvocation(); + const mockAgentPath = yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ); + const runtimeInputs: AcpAdapterV2RuntimeInput[] = []; + let runtimeOrdinalSeen = 0; + const baseMakeRuntime = makeMockRuntime({ + childProcessSpawner, + mockAgentPath, + wrapRuntime: (runtime) => ({ + ...runtime, + terminateProcessGroup: cleanupFails + ? Effect.fail( + new AcpSessionRuntime.AcpProcessGroupTerminationError({ + detail: "Injected scope cleanup failure", + }), + ) + : Effect.void, + }), + environment: (runtimeOrdinal) => { + runtimeOrdinalSeen = runtimeOrdinal; + return {}; + }, + }); + const instanceId = ProviderInstanceId.make("acp-test-unexpected-termination"); + const adapter = yield* makeAcpAdapterV2({ instanceId, + flavor: { + driver: ACP_TEST_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + makeRuntime: (runtimeInput) => + Effect.sync(() => { + runtimeInputs.push(runtimeInput); + }).pipe(Effect.andThen(baseMakeRuntime(runtimeInput))), + }, + selfInvocation, + }); + const threadId = ThreadId.make("thread-acp-unexpected-termination"); + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + }); + const modelSelection = { instanceId, model: "default" } as const; + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("provider-session-acp-unexpected-termination"), + modelSelection, runtimePolicy, - now: yield* DateTime.now, - }), - ); - yield* runtime.events.pipe( - Stream.filter((event) => event.type === "turn.terminal"), - Stream.runHead, - ); - assert.equal(runtimeOrdinalSeen, 2); - assert.lengthOf(runtimeInputs, 2); - }).pipe(Effect.provide(layerTest), Effect.scoped), + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + assert.equal(runtimeOrdinalSeen, 1); + yield* runtimeInputs[0]!.onTermination!( + new EffectAcpErrors.AcpTransportError({ + detail: "Injected unexpected writer termination", + cause: "test", + }), + ); + + const turn = runtime.startTurn( + makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now: yield* DateTime.now, + }), + ); + if (cleanupFails) { + assert.equal((yield* Effect.flip(turn))._tag, "ProviderAdapterTurnStartError"); + assert.equal(runtimeOrdinalSeen, 1); + assert.lengthOf(runtimeInputs, 1); + return; + } + yield* turn; + yield* runtime.events.pipe( + Stream.filter((event) => event.type === "turn.terminal"), + Stream.runHead, + ); + assert.equal(runtimeOrdinalSeen, 2); + assert.lengthOf(runtimeInputs, 2); + }).pipe(Effect.provide(layerTest), Effect.scoped), ); it.effect("negotiates and executes optional native session forks through the ACP runtime", () => diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index b3ce13af0a705..15e89b090868b 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -6313,6 +6313,17 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( resumeSessionId?: string, ) { if (runtimeScope !== undefined) { + yield* ( + runtime?.terminateProcessGroup?.pipe( + Effect.mapError( + (cause) => + new EffectAcpErrors.AcpTransportError({ + detail: "ACP runtime cleanup failed before replacement", + cause, + }), + ), + ) ?? Effect.void + ); yield* Scope.close(runtimeScope, Exit.void); } runtimeScope = yield* Scope.make(); From 4f9ce371971935ca05b3b1b0c9514cff84b0c135 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 30 Oct 2026 00:21:43 +0000 Subject: [PATCH 7/7] fix(acp): reuse Windows cleanup before runtime replacement --- .../provider/acp/AcpJsonRpcConnection.test.ts | 41 ++++++++++++++++++- .../src/server/AcpSessionRuntime.ts | 16 ++++---- 2 files changed, 48 insertions(+), 9 deletions(-) diff --git a/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts b/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts index 6f6ef9746c306..d308905a26693 100644 --- a/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts +++ b/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts @@ -17,7 +17,7 @@ import * as Fiber from "effect/Fiber"; import * as Scope from "effect/Scope"; import * as Result from "effect/Result"; import * as Stream from "effect/Stream"; -import { ChildProcessSpawner } from "effect/process"; +import { ChildProcess, ChildProcessSpawner } from "effect/process"; import { describe, expect } from "vite-plus/test"; import * as AcpSessionRuntime from "@t3tools/provider-acp/server/AcpSessionRuntime"; @@ -297,6 +297,45 @@ describe("AcpSessionRuntime", () => { }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), ); + it.effect.each([0, 1])("reuses Windows finalizer cleanup result %s before replacement", (code) => + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + let kills = 0; + const runtime = yield* AcpSessionRuntime.make({ + ...mockRuntimeOptions, + ownDetachedProcessGroup: true, + processGroupPlatform: "win32", + }).pipe( + Effect.provideService(HostProcess.Platform, "win32"), + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, { + ...spawner, + spawn: (command) => { + if (!ChildProcess.isStandardCommand(command) || command.command !== "taskkill") { + return spawner.spawn(command); + } + return spawner.spawn( + ChildProcess.make(process.execPath, [ + "-e", + `process.exit(${kills++ === 0 ? code : 128})`, + ]), + ); + }, + }), + Effect.scoped, + ); + expect(kills).toBe(1); + if (runtime.terminateProcessGroup === undefined) return expect.fail("Missing ACP teardown"); + if (code === 0) yield* runtime.terminateProcessGroup; + else { + expect(yield* Effect.flip(runtime.terminateProcessGroup)).toMatchObject({ + _tag: "AcpProcessGroupTerminationError", + exitCode: code, + }); + } + expect(kills).toBe(1); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + it.effect.each(["session/new", "session/resume"] as const)( "buffers root metadata while %s startup is still pending", (setupMethod) => diff --git a/packages/provider-acp/src/server/AcpSessionRuntime.ts b/packages/provider-acp/src/server/AcpSessionRuntime.ts index c60c2ffe7f847..de997b2b44098 100644 --- a/packages/provider-acp/src/server/AcpSessionRuntime.ts +++ b/packages/provider-acp/src/server/AcpSessionRuntime.ts @@ -1561,12 +1561,14 @@ export const make = Effect.fn("AcpSessionRuntime.make")(function* ( return cause?.code === "ESRCH" ? Effect.succeed(false) : Effect.fail(error); }), ); + const terminateWindowsProcessTreeWithTaskkillOnce = yield* Effect.cached( + terminateWindowsProcessTreeWithTaskkill(Number(child.pid)).pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + Effect.uninterruptible, + ), + ); const terminateWindowsProcessTree = - options.windowsProcessTreeTerminator ?? - ((pid: number) => - terminateWindowsProcessTreeWithTaskkill(pid).pipe( - Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), - )); + options.windowsProcessTreeTerminator ?? (() => terminateWindowsProcessTreeWithTaskkillOnce); const terminatePosixProcessTree = Effect.fn("AcpSessionRuntime.terminatePosixProcessTree")( function* (grace: Duration.Input, platformOverride?: NodeJS.Platform) { const platform = platformOverride ?? options.processGroupPlatform; @@ -1639,9 +1641,7 @@ export const make = Effect.fn("AcpSessionRuntime.make")(function* ( scopeUnit !== undefined ? terminateScopedProcess : hostPlatform === "win32" - ? terminateWindowsProcessTreeWithTaskkill(Number(child.pid)).pipe( - Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), - ) + ? terminateWindowsProcessTreeWithTaskkillOnce : linuxCgroupLease !== undefined ? terminateLinuxCgroupLease(linuxCgroupLease) : options.ownDescendantProcessGroups === true