Skip to content
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions apps/mobile/src/features/usage/usageProviders.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ export const PROVIDER_ORDER: readonly UsageProviderKind[] = [
"opencode",
"antigravity",
"pi",
"muse",
];

/** The usage series a provider driver reports under, or null for drivers without one. */
Expand All @@ -25,6 +26,7 @@ export const PROVIDER_LABEL: Record<UsageProviderKind, string> = {
opencode: "OpenCode",
antigravity: "Antigravity",
pi: "Pi",
muse: "Muse Code",
};

export function useProviderColors(): Record<UsageProviderKind, string> {
Expand All @@ -37,6 +39,7 @@ export function useProviderColors(): Record<UsageProviderKind, string> {
opencode: "#5b9bbd",
antigravity: "#8c7bd1",
pi: "#4d9f8a",
muse: "#0866ff",
};
}

Expand Down
4 changes: 3 additions & 1 deletion apps/server/src/provider/builtInDrivers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,8 @@ export type BuiltInUsageReadersEnv =
| ProviderUsageReaderEnv<typeof OpenCodeDriver>
| ProviderUsageReaderEnv<typeof AntigravityDriver>
| ProviderUsageReaderEnv<typeof CursorDriver>
| ProviderUsageReaderEnv<typeof PiDriver>;
| ProviderUsageReaderEnv<typeof PiDriver>
| ProviderUsageReaderEnv<typeof MuseDriver>;

