/**
 * Anthropic Messages inbound (/v1/messages + /v1/messages/count_tokens) for Claude Code.
 *
 * Translate-and-replay (devlog/260711_claude_inbound/010): the Anthropic request is
 * converted to a /v1/responses body and replayed through handleResponses on an
 * internal Request, so routing/OAuth/account-pool/failover/sidecars are inherited
 * unchanged. The Responses output (SSE or JSON) is converted back to Anthropic shape.
 */
import { FORWARD_HEADERS } from "../adapters/openai-responses";
import { jsonUtf8Bytes } from "../lib/json-byte-size";
import { sseFieldValue } from "../lib/sse-decoder";
import { enforceAnthropicImageLimits, sniffImageDimensions } from "../adapters/anthropic-image-guard";
import { normalizeAnthropicImages } from "../adapters/anthropic-image-normalize";
import { AnthropicRequestError, DesktopModelMappingUnavailableError, anthropicToResponsesTranslation, extractOcxEffortDirective, extractOcxRouteDirective, verifyAndExtractDirectives, resolveInboundModel, type ClaudeCacheKeySource } from "../claude/inbound";
import { getOrCreateDirectiveSigningKey } from "../claude/directive-key";
import { isAllowedLegacyDirective } from "../claude/agents-inject";
import { isKnownDesktop3pModelId, resolveDesktop3pAlias } from "../claude/desktop-3p";
import { resolveAlias, claudeCodeNativeAlias } from "../claude/alias";
import { recordDesktopRequest } from "../claude/desktop-health";
import { stripOneMillionMarker } from "../claude/context-windows";
import { annotateClaudeInboundDecision, captureClaudeInbound } from "../claude/inbound-debug";
import { isTransientUpstreamStatus } from "../lib/upstream-retry";
import { resolveClientRetryAfter } from "../lib/retry-after";
import { createHash } from "node:crypto";
import {
  analyzeClaudeCompatibility,
  collectClaudeFeatureCodes,
  isToleratedClaudeFeatureCode,
  resolveClaudeCompatibilityMode,
} from "../claude/compatibility";
import {
  anthropicErrorBody,
  anthropicErrorResponse,
  collectAnthropicMessage,
  responsesJsonToAnthropicMessage,
  responsesSseToAnthropicSse,
} from "../claude/outbound";
import { clearableDeadline, idleDeadline } from "../lib/abort";
import { estimateTokens } from "../lib/token-estimate";
import { modelInList } from "../types/tools";
import type { ClaudeSourceEnvelope, OcxConfig, OcxUsage } from "../types";
import { readJsonRequestBody } from "./request-decompress";
import { addFinalRequestLog, httpStatusForRequestLogTerminal, recordFirstOutput, type RequestLogContext, type RequestLogEntry } from "./request-log";
import { conversationIdFromClaudeMetadata } from "./request-log-conversation";
import { normalizeClaudeCompatibilityUsageLog } from "../usage/log";
import { responseWithDeferredRequestLog } from "./relay";
import { handleResponses } from "./responses";
import {
  isApiAuthRequired,
  isDataPlaneAdmissionSecret,
  isProxyAdmissionSecret,
  type RequestPolicyView,
  type DataPlaneAdmission,
} from "./auth-cors";
import type { AdmissionLease } from "../lib/admission";
import { tryClaimNativeMainProfileForTurn } from "../codex/native-main-admission";
import { CODEX_MAIN_PROFILE_MAINTENANCE_MESSAGE } from "../codex/auth-context";
import {
  createTranslatorBudget,
  finalizeTranslatorBudgetResponse,
  isTranslatorBudgetExceededError,
  type TranslatorBudget,
} from "../lib/translator-budget";
import {
  parseRequestEffortRowId,
  type ParsedEffortRowId,
} from "./effort-row";
import {
  parseFastOnlyRowId,
  parseSyntheticRowId,
  type ParsedFastRowId,
} from "./fast-row";
import { supportedLadderFor } from "./effort-policy";
import { NoEligiblePolicyCandidateError, UnknownRoutingPolicyError, routeModel } from "../router";
import { POLICY_NAMESPACE, resolvePolicyProfileId } from "../routing/profile";
import { evidenceFromBody } from "../routing/request-evidence";

type Rec = Record<string, unknown>;

function resolveClaudePolicySelector(
  config: OcxConfig,
  model: string,
): { decodedModel: string; isPolicy: boolean } {
  // Claude Code sends the readable aliases published by /v1/models, not necessarily the
  // underlying route (`claude-ocx-policy--daily` -> `policy/daily`). Policy detection must
  // use the same identity the Messages translator will route, while preserving an exact
  // operator alias such as `claude-smart` when it has no Claude model-map entry.
  const decodedModel = resolveInboundModel(model, config.claudeCode);
  return {
    decodedModel,
    isPolicy: resolvePolicyProfileId(config, decodedModel) !== null
      || decodedModel.startsWith(`${POLICY_NAMESPACE}/`),
  };
}

function isLocalPolicyRoutingError(
  status: number,
  message: string,
  logCtx: Pick<RequestLogContext, "requestedModel" | "routeDecision">,
): boolean {
  if (status !== 404) return false;
  if (
    logCtx.routeDecision?.routeKind === "policy"
    && logCtx.routeDecision.selected.reason === "no-eligible-candidate"
  ) {
    return true;
  }
  return (
    logCtx.requestedModel?.startsWith(`${POLICY_NAMESPACE}/`) === true
    && message.startsWith("Unknown routing policy:")
  );
}

/**
 * Decode a Claude selector that may carry the fast marker.
 *
 * The exact form is tried first, so a real model whose alias genuinely ends in the marker
 * keeps winning. Only then is the marker treated as synthetic and the bare base decoded:
 * a Desktop 3P alias is a HASH registered WITHOUT the marker, so an exact lookup can never
 * resolve a synthetic one.
 */
function decodeClaudeFastSelector(raw: string, cc?: OcxConfig["claudeCode"]): string {
  const model = stripOneMillionMarker(raw);
  const exact = resolveInboundModel(model, cc);
  if (!model.endsWith("--fast")) return exact;
  const fullMapping = cc?.modelMap?.[model];
  if (resolveAlias(model) || isKnownDesktop3pModelId(model)
    || (typeof fullMapping === "string" && fullMapping.length > 0)) return exact;
  const bare = model.slice(0, -"--fast".length);
  // A classifier fallback is not an exact match for a registered Desktop base.
  // Preserve established non-Desktop fallback behavior while decoding that base first.
  if (exact !== model && !resolveDesktop3pAlias(bare)) return exact;
  const decodedBase = resolveInboundModel(bare, cc);
  return decodedBase === bare ? exact : `${decodedBase}--fast`;
}

/** Restore the reversible Fable picker alias before Anthropic passthrough checks. */
function decodeFablePickerAlias(raw: string, cc?: OcxConfig["claudeCode"]): string {
  const decoded = resolveInboundModel(raw, cc);
  if (!decoded.startsWith("claude-fable-")) return raw;
  return claudeCodeNativeAlias(decoded) === raw ? decoded : raw;
}

function isRec(v: unknown): v is Rec {
  return !!v && typeof v === "object" && !Array.isArray(v);
}

function desktopMappingUnavailableResponse(error: DesktopModelMappingUnavailableError): Response {
  const response = anthropicErrorResponse(503, error.message, "api_error", "desktop_model_mapping_unavailable");
  response.headers.set("Retry-After", "1");
  return response;
}

/** Resolve Claude-only sidecar overrides without mutating the shared server config. */
export function buildClaudeReplayConfig(config: OcxConfig): OcxConfig {
  return {
    ...config,
    webSearchSidecar: {
      ...config.webSearchSidecar,
      ...config.claudeCode?.webSearchSidecar,
    },
    visionSidecar: {
      ...config.visionSidecar,
      ...config.claudeCode?.visionSidecar,
    },
  };
}

/**
 * Phase 4 plan 04-02: benchmark-only raw-usage observation.
 *
 * The optional benchmark observer receives a sanitized structural record only —
 * final adapter kind, resolved model id, and the raw OcxUsage reported before
 * Anthropic wire normalization. It never receives request bodies, headers,
 * provider names/aliases, endpoint, account identity, raw provider response, or
 * error text. Standard Messages behavior is unchanged when omitted.
 */
export interface ClaudeBenchmarkRawUsage {
  adapterKind: string;
  modelId: string;
  usage: OcxUsage | undefined;
}

export interface ClaudeBenchmarkObserverOptions {
  onRawUsage?: (observation: ClaudeBenchmarkRawUsage) => void;
}

