feat(devin): admit every inference send through the shared budget

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.

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);

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<C
 
   let resp: Response;
   try {
-    resp = await fetch(`${host}/exa.api_server_pb.ApiServerService/GetChatMessage`, {
+    resp = await (req.executor ?? globalThis.fetch)(`${host}/exa.api_server_pb.ApiServerService/GetChatMessage`, {
       method: 'POST',
       headers: {
         'Content-Type': 'application/connect+proto',

src/adapters/devin/cloud-direct/stated-reset-retry.ts

@@ -5,7 +5,9 @@
  * is idempotent; ambiguous transport failures still propagate unchanged.
  */
 import { parseRetryAfterFromMessage } from '../../../lib/retry-delay.js';
-import { abortError, sleepWithAbort } from '../../../lib/upstream-retry.js';
+import type { AdapterFetchContext } from '../../base.js';
+import { createAdapterPhysicalSend } from '../../physical-send.js';
+import { abortError, SendBudgetExhaustedError, sleepWithAbort } from '../../../lib/upstream-retry.js';
 import { CloudChatError, streamChatEvents, type CloudChatEvent, type CloudChatRequest } from './chat.js';
 
 /** 1 initial attempt plus at most 2 replays. */
@@ -33,6 +35,11 @@ export interface StatedResetRetryOptions {
   maxReplays?: number;
   /** CUMULATIVE wait allowance, not a fresh allowance on every failure. */
   maxWaitMs?: number;
+  /** Request-wide execution authority. Absent preserves context-free callers' unlimited fetch. */
+  execution?: Pick<
+    AdapterFetchContext,
+    "executor" | "sendBudget" | "onPhysicalSend" | "onRecoveryWithheld"
+  >;
 }
 
 function replayLimit(value: number | undefined): number {
@@ -59,30 +66,59 @@ 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;
       }
       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

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);
   };
   /**

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<ResponsesCompletionPolicy, "emptyCompletionGuardEnabled">,
 ): Promise<Response> {
@@ -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,
         );

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

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). |

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