diff --git a/README.md b/README.md index 0a48409e..da259aaf 100644 --- a/README.md +++ b/README.md @@ -83,6 +83,12 @@ The opt-in Images API returns base64 image data and consumes the signed-in accou ## Providers +To keep Claude models on Anthropic while switching GPT models through Codex in the +same session, opt in with `CCP_ALIAS_PROVIDER=anthropic`. See +[Anthropic passthrough](https://claude-code-proxy.raine.dev/providers/anthropic/) +for subscription authentication and thinking compatibility. + + | Provider | Account | Model selection | | ------------ | ------------------------------ | ----------------------------------------------- | | Codex | ChatGPT Plus or Pro | Registered `gpt-*` models and `-fast` variants | diff --git a/docs/astro.config.mjs b/docs/astro.config.mjs index 93880081..536faad6 100644 --- a/docs/astro.config.mjs +++ b/docs/astro.config.mjs @@ -44,6 +44,7 @@ export default defineConfig({ label: 'Providers', items: [ { label: 'Choosing a provider', slug: 'providers/choosing-a-provider' }, + { label: 'Anthropic', slug: 'providers/anthropic' }, { label: 'Codex', slug: 'providers/codex' }, { label: 'Kimi', slug: 'providers/kimi' }, { label: 'Grok', slug: 'providers/grok' }, diff --git a/docs/src/content/docs/providers/anthropic.md b/docs/src/content/docs/providers/anthropic.md new file mode 100644 index 00000000..1e40d728 --- /dev/null +++ b/docs/src/content/docs/providers/anthropic.md @@ -0,0 +1,43 @@ +--- +title: Anthropic +description: Keep Claude models on Anthropic while switching other models through the proxy in the same Claude Code session. +--- + +Anthropic passthrough is opt-in. Set `aliasProvider` to `anthropic` in `config.json`, or start the proxy with: + +```sh +CCP_ALIAS_PROVIDER=anthropic claude-code-proxy serve +``` + +This sends `claude-*` models and the built-in `haiku`, `sonnet`, `opus`, and `fable` aliases to `api.anthropic.com`. Explicit GPT, Kimi, Grok, OpenCode Go and Cursor IDs retain their provider. A preceding GPT or Kimi turn does not change where the Claude aliases go. Restart the proxy after changing `aliasProvider`. + +## Authentication and model switching + +Sign in using Claude Code. The proxy forwards the credentials Claude Code includes in each request; it does not read the keychain, store Anthropic credentials, or refresh them. Codex continues to use its own `claude-code-proxy codex auth login`. + +Launch Claude Code using its existing subscription login: + +```sh +env -u ANTHROPIC_AUTH_TOKEN -u ANTHROPIC_API_KEY \ + ANTHROPIC_BASE_URL=http://127.0.0.1:18765 \ + CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC=1 \ + claude --model claude-opus-5 +``` + +Also remove dummy credentials from Claude Code settings if configured there. `ANTHROPIC_AUTH_TOKEN=unused` replaces the subscription login and causes the Anthropic route to fail. Deliberately configured Anthropic API credentials are forwarded too. + +Use `/model gpt-6-astra` and `/model claude-opus-5` to switch between the providers. Availability still depends on the selected provider account. The model catalog advertises known aliases, but explicit new `claude-*` IDs can be forwarded without a proxy release. + +## Thinking and conversation history + +Native Anthropic thinking and `redacted_thinking` blocks remain unchanged, including signatures. Requests with no local model rewrite or foreign thinking conversion retain their original bytes, including unknown fields, cache controls and beta headers. + +When switching from GPT to Claude, unsigned thinking and thinking with proxy-owned `ccp:` signatures become `` text blocks containing only the visible summary. When switching from Claude to GPT, visible thinking summaries similarly become text; opaque Claude signatures and redacted content are not sent as Codex reasoning. Valid native Codex reasoning continues to replay in its encrypted form on Codex turns. Responses are streamed unchanged, so the conversation keeps the original provider blocks for a later switch back. + +This carries visible summaries across providers, not encrypted internal reasoning. Switch after a completed turn. Changing providers during an unfinished tool-use turn can still fail the receiving provider's history validation. + +## HTTP behavior and monitor + +`POST /v1/messages` and `POST /v1/messages/count_tokens` preserve query parameters and forward Anthropic authentication and beta headers. Local aliases and the `[1m]` suffix are resolved before forwarding. Redirects are not followed. Upstream status codes, request IDs, retry headers, response bodies and SSE events are preserved; connection-specific headers are removed. + +Anthropic requests appear in the existing monitor, including usage and streamed errors. The provider does not expose Anthropic through the proxy's OpenAI-compatible endpoints. With `aliasProvider` set to `codex` (the default) or `kimi`, routing and session affinity keep their existing behavior. diff --git a/docs/src/content/docs/providers/choosing-a-provider.md b/docs/src/content/docs/providers/choosing-a-provider.md index 401dbace..91ff7cd3 100644 --- a/docs/src/content/docs/providers/choosing-a-provider.md +++ b/docs/src/content/docs/providers/choosing-a-provider.md @@ -7,6 +7,7 @@ One `serve` process supports every provider. Choose based on the account you hav | Provider | Account | Upstream protocol | Model selection | Notable capabilities | | --- | --- | --- | --- | --- | +| [Anthropic](/providers/anthropic/) (opt-in) | Claude Code subscription login or Anthropic API credentials | Anthropic Messages passthrough | `claude-*` and built-in Claude aliases | Native tools, thinking and streaming; credentials forwarded from Claude Code | | [Codex](/providers/codex/) | ChatGPT Plus or Pro | OpenAI Responses over WebSocket or HTTP SSE | Named Codex catalog, `-fast` variants | Function tools, image input, hosted web search, reasoning summaries, optional native Responses route | | [Kimi](/providers/kimi/) | kimi.com with Kimi Code access | OpenAI-style chat completions | `kimi-for-coding` and aliases | Function tools, reasoning, image and video input | | [Grok](/providers/grok/) | grok.com | Responses API | `grok-composer-2.5-fast`, `grok-4.5`, `grok-4.6` | Function tools, reasoning, web search, X search, citations | @@ -23,7 +24,7 @@ One `serve` process supports every provider. Choose based on the account you hav ## Shared behavior -All providers route by model ID, use proxy-owned credentials, refresh tokens when supported, stream responses, translate Claude Code tool definitions, and report failures through the same Anthropic-shaped API. +All providers route by model ID, stream responses and report failures through the same Anthropic-shaped API. Translating providers use proxy-owned credentials and refresh tokens when supported. Anthropic forwards Claude Code's credentials and native request fields instead.
Account policy matters. Provider subscriptions, model access, regional availability, rate limits, and rules for unofficial clients can change. Review the terms for your account before using a provider through the proxy. diff --git a/docs/src/content/docs/reference/configuration.md b/docs/src/content/docs/reference/configuration.md index f5936f24..9325b982 100644 --- a/docs/src/content/docs/reference/configuration.md +++ b/docs/src/content/docs/reference/configuration.md @@ -64,7 +64,7 @@ All keys are optional. An unreadable file, malformed JSON, or incompatible field | `CCP_BIND_ADDRESS` | `bindAddress` | `127.0.0.1` | Listener IP address. | | `PORT` | `port` | `18765` | Listener port. | | `CCP_CONFIG_DIR` | none | Platform config directory | Replaces the configuration and file-backed auth root. | -| `CCP_ALIAS_PROVIDER` | `aliasProvider` | `codex` | Routes recognized Anthropic-style aliases through `codex` or `kimi`. | +| `CCP_ALIAS_PROVIDER` | `aliasProvider` | `codex` | Routes recognized Anthropic-style aliases through `codex` or `kimi`; `anthropic` enables [native Anthropic passthrough](/providers/anthropic/). | | `CCP_AUTO_REVIEW_MODEL` | `autoReviewModel` | `gpt-5.6-luna` for Codex | Routes Claude Code's non-streaming, tool-free Bash security-review classifier through a registered model. | | `CCP_LOG_STDERR` | `log.stderr` | `false` | Mirrors logs to stderr when present in the environment, regardless of its value. | | `CCP_LOG_VERBOSE` | `log.verbose` | `false` | Preserves full string fields in structured logs when present, regardless of its value. | diff --git a/docs/src/content/docs/using/models-and-routing.md b/docs/src/content/docs/using/models-and-routing.md index 25419253..200dc67a 100644 --- a/docs/src/content/docs/using/models-and-routing.md +++ b/docs/src/content/docs/using/models-and-routing.md @@ -81,4 +81,6 @@ ANTHROPIC_SMALL_FAST_MODEL=gpt-5.6-luna[1m] \ ## Alias routing -`CCP_ALIAS_PROVIDER=kimi` or `"aliasProvider": "kimi"` routes recognized Anthropic-style aliases to Kimi. Accepted values are `codex` and `kimi`. Explicit provider IDs always use their provider. +`CCP_ALIAS_PROVIDER=kimi` or `"aliasProvider": "kimi"` routes recognized Anthropic-style aliases to Kimi. Accepted values are `codex`, `kimi`, and `anthropic`. Explicit provider IDs always use their provider. + +With `anthropic`, Claude aliases and explicit `claude-*` IDs go to [Anthropic](/providers/anthropic/) using the credentials Claude Code sends. They remain on Anthropic even after a GPT or Kimi turn establishes session affinity. Existing `codex` and `kimi` alias behavior is unchanged. diff --git a/docs/src/content/docs/using/switching-models-and-backends.md b/docs/src/content/docs/using/switching-models-and-backends.md index 83901817..36f8281a 100644 --- a/docs/src/content/docs/using/switching-models-and-backends.md +++ b/docs/src/content/docs/using/switching-models-and-backends.md @@ -12,6 +12,8 @@ Claude Code binds its base URL and client auth when the process starts. A **back | Toggle between proxy and direct Anthropic | Use a launch wrapper controlled by a flag | | Stay on the proxy and change provider or model | Use `/model`, `--model`, or a new `ANTHROPIC_MODEL` | +To use Claude and GPT in one proxy-backed session, enable [Anthropic passthrough](/providers/anthropic/) with `CCP_ALIAS_PROVIDER=anthropic`. Claude Code keeps its subscription login and `/model` selects the provider per request. Switching the base URL itself still requires a new Claude Code process. + ## One-shot aliases ```sh diff --git a/src/config.rs b/src/config.rs index 15c0b072..e10f37b1 100644 --- a/src/config.rs +++ b/src/config.rs @@ -7,6 +7,7 @@ use crate::paths; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum AliasProvider { + Anthropic, Codex, Kimi, } @@ -14,6 +15,7 @@ pub enum AliasProvider { impl AliasProvider { pub fn as_str(&self) -> &str { match self { + AliasProvider::Anthropic => "anthropic", AliasProvider::Codex => "codex", AliasProvider::Kimi => "kimi", } @@ -122,6 +124,7 @@ struct FileLog { fn parse_alias(raw: &str) -> Option { match raw { + "anthropic" => Some(AliasProvider::Anthropic), "codex" => Some(AliasProvider::Codex), "kimi" => Some(AliasProvider::Kimi), _ => None, diff --git a/src/main.rs b/src/main.rs index 68f85548..a6dae2d0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -224,7 +224,7 @@ fn run_provider_cli(name: &str, command: ProviderGroup) -> Result<()> { fn print_models(registry: &Registry, full: bool) { let grouped = registry.grouped_models(); - for provider in ["codex", "kimi", "grok", "opencode", "cursor"] { + for provider in ["anthropic", "codex", "kimi", "grok", "opencode", "cursor"] { let Some(models) = grouped.get(provider) else { continue; }; @@ -277,7 +277,12 @@ fn print_server_banner(bind_address: &str, port: u16, registry: &Registry) { println!(); println!("Configure Claude Code (pick a model from above):"); println!(" export ANTHROPIC_BASE_URL=\"http://localhost:{port}\""); - println!(" export ANTHROPIC_AUTH_TOKEN=\"anything\""); + if registry.provider("anthropic").is_some() { + println!(" # Keep Claude Code's subscription login for Claude models:"); + println!(" unset ANTHROPIC_AUTH_TOKEN ANTHROPIC_API_KEY"); + } else { + println!(" export ANTHROPIC_AUTH_TOKEN=\"anything\""); + } println!(" export ANTHROPIC_MODEL=\"gpt-5.6-sol\""); println!(" export ANTHROPIC_SMALL_FAST_MODEL=\"gpt-5.6-luna\""); println!(" export CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC=1"); diff --git a/src/provider.rs b/src/provider.rs index 2631ce93..9888ae6a 100644 --- a/src/provider.rs +++ b/src/provider.rs @@ -4,10 +4,14 @@ use crate::request_identity::ConversationIdentity; use crate::traffic::TrafficCapture; use anyhow::Result; use async_trait::async_trait; -use axum::{body::Body, http::StatusCode, response::Response}; +use axum::{ + body::Body, + http::{Request, StatusCode}, + response::Response, +}; use bytes::Bytes; use clap::Subcommand; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; #[derive(Debug, Clone, Subcommand)] pub enum AuthCommand { @@ -26,6 +30,16 @@ pub trait Provider: Send + Sync { fn name(&self) -> &'static str; fn supported_models(&self) -> Vec; fn cli(&self) -> &'static dyn CliHandlers; + + async fn handle_request(&self, request: ProviderRequest, ctx: RequestContext) -> Response { + match request.endpoint { + RequestEndpoint::Messages(identity) => { + self.handle_messages_with_conversation_identity(request.body, ctx, identity) + .await + } + RequestEndpoint::CountTokens => self.handle_count_tokens(request.body, ctx).await, + } + } async fn handle_messages(&self, body: MessagesRequest, ctx: RequestContext) -> Response; async fn handle_messages_with_conversation_identity( @@ -56,6 +70,40 @@ pub trait Provider: Send + Sync { } } +/// The parsed routing input and the original HTTP representation travel together. +/// Translating providers use `body`; passthrough providers retain fields unknown +/// to the proxy by forwarding `original`. +pub struct ProviderRequest { + pub body: MessagesRequest, + pub original: Request, + pub endpoint: RequestEndpoint, +} + +pub enum RequestEndpoint { + Messages(Option), + CountTokens, +} + +/// Records a protocol failure discovered after response headers were sent. +#[derive(Clone, Default)] +pub struct ResponseOutcome { + failure: Arc>>, +} + +impl ResponseOutcome { + pub fn failure(&self) -> Option { + self.failure.lock().ok().and_then(|failure| failure.clone()) + } + + pub(crate) fn fail(&self, message: String) { + if let Ok(mut failure) = self.failure.lock() + && failure.is_none() + { + *failure = Some(message); + } + } +} + pub enum GenerationBody { BufferedSse(Bytes), LiveSse(Body), diff --git a/src/providers/anthropic/headers.rs b/src/providers/anthropic/headers.rs new file mode 100644 index 00000000..78187757 --- /dev/null +++ b/src/providers/anthropic/headers.rs @@ -0,0 +1,74 @@ +use http::{HeaderMap, HeaderName, header}; + +pub(super) fn forwarded_headers(headers: &HeaderMap) -> HeaderMap { + let connection_headers: Vec = headers + .get_all(header::CONNECTION) + .iter() + .filter_map(|value| value.to_str().ok()) + .flat_map(|value| value.split(',')) + .filter_map(|name| HeaderName::from_bytes(name.trim().as_bytes()).ok()) + .collect(); + let mut forwarded = HeaderMap::new(); + for (name, value) in headers { + if !connection_headers.contains(name) + && !matches!( + name.as_str(), + "host" + | "connection" + | "keep-alive" + | "proxy-authenticate" + | "proxy-authorization" + | "te" + | "trailer" + | "transfer-encoding" + | "upgrade" + | "content-length" + ) + { + forwarded.append(name.clone(), value.clone()); + } + } + forwarded +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn strips_connection_headers_but_preserves_credentials_and_encoding() { + let mut headers = HeaderMap::new(); + for (name, value) in [ + ("connection", "keep-alive, X-Private"), + ("x-private", "hop"), + ("host", "localhost"), + ("content-length", "12"), + ("authorization", "Bearer token"), + ("x-api-key", "key"), + ("anthropic-beta", "beta"), + ("content-encoding", "gzip"), + ] { + headers.insert(HeaderName::from_static(name), value.parse().unwrap()); + } + headers.append(header::CONNECTION, "X-Other".parse().unwrap()); + headers.insert("x-other", "hop".parse().unwrap()); + let forwarded = forwarded_headers(&headers); + for name in [ + "connection", + "x-private", + "x-other", + "host", + "content-length", + ] { + assert!(!forwarded.contains_key(name)); + } + for name in [ + "authorization", + "x-api-key", + "anthropic-beta", + "content-encoding", + ] { + assert_eq!(forwarded[name], headers[name]); + } + } +} diff --git a/src/providers/anthropic/mod.rs b/src/providers/anthropic/mod.rs new file mode 100644 index 00000000..8196fbb0 --- /dev/null +++ b/src/providers/anthropic/mod.rs @@ -0,0 +1,151 @@ +mod headers; +mod request; +mod response; + +use std::time::Duration; + +use async_trait::async_trait; +use axum::{http::StatusCode, response::Response}; +use reqwest::{Client, Url}; + +use crate::{ + anthropic::{json_error, schema::MessagesRequest}, + provider::{CliHandlers, Provider, ProviderRequest, RequestContext}, + registry::ANTHROPIC_STYLE_ALIASES, +}; + +pub struct AnthropicProvider { + client: Client, + base_url: Url, +} + +impl AnthropicProvider { + pub fn new() -> Self { + let client = Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .connect_timeout(Duration::from_secs(10)) + .read_timeout(Duration::from_secs(300)) + .build() + .expect("Anthropic HTTP client"); + Self::with_client( + client, + Url::parse("https://api.anthropic.com").expect("Anthropic URL"), + ) + } + + pub fn with_client(client: Client, base_url: Url) -> Self { + Self { client, base_url } + } + + async fn relay(&self, request: ProviderRequest, ctx: RequestContext) -> Response { + let Some(model) = request.body.model.as_deref() else { + return json_error( + StatusCode::BAD_REQUEST, + "invalid_request_error", + "Missing model", + ); + }; + let model = request::resolve_model(model); + let (parts, raw) = request.original.into_parts(); + let body = match request::prepare_body(raw, model) { + Ok(body) => body, + Err(error) => { + return json_error( + StatusCode::BAD_REQUEST, + "invalid_request_error", + error.to_string(), + ); + } + }; + let mut url = self.base_url.clone(); + url.set_path(parts.uri.path()); + url.set_query(parts.uri.query()); + let mut headers = headers::forwarded_headers(&parts.headers); + headers.insert( + http::header::ACCEPT_ENCODING, + http::HeaderValue::from_static("identity"), + ); + if let Some(monitor) = &ctx.monitor { + monitor.model_resolved(&ctx.req_id, model); + monitor.upstream_started(&ctx.req_id); + } + match self + .client + .post(url) + .headers(headers) + .body(body) + .send() + .await + { + Ok(upstream) => response::relay(upstream, ctx), + Err(error) => json_error( + StatusCode::BAD_GATEWAY, + "api_error", + format!("Anthropic request failed: {}", error.without_url()), + ), + } + } +} + +impl Default for AnthropicProvider { + fn default() -> Self { + Self::new() + } +} + +#[async_trait] +impl Provider for AnthropicProvider { + fn name(&self) -> &'static str { + "anthropic" + } + + fn supported_models(&self) -> Vec { + ANTHROPIC_STYLE_ALIASES + .iter() + .map(|model| (*model).to_string()) + .collect() + } + + fn cli(&self) -> &'static dyn CliHandlers { + &ANTHROPIC_CLI + } + + async fn handle_request(&self, request: ProviderRequest, ctx: RequestContext) -> Response { + self.relay(request, ctx).await + } + + async fn handle_messages(&self, _body: MessagesRequest, _ctx: RequestContext) -> Response { + original_request_required() + } + + async fn handle_count_tokens(&self, _body: MessagesRequest, _ctx: RequestContext) -> Response { + original_request_required() + } +} + +fn original_request_required() -> Response { + json_error( + StatusCode::BAD_REQUEST, + "invalid_request_error", + "Anthropic passthrough requires the original Messages HTTP request and client credentials", + ) +} + +struct AnthropicCli; +static ANTHROPIC_CLI: AnthropicCli = AnthropicCli; + +impl CliHandlers for AnthropicCli { + fn login(&self) -> anyhow::Result<()> { + anyhow::bail!("Sign in with Claude Code; Anthropic credentials are forwarded per request") + } + fn device(&self) -> anyhow::Result<()> { + self.login() + } + fn status(&self) -> anyhow::Result<()> { + println!("Anthropic credentials are forwarded from Claude Code; the proxy stores none"); + Ok(()) + } + fn logout(&self) -> anyhow::Result<()> { + anyhow::bail!("Sign out in Claude Code; the proxy stores no Anthropic credentials") + } +} diff --git a/src/providers/anthropic/request.rs b/src/providers/anthropic/request.rs new file mode 100644 index 00000000..9aa1f35a --- /dev/null +++ b/src/providers/anthropic/request.rs @@ -0,0 +1,119 @@ +use bytes::Bytes; +use serde_json::Value; + +use crate::providers::translate_shared::previous_reasoning_text; + +pub(super) fn resolve_model(model: &str) -> &str { + match model { + "haiku" => "claude-haiku-4-5", + "sonnet" => "claude-sonnet-5", + "opus" => "claude-opus-5", + "fable" => "claude-fable-5", + _ => model, + } +} + +/// Preserve the original bytes unless a local alias or foreign reasoning needs +/// translation. Unknown Anthropic fields and native signatures remain intact. +pub(super) fn prepare_body(raw: Bytes, model: &str) -> Result { + let mut document: Value = serde_json::from_slice(&raw)?; + let mut changed = document.get("model").and_then(Value::as_str) != Some(model); + if changed { + document["model"] = Value::String(model.to_owned()); + } + if let Some(messages) = document.get_mut("messages").and_then(Value::as_array_mut) { + for message in messages { + if message.get("role").and_then(Value::as_str) != Some("assistant") { + continue; + } + if let Some(blocks) = message.get_mut("content").and_then(Value::as_array_mut) { + for block in blocks { + if let Some(text) = foreign_reasoning(block) { + *block = serde_json::json!({"type": "text", "text": previous_reasoning_text(text)}); + changed = true; + } + } + } + } + } + if changed { + serde_json::to_vec(&document).map(Bytes::from) + } else { + Ok(raw) + } +} + +fn foreign_reasoning(block: &Value) -> Option<&str> { + if block.get("type").and_then(Value::as_str) != Some("thinking") { + return None; + } + let signature = block.get("signature").and_then(Value::as_str); + // Only interpret the proxy's own namespace. Anthropic signatures are opaque. + if signature.is_some_and(|signature| !signature.is_empty() && !is_proxy_signature(signature)) { + return None; + } + block.get("thinking").and_then(Value::as_str) +} + +fn is_proxy_signature(signature: &str) -> bool { + // Kimi and OpenCode encode the complete `ccp:kimi:v1:` prefix as base64url. + signature.starts_with("ccp:") || signature.starts_with("Y2NwOmtpbWk6djE6") +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn encoded_kimi_reasoning_is_not_mistaken_for_anthropic_thinking() { + let signature = + crate::providers::kimi::translate::signature::make_thinking_signature("msg_1", 2); + let raw = Bytes::from(json!({"model":"claude-opus-5", "messages":[{"role":"assistant","content":[{"type":"thinking","thinking":"Kimi summary","signature":signature}]}]}).to_string()); + let document: Value = + serde_json::from_slice(&prepare_body(raw, "claude-opus-5").unwrap()).unwrap(); + assert_eq!( + document["messages"][0]["content"][0], + json!({"type":"text","text":previous_reasoning_text("Kimi summary")}) + ); + } + + #[test] + fn native_requests_preserve_exact_bytes_and_unknown_fields() { + let raw = Bytes::from_static(br#"{ "model":"claude-opus-5", "future":true, "messages":[{"role":"assistant","content":[{"type":"thinking","thinking":"native","signature":"opaque"},{"type":"redacted_thinking","data":"encrypted"},{"type":"future_block","value":12}]}] }"#); + assert_eq!(prepare_body(raw.clone(), "claude-opus-5").unwrap(), raw); + } + + #[test] + fn foreign_reasoning_becomes_text_without_disturbing_tool_history() { + let mut document = json!({"model":"claude-opus-5", "messages":[{"role":"assistant", "content":[ + {"type":"thinking","thinking":"unsigned","signature":""}, + {"type":"thinking","thinking":"summary","signature":"ccp:codex:v1:cnNfMQ:encrypted"}, + {"type":"thinking","thinking":"native","signature":"opaque"}, + {"type":"redacted_thinking","data":"opaque"}, + {"type":"tool_use","id":"call1","name":"Read","input":{"path":"x"}} + ]}, {"role":"user","content":[{"type":"tool_result","tool_use_id":"call1","content":"value"}]}]}); + let raw = Bytes::from(serde_json::to_vec(&document).unwrap()); + let result: Value = + serde_json::from_slice(&prepare_body(raw, "claude-opus-5").unwrap()).unwrap(); + for (index, summary) in [(0, "unsigned"), (1, "summary")] { + document["messages"][0]["content"][index] = + json!({"type":"text","text":previous_reasoning_text(summary)}); + } + assert_eq!(result, document); + assert!(!result.to_string().contains("ccp:codex")); + } + + #[test] + fn model_rewrite_preserves_native_reasoning_and_is_idempotent() { + let raw = Bytes::from_static(br#"{"model":"opus[1m]","messages":[{"role":"assistant","content":[{"type":"thinking","thinking":"native","signature":"opaque"}]}]}"#); + let prepared = prepare_body(raw, resolve_model("opus")).unwrap(); + let document: Value = serde_json::from_slice(&prepared).unwrap(); + assert_eq!(document["model"], "claude-opus-5"); + assert_eq!(document["messages"][0]["content"][0]["signature"], "opaque"); + assert_eq!( + prepare_body(prepared.clone(), "claude-opus-5").unwrap(), + prepared + ); + } +} diff --git a/src/providers/anthropic/response.rs b/src/providers/anthropic/response.rs new file mode 100644 index 00000000..7a5a1db8 --- /dev/null +++ b/src/providers/anthropic/response.rs @@ -0,0 +1,156 @@ +use axum::{body::Body, response::Response}; +use futures_util::StreamExt; +use http::{HeaderMap, header}; +use serde_json::Value; + +use crate::{ + openai_compat::stream::SseDecoder, + provider::{RequestContext, ResponseOutcome}, +}; + +use super::headers::forwarded_headers; + +const MAX_JSON_OBSERVATION_BYTES: usize = 2 * 1024 * 1024; + +enum Observation { + Events { + decoder: SseDecoder, + completed: bool, + }, + Json(Vec), + Opaque, +} + +impl Observation { + fn for_headers(headers: &HeaderMap) -> Self { + if headers + .get(header::CONTENT_ENCODING) + .is_some_and(|encoding| encoding != "identity") + { + return Self::Opaque; + } + let media_type = headers + .get(header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.split(';').next()) + .map(str::trim); + match media_type { + Some("text/event-stream") => Self::Events { + decoder: SseDecoder::default(), + completed: false, + }, + Some("application/json") => Self::Json(Vec::new()), + _ => Self::Opaque, + } + } + + fn receive(&mut self, bytes: &[u8], ctx: &RequestContext, outcome: &ResponseOutcome) { + match self { + Self::Events { decoder, completed } => { + match decoder.push(bytes) { + Ok(events) => { + for event in events { + if event.data.get("type").and_then(Value::as_str) + == Some("message_stop") + { + *completed = true; + } + observe_value(&event.data, ctx, outcome); + } + } + Err(_) => { + // Observation must never alter the relayed bytes or grow without bound. + *self = Self::Opaque; + } + } + if let Some(monitor) = &ctx.monitor { + monitor.stream_progress(&ctx.req_id, bytes.len() as u64, 1, None, None); + } + } + Self::Json(buffer) + if buffer.len().saturating_add(bytes.len()) <= MAX_JSON_OBSERVATION_BYTES => + { + buffer.extend_from_slice(bytes); + } + Self::Json(_) => *self = Self::Opaque, + Self::Opaque => {} + } + } + + fn finish(self, ctx: &RequestContext, outcome: &ResponseOutcome) { + match self { + Self::Json(buffer) => { + if let Ok(value) = serde_json::from_slice(&buffer) { + observe_value(&value, ctx, outcome); + } + } + Self::Events { + completed: false, .. + } => { + outcome.fail("Anthropic stream ended before message_stop".into()); + } + _ => {} + } + } +} + +fn observe_value(value: &Value, ctx: &RequestContext, outcome: &ResponseOutcome) { + if value.get("type").and_then(Value::as_str) == Some("error") { + outcome.fail( + value + .pointer("/error/message") + .and_then(Value::as_str) + .unwrap_or("Anthropic stream error") + .to_owned(), + ); + } + if let Some(monitor) = &ctx.monitor { + for usage in [value.get("usage"), value.pointer("/message/usage")] + .into_iter() + .flatten() + { + monitor.usage_updated( + &ctx.req_id, + usage.get("input_tokens").and_then(Value::as_u64), + usage.get("output_tokens").and_then(Value::as_u64), + ); + } + if let Some(tokens) = value.get("input_tokens").and_then(Value::as_u64) { + monitor.usage_updated(&ctx.req_id, Some(tokens), None); + } + if value.get("type").and_then(Value::as_str) == Some("message_start") { + monitor.generation_started(&ctx.req_id); + } + } +} + +pub(super) fn relay(upstream: reqwest::Response, ctx: RequestContext) -> Response { + let observation = Observation::for_headers(upstream.headers()); + let outcome = ResponseOutcome::default(); + let response = http::Response::builder().status(upstream.status()); + let headers = forwarded_headers(upstream.headers()); + let stream = futures_util::stream::unfold( + (upstream.bytes_stream(), observation, ctx, outcome.clone()), + |(mut upstream, mut observation, ctx, outcome)| async move { + match upstream.next().await { + Some(chunk) => { + if let Ok(bytes) = &chunk { + observation.receive(bytes, &ctx, &outcome); + } + Some((chunk, (upstream, observation, ctx, outcome))) + } + None => { + observation.finish(&ctx, &outcome); + None + } + } + }, + ); + let response = headers.iter().fold(response, |response, (name, value)| { + response.header(name, value) + }); + response + .extension(outcome) + .body(Body::from_stream(stream)) + .expect("valid upstream response") +} diff --git a/src/providers/codex/native.rs b/src/providers/codex/native.rs index 63995b98..afed66e5 100644 --- a/src/providers/codex/native.rs +++ b/src/providers/codex/native.rs @@ -1,6 +1,6 @@ use std::io; use std::pin::Pin; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; use std::time::Duration; use axum::body::Body; @@ -213,24 +213,7 @@ pub fn openai_error( .into_response() } -#[derive(Clone, Default)] -pub struct NativeResponseOutcome { - failure: Arc>>, -} - -impl NativeResponseOutcome { - pub fn failure(&self) -> Option { - self.failure.lock().ok().and_then(|failure| failure.clone()) - } - - pub(crate) fn fail(&self, message: String) { - if let Ok(mut failure) = self.failure.lock() - && failure.is_none() - { - *failure = Some(message); - } - } -} +pub use crate::provider::ResponseOutcome as NativeResponseOutcome; fn passthrough_response( upstream: reqwest::Response, diff --git a/src/providers/codex/translate/request.rs b/src/providers/codex/translate/request.rs index da416353..e2dadcb2 100644 --- a/src/providers/codex/translate/request.rs +++ b/src/providers/codex/translate/request.rs @@ -8,7 +8,7 @@ use crate::anthropic::schema::MessagesRequest; use crate::config; use crate::providers::translate_shared::{ ContentBlock, flatten_system_text, image_source_to_url, normalize_content, parallel_tool_calls, - read_effort, + previous_reasoning_text, read_effort, }; use super::read_rewrite::{ReadOffsetRewrite, read_offset_rewrite}; @@ -898,18 +898,24 @@ fn build_input(req: &MessagesRequest) -> Vec { arguments: args, }); } - ContentBlock::Thinking { signature, .. } => { - let Some(replay) = + ContentBlock::Thinking { + thinking, + signature, + } => { + if let Some(replay) = signature.as_deref().and_then(decode_reasoning_signature) - else { - continue; - }; - flush_text(&mut out, &mut text_parts); - out.push(ResponsesInputItem::Reasoning { - id: replay.id, - summary: Vec::new(), - encrypted_content: replay.encrypted_content, - }); + { + flush_text(&mut out, &mut text_parts); + out.push(ResponsesInputItem::Reasoning { + id: replay.id, + summary: Vec::new(), + encrypted_content: replay.encrypted_content, + }); + } else if !thinking.is_empty() { + text_parts.push(ResponsesContentPart::OutputText { + text: previous_reasoning_text(thinking), + }); + } } _ => {} } @@ -2342,4 +2348,28 @@ mod tests { Some(ResponsesInputItem::Message { role, .. }) if role == "assistant" )); } + + #[test] + fn foreign_thinking_preserves_visible_summary_and_tool_order() { + let req: MessagesRequest = serde_json::from_value(json!({ + "model":"gpt-6-astra", "messages":[{"role":"assistant","content":[ + {"type":"thinking","thinking":"Claude summary","signature":"opaque-native-signature"}, + {"type":"redacted_thinking","data":"opaque-redaction"}, + {"type":"tool_use","id":"call1","name":"Read","input":{"path":"a"}}, + {"type":"text","text":"answer"} + ]},{"role":"user","content":[{"type":"tool_result","tool_use_id":"call1","content":"result"}]}] + })).unwrap(); + let out = serde_json::to_value(translate_request(&req, opts()).unwrap()).unwrap(); + let input = out["input"].as_array().unwrap(); + assert_eq!( + input[0]["content"][0]["text"], + "\nClaude summary\n" + ); + assert_eq!(input[1]["type"], "function_call"); + assert_eq!(input[1]["call_id"], "call1"); + assert_eq!(input[2]["content"][0]["text"], "answer"); + assert_eq!(input[3]["type"], "function_call_output"); + assert!(!out.to_string().contains("opaque-native-signature")); + assert!(!out.to_string().contains("opaque-redaction")); + } } diff --git a/src/providers/mod.rs b/src/providers/mod.rs index 8c4f82fe..1d61fcd1 100644 --- a/src/providers/mod.rs +++ b/src/providers/mod.rs @@ -1,3 +1,4 @@ +pub mod anthropic; pub mod codex; pub mod cursor; pub mod grok; diff --git a/src/providers/translate_shared.rs b/src/providers/translate_shared.rs index e98d8718..fbf4cceb 100644 --- a/src/providers/translate_shared.rs +++ b/src/providers/translate_shared.rs @@ -2,6 +2,10 @@ use serde_json::Value; use crate::anthropic::schema::MessagesRequest; +pub(crate) fn previous_reasoning_text(reasoning: &str) -> String { + format!("\n{reasoning}\n") +} + #[derive(Debug)] pub enum ContentBlock { Text { diff --git a/src/registry.rs b/src/registry.rs index ae052779..f977ecfe 100644 --- a/src/registry.rs +++ b/src/registry.rs @@ -62,6 +62,15 @@ pub struct Registry { impl Registry { pub fn new(alias_provider: AliasProvider) -> Self { let mut models: BTreeMap> = BTreeMap::new(); + if alias_provider == AliasProvider::Anthropic { + models.insert( + "anthropic".into(), + ANTHROPIC_STYLE_ALIASES + .iter() + .map(|model| (*model).to_string()) + .collect(), + ); + } models.insert("codex".into(), expand_codex_models()); models.insert( "kimi".into(), @@ -83,6 +92,7 @@ impl Registry { let mut handlers = BTreeMap::new(); for (name, entries) in &models { let handler: Arc = match name.as_str() { + "anthropic" => Arc::new(crate::providers::anthropic::AnthropicProvider::new()), "codex" => Arc::new(crate::providers::codex::CodexProvider::new()), "kimi" => Arc::new(crate::providers::kimi::KimiProvider::new()), "cursor" => Arc::new(crate::providers::cursor::CursorProvider::new()), @@ -169,6 +179,12 @@ impl Registry { session_affinity: Option<&AliasProvider>, ) -> Option> { let normalized = normalize_incoming_model(raw_model); + // Explicit Claude selection must survive a preceding Codex/Kimi turn. + if self.alias_provider == AliasProvider::Anthropic + && (is_anthropic_alias(&normalized) || normalized.starts_with("claude-")) + { + return self.handlers.get("anthropic").cloned(); + } if is_anthropic_alias(&normalized) { let target = session_affinity.unwrap_or(&self.alias_provider); return self.handlers.get(target.as_str()).cloned(); @@ -341,6 +357,59 @@ fn build_cursor_models() -> Vec { mod tests { use super::*; + #[test] + fn anthropic_routing_is_opt_in_and_ignores_other_provider_affinity() { + let native = Registry::new(AliasProvider::Anthropic); + let default = Registry::new(AliasProvider::Codex); + for model in ["haiku", "opus", "claude-opus-5", "claude-future-model[1m]"] { + assert_eq!( + native + .provider_for_model(model, Some(&AliasProvider::Codex)) + .unwrap() + .name(), + "anthropic" + ); + assert_eq!( + native + .provider_for_model(model, Some(&AliasProvider::Kimi)) + .unwrap() + .name(), + "anthropic" + ); + } + assert_eq!( + native + .provider_for_model("gpt-6-astra", None) + .unwrap() + .name(), + "codex" + ); + assert_eq!( + native.provider_for_model("kimi-k3", None).unwrap().name(), + "kimi" + ); + assert!(native.provider_for_model("unknown", None).is_none()); + assert!(default.provider("anthropic").is_none()); + assert!( + default + .provider_for_model("claude-future-model", None) + .is_none() + ); + assert_eq!( + default + .provider_for_model("opus", Some(&AliasProvider::Kimi)) + .unwrap() + .name(), + "kimi" + ); + assert!( + native + .all_supported_models() + .iter() + .any(|(model, provider)| model == "claude-opus-5" && provider == "anthropic") + ); + } + #[test] fn normalize_model_trims_hint() { assert_eq!(normalize_incoming_model("gpt-5.4-fast[1m]"), "gpt-5.4-fast"); diff --git a/src/server.rs b/src/server.rs index 913a71cc..11d46ee1 100644 --- a/src/server.rs +++ b/src/server.rs @@ -8,7 +8,7 @@ use crate::{ stream::openai_response as render_openai_response, }, project, - provider::RequestContext, + provider::{ProviderRequest, RequestContext, RequestEndpoint, ResponseOutcome}, providers::codex::{ chat_completions::{ChatCompletionsBackend, request::translate_request}, images::{ @@ -16,9 +16,7 @@ use crate::{ MAX_GENERATION_REQUEST_BYTES, MultipartEditInput, UploadedImage, image_error_response, prepare_json_request, prepare_multipart_edit, }, - native::{ - CodexNativeBackend, NativeResponseOutcome, openai_error, validate_native_request_model, - }, + native::{CodexNativeBackend, openai_error, validate_native_request_model}, transcription::{ CodexTranscriptionBackend, MAX_TRANSCRIPTION_REQUEST_BYTES, TranscriptionRequestError, prepare_transcription, transcription_error_response, @@ -1380,14 +1378,15 @@ async fn dispatch_request( let started_at = Instant::now(); let log = create_logger("server"); let req_id = Uuid::new_v4().to_string(); - let method = req.method().clone(); - let uri = req.uri().clone(); - let headers = req.headers().clone(); + let (parts, incoming_body) = req.into_parts(); + let method = &parts.method; + let uri = &parts.uri; + let headers = &parts.headers; let conversation_identity = (!count_tokens) - .then(|| ConversationIdentity::from_headers(&headers)) + .then(|| ConversationIdentity::from_headers(headers)) .flatten(); let path = uri.path().to_string(); - let query = redacted_query(&uri); + let query = redacted_query(uri); let endpoint = if count_tokens { EndpointKind::CountTokens } else { @@ -1402,8 +1401,7 @@ async fn dispatch_request( ("query".to_string(), json!(&query)), ])), ); - let session_id = req - .headers() + let session_id = headers .get("x-claude-code-session-id") .and_then(|value| value.to_str().ok()) .map(std::string::ToString::to_string); @@ -1412,7 +1410,7 @@ async fn dispatch_request( } let request_guard = RequestMonitorGuard::new(state.monitor.clone(), req_id.clone()); let now = current_millis(); - let body_bytes = match axum::body::to_bytes(req.into_body(), MAX_OPENAI_REQUEST_BYTES).await { + let body_bytes = match axum::body::to_bytes(incoming_body, MAX_OPENAI_REQUEST_BYTES).await { Ok(bytes) => bytes, Err(err) => { let response = json_error( @@ -1704,7 +1702,7 @@ async fn dispatch_request( "method": method.as_str(), "path": &path, "query": &query, - "headers": headers_to_record(&headers), + "headers": headers_to_record(headers), }), ); capture.write_json( @@ -1722,21 +1720,25 @@ async fn dispatch_request( monitor: state.monitor.clone(), }; - let response = if count_tokens { - provider.handle_count_tokens(body, context).await + let endpoint = if count_tokens { + RequestEndpoint::CountTokens } else { - provider - .handle_messages_with_conversation_identity( - body, - context, - if auto_review_route.is_some() { - None - } else { - conversation_identity - }, - ) - .await + RequestEndpoint::Messages(if auto_review_route.is_some() { + None + } else { + conversation_identity + }) }; + let response = provider + .handle_request( + ProviderRequest { + body, + original: Request::from_parts(parts, body_bytes), + endpoint, + }, + context, + ) + .await; log_request_completed( &log, RequestLogContext { @@ -1810,10 +1812,7 @@ fn with_request_id(mut response: Response, req_id: &str) -> Response { fn monitor_response_body(response: Response, guard: RequestMonitorGuard) -> Response { let status = response.status(); - let outcome = response - .extensions() - .get::() - .cloned(); + let outcome = response.extensions().get::().cloned(); let (mut parts, body) = response.into_parts(); // Stamped on the parts before the body is streamed, so a streaming SSE // response carries the header too. @@ -1828,8 +1827,7 @@ fn monitor_response_body(response: Response, guard: RequestMonitorGuard) -> Resp Some((Err(err), (body, guard, outcome))) } None => { - if let Some(message) = outcome.as_ref().and_then(NativeResponseOutcome::failure) - { + if let Some(message) = outcome.as_ref().and_then(ResponseOutcome::failure) { guard.failed(status, message); } else if status.is_success() { guard.completed(status); diff --git a/src/tui.rs b/src/tui.rs index 25766c6c..ceb499f9 100644 --- a/src/tui.rs +++ b/src/tui.rs @@ -1662,7 +1662,7 @@ fn mock_setup_text(port: u16, registry: &Registry) -> String { pub fn setup_text(port: u16, registry: &Registry) -> String { let grouped = registry.grouped_models(); - let model_summary = ["codex", "kimi", "cursor"] + let model_summary = ["anthropic", "codex", "kimi", "cursor"] .into_iter() .filter_map(|provider| { grouped @@ -1679,7 +1679,11 @@ pub fn setup_text(port: u16, registry: &Registry) -> String { lines.push(format!( "export ANTHROPIC_BASE_URL=\"http://localhost:{port}\"" )); - lines.push("export ANTHROPIC_AUTH_TOKEN=\"anything\"".to_string()); + lines.push(if registry.provider("anthropic").is_some() { + "unset ANTHROPIC_AUTH_TOKEN ANTHROPIC_API_KEY # use Claude Code's login".to_string() + } else { + "export ANTHROPIC_AUTH_TOKEN=\"anything\"".to_string() + }); lines.push("export ANTHROPIC_MODEL=\"gpt-5.6-sol\"".to_string()); lines.push("export ANTHROPIC_SMALL_FAST_MODEL=\"gpt-5.6-luna\"".to_string()); lines.push("export CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC=1".to_string()); diff --git a/tests/anthropic_passthrough.rs b/tests/anthropic_passthrough.rs new file mode 100644 index 00000000..78ca1292 --- /dev/null +++ b/tests/anthropic_passthrough.rs @@ -0,0 +1,259 @@ +use std::{convert::Infallible, sync::Arc, time::Duration}; + +use axum::{ + Router, + body::{Body, Bytes}, + http::{HeaderMap, Request, StatusCode}, + response::IntoResponse, + routing::post, +}; +use claude_code_proxy::{ + config::AliasProvider, + monitor::{MonitorHandle, RequestStatus}, + providers::anthropic::AnthropicProvider, + registry::Registry, + server::app_with_monitor, +}; +use http_body_util::BodyExt; +use serde_json::{Value, json}; +use tokio::{ + net::TcpListener, + sync::{Mutex, Notify}, + task::JoinHandle, +}; +use tower::ServiceExt; + +struct Upstream { + url: reqwest::Url, + task: JoinHandle<()>, +} + +impl Drop for Upstream { + fn drop(&mut self) { + self.task.abort(); + } +} + +async fn upstream(app: Router) -> Upstream { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()) + .parse() + .unwrap(); + let task = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + Upstream { url, task } +} + +fn proxy(upstream: &Upstream, monitor: &MonitorHandle) -> Router { + let client = reqwest::Client::builder() + .no_proxy() + .redirect(reqwest::redirect::Policy::none()) + .build() + .unwrap(); + let provider: Arc = + Arc::new(AnthropicProvider::with_client(client, upstream.url.clone())); + let registry = Arc::new(Registry::from_providers( + AliasProvider::Anthropic, + [provider], + )); + app_with_monitor(registry, Some(monitor.clone())) +} + +fn request(path: &str, body: impl Into) -> Request { + Request::post(path) + .header("content-type", "application/json") + .header("authorization", "Bearer claude-subscription") + .header( + "anthropic-beta", + "oauth-2025-04-20,prompt-caching-2024-07-31", + ) + .header("anthropic-version", "2023-06-01") + .body(body.into()) + .unwrap() +} + +#[tokio::test] +async fn relays_native_body_credentials_query_and_json_usage() { + let captured = Arc::new(Mutex::new(None)); + let sink = captured.clone(); + let upstream = upstream(Router::new().route("/v1/messages", post(move |req: Request| { + let sink = sink.clone(); + async move { + let (parts, body) = req.into_parts(); + let body = axum::body::to_bytes(body, usize::MAX).await.unwrap(); + *sink.lock().await = Some((parts, body)); + ([("request-id", "req_anthropic"), ("content-type", "application/json")], + r#"{"type":"message","usage":{"input_tokens":17,"output_tokens":8},"content":[{"type":"thinking","signature":"native","thinking":"summary"}]}"#) + } + }))).await; + let monitor = MonitorHandle::default(); + let raw = r#"{ "model":"claude-opus-5", "messages":[], "future_field":{"enabled":true} }"#; + let response = proxy(&upstream, &monitor) + .oneshot(request("/v1/messages?beta=true", raw)) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.headers()["request-id"], "req_anthropic"); + let bytes = response.into_body().collect().await.unwrap().to_bytes(); + assert!( + std::str::from_utf8(&bytes) + .unwrap() + .contains("\"signature\":\"native\"") + ); + let captured = captured.lock().await; + let (parts, body) = captured.as_ref().unwrap(); + assert_eq!(body, raw.as_bytes()); + assert_eq!(parts.uri.query(), Some("beta=true")); + assert_eq!(parts.headers["authorization"], "Bearer claude-subscription"); + assert_eq!(parts.headers["anthropic-version"], "2023-06-01"); + assert!( + parts.headers["anthropic-beta"] + .to_str() + .unwrap() + .contains("oauth") + ); + let snapshot = monitor.snapshot(); + assert_eq!(snapshot.recent[0].provider.as_deref(), Some("anthropic")); + assert_eq!(snapshot.recent[0].input_tokens, Some(17)); + assert_eq!(snapshot.recent[0].output_tokens, Some(8)); + assert_eq!(snapshot.recent[0].status, RequestStatus::Completed); +} + +#[tokio::test] +async fn streams_before_completion_and_observes_fragmented_usage() { + let release = Arc::new(Notify::new()); + let released = release.clone(); + let upstream = upstream(Router::new().route("/v1/messages", post(move || { + let released = released.clone(); + async move { + let chunks = futures_util::stream::unfold(0, move |stage| { + let released = released.clone(); + async move { + let data = match stage { + 0 => "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"usage\":{\"input_tokens\":7,\"output_tokens\":0}}}\n\nevent: message_delta\nda", + 1 => { released.notified().await; "ta: {\"type\":\"message_delta\",\"usage\":{\"output_tokens\":9}}\n\nevent: message_stop\ndata: {\"type\":\"message_stop\"}\n\n" }, + _ => return None, + }; + Some((Ok::<_, Infallible>(Bytes::from_static(data.as_bytes())), stage + 1)) + } + }); + ([("content-type", "text/event-stream")], Body::from_stream(chunks)) + } + }))).await; + let monitor = MonitorHandle::default(); + let response = proxy(&upstream, &monitor) + .oneshot(request( + "/v1/messages", + r#"{"model":"claude-opus-5","stream":true,"messages":[]}"#, + )) + .await + .unwrap(); + let mut body = response.into_body(); + let first = tokio::time::timeout(Duration::from_secs(2), body.frame()) + .await + .unwrap() + .unwrap() + .unwrap() + .into_data() + .unwrap(); + assert!(first.starts_with(b"event: message_start")); + assert_eq!(monitor.snapshot().active[0].input_tokens, Some(7)); + release.notify_one(); + let rest = body.collect().await.unwrap().to_bytes(); + assert!(rest.ends_with(b"data: {\"type\":\"message_stop\"}\n\n")); + let snapshot = monitor.snapshot(); + assert_eq!(snapshot.recent[0].output_tokens, Some(9)); + assert_eq!(snapshot.recent[0].status, RequestStatus::Completed); +} + +#[tokio::test] +async fn token_counting_applies_the_same_model_and_thinking_conversion() { + let upstream = upstream(Router::new().route("/v1/messages/count_tokens", post(|headers: HeaderMap, axum::Json(body): axum::Json| async move { + assert_eq!(headers["authorization"], "Bearer claude-subscription"); + assert_eq!(body["model"], "claude-opus-5"); + assert_eq!(body["messages"][0]["content"][0], json!({"type":"text","text":"\nGPT summary\n"})); + axum::Json(json!({"input_tokens":42})) + }))).await; + let monitor = MonitorHandle::default(); + let body = json!({"model":"opus[1m]","messages":[{"role":"assistant","content":[{"type":"thinking","thinking":"GPT summary","signature":"ccp:codex:v1:cnNfMQ:ciphertext"}]}]}); + let response = proxy(&upstream, &monitor) + .oneshot(request( + "/v1/messages/count_tokens?beta=true", + body.to_string(), + )) + .await + .unwrap(); + let bytes = response.into_body().collect().await.unwrap().to_bytes(); + assert_eq!( + serde_json::from_slice::(&bytes).unwrap(), + json!({"input_tokens":42}) + ); + assert_eq!(monitor.snapshot().recent[0].input_tokens, Some(42)); +} + +#[tokio::test] +async fn preserves_upstream_errors_and_does_not_follow_redirects() { + for (status, body) in [ + ( + StatusCode::TOO_MANY_REQUESTS, + r#"{"type":"error","error":{"type":"rate_limit_error","message":"quota"}}"#, + ), + (StatusCode::TEMPORARY_REDIRECT, "redirect"), + ] { + let upstream = upstream(Router::new().route( + "/v1/messages", + post(move || async move { + ( + status, + [ + ("retry-after", "13"), + ("location", "http://127.0.0.1:1/never"), + ("request-id", "real-id"), + ], + body, + ) + .into_response() + }), + )) + .await; + let monitor = MonitorHandle::default(); + let response = proxy(&upstream, &monitor) + .oneshot(request( + "/v1/messages", + r#"{"model":"claude-opus-5","messages":[]}"#, + )) + .await + .unwrap(); + assert_eq!(response.status(), status); + assert_eq!(response.headers()["retry-after"], "13"); + assert_eq!(response.headers()["request-id"], "real-id"); + assert_eq!( + response.into_body().collect().await.unwrap().to_bytes(), + body.as_bytes() + ); + } +} + +#[tokio::test] +async fn streamed_errors_remain_verbatim_and_fail_the_dashboard_request() { + let raw = "event: error\ndata: {\"type\":\"error\",\"error\":{\"message\":\"overloaded\"}}\n\n"; + let upstream = upstream(Router::new().route( + "/v1/messages", + post(move || async move { ([("content-type", "text/event-stream")], raw) }), + )) + .await; + let monitor = MonitorHandle::default(); + let response = proxy(&upstream, &monitor) + .oneshot(request( + "/v1/messages", + r#"{"model":"claude-opus-5","messages":[],"stream":true}"#, + )) + .await + .unwrap(); + assert_eq!( + response.into_body().collect().await.unwrap().to_bytes(), + raw.as_bytes() + ); + let snapshot = monitor.snapshot(); + assert_eq!(snapshot.recent[0].status, RequestStatus::Failed); + assert_eq!(snapshot.recent[0].error.as_deref(), Some("overloaded")); +} diff --git a/tests/foundation.rs b/tests/foundation.rs index f4bded2d..b25d15a3 100644 --- a/tests/foundation.rs +++ b/tests/foundation.rs @@ -257,6 +257,34 @@ fn alias_provider_has_expected_default() { )); } +#[test] +fn anthropic_alias_provider_can_be_enabled_in_file_or_environment() { + let config = TempDir::new().unwrap(); + std::fs::write( + config.path().join("config.json"), + r#"{"aliasProvider":"anthropic"}"#, + ) + .unwrap(); + let mut env = HashMap::from([( + "CCP_CONFIG_DIR".to_string(), + config.path().to_string_lossy().into_owned(), + )]); + assert_eq!( + load_config_for_env(&env).alias_provider, + AliasProvider::Anthropic + ); + env.insert("CCP_ALIAS_PROVIDER".to_string(), "codex".to_string()); + assert_eq!( + load_config_for_env(&env).alias_provider, + AliasProvider::Codex + ); + env.insert("CCP_ALIAS_PROVIDER".to_string(), "anthropic".to_string()); + assert_eq!( + load_config_for_env(&env).alias_provider, + AliasProvider::Anthropic + ); +} + #[test] fn logger_factory_builds() { let logger = create_logger("server");