function claudeInboundDisabled(config: OcxConfig): Response | null {
  if (config.claudeCode?.enabled === false) {
    return anthropicErrorResponse(403, "Claude inbound is disabled (GUI: Claude ON toggle / config.claudeCode.enabled)", "permission_error");
  }
  return null;
}

// ── Claude source envelope & session precedence (ingress slice) ────────────────

/**
 * Capture sanitized immutable envelope before destructive translation.
 * Charges the same TranslatorBudget for the retained copy + header bytes.
 */
export function captureClaudeSourceEnvelope(
  req: Request,
  rawBody: unknown,
  budget: TranslatorBudget,
): ClaudeSourceEnvelope {
  const beta = req.headers.get("anthropic-beta")?.trim() || undefined;
  const rawVersion = req.headers.get("anthropic-version");
  const version = rawVersion === null ? undefined : rawVersion.trim();
  if (!isRec(rawBody)) throw new AnthropicRequestError("Anthropic request body must be an object");
  const bodyBytes = jsonUtf8Bytes(rawBody);
  let headerBytes = 0;
  if (beta) headerBytes += new TextEncoder().encode(beta).byteLength;
  if (version) headerBytes += new TextEncoder().encode(version).byteLength;
  budget.chargeRetained(bodyBytes + headerBytes, { kind: "request_copies" });
  const bodyClone = structuredClone(rawBody);
  return {
    body: bodyClone,
    headers: {
      ...(beta ? { "anthropic-beta": beta } : {}),
      ...(version !== undefined ? { "anthropic-version": version } : {}),
    },
  };
}

/**
 * Session precedence: x-claude-code-session-id header > metadata.user_id > system cohort.
 * Returns the canonical session id string or null when none is determinable before translation.
 * Agent/parent IDs are NOT session ids — they are only HMAC8 debug tags (see inbound-debug).
 */
export function claudeSessionIdFromRequest(req: Request, body: unknown): string | null {
  const headerSid = req.headers.get("x-claude-code-session-id")?.trim();
  if (headerSid) return headerSid;
  if (body && typeof body === "object" && !Array.isArray(body)) {
    const rec = body as Record<string, unknown>;
    const metadata = rec.metadata;
    if (metadata && typeof metadata === "object" && !Array.isArray(metadata)) {
      const uid = (metadata as Record<string, unknown>).user_id;
      if (typeof uid === "string" && uid.trim().length > 0) return uid.trim();
    }
  }
  return null;
}

function claudeAgentIdsFromRequest(req: Request): { agentId?: string; parentAgentId?: string } {
  const out: { agentId?: string; parentAgentId?: string } = {};
  const headerAgent = req.headers.get("x-claude-code-agent-id")?.trim();
  const headerParent = req.headers.get("x-claude-code-parent-agent-id")?.trim();
  if (headerAgent) out.agentId = headerAgent;
  if (headerParent) out.parentAgentId = headerParent;
  return out;
}

/**
 * Compute prompt_cache_key from a session id (header precedence path).
 * Uses same sha256 hex slice as inbound.ts (32 hex chars) for per-session keys.
 * Returns null when sessionId is null (caller should keep translation's system cohort key).
 */
export function promptCacheKeyForSession(sessionId: string | null): string | null {
  if (!sessionId) return null;
  return createHash("sha256").update(sessionId).digest("hex").slice(0, 32);
}

/** Idempotent, synchronous work for every effective Responses route resolution. */
export function claudeFinalRouteHandler(
  parsed: { options: Record<string, unknown>; modelId: string; _rawBody?: unknown; _promptCacheKeyIsSharedCohort?: boolean },
  route: { provider: OcxConfig["providers"][string]; providerName: string; modelId: string },
  ctx: {
    sourceEnvelope: ClaudeSourceEnvelope;
    cacheKeySource: ClaudeCacheKeySource;
    config: OcxConfig;
    logCtx: RequestLogContext;
  },
): { adapter: string; decision: ReturnType<typeof analyzeClaudeCompatibility>["decision"]; featureCodes: string[] } {
  const adapter = route.provider.adapter;
  // Route-aware fail-closed gate for auto-only models (e.g. muse-spark-1.2-contributor on opencode-go).
  // Must run before any upstream; uses the routed modelId + provider list so a forced/named
  // tool_choice never silently downgrades to auto (Responses core would do that for openai-chat).
  {
    const tc = (ctx.sourceEnvelope.body as Record<string, unknown>).tool_choice as unknown;
    const tcType = tc && typeof tc === "object" && !Array.isArray(tc) && typeof (tc as Record<string, unknown>).type === "string" ? (tc as Record<string, unknown>).type as string : undefined;
    if ((tcType === "any" || tcType === "tool") && modelInList((route.provider as { autoToolChoiceOnlyModels?: string[] }).autoToolChoiceOnlyModels, route.modelId)) {
      throw new AnthropicRequestError(
        `tool_choice type '${tcType}' is not supported for model "${route.modelId}" (provider ${route.providerName}): this model supports only tool_choice auto or none. Remove tool_choice, use {"type":"auto"} or {"type":"none"}, or choose a different model.`,
      );
    }
  }
  // Idempotent sampling strip for openai-responses forward (native ChatGPT pierce)
  if (adapter === "openai-responses") {
    const raw = parsed._rawBody as Record<string, unknown> | undefined;
    if (raw) {
      // _rawBody is snake_case Responses JSON; be idempotent for both snake and camel keys
      delete raw.max_output_tokens;
      delete (raw as Record<string, unknown>).maxOutputTokens;
      delete raw.temperature;
      delete raw.top_p;
      delete (raw as Record<string, unknown>).topP;
      delete raw.stop;
      delete (raw as Record<string, unknown>).stopSequences;
      delete raw.user;
    }
    // OcxRequestOptions is camelCase; _rawBody snake is already handled above. Strip both forms idempotently.
    delete (parsed.options as Record<string, unknown>).max_output_tokens;
    delete (parsed.options as Record<string, unknown>).maxOutputTokens;
    delete (parsed.options as Record<string, unknown>).temperature;
    delete (parsed.options as Record<string, unknown>).top_p;
    delete (parsed.options as Record<string, unknown>).topP;
    delete (parsed.options as Record<string, unknown>).stop;
    delete (parsed.options as Record<string, unknown>).stopSequences;
    delete (parsed.options as Record<string, unknown>).user;
  }
  // Usage estimate only for cursor/kiro (estimated-usage adapters)
  if (adapter === "cursor" || adapter === "kiro") {
    try {
      const bodyRec = ctx.sourceEnvelope.body as Record<string, unknown>;
      const model = typeof bodyRec.model === "string" ? bodyRec.model : ctx.config.claudeCode?.model;
      ctx.logCtx.usageLogInputTokens = estimateClaudeRequestTokens(bodyRec as { system?: unknown; messages?: unknown; tools?: unknown }, model);
    } catch {
      // ignore estimation failures
    }
  }
  // Opus-shaped aliases can make every routed model look reasoning-capable to Claude
  // clients. Strip a forced effort only when the final route explicitly has no ladder.
  if (parsed.options.reasoning !== undefined) {
    const ladder = supportedLadderFor({ provider: route.provider, modelId: route.modelId });
    if (ladder !== undefined && ladder.length === 0) delete parsed.options.reasoning;
  }
  // Compatibility evaluation before network (enforce mode may reject)
  const mode = resolveClaudeCompatibilityMode(ctx.config.claudeCode);
  const anthropicBeta = ctx.sourceEnvelope.headers["anthropic-beta"];
  const result = analyzeClaudeCompatibility(ctx.sourceEnvelope.body, { mode, adapter, anthropicBeta });
  if (result.decision === "reject") {
    ctx.logCtx.errorCode = "claude_compatibility_unsupported";
    throw new AnthropicRequestError(result.reason ?? "incompatible features for routed adapter");
  }
  return { adapter, decision: result.decision, featureCodes: result.shadowFeatureCodes ?? result.featureCodes };
}


async function readAnthropicBody(req: Request, budget: TranslatorBudget): Promise<unknown> {
  try {
    return await readJsonRequestBody(req, budget);
  } catch (err) {
    if (isTranslatorBudgetExceededError(err)) throw err;
    throw new AnthropicRequestError(err instanceof Error && err.message ? err.message : "Invalid JSON body");
  }
}

