1import type {
2 DesktopElectronProcessMetric,
3 DesktopHostTelemetrySnapshot,
4 ResourceMonitorProcessSample,
5 ResourceMonitorSnapshotEvent,
6 ResourceTelemetryAggregate,
7 ResourceTelemetryProcess,
8 ResourceTelemetryProcessCategory,
9} from "@t3tools/contracts";
10import * as DateTime from "effect/DateTime";
11import * as Option from "effect/Option";
12
13const MAX_DELTA_INTERVAL_MS = 30_000;
14const ELECTRON_IDENTITY_TOLERANCE_MS = 2_000;
15
16export interface ProcessState {
17 readonly process: ResourceTelemetryProcess;
18 readonly sampledAtMs: number;
19}
20
21export interface GroupCounters {
22 readonly cpuTimeMs: number;
23 readonly ioReadBytes: number;
24 readonly ioWriteBytes: number;
25 readonly processStarts: number;
26 readonly processExits: number;
27}
28
29export interface TelemetryCounters {
30 readonly backend: GroupCounters;
31 readonly electron: GroupCounters;
32 readonly monitor: GroupCounters;
33 readonly allT3: GroupCounters;
34}
35
36export interface ProcessDelta {
37 readonly identityKey: string;
38 readonly category: ResourceTelemetryProcessCategory;
39 readonly cpuTimeMs: number;
40 readonly ioReadBytes: number;
41 readonly ioWriteBytes: number;
42}
43
44export interface MergeProcessesInput {
45 readonly serverPid: number;
46 readonly sidecarPid: Option.Option<number>;
47 readonly fallbackSampledAtMs: number;
48 readonly nativeSnapshot: Option.Option<ResourceMonitorSnapshotEvent>;
49 readonly desktopSnapshot: Option.Option<DesktopHostTelemetrySnapshot>;
50 readonly electronRootPids?: ReadonlySet<number>;
51 readonly electronRootStartTimes?: ReadonlyMap<number, number>;
52 readonly previous: ReadonlyMap<string, ProcessState>;
53 readonly counters: TelemetryCounters;
54 readonly updatePrevious: boolean;
55 /** History replay needs process deltas, but does not publish lifetime group counters. */
56 readonly accumulateCounters?: boolean;
57}
58
59export interface MergeProcessesResult {
60 readonly sampledAtMs: number;
61 readonly processes: ReadonlyArray<ResourceTelemetryProcess>;
62 readonly previous: ReadonlyMap<string, ProcessState>;
63 readonly counters: TelemetryCounters;
64 readonly groups: {
65 readonly backend: ResourceTelemetryAggregate;
66 readonly electron: ResourceTelemetryAggregate;
67 readonly monitor: ResourceTelemetryAggregate;
68 readonly allT3: ResourceTelemetryAggregate;
69 };
70 readonly deltas: ReadonlyArray<ProcessDelta>;
71}
72
73const emptyGroupCounters = (): GroupCounters => ({
74 cpuTimeMs: 0,
75 ioReadBytes: 0,
76 ioWriteBytes: 0,
77 processStarts: 0,
78 processExits: 0,
79});
80
81export const emptyTelemetryCounters = (): TelemetryCounters => ({
82 backend: emptyGroupCounters(),
83 electron: emptyGroupCounters(),
84 monitor: emptyGroupCounters(),
85 allT3: emptyGroupCounters(),
86});
87
88export function processIdentityKey(pid: number, startTimeMs: number): string {
89 return `${pid}:${startTimeMs}`;
90}
91
92function finiteNonNegative(value: number): number {
93 return Number.isFinite(value) ? Math.max(0, value) : 0;
94}
95
96function categoryGroup(
97 category: ResourceTelemetryProcessCategory,
98): "backend" | "electron" | "monitor" {
99 if (category === "resource-monitor") return "monitor";
100 if (category.startsWith("electron-")) return "electron";
101 return "backend";
102}
103
104function electronCategory(metric: DesktopElectronProcessMetric): ResourceTelemetryProcessCategory {
105 switch (metric.type) {
106 case "Browser":
107 return "electron-main";
108 case "Tab":
109 return "electron-renderer";
110 case "GPU":
111 return "electron-gpu";
112 default:
113 return "electron-utility";
114 }
115}
116
117function inferredElectronCategory(
118 process: ResourceMonitorProcessSample,
119): ResourceTelemetryProcessCategory {
120 const command = process.command.toLowerCase();
121 if (command.includes("--type=renderer")) return "electron-renderer";
122 if (command.includes("--type=gpu-process")) return "electron-gpu";
123 return "electron-utility";
124}
125
126function matchElectronMetric(
127 process: ResourceMonitorProcessSample,
128 metricsByPid: ReadonlyMap<number, DesktopElectronProcessMetric>,
129): DesktopElectronProcessMetric | undefined {
130 const metric = metricsByPid.get(process.pid);
131 if (!metric) return undefined;
132 return Math.abs(metric.creationTimeMs - process.startTimeMs) <= ELECTRON_IDENTITY_TOLERANCE_MS
133 ? metric
134 : undefined;
135}
136
137function syntheticNativeSample(
138 metric: DesktopElectronProcessMetric,
139 sampledAtMs: number,
140 previous: ProcessState | undefined,
141): ResourceMonitorProcessSample {
142 const cpuTimeMs =
143 metric.cumulativeCpuSeconds !== undefined
144 ? Math.max(0, Math.round(metric.cumulativeCpuSeconds * 1_000))
145 : previous
146 ? previous.process.cpuTimeMs +
147 Math.max(0, ((sampledAtMs - previous.sampledAtMs) * metric.cpuPercent) / 100)
148 : 0;
149 return {
150 pid: metric.pid,
151 ppid: 0,
152 startTimeMs: metric.creationTimeMs,
153 runTimeMs: Math.max(0, sampledAtMs - metric.creationTimeMs),
154 name: metric.name ?? metric.serviceName ?? metric.type,
155 command: metric.name ?? metric.serviceName ?? metric.type,
156 status: "Running",
157 cpuPercent: metric.cpuPercent,
158 cpuTimeMs,
159 residentBytes: metric.workingSetBytes,
160 virtualBytes: 0,
161 ioReadBytes: 0,
162 ioWriteBytes: 0,
163 ioSemantics: "storage",
164 };
165}
166
167function processDepths(
168 processes: ReadonlyArray<ResourceMonitorProcessSample>,
169 roots: ReadonlySet<number>,
170): ReadonlyMap<number, number> {
171 const childrenByParent = new Map<number, number[]>();
172 for (const process of processes) {
173 const children = childrenByParent.get(process.ppid) ?? [];
174 children.push(process.pid);
175 childrenByParent.set(process.ppid, children);
176 }
177
178 const depths = new Map<number, number>();
179 const queue = [...roots].map((pid) => ({ pid, depth: 0 }));
180 while (queue.length > 0) {
181 const current = queue.shift();
182 if (!current || depths.has(current.pid)) continue;
183 depths.set(current.pid, current.depth);
184 for (const childPid of childrenByParent.get(current.pid) ?? []) {
185 queue.push({ pid: childPid, depth: current.depth + 1 });
186 }
187 }
188 return depths;
189}
190
191function isElectronDescendant(
192 pid: number,
193 processesByPid: ReadonlyMap<number, ResourceMonitorProcessSample>,
194 electronPids: ReadonlySet<number>,
195): boolean {
196 const visited = new Set<number>();
197 let currentPid = pid;
198 while (!visited.has(currentPid)) {
199 visited.add(currentPid);
200 if (electronPids.has(currentPid)) return true;
201 const current = processesByPid.get(currentPid);
202 if (!current || current.ppid <= 0 || current.ppid === currentPid) return false;
203 currentPid = current.ppid;
204 }
205 return false;
206}
207
208function hasElectronAncestor(
209 process: ResourceMonitorProcessSample,
210 processesByPid: ReadonlyMap<number, ResourceMonitorProcessSample>,
211 electronPids: ReadonlySet<number>,
212): boolean {
213 const visited = new Set<number>();
214 let currentPid = process.ppid;
215 while (currentPid > 0 && !visited.has(currentPid)) {
216 visited.add(currentPid);
217 if (electronPids.has(currentPid)) return true;
218 const current = processesByPid.get(currentPid);
219 if (!current || current.ppid === currentPid) return false;
220 currentPid = current.ppid;
221 }
222 return false;
223}
224
225function orderProcessTree(
226 processes: ReadonlyArray<ResourceTelemetryProcess>,
227 rootPids: ReadonlyArray<number>,
228): ReadonlyArray<ResourceTelemetryProcess> {
229 const processesByPid = new Map(processes.map((process) => [process.identity.pid, process]));
230 const childrenByParent = new Map<number, ResourceTelemetryProcess[]>();
231 for (const process of processes) {
232 const children = childrenByParent.get(process.ppid) ?? [];
233 children.push(process);
234 childrenByParent.set(process.ppid, children);
235 }
236 for (const children of childrenByParent.values()) {
237 children.sort((left, right) => left.identity.pid - right.identity.pid);
238 }
239
240 const ordered: ResourceTelemetryProcess[] = [];
241 const visited = new Set<number>();
242 const visit = (process: ResourceTelemetryProcess): void => {
243 if (visited.has(process.identity.pid)) return;
244 visited.add(process.identity.pid);
245 ordered.push(process);
246 for (const child of childrenByParent.get(process.identity.pid) ?? []) {
247 visit(child);
248 }
249 };
250
251 for (const rootPid of rootPids) {
252 const root = processesByPid.get(rootPid);
253 if (root) visit(root);
254 }
255 for (const process of processes.toSorted(
256 (left, right) => left.depth - right.depth || left.identity.pid - right.identity.pid,
257 )) {
258 visit(process);
259 }
260 return ordered;
261}
262
263function delta(input: {
264 readonly current: number;
265 readonly previous: number;
266 readonly elapsedMs: number;
267}): number {
268 if (
269 input.elapsedMs <= 0 ||
270 input.elapsedMs > MAX_DELTA_INTERVAL_MS ||
271 input.current < input.previous
272 ) {
273 return 0;
274 }
275 return input.current - input.previous;
276}
277
278function incrementCounters(counters: GroupCounters, update: Partial<GroupCounters>): GroupCounters {
279 return {
280 cpuTimeMs: counters.cpuTimeMs + (update.cpuTimeMs ?? 0),
281 ioReadBytes: counters.ioReadBytes + (update.ioReadBytes ?? 0),
282 ioWriteBytes: counters.ioWriteBytes + (update.ioWriteBytes ?? 0),
283 processStarts: counters.processStarts + (update.processStarts ?? 0),
284 processExits: counters.processExits + (update.processExits ?? 0),
285 };
286}
287
288function applyLifecycleCounters(input: {
289 readonly counters: TelemetryCounters;
290 readonly deltas: ReadonlyArray<ProcessDelta>;
291 readonly current: ReadonlyMap<string, ProcessState>;
292 readonly previous: ReadonlyMap<string, ProcessState>;
293}): TelemetryCounters {
294 let backend = input.counters.backend;
295 let electron = input.counters.electron;
296 let monitor = input.counters.monitor;
297 let allT3 = input.counters.allT3;
298 for (const processDelta of input.deltas) {
299 const group = categoryGroup(processDelta.category);
300 switch (group) {
301 case "backend":
302 backend = incrementCounters(backend, processDelta);
303 break;
304 case "electron":
305 electron = incrementCounters(electron, processDelta);
306 break;
307 case "monitor":
308 monitor = incrementCounters(monitor, processDelta);
309 break;
310 }
311 allT3 = incrementCounters(allT3, processDelta);
312 }
313
314 for (const [identityKey, current] of input.current) {
315 if (input.previous.has(identityKey)) continue;
316 const group = categoryGroup(current.process.category);
317 switch (group) {
318 case "backend":
319 backend = incrementCounters(backend, { processStarts: 1 });
320 break;
321 case "electron":
322 electron = incrementCounters(electron, { processStarts: 1 });
323 break;
324 case "monitor":
325 monitor = incrementCounters(monitor, { processStarts: 1 });
326 break;
327 }
328 allT3 = incrementCounters(allT3, { processStarts: 1 });
329 }
330
331 for (const [identityKey, previous] of input.previous) {
332 if (input.current.has(identityKey)) continue;
333 const group = categoryGroup(previous.process.category);
334 switch (group) {
335 case "backend":
336 backend = incrementCounters(backend, { processExits: 1 });
337 break;
338 case "electron":
339 electron = incrementCounters(electron, { processExits: 1 });
340 break;
341 case "monitor":
342 monitor = incrementCounters(monitor, { processExits: 1 });
343 break;
344 }
345 allT3 = incrementCounters(allT3, { processExits: 1 });
346 }
347
348 return { backend, electron, monitor, allT3 };
349}
350
351function aggregate(
352 processes: ReadonlyArray<ResourceTelemetryProcess>,
353 counters: GroupCounters,
354): ResourceTelemetryAggregate {
355 return {
356 processCount: processes.length,
357 currentCpuPercent: processes.reduce((total, process) => total + process.cpuPercent, 0),
358 cpuTimeMs: counters.cpuTimeMs,
359 currentRssBytes: processes.reduce((total, process) => total + process.residentBytes, 0),
360 peakRssBytes: processes.reduce((total, process) => total + process.peakResidentBytes, 0),
361 ioReadBytes: counters.ioReadBytes,
362 ioWriteBytes: counters.ioWriteBytes,
363 ioReadBytesPerSecond: processes.reduce(
364 (total, process) => total + process.ioReadBytesPerSecond,
365 0,
366 ),
367 ioWriteBytesPerSecond: processes.reduce(
368 (total, process) => total + process.ioWriteBytesPerSecond,
369 0,
370 ),
371 processStarts: counters.processStarts,
372 processExits: counters.processExits,
373 };
374}
375
376export function mergeProcesses(input: MergeProcessesInput): MergeProcessesResult {
377 const nativeProcesses = Option.match(input.nativeSnapshot, {
378 onNone: () => [] as ReadonlyArray<ResourceMonitorProcessSample>,
379 onSome: (snapshot) => snapshot.processes,
380 });
381 const electronMetrics = Option.match(input.desktopSnapshot, {
382 onNone: () => [] as ReadonlyArray<DesktopElectronProcessMetric>,
383 onSome: (snapshot) => snapshot.electronProcesses,
384 });
385 const sampledAtMs = Option.match(input.nativeSnapshot, {
386 onNone: () =>
387 Option.match(input.desktopSnapshot, {
388 onNone: () => input.fallbackSampledAtMs,
389 onSome: (snapshot) => snapshot.sampledAtUnixMs,
390 }),
391 onSome: (native) =>
392 Option.match(input.desktopSnapshot, {
393 onNone: () => native.sampledAtUnixMs,
394 onSome: (desktop) => Math.max(native.sampledAtUnixMs, desktop.sampledAtUnixMs),
395 }),
396 });
397 const nativeSampledAtMs = Option.map(
398 input.nativeSnapshot,
399 (snapshot) => snapshot.sampledAtUnixMs,
400 );
401 const desktopSampledAtMs = Option.map(
402 input.desktopSnapshot,
403 (snapshot) => snapshot.sampledAtUnixMs,
404 );
405 const nativeProcessPids = new Set(nativeProcesses.map((process) => process.pid));
406 const nativeByPid = new Map(nativeProcesses.map((process) => [process.pid, process]));
407 const metricsByPid = new Map<number, DesktopElectronProcessMetric>();
408 for (const metric of electronMetrics) {
409 const nativeProcess = nativeByPid.get(metric.pid);
410 if (!nativeProcess) {
411 nativeByPid.set(
412 metric.pid,
413 syntheticNativeSample(
414 metric,
415 Option.getOrElse(desktopSampledAtMs, () => sampledAtMs),
416 input.previous.get(processIdentityKey(metric.pid, metric.creationTimeMs)),
417 ),
418 );
419 metricsByPid.set(metric.pid, metric);
420 continue;
421 }
422 if (
423 Math.abs(metric.creationTimeMs - nativeProcess.startTimeMs) <= ELECTRON_IDENTITY_TOLERANCE_MS
424 ) {
425 metricsByPid.set(metric.pid, metric);
426 }
427 }
428 const processes = [...nativeByPid.values()];
429 const processesByPid = new Map(processes.map((process) => [process.pid, process]));
430 const requestedElectronRootPids = input.electronRootPids ?? new Set<number>();
431 const explicitElectronRootPids = new Set(
432 [...requestedElectronRootPids].filter((pid) => {
433 const process = processesByPid.get(pid);
434 if (!process) return false;
435 const expectedStartTime = input.electronRootStartTimes?.get(pid);
436 if (expectedStartTime !== undefined) {
437 return Math.abs(process.startTimeMs - expectedStartTime) <= ELECTRON_IDENTITY_TOLERANCE_MS;
438 }
439 if (metricsByPid.has(pid)) return true;
440 return [...input.previous.values()].some(
441 (previous) =>
442 previous.process.category === "electron-main" &&
443 previous.process.identity.pid === pid &&
444 previous.process.identity.startTimeMs === process.startTimeMs,
445 );
446 }),
447 );
448 const electronPids = new Set([...metricsByPid.keys(), ...explicitElectronRootPids]);
449 const electronRootPids = [
450 ...explicitElectronRootPids,
451 ...[...electronPids]
452 .filter((pid) => {
453 if (explicitElectronRootPids.has(pid)) return false;
454 const process = processesByPid.get(pid);
455 return process === undefined
456 ? true
457 : !hasElectronAncestor(process, processesByPid, electronPids);
458 })
459 .toSorted((left, right) => left - right),
460 ].filter((pid, index, values) => values.indexOf(pid) === index);
461 const rootPids = [input.serverPid, ...electronRootPids];
462 const roots = new Set(rootPids);
463 const depths = processDepths(processes, roots);
464 const childrenByParent = new Map<number, number[]>();
465 for (const process of processes) {
466 const children = childrenByParent.get(process.ppid) ?? [];
467 children.push(process.pid);
468 childrenByParent.set(process.ppid, children);
469 }
470
471 const nextPrevious = new Map<string, ProcessState>();
472 const processDeltas: ProcessDelta[] = [];
473 const normalized = processes.map((process): ResourceTelemetryProcess => {
474 const identityKey = processIdentityKey(process.pid, process.startTimeMs);
475 const previous = input.previous.get(identityKey);
476 const counterSampledAtMs = nativeProcessPids.has(process.pid)
477 ? Option.getOrElse(nativeSampledAtMs, () => sampledAtMs)
478 : Option.getOrElse(desktopSampledAtMs, () => sampledAtMs);
479 const elapsedMs = previous ? counterSampledAtMs - previous.sampledAtMs : 0;
480 const cpuTimeDelta = previous
481 ? delta({
482 current: process.cpuTimeMs,
483 previous: previous.process.cpuTimeMs,
484 elapsedMs,
485 })
486 : 0;
487 const ioReadDelta = previous
488 ? delta({
489 current: process.ioReadBytes,
490 previous: previous.process.ioReadBytes,
491 elapsedMs,
492 })
493 : 0;
494 const ioWriteDelta = previous
495 ? delta({
496 current: process.ioWriteBytes,
497 previous: previous.process.ioWriteBytes,
498 elapsedMs,
499 })
500 : 0;
501 const electronMetric = matchElectronMetric(process, metricsByPid);
502 const category: ResourceTelemetryProcessCategory =
503 process.pid === input.serverPid
504 ? "server"
505 : Option.contains(input.sidecarPid, process.pid)
506 ? "resource-monitor"
507 : explicitElectronRootPids.has(process.pid)
508 ? "electron-main"
509 : electronMetric
510 ? electronCategory(electronMetric)
511 : isElectronDescendant(process.pid, processesByPid, electronPids)
512 ? inferredElectronCategory(process)
513 : "server-child";
514 const firstSeenAt = previous?.process.firstSeenAt ?? DateTime.makeUnsafe(sampledAtMs);
515 const preservePreviousRates = !input.updatePrevious && previous !== undefined;
516 const cpuPercent = preservePreviousRates
517 ? previous.process.cpuPercent
518 : previous && elapsedMs > 0 && elapsedMs <= MAX_DELTA_INTERVAL_MS
519 ? (cpuTimeDelta / elapsedMs) * 100
520 : finiteNonNegative(process.cpuPercent);
521 const normalizedProcess: ResourceTelemetryProcess = {
522 identity: {
523 pid: process.pid,
524 startTimeMs: process.startTimeMs,
525 },
526 ppid: process.ppid,
527 childPids: [...(childrenByParent.get(process.pid) ?? [])].toSorted(
528 (left, right) => left - right,
529 ),
530 depth: depths.get(process.pid) ?? 0,
531 name: process.name,
532 command: process.command,
533 status: process.status,
534 category,
535 ...(electronMetric ? { electronType: electronMetric.type } : {}),
536 ...(electronMetric?.serviceName ? { electronServiceName: electronMetric.serviceName } : {}),
537 cpuPercent: finiteNonNegative(cpuPercent),
538 cpuTimeMs: process.cpuTimeMs,
539 residentBytes: process.residentBytes,
540 peakResidentBytes: Math.max(
541 process.residentBytes,
542 electronMetric?.peakWorkingSetBytes ?? 0,
543 previous?.process.peakResidentBytes ?? 0,
544 ),
545 virtualBytes: process.virtualBytes,
546 ioReadBytes: process.ioReadBytes,
547 ioWriteBytes: process.ioWriteBytes,
548 ioReadBytesPerSecond: preservePreviousRates
549 ? previous.process.ioReadBytesPerSecond
550 : elapsedMs > 0
551 ? finiteNonNegative((ioReadDelta * 1_000) / elapsedMs)
552 : 0,
553 ioWriteBytesPerSecond: preservePreviousRates
554 ? previous.process.ioWriteBytesPerSecond
555 : elapsedMs > 0
556 ? finiteNonNegative((ioWriteDelta * 1_000) / elapsedMs)
557 : 0,
558 ioSemantics: process.ioSemantics,
559 ...(electronMetric ? { idleWakeupsPerSecond: electronMetric.idleWakeupsPerSecond } : {}),
560 runTimeMs: process.runTimeMs,
561 firstSeenAt,
562 lastSeenAt: DateTime.makeUnsafe(sampledAtMs),
563 };
564 nextPrevious.set(identityKey, {
565 process: normalizedProcess,
566 sampledAtMs: counterSampledAtMs,
567 });
568 processDeltas.push({
569 identityKey,
570 category,
571 cpuTimeMs: cpuTimeDelta,
572 ioReadBytes: ioReadDelta,
573 ioWriteBytes: ioWriteDelta,
574 });
575 return normalizedProcess;
576 });
577 const ordered = orderProcessTree(normalized, rootPids);
578
579 const counters =
580 input.updatePrevious && input.accumulateCounters !== false
581 ? applyLifecycleCounters({
582 counters: input.counters,
583 deltas: processDeltas,
584 current: nextPrevious,
585 previous: input.previous,
586 })
587 : input.counters;
588 const backendProcesses = ordered.filter(
589 (process) => categoryGroup(process.category) === "backend",
590 );
591 const electronProcesses = ordered.filter(
592 (process) => categoryGroup(process.category) === "electron",
593 );
594 const monitorProcesses = ordered.filter(
595 (process) => categoryGroup(process.category) === "monitor",
596 );
597
598 return {
599 sampledAtMs,
600 processes: ordered,
601 previous: input.updatePrevious ? nextPrevious : input.previous,
602 counters,
603 groups: {
604 backend: aggregate(backendProcesses, counters.backend),
605 electron: aggregate(electronProcesses, counters.electron),
606 monitor: aggregate(monitorProcesses, counters.monitor),
607 allT3: aggregate(ordered, counters.allT3),
608 },
609 deltas: processDeltas,
610 };
611}