1import * as HostProcess from "@t3tools/shared/HostProcess";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import * as FileSystem from "effect/FileSystem";
5import * as Layer from "effect/Layer";
6import * as Option from "effect/Option";
7import * as Path from "effect/Path";
8import * as Schema from "effect/Schema";
9import * as ChildProcess from "effect/process/ChildProcess";
10import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
11
12import { writeFileStringAtomically } from "@t3tools/shared/atomicWrite";
13import { signalProcessGroup } from "@t3tools/provider-core/server/processGroup";
14
15const ProcessIdentity = Schema.Struct({ pid: Schema.Int, startTime: Schema.String });
16type ProcessIdentity = typeof ProcessIdentity.Type;
17
18const OpenCodeServerEntry = Schema.Struct({
19 version: Schema.Literal(1),
20 /** The spawned group. Its id is the pid T3 spawned, which also names the entry file. */
21 pgid: Schema.Int,
22 /**
23 * A group member seen at spawn: the spawned process, or a process it left in
24 * the group if it had already exited. Its pid, start time, and command line
25 * must all still match before the group is signalled.
26 */
27 pid: Schema.Int,
28 /** Linux `/proc/<pid>/stat` start ticks, or macOS `ps` lstart (whole seconds). */
29 startTime: Schema.String,
30 command: Schema.String,
31 port: Schema.Int,
32 /** Entries copied along with a state directory are dropped, never acted on. */
33 stateDir: Schema.String,
34 /** The T3 server that spawned it. Entries of a live owner are never touched. */
35 owner: ProcessIdentity,
36});
37type OpenCodeServerEntry = typeof OpenCodeServerEntry.Type;
38const OpenCodeServerEntryJson = Schema.fromJsonString(OpenCodeServerEntry);
39const decodeEntry = Schema.decodeUnknownOption(OpenCodeServerEntryJson);
40const encodeEntry = Schema.encodeEffect(OpenCodeServerEntryJson);
41
42/**
43 * Local `opencode serve` processes run in their own process group so T3 can
44 * stop the whole group, which also means they outlive a T3 server that is
45 * SIGKILLed or crashes. Each spawn is recorded under the state directory and
46 * removed on a graceful stop; the next server start stops whatever a dead
47 * server left behind.
48 */
49export class OpenCodeServerLedger extends Context.Service<
50 OpenCodeServerLedger,
51 {
52 /** Records a spawned server group and returns the effect that forgets it after a graceful stop. */
53 readonly track: (server: {
54 readonly pid: number;
55 readonly port: number;
56 /** The argv after the binary, e.g. `["serve", "--hostname=127.0.0.1", "--port=4096"]`. */
57 readonly args: ReadonlyArray<string>;
58 }) => Effect.Effect<Effect.Effect<void>>;
59 }
60>()("@t3tools/provider-opencode/server/OpenCodeServerLedger") {}
61
62const ENTRY_DIRECTORY = "opencode-servers";
63const ENTRY_FILE = /^\d+\.json$/;
64const STOP_POLL_INTERVAL = "50 millis";
65const STOP_POLL_ATTEMPTS = 40;
66
67interface ObservedProcess {
68 readonly pid: number;
69 readonly pgid: number;
70 readonly startTime: string;
71 readonly command: string;
72 readonly zombie: boolean;
73}
74
75// `ps -o lstart` prints e.g. `Sat Sep 6 20:51:57 2026`; C locale and UTC keep it stable.
76const DARWIN_PS_LINE =
77 /^\s*(\d+)\s+(\d+)\s+(\S+)\s+(\w{3} \w{3} +\d{1,2} \d{2}:\d{2}:\d{2} \d{4})\s+(.*)$/;
78
79const parseDarwinPs = (output: string): ReadonlyArray<ObservedProcess> =>
80 output.split("\n").flatMap((line) => {
81 const match = DARWIN_PS_LINE.exec(line);
82 if (match === null) return [];
83 return [
84 {
85 pid: Number(match[1]),
86 pgid: Number(match[2]),
87 zombie: match[3]!.startsWith("Z"),
88 startTime: match[4]!,
89 command: match[5]!.trimEnd(),
90 },
91 ];
92 });
93
94const signalGroup = (pgid: number, signal: NodeJS.Signals) => {
95 try {
96 signalProcessGroup(pgid, signal);
97 } catch {
98 // The group may already be gone.
99 }
100};
101
102const groupExists = (pgid: number) => {
103 try {
104 signalProcessGroup(pgid, 0);
105 return true;
106 } catch (cause) {
107 return (cause as NodeJS.ErrnoException | undefined)?.code !== "ESRCH";
108 }
109};
110
111/**
112 * Builds a ledger for one state directory. `ownerPid` is the T3 server that
113 * owns the servers it tracks; it defaults to this process.
114 */
115export const make = Effect.fn("OpenCodeServerLedger.make")(function* (input: {
116 readonly stateDir: string;
117 readonly ownerPid?: number;
118}) {
119 const fs = yield* FileSystem.FileSystem;
120 const path = yield* Path.Path;
121 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
122 const platform = yield* HostProcess.Platform;
123 const directory = path.join(input.stateDir, ENTRY_DIRECTORY);
124 // Windows servers are not detached and are never recorded.
125 const recordable = platform === "linux" || platform === "darwin";
126
127 /** Observes `pid`, or nothing when it is not in `group` (if given). */
128 const observeLinux = (pid: number, group?: number) =>
129 Effect.gen(function* () {
130 const stat = yield* fs.readFileString(`/proc/${pid}/stat`);
131 // After the parenthesized comm: state, ppid, pgrp, session, … starttime (index 19).
132 const fields = stat
133 .slice(stat.lastIndexOf(")") + 2)
134 .trim()
135 .split(/\s+/);
136 const pgid = Number(fields[2]);
137 const startTime = fields[19];
138 if (!Number.isSafeInteger(pgid) || startTime === undefined) return undefined;
139 if (group !== undefined && pgid !== group) return undefined;
140 const cmdline = yield* fs.readFileString(`/proc/${pid}/cmdline`);
141 return {
142 pid,
143 pgid,
144 startTime,
145 command: cmdline.replace(/\0$/, "").replaceAll("\0", " "),
146 zombie: fields[0] === "Z",
147 } satisfies ObservedProcess;
148 });
149
150 const psDarwin = (selection: ReadonlyArray<string>) =>
151 spawner
152 .string(
153 ChildProcess.make(
154 "/bin/ps",
155 ["-ww", "-o", "pid=,pgid=,stat=,lstart=,args=", ...selection],
156 {
157 env: { LC_ALL: "C", TZ: "UTC" },
158 extendEnv: true,
159 stdin: "ignore",
160 stderr: "ignore",
161 },
162 ),
163 )
164 .pipe(Effect.timeout("5 seconds"), Effect.map(parseDarwinPs));
165
166 const observe = (pid: number): Effect.Effect<ObservedProcess | undefined> => {
167 if (platform === "linux") return observeLinux(pid).pipe(Effect.orElseSucceed(() => undefined));
168 if (platform === "darwin") {
169 return psDarwin(["-p", String(pid)]).pipe(
170 Effect.map((found) => found.find((candidate) => candidate.pid === pid)),
171 Effect.orElseSucceed(() => undefined),
172 );
173 }
174 return Effect.succeed(undefined);
175 };
176
177 // Scans every process, so it is only used when the spawned process is gone.
178 const observeGroup = (pgid: number): Effect.Effect<ReadonlyArray<ObservedProcess>> => {
179 if (platform === "linux") {
180 return fs.readDirectory("/proc").pipe(
181 Effect.flatMap((names) =>
182 Effect.forEach(
183 names.filter((name) => /^\d+$/.test(name)),
184 (name) => observeLinux(Number(name), pgid).pipe(Effect.orElseSucceed(() => undefined)),
185 { concurrency: 16 },
186 ),
187 ),
188 Effect.map((found) => found.filter((candidate) => candidate !== undefined)),
189 Effect.orElseSucceed(() => []),
190 );
191 }
192 if (platform === "darwin") {
193 return psDarwin(["-A"]).pipe(
194 Effect.map((found) => found.filter((candidate) => candidate.pgid === pgid)),
195 Effect.orElseSucceed(() => []),
196 );
197 }
198 return Effect.succeed([]);
199 };
200
201 const isRunning = (identity: ProcessIdentity) =>
202 observe(identity.pid).pipe(
203 Effect.map(
204 (observed) =>
205 observed !== undefined && !observed.zombie && observed.startTime === identity.startTime,
206 ),
207 );
208
209 const ownerPid = input.ownerPid ?? process.pid;
210 const ownerProcess = yield* observe(ownerPid);
211 const owner =
212 ownerProcess === undefined ? undefined : { pid: ownerPid, startTime: ownerProcess.startTime };
213
214 const track: OpenCodeServerLedger["Service"]["track"] = (server) =>
215 Effect.gen(function* () {
216 if (!recordable) return Effect.void;
217 const pgid = server.pid;
218 // The spawned process leads the group. If it already exited (a wrapper
219 // that left the server running), any live member identifies the group:
220 // a group id is not reused while the group has members.
221 const leader = yield* observe(pgid);
222 const members =
223 leader !== undefined && !leader.zombie && leader.pgid === pgid
224 ? [leader]
225 : (yield* observeGroup(pgid)).filter((candidate) => !candidate.zombie);
226 const member =
227 members.find((candidate) => candidate.command.endsWith(` ${server.args.join(" ")}`)) ??
228 members[0];
229 if (owner === undefined || member === undefined) {
230 if (groupExists(pgid)) {
231 yield* Effect.logWarning(
232 "Could not record an OpenCode server; it will keep running if this server crashes",
233 { pid: pgid, port: server.port },
234 );
235 }
236 return Effect.void;
237 }
238 const entryPath = path.join(directory, `${pgid}.json`);
239 const entry: OpenCodeServerEntry = {
240 version: 1,
241 pgid,
242 pid: member.pid,
243 startTime: member.startTime,
244 command: member.command,
245 port: server.port,
246 stateDir: input.stateDir,
247 owner,
248 };
249 yield* writeFileStringAtomically({
250 filePath: entryPath,
251 contents: yield* encodeEntry(entry),
252 }).pipe(
253 Effect.provideService(FileSystem.FileSystem, fs),
254 Effect.provideService(Path.Path, path),
255 );
256 return fs
257 .remove(entryPath, { force: true })
258 .pipe(
259 Effect.catchCause((cause) =>
260 Effect.logWarning("Could not forget a stopped OpenCode server", { cause }),
261 ),
262 );
263 }).pipe(
264 Effect.catchCause((cause) =>
265 Effect.logWarning("Could not record an OpenCode server process", { cause }).pipe(
266 Effect.as(Effect.void),
267 ),
268 ),
269 );
270
271 // Signals only a group whose recorded member still has the same pid, start
272 // time, group, and full command line, so a recycled pid is never touched.
273 const stopOrphan = (entry: OpenCodeServerEntry) =>
274 Effect.gen(function* () {
275 const observed = yield* observe(entry.pid);
276 if (
277 observed === undefined ||
278 observed.zombie ||
279 observed.startTime !== entry.startTime ||
280 observed.pgid !== entry.pgid ||
281 observed.command !== entry.command
282 ) {
283 return;
284 }
285 yield* Effect.logInfo("Stopping an OpenCode server left by a previous T3 Code server", {
286 pid: entry.pgid,
287 port: entry.port,
288 });
289 signalGroup(entry.pgid, "SIGTERM");
290 for (let attempt = 0; attempt < STOP_POLL_ATTEMPTS && groupExists(entry.pgid); attempt++) {
291 yield* Effect.sleep(STOP_POLL_INTERVAL);
292 }
293 // The group never emptied, so its pgid cannot have been reused.
294 if (groupExists(entry.pgid)) signalGroup(entry.pgid, "SIGKILL");
295 });
296
297 const reapEntry = (entryPath: string) =>
298 Effect.gen(function* () {
299 const entry = decodeEntry(yield* fs.readFileString(entryPath));
300 if (Option.isSome(entry) && entry.value.stateDir === input.stateDir) {
301 if (yield* isRunning(entry.value.owner)) return;
302 yield* stopOrphan(entry.value);
303 }
304 yield* fs.remove(entryPath, { force: true });
305 }).pipe(
306 Effect.catchCause((cause) =>
307 Effect.logWarning("Could not clean up a recorded OpenCode server", { entryPath, cause }),
308 ),
309 );
310
311 /** Stops recorded servers whose owning T3 server is gone and drops stale entries. */
312 const reapOrphans = Effect.gen(function* () {
313 const names = yield* fs.readDirectory(directory).pipe(Effect.orElseSucceed(() => []));
314 yield* Effect.forEach(
315 names.filter((name) => ENTRY_FILE.test(name)),
316 (name) => reapEntry(path.join(directory, name)),
317 { concurrency: "unbounded", discard: true },
318 );
319 });
320
321 return { track, reapOrphans };
322});
323
324/** Tracks servers under `stateDir`, the server's own state directory. */
325export const layer = (input: { readonly stateDir: string }) =>
326 Layer.effect(
327 OpenCodeServerLedger,
328 Effect.gen(function* () {
329 const ledger = yield* make(input);
330 // Reaping waits for orphans to exit, so it must not hold up startup.
331 yield* ledger.reapOrphans.pipe(Effect.forkScoped);
332 return OpenCodeServerLedger.of({ track: ledger.track });
333 }),
334 );
335
336/** Records nothing. For tests that start OpenCode servers without a state directory. */
337export const layerTest = Layer.succeed(
338 OpenCodeServerLedger,
339 OpenCodeServerLedger.of({ track: () => Effect.succeed(Effect.void) }),
340);