Add subscription auth support (Claude Code OAuth + Codex ChatGPT)

config.yaml

@@ -9,6 +9,9 @@ proxy:
 upstream:
   openai_url: "https://api.openai.com"
   anthropic_url: "https://api.anthropic.com"
+  # Used by Codex CLI when signed in with a ChatGPT subscription
+  # (Plus/Pro/Team) instead of an OpenAI API key.
+  chatgpt_backend_url: "https://chatgpt.com/backend-api/codex"
 
 # API keys: can also be set via environment variables
 # OPENAI_API_KEY and ANTHROPIC_API_KEY

pyproject.toml

@@ -31,6 +31,7 @@ dependencies = [
     "tiktoken>=0.5.0",
     "rich>=13.0.0",
     "pyyaml>=6.0",
+    "zstandard>=0.22",
 ]
 
 [project.optional-dependencies]

tests/test_server.py

@@ -145,6 +145,76 @@ def test_messages_compression_with_system(self, mock_post, client: TestClient):
         content = forwarded_body["messages"][0]["content"]
         assert "..." in content  # database.py should be skeletonized
 
+    @patch("httpx.AsyncClient.post")
+    def test_messages_preserves_oauth_authorization(self, mock_post, client: TestClient):
+        mock_response = MagicMock()
+        mock_response.content = json.dumps({"id": "msg-test", "content": []}).encode()
+        mock_response.status_code = 200
+        mock_response.headers = {"content-type": "application/json"}
+        mock_post.return_value = mock_response
+
+        payload = {
+            "model": "claude-sonnet-4-20250514",
+            "messages": [{"role": "user", "content": "Hi"}],
+            "stream": False,
+        }
+
+        response = client.post(
+            "/v1/messages?beta=true",
+            json=payload,
+            headers={
+                "Authorization": "Bearer oauth-token",
+                "anthropic-version": "2023-06-01",
+                "anthropic-beta": "oauth-2025-04-20",
+            },
+        )
+
+        assert response.status_code == 200
+        call_args = mock_post.call_args
+        assert call_args.args[0] == "https://fake-anthropic.com/v1/messages?beta=true"
+        forwarded_headers = call_args.kwargs["headers"]
+        assert forwarded_headers["Authorization"] == "Bearer oauth-token"
+        assert forwarded_headers["anthropic-version"] == "2023-06-01"
+        assert forwarded_headers["anthropic-beta"] == "oauth-2025-04-20"
+        assert "x-api-key" not in forwarded_headers
+
+    @patch("httpx.AsyncClient.post")
+    def test_count_tokens_passes_through_anthropic_auth(
+        self, mock_post, client: TestClient
+    ):
+        mock_response = MagicMock()
+        mock_response.content = json.dumps({"input_tokens": 12}).encode()
+        mock_response.status_code = 200
+        mock_response.headers = {"content-type": "application/json"}
+        mock_post.return_value = mock_response
+
+        payload = {
+            "model": "claude-sonnet-4-20250514",
+            "messages": [{"role": "user", "content": "Count me"}],
+        }
+
+        response = client.post(
+            "/v1/messages/count_tokens?beta=true",
+            json=payload,
+            headers={
+                "Authorization": "Bearer oauth-token",
+                "anthropic-version": "2023-06-01",
+                "anthropic-beta": "oauth-2025-04-20",
+            },
+        )
+
+        assert response.status_code == 200
+        assert response.json() == {"input_tokens": 12}
+        call_args = mock_post.call_args
+        assert call_args.args[0] == (
+            "https://fake-anthropic.com/v1/messages/count_tokens?beta=true"
+        )
+        assert call_args.kwargs["json"] == payload
+        forwarded_headers = call_args.kwargs["headers"]
+        assert forwarded_headers["Authorization"] == "Bearer oauth-token"
+        assert forwarded_headers["anthropic-beta"] == "oauth-2025-04-20"
+        assert "x-api-key" not in forwarded_headers
+
 
 class TestMetrics:
     def test_metrics_tracked(self, client: TestClient, metrics: SessionMetrics):

tests/test_session_cache.py

@@ -239,6 +239,84 @@ def test_double_apply_does_not_add_extra_breakpoints(self, cache):
         markers = _count_cache_markers(body)
         assert markers == info1["breakpoints"]
 
