import { DEFAULT_SERVER_SETTINGS, EnvironmentId, UsageDay, USAGE_CONTRACT_VERSION, WS_METHODS, type ServerConfig, type ServerConfigStreamEvent, type ServerSettings, type UsageSummary, type UsageSummaryInput,} from "@t3tools/contracts";import { expect, it } from "@effect/vitest";import * as Deferred from "effect/Deferred";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Queue from "effect/Queue";import * as Stream from "effect/Stream";import * as SubscriptionRef from "effect/SubscriptionRef";import { AsyncResult, Atom, AtomRegistry } from "effect/unstable/reactivity"; import { AVAILABLE_CONNECTION_STATE, PrimaryConnectionTarget, type PreparedConnection, type SupervisorConnectionState,} from "../connection/model.ts";import { EnvironmentRegistry } from "../connection/registry.ts";import { EnvironmentSupervisor } from "../connection/supervisor.ts";import { EnvironmentCacheStore } from "../platform/persistence.ts";import type { WsRpcProtocolClient } from "../rpc/protocol.ts";import type { RpcSession } from "../rpc/session.ts";import { createServerEnvironmentAtoms } from "./server.ts"; const TARGET = new PrimaryConnectionTarget({ environmentId: EnvironmentId.make("usage-environment"), label: "Usage environment", httpBaseUrl: "https://environment.example.test", wsBaseUrl: "wss://environment.example.test",}); const INPUT: UsageSummaryInput = { sinceDay: UsageDay.make("2026-09-01"), untilDay: UsageDay.make("2026-09-04"), timeZone: "UTC",}; const CONFIG = { settings: DEFAULT_SERVER_SETTINGS, environment: { serverVersion: "0.0.1", capabilities: {} },} as ServerConfig; const makeHarness = Effect.fn("ServerUsageTest.makeHarness")(function* ( beforeRead: (request: number) => Effect.Effect<void> = () => Effect.void,) { const events = yield* Queue.unbounded<ServerConfigStreamEvent>(); let settings = DEFAULT_SERVER_SETTINGS; let requests = 0; const client = { [WS_METHODS.subscribeServerConfig]: () => Stream.concat( Stream.make({ version: 1 as const, type: "snapshot" as const, config: CONFIG }), Stream.fromQueue(events), ), [WS_METHODS.serverGetUsageSummary]: (input: UsageSummaryInput) => Effect.gen(function* () { requests += 1; const price = settings.usagePriceOverrides["custom-model"]?.inputCostPerMillionTokens ?? 0; yield* beforeRead(requests); return { contractVersion: USAGE_CONTRACT_VERSION, readAt: "2026-09-04T12:00:00Z", ...input, buckets: [ { day: input.sinceDay, provider: "codex", model: "custom-model", totals: { uncachedInputTokens: 1_000_000, cachedInputTokens: 0, cacheCreationTokens: 0, outputTokens: 0, reasoningTokens: 0, }, costUsd: price, cacheSavingsUsd: 0, costSource: price === 0 ? "unpriced" : "modelPriced", records: 1, unpricedRecords: price === 0 ? 1 : 0, sessions: 1, }, ], sources: [], pricing: { status: "unavailable", source: "test", fetchedAt: null, knownModels: 0 }, scanDurationMs: 0, } satisfies UsageSummary; }), } as unknown as WsRpcProtocolClient; const session: RpcSession = { client, initialConfig: Effect.succeed(CONFIG), subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, closed: Effect.never, }; const supervisor = EnvironmentSupervisor.of({ target: TARGET, state: yield* SubscriptionRef.make<SupervisorConnectionState>({ ...AVAILABLE_CONNECTION_STATE, phase: "connected", }), session: yield* SubscriptionRef.make(Option.some(session)), prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()), connect: Effect.void, disconnect: Effect.void, retryNow: Effect.void, }); const environments = EnvironmentRegistry.of({ run: (_environmentId, effect) => Effect.provideService(effect, EnvironmentSupervisor, supervisor), followStream: (_environmentId, stream) => Stream.provideService(stream, EnvironmentSupervisor, supervisor), } as EnvironmentRegistry["Service"]); const cache = EnvironmentCacheStore.of({ loadShell: () => Effect.succeed(Option.none()), saveShell: () => Effect.void, loadThread: () => Effect.succeed(Option.none()), saveThread: () => Effect.void, removeThread: () => Effect.void, loadServerConfig: () => Effect.succeed(Option.none()), saveServerConfig: () => Effect.void, loadVcsRefs: () => Effect.succeed(Option.none()), saveVcsRefs: () => Effect.void, removeVcsRefs: () => Effect.void, clearVcsRefs: () => Effect.void, clear: () => Effect.void, }); const runtime = Atom.runtime( Layer.merge( Layer.succeed(EnvironmentRegistry, environments), Layer.succeed(EnvironmentCacheStore, cache), ), ); const initialConfigValueAtom = Atom.make(CONFIG); const atoms = createServerEnvironmentAtoms(runtime, { initialConfigValueAtom: () => initialConfigValueAtom, }); const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) => Effect.sync(() => registry.dispose()), ); const updateSettings = Effect.fn("ServerUsageTest.updateSettings")(function* ( next: ServerSettings, ) { settings = next; yield* Queue.offer(events, { version: 1, type: "settingsUpdated", payload: { settings }, }); yield* AtomRegistry.toStream(registry, atoms.settingsValueAtom(TARGET.environmentId)).pipe( Stream.filter((current) => current === next), Stream.runHead, ); }); return { registry, requests: () => requests, updateSettings, summary: (input = INPUT) => atoms.usageSummary({ environmentId: TARGET.environmentId, input }), };}); function waitForCost<E>( registry: AtomRegistry.AtomRegistry, atom: Atom.Atom<AsyncResult.AsyncResult<UsageSummary, E>>, cost: number,) { return AtomRegistry.toStream(registry, atom).pipe( Stream.filter( (result) => AsyncResult.isSuccess(result) && !result.waiting && result.value.buckets[0]?.costUsd === cost, ), Stream.runHead, );} it.effect("refreshes cached usage windows only when override prices change", () => Effect.scoped( Effect.gen(function* () { const harness = yield* makeHarness(); const current = harness.summary(); const previous = harness.summary({ ...INPUT, sinceDay: UsageDay.make("2026-08-01") }); const unmountCurrent = harness.registry.mount(current); const unmountPrevious = harness.registry.mount(previous); yield* waitForCost(harness.registry, current, 0); yield* waitForCost(harness.registry, previous, 0); unmountPrevious(); expect(harness.requests()).toBe(2); const prices = { "custom-model": { inputCostPerMillionTokens: 3, outputCostPerMillionTokens: 9 }, "other-model": { inputCostPerMillionTokens: 1, outputCostPerMillionTokens: 2 }, }; yield* harness.updateSettings({ ...DEFAULT_SERVER_SETTINGS, usagePriceOverrides: prices }); yield* waitForCost(harness.registry, current, 3); yield* waitForCost(harness.registry, previous, 3); expect(harness.requests()).toBe(4); yield* harness.updateSettings({ ...DEFAULT_SERVER_SETTINGS, usagePriceOverrides: { "other-model": { outputCostPerMillionTokens: 2, inputCostPerMillionTokens: 1 }, "custom-model": { outputCostPerMillionTokens: 9, inputCostPerMillionTokens: 3 }, }, }); yield* harness.updateSettings({ ...DEFAULT_SERVER_SETTINGS, usagePriceOverrides: prices, defaultTheme: "custom-theme", }); yield* harness.updateSettings(DEFAULT_SERVER_SETTINGS); yield* waitForCost(harness.registry, current, 0); yield* waitForCost(harness.registry, previous, 0); expect(harness.requests()).toBe(6); unmountCurrent(); }), ),); it.effect("restarts a pending usage read after a price change", () => Effect.scoped( Effect.gen(function* () { const started = yield* Deferred.make<void>(); const interrupted = yield* Deferred.make<void>(); const harness = yield* makeHarness((request) => request === 1 ? Deferred.succeed(started, undefined).pipe( Effect.andThen(Effect.never), Effect.onInterrupt(() => Deferred.succeed(interrupted, undefined)), ) : Effect.void, ); const summary = harness.summary(); const unmount = harness.registry.mount(summary); yield* Deferred.await(started); yield* harness.updateSettings({ ...DEFAULT_SERVER_SETTINGS, usagePriceOverrides: { "custom-model": { inputCostPerMillionTokens: 5, outputCostPerMillionTokens: 10 }, }, }); yield* waitForCost(harness.registry, summary, 5); yield* Deferred.await(interrupted); expect(harness.requests()).toBe(2); unmount(); }), ),);
import {
DEFAULT_SERVER_SETTINGS,
EnvironmentId,
UsageDay,
USAGE_CONTRACT_VERSION,
WS_METHODS,
type ServerConfig,
type ServerConfigStreamEvent,
type ServerSettings,
type UsageSummary,
type UsageSummaryInput,
} from "@t3tools/contracts";
import { expect, it } from "@effect/vitest";
import * as Deferred from "effect/Deferred";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Queue from "effect/Queue";
import * as Stream from "effect/Stream";
import * as SubscriptionRef from "effect/SubscriptionRef";
import { AsyncResult, Atom, AtomRegistry } from "effect/unstable/reactivity";
import {
AVAILABLE_CONNECTION_STATE,
PrimaryConnectionTarget,
type PreparedConnection,
type SupervisorConnectionState,
} from "../connection/model.ts";
import { EnvironmentRegistry } from "../connection/registry.ts";
import { EnvironmentSupervisor } from "../connection/supervisor.ts";
import { EnvironmentCacheStore } from "../platform/persistence.ts";
import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
import type { RpcSession } from "../rpc/session.ts";
import { createServerEnvironmentAtoms } from "./server.ts";
const TARGET = new PrimaryConnectionTarget({
environmentId: EnvironmentId.make("usage-environment"),
label: "Usage environment",
httpBaseUrl: "https://environment.example.test",
wsBaseUrl: "wss://environment.example.test",
});
const INPUT: UsageSummaryInput = {
sinceDay: UsageDay.make("2026-09-01"),
untilDay: UsageDay.make("2026-09-04"),
timeZone: "UTC",
};
const CONFIG = {
settings: DEFAULT_SERVER_SETTINGS,
environment: { serverVersion: "0.0.1", capabilities: {} },
} as ServerConfig;
const makeHarness = Effect.fn("ServerUsageTest.makeHarness")(function* (
beforeRead: (request: number) => Effect.Effect<void> = () => Effect.void,
) {
const events = yield* Queue.unbounded<ServerConfigStreamEvent>();
let settings = DEFAULT_SERVER_SETTINGS;
let requests = 0;
const client = {
[WS_METHODS.subscribeServerConfig]: () =>
Stream.concat(
Stream.make({ version: 1 as const, type: "snapshot" as const, config: CONFIG }),
Stream.fromQueue(events),
),
[WS_METHODS.serverGetUsageSummary]: (input: UsageSummaryInput) =>
Effect.gen(function* () {
requests += 1;
const price = settings.usagePriceOverrides["custom-model"]?.inputCostPerMillionTokens ?? 0;
yield* beforeRead(requests);
return {
contractVersion: USAGE_CONTRACT_VERSION,
readAt: "2026-09-04T12:00:00Z",
...input,
buckets: [
{
day: input.sinceDay,
provider: "codex",
model: "custom-model",
totals: {
uncachedInputTokens: 1_000_000,
cachedInputTokens: 0,
cacheCreationTokens: 0,
outputTokens: 0,
reasoningTokens: 0,
},
costUsd: price,
cacheSavingsUsd: 0,
costSource: price === 0 ? "unpriced" : "modelPriced",
records: 1,
unpricedRecords: price === 0 ? 1 : 0,
sessions: 1,
},
],
sources: [],
pricing: { status: "unavailable", source: "test", fetchedAt: null, knownModels: 0 },
scanDurationMs: 0,
} satisfies UsageSummary;
}),
} as unknown as WsRpcProtocolClient;
const session: RpcSession = {
client,
initialConfig: Effect.succeed(CONFIG),
subscribeServerConfig: (input) => client.subscribeServerConfig(input),
ready: Effect.void,
probe: Effect.void,
closed: Effect.never,
};
const supervisor = EnvironmentSupervisor.of({
target: TARGET,
state: yield* SubscriptionRef.make<SupervisorConnectionState>({
...AVAILABLE_CONNECTION_STATE,
phase: "connected",
}),
session: yield* SubscriptionRef.make(Option.some(session)),
prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
connect: Effect.void,
disconnect: Effect.void,
retryNow: Effect.void,
});
const environments = EnvironmentRegistry.of({
run: (_environmentId, effect) =>
Effect.provideService(effect, EnvironmentSupervisor, supervisor),
followStream: (_environmentId, stream) =>
Stream.provideService(stream, EnvironmentSupervisor, supervisor),
} as EnvironmentRegistry["Service"]);
const cache = EnvironmentCacheStore.of({
loadShell: () => Effect.succeed(Option.none()),
saveShell: () => Effect.void,
loadThread: () => Effect.succeed(Option.none()),
saveThread: () => Effect.void,
removeThread: () => Effect.void,
loadServerConfig: () => Effect.succeed(Option.none()),
saveServerConfig: () => Effect.void,
loadVcsRefs: () => Effect.succeed(Option.none()),
saveVcsRefs: () => Effect.void,
removeVcsRefs: () => Effect.void,
clearVcsRefs: () => Effect.void,
clear: () => Effect.void,
});
const runtime = Atom.runtime(
Layer.merge(
Layer.succeed(EnvironmentRegistry, environments),
Layer.succeed(EnvironmentCacheStore, cache),
),
);
const initialConfigValueAtom = Atom.make(CONFIG);
const atoms = createServerEnvironmentAtoms(runtime, {
initialConfigValueAtom: () => initialConfigValueAtom,
});
const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
Effect.sync(() => registry.dispose()),
);
const updateSettings = Effect.fn("ServerUsageTest.updateSettings")(function* (
next: ServerSettings,
) {
settings = next;
yield* Queue.offer(events, {
version: 1,
type: "settingsUpdated",
payload: { settings },
});
yield* AtomRegistry.toStream(registry, atoms.settingsValueAtom(TARGET.environmentId)).pipe(
Stream.filter((current) => current === next),
Stream.runHead,
);
});
return {
registry,
requests: () => requests,
updateSettings,
summary: (input = INPUT) => atoms.usageSummary({ environmentId: TARGET.environmentId, input }),
};
});
function waitForCost<E>(
registry: AtomRegistry.AtomRegistry,
atom: Atom.Atom<AsyncResult.AsyncResult<UsageSummary, E>>,
cost: number,
) {
return AtomRegistry.toStream(registry, atom).pipe(
Stream.filter(
(result) =>
AsyncResult.isSuccess(result) &&
!result.waiting &&
result.value.buckets[0]?.costUsd === cost,
),
Stream.runHead,
);
}
it.effect("refreshes cached usage windows only when override prices change", () =>
Effect.scoped(
Effect.gen(function* () {
const harness = yield* makeHarness();
const current = harness.summary();
const previous = harness.summary({ ...INPUT, sinceDay: UsageDay.make("2026-08-01") });
const unmountCurrent = harness.registry.mount(current);
const unmountPrevious = harness.registry.mount(previous);
yield* waitForCost(harness.registry, current, 0);
yield* waitForCost(harness.registry, previous, 0);
unmountPrevious();
expect(harness.requests()).toBe(2);
const prices = {
"custom-model": { inputCostPerMillionTokens: 3, outputCostPerMillionTokens: 9 },
"other-model": { inputCostPerMillionTokens: 1, outputCostPerMillionTokens: 2 },
};
yield* harness.updateSettings({ ...DEFAULT_SERVER_SETTINGS, usagePriceOverrides: prices });
yield* waitForCost(harness.registry, current, 3);
yield* waitForCost(harness.registry, previous, 3);
expect(harness.requests()).toBe(4);
yield* harness.updateSettings({
...DEFAULT_SERVER_SETTINGS,
usagePriceOverrides: {
"other-model": { outputCostPerMillionTokens: 2, inputCostPerMillionTokens: 1 },
"custom-model": { outputCostPerMillionTokens: 9, inputCostPerMillionTokens: 3 },
},
});
yield* harness.updateSettings({
...DEFAULT_SERVER_SETTINGS,
usagePriceOverrides: prices,
defaultTheme: "custom-theme",
});
yield* harness.updateSettings(DEFAULT_SERVER_SETTINGS);
yield* waitForCost(harness.registry, current, 0);
yield* waitForCost(harness.registry, previous, 0);
expect(harness.requests()).toBe(6);
unmountCurrent();
}),
),
);
it.effect("restarts a pending usage read after a price change", () =>
Effect.scoped(
Effect.gen(function* () {
const started = yield* Deferred.make<void>();
const interrupted = yield* Deferred.make<void>();
const harness = yield* makeHarness((request) =>
request === 1
? Deferred.succeed(started, undefined).pipe(
Effect.andThen(Effect.never),
Effect.onInterrupt(() => Deferred.succeed(interrupted, undefined)),
)
: Effect.void,
);
const summary = harness.summary();
const unmount = harness.registry.mount(summary);
yield* Deferred.await(started);
yield* harness.updateSettings({
...DEFAULT_SERVER_SETTINGS,
usagePriceOverrides: {
"custom-model": { inputCostPerMillionTokens: 5, outputCostPerMillionTokens: 10 },
},
});
yield* waitForCost(harness.registry, summary, 5);
yield* Deferred.await(interrupted);
expect(harness.requests()).toBe(2);
unmount();
}),
),
);