// ── Native Anthropic passthrough (subscription OAuth pierce) ──────────────────────
// When Claude Code runs with ONLY ANTHROPIC_BASE_URL set (subscription mode — the
// connectors warning stays off), it sends its OWN claude.ai OAuth Bearer to us.
// Requests for genuine claude/anthropic models that no alias/modelMap claims are
// forwarded VERBATIM to api.anthropic.com with the caller's credential and all
// end-to-end headers, so betas/thinking signatures/billing identity stay native.
// (Evidence: teamclaude --no-mitm + Vercel gateway docs, devlog 003/060.)

const PASSTHROUGH_STRIP_HEADERS = new Set([
  "connection", "keep-alive", "transfer-encoding", "upgrade", "te", "trailer",
  "proxy-authenticate", "proxy-authorization", "host", "content-length",
  "accept-encoding", "x-opencodex-api-key", "origin",
]);

function singleCredentialToken(name: "authorization" | "x-api-key", value: string | null): string | null {
  const raw = value?.trim() ?? "";
  // Fetch Headers comma-joins duplicate fields. Neither Anthropic credential format permits a
  // comma, so treating a joined value as one token could hide an admission secret behind a real
  // provider credential. Ambiguous credential headers fail closed.
  if (!raw || raw.includes(",")) return null;
  if (name === "authorization") {
    const match = /^Bearer\s+(.+)$/i.exec(raw);
    return match?.[1]?.trim() || null;
  }
  return raw;
}

function hasAnthropicNativeCredential(req: Request, config: OcxConfig): boolean {
  const bearer = singleCredentialToken("authorization", req.headers.get("authorization"));
  const apiKey = singleCredentialToken("x-api-key", req.headers.get("x-api-key"));
  return (!!bearer && bearer.startsWith("sk-ant-") && !isProxyAdmissionSecret(bearer, config))
    || (!!apiKey && apiKey.startsWith("sk-ant-") && !isProxyAdmissionSecret(apiKey, config));
}

function wantsNativePassthrough(
  req: Request,
  config: OcxConfig,
  requestPolicy: RequestPolicyView,
  model: unknown,
): model is string {
  if (config.claudeCode?.nativePassthrough === false) return false;
  if (typeof model !== "string" || !/^(claude|anthropic)/i.test(model)) return false;
  // Authorization and x-api-key both belong to the upstream on this branch. An exposed listener
  // therefore requires the dedicated admission header even though the routed Messages surface
  // keeps accepting all three legacy admission forms.
  if (isApiAuthRequired(requestPolicy)) {
    const dedicated = req.headers.get("x-opencodex-api-key")?.trim() ?? "";
    if (!isDataPlaneAdmissionSecret(dedicated, config)) return false;
  }
  if (!hasAnthropicNativeCredential(req, config)) return false;
  // An alias or modelMap hit means the user asked for a ROUTED model: translate instead.
  return resolveInboundModel(model, config.claudeCode) === model;
}

function shouldForwardNativeHeader(name: string, value: string, config: OcxConfig): boolean {
  const lowerName = name.toLowerCase();
  if (PASSTHROUGH_STRIP_HEADERS.has(lowerName)) return false;
  if (lowerName !== "authorization" && lowerName !== "x-api-key") return true;
  const token = singleCredentialToken(lowerName, value);
  return !!token && !isProxyAdmissionSecret(token, config);
}

/** Format a 32-hex cache key as a uuid-shaped session id (version/variant nibbles forced). */
function uuidFromHex(hex32: string): string {
  const h = (hex32 + "0".repeat(32)).slice(0, 32);
  return `${h.slice(0, 8)}-${h.slice(8, 12)}-4${h.slice(13, 16)}-8${h.slice(17, 20)}-${h.slice(20, 32)}`;
}

function anthropicUsageToOcx(usage: Rec | undefined): { inputTokens: number; outputTokens: number; cachedInputTokens?: number; cacheReadInputTokens?: number; cacheCreationInputTokens?: number } | undefined {
  if (!usage) return undefined;
  const num = (v: unknown) => typeof v === "number" ? v : 0;
  const hasCache = usage.cache_read_input_tokens !== undefined || usage.cache_creation_input_tokens !== undefined;
  const read = num(usage.cache_read_input_tokens);
  const write = num(usage.cache_creation_input_tokens);
  // Anthropic input_tokens excludes cache read/write; normalize to the canonical
  // inclusive convention (types.ts OcxUsage / devlog 070). cached = READS only.
  return {
    inputTokens: num(usage.input_tokens) + read + write,
    outputTokens: num(usage.output_tokens),
    ...(hasCache ? {
      cachedInputTokens: read,
      cacheReadInputTokens: read,
      cacheCreationInputTokens: write,
    } : {}),
  };
}

/** Body-occupancy guard for the native passthrough (devlog 260716_passthrough_followups/010). */
export interface PassthroughBodyGuard {
  /** Idle window in ms — raw upstream-byte inactivity while a read is pending. 0 disables. */
  stallMs: number;
  /** Cumulative body byte cap. 0 disables. */
  maxBytes: number;
  /** Client request signal for deterministic cancel classification. */
  reqSignal?: AbortSignal;
}

type PassthroughCloseReason = "terminal" | "client_cancel" | "body_stall" | "body_overflow";

/**
 * Tap an Anthropic-vocabulary SSE stream for the request log (usage + terminal),
 * bounding body occupancy: idle (silence-only, timed ONLY while a reader.read() is
 * pending so downstream backpressure never counts as upstream inactivity) and a
 * cumulative byte cap. On stall/overflow it appends a protocol-compatible Anthropic
 * `event: error` terminal frame after a blank-line boundary, closes, and cancels the
 * upstream reader — never a total-wall-clock bound (slow-but-alive streams live).
 * Exported for deterministic unit tests.
 */
export function tapAnthropicSseForLog(
  upstream: ReadableStream<Uint8Array>,
  logCtx: RequestLogContext,
  finalize: (status: number, meta: { closeReason: PassthroughCloseReason }) => void,
  guard?: PassthroughBodyGuard,
): ReadableStream<Uint8Array> {
  const decoder = new TextDecoder();
  const encoder = new TextEncoder();
  let buffer = "";
  let usageAcc: Rec = {};
  const inspect = (chunk: Uint8Array) => {
    buffer += decoder.decode(chunk, { stream: true });
    let sep: number;
    while ((sep = buffer.indexOf("\n\n")) !== -1) {
      const frame = buffer.slice(0, sep);
      buffer = buffer.slice(sep + 2);
      const dataLine = frame
        .split("\n")
        .map(l => sseFieldValue(l, "data"))
        .filter((v): v is string => v !== null)
        .join("");
      if (!dataLine) continue;
      let data: unknown;
      try { data = JSON.parse(dataLine); } catch { continue; }
      if (!isRec(data)) continue;
      if (data.type === "message_start" && isRec(data.message) && isRec(data.message.usage)) {
        usageAcc = { ...usageAcc, ...data.message.usage };
      } else if (data.type === "message_delta" && isRec(data.usage)) {
        usageAcc = { ...usageAcc, ...data.usage };
      }
    }
  };
  const reader = upstream.getReader();
  let settled = false;
  let bodyBytes = 0;
  let tapController: ReadableStreamDefaultController<Uint8Array> | undefined;

  const recordUsage = () => {
    logCtx.usage = anthropicUsageToOcx(Object.keys(usageAcc).length > 0 ? usageAcc : undefined);
  };
  const failBody = (closeReason: "body_stall" | "body_overflow", errType: string, message: string) => {
    if (settled) return;
    settled = true;
    idle.cancel();
    detachAbort();
    recordUsage();
    finalize(200, { closeReason });
    const payload = JSON.stringify({ type: "error", error: { type: errType, message } });
    try {
      // Leading blank line terminates any partial SSE block so the frame parses cleanly
      // (relaySseWithFailedTail policy, Anthropic wire shape).
      tapController?.enqueue(encoder.encode(`\n\nevent: error\ndata: ${payload}\n\n`));
      tapController?.close();
    } catch { /* client already torn down */ }
    reader.cancel(new DOMException(message, closeReason === "body_stall" ? "TimeoutError" : "QuotaExceededError")).catch(() => {});
  };
  const idle = idleDeadline(guard?.stallMs ?? 0, () => {
    failBody(
      "body_stall",
      "timeout_error",
      `anthropic passthrough body stalled: no upstream bytes for ${Math.round((guard?.stallMs ?? 0) / 1000)}s`,
    );
  });
  // Deterministic client-cancel classification: Bun may surface a client abort as a
  // reader.read() rejection OR a resolved done (src/lib/abort.ts cancelBodyOnAbort
  // rationale), so the listener performs first-wins settlement itself instead of
  // relying on which shape the read takes.
  const onClientAbort = () => {
    if (settled) return;
    settled = true;
    idle.cancel();
    detachAbort();
    finalize(499, { closeReason: "client_cancel" });
    try { tapController?.close(); } catch { /* downstream already torn down */ }
    reader.cancel(guard?.reqSignal?.reason).catch(() => {});
  };
  const detachAbort = (() => {
    const signal = guard?.reqSignal;
    if (!signal) return () => {};
    if (signal.aborted) {
      queueMicrotask(onClientAbort);
      return () => {};
    }
    signal.addEventListener("abort", onClientAbort, { once: true });
    return () => signal.removeEventListener("abort", onClientAbort);
  })();

  return new ReadableStream<Uint8Array>({
    start(controller) {
      tapController = controller;
    },
    async pull(controller) {
      if (settled) return;
      try {
        idle.reset();
        const { done, value } = await reader.read();
        idle.pause();
        if (settled) return; // stall/overflow/abort won the race while we awaited
        if (done) {
          settled = true;
          idle.cancel();
          detachAbort();
          recordUsage();
          finalize(200, { closeReason: "terminal" });
          controller.close();
          return;
        }
        if (value.byteLength > 0) {
          bodyBytes += value.byteLength;
          if (guard && guard.maxBytes > 0 && bodyBytes > guard.maxBytes) {
            failBody(
              "body_overflow",
              "api_error",
              `anthropic passthrough body exceeded ${guard.maxBytes} bytes`,
            );
            return;
          }
        }
        inspect(value);
        controller.enqueue(value);
      } catch (err) {
        if (settled) return;
        settled = true;
        idle.cancel();
        detachAbort();
        recordUsage();
        finalize(200, { closeReason: "terminal" });
        try { controller.error(err); } catch { /* torn down */ }
      }
    },
    cancel(reason) {
      if (!settled) {
        settled = true;
        idle.cancel();
        detachAbort();
        finalize(499, { closeReason: "client_cancel" });
      }
      reader.cancel(reason).catch(() => {});
    },
  });
}

