Open History to see the last commit
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 pgid: Schema.Int,
21 pid: Schema.Int,
22 startTime: Schema.String,
23 command: Schema.String,
24 port: Schema.Int,
25 stateDir: Schema.String,
26 owner: ProcessIdentity,
27});
28type OpenCodeServerEntry = typeof OpenCodeServerEntry.Type;
29const OpenCodeServerEntryJson = Schema.fromJsonString(OpenCodeServerEntry);
30const decodeEntry = Schema.decodeUnknownOption(OpenCodeServerEntryJson);
31const encodeEntry = Schema.encodeEffect(OpenCodeServerEntryJson);
32
33export class OpenCodeServerLedger extends Context.Service<
34 OpenCodeServerLedger,
35 {
36 readonly track: (server: {
37 readonly pid: number;
38 readonly port: number;
39 readonly args: ReadonlyArray<string>;
40 }) => Effect.Effect<Effect.Effect<void>>;
41 }
42>()("@t3tools/provider-opencode/server/OpenCodeServerLedger") {}
43
44const ENTRY_DIRECTORY = "opencode-servers";
45const ENTRY_FILE = /^\d+\.json$/;
46const STOP_POLL_INTERVAL = "50 millis";
47const STOP_POLL_ATTEMPTS = 40;
48
49interface ObservedProcess {
50 readonly pid: number;
51 readonly pgid: number;
52 readonly startTime: string;
53 readonly command: string;
54 readonly zombie: boolean;
55}
56
57// ps lstart needs C locale and UTC for stable parsing.
58const DARWIN_PS_LINE =
59 /^\s*(\d+)\s+(\d+)\s+(\S+)\s+(\w{3} \w{3} +\d{1,2} \d{2}:\d{2}:\d{2} \d{4})\s+(.*)$/;
60
61const parseDarwinPs = (output: string): ReadonlyArray<ObservedProcess> =>
62 output.split("\n").flatMap((line) => {
63 const match = DARWIN_PS_LINE.exec(line);
64 if (match === null) return [];
65 return [
66 {
67 pid: Number(match[1]),
68 pgid: Number(match[2]),
69 zombie: match[3]!.startsWith("Z"),
70 startTime: match[4]!,
71 command: match[5]!.trimEnd(),
72 },
73 ];
74 });
75
76const signalGroup = (pgid: number, signal: NodeJS.Signals) => {
77 try {
78 signalProcessGroup(pgid, signal);
79 } catch {}
80};
81
82const groupExists = (pgid: number) => {
83 try {
84 signalProcessGroup(pgid, 0);
85 return true;
86 } catch (cause) {
87 return (cause as NodeJS.ErrnoException | undefined)?.code !== "ESRCH";
88 }
89};
90
91export const make = Effect.fn("OpenCodeServerLedger.make")(function* (input: {
92 readonly stateDir: string;
93 readonly ownerPid?: number;
94}) {
95 const fs = yield* FileSystem.FileSystem;
96 const path = yield* Path.Path;
97 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
98 const platform = yield* HostProcess.Platform;
99 const directory = path.join(input.stateDir, ENTRY_DIRECTORY);
100 const recordable = platform === "linux" || platform === "darwin";
101
102 // A process-group scan calls this for every /proc entry; avoid a span per process.
103 const observeLinux = Effect.fnUntraced(function* (pid: number, group?: number) {
104 const stat = yield* fs.readFileString(`/proc/${pid}/stat`);
105 const fields = stat
106 .slice(stat.lastIndexOf(")") + 2)
107 .trim()
108 .split(/\s+/);
109 const pgid = Number(fields[2]);
110 const startTime = fields[19];
111 if (!Number.isSafeInteger(pgid) || startTime === undefined) return undefined;
112 if (group !== undefined && pgid !== group) return undefined;
113 const cmdline = yield* fs.readFileString(`/proc/${pid}/cmdline`);
114 return {
115 pid,
116 pgid,
117 startTime,
118 command: cmdline.replace(/\0$/, "").replaceAll("\0", " "),
119 zombie: fields[0] === "Z",
120 } satisfies ObservedProcess;
121 });
122
123 const psDarwin = (selection: ReadonlyArray<string>) =>
124 spawner
125 .string(
126 ChildProcess.make(
127 "/bin/ps",
128 ["-ww", "-o", "pid=,pgid=,stat=,lstart=,args=", ...selection],
129 {
130 env: { LC_ALL: "C", TZ: "UTC" },
131 extendEnv: true,
132 stdin: "ignore",
133 stderr: "ignore",
134 },
135 ),
136 )
137 .pipe(Effect.timeout("5 seconds"), Effect.map(parseDarwinPs));
138
139 const observe = (pid: number): Effect.Effect<ObservedProcess | undefined> => {
140 if (platform === "linux") return observeLinux(pid).pipe(Effect.orElseSucceed(() => undefined));
141 if (platform === "darwin") {
142 return psDarwin(["-p", String(pid)]).pipe(
143 Effect.map((found) => found.find((candidate) => candidate.pid === pid)),
144 Effect.orElseSucceed(() => undefined),
145 );
146 }
147 return Effect.undefined;
148 };
149
150 const observeGroup = (pgid: number): Effect.Effect<ReadonlyArray<ObservedProcess>> => {
151 if (platform === "linux") {
152 return fs.readDirectory("/proc").pipe(
153 Effect.flatMap((names) =>
154 Effect.forEach(
155 names.filter((name) => /^\d+$/.test(name)),
156 (name) => observeLinux(Number(name), pgid).pipe(Effect.orElseSucceed(() => undefined)),
157 { concurrency: 16 },
158 ),
159 ),
160 Effect.map((found) => found.filter((candidate) => candidate !== undefined)),
161 Effect.orElseSucceed(() => []),
162 );
163 }
164 if (platform === "darwin") {
165 return psDarwin(["-A"]).pipe(
166 Effect.map((found) => found.filter((candidate) => candidate.pgid === pgid)),
167 Effect.orElseSucceed(() => []),
168 );
169 }
170 return Effect.succeed([]);
171 };
172
173 const isRunning = (identity: ProcessIdentity) =>
174 observe(identity.pid).pipe(
175 Effect.map(
176 (observed) =>
177 observed !== undefined && !observed.zombie && observed.startTime === identity.startTime,
178 ),
179 );
180
181 const ownerPid = input.ownerPid ?? process.pid;
182 const ownerProcess = yield* observe(ownerPid);
183 const owner =
184 ownerProcess === undefined ? undefined : { pid: ownerPid, startTime: ownerProcess.startTime };
185
186 const track: OpenCodeServerLedger["Service"]["track"] = Effect.fn("OpenCodeServerLedger.track")(
187 function* (server) {
188 if (!recordable) return Effect.void;
189 const pgid = server.pid;
190 // Any live member identifies a leaderless group; pgid cannot be reused while members remain.
191 const leader = yield* observe(pgid);
192 const members =
193 leader !== undefined && !leader.zombie && leader.pgid === pgid
194 ? [leader]
195 : (yield* observeGroup(pgid)).filter((candidate) => !candidate.zombie);
196 const member =
197 members.find((candidate) => candidate.command.endsWith(` ${server.args.join(" ")}`)) ??
198 members[0];
199 if (owner === undefined || member === undefined) {
200 if (groupExists(pgid)) {
201 yield* Effect.logWarning(
202 "Could not record an OpenCode server; it will keep running if this server crashes",
203 { pid: pgid, port: server.port },
204 );
205 }
206 return Effect.void;
207 }
208 const entryPath = path.join(directory, `${pgid}.json`);
209 const entry: OpenCodeServerEntry = {
210 version: 1,
211 pgid,
212 pid: member.pid,
213 startTime: member.startTime,
214 command: member.command,
215 port: server.port,
216 stateDir: input.stateDir,
217 owner,
218 };
219 yield* writeFileStringAtomically({
220 filePath: entryPath,
221 contents: yield* encodeEntry(entry),
222 }).pipe(
223 Effect.provideService(FileSystem.FileSystem, fs),
224 Effect.provideService(Path.Path, path),
225 );
226 return fs
227 .remove(entryPath, { force: true })
228 .pipe(
229 Effect.catchCause((cause) =>
230 Effect.logWarning("Could not forget a stopped OpenCode server", { cause }),
231 ),
232 );
233 },
234 Effect.catchCause((cause) =>
235 Effect.logWarning("Could not record an OpenCode server process", { cause }).pipe(
236 Effect.as(Effect.void),
237 ),
238 ),
239 );
240
241 // Signals only a group whose recorded member still has the same pid, start
242 // time, group, and full command line, so a recycled pid is never touched.
243 const stopOrphan = Effect.fn("OpenCodeServerLedger.stopOrphan")(function* (
244 entry: OpenCodeServerEntry,
245 ) {
246 const observed = yield* observe(entry.pid);
247 if (
248 observed === undefined ||
249 observed.zombie ||
250 observed.startTime !== entry.startTime ||
251 observed.pgid !== entry.pgid ||
252 observed.command !== entry.command
253 ) {
254 return;
255 }
256 yield* Effect.logInfo("Stopping an OpenCode server left by a previous T3 Code server", {
257 pid: entry.pgid,
258 port: entry.port,
259 });
260 signalGroup(entry.pgid, "SIGTERM");
261 for (let attempt = 0; attempt < STOP_POLL_ATTEMPTS && groupExists(entry.pgid); attempt++) {
262 yield* Effect.sleep(STOP_POLL_INTERVAL);
263 }
264 // The group never emptied, so its pgid cannot have been reused.
265 if (groupExists(entry.pgid)) signalGroup(entry.pgid, "SIGKILL");
266 });
267
268 const reapEntry = Effect.fn("OpenCodeServerLedger.reapEntry")(
269 function* (entryPath: string) {
270 const entry = decodeEntry(yield* fs.readFileString(entryPath));
271 if (Option.isSome(entry) && entry.value.stateDir === input.stateDir) {
272 if (yield* isRunning(entry.value.owner)) return;
273 yield* stopOrphan(entry.value);
274 }
275 yield* fs.remove(entryPath, { force: true });
276 },
277 (effect, entryPath) =>
278 effect.pipe(
279 Effect.catchCause((cause) =>
280 Effect.logWarning("Could not clean up a recorded OpenCode server", { entryPath, cause }),
281 ),
282 ),
283 );
284
285 const reapOrphans = Effect.gen(function* () {
286 const names = yield* fs.readDirectory(directory).pipe(Effect.orElseSucceed(() => []));
287 yield* Effect.forEach(
288 names.filter((name) => ENTRY_FILE.test(name)),
289 (name) => reapEntry(path.join(directory, name)),
290 { concurrency: "unbounded", discard: true },
291 );
292 });
293
294 return { track, reapOrphans };
295});
296
297export const layer = (input: { readonly stateDir: string }) =>
298 Layer.effect(
299 OpenCodeServerLedger,
300 Effect.gen(function* () {
301 const ledger = yield* make(input);
302 yield* ledger.reapOrphans.pipe(Effect.forkScoped);
303 return OpenCodeServerLedger.of({ track: ledger.track });
304 }),
305 );
306
307export const layerTest = Layer.succeed(
308 OpenCodeServerLedger,
309 OpenCodeServerLedger.of({
310 track: Effect.fn("OpenCodeServerLedger.track")(function () {
311 return Effect.succeed(Effect.void);
312 }),
313 }),
314);