1import type { ResourceTelemetryProcess } from "@t3tools/contracts";
2import * as Cause from "effect/Cause";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Metric from "effect/Metric";
6import * as Ref from "effect/Ref";
7
8import * as ServerConfig from "../config.ts";
9import { metricAttributes } from "../observability/Metrics.ts";
10import * as ResourceTelemetry from "./ResourceTelemetry.ts";
11
12const SAMPLE_INTERVAL = "1 minute";
13
14const residentBytes = Metric.gauge("t3_process_group_resident_bytes", {
15 description: "Resident memory of the server's processes, summed by group.",
16 attributes: { unit: "By" },
17});
18const cpuPercent = Metric.gauge("t3_process_group_cpu_percent", {
19 description: "CPU use of the server's processes, summed by group; 100 is one full core.",
20 attributes: { unit: "%" },
21});
22const processCount = Metric.gauge("t3_process_group_count", {
23 description: "Processes in each group.",
24 attributes: { unit: "{process}" },
25});
26
27// Each direct child of the server names the group of its whole subtree. A fixed set keeps one
28// series per group per server, however many agents and processes come and go.
29const PROVIDER_GROUPS: ReadonlyArray<readonly [RegExp, string]> = [
30 [/codex/, "codex"],
31 [/claude/, "claude"],
32 [/cursor/, "cursor"],
33 [/opencode/, "opencode"],
34 [/grok/, "grok"],
35 [/antigravity|gemini/, "antigravity"],
36];
37const OTHER_GROUPS: ReadonlyArray<readonly [RegExp, string]> = [
38 [/chrom/, "browser"],
39 [/^(?:ba|z|fi|da|c)?sh(?:\.exe)?$|^pwsh|^powershell|^cmd\.exe$/, "terminal"],
40];
41// A CLI shipped as a script runs under a JavaScript runtime, so its process name is the
42// runtime's (older Claude packages run `node .../@anthropic-ai/claude-code/cli.js`). Only those
43// are matched by command line; any other name is matched by name alone, so a shell or `git`
44// whose arguments mention a provider stays in its own group.
45const SCRIPT_RUNTIME = /^(?:node|bun|deno)(?:\.exe)?$/;
46
47const match = (groups: ReadonlyArray<readonly [RegExp, string]>, text: string) =>
48 groups.find(([pattern]) => pattern.test(text))?.[1];
49
50const childGroup = (name: string, command: string) => {
51 const lowerName = name.toLowerCase();
52 if (SCRIPT_RUNTIME.test(lowerName))
53 return match(PROVIDER_GROUPS, command.toLowerCase()) ?? "other";
54 return match(PROVIDER_GROUPS, lowerName) ?? match(OTHER_GROUPS, lowerName) ?? "other";
55};
56
57export interface ProcessGroupTotals {
58 readonly residentBytes: number;
59 readonly cpuPercent: number;
60 readonly count: number;
61}
62
63export function groupProcesses(
64 processes: ReadonlyArray<ResourceTelemetryProcess>,
65): ReadonlyMap<string, ProcessGroupTotals> {
66 const byPid = new Map(processes.map((process) => [process.identity.pid, process]));
67 const serverPid = processes.find((process) => process.category === "server")?.identity.pid;
68 const groupOf = (process: ResourceTelemetryProcess): string => {
69 if (process.category === "server") return "server";
70 if (process.category === "resource-monitor") return "resource-monitor";
71 if (process.category.startsWith("electron-")) return "electron";
72 let current = process;
73 // Bounded by the table size, in case a reused PID makes the parent chain loop.
74 for (let step = 0; step < processes.length; step++) {
75 if (current.ppid === serverPid) return childGroup(current.name, current.command);
76 const parent = byPid.get(current.ppid);
77 if (parent === undefined) break;
78 current = parent;
79 }
80 return "other";
81 };
82 const totals = new Map<string, ProcessGroupTotals>();
83 for (const process of processes) {
84 const group = groupOf(process);
85 const total = totals.get(group) ?? { residentBytes: 0, cpuPercent: 0, count: 0 };
86 totals.set(group, {
87 residentBytes: total.residentBytes + process.residentBytes,
88 cpuPercent: total.cpuPercent + process.cpuPercent,
89 count: total.count + 1,
90 });
91 }
92 return totals;
93}
94
95/**
96 * Exports process totals by group while OTLP metrics export is on. Sampling walks the process
97 * table, which an idle server otherwise never does, so only a server exporting metrics pays for
98 * it, once a minute.
99 */
100export const layer = Layer.effectDiscard(
101 Effect.gen(function* () {
102 const config = yield* ServerConfig.ServerConfig;
103 if (config.otlpMetricsUrl === undefined) return;
104 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
105 // A group that empties reads zero rather than holding its last value.
106 const seen = new Set<string>();
107 const sample = Effect.fn("ProcessGroupMetrics.sample")(function* () {
108 const totals = groupProcesses((yield* telemetry.refresh).processes);
109 for (const group of totals.keys()) seen.add(group);
110 for (const group of seen) {
111 const total = totals.get(group) ?? { residentBytes: 0, cpuPercent: 0, count: 0 };
112 const attributes = metricAttributes({ group });
113 yield* Metric.update(Metric.withAttributes(residentBytes, attributes), total.residentBytes);
114 yield* Metric.update(Metric.withAttributes(cpuPercent, attributes), total.cpuPercent);
115 yield* Metric.update(Metric.withAttributes(processCount, attributes), total.count);
116 }
117 });
118 // A failing collector would otherwise leave these series silently absent, so the first
119 // failure and the recovery are logged; the repeats stay at debug.
120 const failing = yield* Ref.make(false);
121 const recordSuccess = Ref.getAndSet(failing, false).pipe(
122 Effect.flatMap((wasFailing) =>
123 wasFailing ? Effect.logInfo("process group metrics sample recovered") : Effect.void,
124 ),
125 );
126 const recordFailure = (error: ResourceTelemetry.ResourceTelemetryRefreshFailed) => {
127 const cause = Cause.fail(error);
128 return Ref.getAndSet(failing, true).pipe(
129 Effect.flatMap((wasFailing) =>
130 wasFailing
131 ? Effect.logDebug("process group metrics sample failed", cause)
132 : Effect.logWarning("process group metrics are not exported: sample failed", cause),
133 ),
134 Effect.annotateLogs({ operation: error.operation }),
135 );
136 };
137 yield* sample().pipe(
138 Effect.andThen(recordSuccess),
139 Effect.catch(recordFailure),
140 Effect.andThen(Effect.sleep(SAMPLE_INTERVAL)),
141 Effect.forever,
142 Effect.forkScoped,
143 );
144 }),
145);