1import {
2 DEFAULT_SERVER_SETTINGS,
3 EnvironmentId,
4 UsageDay,
5 UsageReadError,
6 USAGE_CONTRACT_VERSION,
7 WS_METHODS,
8 type ServerConfig,
9 type ServerConfigStreamEvent,
10 type ServerSettings,
11 type UsageSource,
12 type UsageSummary,
13 type UsageSummaryInput,
14} from "@t3tools/contracts";
15import { expect, it } from "@effect/vitest";
16import * as Deferred from "effect/Deferred";
17import * as Effect from "effect/Effect";
18import * as Layer from "effect/Layer";
19import * as Option from "effect/Option";
20import * as Queue from "effect/Queue";
21import * as Result from "effect/Result";
22import * as Stream from "effect/Stream";
23import * as SubscriptionRef from "effect/SubscriptionRef";
24import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
25
26import {
27 AVAILABLE_CONNECTION_STATE,
28 PrimaryConnectionTarget,
29 type PreparedConnection,
30 type SupervisorConnectionState,
31} from "../connection/model.ts";
32import type { EnvironmentPresentation } from "../connection/presentation.ts";
33import * as EnvironmentRegistry from "../connection/registry.ts";
34import * as EnvironmentSupervisor from "../connection/supervisor.ts";
35import * as Persistence from "../platform/persistence.ts";
36import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
37import type { RpcSession } from "../rpc/session.ts";
38import { createServerEnvironmentAtoms } from "./server.ts";
39import { refreshUsage } from "./usage.ts";
40
41const TARGET = new PrimaryConnectionTarget({
42 environmentId: EnvironmentId.make("usage-environment"),
43 label: "Usage environment",
44 httpBaseUrl: "https://environment.example.test",
45 wsBaseUrl: "wss://environment.example.test",
46});
47
48const INPUT: UsageSummaryInput = {
49 sinceDay: UsageDay.make("2026-09-01"),
50 untilDay: UsageDay.make("2026-09-04"),
51 timeZone: "UTC",
52};
53
54const CONFIG = {
55 settings: DEFAULT_SERVER_SETTINGS,
56 environment: { serverVersion: "0.0.1", capabilities: {} },
57} as ServerConfig;
58
59const PRICING = { status: "unavailable" as const, source: "test", fetchedAt: null, knownModels: 0 };
60
61const cursorSource = (refreshing: boolean): UsageSource => ({
62 fingerprint: {
63 hostId: "cursor.com",
64 provider: "cursor",
65 resolvedHomePath: "cursor-account:abc",
66 volumeId: "abc",
67 },
68 status: "ok",
69 scannedFiles: 1,
70 skippedFiles: 0,
71 malformedRecords: 0,
72 distinctSessions: 1,
73 message: null,
74 ...(refreshing ? { refreshing: true as const } : {}),
75});
76
77const makeHarness = Effect.fn("ServerUsageTest.makeHarness")(function* (
78 beforeRead: (
79 request: number,
80 input: UsageSummaryInput,
81 ) => Effect.Effect<void, UsageReadError> = () => Effect.void,
82 sources: (input: UsageSummaryInput) => readonly UsageSource[] = () => [],
83) {
84 const events = yield* Queue.unbounded<ServerConfigStreamEvent>();
85 let settings = DEFAULT_SERVER_SETTINGS;
86 let requests = 0;
87 const inputs: UsageSummaryInput[] = [];
88 const client = {
89 [WS_METHODS.serverRefreshUsageRates]: () => Effect.succeed(PRICING),
90 [WS_METHODS.subscribeServerConfig]: () =>
91 Stream.concat(
92 Stream.make({ version: 1 as const, type: "snapshot" as const, config: CONFIG }),
93 Stream.fromQueue(events),
94 ),
95 [WS_METHODS.serverGetUsageSummary]: (input: UsageSummaryInput) =>
96 Effect.gen(function* () {
97 requests += 1;
98 inputs.push(input);
99 const price = settings.usagePriceOverrides["custom-model"]?.inputCostPerMillionTokens ?? 0;
100 yield* beforeRead(requests, input);
101 return {
102 contractVersion: USAGE_CONTRACT_VERSION,
103 readAt: "2026-09-04T12:00:00Z",
104 ...input,
105 buckets: [
106 {
107 day: input.sinceDay,
108 provider: "codex",
109 model: "custom-model",
110 totals: {
111 uncachedInputTokens: 1_000_000,
112 cachedInputTokens: 0,
113 cacheCreationTokens: 0,
114 outputTokens: 0,
115 reasoningTokens: 0,
116 },
117 costUsd: price,
118 cacheSavingsUsd: 0,
119 costSource: price === 0 ? "unpriced" : "modelPriced",
120 records: 1,
121 unpricedRecords: price === 0 ? 1 : 0,
122 sessions: 1,
123 },
124 ],
125 sources: sources(input),
126 pricing: PRICING,
127 scanDurationMs: 0,
128 } satisfies UsageSummary;
129 }),
130 } as unknown as WsRpcProtocolClient;
131 const session: RpcSession = {
132 client,
133 initialConfig: Effect.succeed(CONFIG),
134 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
135 ready: Effect.void,
136 probe: Effect.void,
137 closed: Effect.never,
138 };
139 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
140 target: TARGET,
141 state: yield* SubscriptionRef.make<SupervisorConnectionState>({
142 ...AVAILABLE_CONNECTION_STATE,
143 phase: "connected",
144 }),
145 session: yield* SubscriptionRef.make(Option.some(session)),
146 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
147 connect: Effect.void,
148 disconnect: Effect.void,
149 retryNow: Effect.void,
150 });
151 const environments = EnvironmentRegistry.EnvironmentRegistry.of({
152 run: (_environmentId, effect) =>
153 Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
154 followStream: (_environmentId, stream) =>
155 Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
156 } as EnvironmentRegistry.EnvironmentRegistry["Service"]);
157 const cache = Persistence.EnvironmentCacheStore.of({
158 loadShell: () => Effect.succeedNone,
159 saveShell: () => Effect.void,
160 loadThread: () => Effect.succeedNone,
161 saveThread: () => Effect.void,
162 removeThread: () => Effect.void,
163 loadServerConfig: () => Effect.succeedNone,
164 saveServerConfig: () => Effect.void,
165 loadVcsRefs: () => Effect.succeedNone,
166 saveVcsRefs: () => Effect.void,
167 removeVcsRefs: () => Effect.void,
168 clearVcsRefs: () => Effect.void,
169 clear: () => Effect.void,
170 });
171 const runtime = Atom.runtime(
172 Layer.merge(
173 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environments),
174 Layer.succeed(Persistence.EnvironmentCacheStore, cache),
175 ),
176 );
177 const initialConfigValueAtom = Atom.make(CONFIG);
178 const atoms = createServerEnvironmentAtoms(runtime, {
179 initialConfigValueAtom: () => initialConfigValueAtom,
180 });
181 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
182 Effect.sync(() => registry.dispose()),
183 );
184 const updateSettings = Effect.fn("ServerUsageTest.updateSettings")(function* (
185 next: ServerSettings,
186 ) {
187 settings = next;
188 yield* Queue.offer(events, {
189 version: 1,
190 type: "settingsUpdated",
191 payload: { settings },
192 });
193 yield* AtomRegistry.toStream(registry, atoms.settingsValueAtom(TARGET.environmentId)).pipe(
194 Stream.filter((current) => current === next),
195 Stream.runHead,
196 );
197 });
198 return {
199 registry,
200 atoms,
201 requests: () => requests,
202 inputs: () => inputs,
203 updateSettings,
204 summary: (input = INPUT) => atoms.usageSummary({ environmentId: TARGET.environmentId, input }),
205 };
206});
207
208function settledSummary<E>(
209 registry: AtomRegistry.AtomRegistry,
210 atom: Atom.Atom<AsyncResult.AsyncResult<UsageSummary, E>>,
211) {
212 return AtomRegistry.toStream(registry, atom).pipe(
213 Stream.filterMap((result) =>
214 AsyncResult.isSuccess(result) && !result.waiting
215 ? Result.succeed(result.value)
216 : Result.failVoid,
217 ),
218 Stream.runHead,
219 Effect.map(Option.getOrThrow),
220 );
221}
222
223/** Answers with a refreshing Cursor source; the `awaitRefresh` request runs `followUp` first. */
224const makeRefreshingHarness = Effect.fn("ServerUsageTest.makeRefreshingHarness")(function* (
225 followUp: Effect.Effect<void, UsageReadError>,
226) {
227 const followUpStarted = yield* Deferred.make<void>();
228 const harness = yield* makeHarness(
229 (_request, input) =>
230 input.awaitRefresh === true
231 ? Deferred.succeed(followUpStarted, undefined).pipe(Effect.andThen(followUp))
232 : Effect.void,
233 (input) => [cursorSource(input.awaitRefresh !== true)],
234 );
235 return { ...harness, followUpStarted };
236});
237
238function waitForCost<E>(
239 registry: AtomRegistry.AtomRegistry,
240 atom: Atom.Atom<AsyncResult.AsyncResult<UsageSummary, E>>,
241 cost: number,
242) {
243 return AtomRegistry.toStream(registry, atom).pipe(
244 Stream.filter(
245 (result) =>
246 AsyncResult.isSuccess(result) &&
247 !result.waiting &&
248 result.value.buckets[0]?.costUsd === cost,
249 ),
250 Stream.runHead,
251 );
252}
253
254it.effect("refreshes cached usage windows only when override prices change", () =>
255 Effect.scoped(
256 Effect.gen(function* () {
257 const harness = yield* makeHarness();
258 const current = harness.summary();
259 const previous = harness.summary({ ...INPUT, sinceDay: UsageDay.make("2026-08-01") });
260 const unmountCurrent = harness.registry.mount(current);
261 const unmountPrevious = harness.registry.mount(previous);
262 yield* waitForCost(harness.registry, current, 0);
263 yield* waitForCost(harness.registry, previous, 0);
264 unmountPrevious();
265 expect(harness.requests()).toBe(2);
266
267 const prices = {
268 "custom-model": { inputCostPerMillionTokens: 3, outputCostPerMillionTokens: 9 },
269 "other-model": { inputCostPerMillionTokens: 1, outputCostPerMillionTokens: 2 },
270 };
271 yield* harness.updateSettings({ ...DEFAULT_SERVER_SETTINGS, usagePriceOverrides: prices });
272 yield* waitForCost(harness.registry, current, 3);
273 yield* waitForCost(harness.registry, previous, 3);
274 expect(harness.requests()).toBe(4);
275
276 yield* harness.updateSettings({
277 ...DEFAULT_SERVER_SETTINGS,
278 usagePriceOverrides: {
279 "other-model": { outputCostPerMillionTokens: 2, inputCostPerMillionTokens: 1 },
280 "custom-model": { outputCostPerMillionTokens: 9, inputCostPerMillionTokens: 3 },
281 },
282 });
283 yield* harness.updateSettings({
284 ...DEFAULT_SERVER_SETTINGS,
285 usagePriceOverrides: prices,
286 defaultTheme: "custom-theme",
287 });
288 yield* harness.updateSettings(DEFAULT_SERVER_SETTINGS);
289 yield* waitForCost(harness.registry, current, 0);
290 yield* waitForCost(harness.registry, previous, 0);
291 expect(harness.requests()).toBe(6);
292 unmountCurrent();
293 }),
294 ),
295);
296
297it.effect("restarts a pending usage read after a price change", () =>
298 Effect.scoped(
299 Effect.gen(function* () {
300 const started = yield* Deferred.make<void>();
301 const interrupted = yield* Deferred.make<void>();
302 const harness = yield* makeHarness((request) =>
303 request === 1
304 ? Deferred.succeed(started, undefined).pipe(
305 Effect.andThen(Effect.never),
306 Effect.onInterrupt(() => Deferred.succeed(interrupted, undefined)),
307 )
308 : Effect.void,
309 );
310 const summary = harness.summary();
311 const unmount = harness.registry.mount(summary);
312 yield* Deferred.await(started);
313 yield* harness.updateSettings({
314 ...DEFAULT_SERVER_SETTINGS,
315 usagePriceOverrides: {
316 "custom-model": { inputCostPerMillionTokens: 5, outputCostPerMillionTokens: 10 },
317 },
318 });
319 yield* waitForCost(harness.registry, summary, 5);
320 yield* Deferred.await(interrupted);
321 expect(harness.requests()).toBe(2);
322 unmount();
323 }),
324 ),
325);
326
327it.effect("settles after one usage read when no source is refreshing", () =>
328 Effect.scoped(
329 Effect.gen(function* () {
330 const harness = yield* makeHarness();
331 const summary = harness.summary();
332 const unmount = harness.registry.mount(summary);
333 yield* settledSummary(harness.registry, summary);
334 expect(harness.inputs()).toEqual([INPUT]);
335 unmount();
336 }),
337 ),
338);
339
340it.effect("shows the cached usage summary until one awaited refresh replaces it", () =>
341 Effect.scoped(
342 Effect.gen(function* () {
343 const refreshed = yield* Deferred.make<void>();
344 const harness = yield* makeRefreshingHarness(Deferred.await(refreshed));
345 const summary = harness.summary();
346 const unmount = harness.registry.mount(summary);
347 yield* Deferred.await(harness.followUpStarted);
348
349 const provisional = harness.registry.get(summary);
350 expect(AsyncResult.isSuccess(provisional) && provisional.waiting).toBe(true);
351 expect(Option.getOrNull(AsyncResult.value(provisional))?.sources).toEqual([
352 cursorSource(true),
353 ]);
354 expect(harness.inputs()).toEqual([INPUT, { ...INPUT, awaitRefresh: true }]);
355
356 yield* Deferred.succeed(refreshed, undefined);
357 expect((yield* settledSummary(harness.registry, summary)).sources).toEqual([
358 cursorSource(false),
359 ]);
360 expect(harness.requests()).toBe(2);
361 unmount();
362 }),
363 ),
364);
365
366it.effect("keeps the cached usage summary when the awaited refresh fails", () =>
367 Effect.scoped(
368 Effect.gen(function* () {
369 const harness = yield* makeRefreshingHarness(
370 Effect.fail(new UsageReadError({ reason: "scanFailed", detail: "Cursor is down" })),
371 );
372 const summary = harness.summary();
373 const unmount = harness.registry.mount(summary);
374 expect((yield* settledSummary(harness.registry, summary)).sources).toEqual([
375 cursorSource(true),
376 ]);
377 expect(AsyncResult.isSuccess(harness.registry.get(summary))).toBe(true);
378 expect(harness.requests()).toBe(2);
379 unmount();
380 }),
381 ),
382);
383
384it.effect("finishes a manual usage refresh only after the awaited refresh", () =>
385 Effect.scoped(
386 Effect.gen(function* () {
387 const refreshed = yield* Deferred.make<void>();
388 const harness = yield* makeRefreshingHarness(Deferred.await(refreshed));
389 const presentation = Atom.make({
390 connection: { phase: "connected" },
391 } as EnvironmentPresentation | null);
392 let finished = false;
393 const refreshing = refreshUsage({
394 registry: harness.registry,
395 server: harness.atoms,
396 presentations: { presentationAtom: () => presentation },
397 environmentIds: [TARGET.environmentId],
398 input: INPUT,
399 }).then(() => {
400 finished = true;
401 });
402 yield* Deferred.await(harness.followUpStarted);
403 expect(finished).toBe(false);
404 yield* Deferred.succeed(refreshed, undefined);
405 yield* Effect.promise(() => refreshing);
406 expect(harness.requests()).toBe(2);
407 }),
408 ),
409);