diff --git a/apps/server/src/process/agentScope.test.ts b/apps/server/src/process/agentScope.test.ts index 2a9998f4cfbf6..b79918a369e9c 100644 --- a/apps/server/src/process/agentScope.test.ts +++ b/apps/server/src/process/agentScope.test.ts @@ -7,7 +7,9 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, describe, expect, it } from "@effect/vitest"; import { AgentScope } from "@t3tools/shared/AgentScope"; import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Logger from "effect/Logger"; @@ -50,12 +52,18 @@ const withFakeSystemd = ( readonly calls: ReadonlyArray>; readonly stateDir: string; readonly cgroupRoot: string; + readonly manager: { stalled: boolean; timeout: Effect.Effect }; }) => Effect.Effect, // A user manager older than systemd 253 rejects OOMPolicy on scopes. { oomPolicy = true, stopFails = false, - }: { readonly oomPolicy?: boolean; readonly stopFails?: boolean } = {}, + stalled = false, + }: { + readonly oomPolicy?: boolean; + readonly stopFails?: boolean; + readonly stalled?: boolean; + } = {}, ) => Effect.gen(function* () { const bin = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-agent-scope-")); @@ -65,6 +73,7 @@ const withFakeSystemd = ( const cgroupRoot = NodePath.join(bin, "cgroup"); const units = new Map(); const calls: Array> = []; + const manager = { stalled, timeout: Effect.void }; const output = (stdout: string, code = 0): ProcessRunner.ProcessRunOutput => ({ stdout, stderr: "", @@ -76,30 +85,42 @@ const withFakeSystemd = ( stderrInvalidUtf8: false, }); const runner = ProcessRunner.ProcessRunner.of({ - run: ({ args }) => - Effect.sync(() => { - calls.push(args); - if (!oomPolicy && args.some((arg) => arg.startsWith("--property=OOMPolicy="))) { - return output("", 1); - } - const [, verb, unit = ""] = args; - if (stopFails && verb === "stop") { - return { - ...output(""), - stderr: `Failed to stop ${args.at(-1)}: Unit not loaded.`, - code: ChildProcessSpawner.ExitCode(5), - }; - } - 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` + - `ControlGroup=${state.controlGroup ?? ""}\n`, - ); - }), + run: ({ command, args }) => + manager.stalled + ? manager.timeout.pipe( + Effect.andThen( + Effect.fail( + new ProcessRunner.ProcessTimeoutError({ + command, + argumentCount: args.length, + timeoutMs: 5_000, + }), + ), + ), + ) + : Effect.sync(() => { + calls.push(args); + if (!oomPolicy && args.some((arg) => arg.startsWith("--property=OOMPolicy="))) { + return output("", 1); + } + const [, verb, unit = ""] = args; + if (stopFails && verb === "stop") { + return { + ...output(""), + stderr: `Failed to stop ${args.at(-1)}: Unit not loaded.`, + code: ChildProcessSpawner.ExitCode(5), + }; + } + 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` + + `ControlGroup=${state.controlGroup ?? ""}\n`, + ); + }), }); const fileSystem = yield* FileSystem.FileSystem; const scope = yield* make.pipe( @@ -115,7 +136,7 @@ const withFakeSystemd = ( }), ); const { stateDir } = yield* ServerConfig.ServerConfig; - return yield* body({ scope, units, calls, stateDir, cgroupRoot }).pipe( + return yield* body({ scope, units, calls, stateDir, cgroupRoot, manager }).pipe( Effect.ensuring(Effect.sync(() => NodeFS.rmSync(bin, { recursive: true, force: true }))), ); }).pipe( @@ -305,6 +326,34 @@ describe("AgentScope service", () => { ), ); + it.effect("keeps a recovered detection when another launch hears the old failure late", () => + withFakeSystemd( + ({ scope, units, calls, stateDir, manager }) => + Effect.gen(function* () { + const timedOut = yield* Deferred.make(); + manager.timeout = Deferred.await(timedOut); + const launch = (threadId: string) => + scope.wrap({ command: "agent", args: [], name: "t", threadId }); + // Both launches share the stalled probe. The first one launches again, + // and finds systemd answering, before the second sees the failure. + const first = yield* launch("a").pipe( + Effect.andThen(Effect.sync(() => (manager.stalled = false))), + Effect.andThen(launch("a")), + Effect.forkChild({ startImmediately: true }), + ); + const second = yield* launch("b").pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.succeed(timedOut, undefined); + expect(sliceOf(yield* Fiber.join(first))).toBe(ownedAgentSlice(stateDir)); + expect(unitOf(yield* Fiber.join(second))).toBe(""); + + units.set(ownedAgentSlice(stateDir), { activeState: "active", result: "success" }); + yield* launch("c"); + expect(stopsOf(calls)).toEqual([]); + }), + { stalled: true }, + ), + ); + it.effect("stops agents a killed server left before the first launch", () => withFakeSystemd(({ scope, units, calls, stateDir }) => Effect.gen(function* () { diff --git a/apps/server/src/process/agentScope.ts b/apps/server/src/process/agentScope.ts index b9b6b56acb140..eeee9e9b24ae4 100644 --- a/apps/server/src/process/agentScope.ts +++ b/apps/server/src/process/agentScope.ts @@ -10,7 +10,9 @@ import { } from "@t3tools/shared/AgentScope"; import * as HostProcess from "@t3tools/shared/HostProcess"; import { resolveCommandPath } from "@t3tools/shared/shell"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; @@ -116,8 +118,10 @@ export const make = Effect.gen(function* () { Effect.provideService(HostProcess.Platform, platform), ); + const runOrFail = (command: string, args: ReadonlyArray) => + runner.run({ command, args, timeout: "5 seconds" }); const run = (command: string, args: ReadonlyArray) => - runner.run({ command, args, timeout: "5 seconds" }).pipe(Effect.option); + runOrFail(command, args).pipe(Effect.option); const readCgroupFile = (controlGroup: string, file: string) => fileSystem @@ -154,7 +158,7 @@ export const make = Effect.gen(function* () { } }); - const systemd = yield* Effect.cached( + const detectSystemd = yield* Effect.cachedWithTTL( Effect.gen(function* () { const runtimeDir = hostEnvironment.XDG_RUNTIME_DIR; if (runtimeDir === undefined || runtimeDir.length === 0) return undefined; @@ -162,8 +166,10 @@ export const make = Effect.gen(function* () { const systemctl = yield* resolve("systemctl", hostEnvironment); if (systemdRun === null || systemctl === null) return undefined; yield* stopLeftoverAgents(systemctl); + // Only a probe systemd answered decides. A busy user manager can miss the + // timeout, and that failure must not turn scopes off for the server's life. const probe = (options: ReadonlyArray) => - run(systemdRun, [ + runOrFail(systemdRun, [ "--user", "--scope", "--quiet", @@ -172,7 +178,7 @@ export const make = Effect.gen(function* () { ...options, "--", "true", - ]).pipe(Effect.map((result) => result._tag === "Some" && result.value.code === 0)); + ]).pipe(Effect.map((result) => result.code === 0)); // Scopes take OOMPolicy from systemd 253; older managers still get scopes. const supportsOomPolicy = yield* probe([...SCOPE_OPTIONS, "--property=OOMPolicy=stop"]); if (!supportsOomPolicy && !(yield* probe(SCOPE_OPTIONS))) { @@ -213,6 +219,16 @@ export const make = Effect.gen(function* () { supportsOomPolicy, }; }), + // Callers sharing a failed probe hear it at different times, so the cache + // expires failures itself; a late caller cannot clear a newer success. + (exit) => (Exit.isSuccess(exit) ? Duration.infinity : Duration.zero), + ); + const systemd = detectSystemd.pipe( + Effect.catch((cause) => + Effect.logWarning("Agent scopes are off until systemd answers", { cause }).pipe( + Effect.as(undefined), + ), + ), ); // `shared` scopes serve other threads too: tracked only for OOM diagnosis on old systemd.