1import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import type { HostPowerSnapshot } from "@t3tools/contracts";
4import {
5 RESOURCE_MONITOR_PROTOCOL_VERSION,
6 ResourceMonitorCommand,
7 type ResourceMonitorHelloEvent,
8} from "@t3tools/contracts";
9import { describe, expect, it } from "@effect/vitest";
10import * as DateTime from "effect/DateTime";
11import * as Deferred from "effect/Deferred";
12import * as Effect from "effect/Effect";
13import * as Exit from "effect/Exit";
14import * as Fiber from "effect/Fiber";
15import * as Layer from "effect/Layer";
16import * as Queue from "effect/Queue";
17import * as Ref from "effect/Ref";
18import * as Schema from "effect/Schema";
19import * as Scope from "effect/Scope";
20import * as Semaphore from "effect/Semaphore";
21import * as Sink from "effect/Sink";
22import * as Stream from "effect/Stream";
23import { ChildProcessSpawner } from "effect/process";
24
25import * as ServerConfig from "../config.ts";
26import * as NativeTelemetryClient from "./NativeTelemetryClient.ts";
27import * as ResourceMonitorBinary from "./ResourceMonitorBinary.ts";
28import {
29 canCommandNativeTelemetrySidecar,
30 canRequestNativeTelemetryRetry,
31 commitCollectionControlUpdate,
32 retainRecentNativeTelemetryFailures,
33 resolveNativeSampleIntervalMs,
34 synchronizeCollectionControlOnStart,
35} from "./NativeTelemetryClient.ts";
36
37const basePower: HostPowerSnapshot = {
38 source: "electron-main",
39 idle: "false",
40 idleSeconds: 0,
41 locked: "false",
42 suspended: false,
43 onBattery: "false",
44 lowPowerMode: "false",
45 thermalState: "nominal",
46 stale: false,
47 updatedAt: DateTime.makeUnsafe("2026-06-17T12:00:00.000Z"),
48};
49
50describe("resolveNativeSampleIntervalMs", () => {
51 it("keeps a recovery cadence while suspended and backs off under host constraints", () => {
52 expect(resolveNativeSampleIntervalMs({ ...basePower, suspended: true }, 1)).toBe(15_000);
53 expect(resolveNativeSampleIntervalMs({ ...basePower, locked: "true" }, 1)).toBe(15_000);
54 expect(resolveNativeSampleIntervalMs({ ...basePower, lowPowerMode: "true" }, 1)).toBe(15_000);
55 expect(resolveNativeSampleIntervalMs({ ...basePower, thermalState: "critical" }, 1)).toBe(
56 15_000,
57 );
58 expect(resolveNativeSampleIntervalMs({ ...basePower, onBattery: "true" }, 1)).toBe(5_000);
59 });
60
61 it("stops timed sampling without a live subscriber and serves live diagnostics at 1Hz", () => {
62 const unknown: HostPowerSnapshot = {
63 ...basePower,
64 source: "unknown",
65 stale: true,
66 };
67 expect(resolveNativeSampleIntervalMs(unknown, 0)).toBe(0);
68 expect(resolveNativeSampleIntervalMs(unknown, 1)).toBe(1_000);
69 expect(resolveNativeSampleIntervalMs({ ...basePower, locked: "true" }, 0)).toBe(0);
70 expect(resolveNativeSampleIntervalMs({ ...basePower, onBattery: "true" }, 0)).toBe(0);
71 expect(resolveNativeSampleIntervalMs(basePower, 0)).toBe(0);
72 expect(resolveNativeSampleIntervalMs(basePower, 1)).toBe(1_000);
73 });
74});
75
76const decodeCommandLine = Schema.decodeUnknownSync(Schema.fromJsonString(ResourceMonitorCommand));
77
78const hello: ResourceMonitorHelloEvent = {
79 version: RESOURCE_MONITOR_PROTOCOL_VERSION,
80 type: "hello",
81 sidecarVersion: "test",
82 sidecarPid: 4242,
83 platform: "linux",
84 arch: "x64",
85 capabilities: {
86 cumulativeCpuTime: true,
87 currentCpuPercent: true,
88 residentMemory: true,
89 virtualMemory: true,
90 ioBytes: true,
91 processStartTime: true,
92 processTree: true,
93 },
94};
95
96describe("NativeTelemetryClient", () => {
97 it.effect("asks the sidecar for timed samples only while a live subscriber exists", () =>
98 Effect.gen(function* () {
99 const commands = yield* Queue.unbounded<ResourceMonitorCommand>();
100 // A sidecar that says hello, stays alive, and records every command it receives.
101 const spawner = ChildProcessSpawner.make(() =>
102 Effect.succeed(
103 ChildProcessSpawner.makeHandle({
104 pid: ChildProcessSpawner.ProcessId(hello.sidecarPid),
105 exitCode: Effect.never,
106 isRunning: Effect.succeed(true),
107 kill: () => Effect.void,
108 unref: Effect.succeed(Effect.void),
109 stdin: Sink.forEach((chunk: Uint8Array) =>
110 Effect.forEach(new TextDecoder().decode(chunk).split("\n").filter(Boolean), (line) =>
111 Queue.offer(commands, decodeCommandLine(line)),
112 ),
113 ),
114 stdout: Stream.encodeText(Stream.make(`${JSON.stringify(hello)}\n`)).pipe(
115 Stream.concat(Stream.never),
116 ),
117 stderr: Stream.empty,
118 all: Stream.empty,
119 getInputFd: () => Sink.drain,
120 getOutputFd: () => Stream.empty,
121 }),
122 ),
123 );
124 const client = yield* NativeTelemetryClient.make().pipe(
125 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
126 Effect.provideService(ResourceMonitorBinary.ResourceMonitorBinary, {
127 resolve: Effect.succeed("/test/t3-resource-monitor"),
128 }),
129 );
130
131 const configure = yield* Queue.take(commands);
132 expect(configure).toMatchObject({ type: "configure", sampleIntervalMs: 0 });
133 expect(yield* Queue.take(commands)).toMatchObject({ type: "setExternalProcesses" });
134
135 const live = yield* Scope.make();
136 yield* client.snapshots.pipe(Stream.runDrain, Effect.forkIn(live));
137 expect(yield* Queue.take(commands)).toMatchObject({
138 type: "setSampleInterval",
139 sampleIntervalMs: 1_000,
140 });
141 expect(yield* Queue.take(commands)).toMatchObject({ type: "setStreaming", enabled: true });
142
143 yield* Scope.close(live, Exit.void);
144 expect(yield* Queue.take(commands)).toMatchObject({
145 type: "setSampleInterval",
146 sampleIntervalMs: 0,
147 });
148 expect(yield* Queue.take(commands)).toMatchObject({ type: "setStreaming", enabled: false });
149 expect((yield* client.health).sampleIntervalMs).toBe(0);
150 }).pipe(
151 Effect.provide(
152 Layer.mergeAll(
153 ServerConfig.layerTest(process.cwd(), { prefix: "t3-native-telemetry-test-" }),
154 NodeCrypto.layer,
155 ).pipe(Layer.provideMerge(NodeServices.layer)),
156 ),
157 ),
158 );
159});
160
161describe("canRequestNativeTelemetryRetry", () => {
162 it("only accepts retry while the supervisor is waiting without a live sidecar", () => {
163 expect(canRequestNativeTelemetryRetry("degraded", false)).toBe(true);
164 expect(canRequestNativeTelemetryRetry("unavailable", false)).toBe(true);
165 expect(canRequestNativeTelemetryRetry("degraded", true)).toBe(false);
166 expect(canRequestNativeTelemetryRetry("healthy", false)).toBe(false);
167 expect(canRequestNativeTelemetryRetry("starting", false)).toBe(false);
168 });
169});
170
171describe("canCommandNativeTelemetrySidecar", () => {
172 it("keeps on-demand recovery commands available while a live sidecar is degraded", () => {
173 expect(canCommandNativeTelemetrySidecar("healthy", true)).toBe(true);
174 expect(canCommandNativeTelemetrySidecar("degraded", true)).toBe(true);
175 expect(canCommandNativeTelemetrySidecar("unavailable", true)).toBe(false);
176 expect(canCommandNativeTelemetrySidecar("degraded", false)).toBe(false);
177 });
178});
179
180describe("retainRecentNativeTelemetryFailures", () => {
181 it("expires old failures so an isolated crash restarts from the initial backoff", () => {
182 expect(retainRecentNativeTelemetryFailures([0, 30_000], 90_001)).toEqual([]);
183 expect(retainRecentNativeTelemetryFailures([30_000, 60_000], 90_000)).toEqual([30_000, 60_000]);
184 });
185});
186
187describe("commitCollectionControlUpdate", () => {
188 it.effect("retains desired demand and retries unapplied sidecar state", () =>
189 Effect.gen(function* () {
190 const initial = {
191 hostPower: basePower,
192 liveSubscriberCount: 0,
193 sampleIntervalMs: 5_000,
194 };
195 const desired = yield* Ref.make(initial);
196 const applied = yield* Ref.make(initial);
197 const failure = new Error("sidecar write failed");
198 const receivedStates: Array<readonly [number, number]> = [];
199
200 const received = yield* commitCollectionControlUpdate(
201 desired,
202 applied,
203 (current) => ({
204 ...current,
205 liveSubscriberCount: 1,
206 sampleIntervalMs: 1_000,
207 }),
208 (previous, next) => {
209 receivedStates.push([previous.sampleIntervalMs, next.sampleIntervalMs]);
210 return Effect.fail(failure);
211 },
212 ).pipe(Effect.flip);
213
214 expect(received).toBe(failure);
215 expect(yield* Ref.get(desired)).toEqual({
216 ...initial,
217 liveSubscriberCount: 1,
218 sampleIntervalMs: 1_000,
219 });
220 expect(yield* Ref.get(applied)).toEqual(initial);
221
222 yield* commitCollectionControlUpdate(
223 desired,
224 applied,
225 (current) => current,
226 (previous, next) => {
227 receivedStates.push([previous.sampleIntervalMs, next.sampleIntervalMs]);
228 return Effect.void;
229 },
230 );
231 expect(receivedStates).toEqual([
232 [5_000, 1_000],
233 [5_000, 1_000],
234 ]);
235 expect(yield* Ref.get(applied)).toEqual(yield* Ref.get(desired));
236 }),
237 );
238
239 it.effect("serializes startup synchronization with runtime control updates", () =>
240 Effect.gen(function* () {
241 const initial = {
242 hostPower: basePower,
243 liveSubscriberCount: 0,
244 sampleIntervalMs: 5_000,
245 };
246 const desired = yield* Ref.make(initial);
247 const applied = yield* Ref.make(initial);
248 const ready = yield* Ref.make(false);
249 const mutex = yield* Semaphore.make(1);
250 const startupApplying = yield* Deferred.make<void>();
251 const releaseStartup = yield* Deferred.make<void>();
252 const appliedIntervals: Array<number> = [];
253
254 const startupFiber = yield* synchronizeCollectionControlOnStart(
255 mutex,
256 desired,
257 applied,
258 (control) =>
259 Effect.sync(() => {
260 appliedIntervals.push(control.sampleIntervalMs);
261 }).pipe(
262 Effect.andThen(Deferred.succeed(startupApplying, undefined)),
263 Effect.andThen(Deferred.await(releaseStartup)),
264 Effect.asVoid,
265 ),
266 Ref.set(ready, true),
267 ).pipe(Effect.forkChild);
268 yield* Deferred.await(startupApplying);
269
270 const updateFiber = yield* mutex
271 .withPermits(1)(
272 commitCollectionControlUpdate(
273 desired,
274 applied,
275 (current) => ({
276 ...current,
277 liveSubscriberCount: 1,
278 sampleIntervalMs: 1_000,
279 }),
280 (_previous, next) =>
281 Effect.gen(function* () {
282 expect(yield* Ref.get(ready)).toBe(true);
283 appliedIntervals.push(next.sampleIntervalMs);
284 }),
285 ),
286 )
287 .pipe(Effect.forkChild);
288 yield* Effect.yieldNow;
289 expect(yield* Ref.get(desired)).toEqual(initial);
290
291 yield* Deferred.succeed(releaseStartup, undefined);
292 yield* Fiber.join(startupFiber);
293 yield* Fiber.join(updateFiber);
294
295 expect(appliedIntervals).toEqual([5_000, 1_000]);
296 expect(yield* Ref.get(applied)).toEqual(yield* Ref.get(desired));
297 }),
298 );
299});