1// @effect-diagnostics nodeBuiltinImport:off -- fake systemd binaries live in a temp dir.
2import * as NodeFS from "node:fs";
3import * as NodeOS from "node:os";
4import * as NodePath from "node:path";
5
6import * as NodeServices from "@effect/platform-node/NodeServices";
7import { assert, describe, expect, it } from "@effect/vitest";
8import { AgentScope } from "@t3tools/shared/AgentScope";
9import * as HostProcess from "@t3tools/shared/HostProcess";
10import * as Effect from "effect/Effect";
11import * as FileSystem from "effect/FileSystem";
12import * as Layer from "effect/Layer";
13import * as Logger from "effect/Logger";
14import { ChildProcess, ChildProcessSpawner } from "effect/process";
15
16import * as ServerConfig from "../config.ts";
17import { makeCodexAppServerSpawnCommand } from "../orchestration-v2/Adapters/CodexAdapterV2.ts";
18import * as ProcessRunner from "../processRunner.ts";
19import * as ProviderProcesses from "./providerProcesses.ts";
20import { classifyScope, make, ownedAgentSlice, type ScopeState } from "./agentScope.ts";
21
22const running: ScopeState = {
23 loadState: "loaded",
24 activeState: "active",
25 result: "success",
26 oomKills: 0,
27 populated: true,
28};
29
30describe("classifyScope", () => {
31 it("reports an OOM kill only once systemd records it", () => {
32 expect(classifyScope({ ...running, activeState: "failed", result: "oom-kill" })).toBe(
33 "oom-killed",
34 );
35 expect(classifyScope({ ...running, oomKills: 1 })).toBe("stopping");
36 expect(classifyScope({ ...running, populated: false })).toBe("stopping");
37 expect(classifyScope({ ...running, activeState: "deactivating" })).toBe("stopping");
38 });
39});
40
41interface FakeUnit {
42 activeState: string;
43 result: string;
44 controlGroup?: string;
45}
46
47const withFakeSystemd = <A, E>(
48 body: (input: {
49 readonly scope: Effect.Success<typeof make>;
50 readonly units: Map<string, FakeUnit>;
51 readonly calls: ReadonlyArray<ReadonlyArray<string>>;
52 readonly stateDir: string;
53 readonly cgroupRoot: string;
54 }) => Effect.Effect<A, E>,
55 // A user manager older than systemd 253 rejects OOMPolicy on scopes.
56 {
57 oomPolicy = true,
58 stopFails = false,
59 leftovers = false,
60 liveOwner = false,
61 }: {
62 readonly oomPolicy?: boolean;
63 readonly stopFails?: boolean;
64 readonly leftovers?: boolean;
65 readonly liveOwner?: boolean;
66 } = {},
67) =>
68 Effect.gen(function* () {
69 const bin = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-agent-scope-"));
70 for (const name of ["systemd-run", "systemctl", "agent"]) {
71 NodeFS.writeFileSync(NodePath.join(bin, name), "", { mode: 0o755 });
72 }
73 const cgroupRoot = NodePath.join(bin, "cgroup");
74 const units = new Map<string, FakeUnit>();
75 const calls: Array<ReadonlyArray<string>> = [];
76 const output = (stdout: string, code = 0): ProcessRunner.ProcessRunOutput => ({
77 stdout,
78 stderr: "",
79 code: ChildProcessSpawner.ExitCode(code),
80 timedOut: false,
81 stdoutTruncated: false,
82 stderrTruncated: false,
83 stdoutInvalidUtf8: false,
84 stderrInvalidUtf8: false,
85 });
86 const runner = ProcessRunner.ProcessRunner.of({
87 run: ({ args }) =>
88 Effect.sync(() => {
89 calls.push(args);
90 if (!oomPolicy && args.some((arg) => arg.startsWith("--property=OOMPolicy="))) {
91 return output("", 1);
92 }
93 const [, verb, unit = ""] = args;
94 if (stopFails && verb === "stop") {
95 return {
96 ...output(""),
97 stderr: `Failed to stop ${args.at(-1)}: Unit not loaded.`,
98 code: ChildProcessSpawner.ExitCode(5),
99 };
100 }
101 if (verb === "reset-failed") units.delete(unit);
102 if (verb !== "show") return output("");
103 const state = units.get(unit);
104 return output(
105 state === undefined
106 ? "LoadState=not-found\nActiveState=inactive\nResult=success\n"
107 : `LoadState=loaded\nActiveState=${state.activeState}\nResult=${state.result}\n` +
108 `ControlGroup=${state.controlGroup ?? ""}\n`,
109 );
110 }),
111 });
112 const fileSystem = yield* FileSystem.FileSystem;
113 const { stateDir } = yield* ServerConfig.ServerConfig;
114 if (leftovers) {
115 units.set(ownedAgentSlice(stateDir), { activeState: "active", result: "success" });
116 }
117 const scope = yield* make.pipe(
118 Effect.provideService(ProviderProcesses.ProviderProcesses, { hasOtherLiveOwner: liveOwner }),
119 Effect.provideService(ProcessRunner.ProcessRunner, runner),
120 Effect.provideService(HostProcess.Platform, "linux"),
121 Effect.provideService(HostProcess.Environment, { PATH: bin, XDG_RUNTIME_DIR: "/run/user/1" }),
122 Effect.provideService(FileSystem.FileSystem, {
123 ...fileSystem,
124 readFileString: (path, encoding) =>
125 fileSystem.readFileString(path.replace("/sys/fs/cgroup", cgroupRoot), encoding),
126 writeFileString: (path, data, options) =>
127 fileSystem.writeFileString(path.replace("/sys/fs/cgroup", cgroupRoot), data, options),
128 }),
129 );
130 return yield* body({ scope, units, calls, stateDir, cgroupRoot }).pipe(
131 Effect.ensuring(Effect.sync(() => NodeFS.rmSync(bin, { recursive: true, force: true }))),
132 );
133 }).pipe(
134 Effect.scoped,
135 Effect.provide(
136 ServerConfig.layerTest(process.cwd(), { prefix: "t3-agent-scope-" }).pipe(
137 Layer.provideMerge(NodeServices.layer),
138 ),
139 ),
140 );
141
142const unitOf = (launch: Pick<ChildProcess.StandardCommand, "args">) =>
143 launch.args
144 .map(String)
145 .find((arg) => arg.startsWith("--unit="))
146 ?.slice("--unit=".length) ?? "";
147
148const sliceOf = (launch: { readonly args: ReadonlyArray<string> }) =>
149 launch.args.find((arg) => arg.startsWith("--slice="))?.slice("--slice=".length) ?? "";
150
151/** An active scope as systemd and the kernel report it, with the agent itself gone. */
152const exitedAgent = (
153 units: Map<string, FakeUnit>,
154 cgroupRoot: string,
155 unit: string,
156 { populated, oomKills }: { readonly populated: boolean; readonly oomKills: number | undefined },
157) => {
158 const controlGroup = `/${unit}`;
159 NodeFS.mkdirSync(NodePath.join(cgroupRoot, controlGroup), { recursive: true });
160 NodeFS.writeFileSync(
161 NodePath.join(cgroupRoot, controlGroup, "cgroup.events"),
162 `populated ${populated ? 1 : 0}\nfrozen 0\n`,
163 );
164 // An undefined count leaves memory.events unreadable.
165 if (oomKills !== undefined) {
166 NodeFS.writeFileSync(
167 NodePath.join(cgroupRoot, controlGroup, "memory.events"),
168 `low 0\nhigh 0\nmax 0\noom ${oomKills}\noom_kill ${oomKills}\noom_group_kill 0\n`,
169 );
170 }
171 units.set(unit, { activeState: "active", result: "success", controlGroup });
172};
173
174const stopsOf = (calls: ReadonlyArray<ReadonlyArray<string>>) =>
175 calls.filter(([, verb]) => verb === "stop");
176const launchCodex = (scope: Effect.Success<typeof make>, threadId: string) =>
177 makeCodexAppServerSpawnCommand({ command: "agent", args: ["app-server"], threadId }).pipe(
178 Effect.provideService(AgentScope, scope),
179 Effect.provideService(HostProcess.Platform, "linux"),
180 Effect.map((command) => {
181 assert(ChildProcess.isStandardCommand(command));
182 return command;
183 }),
184 );
185
186describe("AgentScope service", () => {
187 it.effect("labels a thread from its newest scope until its next session clears it", () =>
188 withFakeSystemd(({ scope, units }) =>
189 Effect.gen(function* () {
190 const first = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
191 units.set(unitOf(first), { activeState: "failed", result: "oom-kill" });
192 const second = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
193 units.set(unitOf(second), { activeState: "active", result: "success" });
194 expect(yield* scope.oomKilled("a")).toBe(false);
195
196 units.set(unitOf(second), { activeState: "failed", result: "oom-kill" });
197 expect(yield* scope.oomKilled("a")).toBe(true);
198 units.delete(unitOf(second));
199 expect(yield* scope.oomKilled("a")).toBe(true);
200
201 yield* scope.clear("a");
202 expect(yield* scope.oomKilled("a")).toBe(false);
203 }),
204 ),
205 );
206
207 it.effect(
208 "keeps shared Codex scopes running without attributing their OOM to the launch thread",
209 () =>
210 withFakeSystemd(({ scope, units, calls }) =>
211 Effect.gen(function* () {
212 const agent = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
213 expect(agent.args).toContain("--property=OOMPolicy=stop");
214 units.set(unitOf(agent), { activeState: "failed", result: "oom-kill" });
215 expect(yield* scope.oomKilled("a")).toBe(true);
216
217 const codex = yield* launchCodex(scope, "a");
218 expect(codex.args).toContain("--property=OOMPolicy=continue");
219 expect(codex.args).not.toContain("--property=OOMPolicy=stop");
220 expect(codex.args).toContain("--property=TimeoutStopSec=10s");
221 expect(codex.args).toContain("--collect");
222 units.set(unitOf(codex), { activeState: "active", result: "success" });
223 expect(yield* scope.oomKilled("a")).toBe(false);
224 units.set(unitOf(codex), { activeState: "failed", result: "oom-kill" });
225 expect(yield* scope.oomKilled("a")).toBe(false);
226 expect(
227 calls.filter(([, verb, unit]) => verb === "show" && unit === unitOf(codex)),
228 ).toEqual([]);
229 }),
230 ),
231 );
232
233 it.effect.each([{ oomPolicy: true }, { oomPolicy: false }])(
234 "Force stop kills a thread's own scope but never a shared Codex scope (OOM policy $oomPolicy)",
235 ({ oomPolicy }) =>
236 withFakeSystemd(
237 ({ scope, units, calls, cgroupRoot }) =>
238 Effect.gen(function* () {
239 const agent = yield* scope.wrap({
240 command: "agent",
241 args: [],
242 name: "t",
243 threadId: "a",
244 });
245 exitedAgent(units, cgroupRoot, unitOf(agent), { populated: true, oomKills: 0 });
246 const codex = yield* launchCodex(scope, "b");
247 exitedAgent(units, cgroupRoot, unitOf(codex), { populated: true, oomKills: 0 });
248
249 expect(yield* scope.kill("a")).toBe(true);
250 expect(
251 NodeFS.readFileSync(NodePath.join(cgroupRoot, unitOf(agent), "cgroup.kill"), "utf8"),
252 ).toBe("1");
253 expect(yield* scope.kill("b")).toBe(false);
254 expect(NodeFS.existsSync(NodePath.join(cgroupRoot, unitOf(codex), "cgroup.kill"))).toBe(
255 false,
256 );
257 expect(calls.filter(([, verb]) => verb === "kill")).toEqual([]);
258 }),
259 { oomPolicy },
260 ),
261 );
262
263 it.effect("keeps a running scope while hundreds of other threads launch", () =>
264 withFakeSystemd(({ scope, units }) =>
265 Effect.gen(function* () {
266 const live = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "live" });
267 units.set(unitOf(live), { activeState: "active", result: "success" });
268 for (let index = 0; index < 600; index++) {
269 yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: `other-${index}` });
270 }
271 units.set(unitOf(live), { activeState: "failed", result: "oom-kill" });
272 expect(yield* scope.oomKilled("live")).toBe(true);
273 }),
274 ),
275 );
276
277 it.effect("delegates every agent scope so rootless containers can manage their cgroups", () =>
278 withFakeSystemd(({ scope }) =>
279 Effect.gen(function* () {
280 const agent = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
281 const codex = yield* launchCodex(scope, "a");
282 const terminal = yield* scope.wrap({
283 command: "agent",
284 args: [],
285 name: "terminal",
286 survivesRestart: true,
287 });
288 for (const launch of [agent, codex, terminal]) {
289 expect(launch.args).toContain("--property=Delegate=yes");
290 }
291 }),
292 ),
293 );
294
295 it.effect("caps agent memory and yields CPU, disk, and tasks to the server", () =>
296 withFakeSystemd(({ scope, calls }) =>
297 Effect.gen(function* () {
298 yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
299 const setProperties = calls.filter(([, verb]) => verb === "set-property");
300 expect(setProperties).toEqual([
301 [
302 "--user",
303 "set-property",
304 "--runtime",
305 "app-t3code-agents.slice",
306 expect.stringMatching(/^MemoryHigh=\d+$/),
307 expect.stringMatching(/^MemoryMax=\d+$/),
308 ],
309 [
310 "--user",
311 "set-property",
312 "--runtime",
313 "app-t3code.slice",
314 "CPUWeight=10",
315 "IOWeight=10",
316 "TasksMax=25%",
317 ],
318 ]);
319 }),
320 ),
321 );
322
323 it.effect("stops agents a killed server left before startup completes", () =>
324 withFakeSystemd(
325 ({ scope, calls, stateDir }) =>
326 Effect.gen(function* () {
327 expect(stopsOf(calls)).toEqual([["--user", "stop", ownedAgentSlice(stateDir)]]);
328 const agent = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
329 expect(sliceOf(agent)).toBe(ownedAgentSlice(stateDir));
330 expect(agent.args).toContain("--property=TimeoutStopSec=10s");
331 const terminal = yield* scope.wrap({
332 command: "agent",
333 args: [],
334 name: "terminal",
335 survivesRestart: true,
336 });
337 expect(sliceOf(terminal)).toBe("app-t3code-agents.slice");
338 expect(stopsOf(calls)).toHaveLength(1);
339 }),
340 { leftovers: true },
341 ),
342 );
343
344 it.effect("leaves the slice to another live server sharing this home", () =>
345 withFakeSystemd(({ calls }) => Effect.sync(() => expect(stopsOf(calls)).toEqual([])), {
346 leftovers: true,
347 liveOwner: true,
348 }),
349 );
350
351 it.effect("stops leftover processes only in an active agent scope", () =>
352 withFakeSystemd(({ scope, units, calls, cgroupRoot }) =>
353 Effect.gen(function* () {
354 const agent = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
355 const unit = unitOf(agent);
356 expect(agent.unit).toBe(unit);
357
358 yield* scope.stop("session-2.scope");
359 expect(calls.some((args) => args.includes("session-2.scope"))).toBe(false);
360 yield* scope.stop(unit);
361 exitedAgent(units, cgroupRoot, unit, { populated: false, oomKills: 0 });
362 yield* scope.stop(unit);
363 exitedAgent(units, cgroupRoot, unit, { populated: true, oomKills: 0 });
364 units.get(unit)!.activeState = "deactivating";
365 yield* scope.stop(unit);
366 expect(stopsOf(calls)).toEqual([]);
367
368 units.get(unit)!.activeState = "active";
369 yield* scope.stop(unit);
370 expect(stopsOf(calls)).toEqual([["--user", "stop", "--no-block", unit]]);
371 expect(yield* scope.oomKilled("a")).toBe(false);
372 }),
373 ),
374 );
375
376 it.effect.each([true, false])(
377 "cleans up an OOM exit without losing its diagnosis (OOMPolicy=%s)",
378 (oomPolicy) =>
379 withFakeSystemd(
380 ({ scope, units, calls, cgroupRoot }) =>
381 Effect.gen(function* () {
382 const agent = yield* scope.wrap({
383 command: "agent",
384 args: [],
385 name: "t",
386 threadId: "a",
387 });
388 const unit = unitOf(agent);
389 exitedAgent(units, cgroupRoot, unit, { populated: true, oomKills: undefined });
390 yield* scope.stop(unit);
391 expect(stopsOf(calls)).toEqual(
392 oomPolicy ? [] : [["--user", "stop", "--no-block", unit]],
393 );
394
395 // The kernel records the kill before systemd records the OOM result.
396 exitedAgent(units, cgroupRoot, unit, { populated: true, oomKills: 1 });
397 yield* scope.stop(unit);
398 expect(stopsOf(calls)).toHaveLength(oomPolicy ? 0 : 2);
399
400 if (oomPolicy) {
401 units.set(unit, { activeState: "failed", result: "oom-kill" });
402 expect(yield* scope.oomKilled("a")).toBe(true);
403 units.delete(unit);
404 expect(yield* scope.oomKilled("a")).toBe(true);
405 }
406 }),
407 { oomPolicy },
408 ),
409 );
410
411 it.effect("reports failed daemon cleanup without failing session teardown", () => {
412 const warnings: Array<unknown> = [];
413 const logger = Logger.make(({ message }) => {
414 warnings.push(message);
415 });
416 return withFakeSystemd(
417 ({ scope, units, cgroupRoot }) =>
418 Effect.gen(function* () {
419 const agent = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
420 exitedAgent(units, cgroupRoot, unitOf(agent), { populated: true, oomKills: 0 });
421 yield* scope.stop(unitOf(agent));
422 expect(warnings.flat()).toContain("Could not stop an exited agent's scope");
423 }).pipe(Effect.provide(Logger.layer([logger], { mergeWithExisting: false }))),
424 { stopFails: true },
425 );
426 });
427
428 it.effect("leaves a quiet slice alone", () =>
429 withFakeSystemd(({ scope, calls }) =>
430 Effect.gen(function* () {
431 yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
432 expect(calls.filter(([, verb]) => verb === "stop")).toEqual([]);
433 }),
434 ),
435 );
436});