diff --git a/packages/provider-pi/src/server/adapter.test.ts b/packages/provider-pi/src/server/adapter.test.ts index 69f2287d8e52..9c1846cc866d 100644 --- a/packages/provider-pi/src/server/adapter.test.ts +++ b/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", diff --git a/packages/provider-pi/src/server/adapter.ts b/packages/provider-pi/src/server/adapter.ts index 4b06ee16bfd8..f2257671780f 100644 --- a/packages/provider-pi/src/server/adapter.ts +++ b/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 = {