diff --git a/apps/server/src/cloud/bootService.ts b/apps/server/src/cloud/bootService.ts index 294f229affa963e4f8dfd84524b4615bd581ffcf..1d7d50b13185fe982acce97414b3b766b00dfa80 100644 --- a/apps/server/src/cloud/bootService.ts +++ b/apps/server/src/cloud/bootService.ts @@ -106,13 +106,22 @@ export function renderBootServiceUnit(plan: BootServicePlan): string { // Let the launcher mark an explicit stop before it signals the server. // systemd still SIGKILLs the whole cgroup if graceful shutdown times out. "KillMode=mixed", - // Agent tool calls run as children of the server, so they share this cgroup. - // With the systemd default of OOMPolicy=stop, the kernel killing one greedy - // child stops the whole unit: the server, every live agent, and the user's - // connection. Keep running and let Restart=always cover the main process. + // Agents and terminals run in their own scopes in app-t3code-agents.slice + // (see process/agentScope.ts). Short helper commands still share this + // cgroup, and with the systemd default of OOMPolicy=stop, the kernel + // killing one of them stops the whole unit. Keep running and let + // Restart=always cover the main process. "OOMPolicy=continue", "Restart=always", "RestartSec=5", + // The agents slice is a sibling in app.slice, so these weights keep the + // server responsive while agents saturate CPU or disk. + "CPUWeight=1000", + "IOWeight=1000", + // systemd-oomd only honors this when the cgroup it watches is owned by the + // same user. Distros watch user-1000.slice, which root owns, so there it is + // ignored. The agent scopes are what keep oomd off the server. + "ManagedOOMPreference=avoid", `StandardOutput=append:${escapeSystemdSpecifiers(plan.logPath)}`, `StandardError=append:${escapeSystemdSpecifiers(plan.logPath)}`, "", diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 79d7bdd7e1152dedd31d5d038995a6a0374733d9..1451ec9ff62e65b52b9838c08e66fb84ca8510ea 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -31,6 +31,7 @@ import type { WebSearchOutput, } from "@anthropic-ai/claude-agent-sdk/sdk-tools"; import { parseCliArgs } from "@t3tools/shared/cliArgs"; +import { AgentScope } from "@t3tools/shared/AgentScope"; import * as HostProcess from "@t3tools/shared/HostProcess"; import { applyClaudePromptEffortPrefix } from "@t3tools/shared/model"; import { @@ -178,6 +179,8 @@ export function claudeProviderTurnTokenUsage( } export const CLAUDE_DEFAULT_INSTANCE_ID = defaultInstanceIdForDriver(CLAUDE_PROVIDER); const DEFAULT_CLAUDE_SETTINGS = Schema.decodeSync(ClaudeSettings)({}); +// Entry extensions the Claude SDK runs through node instead of executing. +const CLAUDE_SCRIPT_EXTENSIONS = [".js", ".mjs", ".tsx", ".ts", ".jsx"]; export const ClaudeProviderCapabilitiesV2 = { sessions: { @@ -605,6 +608,7 @@ export const layerQueryRunner: Layer.Layer< Effect.gen(function* () { const crypto = yield* Crypto.Crypto; const { native: nativeEventLogger } = yield* ProviderEventLoggers.ProviderEventLoggers; + const agentScope = yield* AgentScope; return ClaudeAgentSdkQueryRunner.of({ allocateSessionId: crypto.randomUUIDv4.pipe( @@ -627,11 +631,35 @@ export const layerQueryRunner: Layer.Layer< ), Stream.toAsyncIterable, ); + // The SDK spawns a native binary as `path ...executableArgs ...sdkArgs`, + // which lets the agent scope wrapper go in front of the CLI while the + // SDK keeps its own stderr capture and exit reporting. The SDK runs a + // script entry through node instead, so those launch unwrapped. + const binaryPath = input.options.pathToClaudeCodeExecutable; + const launch = + binaryPath === undefined || + CLAUDE_SCRIPT_EXTENSIONS.some((ext) => binaryPath.endsWith(ext)) + ? undefined + : yield* agentScope.wrap({ + command: binaryPath, + args: [], + name: "claude", + threadId: input.threadId, + env: input.options.env, + }); + const options = + launch === undefined || launch.args.length === 0 + ? input.options + : { + ...input.options, + pathToClaudeCodeExecutable: launch.command, + executableArgs: [...launch.args, ...(input.options.executableArgs ?? [])], + }; const queryRuntime = yield* Effect.try({ try: () => query({ prompt, - options: input.options, + options, }), catch: (cause) => queryRunnerError(cause, "query"), }); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 87c116250c5b7b5d07f6b4452d753422f102dabe..398337b22e2c1658f7cd079b9f89aeba5fc25b56 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -30,6 +30,7 @@ import { type ProviderSetupError, } from "@t3tools/contracts"; import { SKILL_MENTION_PATTERN } from "@t3tools/shared/composerInlineTokens"; +import { AgentScope } from "@t3tools/shared/AgentScope"; import * as HostProcess from "@t3tools/shared/HostProcess"; import { dynamicToolTitle } from "@t3tools/shared/toolActivity"; import { getModelSelectionStringOptionValue, modelSelectionsEqual } from "@t3tools/shared/model"; @@ -1357,8 +1358,16 @@ export const makeCodexAppServerSpawnCommand = Effect.fn( readonly cwd?: string | undefined; readonly env?: NodeJS.ProcessEnv | undefined; readonly extendEnv?: boolean | undefined; + readonly threadId?: string | undefined; }) { - const spawnCommand = yield* resolveSpawnCommand(input.command, input.args, { + const launch = yield* (yield* AgentScope).wrap({ + command: input.command, + args: input.args, + name: "codex", + threadId: input.threadId, + env: input.env, + }); + const spawnCommand = yield* resolveSpawnCommand(launch.command, launch.args, { ...(input.env === undefined ? {} : { env: input.env }), ...(input.extendEnv === undefined ? {} : { extendEnv: input.extendEnv }), }); @@ -1495,6 +1504,7 @@ export const layerAppServerClientFactory: Layer.Layer< Effect.gen(function* () { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; const { native: nativeEventLogger } = yield* ProviderEventLoggers.ProviderEventLoggers; + const agentScope = yield* AgentScope; return CodexAppServerClientFactory.of({ open: (input) => @@ -1510,7 +1520,8 @@ export const layerAppServerClientFactory: Layer.Layer< resolveCodexLaunchArgs(input.settings.launchArgs, input.environment), ), env: environment, - }); + threadId: input.threadId, + }).pipe(Effect.provideService(AgentScope, agentScope)); const handle = yield* spawner.spawn(command).pipe( Effect.provideService(Scope.Scope, scope), Effect.mapError( diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts index 02acf9b51da1abff3160da6929e964e43056a036..0b4d66394e1ced7bd59adad97b41a09852c44491 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts @@ -26,6 +26,7 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; +import { AgentScope } from "@t3tools/shared/AgentScope"; import { getModelSelectionStringOptionValue } from "@t3tools/shared/model"; import * as AnalyticsService from "../telemetry/AnalyticsService.ts"; import * as EventSink from "./EventSink.ts"; @@ -316,6 +317,7 @@ export const layer: Layer.Layer< const idAllocator = yield* IdAllocator.IdAllocatorV2; const threadCommands = yield* ThreadCommandExecutor.ThreadCommandExecutor; const analytics = yield* ProviderTurnAnalytics; + const agentScope = yield* AgentScope; const completedTurnAnalytics = new Set(); const makeDomainEvent = ( @@ -652,6 +654,11 @@ export const layer: Layer.Layer< return dismissed; } const occurredAt = yield* DateTime.now; + // A turn that failed because systemd killed the agent's scope for + // memory would otherwise read as a plain provider crash. + const failure = (yield* agentScope.oomKilled(input.threadId)) + ? { ...input.event.failure, message: "Killed: out of memory.", code: "oom_kill" } + : input.event.failure; return [ ...dismissed, yield* makeDomainEvent(input, { @@ -664,7 +671,7 @@ export const layer: Layer.Layer< providerThreadId: input.event.providerThreadId, providerTurnId: input.event.providerTurnId, itemOrdinal: input.event.failureItemOrdinal, - failure: input.event.failure, + failure, ...(input.event.retry === undefined ? {} : { retry: input.event.retry }), ...(input.event.retryStartedAt === undefined ? {} diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 53513c78e9324a96b078e6ca6eb2f24a3d97300e..04310257a71e6a2cdff09270092ba9058966ba73 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -1,6 +1,7 @@ import * as NetAddress from "effect/net/NetAddress"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; +import { AgentScope } from "@t3tools/shared/AgentScope"; import { EnvironmentId, type ModelSelection, @@ -1092,6 +1093,58 @@ it.effect("ProviderSessionManagerV2 opens a duplicate session only once", () => }), ); +it.effect( + "ProviderSessionManagerV2 starts a thread's session without its last agent's OOM kill", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const owner = ThreadId.make("thread-provider-session-manager-oom-owner"); + const joiner = ThreadId.make("thread-provider-session-manager-oom-joiner"); + // Both threads' previous agents were OOM-killed. A new session may run + // without a scope (Cursor, OpenCode, a shared Codex session), so it must + // not inherit that answer. + const oomKilledThreads = new Set([owner, joiner]); + const agentScope = { + ...AgentScope.defaultValue(), + oomKilled: (id: string) => Effect.sync(() => oomKilledThreads.has(id)), + clear: (id: string) => Effect.sync(() => void oomKilledThreads.delete(id)), + }; + + const effect = Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const now = yield* DateTime.now; + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId: owner, + }); + yield* eventSink.write({ + events: [ + yield* makeThreadCreatedEvent({ idAllocator, threadId: owner, now }), + yield* makeThreadCreatedEvent({ idAllocator, threadId: joiner, now }), + ], + }); + yield* manager.open({ threadId: owner, providerSessionId, modelSelection, runtimePolicy }); + assert.isFalse(yield* agentScope.oomKilled(owner)); + + yield* manager.open({ threadId: joiner, providerSessionId, modelSelection, runtimePolicy }); + assert.isFalse(yield* agentScope.oomKilled(joiner)); + + // Reopening a session the thread already uses keeps its answer. + oomKilledThreads.add(owner); + yield* manager.open({ threadId: owner, providerSessionId, modelSelection, runtimePolicy }); + assert.isTrue(yield* agentScope.oomKilled(owner)); + assert.equal((yield* Ref.get(state)).openCount, 1); + }); + + yield* effect.pipe( + Effect.provide(layerTest({ state, idleTimeoutMs: 60_000 })), + Effect.provideService(AgentScope, agentScope), + ); + }), +); + it.effect("ProviderSessionManagerV2 cleans up an open interrupted mid-handshake", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 53c1c06f13ef0e1a4cec4da279671a2aeb7a623e..0dec68dc1ec9ce6712e70a21339980a8172d1daa 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -1,3 +1,4 @@ +import { AgentScope } from "@t3tools/shared/AgentScope"; import * as KeyedLock from "@t3tools/shared/KeyedLock"; import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { @@ -358,6 +359,7 @@ export const layerWithOptions = ( const idAllocator = yield* IdAllocator.IdAllocatorV2; const providerEventIngestor = yield* ProviderEventIngestor.ProviderEventIngestorV2; const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const agentScope = yield* AgentScope; const agentAccessSettings = Effect.fn("ProviderSessionManagerV2.agentAccessSettings")( function* (threadId: ThreadId) { if (Option.isNone(serverSettings)) return { browser: true, device: false }; @@ -2037,16 +2039,16 @@ export const layerWithOptions = ( const key = sessionKey(input.providerSessionId); const existing = (yield* Ref.get(sessions)).get(key); if (existing !== undefined) { - if ( - !existing.attachedThreadIds.has(input.threadId) && - !existing.supportsMultipleProviderThreads - ) { + const attached = existing.attachedThreadIds.has(input.threadId); + if (!attached && !existing.supportsMultipleProviderThreads) { return yield* new ProviderSessionOpenError({ instanceId: input.modelSelection.instanceId, providerSessionId: input.providerSessionId, cause: `Provider ${existing.runtime.driver} does not support attaching multiple app threads to one session.`, }); } + // Joining a shared session is a new session for this thread. + if (!attached) yield* agentScope.clear(input.threadId); yield* ensureThreadAttached({ providerSessionId: input.providerSessionId, threadId: input.threadId, @@ -2083,6 +2085,9 @@ export const layerWithOptions = ( } }); const sessionScope = yield* Scope.fork(sessionScopes); + // The previous session's agent scope must not explain this + // session's failures, even when this provider runs without one. + yield* agentScope.clear(input.threadId); const runtime = yield* adapter .openSession({ threadId: input.threadId, diff --git a/apps/server/src/process/agentScope.test.ts b/apps/server/src/process/agentScope.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..537bcc646748be12fac5b38edef72ffa896e295a --- /dev/null +++ b/apps/server/src/process/agentScope.test.ts @@ -0,0 +1,145 @@ +// @effect-diagnostics nodeBuiltinImport:off -- fake systemd binaries live in a temp dir. +import * as NodeFS from "node:fs"; +import * as NodeOS from "node:os"; +import * as NodePath from "node:path"; + +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 { ChildProcessSpawner } from "effect/process"; + +import * as ProcessRunner from "../processRunner.ts"; +import { agentSliceMemoryLimits, classifyScope, make, type ScopeState } from "./agentScope.ts"; + +const GiB = 1024 ** 3; + +const running: ScopeState = { + loadState: "loaded", + activeState: "active", + result: "success", + oomKills: 0, + populated: true, +}; + +describe("classifyScope", () => { + it("reports an OOM kill only once systemd records it", () => { + expect(classifyScope({ ...running, activeState: "failed", result: "oom-kill" })).toBe( + "oom-killed", + ); + // The agent pipe can close before systemd handles the kill. + expect(classifyScope({ ...running, oomKills: 1 })).toBe("stopping"); + expect(classifyScope({ ...running, populated: false })).toBe("stopping"); + expect(classifyScope({ ...running, activeState: "deactivating" })).toBe("stopping"); + }); + + it("does not wait on a live agent or a scope that ended for another reason", () => { + expect(classifyScope(running)).toBe("running"); + expect(classifyScope({ ...running, activeState: "inactive", populated: undefined })).toBe( + "gone", + ); + expect(classifyScope({ ...running, loadState: "not-found", activeState: "inactive" })).toBe( + "gone", + ); + }); +}); + +describe("agentSliceMemoryLimits", () => { + it("leaves 6 GB before throttling and 4 GB before a kill on large machines", () => { + expect(agentSliceMemoryLimits(64 * GiB)).toEqual({ high: 58 * GiB, max: 60 * GiB }); + }); + + it("still gives agents most of a small machine", () => { + expect(agentSliceMemoryLimits(8 * GiB)).toEqual({ high: 4 * GiB, max: 6 * GiB }); + }); +}); + +// Runs the real service against a fake user manager. `units` holds the +// ActiveState and Result that `systemctl show` reports for each scope. +const withFakeSystemd = ( + body: (input: { + readonly scope: Effect.Success; + readonly units: Map; + }) => Effect.Effect, +) => + Effect.gen(function* () { + const bin = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-agent-scope-")); + for (const name of ["systemd-run", "systemctl", "agent"]) { + NodeFS.writeFileSync(NodePath.join(bin, name), "", { mode: 0o755 }); + } + const units = new Map(); + const output = (stdout: string): ProcessRunner.ProcessRunOutput => ({ + stdout, + stderr: "", + code: ChildProcessSpawner.ExitCode(0), + timedOut: false, + stdoutTruncated: false, + stderrTruncated: false, + stdoutInvalidUtf8: false, + stderrInvalidUtf8: false, + }); + const runner = ProcessRunner.ProcessRunner.of({ + run: ({ args }) => + Effect.sync(() => { + const [, verb, unit = ""] = args; + 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`, + ); + }), + }); + const scope = yield* make.pipe( + Effect.provideService(ProcessRunner.ProcessRunner, runner), + Effect.provideService(HostProcess.Platform, "linux"), + Effect.provideService(HostProcess.Environment, { PATH: bin, XDG_RUNTIME_DIR: "/run/user/1" }), + ); + return yield* body({ scope, units }).pipe( + Effect.ensuring(Effect.sync(() => NodeFS.rmSync(bin, { recursive: true, force: true }))), + ); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)); + +const unitOf = (launch: { readonly args: ReadonlyArray }) => + launch.args.find((arg) => arg.startsWith("--unit="))?.slice("--unit=".length) ?? ""; + +describe("AgentScope service", () => { + it.effect("labels a thread from its newest scope until its next session clears it", () => + withFakeSystemd(({ scope, units }) => + Effect.gen(function* () { + const first = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" }); + units.set(unitOf(first), { activeState: "failed", result: "oom-kill" }); + const second = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" }); + units.set(unitOf(second), { activeState: "active", result: "success" }); + // The older scope was OOM-killed, but the newest one is still running. + expect(yield* scope.oomKilled("a")).toBe(false); + + units.set(unitOf(second), { activeState: "failed", result: "oom-kill" }); + expect(yield* scope.oomKilled("a")).toBe(true); + // A retried failure reads the same, even after the failed unit is reset. + units.delete(unitOf(second)); + expect(yield* scope.oomKilled("a")).toBe(true); + + // The next session may run without a scope, for example on Cursor. + yield* scope.clear("a"); + expect(yield* scope.oomKilled("a")).toBe(false); + }), + ), + ); + + it.effect("keeps a running scope while hundreds of other threads launch", () => + withFakeSystemd(({ scope, units }) => + Effect.gen(function* () { + const live = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "live" }); + units.set(unitOf(live), { activeState: "active", result: "success" }); + for (let index = 0; index < 600; index++) { + yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: `other-${index}` }); + } + units.set(unitOf(live), { activeState: "failed", result: "oom-kill" }); + expect(yield* scope.oomKilled("live")).toBe(true); + }), + ), + ); +}); diff --git a/apps/server/src/process/agentScope.ts b/apps/server/src/process/agentScope.ts new file mode 100644 index 0000000000000000000000000000000000000000..511ed6446fd3292e127e855ccdd6c8cdc2c7fa18 --- /dev/null +++ b/apps/server/src/process/agentScope.ts @@ -0,0 +1,318 @@ +// @effect-diagnostics nodeBuiltinImport:off -- unit names and RAM size come from the host. +import * as NodeCrypto from "node:crypto"; +import * as NodeOS from "node:os"; + +import { + AgentScope, + agentScopeCommand, + type AgentScopeShape, + type AgentSystemdScope, +} from "@t3tools/shared/AgentScope"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import { resolveCommandPath } from "@t3tools/shared/shell"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; + +import * as ProcessRunner from "../processRunner.ts"; + +/** + * Agents and terminals share this slice. It sits in app.slice next to the + * server's own unit, so the server's CPUWeight and IOWeight compete with the + * agents directly. + */ +const AGENT_SLICE = "app-t3code-agents.slice"; + +const GiB = 1024 ** 3; + +/** + * Memory limits for the agents slice. Agents together leave about 6 GB free + * before they are throttled and 4 GB before the kernel kills one, so the + * server, sshd, and tailscaled keep working. Small machines still give agents + * at least half of RAM before throttling and three quarters before a kill. + */ +export function agentSliceMemoryLimits(totalBytes: number) { + return { + high: Math.max(totalBytes - 6 * GiB, Math.floor(totalBytes / 2)), + max: Math.max(totalBytes - 4 * GiB, Math.floor(totalBytes * 0.75)), + }; +} + +export interface ScopeState { + readonly loadState: string; + readonly activeState: string; + readonly result: string; + /** `oom_kill` from the scope's memory.events, while its cgroup exists. */ + readonly oomKills: number; + /** `populated` from the scope's cgroup.events, while its cgroup exists. */ + readonly populated: boolean | undefined; +} + +/** + * Reads one scope as the server sees it right after the agent process exits. + * Our pipe can close before systemd handles the kill, so a scope that is + * still stopping, is empty, or already counted an OOM kill is "stopping": + * systemd records `Result=oom-kill` for it in a moment. + */ +export function classifyScope(state: ScopeState): "oom-killed" | "gone" | "running" | "stopping" { + if (state.result === "oom-kill") return "oom-killed"; + if ( + state.loadState !== "loaded" || + state.activeState === "inactive" || + state.activeState === "failed" + ) { + return "gone"; + } + if (state.activeState === "deactivating" || state.oomKills > 0 || state.populated === false) { + return "stopping"; + } + return "running"; +} + +function parseKeyValues(text: string): Map { + const values = new Map(); + for (const line of text.split("\n")) { + const index = line.indexOf("="); + if (index > 0) values.set(line.slice(0, index), line.slice(index + 1).trim()); + } + return values; +} + +function parseCgroupCounter(text: string, key: string): number | undefined { + const match = new RegExp(`^${key} (\\d+)$`, "m").exec(text); + return match === null ? undefined : Number(match[1]); +} + +const STOPPING_POLL_INTERVAL = "100 millis"; +const STOPPING_POLL_ATTEMPTS = 20; +// A stopping scope can wait out TimeoutStopSec (90 s by default) for a stuck process. +const CLEAR_FAILED_INTERVAL = "1 second"; +const CLEAR_FAILED_ATTEMPTS = 120; +const MAX_TRACKED_THREADS = 500; + +export const make = Effect.gen(function* () { + const platform = yield* HostProcess.Platform; + if (platform !== "linux") return AgentScope.defaultValue(); + + const runner = yield* ProcessRunner.ProcessRunner; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const hostEnvironment = yield* HostProcess.Environment; + const layerScope = yield* Effect.scope; + + const resolve = (command: string, env: NodeJS.ProcessEnv) => + resolveCommandPath(command, { env }).pipe( + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), + Effect.provideService(HostProcess.Platform, platform), + Effect.option, + ); + + const run = (command: string, args: ReadonlyArray) => + runner.run({ command, args, timeout: "5 seconds" }).pipe(Effect.option); + + // Runs once, on the first agent launch. It proves this process can reach a + // systemd user manager, creates the agents slice, and sets its limits. + const systemd = yield* Effect.cached( + Effect.gen(function* () { + const runtimeDir = hostEnvironment.XDG_RUNTIME_DIR; + if (runtimeDir === undefined || runtimeDir.length === 0) return undefined; + const systemdRun = yield* resolve("systemd-run", hostEnvironment); + const systemctl = yield* resolve("systemctl", hostEnvironment); + if (systemdRun._tag === "None" || systemctl._tag === "None") return undefined; + const probe = (options: ReadonlyArray) => + run(systemdRun.value, [ + "--user", + "--scope", + "--quiet", + "--collect", + `--slice=${AGENT_SLICE}`, + ...options, + "--", + "true", + ]).pipe(Effect.map((result) => result._tag === "Some" && result.value.code === 0)); + // Stop the whole scope when the kernel kills one of its processes, and + // record the result as oom-kill. Scopes take OOMPolicy from systemd 253; + // older user managers still get plain scopes, without the OOM label. + const oomOptions = ["--property=OOMPolicy=stop"]; + const scopeOptions = (yield* probe(oomOptions)) + ? oomOptions + : (yield* probe([])) + ? [] + : undefined; + if (scopeOptions === undefined) { + yield* Effect.logInfo("Agent scopes are off: no systemd user manager"); + return undefined; + } + const limits = agentSliceMemoryLimits(NodeOS.totalmem()); + const limited = yield* run(systemctl.value, [ + "--user", + "set-property", + "--runtime", + AGENT_SLICE, + `MemoryHigh=${limits.high}`, + `MemoryMax=${limits.max}`, + ]); + if (limited._tag === "None" || limited.value.code !== 0) { + yield* Effect.logWarning("Could not set memory limits on the agents slice", { + stderr: limited._tag === "Some" ? limited.value.stderr.trim() : undefined, + }); + } + yield* Effect.logInfo("Agent scopes are on", { + slice: AGENT_SLICE, + ...limits, + oomPolicy: scopeOptions.length > 0, + }); + return { + systemdRun: systemdRun.value, + systemctl: systemctl.value, + runtimeDir, + scopeOptions, + }; + }), + ); + + // The newest scope per thread, oldest thread first. A thread runs one agent + // at a time, so only its latest scope can explain a failure. + const scopeByThread = new Map(); + + const readCgroupFile = (controlGroup: string, file: string) => + fileSystem + .readFileString(path.join("/sys/fs/cgroup", controlGroup, file)) + .pipe(Effect.orElseSucceed(() => "")); + + const readScope = (systemctl: string, unit: string) => + Effect.gen(function* () { + const shown = yield* run(systemctl, [ + "--user", + "show", + unit, + "--property=LoadState,ActiveState,Result,ControlGroup", + ]); + // A failed query says nothing about the scope; the caller keeps the unit. + if (shown._tag === "None" || shown.value.code !== 0) return undefined; + const values = parseKeyValues(shown.value.stdout); + const controlGroup = values.get("ControlGroup") ?? ""; + const memoryEvents = controlGroup ? yield* readCgroupFile(controlGroup, "memory.events") : ""; + const cgroupEvents = controlGroup ? yield* readCgroupFile(controlGroup, "cgroup.events") : ""; + const populated = parseCgroupCounter(cgroupEvents, "populated"); + return { + loadState: values.get("LoadState") ?? "", + activeState: values.get("ActiveState") ?? "", + result: values.get("Result") ?? "", + oomKills: parseCgroupCounter(memoryEvents, "oom_kill") ?? 0, + populated: populated === undefined ? undefined : populated === 1, + } satisfies ScopeState; + }); + + const clearFailed = (systemctl: string, unit: string) => + Effect.gen(function* () { + for (let attempt = 0; attempt < CLEAR_FAILED_ATTEMPTS; attempt++) { + const state = yield* readScope(systemctl, unit); + if (state === undefined) return; + if (state.activeState === "failed") { + yield* run(systemctl, ["--user", "reset-failed", unit]); + return; + } + if (state.activeState !== "deactivating") return; + yield* Effect.sleep(CLEAR_FAILED_INTERVAL); + } + }); + + // Failed scopes stay loaded so we can read their result. Clear them once + // systemd finishes stopping them. + const forget = (systemctl: string, unit: string) => + clearFailed(systemctl, unit).pipe(Effect.forkIn(layerScope), Effect.asVoid); + + // Drops the oldest thread's scope once it has ended. A scope that still + // runs moves to the back, so its thread keeps its OOM answer and cleanup. + const evictOldest = (systemctl: string) => + Effect.gen(function* () { + const oldest = scopeByThread.entries().next(); + if (oldest.done === true) return; + const [threadId, tracked] = oldest.value; + const state = yield* readScope(systemctl, tracked.unit); + // A newer launch replaced it while we read, and forgets it itself. + if (scopeByThread.get(threadId) !== tracked) return; + scopeByThread.delete(threadId); + const status = state === undefined ? "running" : classifyScope(state); + if (status === "running" || status === "stopping") { + scopeByThread.set(threadId, tracked); + return; + } + yield* forget(systemctl, tracked.unit); + }); + + // Records a thread's newest scope, or forgets the thread when `unit` is + // undefined. The replaced scope has its failed state cleared. + const track = (systemctl: string, threadId: string, unit: string | undefined) => + Effect.gen(function* () { + const previous = scopeByThread.get(threadId); + scopeByThread.delete(threadId); + if (unit !== undefined) scopeByThread.set(threadId, { unit, oomKilled: false }); + if (previous !== undefined) yield* forget(systemctl, previous.unit); + if (scopeByThread.size > MAX_TRACKED_THREADS) yield* evictOldest(systemctl); + }); + + const oomKilled: AgentScopeShape["oomKilled"] = (threadId) => + Effect.gen(function* () { + const tracked = scopeByThread.get(threadId); + if (tracked === undefined) return false; + if (tracked.oomKilled) return true; + const scope = yield* systemd; + if (scope === undefined) return false; + for (let attempt = 0; attempt < STOPPING_POLL_ATTEMPTS; attempt++) { + const state = yield* readScope(scope.systemctl, tracked.unit); + if (state === undefined) return false; + const status = classifyScope(state); + if (status === "running") return false; + if (status === "stopping") { + yield* Effect.sleep(STOPPING_POLL_INTERVAL); + continue; + } + yield* forget(scope.systemctl, tracked.unit); + if (status === "gone") return false; + yield* Effect.logWarning("Agent scope was killed: out of memory", { + threadId, + unit: tracked.unit, + }); + tracked.oomKilled = true; + return true; + } + return false; + }); + + const clear: AgentScopeShape["clear"] = (threadId) => + Effect.gen(function* () { + if (!scopeByThread.has(threadId)) return; + const scope = yield* systemd; + if (scope !== undefined) yield* track(scope.systemctl, threadId, undefined); + }); + + return { + wrap: Effect.fn("AgentScope.wrap")(function* ({ command, args, name, threadId, env }) { + const resolved = yield* resolve(command, env ?? hostEnvironment); + // Leave unresolved commands alone so the spawn reports the missing binary. + if (resolved._tag === "None") return { command, args }; + const scope = yield* systemd; + if (scope === undefined) return agentScopeCommand({ command: resolved.value, args }); + const unit = `t3code-${name}-${NodeCrypto.randomUUID().slice(0, 8)}.scope`; + if (threadId !== undefined) yield* track(scope.systemctl, threadId, unit); + const systemdScope: AgentSystemdScope = { + systemdRun: scope.systemdRun, + runtimeDir: scope.runtimeDir, + slice: AGENT_SLICE, + unit, + // Without a thread nothing reads the result, so let systemd unload + // the unit even when it fails. + options: threadId === undefined ? [...scope.scopeOptions, "--collect"] : scope.scopeOptions, + }; + return agentScopeCommand({ command: resolved.value, args, scope: systemdScope }); + }), + oomKilled, + clear, + } satisfies AgentScopeShape; +}); + +export const layer = Layer.effect(AgentScope, make).pipe(Layer.provide(ProcessRunner.layer)); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index a9cbb6cfbbd25cb239542563e54658236550bafc..3760500e40561eb778d56087c19515275a8afa90 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -31,6 +31,7 @@ import * as ServerHttp from "./http.ts"; import { guardHttpResponseWriteErrors } from "./httpResponseErrorGuard.ts"; import { fixPath } from "./os-jank.ts"; import * as Ws from "./ws.ts"; +import * as AgentScopeLive from "./process/agentScope.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; import * as NodePtyAdapter from "./terminal/NodePtyAdapter.ts"; import * as PullRequestHttp from "./pullRequest/http.ts"; @@ -1100,6 +1101,8 @@ const layerMakeServer = Layer.unwrap( Layer.provideMerge(FetchHttpClient.layer), // PR reads, Git operations, and WebSocket discovery share one process limiter. Layer.provide(VcsProcess.layer), + // Every agent and terminal spawn reads this, so it sits below everything. + Layer.provideMerge(AgentScopeLive.layer), Layer.provideMerge(layerPlatformServices), ); }), diff --git a/apps/server/src/terminal/Manager.ts b/apps/server/src/terminal/Manager.ts index d8b2cf3e0553bd6ef5361b118ce5140d6e4b5d25..a109cf5d7540304fb7ad124b8ebe4aa35ff7ebdd 100644 --- a/apps/server/src/terminal/Manager.ts +++ b/apps/server/src/terminal/Manager.ts @@ -42,6 +42,7 @@ import { } from "@t3tools/contracts"; import { makeKeyedCoalescingWorker } from "@t3tools/shared/KeyedCoalescingWorker"; import * as KeyedLock from "@t3tools/shared/KeyedLock"; +import { AgentScope } from "@t3tools/shared/AgentScope"; import * as HostProcess from "@t3tools/shared/HostProcess"; import { mergePathEntries } from "@t3tools/shared/shell"; @@ -1480,6 +1481,7 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func const historyByteLimit = options.historyByteLimit ?? DEFAULT_HISTORY_BYTE_LIMIT; const platform = yield* HostProcess.Platform; const architecture = yield* HostProcess.Architecture; + const agentScope = yield* AgentScope; // Terminals must inherit the user's full environment (minus the blocklist // applied in createTerminalSpawnEnv) — an allowlist here silently strips // things like PSModulePath, DISPLAY, proxies, and toolchain variables. @@ -2180,10 +2182,16 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func ); } + const launch = yield* agentScope.wrap({ + command: candidate.shell, + args: candidate.args ?? [], + name: "terminal", + env: spawnEnv, + }); const attempt = yield* Effect.result( options.ptyAdapter.spawn({ - shell: candidate.shell, - ...(candidate.args ? { args: candidate.args } : {}), + shell: launch.command, + ...(launch.args.length > 0 ? { args: [...launch.args] } : {}), cwd: session.cwd, cols: session.cols, rows: session.rows, diff --git a/packages/provider-acp/src/server/AcpSessionRuntime.ts b/packages/provider-acp/src/server/AcpSessionRuntime.ts index 7f5b724d1f89689341f234a2301794b3f6ac58ae..683bf8c90f41e3690ea9e0ef639a2f894b5c880b 100644 --- a/packages/provider-acp/src/server/AcpSessionRuntime.ts +++ b/packages/provider-acp/src/server/AcpSessionRuntime.ts @@ -27,6 +27,7 @@ import * as EffectAcpErrors from "effect-acp/errors"; import type * as EffectAcpSchema from "effect-acp/compat"; import type * as EffectAcpProtocol from "effect-acp/protocol"; import { resolveSpawnCommand } from "@t3tools/shared/shell"; +import { AgentScope, AgentScopeThreadId, RAISE_OOM_SCORE_LINE } from "@t3tools/shared/AgentScope"; import * as HostProcess from "@t3tools/shared/HostProcess"; import { signalProcessGroup } from "@t3tools/provider-core/server/processGroup"; @@ -275,6 +276,7 @@ export function wrapCommandForLinuxCgroup( "-c", [ "lease_path=$1; expected=$2; shift 2", + RAISE_OOM_SCORE_LINE, 'printf "%s\\n" "$$" > "$lease_path/cgroup.procs" || exit 125', "actual=", "while IFS= read -r line; do", @@ -1552,9 +1554,24 @@ export const make = ( cause: new Error("Contained ACP command was not found on PATH"), }); }); - const containedSpawnCommand = - linuxCgroupLease === undefined + // A cgroup lease is already its own leaf cgroup, so it only needs the + // raised OOM score (in its wrapper). Other agents get their own scope. + 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, + }; + const containedSpawnCommand = + linuxCgroupLease === undefined + ? scopedSpawnCommand : { ...wrapCommandForLinuxCgroup( linuxCgroupLease, diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 5f5d43e56e5c9453dd7269b9779db4578560afd3..bec893b1f4712747d6d195fb5da88155e57e468a 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -31,6 +31,7 @@ import { type ThreadTokenUsageSnapshot, type ThreadId, } from "@t3tools/contracts"; +import { AgentScope, AgentScopeThreadId } from "@t3tools/shared/AgentScope"; import { modelSelectionsEqual } from "@t3tools/shared/model"; import { type SelfInvocation, selfInvocationArgs } from "@t3tools/shared/nodeRuntime"; import { FILE_HEADERS_ONLY, formatPatch, structuredPatch } from "diff"; @@ -1486,6 +1487,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const crypto = yield* Crypto.Crypto; const fileSystem = yield* FileSystem.FileSystem; const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const agentScope = yield* AgentScope; const idAllocator = yield* IdAllocator.IdAllocatorV2; const host = yield* ProviderHost.ProviderHost; const mcpSessions = yield* McpProviderSessions.McpProviderSessions; @@ -6020,6 +6022,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( Effect.provideService(Scope.Scope, runtimeScope), Effect.provideService(Crypto.Crypto, crypto), Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, childProcessSpawner), + Effect.provideService(AgentScope, agentScope), + Effect.provideService(AgentScopeThreadId, threadId ?? undefined), ); }); @@ -6099,6 +6103,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( Effect.provideService(Scope.Scope, replacementScope), Effect.provideService(Crypto.Crypto, crypto), Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, childProcessSpawner), + Effect.provideService(AgentScope, agentScope), + Effect.provideService(AgentScopeThreadId, threadId ?? undefined), ); // Session setup may publish commands before it returns. Buffer those // notifications, but do not expose request or extension handlers diff --git a/packages/provider-muse/src/server/adapter.ts b/packages/provider-muse/src/server/adapter.ts index fe3fbbf57a7442a2c965b9bd7825c442271daa10..ded0b2efd493ce9167112267a6826e19b0f8c5ab 100644 --- a/packages/provider-muse/src/server/adapter.ts +++ b/packages/provider-muse/src/server/adapter.ts @@ -22,6 +22,7 @@ import { type ServerProviderModel, } from "@t3tools/contracts"; import type { MuseSettings } from "../settings.ts"; +import { AgentScope } from "@t3tools/shared/AgentScope"; import { getModelSelectionStringOptionValue } from "@t3tools/shared/model"; import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; @@ -250,6 +251,7 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( const providerHost = yield* ProviderHost.ProviderHost; const fileSystem = yield* FileSystem.FileSystem; const mcpSessions = yield* McpProviderSessions.McpProviderSessions; + const agentScope = yield* AgentScope; const protocolError = (detail: string, payload?: unknown) => new ProviderAdapter.ProviderAdapterProtocolError({ @@ -1367,15 +1369,24 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( const launchHost = Effect.fnUntraced(function* () { const epoch = ++hostEpoch; const mcpSession = yield* mcpSessions.read(input.threadId); + const environment = McpProviderSession.withAgentDeviceEnvironment( + options.environment, + mcpSession, + ); + const launch = yield* agentScope.wrap({ + command: options.settings.binaryPath || "muse", + args: [], + name: "muse", + threadId: input.threadId, + env: environment, + }); const created = yield* Effect.acquireRelease( createMuseSdkHostEffect( { - binaryPath: options.settings.binaryPath || "muse", + binaryPath: launch.command, + launchArgs: launch.args, cwd, - environment: McpProviderSession.withAgentDeviceEnvironment( - options.environment, - mcpSession, - ), + environment, runtimeMode: input.runtimePolicy.runtimeMode, }, options.createHost, diff --git a/packages/provider-muse/src/server/sdk.ts b/packages/provider-muse/src/server/sdk.ts index dcc778d4f77973d25832451b7bcd47f6390342cb..43c958005af137760a8edba5c4d7c956218863ee 100644 --- a/packages/provider-muse/src/server/sdk.ts +++ b/packages/provider-muse/src/server/sdk.ts @@ -29,6 +29,8 @@ export interface MuseSdkHost { export interface MuseSdkHostOptions { readonly binaryPath: string; + /** Arguments before the serve arguments, for a launcher in front of the binary. */ + readonly launchArgs?: ReadonlyArray; readonly cwd?: string; readonly environment?: NodeJS.ProcessEnv; readonly runtimeMode?: RuntimeMode; @@ -89,7 +91,7 @@ export async function createMuseSdkHost( options.signal?.throwIfAborted(); const handshake = spawn({ command: options.binaryPath, - args: museServeArgs(options), + args: [...(options.launchArgs ?? []), ...museServeArgs(options)], ...(options.cwd ? { cwd: options.cwd } : {}), // Callers pass an environment already built with makeMuseEnvironment. env: options.environment ?? makeMuseEnvironment(), diff --git a/packages/provider-opencode/src/server/OpenCodeRuntime.ts b/packages/provider-opencode/src/server/OpenCodeRuntime.ts index c91da0326228b34105366c5739e03e4ef2f3d659..5e87afe5ed598c50c077574d8ba67c2cc486a18a 100644 --- a/packages/provider-opencode/src/server/OpenCodeRuntime.ts +++ b/packages/provider-opencode/src/server/OpenCodeRuntime.ts @@ -36,6 +36,7 @@ import { isWindowsCommandNotFound } from "@t3tools/provider-core/server/snapshot import * as OpenCodeServerLedger from "./OpenCodeServerLedger.ts"; import { collectStreamAsString } from "@t3tools/provider-core/server/snapshotProbe"; import * as NetService from "@t3tools/shared/Net"; +import { AgentScope } from "@t3tools/shared/AgentScope"; import * as HostProcess from "@t3tools/shared/HostProcess"; import { compareSemverVersions, parseSemver } from "@t3tools/shared/semver"; import { resolveSpawnCommand } from "@t3tools/shared/shell"; @@ -597,6 +598,7 @@ const makeOpenCodeRuntime = Effect.gen(function* () { const netService = yield* NetService.NetService; const hostPlatform = yield* HostProcess.Platform; const serverLedger = yield* OpenCodeServerLedger.OpenCodeServerLedger; + const agentScope = yield* AgentScope; const resolveCommand = (command: string, args: ReadonlyArray, env?: NodeJS.ProcessEnv) => resolveSpawnCommand(command, args, env ? { env } : {}); @@ -693,7 +695,14 @@ const makeOpenCodeRuntime = Effect.gen(function* () { )); const timeoutMs = input.timeoutMs ?? DEFAULT_OPENCODE_SERVER_TIMEOUT_MS; const args = ["serve", `--hostname=${hostname}`, `--port=${port}`]; - const spawnCommand = yield* resolveCommand(input.binaryPath, args, input.environment); + // One server runs every session of the instance, so its scope has no thread. + const launch = yield* agentScope.wrap({ + command: input.binaryPath, + args, + name: "opencode", + env: input.environment, + }); + const spawnCommand = yield* resolveCommand(launch.command, launch.args, input.environment); const serverPassword = resolveOpenCodeServerPassword({ external: false, ...(input.serverPassword !== undefined ? { serverPassword: input.serverPassword } : {}), diff --git a/packages/provider-pi/src/server/adapter.ts b/packages/provider-pi/src/server/adapter.ts index c650a444daf9c00ce0f77dcc59f96212360d7cf1..7fd91ebc760933cc31f554d53fc7d14038b08abd 100644 --- a/packages/provider-pi/src/server/adapter.ts +++ b/packages/provider-pi/src/server/adapter.ts @@ -24,6 +24,7 @@ * updates has no matching T3 surface and is ignored. */ import * as HostProcess from "@t3tools/shared/HostProcess"; +import { AgentScope } from "@t3tools/shared/AgentScope"; import { getModelSelectionStringOptionValue } from "@t3tools/shared/model"; import { defaultInstanceIdForDriver, @@ -387,6 +388,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( options: PiAdapterV2Options, ) { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const agentScope = yield* AgentScope; const fileSystem = yield* FileSystem.FileSystem; const idAllocator = yield* IdAllocator.IdAllocatorV2; const host = yield* ProviderHost.ProviderHost; @@ -440,9 +442,16 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( extensionPath, runtimeMode: input.runtimePolicy.runtimeMode, }); - const connection: PiRpcConnection = yield* makePiRpcConnection({ + const scopedLaunch = yield* agentScope.wrap({ command: options.settings.binaryPath || "pi", args: launch.args, + name: "pi", + threadId: input.threadId, + env: launch.env, + }); + const connection: PiRpcConnection = yield* makePiRpcConnection({ + command: scopedLaunch.command, + args: scopedLaunch.args, cwd, env: launch.env, }).pipe( diff --git a/packages/shared/package.json b/packages/shared/package.json index fd68e8dada27945b35d400850500282b95fbcd7c..4ccc2b6910236d428b1bf182fba04f81e9430a40 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -311,6 +311,10 @@ "types": "./src/nodeRuntime.ts", "import": "./src/nodeRuntime.ts" }, + "./AgentScope": { + "types": "./src/AgentScope.ts", + "import": "./src/AgentScope.ts" + }, "./HostProcess": { "types": "./src/HostProcess.ts", "import": "./src/HostProcess.ts" diff --git a/packages/shared/src/AgentScope.test.ts b/packages/shared/src/AgentScope.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..85f454f7ec68db0c39c3df43793b201004d8d030 --- /dev/null +++ b/packages/shared/src/AgentScope.test.ts @@ -0,0 +1,30 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeChildProcess from "node:child_process"; + +import { describe, expect, it } from "@effect/vitest"; + +import { AGENT_OOM_SCORE_ADJ, agentScopeCommand } from "./AgentScope.ts"; +import * as HostProcess from "./HostProcess.ts"; + +describe.skipIf(HostProcess.Platform.defaultValue() !== "linux")("agentScopeCommand", () => { + it("raises the OOM score and execs the command with its arguments in the same process", () => { + const launch = agentScopeCommand({ + command: "/bin/sh", + args: [ + "-c", + 'printf "%s\\n" "$$" "$(cat /proc/self/oom_score_adj)" "$@"', + "agent", + "a b", + "$HOME", + ], + }); + const child = NodeChildProcess.spawnSync(launch.command, [...launch.args], { + encoding: "utf8", + }); + const [pid, score, ...args] = child.stdout.trimEnd().split("\n"); + // The command runs as the spawned process, so signals and PIDs still reach it. + expect(Number(pid)).toBe(child.pid); + expect(Number(score)).toBe(AGENT_OOM_SCORE_ADJ); + expect(args).toEqual(["a b", "$HOME"]); + }); +}); diff --git a/packages/shared/src/AgentScope.ts b/packages/shared/src/AgentScope.ts new file mode 100644 index 0000000000000000000000000000000000000000..403cac7c4a74f27b46393d7acd3c9e92cb18a6fc --- /dev/null +++ b/packages/shared/src/AgentScope.ts @@ -0,0 +1,129 @@ +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; + +/** + * Isolates long-lived agent and terminal processes from the server on Linux. + * + * Each wrapped command runs with a high `oom_score_adj`, so the kernel OOM + * killer picks it before the server. Under a systemd user manager it also runs + * in its own transient scope, so systemd-oomd (which kills whole leaf cgroups) + * kills one agent instead of the server and every other agent. + * + * The server provides the real implementation. The default leaves commands + * unchanged, which is also the behavior on macOS, Windows, and in tests. + */ +export interface AgentScopeShape { + /** + * Wraps the command that starts an agent or terminal. `name` labels the + * scope unit, for example "claude" or "terminal". `env` is the environment + * the command will run with, used to resolve it on PATH. A command that does + * not resolve stays unwrapped, so a missing binary still fails at spawn. + */ + readonly wrap: (input: { + readonly command: string; + readonly args: ReadonlyArray; + readonly name: string; + readonly threadId?: string | undefined; + readonly env?: NodeJS.ProcessEnv | undefined; + }) => Effect.Effect; + /** + * Whether the latest agent scope of this thread was killed because the + * machine or the agents slice ran out of memory. The answer holds until the + * thread launches its next agent, so a retried failure reads the same. + */ + readonly oomKilled: (threadId: 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 + * it, even when the next session runs without one (Cursor, OpenCode). + */ + readonly clear: (threadId: string) => Effect.Effect; +} + +export interface AgentScopeCommand { + readonly command: string; + readonly args: ReadonlyArray; +} + +export const AgentScope = Context.Reference("@t3tools/shared/AgentScope", { + defaultValue: () => ({ + wrap: ({ command, args }) => Effect.succeed({ command, args }), + oomKilled: () => Effect.succeed(false), + clear: () => Effect.void, + }), +}); + +/** + * The thread an ACP runtime is spawned for. ACP runtimes are shared by several + * provider flavors and do not take a thread id, so the ACP adapter provides it. + */ +export const AgentScopeThreadId = Context.Reference( + "@t3tools/shared/AgentScope/ThreadId", + { defaultValue: () => undefined }, +); + +/** + * Agents get this `oom_score_adj` so the kernel kills them before the server. + * An unprivileged process may raise its own score, never lower it. + */ +export const AGENT_OOM_SCORE_ADJ = 800; + +/** Shell line that raises the score of the current process, ignoring failure. */ +export const RAISE_OOM_SCORE_LINE = `{ echo ${AGENT_OOM_SCORE_ADJ} >/proc/self/oom_score_adj; } 2>/dev/null`; + +const LAUNCH_SCRIPT = [RAISE_OOM_SCORE_LINE, 'exec "$@"'].join("\n"); + +// systemd-run needs XDG_RUNTIME_DIR to reach the user manager. Fill it in when +// the agent environment does not carry it; it is the same user either way. +const SCOPED_LAUNCH_SCRIPT = [ + RAISE_OOM_SCORE_LINE, + ': "${XDG_RUNTIME_DIR:=$1}"', + "export XDG_RUNTIME_DIR", + "shift", + 'exec "$@"', +].join("\n"); + +export interface AgentSystemdScope { + readonly systemdRun: string; + readonly slice: string; + readonly unit: string; + readonly runtimeDir: string; + /** Extra `systemd-run` options, such as `--property=OOMPolicy=stop`. */ + readonly options: ReadonlyArray; +} + +/** + * Builds the Linux launch for an agent command. Every step `exec`s, so the + * spawned PID, stdio, and signals all belong to the agent itself. + * `systemd-run --scope` registers its own PID as the scope and then runs the + * command, so the agent and all of its children start inside the scope. + */ +export function agentScopeCommand(input: { + readonly command: string; + readonly args: ReadonlyArray; + readonly scope?: AgentSystemdScope | undefined; +}): AgentScopeCommand { + const { command, args, scope } = input; + if (scope === undefined) { + return { command: "/bin/sh", args: ["-c", LAUNCH_SCRIPT, "t3-agent", command, ...args] }; + } + return { + command: "/bin/sh", + args: [ + "-c", + SCOPED_LAUNCH_SCRIPT, + "t3-agent", + scope.runtimeDir, + scope.systemdRun, + "--user", + "--scope", + "--quiet", + `--slice=${scope.slice}`, + `--unit=${scope.unit}`, + ...scope.options, + "--", + command, + ...args, + ], + }; +}