refactor: extract provider logic into slices

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

README.md

@@ -1,286 +1,285 @@
-<div align="center">
-
-# Headroom
-
-**Compress everything your AI agent reads. Same answers, fraction of the tokens.**
-
-[![CI](https://github.com/chopratejas/headroom/actions/workflows/ci.yml/badge.svg)](https://github.com/chopratejas/headroom/actions/workflows/ci.yml)
-[![codecov](https://codecov.io/gh/chopratejas/headroom/graph/badge.svg)](https://app.codecov.io/gh/chopratejas/headroom)
-[![PyPI](https://img.shields.io/pypi/v/headroom-ai.svg)](https://pypi.org/project/headroom-ai/)
-[![npm](https://img.shields.io/npm/v/headroom-ai.svg)](https://www.npmjs.com/package/headroom-ai)
-[![Model: Kompress-base](https://img.shields.io/badge/model-Kompress--base-yellow.svg)](https://huggingface.co/chopratejas/kompress-base)
-[![Tokens saved: 60B+](https://img.shields.io/badge/tokens%20saved-60B%2B-2ea44f)](https://headroomlabs.ai/dashboard)
-[![License: Apache 2.0](https://img.shields.io/badge/license-Apache%202.0-blue.svg)](LICENSE)
-[![Docs](https://img.shields.io/badge/docs-online-blue.svg)](https://headroom-docs.vercel.app/docs)
-
-<img src="HeadroomDemo-Fast.gif" alt="Headroom in action" width="820">
-
-</div>
-
----
-
-Every tool call, log line, DB read, RAG chunk, and file your agent injects into a prompt is mostly boilerplate. Headroom strips the noise and keeps the signal — **losslessly, locally, and without touching accuracy.**
-
-> **100 logs. One FATAL error buried at position 67. Both runs found it.**
-> Baseline **10,144 tokens** → Headroom **1,260 tokens** — **87% fewer, identical answer.**
-> `python examples/needle_in_haystack_test.py`
-
----
-
-## Quick start
-
-Works with Anthropic, OpenAI, Google, Bedrock, Vertex, Azure, OpenRouter, and 100+ models via LiteLLM.
-
-**Wrap your coding agent — one command:**
-
-```bash
-pip install "headroom-ai[all]"
-
-headroom wrap claude      # Claude Code
-headroom wrap codex       # Codex
-headroom wrap cursor      # Cursor
-headroom wrap aider       # Aider
-headroom wrap copilot     # GitHub Copilot CLI
-```
-
-**Prefer a one-time durable install instead of wrapping every launch:**
-
-```bash
-headroom init -g          # Detect installed user-scoped agents and wire them to Headroom
-headroom init claude      # Install repo-local Claude hooks for just this project
-headroom init copilot -g  # Install user-scoped Copilot hooks and provider routing
-```
-
-**Drop it into your own code — Python or TypeScript:**
-
-```python
-from headroom import compress
-
-result = compress(messages, model="claude-sonnet-4-5")
-response = client.messages.create(model="claude-sonnet-4-5", messages=result.messages)
-print(f"Saved {result.tokens_saved} tokens ({result.compression_ratio:.0%})")
-```
-
-```typescript
-import { compress } from 'headroom-ai';
-const result = await compress(messages, { model: 'gpt-4o' });
-```
-
-**Or run it as a proxy — zero code changes, any language:**
-
-```bash
-headroom proxy --port 8787
-ANTHROPIC_BASE_URL=http://localhost:8787 your-app
-OPENAI_BASE_URL=http://localhost:8787/v1 your-app
-```
-
----
-
-## Why Headroom
-
-- **Accuracy-preserving.** GSM8K **0.870 → 0.870** (±0.000). TruthfulQA **+0.030**. SQuAD v2 and BFCL both **97%** accuracy after compression. Validated on public OSS benchmarks you can rerun yourself.
-- **Runs on your machine.** No cloud API, no data egress. Compression latency is milliseconds — faster end-to-end for Sonnet / Opus / GPT-4 class models than a hosted service round-trip.
-- **[Kompress-base](https://huggingface.co/chopratejas/kompress-base) on HuggingFace.** Our open-source text compressor, fine-tuned on real agentic traces — tool outputs, logs, RAG chunks, code. Install with `pip install "headroom-ai[ml]"`.
-- **Cross-agent memory and learning.** Claude Code saves a fact, Codex reads it back. `headroom learn` mines failed sessions and writes corrections straight to `CLAUDE.md` / `AGENTS.md` / `GEMINI.md` — reliability compounds over time.
-- **Reversible (CCR).** Compression is not deletion. The model can always call `headroom_retrieve` to pull the original bytes. Nothing is thrown away.
-
-Bundles the [RTK](https://github.com/rtk-ai/rtk) binary for shell-output rewriting — full [attribution below](#compared-to).
-
----
-
-## How it fits
-
-```
- Your agent / app
-   (Claude Code, Cursor, Codex, LangChain, Agno, Strands, your own code…)
-        │   prompts · tool outputs · logs · RAG results · files
-        ▼
-    ┌────────────────────────────────────────────────────┐
-    │  Headroom   (runs locally — your data stays here)  │
-    │  ───────────────────────────────────────────────   │
-    │  CacheAligner  →  ContentRouter  →  CCR             │
-    │                    ├─ SmartCrusher   (JSON)         │
-    │                    ├─ CodeCompressor (AST)          │
-    │                    └─ Kompress-base  (text, HF)     │
-    │                                                     │
-    │  Cross-agent memory  ·  headroom learn  ·  MCP      │
-    └────────────────────────────────────────────────────┘
-        │   compressed prompt  +  retrieval tool
-        ▼
- LLM provider  (Anthropic · OpenAI · Bedrock · …)
-```
-
-→ [Architecture](https://headroom-docs.vercel.app/docs/architecture) · [CCR reversible compression](https://headroom-docs.vercel.app/docs/ccr) · [Kompress-base model card](https://huggingface.co/chopratejas/kompress-base)
-
-### Canonical pipeline lifecycle
-
-Headroom now exposes one stable request lifecycle across `compress()`, the SDK, and the proxy:
-
-`Setup` → `Pre-Start` → `Post-Start` → `Input Received` → `Input Cached` → `Input Routed` → `Input Compressed` → `Input Remembered` → `Pre-Send` → `Post-Send` → `Response Received`
-
-- **Transforms** still do the work: CacheAligner, ContentRouter, SmartCrusher, CodeCompressor, Kompress-base, IntelligentContext / RollingWindow.
-- **Pipeline extensions** observe or customize those lifecycle stages via `on_pipeline_event(...)`.
-- **Compression hooks** still work and now sit alongside the canonical lifecycle instead of being the only extension seam.
-- **Proxy extensions** remain the server/app integration seam for ASGI middleware, routes, and startup policy.
-
----
-
-## Proof
-
-**Savings on real agent workloads:**
-
-| Workload                      | Before | After  | Savings |
-|-------------------------------|-------:|-------:|--------:|
-| Code search (100 results)     | 17,765 |  1,408 | **92%** |
-| SRE incident debugging        | 65,694 |  5,118 | **92%** |
-| GitHub issue triage           | 54,174 | 14,761 | **73%** |
-| Codebase exploration          | 78,502 | 41,254 | **47%** |
-
-**Accuracy preserved on standard benchmarks:**
-
-| Benchmark  | Category | N   | Baseline | Headroom | Delta     |
-|------------|----------|----:|---------:|---------:|----------:|
-| GSM8K      | Math     | 100 |    0.870 |    0.870 | **±0.000**|
-| TruthfulQA | Factual  | 100 |    0.530 |    0.560 | **+0.030**|
-| SQuAD v2   | QA       | 100 |        — |  **97%** | 19% compression |
-| BFCL       | Tools    | 100 |        — |  **97%** | 32% compression |
-
-Reproduce:
-
-```bash
-python -m headroom.evals suite --tier 1
-```
-
-**Community, live:**
-
-<div align="center">
-  <a href="https://headroomlabs.ai/dashboard">
-    <img src="headroom-savings.png" alt="60B+ tokens saved — community leaderboard" width="820">
-  </a>
-  <p><b><a href="https://headroomlabs.ai/dashboard">60B+ tokens saved by the community in the last 20 days — live leaderboard →</a></b></p>
-</div>
-
-→ [Full benchmarks & methodology](https://headroom-docs.vercel.app/docs/benchmarks)
-
----
-
-## Built for coding agents
-
-| Agent              | Durable init / one-shot wrap       | Notes                                                            |
-|--------------------|------------------------------------|------------------------------------------------------------------|
-| **Claude Code**    | `headroom init claude -g` / `headroom wrap claude` | `init` installs user or repo-local hooks; `wrap` is still useful for ad hoc sessions |
-| **Codex**          | `headroom init codex -g` / `headroom wrap codex --memory` | `init` installs provider config plus lifecycle hooks where supported |
-| **Cursor**         | `headroom wrap cursor`             | Prints Cursor config — durable init not available yet            |
-| **Aider**          | `headroom wrap aider`              | Starts proxy, launches Aider                                     |
-| **Copilot CLI**    | `headroom init copilot -g` / `headroom wrap copilot` | `init` installs hooks and BYOK provider routing for the current user |
-| **OpenClaw**       | `headroom init openclaw -g` / `headroom wrap openclaw` | Installs Headroom as ContextEngine plugin                        |
-
-MCP-native too — `headroom mcp install` exposes `headroom_compress`, `headroom_retrieve`, and `headroom_stats` to any MCP client.
-
-<div align="center">
-  <img src="headroom_learn.gif" alt="headroom learn in action" width="720">
-</div>
-
----
-
-## Integrations
-
-<details>
-<summary><b>Drop Headroom into any stack</b></summary>
-
-| Your setup              | Hook in with                                                     |
-|-------------------------|------------------------------------------------------------------|
-| Any Python app          | `compress(messages, model=…)`                                    |
-| Any TypeScript app      | `await compress(messages, { model })`                            |
-| Anthropic / OpenAI SDK  | `withHeadroom(new Anthropic())` · `withHeadroom(new OpenAI())`   |
-| Vercel AI SDK           | `wrapLanguageModel({ model, middleware: headroomMiddleware() })` |
-| LiteLLM                 | `litellm.callbacks = [HeadroomCallback()]`                       |
-| LangChain               | `HeadroomChatModel(your_llm)`                                    |
-| Agno                    | `HeadroomAgnoModel(your_model)`                                  |
-| Strands                 | [Strands guide](https://headroom-docs.vercel.app/docs/strands) |
-| ASGI apps               | `app.add_middleware(CompressionMiddleware)`                      |
-| Multi-agent             | `SharedContext().put / .get`                                     |
-| MCP clients             | `headroom mcp install`                                           |
-
-</details>
-
-<details>
-<summary><b>What's inside</b></summary>
-
-- **SmartCrusher** — universal JSON: arrays of dicts, nested objects, mixed types.
-- **CodeCompressor** — AST-aware for Python, JS, Go, Rust, Java, C++.
-- **Kompress-base** — our HuggingFace model, trained on agentic traces.
-- **Image compression** — 40–90% reduction via trained ML router.
-- **CacheAligner** — stabilizes prefixes so Anthropic/OpenAI KV caches actually hit.
-- **IntelligentContext** — score-based context fitting with learned importance.
-- **CCR** — reversible compression; LLM retrieves originals on demand.
-- **Cross-agent memory** — shared store, agent provenance, auto-dedup.
-- **SharedContext** — compressed context passing across multi-agent workflows.
-- **`headroom learn`** — plugin-based failure mining for Claude, Codex, Gemini.
-
-</details>
-
----
-
-## Install
-
-```bash
-pip install "headroom-ai[all]"          # Python, everything
-npm  install headroom-ai                # TypeScript / Node
-docker pull ghcr.io/chopratejas/headroom:latest
-```
-
-Granular extras: `[proxy]`, `[mcp]`, `[ml]` (Kompress-base), `[agno]`, `[langchain]`, `[evals]`. Requires **Python 3.10+**.
-
-→ [Installation guide](https://headroom-docs.vercel.app/docs/installation) — Docker tags, persistent service, PowerShell, devcontainers.
-
----
-
-## Documentation
-
-| Start here                                                              | Go deeper                                                              |
-|-------------------------------------------------------------------------|------------------------------------------------------------------------|
-| [Quickstart](https://headroom-docs.vercel.app/docs/quickstart)    | [Architecture](https://headroom-docs.vercel.app/docs/architecture) |
-| [Proxy](https://headroom-docs.vercel.app/docs/proxy)              | [How compression works](https://headroom-docs.vercel.app/docs/how-compression-works) |
-| [MCP tools](https://headroom-docs.vercel.app/docs/mcp)            | [CCR — reversible compression](https://headroom-docs.vercel.app/docs/ccr) |
-| [Memory](https://headroom-docs.vercel.app/docs/memory)            | [Cache optimization](https://headroom-docs.vercel.app/docs/cache-optimization) |
-| [Failure learning](https://headroom-docs.vercel.app/docs/failure-learning) | [Benchmarks](https://headroom-docs.vercel.app/docs/benchmarks) |
-| [Configuration](https://headroom-docs.vercel.app/docs/configuration) | [Limitations](https://headroom-docs.vercel.app/docs/limitations) |
-
----
-
-## Compared to
-
-Headroom runs **locally**, covers **every** content type (not just CLI or text), works with every major framework, and is **reversible**.
-
-|                                  | Scope                                           | Deploy                              | Local | Reversible |
-|----------------------------------|-------------------------------------------------|-------------------------------------|:-----:|:----------:|
-| **Headroom**                     | All context — tools, RAG, logs, files, history  | Proxy · library · middleware · MCP  |  Yes  |    Yes     |
-| [RTK](https://github.com/rtk-ai/rtk) | CLI command outputs                         | CLI wrapper                         |  Yes  |    No      |
-| [Compresr](https://compresr.ai), [Token Co.](https://thetokencompany.ai) | Text sent to their API | Hosted API call         |  No   |    No      |
-| OpenAI Compaction                | Conversation history                            | Provider-native                     |  No   |    No      |
-
-> **Attribution.** Headroom ships with the excellent [RTK](https://github.com/rtk-ai/rtk) binary for shell-output rewriting — `git show` → `git show --short`, noisy `ls` → scoped, chatty installers → summarized. Huge thanks to the RTK team; their tool is a first-class part of our stack, and Headroom compresses everything downstream of it.
-
----
-
-## Contributing
-
-```bash
-git clone https://github.com/chopratejas/headroom.git && cd headroom
-pip install -e ".[dev]" && pytest
-```
-
-Devcontainers in `.devcontainer/` (default + `memory-stack` with Qdrant & Neo4j). See [CONTRIBUTING.md](CONTRIBUTING.md).
-
----
-
-## Community
-
-- **[Live leaderboard](https://headroomlabs.ai/dashboard)** — 60B+ tokens saved and counting.
-- **[Discord](https://discord.gg/yRmaUNpsPJ)** — questions, feedback, war stories.
-- **[Kompress-base on HuggingFace](https://huggingface.co/chopratejas/kompress-base)** — the model behind our text compression.
-
-## License
-
-Apache 2.0 — see [LICENSE](LICENSE).
+<div align="center">
+
+# Headroom
+
+**Compress everything your AI agent reads. Same answers, fraction of the tokens.**
+
+[![CI](https://github.com/chopratejas/headroom/actions/workflows/ci.yml/badge.svg)](https://github.com/chopratejas/headroom/actions/workflows/ci.yml)
+[![PyPI](https://img.shields.io/pypi/v/headroom-ai.svg)](https://pypi.org/project/headroom-ai/)
+[![npm](https://img.shields.io/npm/v/headroom-ai.svg)](https://www.npmjs.com/package/headroom-ai)
+[![Model: Kompress-base](https://img.shields.io/badge/model-Kompress--base-yellow.svg)](https://huggingface.co/chopratejas/kompress-base)
+[![Tokens saved: 60B+](https://img.shields.io/badge/tokens%20saved-60B%2B-2ea44f)](https://headroomlabs.ai/dashboard)
+[![License: Apache 2.0](https://img.shields.io/badge/license-Apache%202.0-blue.svg)](LICENSE)
+[![Docs](https://img.shields.io/badge/docs-online-blue.svg)](https://headroom-docs.vercel.app/docs)
+
+<img src="HeadroomDemo-Fast.gif" alt="Headroom in action" width="820">
+
+</div>
+
+---
+
+Every tool call, log line, DB read, RAG chunk, and file your agent injects into a prompt is mostly boilerplate. Headroom strips the noise and keeps the signal — **losslessly, locally, and without touching accuracy.**
+
+> **100 logs. One FATAL error buried at position 67. Both runs found it.**
+> Baseline **10,144 tokens** → Headroom **1,260 tokens** — **87% fewer, identical answer.**
+> `python examples/needle_in_haystack_test.py`
+
+---
+
+## Quick start
+
+Works with Anthropic, OpenAI, Google, Bedrock, Vertex, Azure, OpenRouter, and 100+ models via LiteLLM.
+
+**Wrap your coding agent — one command:**
+
+```bash
+pip install "headroom-ai[all]"
+
+headroom wrap claude      # Claude Code
+headroom wrap codex       # Codex
+headroom wrap cursor      # Cursor
+headroom wrap aider       # Aider
+headroom wrap copilot     # GitHub Copilot CLI
+```
+
+**Drop it into your own code — Python or TypeScript:**
+
+```python
+from headroom import compress
+
+result = compress(messages, model="claude-sonnet-4-5")
+response = client.messages.create(model="claude-sonnet-4-5", messages=result.messages)
+print(f"Saved {result.tokens_saved} tokens ({result.compression_ratio:.0%})")
+```
+
+```typescript
+import { compress } from 'headroom-ai';
+const result = await compress(messages, { model: 'gpt-4o' });
+```
+
+**Or run it as a proxy — zero code changes, any language:**
+
+```bash
+headroom proxy --port 8787
+ANTHROPIC_BASE_URL=http://localhost:8787 your-app
+OPENAI_BASE_URL=http://localhost:8787/v1 your-app
+```
+
+---
+
+## Why Headroom
+
+- **Accuracy-preserving.** GSM8K **0.870 → 0.870** (±0.000). TruthfulQA **+0.030**. SQuAD v2 and BFCL both **97%** accuracy after compression. Validated on public OSS benchmarks you can rerun yourself.
+- **Runs on your machine.** No cloud API, no data egress. Compression latency is milliseconds — faster end-to-end for Sonnet / Opus / GPT-4 class models than a hosted service round-trip.
+- **[Kompress-base](https://huggingface.co/chopratejas/kompress-base) on HuggingFace.** Our open-source text compressor, fine-tuned on real agentic traces — tool outputs, logs, RAG chunks, code. Install with `pip install "headroom-ai[ml]"`.
+- **Cross-agent memory and learning.** Claude Code saves a fact, Codex reads it back. `headroom learn` mines failed sessions and writes corrections straight to `CLAUDE.md` / `AGENTS.md` / `GEMINI.md` — reliability compounds over time.
+- **Reversible (CCR).** Compression is not deletion. The model can always call `headroom_retrieve` to pull the original bytes. Nothing is thrown away.
+
+Bundles the [RTK](https://github.com/rtk-ai/rtk) binary for shell-output rewriting — full [attribution below](#compared-to).
+
+---
+
+## How it fits
+
+```
+ Your agent / app
+   (Claude Code, Cursor, Codex, LangChain, Agno, Strands, your own code…)
+        │   prompts · tool outputs · logs · RAG results · files
+        ▼
+    ┌────────────────────────────────────────────────────┐
+    │  Headroom   (runs locally — your data stays here)  │
+    │  ───────────────────────────────────────────────   │
+    │  CacheAligner  →  ContentRouter  →  CCR             │
+    │                    ├─ SmartCrusher   (JSON)         │
+    │                    ├─ CodeCompressor (AST)          │
+    │                    └─ Kompress-base  (text, HF)     │
+    │                                                     │
+    │  Cross-agent memory  ·  headroom learn  ·  MCP      │
+    └────────────────────────────────────────────────────┘
+        │   compressed prompt  +  retrieval tool
+        ▼
+ LLM provider  (Anthropic · OpenAI · Bedrock · …)
+```
+
+→ [Architecture](https://headroom-docs.vercel.app/docs/architecture) · [CCR reversible compression](https://headroom-docs.vercel.app/docs/ccr) · [Kompress-base model card](https://huggingface.co/chopratejas/kompress-base)
+
+### Canonical pipeline lifecycle
+
+Headroom now exposes one stable request lifecycle across `compress()`, the SDK, and the proxy:
+
+`Setup` → `Pre-Start` → `Post-Start` → `Input Received` → `Input Cached` → `Input Routed` → `Input Compressed` → `Input Remembered` → `Pre-Send` → `Post-Send` → `Response Received`
+
+- **Transforms** still do the work: CacheAligner, ContentRouter, SmartCrusher, CodeCompressor, Kompress-base, IntelligentContext / RollingWindow.
+- **Pipeline extensions** observe or customize those lifecycle stages via `on_pipeline_event(...)`.
+- **Compression hooks** still work and now sit alongside the canonical lifecycle instead of being the only extension seam.
+- **Proxy extensions** remain the server/app integration seam for ASGI middleware, routes, and startup policy.
+
+### Provider slices
+
+Provider and tool-specific behavior is being moved behind dedicated modules under `headroom/providers/` so core orchestration stays focused on lifecycle, sequencing, and policy.
+
+- **CLI/tool slices**: `headroom/providers/claude`, `copilot`, `codex`, `openclaw`
+- **Provider runtime slices**: `headroom/providers/claude`, `gemini`, plus shared backend/runtime dispatch in `headroom/providers/registry.py`
+- **Core files stay orchestration-first**: `wrap.py`, `client.py`, `cli/proxy.py`, and `proxy/server.py` now delegate provider-specific env shaping, API target normalization, backend selection, and transport dispatch instead of inlining those rules.
+
+---
+
+## Proof
+
+**Savings on real agent workloads:**
+
+| Workload                      | Before | After  | Savings |
+|-------------------------------|-------:|-------:|--------:|
+| Code search (100 results)     | 17,765 |  1,408 | **92%** |
+| SRE incident debugging        | 65,694 |  5,118 | **92%** |
+| GitHub issue triage           | 54,174 | 14,761 | **73%** |
+| Codebase exploration          | 78,502 | 41,254 | **47%** |
+
+**Accuracy preserved on standard benchmarks:**
+
+| Benchmark  | Category | N   | Baseline | Headroom | Delta     |
+|------------|----------|----:|---------:|---------:|----------:|
+| GSM8K      | Math     | 100 |    0.870 |    0.870 | **±0.000**|
+| TruthfulQA | Factual  | 100 |    0.530 |    0.560 | **+0.030**|
+| SQuAD v2   | QA       | 100 |        — |  **97%** | 19% compression |
+| BFCL       | Tools    | 100 |        — |  **97%** | 32% compression |
+
+Reproduce:
+
+```bash
+python -m headroom.evals suite --tier 1
+```
+
+**Community, live:**
+
+<div align="center">
+  <a href="https://headroomlabs.ai/dashboard">
+    <img src="headroom-savings.png" alt="60B+ tokens saved — community leaderboard" width="820">
+  </a>
+  <p><b><a href="https://headroomlabs.ai/dashboard">60B+ tokens saved by the community in the last 20 days — live leaderboard →</a></b></p>
+</div>
+
+→ [Full benchmarks & methodology](https://headroom-docs.vercel.app/docs/benchmarks)
+
+---
+
+## Built for coding agents
+
+| Agent              | One-command wrap                   | Notes                                                            |
+|--------------------|------------------------------------|------------------------------------------------------------------|
+| **Claude Code**    | `headroom wrap claude`             | `--memory` for cross-agent memory, `--code-graph` for codebase intel |
+| **Codex**          | `headroom wrap codex --memory`     | Shares the same memory store as Claude                           |
+| **Cursor**         | `headroom wrap cursor`             | Prints Cursor config — paste once, done                          |
+| **Aider**          | `headroom wrap aider`              | Starts proxy, launches Aider                                     |
+| **Copilot CLI**    | `headroom wrap copilot`            | Starts proxy, launches Copilot                                   |
+| **OpenClaw**       | `headroom wrap openclaw`           | Installs Headroom as ContextEngine plugin                        |
+
+MCP-native too — `headroom mcp install` exposes `headroom_compress`, `headroom_retrieve`, and `headroom_stats` to any MCP client.
+
+<div align="center">
+  <img src="headroom_learn.gif" alt="headroom learn in action" width="720">
+</div>
+
+---
+
+## Integrations
+
+<details>
+<summary><b>Drop Headroom into any stack</b></summary>
+
+| Your setup              | Hook in with                                                     |
+|-------------------------|------------------------------------------------------------------|
+| Any Python app          | `compress(messages, model=…)`                                    |
+| Any TypeScript app      | `await compress(messages, { model })`                            |
+| Anthropic / OpenAI SDK  | `withHeadroom(new Anthropic())` · `withHeadroom(new OpenAI())`   |
+| Vercel AI SDK           | `wrapLanguageModel({ model, middleware: headroomMiddleware() })` |
+| LiteLLM                 | `litellm.callbacks = [HeadroomCallback()]`                       |
+| LangChain               | `HeadroomChatModel(your_llm)`                                    |
+| Agno                    | `HeadroomAgnoModel(your_model)`                                  |
+| Strands                 | [Strands guide](https://headroom-docs.vercel.app/docs/strands) |
+| ASGI apps               | `app.add_middleware(CompressionMiddleware)`                      |
+| Multi-agent             | `SharedContext().put / .get`                                     |
+| MCP clients             | `headroom mcp install`                                           |
+
+</details>
+
+<details>
+<summary><b>What's inside</b></summary>
+
+- **SmartCrusher** — universal JSON: arrays of dicts, nested objects, mixed types.
+- **CodeCompressor** — AST-aware for Python, JS, Go, Rust, Java, C++.
+- **Kompress-base** — our HuggingFace model, trained on agentic traces.
+- **Image compression** — 40–90% reduction via trained ML router.
+- **CacheAligner** — stabilizes prefixes so Anthropic/OpenAI KV caches actually hit.
+- **IntelligentContext** — score-based context fitting with learned importance.
+- **CCR** — reversible compression; LLM retrieves originals on demand.
+- **Cross-agent memory** — shared store, agent provenance, auto-dedup.
+- **SharedContext** — compressed context passing across multi-agent workflows.
+- **`headroom learn`** — plugin-based failure mining for Claude, Codex, Gemini.
+
+</details>
+
+---
+
+## Install
+
+```bash
+pip install "headroom-ai[all]"          # Python, everything
+npm  install headroom-ai                # TypeScript / Node
+docker pull ghcr.io/chopratejas/headroom:latest
+```
+
+Granular extras: `[proxy]`, `[mcp]`, `[ml]` (Kompress-base), `[agno]`, `[langchain]`, `[evals]`. Requires **Python 3.10+**.
+
+→ [Installation guide](https://headroom-docs.vercel.app/docs/installation) — Docker tags, persistent service, PowerShell, devcontainers.
+
+---
+
+## Documentation
+
+| Start here                                                              | Go deeper                                                              |
+|-------------------------------------------------------------------------|------------------------------------------------------------------------|
+| [Quickstart](https://headroom-docs.vercel.app/docs/quickstart)    | [Architecture](https://headroom-docs.vercel.app/docs/architecture) |
+| [Proxy](https://headroom-docs.vercel.app/docs/proxy)              | [How compression works](https://headroom-docs.vercel.app/docs/how-compression-works) |
+| [MCP tools](https://headroom-docs.vercel.app/docs/mcp)            | [CCR — reversible compression](https://headroom-docs.vercel.app/docs/ccr) |
+| [Memory](https://headroom-docs.vercel.app/docs/memory)            | [Cache optimization](https://headroom-docs.vercel.app/docs/cache-optimization) |
+| [Failure learning](https://headroom-docs.vercel.app/docs/failure-learning) | [Benchmarks](https://headroom-docs.vercel.app/docs/benchmarks) |
+| [Configuration](https://headroom-docs.vercel.app/docs/configuration) | [Limitations](https://headroom-docs.vercel.app/docs/limitations) |
+
+---
+
+## Compared to
+
+Headroom runs **locally**, covers **every** content type (not just CLI or text), works with every major framework, and is **reversible**.
+
+|                                  | Scope                                           | Deploy                              | Local | Reversible |
+|----------------------------------|-------------------------------------------------|-------------------------------------|:-----:|:----------:|
+| **Headroom**                     | All context — tools, RAG, logs, files, history  | Proxy · library · middleware · MCP  |  Yes  |    Yes     |
+| [RTK](https://github.com/rtk-ai/rtk) | CLI command outputs                         | CLI wrapper                         |  Yes  |    No      |
+| [Compresr](https://compresr.ai), [Token Co.](https://thetokencompany.ai) | Text sent to their API | Hosted API call         |  No   |    No      |
+| OpenAI Compaction                | Conversation history                            | Provider-native                     |  No   |    No      |
+
+> **Attribution.** Headroom ships with the excellent [RTK](https://github.com/rtk-ai/rtk) binary for shell-output rewriting — `git show` → `git show --short`, noisy `ls` → scoped, chatty installers → summarized. Huge thanks to the RTK team; their tool is a first-class part of our stack, and Headroom compresses everything downstream of it.
+
+---
+
+## Contributing
+
+```bash
+git clone https://github.com/chopratejas/headroom.git && cd headroom
+pip install -e ".[dev]" && pytest
+```
+
+Devcontainers in `.devcontainer/` (default + `memory-stack` with Qdrant & Neo4j). See [CONTRIBUTING.md](CONTRIBUTING.md).
+
+---
+
+## Community
+
+- **[Live leaderboard](https://headroomlabs.ai/dashboard)** — 60B+ tokens saved and counting.
+- **[Discord](https://discord.gg/yRmaUNpsPJ)** — questions, feedback, war stories.
+- **[Kompress-base on HuggingFace](https://huggingface.co/chopratejas/kompress-base)** — the model behind our text compression.
+
+## License
+
+Apache 2.0 — see [LICENSE](LICENSE).

headroom/providers/aider/__init__.py

@@ -0,0 +1,5 @@
+"""Aider-specific provider helpers."""
+
+from .runtime import build_launch_env
+
+__all__ = ["build_launch_env"]

headroom/providers/aider/runtime.py

@@ -0,0 +1,24 @@
+"""Runtime helpers for Aider integrations."""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Mapping
+
+from headroom.providers.claude import proxy_base_url as claude_proxy_base_url
+from headroom.providers.codex import proxy_base_url as codex_proxy_base_url
+
+
+def build_launch_env(
+    port: int, environ: Mapping[str, str] | None = None
+) -> tuple[dict[str, str], list[str]]:
+    """Build environment variables for Aider through the local proxy."""
+    env = dict(environ or os.environ)
+    openai_base_url = codex_proxy_base_url(port)
+    anthropic_base_url = claude_proxy_base_url(port)
+    env["OPENAI_API_BASE"] = openai_base_url
+    env["ANTHROPIC_BASE_URL"] = anthropic_base_url
+    return env, [
+        f"OPENAI_API_BASE={openai_base_url}",
+        f"ANTHROPIC_BASE_URL={anthropic_base_url}",
+    ]

headroom/providers/claude/__init__.py

@@ -0,0 +1,5 @@
+"""Claude-specific provider helpers."""
+
+from .runtime import DEFAULT_API_URL, proxy_base_url
+
+__all__ = ["DEFAULT_API_URL", "proxy_base_url"]

headroom/providers/claude/runtime.py

@@ -0,0 +1,10 @@
+"""Runtime helpers for Claude-facing integrations."""
+
+from __future__ import annotations
+
+DEFAULT_API_URL = "https://api.anthropic.com"
+
+
+def proxy_base_url(port: int) -> str:
+    """Return the local proxy base URL used by Claude integrations."""
+    return f"http://127.0.0.1:{port}"

headroom/providers/codex/__init__.py

@@ -0,0 +1,5 @@
+"""Codex-specific provider helpers."""
+
+from .runtime import DEFAULT_API_URL, build_launch_env, proxy_base_url
+
+__all__ = ["DEFAULT_API_URL", "build_launch_env", "proxy_base_url"]

headroom/providers/codex/runtime.py

@@ -0,0 +1,23 @@
+"""Runtime helpers for Codex/OpenAI-facing integrations."""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Mapping
+
+DEFAULT_API_URL = "https://api.openai.com"
+
+
+def proxy_base_url(port: int) -> str:
+    """Return the local proxy base URL used by OpenAI-compatible integrations."""
+    return f"http://127.0.0.1:{port}/v1"
+
+
+def build_launch_env(
+    port: int, environ: Mapping[str, str] | None = None
+) -> tuple[dict[str, str], list[str]]:
+    """Build environment variables for Codex through the local proxy."""
+    env = dict(environ or os.environ)
+    base_url = proxy_base_url(port)
+    env["OPENAI_BASE_URL"] = base_url
+    return env, [f"OPENAI_BASE_URL={base_url}"]

headroom/providers/copilot/__init__.py

@@ -0,0 +1,21 @@
+"""Copilot-specific provider helpers."""
+
+from .wrap import (
+    build_launch_env,
+    detect_running_proxy_backend,
+    model_configured,
+    provider_key_source,
+    query_proxy_config,
+    resolve_provider_type,
+    validate_configuration,
+)
+
+__all__ = [
+    "build_launch_env",
+    "detect_running_proxy_backend",
+    "model_configured",
+    "provider_key_source",
+    "query_proxy_config",
+    "resolve_provider_type",
+    "validate_configuration",
+]

headroom/providers/copilot/wrap.py

@@ -0,0 +1,120 @@
+"""Copilot wrapper provider helpers."""
+
+from __future__ import annotations
+
+import json
+import os
+import urllib.error
+import urllib.request
+from collections.abc import Mapping
+from typing import Any
+
+import click
+
+
+def resolve_provider_type(
+    backend: str | None, provider_type: str, environ: Mapping[str, str] | None = None
+) -> str:
+    """Resolve Copilot BYOK provider type for the current proxy backend."""
+    if provider_type != "auto":
+        return provider_type
+
+    env = environ or os.environ
+    effective_backend = backend or env.get("HEADROOM_BACKEND") or "anthropic"
+    return "anthropic" if effective_backend == "anthropic" else "openai"
+
+
+def query_proxy_config(port: int) -> dict[str, Any] | None:
+    """Query the running proxy's feature configuration via /health."""
+    url = f"http://127.0.0.1:{port}/health"
+    try:
+        with urllib.request.urlopen(url, timeout=2) as response:
+            payload = json.loads(response.read().decode("utf-8"))
+    except (OSError, urllib.error.URLError, ValueError, json.JSONDecodeError):
+        return None
+
+    config = payload.get("config")
+    if not isinstance(config, dict):
+        return None
+    return config
+
+
+def detect_running_proxy_backend(port: int) -> str | None:
+    """Read the backend of an already-running proxy from its health endpoint."""
+    config = query_proxy_config(port)
+    if config is None:
+        return None
+    backend = config.get("backend")
+    return backend if isinstance(backend, str) else None
+
+
+def validate_configuration(
+    *,
+    provider_type: str,
+    wire_api: str | None,
+    backend: str | None,
+) -> None:
+    """Validate Copilot BYOK provider and wire-api settings."""
+    if provider_type == "anthropic" and wire_api is not None:
+        raise click.ClickException(
+            "--wire-api is only valid when Copilot is using the openai provider type."
+        )
+    if wire_api == "responses" and backend not in (None, "anthropic"):
+        raise click.ClickException(
+            "--wire-api responses is not supported with translated backends; use completions."
+        )
+
+
+def provider_key_source(provider_type: str) -> str:
+    """Return the preferred provider key variable for the selected provider type."""
+    return "ANTHROPIC_API_KEY" if provider_type == "anthropic" else "OPENAI_API_KEY"
+
+
+def build_launch_env(
+    *,
+    port: int,
+    provider_type: str,
+    wire_api: str | None,
+    environ: Mapping[str, str] | None = None,
+) -> tuple[dict[str, str], list[str]]:
+    """Build the Copilot BYOK environment for the selected provider type."""
+    env = dict(environ or os.environ)
+    env["COPILOT_PROVIDER_TYPE"] = provider_type
+    env.pop("COPILOT_PROVIDER_WIRE_API", None)
+
+    if not env.get("COPILOT_PROVIDER_API_KEY"):
+        key = env.get(provider_key_source(provider_type), "")
+        if key:
+            env["COPILOT_PROVIDER_API_KEY"] = key
+
+    if provider_type == "anthropic":
+        base_url = f"http://127.0.0.1:{port}"
+        env["COPILOT_PROVIDER_BASE_URL"] = base_url
+        return env, [
+            "COPILOT_PROVIDER_TYPE=anthropic",
+            f"COPILOT_PROVIDER_BASE_URL={base_url}",
+        ]
+
+    effective_wire_api = wire_api or "completions"
+    base_url = f"http://127.0.0.1:{port}/v1"
+    env["COPILOT_PROVIDER_BASE_URL"] = base_url
+    env["COPILOT_PROVIDER_WIRE_API"] = effective_wire_api
+    return env, [
+        "COPILOT_PROVIDER_TYPE=openai",
+        f"COPILOT_PROVIDER_BASE_URL={base_url}",
+        f"COPILOT_PROVIDER_WIRE_API={effective_wire_api}",
+    ]
+
+
+def model_configured(copilot_args: tuple[str, ...], env: Mapping[str, str]) -> bool:
+    """Return True when Copilot BYOK model selection is configured."""
+    if env.get("COPILOT_MODEL") or env.get("COPILOT_PROVIDER_MODEL_ID"):
+        return True
+
+    for idx, arg in enumerate(copilot_args):
+        if arg == "--model" and idx + 1 < len(copilot_args):
+            return True
+        if arg.startswith("--model="):
+            return True
+
+    return False

headroom/providers/cursor/__init__.py

@@ -0,0 +1,5 @@
+"""Cursor-specific provider helpers."""
+
+from .runtime import CursorProxyTargets, build_proxy_targets, render_setup_lines
+
+__all__ = ["CursorProxyTargets", "build_proxy_targets", "render_setup_lines"]

headroom/providers/cursor/runtime.py

@@ -0,0 +1,44 @@
+"""Runtime helpers for Cursor integrations."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from headroom.providers.claude import proxy_base_url as claude_proxy_base_url
+from headroom.providers.codex import proxy_base_url as codex_proxy_base_url
+
+
+@dataclass(frozen=True)
+class CursorProxyTargets:
+    """Resolved local proxy targets shown in Cursor setup instructions."""
+
+    openai_base_url: str
+    anthropic_base_url: str
+
+
+def build_proxy_targets(port: int) -> CursorProxyTargets:
+    """Build the local proxy URLs shown to Cursor users."""
+    return CursorProxyTargets(
+        openai_base_url=codex_proxy_base_url(port),
+        anthropic_base_url=claude_proxy_base_url(port),
+    )
+
+
+def render_setup_lines(port: int) -> list[str]:
+    """Render the Cursor setup instructions for the local proxy."""
+    targets = build_proxy_targets(port)
+    return [
+        "  Headroom proxy is running. Configure Cursor:",
+        "",
+        "  For OpenAI models:",
+        f"    Base URL:  {targets.openai_base_url}",
+        "    API Key:   your-openai-api-key",
+        "",
+        "  For Anthropic models:",
+        f"    Base URL:  {targets.anthropic_base_url}",
+        "    API Key:   your-anthropic-api-key",
+        "",
+        "  In Cursor:",
+        "    Settings > Models > OpenAI API Key > Override OpenAI Base URL",
+        f"    Set to: {targets.openai_base_url}",
+    ]

headroom/providers/gemini/__init__.py

@@ -0,0 +1,5 @@
+"""Gemini-specific provider helpers."""
+
+from .runtime import DEFAULT_API_URL
+
+__all__ = ["DEFAULT_API_URL"]

headroom/providers/gemini/runtime.py

@@ -0,0 +1,5 @@
+"""Runtime helpers for Gemini-facing integrations."""
+
+from __future__ import annotations
+
+DEFAULT_API_URL = "https://generativelanguage.googleapis.com"

headroom/providers/openclaw/__init__.py

@@ -0,0 +1,15 @@
+"""OpenClaw-specific provider helpers."""
+
+from .wrap import (
+    build_plugin_entry,
+    build_unwrap_entry,
+    decode_entry_json,
+    normalize_gateway_provider_ids,
+)
+
+__all__ = [
+    "build_plugin_entry",
+    "build_unwrap_entry",
+    "decode_entry_json",
+    "normalize_gateway_provider_ids",
+]

headroom/providers/openclaw/wrap.py

@@ -0,0 +1,89 @@
+"""OpenClaw wrapper provider helpers."""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+DEFAULT_GATEWAY_PROVIDER_IDS = ["openai-codex"]
+
+
+def normalize_gateway_provider_ids(provider_ids: tuple[str, ...] | None) -> list[str]:
+    """Normalize configured OpenClaw provider ids."""
+    values = provider_ids or ()
+    seen: set[str] = set()
+    normalized: list[str] = []
+
+    for entry in values:
+        provider_id = entry.strip()
+        if not provider_id or provider_id in seen:
+            continue
+        seen.add(provider_id)
+        normalized.append(provider_id)
+
+    return normalized or DEFAULT_GATEWAY_PROVIDER_IDS.copy()
+
+
+def decode_entry_json(raw_value: str | None) -> Any | None:
+    """Decode a JSON payload captured from `openclaw config get` when available."""
+    if not raw_value:
+        return None
+
+    try:
+        return json.loads(raw_value)
+    except json.JSONDecodeError:
+        return raw_value
+
+
+def build_plugin_entry(
+    *,
+    existing_entry: Any,
+    proxy_port: int,
+    startup_timeout_ms: int,
+    python_path: str | None,
+    no_auto_start: bool,
+    gateway_provider_ids: tuple[str, ...] | None,
+    enabled: bool,
+) -> dict[str, object]:
+    """Merge managed Headroom plugin settings with any existing entry payload."""
+    base_entry = existing_entry if isinstance(existing_entry, dict) else {}
+    existing_config = base_entry.get("config")
+    next_config = dict(existing_config) if isinstance(existing_config, dict) else {}
+
+    next_config["proxyPort"] = proxy_port
+    next_config["autoStart"] = not no_auto_start
+    next_config["startupTimeoutMs"] = startup_timeout_ms
+    next_config["gatewayProviderIds"] = normalize_gateway_provider_ids(gateway_provider_ids)
+
+    if python_path:
+        next_config["pythonPath"] = python_path
+    else:
+        next_config.pop("pythonPath", None)
+
+    return {
+        **base_entry,
+        "enabled": enabled,
+        "config": next_config,
+    }
+
+
+def build_unwrap_entry(existing_entry: Any) -> dict[str, object]:
+    """Disable the managed plugin while preserving unrelated user config."""
+    base_entry = existing_entry if isinstance(existing_entry, dict) else {}
+    existing_config: dict[str, object] = {}
+    if isinstance(existing_entry, dict) and isinstance(existing_entry.get("config"), dict):
+        existing_config = {
+            key: value
+            for key, value in existing_entry["config"].items()
+            if key
+            not in {
+                "gatewayProviderIds",
+                "proxyUrl",
+                "proxyPort",
+                "autoStart",
+                "startupTimeoutMs",
+                "pythonPath",
+            }
+        }
+
+    return {**base_entry, "enabled": False, "config": existing_config}

headroom/providers/proxy_routes.py

@@ -0,0 +1,337 @@
+# mypy: disable-error-code=no-untyped-def
+"""Provider-specific proxy route registration."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from fastapi import FastAPI, Request, WebSocket
+from fastapi.responses import Response
+
+from headroom.proxy.handlers.openai import _resolve_codex_routing_headers
+
+logger = logging.getLogger("headroom.proxy.routes")
+
+
+def register_provider_routes(app: FastAPI, proxy: Any) -> None:
+    """Register provider-specific proxy endpoints."""
+
+    @app.post("/v1/messages")
+    async def anthropic_messages(request: Request):
+        return await proxy.handle_anthropic_messages(request)
+
+    @app.post("/v1/messages/count_tokens")
+    async def anthropic_count_tokens(request: Request):
+        return await proxy.handle_passthrough(
+            request,
+            proxy.provider_runtime.api_target("anthropic"),
+            "count_tokens",
+            "anthropic",
+        )
+
+    @app.post("/v1/messages/batches")
+    async def anthropic_batch_create(request: Request):
+        return await proxy.handle_anthropic_batch_create(request)
+
+    @app.get("/v1/messages/batches")
+    async def anthropic_batch_list(request: Request):
+        return await proxy.handle_anthropic_batch_passthrough(request)
+
+    @app.get("/v1/messages/batches/{batch_id}")
+    async def anthropic_batch_get(request: Request, batch_id: str):
+        return await proxy.handle_anthropic_batch_passthrough(request, batch_id)
+
+    @app.get("/v1/messages/batches/{batch_id}/results")
+    async def anthropic_batch_results(request: Request, batch_id: str):
+        return await proxy.handle_anthropic_batch_results(request, batch_id)
+
+    @app.post("/v1/messages/batches/{batch_id}/cancel")
+    async def anthropic_batch_cancel(request: Request, batch_id: str):
+        return await proxy.handle_anthropic_batch_passthrough(request, batch_id)
+
+    @app.post("/v1/chat/completions")
+    async def openai_chat(request: Request):
+        return await proxy.handle_openai_chat(request)
+
+    @app.post("/v1/responses")
+    async def openai_responses(request: Request):
+        return await proxy.handle_openai_responses(request)
+
+    @app.post("/v1/codex/responses")
+    async def openai_v1_codex_responses(request: Request):
+        return await proxy.handle_openai_responses(request)
+
+    @app.post("/backend-api/responses")
+    async def openai_codex_responses(request: Request):
+        return await proxy.handle_openai_responses(request)
+
+    @app.post("/backend-api/codex/responses")
+    async def openai_codex_nested_responses(request: Request):
+        return await proxy.handle_openai_responses(request)
+
+    @app.websocket("/v1/responses")
+    async def openai_responses_ws(websocket: WebSocket):
+        await proxy.handle_openai_responses_ws(websocket)
+
+    @app.websocket("/v1/codex/responses")
+    async def openai_v1_codex_responses_ws(websocket: WebSocket):
+        await proxy.handle_openai_responses_ws(websocket)
+
+    @app.api_route("/v1/responses/{sub_path:path}", methods=["GET", "POST", "DELETE"])
+    async def openai_responses_sub(request: Request, sub_path: str):
+        headers = dict(request.headers.items())
+        headers.pop("host", None)
+        headers, is_chatgpt_auth = _resolve_codex_routing_headers(headers)
+
+        if is_chatgpt_auth:
+            url = f"https://chatgpt.com/backend-api/codex/responses/{sub_path}"
+        else:
+            url = f"{proxy.provider_runtime.api_target('openai')}/v1/responses/{sub_path}"
+
+        if request.url.query:
+            url = f"{url}?{request.url.query}"
+
+        body = await request.body()
+        try:
+            assert proxy.http_client is not None
+            resp = await proxy.http_client.request(
+                request.method,
+                url,
+                headers=headers,
+                content=body,
+                timeout=120.0,
+            )
+            return Response(
+                content=resp.content,
+                status_code=resp.status_code,
+                headers=dict(resp.headers),
+            )
+        except Exception as exc:
+            logger.error("Passthrough /v1/responses/%s failed: %s", sub_path, exc)
+            return Response(content=str(exc), status_code=502)
+
+    @app.api_route("/v1/codex/responses/{sub_path:path}", methods=["GET", "POST", "DELETE"])
+    async def openai_v1_codex_responses_sub(request: Request, sub_path: str):
+        return await openai_responses_sub(request, sub_path)
+
+    @app.websocket("/backend-api/responses")
+    async def openai_codex_responses_ws(websocket: WebSocket):
+        await proxy.handle_openai_responses_ws(websocket)
+
+    @app.websocket("/backend-api/codex/responses")
+    async def openai_codex_nested_responses_ws(websocket: WebSocket):
+        await proxy.handle_openai_responses_ws(websocket)
+
+    @app.api_route("/backend-api/responses/{sub_path:path}", methods=["GET", "POST", "DELETE"])
+    async def openai_codex_responses_sub(request: Request, sub_path: str):
+        return await openai_responses_sub(request, sub_path)
+
+    @app.api_route(
+        "/backend-api/codex/responses/{sub_path:path}", methods=["GET", "POST", "DELETE"]
+    )
+    async def openai_codex_nested_responses_sub(request: Request, sub_path: str):
+        return await openai_responses_sub(request, sub_path)
+
+    @app.post("/v1/batches")
+    async def create_batch(request: Request):
+        return await proxy.handle_batch_create(request)
+
+    @app.get("/v1/batches")
+    async def list_batches(request: Request):
+        return await proxy.handle_batch_list(request)
+
+    @app.get("/v1/batches/{batch_id}")
+    async def get_batch(request: Request, batch_id: str):
+        return await proxy.handle_batch_get(request, batch_id)
+
+    @app.post("/v1/batches/{batch_id}/cancel")
+    async def cancel_batch(request: Request, batch_id: str):
+        return await proxy.handle_batch_cancel(request, batch_id)
+
+    @app.post("/v1beta/models/{model}:generateContent")
+    async def gemini_generate_content(request: Request, model: str):
+        return await proxy.handle_gemini_generate_content(request, model)
+
+    @app.post("/v1beta/models/{model}:streamGenerateContent")
+    async def gemini_stream_generate_content(request: Request, model: str):
+        return await proxy.handle_gemini_stream_generate_content(request, model)
+
+    @app.post("/v1beta/models/{model}:countTokens")
+    async def gemini_count_tokens(request: Request, model: str):
+        return await proxy.handle_gemini_count_tokens(request, model)
+
+    @app.post("/v1internal:streamGenerateContent")
+    async def google_cloudcode_stream_generate_content(request: Request):
+        return await proxy.handle_google_cloudcode_stream(request)
+
+    @app.post("/v1/v1internal:streamGenerateContent")
+    async def google_cloudcode_stream_generate_content_v1(request: Request):
+        return await proxy.handle_google_cloudcode_stream(request)
+
+    @app.post("/serving-endpoints/{model}/invocations")
+    async def databricks_invocations(request: Request, model: str):
+        return await proxy.handle_databricks_invocations(request, model)
+
+    @app.get("/v1/models")
+    async def list_models(request: Request):
+        provider_name = proxy.provider_runtime.model_metadata_provider(dict(request.headers))
+        return await proxy.handle_passthrough(
+            request,
+            proxy.provider_runtime.api_target(provider_name),
+            "models",
+            provider_name,
+        )
+
+    @app.get("/v1/models/{model_id}")
+    async def get_model(request: Request, model_id: str):
+        provider_name = proxy.provider_runtime.model_metadata_provider(dict(request.headers))
+        return await proxy.handle_passthrough(
+            request,
+            proxy.provider_runtime.api_target(provider_name),
+            "models",
+            provider_name,
+        )
+
+    @app.post("/v1/embeddings")
+    async def openai_embeddings(request: Request):
+        return await proxy.handle_passthrough(
+            request,
+            proxy.provider_runtime.api_target("openai"),
+            "embeddings",
+            "openai",
+        )
+
+    @app.post("/v1/moderations")
+    async def openai_moderations(request: Request):
+        return await proxy.handle_passthrough(
+            request,
+            proxy.provider_runtime.api_target("openai"),
+            "moderations",
+            "openai",
+        )
+
+    @app.post("/v1/images/generations")
+    async def openai_images_generations(request: Request):
+        return await proxy.handle_passthrough(
+            request,
+            proxy.provider_runtime.api_target("openai"),
+            "images/generations",
+            "openai",
+        )
+
+    @app.post("/v1/audio/transcriptions")
+    async def openai_audio_transcriptions(request: Request):
+        return await proxy.handle_passthrough(
+            request,
+            proxy.provider_runtime.api_target("openai"),
+            "audio/transcriptions",
+            "openai",
+        )
+
+    @app.post("/v1/audio/speech")
+    async def openai_audio_speech(request: Request):
+        return await proxy.handle_passthrough(
+            request,
+            proxy.provider_runtime.api_target("openai"),
+            "audio/speech",
+            "openai",
+        )
+
+    @app.get("/v1beta/models")
+    async def gemini_list_models(request: Request):
+        return await proxy.handle_passthrough(
+            request,
+            proxy.provider_runtime.api_target("gemini"),
+            "models",
+            "gemini",
+        )
+
+    @app.get("/v1beta/models/{model_name}")
+    async def gemini_get_model(request: Request, model_name: str):
+        return await proxy.handle_passthrough(
+            request,
+            proxy.provider_runtime.api_target("gemini"),
+            "models",
+            "gemini",
+        )
+
+    @app.post("/v1beta/models/{model}:embedContent")
+    async def gemini_embed_content(request: Request, model: str):
+        return await proxy.handle_passthrough(
+            request,
+            proxy.provider_runtime.api_target("gemini"),
+            "embedContent",
+            "gemini",
+        )
+
+    @app.post("/v1beta/models/{model}:batchEmbedContents")
+    async def gemini_batch_embed_contents(request: Request, model: str):
+        return await proxy.handle_passthrough(
+            request,
+            proxy.provider_runtime.api_target("gemini"),
+            "batchEmbedContents",
+            "gemini",
+        )
+
+    @app.post("/v1beta/models/{model}:batchGenerateContent")
+    async def gemini_batch_create(request: Request, model: str):
+        return await proxy.handle_google_batch_create(request, model)
+
+    @app.get("/v1beta/batches/{batch_name}")
+    async def gemini_batch_get(request: Request, batch_name: str):
+        return await proxy.handle_google_batch_results(request, batch_name)
+
+    @app.post("/v1beta/batches/{batch_name}:cancel")
+    async def gemini_batch_cancel(request: Request, batch_name: str):
+        return await proxy.handle_google_batch_passthrough(request, batch_name)
+
+    @app.delete("/v1beta/batches/{batch_name}")
+    async def gemini_batch_delete(request: Request, batch_name: str):
+        return await proxy.handle_google_batch_passthrough(request, batch_name)
+
+    @app.post("/v1beta/cachedContents")
+    async def gemini_create_cached_content(request: Request):
+        return await proxy.handle_passthrough(
+            request,
+            proxy.provider_runtime.api_target("gemini"),
+            "cachedContents",
+            "gemini",
+        )
+
+    @app.get("/v1beta/cachedContents")
+    async def gemini_list_cached_contents(request: Request):
+        return await proxy.handle_passthrough(
+            request,
+            proxy.provider_runtime.api_target("gemini"),
+            "cachedContents",
+            "gemini",
+        )
+
+    @app.get("/v1beta/cachedContents/{cache_id}")
+    async def gemini_get_cached_content(request: Request, cache_id: str):
+        return await proxy.handle_passthrough(
+            request,
+            proxy.provider_runtime.api_target("gemini"),
+            "cachedContents",
+            "gemini",
+        )
+
+    @app.delete("/v1beta/cachedContents/{cache_id}")
+    async def gemini_delete_cached_content(request: Request, cache_id: str):
+        return await proxy.handle_passthrough(
+            request,
+            proxy.provider_runtime.api_target("gemini"),
+            "cachedContents",
+            "gemini",
+        )
+
+    @app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
+    async def passthrough(request: Request, path: str):
+        custom_base = request.headers.get("x-headroom-base-url")
+        if custom_base:
+            return await proxy.handle_passthrough(request, custom_base.rstrip("/"))
+        return await proxy.handle_passthrough(
+            request,
+            proxy.provider_runtime.select_passthrough_base_url(dict(request.headers)),
+        )

headroom/providers/registry.py

@@ -0,0 +1,316 @@
+"""Provider runtime registry and transport helpers."""
+
+from __future__ import annotations
+
+import logging
+import os
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, cast
+
+from headroom.providers.claude import DEFAULT_API_URL as DEFAULT_ANTHROPIC_API_URL
+from headroom.providers.codex import DEFAULT_API_URL as DEFAULT_OPENAI_API_URL
+from headroom.providers.gemini import DEFAULT_API_URL as DEFAULT_GEMINI_API_URL
+
+DEFAULT_CLOUDCODE_API_URL = "https://cloudcode-pa.googleapis.com"
+
+if TYPE_CHECKING:
+    from headroom.backends.base import Backend
+    from headroom.providers.base import Provider
+
+AnyLLMBackendType: Any = None
+LiteLLMBackendType: Any = None
+
+
+@dataclass(frozen=True)
+class ProviderApiOverrides:
+    """Optional upstream API URL overrides configured for the proxy."""
+
+    anthropic: str | None = None
+    openai: str | None = None
+    gemini: str | None = None
+    cloudcode: str | None = None
+
+
+@dataclass(frozen=True)
+class ProviderApiTargets:
+    """Resolved upstream API targets after provider normalization."""
+
+    anthropic: str = DEFAULT_ANTHROPIC_API_URL
+    openai: str = DEFAULT_OPENAI_API_URL
+    gemini: str = DEFAULT_GEMINI_API_URL
+    cloudcode: str = DEFAULT_CLOUDCODE_API_URL
+
+
+@dataclass(frozen=True)
+class ProxyProviderRuntime:
+    """Provider runtime state used by the proxy server."""
+
+    api_targets: ProviderApiTargets
+    pipeline_providers: dict[str, Provider]
+
+    def api_target(self, provider_name: str) -> str:
+        """Return the resolved upstream target for a provider."""
+        return {
+            "anthropic": self.api_targets.anthropic,
+            "openai": self.api_targets.openai,
+            "gemini": self.api_targets.gemini,
+            "cloudcode": self.api_targets.cloudcode,
+        }[provider_name]
+
+    def pipeline_provider(self, provider_name: str) -> Provider:
+        """Return the pipeline provider instance for a provider."""
+        return self.pipeline_providers[provider_name]
+
+    def model_metadata_provider(self, headers: Mapping[str, str]) -> str:
+        """Resolve the upstream provider that should serve OpenAI-style model metadata."""
+        return "anthropic" if _is_anthropic_auth(headers) else "openai"
+
+    def select_passthrough_base_url(self, headers: Mapping[str, str]) -> str:
+        """Resolve the upstream base URL for catch-all passthrough requests."""
+        if _is_anthropic_auth(headers):
+            return self.api_targets.anthropic
+        if headers.get("x-goog-api-key"):
+            return self.api_targets.gemini
+        if headers.get("api-key"):
+            azure_base = headers.get("x-headroom-base-url", "")
+            if azure_base:
+                return azure_base.rstrip("/")
+        return self.api_targets.openai
+
+
+def _normalize_api_url(url: str | None, *, default: str) -> str:
+    if not url:
+        return default
+
+    normalized = url.rstrip("/")
+    if normalized.endswith("/v1"):
+        normalized = normalized[:-3]
+    return normalized
+
+
+def resolve_api_overrides(
+    *,
+    anthropic_api_url: str | None,
+    openai_api_url: str | None,
+    gemini_api_url: str | None,
+    cloudcode_api_url: str | None,
+    environ: Mapping[str, str] | None = None,
+) -> ProviderApiOverrides:
+    """Resolve provider API URL overrides from CLI/config inputs and environment."""
+    env = environ or os.environ
+    return ProviderApiOverrides(
+        anthropic=anthropic_api_url or env.get("ANTHROPIC_TARGET_API_URL"),
+        openai=openai_api_url or env.get("OPENAI_TARGET_API_URL"),
+        gemini=gemini_api_url or env.get("GEMINI_TARGET_API_URL"),
+        cloudcode=cloudcode_api_url or env.get("CLOUDCODE_TARGET_API_URL"),
+    )
+
+
+def resolve_api_targets(overrides: ProviderApiOverrides) -> ProviderApiTargets:
+    """Resolve normalized upstream provider targets from configured overrides."""
+    return ProviderApiTargets(
+        anthropic=_normalize_api_url(overrides.anthropic, default=DEFAULT_ANTHROPIC_API_URL),
+        openai=_normalize_api_url(overrides.openai, default=DEFAULT_OPENAI_API_URL),
+        gemini=_normalize_api_url(overrides.gemini, default=DEFAULT_GEMINI_API_URL),
+        cloudcode=_normalize_api_url(overrides.cloudcode, default=DEFAULT_CLOUDCODE_API_URL),
+    )
+
+
+def build_proxy_provider_runtime(config: Any) -> ProxyProviderRuntime:
+    """Build provider runtime objects and resolved targets for the proxy."""
+    from headroom.providers.anthropic import AnthropicProvider
+    from headroom.providers.openai import OpenAIProvider
+
+    api_targets = resolve_api_targets(config.provider_api_overrides)
+    return ProxyProviderRuntime(
+        api_targets=api_targets,
+        pipeline_providers={
+            "anthropic": AnthropicProvider(),
+            "openai": OpenAIProvider(),
+        },
+    )
+
+
+def create_proxy_backend(
+    *,
+    backend: str,
+    anyllm_provider: str,
+    bedrock_region: str | None,
+    logger: logging.Logger,
+) -> Backend | None:
+    """Create the optional translated backend for Anthropic proxy requests."""
+    if backend == "anthropic":
+        return None
+
+    if backend == "anyllm" or backend.startswith("anyllm-"):
+        provider = anyllm_provider
+        try:
+            backend_cls = _load_anyllm_backend()
+            instance = cast("Backend", backend_cls(provider=provider))
+            logger.info("any-llm backend enabled (provider=%s)", provider)
+            return instance
+        except ImportError as exc:
+            logger.warning("any-llm backend not available: %s", exc)
+            return None
+        except Exception as exc:  # pragma: no cover - defensive logging
+            logger.error("Failed to initialize any-llm backend: %s", exc)
+            return None
+
+    normalized_backend = backend if backend.startswith("litellm-") else f"litellm-{backend}"
+    provider = normalized_backend.replace("litellm-", "")
+    try:
+        backend_cls = _load_litellm_backend()
+        instance = cast("Backend", backend_cls(provider=provider, region=bedrock_region))
+        logger.info("LiteLLM backend enabled (provider=%s, region=%s)", provider, bedrock_region)
+        return instance
+    except ImportError as exc:
+        logger.warning("LiteLLM backend not available: %s", exc)
+        return None
+    except Exception as exc:  # pragma: no cover - defensive logging
+        logger.error("Failed to initialize LiteLLM backend: %s", exc)
+        return None
+
+
+def format_backend_status(*, backend: str, anyllm_provider: str, bedrock_region: str | None) -> str:
+    """Build the human-readable backend status string shown in CLI/server output."""
+    if backend == "anthropic":
+        return "ANTHROPIC (direct API)"
+    if backend == "anyllm" or backend.startswith("anyllm-"):
+        return f"{anyllm_provider.title()} via any-llm"
+
+    from headroom.backends.litellm import get_provider_config
+
+    provider = backend.replace("litellm-", "")
+    provider_config = get_provider_config(provider)
+    if provider_config.uses_region:
+        return f"{provider_config.display_name} via LiteLLM (region={bedrock_region})"
+    return f"{provider_config.display_name} via LiteLLM"
+
+
+def call_client_transport(
+    api_style: str,
+    client: Any,
+    *,
+    model: str,
+    messages: list[dict[str, Any]],
+    stream: bool,
+    metrics: Any,
+    **kwargs: Any,
+) -> Any:
+    """Dispatch the SDK request to the provider-specific transport handler."""
+    try:
+        transport = _CLIENT_TRANSPORTS[api_style]
+    except KeyError as exc:
+        raise ValueError(f"Unsupported api_style: {api_style}") from exc
+
+    return transport(
+        client,
+        model=model,
+        messages=messages,
+        stream=stream,
+        metrics=metrics,
+        **kwargs,
+    )
+
+
+def _load_anyllm_backend() -> Any:
+    global AnyLLMBackendType
+    if AnyLLMBackendType is None:
+        from headroom.backends.anyllm import AnyLLMBackend
+
+        AnyLLMBackendType = AnyLLMBackend
+    return AnyLLMBackendType
+
+
+def _load_litellm_backend() -> Any:
+    global LiteLLMBackendType
+    if LiteLLMBackendType is None:
+        from headroom.backends.litellm import LiteLLMBackend
+
+        LiteLLMBackendType = LiteLLMBackend
+    return LiteLLMBackendType
+
+
+def _call_openai_transport(
+    client: Any,
+    *,
+    model: str,
+    messages: list[dict[str, Any]],
+    stream: bool,
+    metrics: Any,
+    **kwargs: Any,
+) -> Any:
+    if stream:
+        response = client._original.chat.completions.create(
+            model=model,
+            messages=messages,
+            stream=True,
+            **kwargs,
+        )
+        return client._wrap_stream(response, metrics)
+
+    response = client._original.chat.completions.create(
+        model=model,
+        messages=messages,
+        stream=False,
+        **kwargs,
+    )
+
+    if hasattr(response, "usage") and response.usage:
+        metrics.tokens_output = response.usage.completion_tokens
+        if hasattr(response.usage, "prompt_tokens_details"):
+            details = response.usage.prompt_tokens_details
+            if hasattr(details, "cached_tokens"):
+                metrics.cached_tokens = details.cached_tokens
+
+    client._storage.save(metrics)
+    return response
+
+
+def _call_anthropic_transport(
+    client: Any,
+    *,
+    model: str,
+    messages: list[dict[str, Any]],
+    stream: bool,
+    metrics: Any,
+    **kwargs: Any,
+) -> Any:
+    if stream:
+        stream_manager = client._original.messages.stream(
+            model=model,
+            messages=messages,
+            **kwargs,
+        )
+        client._storage.save(metrics)
+        return stream_manager
+
+    response = client._original.messages.create(
+        model=model,
+        messages=messages,
+        **kwargs,
+    )
+
+    if hasattr(response, "usage") and response.usage:
+        metrics.tokens_output = response.usage.output_tokens
+        if hasattr(response.usage, "cache_read_input_tokens"):
+            metrics.cached_tokens = response.usage.cache_read_input_tokens
+
+    client._storage.save(metrics)
+    return response
+
+
+_ClientTransport = Callable[..., Any]
+_CLIENT_TRANSPORTS: dict[str, _ClientTransport] = {
+    "anthropic": _call_anthropic_transport,
+    "openai": _call_openai_transport,
+}
+
+
+def _is_anthropic_auth(headers: Mapping[str, str]) -> bool:
+    return bool(
+        headers.get("x-api-key")
+        or headers.get("anthropic-version")
+        or headers.get("Authorization", "").startswith("Bearer sk-ant-")
+    )

headroom/proxy/models.py

@@ -1,250 +1,262 @@
-"""Data models for the Headroom proxy.
-
-Contains configuration and data classes used across the proxy modules.
-Extracted from server.py to keep the codebase maintainable.
-"""
-
-from __future__ import annotations
-
-from dataclasses import dataclass, field
-from datetime import datetime
-from typing import Any, Literal
-
-# =============================================================================
-# Data Models
-# =============================================================================
-
-
-@dataclass
-class RequestLog:
-    """Complete log of a single request."""
-
-    request_id: str
-    timestamp: str
-    provider: str
-    model: str
-
-    # Tokens
-    input_tokens_original: int
-    input_tokens_optimized: int
-    output_tokens: int | None
-    tokens_saved: int
-    savings_percent: float
-
-    # Performance
-    optimization_latency_ms: float
-    total_latency_ms: float | None
-
-    # Metadata
-    tags: dict[str, str]
-    cache_hit: bool
-    transforms_applied: list[str]
-
-    # Waste signals detected in original messages
-    waste_signals: dict[str, int] | None = None
-
-    # Request/Response (optional, for debugging)
-    request_messages: list[dict] | None = None
-    response_content: str | None = None
-    error: str | None = None
-
-    # NOTE (Unit 2 follow-up): stage timings and session_id were briefly
-    # added here but are now emitted exclusively through
-    # ``emit_stage_timings_log`` (structured log line) and Prometheus.
-    # They were never populated on ``RequestLog`` instances, so the
-    # fields were removed to avoid confusing readers who expect
-    # them to be set. If a JSONL consumer needs them, have the consumer
-    # merge ``stage_timings`` log lines by ``request_id``.
-
-
-@dataclass
-class CacheEntry:
-    """Cached response entry."""
-
-    response_body: bytes
-    response_headers: dict[str, str]
-    created_at: datetime
-    ttl_seconds: int
-    hit_count: int = 0
-    tokens_saved_per_hit: int = 0
-
-
-@dataclass
-class RateLimitState:
-    """Token bucket rate limiter state."""
-
-    tokens: float
-    last_update: float
-
-
-@dataclass
-class ProxyConfig:
-    """Proxy configuration."""
-
-    # Server
-    host: str = "127.0.0.1"
-    port: int = 8787
-    anthropic_api_url: str | None = None  # Custom Anthropic API URL override
-    openai_api_url: str | None = None  # Custom OpenAI API URL override
-    gemini_api_url: str | None = None  # Custom Gemini API URL override
-    cloudcode_api_url: str | None = None  # Custom Cloud Code Assist API URL override
-
-    # Backend: "anthropic" (direct API), "litellm-*" (via LiteLLM), or "anyllm" (via any-llm)
-    backend: str = "anthropic"
-    bedrock_region: str = "us-west-2"
-    bedrock_profile: str | None = None
-    anyllm_provider: str = "openai"
-
-    # Optimization mode: "token" (rewrite for max compression) or
-    # "cache" (freeze prior turns for prefix-cache stability).
-    mode: str = "token"
-
-    # Optimization
-    optimize: bool = True
-    image_optimize: bool = True
-    min_tokens_to_crush: int = 500
-    max_items_after_crush: int = 50
-    keep_last_turns: int = 4
-
-    # CCR Tool Injection
-    ccr_inject_tool: bool = True
-    ccr_inject_system_instructions: bool = False
-
-    # CCR Response Handling
-    ccr_handle_responses: bool = True
-    ccr_max_retrieval_rounds: int = 3
-
-    # CCR Context Tracking
-    ccr_context_tracking: bool = True
-    ccr_proactive_expansion: bool = True
-    ccr_max_proactive_expansions: int = 2
-
-    # Code-aware compression (disabled by default — use code graph tools instead)
-    code_aware_enabled: bool = False
-
-    # Code graph live watcher (triggers incremental reindex on file changes)
-    code_graph_watcher: bool = False
-
-    # Per-tool compression profiles
-    tool_profiles: dict[str, Any] | None = None
-
-    # Read lifecycle management
-    read_lifecycle: bool = True
-
-    # Smart content routing
-    smart_routing: bool = True
-
-    # Intelligent context management
-    intelligent_context: bool = True
-    intelligent_context_scoring: bool = True
-    intelligent_context_compress_first: bool = True
-
-    # Caching
-    cache_enabled: bool = True
-    cache_ttl_seconds: int = 3600
-    cache_max_entries: int = 1000
-
-    # Rate limiting
-    rate_limit_enabled: bool = True
-    rate_limit_requests_per_minute: int = 60
-    rate_limit_tokens_per_minute: int = 100000
-
-    # Retry
-    retry_enabled: bool = True
-    retry_max_attempts: int = 3
-    retry_base_delay_ms: int = 1000
-    retry_max_delay_ms: int = 30000
-
-    # Prefix freeze
-    prefix_freeze_enabled: bool = True
-    prefix_freeze_session_ttl: int = 600
-
-    # Cost tracking
-    cost_tracking_enabled: bool = True
-    budget_limit_usd: float | None = None
-    budget_period: Literal["hourly", "daily", "monthly"] = "daily"
-
-    # Logging
-    log_requests: bool = True
-    log_file: str | None = None
-    log_full_messages: bool = False
-
-    # Fallback
-    fallback_enabled: bool = False
-    fallback_provider: str | None = None
-
-    # Timeouts
-    request_timeout_seconds: int = 300
-    connect_timeout_seconds: int = 10
-
-    # Connection pool
-    max_connections: int = 500
-    max_keepalive_connections: int = 100
-    http2: bool = True
-
-    # Memory System
-    memory_enabled: bool = False
-    memory_backend: Literal["local", "qdrant-neo4j"] = "local"
-    memory_db_path: str = ""  # Empty = auto: {cwd}/.headroom/memory.db
-    memory_inject_tools: bool = True
-    traffic_learning_enabled: bool = False
-    traffic_learning_agent_type: str = "unknown"  # Which agent is being wrapped
-    memory_use_native_tool: bool = False
-    memory_inject_context: bool = True
-    memory_top_k: int = 10
-    memory_min_similarity: float = 0.3
-    memory_qdrant_host: str = "localhost"
-    memory_qdrant_port: int = 6333
-    memory_neo4j_uri: str = "neo4j://localhost:7687"
-    memory_neo4j_user: str = "neo4j"
-    memory_neo4j_password: str = "password"
-    memory_bridge_enabled: bool = False
-    memory_bridge_md_paths: list[str] = field(default_factory=list)
-    memory_bridge_md_format: str = "auto"
-    memory_bridge_auto_import: bool = False
-    memory_bridge_export_path: str = ""
-
-    # License / Usage Reporting
-    license_key: str | None = None
-    license_cloud_url: str = "https://app.headroomlabs.ai"
-    license_report_interval: int = 300
-
-    # Compression Hooks
-    hooks: Any = None
-    pipeline_extensions: list[Any] = field(default_factory=list)
-    discover_pipeline_extensions: bool = True
-
-    # Subscription Window Tracking (Anthropic OAuth accounts)
-    subscription_tracking_enabled: bool = True
-    subscription_poll_interval_s: int = 10
-    subscription_active_window_s: int = 60
-
-    # Stateless mode — disable all filesystem writes for read-only / container deployments
-    stateless: bool = False
-
-    # Unit 4: Bounded pre-upstream concurrency for Anthropic replay storms.
-    #
-    # Caps the number of simultaneous requests allowed to run the
-    # pre-upstream phase of ``handle_anthropic_messages`` (request JSON
-    # read → deep-copy → first compression stage → memory-context lookup
-    # → first upstream connect). Prevents cold-start replay storms from
-    # monopolising the event loop / thread pool and starving ``/livez``,
-    # ``/readyz``, and new Codex WS opens. Compression stays on.
-    #
-    # ``None`` (default) -> auto-compute ``max(2, min(8, os.cpu_count() or 4))``.
-    # ``0`` or negative  -> disables the semaphore (unbounded); useful for
-    # the Unit 6 counter-factual and for deliberately reproducing the
-    # original starvation. Any positive integer is honored verbatim.
-    #
-    # CLI: ``--anthropic-pre-upstream-concurrency``.
-    # Env: ``HEADROOM_ANTHROPIC_PRE_UPSTREAM_CONCURRENCY``.
-    # Precedence: CLI > env > auto-compute.
-    anthropic_pre_upstream_concurrency: int | None = None
-    # Upper bound for waiting on the Anthropic pre-upstream semaphore
-    # before failing fast with a 503 + Retry-After. Keeps the queue bounded
-    # when all pre-upstream slots are occupied by slow/hung work.
-    anthropic_pre_upstream_acquire_timeout_seconds: float = 15.0
-    # Fail-open timeout for Anthropic memory-context lookup while the request
-    # is still holding a pre-upstream slot. Compression already has its own
-    # COMPRESSION_TIMEOUT_SECONDS guard; this bounds the memory leg too.
-    anthropic_pre_upstream_memory_context_timeout_seconds: float = 2.0
+"""Data models for the Headroom proxy.
+
+Contains configuration and data classes used across the proxy modules.
+Extracted from server.py to keep the codebase maintainable.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Any, Literal
+
+from headroom.providers.registry import ProviderApiOverrides
+
+# =============================================================================
+# Data Models
+# =============================================================================
+
+
+@dataclass
+class RequestLog:
+    """Complete log of a single request."""
+
+    request_id: str
+    timestamp: str
+    provider: str
+    model: str
+
+    # Tokens
+    input_tokens_original: int
+    input_tokens_optimized: int
+    output_tokens: int | None
+    tokens_saved: int
+    savings_percent: float
+
+    # Performance
+    optimization_latency_ms: float
+    total_latency_ms: float | None
+
+    # Metadata
+    tags: dict[str, str]
+    cache_hit: bool
+    transforms_applied: list[str]
+
+    # Waste signals detected in original messages
+    waste_signals: dict[str, int] | None = None
+
+    # Request/Response (optional, for debugging)
+    request_messages: list[dict] | None = None
+    response_content: str | None = None
+    error: str | None = None
+
+    # NOTE (Unit 2 follow-up): stage timings and session_id were briefly
+    # added here but are now emitted exclusively through
+    # ``emit_stage_timings_log`` (structured log line) and Prometheus.
+    # They were never populated on ``RequestLog`` instances, so the
+    # fields were removed to avoid confusing readers who expect
+    # them to be set. If a JSONL consumer needs them, have the consumer
+    # merge ``stage_timings`` log lines by ``request_id``.
+
+
+@dataclass
+class CacheEntry:
+    """Cached response entry."""
+
+    response_body: bytes
+    response_headers: dict[str, str]
+    created_at: datetime
+    ttl_seconds: int
+    hit_count: int = 0
+    tokens_saved_per_hit: int = 0
+
+
+@dataclass
+class RateLimitState:
+    """Token bucket rate limiter state."""
+
+    tokens: float
+    last_update: float
+
+
+@dataclass
+class ProxyConfig:
+    """Proxy configuration."""
+
+    # Server
+    host: str = "127.0.0.1"
+    port: int = 8787
+    anthropic_api_url: str | None = None  # Custom Anthropic API URL override
+    openai_api_url: str | None = None  # Custom OpenAI API URL override
+    gemini_api_url: str | None = None  # Custom Gemini API URL override
+    cloudcode_api_url: str | None = None  # Custom Cloud Code Assist API URL override
+
+    # Backend: "anthropic" (direct API), "litellm-*" (via LiteLLM), or "anyllm" (via any-llm)
+    backend: str = "anthropic"
+    bedrock_region: str = "us-west-2"
+    bedrock_profile: str | None = None
+    anyllm_provider: str = "openai"
+
+    # Optimization mode: "token" (rewrite for max compression) or
+    # "cache" (freeze prior turns for prefix-cache stability).
+    mode: str = "token"
+
+    # Optimization
+    optimize: bool = True
+    image_optimize: bool = True
+    min_tokens_to_crush: int = 500
+    max_items_after_crush: int = 50
+    keep_last_turns: int = 4
+
+    # CCR Tool Injection
+    ccr_inject_tool: bool = True
+    ccr_inject_system_instructions: bool = False
+
+    # CCR Response Handling
+    ccr_handle_responses: bool = True
+    ccr_max_retrieval_rounds: int = 3
+
+    # CCR Context Tracking
+    ccr_context_tracking: bool = True
+    ccr_proactive_expansion: bool = True
+    ccr_max_proactive_expansions: int = 2
+
+    # Code-aware compression (disabled by default — use code graph tools instead)
+    code_aware_enabled: bool = False
+
+    # Code graph live watcher (triggers incremental reindex on file changes)
+    code_graph_watcher: bool = False
+
+    # Per-tool compression profiles
+    tool_profiles: dict[str, Any] | None = None
+
+    # Read lifecycle management
+    read_lifecycle: bool = True
+
+    # Smart content routing
+    smart_routing: bool = True
+
+    # Intelligent context management
+    intelligent_context: bool = True
+    intelligent_context_scoring: bool = True
+    intelligent_context_compress_first: bool = True
+
+    # Caching
+    cache_enabled: bool = True
+    cache_ttl_seconds: int = 3600
+    cache_max_entries: int = 1000
+
+    # Rate limiting
+    rate_limit_enabled: bool = True
+    rate_limit_requests_per_minute: int = 60
+    rate_limit_tokens_per_minute: int = 100000
+
+    # Retry
+    retry_enabled: bool = True
+    retry_max_attempts: int = 3
+    retry_base_delay_ms: int = 1000
+    retry_max_delay_ms: int = 30000
+
+    # Prefix freeze
+    prefix_freeze_enabled: bool = True
+    prefix_freeze_session_ttl: int = 600
+
+    # Cost tracking
+    cost_tracking_enabled: bool = True
+    budget_limit_usd: float | None = None
+    budget_period: Literal["hourly", "daily", "monthly"] = "daily"
+
+    # Logging
+    log_requests: bool = True
+    log_file: str | None = None
+    log_full_messages: bool = False
+
+    # Fallback
+    fallback_enabled: bool = False
+    fallback_provider: str | None = None
+
+    # Timeouts
+    request_timeout_seconds: int = 300
+    connect_timeout_seconds: int = 10
+
+    # Connection pool
+    max_connections: int = 500
+    max_keepalive_connections: int = 100
+    http2: bool = True
+
+    # Memory System
+    memory_enabled: bool = False
+    memory_backend: Literal["local", "qdrant-neo4j"] = "local"
+    memory_db_path: str = ""  # Empty = auto: {cwd}/.headroom/memory.db
+    memory_inject_tools: bool = True
+    traffic_learning_enabled: bool = False
+    traffic_learning_agent_type: str = "unknown"  # Which agent is being wrapped
+    memory_use_native_tool: bool = False
+    memory_inject_context: bool = True
+    memory_top_k: int = 10
+    memory_min_similarity: float = 0.3
+    memory_qdrant_host: str = "localhost"
+    memory_qdrant_port: int = 6333
+    memory_neo4j_uri: str = "neo4j://localhost:7687"
+    memory_neo4j_user: str = "neo4j"
+    memory_neo4j_password: str = "password"
+    memory_bridge_enabled: bool = False
+    memory_bridge_md_paths: list[str] = field(default_factory=list)
+    memory_bridge_md_format: str = "auto"
+    memory_bridge_auto_import: bool = False
+    memory_bridge_export_path: str = ""
+
+    # License / Usage Reporting
+    license_key: str | None = None
+    license_cloud_url: str = "https://app.headroomlabs.ai"
+    license_report_interval: int = 300
+
+    # Compression Hooks
+    hooks: Any = None
+    pipeline_extensions: list[Any] = field(default_factory=list)
+    discover_pipeline_extensions: bool = True
+
+    # Subscription Window Tracking (Anthropic OAuth accounts)
+    subscription_tracking_enabled: bool = True
+    subscription_poll_interval_s: int = 10
+    subscription_active_window_s: int = 60
+
+    # Stateless mode — disable all filesystem writes for read-only / container deployments
+    stateless: bool = False
+
+    # Unit 4: Bounded pre-upstream concurrency for Anthropic replay storms.
+    #
+    # Caps the number of simultaneous requests allowed to run the
+    # pre-upstream phase of ``handle_anthropic_messages`` (request JSON
+    # read → deep-copy → first compression stage → memory-context lookup
+    # → first upstream connect). Prevents cold-start replay storms from
+    # monopolising the event loop / thread pool and starving ``/livez``,
+    # ``/readyz``, and new Codex WS opens. Compression stays on.
+    #
+    # ``None`` (default) -> auto-compute ``max(2, min(8, os.cpu_count() or 4))``.
+    # ``0`` or negative  -> disables the semaphore (unbounded); useful for
+    # the Unit 6 counter-factual and for deliberately reproducing the
+    # original starvation. Any positive integer is honored verbatim.
+    #
+    # CLI: ``--anthropic-pre-upstream-concurrency``.
+    # Env: ``HEADROOM_ANTHROPIC_PRE_UPSTREAM_CONCURRENCY``.
+    # Precedence: CLI > env > auto-compute.
+    anthropic_pre_upstream_concurrency: int | None = None
+    # Upper bound for waiting on the Anthropic pre-upstream semaphore
+    # before failing fast with a 503 + Retry-After. Keeps the queue bounded
+    # when all pre-upstream slots are occupied by slow/hung work.
+    anthropic_pre_upstream_acquire_timeout_seconds: float = 15.0
+    # Fail-open timeout for Anthropic memory-context lookup while the request
+    # is still holding a pre-upstream slot. Compression already has its own
+    # COMPRESSION_TIMEOUT_SECONDS guard; this bounds the memory leg too.
+    anthropic_pre_upstream_memory_context_timeout_seconds: float = 2.0
+
+    @property
+    def provider_api_overrides(self) -> ProviderApiOverrides:
+        """Return provider API URL overrides as a dedicated provider config object."""
+        return ProviderApiOverrides(
+            anthropic=self.anthropic_api_url,
+            openai=self.openai_api_url,
+            gemini=self.gemini_api_url,
+            cloudcode=self.cloudcode_api_url,
+        )

tests/test_cli/test_wrap_aider.py

@@ -0,0 +1,39 @@
+"""Tests for `headroom wrap aider` command."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+from click.testing import CliRunner
+
+from headroom.cli.main import main
+
+
+@pytest.fixture
+def runner() -> CliRunner:
+    return CliRunner()
+
+
+def test_wrap_aider_sets_provider_envs(
+    runner: CliRunner, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.chdir(tmp_path)
+    captured: dict[str, object] = {}
+
+    def fake_launch_tool(**kwargs):  # noqa: ANN003
+        captured.update(kwargs)
+
+    with patch("headroom.cli.wrap.shutil.which", return_value="aider"):
+        with patch("headroom.cli.wrap._launch_tool", side_effect=fake_launch_tool):
+            result = runner.invoke(main, ["wrap", "aider", "--no-rtk", "--", "--model", "gpt-4o"])
+
+    assert result.exit_code == 0, result.output
+    env = captured["env"]
+    assert isinstance(env, dict)
+    assert env["OPENAI_API_BASE"] == "http://127.0.0.1:8787/v1"
+    assert env["ANTHROPIC_BASE_URL"] == "http://127.0.0.1:8787"
+    assert captured["tool_label"] == "AIDER"
+    assert captured["agent_type"] == "aider"
+    assert captured["args"] == ("--model", "gpt-4o")

tests/test_cli/test_wrap_copilot.py

@@ -1,264 +1,198 @@
-"""Tests for `headroom wrap copilot` command."""
-
-from __future__ import annotations
-
-import importlib
-import sys
-import types
-from pathlib import Path
-from unittest.mock import patch
-
-import click
-import pytest
-from click.testing import CliRunner
-
-from headroom.copilot_auth import DEFAULT_API_URL
-
-fake_main_module = types.ModuleType("headroom.cli.main")
-fake_main_module.main = click.Group()
-sys.modules["headroom.cli.main"] = fake_main_module
-sys.modules.pop("headroom.cli", None)
-sys.modules.pop("headroom.cli.wrap", None)
-
-wrap_cli = importlib.import_module("headroom.cli.wrap")
-main = fake_main_module.main
-
-
-@pytest.fixture
-def runner() -> CliRunner:
-    return CliRunner()
-
-
-@pytest.fixture(autouse=True)
-def no_running_proxy(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(wrap_cli, "_check_proxy", lambda _port: False)
-
-
-def test_wrap_copilot_auto_anthropic_injects_instructions(
-    runner: CliRunner, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    monkeypatch.chdir(tmp_path)
-    monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test-dummy")
-    captured: dict[str, object] = {}
-
-    def fake_launch_tool(**kwargs):  # noqa: ANN003
-        captured.update(kwargs)
-
-    with patch("headroom.cli.wrap.shutil.which", return_value="copilot"):
-        with patch("headroom.cli.wrap._ensure_rtk_binary", return_value=Path("/tmp/rtk")):
-            with patch("headroom.cli.wrap._launch_tool", side_effect=fake_launch_tool):
-                result = runner.invoke(
-                    main,
-                    ["wrap", "copilot", "--", "--model", "claude-sonnet-4-20250514"],
-                )
-
-    assert result.exit_code == 0, result.output
-    instructions = tmp_path / ".github" / "copilot-instructions.md"
-    assert instructions.exists()
-    content = instructions.read_text()
-    assert wrap_cli._RTK_MARKER in content
-    assert "RTK (Rust Token Killer)" in content
-
-    env = captured["env"]
-    assert isinstance(env, dict)
-    assert env["COPILOT_PROVIDER_TYPE"] == "anthropic"
-    assert env["COPILOT_PROVIDER_BASE_URL"] == "http://127.0.0.1:8787"
-    assert "COPILOT_PROVIDER_WIRE_API" not in env
-    assert captured["agent_type"] == "copilot"
-    assert captured["tool_label"] == "COPILOT"
-    assert captured["args"] == ("--model", "claude-sonnet-4-20250514")
-
-
-def test_wrap_copilot_openai_backend_sets_completions_env(
-    runner: CliRunner, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    monkeypatch.setenv("OPENAI_API_KEY", "sk-test-dummy")
-    captured: dict[str, object] = {}
-
-    def fake_launch_tool(**kwargs):  # noqa: ANN003
-        captured.update(kwargs)
-
-    with patch("headroom.cli.wrap.shutil.which", return_value="copilot"):
-        with patch("headroom.cli.wrap._launch_tool", side_effect=fake_launch_tool):
-            result = runner.invoke(
-                main,
-                [
-                    "wrap",
-                    "copilot",
-                    "--no-rtk",
-                    "--backend",
-                    "anyllm",
-                    "--anyllm-provider",
-                    "groq",
-                    "--region",
-                    "us-central1",
-                    "--",
-                    "--model",
-                    "gpt-4o",
-                ],
-            )
-
-    assert result.exit_code == 0, result.output
-
-    env = captured["env"]
-    assert isinstance(env, dict)
-    assert env["COPILOT_PROVIDER_TYPE"] == "openai"
-    assert env["COPILOT_PROVIDER_BASE_URL"] == "http://127.0.0.1:8787/v1"
-    assert env["COPILOT_PROVIDER_WIRE_API"] == "completions"
-    assert captured["backend"] == "anyllm"
-    assert captured["anyllm_provider"] == "groq"
-    assert captured["region"] == "us-central1"
-    assert captured["args"] == ("--model", "gpt-4o")
-
-
-def test_wrap_copilot_auto_detects_running_proxy_backend(
-    runner: CliRunner, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    monkeypatch.setenv("OPENAI_API_KEY", "sk-test-dummy")
-    captured: dict[str, object] = {}
-
-    def fake_launch_tool(**kwargs):  # noqa: ANN003
-        captured.update(kwargs)
-
-    with patch("headroom.cli.wrap.shutil.which", return_value="copilot"):
-        with patch("headroom.cli.wrap._check_proxy", return_value=True):
-            with patch("headroom.cli.wrap._detect_running_proxy_backend", return_value="anyllm"):
-                with patch("headroom.cli.wrap._launch_tool", side_effect=fake_launch_tool):
-                    result = runner.invoke(
-                        main,
-                        ["wrap", "copilot", "--no-rtk", "--", "--model", "gpt-4o"],
-                    )
-
-    assert result.exit_code == 0, result.output
-    env = captured["env"]
-    assert isinstance(env, dict)
-    assert env["COPILOT_PROVIDER_TYPE"] == "openai"
-    assert env["COPILOT_PROVIDER_BASE_URL"] == "http://127.0.0.1:8787/v1"
-    assert env["COPILOT_PROVIDER_WIRE_API"] == "completions"
-
-
-def test_wrap_copilot_prefers_existing_oauth_session(
-    runner: CliRunner, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test-dummy")
-    captured: dict[str, object] = {}
-
-    def fake_launch_tool(**kwargs):  # noqa: ANN003
-        captured.update(kwargs)
-
-    with patch("headroom.cli.wrap.shutil.which", return_value="copilot"):
-        with patch("headroom.cli.wrap.resolve_client_bearer_token", return_value="gho-existing"):
-            with patch("headroom.cli.wrap.has_oauth_auth", return_value=True):
-                with patch("headroom.cli.wrap._launch_tool", side_effect=fake_launch_tool):
-                    result = runner.invoke(
-                        main,
-                        ["wrap", "copilot", "--no-rtk", "--", "--model", "claude-sonnet-4.6"],
-                    )
-
-    assert result.exit_code == 0, result.output
-    env = captured["env"]
-    assert isinstance(env, dict)
-    assert env["COPILOT_PROVIDER_TYPE"] == "openai"
-    assert env["COPILOT_PROVIDER_BASE_URL"] == "http://127.0.0.1:8787/v1"
-    assert env["COPILOT_PROVIDER_WIRE_API"] == "completions"
-    assert env["COPILOT_PROVIDER_BEARER_TOKEN"] == "gho-existing"
-    assert "COPILOT_PROVIDER_API_KEY" not in env
-    assert captured["openai_api_url"] == DEFAULT_API_URL
-
-
-def test_wrap_copilot_translated_backend_still_requires_byok(
-    runner: CliRunner,
-) -> None:
-    with patch("headroom.cli.wrap.shutil.which", return_value="copilot"):
-        with patch("headroom.cli.wrap.has_oauth_auth", return_value=True):
-            result = runner.invoke(
-                main,
-                [
-                    "wrap",
-                    "copilot",
-                    "--backend",
-                    "anyllm",
-                    "--",
-                    "--model",
-                    "gpt-4o",
-                ],
-            )
-
-    assert result.exit_code == 1
-    assert "Copilot BYOK mode requires a provider API key" in result.output
-
-
-def test_wrap_copilot_rejects_wire_api_for_anthropic_provider(runner: CliRunner) -> None:
-    with patch("headroom.cli.wrap.shutil.which", return_value="copilot"):
-        result = runner.invoke(
-            main,
-            [
-                "wrap",
-                "copilot",
-                "--wire-api",
-                "responses",
-                "--",
-                "--model",
-                "claude-sonnet-4-20250514",
-            ],
-        )
-
-    assert result.exit_code != 0
-    assert "--wire-api is only valid" in result.output
-
-
-def test_wrap_copilot_rejects_responses_for_translated_backends(runner: CliRunner) -> None:
-    with patch("headroom.cli.wrap.shutil.which", return_value="copilot"):
-        result = runner.invoke(
-            main,
-            [
-                "wrap",
-                "copilot",
-                "--backend",
-                "anyllm",
-                "--wire-api",
-                "responses",
-                "--",
-                "--model",
-                "gpt-4o",
-            ],
-        )
-
-    assert result.exit_code != 0
-    assert "not supported with translated backends" in result.output
-
-
-def test_wrap_copilot_clears_stale_wire_api_in_anthropic_mode(
-    runner: CliRunner, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test-dummy")
-    captured: dict[str, object] = {}
-
-    def fake_launch_tool(**kwargs):  # noqa: ANN003
-        captured.update(kwargs)
-
-    with patch("headroom.cli.wrap.shutil.which", return_value="copilot"):
-        with patch("headroom.cli.wrap._launch_tool", side_effect=fake_launch_tool):
-            result = runner.invoke(
-                main,
-                ["wrap", "copilot", "--no-rtk", "--", "--model", "claude-sonnet-4-20250514"],
-                env={
-                    "COPILOT_PROVIDER_WIRE_API": "responses",
-                    "ANTHROPIC_API_KEY": "sk-test-dummy",
-                },
-            )
-
-    assert result.exit_code == 0, result.output
-    env = captured["env"]
-    assert isinstance(env, dict)
-    assert env["COPILOT_PROVIDER_TYPE"] == "anthropic"
-    assert "COPILOT_PROVIDER_WIRE_API" not in env
-
-
-def test_wrap_copilot_fails_when_binary_missing(runner: CliRunner) -> None:
-    with patch("headroom.cli.wrap.shutil.which", return_value=None):
-        result = runner.invoke(main, ["wrap", "copilot", "--", "--model", "gpt-4o"])
-
-    assert result.exit_code == 1
-    assert "'copilot' not found in PATH" in result.output
-    assert "Install GitHub Copilot CLI" in result.output
+"""Tests for `headroom wrap copilot` command."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+from click.testing import CliRunner
+
+from headroom.cli import wrap as wrap_cli
+from headroom.cli.main import main
+
+
+@pytest.fixture
+def runner() -> CliRunner:
+    return CliRunner()
+
+
+def test_wrap_copilot_auto_anthropic_injects_instructions(
+    runner: CliRunner, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test-dummy")
+    captured: dict[str, object] = {}
+
+    def fake_launch_tool(**kwargs):  # noqa: ANN003
+        captured.update(kwargs)
+
+    with patch("headroom.cli.wrap.shutil.which", return_value="copilot"):
+        with patch("headroom.cli.wrap._ensure_rtk_binary", return_value=Path("/tmp/rtk")):
+            with patch("headroom.cli.wrap._launch_tool", side_effect=fake_launch_tool):
+                result = runner.invoke(
+                    main,
+                    ["wrap", "copilot", "--", "--model", "claude-sonnet-4-20250514"],
+                )
+
+    assert result.exit_code == 0, result.output
+    instructions = tmp_path / ".github" / "copilot-instructions.md"
+    assert instructions.exists()
+    content = instructions.read_text()
+    assert wrap_cli._RTK_MARKER in content
+    assert "RTK (Rust Token Killer)" in content
+
+    env = captured["env"]
+    assert isinstance(env, dict)
+    assert env["COPILOT_PROVIDER_TYPE"] == "anthropic"
+    assert env["COPILOT_PROVIDER_BASE_URL"] == "http://127.0.0.1:8787"
+    assert "COPILOT_PROVIDER_WIRE_API" not in env
+    assert captured["agent_type"] == "copilot"
+    assert captured["tool_label"] == "COPILOT"
+    assert captured["args"] == ("--model", "claude-sonnet-4-20250514")
+
+
+def test_wrap_copilot_openai_backend_sets_completions_env(
+    runner: CliRunner, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-test-dummy")
+    captured: dict[str, object] = {}
+
+    def fake_launch_tool(**kwargs):  # noqa: ANN003
+        captured.update(kwargs)
+
+    with patch("headroom.cli.wrap.shutil.which", return_value="copilot"):
+        with patch("headroom.cli.wrap._check_proxy", return_value=False):
+            with patch("headroom.cli.wrap._launch_tool", side_effect=fake_launch_tool):
+                result = runner.invoke(
+                    main,
+                    [
+                        "wrap",
+                        "copilot",
+                        "--no-rtk",
+                        "--backend",
+                        "anyllm",
+                        "--anyllm-provider",
+                        "groq",
+                        "--region",
+                        "us-central1",
+                        "--",
+                        "--model",
+                        "gpt-4o",
+                    ],
+                )
+
+    assert result.exit_code == 0, result.output
+
+    env = captured["env"]
+    assert isinstance(env, dict)
+    assert env["COPILOT_PROVIDER_TYPE"] == "openai"
+    assert env["COPILOT_PROVIDER_BASE_URL"] == "http://127.0.0.1:8787/v1"
+    assert env["COPILOT_PROVIDER_WIRE_API"] == "completions"
+    assert captured["backend"] == "anyllm"
+    assert captured["anyllm_provider"] == "groq"
+    assert captured["region"] == "us-central1"
+    assert captured["args"] == ("--model", "gpt-4o")
+
+
+def test_wrap_copilot_auto_detects_running_proxy_backend(
+    runner: CliRunner, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-test-dummy")
+    captured: dict[str, object] = {}
+
+    def fake_launch_tool(**kwargs):  # noqa: ANN003
+        captured.update(kwargs)
+
+    with patch("headroom.cli.wrap.shutil.which", return_value="copilot"):
+        with patch("headroom.cli.wrap._check_proxy", return_value=True):
+            with patch("headroom.cli.wrap._detect_running_proxy_backend", return_value="anyllm"):
+                with patch("headroom.cli.wrap._launch_tool", side_effect=fake_launch_tool):
+                    result = runner.invoke(
+                        main,
+                        ["wrap", "copilot", "--no-rtk", "--", "--model", "gpt-4o"],
+                    )
+
+    assert result.exit_code == 0, result.output
+    env = captured["env"]
+    assert isinstance(env, dict)
+    assert env["COPILOT_PROVIDER_TYPE"] == "openai"
+    assert env["COPILOT_PROVIDER_BASE_URL"] == "http://127.0.0.1:8787/v1"
+    assert env["COPILOT_PROVIDER_WIRE_API"] == "completions"
+
+
+def test_wrap_copilot_rejects_wire_api_for_anthropic_provider(runner: CliRunner) -> None:
+    with patch("headroom.cli.wrap.shutil.which", return_value="copilot"):
+        result = runner.invoke(
+            main,
+            [
+                "wrap",
+                "copilot",
+                "--wire-api",
+                "responses",
+                "--",
+                "--model",
+                "claude-sonnet-4-20250514",
+            ],
+        )
+
+    assert result.exit_code != 0
+    assert "--wire-api is only valid" in result.output
+
+
+def test_wrap_copilot_rejects_responses_for_translated_backends(runner: CliRunner) -> None:
+    with patch("headroom.cli.wrap.shutil.which", return_value="copilot"):
+        with patch("headroom.cli.wrap._check_proxy", return_value=False):
+            result = runner.invoke(
+                main,
+                [
+                    "wrap",
+                    "copilot",
+                    "--backend",
+                    "anyllm",
+                    "--wire-api",
+                    "responses",
+                    "--",
+                    "--model",
+                    "gpt-4o",
+                ],
+            )
+
+    assert result.exit_code != 0
+    assert "not supported with translated backends" in result.output
+
+
+def test_wrap_copilot_clears_stale_wire_api_in_anthropic_mode(
+    runner: CliRunner, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test-dummy")
+    captured: dict[str, object] = {}
+
+    def fake_launch_tool(**kwargs):  # noqa: ANN003
+        captured.update(kwargs)
+
+    with patch("headroom.cli.wrap.shutil.which", return_value="copilot"):
+        with patch("headroom.cli.wrap._launch_tool", side_effect=fake_launch_tool):
+            result = runner.invoke(
+                main,
+                ["wrap", "copilot", "--no-rtk", "--", "--model", "claude-sonnet-4-20250514"],
+                env={
+                    "COPILOT_PROVIDER_WIRE_API": "responses",
+                    "ANTHROPIC_API_KEY": "sk-test-dummy",
+                },
+            )
+
+    assert result.exit_code == 0, result.output
+    env = captured["env"]
+    assert isinstance(env, dict)
+    assert env["COPILOT_PROVIDER_TYPE"] == "anthropic"
+    assert "COPILOT_PROVIDER_WIRE_API" not in env
+
+
+def test_wrap_copilot_fails_when_binary_missing(runner: CliRunner) -> None:
+    with patch("headroom.cli.wrap.shutil.which", return_value=None):
+        result = runner.invoke(main, ["wrap", "copilot", "--", "--model", "gpt-4o"])
+
+    assert result.exit_code == 1
+    assert "'copilot' not found in PATH" in result.output
+    assert "Install GitHub Copilot CLI" in result.output

tests/test_provider_cursor.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+from headroom.providers.cursor import build_proxy_targets, render_setup_lines
+
+
+def test_cursor_proxy_targets_use_local_headroom_proxy() -> None:
+    targets = build_proxy_targets(9999)
+
+    assert targets.openai_base_url == "http://127.0.0.1:9999/v1"
+    assert targets.anthropic_base_url == "http://127.0.0.1:9999"
+
+
+def test_cursor_setup_lines_include_both_provider_urls() -> None:
+    lines = render_setup_lines(8787)
+    joined = "\n".join(lines)
+
+    assert "http://127.0.0.1:8787/v1" in joined
+    assert "http://127.0.0.1:8787" in joined

tests/test_provider_registry.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+from headroom.providers.registry import (
+    ProviderApiOverrides,
+    build_proxy_provider_runtime,
+    format_backend_status,
+    resolve_api_overrides,
+    resolve_api_targets,
+)
+from headroom.proxy.models import ProxyConfig
+
+
+def test_resolve_api_overrides_prefers_explicit_values_over_environment(monkeypatch) -> None:
+    monkeypatch.setenv("ANTHROPIC_TARGET_API_URL", "https://env.anthropic.example/v1")
+    monkeypatch.setenv("OPENAI_TARGET_API_URL", "https://env.openai.example/v1")
+
+    overrides = resolve_api_overrides(
+        anthropic_api_url="https://cli.anthropic.example/v1",
+        openai_api_url=None,
+        gemini_api_url=None,
+        cloudcode_api_url=None,
+    )
+
+    assert overrides == ProviderApiOverrides(
+        anthropic="https://cli.anthropic.example/v1",
+        openai="https://env.openai.example/v1",
+        gemini=None,
+        cloudcode=None,
+    )
+
+
+def test_resolve_api_targets_normalizes_trailing_v1() -> None:
+    targets = resolve_api_targets(
+        ProviderApiOverrides(
+            anthropic="https://anthropic.example/v1/",
+            openai="https://openai.example/v1",
+            gemini="https://gemini.example/v1",
+            cloudcode="https://cloudcode.example/v1/",
+        )
+    )
+
+    assert targets.anthropic == "https://anthropic.example"
+    assert targets.openai == "https://openai.example"
+    assert targets.gemini == "https://gemini.example"
+    assert targets.cloudcode == "https://cloudcode.example"
+
+
+def test_proxy_config_exposes_provider_api_overrides() -> None:
+    config = ProxyConfig(
+        anthropic_api_url="https://anthropic.example",
+        openai_api_url="https://openai.example",
+        gemini_api_url=None,
+        cloudcode_api_url="https://cloudcode.example",
+    )
+
+    assert config.provider_api_overrides == ProviderApiOverrides(
+        anthropic="https://anthropic.example",
+        openai="https://openai.example",
+        gemini=None,
+        cloudcode="https://cloudcode.example",
+    )
+
+
+def test_format_backend_status_for_anyllm() -> None:
+    assert (
+        format_backend_status(
+            backend="anyllm",
+            anyllm_provider="groq",
+            bedrock_region="us-central1",
+        )
+        == "Groq via any-llm"
+    )
+
+
+def test_proxy_provider_runtime_routes_model_metadata_and_passthrough() -> None:
+    runtime = build_proxy_provider_runtime(ProxyConfig())
+
+    assert runtime.model_metadata_provider({"x-api-key": "test"}) == "anthropic"
+    assert runtime.model_metadata_provider({}) == "openai"
+    assert (
+        runtime.select_passthrough_base_url({"x-goog-api-key": "test"})
+        == runtime.api_targets.gemini
+    )