ProcessGroupMetrics.ts

import type { ResourceTelemetryProcess } from "@t3tools/contracts";
import * as Cause from "effect/Cause";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Metric from "effect/Metric";
import * as Ref from "effect/Ref";

import * as ServerConfig from "../config.ts";
import { metricAttributes } from "../observability/Metrics.ts";
import * as ResourceTelemetry from "./ResourceTelemetry.ts";

const SAMPLE_INTERVAL = "1 minute";

const residentBytes = Metric.gauge("t3_process_group_resident_bytes", {
  description: "Resident memory of the server's processes, summed by group.",
  attributes: { unit: "By" },
});
const cpuPercent = Metric.gauge("t3_process_group_cpu_percent", {
  description: "CPU use of the server's processes, summed by group; 100 is one full core.",
  attributes: { unit: "%" },
});
const processCount = Metric.gauge("t3_process_group_count", {
  description: "Processes in each group.",
  attributes: { unit: "{process}" },
});

// Each direct child of the server names the group of its whole subtree. A fixed set keeps one
// series per group per server, however many agents and processes come and go.
const PROVIDER_GROUPS: ReadonlyArray<readonly [RegExp, string]> = [
  [/codex/, "codex"],
  [/claude/, "claude"],
  [/cursor/, "cursor"],
  [/opencode/, "opencode"],
  [/grok/, "grok"],
  [/antigravity|gemini/, "antigravity"],
];
const OTHER_GROUPS: ReadonlyArray<readonly [RegExp, string]> = [
  [/chrom/, "browser"],
  [/^(?:ba|z|fi|da|c)?sh(?:\.exe)?$|^pwsh|^powershell|^cmd\.exe$/, "terminal"],
];
// A CLI shipped as a script runs under a JavaScript runtime, so its process name is the
// runtime's (older Claude packages run `node .../@anthropic-ai/claude-code/cli.js`). Only those
// are matched by command line; any other name is matched by name alone, so a shell or `git`
// whose arguments mention a provider stays in its own group.
const SCRIPT_RUNTIME = /^(?:node|bun|deno)(?:\.exe)?$/;

const match = (groups: ReadonlyArray<readonly [RegExp, string]>, text: string) =>
  groups.find(([pattern]) => pattern.test(text))?.[1];

const childGroup = (name: string, command: string) => {
  const lowerName = name.toLowerCase();
  if (SCRIPT_RUNTIME.test(lowerName))
    return match(PROVIDER_GROUPS, command.toLowerCase()) ?? "other";
  return match(PROVIDER_GROUPS, lowerName) ?? match(OTHER_GROUPS, lowerName) ?? "other";
};

export interface ProcessGroupTotals {
  readonly residentBytes: number;
  readonly cpuPercent: number;
  readonly count: number;
}

export function groupProcesses(
  processes: ReadonlyArray<ResourceTelemetryProcess>,
): ReadonlyMap<string, ProcessGroupTotals> {
  const byPid = new Map(processes.map((process) => [process.identity.pid, process]));
  const serverPid = processes.find((process) => process.category === "server")?.identity.pid;
  const groupOf = (process: ResourceTelemetryProcess): string => {
    if (process.category === "server") return "server";
    if (process.category === "resource-monitor") return "resource-monitor";
    if (process.category.startsWith("electron-")) return "electron";
    let current = process;
    // Bounded by the table size, in case a reused PID makes the parent chain loop.
    for (let step = 0; step < processes.length; step++) {
      if (current.ppid === serverPid) return childGroup(current.name, current.command);
      const parent = byPid.get(current.ppid);
      if (parent === undefined) break;
      current = parent;
    }
    return "other";
  };
  const totals = new Map<string, ProcessGroupTotals>();
  for (const process of processes) {
    const group = groupOf(process);
    const total = totals.get(group) ?? { residentBytes: 0, cpuPercent: 0, count: 0 };
    totals.set(group, {
      residentBytes: total.residentBytes + process.residentBytes,
      cpuPercent: total.cpuPercent + process.cpuPercent,
      count: total.count + 1,
    });
  }
  return totals;
}

/**
 * Exports process totals by group while OTLP metrics export is on. Sampling walks the process
 * table, which an idle server otherwise never does, so only a server exporting metrics pays for
 * it, once a minute.
 */
export const layer = Layer.effectDiscard(
  Effect.gen(function* () {
    const config = yield* ServerConfig.ServerConfig;
    if (config.otlpMetricsUrl === undefined) return;
    const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
    // A group that empties reads zero rather than holding its last value.
    const seen = new Set<string>();
    const sample = Effect.fn("ProcessGroupMetrics.sample")(function* () {
      const totals = groupProcesses((yield* telemetry.refresh).processes);
      for (const group of totals.keys()) seen.add(group);
      for (const group of seen) {
        const total = totals.get(group) ?? { residentBytes: 0, cpuPercent: 0, count: 0 };
        const attributes = metricAttributes({ group });
        yield* Metric.update(Metric.withAttributes(residentBytes, attributes), total.residentBytes);
        yield* Metric.update(Metric.withAttributes(cpuPercent, attributes), total.cpuPercent);
        yield* Metric.update(Metric.withAttributes(processCount, attributes), total.count);
      }
    });
    // A failing collector would otherwise leave these series silently absent, so the first
    // failure and the recovery are logged; the repeats stay at debug.
    const failing = yield* Ref.make(false);
    const recordSuccess = Ref.getAndSet(failing, false).pipe(
      Effect.flatMap((wasFailing) =>
        wasFailing ? Effect.logInfo("process group metrics sample recovered") : Effect.void,
      ),
    );
    const recordFailure = (error: ResourceTelemetry.ResourceTelemetryRefreshFailed) => {
      const cause = Cause.fail(error);
      return Ref.getAndSet(failing, true).pipe(
        Effect.flatMap((wasFailing) =>
          wasFailing
            ? Effect.logDebug("process group metrics sample failed", cause)
            : Effect.logWarning("process group metrics are not exported: sample failed", cause),
        ),
        Effect.annotateLogs({ operation: error.operation }),
      );
    };
    yield* sample().pipe(
      Effect.andThen(recordSuccess),
      Effect.catch(recordFailure),
      Effect.andThen(Effect.sleep(SAMPLE_INTERVAL)),
      Effect.forever,
      Effect.forkScoped,
    );
  }),
);