feat(providers): show Muse Code subscription usage in Limits

apps/mobile/src/features/usage/UsageLimitsPooled.tsx

@@ -27,7 +27,11 @@ import { environmentPresentations } from "../../state/presentation";
 import { ResetCredits } from "./UsageLimitsSection";
 import { useProviderColors } from "./usageProviders";
 
-const DRIVER_LABEL: Partial<Record<string, string>> = { codex: "Codex", claudeAgent: "Claude" };
+const DRIVER_LABEL: Partial<Record<string, string>> = {
+  codex: "Codex",
+  claudeAgent: "Claude",
+  muse: "Muse Code",
+};
 const PACE_LABEL = { ahead: "Ahead of pace", on: "On pace", under: "Under pace" } as const;
 
 function accountName(account: LimitAccount) {

apps/server/src/orchestration-v2/Adapters/MuseAdapterV2.test.ts

@@ -33,7 +33,7 @@ import * as Stream from "effect/Stream";
 
 import * as ServerConfig from "../../config.ts";
 import * as McpProviderSession from "@t3tools/provider-core/server/mcpSession";
-import type { MuseItem } from "../../provider/museProtocol.ts";
+import type { MuseItem, MuseSubscriptionUsage } from "../../provider/museProtocol.ts";
 import type { MuseSdkHost } from "../../provider/museSdk.ts";
 import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
 import {
@@ -178,6 +178,11 @@ const makeFakeMuse = Effect.fnUntraced(function* (idPrefix = "native") {
       Effect.sync(() => {
         notify({ jsonrpc: "2.0", method, params: { sessionId, ...params } });
       }),
+    /** A notification about the host or account rather than a session, as `usage/changed` is. */
+    emitHostWide: (method: string, params: Record<string, unknown>) =>
+      Effect.sync(() => {
+        notify({ jsonrpc: "2.0", method, params });
+      }),
     takeCall: Effect.fnUntraced(function* (method: string) {
       while (true) {
         const call = yield* Queue.take(requests);
@@ -196,7 +201,13 @@ const makeHarness = Effect.fnUntraced(function* (
   policy = runtimePolicy,
   overrides: Pick<
     MuseAdapterV2Options,
-    "createHost" | "nativeEventLogger" | "modelCatalog" | "continuationRequests"
+    | "createHost"
+    | "nativeEventLogger"
+    | "modelCatalog"
+    | "continuationRequests"
+    | "onSubscriptionUsage"
+    | "latestSubscriptionUsage"
+    | "usageAccountGeneration"
   > = {},
 ) {
   let hostCount = 0;
@@ -340,6 +351,17 @@ const startConversation = Effect.fnUntraced(function* (
   return { nativeId: call.commandId!, providerTurn: event.providerTurn };
 });
 
+// `usage/changed` as Muse 1.4.3 sent it after a reply (2026-10-08), with a placeholder tier.
+const SUBSCRIPTION_USAGE = {
+  window: { usedPercent: 0, windowDurationMins: 300, resetsAtMs: 1791465557000 },
+  weekly: { usedPercent: 5, resetsAtMs: 1791763200000 },
+  tier: "tier-1",
+  observedAtMs: 1791447762610,
+} satisfies MuseSubscriptionUsage;
+// How Muse 1.4.3 ends a turn Meta refuses for quota (from a simulated Meta 429).
+const QUOTA_MESSAGE =
+  "API error 429: Subscription quota exhausted. Your usage window resets soon. (rate_limit_error)";
+
 const approval = (turnId: string) => ({
   approvalId: "approval-1",
   turnId,
@@ -523,6 +545,87 @@ describe("MuseAdapterV2", () => {
     }).pipe(Effect.scoped, Effect.provide(testLayer)),
   );
 
+  it.effect("hands Meta's subscription usage to the instance, with or without a turn", () =>
+    Effect.gen(function* () {
+      const fake = yield* makeFakeMuse();
+      const reports: Array<readonly [MuseSubscriptionUsage, number]> = [];
+      let generation = 3;
+      const harness = yield* makeHarness(
+        fake,
+        INSTANCE_ID,
+        undefined,
+        undefined,
+        undefined,
+        runtimePolicy,
+        {
+          onSubscriptionUsage: (usage, accountGeneration) =>
+            Effect.sync(() => void reports.push([usage, accountGeneration])),
+          usageAccountGeneration: Effect.sync(() => generation),
+        },
+      );
+      // A report counts for the generation its host started under, not the current one.
+      generation = 4;
+      yield* fake.emitHostWide("usage/changed", SUBSCRIPTION_USAGE);
+      yield* fake.emitHostWide("usage/changed", { window: "not usage" });
+      const { nativeId } = yield* startConversation(harness, fake);
+      yield* fake.emit("turn/completed", { turnId: nativeId, terminal: "completed" });
+      const terminal = yield* harness.takeEvent("turn.terminal");
+      assert.strictEqual(terminal.status, "completed");
+      assert.deepStrictEqual(reports, [[SUBSCRIPTION_USAGE, 3]]);
+      assert.strictEqual(fake.closeCount(), 0);
+    }).pipe(Effect.scoped, Effect.provide(testLayer)),
+  );
+
+  it.effect("stops at a Meta usage limit until every exhausted window resets", () =>
+    Effect.gen(function* () {
+      const fake = yield* makeFakeMuse();
+      const harness = yield* makeHarness(
+        fake,
+        INSTANCE_ID,
+        undefined,
+        undefined,
+        undefined,
+        runtimePolicy,
+        {
+          latestSubscriptionUsage: Effect.succeed({
+            ...SUBSCRIPTION_USAGE,
+            window: { ...SUBSCRIPTION_USAGE.window, usedPercent: 103 },
+          }),
+        },
+      );
+      const { nativeId } = yield* startConversation(harness, fake);
+      yield* fake.emit("turn/completed", {
+        turnId: nativeId,
+        terminal: "failed",
+        reason: QUOTA_MESSAGE,
+        error: { kind: "modelError", message: QUOTA_MESSAGE, retryable: false },
+      });
+      const terminal = yield* harness.takeEvent("turn.terminal");
+      assert.strictEqual(terminal.status, "failed");
+      assert.strictEqual(terminal.threadDisposition, "reusable");
+      assert.strictEqual(terminal.failure?.class, "usage_limit");
+      // The exhausted session window's reset, not the weekly one.
+      assert.strictEqual(terminal.failure?.resetAt, "2026-10-08T13:19:17.000Z");
+      assert.isTrue(
+        harness.allEvents.some(
+          (event) =>
+            event.type === "turn_item.updated" &&
+            event.turnItem.type === "error" &&
+            event.turnItem.title === "Usage limit reached",
+        ),
+      );
+
+      const next = yield* startConversation(harness, fake, 2);
+      yield* fake.emit("turn/completed", {
+        turnId: next.nativeId,
+        terminal: "failed",
+        error: { message: "API error 429: Rate limit exceeded. (rate_limit_error)" },
+      });
+      const throttled = yield* harness.takeEvent("turn.terminal");
+      assert.strictEqual(throttled.failure?.class, "provider_error");
+    }).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();

apps/server/src/orchestration-v2/Adapters/MuseAdapterV2.ts

@@ -49,6 +49,7 @@ import {
   MuseDelta,
   MuseItemEvent,
   MuseSessionResult,
+  MuseSubscriptionUsage,
   MuseTokenUsageEvent,
   MuseTodoList,
   MuseTurnRetryScheduled,
@@ -60,6 +61,7 @@ import {
   museApprovalOptions,
   type MuseItem,
 } from "../../provider/museProtocol.ts";
+import { isMuseUsageLimitFailure, museUsageLimitResetAt } from "../../provider/museUsageLimits.ts";
 import {
   createMuseSdkHostEffect,
   museApprovalMode,
@@ -216,6 +218,18 @@ export interface MuseAdapterV2Options {
       request: ProviderContinuationRequests.ProviderContinuationRequest,
     ) => Effect.Effect<void>;
   };
+  /**
+   * Receives each subscription usage report (`usage/changed`) from this adapter's hosts,
+   * with the account generation the reporting host started under.
+   */
+  readonly onSubscriptionUsage?: (
+    usage: MuseSubscriptionUsage,
+    accountGeneration: number,
+  ) => Effect.Effect<void>;
+  /** The instance's account generation, read as each host starts, since reports name no account. */
+  readonly usageAccountGeneration?: Effect.Effect<number>;
+  /** The newest report across the instance, for dating a turn stopped at a usage limit. */
+  readonly latestSubscriptionUsage?: Effect.Effect<MuseSubscriptionUsage | undefined>;
 }
 
 interface ActiveTurn {
@@ -260,6 +274,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.
@@ -319,6 +334,8 @@ export function makeMuseAdapterV2(options: MuseAdapterV2Options): ProviderAdapte
       const eventPermit = yield* Semaphore.make(1);
       let host: MuseSdkHost;
       let hostEpoch = 0;
+      // The account generation the current host started under; see usageAccountGeneration.
+      let hostAccountGeneration = 0;
       let closed = false;
       let broken = false;
       let nativeSessionId: string | undefined;
@@ -736,9 +753,23 @@ export function makeMuseAdapterV2(options: MuseAdapterV2Options): ProviderAdapte
         yield* updateSession(disposition === "broken" ? "error" : "ready", detail ?? null);
         active = undefined;
         if (status === "failed") {
+          const usageLimited = disposition === "reusable" && isMuseUsageLimitFailure(detail);
           const failure = makeProviderFailure({
-            class: disposition === "broken" ? "transport_error" : "provider_error",
+            class:
+              disposition === "broken"
+                ? "transport_error"
+                : usageLimited
+                  ? "usage_limit"
+                  : "provider_error",
             message: detail ?? "Muse turn failed.",
+            ...(usageLimited && options.latestSubscriptionUsage
+              ? {
+                  resetAt: museUsageLimitResetAt(
+                    yield* options.latestSubscriptionUsage,
+                    DateTime.toEpochMillis(completedAt),
+                  ),
+                }
+              : {}),
           });
           const base = baseItem(turn, `failure:${turn.nativeId}`, completedAt);
           yield* emit({
@@ -748,7 +779,7 @@ export function makeMuseAdapterV2(options: MuseAdapterV2Options): ProviderAdapte
               ...base,
               type: "error",
               status: "failed",
-              title: null,
+              title: usageLimited ? "Usage limit reached" : null,
               completedAt,
               failure,
             },
@@ -922,6 +953,13 @@ export function makeMuseAdapterV2(options: MuseAdapterV2Options): ProviderAdapte
       });
       const handleNotification = Effect.fnUntraced(function* (method: string, data: unknown) {
         const params = yield* decode(recordSchema, data);
+        // Account-wide, so it names no session: Meta's subscription windows after a response.
+        if (method === "usage/changed") {
+          const usage = yield* decode(MuseSubscriptionUsage, params);
+          if (options.onSubscriptionUsage)
+            yield* options.onSubscriptionUsage(usage, hostAccountGeneration);
+          return;
+        }
         if (params.sessionId !== nativeSessionId) return;
         if (method === "view/gap")
           return yield* protocolError("Muse delivery gap requires session recovery");
@@ -1315,6 +1353,9 @@ export function makeMuseAdapterV2(options: MuseAdapterV2Options): ProviderAdapte
       );
       const launchHost = Effect.fnUntraced(function* () {
         const epoch = ++hostEpoch;
+        hostAccountGeneration = options.usageAccountGeneration
+          ? yield* options.usageAccountGeneration
+          : 0;
         const mcpSession = McpProviderSession.readMcpProviderSession(input.threadId);
         const created = yield* Effect.acquireRelease(
           createMuseSdkHostEffect(

apps/server/src/provider/Drivers/MuseDriver.ts

@@ -1,9 +1,12 @@
 import { MuseSettings, ProviderDriverKind } from "@t3tools/contracts";
 import { HostProcessEnvironment } 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 { HttpClient } from "effect/http";
 import { ChildProcessSpawner } from "effect/process";
 
@@ -19,7 +22,15 @@ import { checkMuseProviderStatus, makePendingMuseProvider } from "../MuseProvide
 import * as ProviderEventLoggers from "../ProviderEventLoggers.ts";
 import { makeManagedServerProvider } from "@t3tools/provider-core/server/managedProvider";
 import { enrichMuseSnapshot, latestMuseVersion, museMaintenance } from "../museMaintenance.ts";
+import type { MuseSubscriptionUsage } from "../museProtocol.ts";
 import { makeMuseEnvironment } from "../museSdk.ts";
+import {
+  museStatusUsageLimits,
+  museUsageWindows,
+  nextMuseUsageAccount,
+  type MuseUsageAccount,
+} from "../museUsageLimits.ts";
+import type { ServerProviderDraft } from "@t3tools/provider-core/server/snapshotProbe";
 import {
   defaultProviderContinuationIdentity,
   type ProviderDriver,
@@ -90,6 +101,41 @@ export const MuseDriver: ProviderDriver<MuseSettings, MuseDriverEnv> = {
         continuationGroupKey: continuationIdentity.continuationKey,
       });
       const snapshotSettings = makeProviderSnapshotSettingsSource(effectiveConfig, host.settings);
+      // Meta reports subscription usage only with a model response, 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.
+      // Each report counts for the account generation its host started under.
+      const usageAccount = yield* Ref.make<MuseUsageAccount>({
+        generation: 0,
+        identity: undefined,
+      });
+      const latestUsage = yield* Ref.make<
+        { readonly usage: MuseSubscriptionUsage; readonly generation: number } | undefined
+      >(undefined);
+      // Reports and status checks take turns, so an older report cannot publish after a
+      // newer one, and none publishes for a generation a check has just replaced.
+      const usagePermit = yield* Semaphore.make(1);
+      const withUsageLimits = (provider: ServerProviderDraft) =>
+        usagePermit.withPermits(1)(
+          Effect.gen(function* () {
+            const { generation } = yield* Ref.updateAndGet(usageAccount, (account) =>
+              nextMuseUsageAccount(account, provider.auth),
+            );
+            const { retained, dropped } = yield* Ref.modify(latestUsage, (current) => {
+              const drop = current !== undefined && current.generation !== generation;
+              const kept = drop ? undefined : current;
+              return [{ retained: kept, dropped: drop }, kept] as const;
+            });
+            const usageLimits = museStatusUsageLimits({
+              auth: provider.auth,
+              enabled: provider.enabled,
+              observation: retained?.usage,
+              dropped,
+              nowMs: DateTime.toEpochMillis(yield* DateTime.now),
+            });
+            return usageLimits ? { ...provider, usageLimits } : provider;
+          }),
+        );
       const resolveInstallation = yield* makeCachedProviderMaintenanceResolution(
         resolveProviderMaintenanceCapabilitiesEffect(museMaintenance, {
           binaryPath: effectiveConfig.binaryPath,
@@ -120,6 +166,7 @@ export const MuseDriver: ProviderDriver<MuseSettings, MuseDriverEnv> = {
         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),
@@ -158,6 +205,37 @@ export const MuseDriver: ProviderDriver<MuseSettings, MuseDriverEnv> = {
         serverConfig,
         fileSystem,
         modelCatalog,
+        latestSubscriptionUsage: Ref.get(latestUsage).pipe(
+          Effect.map((retained) => retained?.usage),
+        ),
+        usageAccountGeneration: Ref.get(usageAccount).pipe(
+          Effect.map((account) => account.generation),
+        ),
+        onSubscriptionUsage: (usage, hostGeneration) =>
+          usagePermit.withPermits(1)(
+            Effect.gen(function* () {
+              const account = yield* Ref.get(usageAccount);
+              // A host from before a logout or another login reports for an account
+              // that is gone, and signed out there is no account to report for.
+              if (hostGeneration !== account.generation || account.identity === null) return;
+              // Every session's host reports; an older report arriving late must not win.
+              const newer = yield* Ref.modify(latestUsage, (previous) =>
+                previous && previous.usage.observedAtMs >= usage.observedAtMs
+                  ? [false, previous]
+                  : [true, { usage, generation: hostGeneration }],
+              );
+              // An API-key instance leaves the account to its hub; see museStatusUsageLimits.
+              if (!newer || (yield* snapshot.getSnapshot).auth.type === "apiKey") return;
+              const now = yield* DateTime.now;
+              yield* snapshot.applyUsageLimits({
+                windows: museUsageWindows(usage, DateTime.toEpochMillis(now)),
+                checkedAt: DateTime.formatIso(DateTime.makeUnsafe(usage.observedAtMs)),
+                // A report is the account's whole state, so a window that has reset
+                // since loses its old reset time instead of keeping it.
+                replace: true,
+              });
+            }),
+          ),
         ...(eventLoggers.native ? { nativeEventLogger: eventLoggers.native } : {}),
         continuationRequests,
       });