1import type {
2 DesktopHostTelemetrySnapshot,
3 ResourceMonitorProcessSample,
4 ResourceMonitorSnapshotEvent,
5} from "@t3tools/contracts";
6import { describe, expect, it } from "@effect/vitest";
7import * as DateTime from "effect/DateTime";
8import * as Deferred from "effect/Deferred";
9import * as Effect from "effect/Effect";
10import * as Fiber from "effect/Fiber";
11import * as Layer from "effect/Layer";
12import * as Option from "effect/Option";
13import * as PubSub from "effect/PubSub";
14import * as Ref from "effect/Ref";
15import * as Stream from "effect/Stream";
16
17import * as DesktopTelemetryReceiver from "./DesktopTelemetryReceiver.ts";
18import * as NativeTelemetryClient from "./NativeTelemetryClient.ts";
19import * as ResourceAttribution from "./ResourceAttribution.ts";
20import * as ResourceTelemetry from "./ResourceTelemetry.ts";
21
22// ResourceTelemetry classifies process.pid as the server, so fixture PIDs are
23// offsets from it. Fixed PIDs collide whenever the test worker is assigned one.
24const SERVER_PID = process.pid;
25const CODEX_PID = SERVER_PID + 1;
26const SHORT_LIVED_CHILD_PID = SERVER_PID + 2;
27const ELECTRON_PID = SERVER_PID + 3;
28const MONITOR_PID = SERVER_PID + 4;
29
30function processSample(
31 input: Partial<ResourceMonitorProcessSample> &
32 Pick<ResourceMonitorProcessSample, "pid" | "ppid" | "startTimeMs">,
33): ResourceMonitorProcessSample {
34 return {
35 runTimeMs: 1_000,
36 name: `process-${input.pid}`,
37 command: `process-${input.pid}`,
38 status: "Running",
39 cpuPercent: 0,
40 cpuTimeMs: 0,
41 residentBytes: 1_024,
42 virtualBytes: 2_048,
43 ioReadBytes: 0,
44 ioWriteBytes: 0,
45 ioSemantics: "storage",
46 ...input,
47 };
48}
49
50function nativeSnapshot(input: {
51 readonly sequence: number;
52 readonly sampledAtUnixMs: number;
53 readonly childCpuTimeMs: number;
54 readonly childWriteBytes: number;
55 readonly externalProcesses?: ResourceMonitorSnapshotEvent["externalProcesses"];
56}): ResourceMonitorSnapshotEvent {
57 const processes = [
58 processSample({
59 pid: SERVER_PID,
60 ppid: 1,
61 startTimeMs: 100,
62 cpuTimeMs: input.sequence * 10,
63 }),
64 processSample({
65 pid: CODEX_PID,
66 ppid: SERVER_PID,
67 startTimeMs: 200,
68 name: "codex",
69 command: "codex app-server",
70 cpuTimeMs: input.childCpuTimeMs,
71 ioWriteBytes: input.childWriteBytes,
72 }),
73 processSample({
74 pid: ELECTRON_PID,
75 ppid: 1,
76 startTimeMs: 300,
77 name: "electron",
78 command: "electron",
79 cpuTimeMs: input.sequence * 20,
80 }),
81 processSample({
82 pid: MONITOR_PID,
83 ppid: SERVER_PID,
84 startTimeMs: 400,
85 name: "t3-resource-monitor",
86 command: "t3-resource-monitor",
87 cpuTimeMs: input.sequence * 5,
88 }),
89 ];
90 return {
91 version: 3,
92 type: "snapshot",
93 sequence: input.sequence,
94 sampledAtUnixMs: input.sampledAtUnixMs,
95 collectionDurationMicros: 300,
96 scannedProcessCount: 80,
97 retainedProcessCount: processes.length,
98 inaccessibleProcessCount: 1,
99 ...(input.externalProcesses === undefined
100 ? {}
101 : { externalProcesses: input.externalProcesses }),
102 processes,
103 };
104}
105
106function nativeGeneration(
107 snapshot: ResourceMonitorSnapshotEvent,
108 generation: number,
109): NativeTelemetryClient.NativeTelemetrySnapshot {
110 return { generation, snapshot };
111}
112
113function desktopSnapshot(sampledAtUnixMs: number): DesktopHostTelemetrySnapshot {
114 const sampledAt = DateTime.makeUnsafe(sampledAtUnixMs);
115 return {
116 version: 1,
117 type: "desktopTelemetry",
118 sequence: 1,
119 sampledAtUnixMs,
120 electronPid: ELECTRON_PID,
121 power: {
122 source: "electron-main",
123 idle: "false",
124 idleSeconds: 2,
125 locked: "false",
126 suspended: false,
127 onBattery: "true",
128 lowPowerMode: "unknown",
129 thermalState: "fair",
130 stale: false,
131 updatedAt: sampledAt,
132 },
133 speedLimitPercent: Option.some(90),
134 electronProcesses: [
135 {
136 pid: ELECTRON_PID,
137 creationTimeMs: 300,
138 type: "Browser",
139 name: "electron",
140 cpuPercent: 2,
141 cumulativeCpuSeconds: 0.02,
142 idleWakeupsPerSecond: 3,
143 workingSetBytes: 4_096,
144 peakWorkingSetBytes: 8_192,
145 },
146 ],
147 };
148}
149
150describe("ResourceTelemetry", () => {
151 it.effect("replays churning history and retains baselines for returning processes", () =>
152 Effect.gen(function* () {
153 const readAtMs = DateTime.toEpochMillis(yield* DateTime.now);
154 const sampleCount = 100;
155 const samples = Array.from({ length: sampleCount }, (_, index) => {
156 const sample = nativeSnapshot({
157 sequence: index + 1,
158 sampledAtUnixMs: readAtMs - (sampleCount - index) * 1_000,
159 childCpuTimeMs: index * 10,
160 childWriteBytes: index * 100,
161 });
162 const processes = sample.processes
163 .filter((entry) => entry.pid === SERVER_PID || entry.pid === CODEX_PID)
164 .map((entry) =>
165 entry.pid === CODEX_PID
166 ? { ...entry, startTimeMs: index }
167 : { ...entry, ioWriteBytes: index * 100 },
168 );
169 if (index === 0 || index === 2) {
170 processes.push(
171 processSample({
172 pid: SHORT_LIVED_CHILD_PID,
173 ppid: SERVER_PID,
174 startTimeMs: 500,
175 cpuTimeMs: index === 0 ? 100 : 350,
176 ioWriteBytes: index === 0 ? 1_000 : 5_000,
177 }),
178 );
179 }
180 return { ...sample, processes };
181 });
182 const layerTelemetry = ResourceTelemetry.layer.pipe(
183 Layer.provide(
184 Layer.mergeAll(
185 NativeTelemetryClient.layerTest({ readHistory: () => Effect.succeed(samples) }),
186 DesktopTelemetryReceiver.layerTest(),
187 ResourceAttribution.layer,
188 ),
189 ),
190 );
191
192 yield* Effect.gen(function* () {
193 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
194 const history = yield* telemetry.readHistory({
195 windowMs: sampleCount * 1_000,
196 bucketMs: 10_000,
197 });
198
199 expect(history.topProcesses).toHaveLength(sampleCount + 2);
200 expect(
201 history.topProcesses.find((entry) => entry.identity.pid === SERVER_PID),
202 ).toMatchObject({
203 sampleCount,
204 cpuTimeMs: (sampleCount - 1) * 10,
205 ioWriteBytes: (sampleCount - 1) * 100,
206 });
207 const children = history.topProcesses.filter((entry) => entry.identity.pid === CODEX_PID);
208 expect(children).toHaveLength(sampleCount);
209 for (const child of children) {
210 expect(child).toMatchObject({ sampleCount: 1, cpuTimeMs: 0, ioWriteBytes: 0 });
211 }
212 expect(
213 history.topProcesses.find((entry) => entry.identity.pid === SHORT_LIVED_CHILD_PID),
214 ).toMatchObject({
215 sampleCount: 2,
216 cpuTimeMs: 250,
217 ioWriteBytes: 4_000,
218 });
219 expect(Math.max(...history.buckets.map((bucket) => bucket.maxProcessCount))).toBe(3);
220 expect(history.buckets.reduce((total, bucket) => total + bucket.ioWriteBytes, 0)).toBe(
221 (sampleCount - 1) * 100 + 4_000,
222 );
223 }).pipe(Effect.provide(layerTelemetry));
224 }),
225 );
226
227 it.effect(
228 "attributes an initial Electron root from the identity recorded by the native snapshot",
229 () =>
230 Effect.gen(function* () {
231 const sampledAtUnixMs = DateTime.toEpochMillis(yield* DateTime.now);
232 const sample = nativeSnapshot({
233 sequence: 1,
234 sampledAtUnixMs,
235 childCpuTimeMs: 100,
236 childWriteBytes: 1_000,
237 externalProcesses: [{ pid: ELECTRON_PID, startTimeMs: 300 }],
238 });
239 const desktop = {
240 ...desktopSnapshot(sampledAtUnixMs),
241 electronProcesses: [],
242 };
243 const layerNative = NativeTelemetryClient.layerTest({
244 sampleNow: Effect.succeed(nativeGeneration(sample, 0)),
245 health: Effect.succeed({
246 status: "healthy",
247 hello: Option.none(),
248 lastSampleAt: Option.none(),
249 lastError: Option.none(),
250 restartCount: 0,
251 sampleIntervalMs: 1_000,
252 }),
253 });
254 const layerDesktop = DesktopTelemetryReceiver.layerTest({
255 latest: Effect.succeedSome(desktop),
256 });
257 const layerTelemetry = ResourceTelemetry.layer.pipe(
258 Layer.provide(Layer.mergeAll(layerNative, layerDesktop, ResourceAttribution.layer)),
259 );
260
261 const snapshot = yield* Effect.gen(function* () {
262 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
263 return yield* telemetry.refresh;
264 }).pipe(Effect.provide(layerTelemetry));
265
266 expect(
267 snapshot.processes.find((entry) => entry.identity.pid === ELECTRON_PID)?.category,
268 ).toBe("electron-main");
269 expect(snapshot.groups.electron.processStarts).toBe(1);
270 expect(snapshot.groups.backend.processStarts).toBe(3);
271 }),
272 );
273
274 it.effect("rejects buffered snapshots from an earlier sidecar generation", () =>
275 Effect.gen(function* () {
276 const startedAt = DateTime.toEpochMillis(yield* DateTime.now);
277 const stale = nativeSnapshot({
278 sequence: 100,
279 sampledAtUnixMs: startedAt + 1_000,
280 childCpuTimeMs: 100,
281 childWriteBytes: 1_000,
282 });
283 const current = nativeSnapshot({
284 sequence: 1,
285 sampledAtUnixMs: startedAt + 2_000,
286 childCpuTimeMs: 200,
287 childWriteBytes: 2_000,
288 });
289 const nativeSnapshots =
290 yield* PubSub.unbounded<NativeTelemetryClient.NativeTelemetrySnapshot>();
291 const layerNative = NativeTelemetryClient.layerTest({
292 snapshots: Stream.fromPubSub(nativeSnapshots),
293 sampleNow: Effect.never,
294 health: Effect.succeed({
295 status: "healthy",
296 hello: Option.none(),
297 lastSampleAt: Option.none(),
298 lastError: Option.none(),
299 restartCount: 1,
300 sampleIntervalMs: 1_000,
301 }),
302 });
303 const layerTelemetry = ResourceTelemetry.layer.pipe(
304 Layer.provide(
305 Layer.mergeAll(
306 layerNative,
307 DesktopTelemetryReceiver.layerTest(),
308 ResourceAttribution.layer,
309 ),
310 ),
311 );
312
313 yield* Effect.gen(function* () {
314 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
315 const subscription = yield* telemetry.subscribe;
316 const nextSnapshot = yield* Stream.runHead(subscription.changes).pipe(Effect.forkChild);
317
318 yield* PubSub.publish(nativeSnapshots, nativeGeneration(stale, 0));
319 yield* Effect.yieldNow;
320 expect((yield* telemetry.latest).health.scannedProcessCount).toBe(0);
321
322 yield* PubSub.publish(nativeSnapshots, nativeGeneration(current, 1));
323 const received = yield* Fiber.join(nextSnapshot);
324
325 expect(Option.isSome(received)).toBe(true);
326 expect(DateTime.toEpochMillis(Option.getOrThrow(received).readAt)).toBe(
327 current.sampledAtUnixMs,
328 );
329 }).pipe(Effect.provide(layerTelemetry));
330 }),
331 );
332
333 it.effect("atomically subscribes while a desktop update is being rebuilt", () =>
334 Effect.gen(function* () {
335 const startedAt = DateTime.toEpochMillis(yield* DateTime.now);
336 const nextDesktopSnapshot = desktopSnapshot(startedAt + 1_000);
337 const desktopChanges = yield* PubSub.unbounded<DesktopHostTelemetrySnapshot>();
338 const rebuildStarted = yield* Deferred.make<void>();
339 const finishRebuild = yield* Deferred.make<void>();
340 const attributionReads = yield* Ref.make(0);
341 const layerAttribution = Layer.succeed(
342 ResourceAttribution.ResourceAttribution,
343 ResourceAttribution.ResourceAttribution.of({
344 record: () => Effect.void,
345 snapshot: Ref.updateAndGet(attributionReads, (count) => count + 1).pipe(
346 Effect.flatMap((count) =>
347 count === 1
348 ? Effect.void
349 : Deferred.succeed(rebuildStarted, undefined).pipe(
350 Effect.andThen(Deferred.await(finishRebuild)),
351 ),
352 ),
353 Effect.andThen(
354 Effect.succeed({
355 readAt: DateTime.makeUnsafe(startedAt),
356 entries: [],
357 }),
358 ),
359 ),
360 }),
361 );
362 const layerNative = NativeTelemetryClient.layerTest({
363 sampleNow: Effect.never,
364 health: Effect.succeed({
365 status: "healthy",
366 hello: Option.none(),
367 lastSampleAt: Option.none(),
368 lastError: Option.none(),
369 restartCount: 0,
370 sampleIntervalMs: 1_000,
371 }),
372 });
373 const layerDesktop = DesktopTelemetryReceiver.layerTest({
374 changes: Stream.fromPubSub(desktopChanges),
375 });
376 const layerTelemetry = ResourceTelemetry.layer.pipe(
377 Layer.provide(Layer.mergeAll(layerNative, layerDesktop, layerAttribution)),
378 );
379
380 yield* Effect.gen(function* () {
381 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
382 yield* Effect.yieldNow;
383 yield* PubSub.publish(desktopChanges, nextDesktopSnapshot);
384 yield* Deferred.await(rebuildStarted);
385
386 const subscriptionFiber = yield* telemetry.subscribe.pipe(Effect.forkChild);
387 yield* Effect.yieldNow;
388 yield* Deferred.succeed(finishRebuild, undefined);
389 const subscription = yield* Fiber.join(subscriptionFiber);
390
391 expect(DateTime.toEpochMillis(subscription.latest.readAt)).toBe(
392 nextDesktopSnapshot.sampledAtUnixMs,
393 );
394 expect(subscription.latest.power).toEqual(nextDesktopSnapshot.power);
395 }).pipe(Effect.provide(layerTelemetry));
396 }),
397 );
398});