async function anthropicNativePassthrough(
  req: Request,
  config: OcxConfig,
  logCtx: RequestLogContext,
  logIds: { requestId: string; start: number } | undefined,
  body: Rec,
  pathname: string,
): Promise<Response> {
  const model = typeof body.model === "string" ? body.model : "unknown";
  logCtx.model = model;
  logCtx.provider = "anthropic-native";
  logCtx.requestedModel = model;
  let logged = false;
  const finalize = (status: number, meta: { closeReason: PassthroughCloseReason | "non_stream" }) => {
    if (!logIds || logged) return;
    logged = true;
    addFinalRequestLog(logIds.requestId, logIds.start, logCtx, status, meta);
  };

  const base = (config.claudeCode?.anthropicBaseUrl ?? "https://api.anthropic.com").replace(/\/$/, "");
  const search = new URL(req.url).search;
  // Native passthrough bypasses the anthropic adapter, so the generous image pipeline
  // (devlog/260714_image_normalization_pipeline/040) must run here: tier-normalize then
  // guard the already-Anthropic-wire messages before serialization. Applies to
  // count_tokens too — counts must match what the real send will contain, and the 32MB
  // body cap applies to it equally. Non-message bodies pass through untouched.
  if (Array.isArray(body.messages)) {
    await normalizeAnthropicImages(body.messages);
    enforceAnthropicImageLimits(body.messages);
  }
  const headers = new Headers();
  req.headers.forEach((value, name) => {
    if (shouldForwardNativeHeader(name, value, config)) headers.set(name, value);
  });
  headers.set("content-type", "application/json");

  const result = await fetchWithHeaderDeadline(
    `${base}${pathname}${search}`,
    { method: "POST", headers, body: JSON.stringify(body) },
    config.connectTimeoutMs ?? 200_000,
    req.signal,
  );
  if (result.kind === "timeout") {
    finalize(504, { closeReason: "non_stream" });
    return anthropicErrorResponse(504, "anthropic passthrough timed out waiting for response headers", "timeout_error");
  }
  if (result.kind === "error") {
    const err = result.error;
    finalize(502, { closeReason: "non_stream" });
    return anthropicErrorResponse(502, `anthropic passthrough failed: ${err instanceof Error ? err.message : String(err)}`, "api_error");
  }
  const upstream = result.upstream;

  const contentType = upstream.headers.get("content-type") ?? "application/json";
  const bodyGuard = resolvePassthroughBodyGuard(config, req.signal);
  if (upstream.ok && contentType.includes("text/event-stream") && upstream.body) {
    return new Response(tapAnthropicSseForLog(upstream.body, logCtx, finalize, bodyGuard), {
      status: upstream.status,
      headers: {
        "Content-Type": contentType,
        "Cache-Control": "no-cache",
        "Connection": "keep-alive",
      },
    });
  }
  // Non-stream (count_tokens, errors, stream:false): relay verbatim under the same
  // idle/size bounds — headers are NOT yet sent here, so real statuses are available.
  const bodyResult = await readBoundedPassthroughBody(upstream, bodyGuard);
  if (bodyResult.kind === "client_cancel") {
    finalize(499, { closeReason: "client_cancel" });
    return anthropicErrorResponse(499, "client closed request during anthropic passthrough", "api_error");
  }
  if (bodyResult.kind === "stall") {
    finalize(504, { closeReason: "body_stall" });
    return anthropicErrorResponse(504, `anthropic passthrough body stalled: no upstream bytes for ${Math.round(bodyGuard.stallMs / 1000)}s`, "timeout_error");
  }
  if (bodyResult.kind === "overflow") {
    finalize(502, { closeReason: "body_overflow" });
    return anthropicErrorResponse(502, `anthropic passthrough body exceeded ${bodyGuard.maxBytes} bytes`, "api_error");
  }
  const text = bodyResult.text;
  if (upstream.ok) {
    try {
      const parsed = JSON.parse(text) as { usage?: Rec };
      if (isRec(parsed?.usage)) logCtx.usage = anthropicUsageToOcx(parsed.usage);
    } catch { /* count_tokens etc. */ }
  }
  finalize(upstream.status, { closeReason: "non_stream" });
  const retryAfter = upstream.headers.get("retry-after");
  return new Response(text, {
    status: upstream.status,
    headers: { "Content-Type": contentType, ...(retryAfter ? { "Retry-After": retryAfter } : {}) },
  });
}

const DEFAULT_BODY_STALL_SEC = 90;
const DEFAULT_BODY_MAX_BYTES = 64 * 1024 * 1024;

/**
 * Normalize the claudeCode body-guard config (devlog 260716_passthrough_followups/010).
 * Policy: exactly 0 disables; finite positive values are honored (stall clamped to
 * min 1s); negative/non-finite/absent values fall back to the defaults.
 */
export function resolvePassthroughBodyGuard(config: OcxConfig, reqSignal?: AbortSignal): PassthroughBodyGuard {
  const rawSec = config.claudeCode?.bodyStallSec;
  const stallSec = rawSec === 0
    ? 0
    : typeof rawSec === "number" && Number.isFinite(rawSec) && rawSec > 0
      ? Math.max(1, rawSec)
      : DEFAULT_BODY_STALL_SEC;
  const rawBytes = config.claudeCode?.bodyMaxBytes;
  const maxBytes = rawBytes === 0
    ? 0
    : typeof rawBytes === "number" && Number.isFinite(rawBytes) && rawBytes > 0
      ? Math.floor(rawBytes)
      : DEFAULT_BODY_MAX_BYTES;
  return { stallMs: stallSec * 1000, maxBytes, ...(reqSignal ? { reqSignal } : {}) };
}

type BoundedPassthroughBody =
  | { kind: "ok"; text: string }
  | { kind: "stall" }
  | { kind: "overflow" }
  | { kind: "client_cancel" };

/**
 * Bounded replacement for `await upstream.text()` on the non-stream passthrough
 * branch: same idle-only + size-cap semantics as the SSE tap. NOTE: reader.cancel()
 * resolves a pending read as done rather than rejecting, so the stalled flag is
 * re-checked after every read settlement (audit round 3).
 */