+    def test_existing_breakpoints_consume_budget(self, cache):
+        msgs = [
+            _msg("user", [{"type": "text", "text": f"turn {i}"}])
+            for i in range(8)
+        ]
+        for msg in msgs[:3]:
+            msg["content"][0]["cache_control"] = {"type": "ephemeral"}
+        body = {
+            "system": "System prompt",
+            "messages": msgs,
+            "tools": [{"name": "Read"}],
+        }
+
+        modified, info = cache.apply(body)
+
+        assert _count_cache_markers(modified) <= MAX_BREAKPOINTS
+        assert info["breakpoints"] <= 1
+
+    def test_surplus_existing_breakpoints_are_trimmed(self, cache):
+        msgs = [
+            _msg("user", [{"type": "text", "text": f"turn {i}"}])
+            for i in range(5)
+        ]
+        body = {
+            "system": [
+                {"type": "text", "text": "system", "cache_control": {"type": "ephemeral"}},
+            ],
+            "messages": msgs,
+            "tools": [
+                {"name": "Read", "cache_control": {"type": "ephemeral"}},
+            ],
+        }
+        for msg in msgs[:3]:
+            msg["content"][0]["cache_control"] = {"type": "ephemeral"}
+
+        modified, info = cache.apply(body)
+
+        assert _count_cache_markers(modified) == MAX_BREAKPOINTS
+        assert info["breakpoints"] == 0
+
+    def test_injected_breakpoint_before_one_hour_ttl_is_promoted(self, cache):
+        body = {
+            "system": [
+                {"type": "text", "text": "first"},
+                {
+                    "type": "text",
+                    "text": "second",
+                    "cache_control": {"type": "ephemeral", "ttl": "1h"},
+                },
+            ],
+            "messages": [_msg("user", f"turn {i}") for i in range(5)],
+            "tools": [{"name": "Read"}],
+        }
+
+        modified, info = cache.apply(body)
+
+        assert modified["tools"][0]["cache_control"]["ttl"] == "1h"
+        assert modified["system"][1]["cache_control"]["ttl"] == "1h"
+        assert _cache_ttls_are_valid(modified)
+
+    def test_existing_short_request_ttls_are_normalized(self, cache):
+        body = {
+            "system": [
+                {"type": "text", "text": "a", "cache_control": {"type": "ephemeral"}},
+                {
+                    "type": "text",
+                    "text": "b",
+                    "cache_control": {"type": "ephemeral", "ttl": "1h"},
+                },
+            ],
+            "messages": [_msg("user", "short"), _msg("assistant", "ok")],
+        }
+
+        modified, info = cache.apply(body)
+
+        assert modified["system"][0]["cache_control"]["ttl"] == "1h"
+        assert _cache_ttls_are_valid(modified)
+
 
 def _count_cache_markers(body: dict) -> int:
     count = 0
@@ -259,6 +337,20 @@ def _count_cache_markers(body: dict) -> int:
     return count
 
 
+def _cache_ttls_are_valid(body: dict) -> bool:
+    seen_short = False
+    owners = SessionCache._iter_cache_control_owners(body)
+    for owner in owners:
+        cache_control = owner.get("cache_control")
+        ttl = cache_control.get("ttl") if isinstance(cache_control, dict) else None
+        is_one_hour = ttl == "1h"
+        if seen_short and is_one_hour:
+            return False
+        if not is_one_hour:
+            seen_short = True
+    return True
+
+
 # ──────────────────────────────────────────────────────────
 #  Session tracking
 # ──────────────────────────────────────────────────────────

token_tamer/config.py

@@ -25,6 +25,9 @@ class ProxyConfig:
 class UpstreamConfig:
     openai_url: str = "https://api.openai.com"
     anthropic_url: str = "https://api.anthropic.com"
+    # ChatGPT-subscription backend used by Codex CLI when the user is signed in
+    # with a ChatGPT Plus/Pro/Team account instead of an OpenAI API key.
+    chatgpt_backend_url: str = "https://chatgpt.com/backend-api/codex"
 
 
 @dataclass
@@ -126,6 +129,9 @@ def load_config(config_path: Union[str, Path, None] = None) -> Config:
         config.upstream = UpstreamConfig(
             openai_url=raw["upstream"].get("openai_url", config.upstream.openai_url),
             anthropic_url=raw["upstream"].get("anthropic_url", config.upstream.anthropic_url),
+            chatgpt_backend_url=raw["upstream"].get(
+                "chatgpt_backend_url", config.upstream.chatgpt_backend_url
+            ),
         )
 
     # API keys (env vars take priority)