diff --git a/apps/mobile/src/features/threads/ComposerUsageLimits.tsx b/apps/mobile/src/features/threads/ComposerUsageLimits.tsx index 31d3a0e3f277..f3ca830c916b 100644 --- a/apps/mobile/src/features/threads/ComposerUsageLimits.tsx +++ b/apps/mobile/src/features/threads/ComposerUsageLimits.tsx @@ -5,7 +5,11 @@ import { SymbolView } from "../../components/AppSymbol"; import { AppText as Text } from "../../components/AppText"; import { AccountLimits, ResetCredits } from "../usage/UsageLimitsSection"; -const DRIVER_LABEL: Partial> = { codex: "Codex", claudeAgent: "Claude" }; +const DRIVER_LABEL: Partial> = { + codex: "Codex", + claudeAgent: "Claude", + muse: "Muse Code", +}; /** * The /usage-limits result, docked above the composer. It is the Usage → Limits diff --git a/apps/mobile/src/features/usage/UsageLimitsPooled.tsx b/apps/mobile/src/features/usage/UsageLimitsPooled.tsx index be9adb283488..c33eaffefa18 100644 --- a/apps/mobile/src/features/usage/UsageLimitsPooled.tsx +++ b/apps/mobile/src/features/usage/UsageLimitsPooled.tsx @@ -30,6 +30,7 @@ const DRIVER_LABEL: Partial> = { codex: "Codex", claudeAgent: "Claude", antigravity: "Antigravity", + muse: "Muse Code", }; const PACE_LABEL = { ahead: "Ahead of pace", on: "On pace", under: "Under pace" } as const; diff --git a/docs/user/providers-muse.md b/docs/user/providers-muse.md index 9013e97dfa97..dd9c156ce677 100644 --- a/docs/user/providers-muse.md +++ b/docs/user/providers-muse.md @@ -43,6 +43,13 @@ Muse can list models that your account cannot use. If a message fails with "does not exist or you lack access", pick another model, or add one your account can use under **Custom models** in the instance settings. +## Usage limits + +**Usage → Limits** shows your Meta subscription's session and weekly windows. +Muse reports them with each reply, so they appear after Muse's first reply +since the T3 server started. An instance configured with `META_API_KEY` does +not report subscription limits. + ## Permissions and limitations Muse offers two [permission modes](./permission-modes.md): **Supervised** asks diff --git a/docs/user/usage.md b/docs/user/usage.md index ed1dd19ce1a1..6d37644abc4f 100644 --- a/docs/user/usage.md +++ b/docs/user/usage.md @@ -118,6 +118,10 @@ Grok reports the remaining subscription allowance and reset time for its current after signing in with `grok login`. Explicit `XAI_API_KEY` connections and custom authentication or endpoint configurations do not report subscription limits. +Muse Code reports your Meta subscription's session and weekly windows with each reply, not on +request, so they appear after Muse's first reply since the server started. A Muse instance +configured with `META_API_KEY` does not report subscription limits. + API-key accounts may not report subscription limits. This also applies to Claude connections using a proxy through `ANTHROPIC_AUTH_TOKEN`. diff --git a/packages/provider-core/src/server/managedProvider.ts b/packages/provider-core/src/server/managedProvider.ts index bac7fbd3a52a..e565783afbe8 100644 --- a/packages/provider-core/src/server/managedProvider.ts +++ b/packages/provider-core/src/server/managedProvider.ts @@ -207,6 +207,7 @@ export const makeManagedServerProvider = Effect.fn("makeManagedServerProvider")( previous: state.snapshot.usageLimits, update, checkedAt: update.checkedAt, + ...(update.replace ? { replace: true } : {}), }); // `applyUsageLimitsUpdate` hands back the same object when nothing // moved, which is the common case for Codex's per-tick notification. diff --git a/packages/provider-core/src/server/snapshot.ts b/packages/provider-core/src/server/snapshot.ts index 7832e944410f..641e37616c8b 100644 --- a/packages/provider-core/src/server/snapshot.ts +++ b/packages/provider-core/src/server/snapshot.ts @@ -18,9 +18,13 @@ export interface ManagedServerProvider { /** * Fold a runtime rate-limit update into the published snapshot without * waiting for the next status probe. Sparse: windows merge by id and an - * update with no usable window leaves the snapshot untouched. + * update with no usable window leaves the snapshot untouched. With + * `replace`, the update's windows are the whole set instead. */ readonly applyUsageLimits: ( - update: ProviderUsageLimitsUpdate & { readonly checkedAt: string }, + update: ProviderUsageLimitsUpdate & { + readonly checkedAt: string; + readonly replace?: boolean; + }, ) => Effect.Effect; } diff --git a/packages/provider-core/src/server/usageLimits.test.ts b/packages/provider-core/src/server/usageLimits.test.ts index 9b305a74b98e..2a9daac467b3 100644 --- a/packages/provider-core/src/server/usageLimits.test.ts +++ b/packages/provider-core/src/server/usageLimits.test.ts @@ -62,6 +62,45 @@ describe("applyUsageLimitsUpdate", () => { ).toBe(published); }); + it("replaces the windows outright when the update is the account's whole state", () => { + // Muse reports a window that has reset since with no reset time; a sparse + // merge would keep the expired one. + const restarted = { + id: "five_hour", + kind: "session", + label: "Session", + usedPercent: 0, + windowDurationMins: 300, + } as const; + expect( + applyUsageLimitsUpdate({ + previous: published, + checkedAt: "2026-09-03T15:00:00.000Z", + update: { windows: [restarted] }, + replace: true, + }), + ).toEqual({ checkedAt: "2026-09-03T15:00:00.000Z", windows: [restarted] }); + // The same report again publishes nothing new. + expect( + applyUsageLimitsUpdate({ + previous: published, + checkedAt, + update: { windows: [session, weekly] }, + replace: true, + }), + ).toBe(published); + // A later report with the same windows publishes its time, so this account's + // report still counts as the freshest when Limits merges environments. + expect( + applyUsageLimitsUpdate({ + previous: published, + checkedAt: "2026-09-03T12:00:05.000Z", + update: { windows: [session, weekly] }, + replace: true, + }), + ).toEqual({ ...published, checkedAt: "2026-09-03T12:00:05.000Z" }); + }); + it("preserves reset credits when a streamed window update changes usage", () => { const resetCredits = { availableCount: 2, nextExpiresAt: "2026-10-01T00:00:00.000Z" }; const next = applyUsageLimitsUpdate({ diff --git a/packages/provider-core/src/server/usageLimits.ts b/packages/provider-core/src/server/usageLimits.ts index ea8d0d1d029f..41a28ff17b0a 100644 --- a/packages/provider-core/src/server/usageLimits.ts +++ b/packages/provider-core/src/server/usageLimits.ts @@ -61,31 +61,48 @@ export function applyUsageLimitsUpdate(input: { readonly previous: ServerProviderUsageLimits | undefined; readonly update: ProviderUsageLimitsUpdate; readonly checkedAt: string; + /** + * The update is the account's whole state: windows it leaves out are + * dropped, and a reset time it omits is cleared rather than kept. + */ + readonly replace?: boolean; }): ServerProviderUsageLimits | undefined { const { previous, update } = input; + const replace = input.replace === true; if (update.windows.length === 0 || previous?.unavailable?.reason === "unsupported") { return previous; } - const merged = new Map(previous?.windows.map((window) => [window.id, window] as const)); + const published = new Map(previous?.windows.map((window) => [window.id, window] as const)); + const merged = replace + ? new Map() + : new Map(published); // Codex sends this notification beside every token-usage tick, almost // always with unchanged numbers. Decide "nothing changed" per window on - // the way through so the no-op case never allocates a new snapshot. - let changed = false; + // the way through so the no-op case never allocates a new snapshot. A + // whole-state report that only moves the time is still news: Limits picks + // the freshest report for an account across environments by `checkedAt`. + let changed = + replace && + (update.windows.length !== published.size || input.checkedAt !== previous?.checkedAt); for (const window of update.windows) { - const existing = merged.get(window.id); + const existing = published.get(window.id); const next: ServerProviderUsageWindow = { ...window, usedPercent: clampPercent(window.usedPercent), - ...(window.resetsAt === undefined && existing?.resetsAt !== undefined + ...(!replace && window.resetsAt === undefined && existing?.resetsAt !== undefined ? { resetsAt: existing.resetsAt } : {}), - ...(window.windowDurationMins === undefined && existing?.windowDurationMins !== undefined + ...(!replace && + window.windowDurationMins === undefined && + existing?.windowDurationMins !== undefined ? { windowDurationMins: existing.windowDurationMins } : {}), }; if (existing === undefined || !usageWindowEquals(existing, next)) { merged.set(window.id, next); changed = true; + } else if (replace) { + merged.set(window.id, existing); } } if (!changed && previous !== undefined && previous.unavailable === undefined) { diff --git a/packages/provider-muse/src/server/adapter.test.ts b/packages/provider-muse/src/server/adapter.test.ts index c09ac92efa50..63ad6974a502 100644 --- a/packages/provider-muse/src/server/adapter.test.ts +++ b/packages/provider-muse/src/server/adapter.test.ts @@ -33,7 +33,7 @@ import * as Stream from "effect/Stream"; import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; -import type { MuseItem } from "./protocol.ts"; +import type { MuseItem, MuseSubscriptionUsage } from "./protocol.ts"; import type { MuseSdkHost } from "./sdk.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; @@ -191,7 +191,11 @@ const makeHarness = Effect.fnUntraced(function* ( policy = runtimePolicy, overrides: Pick< MuseAdapterV2Options, - "createHost" | "nativeEventLogger" | "modelCatalog" | "continuationRequests" + | "createHost" + | "nativeEventLogger" + | "modelCatalog" + | "continuationRequests" + | "onSubscriptionUsage" > = {}, ) { let hostCount = 0; @@ -548,6 +552,39 @@ describe("MuseAdapterV2", () => { }).pipe(Effect.scoped, Effect.provide(testLayer)), ); + it.effect("forwards subscription usage reports and skips unreadable ones", () => + Effect.gen(function* () { + const fake = yield* makeFakeMuse(); + const reported = yield* Deferred.make(); + const harness = yield* makeHarness( + fake, + INSTANCE_ID, + undefined, + undefined, + undefined, + runtimePolicy, + { + onSubscriptionUsage: (usage) => Deferred.succeed(reported, usage).pipe(Effect.asVoid), + }, + ); + const { nativeId } = yield* startConversation(harness, fake); + yield* fake.emit("usage/changed", { tier: 1 }); + yield* fake.emit("usage/changed", { + observedAtMs: 1_000, + tier: "meta-subscription", + window: { usedPercent: 3, resetsAtMs: 2_000, windowDurationMins: 300 }, + weekly: { usedPercent: 46, resetsAtMs: 3_000 }, + }); + const usage = yield* Deferred.await(reported); + assert.strictEqual(usage.window.usedPercent, 3); + assert.strictEqual(usage.weekly.usedPercent, 46); + yield* fake.emit("turn/completed", { turnId: nativeId, terminal: "completed" }); + const terminal = yield* harness.takeEvent("turn.terminal"); + assert.strictEqual(terminal.status, "completed"); + assert.strictEqual(fake.closeCount(), 0); + }).pipe(Effect.scoped, Effect.provide(testLayer)), + ); + it.effect("moves a resumed session on another model back to the catalog default", () => Effect.gen(function* () { const fake = yield* makeFakeMuse(); diff --git a/packages/provider-muse/src/server/adapter.ts b/packages/provider-muse/src/server/adapter.ts index 2694b5990784..abde3ac2369b 100644 --- a/packages/provider-muse/src/server/adapter.ts +++ b/packages/provider-muse/src/server/adapter.ts @@ -50,6 +50,7 @@ import { MuseDelta, MuseItemEvent, MuseSessionResult, + MuseSubscriptionUsage, MuseTokenUsageEvent, MuseTodoList, MuseTurnRetryScheduled, @@ -195,6 +196,8 @@ export interface MuseAdapterV2Options { readonly createHost?: typeof createMuseSdkHost; readonly requestTimeoutMs?: number; readonly nativeEventLogger?: ProviderEventLoggers.EventNdjsonLogger; + /** Each subscription-usage report Muse forwards (`usage/changed`); the driver publishes it. */ + readonly onSubscriptionUsage?: (usage: MuseSubscriptionUsage) => Effect.Effect; /** Where a turn Muse starts on its own (a finished workflow's report) asks for its run. */ readonly continuationRequests?: { readonly offer: ( @@ -261,6 +264,7 @@ const INFORMATIONAL_NOTIFICATIONS = new Set([ "session/tokenUsage", "session/todoListChanged", "turn/retryScheduled", + "usage/changed", ]); // Tools whose results already show as their own rows: T3's todo list and question // rows, and the workflow item a `workflow` call launches. @@ -1284,6 +1288,12 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( } }); const handleNotification = Effect.fnUntraced(function* (method: string, data: unknown) { + // Subscription usage is connection-scoped: its payload names no session. + if (method === "usage/changed") { + const usage = yield* decode(MuseSubscriptionUsage, data); + if (options.onSubscriptionUsage) yield* options.onSubscriptionUsage(usage); + return; + } const params = yield* decode(recordSchema, data); if (params.sessionId !== nativeSessionId) return; if (method === "view/gap") diff --git a/packages/provider-muse/src/server/driver.ts b/packages/provider-muse/src/server/driver.ts index 9a70f74eed83..7b3060e5e295 100644 --- a/packages/provider-muse/src/server/driver.ts +++ b/packages/provider-muse/src/server/driver.ts @@ -1,10 +1,13 @@ import { ProviderDriverKind } from "@t3tools/contracts"; import { MuseSettings } from "../settings.ts"; import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; +import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; import * as HttpClient from "effect/http/HttpClient"; import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; @@ -22,6 +25,14 @@ import * as ProviderEventLoggers from "@t3tools/provider-core/server/ProviderEve import { makeManagedServerProvider } from "@t3tools/provider-core/server/managedProvider"; import { enrichMuseSnapshot, latestMuseVersion, museMaintenance } from "./maintenance.ts"; import { makeMuseEnvironment } from "./sdk.ts"; +import type { MuseSubscriptionUsage } from "./protocol.ts"; +import { + isNewerMuseUsageReport, + museSubscriptionUsageToLimits, + museSubscriptionUsageToUpdate, +} from "./usageLimits.ts"; +import { makeUnavailableUsageLimits } from "@t3tools/provider-core/server/usageLimits"; +import type { ServerProviderDraft } from "@t3tools/provider-core/server/snapshotProbe"; import { defaultProviderContinuationIdentity, type ProviderDriver, @@ -92,6 +103,46 @@ export const MuseDriver: ProviderDriver = { continuationGroupKey: continuationIdentity.continuationKey, }); const snapshotSettings = yield* makeProviderSnapshotSettingsSource(effectiveConfig); + // Meta reports subscription usage only with a reply, and a status check + // makes none. Keep the newest report from any of this instance's + // sessions so a refresh re-derives the windows instead of dropping + // them. Reports and status checks take turns through one permit, so + // comparing, retaining and publishing a report is a single step: an + // older report arriving late can neither overwrite nor publish over a + // newer one. + const retainedUsage = yield* Ref.make(undefined); + const usagePermit = yield* Semaphore.make(1); + // Only the instance's own environment counts: an inherited META_API_KEY + // was already dropped so Muse uses its login. + const usesApiKey = Object.entries(processEnvironment).some( + ([key, value]) => key.toUpperCase() === "META_API_KEY" && (value ?? "").trim() !== "", + ); + const withUsageLimits = (snapshot: ServerProviderDraft) => + usagePermit.withPermits(1)( + Effect.gen(function* () { + if (!effectiveConfig.enabled || !snapshot.installed) return snapshot; + const now = yield* DateTime.now; + if (usesApiKey) { + return { + ...snapshot, + usageLimits: makeUnavailableUsageLimits({ + checkedAt: DateTime.formatIso(now), + reason: "unsupported", + message: "Muse uses an API key here, which does not report subscription limits.", + }), + }; + } + const usage = yield* Ref.get(retainedUsage); + const usageLimits = usage + ? museSubscriptionUsageToLimits({ usage, nowMs: DateTime.toEpochMillis(now) }) + : makeUnavailableUsageLimits({ + checkedAt: DateTime.formatIso(now), + reason: "probeFailed", + message: "Muse reports usage limits with its next reply.", + }); + return { ...snapshot, usageLimits }; + }), + ); const resolveInstallation = yield* makeCachedProviderMaintenanceResolution( resolveProviderMaintenanceCapabilitiesEffect(museMaintenance, { binaryPath: effectiveConfig.binaryPath, @@ -126,6 +177,7 @@ export const MuseDriver: ProviderDriver = { initialSnapshot: (settings) => makePendingMuseProvider(settings.provider).pipe(Effect.map(stampIdentity)), checkProvider: checkMuseProviderStatus(effectiveConfig, processEnvironment, cwd).pipe( + Effect.flatMap(withUsageLimits), Effect.map(stampIdentity), Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), Effect.provideService(FileSystem.FileSystem, fileSystem), @@ -163,6 +215,24 @@ export const MuseDriver: ProviderDriver = { environment: processEnvironment, modelCatalog, ...(eventLoggers.native ? { nativeEventLogger: eventLoggers.native } : {}), + onSubscriptionUsage: (usage) => + usagePermit + .withPermits(1)( + Effect.gen(function* () { + // Every session's host reports; only the newest wins. + const newer = yield* Ref.modify(retainedUsage, (previous) => + isNewerMuseUsageReport(previous, usage) + ? ([true, usage] as const) + : ([false, previous] as const), + ); + if (!newer || usesApiKey) return; + const now = yield* DateTime.now; + yield* snapshot.applyUsageLimits( + museSubscriptionUsageToUpdate({ usage, nowMs: DateTime.toEpochMillis(now) }), + ); + }), + ) + .pipe(Effect.ignoreCause({ log: true })), continuationRequests, }); const textGeneration = yield* makeMuseTextGeneration(effectiveConfig, { diff --git a/packages/provider-muse/src/server/protocol.test.ts b/packages/provider-muse/src/server/protocol.test.ts index d583d84de495..36ae08278e96 100644 --- a/packages/provider-muse/src/server/protocol.test.ts +++ b/packages/provider-muse/src/server/protocol.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from "vite-plus/test"; +import * as Schema from "effect/Schema"; -import { museApprovalChoices } from "./protocol.ts"; +import { MuseSubscriptionUsage, museApprovalChoices } from "./protocol.ts"; // Muse 1.4.3 shell approvals offer only `abort`; the muse_permission replay covers that case. describe("Muse approval choices", () => { @@ -15,3 +16,30 @@ describe("Muse approval choices", () => { expect(choices.get("cancel")?.choiceId).toBe("abort"); }); }); + +describe("Muse subscription usage", () => { + const report = { + observedAtMs: 1_000, + tier: "meta-subscription", + window: { usedPercent: 3, resetsAtMs: 2_000, windowDurationMins: 300 }, + weekly: { usedPercent: 46, resetsAtMs: 3_000 }, + }; + const decode = Schema.decodeUnknownResult(MuseSubscriptionUsage); + + it("decodes a report, including over-quota percentages", () => { + expect(decode({ ...report, weekly: { ...report.weekly, usedPercent: 130 } })._tag).toBe( + "Success", + ); + }); + + it("rejects a fractional window duration at the boundary", () => { + expect( + decode({ ...report, window: { ...report.window, windowDurationMins: 300.5 } })._tag, + ).toBe("Failure"); + }); + + it("rejects timestamps a Date cannot hold", () => { + expect(decode({ ...report, observedAtMs: Number.MAX_VALUE })._tag).toBe("Failure"); + expect(decode({ ...report, window: { ...report.window, resetsAtMs: 0 } })._tag).toBe("Failure"); + }); +}); diff --git a/packages/provider-muse/src/server/protocol.ts b/packages/provider-muse/src/server/protocol.ts index e7413a181a04..49b636cfa16c 100644 --- a/packages/provider-muse/src/server/protocol.ts +++ b/packages/provider-muse/src/server/protocol.ts @@ -151,6 +151,32 @@ export const MuseContextUsage = Schema.Struct({ export const MuseTodoList = Schema.Struct({ items: Schema.Array(Schema.Struct({ text: Schema.String, status: Schema.String })), }); +/** The latest instant a JavaScript Date holds; formatting a later one throws. */ +const LATEST_DATE_MS = 8.64e15; +const DateMillis = Schema.Finite.check( + Schema.isGreaterThan(0), + Schema.isLessThanOrEqualTo(LATEST_DATE_MS), +); +/** + * `usage/changed` params (the same payload `usage/read` returns): the Meta + * subscription windows the latest model response reported. Percentages are + * Meta's, verbatim, and may exceed 100. The window duration must be a whole + * number of minutes, as the usage-limits contract requires an integer. + */ +export const MuseSubscriptionUsage = Schema.Struct({ + observedAtMs: DateMillis, + tier: Schema.String, + window: Schema.Struct({ + usedPercent: Schema.Finite.check(Schema.isGreaterThanOrEqualTo(0)), + windowDurationMins: Schema.Int.check(Schema.isGreaterThan(0)), + resetsAtMs: DateMillis, + }), + weekly: Schema.Struct({ + usedPercent: Schema.Finite.check(Schema.isGreaterThanOrEqualTo(0)), + resetsAtMs: DateMillis, + }), +}); +export type MuseSubscriptionUsage = typeof MuseSubscriptionUsage.Type; export const MuseTurnCompleted = Schema.Struct({ turnId: NonEmptyString, terminal: NonEmptyString, diff --git a/packages/provider-muse/src/server/usageLimits.test.ts b/packages/provider-muse/src/server/usageLimits.test.ts new file mode 100644 index 000000000000..a170fd05a872 --- /dev/null +++ b/packages/provider-muse/src/server/usageLimits.test.ts @@ -0,0 +1,107 @@ +import { describe, expect, it } from "@effect/vitest"; + +import { applyUsageLimitsUpdate } from "@t3tools/provider-core/server/usageLimits"; +import type { MuseSubscriptionUsage } from "./protocol.ts"; +import { + isNewerMuseUsageReport, + museSubscriptionUsageToLimits, + museSubscriptionUsageToUpdate, +} from "./usageLimits.ts"; + +const NOW_MS = Date.parse("2026-10-12T12:00:00.000Z"); + +const usage: MuseSubscriptionUsage = { + observedAtMs: NOW_MS, + tier: "meta-subscription", + window: { + usedPercent: 3, + resetsAtMs: Date.parse("2026-10-12T14:19:00.000Z"), + windowDurationMins: 300, + }, + weekly: { + usedPercent: 46, + resetsAtMs: Date.parse("2026-10-19T00:00:00.000Z"), + }, +}; + +describe("Muse subscription usage limits", () => { + it("maps the session and weekly windows, dated when Meta reported them", () => { + expect(museSubscriptionUsageToLimits({ usage, nowMs: NOW_MS })).toEqual({ + checkedAt: "2026-10-12T12:00:00.000Z", + windows: [ + { + id: "session", + kind: "session", + label: "Session", + usedPercent: 3, + windowDurationMins: 300, + resetsAt: "2026-10-12T14:19:00.000Z", + }, + { + id: "weekly", + kind: "weekly", + label: "Weekly", + usedPercent: 46, + windowDurationMins: 10080, + resetsAt: "2026-10-19T00:00:00.000Z", + }, + ], + }); + }); + + it("starts a window over empty once its reset has passed, without touching the other", () => { + const limits = museSubscriptionUsageToLimits({ + usage: { ...usage, window: { ...usage.window, usedPercent: 97, resetsAtMs: NOW_MS - 1 } }, + nowMs: NOW_MS, + }); + expect(limits.windows[0]).toEqual({ + id: "session", + kind: "session", + label: "Session", + usedPercent: 0, + windowDurationMins: 300, + }); + expect(limits.windows[1]?.usedPercent).toBe(46); + }); + + it("clamps over-quota percentages to the contract range", () => { + const update = museSubscriptionUsageToUpdate({ + usage: { ...usage, weekly: { ...usage.weekly, usedPercent: 130 } }, + nowMs: NOW_MS, + }); + expect(update.windows.map((window) => window.usedPercent)).toEqual([3, 100]); + expect(update.replace).toBe(true); + expect(update.checkedAt).toBe("2026-10-12T12:00:00.000Z"); + }); + + it("clears the stale reset time when a whole-state update replaces an expired window", () => { + const published = museSubscriptionUsageToLimits({ usage, nowMs: NOW_MS }); + const next = applyUsageLimitsUpdate({ + previous: published, + checkedAt: "2026-10-12T15:00:00.000Z", + replace: true, + update: museSubscriptionUsageToUpdate({ + usage: { + ...usage, + observedAtMs: NOW_MS + 60_000, + window: { ...usage.window, usedPercent: 97, resetsAtMs: NOW_MS - 1 }, + }, + nowMs: NOW_MS, + }), + }); + expect(next?.windows[0]).toEqual({ + id: "session", + kind: "session", + label: "Session", + usedPercent: 0, + windowDurationMins: 300, + }); + }); + + it("only accepts reports newer than the retained one", () => { + expect(isNewerMuseUsageReport(undefined, usage)).toBe(true); + expect(isNewerMuseUsageReport(usage, { ...usage, observedAtMs: NOW_MS + 1 })).toBe(true); + expect(isNewerMuseUsageReport(usage, { ...usage, observedAtMs: NOW_MS })).toBe(false); + expect(isNewerMuseUsageReport(usage, { ...usage, observedAtMs: NOW_MS - 1 })).toBe(false); + }); +}); diff --git a/packages/provider-muse/src/server/usageLimits.ts b/packages/provider-muse/src/server/usageLimits.ts new file mode 100644 index 000000000000..a8aca648cc93 --- /dev/null +++ b/packages/provider-muse/src/server/usageLimits.ts @@ -0,0 +1,114 @@ +/** + * Muse subscription usage. Meta has no usage endpoint T3 can poll: Muse + * reports the signed-in account's windows with each reply and forwards them + * as the MSP `usage/changed` notification. The adapter hands each report to + * the driver, which publishes it as the instance's `usageLimits` and + * re-derives the windows from the newest report on every status check. + * + * A report is the account's whole state, so live updates replace the + * published windows: a window whose reset has passed starts over empty, + * with no reset time until Meta reports the next one. Limits are dated when + * Meta reported them (`observedAtMs`), so the freshest report wins when + * Limits merges environments. + * + * Which Meta account a report belongs to is not decided here: usage + * attribution per session host is tracked separately in #17397. + */ +import type { + ProviderUsageLimitsUpdate, + ServerProviderUsageLimits, + ServerProviderUsageWindow, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Option from "effect/Option"; + +import { clampPercent, makeUsageLimits } from "@t3tools/provider-core/server/usageLimits"; +import type { MuseSubscriptionUsage } from "./protocol.ts"; + +const WEEK_MINS = 7 * 24 * 60; + +function isoFromEpochMs(value: number): string | undefined { + const dt = DateTime.make(value); + return Option.isSome(dt) ? DateTime.formatIso(dt.value) : undefined; +} + +function window(input: { + readonly id: string; + readonly kind: ServerProviderUsageWindow["kind"]; + readonly label: string; + readonly usedPercent: number; + readonly resetsAtMs: number; + readonly windowDurationMins: number; + readonly nowMs: number; +}): ServerProviderUsageWindow { + // Meta opens the next window with the account's next request, so a window + // that has reset since the report is empty and its new reset is unknown. + const expired = input.resetsAtMs <= input.nowMs; + const resetsAt = expired ? undefined : isoFromEpochMs(input.resetsAtMs); + return { + id: input.id, + kind: input.kind, + label: input.label, + usedPercent: expired ? 0 : clampPercent(input.usedPercent), + windowDurationMins: input.windowDurationMins, + ...(resetsAt ? { resetsAt } : {}), + }; +} + +/** The two subscription windows a Muse report names, as T3 limit windows. */ +function museSubscriptionUsageWindows( + usage: MuseSubscriptionUsage, + nowMs: number, +): ReadonlyArray { + return [ + window({ + id: "session", + kind: "session", + label: "Session", + usedPercent: usage.window.usedPercent, + resetsAtMs: usage.window.resetsAtMs, + windowDurationMins: usage.window.windowDurationMins, + nowMs, + }), + window({ + id: "weekly", + kind: "weekly", + label: "Weekly", + usedPercent: usage.weekly.usedPercent, + resetsAtMs: usage.weekly.resetsAtMs, + windowDurationMins: WEEK_MINS, + nowMs, + }), + ]; +} + +/** Whether a report is newer than the one already retained, by Meta's observation time. */ +export function isNewerMuseUsageReport( + previous: MuseSubscriptionUsage | undefined, + next: MuseSubscriptionUsage, +): boolean { + return previous === undefined || next.observedAtMs > previous.observedAtMs; +} + +/** Limits as of `nowMs`, dated when Meta reported them. */ +export function museSubscriptionUsageToLimits(input: { + readonly usage: MuseSubscriptionUsage; + readonly nowMs: number; +}): ServerProviderUsageLimits { + return makeUsageLimits({ + checkedAt: DateTime.formatIso(DateTime.makeUnsafe(input.usage.observedAtMs)), + windows: museSubscriptionUsageWindows(input.usage, input.nowMs), + }); +} + +/** A live whole-state update for a report, to publish with `replace: true`. */ +export function museSubscriptionUsageToUpdate(input: { + readonly usage: MuseSubscriptionUsage; + readonly nowMs: number; +}): ProviderUsageLimitsUpdate & { readonly checkedAt: string; readonly replace: true } { + return { + windows: [...museSubscriptionUsageWindows(input.usage, input.nowMs)], + checkedAt: DateTime.formatIso(DateTime.makeUnsafe(input.usage.observedAtMs)), + replace: true, + }; +}