export async function readBoundedPassthroughBody(
  upstream: Response,
  guard: PassthroughBodyGuard,
): Promise<BoundedPassthroughBody> {
  if (!upstream.body) return { kind: "ok", text: await upstream.text() };
  const reader = upstream.body.getReader();
  const decoder = new TextDecoder();
  let text = "";
  let bytes = 0;
  let stalled = false;
  let aborted = false;
  const idle = idleDeadline(guard.stallMs, () => {
    stalled = true;
    reader.cancel(new DOMException("anthropic passthrough body stalled", "TimeoutError")).catch(() => {});
  });
  // Deterministic client-abort classification (audit round 4): Bun may surface the
  // abort as a read rejection OR a resolved done, so we cancel the reader ourselves
  // and classify via the flag rather than the read's settlement shape.
  const signal = guard.reqSignal;
  const onAbort = () => {
    aborted = true;
    reader.cancel(signal?.reason).catch(() => {});
  };
  if (signal?.aborted) onAbort();
  else signal?.addEventListener("abort", onAbort, { once: true });
  try {
    while (true) {
      idle.reset();
      let result: Awaited<ReturnType<typeof reader.read>>;
      try {
        result = await reader.read();
      } catch (err) {
        if (aborted) return { kind: "client_cancel" };
        if (stalled) return { kind: "stall" };
        throw err;
      } finally {
        idle.pause();
      }
      if (aborted) return { kind: "client_cancel" };
      if (stalled) return { kind: "stall" };
      if (result.done) break;
      if (result.value.byteLength === 0) continue;
      bytes += result.value.byteLength;
      if (guard.maxBytes > 0 && bytes > guard.maxBytes) {
        reader.cancel(new DOMException("anthropic passthrough body exceeded byte cap", "QuotaExceededError")).catch(() => {});
        return { kind: "overflow" };
      }
      text += decoder.decode(result.value, { stream: true });
    }
    text += decoder.decode();
    return { kind: "ok", text };
  } finally {
    idle.cancel();
    signal?.removeEventListener("abort", onAbort);
  }
}

/**
 * Header-phase fetch guarded by a clearable deadline (PR #136 follow-up hardening).
 *
 * The deadline covers ONLY the wait for response headers; once `fetch` settles —
 * fulfilled OR rejected — the timer must die. The `finally` block guarantees
 * `clear()` on every path (success, upstream reject, deadline expiry), fixing the
 * timer leak where a rejected fetch left the deadline running until expiry.
 * `didExpire()` stays truthful after `clear()` (see src/lib/abort.ts), so timeout
 * classification inside the catch is unaffected by the finally cleanup.
 *
 * `makeDeadline`/`fetchImpl` are injectable for deterministic unit tests.
 */
export type HeaderDeadlineFetchResult =
  | { kind: "response"; upstream: Response }
  | { kind: "timeout" }
  | { kind: "error"; error: unknown };

export async function fetchWithHeaderDeadline(
  input: string | URL,
  init: RequestInit,
  timeoutMs: number,
  parent?: AbortSignal,
  makeDeadline: typeof clearableDeadline = clearableDeadline,
  fetchImpl: typeof fetch = fetch,
): Promise<HeaderDeadlineFetchResult> {
  const deadline = makeDeadline(timeoutMs, parent);
  try {
    const upstream = await fetchImpl(input, { ...init, signal: deadline.signal, timeout: 0 });
    return { kind: "response", upstream };
  } catch (error) {
    if (deadline.didExpire()) return { kind: "timeout" };
    return { kind: "error", error };
  } finally {
    deadline.clear();
  }
}

export async function handleClaudeMessages(
  req: Request,
  config: OcxConfig,
  logCtx: RequestLogContext,
  logIds?: { requestId: string; start: number; turnAdmissionLease?: AdmissionLease; admission?: DataPlaneAdmission },
  requestPolicy: RequestPolicyView = config,
  benchmark?: ClaudeBenchmarkObserverOptions,
): Promise<Response> {
  const translatorBudget = createTranslatorBudget();
  try {
    return finalizeTranslatorBudgetResponse(
      await handleClaudeMessagesWithBudget(req, config, logCtx, translatorBudget, logIds, requestPolicy, benchmark),
      translatorBudget,
    );
  } catch (error) {
    translatorBudget.dispose();
    throw error;
  }
}

