1import type {
2 ResourceTelemetryProcessCategory,
3 ServerProcessDiagnosticsEntry,
4 ServerProcessDiagnosticsResult,
5 ServerProcessSignal,
6 ServerSignalProcessResult,
7} from "@t3tools/contracts";
8import * as Context from "effect/Context";
9import * as Effect from "effect/Effect";
10import * as Layer from "effect/Layer";
11import * as Option from "effect/Option";
12import * as Schema from "effect/Schema";
13
14import * as ResourceTelemetry from "../resourceTelemetry/ResourceTelemetry.ts";
15
16export class ProcessSignalFailed extends Schema.TaggedError<ProcessSignalFailed>()(
17 "ProcessSignalFailed",
18 {
19 pid: Schema.Number,
20 signal: Schema.String,
21 cause: Schema.Defect(),
22 },
23) {
24 override get message(): string {
25 return `Failed to signal process ${this.pid} with ${this.signal}.`;
26 }
27}
28
29export class ProcessDiagnostics extends Context.Service<
30 ProcessDiagnostics,
31 {
32 readonly read: Effect.Effect<ServerProcessDiagnosticsResult>;
33 readonly signal: (input: {
34 readonly pid: number;
35 readonly startTimeMs: number;
36 readonly signal: ServerProcessSignal;
37 }) => Effect.Effect<ServerSignalProcessResult>;
38 }
39>()("t3/diagnostics/ProcessDiagnostics") {}
40
41function formatElapsed(runTimeMs: number): string {
42 const totalSeconds = Math.max(0, Math.floor(runTimeMs / 1_000));
43 const hours = Math.floor(totalSeconds / 3_600);
44 const minutes = Math.floor((totalSeconds % 3_600) / 60);
45 const seconds = totalSeconds % 60;
46 return hours > 0
47 ? `${hours}:${String(minutes).padStart(2, "0")}:${String(seconds).padStart(2, "0")}`
48 : `${minutes}:${String(seconds).padStart(2, "0")}`;
49}
50
51function canSignalCategory(category: ResourceTelemetryProcessCategory): boolean {
52 return (
53 category === "server-child" || category === "provider-root" || category === "terminal-root"
54 );
55}
56
57/** @public Service construction is part of the canonical Effect module API. */
58export const make = Effect.fn("makeProcessDiagnostics")(function* () {
59 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
60 const refreshedTelemetry = telemetry.refresh.pipe(Effect.catch(() => telemetry.latest));
61 const read: ProcessDiagnostics["Service"]["read"] = refreshedTelemetry.pipe(
62 Effect.map((snapshot) => {
63 const processes = snapshot.processes
64 .filter((entry) => canSignalCategory(entry.category))
65 .map((entry): ServerProcessDiagnosticsEntry => ({
66 pid: entry.identity.pid,
67 startTimeMs: entry.identity.startTimeMs,
68 ppid: entry.ppid,
69 pgid: Option.none(),
70 status: entry.status || "Unknown",
71 cpuPercent: entry.cpuPercent,
72 rssBytes: entry.residentBytes,
73 elapsed: formatElapsed(entry.runTimeMs),
74 command: entry.command || entry.name || "unknown",
75 depth: Math.max(0, entry.depth - 1),
76 childPids: entry.childPids,
77 }));
78 return {
79 serverPid: process.pid,
80 readAt: snapshot.readAt,
81 processCount: processes.length,
82 totalRssBytes: processes.reduce((total, entry) => total + entry.rssBytes, 0),
83 totalCpuPercent: processes.reduce((total, entry) => total + entry.cpuPercent, 0),
84 processes,
85 error: Option.map(snapshot.health.native.lastError, (message) => ({ message })),
86 };
87 }),
88 );
89
90 const signal: ProcessDiagnostics["Service"]["signal"] = Effect.fn("ProcessDiagnostics.signal")(
91 function* (input) {
92 if (input.pid === process.pid) {
93 return {
94 pid: input.pid,
95 signal: input.signal,
96 signaled: false,
97 message: Option.some("Refusing to signal the T3 server process."),
98 };
99 }
100 const current = yield* telemetry.refresh.pipe(Effect.option);
101 if (Option.isNone(current)) {
102 return {
103 pid: input.pid,
104 signal: input.signal,
105 signaled: false,
106 message: Option.some(
107 `Could not refresh process ${input.pid}; refusing to signal a stale identity.`,
108 ),
109 };
110 }
111 const selected = current.value.processes.find(
112 (entry) =>
113 entry.identity.pid === input.pid && entry.identity.startTimeMs === input.startTimeMs,
114 );
115 if (!selected) {
116 return {
117 pid: input.pid,
118 signal: input.signal,
119 signaled: false,
120 message: Option.some(
121 `Process ${input.pid} no longer matches the selected process identity.`,
122 ),
123 };
124 }
125 if (!canSignalCategory(selected.category)) {
126 return {
127 pid: input.pid,
128 signal: input.signal,
129 signaled: false,
130 message: Option.some(`Process ${input.pid} is not a signalable T3 backend descendant.`),
131 };
132 }
133 return yield* Effect.try({
134 try: () => {
135 process.kill(input.pid, input.signal);
136 return {
137 pid: input.pid,
138 signal: input.signal,
139 signaled: true,
140 message: Option.none(),
141 };
142 },
143 catch: (cause) =>
144 new ProcessSignalFailed({
145 pid: input.pid,
146 signal: input.signal,
147 cause,
148 }),
149 }).pipe(
150 Effect.catch((error) =>
151 Effect.succeed({
152 pid: input.pid,
153 signal: input.signal,
154 signaled: false,
155 message: Option.some(
156 error instanceof Error ? error.message : "Failed to signal process.",
157 ),
158 }),
159 ),
160 );
161 },
162 );
163
164 return ProcessDiagnostics.of({ read, signal });
165});
166
167export const layer = Layer.effect(ProcessDiagnostics, make());