diff --git a/src/adapters/base.ts b/src/adapters/base.ts index fff735ed98..0efce92e5b 100644 --- a/src/adapters/base.ts +++ b/src/adapters/base.ts @@ -31,11 +31,34 @@ export interface IncomingMeta { * behind it (#4546). */ sendBudget?: RequestExecutionBudget; + /** + * Physical-send observations for runTurn adapters. Without the same callback carried by + * AdapterFetchContext, an adapter-owned replay spends the shared budget but remains absent + * from the request's sendCount. + */ + onPhysicalSend?: (send: { ordinal: number; recovery?: AttemptRecoveryKind }) => void; + /** + * Recovery refusals for runTurn adapters. A refused replay is not a send, so this separate + * channel explains why recovery stopped without inflating physical-send telemetry. + */ + onRecoveryWithheld?: (withheld: { reason: AttemptRecoveryWithheld }) => void; } export interface ProviderAdapter { name: string; + /** + * This adapter reports every physical inference send through `IncomingMeta.onPhysicalSend`, + * including its first. + * + * The caller normally logs the first send before handing control over, which is correct for a + * transport whose sends it can see. An adapter that admits its own sends through the shared + * budget can have that first send refused, and a send logged before admission is a send the + * log claims and the wire never made. Setting this moves the first send's accounting to the + * boundary where it is actually dispatched. + */ + reportsPhysicalSends?: boolean; + /** * Convert an already-read provider HTTP error into client-safe text. This hook must be pure and * return fully redacted output: callers may pass untrusted provider headers and payload text. diff --git a/src/adapters/devin.ts b/src/adapters/devin.ts index 0ff3b7da33..082cde39d0 100644 --- a/src/adapters/devin.ts +++ b/src/adapters/devin.ts @@ -15,6 +15,7 @@ import { getCachedCatalog, type CacheEntry } from "./devin/cloud-direct/catalog" import { collapseDevinModelUid } from "./devin/live-models"; import { buildNonOpenAIToolCatalogNudgeForTools } from "./tool-catalog-nudge"; import { DEVIN_DEFAULT_API_SERVER, resolveDevinApiServer } from "../oauth/devin"; +import { SendBudgetExhaustedError } from "../lib/upstream-retry"; /** * Combine two usage frames from one turn by keeping the larger count per field. @@ -490,6 +491,10 @@ export function createDevinAdapter( return { name: "devin", + // Every GetChatMessage send, including the first, is admitted through the shared budget and + // reported from the executor that dispatches it. The caller therefore leaves the first + // send's accounting here rather than logging it before admission can refuse it. + reportsPhysicalSends: true, buildRequest() { return { @@ -584,6 +589,13 @@ export function createDevinAdapter( ...(typeof parsed.options.topP === "number" ? { topP: parsed.options.topP } : {}), }, signal: incoming.abortSignal, + }, { + execution: { + executor: incoming.providerFetch, + sendBudget: incoming.sendBudget, + onPhysicalSend: incoming.onPhysicalSend, + onRecoveryWithheld: incoming.onRecoveryWithheld, + }, })) { if (incoming.abortSignal?.aborted) { // Emitting nothing here left the bridge to synthesize adapter_eof. @@ -662,6 +674,9 @@ export function createDevinAdapter( emit({ type: "error", message: DEVIN_CLIENT_CLOSED_MESSAGE, status: 499, retryable: false, ...(usage ? { usage } : {}) }); return; } + // The Responses boundary already maps this local refusal to its structured 429 code. + // Converting it to an adapter event would make it an ordinary untyped upstream error. + if (error instanceof SendBudgetExhaustedError) throw error; const message = error instanceof CloudChatError ? ("Devin cloud error" + (error.code ? " " + error.code : "") + ": " + error.message) : error instanceof Error ? error.message : String(error); diff --git a/src/adapters/devin/cloud-direct/chat.ts b/src/adapters/devin/cloud-direct/chat.ts index 794c5cddf2..34547ecc19 100644 --- a/src/adapters/devin/cloud-direct/chat.ts +++ b/src/adapters/devin/cloud-direct/chat.ts @@ -1055,6 +1055,8 @@ export interface CloudChatRequest { catalog?: CacheEntry | null; /** Abort signal — closes the fetch stream. */ signal?: AbortSignal; + /** Executor for the inference POST only; catalog and JWT RPCs retain their own transport. */ + executor?: typeof globalThis.fetch; } export class CloudChatError extends Error { @@ -1216,7 +1218,7 @@ export async function* streamChatEvents(req: CloudChatRequest): AsyncGenerator; } function replayLimit(value: number | undefined): number { @@ -59,15 +66,34 @@ export async function* streamChatEventsWithResetRetry( const sleep = options?.sleep ?? sleepWithAbort; const maxReplays = replayLimit(options?.maxReplays); const maxWaitMs = waitLimit(options?.maxWaitMs); + const execution = options?.execution; + // One sender owns the whole invocation so ordinals span the initial POST and both replays. + // Its executor remains lazy: replay admission happens after the provider-stated wait, never + // while a reservation could be held for up to an hour. + const send = execution + ? createAdapterPhysicalSend({ ...execution, abortSignal: req.signal }) + : undefined; let replays = 0; let waitedMs = 0; + let replaySourceError: CloudChatError | undefined; while (true) { // Check again after sleeping: cancellation can race with timer completion. // A pre-aborted request must not even enter a custom transport. if (req.signal?.aborted) throw abortError(req.signal); let yielded = false; try { - for await (const event of stream(req)) { + const recovery = replays > 0 ? "rate-limit-429" as const : undefined; + const attemptRequest = send + ? { + ...req, + executor: ((input, init) => send({ + url: typeof input === "string" ? input : input instanceof URL ? input.href : input.url, + ...(recovery ? { sendClass: "auth-recovery" as const, recovery } : {}), + dispatch: executor => executor(input, init), + })) as typeof globalThis.fetch, + } + : req; + for await (const event of stream(attemptRequest)) { // Latch before yielding, so a consumer-injected error is post-output. yielded = true; yield event; @@ -75,14 +101,24 @@ export async function* streamChatEventsWithResetRetry( return; } catch (error) { if (req.signal?.aborted) throw abortError(req.signal); - const waitSec = !yielded + if (!yielded && error instanceof SendBudgetExhaustedError && replaySourceError) { + // The provider's refusal is the real upstream answer. A local cap can withhold its + // recovery, but replacing the 429 would erase the status and stated reset metadata. + execution?.onRecoveryWithheld?.({ reason: "retry-send-budget" }); + throw replaySourceError; + } + const retryableError = !yielded && error instanceof CloudChatError && error.status === 429 - ? parseRetryAfterFromMessage(error.message) + ? error + : undefined; + const waitSec = retryableError + ? parseRetryAfterFromMessage(retryableError.message) : undefined; const waitMs = waitSec === undefined ? undefined : waitSec * 1000; if ( - waitMs === undefined + retryableError === undefined + || waitMs === undefined || replays >= maxReplays || waitMs > maxWaitMs - waitedMs ) { @@ -91,6 +127,7 @@ export async function* streamChatEventsWithResetRetry( throw error; } replays += 1; + replaySourceError = retryableError; // Charge the complete scheduled wait once, before sleeping. This is a // sleep allowance, not a wall-clock deadline on generation or timer // scheduling: waking a few milliseconds late must not reject an already diff --git a/src/server/responses/request-send-budget.ts b/src/server/responses/request-send-budget.ts index 7af8b25598..9250162da2 100644 --- a/src/server/responses/request-send-budget.ts +++ b/src/server/responses/request-send-budget.ts @@ -109,8 +109,12 @@ export function createResponsesSendBudget( const noteAdapterPhysicalSend = ( inputTokens: number | undefined, send: { ordinal: number; recovery?: AttemptRecoveryKind }, + options: { readonly includeFirst?: boolean } = {}, ): void => { - if (send.ordinal <= 1) return; + // Ordinal 1 is skipped because the caller normally records it before dispatch. An adapter + // that reports every send asks for it to be counted here instead, so that the first send is + // logged where it actually happens rather than before admission could still refuse it. + if (send.ordinal <= 1 && options.includeFirst !== true) return; noteAttemptSend(logCtx.activeAttempt, inputTokens, send.recovery); }; /** diff --git a/src/server/responses/run-turn-execution.ts b/src/server/responses/run-turn-execution.ts index 0eda091879..dd693a3013 100644 --- a/src/server/responses/run-turn-execution.ts +++ b/src/server/responses/run-turn-execution.ts @@ -79,7 +79,11 @@ export async function executeResponsesRunTurn( >, sendBudgetState: Pick< ResponsesSendBudget, - "adapterDispatchBudget" | "reserveCredentialHop" | "pendingHopPermit" + | "adapterDispatchBudget" + | "noteAdapterPhysicalSend" + | "noteAdapterRecoveryWithheld" + | "reserveCredentialHop" + | "pendingHopPermit" >, completionPolicy: Pick, ): Promise { @@ -100,7 +104,12 @@ export async function executeResponsesRunTurn( rememberKiroDeliveredFinalAnswer, responseStateOptions, } = requestState; - const { adapterDispatchBudget, reserveCredentialHop } = sendBudgetState; + const { + adapterDispatchBudget, + noteAdapterPhysicalSend, + noteAdapterRecoveryWithheld, + reserveCredentialHop, + } = sendBudgetState; const { emptyCompletionGuardEnabled } = completionPolicy; const { cancelResponseCompletion, @@ -146,7 +155,11 @@ export async function executeResponsesRunTurn( await waitForProviderRequestSlot(route.providerName, route.provider, route.modelId, runTurnAbort.signal); } await refreshRunTurnSelection(); - transportState.noteRoutedAttemptSend(logCtx.usageLogInputTokens, recovery); + // An adapter that reports its own sends accounts for the first one at the boundary that + // dispatches it. Logging here would claim a send that the adapter's own budget can still + // refuse, which is exactly what happens once earlier recovery has spent the allowance. + const reportsOwnSends = transportState.runTurnAdapter.reportsPhysicalSends === true; + if (!reportsOwnSends) transportState.noteRoutedAttemptSend(logCtx.usageLogInputTokens, recovery); const runTurnProviderFetch = providerFetch( route.provider, options.codexWsRuntimeIdentity, @@ -169,6 +182,14 @@ export async function executeResponsesRunTurn( // The only way the request budget reaches a transport the adapter owns. Without it // a Cursor turn's inner ladder was three physical sends the cap read as one. ...(adapterDispatchBudget ? { sendBudget: adapterDispatchBudget } : {}), + onPhysicalSend: send => noteAdapterPhysicalSend( + logCtx.usageLogInputTokens, + // The attempt's own recovery kind still labels its first send when the adapter + // does not supply one of its own. + { ...send, ...(send.recovery ?? recovery ? { recovery: send.recovery ?? recovery } : {}) }, + { includeFirst: reportsOwnSends }, + ), + onRecoveryWithheld: noteAdapterRecoveryWithheld, }, targetQueue.push, ); diff --git a/structure/adapters/registry.md b/structure/adapters/registry.md index 34d04d9eb6..75238008df 100644 --- a/structure/adapters/registry.md +++ b/structure/adapters/registry.md @@ -39,6 +39,9 @@ Some adapters share another adapter's routed-tool semantics while retaining inde does not. `devin-cli` survives only as a deprecated alias — `ocx login devin-cli` routes to `devin`, and a startup merge migration rewrites any saved row still keyed under the old provider id, so the registry carries one Devin provider, not two. + Its `GetChatMessage` inference POSTs, including the two bounded pre-output stated-reset + replays, pass through the request's provider executor and shared physical-send budget. + Catalog and JWT RPCs remain adapter support traffic rather than inference sends. `AdapterFactoryContext.providerId` still tells the shared adapter which configured row it is serving: the Cognition tenant is recorded on the credential, not in the registry, so the adapter has to know the row before it can resolve a host. That adapter advertises bare local diff --git a/structure/providers-and-adapters.md b/structure/providers-and-adapters.md index 1bc9a86aaf..cf5000eb15 100644 --- a/structure/providers-and-adapters.md +++ b/structure/providers-and-adapters.md @@ -17,6 +17,7 @@ the [bounded ingestion contract](transports/inventory.md#bounded-response-ingest | `src/adapters/google.ts` | Gemini bridge. | | `src/adapters/azure.ts` | Azure OpenAI bridge. | | `src/adapters/cursor.ts`, `src/adapters/cursor/` | Cursor protobuf transport: discovery, request builder, event decoding, MCP, thread continuity, native-exec policy. | +| `src/adapters/devin.ts`, `src/adapters/devin/cloud-direct/` | Devin runTurn transport over Cognition Connect-RPC. `GetChatMessage` uses the Responses provider executor and shared physical-send budget; catalog and JWT support RPCs remain outside inference-send accounting. | | `src/adapters/kiro.ts` and `src/adapters/kiro/` | Kiro event/tool/thinking/truncation/retry handling. The original path is a facade over leaves for wire identity, reasoning, conversation state, token estimation, payload assembly, streaming, and the adapter. | | `src/adapters/mimo-free.ts` | Mimo Free transport (client identity + JWT). | | `src/adapters/image.ts`, `src/adapters/anthropic-image-guard.ts`, `src/adapters/anthropic-image-normalize.ts`, `src/adapters/anthropic-image-codec.ts` | Image conversion for adapter ingress and Anthropic-specific normalization/limits. An image's ladder position is pinned to its own identity (content hash + media type), so appending a newer image cannot re-encode older ones and bust Anthropic's prompt prefix cache (#4532). | diff --git a/structure/transports/responses.md b/structure/transports/responses.md index 1d954bcc44..b6e45cd813 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -941,14 +941,23 @@ The hop pays for a replay that some *other* layer dispatches, so which layer set reservation follows the dispatcher, not the ladder. A helper-routed replay reports the same physical send back through `onSendsConsumed`; that is what `countedExternally: true` names, and the reporter's first send settles the pending booking instead of adding a second charge. An adapter -that owns its transport — Kiro's reset ladder, Cursor's transport ladder — reserves once per -physical send instead, so no reporter ever arrives. Those ladders are handed +that owns its transport — Kiro's reset ladder, Cursor's transport ladder, or Devin's bounded +pre-output stated-reset replay — reserves once per physical send instead, so no reporter ever +arrives. Those ladders are handed `adapterDispatchBudget`, a live delegating view of the same budget that spends a permit passed down through `pendingHopPermit` on the adapter's first reservation and closes the booking through `permit.assumeCharge()`. Letting both charge is how one physical send became two charges, and how a spent allowance answered a 429 with a synthetic error instead of the rate limit it was recovering from (#4709). +`run-turn-execution.ts` passes the same physical-send and recovery-withheld observers used by the +request-building adapter path. Devin builds one `createAdapterPhysicalSend` for the whole +`GetChatMessage` invocation, so its initial POST and at most two same-target replays report ordinals +1, 2, and 3. The outer runTurn attempt already records ordinal 1, and the shared observer therefore +adds only ordinals above 1 to `sendCount`; the execution budget still reserves every ordinal. A +replay reserves only after its server-stated wait. If admission is refused, no inference I/O occurs, +`retry-send-budget` is recorded, and the preceding provider 429 remains the returned error. + Confirmation happens at the dispatch boundary rather than at the rotation. `adapter-dispatch.ts` passes an `onDispatch` callback that the rebuild invokes immediately before the wire, and skips it when the adapter owns dispatch: settling there first would hand that adapter a dead permit, which diff --git a/tests/adapters/adapter-inner-send-budget-wiring.test.ts b/tests/adapters/adapter-inner-send-budget-wiring.test.ts index 4b86734113..9885c73af9 100644 --- a/tests/adapters/adapter-inner-send-budget-wiring.test.ts +++ b/tests/adapters/adapter-inner-send-budget-wiring.test.ts @@ -9,6 +9,8 @@ import { } from "../../src/adapters/cursor/thread-continuity"; import type { CursorTransport } from "../../src/adapters/cursor/transport"; import { createKiroAdapter } from "../../src/adapters/kiro"; +import { createDevinAdapter, DEVIN_API_SERVER } from "../../src/adapters/devin"; +import { setCachedCatalogForTests } from "../../src/adapters/devin/cloud-direct/catalog"; import { resetKiroThrottleStateForTests } from "../../src/adapters/kiro-retry"; import type { AdapterFetchContext } from "../../src/adapters/base"; import { encodeMessage } from "../../src/lib/eventstream-decoder"; @@ -122,6 +124,121 @@ describe("Cursor runTurn and the request send budget", () => { }); }); +describe("Devin runTurn execution wiring", () => { + test("forwards the provider executor, shared budget, and physical-send observer", async () => { + const previousHome = process.env.OPENCODEX_HOME; + const previousJwtFlag = process.env.OPENCODEX_DEVIN_SEND_USER_JWT; + const home = mkdtempSync(join(tmpdir(), "devin-send-wiring-")); + process.env.OPENCODEX_HOME = home; + delete process.env.OPENCODEX_DEVIN_SEND_USER_JWT; + const apiKey = "devin-test-key"; + setCachedCatalogForTests({ + apiKey, + host: DEVIN_API_SERVER, + fetchedAt: Date.now(), + byUid: new Map([["swe-2", { modelUid: "swe-2", label: "SWE-2", disabled: false }]]), + }); + const budget = budgetOf(1); + const observed: Array<{ ordinal: number; recovery?: string }> = []; + const urls: string[] = []; + const events: AdapterEvent[] = []; + const adapter = createDevinAdapter({ + adapter: "devin", baseUrl: DEVIN_API_SERVER, apiKey, + } as unknown as OcxProviderConfig); + + try { + await adapter.runTurn?.( + { + modelId: "swe-2", stream: true, options: {}, + context: { messages: [{ role: "user", content: "hi" }] }, + } as unknown as OcxParsedRequest, + { + headers: new Headers(), + translatorBudget: createTestTranslatorBudget(), + sendBudget: budget, + providerFetch: (async input => { + urls.push(String(input)); + return new Response("busy", { status: 500 }); + }) as typeof fetch, + onPhysicalSend: send => { observed.push(send); }, + }, + event => events.push(event), + ); + } finally { + setCachedCatalogForTests(null); + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + if (previousJwtFlag === undefined) delete process.env.OPENCODEX_DEVIN_SEND_USER_JWT; + else process.env.OPENCODEX_DEVIN_SEND_USER_JWT = previousJwtFlag; + removeTreeWithRetry(home); + } + + expect(urls).toHaveLength(1); + expect(urls[0]).toContain("GetChatMessage"); + expect(budget.used).toBe(1); + expect(observed).toEqual([{ ordinal: 1 }]); + expect(events.at(-1)).toMatchObject({ type: "error", status: 500 }); + }); + + test("a first send the budget refuses makes no request and reports no send", async () => { + // The accounting defect this pins: the caller used to log this turn's first send before the + // adapter ran, so an allowance already spent by earlier recovery produced a logged send the + // wire never made. Nothing is dispatched here, so nothing may be observed either. + const previousHome = process.env.OPENCODEX_HOME; + const previousJwtFlag = process.env.OPENCODEX_DEVIN_SEND_USER_JWT; + const home = mkdtempSync(join(tmpdir(), "devin-send-denied-")); + process.env.OPENCODEX_HOME = home; + delete process.env.OPENCODEX_DEVIN_SEND_USER_JWT; + const apiKey = "devin-denied-key"; + setCachedCatalogForTests({ + apiKey, + host: DEVIN_API_SERVER, + fetchedAt: Date.now(), + byUid: new Map([["swe-2", { modelUid: "swe-2", label: "SWE-2", disabled: false }]]), + }); + // Nothing left to spend: the same state an earlier combo fan-out or empty-response recovery + // leaves behind before this turn starts. + const budget = budgetOf(0); + const observed: Array<{ ordinal: number; recovery?: string }> = []; + const urls: string[] = []; + const events: AdapterEvent[] = []; + const adapter = createDevinAdapter({ + adapter: "devin", baseUrl: DEVIN_API_SERVER, apiKey, + } as unknown as OcxProviderConfig); + + try { + await adapter.runTurn?.( + { + modelId: "swe-2", stream: true, options: {}, + context: { messages: [{ role: "user", content: "hi" }] }, + } as unknown as OcxParsedRequest, + { + headers: new Headers(), + translatorBudget: createTestTranslatorBudget(), + sendBudget: budget, + providerFetch: (async input => { + urls.push(String(input)); + return new Response("busy", { status: 500 }); + }) as typeof fetch, + onPhysicalSend: send => { observed.push(send); }, + }, + event => events.push(event), + ).catch(() => { /* the refusal escapes the adapter for the caller to map */ }); + } finally { + setCachedCatalogForTests(null); + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + if (previousJwtFlag === undefined) delete process.env.OPENCODEX_DEVIN_SEND_USER_JWT; + else process.env.OPENCODEX_DEVIN_SEND_USER_JWT = previousJwtFlag; + removeTreeWithRetry(home); + } + + expect(urls.filter(url => url.includes("GetChatMessage"))).toHaveLength(0); + expect(observed).toEqual([]); + expect(budget.used).toBe(0); + }); +}); + const kiroProvider = { adapter: "kiro", baseUrl: "https://runtime.us-east-1.kiro.dev", diff --git a/tests/adapters/adapter-inner-send-budget.test.ts b/tests/adapters/adapter-inner-send-budget.test.ts index 09406cb506..2da2597c03 100644 --- a/tests/adapters/adapter-inner-send-budget.test.ts +++ b/tests/adapters/adapter-inner-send-budget.test.ts @@ -6,6 +6,8 @@ import type { CursorRunRequest, CursorServerMessage } from "../../src/adapters/c import type { CursorTransport } from "../../src/adapters/cursor/transport"; import { createRequestExecutionBudget, type RequestExecutionBudgetPolicy } from "../../src/lib/request-execution-budget"; import { SendBudgetExhaustedError } from "../../src/lib/upstream-retry"; +import { CloudChatError, type CloudChatEvent, type CloudChatRequest } from "../../src/adapters/devin/cloud-direct"; +import { streamChatEventsWithResetRetry } from "../../src/adapters/devin/cloud-direct/stated-reset-retry"; /** * Adapters that retry INSIDE one adapter call are the layer a per-request cap cannot see from @@ -145,3 +147,22 @@ describe("Cursor inner retries and the request send budget", () => { expect(observed.map(send => send.recovery)).toEqual([undefined, "connection-reset"]); }); }); + +describe("Devin inner retries and the request send budget", () => { + test("an exhausted initial send escapes as the local budget error", async () => { + const request = { + apiKey: "test", apiServerUrl: "https://example.invalid", modelUid: "swe-2", messages: [], + } as unknown as CloudChatRequest; + const stream = (req: CloudChatRequest) => (async function* (): AsyncGenerator { + await req.executor!("https://example.invalid/GetChatMessage"); + throw new CloudChatError("must not replace the budget refusal", undefined, undefined, 429); + })(); + + await expect((async () => { + for await (const _event of streamChatEventsWithResetRetry(request, { + stream, + execution: { sendBudget: budgetOf(0) }, + })) { /* drain */ } + })()).rejects.toBeInstanceOf(SendBudgetExhaustedError); + }); +}); diff --git a/tests/providers/devin-adapter.test.ts b/tests/providers/devin-adapter.test.ts index a50a3d96e4..7407252db2 100644 --- a/tests/providers/devin-adapter.test.ts +++ b/tests/providers/devin-adapter.test.ts @@ -483,9 +483,8 @@ describe("devin adapter api-server host resolution (#4503)", () => { home = mkdtempSync(join(tmpdir(), "ocx-devin-host-")); process.env.OPENCODEX_HOME = home; seenUrls = []; - // This adapter's transport fetches through the global fetch — it does not - // consume IncomingMeta.providerFetch — so the stub observes every upstream - // URL the turn dispatches to. + // No providerFetch is supplied by this direct adapter test, so inference falls back to the + // global fetch alongside catalog/JWT RPCs and this stub observes every upstream URL. globalThis.fetch = (async (input: RequestInfo | URL) => { seenUrls.push(String(input)); return new Response("down", { status: 500 }); diff --git a/tests/providers/devin-stated-reset-hardening.test.ts b/tests/providers/devin-stated-reset-hardening.test.ts index 0dac8ed0e3..2618b92041 100644 --- a/tests/providers/devin-stated-reset-hardening.test.ts +++ b/tests/providers/devin-stated-reset-hardening.test.ts @@ -74,14 +74,20 @@ describe("Devin cumulative stated-reset allowance", () => { test("cancellation at the sleep-completion boundary prevents replay", async () => { const controller = new AbortController(); let calls = 0; - const stream = async function* (): AsyncGenerator { + let inferenceSends = 0; + const stream = async function* (req: CloudChatRequest): AsyncGenerator { calls += 1; + await req.executor!("https://example.invalid/GetChatMessage"); throw cap("reset in 1 second"); }; await expect(drain(streamChatEventsWithResetRetry({ ...request, signal: controller.signal }, { stream, sleep: async () => { controller.abort(); }, + execution: { + executor: (async () => { inferenceSends += 1; return new Response(); }) as typeof fetch, + }, }))).rejects.toHaveProperty("name", "AbortError"); expect(calls).toBe(1); + expect(inferenceSends).toBe(1); }); test.each([ @@ -91,13 +97,21 @@ describe("Devin cumulative stated-reset allowance", () => { { kind: "reasoning_signature", signature: "sig" }, ] as CloudChatEvent[])("never retries after an event: %j", async event => { let calls = 0; - const stream = async function* (): AsyncGenerator { + let inferenceSends = 0; + const stream = async function* (req: CloudChatRequest): AsyncGenerator { calls += 1; + await req.executor!("https://example.invalid/GetChatMessage"); yield event; throw cap("reset in 1 second"); }; - await expect(drain(streamChatEventsWithResetRetry(request, { stream }))).rejects.toThrow("reset in 1 second"); + await expect(drain(streamChatEventsWithResetRetry(request, { + stream, + execution: { + executor: (async () => { inferenceSends += 1; return new Response(); }) as typeof fetch, + }, + }))).rejects.toThrow("reset in 1 second"); expect(calls).toBe(1); + expect(inferenceSends).toBe(1); }); test("explicit zero disables waiting; overlarge overrides stay bounded", () => { diff --git a/tests/providers/devin-stated-reset-retry.test.ts b/tests/providers/devin-stated-reset-retry.test.ts index 18e7ef9ce6..305372ad0e 100644 --- a/tests/providers/devin-stated-reset-retry.test.ts +++ b/tests/providers/devin-stated-reset-retry.test.ts @@ -7,6 +7,7 @@ import { describe, expect, test } from "bun:test"; import { CloudChatError, type CloudChatEvent, type CloudChatRequest } from "../../src/adapters/devin/cloud-direct"; import { streamChatEventsWithResetRetry } from "../../src/adapters/devin/cloud-direct/stated-reset-retry"; +import { createRequestExecutionBudget } from "../../src/lib/request-execution-budget"; const REQ = { apiKey: "k", apiServerUrl: "https://example.invalid", modelUid: "swe-2", messages: [] } as unknown as CloudChatRequest; @@ -27,6 +28,77 @@ async function drain(source: AsyncGenerator): Promise { + test("admits and reports every inference POST through one shared budget", async () => { + let spendReservations = 0; + let spendRefunds = 0; + const budget = createRequestExecutionBudget({ + maxTotalModelSends: 3, baseSendAllowance: 3, finalRecoveryAllowance: 0, + maxAlternateTargetSends: 0, maxTargetTransitions: 0, + }, "devin-reset-accounting", { + charge: () => { spendReservations += 1; return true; }, + refund: () => { spendRefunds += 1; }, + }); + const sends: number[] = []; + const usedDuringWait: number[] = []; + const observed: Array<{ ordinal: number; recovery?: string }> = []; + let attempts = 0; + const stream = (req: CloudChatRequest) => (async function* (): AsyncGenerator { + attempts += 1; + await req.executor!("https://example.invalid/GetChatMessage"); + if (attempts < 3) throw new CloudChatError("reset in 1 second", "resource_exhausted", "t", 429); + yield { kind: "finish", reason: "stop" } as CloudChatEvent; + })(); + + await drain(streamChatEventsWithResetRetry(REQ, { + stream, + sleep: async () => { usedDuringWait.push(budget.used); }, + execution: { + executor: (async () => { sends.push(sends.length + 1); return new Response(); }) as typeof fetch, + sendBudget: budget, + onPhysicalSend: send => { observed.push(send); }, + }, + })); + + expect(sends).toEqual([1, 2, 3]); + expect(budget.used).toBe(3); + expect(usedDuringWait).toEqual([1, 2]); + expect(spendReservations).toBe(3); + expect(spendRefunds).toBe(0); + expect(observed).toEqual([ + { ordinal: 1 }, + { ordinal: 2, recovery: "rate-limit-429" }, + { ordinal: 3, recovery: "rate-limit-429" }, + ]); + }); + + test("a refused replay performs no inference I/O and preserves the provider 429", async () => { + const budget = createRequestExecutionBudget({ + maxTotalModelSends: 1, baseSendAllowance: 1, finalRecoveryAllowance: 0, + maxAlternateTargetSends: 0, maxTargetTransitions: 0, + }, "devin-reset-refusal"); + const refusal = new CloudChatError("Your limit will reset in 1 second", "resource_exhausted", "trace", 429); + const withheld: string[] = []; + let inferenceSends = 0; + const stream = (req: CloudChatRequest) => (async function* (): AsyncGenerator { + await req.executor!("https://example.invalid/GetChatMessage"); + throw refusal; + })(); + + await expect(drain(streamChatEventsWithResetRetry(REQ, { + stream, + sleep: async () => {}, + execution: { + executor: (async () => { inferenceSends += 1; return new Response(); }) as typeof fetch, + sendBudget: budget, + onRecoveryWithheld: event => { withheld.push(event.reason); }, + }, + }))).rejects.toBe(refusal); + + expect(inferenceSends).toBe(1); + expect(budget.used).toBe(1); + expect(withheld).toEqual(["retry-send-budget"]); + }); + test("waits the stated delay and replays a zero-event 429", async () => { const waits: number[] = []; let calls = 0; diff --git a/tests/responses/responses-send-budget-counts.test.ts b/tests/responses/responses-send-budget-counts.test.ts index 10b08864f3..d6119caa0f 100644 --- a/tests/responses/responses-send-budget-counts.test.ts +++ b/tests/responses/responses-send-budget-counts.test.ts @@ -8,6 +8,14 @@ import { handleResponses } from "../../src/server/responses/core"; import { COMBO_TARGET_BASE_SENDS, comboExecutionBudgetPolicy } from "../../src/server/responses/core-combo"; import type { RequestLogContext } from "../../src/server/request-log"; import type { OcxConfig } from "../../src/types"; +import { DEVIN_API_SERVER } from "../../src/adapters/devin"; +import { setCachedCatalogForTests } from "../../src/adapters/devin/cloud-direct/catalog"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; +import { saveCredential } from "../../src/oauth/store"; +import { createRequestExecutionBudget } from "../../src/lib/request-execution-budget"; /** * One logical request, one send budget -- asserted as a COUNT, because the defect in #4546 is a @@ -31,6 +39,7 @@ beforeEach(() => { afterEach(() => { globalThis.fetch = originalFetch; + setCachedCatalogForTests(null); clearComboSelectionState(); clearComboTargetCooldowns(); clearKeyCooldowns(); @@ -91,6 +100,145 @@ const totalSends = (logCtx: RequestLogContext): number => sendCounts(logCtx).reduce((sum, count) => sum + count, 0); describe("upstream sends per logical request", () => { + test("Devin's initial inner send is recorded once, not omitted or double-counted", async () => { + const previousHome = process.env.OPENCODEX_HOME; + const previousJwtFlag = process.env.OPENCODEX_DEVIN_SEND_USER_JWT; + const home = mkdtempSync(join(tmpdir(), "devin-send-count-")); + process.env.OPENCODEX_HOME = home; + delete process.env.OPENCODEX_DEVIN_SEND_USER_JWT; + const apiKey = "devin-count-test"; + // Devin is an OAuth-kind provider: the key the adapter ends up using is injected onto the + // row from the stored credential, so a config that only carries `apiKey` never routes. The + // credential is what makes this the path production takes. + await saveCredential("devin", { + access: apiKey, + refresh: apiKey, + expires: Number.MAX_SAFE_INTEGER, + source: "oauth", + apiBaseUrl: DEVIN_API_SERVER, + }); + setCachedCatalogForTests({ + apiKey, + host: DEVIN_API_SERVER, + fetchedAt: Date.now(), + byUid: new Map([["swe-2", { modelUid: "swe-2", label: "SWE-2", disabled: false }]]), + }); + const urls: string[] = []; + globalThis.fetch = (async input => { + urls.push(String(input)); + return new Response("busy", { status: 500 }); + }) as typeof fetch; + const logCtx: RequestLogContext = { model: "", provider: "" }; + const config = { + defaultProvider: "devin", + providers: { + devin: { + adapter: "devin", baseUrl: DEVIN_API_SERVER, models: ["swe-2"], + }, + }, + } as unknown as OcxConfig; + + try { + const response = await handleResponses(responsesRequest("devin/swe-2"), config, logCtx); + const body = await response.text(); + + // Reported together, with the status, so a turn that never reaches the adapter says so + // instead of presenting as an empty URL list. + expect({ + chatCalls: urls.filter(url => url.includes("GetChatMessage")).length, + totalSends: totalSends(logCtx), + }, `status ${response.status}: ${body.slice(0, 200)}`).toEqual({ chatCalls: 1, totalSends: 1 }); + } finally { + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + if (previousJwtFlag === undefined) delete process.env.OPENCODEX_DEVIN_SEND_USER_JWT; + else process.env.OPENCODEX_DEVIN_SEND_USER_JWT = previousJwtFlag; + removeTreeWithRetry(home); + } + }); + + test("a Devin turn the budget refuses logs no send at the request boundary", async () => { + // The defect this pins lives in the outer runTurn path, not in the adapter: the attempt's + // first send was logged before the adapter ran, so a request with nothing left to spend + // recorded a send Devin never made. The direct-adapter case in tests/adapters covers the + // executor side; only this one can see `sendCount`. The budget is handed in already spent + // rather than arranged through combo arithmetic, which is how an earlier attempt at this + // case ended up admitting the send it meant to refuse. + const previousHome = process.env.OPENCODEX_HOME; + const previousJwtFlag = process.env.OPENCODEX_DEVIN_SEND_USER_JWT; + const home = mkdtempSync(join(tmpdir(), "devin-send-denied-")); + process.env.OPENCODEX_HOME = home; + delete process.env.OPENCODEX_DEVIN_SEND_USER_JWT; + const apiKey = "devin-denied-test"; + await saveCredential("devin", { + access: apiKey, + refresh: apiKey, + expires: Number.MAX_SAFE_INTEGER, + source: "oauth", + apiBaseUrl: DEVIN_API_SERVER, + }); + setCachedCatalogForTests({ + apiKey, + host: DEVIN_API_SERVER, + fetchedAt: Date.now(), + byUid: new Map([["swe-2", { modelUid: "swe-2", label: "SWE-2", disabled: false }]]), + }); + const urls: string[] = []; + globalThis.fetch = (async input => { + urls.push(String(input)); + return new Response(JSON.stringify({ error: { message: "busy", type: "server_error" } }), { + status: 502, + headers: { "content-type": "application/json" }, + }); + }) as typeof fetch; + const logCtx: RequestLogContext = { model: "", provider: "" }; + const config = { + defaultProvider: "devin", + providers: { devin: { adapter: "devin", baseUrl: DEVIN_API_SERVER, models: ["swe-2"] } }, + } as unknown as OcxConfig; + // The real budget factory with nothing to give: the state an earlier combo fan-out or + // empty-response recovery leaves behind, stated directly instead of inferred. + const spent = createRequestExecutionBudget({ + maxTotalModelSends: 0, + baseSendAllowance: 0, + finalRecoveryAllowance: 0, + maxAlternateTargetSends: 0, + maxTargetTransitions: 0, + }, "devin-denied-initial-send"); + + try { + const response = await handleResponses( + responsesRequest("devin/swe-2"), config, logCtx, { sendBudget: spent }, + ); + const body = await response.text(); + const attempts = logCtx.attempts ?? []; + + // The attempt must EXIST and be empty. An absent attempt would satisfy a zero count + // without proving the refusal was recorded against the turn that was refused. + expect(attempts).toHaveLength(1); + expect({ + adapter: attempts[0]?.adapter, + sendCount: attempts[0]?.sendCount, + chatCalls: urls.filter(url => url.includes("GetChatMessage")).length, + totalSends: totalSends(logCtx), + refused: body.includes("request_send_budget_exhausted"), + }, `status ${response.status}: ${body.slice(0, 240)}`).toEqual({ + adapter: "devin", + sendCount: 0, + chatCalls: 0, + totalSends: 0, + refused: true, + }); + } finally { + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + if (previousJwtFlag === undefined) delete process.env.OPENCODEX_DEVIN_SEND_USER_JWT; + else process.env.OPENCODEX_DEVIN_SEND_USER_JWT = previousJwtFlag; + setCachedCatalogForTests(null); + removeTreeWithRetry(home); + } + }); + test("a 5xx streak on a single target spends the base allowance and stops", async () => { const upstream = alwaysFailing(502, "upstream busy"); const logCtx: RequestLogContext = { model: "", provider: "" };