async function handleClaudeMessagesWithBudget(
  req: Request,
  config: OcxConfig,
  logCtx: RequestLogContext,
  translatorBudget: TranslatorBudget,
  logIds?: { requestId: string; start: number; turnAdmissionLease?: AdmissionLease; admission?: DataPlaneAdmission },
  requestPolicy: RequestPolicyView = config,
  benchmark?: ClaudeBenchmarkObserverOptions,
): Promise<Response> {
  logCtx.surface = "claude";
  const disabled = claudeInboundDisabled(config);
  if (disabled) {
    if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 403, { closeReason: "non_stream" });
    return disabled;
  }

  let anthropicBody: unknown;
  let internalBody: Rec;
  let cacheKeySource: ClaudeCacheKeySource = null;
  let effortOverride: string | null = null;
  let effortRow: ParsedEffortRowId | null = null;
  let fastRow: ParsedFastRowId | null = null;
  let requestedModel = "";
  let sourceEnvelope: ClaudeSourceEnvelope | null = null;
  let headerSessionId: string | null = null;
  let featureCodesEarly: string[] = [];
  let agentIds: { agentId?: string; parentAgentId?: string } = {};
  let debugCaptureId: number | undefined;
  try {
    anthropicBody = await readAnthropicBody(req, translatorBudget);
    // Defensive [1m] strip (devlog 138): clients normally remove the context-variant
    // marker themselves; the 1M signal we act on is the anthropic-beta header.
    // Case-insensitive — the CLI matches /\[1m\]/i (audit 021 #7).
    if (isRec(anthropicBody) && typeof anthropicBody.model === "string") {
      anthropicBody.model = stripOneMillionMarker(anthropicBody.model);
    }
    // ocx-route override (devlog 072 + TRUST-01..05): injected agent bodies pin their model via a
    // system-prompt directive because 2.1.207 ignores custom ids in agent
    // frontmatter. Must run BEFORE the native-passthrough branch — the CLI sends
    // these subagent turns under a fallback claude model id.
    if (isRec(anthropicBody)) {
      const directives = verifyAndExtractDirectives(
        anthropicBody,
        getOrCreateDirectiveSigningKey(),
        (route, effort) => isAllowedLegacyDirective(route, effort, config),
      );
      if (directives.route && typeof anthropicBody.model === "string") {
        anthropicBody.model = stripOneMillionMarker(directives.route);
        if (directives.effort) {
          effortOverride = directives.effort;
        }
      }
    }
    if (isRec(anthropicBody) && typeof anthropicBody.model === "string") {
      anthropicBody.model = decodeFablePickerAlias(anthropicBody.model, config.claudeCode);
    }
    if (isRec(anthropicBody) && typeof anthropicBody.model === "string") {
      requestedModel = anthropicBody.model;
      // Decode for Fast only. A Claude alias is `claude-ocx-<provider>--<model>`, so it
      // already uses `--` as its own separator: stripping the marker off the RAW alias would
      // turn `claude-ocx-p--foo--fast` into `claude-ocx-p--foo` and route a DIFFERENT model.
      // Effort parsing keeps the raw selector, so its behaviour is untouched.
      ({ fastRow, effortRow } = parseSyntheticRowId(
        requestedModel,
        config,
        () => decodeClaudeFastSelector(requestedModel, config.claudeCode),
      ));
      if (effortRow) {
        anthropicBody.model = effortRow.baseId;
        effortOverride = effortRow.effort;
      }
      if (fastRow) anthropicBody.model = fastRow.baseId;
    }
    headerSessionId = claudeSessionIdFromRequest(req, anthropicBody);
    agentIds = claudeAgentIdsFromRequest(req);
    featureCodesEarly = collectClaudeFeatureCodes(anthropicBody, req.headers.get("anthropic-beta") ?? undefined);
    // Debug capture (opt-in allowlist scalars) BEFORE the passthrough branch so
    // native, routed, and disabled-alias paths are all observable (devlog 130 B1).
    // Extended ring carries featureCodes/adapter/decision + HMAC8 session/agent tags.
    debugCaptureId = captureClaudeInbound(
      "messages",
      anthropicBody,
      isRec(anthropicBody) && typeof anthropicBody.model === "string"
        ? resolveInboundModel(anthropicBody.model, config.claudeCode)
        : undefined,
      req.headers.get("anthropic-beta") ?? undefined,
      {
        ...(headerSessionId ? { sessionId: headerSessionId } : {}),
        ...(agentIds.agentId ? { agentId: agentIds.agentId } : {}),
        ...(agentIds.parentAgentId ? { parentAgentId: agentIds.parentAgentId } : {}),
        ...(featureCodesEarly.length > 0 ? { featureCodes: featureCodesEarly } : {}),
      },
    );
    // Client surface discrimination: Desktop 3P aliases resolve through the
    // desktop registry; Code uses readable aliases or direct model names.
    if (isRec(anthropicBody) && typeof anthropicBody.model === "string" && resolveDesktop3pAlias(anthropicBody.model)) {
      logCtx.surface = "claude-desktop";
      recordDesktopRequest();
    }
    // Correlate before native passthrough so Anthropic-credential turns still filter/total (#330 / #522).
    if (isRec(anthropicBody)) {
      const claudeConversationId = conversationIdFromClaudeMetadata(
        isRec(anthropicBody.metadata) ? anthropicBody.metadata : undefined,
      );
      if (claudeConversationId) logCtx.conversationId = claudeConversationId;
    }
    // A fast row blocks passthrough, unlike the chat case: this path forwards to Anthropic's
    // own API, whose wire has no service_tier field and whose FastWire kind has an empty
    // adapter set by design, so the tier would be silently dropped.
    const policySelector = isRec(anthropicBody) && typeof anthropicBody.model === "string"
      ? resolveClaudePolicySelector(config, anthropicBody.model)
      : null;
    if (
      !effortRow
      && !fastRow
      && policySelector?.isPolicy !== true
      && isRec(anthropicBody)
      && wantsNativePassthrough(req, config, requestPolicy, anthropicBody.model)
    ) {
      annotateClaudeInboundDecision(debugCaptureId, "anthropic", "native", featureCodesEarly);
      return await anthropicNativePassthrough(req, config, logCtx, logIds, anthropicBody, "/v1/messages");
    }
    if (isRec(anthropicBody) && effortOverride) {
      anthropicBody.output_config = {
        ...(isRec(anthropicBody.output_config) ? anthropicBody.output_config : {}),
        effort: effortOverride,
      };
      delete anthropicBody.thinking;
    }
    // Routed requests retain the post-directive source body. Native passthrough never
    // pays for or observes this clone, preserving its existing byte-for-byte path.
    sourceEnvelope = captureClaudeSourceEnvelope(req, anthropicBody, translatorBudget);
    const translation = anthropicToResponsesTranslation(anthropicBody, config.claudeCode, translatorBudget);
    internalBody = translation.body;
    // The Anthropic translator builds its body from model/input/store/stream plus sampling
    // fields only, so the caller intent is applied to the TRANSLATED body rather than the
    // inbound one.
    if (fastRow) internalBody.service_tier = "priority";
    translatorBudget.chargeRetained(jsonUtf8Bytes(internalBody), { kind: "request_copies" });
    // Session header precedence feeds prompt_cache_key (header > metadata > system cohort).
    // When the x-claude-code-session-id header is present, it replaces the
    // translation's per-session/system key with a stable per-header sha256 key
    // and promotes cacheKeySource to "metadata" so downstream affinity/session_id
    // logic treats it as a real per-session key (not the shared system cohort).
    if (headerSessionId) {
      const sessionCacheKey = promptCacheKeyForSession(headerSessionId);
      if (sessionCacheKey) {
        internalBody.prompt_cache_key = sessionCacheKey;
        cacheKeySource = "metadata";
      } else {
        cacheKeySource = translation.cacheKeySource;
      }
    } else {
      cacheKeySource = translation.cacheKeySource;
    }
  } catch (err) {
    const overflow = isTranslatorBudgetExceededError(err);
    const unavailable = err instanceof DesktopModelMappingUnavailableError;
    const status = overflow ? 413 : unavailable ? 503 : err instanceof AnthropicRequestError ? 400 : 500;
    if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, status, { closeReason: "non_stream" });
    if (unavailable) return desktopMappingUnavailableResponse(err);
    return anthropicErrorResponse(
      status,
      overflow ? "request translation buffer exceeded the safe limit" : err instanceof Error ? err.message : String(err),
      overflow ? "request_too_large" : undefined,
      overflow ? "translation_buffer_limit" : undefined,
    );
  }

  if (!requestedModel) requestedModel = (anthropicBody as Rec).model as string;
  const stream = internalBody.stream === true;
  // Routed adapters only support streamed turns; always stream internally and fold
  // the translated Anthropic SSE into a message JSON for non-streaming clients.
  internalBody.stream = true;

  // ── Final-route callback (post-route, pre-network) ───────────────────────────────────────
  // Adapter-specific work runs only after Responses core owns the final route.
  const claudeOnResolvedRoute = (info: import("./responses/core").ResolvedRouteInfo): void => {
    if (!sourceEnvelope) return;
    // Each attempt owns its admission evidence; a later native route must not
    // inherit a translated shadow decision from an earlier fallback target.
    delete logCtx.claudeCompatibility;
    const result = claudeFinalRouteHandler(
      info.parsed as unknown as Parameters<typeof claudeFinalRouteHandler>[0],
      { provider: { ...info.provider, adapter: info.adapterName }, providerName: info.route.providerName, modelId: info.modelId } as Parameters<typeof claudeFinalRouteHandler>[1],
      {
        sourceEnvelope,
        cacheKeySource,
        config,
        logCtx,
      },
    );
    if (result.decision === "shadow") {
      // Persisted shadow evidence must be backed by the final per-attempt
      // evaluation: every non-tolerated code comes from result.featureCodes,
      // so an adapter-tolerated feature (e.g. deferred_tools on
      // openai-responses) can never be relabeled unsupported. Early pre-route
      // codes only contribute tolerated diagnostics that the effort override
      // legitimately removed before the final evaluation.
      logCtx.claudeCompatibility = normalizeClaudeCompatibilityUsageLog({
        decision: "shadow",
        featureCodes: [...new Set([
          ...result.featureCodes,
          ...featureCodesEarly.filter(isToleratedClaudeFeatureCode),
        ])],
      });
    }
    // Session_id header synthesis: only for openai-responses and only for a
    // real per-session prompt_cache_key (metadata), never the system-hash
    // cohort. Idempotent: check headers.has before set.
    {
      const pck = (info.parsed.options as Rec).promptCacheKey ?? (info.parsed.options as Rec).prompt_cache_key ?? ((info.parsed as unknown as Rec)._rawBody as Rec | undefined)?.prompt_cache_key;
      if (info.adapterName === "openai-responses" && cacheKeySource === "metadata" && typeof pck === "string" && !info.headers.has("session_id")) {
        info.headers.set("session_id", uuidFromHex(pck as string));
      } else if (info.adapterName !== "openai-responses") {
        info.headers.delete("session_id");
      }
    }
    annotateClaudeInboundDecision(debugCaptureId, result.adapter, result.decision, result.featureCodes);
  };

  const headers = new Headers({ "content-type": "application/json" });
  for (const name of FORWARD_HEADERS) {
    // The caller's bearer is the proxy admission token (ocx claude placeholder), never a
    // ChatGPT credential — forwarding it upstream turns into {"detail":"Unauthorized"}.
    if (name === "authorization") continue;
    const value = req.headers.get(name);
    if (value) headers.set(name, value);
  }
  // Routed replays need main ChatGPT auth so OpenAI-backed sidecars remain reachable;
  // native replays have no caller ChatGPT credential. This enrichment is optional:
  // auth-context later rejects a real physical-main selection, while routed/pool
  // traffic continues without reading native credentials during a fence/recovery.
  if (tryClaimNativeMainProfileForTurn(logIds?.turnAdmissionLease)) {
    const { getMainAccountToken } = await import("../codex/main-account");
    const token = getMainAccountToken();
    if (token) {
      headers.set("authorization", `Bearer ${token.accessToken}`);
      headers.set("chatgpt-account-id", token.chatgptAccountId);
    }
  }
  let internalReq: Request;
  try {
    // The UTF-16 JSON string and the Request's UTF-8 body coexist until dispatch.
    const bodyBytes = jsonUtf8Bytes(internalBody);
    const reservation = translatorBudget.reserveTransient(3 * bodyBytes, { kind: "request_copies" });
    try {
      internalReq = new Request("http://localhost/v1/responses", {
        method: "POST",
        headers,
        body: JSON.stringify(internalBody),
      });
    } finally {
      reservation.release();
    }
    translatorBudget.chargeRetained(bodyBytes, { kind: "request_copies" });
  } catch (err) {
    if (!isTranslatorBudgetExceededError(err)) throw err;
    if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 413, { closeReason: "non_stream" });
    return anthropicErrorResponse(413, "request translation buffer exceeded the safe limit", "request_too_large", "translation_buffer_limit");
  }

  // Request-log wiring mirrors the /v1/responses route: native passthrough finalizes
  // via the terminal callbacks; routed streams get the Responses-vocabulary log tap
  // BEFORE translation (the translated Anthropic stream has no response.completed
  // frame, so tapping it records a bogus 502 with no usage/cache detail).
  let nativeLogged = false;
  const finalizeNativeLog = (status: number, meta: { terminalStatus?: RequestLogEntry["terminalStatus"]; closeReason: "terminal" | "client_cancel" }) => {
    if (!logIds || nativeLogged) return;
    nativeLogged = true;
    addFinalRequestLog(logIds.requestId, logIds.start, logCtx, status, meta);
  };
  const upstream = await handleResponses(internalReq, buildClaudeReplayConfig(config), logCtx, {
    // Routing keeps Claude-only sidecar overrides; admission policy must follow the live owner.
    codexAuthPolicy: config,
    ...(logIds?.admission ? { admission: logIds.admission } : {}),
    ...(logIds?.turnAdmissionLease ? { turnAdmissionLease: logIds.turnAdmissionLease } : {}),
    abortSignal: req.signal,
    promptCacheKeyIsSharedCohort: cacheKeySource === "system",
    // The body is Responses-shaped by now, but the client spoke Anthropic Messages.
    // Without this the replay would look native and a Responses-scoped wire default
    // would fire, disagreeing with the pre-flight decision above.
    inboundWire: "anthropic",
    stripClaudeMainAuthForNoncanonicalForward: true,
    translatorBudget,
    // Forward the sanitized immutable source envelope (charged to the same budget)
    // so core can perform fidelity-preserving work. The envelope is the
    // pre-translation clone (body + anthropic-beta/version headers only).
    ...(sourceEnvelope ? { claudeSourceEnvelope: sourceEnvelope } : {}),
    // Adapter-aware final-route gate: idempotent strip / session_id synthesis /
    // usage estimate / compatibility enforce. Core should invoke this after
    // each routeModel+resolveWireProtocolOverride (see note above).
    onResolvedRoute: claudeOnResolvedRoute,
    ...(logIds ? { onFirstOutput: () => recordFirstOutput(logCtx, logIds.start) } : {}),
    onNativePassthroughTerminal: status => finalizeNativeLog(httpStatusForRequestLogTerminal(status, logCtx), { terminalStatus: status, closeReason: "terminal" }),
    onNativePassthroughCancel: () => finalizeNativeLog(499, { closeReason: "client_cancel" }),
    ...(benchmark?.onRawUsage ? { claudeBenchmarkObserver: benchmark.onRawUsage } : {}),
  });
  const response = logIds ? responseWithDeferredRequestLog(upstream, logIds.requestId, logIds.start, logCtx) : upstream;

  if (!response.ok) {
    // Re-shape the OpenAI-style error envelope into the Anthropic one, preserving status.
    let message = `upstream error (${response.status})`;
    try {
      const text = await response.text();
      try {
        const parsed = JSON.parse(text) as { error?: { message?: string; type?: string } | string; message?: string };
        const nested = typeof parsed?.error === "object" && parsed.error ? parsed.error.message : undefined;
        const flat = typeof parsed?.error === "string" ? parsed.error : parsed?.message;
        message = nested || flat || (text ? `upstream error (${response.status}): ${text.slice(0, 400)}` : message);
      } catch {
        if (text) message = `upstream error (${response.status}): ${text.slice(0, 400)}`;
      }
    } catch { /* keep fallback message */ }
    const upstreamRetryAfter = response.headers.get("retry-after");
    const retryAfter = resolveClientRetryAfter({
      status: response.status,
      message,
      upstreamRetryAfter,
    })
      // Instant-retry "0" is a valid client directive but rejected by cooldown parsers.
      // Preserve it so it still wins over the transient "2" fallback (claude-529 mapping).
      ?? (upstreamRetryAfter?.trim() === "0" ? "0" : undefined);
    // Transient upstream 5xx (already retried pre-stream, 010): reclassify as Anthropic
    // 529 overloaded_error so the Claude Code client applies its built-in backoff retry
    // instead of dying on a fatal api_error (260716 sol-builder incident). The request
    // log keeps the upstream status (captured in the deferred-log closure before this
    // rewrite): log = upstream truth, client = retry signal.
    // Retryable 429s also get Retry-After (#507) so Codex-shaped clients and Claude Code
    // share a backoff hint when the upstream omitted the header.
    const nativeMainFence = response.status === 503
      && upstreamRetryAfter?.trim() === "1"
      && message === CODEX_MAIN_PROFILE_MAINTENANCE_MESSAGE;
    const transient = !nativeMainFence && isTransientUpstreamStatus(response.status);
    const outStatus = nativeMainFence ? 503 : transient ? 529 : response.status;
    const errorType = isLocalPolicyRoutingError(response.status, message, logCtx)
      ? "invalid_request_error"
      : undefined;
    const out = new Response(JSON.stringify(anthropicErrorBody(outStatus, message, errorType)), {
      status: outStatus,
      headers: {
        "Content-Type": "application/json",
        ...(retryAfter ? { "Retry-After": retryAfter } : (transient ? { "Retry-After": "2" } : {})),
      },
    });
    return out;
  }

  const contentType = response.headers.get("content-type") ?? "";
  if (contentType.includes("text/event-stream") && response.body) {
    const anthropicSse = responsesSseToAnthropicSse(response.body, requestedModel, { translatorBudget });
    if (stream) {
      return new Response(anthropicSse, {
        status: 200,
        headers: {
          "Content-Type": "text/event-stream; charset=utf-8",
          "Cache-Control": "no-cache",
          "Connection": "keep-alive",
        },
      });
    }
    let message: Rec;
    try {
      message = await collectAnthropicMessage(anthropicSse, requestedModel, translatorBudget);
    } catch (error) {
      if (isTranslatorBudgetExceededError(error)) {
        return anthropicErrorResponse(413, error.message, "request_too_large", error.code);
      }
      return anthropicErrorResponse(502, error instanceof Error ? error.message : String(error), "api_error");
    }
    const isError = (message as Rec).type === "error";
    const translatedError = isError && typeof (message as Rec).error === "object"
      ? (message as { error: { code?: unknown; message?: unknown } }).error
      : undefined;
    if (translatedError?.code === "translation_buffer_limit") {
      return anthropicErrorResponse(
        413,
        typeof translatedError.message === "string"
          ? translatedError.message
          : "upstream translation buffer exceeded the safe limit",
        "request_too_large",
        "translation_buffer_limit",
      );
    }
    return new Response(JSON.stringify(message), {
      status: isError ? 502 : 200,
      headers: { "Content-Type": "application/json" },
    });
  }

  // Defensive: some passthrough paths may answer JSON despite stream:true.
  let json: unknown;
  try {
    json = await response.json();
  } catch {
    return anthropicErrorResponse(502, "internal replay returned a non-JSON response", "api_error");
  }
  const status = (json as Rec)?.status;
  if (status === "failed") {
    const error = (json as { error?: { message?: string; code?: string } }).error;
    if (error?.code === "translation_buffer_limit") {
      return anthropicErrorResponse(
        413,
        error.message ?? "upstream translation buffer exceeded the safe limit",
        "request_too_large",
        "translation_buffer_limit",
      );
    }
    return anthropicErrorResponse(502, error?.message ?? "upstream request failed", "api_error");
  }
  let message: Rec;
  try {
    message = responsesJsonToAnthropicMessage(json, requestedModel, translatorBudget);
  } catch (err) {
    if (!isTranslatorBudgetExceededError(err)) throw err;
    return anthropicErrorResponse(413, "upstream translation buffer exceeded the safe limit", "request_too_large", "translation_buffer_limit");
  }
  if ((message as Rec).type === "error") {
    return new Response(JSON.stringify(message), {
      status: 529,
      headers: { "Content-Type": "application/json", "Retry-After": "2" },
    });
  }
  if (!stream) {
    return new Response(JSON.stringify(message), { status: 200, headers: { "Content-Type": "application/json" } });
  }
  // Streaming client + JSON upstream: synthesize a minimal valid Anthropic stream.
  const encoder = new TextEncoder();
  const frames: string[] = [];
  const emit = (name: string, data: Rec) => frames.push(`event: ${name}\ndata: ${JSON.stringify(data)}\n\n`);
  emit("message_start", { type: "message_start", message: { ...message, content: [], stop_reason: null, usage: { input_tokens: 0, output_tokens: 0 } } });
  const blocks = Array.isArray((message as Rec).content) ? (message as Rec).content as Rec[] : [];
  blocks.forEach((block, index) => {
    emit("content_block_start", { type: "content_block_start", index, content_block: block });
    emit("content_block_stop", { type: "content_block_stop", index });
  });
  emit("message_delta", { type: "message_delta", delta: { stop_reason: (message as Rec).stop_reason ?? "end_turn", stop_sequence: null }, usage: (message as Rec).usage ?? {} });
  emit("message_stop", { type: "message_stop" });
  return new Response(encoder.encode(frames.join("")), {
    status: 200,
    headers: { "Content-Type": "text/event-stream; charset=utf-8", "Cache-Control": "no-cache" },
  });
}

