diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index b07bb23c85eb..6445328835df 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -24,6 +24,7 @@ import { type OrchestrationV2ProviderThread, type OrchestrationV2ProviderTurn, type ProviderReplayEntry, + type ServerProviderUsageLimits, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; @@ -326,6 +327,7 @@ const resumed = ( readonly resolveExtractedTextPath?: (path: string) => Effect.Effect; readonly fileSystem?: FileSystem.FileSystem; readonly onPrompt?: (input: unknown) => void; + readonly readUsageLimits?: Effect.Effect; readonly borrowers?: { current: number }; }, ) => @@ -364,6 +366,9 @@ const resumed = ( ...(options?.resolveExtractedTextPath === undefined ? {} : { resolveExtractedTextPath: options.resolveExtractedTextPath }), + ...(options?.readUsageLimits === undefined + ? {} + : { readUsageLimits: options.readUsageLimits }), }, ); const thread = yield* runtime.resumeThread({ @@ -528,6 +533,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { it.effect.each([ ["provider.quota", "usage_limit"], ["provider.rate-limit", "usage_limit"], + ["provider.go-limit", "usage_limit"], ["provider", "provider_error"], ] as const)("classifies a %s execution failure as %s", ([code, failureClass]) => Effect.gen(function* () { @@ -579,6 +585,15 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }, }, }, + { + name: "an out-of-funds provider.go-limit (402)", + failureClass: "provider_error", + error: { + type: "provider.go-limit", + message: "Upstream request failed: Insufficient account funds", + status: 402, + }, + }, { name: "a provider.quota account budget (429)", failureClass: "usage_limit", @@ -662,6 +677,265 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + const goLimitMessage = + "OpenCode Go 5 hour usage limit reached. Switch to a free model, or turn on Extra Usage on the Go page in OpenCode Console to keep going."; + const goUsage = (rollingPercent: number): ServerProviderUsageLimits => ({ + checkedAt: "2026-11-06T01:49:10.000Z", + windows: [ + { + id: "go_rolling", + kind: "session", + label: "Go · Session", + usedPercent: rollingPercent, + resetsAt: "2026-11-06T04:12:00.000Z", + }, + { + id: "go_weekly", + kind: "weekly", + label: "Go · Weekly", + usedPercent: 41, + resetsAt: "2026-11-09T00:00:00.000Z", + }, + ], + }); + const failWith = (error: Record) => + event("session.execution.failed", { sessionID: SESSION, error }); + const endedBy = ( + error: Record, + readUsageLimits: Effect.Effect, + ) => + Effect.gen(function* () { + const { runtime, thread } = yield* resumed( + [ + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + failWith(error), + ], + { readUsageLimits }, + ); + const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped); + yield* runtime.startTurn(turnInput(thread)); + return yield* Fiber.join(terminal); + }); + + // regression: OpenCode Go threads stayed failed after the Go window reset instead of resuming + it.effect("reports a spent Go window as a usage limit that resets with that window", () => + Effect.gen(function* () { + // OpenCode's usage endpoint can lag the gateway, so the named window wins even below 100%. + const ended = yield* endedBy( + { type: "provider.go-limit", message: goLimitMessage }, + Effect.succeed(goUsage(97)), + ); + assert.equal(ended?.status, "failed"); + assert.deepEqual(ended?.failure, { + class: "usage_limit", + code: "provider.go-limit", + message: goLimitMessage, + retryable: null, + resetAt: "2026-11-06T04:12:00.000Z", + }); + }).pipe(Effect.scoped), + ); + + it.effect("reads the reset for an older provider.quota Go limit from the spent window", () => + Effect.gen(function* () { + const ended = yield* endedBy( + { + type: "provider.quota", + message: "Subscription quota exceeded.", + status: 429, + response: { + body: JSON.stringify({ + type: "error", + error: { type: "GoUsageLimitError", message: "Subscription quota exceeded." }, + }), + }, + }, + Effect.succeed(goUsage(100)), + ); + assert.equal(ended?.failure?.class, "usage_limit"); + assert.equal(ended?.failure?.resetAt, "2026-11-06T04:12:00.000Z"); + }).pipe(Effect.scoped), + ); + + it.effect("keeps a Go limit without a reset when Go usage cannot be read", () => + Effect.gen(function* () { + const ended = yield* endedBy( + { type: "provider.go-limit", message: goLimitMessage }, + Effect.succeed({ + checkedAt: "2026-11-06T01:49:10.000Z", + windows: [], + unavailable: { reason: "probeFailed" }, + }), + ); + assert.deepEqual(ended?.failure, { + class: "usage_limit", + code: "provider.go-limit", + message: goLimitMessage, + retryable: null, + }); + }).pipe(Effect.scoped), + ); + + it.effect("reads Go usage off the event lock, so Stop still ends the limited turn", () => + Effect.gen(function* () { + const reading = yield* Deferred.make(); + const { runtime, thread } = yield* resumed( + [ + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + failWith({ type: "provider.go-limit", message: goLimitMessage, status: 429 }), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: false }), + ], + { + readUsageLimits: Deferred.succeed(reading, undefined).pipe(Effect.andThen(Effect.never)), + }, + ); + const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped); + yield* runtime.startTurn(turnInput(thread)); + yield* Deferred.await(reading); + yield* runtime.interruptTurn({ + providerThread: thread, + providerTurnId: yield* providerTurnId, + }); + assert.equal((yield* Fiber.join(terminal))?.status, "interrupted"); + }).pipe(Effect.scoped), + ); + + it.effect("does not read Go usage for limits from other OpenCode providers", () => + Effect.gen(function* () { + const ended = yield* endedBy( + { + type: "provider.rate-limit", + message: "Rate limit exceeded: free-models-per-min.", + status: 429, + }, + Effect.die("Go usage must not be read for a non-Go limit"), + ); + assert.deepEqual(ended?.failure, { + class: "usage_limit", + code: "provider.rate-limit", + message: "Rate limit exceeded: free-models-per-min.", + retryable: null, + }); + }).pipe(Effect.scoped), + ); + + // An event for a session T3 Code does not know; it only holds the replay at its label. + const marker = (label: string) => + event( + "session.model.selected", + { + sessionID: "ses_f148ca2deffeUnknownSession0", + model: { providerID: "opencode", id: "big-pickle", variant: "default" }, + }, + label, + ); + + // regression: a Stop sent during the Go usage read ended the turn failed, so it resumed at the reset + it.effect("keeps a Stop sent while Go usage is read, so the thread does not resume", () => + Effect.gen(function* () { + const gate = makeProviderReplayGate(["interrupting"]); + const reading = yield* Deferred.make(); + const usage = yield* Deferred.make(); + const { runtime, thread } = yield* resumed( + [ + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + failWith({ type: "provider.go-limit", message: goLimitMessage, status: 429 }), + out("session.interrupt", { sessionID: SESSION }), + // Holds the interrupt reply until the usage read has finished. + marker("interrupting"), + reply("session.interrupt", { interrupted: false }), + ], + { + replayGate: gate, + readUsageLimits: Deferred.succeed(reading, undefined).pipe( + Effect.andThen(Deferred.await(usage)), + ), + }, + ); + const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped); + yield* runtime.startTurn(turnInput(thread)); + yield* Deferred.await(reading); + const stop = yield* runtime + .interruptTurn({ providerThread: thread, providerTurnId: yield* providerTurnId }) + .pipe(Effect.forkScoped); + assert.isTrue(yield* Effect.promise(() => gate.waitForReached("interrupting"))); + yield* Deferred.succeed(usage, goUsage(100)); + const ended = yield* Fiber.join(terminal); + gate.release("interrupting"); + yield* Fiber.join(stop); + assert.equal(ended?.status, "interrupted"); + assert.notExists(ended?.failure); + }).pipe(Effect.scoped), + ); + + // regression: an undecodable start left the Go usage read able to end the restarted execution + it.effect("keeps a Go-limited turn that restarts with a start this build cannot decode", () => + Effect.gen(function* () { + const gate = makeProviderReplayGate(["end"]); + const restarted = yield* Deferred.make(); + const { runtime, thread } = yield* resumed( + [ + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + failWith({ type: "provider.go-limit", message: goLimitMessage, status: 429 }), + { + type: "emit_inbound", + frame: { + type: "sdk.event", + event: { + id: "evt_executionstartedx", + created: 1, + type: "session.execution.started", + data: { sessionID: SESSION }, + durable: "not-an-envelope", + }, + }, + }, + event("session.text.started", { + sessionID: SESSION, + assistantMessageID: "msg_0eb735d5b001oAFVeY5jz3WD4Z", + ordinal: 0, + }), + event("session.text.ended", { + sessionID: SESSION, + assistantMessageID: "msg_0eb735d5b001oAFVeY5jz3WD4Z", + ordinal: 0, + text: "DONE", + }), + event("session.execution.succeeded", { sessionID: SESSION }, "end"), + ], + { + replayGate: gate, + // The restarted execution ends only after the read, whose end is queued first. + readUsageLimits: Deferred.await(restarted).pipe( + Effect.as(goUsage(100)), + Effect.ensuring(Effect.sync(() => gate.release("end"))), + ), + }, + ); + const terminal = yield* runtime.events.pipe( + Stream.tap((event) => + event.type === "turn_item.updated" && + event.turnItem.type === "assistant_message" && + event.turnItem.text === "DONE" + ? Deferred.succeed(restarted, undefined) + : Effect.void, + ), + Stream.filter((event) => event.type === "turn.terminal"), + Stream.runHead, + Effect.map(Option.getOrUndefined), + Effect.forkScoped, + ); + yield* runtime.startTurn(turnInput(thread)); + const ended = yield* Fiber.join(terminal); + assert.deepInclude(ended, { type: "turn.terminal", status: "completed" }); + }).pipe(Effect.scoped), + ); + it.effect("ends the turn when its terminal event is one this build cannot decode", () => Effect.gen(function* () { const { runtime, thread } = yield* resumed([ @@ -2696,6 +2970,39 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped), ); + // OpenCode already delivered this execution's end; a listing taken at reconnect does not undo it. + it.effect("ends a Go limit after a reconnect even while OpenCode still lists the session", () => + Effect.gen(function* () { + const gate = makeProviderReplayGate(["reconnected"]); + const { runtime, thread } = yield* resumed( + [ + out("session.prompt", { sessionID: SESSION, text: "" }), + promptAccepted, + failWith({ type: "provider.go-limit", message: goLimitMessage, status: 429 }), + { type: "runtime_exit", status: "success" } as const, + out("event.subscribe"), + out("session.active"), + replyData("session.active", { [SESSION]: { type: "running" } }), + ...openRequests([]), + marker("reconnected"), + ], + { + replayGate: gate, + // The read ends after the reconnect has listed the session as running. + readUsageLimits: Effect.promise(() => gate.waitForReached("reconnected")).pipe( + Effect.tap(() => Effect.sync(() => gate.release("reconnected"))), + Effect.as(goUsage(100)), + ), + }, + ); + const terminal = yield* terminalOf(runtime).pipe(Effect.forkScoped); + yield* runtime.startTurn(turnInput(thread)); + const ended = yield* Fiber.join(terminal); + assert.equal(ended?.failure?.class, "usage_limit"); + assert.equal(ended?.failure?.resetAt, "2026-11-06T04:12:00.000Z"); + }).pipe(Effect.scoped), + ); + it.effect("keeps a turn still running after a reconnect open for its next events", () => Effect.gen(function* () { const { runtime, thread } = yield* resumed([ diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts index 2714b3a360d1..0d648d43d04e 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts @@ -6,6 +6,7 @@ import { ProviderSessionId, ThreadId, type ProviderReplayTranscript, + type ServerProviderUsageLimits, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; @@ -257,13 +258,17 @@ const makeReplayAdapter = ( readonly onPrompt?: (input: unknown) => void; readonly replayGate?: ProviderReplayGate; readonly borrowers?: { current: number }; + readonly readUsageLimits?: Effect.Effect; }, ) => Effect.gen(function* () { const server = yield* replayServer(transcript, options); - return yield* OpenCode2Adapter.make(ProviderInstanceId.make("opencode")).pipe( - Effect.provideService(OpenCode2Server.OpenCode2Server, server), - ); + return yield* OpenCode2Adapter.make( + ProviderInstanceId.make("opencode"), + options?.readUsageLimits === undefined + ? undefined + : { readUsageLimits: options.readUsageLimits }, + ).pipe(Effect.provideService(OpenCode2Server.OpenCode2Server, server)); }); const layerReplayHost = TestProviderHost.layer().pipe( @@ -314,6 +319,7 @@ export const openCode2ReplayRuntime = ( readonly attachmentsDir?: string; readonly resolveExtractedTextPath?: (path: string) => Effect.Effect; readonly fileSystem?: FileSystem.FileSystem; + readonly readUsageLimits?: Effect.Effect; }, ) => Effect.gen(function* () { diff --git a/packages/provider-opencode/src/server/adapter.ts b/packages/provider-opencode/src/server/adapter.ts index a1b5ccb008eb..be0c3f10cd3a 100644 --- a/packages/provider-opencode/src/server/adapter.ts +++ b/packages/provider-opencode/src/server/adapter.ts @@ -2664,6 +2664,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function (state): state is OpenCodeThreadState => state !== undefined, ); const error = event.properties.error; + // OpenCode 1 does not say whose limit a 429 is, so it gets no Go window reset time. const usageLimit = error?.name === "APIError" && error.data.statusCode === 429; const failure = makeProviderFailure({ message: openCodeErrorMessage(event), diff --git a/packages/provider-opencode/src/server/driver.ts b/packages/provider-opencode/src/server/driver.ts index 5c60ccfe4327..a4f0b667d49f 100644 --- a/packages/provider-opencode/src/server/driver.ts +++ b/packages/provider-opencode/src/server/driver.ts @@ -325,11 +325,21 @@ export const OpenCodeDriver: ProviderDriver< yield* OpenCode2Client.make.pipe(Effect.provideService(HttpClient.HttpClient, httpClient)), ), ); + const checkUsageLimits = readOpenCodeGoUsageLimits({ + enabled: effectiveConfig.enabled, + serverUrl: effectiveConfig.serverUrl, + environment: processEnv, + }).pipe( + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, pathService), + Effect.provideService(HttpClient.HttpClient, httpClient), + Effect.provideService(Digest.Digest, digest), + ); const orchestrationAdapter = selectOpenCodeRuntimeAdapter({ probe: runtimeProbe, resolve: resolveRuntime, v1: openCodeV1Adapter, - v2: yield* OpenCode2AdapterV2.make(instanceId).pipe( + v2: yield* OpenCode2AdapterV2.make(instanceId, { readUsageLimits: checkUsageLimits }).pipe( Effect.provideService(OpenCode2Server.OpenCode2Server, openCode2Server), ), }); @@ -385,16 +395,6 @@ export const OpenCodeDriver: ProviderDriver< ), ); - const checkUsageLimits = readOpenCodeGoUsageLimits({ - enabled: effectiveConfig.enabled, - serverUrl: effectiveConfig.serverUrl, - environment: processEnv, - }).pipe( - Effect.provideService(FileSystem.FileSystem, fileSystem), - Effect.provideService(Path.Path, pathService), - Effect.provideService(HttpClient.HttpClient, httpClient), - Effect.provideService(Digest.Digest, digest), - ); const checkProvider = Effect.all( { provider: checkOpenCodeProviderStatus( diff --git a/packages/provider-opencode/src/server/usageLimits.test.ts b/packages/provider-opencode/src/server/usageLimits.test.ts new file mode 100644 index 000000000000..3791b78dc345 --- /dev/null +++ b/packages/provider-opencode/src/server/usageLimits.test.ts @@ -0,0 +1,84 @@ +import type { ServerProviderUsageLimits, ServerProviderUsageWindow } from "@t3tools/contracts"; +import { describe, expect, it } from "vite-plus/test"; + +import { openCodeGoLimitResetAt } from "./usageLimits.ts"; + +const ROLLING_RESET = "2026-11-06T04:12:00.000Z"; +const WEEKLY_RESET = "2026-11-09T00:00:00.000Z"; +const MONTHLY_RESET = "2026-12-01T00:00:00.000Z"; + +const usage = (percents: { + readonly rolling: number; + readonly weekly: number; + readonly monthly: number; +}): ServerProviderUsageLimits => { + const window = ( + id: string, + kind: ServerProviderUsageWindow["kind"], + usedPercent: number, + resetsAt: string, + ): ServerProviderUsageWindow => ({ id, kind, label: id, usedPercent, resetsAt }); + return { + checkedAt: "2026-11-06T01:49:10.000Z", + windows: [ + window("go_rolling", "session", percents.rolling, ROLLING_RESET), + window("go_weekly", "weekly", percents.weekly, WEEKLY_RESET), + window("go_monthly", "monthly", percents.monthly, MONTHLY_RESET), + ], + }; +}; + +describe("openCodeGoLimitResetAt", () => { + it.each([ + ["OpenCode Go 5 hour usage limit reached.", ROLLING_RESET], + ["OpenCode Go weekly usage limit reached.", WEEKLY_RESET], + ["OpenCode Go monthly usage limit reached.", MONTHLY_RESET], + ])("resumes %s when that window resets", (message, expected) => { + expect(openCodeGoLimitResetAt(message, usage({ rolling: 90, weekly: 60, monthly: 30 }))).toBe( + expected, + ); + }); + + it("waits for every spent window, not only the one the error names", () => { + expect( + openCodeGoLimitResetAt( + "OpenCode Go 5 hour usage limit reached.", + usage({ rolling: 100, weekly: 100, monthly: 30 }), + ), + ).toBe(WEEKLY_RESET); + }); + + it.each([ + "OpenCode Go weekly usage limit reached for this session.", + "OpenCode Go 5 hour and weekly usage limits reached.", + ])("waits for the latest window %s names while usage lags", (message) => { + expect(openCodeGoLimitResetAt(message, usage({ rolling: 40, weekly: 98, monthly: 30 }))).toBe( + WEEKLY_RESET, + ); + }); + + it("uses the spent windows when the error does not name one", () => { + expect( + openCodeGoLimitResetAt( + "Subscription quota exceeded.", + usage({ rolling: 100, weekly: 60, monthly: 30 }), + ), + ).toBe(ROLLING_RESET); + }); + + it("has no reset when no window is named or spent", () => { + expect( + openCodeGoLimitResetAt( + "Subscription quota exceeded.", + usage({ rolling: 90, weekly: 60, monthly: 30 }), + ), + ).toBeNull(); + expect( + openCodeGoLimitResetAt("OpenCode Go 5 hour usage limit reached.", { + checkedAt: "2026-11-06T01:49:10.000Z", + windows: [], + unavailable: { reason: "probeFailed" }, + }), + ).toBeNull(); + }); +}); diff --git a/packages/provider-opencode/src/server/usageLimits.ts b/packages/provider-opencode/src/server/usageLimits.ts index 03488e515a79..f8fef0c13445 100644 --- a/packages/provider-opencode/src/server/usageLimits.ts +++ b/packages/provider-opencode/src/server/usageLimits.ts @@ -1,6 +1,6 @@ import * as NodeSqlite from "node:sqlite"; -import type { ServerProviderUsageWindow } from "@t3tools/contracts"; +import type { ServerProviderUsageLimits, ServerProviderUsageWindow } from "@t3tools/contracts"; import * as Digest from "effect/crypto/Digest"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -83,6 +83,36 @@ const readStoredApiKey = (databasePath: string, integrationId: "opencode" | "ope catch: (cause) => new OpenCodeCredentialReadError({ databasePath, cause }), }).pipe(Effect.orElseSucceed(() => undefined)); +// Matched against OpenCode's English error text ("OpenCode Go 5 hour usage limit reached"). +const GO_LIMIT_WINDOWS = [ + { id: "go_rolling", pattern: /\b5[ -]?hours?\b|\brolling\b/i }, + { id: "go_weekly", pattern: /\bweek(?:ly)?\b/i }, + { id: "go_monthly", pattern: /\bmonth(?:ly)?\b/i }, +] as const; + +/** + * OpenCode Go names the spent window in its limit error but not when it resets. The + * turn can run again once every window the error names and every spent window has + * reset, so a message naming several windows waits for the latest. + */ +export function openCodeGoLimitResetAt( + message: string, + limits: ServerProviderUsageLimits, +): string | null { + const named = new Set( + GO_LIMIT_WINDOWS.filter((window) => window.pattern.test(message)).map((window) => window.id), + ); + let latest: string | null = null; + for (const window of limits.windows) { + if (!named.has(window.id) && window.usedPercent < 100) continue; + if (window.resetsAt === undefined) continue; + if (latest === null || Date.parse(window.resetsAt) > Date.parse(latest)) { + latest = window.resetsAt; + } + } + return latest; +} + /** External OpenCode servers own their credentials; never read the host's account for them. */ export const readOpenCodeGoUsageLimits = Effect.fn("readOpenCodeGoUsageLimits")(function* (input: { readonly enabled: boolean; diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index 19673f2597fe..a685a656d84c 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -42,6 +42,7 @@ import { type ProviderInstanceId, type RunId, type RuntimeRequestId, + type ServerProviderUsageLimits, } from "@t3tools/contracts"; import type * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; @@ -95,6 +96,7 @@ import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter" import { turnScopedSelectionTransition } from "@t3tools/provider-core/server/selectionTransition"; import { OPENCODE_PROVIDER, openCodePermissionRequestKind } from "../adapter.ts"; import { openCodeCommandFailed, openCodeToolTurnItem } from "../toolItems.ts"; +import { openCodeGoLimitResetAt } from "../usageLimits.ts"; const NativeGoal = Schema.NullOr( Schema.Struct({ @@ -259,6 +261,8 @@ interface ActiveTurn { readonly steers: Set; readonly settledInbox: Set; heldEnd: TurnTerminal | undefined; + /** Set while a failed Go turn reads its reset time off the event lock; a new execution clears it. */ + limitRead: object | undefined; awaitingStart: boolean; /** Scope child tool IDs because model-generated IDs can repeat parent IDs. */ readonly scope: string; @@ -587,6 +591,7 @@ const LOST_BACKGROUND = "T3 Code lost its connection to OpenCode while this subagent ran, so its result is not shown."; const INVENTORY_TIMEOUT = "5 seconds"; const ACTIVE_CHECK_TIMEOUT = "5 seconds"; +const USAGE_READ_TIMEOUT = "10 seconds"; // Other failures may still have admitted the prompt. const CLEAR_PROMPT_REJECTIONS: ReadonlySet = new Set([ "InvalidRequestError", @@ -710,7 +715,13 @@ const turnTokenUsage = (turn: ActiveTurn, status: OrchestrationV2ProviderTurn["s }; }; -export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: ProviderInstanceId) { +export const make = Effect.fn("OpenCode2Adapter.make")(function* ( + instanceId: ProviderInstanceId, + options?: { + /** OpenCode Go's usage windows, which carry the reset times its limit errors omit. */ + readonly readUsageLimits?: Effect.Effect; + }, +) { const server = yield* OpenCode2Server.OpenCode2Server; const idAllocator = yield* IdAllocator.IdAllocatorV2; const host = yield* ProviderHost.ProviderHost; @@ -1366,6 +1377,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr steers: new Set(), settledInbox: new Set(), heldEnd: undefined, + limitRead: undefined, awaitingStart: options.awaitingStart, scope: options.scope, usedSubagents: false, @@ -2384,6 +2396,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr return yield* updateRetry(state, turn, event.data, event.data.at - event.created); case "session.execution.started": turn.heldEnd = undefined; + turn.limitRead = undefined; return; case "session.execution.succeeded": return yield* endExecution(state, turn, { @@ -2396,19 +2409,60 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr // OpenCode types an empty balance as provider.quota too, but it answers // that with HTTP 402 and it never resets. Usage windows and budgets are 429. const claudeLimit = claudeUsageLimit(message); + // Newer OpenCode builds report a spent Go window as provider.go-limit; + // older ones sent provider.quota with a GoUsageLimitError body. + const goLimit = + (type === "provider.go-limit" && status !== 402) || + (type === "provider.quota" && + status === 429 && + response?.body.includes("GoUsageLimitError") === true); const usageLimit = + goLimit || (type === "provider.quota" && status !== 402) || type === "provider.rate-limit" || claudeLimit !== undefined; - return yield* endExecution(state, turn, { + const failed = (resetAt: string | null): TurnTerminal => ({ status: "failed", failure: makeProviderFailure({ message: readableProviderErrorMessage(message, status, response?.body), code: type, class: usageLimit ? "usage_limit" : "provider_error", - resetAt: claudeLimit?.resetAt ?? null, + resetAt, }), }); + const readUsageLimits = options?.readUsageLimits; + if (claudeLimit !== undefined || !goLimit || readUsageLimits === undefined) { + return yield* endExecution(state, turn, failed(claudeLimit?.resetAt ?? null)); + } + // Go errors omit the reset time, so it comes from Go's usage windows. That read + // is a network call, so the turn ends after it, off the lock all threads share. + const read = {}; + turn.limitRead = read; + const endAfterRead = (resetAt: string | null) => + lock.withPermit( + Effect.suspend(() => { + if (state.active !== turn || turn.limitRead !== read) return Effect.void; + turn.limitRead = undefined; + // A Stop sent during the read wins, so the stopped thread does not resume at the reset. + const end = turn.interrupted ? { status: "interrupted" as const } : failed(resetAt); + return endExecution(state, turn, end).pipe( + Effect.andThen(syncRosterAncestors(state)), + ); + }), + ); + yield* Effect.forkIn( + readUsageLimits.pipe( + Effect.timeout(USAGE_READ_TIMEOUT), + Effect.exit, + Effect.flatMap((exit) => + endAfterRead( + Exit.isSuccess(exit) ? openCodeGoLimitResetAt(message, exit.value) : null, + ), + ), + ), + sessionScope, + ); + return; } default: return; @@ -2682,6 +2736,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr if (turn !== undefined) { turn.awaitingStart = false; turn.heldEnd = undefined; + turn.limitRead = undefined; return; } } @@ -2999,7 +3054,8 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr for (const childId of strays.keys()) if (!(childId in active)) strays.delete(childId); for (const [sessionId, state] of running) { const turn = state.active; - if (turn === undefined) continue; + // Its end already arrived; the pending usage read finishes it. + if (turn === undefined || turn.limitRead !== undefined) continue; turn.missedEvents = true; const ended = yield* backfill(sessionId, state); if (state.active !== turn) continue; @@ -4287,7 +4343,11 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr yield* lock.withPermit(finishTurn(state, { status: "interrupted" })); return yield* cancelStrandedSteers(state); } - if (!reply.value.interrupted && state.active === turn && turn.unsent) { + if ( + !reply.value.interrupted && + state.active === turn && + (turn.unsent || turn.limitRead !== undefined) + ) { return yield* lock.withPermit(finishTurn(state, { status: "interrupted" })); } if (!reply.value.interrupted && state.active === turn) {