1import { describe, it, assert } from "@effect/vitest";
2import {
3 DEFAULT_SERVER_SETTINGS,
4 ProviderDriverKind,
5 ProviderInstanceId,
6 ServerSettingsError,
7 type ServerProvider,
8} from "@t3tools/contracts";
9import { createModelCapabilities } from "@t3tools/shared/model";
10import * as Cache from "effect/Cache";
11import * as Deferred from "effect/Deferred";
12import * as Effect from "effect/Effect";
13import * as Fiber from "effect/Fiber";
14import * as Layer from "effect/Layer";
15import * as PubSub from "effect/PubSub";
16import * as Ref from "effect/Ref";
17import * as Stream from "effect/Stream";
18import * as TestClock from "effect/testing/TestClock";
19
20import * as ProviderHost from "./ProviderHost.ts";
21import { makeManagedServerProvider } from "./managedProvider.ts";
22
23const emptyCapabilities = createModelCapabilities({ optionDescriptors: [] });
24const fastModeCapabilities = createModelCapabilities({
25 optionDescriptors: [
26 {
27 id: "fastMode",
28 label: "Fast Mode",
29 type: "boolean",
30 },
31 ],
32});
33
34interface TestSettings {
35 readonly enabled: boolean;
36}
37
38const maintenanceCapabilities = {
39 provider: ProviderDriverKind.make("codex"),
40 packageName: "@openai/codex",
41 update: {
42 command: "npm install -g @openai/codex@latest",
43
44 executable: "npm",
45
46 args: ["install", "-g", "@openai/codex@latest"],
47
48 lockKey: "npm-global",
49 },
50} as const;
51
52const initialSnapshot: ServerProvider = {
53 instanceId: ProviderInstanceId.make("codex"),
54 driver: ProviderDriverKind.make("codex"),
55 enabled: true,
56 installed: true,
57 version: null,
58 status: "warning",
59 auth: { status: "unknown" },
60 checkedAt: "2026-04-10T00:00:00.000Z",
61 message: "Checking provider availability...",
62 models: [],
63 slashCommands: [],
64 skills: [],
65};
66
67const refreshedSnapshot: ServerProvider = {
68 instanceId: ProviderInstanceId.make("codex"),
69 driver: ProviderDriverKind.make("codex"),
70 enabled: true,
71 installed: true,
72 version: "1.0.0",
73 status: "ready",
74 auth: { status: "authenticated" },
75 checkedAt: "2026-04-10T00:00:01.000Z",
76 models: [],
77 slashCommands: [],
78 skills: [],
79};
80
81const enrichedSnapshot: ServerProvider = {
82 ...refreshedSnapshot,
83 checkedAt: "2026-04-10T00:00:02.000Z",
84 models: [
85 {
86 slug: "composer-2",
87 name: "Composer 2",
88 isCustom: false,
89 capabilities: fastModeCapabilities,
90 },
91 ],
92};
93
94const refreshedSnapshotSecond: ServerProvider = {
95 ...refreshedSnapshot,
96 checkedAt: "2026-04-10T00:00:03.000Z",
97 message: "Refreshed provider availability again.",
98};
99
100function layerProviderHost(input: {
101 readonly runBackgroundWork: boolean;
102 readonly settings?: Pick<ProviderHost.ProviderHost["Service"], "settings">["settings"];
103}) {
104 return Layer.succeed(
105 ProviderHost.ProviderHost,
106 ProviderHost.ProviderHost.of({
107 paths: {
108 cwd: process.cwd(),
109 baseDir: "/t3",
110 stateDir: "/t3/userdata",
111 providerStatusCacheDir: "/t3/caches",
112 attachmentsDir: "/t3/userdata/attachments",
113 },
114 settings: input.settings ?? {
115 get: Effect.succeed(DEFAULT_SERVER_SETTINGS),
116 withSnapshot: (use) => use(DEFAULT_SERVER_SETTINGS),
117 changes: Stream.empty,
118 subscribe: Effect.succeed(Stream.empty),
119 },
120 shouldRunBackgroundWork: () => Effect.succeed(input.runBackgroundWork),
121 resolveAttachmentPath: () => null,
122 credentials: () => Effect.die("unused"),
123 }),
124 );
125}
126
127const layerAlwaysRunTest = layerProviderHost({ runBackgroundWork: true });
128
129const enrichedSnapshotSecond: ServerProvider = {
130 ...refreshedSnapshotSecond,
131 checkedAt: "2026-04-10T00:00:04.000Z",
132 models: [
133 {
134 slug: "gpt-5.4",
135 name: "GPT-5.4",
136 isCustom: false,
137 capabilities: emptyCapabilities,
138 },
139 ],
140};
141
142describe("makeManagedServerProvider", () => {
143 it.effect("invalidates cached usage and queues one refresh behind an in-flight refresh", () =>
144 Effect.scoped(
145 Effect.gen(function* () {
146 const calls = yield* Ref.make(0);
147 const probeStarted = yield* Deferred.make<void>();
148 const releaseProbe = yield* Deferred.make<void>();
149 const cache = yield* Cache.makeWith(
150 () =>
151 Effect.gen(function* () {
152 const count = yield* Ref.updateAndGet(calls, (count) => count + 1);
153 if (count === 2) {
154 yield* Deferred.succeed(probeStarted, undefined);
155 yield* Deferred.await(releaseProbe);
156 }
157 return {
158 ...refreshedSnapshot,
159 usageLimits: {
160 checkedAt: refreshedSnapshot.checkedAt,
161 windows: [
162 {
163 id: "five_hour",
164 kind: "session",
165 label: "Session",
166 usedPercent: count === 1 ? 20 : 100,
167 },
168 ],
169 },
170 } satisfies ServerProvider;
171 }),
172 { capacity: 1, timeToLive: () => "5 minutes" },
173 );
174 const warmed = yield* Cache.get(cache, "claude");
175 const provider = yield* makeManagedServerProvider<TestSettings>({
176 resolveMaintenance: () => Effect.die("usage refresh must not resolve maintenance"),
177 getSettings: Effect.succeed({ enabled: true }),
178 streamSettings: Stream.empty,
179 haveSettingsChanged: () => false,
180 initialSnapshot: () => Effect.succeed(warmed),
181 checkProvider: Cache.get(cache, "claude"),
182 invalidateUsageCache: Cache.invalidateAll(cache),
183 refreshOnInterval: false,
184 });
185 yield* provider.streamChanges.pipe(Stream.take(1), Stream.runDrain);
186 assert.strictEqual(yield* Ref.get(calls), 1);
187 const refresh = yield* provider.refreshUsage!.pipe(Effect.forkChild);
188 yield* Deferred.await(probeStarted);
189 const queued = yield* provider.refreshUsage!.pipe(
190 Effect.forkChild({ startImmediately: true }),
191 );
192 yield* provider.refreshUsage!;
193 assert.strictEqual(yield* Ref.get(calls), 2);
194 assert.strictEqual((yield* provider.getSnapshot).usageLimits?.windows[0]?.usedPercent, 20);
195 yield* Deferred.succeed(releaseProbe, undefined);
196 yield* Fiber.join(refresh);
197 yield* Fiber.join(queued);
198 assert.strictEqual(yield* Ref.get(calls), 3);
199 assert.strictEqual((yield* provider.getSnapshot).usageLimits?.windows[0]?.usedPercent, 100);
200 }),
201 ).pipe(Effect.provide(layerAlwaysRunTest)),
202 );
203 it.effect("keeps the waiting usage refresh's claim when the running one is interrupted", () =>
204 Effect.scoped(
205 Effect.gen(function* () {
206 const calls = yield* Ref.make(0);
207 const usageProbeStarted = yield* Deferred.make<void>();
208 const statusProbeStarted = yield* Deferred.make<void>();
209 const releaseStatusProbe = yield* Deferred.make<void>();
210 const provider = yield* makeManagedServerProvider<TestSettings>({
211 resolveMaintenance: () => Effect.succeed(maintenanceCapabilities),
212 getSettings: Effect.succeed({ enabled: true }),
213 streamSettings: Stream.empty,
214 haveSettingsChanged: () => false,
215 initialSnapshot: () => Effect.succeed(initialSnapshot),
216 checkProvider: Effect.gen(function* () {
217 const count = yield* Ref.updateAndGet(calls, (count) => count + 1);
218 if (count === 2) {
219 yield* Deferred.succeed(usageProbeStarted, undefined);
220 return yield* Effect.never;
221 }
222 if (count === 3) {
223 yield* Deferred.succeed(statusProbeStarted, undefined);
224 yield* Deferred.await(releaseStatusProbe);
225 }
226 return {
227 ...refreshedSnapshot,
228 usageLimits: { checkedAt: refreshedSnapshot.checkedAt, windows: [] },
229 } satisfies ServerProvider;
230 }),
231 refreshOnInterval: false,
232 });
233 yield* provider.streamChanges.pipe(Stream.take(1), Stream.runDrain);
234 const running = yield* provider.refreshUsage!.pipe(Effect.forkChild);
235 yield* Deferred.await(usageProbeStarted);
236 const status = yield* provider.refresh.pipe(Effect.forkChild({ startImmediately: true }));
237 const waiting = yield* provider.refreshUsage!.pipe(
238 Effect.forkChild({ startImmediately: true }),
239 );
240 yield* Fiber.interrupt(running);
241 yield* Deferred.await(statusProbeStarted);
242 const extra = yield* provider.refreshUsage!.pipe(
243 Effect.forkChild({ startImmediately: true }),
244 );
245 yield* Deferred.succeed(releaseStatusProbe, undefined);
246 yield* Fiber.join(status);
247 yield* Fiber.join(waiting);
248 yield* Fiber.join(extra);
249 assert.strictEqual(yield* Ref.get(calls), 4);
250 }),
251 ).pipe(Effect.provide(layerAlwaysRunTest)),
252 );
253 it.effect(
254 "runs the initial provider check in the background and streams the refreshed snapshot",
255 () =>
256 Effect.scoped(
257 Effect.gen(function* () {
258 const checkCalls = yield* Ref.make(0);
259 const releaseCheck = yield* Deferred.make<void>();
260 const provider = yield* makeManagedServerProvider<TestSettings>({
261 resolveMaintenance: () => Effect.succeed(maintenanceCapabilities),
262 getSettings: Effect.succeed({ enabled: true }),
263 streamSettings: Stream.empty,
264 haveSettingsChanged: (previous, next) => previous.enabled !== next.enabled,
265 initialSnapshot: () => Effect.succeed(initialSnapshot),
266 checkProvider: Ref.update(checkCalls, (count) => count + 1).pipe(
267 Effect.flatMap(() => Deferred.await(releaseCheck)),
268 Effect.as(refreshedSnapshot),
269 ),
270 refreshInterval: "1 hour",
271 });
272
273 const initial = yield* provider.getSnapshot;
274 assert.deepStrictEqual(initial, initialSnapshot);
275 assert.strictEqual(yield* provider.isAwaitingFirstCheck!, true);
276
277 const updatesFiber = yield* Stream.take(provider.streamChanges, 1).pipe(
278 Stream.runCollect,
279 Effect.forkChild,
280 );
281 yield* Effect.yieldNow;
282 assert.strictEqual(yield* Ref.get(checkCalls), 1);
283
284 yield* Deferred.succeed(releaseCheck, undefined);
285
286 const updates = Array.from(yield* Fiber.join(updatesFiber));
287 const latest = yield* provider.getSnapshot;
288
289 assert.deepStrictEqual(updates, [refreshedSnapshot]);
290 assert.deepStrictEqual(latest, refreshedSnapshot);
291 assert.strictEqual(yield* provider.isAwaitingFirstCheck!, false);
292 assert.strictEqual(yield* Ref.get(checkCalls), 1);
293 }),
294 ).pipe(Effect.provide(layerAlwaysRunTest)),
295 );
296
297 it.effect("skips periodic provider refreshes without foreground provider-status demand", () =>
298 Effect.scoped(
299 Effect.gen(function* () {
300 const checkCalls = yield* Ref.make(0);
301 const initialCheckDone = yield* Deferred.make<void>();
302 yield* makeManagedServerProvider<TestSettings>({
303 resolveMaintenance: () => Effect.succeed(maintenanceCapabilities),
304 getSettings: Effect.succeed({ enabled: true }),
305 streamSettings: Stream.empty,
306 haveSettingsChanged: (previous, next) => previous.enabled !== next.enabled,
307 refreshOnInterval: () => true,
308 initialSnapshot: () => Effect.succeed(initialSnapshot),
309 checkProvider: Ref.updateAndGet(checkCalls, (count) => count + 1).pipe(
310 Effect.tap((count) =>
311 count === 1
312 ? Deferred.succeed(initialCheckDone, undefined).pipe(Effect.ignore)
313 : Effect.void,
314 ),
315 Effect.as(refreshedSnapshot),
316 ),
317 refreshInterval: "1 second",
318 });
319
320 yield* Deferred.await(initialCheckDone);
321 yield* TestClock.adjust("1 second");
322 yield* Effect.yieldNow;
323
324 assert.strictEqual(yield* Ref.get(checkCalls), 1);
325 }),
326 ).pipe(
327 Effect.provide(
328 Layer.mergeAll(layerProviderHost({ runBackgroundWork: false }), TestClock.layer()),
329 ),
330 ),
331 );
332
333 it.effect("stops periodic recovery checks once the snapshot is ready", () =>
334 Effect.scoped(
335 Effect.gen(function* () {
336 const checkCalls = yield* Ref.make(0);
337 const releaseInitialCheck = yield* Deferred.make<void>();
338 const failedSnapshot = { ...refreshedSnapshot, status: "error" as const };
339 const provider = yield* makeManagedServerProvider<TestSettings>({
340 resolveMaintenance: () => Effect.succeed(maintenanceCapabilities),
341 getSettings: Effect.succeed({ enabled: true }),
342 streamSettings: Stream.empty,
343 haveSettingsChanged: (previous, next) => previous.enabled !== next.enabled,
344 refreshOnInterval: (snapshot) => snapshot.status === "error",
345 initialSnapshot: () => Effect.succeed(initialSnapshot),
346 checkProvider: Deferred.await(releaseInitialCheck).pipe(
347 Effect.andThen(Ref.updateAndGet(checkCalls, (count) => count + 1)),
348 Effect.map((count) => (count === 1 ? failedSnapshot : refreshedSnapshot)),
349 ),
350 refreshInterval: "1 second",
351 });
352 const updates = yield* provider.streamChanges.pipe(
353 Stream.take(2),
354 Stream.runCollect,
355 Effect.forkChild,
356 );
357 yield* Effect.yieldNow;
358 yield* Deferred.succeed(releaseInitialCheck, undefined);
359 yield* TestClock.adjust("1 second");
360
361 assert.deepStrictEqual(Array.from(yield* Fiber.join(updates)), [
362 failedSnapshot,
363 refreshedSnapshot,
364 ]);
365 yield* TestClock.adjust("3 seconds");
366 assert.strictEqual(yield* Ref.get(checkCalls), 2);
367 }),
368 ).pipe(Effect.provide(Layer.mergeAll(layerAlwaysRunTest, TestClock.layer()))),
369 );
370
371 it.effect("rechecks a ready provider after one failed periodic check before publishing it", () =>
372 Effect.scoped(
373 Effect.gen(function* () {
374 const checkCalls = yield* Ref.make(0);
375 const initialCheckDone = yield* Deferred.make<void>();
376 const failedSnapshot: ServerProvider = {
377 ...refreshedSnapshot,
378 status: "error",
379 auth: { status: "unknown" },
380 message: "Codex app-server provider probe failed: workspace routing discovery timed out.",
381 };
382 const provider = yield* makeManagedServerProvider<TestSettings>({
383 resolveMaintenance: () => Effect.succeed(maintenanceCapabilities),
384 getSettings: Effect.succeed({ enabled: true }),
385 streamSettings: Stream.empty,
386 haveSettingsChanged: (previous, next) => previous.enabled !== next.enabled,
387 initialSnapshot: () => Effect.succeed(initialSnapshot),
388 checkProvider: Ref.updateAndGet(checkCalls, (count) => count + 1).pipe(
389 Effect.tap((count) =>
390 count === 1
391 ? Deferred.succeed(initialCheckDone, undefined).pipe(Effect.ignore)
392 : Effect.void,
393 ),
394 Effect.map((count) => (count === 2 ? failedSnapshot : refreshedSnapshot)),
395 ),
396 refreshInterval: "5 minutes",
397 });
398 yield* Deferred.await(initialCheckDone);
399 const published = yield* Ref.make<ReadonlyArray<ServerProvider>>([]);
400 yield* provider.streamChanges.pipe(
401 Stream.runForEach((snapshot) => Ref.update(published, (all) => [...all, snapshot])),
402 Effect.forkChild,
403 );
404 yield* Effect.yieldNow;
405
406 yield* TestClock.adjust("5 minutes");
407 yield* TestClock.adjust("1 minute");
408
409 assert.strictEqual((yield* provider.getSnapshot).status, "ready");
410 assert.deepStrictEqual(
411 (yield* Ref.get(published)).filter((snapshot) => snapshot.status === "error"),
412 [],
413 );
414 }),
415 ).pipe(Effect.provide(Layer.mergeAll(layerAlwaysRunTest, TestClock.layer()))),
416 );
417
418 it.effect("reruns the provider check when streamed settings change", () =>
419 Effect.scoped(
420 Effect.gen(function* () {
421 const settingsRef = yield* Ref.make<TestSettings>({ enabled: true });
422 const settingsChanges = yield* PubSub.unbounded<TestSettings>();
423 const checkCalls = yield* Ref.make(0);
424 const releaseInitialCheck = yield* Deferred.make<void>();
425 const releaseSettingsCheck = yield* Deferred.make<void>();
426 const provider = yield* makeManagedServerProvider<TestSettings>({
427 resolveMaintenance: () => Effect.succeed(maintenanceCapabilities),
428 getSettings: Ref.get(settingsRef),
429 streamSettings: Stream.fromPubSub(settingsChanges),
430 haveSettingsChanged: (previous, next) => previous.enabled !== next.enabled,
431 initialSnapshot: () => Effect.succeed(initialSnapshot),
432 checkProvider: Ref.updateAndGet(checkCalls, (count) => count + 1).pipe(
433 Effect.flatMap((count) =>
434 count === 1
435 ? Deferred.await(releaseInitialCheck).pipe(Effect.as(refreshedSnapshot))
436 : Deferred.await(releaseSettingsCheck).pipe(Effect.as(refreshedSnapshotSecond)),
437 ),
438 ),
439 refreshInterval: "1 hour",
440 });
441
442 const updatesFiber = yield* Stream.take(provider.streamChanges, 2).pipe(
443 Stream.runCollect,
444 Effect.forkChild,
445 );
446 yield* Effect.yieldNow;
447
448 yield* Deferred.succeed(releaseInitialCheck, undefined);
449 yield* Ref.set(settingsRef, { enabled: false });
450 yield* PubSub.publish(settingsChanges, { enabled: false });
451 yield* Deferred.succeed(releaseSettingsCheck, undefined);
452
453 const updates = Array.from(yield* Fiber.join(updatesFiber));
454 const latest = yield* provider.getSnapshot;
455
456 assert.deepStrictEqual(updates, [refreshedSnapshot, refreshedSnapshotSecond]);
457 assert.deepStrictEqual(latest, refreshedSnapshotSecond);
458 assert.strictEqual(yield* Ref.get(checkCalls), 2);
459 }),
460 ).pipe(Effect.provide(layerAlwaysRunTest)),
461 );
462
463 it.effect("keeps consuming settings changes after a provider check fails", () =>
464 Effect.scoped(
465 Effect.gen(function* () {
466 const settingsChanges = yield* PubSub.unbounded<TestSettings>();
467 const checkCalls = yield* Ref.make(0);
468 const releaseInitialCheck = yield* Deferred.make<void>();
469 const thirdCheckStarted = yield* Deferred.make<void>();
470 const releaseRecoveredCheck = yield* Deferred.make<void>();
471 const provider = yield* makeManagedServerProvider<TestSettings>({
472 resolveMaintenance: () => Effect.succeed(maintenanceCapabilities),
473 getSettings: Effect.succeed({ enabled: true }),
474 streamSettings: Stream.fromPubSub(settingsChanges),
475 haveSettingsChanged: (previous, next) => previous.enabled !== next.enabled,
476 initialSnapshot: () => Effect.succeed(initialSnapshot),
477 checkProvider: Ref.updateAndGet(checkCalls, (count) => count + 1).pipe(
478 Effect.flatMap((count) =>
479 count === 1
480 ? Deferred.await(releaseInitialCheck).pipe(Effect.as(refreshedSnapshot))
481 : count === 2
482 ? Effect.fail(
483 new ServerSettingsError({
484 settingsPath: "/t3/userdata/settings.json",
485 operation: "read-file",
486 }),
487 )
488 : Deferred.succeed(thirdCheckStarted, undefined).pipe(
489 Effect.andThen(Deferred.await(releaseRecoveredCheck)),
490 Effect.as(refreshedSnapshotSecond),
491 ),
492 ),
493 ),
494 refreshInterval: "1 hour",
495 });
496
497 const updatesFiber = yield* Stream.take(provider.streamChanges, 2).pipe(
498 Stream.runCollect,
499 Effect.forkChild,
500 );
501 yield* Effect.yieldNow;
502 yield* Deferred.succeed(releaseInitialCheck, undefined);
503
504 // The first change fails its provider check; the failed application never
505 // advances the stored settings, so the same change still differs next time.
506 yield* PubSub.publish(settingsChanges, { enabled: false });
507 yield* Effect.yieldNow;
508 assert.strictEqual(yield* Ref.get(checkCalls), 2);
509
510 // The subscription must survive that failure and retry on the next change;
511 // on main the dead consumer never starts the third check and this times out.
512 yield* PubSub.publish(settingsChanges, { enabled: false });
513 yield* Deferred.await(thirdCheckStarted);
514 assert.strictEqual(yield* Ref.get(checkCalls), 3);
515 yield* Deferred.succeed(releaseRecoveredCheck, undefined);
516
517 const updates = Array.from(yield* Fiber.join(updatesFiber));
518 assert.deepStrictEqual(updates, [refreshedSnapshot, refreshedSnapshotSecond]);
519 assert.deepStrictEqual(yield* provider.getSnapshot, refreshedSnapshotSecond);
520 }),
521 ).pipe(Effect.provide(layerAlwaysRunTest)),
522 );
523
524 it.effect(
525 "rechecks an auth failure when settings are applied, even if this provider's did not change",
526 () =>
527 Effect.scoped(
528 Effect.gen(function* () {
529 const settingsChanges = yield* PubSub.unbounded<
530 TestSettings & { readonly revision: number }
531 >();
532 const checkCalls = yield* Ref.make(0);
533 const seenRevisions: Array<number> = [];
534 const releaseInitialCheck = yield* Deferred.make<void>();
535 const markerApplied = yield* Deferred.make<void>();
536 const unauthenticatedSnapshot: ServerProvider = {
537 ...refreshedSnapshot,
538 status: "error",
539 auth: { status: "unauthenticated" },
540 message: "Codex CLI is not authenticated. Run `codex login` and try again.",
541 };
542 const provider = yield* makeManagedServerProvider<
543 TestSettings & { readonly revision: number }
544 >({
545 resolveMaintenance: () => Effect.succeed(maintenanceCapabilities),
546 getSettings: Effect.succeed({ enabled: true, revision: 0 }),
547 streamSettings: Stream.fromPubSub(settingsChanges),
548 haveSettingsChanged: (previous, next) => {
549 seenRevisions.push(next.revision);
550 if (next.revision === 2) Deferred.doneUnsafe(markerApplied, Effect.void);
551 return previous.enabled !== next.enabled;
552 },
553 initialSnapshot: () => Effect.succeed(initialSnapshot),
554 // The credential is fixed (e.g. `codex login`) after the first, failed probe.
555 checkProvider: Deferred.await(releaseInitialCheck).pipe(
556 Effect.andThen(Ref.updateAndGet(checkCalls, (count) => count + 1)),
557 Effect.map((count) => (count === 1 ? unauthenticatedSnapshot : refreshedSnapshot)),
558 ),
559 refreshInterval: "1 hour",
560 });
561 const initialProbe = yield* provider.streamChanges.pipe(
562 Stream.take(1),
563 Stream.runDrain,
564 Effect.forkChild,
565 );
566 yield* Effect.yieldNow;
567 yield* Deferred.succeed(releaseInitialCheck, undefined);
568 yield* Fiber.join(initialProbe);
569 assert.deepStrictEqual(yield* provider.getSnapshot, unauthenticatedSnapshot);
570
571 // A settings apply that does not touch this provider's settings, then a marker:
572 // the consumer is sequential, so the first apply has finished once the marker is seen.
573 yield* PubSub.publish(settingsChanges, { enabled: true, revision: 1 });
574 yield* PubSub.publish(settingsChanges, { enabled: true, revision: 2 });
575 yield* Deferred.await(markerApplied);
576
577 assert.deepStrictEqual(seenRevisions, [1, 2]);
578 assert.isAtLeast(yield* Ref.get(checkCalls), 2);
579 assert.deepStrictEqual((yield* provider.getSnapshot).auth, refreshedSnapshot.auth);
580 }),
581 ).pipe(Effect.provide(layerAlwaysRunTest)),
582 );
583
584 it.effect("streams supplemental snapshot updates after the base provider check completes", () =>
585 Effect.scoped(
586 Effect.gen(function* () {
587 const releaseEnrichment = yield* Deferred.make<void>();
588 const releaseCheck = yield* Deferred.make<void>();
589 const provider = yield* makeManagedServerProvider<TestSettings>({
590 resolveMaintenance: () => Effect.succeed(maintenanceCapabilities),
591 getSettings: Effect.succeed({ enabled: true }),
592 streamSettings: Stream.empty,
593 haveSettingsChanged: (previous, next) => previous.enabled !== next.enabled,
594 initialSnapshot: () => Effect.succeed(initialSnapshot),
595 checkProvider: Deferred.await(releaseCheck).pipe(Effect.as(refreshedSnapshot)),
596 enrichSnapshot: ({ publishSnapshot }) =>
597 Deferred.await(releaseEnrichment).pipe(
598 Effect.flatMap(() => publishSnapshot(enrichedSnapshot)),
599 ),
600 refreshInterval: "1 hour",
601 });
602
603 const updatesFiber = yield* Stream.take(provider.streamChanges, 2).pipe(
604 Stream.runCollect,
605 Effect.forkChild,
606 );
607 yield* Effect.yieldNow;
608
609 yield* Deferred.succeed(releaseCheck, undefined);
610
611 yield* Deferred.succeed(releaseEnrichment, undefined);
612
613 const updates = Array.from(yield* Fiber.join(updatesFiber));
614 const latest = yield* provider.getSnapshot;
615
616 assert.deepStrictEqual(updates, [refreshedSnapshot, enrichedSnapshot]);
617 assert.deepStrictEqual(latest, enrichedSnapshot);
618 }),
619 ).pipe(Effect.provide(layerAlwaysRunTest)),
620 );
621
622 it.effect("ignores stale enrichment callbacks after a newer refresh advances generation", () =>
623 Effect.scoped(
624 Effect.gen(function* () {
625 const publishCallbacks: Array<(snapshot: ServerProvider) => Effect.Effect<void>> = [];
626 const refreshCount = yield* Ref.make(0);
627 const firstCallbackReady = yield* Deferred.make<void>();
628 const secondCallbackReady = yield* Deferred.make<void>();
629 const allowFirstRefresh = yield* Deferred.make<void>();
630 const provider = yield* makeManagedServerProvider<TestSettings>({
631 resolveMaintenance: () => Effect.succeed(maintenanceCapabilities),
632 getSettings: Effect.succeed({ enabled: true }),
633 streamSettings: Stream.empty,
634 haveSettingsChanged: (previous, next) => previous.enabled !== next.enabled,
635 initialSnapshot: () => Effect.succeed(initialSnapshot),
636 checkProvider: Ref.updateAndGet(refreshCount, (count) => count + 1).pipe(
637 Effect.flatMap((count) =>
638 count === 1
639 ? Deferred.await(allowFirstRefresh).pipe(Effect.as(refreshedSnapshot))
640 : Effect.succeed(refreshedSnapshotSecond),
641 ),
642 ),
643 enrichSnapshot: ({ publishSnapshot }) =>
644 Effect.gen(function* () {
645 publishCallbacks.push(publishSnapshot);
646 if (publishCallbacks.length === 1) {
647 yield* Deferred.succeed(firstCallbackReady, undefined).pipe(Effect.ignore);
648 } else if (publishCallbacks.length === 2) {
649 yield* Deferred.succeed(secondCallbackReady, undefined).pipe(Effect.ignore);
650 }
651 }),
652 refreshInterval: "1 hour",
653 });
654
655 const updatesFiber = yield* Stream.take(provider.streamChanges, 3).pipe(
656 Stream.runCollect,
657 Effect.forkChild,
658 );
659 yield* Effect.yieldNow;
660
661 yield* Deferred.succeed(allowFirstRefresh, undefined);
662 yield* Deferred.await(firstCallbackReady);
663
664 yield* provider.refresh;
665 yield* Deferred.await(secondCallbackReady);
666
667 yield* publishCallbacks[0]!(enrichedSnapshot);
668 yield* publishCallbacks[1]!(enrichedSnapshotSecond);
669
670 const updates = Array.from(yield* Fiber.join(updatesFiber));
671 const latest = yield* provider.getSnapshot;
672
673 assert.deepStrictEqual(updates, [
674 refreshedSnapshot,
675 refreshedSnapshotSecond,
676 enrichedSnapshotSecond,
677 ]);
678 assert.deepStrictEqual(latest, enrichedSnapshotSecond);
679 }),
680 ).pipe(Effect.provide(layerAlwaysRunTest)),
681 );
682
683 it.effect("applies runtime usage updates onto the published snapshot", () =>
684 Effect.scoped(
685 Effect.gen(function* () {
686 const provider = yield* makeManagedServerProvider<TestSettings>({
687 resolveMaintenance: () => Effect.succeed(maintenanceCapabilities),
688 getSettings: Effect.succeed({ enabled: true }),
689 streamSettings: Stream.empty,
690 haveSettingsChanged: (previous, next) => previous.enabled !== next.enabled,
691 initialSnapshot: () => Effect.succeed(initialSnapshot),
692 checkProvider: Effect.succeed({
693 ...refreshedSnapshot,
694 usageLimits: {
695 checkedAt: "2026-04-10T00:00:01.000Z",
696 windows: [
697 { id: "five_hour", kind: "session", label: "Session", usedPercent: 10 },
698 {
699 id: "seven_day",
700 kind: "weekly",
701 label: "Weekly",
702 usedPercent: 20,
703 resetsAt: "2026-04-17T00:00:00.000Z",
704 },
705 ],
706 },
707 } satisfies ServerProvider),
708 refreshInterval: "1 hour",
709 });
710 yield* Stream.take(provider.streamChanges, 1).pipe(Stream.runDrain);
711
712 const updatesFiber = yield* Stream.take(provider.streamChanges, 1).pipe(
713 Stream.runCollect,
714 Effect.forkChild,
715 );
716 yield* Effect.yieldNow;
717
718 yield* provider.applyUsageLimits({
719 checkedAt: "2026-04-10T00:05:00.000Z",
720 windows: [{ id: "seven_day", kind: "weekly", label: "Weekly", usedPercent: 25 }],
721 authLabel: "tier-1",
722 });
723 yield* provider.applyUsageLimits({
724 checkedAt: "2026-04-10T00:06:00.000Z",
725 windows: [],
726 authLabel: "tier-2",
727 });
728
729 const [update] = Array.from(yield* Fiber.join(updatesFiber));
730 assert.strictEqual(update?.auth.label, "tier-1");
731 assert.deepStrictEqual(update?.usageLimits, {
732 checkedAt: "2026-04-10T00:05:00.000Z",
733 windows: [
734 { id: "five_hour", kind: "session", label: "Session", usedPercent: 10 },
735 {
736 id: "seven_day",
737 kind: "weekly",
738 label: "Weekly",
739 usedPercent: 25,
740 resetsAt: "2026-04-17T00:00:00.000Z",
741 },
742 ],
743 });
744 assert.deepStrictEqual(yield* provider.getSnapshot, update);
745 }),
746 ).pipe(Effect.provide(layerAlwaysRunTest)),
747 );
748
749 it.effect.each([
750 { name: "a live plan", clearPlan: false },
751 { name: "a cleared plan", clearPlan: true },
752 ])("keeps $name and usage across a failed probe and stale enrichment", ({ clearPlan }) =>
753 Effect.scoped(
754 Effect.gen(function* () {
755 const releaseEnrichment = yield* Deferred.make<void>();
756 const refreshCount = yield* Ref.make(0);
757 const probedLimits = {
758 checkedAt: "2026-04-10T00:00:01.000Z",
759 windows: [{ id: "primary", kind: "session", label: "Session", usedPercent: 10 }],
760 } as const;
761 const provider = yield* makeManagedServerProvider<TestSettings>({
762 resolveMaintenance: () => Effect.succeed(maintenanceCapabilities),
763 getSettings: Effect.succeed({ enabled: true }),
764 streamSettings: Stream.empty,
765 haveSettingsChanged: (previous, next) => previous.enabled !== next.enabled,
766 initialSnapshot: () => Effect.succeed(initialSnapshot),
767 checkProvider: Ref.updateAndGet(refreshCount, (count) => count + 1).pipe(
768 Effect.map((count) =>
769 count === 1
770 ? {
771 ...refreshedSnapshot,
772 auth: { ...refreshedSnapshot.auth, workspaceId: "ws-a", label: "tier-1" },
773 usageLimits: probedLimits,
774 }
775 : {
776 ...refreshedSnapshotSecond,
777 ...(count === 3
778 ? { auth: { ...refreshedSnapshotSecond.auth, workspaceId: "ws-b" } }
779 : {}),
780 usageLimits: {
781 checkedAt: "2026-04-10T00:00:03.000Z",
782 windows: [],
783 unavailable: { reason: "probeFailed" },
784 },
785 },
786 ),
787 ),
788 enrichSnapshot: ({ snapshot, publishSnapshot }) =>
789 Deferred.await(releaseEnrichment).pipe(
790 Effect.flatMap(() =>
791 publishSnapshot({
792 ...enrichedSnapshot,
793 ...snapshot,
794 models: enrichedSnapshot.models,
795 }),
796 ),
797 ),
798 refreshInterval: "1 hour",
799 });
800 yield* Stream.take(provider.streamChanges, 1).pipe(Stream.runDrain);
801
802 const liveWindow = {
803 id: "primary",
804 kind: "session",
805 label: "Session",
806 usedPercent: 60,
807 } as const;
808 yield* provider.applyUsageLimits({
809 checkedAt: "2026-04-10T00:00:02.000Z",
810 authLabel: "tier-2",
811 windows: [liveWindow],
812 });
813
814 if (clearPlan) {
815 yield* provider.applyUsageLimits({
816 checkedAt: "2026-04-10T00:00:02.000Z",
817 complete: true,
818 windows: [liveWindow],
819 });
820 }
821
822 yield* Deferred.succeed(releaseEnrichment, undefined);
823 const enriched = yield* Stream.take(provider.streamChanges, 1).pipe(
824 Stream.runCollect,
825 Effect.map((chunk) => Array.from(chunk)[0]!),
826 );
827 assert.deepStrictEqual(enriched.models, enrichedSnapshot.models);
828 assert.deepStrictEqual(enriched.usageLimits?.windows, [liveWindow]);
829 assert.strictEqual(enriched.auth.label, clearPlan ? undefined : "tier-2");
830
831 const refreshed = yield* provider.refresh;
832 assert.strictEqual(refreshed.message, refreshedSnapshotSecond.message);
833 assert.deepStrictEqual(refreshed.usageLimits?.windows, [liveWindow]);
834 assert.strictEqual(refreshed.auth.workspaceId, "ws-a");
835 assert.strictEqual(refreshed.auth.label, clearPlan ? undefined : "tier-2");
836
837 const switched = yield* provider.refresh;
838 assert.strictEqual(switched.auth.workspaceId, "ws-b");
839 assert.strictEqual(switched.usageLimits?.unavailable?.reason, "probeFailed");
840 }),
841 ).pipe(Effect.provide(layerAlwaysRunTest)),
842 );
843
844 it.effect.each([
845 { name: "an empty reading", hasUsage: true, clearPlan: false },
846 { name: "no reading", hasUsage: false, clearPlan: false },
847 { name: "an empty reading after clearing the plan", hasUsage: true, clearPlan: true },
848 { name: "no reading after clearing the plan", hasUsage: false, clearPlan: true },
849 ])("keeps the live plan and limits after $name", ({ hasUsage, clearPlan }) =>
850 Effect.scoped(
851 Effect.gen(function* () {
852 const started = yield* Deferred.make<void>();
853 const release = yield* Deferred.make<void>();
854 const count = yield* Ref.make(0);
855 const provider = yield* makeManagedServerProvider<TestSettings>({
856 resolveMaintenance: () => Effect.succeed(maintenanceCapabilities),
857 getSettings: Effect.succeed({ enabled: true }),
858 streamSettings: Stream.empty,
859 haveSettingsChanged: (previous, next) => previous.enabled !== next.enabled,
860 initialSnapshot: () => Effect.succeed(initialSnapshot),
861 checkProvider: Effect.gen(function* () {
862 if ((yield* Ref.updateAndGet(count, (n) => n + 1)) > 1) {
863 yield* Deferred.succeed(started, undefined);
864 yield* Deferred.await(release);
865 }
866 return {
867 ...refreshedSnapshot,
868 auth: { status: "authenticated", label: "tier-1" },
869 ...(hasUsage
870 ? { usageLimits: { checkedAt: refreshedSnapshot.checkedAt, windows: [] } }
871 : {}),
872 } satisfies ServerProvider;
873 }),
874 });
875 yield* Stream.take(provider.streamChanges, 1).pipe(Stream.runDrain);
876 const refresh = yield* provider.refresh.pipe(Effect.forkChild);
877 yield* Deferred.await(started);
878 yield* provider.applyUsageLimits({
879 checkedAt: "2026-04-10T00:00:04.000Z",
880 authLabel: "tier-2",
881 complete: true,
882 windows: [{ id: "session", kind: "session", label: "Session", usedPercent: 70 }],
883 });
884 if (clearPlan) {
885 yield* provider.applyUsageLimits({
886 checkedAt: "2026-04-10T00:00:04.000Z",
887 complete: true,
888 windows: [{ id: "session", kind: "session", label: "Session", usedPercent: 70 }],
889 });
890 }
891 yield* Deferred.succeed(release, undefined);
892 const refreshed = yield* Fiber.join(refresh);
893 assert.strictEqual(refreshed.auth.label, clearPlan ? undefined : "tier-2");
894 assert.strictEqual(refreshed.usageLimits?.windows[0]?.usedPercent, 70);
895 assert.strictEqual(refreshed.usageLimits?.unavailable, undefined);
896 }),
897 ).pipe(Effect.provide(layerAlwaysRunTest)),
898 );
899
900 it.effect("drops an earlier plan override after a complete reading without a label", () =>
901 Effect.scoped(
902 Effect.gen(function* () {
903 const usageLimits = {
904 checkedAt: "2026-04-10T00:00:04.000Z",
905 windows: [{ id: "session", kind: "session", label: "Session", usedPercent: 70 }],
906 } as const;
907 const provider = yield* makeManagedServerProvider<TestSettings>({
908 resolveMaintenance: () => Effect.succeed(maintenanceCapabilities),
909 getSettings: Effect.succeed({ enabled: true }),
910 streamSettings: Stream.empty,
911 haveSettingsChanged: (previous, next) => previous.enabled !== next.enabled,
912 initialSnapshot: () => Effect.succeed(initialSnapshot),
913 checkProvider: Effect.succeed({ ...refreshedSnapshot, usageLimits }),
914 refreshOnInterval: false,
915 });
916 yield* Stream.take(provider.streamChanges, 1).pipe(Stream.runDrain);
917 yield* provider.applyUsageLimits({ ...usageLimits, complete: true, authLabel: "tier-2" });
918 assert.strictEqual((yield* provider.refresh).auth.label, "tier-2");
919
920 yield* provider.applyUsageLimits({ ...usageLimits, complete: true });
921 assert.strictEqual((yield* provider.getSnapshot).auth.label, undefined);
922 const refreshed = yield* provider.refresh;
923 assert.strictEqual(refreshed.auth.label, undefined);
924 assert.deepStrictEqual(refreshed.usageLimits, usageLimits);
925 }),
926 ).pipe(Effect.provide(layerAlwaysRunTest)),
927 );
928
929 it.effect("keeps sparse usage unavailable after probe failure until a full probe recovers", () =>
930 Effect.scoped(
931 Effect.gen(function* () {
932 const failedLimits = {
933 checkedAt: "2026-04-10T00:00:01.000Z",
934 windows: [],
935 unavailable: { reason: "probeFailed", message: "Usage probe timed out." },
936 } as const;
937 const probeSnapshot = yield* Ref.make<ServerProvider>({
938 ...refreshedSnapshot,
939 usageLimits: failedLimits,
940 });
941 const provider = yield* makeManagedServerProvider<TestSettings>({
942 resolveMaintenance: () => Effect.succeed(maintenanceCapabilities),
943 getSettings: Effect.succeed({ enabled: true }),
944 streamSettings: Stream.empty,
945 haveSettingsChanged: (previous, next) => previous.enabled !== next.enabled,
946 initialSnapshot: () => Effect.succeed(initialSnapshot),
947 checkProvider: Ref.get(probeSnapshot),
948 refreshOnInterval: false,
949 });
950 yield* Stream.take(provider.streamChanges, 1).pipe(Stream.runDrain);
951
952 const weekly = {
953 id: "seven_day",
954 kind: "weekly",
955 label: "Weekly",
956 usedPercent: 20,
957 } as const;
958 yield* provider.applyUsageLimits({
959 checkedAt: "2026-04-10T00:00:02.000Z",
960 windows: [weekly],
961 });
962 const partial = yield* provider.getSnapshot;
963 assert.deepStrictEqual(partial.usageLimits, {
964 checkedAt: "2026-04-10T00:00:02.000Z",
965 windows: [weekly],
966 unavailable: failedLimits.unavailable,
967 });
968 yield* provider.applyUsageLimits({
969 checkedAt: "2026-04-10T00:00:03.000Z",
970 windows: [{ ...weekly }],
971 });
972 assert.strictEqual(yield* provider.getSnapshot, partial);
973 assert.deepStrictEqual((yield* provider.refresh).usageLimits, failedLimits);
974
975 const completeLimits = {
976 checkedAt: "2026-04-10T00:00:04.000Z",
977 windows: [
978 { id: "five_hour", kind: "session", label: "Session", usedPercent: 10 },
979 { ...weekly, usedPercent: 25 },
980 ],
981 } as const;
982 yield* Ref.set(probeSnapshot, { ...refreshedSnapshot, usageLimits: completeLimits });
983 assert.deepStrictEqual((yield* provider.refresh).usageLimits, completeLimits);
984 }),
985 ).pipe(Effect.provide(layerAlwaysRunTest)),
986 );
987});