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 Duration from "effect/Duration";
10import * as Effect from "effect/Effect";
11import * as Fiber from "effect/Fiber";
12import * as Layer from "effect/Layer";
13import * as Option from "effect/Option";
14import * as PubSub from "effect/PubSub";
15import * as Ref from "effect/Ref";
16import * as Stream from "effect/Stream";
17import * as TestClock from "effect/testing/TestClock";
18
19import * as DesktopTelemetryReceiver from "./DesktopTelemetryReceiver.ts";
20import * as NativeTelemetryClient from "./NativeTelemetryClient.ts";
21import * as ResourceAttribution from "./ResourceAttribution.ts";
22import * as ResourceTelemetry from "./ResourceTelemetry.ts";
23
24function processSample(
25 input: Partial<ResourceMonitorProcessSample> &
26 Pick<ResourceMonitorProcessSample, "pid" | "ppid" | "startTimeMs">,
27): ResourceMonitorProcessSample {
28 return {
29 runTimeMs: 1_000,
30 name: `process-${input.pid}`,
31 command: `process-${input.pid}`,
32 status: "Running",
33 cpuPercent: 0,
34 cpuTimeMs: 0,
35 residentBytes: 1_024,
36 virtualBytes: 2_048,
37 ioReadBytes: 0,
38 ioWriteBytes: 0,
39 ioSemantics: "storage",
40 ...input,
41 };
42}
43
44function nativeSnapshot(input: {
45 readonly sequence: number;
46 readonly sampledAtUnixMs: number;
47 readonly childCpuTimeMs: number;
48 readonly childWriteBytes: number;
49 readonly externalProcesses?: ResourceMonitorSnapshotEvent["externalProcesses"];
50}): ResourceMonitorSnapshotEvent {
51 const processes = [
52 processSample({
53 pid: process.pid,
54 ppid: 1,
55 startTimeMs: 100,
56 cpuTimeMs: input.sequence * 10,
57 }),
58 processSample({
59 pid: 4_242,
60 ppid: process.pid,
61 startTimeMs: 200,
62 name: "codex",
63 command: "codex app-server",
64 cpuTimeMs: input.childCpuTimeMs,
65 ioWriteBytes: input.childWriteBytes,
66 }),
67 processSample({
68 pid: 5_000,
69 ppid: 1,
70 startTimeMs: 300,
71 name: "electron",
72 command: "electron",
73 cpuTimeMs: input.sequence * 20,
74 }),
75 processSample({
76 pid: 9_000,
77 ppid: process.pid,
78 startTimeMs: 400,
79 name: "t3-resource-monitor",
80 command: "t3-resource-monitor",
81 cpuTimeMs: input.sequence * 5,
82 }),
83 ];
84 return {
85 version: 3,
86 type: "snapshot",
87 sequence: input.sequence,
88 sampledAtUnixMs: input.sampledAtUnixMs,
89 collectionDurationMicros: 300,
90 scannedProcessCount: 80,
91 retainedProcessCount: processes.length,
92 inaccessibleProcessCount: 1,
93 ...(input.externalProcesses === undefined
94 ? {}
95 : { externalProcesses: input.externalProcesses }),
96 processes,
97 };
98}
99
100function nativeGeneration(
101 snapshot: ResourceMonitorSnapshotEvent,
102 generation: number,
103): NativeTelemetryClient.NativeTelemetrySnapshot {
104 return { generation, snapshot };
105}
106
107function desktopSnapshot(sampledAtUnixMs: number): DesktopHostTelemetrySnapshot {
108 const sampledAt = DateTime.makeUnsafe(sampledAtUnixMs);
109 return {
110 version: 1,
111 type: "desktopTelemetry",
112 sequence: 1,
113 sampledAtUnixMs,
114 electronPid: 5_000,
115 power: {
116 source: "electron-main",
117 idle: "false",
118 idleSeconds: 2,
119 locked: "false",
120 suspended: false,
121 onBattery: "true",
122 lowPowerMode: "unknown",
123 thermalState: "fair",
124 stale: false,
125 updatedAt: sampledAt,
126 },
127 speedLimitPercent: Option.some(90),
128 electronProcesses: [
129 {
130 pid: 5_000,
131 creationTimeMs: 300,
132 type: "Browser",
133 name: "electron",
134 cpuPercent: 2,
135 cumulativeCpuSeconds: 0.02,
136 idleWakeupsPerSecond: 3,
137 workingSetBytes: 4_096,
138 peakWorkingSetBytes: 8_192,
139 },
140 ],
141 };
142}
143
144describe("ResourceTelemetry", () => {
145 it.effect("replays churning history and retains baselines for returning processes", () =>
146 Effect.gen(function* () {
147 const readAtMs = DateTime.toEpochMillis(yield* DateTime.now);
148 const sampleCount = 100;
149 const samples = Array.from({ length: sampleCount }, (_, index) => {
150 const sample = nativeSnapshot({
151 sequence: index + 1,
152 sampledAtUnixMs: readAtMs - (sampleCount - index) * 1_000,
153 childCpuTimeMs: index * 10,
154 childWriteBytes: index * 100,
155 });
156 const processes = sample.processes
157 .filter((entry) => entry.pid === process.pid || entry.pid === 4_242)
158 .map((entry) =>
159 entry.pid === 4_242
160 ? { ...entry, startTimeMs: index }
161 : { ...entry, ioWriteBytes: index * 100 },
162 );
163 if (index === 0 || index === 2) {
164 processes.push(
165 processSample({
166 pid: 4_243,
167 ppid: process.pid,
168 startTimeMs: 500,
169 cpuTimeMs: index === 0 ? 100 : 350,
170 ioWriteBytes: index === 0 ? 1_000 : 5_000,
171 }),
172 );
173 }
174 return { ...sample, processes };
175 });
176 const layerTelemetry = ResourceTelemetry.layer.pipe(
177 Layer.provide(
178 Layer.mergeAll(
179 NativeTelemetryClient.layerTest({ readHistory: () => Effect.succeed(samples) }),
180 DesktopTelemetryReceiver.layerTest(),
181 ResourceAttribution.layer,
182 ),
183 ),
184 );
185
186 yield* Effect.gen(function* () {
187 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
188 const history = yield* telemetry.readHistory({
189 windowMs: sampleCount * 1_000,
190 bucketMs: 10_000,
191 });
192
193 expect(history.topProcesses).toHaveLength(sampleCount + 2);
194 expect(
195 history.topProcesses.find((entry) => entry.identity.pid === process.pid),
196 ).toMatchObject({
197 sampleCount,
198 cpuTimeMs: (sampleCount - 1) * 10,
199 ioWriteBytes: (sampleCount - 1) * 100,
200 });
201 const children = history.topProcesses.filter((entry) => entry.identity.pid === 4_242);
202 expect(children).toHaveLength(sampleCount);
203 for (const child of children) {
204 expect(child).toMatchObject({ sampleCount: 1, cpuTimeMs: 0, ioWriteBytes: 0 });
205 }
206 expect(history.topProcesses.find((entry) => entry.identity.pid === 4_243)).toMatchObject({
207 sampleCount: 2,
208 cpuTimeMs: 250,
209 ioWriteBytes: 4_000,
210 });
211 expect(Math.max(...history.buckets.map((bucket) => bucket.maxProcessCount))).toBe(3);
212 expect(history.buckets.reduce((total, bucket) => total + bucket.ioWriteBytes, 0)).toBe(
213 (sampleCount - 1) * 100 + 4_000,
214 );
215 }).pipe(Effect.provide(layerTelemetry));
216 }),
217 );
218
219 it.effect("enables live native and Electron collection only while changes are retained", () =>
220 Effect.gen(function* () {
221 const sampledAtUnixMs = DateTime.toEpochMillis(yield* DateTime.now);
222 const sample = nativeSnapshot({
223 sequence: 1,
224 sampledAtUnixMs,
225 childCpuTimeMs: 100,
226 childWriteBytes: 1_000,
227 });
228 const demandChanges = yield* Ref.make<ReadonlyArray<boolean>>([]);
229 const layerNative = NativeTelemetryClient.layerTest({
230 sampleNow: Effect.succeed(nativeGeneration(sample, 0)),
231 health: Effect.succeed({
232 status: "healthy",
233 hello: Option.none(),
234 lastSampleAt: Option.none(),
235 lastError: Option.none(),
236 restartCount: 0,
237 sampleIntervalMs: 1_000,
238 }),
239 });
240 const layerDesktop = DesktopTelemetryReceiver.layerTest({
241 latest: Effect.succeedSome(desktopSnapshot(sampledAtUnixMs)),
242 setDiagnosticsDemand: (enabled) =>
243 Ref.update(demandChanges, (changes) => [...changes, enabled]),
244 });
245 const layerTelemetry = ResourceTelemetry.layer.pipe(
246 Layer.provide(Layer.mergeAll(layerNative, layerDesktop, ResourceAttribution.layer)),
247 );
248
249 const live = yield* Stream.runHead(
250 Effect.gen(function* () {
251 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
252 return telemetry.changes;
253 }).pipe(Stream.unwrap),
254 ).pipe(Effect.provide(layerTelemetry));
255
256 expect(Option.isSome(live)).toBe(true);
257 expect(yield* Ref.get(demandChanges)).toEqual([true, false]);
258 }),
259 );
260
261 it.effect("releases live demand when acquisition is interrupted by an idle collector", () =>
262 Effect.gen(function* () {
263 const sampledAtUnixMs = DateTime.toEpochMillis(yield* DateTime.now);
264 const demandChanges = yield* Ref.make<ReadonlyArray<boolean>>([]);
265 const layerNative = NativeTelemetryClient.layerTest({
266 sampleNow: Effect.never,
267 health: Effect.succeed({
268 status: "healthy",
269 hello: Option.none(),
270 lastSampleAt: Option.none(),
271 lastError: Option.none(),
272 restartCount: 0,
273 sampleIntervalMs: 1_000,
274 }),
275 });
276 const layerDesktop = DesktopTelemetryReceiver.layerTest({
277 latest: Effect.succeedSome(desktopSnapshot(sampledAtUnixMs)),
278 setDiagnosticsDemand: (enabled) =>
279 Ref.update(demandChanges, (changes) => [...changes, enabled]),
280 });
281 const layerTelemetry = ResourceTelemetry.layer.pipe(
282 Layer.provide(Layer.mergeAll(layerNative, layerDesktop, ResourceAttribution.layer)),
283 );
284
285 const resultFiber = yield* Stream.runHead(
286 Effect.gen(function* () {
287 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
288 return telemetry.changes;
289 }).pipe(Stream.unwrap),
290 ).pipe(Effect.timeoutOption("10 millis"), Effect.provide(layerTelemetry), Effect.forkChild);
291 yield* Effect.yieldNow;
292 yield* TestClock.adjust("10 millis");
293 const result = yield* Fiber.join(resultFiber);
294
295 expect(Option.isNone(result)).toBe(true);
296 expect(yield* Ref.get(demandChanges)).toEqual([true, false]);
297 }),
298 );
299
300 it.effect(
301 "attributes an initial Electron root from the identity recorded by the native snapshot",
302 () =>
303 Effect.gen(function* () {
304 const sampledAtUnixMs = DateTime.toEpochMillis(yield* DateTime.now);
305 const sample = nativeSnapshot({
306 sequence: 1,
307 sampledAtUnixMs,
308 childCpuTimeMs: 100,
309 childWriteBytes: 1_000,
310 externalProcesses: [{ pid: 5_000, startTimeMs: 300 }],
311 });
312 const desktop = {
313 ...desktopSnapshot(sampledAtUnixMs),
314 electronProcesses: [],
315 };
316 const layerNative = NativeTelemetryClient.layerTest({
317 sampleNow: Effect.succeed(nativeGeneration(sample, 0)),
318 health: Effect.succeed({
319 status: "healthy",
320 hello: Option.none(),
321 lastSampleAt: Option.none(),
322 lastError: Option.none(),
323 restartCount: 0,
324 sampleIntervalMs: 1_000,
325 }),
326 });
327 const layerDesktop = DesktopTelemetryReceiver.layerTest({
328 latest: Effect.succeedSome(desktop),
329 });
330 const layerTelemetry = ResourceTelemetry.layer.pipe(
331 Layer.provide(Layer.mergeAll(layerNative, layerDesktop, ResourceAttribution.layer)),
332 );
333
334 const snapshot = yield* Effect.gen(function* () {
335 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
336 return yield* telemetry.refresh;
337 }).pipe(Effect.provide(layerTelemetry));
338
339 expect(snapshot.processes.find((entry) => entry.identity.pid === 5_000)?.category).toBe(
340 "electron-main",
341 );
342 expect(snapshot.groups.electron.processStarts).toBe(1);
343 expect(snapshot.groups.backend.processStarts).toBe(3);
344 }),
345 );
346
347 it.effect("rejects buffered snapshots from an earlier sidecar generation", () =>
348 Effect.scoped(
349 Effect.gen(function* () {
350 const startedAt = DateTime.toEpochMillis(yield* DateTime.now);
351 const stale = nativeSnapshot({
352 sequence: 100,
353 sampledAtUnixMs: startedAt + 1_000,
354 childCpuTimeMs: 100,
355 childWriteBytes: 1_000,
356 });
357 const current = nativeSnapshot({
358 sequence: 1,
359 sampledAtUnixMs: startedAt + 2_000,
360 childCpuTimeMs: 200,
361 childWriteBytes: 2_000,
362 });
363 const nativeSnapshots =
364 yield* PubSub.unbounded<NativeTelemetryClient.NativeTelemetrySnapshot>();
365 const layerNative = NativeTelemetryClient.layerTest({
366 snapshots: Stream.fromPubSub(nativeSnapshots),
367 sampleNow: Effect.never,
368 health: Effect.succeed({
369 status: "healthy",
370 hello: Option.none(),
371 lastSampleAt: Option.none(),
372 lastError: Option.none(),
373 restartCount: 1,
374 sampleIntervalMs: 1_000,
375 }),
376 });
377 const layerTelemetry = ResourceTelemetry.layer.pipe(
378 Layer.provide(
379 Layer.mergeAll(
380 layerNative,
381 DesktopTelemetryReceiver.layerTest(),
382 ResourceAttribution.layer,
383 ),
384 ),
385 );
386
387 yield* Effect.gen(function* () {
388 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
389 const subscription = yield* telemetry.subscribe;
390 const nextSnapshot = yield* Stream.runHead(subscription.changes).pipe(Effect.forkChild);
391
392 yield* PubSub.publish(nativeSnapshots, nativeGeneration(stale, 0));
393 yield* Effect.yieldNow;
394 expect((yield* telemetry.latest).health.scannedProcessCount).toBe(0);
395
396 yield* PubSub.publish(nativeSnapshots, nativeGeneration(current, 1));
397 const received = yield* Fiber.join(nextSnapshot);
398
399 expect(Option.isSome(received)).toBe(true);
400 expect(DateTime.toEpochMillis(Option.getOrThrow(received).readAt)).toBe(
401 current.sampledAtUnixMs,
402 );
403 }).pipe(Effect.provide(layerTelemetry));
404 }),
405 ),
406 );
407
408 it.effect("retains desktop health changes while aggregate telemetry initializes", () =>
409 Effect.scoped(
410 Effect.gen(function* () {
411 const health = yield* Ref.make<DesktopTelemetryReceiver.DesktopTelemetryReceiverHealth>({
412 status: "starting",
413 lastSampleAt: Option.none(),
414 lastError: Option.none(),
415 });
416 const healthChanges =
417 yield* PubSub.sliding<DesktopTelemetryReceiver.DesktopTelemetryReceiverHealth>(4);
418 const healthSubscribed = yield* Deferred.make<void>();
419 const finishDesktopSnapshot = yield* Deferred.make<void>();
420 const layerDesktop = DesktopTelemetryReceiver.layerTest({
421 health: Ref.get(health),
422 subscribeHealth: Effect.gen(function* () {
423 const subscription = yield* PubSub.subscribe(healthChanges);
424 const latest = yield* Ref.get(health);
425 yield* Deferred.succeed(healthSubscribed, undefined);
426 return {
427 latest,
428 changes: Stream.fromSubscription(subscription),
429 };
430 }),
431 subscribe: Deferred.await(finishDesktopSnapshot).pipe(
432 Effect.as({
433 latest: Option.none(),
434 changes: Stream.empty,
435 }),
436 ),
437 });
438 const layerTelemetry = ResourceTelemetry.layer.pipe(
439 Layer.provide(
440 Layer.mergeAll(
441 NativeTelemetryClient.layerTest(),
442 layerDesktop,
443 ResourceAttribution.layer,
444 ),
445 ),
446 );
447 const resultFiber = yield* Effect.gen(function* () {
448 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
449 while (true) {
450 const snapshot = yield* telemetry.latest;
451 if (snapshot.health.desktop.status === "degraded") return snapshot;
452 yield* Effect.yieldNow;
453 }
454 }).pipe(Effect.provide(layerTelemetry), Effect.timeout("1 second"), Effect.forkChild);
455
456 yield* Deferred.await(healthSubscribed);
457 const degraded: DesktopTelemetryReceiver.DesktopTelemetryReceiverHealth = {
458 status: "degraded",
459 lastSampleAt: Option.none(),
460 lastError: Option.some("desktop telemetry failed"),
461 };
462 yield* Ref.set(health, degraded);
463 yield* PubSub.publish(healthChanges, degraded);
464 yield* Deferred.succeed(finishDesktopSnapshot, undefined);
465
466 const snapshot = yield* Fiber.join(resultFiber);
467 expect(snapshot.health.desktop.status).toBe("degraded");
468 expect(Option.getOrNull(snapshot.health.desktop.lastError)).toBe(
469 "desktop telemetry failed",
470 );
471 }),
472 ),
473 );
474
475 it.effect("atomically subscribes while a desktop update is being rebuilt", () =>
476 Effect.scoped(
477 Effect.gen(function* () {
478 const startedAt = DateTime.toEpochMillis(yield* DateTime.now);
479 const nextDesktopSnapshot = desktopSnapshot(startedAt + 1_000);
480 const desktopChanges = yield* PubSub.unbounded<DesktopHostTelemetrySnapshot>();
481 const rebuildStarted = yield* Deferred.make<void>();
482 const finishRebuild = yield* Deferred.make<void>();
483 const attributionReads = yield* Ref.make(0);
484 const layerAttribution = Layer.succeed(
485 ResourceAttribution.ResourceAttribution,
486 ResourceAttribution.ResourceAttribution.of({
487 record: () => Effect.void,
488 snapshot: Ref.updateAndGet(attributionReads, (count) => count + 1).pipe(
489 Effect.flatMap((count) =>
490 count === 1
491 ? Effect.void
492 : Deferred.succeed(rebuildStarted, undefined).pipe(
493 Effect.andThen(Deferred.await(finishRebuild)),
494 ),
495 ),
496 Effect.andThen(
497 Effect.succeed({
498 readAt: DateTime.makeUnsafe(startedAt),
499 entries: [],
500 }),
501 ),
502 ),
503 }),
504 );
505 const layerNative = NativeTelemetryClient.layerTest({
506 sampleNow: Effect.never,
507 health: Effect.succeed({
508 status: "healthy",
509 hello: Option.none(),
510 lastSampleAt: Option.none(),
511 lastError: Option.none(),
512 restartCount: 0,
513 sampleIntervalMs: 1_000,
514 }),
515 });
516 const layerDesktop = DesktopTelemetryReceiver.layerTest({
517 changes: Stream.fromPubSub(desktopChanges),
518 });
519 const layerTelemetry = ResourceTelemetry.layer.pipe(
520 Layer.provide(Layer.mergeAll(layerNative, layerDesktop, layerAttribution)),
521 );
522
523 yield* Effect.gen(function* () {
524 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
525 yield* Effect.yieldNow;
526 yield* PubSub.publish(desktopChanges, nextDesktopSnapshot);
527 yield* Deferred.await(rebuildStarted);
528
529 const subscriptionFiber = yield* telemetry.subscribe.pipe(Effect.forkChild);
530 yield* Effect.yieldNow;
531 yield* Deferred.succeed(finishRebuild, undefined);
532 const subscription = yield* Fiber.join(subscriptionFiber);
533
534 expect(DateTime.toEpochMillis(subscription.latest.readAt)).toBe(
535 nextDesktopSnapshot.sampledAtUnixMs,
536 );
537 expect(subscription.latest.power).toEqual(nextDesktopSnapshot.power);
538 }).pipe(Effect.provide(layerTelemetry));
539 }),
540 ),
541 );
542
543 it.effect("combines native, Electron, attribution, retry, and history data", () =>
544 Effect.gen(function* () {
545 const startedAt = DateTime.toEpochMillis(yield* DateTime.now);
546 const samples = [
547 nativeSnapshot({
548 sequence: 1,
549 sampledAtUnixMs: startedAt,
550 childCpuTimeMs: 100,
551 childWriteBytes: 1_000,
552 }),
553 nativeSnapshot({
554 sequence: 2,
555 sampledAtUnixMs: startedAt + 1_000,
556 childCpuTimeMs: 350,
557 childWriteBytes: 5_000,
558 }),
559 nativeSnapshot({
560 sequence: 1,
561 sampledAtUnixMs: startedAt + 2_000,
562 childCpuTimeMs: 500,
563 childWriteBytes: 7_000,
564 }),
565 ] as const;
566 const sampleIndex = yield* Ref.make(0);
567 const externalProcesses = yield* Ref.make<
568 ReadonlyArray<{ readonly pid: number; readonly startTimeMs?: number }>
569 >([]);
570 const retryCount = yield* Ref.make(0);
571 const nativeHealth = yield* Ref.make<NativeTelemetryClient.NativeTelemetryClientHealth>({
572 status: "healthy",
573 hello: Option.some({
574 version: 3,
575 type: "hello",
576 sidecarVersion: "0.1.0",
577 sidecarPid: 9_000,
578 platform: "test",
579 arch: "test",
580 capabilities: {
581 cumulativeCpuTime: true,
582 currentCpuPercent: true,
583 residentMemory: true,
584 virtualMemory: true,
585 ioBytes: true,
586 processStartTime: true,
587 processTree: true,
588 },
589 }),
590 lastSampleAt: Option.some(DateTime.makeUnsafe(startedAt)),
591 lastError: Option.none(),
592 restartCount: 2,
593 sampleIntervalMs: 1_000,
594 });
595 const nativeHealthChanges =
596 yield* PubSub.sliding<NativeTelemetryClient.NativeTelemetryClientHealth>(4);
597 const layerNative = NativeTelemetryClient.layerTest({
598 setExternalProcesses: (processes) => Ref.set(externalProcesses, processes),
599 readHistory: () => Effect.succeed(samples.slice(0, 2)),
600 sampleNow: Ref.modify(sampleIndex, (index) => {
601 const sampleIndex = Math.min(index, samples.length - 1);
602 return [nativeGeneration(samples[sampleIndex]!, sampleIndex === 2 ? 3 : 2), index + 1];
603 }),
604 retry: Ref.updateAndGet(retryCount, (count) => count + 1).pipe(Effect.as(true)),
605 health: Ref.get(nativeHealth),
606 subscribeHealth: Effect.gen(function* () {
607 const subscription = yield* PubSub.subscribe(nativeHealthChanges);
608 const latest = yield* Ref.get(nativeHealth);
609 return {
610 latest,
611 changes: Stream.fromSubscription(subscription),
612 };
613 }),
614 });
615 const layerDesktop = DesktopTelemetryReceiver.layerTest({
616 latest: Effect.succeedSome(desktopSnapshot(startedAt)),
617 health: Effect.succeed({
618 status: "healthy",
619 lastSampleAt: Option.some(DateTime.makeUnsafe(startedAt)),
620 lastError: Option.none(),
621 }),
622 });
623 const layerAttribution = ResourceAttribution.layer;
624 const layerDependencies = Layer.mergeAll(layerNative, layerDesktop, layerAttribution);
625 const layerTelemetry = ResourceTelemetry.layer.pipe(Layer.provide(layerDependencies));
626 const layer = Layer.mergeAll(layerDependencies, layerTelemetry);
627
628 yield* Effect.gen(function* () {
629 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
630 const attribution = yield* ResourceAttribution.ResourceAttribution;
631
632 expect(yield* Ref.get(externalProcesses)).toEqual([{ pid: 5_000, startTimeMs: 300 }]);
633
634 yield* attribution.record({
635 component: "provider-event-log",
636 operation: "append",
637 logicalWriteBytes: 512,
638 count: 2,
639 durationMs: 4,
640 });
641 const first = yield* telemetry.refresh;
642 expect(first.groups.backend.processCount).toBe(2);
643 expect(first.groups.electron.processCount).toBe(1);
644 expect(first.groups.monitor.processCount).toBe(1);
645 expect(first.power.onBattery).toBe("true");
646 expect(Option.getOrNull(first.speedLimitPercent)).toBe(90);
647 expect(first.attribution.entries).toEqual([
648 {
649 component: "provider-event-log",
650 operation: "append",
651 logicalReadBytes: 0,
652 logicalWriteBytes: 512,
653 count: 2,
654 durationMs: 4,
655 },
656 ]);
657
658 yield* TestClock.adjust(Duration.seconds(1));
659 const second = yield* telemetry.refresh;
660 const codex = second.processes.find((entry) => entry.identity.pid === 4_242);
661 expect(codex?.cpuPercent).toBe(25);
662 expect(codex?.ioWriteBytesPerSecond).toBe(4_000);
663 expect(second.groups.backend.ioWriteBytes).toBe(4_000);
664 expect(second.health.collectionDurationMicros).toBe(300);
665 expect(second.health.scannedProcessCount).toBe(80);
666 expect(second.health.inaccessibleProcessCount).toBe(1);
667
668 const history = yield* telemetry.readHistory({
669 windowMs: 60_000,
670 bucketMs: 10_000,
671 });
672 expect(history.retainedSampleCount).toBeGreaterThan(0);
673 expect(
674 history.topProcesses.find((entry) => entry.identity.pid === 4_242)?.sampleCount,
675 ).toBe(2);
676 expect(history.topProcesses.find((entry) => entry.identity.pid === 4_242)?.cpuTimeMs).toBe(
677 250,
678 );
679 expect(
680 history.topProcesses.find((entry) => entry.identity.pid === 4_242)?.ioWriteBytes,
681 ).toBe(4_000);
682 expect(history.buckets.reduce((total, bucket) => total + bucket.ioWriteBytes, 0)).toBe(
683 4_000,
684 );
685
686 const retry = yield* telemetry.retry;
687 expect(retry.accepted).toBe(true);
688 expect(yield* Ref.get(retryCount)).toBe(1);
689
690 yield* Ref.update(nativeHealth, (current) => ({
691 ...current,
692 hello: Option.map(current.hello, (hello) => ({
693 ...hello,
694 sidecarPid: 9_001,
695 })),
696 restartCount: 3,
697 }));
698 yield* TestClock.adjust(Duration.seconds(1));
699 const restarted = yield* telemetry.refresh;
700 expect(DateTime.toEpochMillis(restarted.readAt)).toBe(startedAt + 2_000);
701 expect(Option.getOrNull(restarted.health.sidecarPid)).toBe(9_001);
702
703 yield* Ref.update(nativeHealth, (current) => ({
704 ...current,
705 status: "degraded" as const,
706 lastError: Option.some("collector exited"),
707 }));
708 yield* PubSub.publish(nativeHealthChanges, yield* Ref.get(nativeHealth));
709 yield* Effect.yieldNow;
710 const healthUpdate = yield* telemetry.latest;
711 expect(healthUpdate.health.native.status).toBe("degraded");
712 expect(Option.getOrNull(healthUpdate.health.native.lastError)).toBe("collector exited");
713 const degradedHistory = yield* telemetry.readHistory({
714 windowMs: 60_000,
715 bucketMs: 10_000,
716 });
717 expect(degradedHistory.health.native.status).toBe("degraded");
718 }).pipe(Effect.provide(layer));
719 }),
720 );
721});