fix(server): Muse turns stopped by the subscription quota are Limited, not failed

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";
@@ -173,6 +173,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);
@@ -339,6 +344,20 @@ 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 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)";
+// How Muse 1.4.6 ended a turn at a real quota stop (2026-10-13), relayed by a credential proxy.
+const RELAYED_QUOTA_MESSAGE =
+  "API error 429: All credentials for model muse-wonder-1.0-contributor are cooling down (last error: rate_limit_exceeded: Subscription quota exhausted. Your usage window resets at 2026-10-13T10:45:52Z.) (after 10 provider attempts)";
+
 const approval = (turnId: string) => ({
   approvalId: "approval-1",
   turnId,
@@ -548,6 +567,129 @@ describe("MuseAdapterV2", () => {
     }).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);
+      // Meta's report after the last reply that went through: the session window ran out.
+      yield* fake.emitHostWide("usage/changed", {
+        ...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("limits a quota stop without a reset time when no report shows a used-up window", () =>
+    Effect.gen(function* () {
+      const fake = yield* makeFakeMuse();
+      const harness = yield* makeHarness(fake);
+      // The last report still had room in both windows, so it cannot say when Meta will allow more.
+      yield* fake.emitHostWide("usage/changed", SUBSCRIPTION_USAGE);
+      const { nativeId } = yield* startConversation(harness, fake);
+      yield* fake.emit("turn/completed", {
+        turnId: nativeId,
+        terminal: "failed",
+        error: { message: QUOTA_MESSAGE },
+      });
+      const terminal = yield* harness.takeEvent("turn.terminal");
+      assert.strictEqual(terminal.failure?.class, "usage_limit");
+      assert.isUndefined(terminal.failure?.resetAt);
+    }).pipe(Effect.scoped, Effect.provide(testLayer)),
+  );
+
+  it.effect("resumes a quota stop at the reset the refusal names", () =>
+    Effect.gen(function* () {
+      const fake = yield* makeFakeMuse();
+      const harness = yield* makeHarness(fake);
+      // A fresh host has no report yet; the refusal itself says when the window resets.
+      const { nativeId } = yield* startConversation(harness, fake);
+      yield* fake.emit("turn/completed", {
+        turnId: nativeId,
+        terminal: "failed",
+        reason: RELAYED_QUOTA_MESSAGE,
+        error: { kind: "modelError", message: RELAYED_QUOTA_MESSAGE, retryable: true },
+      });
+      const terminal = yield* harness.takeEvent("turn.terminal");
+      assert.strictEqual(terminal.failure?.class, "usage_limit");
+      assert.strictEqual(terminal.failure?.resetAt, "2026-10-13T10:45:52.000Z");
+    }).pipe(Effect.scoped, Effect.provide(testLayer)),
+  );
+
+  it.effect("limits a stop Muse types as quota at the reset it reports", () =>
+    Effect.gen(function* () {
+      const fake = yield* makeFakeMuse();
+      const harness = yield* makeHarness(fake);
+      const { nativeId } = yield* startConversation(harness, fake);
+      // The typed quota decides even when the text names no quota.
+      yield* fake.emit("turn/completed", {
+        turnId: nativeId,
+        terminal: "failed",
+        error: {
+          kind: "modelError",
+          message: "Model request refused.",
+          quota: { limitKind: "tierLimit", resetsAtMs: SUBSCRIPTION_USAGE.window.resetsAtMs },
+          retryable: true,
+        },
+      });
+      const terminal = yield* harness.takeEvent("turn.terminal");
+      assert.strictEqual(terminal.failure?.class, "usage_limit");
+      assert.strictEqual(terminal.failure?.resetAt, "2026-10-08T13:19:17.000Z");
+    }).pipe(Effect.scoped, Effect.provide(testLayer)),
+  );
+
+  it.effect("skips an unreadable usage report without ending the session", () =>
+    Effect.gen(function* () {
+      const fake = yield* makeFakeMuse();
+      const harness = yield* makeHarness(fake);
+      yield* fake.emitHostWide("usage/changed", { window: "not usage" });
+      // A reset no Date can hold would throw once formatted, so that report is skipped too.
+      yield* fake.emitHostWide("usage/changed", {
+        ...SUBSCRIPTION_USAGE,
+        window: { ...SUBSCRIPTION_USAGE.window, usedPercent: 103, resetsAtMs: 1e16 },
+      });
+      const { nativeId } = yield* startConversation(harness, fake);
+      yield* fake.emit("turn/completed", {
+        turnId: nativeId,
+        terminal: "failed",
+        error: { message: QUOTA_MESSAGE },
+      });
+      const terminal = yield* harness.takeEvent("turn.terminal");
+      assert.strictEqual(terminal.failure?.class, "usage_limit");
+      assert.isUndefined(terminal.failure?.resetAt);
+      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();

packages/provider-muse/src/server/adapter.ts

@@ -50,6 +50,7 @@ import {
   MuseDelta,
   MuseItemEvent,
   MuseSessionResult,
+  MuseSubscriptionUsage,
   MuseTokenUsageEvent,
   MuseTodoList,
   MuseTurnRetryScheduled,
@@ -61,6 +62,7 @@ import {
   museApprovalOptions,
   type MuseItem,
 } from "./protocol.ts";
+import { isMuseUsageLimitFailure, museUsageLimitResetAt } from "./usageLimits.ts";
 import {
   createMuseSdkHostEffect,
   museApprovalMode,
@@ -261,6 +263,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.
@@ -335,6 +338,8 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* (
       let active: ActiveTurn | undefined;
       // The last finished turn, so context usage reported after it still lands on it.
       let lastProviderTurn: OrchestrationV2ProviderTurn | undefined;
+      // Meta's newest subscription windows, for when a turn stopped at a usage limit can resume.
+      let latestSubscriptionUsage: MuseSubscriptionUsage | undefined;
       const agentEndedAt = new Map<string, DateTime.Utc>();
       // "default" is the catalog's default model, from the provider snapshot. Without
       // one, Muse keeps whatever model the session already runs.
@@ -966,6 +971,7 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* (
         status: "completed" | "cancelled" | "interrupted" | "failed",
         detail?: string,
         disposition: "reusable" | "broken" = "reusable",
+        quota?: { readonly resetsAtMs: number },
       ) {
         if (active !== turn || seenTerminals.has(turn.nativeId)) return;
         seenTerminals.add(turn.nativeId);
@@ -1004,9 +1010,25 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* (
         yield* updateSession(disposition === "broken" ? "error" : "ready", detail ?? null);
         active = undefined;
         if (status === "failed") {
+          const usageLimited =
+            disposition === "reusable" && (quota !== undefined || 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
+              ? {
+                  resetAt: museUsageLimitResetAt(
+                    latestSubscriptionUsage,
+                    DateTime.toEpochMillis(completedAt),
+                    { quota, message: detail },
+                  ),
+                }
+              : {}),
           });
           const base = baseItem(turn, `failure:${turn.nativeId}`, completedAt);
           yield* emit({
@@ -1016,7 +1038,7 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* (
               ...base,
               type: "error",
               status: "failed",
-              title: null,
+              title: usageLimited ? "Usage limit reached" : null,
               completedAt,
               failure,
             },
@@ -1285,6 +1307,11 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* (
       });
       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") {
+          latestSubscriptionUsage = yield* decode(MuseSubscriptionUsage, params);
+          return;
+        }
         if (params.sessionId !== nativeSessionId) return;
         if (method === "view/gap")
           return yield* protocolError("Muse delivery gap requires session recovery");
@@ -1459,7 +1486,13 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* (
                     ? "interrupted"
                     : "cancelled"
                   : "failed";
-            yield* finish(turn, status, result.error?.message ?? result.reason);
+            yield* finish(
+              turn,
+              status,
+              result.error?.message ?? result.reason,
+              "reusable",
+              result.error?.quota,
+            );
             break;
           }
           case "approval/requested":

packages/provider-muse/src/server/protocol.ts

@@ -155,9 +155,39 @@ export const MuseTurnCompleted = Schema.Struct({
   turnId: NonEmptyString,
   terminal: NonEmptyString,
   reason: Schema.optional(Schema.String),
-  error: Schema.optional(Schema.Struct({ message: Schema.String })),
+  error: Schema.optional(
+    Schema.Struct({
+      message: Schema.String,
+      // Present when Muse recognized the stop as a used-up quota.
+      quota: Schema.optional(Schema.Struct({ resetsAtMs: Schema.Finite })),
+    }),
+  ),
   usage: Schema.optional(MuseUsage),
 });
+/** The latest instant a JavaScript Date holds; formatting a later one throws. */
+export const LATEST_DATE_MS = 8.64e15;
+const DateMillis = Schema.Finite.check(
+  Schema.isGreaterThan(0),
+  Schema.isLessThanOrEqualTo(LATEST_DATE_MS),
+);
+/**
+ * `usage/changed` params: the Meta subscription windows the latest model
+ * response reported. Percentages are Meta's, verbatim, and may exceed 100.
+ */
+export const MuseSubscriptionUsage = Schema.Struct({
+  observedAtMs: DateMillis,
+  tier: Schema.String,
+  window: Schema.Struct({
+    usedPercent: Schema.Finite.check(Schema.isGreaterThanOrEqualTo(0)),
+    windowDurationMins: Schema.Finite.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 MuseTurnRetryScheduled = Schema.Struct({
   turnId: NonEmptyString,
   nextAttempt: Schema.Int.check(Schema.isGreaterThan(0)),

packages/provider-muse/src/server/usageLimits.test.ts

@@ -0,0 +1,60 @@
+import { describe, expect, it } from "@effect/vitest";
+
+import type { MuseSubscriptionUsage } from "./protocol.ts";
+import { isMuseUsageLimitFailure, museUsageLimitResetAt } from "./usageLimits.ts";
+
+// What Meta reported after a reply on 2026-10-08, as Muse forwards it.
+const usage = {
+  observedAtMs: 1791447762610,
+  tier: "tier-1",
+  window: { usedPercent: 0, windowDurationMins: 300, resetsAtMs: 1791465557000 },
+  weekly: { usedPercent: 5, resetsAtMs: 1791763200000 },
+} satisfies MuseSubscriptionUsage;
+const nowMs = usage.observedAtMs;
+// How Muse 1.4.6 ended a turn at a real quota stop (2026-10-13), relayed by a credential proxy.
+const realRefusal =
+  "API error 429: All credentials for model muse-wonder-1.0-contributor are cooling down (last error: rate_limit_exceeded: Subscription quota exhausted. Your usage window resets at 2026-10-13T10:45:52Z.) (after 10 provider attempts)";
+
+describe("Muse usage limit failures", () => {
+  it("recognizes Meta's quota refusal and no other failure", () => {
+    expect(
+      isMuseUsageLimitFailure(
+        "API error 429: Subscription quota exhausted. Your usage window resets soon. (rate_limit_error)",
+      ),
+    ).toBe(true);
+    expect(isMuseUsageLimitFailure(realRefusal)).toBe(true);
+    // A plain rate limit retries on its own; it is not a used-up subscription window.
+    expect(isMuseUsageLimitFailure("API error 429: Rate limit exceeded. (rate_limit_error)")).toBe(
+      false,
+    );
+    expect(isMuseUsageLimitFailure("Subscription quota exhausted.")).toBe(false);
+    expect(isMuseUsageLimitFailure(undefined)).toBe(false);
+  });
+
+  it("resumes once every exhausted window has reset", () => {
+    const sessionOut = { ...usage, window: { ...usage.window, usedPercent: 103 } };
+    expect(museUsageLimitResetAt(sessionOut, nowMs)).toBe("2026-10-08T13:19:17.000Z");
+    const bothOut = { ...sessionOut, weekly: { ...usage.weekly, usedPercent: 100 } };
+    expect(museUsageLimitResetAt(bothOut, nowMs)).toBe("2026-10-12T00:00:00.000Z");
+  });
+
+  it("takes the reset a refusal names or types when no report has one", () => {
+    expect(museUsageLimitResetAt(undefined, nowMs, { message: realRefusal })).toBe(
+      "2026-10-13T10:45:52.000Z",
+    );
+    const quota = { resetsAtMs: usage.window.resetsAtMs };
+    expect(museUsageLimitResetAt(usage, nowMs, { quota })).toBe("2026-10-08T13:19:17.000Z");
+    // A used-up weekly window still holds the turn back after the typed reset.
+    const weeklyOut = { ...usage, weekly: { ...usage.weekly, usedPercent: 100 } };
+    expect(museUsageLimitResetAt(weeklyOut, nowMs, { quota })).toBe("2026-10-12T00:00:00.000Z");
+  });
+
+  it("has no reset time unless a report shows a used-up window that resets later", () => {
+    expect(museUsageLimitResetAt(undefined, nowMs)).toBeNull();
+    expect(museUsageLimitResetAt(usage, nowMs)).toBeNull();
+    const sessionOut = { ...usage, window: { ...usage.window, usedPercent: 100 } };
+    expect(museUsageLimitResetAt(sessionOut, usage.window.resetsAtMs)).toBeNull();
+    const afterNamedReset = Date.parse("2026-10-14T00:00:00Z");
+    expect(museUsageLimitResetAt(undefined, afterNamedReset, { message: realRefusal })).toBeNull();
+  });
+});