/** Per-attachment token estimate for a base64 payload: real image dimensions when the
 * header is sniffable (Anthropic prices images at ~pixels/750), else decoded bytes/512,
 * min 256 — the same shape as the Kiro usage estimator (estimateKiroImageTokens). */
function estimateBase64AttachmentTokens(data: string): number {
  const dims = sniffImageDimensions(data);
  if (dims) return Math.max(256, Math.ceil((dims.width * dims.height) / 750));
  const unpadded = data.endsWith("==") ? data.length - 2 : data.endsWith("=") ? data.length - 1 : data.length;
  return Math.max(256, Math.ceil(Math.floor((unpadded * 3) / 4) / 512));
}

/**
 * Char-based token estimate for an Anthropic-shaped request body. Base64 attachment
 * payloads (image/document blocks in message content, including blocks nested in
 * tool_result.content) are counted as a bounded per-attachment estimate instead of raw
 * characters: one 2MB screenshot is ~2.7M base64 chars, which the plain chars/token
 * divide reports as hundreds of thousands of tokens versus a real cost around 1.6k.
 * That breaks the >2x drift bound the estimator is held to (devlog 260711_claude_inbound
 * 040 §3). Text and url sources are left in place and counted as characters, as is
 * anything outside protocol content positions (tool_use.input, tool schemas).
 */
