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 { describe, expect, it } from "@effect/vitest";
8import * as HostProcess from "@t3tools/shared/HostProcess";
9import * as Effect from "effect/Effect";
10import * as Layer from "effect/Layer";
11import { ChildProcessSpawner } from "effect/process";
12
13import * as ServerConfig from "../config.ts";
14import * as ProcessRunner from "../processRunner.ts";
15import {
16 agentSliceMemoryLimits,
17 classifyScope,
18 make,
19 ownedAgentSlice,
20 type ScopeState,
21} from "./agentScope.ts";
22
23const GiB = 1024 ** 3;
24
25const running: ScopeState = {
26 loadState: "loaded",
27 activeState: "active",
28 result: "success",
29 oomKills: 0,
30 populated: true,
31};
32
33describe("classifyScope", () => {
34 it("reports an OOM kill only once systemd records it", () => {
35 expect(classifyScope({ ...running, activeState: "failed", result: "oom-kill" })).toBe(
36 "oom-killed",
37 );
38 // The agent pipe can close before systemd handles the kill.
39 expect(classifyScope({ ...running, oomKills: 1 })).toBe("stopping");
40 expect(classifyScope({ ...running, populated: false })).toBe("stopping");
41 expect(classifyScope({ ...running, activeState: "deactivating" })).toBe("stopping");
42 });
43
44 it("does not wait on a live agent or a scope that ended for another reason", () => {
45 expect(classifyScope(running)).toBe("running");
46 expect(classifyScope({ ...running, activeState: "inactive", populated: undefined })).toBe(
47 "gone",
48 );
49 expect(classifyScope({ ...running, loadState: "not-found", activeState: "inactive" })).toBe(
50 "gone",
51 );
52 });
53});
54
55describe("agentSliceMemoryLimits", () => {
56 it("leaves 6 GB before throttling and 4 GB before a kill on large machines", () => {
57 expect(agentSliceMemoryLimits(64 * GiB)).toEqual({ high: 58 * GiB, max: 60 * GiB });
58 });
59
60 it("still gives agents most of a small machine", () => {
61 expect(agentSliceMemoryLimits(8 * GiB)).toEqual({ high: 4 * GiB, max: 6 * GiB });
62 });
63});
64
65// Runs the real service against a fake user manager. `units` holds the
66// ActiveState and Result that `systemctl show` reports for each scope or
67// slice, and `calls` records every command in order.
68const withFakeSystemd = <A, E>(
69 body: (input: {
70 readonly scope: Effect.Success<typeof make>;
71 readonly units: Map<string, { activeState: string; result: string }>;
72 readonly calls: ReadonlyArray<ReadonlyArray<string>>;
73 readonly stateDir: string;
74 }) => Effect.Effect<A, E>,
75) =>
76 Effect.gen(function* () {
77 const bin = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-agent-scope-"));
78 for (const name of ["systemd-run", "systemctl", "agent"]) {
79 NodeFS.writeFileSync(NodePath.join(bin, name), "", { mode: 0o755 });
80 }
81 const units = new Map<string, { activeState: string; result: string }>();
82 const calls: Array<ReadonlyArray<string>> = [];
83 const output = (stdout: string): ProcessRunner.ProcessRunOutput => ({
84 stdout,
85 stderr: "",
86 code: ChildProcessSpawner.ExitCode(0),
87 timedOut: false,
88 stdoutTruncated: false,
89 stderrTruncated: false,
90 stdoutInvalidUtf8: false,
91 stderrInvalidUtf8: false,
92 });
93 const runner = ProcessRunner.ProcessRunner.of({
94 run: ({ args }) =>
95 Effect.sync(() => {
96 calls.push(args);
97 const [, verb, unit = ""] = args;
98 if (verb === "reset-failed") units.delete(unit);
99 if (verb !== "show") return output("");
100 const state = units.get(unit);
101 return output(
102 state === undefined
103 ? "LoadState=not-found\nActiveState=inactive\nResult=success\n"
104 : `LoadState=loaded\nActiveState=${state.activeState}\nResult=${state.result}\n`,
105 );
106 }),
107 });
108 const scope = yield* make.pipe(
109 Effect.provideService(ProcessRunner.ProcessRunner, runner),
110 Effect.provideService(HostProcess.Platform, "linux"),
111 Effect.provideService(HostProcess.Environment, { PATH: bin, XDG_RUNTIME_DIR: "/run/user/1" }),
112 );
113 const { stateDir } = yield* ServerConfig.ServerConfig;
114 return yield* body({ scope, units, calls, stateDir }).pipe(
115 Effect.ensuring(Effect.sync(() => NodeFS.rmSync(bin, { recursive: true, force: true }))),
116 );
117 }).pipe(
118 Effect.scoped,
119 Effect.provide(
120 ServerConfig.layerTest(process.cwd(), { prefix: "t3-agent-scope-" }).pipe(
121 Layer.provideMerge(NodeServices.layer),
122 ),
123 ),
124 );
125
126const unitOf = (launch: { readonly args: ReadonlyArray<string> }) =>
127 launch.args.find((arg) => arg.startsWith("--unit="))?.slice("--unit=".length) ?? "";
128
129const sliceOf = (launch: { readonly args: ReadonlyArray<string> }) =>
130 launch.args.find((arg) => arg.startsWith("--slice="))?.slice("--slice=".length) ?? "";
131
132describe("AgentScope service", () => {
133 it.effect("labels a thread from its newest scope until its next session clears it", () =>
134 withFakeSystemd(({ scope, units }) =>
135 Effect.gen(function* () {
136 const first = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
137 units.set(unitOf(first), { activeState: "failed", result: "oom-kill" });
138 const second = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
139 units.set(unitOf(second), { activeState: "active", result: "success" });
140 // The older scope was OOM-killed, but the newest one is still running.
141 expect(yield* scope.oomKilled("a")).toBe(false);
142
143 units.set(unitOf(second), { activeState: "failed", result: "oom-kill" });
144 expect(yield* scope.oomKilled("a")).toBe(true);
145 // A retried failure reads the same, even after the failed unit is reset.
146 units.delete(unitOf(second));
147 expect(yield* scope.oomKilled("a")).toBe(true);
148
149 // The next session may run without a scope, for example on Cursor.
150 yield* scope.clear("a");
151 expect(yield* scope.oomKilled("a")).toBe(false);
152 }),
153 ),
154 );
155
156 it.effect("keeps a running scope while hundreds of other threads launch", () =>
157 withFakeSystemd(({ scope, units }) =>
158 Effect.gen(function* () {
159 const live = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "live" });
160 units.set(unitOf(live), { activeState: "active", result: "success" });
161 for (let index = 0; index < 600; index++) {
162 yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: `other-${index}` });
163 }
164 units.set(unitOf(live), { activeState: "failed", result: "oom-kill" });
165 expect(yield* scope.oomKilled("live")).toBe(true);
166 }),
167 ),
168 );
169
170 it.effect("stops agents a killed server left before the first launch", () =>
171 withFakeSystemd(({ scope, units, calls, stateDir }) =>
172 Effect.gen(function* () {
173 units.set(ownedAgentSlice(stateDir), { activeState: "active", result: "success" });
174 const agent = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
175 const stops = calls.filter(([, verb]) => verb === "stop");
176 expect(stops).toEqual([["--user", "stop", ownedAgentSlice(stateDir)]]);
177 // Agents run in this server's own slice; terminals stay outside it.
178 expect(sliceOf(agent)).toBe(ownedAgentSlice(stateDir));
179 // An agent that ignores SIGTERM is killed well inside the stop's wait.
180 expect(agent.args).toContain("--property=TimeoutStopSec=10s");
181 const terminal = yield* scope.wrap({
182 command: "agent",
183 args: [],
184 name: "terminal",
185 survivesRestart: true,
186 });
187 expect(sliceOf(terminal)).toBe("app-t3code-agents.slice");
188 // Only the first launch looks for leftovers.
189 expect(calls.filter(([, verb]) => verb === "stop")).toHaveLength(1);
190 }),
191 ),
192 );
193
194 it.effect("leaves a quiet slice alone", () =>
195 withFakeSystemd(({ scope, calls }) =>
196 Effect.gen(function* () {
197 yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
198 expect(calls.filter(([, verb]) => verb === "stop")).toEqual([]);
199 }),
200 ),
201 );
202});