/**
* Aggregation keeps the first copy of a duplicate record, so the order is part
Expand All @@ -59,6 +60,7 @@ export const BUILT_IN_USAGE_DRIVERS: ReadonlyArray<
GrokDriver,
PiDriver,
OpenCodeDriver,
MuseDriver,
AntigravityDriver,
CursorDriver,
];
2 changes: 2 additions & 0 deletions apps/server/src/server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -168,6 +168,7 @@ import * as ResourceMonitorBinary from "./resourceTelemetry/ResourceMonitorBinar
import * as ProcessGroupMetrics from "./resourceTelemetry/ProcessGroupMetrics.ts";
import * as ResourceTelemetry from "./resourceTelemetry/ResourceTelemetry.ts";
import * as AntigravityUsage from "./provider/Drivers/AntigravityUsage.ts";
import { MuseUsageHome } from "@t3tools/provider-muse/server";
import * as CursorAccountReader from "@t3tools/provider-cursor/server/CursorAccountReader";
import * as CursorKeychain from "@t3tools/provider-cursor/server/CursorKeychain";
import * as CursorUsageAccounts from "@t3tools/provider-cursor/server/CursorUsageAccounts";
Expand Down Expand Up @@ -269,6 +270,7 @@ const layerUsage = UsageService.layer.pipe(
Layer.provide(
Layer.mergeAll(
AntigravityUsage.layer,
MuseUsageHome.layer,
CodexUsageHistory.layer,
CursorUsageAccounts.layer.pipe(
Layer.provide(CursorAccountReader.layer.pipe(Layer.provide(CursorKeychain.layer))),
Expand Down
232 changes: 232 additions & 0 deletions apps/server/src/usage/UsageService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -48,9 +48,19 @@ import * as AntigravityUsage from "../provider/Drivers/AntigravityUsage.ts";
import * as ProviderHostLive from "../provider/ProviderHostLive.ts";
import { totalTokens, type UsageRecord } from "@t3tools/provider-core/server/usage";
import * as UsageService from "./UsageService.ts";
import * as MuseSdk from "../../../../packages/provider-muse/src/server/sdk.ts";
import { MuseUsageHome } from "@t3tools/provider-muse/server";
import { encodeScanCache, SCAN_CACHE_FILE_NAME } from "./usageScanCache.ts";

vi.mock("node:fs/promises", { spy: true });
vi.mock("../../../../packages/provider-muse/src/server/sdk.ts", async (importOriginal) => ({
...(await importOriginal<typeof MuseSdk>()),
createMuseSdkHostEffect: vi.fn(() =>
Effect.try(() => {
throw new Error("Muse is not installed in this test");
}),
),
}));

const encodeUnknownJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown));
const encodeUnknownJsonString = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
Expand Down Expand Up @@ -137,6 +147,7 @@ const layerService = (input: {
CursorAccountReader.layer.pipe(Layer.provide(CursorKeychain.layer)),
).pipe(
Layer.provideMerge(AntigravityUsage.layer),
Layer.provideMerge(MuseUsageHome.layer),
Layer.provideMerge(ProviderHostLive.layer),
Layer.provideMerge(CodexUsageHistory.layer.pipe(Layer.provideMerge(Sqlite.layerMemory))),
Layer.provideMerge(Layer.mock(BackgroundPolicy.BackgroundPolicy)({})),
Expand Down Expand Up @@ -797,6 +808,227 @@ describe("UsageService", () => {
}).pipe(Effect.scoped),
);

it.live(
"reads Muse inference history once, persists appended usage, and reprices unknown models",
() =>
Effect.gen(function* () {
const { home, settings } = yield* setup;
const museHome = NodePath.join(home, "configured-data", "muse");
const dir = NodePath.join(museHome, "sessions", "2026", "08", "01", "parent");
const child = NodePath.join(dir, "subagent", "child");
const entry = (
id: string,
sessionId = "parent",
timestamp = "2026-08-01T00:00:00Z",
model = "muse-unknown",
) => ({
schema_version: 1,
record_type: "event",
payload_type: "runtime.session",
payload_schema_version: 1,
id,
recorded_at: Date.parse(timestamp) * 1000 + 999,
stream: { kind: "session", id: sessionId },
payload: {
source_run_record_id: id,
event: {
kind: "model_completed",
model,
usage: {
input_tokens: 100,
cached_tokens: 60,
cache_read_tokens: 60,
cache_write_tokens: 10,
output_tokens: 20,
reasoning_tokens: 8,
},
},
},
});
const parent = entry("inference-1");
const review = {
...entry("review", "child"),
payload_schema_version: 3,
payload: {
event: {
...parent.payload.event,
kind: "automated_review_completed",
model: { model_id: "priced-model" },
usage: {
input_tokens: 100,
cached_input_tokens: 60,
output_tokens: 20,
reasoning_tokens: 8,
},
},
},
};
yield* Effect.promise(async () => {
await NodeFSP.mkdir(child, { recursive: true });
await NodeFSP.writeFile(
NodePath.join(dir, "session.jsonl"),
[
parent,
parent,
{
...parent,
payload: { event: { ...parent.payload.event, kind: "workflow_completed" } },
},
{ ...parent, payload_type: "goal_usage_attribution" },
entry("before", "parent", "2026-07-31T23:59:59.999Z"),
entry("after", "parent", "2026-08-02T00:00:00Z"),
]
.map((line) => encodeUnknownJsonString(line))
.join("\n") + "\n{broken}\n",
);
await NodeFSP.writeFile(
NodePath.join(child, "session.jsonl"),
[
{ ...parent, id: "copied-envelope", stream: { kind: "session", id: "child" } },
entry("child-inference", "child"),
review,
]
.map((line) => encodeUnknownJsonString(line))
.join("\n") + "\n",
);
await NodeFSP.writeFile(
NodePath.join(child, "telemetry.jsonl"),
encodeUnknownJsonString(entry("not-a-session")) + "\n",
);
});
using _create = vi.spyOn(MuseSdk, "createMuseSdkHostEffect").mockImplementation((options) =>
Effect.succeed({
initializeResult: {
grantedCapabilities: [],
museHome: NodePath.join(options.environment?.XDG_DATA_HOME ?? home, "muse"),
},
close: async () => {},
} as unknown as MuseSdk.MuseSdkHost),
);
yield* Effect.gen(function* () {
const service = yield* UsageService.make;
const window = {
timeZone: "UTC",
sinceDay: UsageDay.make("2026-08-01"),
untilDay: UsageDay.make("2026-08-01"),
};
const muse = (summary: UsageSummary) =>
summary.buckets.filter((bucket) => bucket.provider === "muse");
const first = yield* service.readSummary(window);
const buckets = muse(first);
assert.strictEqual(
buckets.reduce((sum, b) => sum + b.records, 0),
3,
);
assert.strictEqual(
buckets.reduce((sum, b) => sum + totalTokens(b.totals), 0),
360,
);
const unknown = buckets.find((b) => b.model === "muse-unknown");
assert.strictEqual(unknown?.costSource, "unpriced");
assert.strictEqual(unknown?.unpricedRecords, 2);
assert.deepStrictEqual(unknown?.totals, {
uncachedInputTokens: 60,
cachedInputTokens: 120,
cacheCreationTokens: 20,
outputTokens: 40,
reasoningTokens: 16,
});
const priced = buckets.find((b) => b.model === "priced-model");
assert.strictEqual(priced?.costSource, "modelPriced");
assert.closeTo(priced?.costUsd ?? NaN, 0.000172, 1e-12);
assert.closeTo(priced?.cacheSavingsUsd ?? NaN, 0.000108, 1e-12);
const sources = first.sources.filter((source) => source.fingerprint.provider === "muse");
assert.strictEqual(sources.length, 1);
assert.strictEqual(sources[0]?.scannedFiles, 2);
assert.strictEqual(sources[0]?.distinctSessions, 2);
const local = muse(
yield* service.readSummary({ ...window, timeZone: "America/Los_Angeles" }),
);
assert.strictEqual(
local.reduce((sum, b) => sum + b.records, 0),
1,
);
assert.strictEqual(local[0]?.day, "2026-08-01");
const hourly = muse(
yield* service.readSummary({
...window,
resolution: "hour",
sinceTime: "2026-08-01T00:00:00.000Z",
untilTime: "2026-08-02T00:00:00.000Z",
}),
);
assert.strictEqual(
hourly.reduce((sum, b) => sum + b.records, 0),
3,
);
const appended = encodeUnknownJsonString(entry("appended", "child"));
const file = NodePath.join(child, "session.jsonl");
yield* Effect.promise(() => NodeFSP.appendFile(file, appended.slice(0, -10)));
assert.deepStrictEqual(muse(yield* service.readSummary(window)), buckets);
yield* service.awaitPersisted;
const restarted = yield* UsageService.make;
yield* Effect.promise(() => NodeFSP.appendFile(file, appended.slice(-10) + "\n"));
const grown = muse(yield* restarted.readSummary(window));
assert.strictEqual(
grown.reduce((sum, b) => sum + b.records, 0),
4,
);
assert.strictEqual(
grown.reduce((sum, b) => sum + totalTokens(b.totals), 0),
480,
);
const settingsService = yield* ServerSettings.ServerSettingsService;
yield* settingsService.updateSettings({
usagePriceOverrides: {
"muse-unknown": {
inputCostPerMillionTokens: 2,
outputCostPerMillionTokens: 4,
cacheReadCostPerMillionTokens: 0.2,
cacheWriteCostPerMillionTokens: 1,
},
},
});
const repriced = muse(yield* restarted.readSummary(window)).find(
(b) => b.model === "muse-unknown",
);
assert.strictEqual(repriced?.unpricedRecords, 0);
assert.closeTo(repriced?.costUsd ?? NaN, 0.000486, 1e-12);
}).pipe(
Effect.scoped,
Effect.provide(
layerService({
prefix: "usage-muse",
home,
environment: { XDG_DATA_HOME: NodePath.dirname(museHome) },
settings: {
...settings,
providerInstances: {
...settings.providerInstances,
[ProviderInstanceId.make("muse")]: {
driver: ProviderDriverKind.make("muse"),
enabled: false,
},
[ProviderInstanceId.make("muse-alias")]: {
driver: ProviderDriverKind.make("muse"),
enabled: false,
},
},
},
ratesDocument: {
"priced-model": {
input_cost_per_token: 0.000002,
output_cost_per_token: 0.000004,
cache_read_input_token_cost: 0.0000002,
cache_creation_input_token_cost: 0.0000002,
},
},
}),
),
);
}).pipe(Effect.scoped),
);

it.live("reports missing Pi history and skips malformed records without losing valid usage", () =>
Effect.gen(function* () {
const { home, settings } = yield* setup;
Expand Down
4 changes: 3 additions & 1 deletion apps/server/src/usage/usageTestFormats.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,18 +2,20 @@ import type { UsageProviderKind } from "@t3tools/contracts";
import type { TranscriptUsageFormat } from "@t3tools/provider-core/server/usage";
import { piUsageFormat } from "@t3tools/provider-pi/server/usage";
import { grokUsageFormat } from "@t3tools/provider-grok/server/usage";
import { museUsageFormat } from "@t3tools/provider-muse/server/usage";

import { claudeUsageFormat } from "../provider/Drivers/claudeUsage.ts";
import { codexUsageFormat } from "../provider/Drivers/codexUsage.ts";

export const TEST_FORMATS: Record<
"claude" | "codex" | "grok" | "pi",
"claude" | "codex" | "grok" | "pi" | "muse",
TranscriptUsageFormat<unknown>
> = {
claude: claudeUsageFormat,
codex: codexUsageFormat,
grok: grokUsageFormat,
pi: piUsageFormat,
muse: museUsageFormat,
};

export const TEST_FORMAT_MAP = new Map<UsageProviderKind, TranscriptUsageFormat<unknown>>(
Expand Down
47 changes: 47 additions & 0 deletions apps/server/src/usage/usageTranscriptStreaming.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,53 @@ async function scan(
}

describe("large usage records", () => {
it("streams Muse completion envelopes and resumes a partial appended event", async () => {
const entry = {
schema_version: 1,
record_type: "event",
payload_type: "runtime.session",
payload_schema_version: 3,
id: "envelope-1",
recorded_at: 1_785_578_400_123_456,
stream: { kind: "session", id: "parent" },
payload: {
source_run_record_id: "inference-1",
event: {
kind: "automated_review_completed",
model: { model_id: "review-model" },
usage: {
input_tokens: 100,
cached_input_tokens: 60,
output_tokens: 20,
reasoning_tokens: 8,
},
},
},
};
const small = await scan([entry], "muse", "small-muse");
expect(small.records[0]?.totals).toMatchObject({
uncachedInputTokens: 40,
cachedInputTokens: 60,
});
const path = NodePath.join(dir, "large-muse.jsonl");
await NodeFSP.writeFile(path, JSON.stringify({ ...entry, discarded: { content } }) + "\n");
const large = await readTranscriptRecords(path, "muse");
expect(large?.records).toEqual(small.records);
const appended = JSON.stringify({
...entry,
id: "envelope-2",
payload: { ...entry.payload, source_run_record_id: "inference-2" },
});
await NodeFSP.appendFile(path, appended.slice(0, -7));
const partial = await readTranscriptRecords(path, "muse", large!.position);
expect(partial?.records).toEqual([]);
await NodeFSP.appendFile(path, appended.slice(-7) + "\n");
const completed = await readTranscriptRecords(path, "muse", partial!.position);
expect(completed?.records).toHaveLength(1);
expect(completed?.records[0]?.dedupeKey).not.toBe(large?.records[0]?.dedupeKey);
expect(completed?.records[0]?.totals).toEqual(small.records[0]?.totals);
});

it("keeps Pi model, cost, and fork identity after large assistant content", async () => {
const entry = {
type: "message",
Expand Down
Loading
Loading