diff --git a/README.md b/README.md index c73cc0d..9fa3e5b 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,7 @@ # claude-code-proxy -Claude Code, powered by **OpenAI**, **Kimi**, **Grok**, or **Cursor**. +Claude Code, powered by **OpenAI**, **Kimi**, **Grok**, **OpenCode Go**, or +**Cursor**. Claude Code running through claude-code-proxy @@ -43,8 +44,8 @@ artifacts are published as `claude-code-proxy-windows-amd64.zip` and ### 2. Pick a provider and authenticate -The proxy supports four upstream providers. Pick one and run its login flow; the -proxy will refuse to start traffic until a token is stored. +The proxy supports five upstream providers. OAuth providers use the login +commands below; OpenCode Go uses a static workspace API key. **Codex (ChatGPT Plus/Pro):** @@ -78,6 +79,17 @@ OAuth session and does not use the official Grok CLI credential file. On a headless host, `grok auth device` prints a verification URL and code to enter on any other device, then polls until authorization completes. +**OpenCode Go:** + +```sh +export OPENCODE_API_KEY=sk-... +``` + +Copy the workspace key from the OpenCode console after subscribing to Go. The +proxy does not implement an OpenCode login flow; `CCP_OPENCODE_API_KEY` is an +equivalent, proxy-specific variable, and `opencode.apiKey` can be used in +`config.json` instead. + **Cursor Agent:** ```sh @@ -362,6 +374,29 @@ refreshes them five minutes before expiry, and does not use `~/.grok/auth.json`. | `grok auth status` | Show token expiry and storage path | | `grok auth logout` | Delete proxy-owned credentials | +### OpenCode Go + +Upstream: `https://opencode.ai/zen/go/v1`. OpenCode Go uses one API key but +splits its catalog across two wire protocols. The proxy routes each model to +its verified endpoint automatically: + +- OpenAI-compatible `/chat/completions`: `glm-5.2`, `glm-5.1`, `glm-5`, + `kimi-k2.7-code`, `kimi-k2.6`, `kimi-k2.5`, `deepseek-v4-pro`, + `deepseek-v4-flash`, `mimo-v2.5`, `mimo-v2.5-pro`, and `qwen3.5-plus` +- Anthropic-compatible `/messages`: `minimax-m3`, `minimax-m2.7`, + `minimax-m2.5`, `qwen3.7-max`, `qwen3.7-plus`, `qwen3.6-plus` + +Both bare model ids and the official `opencode-go/` form are +accepted. The existing Kimi Code provider already owns the bare `kimi-k2.6` +alias, so use `opencode-go/kimi-k2.6` to select the Go version explicitly. +`count_tokens` remains local and does not consume Go usage. + +Authentication is API-key only. Set `CCP_OPENCODE_API_KEY` or +`OPENCODE_API_KEY`, or configure `opencode.apiKey`. The proxy sends it as a +Bearer token to `/chat/completions` and as `x-api-key` to the +Anthropic-compatible `/messages` endpoint. There is no CCP browser/device login +for this provider. + ### Cursor Agent Upstream: `https://api2.cursor.sh/agent.v1.AgentService/Run` (Cursor Agent's @@ -428,9 +463,9 @@ sequenceDiagram participant CC as Claude Code participant P as claude-code-proxy participant AUTH as OAuth host / credential store - participant U as Upstream API
(Codex, Kimi, or Cursor) + participant U as Upstream API
(Codex, Kimi, Grok, OpenCode, or Cursor) - Note over P,AUTH: One-time: PKCE / device OAuth
tokens cached locally for reuse + Note over P,AUTH: One-time: OAuth login or API-key setup
credentials cached or supplied locally CC->>P: POST /v1/messages (Anthropic shape, stream: true) @@ -699,6 +734,10 @@ Windows, and at "baseUrl": "https://cli-chat-proxy.grok.com/v1", "clientVersion": "0.2.93" }, + "opencode": { + "apiKey": "sk-...", + "baseUrl": "https://opencode.ai/zen/go/v1" + }, "cursor": { "baseUrl": "https://api2.cursor.sh", "clientVersion": "cli-2026.06.04-5fd875e", @@ -735,6 +774,9 @@ Windows, and at | `CCP_KIMI_USER_AGENT` | `kimi.userAgent` | `KimiCLI/1.37.0` | Override the `User-Agent` header sent to Kimi | | `CCP_GROK_BASE_URL` | `grok.baseUrl` | `https://cli-chat-proxy.grok.com/v1` | Override the Grok Responses API base URL | | `CCP_GROK_CLIENT_VERSION` | `grok.clientVersion` | `0.2.93` | Override the Grok client version header | +| `CCP_OPENCODE_API_KEY` | `opencode.apiKey` | unset | OpenCode Go workspace API key; takes precedence over `OPENCODE_API_KEY` and config | +| `OPENCODE_API_KEY` | `opencode.apiKey` | unset | Standard OpenCode Go API key variable; used when `CCP_OPENCODE_API_KEY` is unset | +| `CCP_OPENCODE_BASE_URL` | `opencode.baseUrl` | `https://opencode.ai/zen/go/v1` | Override the OpenCode Go API base URL | | `CCP_CURSOR_BASE_URL` | `cursor.baseUrl` | `https://api2.cursor.sh` | Override Cursor's API base URL | | `CCP_CURSOR_CLIENT_VERSION` | `cursor.clientVersion` | `cli-2026.06.04-5fd875e` | Override Cursor client version headers | | `CCP_CURSOR_AGENT_BUNDLE` | `cursor.agentBundle` | auto-detected | Path to Cursor Agent's bundled `index.js` used only for protobuf schemas | diff --git a/src/anthropic/sse.rs b/src/anthropic/sse.rs index 6abd619..aef9612 100644 --- a/src/anthropic/sse.rs +++ b/src/anthropic/sse.rs @@ -13,6 +13,94 @@ pub struct SseParseStats { pub event_count: usize, } +pub const MAX_SSE_FRAME_BYTES: usize = 1024 * 1024; + +#[derive(Default)] +pub struct SseDecoder { + frame: Vec, + line_start: usize, + skip_lf: bool, +} + +impl SseDecoder { + pub fn push(&mut self, input: &[u8]) -> anyhow::Result> { + let mut events = Vec::new(); + for &byte in input { + if self.skip_lf { + self.skip_lf = false; + if byte == b'\n' { + continue; + } + } + match byte { + b'\n' => self.end_line(&mut events)?, + b'\r' => { + self.end_line(&mut events)?; + self.skip_lf = true; + } + _ => self.push_byte(byte)?, + } + } + Ok(events) + } + + pub fn finish(&mut self) -> anyhow::Result<()> { + if self.frame.is_empty() { + Ok(()) + } else { + anyhow::bail!("SSE stream ended with an incomplete frame") + } + } + + fn push_byte(&mut self, byte: u8) -> anyhow::Result<()> { + if self.frame.len() >= MAX_SSE_FRAME_BYTES { + anyhow::bail!("SSE frame exceeds the size limit"); + } + self.frame.push(byte); + Ok(()) + } + + fn end_line(&mut self, events: &mut Vec) -> anyhow::Result<()> { + if self.frame.len() == self.line_start { + if !self.frame.is_empty() { + events.push(parse_incremental_frame(&self.frame)?); + } + self.frame.clear(); + self.line_start = 0; + return Ok(()); + } + self.push_byte(b'\n')?; + self.line_start = self.frame.len(); + Ok(()) + } +} + +fn parse_incremental_frame(frame: &[u8]) -> anyhow::Result { + let frame = std::str::from_utf8(frame) + .map_err(|_| anyhow::anyhow!("SSE frame contains invalid UTF-8"))?; + let mut event = None; + let mut data = Vec::new(); + for line in frame.lines() { + if line.starts_with(':') { + continue; + } + let (field, value) = line.split_once(':').unwrap_or((line, "")); + let value = value.strip_prefix(' ').unwrap_or(value); + match field { + "event" => event = Some(value.to_owned()), + "data" => data.push(value), + _ => {} + } + } + if data.is_empty() { + anyhow::bail!("SSE frame lacks data") + } + Ok(SseEvent { + event, + data: data.join("\n"), + }) +} + pub fn parse_sse_events(input: &[u8]) -> Vec { parse_sse_events_with_stats(input).0 } @@ -128,3 +216,48 @@ fn parse_block(block: &str) -> Vec { events } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn incremental_decoder_accepts_chunk_boundaries_and_crlf() { + let input = b": note\r\nevent: update\r\ndata: first\r\ndata: second\r\n\r\n"; + let expected = vec![SseEvent { + event: Some("update".to_string()), + data: "first\nsecond".to_string(), + }]; + for split in 0..=input.len() { + let mut decoder = SseDecoder::default(); + let mut events = decoder.push(&input[..split]).unwrap(); + events.extend(decoder.push(&input[split..]).unwrap()); + decoder.finish().unwrap(); + assert_eq!(events, expected, "split at byte {split}"); + } + } + + #[test] + fn incremental_decoder_rejects_invalid_incomplete_and_oversized_frames() { + assert!(SseDecoder::default().push(b"data: \xff\n\n").is_err()); + + let mut incomplete = SseDecoder::default(); + incomplete.push(b"data: partial").unwrap(); + assert!(incomplete.finish().is_err()); + + let mut oversized = SseDecoder::default(); + let exact = vec![b'x'; MAX_SSE_FRAME_BYTES - b"data: \n".len()]; + let mut frame = b"data: ".to_vec(); + frame.extend(exact); + frame.extend(b"\n\n"); + oversized.push(&frame).unwrap(); + oversized.finish().unwrap(); + + let mut too_large = SseDecoder::default(); + assert!( + too_large + .push(&vec![b'x'; MAX_SSE_FRAME_BYTES + 1]) + .is_err() + ); + } +} diff --git a/src/config.rs b/src/config.rs index df1b028..338b167 100644 --- a/src/config.rs +++ b/src/config.rs @@ -42,6 +42,7 @@ struct FileConfig { pub codex: Option, pub cursor: Option, pub grok: Option, + pub opencode: Option, } #[derive(Deserialize, Clone)] @@ -93,6 +94,14 @@ struct GrokConfig { pub client_version: Option, } +#[derive(Deserialize, Clone)] +struct OpenCodeConfig { + #[serde(rename = "apiKey")] + pub api_key: Option, + #[serde(rename = "baseUrl")] + pub base_url: Option, +} + #[derive(Deserialize)] struct FileLog { pub verbose: Option, @@ -238,6 +247,14 @@ pub fn config_override_summary_lines(cfg: &LoadedConfig) -> Vec { if env.contains_key("CCP_GROK_CLIENT_VERSION") { out.push("grok.clientVersion (env)".to_string()); } + if env.contains_key("CCP_OPENCODE_API_KEY") { + out.push("opencode.apiKey (env)".to_string()); + } else if env.contains_key("OPENCODE_API_KEY") { + out.push("opencode.apiKey (OpenCode env)".to_string()); + } + if env.contains_key("CCP_OPENCODE_BASE_URL") { + out.push("opencode.baseUrl (env)".to_string()); + } if env .get("CCP_CODEX_REASONING_SUMMARY") .is_some_and(|raw| !raw.is_empty()) @@ -262,6 +279,14 @@ pub fn config_override_summary_lines(cfg: &LoadedConfig) -> Vec { out.push(format!("log.stderr: {v}")); } } + if let Some(opencode) = file_cfg.opencode { + if opencode.api_key.is_some_and(|raw| !raw.is_empty()) { + out.push("opencode.apiKey (config)".to_string()); + } + if let Some(url) = opencode.base_url.filter(|raw| !raw.is_empty()) { + out.push(format!("opencode.baseUrl: {url}")); + } + } if let Some(codex) = file_cfg.codex && let Some(summary) = codex.reasoning_summary && !summary.is_empty() @@ -298,6 +323,65 @@ pub fn grok_client_version() -> String { "0.2.93".to_string() } +pub fn opencode_api_key() -> Option { + let env: HashMap<_, _> = std::env::vars().collect(); + if let Some(raw) = env + .get("CCP_OPENCODE_API_KEY") + .filter(|raw| !raw.is_empty()) + { + return Some(raw.clone()); + } + if let Some(raw) = env.get("OPENCODE_API_KEY").filter(|raw| !raw.is_empty()) { + return Some(raw.clone()); + } + read_file_config(&paths::config_dir()) + .and_then(|file| file.opencode) + .and_then(|opencode| opencode.api_key) + .filter(|raw| !raw.is_empty()) +} + +pub fn opencode_api_key_source() -> Option<&'static str> { + let env: HashMap<_, _> = std::env::vars().collect(); + if env + .get("CCP_OPENCODE_API_KEY") + .is_some_and(|raw| !raw.is_empty()) + { + return Some("CCP_OPENCODE_API_KEY"); + } + if env + .get("OPENCODE_API_KEY") + .is_some_and(|raw| !raw.is_empty()) + { + return Some("OPENCODE_API_KEY"); + } + if read_file_config(&paths::config_dir()) + .and_then(|file| file.opencode) + .and_then(|opencode| opencode.api_key) + .is_some_and(|raw| !raw.is_empty()) + { + return Some("config.json"); + } + None +} + +pub fn opencode_base_url() -> String { + let env: HashMap<_, _> = std::env::vars().collect(); + if let Some(raw) = env + .get("CCP_OPENCODE_BASE_URL") + .filter(|raw| !raw.is_empty()) + { + return raw.clone(); + } + if let Some(url) = read_file_config(&paths::config_dir()) + .and_then(|file| file.opencode) + .and_then(|opencode| opencode.base_url) + .filter(|raw| !raw.is_empty()) + { + return url; + } + "https://opencode.ai/zen/go/v1".to_string() +} + pub fn is_verbose() -> bool { log_verbose() } @@ -592,9 +676,39 @@ mod tests { std::env::remove_var("CCP_LOG_VERBOSE"); std::env::remove_var("CCP_LOG_STDERR"); std::env::remove_var("CCP_CODEX_REASONING_SUMMARY"); + std::env::remove_var("CCP_OPENCODE_API_KEY"); + std::env::remove_var("OPENCODE_API_KEY"); + std::env::remove_var("CCP_OPENCODE_BASE_URL"); } } + #[test] + fn opencode_config_reads_file_and_env_precedence() { + let _guard = ENV_LOCK.lock().unwrap(); + clear_env(); + let config = tempfile::TempDir::new().unwrap(); + std::fs::write( + config.path().join("config.json"), + r#"{"opencode":{"apiKey":"file-key","baseUrl":"https://file.example/v1"}}"#, + ) + .unwrap(); + let _config_env = EnvGuard::set("CCP_CONFIG_DIR", config.path()); + + assert_eq!(opencode_api_key().as_deref(), Some("file-key")); + assert_eq!(opencode_api_key_source(), Some("config.json")); + assert_eq!(opencode_base_url(), "https://file.example/v1"); + + let _standard_key = EnvGuard::set("OPENCODE_API_KEY", "standard-key"); + assert_eq!(opencode_api_key().as_deref(), Some("standard-key")); + assert_eq!(opencode_api_key_source(), Some("OPENCODE_API_KEY")); + + let _ccp_key = EnvGuard::set("CCP_OPENCODE_API_KEY", "ccp-key"); + let _base_url = EnvGuard::set("CCP_OPENCODE_BASE_URL", "https://env.example/v1"); + assert_eq!(opencode_api_key().as_deref(), Some("ccp-key")); + assert_eq!(opencode_api_key_source(), Some("CCP_OPENCODE_API_KEY")); + assert_eq!(opencode_base_url(), "https://env.example/v1"); + } + #[test] fn bind_address_defaults_to_loopback() { let _guard = ENV_LOCK.lock().unwrap(); diff --git a/src/main.rs b/src/main.rs index 6ce6794..097cd3f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -216,7 +216,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", "cursor"] { + for provider in ["codex", "kimi", "grok", "opencode", "cursor"] { let Some(models) = grouped.get(provider) else { continue; }; diff --git a/src/providers/grok/translate/stream.rs b/src/providers/grok/translate/stream.rs index 62d0034..3e5aeed 100644 --- a/src/providers/grok/translate/stream.rs +++ b/src/providers/grok/translate/stream.rs @@ -1,93 +1,6 @@ use super::reducer::{Reducer, ReducerEvent}; -use crate::anthropic::sse::{SseEvent, encode_sse_event}; - -pub const MAX_SSE_FRAME_BYTES: usize = 1024 * 1024; - -#[derive(Default)] -pub struct SseDecoder { - frame: Vec, - line_start: usize, - skip_lf: bool, -} - -impl SseDecoder { - pub fn push(&mut self, input: &[u8]) -> anyhow::Result> { - let mut events = Vec::new(); - for &byte in input { - if self.skip_lf { - self.skip_lf = false; - if byte == b'\n' { - continue; - } - } - match byte { - b'\n' => self.end_line(&mut events)?, - b'\r' => { - self.end_line(&mut events)?; - self.skip_lf = true; - } - _ => self.push_byte(byte)?, - } - } - Ok(events) - } - - pub fn finish(&mut self) -> anyhow::Result<()> { - if self.frame.is_empty() { - Ok(()) - } else { - anyhow::bail!("Grok SSE stream ended with an incomplete frame") - } - } - - fn push_byte(&mut self, byte: u8) -> anyhow::Result<()> { - if self.frame.len() >= MAX_SSE_FRAME_BYTES { - anyhow::bail!("Grok SSE frame exceeds the size limit"); - } - self.frame.push(byte); - Ok(()) - } - - fn end_line(&mut self, events: &mut Vec) -> anyhow::Result<()> { - if self.frame.len() == self.line_start { - if !self.frame.is_empty() { - events.push(parse_frame(&self.frame)?); - } - self.frame.clear(); - self.line_start = 0; - return Ok(()); - } - self.push_byte(b'\n')?; - self.line_start = self.frame.len(); - Ok(()) - } -} - -fn parse_frame(frame: &[u8]) -> anyhow::Result { - let frame = std::str::from_utf8(frame) - .map_err(|_| anyhow::anyhow!("Grok SSE frame contains invalid UTF-8"))?; - let mut event = None; - let mut data = Vec::new(); - for line in frame.lines() { - if line.starts_with(':') { - continue; - } - let (field, value) = line.split_once(':').unwrap_or((line, "")); - let value = value.strip_prefix(' ').unwrap_or(value); - match field { - "event" => event = Some(value.to_owned()), - "data" => data.push(value), - _ => {} - } - } - if data.is_empty() { - anyhow::bail!("Grok SSE frame lacks data") - } - Ok(SseEvent { - event, - data: data.join("\n"), - }) -} +use crate::anthropic::sse::encode_sse_event; +pub use crate::anthropic::sse::{MAX_SSE_FRAME_BYTES, SseDecoder, SseEvent}; pub struct StreamTranslator { message_id: String, diff --git a/src/providers/kimi/client.rs b/src/providers/kimi/client.rs index 011a8ac..485569c 100644 --- a/src/providers/kimi/client.rs +++ b/src/providers/kimi/client.rs @@ -4,7 +4,7 @@ use crate::providers::kimi::auth::constants::api_base_url; use crate::providers::kimi::auth::headers::common_headers; use crate::providers::kimi::auth::manager::KimiAuthManager; use crate::providers::kimi::auth::token_store::{StoredAuth, file_store}; -use crate::providers::kimi::translate::request::KimiChatRequest; +use crate::providers::kimi::translate::request::OpenAiChatRequest; use crate::retry::{MAX_RATE_LIMIT_RETRIES, compute_backoff_delay}; #[derive(Debug)] @@ -47,7 +47,7 @@ impl KimiHttpClient { &self.auth_manager } - pub fn post_kimi(&self, body: &KimiChatRequest) -> Result { + pub fn post_kimi(&self, body: &OpenAiChatRequest) -> Result { let mut auth = self.auth_manager.get_auth().map_err(|e| KimiError { status: 401, message: "Auth error".to_string(), @@ -96,7 +96,7 @@ impl KimiHttpClient { fn attempt_post( &self, access_token: &str, - body: &KimiChatRequest, + body: &OpenAiChatRequest, ) -> Result { let headers = common_headers().map_err(|e| KimiError { status: 500, diff --git a/src/providers/kimi/translate/accumulate.rs b/src/providers/kimi/translate/accumulate.rs index db246b4..5b4b45a 100644 --- a/src/providers/kimi/translate/accumulate.rs +++ b/src/providers/kimi/translate/accumulate.rs @@ -202,6 +202,27 @@ mod tests { assert_eq!(content[2]["type"], "tool_use"); } + #[test] + fn accumulate_fragmented_tool_arguments_after_thinking() { + let upstream = concat!( + "data: {\"choices\":[{\"delta\":{\"reasoning_content\":\"thinking\"}}]}\n\n", + "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"function\":{\"name\":\"Read\",\"arguments\":\"\"}}]}}]}\n\n", + "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"{\\\"file_path\\\":\"}}]}}]}\n\n", + "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"\\\"input.txt\\\"}\"}}]}}]}\n\n", + "data: {\"choices\":[{\"finish_reason\":\"tool_calls\"}]}\n\n", + "data: [DONE]\n\n" + ); + let response = accumulate_response(upstream.as_bytes(), "msg_1", "model").unwrap(); + let tool = response["content"] + .as_array() + .unwrap() + .iter() + .find(|block| block["type"] == "tool_use") + .unwrap(); + assert_eq!(tool["name"], "Read"); + assert_eq!(tool["input"]["file_path"], "input.txt"); + } + #[test] fn accumulate_handles_upstream_error() { let upstream = "data: {\"error\":{\"message\":\"upstream failure\"}}\n\n"; diff --git a/src/providers/kimi/translate/reducer.rs b/src/providers/kimi/translate/reducer.rs index 89fc449..0368d17 100644 --- a/src/providers/kimi/translate/reducer.rs +++ b/src/providers/kimi/translate/reducer.rs @@ -157,33 +157,39 @@ struct StreamError { r#type: Option, } -#[allow(dead_code)] struct ToolSlot { + upstream_index: usize, block_index: usize, - id: String, - name: String, } -pub fn reduce_upstream_bytes(input: &[u8]) -> Result, UpstreamStreamError> { - let sse_events = parse_sse_events(input); - let mut out = Vec::new(); - let mut next_block_index = 0usize; - let mut thinking_index: Option = None; - let mut text_index: Option = None; - let mut tool_slots: Vec = Vec::new(); - let mut saw_tool_calls = false; - let mut finish_reason: Option = None; - let mut final_usage: Option = None; - - for evt in &sse_events { - let data = evt.data.trim(); - if data.is_empty() || data == "[DONE]" { - continue; +#[derive(Default)] +pub struct Reducer { + next_block_index: usize, + thinking_index: Option, + text_index: Option, + tool_slots: Vec, + saw_tool_calls: bool, + finish_reason: Option, + final_usage: Option, + finished: bool, +} + +impl Reducer { + pub fn push_data(&mut self, data: &str) -> Result, UpstreamStreamError> { + if self.finished { + return Ok(Vec::new()); + } + let data = data.trim(); + if data.is_empty() { + return Ok(Vec::new()); + } + if data == "[DONE]" { + return Ok(self.finish()); } let chunk: StreamChunk = match serde_json::from_str(data) { Ok(c) => c, - Err(_) => continue, + Err(_) => return Ok(Vec::new()), }; if let Some(ref err) = chunk.error { @@ -198,24 +204,25 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream } if chunk.usage.is_some() && chunk.choices.as_ref().map(|c| c.is_empty()).unwrap_or(true) { - final_usage = chunk.usage.map(|u| kimi_usage_from_stream(&u)); - continue; + self.final_usage = chunk.usage.map(|u| kimi_usage_from_stream(&u)); + return Ok(Vec::new()); } let choice = match chunk.choices.as_ref().and_then(|c| c.first()) { Some(c) => c, - None => continue, + None => return Ok(Vec::new()), }; + let mut out = Vec::new(); let delta = match choice.delta.as_ref() { Some(d) => d, None => { if choice.finish_reason.is_some() { - finish_reason = choice.finish_reason.clone(); + self.finish_reason = choice.finish_reason.clone(); if chunk.usage.is_some() { - final_usage = chunk.usage.map(|u| kimi_usage_from_stream(&u)); + self.final_usage = chunk.usage.map(|u| kimi_usage_from_stream(&u)); } } - continue; + return Ok(out); } }; @@ -223,15 +230,15 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream if let Some(ref reasoning) = delta.reasoning_content && !reasoning.is_empty() { - if thinking_index.is_none() { - thinking_index = Some(next_block_index); - next_block_index += 1; + if self.thinking_index.is_none() { + self.thinking_index = Some(self.next_block_index); + self.next_block_index += 1; out.push(ReducerEvent::ThinkingStart { - index: thinking_index.unwrap(), + index: self.thinking_index.unwrap(), }); } out.push(ReducerEvent::ThinkingDelta { - index: thinking_index.unwrap(), + index: self.thinking_index.unwrap(), text: reasoning.clone(), }); } @@ -241,18 +248,18 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream && !content.is_empty() { // Close thinking before text - if let Some(ti) = thinking_index.take() { + if let Some(ti) = self.thinking_index.take() { out.push(ReducerEvent::ThinkingStop { index: ti }); } - if text_index.is_none() { - text_index = Some(next_block_index); - next_block_index += 1; + if self.text_index.is_none() { + self.text_index = Some(self.next_block_index); + self.next_block_index += 1; out.push(ReducerEvent::TextStart { - index: text_index.unwrap(), + index: self.text_index.unwrap(), }); } out.push(ReducerEvent::TextDelta { - index: text_index.unwrap(), + index: self.text_index.unwrap(), text: content.clone(), }); } @@ -262,17 +269,20 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream && !tool_calls.is_empty() { // Close thinking and text before tools - if let Some(ti) = thinking_index.take() { + if let Some(ti) = self.thinking_index.take() { out.push(ReducerEvent::ThinkingStop { index: ti }); } - if let Some(ti) = text_index.take() { + if let Some(ti) = self.text_index.take() { out.push(ReducerEvent::TextStop { index: ti }); } for tc in tool_calls { - let existing_pos = tool_slots.iter().position(|s| s.block_index == tc.index); + let existing_pos = self + .tool_slots + .iter() + .position(|slot| slot.upstream_index == tc.index); let block_index = if let Some(pos) = existing_pos { - tool_slots[pos].block_index + self.tool_slots[pos].block_index } else { let id = tc.id.clone().unwrap_or_default(); let name = tc @@ -284,13 +294,12 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream // Defensive: skip out-of-order fragments continue; } - saw_tool_calls = true; - let bi = next_block_index; - next_block_index += 1; - tool_slots.push(ToolSlot { + self.saw_tool_calls = true; + let bi = self.next_block_index; + self.next_block_index += 1; + self.tool_slots.push(ToolSlot { + upstream_index: tc.index, block_index: bi, - id: id.clone(), - name: name.clone(), }); out.push(ReducerEvent::ToolStart { index: bi, @@ -313,38 +322,58 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream // Finish reason on choice level if let Some(ref reason) = choice.finish_reason { - finish_reason = Some(reason.clone()); + self.finish_reason = Some(reason.clone()); if chunk.usage.is_some() { - final_usage = chunk.usage.map(|u| kimi_usage_from_stream(&u)); + self.final_usage = chunk.usage.map(|u| kimi_usage_from_stream(&u)); } } - } - // Close any open blocks - if let Some(ti) = thinking_index.take() { - out.push(ReducerEvent::ThinkingStop { index: ti }); - } - if let Some(ti) = text_index.take() { - out.push(ReducerEvent::TextStop { index: ti }); + Ok(out) } - for slot in tool_slots.iter() { - out.push(ReducerEvent::ToolStop { - index: slot.block_index, + + pub fn finish(&mut self) -> Vec { + if self.finished { + return Vec::new(); + } + self.finished = true; + let mut out = Vec::new(); + if let Some(index) = self.thinking_index.take() { + out.push(ReducerEvent::ThinkingStop { index }); + } + if let Some(index) = self.text_index.take() { + out.push(ReducerEvent::TextStop { index }); + } + for slot in &self.tool_slots { + out.push(ReducerEvent::ToolStop { + index: slot.block_index, + }); + } + + let stop_reason = match self.finish_reason.as_deref() { + Some("length") => StopReason::MaxTokens, + Some("tool_calls") => StopReason::ToolUse, + _ if self.saw_tool_calls => StopReason::ToolUse, + _ => StopReason::EndTurn, + }; + out.push(ReducerEvent::Finish { + stop_reason, + usage: self.final_usage.take(), }); + out } - let stop_reason = match finish_reason.as_deref() { - Some("length") => StopReason::MaxTokens, - Some("tool_calls") => StopReason::ToolUse, - _ if saw_tool_calls => StopReason::ToolUse, - _ => StopReason::EndTurn, - }; - - out.push(ReducerEvent::Finish { - stop_reason, - usage: final_usage, - }); + pub fn finished(&self) -> bool { + self.finished + } +} +pub fn reduce_upstream_bytes(input: &[u8]) -> Result, UpstreamStreamError> { + let mut reducer = Reducer::default(); + let mut out = Vec::new(); + for event in parse_sse_events(input) { + out.extend(reducer.push_data(&event.data)?); + } + out.extend(reducer.finish()); Ok(out) } @@ -400,7 +429,7 @@ mod tests { let upstream = concat!( "data: {\"choices\":[{\"delta\":{\"reasoning_content\":\"think\"}}]}\n\n", "data: {\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\n", - "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"function\":{\"name\":\"search\",\"arguments\":\"{\\\"q\\\"\"}}]}}}]}\n\n", + "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"function\":{\"name\":\"search\",\"arguments\":\"{\\\"q\\\"\"}}]}}]}\n\n", "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\":\\\"rust\\\"}\"}}]}}]}\n\n", "data: {\"choices\":[{\"finish_reason\":\"tool_calls\"}],\"usage\":{\"prompt_tokens\":10,\"completion_tokens\":3,\"prompt_tokens_details\":{\"cached_tokens\":4}}}\n\n", "data: [DONE]\n\n" @@ -418,6 +447,14 @@ mod tests { .. } ))); + let arguments = events + .iter() + .filter_map(|event| match event { + ReducerEvent::ToolDelta { partial_json, .. } => Some(partial_json.as_str()), + _ => None, + }) + .collect::(); + assert_eq!(arguments, "{\"q\":\"rust\"}"); } #[test] diff --git a/src/providers/kimi/translate/request.rs b/src/providers/kimi/translate/request.rs index 7509b03..799e7b9 100644 --- a/src/providers/kimi/translate/request.rs +++ b/src/providers/kimi/translate/request.rs @@ -9,11 +9,11 @@ use crate::providers::translate_shared::{ }; // --------------------------------------------------------------------------- -// Kimi OpenAI-compatible chat-completions types +// Shared OpenAI-compatible chat-completions types // --------------------------------------------------------------------------- #[derive(Debug, Clone, Serialize, Deserialize)] -pub struct KimiChatRequest { +pub struct OpenAiChatRequest { pub model: String, pub messages: Vec, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -129,7 +129,7 @@ const DEFAULT_MAX_TOKENS: u32 = 32000; pub fn translate_request( req: &MessagesRequest, opts: TranslateOptions, -) -> Result { +) -> Result { let model = req.model.as_deref().unwrap_or(KIMI_DEFAULT_MODEL); let resolved = resolve_model(model); assert_allowed_model(&resolved).map_err(|e| anyhow::anyhow!("{e}"))?; @@ -138,7 +138,7 @@ pub fn translate_request( let tools = read_tools(req)?; let tool_choice = read_tool_choice(req)?; - let mut out = KimiChatRequest { + let mut out = OpenAiChatRequest { model: resolved, messages, stream: true, @@ -163,6 +163,36 @@ pub fn translate_request( Ok(out) } +pub fn translate_openai_compatible_request( + req: &MessagesRequest, + model: String, +) -> Result { + let messages = build_messages(req)?; + let tools = read_tools(req)?; + let mut tool_choice = read_tool_choice(req)?; + if matches!(tool_choice, Some(KimiToolChoice::Auto)) { + tool_choice = None; + } + + Ok(OpenAiChatRequest { + model, + messages, + tools: if tools.is_empty() { None } else { Some(tools) }, + tool_choice, + stream: true, + stream_options: KimiStreamOptions { + include_usage: true, + }, + max_tokens: req + .max_tokens + .filter(|value| *value > 0) + .unwrap_or(DEFAULT_MAX_TOKENS), + reasoning_effort: None, + thinking: None, + prompt_cache_key: None, + }) +} + fn clamp_max_tokens(requested: Option) -> u32 { match requested { Some(v) if v > 0 => v.min(DEFAULT_MAX_TOKENS), @@ -257,27 +287,50 @@ fn read_tools(req: &MessagesRequest) -> Result, anyhow::Error> { fn build_messages(req: &MessagesRequest) -> Result, anyhow::Error> { let mut out: Vec = Vec::new(); + let mut system_parts: Vec = Vec::new(); - // System message if let Some(system) = flatten_system_text(req.extra.get("system")) { - out.push(KimiMessage::System { - role: "system".to_string(), - content: system, - }); + system_parts.push(system); } - // Convert each message for msg in &req.messages { let blocks = normalize_content(&msg.content, serde_json::json!({})); match msg.role.as_str() { "user" => push_user_messages(&mut out, &blocks), "assistant" => push_assistant_message(&mut out, &blocks), + // Recent Claude Code versions can append dynamic system-prompt + // sections as a message instead of placing all of them in the + // top-level `system` field. OpenAI-compatible upstreams expect + // those sections to remain system instructions. + "system" => { + let text = blocks + .iter() + .filter_map(|block| match block { + ContentBlock::Text { text } => Some(text.as_str()), + _ => None, + }) + .collect::>() + .join("\n\n"); + if !text.is_empty() { + system_parts.push(text); + } + } other => { anyhow::bail!("unexpected message role: {other}"); } } } + if !system_parts.is_empty() { + out.insert( + 0, + KimiMessage::System { + role: "system".to_string(), + content: system_parts.join("\n\n"), + }, + ); + } + Ok(out) } @@ -518,6 +571,49 @@ mod tests { assert_eq!(translated.max_tokens, 10); } + #[test] + fn generic_openai_translation_uses_exact_model_without_kimi_extensions() { + let req: MessagesRequest = serde_json::from_value(json!({ + "model": "opencode-go/glm-5.2", + "max_tokens": 64000, + "messages": [{"role": "user", "content": "hello"}], + "output_config": {"effort":"xhigh"} + })) + .unwrap(); + + let translated = translate_openai_compatible_request(&req, "glm-5.2".to_string()).unwrap(); + assert_eq!(translated.model, "glm-5.2"); + assert_eq!(translated.max_tokens, 64000); + assert!(translated.stream); + assert!(translated.reasoning_effort.is_none()); + assert!(translated.thinking.is_none()); + assert!(translated.prompt_cache_key.is_none()); + } + + #[test] + fn generic_openai_translation_preserves_message_role_system_sections() { + let req: MessagesRequest = serde_json::from_value(json!({ + "model": "opencode-go/glm-5.2", + "system": [{"type": "text", "text": "base instructions"}], + "messages": [ + {"role": "user", "content": "hello"}, + {"role": "system", "content": "dynamic instructions"} + ] + })) + .unwrap(); + + let translated = translate_openai_compatible_request(&req, "glm-5.2".to_string()).unwrap(); + assert_eq!(translated.messages.len(), 2); + match &translated.messages[0] { + KimiMessage::System { role, content } => { + assert_eq!(role, "system"); + assert_eq!(content, "base instructions\n\ndynamic instructions"); + } + _ => panic!("expected merged system message first"), + } + assert!(matches!(translated.messages[1], KimiMessage::User { .. })); + } + #[test] fn translate_tool_result_with_unsupported_blocks() { let req: MessagesRequest = serde_json::from_value(json!({ diff --git a/src/providers/kimi/translate/stream.rs b/src/providers/kimi/translate/stream.rs index e055976..b205a0e 100644 --- a/src/providers/kimi/translate/stream.rs +++ b/src/providers/kimi/translate/stream.rs @@ -1,296 +1,238 @@ -use super::reducer::{map_usage_to_anthropic, reduce_upstream_bytes}; +use super::reducer::{Reducer, ReducerEvent, StopReason, map_usage_to_anthropic}; use super::signature::make_thinking_signature; -use crate::anthropic::sse::encode_sse_event; +use crate::anthropic::sse::{SseDecoder, encode_sse_event}; -pub fn translate_stream_bytes( - input: &[u8], - message_id: &str, - model: &str, -) -> Result, anyhow::Error> { - let events = reduce_upstream_bytes(input) - .map_err(|e| anyhow::anyhow!("upstream stream error: {} ({:?})", e.message, e.kind))?; - - let mut out = Vec::new(); - let mut message_started = false; - let mut active_tools: Vec<(usize, String, String)> = Vec::new(); +pub struct StreamTranslator { + message_id: String, + model: String, + message_started: bool, + finished: bool, +} - let mut emit = |event: Option<&str>, data: &str| { - out.extend_from_slice(&encode_sse_event(event, data)); - }; +impl StreamTranslator { + pub fn new(message_id: String, model: String) -> Self { + Self { + message_id, + model, + message_started: false, + finished: false, + } + } - for event in &events { - match event { - super::reducer::ReducerEvent::ThinkingStart { index } => { - if !message_started { - let data = serde_json::json!({ - "type": "message_start", - "message": { - "id": message_id, - "type": "message", - "role": "assistant", - "model": model, - "content": [], - "stop_reason": null, - "stop_sequence": null, - "usage": { - "input_tokens": 0, - "output_tokens": 0, - } - } - }); - emit(Some("message_start"), &data.to_string()); - message_started = true; - } - let data = serde_json::json!({ - "type": "content_block_start", - "index": index, - "content_block": { - "type": "thinking", - "thinking": "" - } - }); - emit(Some("content_block_start"), &data.to_string()); - } - super::reducer::ReducerEvent::ThinkingDelta { index, text } => { - // Ensure message start - if !message_started { - let data = serde_json::json!({ - "type": "message_start", - "message": { - "id": message_id, - "type": "message", - "role": "assistant", - "model": model, - "content": [], - "stop_reason": null, - "stop_sequence": null, - "usage": { - "input_tokens": 0, - "output_tokens": 0, - } - } - }); - emit(Some("message_start"), &data.to_string()); - message_started = true; - } - let data = serde_json::json!({ - "type": "content_block_delta", - "index": index, - "delta": { - "type": "thinking_delta", - "thinking": text - } - }); - emit(Some("content_block_delta"), &data.to_string()); - } - super::reducer::ReducerEvent::ThinkingStop { index } => { - let data = serde_json::json!({ - "type": "content_block_delta", - "index": index, - "delta": { - "type": "signature_delta", - "signature": make_thinking_signature(message_id, *index) - } - }); - emit(Some("content_block_delta"), &data.to_string()); - let data = serde_json::json!({ - "type": "content_block_stop", - "index": index, - }); - emit(Some("content_block_stop"), &data.to_string()); - } - super::reducer::ReducerEvent::TextStart { index } => { - if !message_started { - let data = serde_json::json!({ - "type": "message_start", - "message": { - "id": message_id, - "type": "message", - "role": "assistant", - "model": model, - "content": [], - "stop_reason": null, - "stop_sequence": null, - "usage": { - "input_tokens": 0, - "output_tokens": 0, - } - } - }); - emit(Some("message_start"), &data.to_string()); - message_started = true; - } - let data = serde_json::json!({ - "type": "content_block_start", - "index": index, - "content_block": { - "type": "text", - "text": "" - } - }); - emit(Some("content_block_start"), &data.to_string()); - } - super::reducer::ReducerEvent::TextDelta { index, text } => { - if !message_started { - let data = serde_json::json!({ - "type": "message_start", - "message": { - "id": message_id, - "type": "message", - "role": "assistant", - "model": model, - "content": [], - "stop_reason": null, - "stop_sequence": null, - "usage": { - "input_tokens": 0, - "output_tokens": 0, + pub fn render(&mut self, events: Vec) -> anyhow::Result> { + if self.finished && !events.is_empty() { + anyhow::bail!("event after terminal completion"); + } + let mut out = Vec::new(); + for event in events { + self.ensure_message_started(&mut out); + match event { + ReducerEvent::ThinkingStart { index } => emit( + &mut out, + "content_block_start", + serde_json::json!({ + "type": "content_block_start", + "index": index, + "content_block": {"type": "thinking", "thinking": ""} + }), + ), + ReducerEvent::ThinkingDelta { index, text } => emit( + &mut out, + "content_block_delta", + serde_json::json!({ + "type": "content_block_delta", + "index": index, + "delta": {"type": "thinking_delta", "thinking": text} + }), + ), + ReducerEvent::ThinkingStop { index } => { + emit( + &mut out, + "content_block_delta", + serde_json::json!({ + "type": "content_block_delta", + "index": index, + "delta": { + "type": "signature_delta", + "signature": make_thinking_signature(&self.message_id, index) } - } - }); - emit(Some("message_start"), &data.to_string()); - message_started = true; + }), + ); + emit_block_stop(&mut out, index); } - let data = serde_json::json!({ - "type": "content_block_delta", - "index": index, - "delta": { - "type": "text_delta", - "text": text - } - }); - emit(Some("content_block_delta"), &data.to_string()); - } - super::reducer::ReducerEvent::TextStop { index } => { - let data = serde_json::json!({ - "type": "content_block_stop", - "index": index, - }); - emit(Some("content_block_stop"), &data.to_string()); - } - super::reducer::ReducerEvent::ToolStart { index, id, name } => { - if !message_started { - let data = serde_json::json!({ - "type": "message_start", - "message": { - "id": message_id, - "type": "message", - "role": "assistant", - "model": model, - "content": [], - "stop_reason": null, - "stop_sequence": null, - "usage": { - "input_tokens": 0, - "output_tokens": 0, - } - } - }); - emit(Some("message_start"), &data.to_string()); - message_started = true; + ReducerEvent::TextStart { index } => emit( + &mut out, + "content_block_start", + serde_json::json!({ + "type": "content_block_start", + "index": index, + "content_block": {"type": "text", "text": ""} + }), + ), + ReducerEvent::TextDelta { index, text } => emit( + &mut out, + "content_block_delta", + serde_json::json!({ + "type": "content_block_delta", + "index": index, + "delta": {"type": "text_delta", "text": text} + }), + ), + ReducerEvent::TextStop { index } | ReducerEvent::ToolStop { index } => { + emit_block_stop(&mut out, index); } - active_tools.push((*index, id.clone(), name.clone())); - let data = serde_json::json!({ - "type": "content_block_start", - "index": index, - "content_block": { - "type": "tool_use", - "id": id, - "name": name, - "input": {} - } - }); - emit(Some("content_block_start"), &data.to_string()); - } - super::reducer::ReducerEvent::ToolDelta { - index, - partial_json, - } => { - if !message_started { - let data = serde_json::json!({ - "type": "message_start", - "message": { - "id": message_id, - "type": "message", - "role": "assistant", - "model": model, - "content": [], - "stop_reason": null, - "stop_sequence": null, - "usage": { - "input_tokens": 0, - "output_tokens": 0, - } + ReducerEvent::ToolStart { index, id, name } => emit( + &mut out, + "content_block_start", + serde_json::json!({ + "type": "content_block_start", + "index": index, + "content_block": { + "type": "tool_use", + "id": id, + "name": name, + "input": {} } - }); - emit(Some("message_start"), &data.to_string()); - message_started = true; + }), + ), + ReducerEvent::ToolDelta { + index, + partial_json, + } => emit( + &mut out, + "content_block_delta", + serde_json::json!({ + "type": "content_block_delta", + "index": index, + "delta": {"type": "input_json_delta", "partial_json": partial_json} + }), + ), + ReducerEvent::Finish { stop_reason, usage } => { + let stop_reason = match stop_reason { + StopReason::EndTurn => "end_turn", + StopReason::ToolUse => "tool_use", + StopReason::MaxTokens => "max_tokens", + }; + emit( + &mut out, + "message_delta", + serde_json::json!({ + "type": "message_delta", + "delta": {"stop_reason": stop_reason, "stop_sequence": null}, + "usage": map_usage_to_anthropic(&usage) + }), + ); + emit( + &mut out, + "message_stop", + serde_json::json!({"type": "message_stop"}), + ); + self.finished = true; } - let data = serde_json::json!({ - "type": "content_block_delta", - "index": index, - "delta": { - "type": "input_json_delta", - "partial_json": partial_json - } - }); - emit(Some("content_block_delta"), &data.to_string()); - } - super::reducer::ReducerEvent::ToolStop { index } => { - active_tools.retain(|(i, _, _)| *i != *index); - let data = serde_json::json!({ - "type": "content_block_stop", - "index": index, - }); - emit(Some("content_block_stop"), &data.to_string()); } - super::reducer::ReducerEvent::Finish { stop_reason, usage } => { - if !message_started { - let data = serde_json::json!({ - "type": "message_start", - "message": { - "id": message_id, - "type": "message", - "role": "assistant", - "model": model, - "content": [], - "stop_reason": null, - "stop_sequence": null, - "usage": { - "input_tokens": 0, - "output_tokens": 0, - } - } - }); - emit(Some("message_start"), &data.to_string()); - message_started = true; + } + Ok(out) + } + + fn ensure_message_started(&mut self, out: &mut Vec) { + if self.message_started { + return; + } + self.message_started = true; + emit( + out, + "message_start", + serde_json::json!({ + "type": "message_start", + "message": { + "id": self.message_id, + "type": "message", + "role": "assistant", + "model": self.model, + "content": [], + "stop_reason": null, + "stop_sequence": null, + "usage": {"input_tokens": 0, "output_tokens": 0} } - let sr = match stop_reason { - super::reducer::StopReason::EndTurn => "end_turn", - super::reducer::StopReason::ToolUse => "tool_use", - super::reducer::StopReason::MaxTokens => "max_tokens", - }; - let mapped = map_usage_to_anthropic(usage); - let data = serde_json::json!({ - "type": "message_delta", - "delta": { - "stop_reason": sr, - "stop_sequence": null - }, - "usage": mapped - }); - emit(Some("message_delta"), &data.to_string()); - let data = serde_json::json!({ - "type": "message_stop" - }); - emit(Some("message_stop"), &data.to_string()); - } + }), + ); + } +} + +pub struct LiveStreamTranslator { + decoder: SseDecoder, + reducer: Reducer, + renderer: StreamTranslator, +} + +impl LiveStreamTranslator { + pub fn new(message_id: String, model: String) -> Self { + Self { + decoder: SseDecoder::default(), + reducer: Reducer::default(), + renderer: StreamTranslator::new(message_id, model), + } + } + + pub fn push(&mut self, chunk: &[u8]) -> anyhow::Result> { + let mut out = Vec::new(); + for event in self.decoder.push(chunk)? { + let reduced = self.reducer.push_data(&event.data).map_err(|error| { + anyhow::anyhow!( + "upstream stream error: {} ({:?})", + error.message, + error.kind + ) + })?; + out.extend(self.renderer.render(reduced)?); } + Ok(out) + } + + pub fn finish(&mut self) -> anyhow::Result> { + self.decoder.finish()?; + self.renderer.render(self.reducer.finish()) + } + + pub fn is_finished(&self) -> bool { + self.reducer.finished() } +} +pub fn translate_stream_bytes( + input: &[u8], + message_id: &str, + model: &str, +) -> anyhow::Result> { + let mut translator = LiveStreamTranslator::new(message_id.to_string(), model.to_string()); + let mut out = translator.push(input)?; + out.extend(translator.finish()?); Ok(out) } +pub fn stream_error(message: &str) -> Vec { + encode_sse_event( + Some("error"), + &serde_json::json!({ + "type": "error", + "error": {"type": "api_error", "message": message} + }) + .to_string(), + ) +} + +fn emit(out: &mut Vec, event: &str, data: serde_json::Value) { + out.extend(encode_sse_event(Some(event), &data.to_string())); +} + +fn emit_block_stop(out: &mut Vec, index: usize) { + emit( + out, + "content_block_stop", + serde_json::json!({"type": "content_block_stop", "index": index}), + ); +} + #[cfg(test)] mod tests { use super::*; @@ -306,21 +248,13 @@ mod tests { translate_stream_bytes(upstream.as_bytes(), "msg_1", "kimi-for-coding").unwrap(); let output = String::from_utf8_lossy(&result); - // Should contain message_start, content_block_start, content_block_delta, - // content_block_stop, message_delta, message_stop - assert!(output.contains("message_start"), "missing message_start"); - assert!( - output.contains("content_block_start"), - "missing content_block_start" - ); - assert!(output.contains("text_delta"), "missing text_delta"); - assert!( - output.contains("content_block_stop"), - "missing content_block_stop" - ); - assert!(output.contains("message_delta"), "missing message_delta"); - assert!(output.contains("message_stop"), "missing message_stop"); - assert!(output.contains("end_turn"), "missing end_turn"); + assert!(output.contains("message_start")); + assert!(output.contains("content_block_start")); + assert!(output.contains("text_delta")); + assert!(output.contains("content_block_stop")); + assert!(output.contains("message_delta")); + assert!(output.contains("message_stop")); + assert!(output.contains("end_turn")); } #[test] @@ -334,12 +268,9 @@ mod tests { let result = translate_stream_bytes(upstream.as_bytes(), "msg_2", "model").unwrap(); let output = String::from_utf8_lossy(&result); - assert!(output.contains("thinking_delta"), "missing thinking_delta"); - assert!( - output.contains("signature_delta"), - "missing signature_delta" - ); - assert!(output.contains("text_delta"), "missing text_delta"); + assert!(output.contains("thinking_delta")); + assert!(output.contains("signature_delta")); + assert!(output.contains("text_delta")); } #[test] @@ -352,15 +283,8 @@ mod tests { let result = translate_stream_bytes(upstream.as_bytes(), "msg_3", "model").unwrap(); let output = String::from_utf8_lossy(&result); - assert!(output.contains("tool_use"), "missing tool_use"); - assert!( - output.contains("input_json_delta"), - "missing input_json_delta" - ); - assert!( - output.contains("tool_use"), - "missing tool_use in stop_reason" - ); + assert!(output.contains("tool_use")); + assert!(output.contains("input_json_delta")); } #[test] @@ -371,13 +295,53 @@ mod tests { ); let result = translate_stream_bytes(upstream.as_bytes(), "msg_4", "model").unwrap(); let output = String::from_utf8_lossy(&result); - - // message_start should come before content_block_start - let msg_start_pos = output.find("message_start").unwrap(); - let cb_start_pos = output.find("content_block_start").unwrap(); assert!( - msg_start_pos < cb_start_pos, - "message_start should be before content_block_start" + output.find("message_start").unwrap() < output.find("content_block_start").unwrap() ); } + + #[test] + fn live_stream_emits_text_before_terminal_frame_arrives() { + let mut translator = LiveStreamTranslator::new("msg_live".into(), "model".into()); + let first = translator + .push(b"data: {\"choices\":[{\"delta\":{\"content\":\"early\"}}]}\n\n") + .unwrap(); + let first = String::from_utf8(first).unwrap(); + assert!(first.contains("text_delta")); + assert!(!first.contains("message_stop")); + + let last = translator + .push( + b"data: {\"choices\":[{\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":2,\"completion_tokens\":1}}\n\ndata: [DONE]\n\n", + ) + .unwrap(); + let last = String::from_utf8(last).unwrap(); + assert!(last.contains("content_block_stop")); + assert!(last.contains("message_stop")); + assert!(translator.is_finished()); + } + + #[test] + fn live_stream_keeps_tool_indices_across_chunks() { + let mut translator = LiveStreamTranslator::new("msg_tools".into(), "model".into()); + let first = translator + .push(b"data: {\"choices\":[{\"delta\":{\"reasoning_content\":\"think\"}}]}\n\ndata: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"function\":{\"name\":\"search\",\"arguments\":\"{\\\"q\\\"\"}}]}}]}\n\n") + .unwrap(); + let second = translator + .push(b"data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\":\\\"rust\\\"}\"}}]}}]}\n\ndata: [DONE]\n\n") + .unwrap(); + let output = String::from_utf8([first, second].concat()).unwrap(); + let partials = crate::anthropic::sse::parse_sse_events(output.as_bytes()) + .into_iter() + .filter_map(|event| serde_json::from_str::(&event.data).ok()) + .filter_map(|value| { + value + .pointer("/delta/partial_json") + .and_then(serde_json::Value::as_str) + .map(str::to_string) + }) + .collect::>(); + assert_eq!(partials, ["{\"q\"", ":\"rust\"}"]); + assert!(output.contains("message_stop")); + } } diff --git a/src/providers/mod.rs b/src/providers/mod.rs index 438af71..8c4f82f 100644 --- a/src/providers/mod.rs +++ b/src/providers/mod.rs @@ -2,4 +2,5 @@ pub mod codex; pub mod cursor; pub mod grok; pub mod kimi; +pub mod opencode; pub mod translate_shared; diff --git a/src/providers/opencode/client.rs b/src/providers/opencode/client.rs new file mode 100644 index 0000000..11ed2fe --- /dev/null +++ b/src/providers/opencode/client.rs @@ -0,0 +1,327 @@ +use std::sync::Arc; +use std::time::Duration; + +use futures_util::StreamExt; +use http::StatusCode; +use serde::Serialize; + +use super::model::EndpointKind; +use crate::traffic::TrafficCapture; + +const MAX_BUFFERED_RESPONSE_BYTES: usize = 8 * 1024 * 1024; + +pub struct OpenCodeClient { + client: Arc, + base_url: reqwest::Url, + api_key: Option, +} + +pub struct OpenCodeResponse { + response: reqwest::Response, +} + +#[derive(Debug)] +pub struct OpenCodeError { + pub status: StatusCode, + pub retry_after: Option, + pub message: String, +} + +impl OpenCodeResponse { + pub fn into_stream( + self, + ) -> impl futures_util::Stream> + Send { + self.response.bytes_stream().map(|chunk| { + chunk.map_err(|_| OpenCodeError { + status: StatusCode::BAD_GATEWAY, + retry_after: None, + message: "OpenCode Go upstream stream failed".to_string(), + }) + }) + } + + pub async fn into_bytes(self) -> Result, OpenCodeError> { + let mut stream = self.into_stream(); + let mut bytes = Vec::new(); + while let Some(chunk) = stream.next().await { + let chunk = chunk?; + if bytes.len().saturating_add(chunk.len()) > MAX_BUFFERED_RESPONSE_BYTES { + return Err(OpenCodeError { + status: StatusCode::BAD_GATEWAY, + retry_after: None, + message: "OpenCode Go upstream response exceeds the size limit".to_string(), + }); + } + bytes.extend_from_slice(&chunk); + } + Ok(bytes) + } +} + +impl OpenCodeClient { + pub fn new(base_url: String, api_key: Option) -> anyhow::Result { + let base_url = reqwest::Url::parse(base_url.trim_end_matches('/'))?; + let client = reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .connect_timeout(Duration::from_secs(10)) + .build()?; + Ok(Self { + client: Arc::new(client), + base_url, + api_key, + }) + } + + pub async fn post( + &self, + endpoint: EndpointKind, + body: &T, + stream: bool, + traffic: Option>, + ) -> Result { + let Some(api_key) = self.api_key.as_deref().filter(|key| !key.is_empty()) else { + return Err(OpenCodeError { + status: StatusCode::UNAUTHORIZED, + retry_after: None, + message: "OpenCode Go API key is not configured; set CCP_OPENCODE_API_KEY, OPENCODE_API_KEY, or opencode.apiKey in config.json".to_string(), + }); + }; + let url = self.endpoint_url(endpoint); + let accept = if stream { + "text/event-stream" + } else { + "application/json" + }; + + if let Some(capture) = traffic.as_ref() { + let value = serde_json::to_value(body).unwrap_or(serde_json::Value::Null); + capture.write_json("020-upstream-request", &value); + let auth_header = match endpoint { + EndpointKind::ChatCompletions => "authorization", + EndpointKind::Messages => "x-api-key", + }; + capture.write_json( + "021-upstream-request-metadata", + &serde_json::json!({ + "method": "POST", + "url": url.as_str(), + "provider": "opencode", + "transport": "http", + "headers": { + "accept": accept, + auth_header: "[redacted]", + "content-type": "application/json" + } + }), + ); + } + + let mut request = self + .client + .post(url) + .header(http::header::ACCEPT, accept) + .header(http::header::CONTENT_TYPE, "application/json") + .json(body); + match endpoint { + EndpointKind::ChatCompletions => { + request = request.header(http::header::AUTHORIZATION, format!("Bearer {api_key}")); + } + EndpointKind::Messages => { + request = request + .header("x-api-key", api_key) + .header("anthropic-version", "2023-06-01"); + } + } + let response = request.send().await.map_err(|_| OpenCodeError { + status: StatusCode::BAD_GATEWAY, + retry_after: None, + message: "OpenCode Go upstream request failed".to_string(), + })?; + + if let Some(capture) = traffic.as_ref() { + capture.write_json( + "030-upstream-response-headers", + &serde_json::json!({ + "status": response.status().as_u16(), + "headers": safe_headers(response.headers()) + }), + ); + } + + if !response.status().is_success() { + return Err(rejected_response(response).await); + } + Ok(OpenCodeResponse { response }) + } + + fn endpoint_url(&self, endpoint: EndpointKind) -> reqwest::Url { + let mut url = self.base_url.clone(); + let base_path = url.path().trim_end_matches('/'); + let suffix = match endpoint { + EndpointKind::ChatCompletions => "chat/completions", + EndpointKind::Messages => "messages", + }; + url.set_path(&format!("{base_path}/{suffix}")); + url + } +} + +async fn rejected_response(response: reqwest::Response) -> OpenCodeError { + let status = response.status(); + let retry_after = response + .headers() + .get(http::header::RETRY_AFTER) + .and_then(|value| value.to_str().ok()) + .map(str::to_string); + let body = response.bytes().await.unwrap_or_default(); + let body = &body[..body.len().min(64 * 1024)]; + let message = serde_json::from_slice::(body) + .ok() + .and_then(|value| { + value + .pointer("/error/message") + .or_else(|| value.get("message")) + .and_then(|value| value.as_str()) + .map(str::to_string) + }) + .filter(|value| !value.is_empty()) + .unwrap_or_else(|| format!("OpenCode Go upstream returned HTTP {status}")); + OpenCodeError { + status, + retry_after, + message, + } +} + +fn safe_headers(headers: &reqwest::header::HeaderMap) -> serde_json::Value { + let mut result = serde_json::Map::new(); + for name in [ + "content-type", + "content-length", + "retry-after", + "x-request-id", + ] { + if let Some(value) = headers.get(name).and_then(|value| value.to_str().ok()) { + result.insert( + name.to_string(), + serde_json::Value::String(value.to_string()), + ); + } + } + serde_json::Value::Object(result) +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::{ + Json, Router, + extract::{OriginalUri, State}, + http::HeaderMap, + routing::post, + }; + use std::sync::Mutex; + + #[derive(Debug)] + struct SeenRequest { + path: String, + authorization: String, + x_api_key: String, + anthropic_version: String, + body: serde_json::Value, + } + + type Seen = Arc>>; + + async fn capture_request( + State(seen): State, + OriginalUri(uri): OriginalUri, + headers: HeaderMap, + Json(body): Json, + ) -> Json { + seen.lock().unwrap().push(SeenRequest { + path: uri.path().to_string(), + authorization: headers + .get(http::header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + .unwrap_or_default() + .to_string(), + x_api_key: headers + .get("x-api-key") + .and_then(|value| value.to_str().ok()) + .unwrap_or_default() + .to_string(), + anthropic_version: headers + .get("anthropic-version") + .and_then(|value| value.to_str().ok()) + .unwrap_or_default() + .to_string(), + body, + }); + Json(serde_json::json!({"ok": true})) + } + + #[test] + fn endpoint_urls_preserve_go_base_path() { + let client = OpenCodeClient::new( + "https://opencode.ai/zen/go/v1/".to_string(), + Some("test".to_string()), + ) + .unwrap(); + assert_eq!( + client.endpoint_url(EndpointKind::ChatCompletions).as_str(), + "https://opencode.ai/zen/go/v1/chat/completions" + ); + assert_eq!( + client.endpoint_url(EndpointKind::Messages).as_str(), + "https://opencode.ai/zen/go/v1/messages" + ); + } + + #[tokio::test] + async fn endpoints_use_protocol_native_auth_and_wire_model_ids() { + let seen: Seen = Arc::new(Mutex::new(Vec::new())); + let app = Router::new() + .route("/v1/chat/completions", post(capture_request)) + .route("/v1/messages", post(capture_request)) + .with_state(seen.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + + let client = + OpenCodeClient::new(format!("http://{address}/v1"), Some("test-key".to_string())) + .unwrap(); + for (endpoint, model) in [ + (EndpointKind::ChatCompletions, "glm-5.2"), + (EndpointKind::Messages, "minimax-m3"), + ] { + client + .post( + endpoint, + &serde_json::json!({"model": model, "messages": []}), + false, + None, + ) + .await + .unwrap() + .into_bytes() + .await + .unwrap(); + } + server.abort(); + + let seen = seen.lock().unwrap(); + assert_eq!(seen.len(), 2); + assert_eq!(seen[0].path, "/v1/chat/completions"); + assert_eq!(seen[1].path, "/v1/messages"); + assert_eq!(seen[0].authorization, "Bearer test-key"); + assert!(seen[0].x_api_key.is_empty()); + assert!(seen[1].authorization.is_empty()); + assert_eq!(seen[1].x_api_key, "test-key"); + assert!(seen[0].anthropic_version.is_empty()); + assert_eq!(seen[1].anthropic_version, "2023-06-01"); + assert_eq!(seen[0].body["model"], "glm-5.2"); + assert_eq!(seen[1].body["model"], "minimax-m3"); + } +} diff --git a/src/providers/opencode/mod.rs b/src/providers/opencode/mod.rs new file mode 100644 index 0000000..4841086 --- /dev/null +++ b/src/providers/opencode/mod.rs @@ -0,0 +1,919 @@ +pub mod client; +pub mod model; + +use std::convert::Infallible; +use std::sync::Arc; + +use async_trait::async_trait; +use axum::{ + Json, + body::Body, + http::StatusCode, + response::{IntoResponse, Response}, +}; +use bytes::Bytes; +use futures_util::StreamExt; + +use crate::anthropic::{ + error::json_error, + schema::{CountTokensResponse, MessagesRequest}, + sse::SseDecoder, +}; +use crate::monitor::{MonitorHandle, usage_from_anthropic_sse}; +use crate::provider::{CliHandlers, Provider, RequestContext}; +use crate::providers::kimi::{ + count_tokens, + translate::{ + accumulate::accumulate_response, + request::translate_openai_compatible_request, + stream::{LiveStreamTranslator, stream_error}, + }, +}; +use crate::traffic::StreamTrafficCapture; + +use self::client::{OpenCodeClient, OpenCodeError, OpenCodeResponse}; +use self::model::EndpointKind; + +enum ClientState { + Ready(Arc), + Invalid(String), +} + +pub struct OpenCodeProvider { + client: ClientState, +} + +impl OpenCodeProvider { + pub fn new() -> Self { + let client = OpenCodeClient::new( + crate::config::opencode_base_url(), + crate::config::opencode_api_key(), + ) + .map(Arc::new) + .map(ClientState::Ready) + .unwrap_or_else(|error| ClientState::Invalid(error.to_string())); + Self { client } + } + + #[cfg(test)] + fn with_client(client: OpenCodeClient) -> Self { + Self { + client: ClientState::Ready(Arc::new(client)), + } + } + + fn client(&self) -> Result, String> { + match &self.client { + ClientState::Ready(client) => Ok(client.clone()), + ClientState::Invalid(error) => Err(error.clone()), + } + } +} + +impl Default for OpenCodeProvider { + fn default() -> Self { + Self::new() + } +} + +pub fn advertised_models() -> Vec { + model::advertised_models() +} + +#[async_trait] +impl Provider for OpenCodeProvider { + fn name(&self) -> &'static str { + "opencode" + } + + fn supported_models(&self) -> Vec { + advertised_models() + } + + fn cli(&self) -> &'static dyn CliHandlers { + &OPENCODE_CLI + } + + async fn handle_messages(&self, body: MessagesRequest, ctx: RequestContext) -> Response { + let requested = body.model.as_deref().unwrap_or_default(); + let Some(spec) = model::resolve(requested) else { + return json_error( + StatusCode::BAD_REQUEST, + "invalid_request_error", + format!("Unsupported OpenCode Go model: {requested}"), + ); + }; + if let Some(monitor) = ctx.monitor.as_ref() { + monitor.model_resolved(&ctx.req_id, spec.id); + monitor.upstream_started(&ctx.req_id); + } + let client = match self.client() { + Ok(client) => client, + Err(error) => { + return json_error( + StatusCode::INTERNAL_SERVER_ERROR, + "api_error", + format!("Invalid OpenCode Go configuration: {error}"), + ); + } + }; + + match spec.endpoint { + EndpointKind::ChatCompletions => { + let translated = match translate_openai_compatible_request(&body, spec.id.into()) { + Ok(translated) => translated, + Err(error) => { + return json_error( + StatusCode::BAD_REQUEST, + "invalid_request_error", + error.to_string(), + ); + } + }; + let upstream = match client + .post(spec.endpoint, &translated, true, ctx.traffic.clone()) + .await + { + Ok(upstream) => upstream, + Err(error) => return map_error(error), + }; + chat_response(upstream, &body, requested, &ctx).await + } + EndpointKind::Messages => { + let translated = match prepare_messages_request(&body, spec) { + Ok(value) => value, + Err(error) => { + return json_error( + StatusCode::BAD_REQUEST, + "invalid_request_error", + error.to_string(), + ); + } + }; + let upstream = match client + .post(spec.endpoint, &translated, body.stream, ctx.traffic.clone()) + .await + { + Ok(upstream) => upstream, + Err(error) => return map_error(error), + }; + messages_response(upstream, body.stream, &ctx).await + } + } + } + + async fn handle_count_tokens(&self, body: MessagesRequest, ctx: RequestContext) -> Response { + let requested = body.model.as_deref().unwrap_or_default(); + let Some(spec) = model::resolve(requested) else { + return json_error( + StatusCode::BAD_REQUEST, + "invalid_request_error", + format!("Unsupported OpenCode Go model: {requested}"), + ); + }; + if let Some(monitor) = ctx.monitor.as_ref() { + monitor.model_resolved(&ctx.req_id, spec.id); + } + let tokens = count_tokens::count_tokens(&body); + if let Some(monitor) = ctx.monitor.as_ref() { + monitor.usage_updated(&ctx.req_id, Some(tokens), None); + } + ( + StatusCode::OK, + Json(CountTokensResponse { + input_tokens: tokens, + }), + ) + .into_response() + } +} + +fn prepare_messages_request( + body: &MessagesRequest, + spec: model::ModelSpec, +) -> Result { + let mut translated = serde_json::to_value(body)?; + translated["model"] = serde_json::Value::String(spec.id.to_string()); + if spec.id == "minimax-m3" && translated.get("thinking").is_none() { + translated["thinking"] = serde_json::json!({"type": "adaptive"}); + } + Ok(translated) +} + +async fn chat_response( + upstream: OpenCodeResponse, + body: &MessagesRequest, + requested: &str, + ctx: &RequestContext, +) -> Response { + let message_id = format!("msg_{}", uuid::Uuid::new_v4().simple()); + if body.stream { + return chat_stream_response( + upstream, + message_id, + requested.to_string(), + ctx.monitor.clone(), + ctx.req_id.clone(), + ctx.traffic.clone(), + ); + } + let bytes = match upstream.into_bytes().await { + Ok(bytes) => bytes, + Err(error) => return map_error(error), + }; + if let Some(traffic) = ctx.traffic.as_ref() { + traffic.write_bytes("032-upstream-response-body.sse", &bytes); + } + match accumulate_response(&bytes, &message_id, requested) { + Ok(value) => { + if let Some(monitor) = ctx.monitor.as_ref() { + monitor.usage_updated( + &ctx.req_id, + value + .pointer("/usage/input_tokens") + .and_then(|value| value.as_u64()), + value + .pointer("/usage/output_tokens") + .and_then(|value| value.as_u64()), + ); + } + (StatusCode::OK, Json(value)).into_response() + } + Err(error) => json_error( + StatusCode::BAD_GATEWAY, + "api_error", + format!("OpenCode Go response translation failed: {error}"), + ), + } +} + +fn chat_stream_response( + upstream: OpenCodeResponse, + message_id: String, + model: String, + monitor: Option, + req_id: String, + traffic: Option>, +) -> Response { + let state = OpenCodeChatStreamState { + upstream: upstream.into_stream(), + translator: LiveStreamTranslator::new(message_id, model), + capture_decoder: SseDecoder::default(), + terminal: false, + error_sent: false, + monitor, + req_id, + bytes: 0, + chunks: 0, + stream_capture: traffic.as_ref().map(|traffic| traffic.stream_capture()), + traffic, + }; + let stream = futures_util::stream::unfold(state, |mut state| async move { + state + .next_output() + .await + .map(|bytes| (Ok::(Bytes::from(bytes)), state)) + }); + ( + [ + (http::header::CONTENT_TYPE, "text/event-stream"), + (http::header::CACHE_CONTROL, "no-cache"), + (http::header::CONNECTION, "keep-alive"), + ], + Body::from_stream(stream), + ) + .into_response() +} + +struct OpenCodeChatStreamState { + upstream: S, + translator: LiveStreamTranslator, + capture_decoder: SseDecoder, + terminal: bool, + error_sent: bool, + monitor: Option, + req_id: String, + bytes: u64, + chunks: u64, + stream_capture: Option, + traffic: Option>, +} + +impl OpenCodeChatStreamState +where + S: futures_util::Stream> + Unpin, +{ + async fn next_output(&mut self) -> Option> { + if self.terminal { + return None; + } + if self.error_sent { + self.terminal = true; + return None; + } + loop { + let chunk = match self.upstream.next().await { + Some(Ok(chunk)) => chunk, + Some(Err(_)) => return Some(self.fail_at("transport", "upstream_stream")), + None => { + let output = match self.translator.finish() { + Ok(output) => output, + Err(_) => return Some(self.fail_at("decoder", "incomplete_stream")), + }; + if self.capture_decoder.finish().is_err() { + return Some(self.fail_at("capture", "incomplete_stream")); + } + self.terminal = true; + self.capture_downstream(&output); + self.finish_capture(true); + return (!output.is_empty()).then_some(output); + } + }; + if self.bytes == 0 + && let Some(monitor) = self.monitor.as_ref() + { + monitor.generation_started(&self.req_id); + } + self.bytes = self.bytes.saturating_add(chunk.len() as u64); + self.chunks = self.chunks.saturating_add(1); + self.capture_upstream(&chunk); + + let output = match self.translator.push(&chunk) { + Ok(output) => output, + Err(_) => return Some(self.fail_at("translation", "invalid_event")), + }; + if !output.is_empty() { + let (input_tokens, output_tokens) = usage_from_anthropic_sse(&output); + if let Some(monitor) = self.monitor.as_ref() { + monitor.stream_progress( + &self.req_id, + output.len() as u64, + count_sse_events(&output), + input_tokens, + output_tokens, + ); + } + self.capture_downstream(&output); + } + if self.translator.is_finished() { + self.terminal = true; + self.finish_capture(true); + return (!output.is_empty()).then_some(output); + } + if !output.is_empty() { + return Some(output); + } + } + } + + fn capture_upstream(&mut self, bytes: &[u8]) { + let events = match self.capture_decoder.push(bytes) { + Ok(events) => events, + Err(_) => { + if let Some(capture) = self.stream_capture.as_mut() { + capture.malformed("decoder", "malformed_sse"); + } + return; + } + }; + let Some(capture) = self.stream_capture.as_mut() else { + return; + }; + for event in events { + match serde_json::from_str(&event.data) { + Ok(value) => capture.upstream_event(event.event.as_deref(), &value), + Err(_) if event.data.trim() == "[DONE]" => { + capture.upstream_event(event.event.as_deref(), &serde_json::json!("[DONE]")); + } + Err(_) => capture.malformed("json", "malformed_event"), + } + } + } + + fn capture_downstream(&mut self, bytes: &[u8]) { + let Some(capture) = self.stream_capture.as_mut() else { + return; + }; + let mut decoder = SseDecoder::default(); + if let Ok(events) = decoder.push(bytes) { + for event in events { + if let Ok(value) = serde_json::from_str(&event.data) { + capture.downstream_event(event.event.as_deref().unwrap_or("message"), value); + } + } + } + } + + fn fail_at(&mut self, stage: &str, kind: &str) -> Vec { + self.error_sent = true; + let output = stream_error("OpenCode Go stream is invalid"); + if let Some(capture) = self.stream_capture.as_mut() { + capture.malformed(stage, kind); + } + self.capture_downstream(&output); + self.finish_capture(false); + output + } + + fn finish_capture(&mut self, completed: bool) { + if let (Some(capture), Some(traffic)) = (self.stream_capture.take(), self.traffic.as_ref()) + { + capture.finish_named( + traffic, + serde_json::json!({ + "kind": if completed { "stream_completion" } else { "stream_error" }, + "bytes": self.bytes, + "chunks": self.chunks, + }), + "061-opencode-stream-summary", + ); + } + } +} + +impl Drop for OpenCodeChatStreamState { + fn drop(&mut self) { + if self.terminal || self.stream_capture.is_none() { + return; + } + if let (Some(capture), Some(traffic)) = (self.stream_capture.take(), self.traffic.as_ref()) + { + capture.finish_named( + traffic, + serde_json::json!({ + "kind": "stream_abandoned", + "reason": "downstream_body_dropped", + "bytes": self.bytes, + "chunks": self.chunks, + }), + "061-opencode-stream-summary", + ); + } + } +} + +async fn messages_response( + upstream: OpenCodeResponse, + stream: bool, + ctx: &RequestContext, +) -> Response { + if stream { + let monitor = ctx.monitor.clone(); + let req_id = ctx.req_id.clone(); + let mut bytes = 0u64; + let mut chunks = 0u64; + let stream = upstream.into_stream().map(move |chunk| { + let output = match chunk { + Ok(chunk) => { + bytes = bytes.saturating_add(chunk.len() as u64); + chunks = chunks.saturating_add(1); + if let Some(monitor) = monitor.as_ref() { + let (input_tokens, output_tokens) = usage_from_anthropic_sse(&chunk); + monitor.stream_progress( + &req_id, + bytes, + chunks, + input_tokens, + output_tokens, + ); + } + chunk + } + Err(_) => Bytes::from_static( + b"event: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"api_error\",\"message\":\"OpenCode Go upstream stream failed\"}}\n\n", + ), + }; + Ok::(output) + }); + ( + [ + (http::header::CONTENT_TYPE, "text/event-stream"), + (http::header::CACHE_CONTROL, "no-cache"), + (http::header::CONNECTION, "keep-alive"), + ], + Body::from_stream(stream), + ) + .into_response() + } else { + let bytes = match upstream.into_bytes().await { + Ok(bytes) => bytes, + Err(error) => return map_error(error), + }; + if let Some(traffic) = ctx.traffic.as_ref() { + traffic.write_bytes("032-upstream-response-body.json", &bytes); + } + let value = match serde_json::from_slice::(&bytes) { + Ok(value) => value, + Err(_) => { + return json_error( + StatusCode::BAD_GATEWAY, + "api_error", + "OpenCode Go returned invalid JSON", + ); + } + }; + if let Some(monitor) = ctx.monitor.as_ref() { + monitor.usage_updated( + &ctx.req_id, + value + .pointer("/usage/input_tokens") + .and_then(|value| value.as_u64()), + value + .pointer("/usage/output_tokens") + .and_then(|value| value.as_u64()), + ); + } + (StatusCode::OK, Json(value)).into_response() + } +} + +fn map_error(error: OpenCodeError) -> Response { + let error_type = match error.status { + StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN => "authentication_error", + StatusCode::TOO_MANY_REQUESTS => "rate_limit_error", + status if status.is_client_error() => "invalid_request_error", + _ => "api_error", + }; + let status = if error.status.is_server_error() { + StatusCode::BAD_GATEWAY + } else { + error.status + }; + let response = json_error(status, error_type, error.message); + if let Some(retry_after) = error.retry_after { + ([(http::header::RETRY_AFTER, retry_after)], response).into_response() + } else { + response + } +} + +fn count_sse_events(bytes: &[u8]) -> u64 { + String::from_utf8_lossy(bytes).matches("event:").count() as u64 +} + +struct OpenCodeCli; + +impl CliHandlers for OpenCodeCli { + fn login(&self) -> anyhow::Result<()> { + anyhow::bail!( + "OpenCode Go uses an API key; set CCP_OPENCODE_API_KEY, OPENCODE_API_KEY, or opencode.apiKey in config.json" + ) + } + + fn device(&self) -> anyhow::Result<()> { + self.login() + } + + fn status(&self) -> anyhow::Result<()> { + let Some(source) = crate::config::opencode_api_key_source() else { + anyhow::bail!("Not authenticated"); + }; + println!("API key configured: true"); + println!("Source: {source}"); + println!("Base URL: {}", crate::config::opencode_base_url()); + Ok(()) + } + + fn logout(&self) -> anyhow::Result<()> { + anyhow::bail!( + "OpenCode Go credentials are managed through environment variables or config.json" + ) + } +} + +static OPENCODE_CLI: OpenCodeCli = OpenCodeCli; + +#[cfg(test)] +mod tests { + use super::*; + use crate::provider::RequestContext; + use axum::{Json, Router, extract::OriginalUri, http::HeaderMap, routing::post}; + use serde_json::json; + + fn context() -> RequestContext { + RequestContext { + req_id: "req_test".to_string(), + provider: "opencode".to_string(), + session_id: None, + session_seq: None, + monitor: None, + traffic: None, + } + } + + #[test] + fn minimax_m3_defaults_to_adaptive_thinking_without_overriding_the_caller() { + let body: MessagesRequest = serde_json::from_value(json!({ + "model": "minimax-m3", + "messages": [{"role": "user", "content": "hello"}] + })) + .unwrap(); + let translated = + prepare_messages_request(&body, model::resolve("minimax-m3").unwrap()).unwrap(); + assert_eq!(translated["thinking"], json!({"type": "adaptive"})); + + let body: MessagesRequest = serde_json::from_value(json!({ + "model": "minimax-m3", + "messages": [{"role": "user", "content": "hello"}], + "thinking": {"type": "enabled", "budget_tokens": 2048} + })) + .unwrap(); + let translated = + prepare_messages_request(&body, model::resolve("minimax-m3").unwrap()).unwrap(); + assert_eq!( + translated["thinking"], + json!({"type": "enabled", "budget_tokens": 2048}) + ); + + let body: MessagesRequest = serde_json::from_value(json!({ + "model": "qwen3.7-max", + "messages": [{"role": "user", "content": "hello"}] + })) + .unwrap(); + let translated = + prepare_messages_request(&body, model::resolve("qwen3.7-max").unwrap()).unwrap(); + assert!(translated.get("thinking").is_none()); + } + + #[tokio::test] + async fn missing_key_is_actionable() { + let client = OpenCodeClient::new("https://example.com/v1".into(), None).unwrap(); + let provider = OpenCodeProvider::with_client(client); + let body: MessagesRequest = serde_json::from_value(json!({ + "model": "glm-5.2", + "messages": [{"role": "user", "content": "hello"}] + })) + .unwrap(); + let response = provider.handle_messages(body, context()).await; + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let value: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert!( + value["error"]["message"] + .as_str() + .unwrap() + .contains("OPENCODE_API_KEY") + ); + } + + #[tokio::test] + async fn count_tokens_is_local_for_both_protocol_families() { + let provider = OpenCodeProvider::with_client( + OpenCodeClient::new("https://example.com/v1".into(), None).unwrap(), + ); + for model in ["glm-5.2", "minimax-m3"] { + let body: MessagesRequest = serde_json::from_value(json!({ + "model": model, + "messages": [{"role": "user", "content": "hello world"}] + })) + .unwrap(); + let response = provider.handle_count_tokens(body, context()).await; + assert_eq!(response.status(), StatusCode::OK); + } + } + + async fn mock_go_upstream( + OriginalUri(uri): OriginalUri, + headers: HeaderMap, + Json(body): Json, + ) -> Response { + if uri.path().ends_with("/chat/completions") { + assert_eq!( + headers + .get(http::header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()), + Some("Bearer test-key") + ); + } else { + assert_eq!( + headers + .get("x-api-key") + .and_then(|value| value.to_str().ok()), + Some("test-key") + ); + } + if body["model"] == "qwen3.7-max" { + return ( + StatusCode::TOO_MANY_REQUESTS, + [(http::header::RETRY_AFTER, "17")], + Json(json!({"error":{"message":"Go limit reached"}})), + ) + .into_response(); + } + if uri.path().ends_with("/chat/completions") { + return ( + [(http::header::CONTENT_TYPE, "text/event-stream")], + concat!( + "data: {\"choices\":[{\"delta\":{\"content\":\"hello from chat\"}}]}\n\n", + "data: {\"choices\":[{\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":5,\"completion_tokens\":3}}\n\n", + "data: [DONE]\n\n" + ), + ) + .into_response(); + } + if body["stream"] == true { + return ( + [(http::header::CONTENT_TYPE, "text/event-stream")], + concat!( + "event: message_start\n", + "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_native\",\"type\":\"message\",\"role\":\"assistant\",\"model\":\"minimax-m3\",\"content\":[],\"stop_reason\":null,\"usage\":{\"input_tokens\":4,\"output_tokens\":0}}}\n\n", + "event: message_stop\n", + "data: {\"type\":\"message_stop\"}\n\n" + ), + ) + .into_response(); + } + Json(json!({ + "id": "msg_native", + "type": "message", + "role": "assistant", + "model": body["model"], + "content": [{"type":"text","text":"hello from messages"}], + "stop_reason": "end_turn", + "usage": {"input_tokens":4,"output_tokens":3} + })) + .into_response() + } + + async fn delayed_chat_upstream( + axum::extract::State(release): axum::extract::State>, + ) -> Response { + let stream = futures_util::stream::unfold(0u8, move |step| { + let release = release.clone(); + async move { + match step { + 0 => Some(( + Ok::(Bytes::from_static( + b"data: {\"choices\":[{\"delta\":{\"content\":\"early\"}}]}\n\n", + )), + 1, + )), + 1 => { + release.notified().await; + Some(( + Ok(Bytes::from_static( + b"data: {\"choices\":[{\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":2,\"completion_tokens\":1}}\n\ndata: [DONE]\n\n", + )), + 2, + )) + } + _ => None, + } + } + }); + ( + [(http::header::CONTENT_TYPE, "text/event-stream")], + Body::from_stream(stream), + ) + .into_response() + } + + async fn mock_provider() -> (OpenCodeProvider, tokio::task::JoinHandle<()>) { + let app = Router::new() + .route("/v1/chat/completions", post(mock_go_upstream)) + .route("/v1/messages", post(mock_go_upstream)); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + let client = + OpenCodeClient::new(format!("http://{address}/v1"), Some("test-key".to_string())) + .unwrap(); + (OpenCodeProvider::with_client(client), server) + } + + async fn delayed_provider() -> ( + OpenCodeProvider, + Arc, + tokio::task::JoinHandle<()>, + ) { + let release = Arc::new(tokio::sync::Notify::new()); + let app = Router::new() + .route("/v1/chat/completions", post(delayed_chat_upstream)) + .with_state(release.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + let client = + OpenCodeClient::new(format!("http://{address}/v1"), Some("test-key".to_string())) + .unwrap(); + (OpenCodeProvider::with_client(client), release, server) + } + + #[tokio::test] + async fn chat_stream_forwards_the_first_event_before_upstream_completion() { + let (provider, release, server) = delayed_provider().await; + let body: MessagesRequest = serde_json::from_value(json!({ + "model": "glm-5.2", + "stream": true, + "messages": [{"role": "user", "content": "hello"}] + })) + .unwrap(); + let response = provider.handle_messages(body, context()).await; + assert_eq!(response.status(), StatusCode::OK); + let mut stream = response.into_body().into_data_stream(); + let first = tokio::time::timeout(std::time::Duration::from_secs(1), stream.next()) + .await + .expect("the proxy buffered the stream until upstream completion") + .expect("stream ended before the first translated event") + .unwrap(); + let first = String::from_utf8(first.to_vec()).unwrap(); + assert!(first.contains("early")); + assert!(!first.contains("message_stop")); + + release.notify_one(); + let rest = tokio::time::timeout(std::time::Duration::from_secs(1), async move { + let mut output = Vec::new(); + while let Some(chunk) = stream.next().await { + output.extend_from_slice(&chunk.unwrap()); + } + output + }) + .await + .expect("stream did not finish after the upstream terminal event"); + assert!(String::from_utf8(rest).unwrap().contains("message_stop")); + server.abort(); + } + + #[tokio::test] + async fn provider_translates_chat_and_passes_messages_through() { + let (provider, server) = mock_provider().await; + + let chat: MessagesRequest = serde_json::from_value(json!({ + "model": "glm-5.2", + "stream": false, + "messages": [{"role":"user","content":"hello"}] + })) + .unwrap(); + let response = provider.handle_messages(chat, context()).await; + assert_eq!(response.status(), StatusCode::OK); + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let value: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(value["content"][0]["text"], "hello from chat"); + + let messages: MessagesRequest = serde_json::from_value(json!({ + "model": "opencode-go/minimax-m3", + "stream": false, + "messages": [{"role":"user","content":"hello"}] + })) + .unwrap(); + let response = provider.handle_messages(messages, context()).await; + assert_eq!(response.status(), StatusCode::OK); + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let value: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(value["model"], "minimax-m3"); + assert_eq!(value["content"][0]["text"], "hello from messages"); + server.abort(); + } + + #[tokio::test] + async fn messages_stream_is_passthrough_and_rate_limit_is_preserved() { + let (provider, server) = mock_provider().await; + let streaming: MessagesRequest = serde_json::from_value(json!({ + "model": "minimax-m3", + "stream": true, + "messages": [{"role":"user","content":"hello"}] + })) + .unwrap(); + let response = provider.handle_messages(streaming, context()).await; + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + response + .headers() + .get(http::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()), + Some("text/event-stream") + ); + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + assert!(String::from_utf8_lossy(&bytes).contains("event: message_stop")); + + let limited: MessagesRequest = serde_json::from_value(json!({ + "model": "qwen3.7-max", + "messages": [{"role":"user","content":"hello"}] + })) + .unwrap(); + let response = provider.handle_messages(limited, context()).await; + assert_eq!(response.status(), StatusCode::TOO_MANY_REQUESTS); + assert_eq!( + response + .headers() + .get(http::header::RETRY_AFTER) + .and_then(|value| value.to_str().ok()), + Some("17") + ); + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let value: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(value["error"]["type"], "rate_limit_error"); + assert_eq!(value["error"]["message"], "Go limit reached"); + server.abort(); + } +} diff --git a/src/providers/opencode/model.rs b/src/providers/opencode/model.rs new file mode 100644 index 0000000..685675d --- /dev/null +++ b/src/providers/opencode/model.rs @@ -0,0 +1,142 @@ +pub const MODEL_PREFIX: &str = "opencode-go/"; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum EndpointKind { + ChatCompletions, + Messages, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ModelSpec { + pub id: &'static str, + pub endpoint: EndpointKind, +} + +pub const MODELS: &[ModelSpec] = &[ + ModelSpec { + id: "glm-5.2", + endpoint: EndpointKind::ChatCompletions, + }, + ModelSpec { + id: "glm-5.1", + endpoint: EndpointKind::ChatCompletions, + }, + ModelSpec { + id: "glm-5", + endpoint: EndpointKind::ChatCompletions, + }, + ModelSpec { + id: "kimi-k2.7-code", + endpoint: EndpointKind::ChatCompletions, + }, + ModelSpec { + id: "kimi-k2.6", + endpoint: EndpointKind::ChatCompletions, + }, + ModelSpec { + id: "kimi-k2.5", + endpoint: EndpointKind::ChatCompletions, + }, + ModelSpec { + id: "deepseek-v4-pro", + endpoint: EndpointKind::ChatCompletions, + }, + ModelSpec { + id: "deepseek-v4-flash", + endpoint: EndpointKind::ChatCompletions, + }, + ModelSpec { + id: "mimo-v2.5", + endpoint: EndpointKind::ChatCompletions, + }, + ModelSpec { + id: "mimo-v2.5-pro", + endpoint: EndpointKind::ChatCompletions, + }, + ModelSpec { + id: "qwen3.5-plus", + endpoint: EndpointKind::Messages, + }, + ModelSpec { + id: "minimax-m3", + endpoint: EndpointKind::Messages, + }, + ModelSpec { + id: "minimax-m2.7", + endpoint: EndpointKind::Messages, + }, + ModelSpec { + id: "minimax-m2.5", + endpoint: EndpointKind::Messages, + }, + ModelSpec { + id: "qwen3.7-max", + endpoint: EndpointKind::Messages, + }, + ModelSpec { + id: "qwen3.7-plus", + endpoint: EndpointKind::Messages, + }, + ModelSpec { + id: "qwen3.6-plus", + endpoint: EndpointKind::Messages, + }, +]; + +pub fn resolve(raw: &str) -> Option { + let id = raw.strip_prefix(MODEL_PREFIX).unwrap_or(raw); + MODELS.iter().copied().find(|model| model.id == id) +} + +pub fn advertised_models() -> Vec { + let mut result = Vec::with_capacity(MODELS.len() * 2 - 1); + for model in MODELS { + // `kimi-k2.6` is already a public alias of the Kimi Code provider. + // Keep that behavior stable and expose the Go model via its canonical, + // provider-qualified OpenCode id. + if model.id != "kimi-k2.6" { + result.push(model.id.to_string()); + } + result.push(format!("{MODEL_PREFIX}{}", model.id)); + } + result.sort_unstable(); + result +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn supported_catalog_is_partitioned_by_wire_protocol() { + let chat = MODELS + .iter() + .filter(|model| model.endpoint == EndpointKind::ChatCompletions) + .count(); + let messages = MODELS + .iter() + .filter(|model| model.endpoint == EndpointKind::Messages) + .count(); + assert_eq!(chat, 10); + assert_eq!(messages, 7); + } + + #[test] + fn canonical_prefix_resolves_and_unknown_models_do_not() { + let spec = resolve("opencode-go/minimax-m3").expect("known model"); + assert_eq!(spec.id, "minimax-m3"); + assert_eq!(spec.endpoint, EndpointKind::Messages); + assert_eq!( + resolve("qwen3.5-plus").unwrap().endpoint, + EndpointKind::Messages + ); + assert!(resolve("opencode-go/not-a-model").is_none()); + } + + #[test] + fn conflicting_kimi_id_is_only_advertised_with_prefix() { + let models = advertised_models(); + assert!(!models.iter().any(|model| model == "kimi-k2.6")); + assert!(models.iter().any(|model| model == "opencode-go/kimi-k2.6")); + } +} diff --git a/src/registry.rs b/src/registry.rs index 4b149a3..027c38c 100644 --- a/src/registry.rs +++ b/src/registry.rs @@ -73,6 +73,10 @@ impl Registry { .map(|model| (*model).to_string()) .collect(), ); + models.insert( + "opencode".into(), + crate::providers::opencode::advertised_models(), + ); let mut handlers = BTreeMap::new(); for (name, entries) in &models { @@ -81,6 +85,7 @@ impl Registry { "kimi" => Arc::new(crate::providers::kimi::KimiProvider::new()), "cursor" => Arc::new(crate::providers::cursor::CursorProvider::new()), "grok" => Arc::new(crate::providers::grok::GrokProvider::new()), + "opencode" => Arc::new(crate::providers::opencode::OpenCodeProvider::new()), _ => Arc::new(PlaceholderProvider::new(name, entries.clone())), }; handlers.insert(name.clone(), handler); @@ -206,6 +211,7 @@ impl PlaceholderProvider { "kimi" => "kimi", "cursor" => "cursor", "grok" => "grok", + "opencode" => "opencode", _ => "codex", }; Self { name, models } @@ -228,6 +234,7 @@ impl Provider for PlaceholderProvider { "kimi" => &KIMI_CLI, "cursor" => &CURSOR_CLI, "grok" => &GROK_CLI, + "opencode" => &OPENCODE_CLI, _ => &CODEX_CLI, } } @@ -287,6 +294,9 @@ const CODEX_CLI: PlaceholderCli = PlaceholderCli { provider: "codex" }; const KIMI_CLI: PlaceholderCli = PlaceholderCli { provider: "kimi" }; const CURSOR_CLI: PlaceholderCli = PlaceholderCli { provider: "cursor" }; const GROK_CLI: PlaceholderCli = PlaceholderCli { provider: "grok" }; +const OPENCODE_CLI: PlaceholderCli = PlaceholderCli { + provider: "opencode", +}; fn expand_codex_models() -> Vec { let mut set = HashSet::new(); @@ -374,4 +384,30 @@ mod tests { "cursor" ); } + + #[test] + fn opencode_models_route_without_stealing_existing_kimi_alias() { + let registry = Registry::new(AliasProvider::Codex); + assert_eq!( + registry + .provider_for_model("kimi-k2.7-code", None) + .unwrap() + .name(), + "opencode" + ); + assert_eq!( + registry + .provider_for_model("opencode-go/kimi-k2.6", None) + .unwrap() + .name(), + "opencode" + ); + assert_eq!( + registry + .provider_for_model("kimi-k2.6", None) + .unwrap() + .name(), + "kimi" + ); + } } diff --git a/src/traffic.rs b/src/traffic.rs index 780d2b8..22da598 100644 --- a/src/traffic.rs +++ b/src/traffic.rs @@ -270,6 +270,10 @@ impl StreamTrafficCapture { } pub fn finish(self, traffic: &TrafficCapture, completion: Value) { + self.finish_named(traffic, completion, "061-grok-stream-summary"); + } + + pub fn finish_named(self, traffic: &TrafficCapture, completion: Value, summary_name: &str) { let upstream_event_count = self.upstream_events.len(); let downstream_event_count = self.downstream_events.len(); if !self.upstream_sse.is_empty() { @@ -296,7 +300,7 @@ impl StreamTrafficCapture { traffic.write_json_event("050-downstream-event", &value); } traffic.write_json( - "061-grok-stream-summary", + summary_name, &serde_json::json!({ "completion": completion, "upstream_sse": { diff --git a/tests/cli.rs b/tests/cli.rs index f3048fb..07190b0 100644 --- a/tests/cli.rs +++ b/tests/cli.rs @@ -24,6 +24,7 @@ fn models_prints_all_providers() -> Result<(), Box> { let out = String::from_utf8(cmd.output()?.stdout)?; assert!(out.contains("codex:")); assert!(out.contains("kimi:")); + assert!(out.contains("opencode:")); assert!(out.contains("cursor:")); let mut cmd = Command::cargo_bin("claude-code-proxy")?;