fix(provider-pi): Pi usage-limit failures now arm usage-limit recovery

packages/provider-pi/src/server/adapter.test.ts

@@ -32,6 +32,7 @@ import * as Queue from "effect/Queue";
 import * as Schema from "effect/Schema";
 import * as Sink from "effect/Sink";
 import * as Stream from "effect/Stream";
+import * as TestClock from "effect/testing/TestClock";
 import * as ChildProcess from "effect/process/ChildProcess";
 import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
 
@@ -847,6 +848,13 @@ describe("PiAdapterV2", () => {
         }
         const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
         assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "failed");
+        if (
+          failure === "model error" &&
+          terminal.type === "turn.terminal" &&
+          terminal.status === "failed"
+        ) {
+          assert.strictEqual(terminal.failure.class, "provider_error");
+        }
         assert.deepInclude(runtime.providerSession, {
           status: "error",
           extensionStatuses: [{ key: "watch", text: "Watching 3 files" }],
@@ -4693,6 +4701,104 @@ describe("PiAdapterV2", () => {
     }).pipe(Effect.scoped, Effect.provide(layerTest)),
   );
 
+  it.effect.each([
+    {
+      name: "a streamed usage limit",
+      errorMessage: "Codex error: The usage limit has been reached",
+      resetsInMinutes: null,
+    },
+    {
+      name: "a ChatGPT plan limit with a reset time",
+      errorMessage: "You have hit your ChatGPT usage limit (plus plan). Try again in ~137 min.",
+      resetsInMinutes: 137,
+    },
+    {
+      name: "a ChatGPT plan limit whose reset rounds to ~0 min",
+      errorMessage: "You have hit your ChatGPT usage limit (plus plan). Try again in ~0 min.",
+      resetsInMinutes: null,
+    },
+  ])("fails the turn as a usage limit on $name", ({ errorMessage, resetsInMinutes }) =>
+    Effect.gen(function* () {
+      yield* TestClock.setTime(Date.parse("2026-11-09T12:00:00.000Z"));
+      const fake = yield* makeFakePi;
+      const { runtime, takeEvent } = yield* openRuntime(fake);
+      const providerThread = yield* runtime.ensureThread({
+        threadId: THREAD_ID,
+        modelSelection: modelSelection("default"),
+        runtimePolicy,
+      });
+      yield* startTurn(runtime, providerThread);
+      yield* fake.takeRequest("prompt");
+      const receivedAt = yield* DateTime.now;
+      yield* fake.emit({
+        type: "message_end",
+        message: { role: "assistant", stopReason: "error", errorMessage },
+      });
+      yield* fake.emit({ type: "agent_settled" });
+
+      const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
+      assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "failed");
+      if (terminal.type !== "turn.terminal" || terminal.status !== "failed") return;
+      assert.strictEqual(terminal.failure.class, "usage_limit");
+      if (resetsInMinutes === null) {
+        assert.isUndefined(terminal.failure.resetAt);
+        return;
+      }
+      assert.strictEqual(
+        terminal.failure.resetAt,
+        DateTime.formatIso(
+          DateTime.add(receivedAt, { milliseconds: resetsInMinutes * 60_000 + 30_000 }),
+        ),
+      );
+    }).pipe(Effect.scoped, Effect.provide(layerTest)),
+  );
+
+  it.effect("fails the turn as a usage limit when Pi's retry hits the limit", () =>
+    Effect.gen(function* () {
+      const fake = yield* makeFakePi;
+      const { runtime, takeEvent } = yield* openRuntime(fake);
+      const providerThread = yield* runtime.ensureThread({
+        threadId: THREAD_ID,
+        modelSelection: modelSelection("default"),
+        runtimePolicy,
+      });
+      yield* startTurn(runtime, providerThread);
+      yield* fake.takeRequest("prompt");
+      yield* fake.emit({
+        type: "message_end",
+        message: {
+          role: "assistant",
+          stopReason: "error",
+          errorMessage: "stream disconnected before completion",
+        },
+      });
+      yield* fake.emit({
+        type: "auto_retry_start",
+        attempt: 1,
+        maxAttempts: 3,
+        delayMs: 2_000,
+        errorMessage: "stream disconnected before completion",
+      });
+      const errorMessage = "Codex error: The usage limit has been reached";
+      yield* fake.emit({
+        type: "message_end",
+        message: { role: "assistant", stopReason: "error", errorMessage },
+      });
+      yield* fake.emit({
+        type: "auto_retry_end",
+        success: false,
+        attempt: 1,
+        finalError: errorMessage,
+      });
+      yield* fake.emit({ type: "agent_settled" });
+
+      const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
+      assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "failed");
+      if (terminal.type !== "turn.terminal" || terminal.status !== "failed") return;
+      assert.strictEqual(terminal.failure.class, "usage_limit");
+    }).pipe(Effect.scoped, Effect.provide(layerTest)),
+  );
+
   it.effect.each([
     {
       pi: "clears its native queue",

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

@@ -94,6 +94,35 @@ const PI_UNSOLICITED_ACTIVITY_ERROR =
 const SETTLE_PROBE_MAX_ATTEMPTS = 3;
 const SETTLE_PROBE_RETRY_DELAY = Duration.millis(100);
 
+const PI_USAGE_LIMIT = /usage[ _]limit/i;
+// Pi rounds the server's resets_at to the nearest minute, so the margin keeps recovery after the
+// real reset; ~0 (a reset past or under 30 s away) carries no usable reset.
+const PI_USAGE_LIMIT_RESET = /Try again in ~([1-9]\d{0,5}) min\b/;
+const PI_USAGE_LIMIT_RESET_MARGIN_MS = 30_000;
+
+function piModelErrorFailure(
+  message: string,
+  receivedAt: DateTime.Utc,
+  retryable: boolean | null = null,
+) {
+  if (!PI_USAGE_LIMIT.test(message))
+    return makeProviderFailure({ message, class: "provider_error", retryable });
+  const minutes = PI_USAGE_LIMIT_RESET.exec(message)?.[1];
+  return makeProviderFailure({
+    message,
+    class: "usage_limit",
+    retryable,
+    resetAt:
+      minutes === undefined
+        ? null
+        : DateTime.formatIso(
+            DateTime.add(receivedAt, {
+              milliseconds: Number(minutes) * 60_000 + PI_USAGE_LIMIT_RESET_MARGIN_MS,
+            }),
+          ),
+  });
+}
+
 const PiProviderCapabilitiesV2 = {
   runtimePolicy: { enforcement: "client-boundary" },
   sessions: {
@@ -2002,10 +2031,10 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (
             if (recordString(message, "role") !== "assistant") return;
             yield* completeOpenStreamItems(turn);
             if (recordString(message, "stopReason") === "error" && turn.failure === null) {
-              turn.failure = makeProviderFailure({
-                message: recordString(message, "errorMessage") ?? "Pi reported a model error.",
-                class: "provider_error",
-              });
+              turn.failure = piModelErrorFailure(
+                recordString(message, "errorMessage") ?? "Pi reported a model error.",
+                yield* DateTime.now,
+              );
             }
             return;
           }
@@ -2139,11 +2168,11 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (
               turn.failure = null;
               return;
             }
-            const failure = makeProviderFailure({
-              message: recordString(event, "finalError") ?? "Pi auto-retry failed.",
-              class: "provider_error",
-              retryable: false,
-            });
+            const failure = piModelErrorFailure(
+              recordString(event, "finalError") ?? "Pi auto-retry failed.",
+              emittedAt,
+              false,
+            );
             const attempt = Math.max(1, Math.trunc(recordNumber(event, "attempt") ?? 1));
             const current = turn.activeProviderRetry;
             const providerRetry = {