feat: harvest quota headers on relay transports

Relay responses now feed the passive quota-header harvester: HTTP relay
responses expose upstream headers directly, and the WebSocket relay
delivers genuine upstream headers from response_start control messages
via an onResponseHeaders callback in sendViaRelay. Synthetic optimistic
response headers are never harvested; relay-to-direct fallback keeps
header harvesting on the direct path.

The WebSocket callback is idempotent per logical request with
last-accepted-response wins: headers are staged at response_start and
fire exactly once when the response commits (first downstream bytes or
completion). A pre-byte recovery that resends full_sync replaces the
staged headers with the retry's frame — the response that actually
streams is also the fresher quota evidence — and duplicate
response_start frames within one attempt are ignored.

Author: iceteaSA

ARCHITECTURE.md

@@ -50,7 +50,7 @@
 5. **URL rewrite** — `rewriteUrl()` adds `?beta=true` to `/v1/messages` and overrides base URL when `ANTHROPIC_BASE_URL` is set — `packages/opencode/src/transform.ts`
 6. **Request body rewrite** — `rewriteRequestBody()` strips trailing assistant messages, normalizes Fable/Mythos thinking and Sonnet 5 adaptive thinking, injects billing header, sanitizes system prompt (removes OpenCode identity), prepends Claude Code identity, applies cache strategy (explicit/automatic/hybrid), adds fast mode, prefixes tool names with `mcp_`, creates `cch` over serialized body — `packages/opencode/src/transform.ts`
 7. **Routing** — `shouldFallbackStatus()` checks if response should trigger fallback; `FallbackAccountManager` iterates accounts in ordered modes, while `StickySessionRouter` assigns cold sessions by reset-normalized spendable OAuth quota and weighted initial-prompt deficit, then persists hashed session affinity across processes/restarts. Sticky routes retain transient failures, hold confirmed 5h exhaustion when reset is within 15 minutes, and migrate for longer confirmed exhaustion/permanent account failure. All modes respect model-scoped quotas and killswitch thresholds (including per-model scoped thresholds). If all accounts fail the killswitch policy, a 429 block response is returned immediately; this block is classified as scoped-driven (matching a specific model's weekly limit) or account-level (5h/7d limits) with a model-specific or generic retry hint — `packages/core/src/routing.ts`, `packages/core/src/accounts.ts`, `packages/opencode/src/index.ts`
-8. **Relay** — `sendViaRelay()` sends full or patched body to Cloudflare Worker, which streams Anthropic response back — `packages/core/src/relay.ts`
+8. **Relay** — `sendViaRelay()` sends full or patched body to Cloudflare Worker, which streams Anthropic response back. HTTP relay responses and WebSocket `response_start` control messages deliver genuine upstream headers to the OpenCode account-bound quota harvester; synthetic optimistic response headers are never harvest input, and relay-to-direct fallback stays owned by the direct path — `packages/core/src/relay.ts`, `packages/opencode/src/index.ts`
 9. **SSE stream and Fable recovery** — Response body is wrapped in `createStrippedStream()`, which reverses the tool name prefix and detects Anthropic `refusal` finishes. For selected Fable sessions, a refusal activates a session-local 10-response Opus 4.8 downgrade; every successful downgraded response triggers a zero-output Fable prewarm through `CacheKeepManager` using the OAuth account that served the filtered Fable request. The last successful Opus tail anchor is retained per session and OAuth account; if Fable later refuses after that anchor has moved outside Anthropic's 20-block lookback, the retry spends the system cache slot on an explicit old-Opus-to-current-tail bridge rather than rewriting the intervening Opus cache. Recovery transitions are written per session to the TUI sidebar state; when no matching TUI is connected, OpenCode Desktop receives immediate ignored/no-reply `promptAsync` notices for the switch to Opus and return to Fable. Each notice is assigned a message ID immediately before the active assistant message: Desktop displays it immediately by creation time, while OpenCode's run loop sees it as older than the active assistant and cannot mistake it for pending user work or create an extra provider response — `packages/opencode/src/transform.ts`, `packages/opencode/src/fable-fallback.ts`, `packages/opencode/src/prompt-context.ts`, `packages/opencode/src/index.ts`
 10. **Sidebar update** — `writeSidebarState()` writes quota/routing/cache state plus bounded per-session Fable recovery status to a JSON file read by the TUI sidebar widget (separate process via RPC). Routing-authoritative writes (e.g. active routing decisions) are distinguished from display-only/metadata writes (e.g. quota refreshes or command paths). Display-only writes re-read the file and merge state to preserve any live routing session's `activeId` and route. Cross-process writes are synchronized using an atomic `mkdir` directory lock with jittered retries, rename-claim eviction, and lock-budget exhaustion skips. The write is fenced: ownership is verified before and after the rename, triggering one bounded locked repair of routing-authoritative fields on post-rename loss. — `packages/opencode/src/sidebar-state.ts`, `packages/opencode/src/index.ts`
 

docs/parity-backlog.md

@@ -337,23 +337,20 @@ production-only regression (env-override bypass) — first-hand DIFF review caug
 
 ---
 
-## 8. Relay response eligibility for quota harvest — FOLLOW-UP
+## 8. Relay response eligibility for quota harvest — DONE
 
-The corrected Miniflare gate confirms both relay transports preserve
-`anthropic-ratelimit-unified-*`: the HTTP Worker copies `upstream.headers` into its response and
-WebSocket sends them in `response_start`. The earlier HTTP-negative test omitted
-`x-session-affinity`, so `sendViaRelay` returned its direct fallback without reaching the Worker.
-Quota harvest remains direct-only at the client `usedRelay` guard because WebSocket response
-headers are transport-reconstructed and are not yet canonical harvest evidence. Resolve that
-eligibility/synthetic-header question at the client guard before enabling relay harvest. Do not add
-a quota side channel.
+OpenCode harvests genuine upstream headers from both relay transports without a quota side channel.
+HTTP delivery is gated on `usedRelay`; WebSocket delivery comes from `response_start` before the
+optimistic-response early return. Relay-to-direct fallback invokes only the direct-path harvester,
+and synthetic optimistic headers remain ineligible.
 
 ## 9. Pi quota-header harvest parity — FOLLOW-UP
 
 Pi uses the distinct `packages/pi/src/stream.ts` response path. Header harvest, served-account
 attribution, sidecar persistence, and quota display parity remain out of scope for v1. Port the
 OpenCode direct-path behavior without sharing request-path state implicitly, then gate Pi's own
 streaming response headers and malformed-header handling.
+Relay-header callback wiring is also OpenCode-only; Pi parity remains outstanding.
 
 ## Implementation phase (operator directive)
 

docs/quota-surfaces.md

@@ -7,7 +7,7 @@ Three independent surfaces expose Claude plan quota and account identity to OAut
 | Usage API (`GET /api/oauth/usage`) | active poll, per token | on demand | **yes** (`limits[]`) | **yes** — pollable without traffic |
 | `anthropic-ratelimit-unified-*` headers | passive, on every `/v1/messages` response | every request | no | no — only accounts you send through |
 
-The plugin combines surface 1 background polling (`fetchOAuthQuotaSnapshot` + `QuotaManager`) with passive direct-path harvest from surface 2. Relay responses remain gated from harvest.
+The plugin combines surface 1 background polling (`fetchOAuthQuotaSnapshot` + `QuotaManager`) with passive response-header harvest from surface 2 across direct and relay transports.
 
 ---
 
@@ -143,7 +143,7 @@ Conditional headers (absent on main, present on work-alt): `fallback` appears on
 | Scoped per-model windows | `limits[]` `weekly_scoped` | n/a |
 | Representative/binding marker | `representative-claim` header + `is_active` (inferred) | `x-codex-…-over-…` style flags |
 
-openai-auth is push-based by necessity (QuotaManager fed via `setMain`/`setFallback`); anthropic-auth now combines the usage poll with passive header pushes on direct requests.
+openai-auth is push-based by necessity (QuotaManager fed via `setMain`/`setFallback`); anthropic-auth combines the usage poll with passive header pushes on direct and relayed requests.
 
 ---
 
@@ -171,17 +171,16 @@ Also returned: account/org uuids, email, subscription status/created, `enabled_p
 
 ## Implemented behavior
 
-- Direct `/v1/messages` responses harvest the unified 5h and 7d windows. Utilization fractions are multiplied by 100, then rounded; reset values are epoch seconds converted to ISO timestamps.
+- Direct and relayed `/v1/messages` responses harvest the unified 5h and 7d windows. Utilization fractions are multiplied by 100, then rounded; reset values are epoch seconds converted to ISO timestamps.
 - Header pushes merge into the last poll snapshot. They preserve poll-owned `scoped`, including meaningful empty `[]`, and `extraUsage` credit data.
 - Poll `limits[].is_active` owns `bindingWindow` when present. The header `representative-claim` fills the marker only when the poll did not supply one.
 - Money stays in integer minor units with an explicit currency exponent. Formatting happens at the display boundary.
 - `fallback: available` becomes `fallbackAdvised`; it appears only in expanded quota views and does not change routing.
 - Profile metadata is sidecar-persisted, uses a seven-day TTL, and is absent from the request path.
-- Relay transport is direct-only for harvest in v1. Both the HTTP Worker (`upstream.headers` copied into its response) and WebSocket `response_start` preserve unified headers, but relay responses remain gated because transport-reconstructed WebSocket headers are not yet treated as canonical harvest evidence. See the relay parity item in `docs/parity-backlog.md`.
+- Relay HTTP harvest reads the Worker's genuine upstream response headers only when the relay handled the request. WebSocket harvest reads the genuine `response_start` headers through a callback even when the caller already received an optimistic response with synthetic headers. Relay-to-direct fallback remains owned by the direct-path harvest, so one upstream response produces one push.
 
 ## Gaps / opportunities
 
-- Relay-side harvest requires a client eligibility decision and synthetic-header safety gate at the `usedRelay` guard, not an HTTP Worker passthrough fix.
 - Pi has a separate streaming response path and does not harvest quota headers in v1.
 
 ## Probe recipes

packages/core/src/relay.ts

@@ -497,6 +497,9 @@ type PendingWebSocketRequest = {
   streamByteCount: number
   retryAttempts: number
   retryingBeforeResponse: boolean
+  onResponseHeaders?: (headers: Headers) => void
+  stagedResponseHeaders?: Headers
+  responseHeadersDelivered: boolean
 }
 
 class PersistentRelaySession {
@@ -535,13 +538,20 @@ class PersistentRelaySession {
     payload: RelayPayload,
     bodyText: string,
     optimisticResponse = false,
+    onResponseHeaders?: (headers: Headers) => void,
   ): Promise<RelaySendResult> {
     this.touch()
     const enqueuedAt = perfNowMs()
     const start = this.queue
       .catch(() => {})
       .then(() =>
-        this.startQueued(payload, bodyText, enqueuedAt, optimisticResponse),
+        this.startQueued(
+          payload,
+          bodyText,
+          enqueuedAt,
+          optimisticResponse,
+          onResponseHeaders,
+        ),
       )
     const result = start.then(async ({ response, getPayload }) => ({
       response: await response,
@@ -562,6 +572,7 @@ class PersistentRelaySession {
     bodyText: string,
     enqueuedAt: number,
     optimisticResponse: boolean,
+    onResponseHeaders?: (headers: Headers) => void,
   ) {
     const connectStart = perfNowMs()
     await this.ensureConnected()
@@ -593,7 +604,12 @@ class PersistentRelaySession {
     )
 
     let activePayload = requestPayload
-    const first = this.sendPayload(requestPayload, bodyText, optimisticResponse)
+    const first = this.sendPayload(
+      requestPayload,
+      bodyText,
+      optimisticResponse,
+      onResponseHeaders,
+    )
     void first.done.catch(() => {})
     let activeDone = first.done
     const response = first.response.catch((error) => {
@@ -608,7 +624,12 @@ class PersistentRelaySession {
       fullSync.id = createRequestId()
       fullSync.revision = (this.serverState?.revision ?? 0) + 1
       activePayload = fullSync
-      const retry = this.sendPayload(fullSync, bodyText, optimisticResponse)
+      const retry = this.sendPayload(
+        fullSync,
+        bodyText,
+        optimisticResponse,
+        onResponseHeaders,
+      )
       void retry.done.catch(() => {})
       activeDone = retry.done
       return retry.response
@@ -711,6 +732,7 @@ class PersistentRelaySession {
     payload: RelayPayload,
     bodyText: string,
     optimisticResponse: boolean,
+    onResponseHeaders?: (headers: Headers) => void,
   ) {
     const socket = this.socket
     if (!socket || socket.readyState !== WebSocket.OPEN) {
@@ -741,6 +763,9 @@ class PersistentRelaySession {
         streamByteCount: 0,
         retryAttempts: 0,
         retryingBeforeResponse: false,
+        onResponseHeaders,
+        stagedResponseHeaders: undefined,
+        responseHeadersDelivered: false,
       }
       this.pending = pending
       this.resetPendingTimeout(pending)
@@ -805,6 +830,8 @@ class PersistentRelaySession {
         pending.accepted = false
         pending.acceptedAt = undefined
         pending.responseStartedAt = undefined
+        pending.stagedResponseHeaders = undefined
+        pending.responseHeadersDelivered = false
         pending.sentAt = perfNowMs()
         pending.retryingBeforeResponse = false
 
@@ -830,11 +857,24 @@ class PersistentRelaySession {
     pending: PendingWebSocketRequest,
     chunk: Uint8Array,
   ) {
+    this.deliverResponseHeaders(pending)
     pending.streamChunkCount += 1
     pending.streamByteCount += chunk.byteLength
     pending.streamController?.enqueue(chunk)
   }
 
+  private deliverResponseHeaders(pending: PendingWebSocketRequest) {
+    if (
+      pending.responseHeadersDelivered ||
+      pending.stagedResponseHeaders == null
+    )
+      return
+    pending.responseHeadersDelivered = true
+    try {
+      pending.onResponseHeaders?.(pending.stagedResponseHeaders)
+    } catch {}
+  }
+
   private handleBinaryChunk(data: unknown) {
     const pending = this.pending
     if (!pending?.responseStarted) return
@@ -894,19 +934,23 @@ class PersistentRelaySession {
       return
     }
     if (message.type === 'response_start') {
+      if (pending.responseStartedAt != null) return
       const responseStartedAt = perfNowMs()
       pending.responseStartedAt = responseStartedAt
       relayLog(
         `perf websocket response_start session=${shortAffinity(this.affinity)} request=${pending.payload.id} sentMs=${formatMs(responseStartedAt - pending.sentAt)} upstreamMs=${pending.acceptedAt == null ? 'unknown' : formatMs(responseStartedAt - pending.acceptedAt)} status=${message.status}`,
       )
       clearTimeout(pending.timeout)
+      if (message.headers)
+        pending.stagedResponseHeaders = new Headers(message.headers)
       this.resolvePendingResponse(
         pending,
         message.status,
         message.statusText,
         message.headers,
       )
       if (pending.optimisticResponse && message.status >= 400) {
+        this.deliverResponseHeaders(pending)
         pending.streamController?.enqueue(
           new TextEncoder().encode(
             `event: error\ndata: ${JSON.stringify({ type: 'error', error: { type: 'relay_upstream_error', message: `Relay upstream returned HTTP ${message.status}` } })}\n\n`,
@@ -933,6 +977,7 @@ class PersistentRelaySession {
         relayLog(
           `websocket relay error during optimistic response session=${shortAffinity(this.affinity)} request=${pending.payload.id} status=${message.status} message=${message.message || 'unknown'} chunks=${pending.streamChunkCount} bytes=${pending.streamByteCount}`,
         )
+        this.deliverResponseHeaders(pending)
         pending.streamController?.enqueue(
           new TextEncoder().encode(
             `event: error\ndata: ${JSON.stringify({ type: 'error', error: { type: 'relay_error', message: message.message || 'relay error' } })}\n\n`,
@@ -972,6 +1017,7 @@ class PersistentRelaySession {
     )
     clearTimeout(pending.timeout)
     if (!pending.streamDone) {
+      this.deliverResponseHeaders(pending)
       pending.streamDone = true
       pending.streamController?.close()
     }
@@ -1046,6 +1092,13 @@ export async function sendViaRelay(options: {
   fallback: () => Promise<Response>
   affinity?: string | null
   optimisticResponse?: boolean
+  /**
+   * Observes genuine upstream headers once per delivered response. WebSocket
+   * retries before downstream bytes replace staged headers, so the callback
+   * receives only the final accepted response; duplicate response_start frames
+   * within an attempt are ignored.
+   */
+  onResponseHeaders?: (headers: Headers) => void
 }): Promise<Response> {
   const {
     config,
@@ -1056,6 +1109,7 @@ export async function sendViaRelay(options: {
     fallback,
     affinity: explicitAffinity,
     optimisticResponse,
+    onResponseHeaders,
   } = options
   if (!config || !isRelayableAnthropicRequest(input, body)) return fallback()
 
@@ -1104,6 +1158,7 @@ export async function sendViaRelay(options: {
           payload,
           bodyText,
           optimisticResponse === true,
+          onResponseHeaders,
         )
       } catch (error) {
         relayLog(
@@ -1139,6 +1194,12 @@ export async function sendViaRelay(options: {
 
     if (!result.usedRelay) return result.response
 
+    if (result.transport === 'http') {
+      try {
+        onResponseHeaders?.(result.response.headers)
+      } catch {}
+    }
+
     if (result.transport === 'http') {
       updateLocalRelayState(
         affinity,