export function estimateClaudeRequestTokens(
  raw: { system?: unknown; messages?: unknown; tools?: unknown },
  modelId: string | undefined,
): number {
  let attachmentTokens = 0;
  // Blank base64 payloads ONLY in protocol content positions: message content blocks and
  // blocks nested in tool_result.content. tool_use.input and tool schemas can legitimately
  // contain attachment-shaped JSON, and those bytes ARE serialized into function_call
  // arguments / tool definitions for routed providers, so they must keep counting as text.
  // system is text-only per the Anthropic protocol (no attachment sources), so it is
  // stringified as-is.
  const sanitizeBlock = (block: unknown): unknown => {
    if (!block || typeof block !== "object") return block;
    const b = block as Record<string, unknown>;
    if (b.type === "image" || b.type === "document") {
      const source = b.source as { type?: unknown; data?: unknown } | undefined;
      if (source && typeof source === "object" && source.type === "base64" && typeof source.data === "string") {
        attachmentTokens += estimateBase64AttachmentTokens(source.data);
        return { ...b, source: { ...(source as Record<string, unknown>), data: "" } };
      }
      return block;
    }
    if (b.type === "tool_result" && Array.isArray(b.content)) {
      return { ...b, content: (b.content as unknown[]).map(sanitizeBlock) };
    }
    return block;
  };
  const sanitizedMessages = (messages: unknown): unknown =>
    Array.isArray(messages)
      ? messages.map(message => {
          if (!message || typeof message !== "object") return message;
          const m = message as Record<string, unknown>;
          return Array.isArray(m.content) ? { ...m, content: (m.content as unknown[]).map(sanitizeBlock) } : message;
        })
      : messages;
  const parts: string[] = [];
  if (raw.system !== undefined) parts.push(typeof raw.system === "string" ? raw.system : JSON.stringify(raw.system));
  if (raw.messages !== undefined) parts.push(JSON.stringify(sanitizedMessages(raw.messages)));
  if (raw.tools !== undefined) parts.push(JSON.stringify(raw.tools));
  return Math.max(1, estimateTokens(parts.join("\n"), modelId) + attachmentTokens);
}

export async function handleClaudeCountTokens(
  req: Request,
  config: OcxConfig,
  requestPolicy: RequestPolicyView = config,
): Promise<Response> {
  const disabled = claudeInboundDisabled(config);
  if (disabled) return disabled;

  let body: unknown;
  const translatorBudget = createTranslatorBudget();
  try {
    body = await readAnthropicBody(req, translatorBudget);
  } catch (err) {
    if (err instanceof DesktopModelMappingUnavailableError) return desktopMappingUnavailableResponse(err);
    if (err instanceof AnthropicRequestError) return anthropicErrorResponse(400, err.message);
    if (isTranslatorBudgetExceededError(err)) {
      return anthropicErrorResponse(413, "request translation buffer exceeded the safe limit", "request_too_large", "translation_buffer_limit");
    }
    return anthropicErrorResponse(500, err instanceof Error ? err.message : String(err));
  } finally { translatorBudget.dispose(); }
  if (!body || typeof body !== "object" || Array.isArray(body)) {
    return anthropicErrorResponse(400, "request body must be a JSON object");
  }
  const raw = body as Rec;
  if (typeof raw.model !== "string" || raw.model.length === 0) {
    return anthropicErrorResponse(400, "model is required");
  }
  try {
  let model = raw.model;
  // Case-insensitive [1m] strip (audit 021 #7 — the CLI matches /\[1m\]/i).
  const stripped = stripOneMillionMarker(model);
  if (stripped !== model) {
    model = stripped;
    raw.model = model;
  }
  // ocx-route override (devlog 072 + TRUST-01..05): keep count_tokens consistent with messages.
  let directives;
  try {
    directives = verifyAndExtractDirectives(
      raw,
      getOrCreateDirectiveSigningKey(),
      (route, effort) => isAllowedLegacyDirective(route, effort, config),
    );
  } catch (err) {
    if (err instanceof AnthropicRequestError) {
      return anthropicErrorResponse(400, err.message, "invalid_request_error");
    }
    throw err;
  }
  if (directives.route) {
    model = stripOneMillionMarker(directives.route);
    raw.model = model;
  }
  model = decodeFablePickerAlias(model, config.claudeCode);
  raw.model = model;
  // Fast-only count requests carry a synthetic selector but never parsed an effort row.
  // Normalize the identity before native passthrough or estimation; no tier is sent here.
  const countFastRow = parseFastOnlyRowId(
    config, () => decodeClaudeFastSelector(model, config.claudeCode),
  );
  if (countFastRow) {
    model = countFastRow.baseId;
    raw.model = model;
  }
  const policySelector = resolveClaudePolicySelector(config, model);
  let resolvedPolicy = false;
  if (policySelector.isPolicy) {
    try {
      const route = routeModel(config, policySelector.decodedModel, evidenceFromBody(raw));
      model = route.modelId;
      raw.model = model;
      resolvedPolicy = true;
    } catch (err) {
      if (err instanceof UnknownRoutingPolicyError || err instanceof NoEligiblePolicyCandidateError) {
        return anthropicErrorResponse(404, err.message, "invalid_request_error");
      }
      throw err;
    }
  }
  const ctHeaderSessionId = claudeSessionIdFromRequest(req, raw);
  const ctAgentIds = claudeAgentIdsFromRequest(req);
  const ctAnthropicBeta = req.headers.get("anthropic-beta") ?? undefined;
  const ctFeatureCodes = collectClaudeFeatureCodes(raw, ctAnthropicBeta);
  captureClaudeInbound("count_tokens", raw, resolveInboundModel(model, config.claudeCode), ctAnthropicBeta, {
    ...(ctHeaderSessionId ? { sessionId: ctHeaderSessionId } : {}),
    ...(ctAgentIds.agentId ? { agentId: ctAgentIds.agentId } : {}),
    ...(ctAgentIds.parentAgentId ? { parentAgentId: ctAgentIds.parentAgentId } : {}),
    ...(ctFeatureCodes.length > 0 ? { featureCodes: ctFeatureCodes } : {}),
  });
  if (!resolvedPolicy && wantsNativePassthrough(req, config, requestPolicy, model)) {
    return await anthropicNativePassthrough(req, config, { model, provider: "anthropic-native", surface: "claude" }, undefined, raw, "/v1/messages/count_tokens");
  }
  const inputTokens = estimateClaudeRequestTokens(raw, model);
  return new Response(JSON.stringify({ input_tokens: inputTokens }), {
    status: 200,
    headers: { "Content-Type": "application/json" },
  });
  } catch (error) {
    if (error instanceof DesktopModelMappingUnavailableError) return desktopMappingUnavailableResponse(error);
    if (error instanceof AnthropicRequestError) return anthropicErrorResponse(400, error.message);
    throw error;
  }
}