diff --git a/README.md b/README.md index 9c3067a..0a48409 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ # claude-code-proxy -Claude Code, powered by **OpenAI Codex**, **Kimi**, **Grok**, or **Cursor -Agent**. +Claude Code, powered by **OpenAI Codex**, **Kimi**, **Grok**, **OpenCode Go**, +or **Cursor Agent**. Docs: @@ -88,6 +88,7 @@ The opt-in Images API returns base64 image data and consumes the signed-in accou | Codex | ChatGPT Plus or Pro | Registered `gpt-*` models and `-fast` variants | | Kimi | kimi.com with Kimi Code access | `kimi-for-coding` and aliases | | Grok | grok.com | Registered Grok models | +| OpenCode Go | OpenCode Go subscription | Non-conflicting IDs and `opencode-go/` | | Cursor Agent | Cursor account | Cursor aliases and `cursor:` prefixes | Run `claude-code-proxy models` for the current catalog or diff --git a/docs/astro.config.mjs b/docs/astro.config.mjs index 58b08be..9388008 100644 --- a/docs/astro.config.mjs +++ b/docs/astro.config.mjs @@ -8,7 +8,7 @@ export default defineConfig({ integrations: [ starlight({ title: 'claude-code-proxy', - description: 'Run Claude Code with Codex, Kimi, Grok, or Cursor Agent.', + description: 'Run Claude Code with Codex, Kimi, Grok, OpenCode Go, or Cursor Agent.', plugins: [starlightLlmsTxt()], favicon: '/favicon.svg', head: [ @@ -47,6 +47,7 @@ export default defineConfig({ { label: 'Codex', slug: 'providers/codex' }, { label: 'Kimi', slug: 'providers/kimi' }, { label: 'Grok', slug: 'providers/grok' }, + { label: 'OpenCode Go', slug: 'providers/opencode-go' }, { label: 'Cursor Agent', slug: 'providers/cursor-agent' }, ], }, diff --git a/docs/src/content/docs/getting-started.md b/docs/src/content/docs/getting-started.md index a8bb3b4..f6943a5 100644 --- a/docs/src/content/docs/getting-started.md +++ b/docs/src/content/docs/getting-started.md @@ -3,7 +3,7 @@ title: Getting started description: Install claude-code-proxy, authenticate with Codex, start the server, and open one working Claude Code session. --- -This path gets one Codex-backed Claude Code session working. See [Choosing a provider](/providers/choosing-a-provider/) for Kimi, Grok, and Cursor Agent. +This path gets one Codex-backed Claude Code session working. See [Choosing a provider](/providers/choosing-a-provider/) for Kimi, Grok, OpenCode Go, and Cursor Agent. ## 1. Install diff --git a/docs/src/content/docs/how-it-works.md b/docs/src/content/docs/how-it-works.md index 124d0c5..598c576 100644 --- a/docs/src/content/docs/how-it-works.md +++ b/docs/src/content/docs/how-it-works.md @@ -10,7 +10,7 @@ claude-code-proxy exposes Anthropic Messages and optional OpenAI-compatible rout
Proxy pipelineroute model
refresh auth
translate events
-
Codex ResponsesKimi Chat CompletionsGrok ResponsesCursor Connect
+
Codex ResponsesKimi Chat CompletionsGrok ResponsesOpenCode Chat / Messages / ResponsesCursor Connect
## Anthropic requests @@ -25,17 +25,17 @@ claude-code-proxy exposes Anthropic Messages and optional OpenAI-compatible rout ## OpenAI-compatible requests -Enable the OpenAI routes to use `/v1/chat/completions` or `/v1/responses`. The `model` field chooses Codex, Kimi, Grok, or Cursor in the same way it does on `/v1/messages`. +Enable the OpenAI routes to use `/v1/chat/completions` or `/v1/responses`. The `model` field chooses Codex, Kimi, Grok, OpenCode Go, or Cursor in the same way it does on `/v1/messages`. Codex Responses requests go directly to the native Codex API. The proxy translates other OpenAI requests to the selected provider and returns either Chat Completions or Responses output. Streaming and non-streaming requests use the same translation rules, and unsupported fields return an error instead of being ignored. ## Authentication boundary -Each provider login belongs to claude-code-proxy. The proxy does not read native Codex, Grok, or Cursor Agent credentials. Credentials live in the platform credential store described in [Files and storage](/reference/files-and-storage/). Incoming `ANTHROPIC_AUTH_TOKEN` values are accepted for client compatibility and are not used as upstream credentials. +Each provider login belongs to claude-code-proxy. The proxy does not read native Codex, Grok, or Cursor Agent credentials. Credentials live in the platform credential store described in [Files and storage](/reference/files-and-storage/). OpenCode Go instead uses the configured subscription API key. Incoming `ANTHROPIC_AUTH_TOKEN` values are accepted for client compatibility and are not used as upstream credentials. ## Routing boundary -Routing happens per request, not per server process or API surface. Codex IDs, Kimi IDs, Grok IDs, Cursor prefixes, and configured Anthropic-style aliases can share one listener across `/v1/messages`, `/v1/chat/completions`, and `/v1/responses`. Unknown model IDs return HTTP 400 with the supported catalog. +Routing happens per request, not per server process or API surface. Codex IDs, Kimi IDs, Grok IDs, OpenCode Go IDs, Cursor prefixes, and configured Anthropic-style aliases can share one listener across `/v1/messages`, `/v1/chat/completions`, and `/v1/responses`. Unknown model IDs return HTTP 400 with the supported catalog. ## Session state diff --git a/docs/src/content/docs/index.md b/docs/src/content/docs/index.md index dfe5f39..fbd9192 100644 --- a/docs/src/content/docs/index.md +++ b/docs/src/content/docs/index.md @@ -1,10 +1,10 @@ --- title: What is claude-code-proxy? -description: Run Claude Code with Codex, Kimi, Grok, or Cursor Agent through one local Anthropic-compatible proxy. +description: Run Claude Code with Codex, Kimi, Grok, OpenCode Go, or Cursor Agent through one local Anthropic-compatible proxy. ---
-claude-code-proxy lets you use Claude Code with Codex, Kimi, Grok, or Cursor Agent. Start one local app, choose a model, and keep working in the Claude Code interface you already know. +claude-code-proxy lets you use Claude Code with Codex, Kimi, Grok, OpenCode Go, or Cursor Agent. Start one local app, choose a model, and keep working in the Claude Code interface you already know.
@@ -21,7 +21,7 @@ claude-code-proxy lets you use Claude Code with Codex, Kimi, Grok, or Cursor Age
Supported providers -
CodexKimiGrokCursor Agent
+
CodexKimiGrokOpenCode GoCursor Agent
@@ -30,7 +30,7 @@ claude-code-proxy lets you use Claude Code with Codex, Kimi, Grok, or Cursor Age - **Keep the Claude Code experience.** Skills, tools, hooks, subagents, IDE integrations, and the terminal interface stay on the client side. - **Use subscription-backed providers.** Authenticate with supported consumer accounts instead of putting provider API keys into Claude Code. - **Switch providers by model.** A single proxy process routes every request from its model ID. -- **Use Claude Code normally.** Tools, images, streaming responses, and reasoning work across supported providers. +- **Use Claude Code normally.** Tools and streaming are translated across providers; images and reasoning depend on the selected provider and model. - **See what is happening.** The monitor TUI shows sessions, requests, errors, models, token use, and throughput. Structured logs and optional traffic captures support deeper diagnosis. ![Claude Code running through claude-code-proxy](/claude-code-screenshot.webp) diff --git a/docs/src/content/docs/providers/choosing-a-provider.md b/docs/src/content/docs/providers/choosing-a-provider.md index 09c14bb..4ed6627 100644 --- a/docs/src/content/docs/providers/choosing-a-provider.md +++ b/docs/src/content/docs/providers/choosing-a-provider.md @@ -10,6 +10,7 @@ One `serve` process supports every provider. Choose based on the account you hav | [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` | Function tools, reasoning, web search, X search, citations | +| [OpenCode Go](/providers/opencode-go/) | OpenCode Go subscription | OpenAI-compatible Chat Completions, OpenAI Responses, or Anthropic-compatible Messages | Non-conflicting bare IDs and `opencode-go/` forms | Curated models tested and benchmarked for coding-agent use | | [Cursor Agent](/providers/cursor-agent/) | Cursor account | HTTP/2 Connect stream | Cursor modes and `cursor:` prefixes | Dynamic model catalog, effort variants, images, plan and ask modes, session continuation | ## Practical guidance @@ -17,6 +18,7 @@ One `serve` process supports every provider. Choose based on the account you hav - Start with **Codex** when you have a ChatGPT subscription and want the most developed Claude Code translation path. - Choose **Kimi** for the Kimi Code model and multimodal coding input. - Choose **Grok** for Grok models and hosted web or X search. +- Choose **OpenCode Go** when you have a Go subscription and want its documented model catalog in Claude Code. - Choose **Cursor Agent** when you want Cursor's model catalog and agent modes. It depends on an installed Cursor Agent bundle for protobuf schemas. ## Shared behavior diff --git a/docs/src/content/docs/providers/codex.md b/docs/src/content/docs/providers/codex.md index 0541c0c..7608258 100644 --- a/docs/src/content/docs/providers/codex.md +++ b/docs/src/content/docs/providers/codex.md @@ -100,7 +100,7 @@ While the native request is active, the monitor shows `compacting`. Structured l ## OpenAI-compatible APIs -`CCP_CODEX_RESPONSES_API=1` enables both `POST /v1/responses` and `POST /v1/chat/completions`. The setting is under Codex configuration, but the routes also accept Kimi, Grok, and Cursor models. +`CCP_CODEX_RESPONSES_API=1` enables both `POST /v1/responses` and `POST /v1/chat/completions`. The setting is under Codex configuration, but the routes also accept Kimi, Grok, OpenCode Go, and Cursor models. The Responses route preserves native JSON or SSE response bodies for registered Codex models. The Chat Completions route translates standard text messages, reasoning effort, JSON object or JSON Schema output, and buffered or streaming responses. Its omitted reasoning effort defaults to `medium`; the proxy-wide Codex effort override still takes precedence. diff --git a/docs/src/content/docs/providers/opencode-go.md b/docs/src/content/docs/providers/opencode-go.md new file mode 100644 index 0000000..e047038 --- /dev/null +++ b/docs/src/content/docs/providers/opencode-go.md @@ -0,0 +1,62 @@ +--- +title: OpenCode Go +description: Configure an OpenCode Go API key, model routing, streaming, tools, and provider overrides. +--- + +OpenCode Go uses the API at `https://opencode.ai/zen/go/v1`. Its catalog spans +OpenAI-compatible chat completions, Anthropic-compatible messages, and OpenAI +Responses; the proxy selects the wire protocol for each registered model. The +registered mapping follows the [official OpenCode Go endpoint table](https://opencode.ai/docs/go/#endpoints). + +## Account and authentication + +Subscribe to OpenCode Go, copy your API key, and provide it to the proxy: + +```sh +export OPENCODE_API_KEY=YOUR_OPENCODE_GO_API_KEY +claude-code-proxy serve +``` + +`CCP_OPENCODE_API_KEY` takes precedence over `OPENCODE_API_KEY`. The +`opencode.apiKey` configuration key is also supported. The proxy does not +implement an OpenCode login flow. + +## Models + +Run `claude-code-proxy models` for the statically registered catalog based on +OpenCode's documented endpoint table. Every registered model has a +provider-qualified form. Bare IDs are also accepted when they do not belong to +another provider: + +```sh +ANTHROPIC_MODEL=opencode-go/glm-5.2 \ +ANTHROPIC_SMALL_FAST_MODEL=opencode-go/glm-5.2 \ + claude --model opencode-go/glm-5.2 +``` + +The bare IDs `gpt-5.6-luna`, `grok-4.5`, `kimi-k3`, and `kimi-k2.6` remain +owned by the existing Codex, Grok, or Kimi providers. Prefix those IDs with +`opencode-go/` to select the OpenCode Go version. + +## Tools and streaming + +Claude function definitions, tool choices, tool calls, and tool results are +translated for chat-completions models. Tool-call argument fragments are +streamed incrementally and reassembled into Anthropic `tool_use` blocks. +Upstream tool behavior remains model-dependent. + +Models served through the Anthropic-compatible endpoint retain their native +messages stream. GPT 5.6 Luna uses OpenCode Go's Responses endpoint and is +translated to the same Anthropic event stream as other providers. The proxy +handles `/v1/messages/count_tokens` locally and does not send that request to +OpenCode Go. + +## Configuration + +- `CCP_OPENCODE_API_KEY`, `OPENCODE_API_KEY`, or `opencode.apiKey` supplies the key. +- `CCP_OPENCODE_BASE_URL` or `opencode.baseUrl` changes the API base URL. + +OpenCode Go may expose additional model IDs through `/models`, but the proxy +registers only models whose wire protocol is documented. Unknown IDs are +rejected locally. Access or usage-limit errors for registered models are +surfaced from OpenCode. diff --git a/docs/src/content/docs/reference/compatibility-and-limitations.md b/docs/src/content/docs/reference/compatibility-and-limitations.md index bd39757..9710a43 100644 --- a/docs/src/content/docs/reference/compatibility-and-limitations.md +++ b/docs/src/content/docs/reference/compatibility-and-limitations.md @@ -31,8 +31,8 @@ claude-code-proxy targets Claude Code's practical Anthropic API usage rather tha ## OpenAI API scope -- `CCP_CODEX_RESPONSES_API=1` enables `/v1/chat/completions` and `/v1/responses` for Codex, Kimi, Grok, and Cursor models. -- Codex Responses requests use native passthrough. Requests for the other providers support text, reasoning, function tools, tool results, token limits, usage, streaming, aliases, and `[1m]` model hints. +- `CCP_CODEX_RESPONSES_API=1` enables `/v1/chat/completions` and `/v1/responses` for Codex, Kimi, Grok, OpenCode Go, and Cursor models. +- Codex Responses requests use native passthrough. Requests for the other providers support their mapped subsets of text, reasoning, function tools, tool results, token limits, usage, streaming, aliases, and `[1m]` model hints; exact support depends on the selected provider and model. - Unsupported non-null request fields return an error instead of being ignored. - Grok search calls appear as Responses `web_search_call` items. Chat Completions returns the citations without a separate search item. - Cursor tool bridging supports `Read`, `Write`, and `Bash`. It requires streaming and a stable session ID. diff --git a/docs/src/content/docs/reference/configuration.md b/docs/src/content/docs/reference/configuration.md index 90c2161..4165a87 100644 --- a/docs/src/content/docs/reference/configuration.md +++ b/docs/src/content/docs/reference/configuration.md @@ -39,6 +39,10 @@ These settings configure the proxy process. Claude Code client settings such as "baseUrl": "https://cli-chat-proxy.grok.com/v1", "clientVersion": "0.2.93" }, + "opencode": { + "apiKey": "YOUR_OPENCODE_GO_API_KEY", + "baseUrl": "https://opencode.ai/zen/go/v1" + }, "cursor": { "baseUrl": "https://api2.cursor.sh", "clientVersion": "0.48.5", @@ -130,6 +134,14 @@ Proxy URLs may use `http`, `https`, `socks4`, `socks4a`, `socks5`, or `socks5h`. | `CCP_GROK_CLIENT_VERSION` | `grok.clientVersion` | `0.2.93` | Changes the Grok client version header. | | `CCP_GROK_TOOL_IMAGE` | none | `omit` | Selects `omit`, `reattach`, `inline`, or `reject` image handling. | +## OpenCode Go + +| Environment | Config key | Default | Purpose | +| --- | --- | --- | --- | +| `CCP_OPENCODE_API_KEY` | `opencode.apiKey` | unset | OpenCode Go API key; takes precedence over `OPENCODE_API_KEY` and config. | +| `OPENCODE_API_KEY` | `opencode.apiKey` | unset | Fallback API-key variable accepted by the proxy when the CCP-specific variable is unset. | +| `CCP_OPENCODE_BASE_URL` | `opencode.baseUrl` | `https://opencode.ai/zen/go/v1` | Changes the OpenCode Go API base URL. | + ## Cursor Agent | Environment | Config key | Default | Purpose | diff --git a/docs/src/content/docs/reference/files-and-storage.md b/docs/src/content/docs/reference/files-and-storage.md index fc3c229..4ac7a9a 100644 --- a/docs/src/content/docs/reference/files-and-storage.md +++ b/docs/src/content/docs/reference/files-and-storage.md @@ -30,7 +30,14 @@ On macOS, Codex and Cursor use Keychain services: Kimi and Grok use `//auth.json` on every platform. Codex and Cursor use the same file layout on Linux and Windows. File-backed credentials are written with restrictive permissions where supported. -When `CCP_CONFIG_DIR` is set, every provider uses `//auth.json`, including Codex and Cursor on macOS. `CCP_CURSOR_AUTH_TOKEN` bypasses Cursor's local credential store for that process. +When `CCP_CONFIG_DIR` is set, file-backed provider credentials use +`//auth.json`, including Codex and Cursor on macOS. +`CCP_CURSOR_AUTH_TOKEN` bypasses Cursor's local credential store for that +process. + +OpenCode Go is the exception: it reads its API key from +`CCP_OPENCODE_API_KEY`, `OPENCODE_API_KEY`, or `opencode.apiKey` in +`config.json` and does not create a provider auth store. The proxy owns these credentials independently of native Codex, Grok, and Cursor Agent stores. diff --git a/docs/src/content/docs/reference/http-api.md b/docs/src/content/docs/reference/http-api.md index 3b25b60..4b4f558 100644 --- a/docs/src/content/docs/reference/http-api.md +++ b/docs/src/content/docs/reference/http-api.md @@ -3,7 +3,7 @@ title: HTTP API description: Local routes for health checks, Anthropic Messages, token counts, model discovery, OpenAI-compatible requests, and Codex images. --- -The server exposes the Anthropic and OpenAI routes supported by the proxy. Each route uses the stored login for the model's provider. +The server exposes the Anthropic and OpenAI routes supported by the proxy. Each route uses the configured provider credential for the selected model.
No client authentication. The listener accepts requests without validating `Authorization` or `x-api-key`. Loopback is the default. Protect every non-loopback listener with a firewall or authenticating reverse proxy. @@ -37,7 +37,7 @@ Accepts the same basic Anthropic request shape and returns: {"input_tokens":1234} ``` -Codex tokenizes text locally with `o200k_base` and adds estimates for images, encrypted reasoning, and protocol framing. Kimi and Grok use local text heuristics, while Cursor estimates the rendered prompt from its character length. Counts support Claude Code compaction behavior and are estimates rather than provider billing values. +Codex tokenizes text locally with `o200k_base` and adds estimates for images, encrypted reasoning, and protocol framing. Kimi, Grok, and OpenCode Go use local text heuristics, while Cursor estimates the rendered prompt from its character length. Counts support Claude Code compaction behavior and are estimates rather than provider billing values. ## `GET /v1/models` @@ -64,7 +64,7 @@ Claude Code gateway discovery filters IDs according to its own model rules. See ## `POST /v1/chat/completions` -Enable this route with `CCP_CODEX_RESPONSES_API=1` or `codex.responsesApi: true`. The `model` field selects Codex, Kimi, Grok, or Cursor. The proxy ignores incoming bearer credentials and uses the stored login for that provider. +Enable this route with `CCP_CODEX_RESPONSES_API=1` or `codex.responsesApi: true`. The `model` field selects Codex, Kimi, Grok, OpenCode Go, or Cursor. The proxy ignores incoming bearer credentials and uses the configured provider credential. ```sh curl http://127.0.0.1:18765/v1/chat/completions \ @@ -72,7 +72,7 @@ curl http://127.0.0.1:18765/v1/chat/completions \ -d '{"model":"kimi-k2.6","messages":[{"role":"user","content":"Hello"}]}' ``` -For Kimi, Grok, and Cursor, the route accepts: +For Kimi, Grok, OpenCode Go, and Cursor, the route accepts: - `system`, `developer`, `user`, `assistant`, and `tool` messages - text, supported user images, function calls, and tool results @@ -116,7 +116,7 @@ The Images API is an internal ChatGPT Codex integration, not the public OpenAI P ## `POST /v1/responses` -Enable this route with `CCP_CODEX_RESPONSES_API=1` or `codex.responsesApi: true`. The `model` field selects Codex, Kimi, Grok, or Cursor. +Enable this route with `CCP_CODEX_RESPONSES_API=1` or `codex.responsesApi: true`. The `model` field selects Codex, Kimi, Grok, OpenCode Go, or Cursor. ```sh curl http://127.0.0.1:18765/v1/responses \ @@ -124,7 +124,7 @@ curl http://127.0.0.1:18765/v1/responses \ -d '{"model":"grok-4.5","input":"Hello"}' ``` -Codex models use native Responses passthrough, including native JSON and SSE output. For Kimi, Grok, and Cursor, the route accepts: +Codex models use native Responses passthrough, including native JSON and SSE output. For Kimi, Grok, OpenCode Go, and Cursor, the route accepts: - string input or message items - `instructions` diff --git a/docs/src/content/docs/using/for-coding-agents.md b/docs/src/content/docs/using/for-coding-agents.md index 6f1453d..cc7a5c1 100644 --- a/docs/src/content/docs/using/for-coding-agents.md +++ b/docs/src/content/docs/using/for-coding-agents.md @@ -55,7 +55,7 @@ For a source checkout, implementation and tests are the final authority when doc 1. Check liveness with `curl http://127.0.0.1:18765/healthz`. 2. Check the selected model with `claude-code-proxy models`. -3. Check provider credentials with ` auth status`. +3. Check stored provider credentials with ` auth status`. For OpenCode Go, inspect whether its API-key environment variable or config key is configured without printing the secret. 4. Read the monitor request detail and structured `proxy.log`. 5. Read the redacted payload under `errors/` for a failed response. 6. Enable verbose logging or traffic capture only for a focused reproduction. diff --git a/docs/src/content/docs/using/models-and-routing.md b/docs/src/content/docs/using/models-and-routing.md index bff6b47..a6dc6f0 100644 --- a/docs/src/content/docs/using/models-and-routing.md +++ b/docs/src/content/docs/using/models-and-routing.md @@ -12,6 +12,7 @@ The model ID in each request selects its provider. One proxy listener can serve | Registered `gpt-*` IDs and their `-fast` forms | Codex | | `kimi-for-coding`, `kimi-k2.6`, `k2.6` | Kimi | | `grok-composer-2.5-fast`, `grok-4.5` | Grok | +| Non-conflicting registered OpenCode Go IDs and every `opencode-go/` | OpenCode Go | | `cursor`, Cursor legacy aliases, `cursor:`, `cursor-plan:`, `cursor-ask:` | Cursor Agent | | Anthropic-style aliases such as `haiku`, `sonnet`, `opus`, `fable`, and registered `claude-*` aliases | The `aliasProvider`, Codex by default | diff --git a/docs/src/content/docs/using/troubleshooting.md b/docs/src/content/docs/using/troubleshooting.md index e127052..c44b707 100644 --- a/docs/src/content/docs/using/troubleshooting.md +++ b/docs/src/content/docs/using/troubleshooting.md @@ -25,6 +25,9 @@ claude-code-proxy cursor auth status ``` Use that provider's login command when credentials are missing or expired. Codex requires ChatGPT subscription auth, not an OpenAI API key. Each provider uses proxy-owned credentials. +OpenCode Go has no `auth status` command; configure its API key through +`CCP_OPENCODE_API_KEY`, `OPENCODE_API_KEY`, or `opencode.apiKey` in +`config.json`. ## Model returns HTTP 400 diff --git a/src/config.rs b/src/config.rs index 4ed20ab..10c7dc6 100644 --- a/src/config.rs +++ b/src/config.rs @@ -44,6 +44,7 @@ struct FileConfig { pub codex: Option, pub cursor: Option, pub grok: Option, + pub opencode: Option, } #[derive(Deserialize, Clone)] @@ -105,6 +106,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, @@ -281,6 +290,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()) @@ -320,6 +337,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 { if codex .reasoning_summary @@ -429,6 +454,70 @@ pub fn warn_grok_tool_image_mode_once(log: &crate::logging::Logger) { } } +struct ResolvedOpenCodeConfig { + api_key: Option, + api_key_source: Option<&'static str>, + base_url: String, +} + +fn resolve_opencode_config( + env: &HashMap, + config_dir: &Path, +) -> ResolvedOpenCodeConfig { + let file = read_file_config(config_dir).and_then(|file| file.opencode); + let file_key = file + .as_ref() + .and_then(|config| config.api_key.as_ref()) + .filter(|value| !value.is_empty()); + let (api_key, api_key_source) = if let Some(value) = env + .get("CCP_OPENCODE_API_KEY") + .filter(|value| !value.is_empty()) + { + (Some(value.clone()), Some("CCP_OPENCODE_API_KEY")) + } else if let Some(value) = env + .get("OPENCODE_API_KEY") + .filter(|value| !value.is_empty()) + { + (Some(value.clone()), Some("OPENCODE_API_KEY")) + } else if let Some(value) = file_key { + (Some(value.clone()), Some("config.json")) + } else { + (None, None) + }; + let base_url = env + .get("CCP_OPENCODE_BASE_URL") + .filter(|value| !value.is_empty()) + .cloned() + .or_else(|| { + file.as_ref() + .and_then(|config| config.base_url.as_ref()) + .filter(|value| !value.is_empty()) + .cloned() + }) + .unwrap_or_else(|| "https://opencode.ai/zen/go/v1".to_string()); + + ResolvedOpenCodeConfig { + api_key, + api_key_source, + base_url, + } +} + +pub fn opencode_api_key() -> Option { + let env: HashMap<_, _> = std::env::vars().collect(); + resolve_opencode_config(&env, &paths::config_dir()).api_key +} + +pub fn opencode_api_key_source() -> Option<&'static str> { + let env: HashMap<_, _> = std::env::vars().collect(); + resolve_opencode_config(&env, &paths::config_dir()).api_key_source +} + +pub fn opencode_base_url() -> String { + let env: HashMap<_, _> = std::env::vars().collect(); + resolve_opencode_config(&env, &paths::config_dir()).base_url +} + pub fn is_verbose() -> bool { log_verbose() } @@ -851,6 +940,36 @@ mod tests { )]) } + #[test] + fn opencode_config_reads_file_and_env_precedence() { + 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 mut env = HashMap::new(); + let resolved = resolve_opencode_config(&env, config.path()); + assert_eq!(resolved.api_key.as_deref(), Some("file-key")); + assert_eq!(resolved.api_key_source, Some("config.json")); + assert_eq!(resolved.base_url, "https://file.example/v1"); + + env.insert("OPENCODE_API_KEY".into(), "standard-key".into()); + let resolved = resolve_opencode_config(&env, config.path()); + assert_eq!(resolved.api_key.as_deref(), Some("standard-key")); + assert_eq!(resolved.api_key_source, Some("OPENCODE_API_KEY")); + + env.insert("CCP_OPENCODE_API_KEY".into(), "ccp-key".into()); + env.insert( + "CCP_OPENCODE_BASE_URL".into(), + "https://env.example/v1".into(), + ); + let resolved = resolve_opencode_config(&env, config.path()); + assert_eq!(resolved.api_key.as_deref(), Some("ccp-key")); + assert_eq!(resolved.api_key_source, Some("CCP_OPENCODE_API_KEY")); + assert_eq!(resolved.base_url, "https://env.example/v1"); + } + #[test] fn bind_address_defaults_to_loopback() { let config = tempfile::TempDir::new().unwrap(); diff --git a/src/main.rs b/src/main.rs index ed905c5..68f8554 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", "cursor"] { + for provider in ["codex", "kimi", "grok", "opencode", "cursor"] { let Some(models) = grouped.get(provider) else { continue; }; diff --git a/src/providers/codex/translate/request.rs b/src/providers/codex/translate/request.rs index b998039..d3677f3 100644 --- a/src/providers/codex/translate/request.rs +++ b/src/providers/codex/translate/request.rs @@ -427,6 +427,31 @@ pub fn has_hosted_web_search(req: &MessagesRequest) -> bool { pub fn translate_request( req: &MessagesRequest, opts: TranslateOptions, +) -> Result { + translate_request_inner(req, opts, true) +} + +pub fn translate_openai_compatible_request( + req: &MessagesRequest, + model: String, + session_id: Option, +) -> Result { + translate_request_inner( + req, + TranslateOptions { + session_id, + service_tier: None, + model, + use_responses_lite: false, + }, + false, + ) +} + +fn translate_request_inner( + req: &MessagesRequest, + opts: TranslateOptions, + apply_codex_config: bool, ) -> Result { let instructions = flatten_system_text(req.extra.get("system")); let is_compact = is_compact_messages_request(req); @@ -519,15 +544,22 @@ pub fn translate_request( out.prompt_cache_key = Some(sid); } - let service_tier = resolve_service_tier(opts.service_tier)?; - if let Some(ref tier) = service_tier { - out.service_tier = Some(tier.clone()); + if apply_codex_config { + let service_tier = resolve_service_tier(opts.service_tier)?; + if let Some(ref tier) = service_tier { + out.service_tier = Some(tier.clone()); + } } let effort = read_effort(req)?; let codex_effort = to_codex_effort(effort); - let mut resolved_effort = resolve_effort(codex_effort)?; - if is_compact + let mut resolved_effort = if apply_codex_config { + resolve_effort(codex_effort)? + } else { + codex_effort + }; + if apply_codex_config + && is_compact && let Some(cap) = compact_effort_cap() && resolved_effort.as_ref().is_some_and(|e| *e > cap) { @@ -535,7 +567,8 @@ pub fn translate_request( } if resolved_effort.is_some() || opts.use_responses_lite { let summary = if resolved_effort.is_some() - && reasoning_summary_requested(config::codex_reasoning_summary().as_deref()) + && (!apply_codex_config + || reasoning_summary_requested(config::codex_reasoning_summary().as_deref())) { Some("auto".to_string()) } else { 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/chat.rs b/src/providers/opencode/chat.rs new file mode 100644 index 0000000..1709e15 --- /dev/null +++ b/src/providers/opencode/chat.rs @@ -0,0 +1,1551 @@ +use std::convert::Infallible; +use std::sync::Arc; + +use axum::body::Body; +use base64::Engine; +use bytes::Bytes; +use futures_util::StreamExt; +use serde::Serialize; +use serde_json::{Value, json}; + +use super::client::{OpenCodeError, OpenCodeResponse}; +use crate::anthropic::{ + schema::MessagesRequest, + sse::{encode_sse_event, parse_sse_events}, +}; +use crate::monitor::{MonitorHandle, usage_from_anthropic_sse}; +use crate::providers::{ + grok::translate::stream::SseDecoder, + translate_shared::{ + ContentBlock, flatten_system_text, image_source_to_url, normalize_content, read_effort, + }, +}; +use crate::traffic::{StreamTrafficCapture, TrafficCapture}; + +const DEFAULT_MAX_TOKENS: u32 = 32_000; + +#[derive(Debug, Clone, Serialize)] +pub struct ChatRequest { + pub model: String, + pub messages: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub tools: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub tool_choice: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub parallel_tool_calls: Option, + pub stream: bool, + pub stream_options: StreamOptions, + pub max_tokens: u32, + #[serde(skip_serializing_if = "Option::is_none")] + pub reasoning_effort: Option, +} + +#[derive(Debug, Clone, Serialize)] +pub struct StreamOptions { + pub include_usage: bool, +} + +/// Translate the Anthropic Messages wire model to the OpenAI-compatible wire +/// model accepted by OpenCode Go's `/chat/completions` endpoint. +pub fn prepare_request(req: &MessagesRequest, model: &str) -> anyhow::Result { + let messages = build_messages(req, model)?; + let tools = read_tools(req)?; + let (tool_choice, parallel_tool_calls) = read_tool_choice(req)?; + if tools.is_empty() + && tool_choice + .as_ref() + .is_some_and(|choice| choice.as_str() != Some("none")) + { + anyhow::bail!("tool_choice requires at least one tool"); + } + + Ok(ChatRequest { + model: model.to_string(), + messages, + tools: (!tools.is_empty()).then_some(tools), + tool_choice, + parallel_tool_calls, + stream: true, + stream_options: StreamOptions { + include_usage: true, + }, + max_tokens: req + .max_tokens + .filter(|value| *value > 0) + .unwrap_or(DEFAULT_MAX_TOKENS), + reasoning_effort: map_reasoning_effort(req, model)?, + }) +} + +fn build_messages(req: &MessagesRequest, model: &str) -> anyhow::Result> { + let deepseek = model.to_ascii_lowercase().contains("deepseek"); + let mut system = Vec::new(); + if let Some(text) = flatten_system_text(req.extra.get("system")) { + system.push(text); + } + + let mut messages = Vec::new(); + for message in &req.messages { + let blocks = normalize_content(&message.content, json!({})); + match message.role.as_str() { + "system" | "developer" => { + let text = blocks + .iter() + .filter_map(|block| match block { + ContentBlock::Text { text } => Some(text.as_str()), + _ => None, + }) + .collect::>() + .join("\n"); + if !text.is_empty() { + system.push(text); + } + } + "user" => push_user_messages(&mut messages, &blocks), + "assistant" => { + if let Some(message) = assistant_message(&blocks, deepseek)? { + messages.push(message); + } + } + other => anyhow::bail!("unexpected message role: {other}"), + } + } + + if !system.is_empty() { + messages.insert( + 0, + json!({ + "role": "system", + "content": system.join("\n\n"), + }), + ); + } + Ok(messages) +} + +fn push_user_messages(messages: &mut Vec, blocks: &[ContentBlock]) { + let mut content = Vec::new(); + let mut tool_messages = Vec::new(); + let flush = |messages: &mut Vec, content: &mut Vec| { + if content.is_empty() { + return; + } + let value = if content + .iter() + .all(|part| part.get("type").and_then(Value::as_str) == Some("text")) + { + Value::String( + content + .iter() + .filter_map(|part| part.get("text").and_then(Value::as_str)) + .collect(), + ) + } else { + Value::Array(std::mem::take(content)) + }; + messages.push(json!({"role":"user", "content":value})); + content.clear(); + }; + + for block in blocks { + match block { + ContentBlock::Text { text } => content.push(json!({"type":"text", "text":text})), + ContentBlock::Image { source } => content.push(json!({ + "type":"image_url", + "image_url":{"url":image_source_to_url(source)}, + })), + ContentBlock::ToolResult { + tool_use_id, + content: result, + is_error, + } => { + let rendered = render_tool_result(result, is_error.unwrap_or(false)); + tool_messages.push(json!({ + "role":"tool", + "tool_call_id":tool_use_id, + "content":rendered.text, + })); + content.extend(rendered.images); + } + ContentBlock::Thinking { .. } | ContentBlock::ToolUse { .. } => {} + } + } + // A parallel assistant tool-call turn requires every matching tool + // response before any subsequent user message. Keep tool results in their + // original order, then reattach vision parts and ordinary user content in + // one protocol-compatible user message. + messages.extend(tool_messages); + flush(messages, &mut content); +} + +fn assistant_message(blocks: &[ContentBlock], deepseek: bool) -> anyhow::Result> { + let mut text = String::new(); + let mut reasoning = String::new(); + let mut tool_calls = Vec::new(); + for block in blocks { + match block { + ContentBlock::Text { text: value } => text.push_str(value), + ContentBlock::Thinking { + thinking, + signature: _, + } => { + if !reasoning.is_empty() && !thinking.is_empty() { + reasoning.push_str("\n\n"); + } + reasoning.push_str(thinking); + } + ContentBlock::ToolUse { id, name, input } => { + if id.is_empty() || name.is_empty() { + anyhow::bail!("assistant tool_use requires non-empty id and name"); + } + tool_calls.push(json!({ + "id":id, + "type":"function", + "function":{ + "name":name, + "arguments":serde_json::to_string(input)?, + }, + })); + } + ContentBlock::Image { .. } | ContentBlock::ToolResult { .. } => {} + } + } + if text.is_empty() && reasoning.is_empty() && tool_calls.is_empty() { + return Ok(None); + } + + let mut message = serde_json::Map::new(); + message.insert("role".into(), Value::String("assistant".into())); + message.insert("content".into(), Value::String(text)); + // DeepSeek requires the field on every replayed assistant message, even + // when that turn did not contain visible reasoning. + if deepseek || !reasoning.is_empty() { + message.insert("reasoning_content".into(), Value::String(reasoning)); + } + if !tool_calls.is_empty() { + message.insert("tool_calls".into(), Value::Array(tool_calls)); + } + Ok(Some(Value::Object(message))) +} + +struct RenderedToolResult { + text: String, + images: Vec, +} + +fn render_tool_result(value: &Value, is_error: bool) -> RenderedToolResult { + let mut text = String::new(); + let mut images = Vec::new(); + if is_error { + text.push_str("[tool execution error]\n"); + } + match value { + Value::String(value) => text.push_str(value), + Value::Array(parts) => { + for part in parts { + match part.get("type").and_then(Value::as_str) { + Some("text") => { + if let Some(value) = part.get("text").and_then(Value::as_str) { + text.push_str(value); + } + } + Some("image") => match normalized_tool_result_image(part) { + Some(image_url) => { + images.push(json!({ + "type":"image_url", + "image_url":{"url":image_url}, + })); + } + None => text.push_str("[unsupported tool result block omitted: image]"), + }, + Some(kind) => { + text.push_str(&format!("[unsupported tool result block omitted: {kind}]")) + } + None => text.push_str("[unsupported tool result block omitted]"), + } + } + } + other => text.push_str(&other.to_string()), + } + RenderedToolResult { text, images } +} + +fn normalized_tool_result_image(part: &Value) -> Option { + let block = normalize_content(&Value::Array(vec![part.clone()]), json!({})) + .into_iter() + .next()?; + let ContentBlock::Image { source } = block else { + return None; + }; + Some(image_source_to_url(&source)) +} + +fn read_tools(req: &MessagesRequest) -> anyhow::Result> { + let Some(value) = req.extra.get("tools") else { + return Ok(Vec::new()); + }; + let tools = value + .as_array() + .ok_or_else(|| anyhow::anyhow!("tools must be an array"))?; + tools + .iter() + .map(|tool| { + let name = tool + .get("name") + .and_then(Value::as_str) + .filter(|name| !name.is_empty()) + .ok_or_else(|| anyhow::anyhow!("tool name must be a non-empty string"))?; + let description = tool.get("description").cloned(); + let parameters = tool + .get("input_schema") + .cloned() + .unwrap_or_else(|| json!({})); + let mut function = serde_json::Map::new(); + function.insert("name".into(), Value::String(name.to_string())); + if let Some(description) = description.filter(|value| !value.is_null()) { + function.insert("description".into(), description); + } + function.insert("parameters".into(), parameters); + Ok(json!({"type":"function", "function":function})) + }) + .collect() +} + +fn read_tool_choice(req: &MessagesRequest) -> anyhow::Result<(Option, Option)> { + let Some(value) = req.extra.get("tool_choice") else { + return Ok((None, None)); + }; + if value.is_null() { + return Ok((None, None)); + } + let choice = value + .as_object() + .ok_or_else(|| anyhow::anyhow!("tool_choice must be an object"))?; + let parallel = match choice.get("disable_parallel_tool_use") { + Some(Value::Bool(disabled)) => Some(!disabled), + Some(_) => anyhow::bail!("tool_choice.disable_parallel_tool_use must be a boolean"), + None => None, + }; + let translated = match choice.get("type").and_then(Value::as_str) { + Some("auto") => Some(Value::String("auto".into())), + Some("none") => Some(Value::String("none".into())), + Some("any") => Some(Value::String("required".into())), + Some("tool") => { + let name = choice + .get("name") + .and_then(Value::as_str) + .filter(|name| !name.is_empty()) + .ok_or_else(|| anyhow::anyhow!("tool_choice.name must be a non-empty string"))?; + Some(json!({"type":"function", "function":{"name":name}})) + } + Some(kind) => anyhow::bail!("unsupported tool_choice type: {kind}"), + None => anyhow::bail!("tool_choice.type must be a string"), + }; + Ok((translated, parallel)) +} + +fn map_reasoning_effort(req: &MessagesRequest, model: &str) -> anyhow::Result> { + let Some(effort) = read_effort(req)? else { + return Ok(None); + }; + let id = model.to_ascii_lowercase(); + if ["glm-5.2", "glm-5-2", "glm-5p2"] + .iter() + .any(|needle| id.contains(needle)) + { + return match effort { + "high" => Ok(Some("high".into())), + "xhigh" | "max" => Ok(Some("max".into())), + other => anyhow::bail!( + "OpenCode Go model {model} does not support reasoning effort {other}; use high, xhigh, or max" + ), + }; + } + if id.contains("deepseek-v4") { + return match effort { + "low" | "medium" | "high" | "max" => Ok(Some(effort.into())), + "xhigh" => Ok(Some("max".into())), + _ => unreachable!("read_effort validates the effort"), + }; + } + if id.contains("mimo") { + return match effort { + "low" | "medium" | "high" => Ok(Some(effort.into())), + other => anyhow::bail!( + "OpenCode Go model {model} does not support reasoning effort {other}; use low, medium, or high" + ), + }; + } + // OpenCode exposes no selectable effort variants for the remaining chat + // models. Their native/default reasoning behavior remains in effect. + Ok(None) +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum StopReason { + EndTurn, + ToolUse, + MaxTokens, +} + +impl StopReason { + fn anthropic(self) -> &'static str { + match self { + Self::EndTurn => "end_turn", + Self::ToolUse => "tool_use", + Self::MaxTokens => "max_tokens", + } + } +} + +#[derive(Debug, Clone, Default, serde::Deserialize)] +struct Usage { + prompt_tokens: Option, + completion_tokens: Option, + cached_tokens: Option, + prompt_tokens_details: Option, +} + +#[derive(Debug, Clone, Default, serde::Deserialize)] +struct PromptTokensDetails { + cached_tokens: Option, +} + +impl Usage { + fn anthropic(&self) -> Value { + let cached = self + .prompt_tokens_details + .as_ref() + .and_then(|details| details.cached_tokens) + .or(self.cached_tokens) + .unwrap_or(0); + json!({ + "input_tokens":self.prompt_tokens.unwrap_or(0).saturating_sub(cached), + "output_tokens":self.completion_tokens.unwrap_or(0), + "cache_creation_input_tokens":0, + "cache_read_input_tokens":cached, + }) + } +} + +#[derive(Debug, serde::Deserialize)] +struct StreamChunk { + #[serde(default)] + choices: Option>, + #[serde(default)] + usage: Option, + #[serde(default)] + error: Option, +} + +#[derive(Debug, serde::Deserialize)] +struct UpstreamError { + #[serde(default)] + message: Option, +} + +#[derive(Debug, serde::Deserialize)] +struct Choice { + #[serde(default)] + delta: Option, + #[serde(default)] + finish_reason: Option, +} + +#[derive(Debug, Default, serde::Deserialize)] +struct Delta { + #[serde(default)] + content: Option, + #[serde(default)] + reasoning_content: Option, + #[serde(default)] + tool_calls: Option>, +} + +#[derive(Debug, serde::Deserialize)] +struct ToolCallDelta { + index: usize, + #[serde(default)] + id: Option, + #[serde(default)] + function: Option, +} + +#[derive(Debug, Default, serde::Deserialize)] +struct ToolFunctionDelta { + #[serde(default)] + name: Option, + #[serde(default)] + arguments: Option, +} + +#[derive(Debug)] +struct Block { + index: usize, + kind: BlockKind, +} + +#[derive(Debug)] +enum BlockKind { + Thinking { + text: String, + }, + Text { + text: String, + }, + Tool { + id: String, + name: String, + args: String, + }, +} + +#[derive(Debug)] +struct ToolSlot { + upstream_index: usize, + block_index: usize, + upstream_id: String, + downstream_id: String, + name: String, + args: String, + started: bool, +} + +struct TranslationState { + message_id: String, + model: String, + message_started: bool, + finished: bool, + next_block_index: usize, + thinking_index: Option, + text_index: Option, + blocks: Vec, + tools: Vec, + pending_stop: Option, + usage: Usage, +} + +impl TranslationState { + fn new(message_id: String, model: String) -> Self { + Self { + message_id, + model, + message_started: false, + finished: false, + next_block_index: 0, + thinking_index: None, + text_index: None, + blocks: Vec::new(), + tools: Vec::new(), + pending_stop: None, + usage: Usage::default(), + } + } + + fn apply_chunk(&mut self, chunk: StreamChunk) -> anyhow::Result> { + if self.finished { + anyhow::bail!("OpenCode Go event after terminal completion"); + } + if let Some(error) = chunk.error { + anyhow::bail!( + "OpenCode Go upstream error: {}", + error.message.unwrap_or_else(|| "unknown error".into()) + ); + } + if let Some(usage) = chunk.usage { + self.usage = usage; + } + let Some(choices) = chunk.choices else { + return Ok(Vec::new()); + }; + if choices.is_empty() { + return Ok(Vec::new()); + } + if choices.len() != 1 { + anyhow::bail!("OpenCode Go stream returned multiple choices"); + } + let choice = choices.into_iter().next().expect("one choice"); + let mut out = Vec::new(); + if let Some(delta) = choice.delta { + self.apply_delta(delta, &mut out)?; + } + if let Some(reason) = choice.finish_reason { + let reason = parse_finish_reason(&reason)?; + if self.pending_stop.is_some_and(|current| current != reason) { + anyhow::bail!("OpenCode Go stream changed finish_reason"); + } + self.pending_stop = Some(reason); + } + Ok(out) + } + + fn apply_delta(&mut self, delta: Delta, out: &mut Vec) -> anyhow::Result<()> { + if let Some(reasoning) = delta.reasoning_content.filter(|value| !value.is_empty()) { + self.close_text(out); + self.close_tools(out); + let index = match self.thinking_index { + Some(index) => index, + None => { + let index = self.allocate_block(); + self.ensure_message_start(out); + emit( + out, + "content_block_start", + json!({"type":"content_block_start","index":index,"content_block":{"type":"thinking","thinking":"","signature":""}}), + ); + self.blocks.push(Block { + index, + kind: BlockKind::Thinking { + text: String::new(), + }, + }); + self.thinking_index = Some(index); + index + } + }; + if let Some(Block { + kind: BlockKind::Thinking { text }, + .. + }) = self.blocks.iter_mut().find(|block| block.index == index) + { + text.push_str(&reasoning); + } + emit( + out, + "content_block_delta", + json!({"type":"content_block_delta","index":index,"delta":{"type":"thinking_delta","thinking":reasoning}}), + ); + } + + if let Some(content) = delta.content.filter(|value| !value.is_empty()) { + self.close_thinking(out); + self.close_tools(out); + let index = match self.text_index { + Some(index) => index, + None => { + let index = self.allocate_block(); + self.ensure_message_start(out); + emit( + out, + "content_block_start", + json!({"type":"content_block_start","index":index,"content_block":{"type":"text","text":""}}), + ); + self.blocks.push(Block { + index, + kind: BlockKind::Text { + text: String::new(), + }, + }); + self.text_index = Some(index); + index + } + }; + if let Some(Block { + kind: BlockKind::Text { text }, + .. + }) = self.blocks.iter_mut().find(|block| block.index == index) + { + text.push_str(&content); + } + emit( + out, + "content_block_delta", + json!({"type":"content_block_delta","index":index,"delta":{"type":"text_delta","text":content}}), + ); + } + + if let Some(tool_calls) = delta.tool_calls.filter(|value| !value.is_empty()) { + self.close_thinking(out); + self.close_text(out); + for call in tool_calls { + self.apply_tool_delta(call, out)?; + } + } + Ok(()) + } + + fn apply_tool_delta(&mut self, delta: ToolCallDelta, out: &mut Vec) -> anyhow::Result<()> { + let ToolCallDelta { + index: upstream_index, + id, + function, + } = delta; + let position = match self + .tools + .iter() + .position(|slot| slot.upstream_index == upstream_index) + { + Some(position) => position, + None => { + let block_index = self.allocate_block(); + self.tools.push(ToolSlot { + upstream_index, + block_index, + upstream_id: String::new(), + downstream_id: make_tool_use_id(&self.message_id, upstream_index), + name: String::new(), + args: String::new(), + started: false, + }); + self.tools.len() - 1 + } + }; + let mut new_arguments = String::new(); + { + let slot = &mut self.tools[position]; + if let Some(id) = id { + slot.upstream_id.push_str(&id); + } + if let Some(function) = function { + if let Some(name) = function.name { + slot.name.push_str(&name); + } + if let Some(arguments) = function.arguments { + new_arguments = arguments; + slot.args.push_str(&new_arguments); + } + } + } + let should_start = { + let slot = &self.tools[position]; + !slot.started && !slot.upstream_id.is_empty() && !slot.name.is_empty() + }; + if should_start { + self.ensure_message_start(out); + let (block_index, id, name, args) = { + let slot = &mut self.tools[position]; + slot.started = true; + ( + slot.block_index, + slot.downstream_id.clone(), + slot.name.clone(), + slot.args.clone(), + ) + }; + emit( + out, + "content_block_start", + json!({"type":"content_block_start","index":block_index,"content_block":{"type":"tool_use","id":id,"name":name,"input":{}}}), + ); + self.blocks.push(Block { + index: block_index, + kind: BlockKind::Tool { + id, + name, + args: String::new(), + }, + }); + if !args.is_empty() { + emit( + out, + "content_block_delta", + json!({"type":"content_block_delta","index":block_index,"delta":{"type":"input_json_delta","partial_json":args}}), + ); + if let Some(Block { + kind: BlockKind::Tool { args: stored, .. }, + .. + }) = self + .blocks + .iter_mut() + .find(|block| block.index == block_index) + { + stored.push_str(&args); + } + } + } else if self.tools[position].started && !new_arguments.is_empty() { + let block_index = self.tools[position].block_index; + emit( + out, + "content_block_delta", + json!({"type":"content_block_delta","index":block_index,"delta":{"type":"input_json_delta","partial_json":new_arguments}}), + ); + if let Some(Block { + kind: BlockKind::Tool { args, .. }, + .. + }) = self + .blocks + .iter_mut() + .find(|block| block.index == block_index) + { + args.push_str(&new_arguments); + } + } + Ok(()) + } + + fn finalize(&mut self) -> anyhow::Result> { + if self.finished { + return Ok(Vec::new()); + } + for slot in &self.tools { + if !slot.started { + anyhow::bail!( + "OpenCode Go tool call {} ended without id or function name", + slot.upstream_index + ); + } + } + for block in &self.blocks { + if let BlockKind::Tool { args, .. } = &block.kind { + let value: Value = serde_json::from_str(if args.is_empty() { "{}" } else { args }) + .map_err(|_| anyhow::anyhow!("OpenCode Go tool arguments are invalid JSON"))?; + if !value.is_object() { + anyhow::bail!("OpenCode Go tool arguments must be a JSON object"); + } + } + } + let mut out = Vec::new(); + self.close_thinking(&mut out); + self.close_text(&mut out); + self.close_tools(&mut out); + self.ensure_message_start(&mut out); + let stop = self.pending_stop.unwrap_or({ + if self.tools.is_empty() { + StopReason::EndTurn + } else { + StopReason::ToolUse + } + }); + emit( + &mut out, + "message_delta", + json!({"type":"message_delta","delta":{"stop_reason":stop.anthropic(),"stop_sequence":null},"usage":self.usage.anthropic()}), + ); + emit(&mut out, "message_stop", json!({"type":"message_stop"})); + self.finished = true; + Ok(out) + } + + fn response(&self) -> anyhow::Result { + if !self.finished { + anyhow::bail!("OpenCode Go response is not complete"); + } + let mut content = Vec::new(); + for block in &self.blocks { + match &block.kind { + BlockKind::Thinking { text } => content.push(json!({ + "type":"thinking", + "thinking":text, + "signature":make_thinking_signature(&self.message_id, block.index), + })), + BlockKind::Text { text } => { + content.push(json!({"type":"text", "text":text})); + } + BlockKind::Tool { id, name, args } => { + let input: Value = + serde_json::from_str(if args.is_empty() { "{}" } else { args })?; + content.push(json!({ + "type":"tool_use", + "id":id, + "name":name, + "input":input, + })); + } + } + } + let stop = self.pending_stop.unwrap_or({ + if self.tools.is_empty() { + StopReason::EndTurn + } else { + StopReason::ToolUse + } + }); + Ok(json!({ + "id":self.message_id, + "type":"message", + "role":"assistant", + "model":self.model, + "content":content, + "stop_reason":stop.anthropic(), + "stop_sequence":null, + "usage":self.usage.anthropic(), + })) + } + + fn ensure_message_start(&mut self, out: &mut Vec) { + if self.message_started { + return; + } + emit( + out, + "message_start", + 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}}}), + ); + self.message_started = true; + } + + fn allocate_block(&mut self) -> usize { + let index = self.next_block_index; + self.next_block_index += 1; + index + } + + fn close_thinking(&mut self, out: &mut Vec) { + let Some(index) = self.thinking_index.take() else { + return; + }; + emit( + out, + "content_block_delta", + json!({"type":"content_block_delta","index":index,"delta":{"type":"signature_delta","signature":make_thinking_signature(&self.message_id, index)}}), + ); + emit( + out, + "content_block_stop", + json!({"type":"content_block_stop","index":index}), + ); + } + + fn close_text(&mut self, out: &mut Vec) { + let Some(index) = self.text_index.take() else { + return; + }; + emit( + out, + "content_block_stop", + json!({"type":"content_block_stop","index":index}), + ); + } + + fn close_tools(&mut self, out: &mut Vec) { + for slot in &mut self.tools { + if slot.started { + emit( + out, + "content_block_stop", + json!({"type":"content_block_stop","index":slot.block_index}), + ); + slot.started = false; + } + } + } +} + +fn parse_finish_reason(reason: &str) -> anyhow::Result { + match reason { + "stop" => Ok(StopReason::EndTurn), + "tool_calls" | "function_call" => Ok(StopReason::ToolUse), + "length" => Ok(StopReason::MaxTokens), + "content_filter" => anyhow::bail!("OpenCode Go response was blocked by a content filter"), + other => anyhow::bail!("unsupported OpenCode Go finish_reason: {other}"), + } +} + +fn make_thinking_signature(message_id: &str, index: usize) -> String { + base64::engine::general_purpose::URL_SAFE_NO_PAD + .encode(format!("ccp:opencode:v1:{message_id}:{index}")) +} + +fn make_tool_use_id(message_id: &str, upstream_index: usize) -> String { + // Several Chat Completions providers restart tool-call IDs (for example, + // `Agent_0`) in every independent completion. Nested Claude Code agents + // are flattened into one downstream stream, so forwarding those IDs can + // make a child tool call appear to reference itself as its parent. The + // response message ID is generated uniquely by CCP; namespacing the slot + // with it keeps split deltas stable and independent completions distinct. + format!("toolu_{message_id}_{upstream_index}") +} + +fn emit(out: &mut Vec, event: &str, value: Value) { + out.extend(encode_sse_event(Some(event), &value.to_string())); +} + +pub struct LiveStreamTranslator { + decoder: SseDecoder, + state: TranslationState, +} + +impl LiveStreamTranslator { + pub fn new(message_id: String, model: String) -> Self { + Self { + decoder: SseDecoder::default(), + state: TranslationState::new(message_id, model), + } + } + + pub fn push(&mut self, input: &[u8]) -> anyhow::Result> { + let mut out = Vec::new(); + for event in self.decoder.push(input)? { + let data = event.data.trim(); + if data == "[DONE]" { + out.extend(self.state.finalize()?); + continue; + } + if self.state.finished { + validate_post_done_metadata(data)?; + continue; + } + let chunk: StreamChunk = serde_json::from_str(data) + .map_err(|_| anyhow::anyhow!("malformed OpenCode Go SSE event"))?; + out.extend(self.state.apply_chunk(chunk)?); + } + Ok(out) + } + + pub fn finish(&mut self) -> anyhow::Result> { + self.decoder.finish()?; + if self.state.finished { + return Ok(Vec::new()); + } + if self.state.pending_stop.is_none() { + anyhow::bail!("OpenCode Go stream ended without [DONE] or finish_reason"); + } + self.state.finalize() + } + + pub fn is_finished(&self) -> bool { + self.state.finished + } +} + +fn validate_post_done_metadata(data: &str) -> anyhow::Result<()> { + let value: Value = serde_json::from_str(data) + .map_err(|_| anyhow::anyhow!("malformed OpenCode Go SSE event after [DONE]"))?; + let object = value + .as_object() + .ok_or_else(|| anyhow::anyhow!("OpenCode Go event after terminal completion"))?; + let has_empty_choices = object + .get("choices") + .and_then(Value::as_array) + .is_some_and(Vec::is_empty); + let has_cost = object + .get("cost") + .is_some_and(|cost| cost.is_string() || cost.is_number()); + let known_keys = object.keys().all(|key| { + matches!( + key.as_str(), + "choices" | "cost" | "x-opencode-type" | "normalizedUsage" + ) + }); + if has_empty_choices && has_cost && known_keys { + return Ok(()); + } + anyhow::bail!("OpenCode Go event after terminal completion") +} + +pub fn accumulate_response(input: &[u8], message_id: &str, model: &str) -> anyhow::Result { + let mut translator = LiveStreamTranslator::new(message_id.into(), model.into()); + translator.push(input)?; + translator.finish()?; + translator.state.response() +} + +pub fn stream_error(message: &str) -> Vec { + encode_sse_event( + Some("error"), + &json!({"type":"error","error":{"type":"api_error","message":message}}).to_string(), + ) +} + +pub fn stream_body( + upstream: OpenCodeResponse, + message_id: String, + model: String, + monitor: Option, + req_id: String, + traffic: Option>, +) -> Body { + 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)) + }); + Body::from_stream(stream) +} + +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() { + if self.translator.finish().is_err() || self.capture_decoder.finish().is_err() { + return Some(self.fail_at("decoder", "trailing_incomplete_frame")); + } + 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(), &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; + }; + for event in parse_sse_events(bytes) { + 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, + 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, + json!({ + "kind":"stream_abandoned", + "reason":"downstream_body_dropped", + "bytes":self.bytes, + "chunks":self.chunks, + }), + "061-opencode-stream-summary", + ); + } + } +} + +fn count_sse_events(bytes: &[u8]) -> u64 { + parse_sse_events(bytes).len() as u64 +} + +#[cfg(test)] +mod tests { + use super::*; + + fn request(value: Value) -> MessagesRequest { + serde_json::from_value(value).unwrap() + } + + #[test] + fn request_maps_glm_effort_tools_parallelism_and_replay() { + let req = request(json!({ + "model":"opencode-go/glm-5.2", + "max_tokens":123, + "system":"system", + "messages":[ + {"role":"assistant","content":[ + {"type":"thinking","thinking":"reason","signature":"opaque"}, + {"type":"tool_use","id":"call_1","name":"lookup","input":{"q":"rust"}} + ]}, + {"role":"user","content":[{"type":"tool_result","tool_use_id":"call_1","content":"ok"}]} + ], + "tools":[{"name":"lookup","description":"Lookup","input_schema":{"type":"object"}}], + "tool_choice":{"type":"tool","name":"lookup","disable_parallel_tool_use":true}, + "output_config":{"effort":"xhigh"} + })); + let wire = serde_json::to_value(prepare_request(&req, "glm-5.2").unwrap()).unwrap(); + assert_eq!(wire["model"], "glm-5.2"); + assert_eq!(wire["max_tokens"], 123); + assert_eq!(wire["reasoning_effort"], "max"); + assert_eq!(wire["parallel_tool_calls"], false); + assert_eq!(wire["messages"][1]["reasoning_content"], "reason"); + assert_eq!(wire["messages"][1]["tool_calls"][0]["id"], "call_1"); + assert_eq!(wire["messages"][2]["tool_call_id"], "call_1"); + assert_eq!( + wire["messages"][1]["tool_calls"][0]["function"]["arguments"], + "{\"q\":\"rust\"}" + ); + } + + #[test] + fn tool_result_images_follow_the_required_tool_response_in_image_order() { + let req = request(json!({ + "messages":[ + {"role":"assistant","content":[ + {"type":"tool_use","id":"call_image","name":"Read","input":{"file_path":"image.png"}} + ]}, + {"role":"user","content":[ + {"type":"tool_result","tool_use_id":"call_image","is_error":true,"content":[ + {"type":"text","text":"before"}, + {"type":"image","source":{"type":"base64","media_type":"image/png","data":"YWJj"}}, + {"type":"text","text":"between"}, + {"type":"image","source":{"type":"url","url":"https://example.invalid/image.webp"}}, + {"type":"text","text":"after"} + ]} + ]} + ] + })); + + let wire = serde_json::to_value(prepare_request(&req, "glm-5.2").unwrap()).unwrap(); + let messages = wire["messages"].as_array().unwrap(); + assert_eq!(messages.len(), 3); + assert_eq!(messages[1]["role"], "tool"); + assert_eq!(messages[1]["tool_call_id"], "call_image"); + assert_eq!( + messages[1]["content"], + concat!("[tool execution error]\n", "before", "between", "after") + ); + assert!( + !messages[1]["content"] + .as_str() + .unwrap() + .contains("unsupported tool result block omitted: image") + ); + + assert_eq!(messages[2]["role"], "user"); + assert_eq!( + messages[2]["content"], + json!([ + { + "type":"image_url", + "image_url":{"url":"data:image/png;base64,YWJj"} + }, + { + "type":"image_url", + "image_url":{"url":"https://example.invalid/image.webp"} + } + ]) + ); + } + + #[test] + fn tool_result_image_reattachment_preserves_surrounding_user_message_order() { + let req = request(json!({ + "messages":[{"role":"user","content":[ + {"type":"text","text":"before tool"}, + {"type":"tool_result","tool_use_id":"call_image","content":[ + {"type":"image","source":{"type":"base64","media_type":"image/jpeg","data":"YWJj"}} + ]}, + {"type":"text","text":"after tool"} + ]}] + })); + + let wire = serde_json::to_value(prepare_request(&req, "glm-5.2").unwrap()).unwrap(); + let messages = wire["messages"].as_array().unwrap(); + assert_eq!(messages.len(), 2); + assert_eq!(messages[0]["role"], "tool"); + assert_eq!(messages[0]["tool_call_id"], "call_image"); + assert_eq!(messages[0]["content"], ""); + assert_eq!(messages[1]["role"], "user"); + assert_eq!( + messages[1]["content"], + json!([ + {"type":"text","text":"before tool"}, + { + "type":"image_url", + "image_url":{"url":"data:image/jpeg;base64,YWJj"} + }, + {"type":"text","text":"after tool"} + ]) + ); + } + + #[test] + fn parallel_tool_results_stay_contiguous_before_reattached_images() { + let req = request(json!({ + "messages":[ + {"role":"assistant","content":[ + {"type":"tool_use","id":"call_input","name":"Read","input":{"file_path":"input.txt"}}, + {"type":"tool_use","id":"call_image","name":"Read","input":{"file_path":"image.png"}}, + {"type":"tool_use","id":"call_bash","name":"Bash","input":{"command":"pwd"}} + ]}, + {"role":"user","content":[ + {"type":"tool_result","tool_use_id":"call_input","content":"input ok"}, + {"type":"tool_result","tool_use_id":"call_image","content":[ + {"type":"image","source":{"type":"base64","media_type":"image/png","data":"YWJj"}} + ]}, + {"type":"tool_result","tool_use_id":"call_bash","content":"bash ok"} + ]} + ] + })); + + let wire = serde_json::to_value(prepare_request(&req, "deepseek-v4-pro").unwrap()).unwrap(); + let messages = wire["messages"].as_array().unwrap(); + assert_eq!( + messages + .iter() + .map(|message| message["role"].as_str().unwrap()) + .collect::>(), + ["assistant", "tool", "tool", "tool", "user"] + ); + assert_eq!(messages[1]["tool_call_id"], "call_input"); + assert_eq!(messages[2]["tool_call_id"], "call_image"); + assert_eq!(messages[2]["content"], ""); + assert_eq!(messages[3]["tool_call_id"], "call_bash"); + assert_eq!( + messages[4]["content"], + json!([{ + "type":"image_url", + "image_url":{"url":"data:image/png;base64,YWJj"} + }]) + ); + } + + #[test] + fn deepseek_replay_always_has_reasoning_content() { + let req = request(json!({ + "messages":[ + {"role":"assistant","content":[{"type":"text","text":"answer"}]}, + {"role":"user","content":"next"} + ] + })); + let wire = serde_json::to_value(prepare_request(&req, "deepseek-v4-pro").unwrap()).unwrap(); + assert_eq!(wire["messages"][0]["reasoning_content"], ""); + } + + #[test] + fn live_stream_preserves_fragmented_reasoning_text_tools_usage_and_indices() { + let upstream = concat!( + "data: {\"choices\":[{\"delta\":{\"reasoning_content\":\"think\"}}]}\n\n", + "data: {\"choices\":[{\"delta\":{\"content\":\"answer\"}}]}\n\n", + "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":7,\"id\":\"call_1\",\"function\":{\"name\":\"search\",\"arguments\":\"{\\\"q\\\"\"}}]}}]}\n\n", + "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":7,\"function\":{\"arguments\":\":\\\"rust\\\"}\"}}]}}]}\n\n", + "data: {\"choices\":[{\"finish_reason\":\"tool_calls\"}]}\n\n", + "data: {\"choices\":[],\"usage\":{\"prompt_tokens\":10,\"completion_tokens\":4,\"prompt_tokens_details\":{\"cached_tokens\":3}}}\n\n", + "data: [DONE]\n\n" + ); + for split in 0..=upstream.len() { + let mut translator = LiveStreamTranslator::new("msg_1".into(), "glm-5.2".into()); + let mut output = translator.push(&upstream.as_bytes()[..split]).unwrap(); + output.extend(translator.push(&upstream.as_bytes()[split..]).unwrap()); + output.extend(translator.finish().unwrap()); + let rendered = String::from_utf8(output).unwrap(); + assert!(rendered.contains("thinking_delta"), "split {split}"); + assert!(rendered.contains("signature_delta"), "split {split}"); + assert!(rendered.contains("text_delta"), "split {split}"); + assert!(rendered.contains("input_json_delta"), "split {split}"); + assert!(rendered.contains("tool_use"), "split {split}"); + assert_eq!( + rendered.matches("\"id\":\"toolu_msg_1_7\"").count(), + 1, + "split {split}" + ); + assert!( + rendered.contains("cache_read_input_tokens"), + "split {split}" + ); + assert!(rendered.contains("message_stop"), "split {split}"); + } + } + + #[test] + fn chat_tool_ids_are_namespaced_per_response() { + let upstream = concat!( + "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"Agent_0\",\"function\":{\"name\":\"Agent\",\"arguments\":\"{}\"}}]}}]}\n\n", + "data: {\"choices\":[{\"finish_reason\":\"tool_calls\"}]}\n\n", + "data: [DONE]\n\n" + ); + let main = accumulate_response(upstream.as_bytes(), "msg_main", "kimi-k3").unwrap(); + let child = accumulate_response(upstream.as_bytes(), "msg_child", "kimi-k3").unwrap(); + let main_id = main["content"][0]["id"].as_str().unwrap(); + let child_id = child["content"][0]["id"].as_str().unwrap(); + + assert_eq!(main_id, "toolu_msg_main_0"); + assert_eq!(child_id, "toolu_msg_child_0"); + assert_ne!(main_id, child_id); + assert_ne!(main_id, "Agent_0"); + assert_ne!(child_id, "Agent_0"); + } + + #[test] + fn buffered_response_is_strict_and_complete() { + let upstream = concat!( + "data: {\"choices\":[{\"delta\":{\"content\":\"hello\"}}]}\n\n", + "data: {\"choices\":[{\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":2,\"completion_tokens\":1}}\n\n" + ); + let response = accumulate_response(upstream.as_bytes(), "msg_2", "glm-5.2").unwrap(); + assert_eq!(response["content"][0]["text"], "hello"); + assert_eq!(response["stop_reason"], "end_turn"); + assert_eq!(response["usage"]["output_tokens"], 1); + } + + #[test] + fn malformed_incomplete_and_unrecognized_terminal_streams_fail() { + let malformed = b"data: not-json\n\n"; + let incomplete = b"data: {\"choices\":[{\"delta\":{\"content\":\"x\"}}]}\n\n"; + let partial_frame = b"data: {\"choices\":[]}"; + let filtered = b"data: {\"choices\":[{\"finish_reason\":\"content_filter\"}]}\n\n"; + let unknown = b"data: {\"choices\":[{\"finish_reason\":\"mystery\"}]}\n\n"; + + assert!(accumulate_response(malformed, "m", "model").is_err()); + assert!(accumulate_response(incomplete, "m", "model").is_err()); + assert!(accumulate_response(partial_frame, "m", "model").is_err()); + assert!(accumulate_response(filtered, "m", "model").is_err()); + assert!(accumulate_response(unknown, "m", "model").is_err()); + } + + #[test] + fn done_is_a_real_terminal_even_without_finish_reason() { + let upstream = concat!( + "data: {\"choices\":[{\"delta\":{\"content\":\"hello\"}}]}\n\n", + "data: [DONE]\n\n" + ); + let response = accumulate_response(upstream.as_bytes(), "msg_3", "model").unwrap(); + assert_eq!(response["stop_reason"], "end_turn"); + } + + #[test] + fn accepts_opencode_cost_metadata_after_done() { + let upstream = concat!( + "data: {\"choices\":[{\"delta\":{\"content\":\"hello\"},\"finish_reason\":\"stop\"}]}\n\n", + "data: {\"choices\":[],\"usage\":{\"prompt_tokens\":2,\"completion_tokens\":1}}\n\n", + "data: [DONE]\n\n", + "data: {\"choices\":[],\"cost\":\"0\"}\n\n" + ); + let response = accumulate_response(upstream.as_bytes(), "msg_4", "glm-5.2").unwrap(); + assert_eq!(response["content"][0]["text"], "hello"); + assert_eq!(response["usage"]["output_tokens"], 1); + } + + #[test] + fn rejects_content_after_done() { + let upstream = concat!( + "data: [DONE]\n\n", + "data: {\"choices\":[{\"delta\":{\"content\":\"late\"}}]}\n\n" + ); + assert!(accumulate_response(upstream.as_bytes(), "msg_5", "glm-5.2").is_err()); + } + + #[tokio::test] + async fn live_stream_rejects_an_incomplete_frame_after_done() { + let upstream = futures_util::stream::iter([Ok::( + Bytes::from_static(b"data: [DONE]\n\ndata: {"), + )]); + let mut state = OpenCodeChatStreamState { + upstream, + translator: LiveStreamTranslator::new("msg_4".into(), "model".into()), + capture_decoder: SseDecoder::default(), + terminal: false, + error_sent: false, + monitor: None, + req_id: "req".into(), + bytes: 0, + chunks: 0, + stream_capture: None, + traffic: None, + }; + let output = state.next_output().await.expect("error event"); + assert!(String::from_utf8_lossy(&output).contains("OpenCode Go stream is invalid")); + } +} diff --git a/src/providers/opencode/client.rs b/src/providers/opencode/client.rs new file mode 100644 index 0000000..b28ca99 --- /dev/null +++ b/src/providers/opencode/client.rs @@ -0,0 +1,349 @@ +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 | EndpointKind::Responses => "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 | EndpointKind::Responses => { + 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", + EndpointKind::Responses => "responses", + }; + 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 mut stream = response.bytes_stream(); + let mut body = Vec::new(); + while body.len() < 64 * 1024 { + let Some(chunk) = stream.next().await else { + break; + }; + let Ok(chunk) = chunk else { + break; + }; + let remaining = 64 * 1024 - body.len(); + body.extend_from_slice(&chunk[..chunk.len().min(remaining)]); + } + 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" + ); + assert_eq!( + client.endpoint_url(EndpointKind::Responses).as_str(), + "https://opencode.ai/zen/go/v1/responses" + ); + } + + #[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)) + .route("/v1/responses", 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"), + (EndpointKind::Responses, "gpt-5.6-luna"), + ] { + 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(), 3); + assert_eq!(seen[0].path, "/v1/chat/completions"); + assert_eq!(seen[1].path, "/v1/messages"); + assert_eq!(seen[2].path, "/v1/responses"); + 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_eq!(seen[2].authorization, "Bearer test-key"); + assert!(seen[2].x_api_key.is_empty()); + assert!(seen[0].anthropic_version.is_empty()); + assert_eq!(seen[1].anthropic_version, "2023-06-01"); + assert!(seen[2].anthropic_version.is_empty()); + assert_eq!(seen[0].body["model"], "glm-5.2"); + assert_eq!(seen[1].body["model"], "minimax-m3"); + assert_eq!(seen[2].body["model"], "gpt-5.6-luna"); + } +} diff --git a/src/providers/opencode/messages.rs b/src/providers/opencode/messages.rs new file mode 100644 index 0000000..b0d6a07 --- /dev/null +++ b/src/providers/opencode/messages.rs @@ -0,0 +1,314 @@ +use std::convert::Infallible; +use std::sync::Arc; + +use axum::body::Body; +use bytes::Bytes; +use futures_util::StreamExt; + +use crate::anthropic::schema::MessagesRequest; +use crate::anthropic::sse::encode_sse_event; +use crate::monitor::MonitorHandle; +use crate::providers::grok::translate::stream::SseDecoder; +use crate::traffic::{StreamTrafficCapture, TrafficCapture}; + +use super::client::{OpenCodeError, OpenCodeResponse}; + +const DEFAULT_MAX_TOKENS: u32 = 32_000; + +pub fn prepare_request( + body: &MessagesRequest, + model: &str, +) -> Result { + let mut translated = serde_json::to_value(body)?; + translated["model"] = serde_json::Value::String(model.to_string()); + translated["max_tokens"] = serde_json::json!( + body.max_tokens + .filter(|value| *value > 0) + .unwrap_or(DEFAULT_MAX_TOKENS) + ); + if model == "minimax-m3" && translated.get("thinking").is_none() { + translated["thinking"] = serde_json::json!({"type": "adaptive"}); + } + Ok(translated) +} + +pub fn stream_body( + upstream: OpenCodeResponse, + monitor: Option, + req_id: String, + traffic: Option>, +) -> Body { + let state = MessagesStreamState { + upstream: upstream.into_stream(), + 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), state)) + }); + Body::from_stream(stream) +} + +struct MessagesStreamState { + upstream: S, + decoder: SseDecoder, + terminal: bool, + error_sent: bool, + monitor: Option, + req_id: String, + bytes: u64, + chunks: u64, + stream_capture: Option, + traffic: Option>, +} + +impl MessagesStreamState +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; + } + + let chunk = match self.upstream.next().await { + Some(Ok(chunk)) => chunk, + Some(Err(_)) => return Some(self.fail_at("transport", "upstream_stream")), + None => { + if self.decoder.finish().is_err() { + return Some(self.fail_at("decoder", "incomplete_stream")); + } + return Some(self.fail_at("protocol", "missing_message_stop")); + } + }; + 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); + + let events = match self.decoder.push(&chunk) { + Ok(events) => events, + Err(_) => return Some(self.fail_at("decoder", "malformed_sse")), + }; + let mut input_tokens = None; + let mut output_tokens = None; + let mut terminal = false; + for event in events { + if terminal { + return Some(self.fail_at("protocol", "event_after_message_stop")); + } + let value: serde_json::Value = match serde_json::from_str(&event.data) { + Ok(value) => value, + Err(_) => return Some(self.fail_at("json", "malformed_event")), + }; + if let Some(capture) = self.stream_capture.as_mut() { + capture.upstream_event(event.event.as_deref(), &value); + capture + .downstream_event(event.event.as_deref().unwrap_or("message"), value.clone()); + } + input_tokens = value + .pointer("/message/usage/input_tokens") + .or_else(|| value.pointer("/usage/input_tokens")) + .and_then(serde_json::Value::as_u64) + .or(input_tokens); + output_tokens = value + .pointer("/message/usage/output_tokens") + .or_else(|| value.pointer("/usage/output_tokens")) + .and_then(serde_json::Value::as_u64) + .or(output_tokens); + let kind = value.get("type").and_then(serde_json::Value::as_str); + if event.event.as_deref() == Some("error") || kind == Some("error") { + return Some(self.fail_at("upstream", "error_event")); + } + if event.event.as_deref() == Some("message_stop") || kind == Some("message_stop") { + terminal = true; + } + } + if let Some(monitor) = self.monitor.as_ref() { + monitor.stream_progress( + &self.req_id, + chunk.len() as u64, + 1, + input_tokens, + output_tokens, + ); + } + if terminal { + if self.decoder.finish().is_err() { + return Some(self.fail_at("decoder", "trailing_incomplete_frame")); + } + self.terminal = true; + self.finish_capture(true); + } + Some(chunk) + } + + fn fail_at(&mut self, stage: &str, kind: &str) -> Bytes { + self.error_sent = true; + if let Some(capture) = self.stream_capture.as_mut() { + capture.malformed(stage, kind); + } + if let Some(traffic) = self.traffic.as_ref() { + traffic.write_json( + "060-opencode-messages-stream-error", + &serde_json::json!({ + "stage": stage, + "kind": kind, + "bytes": self.bytes, + "chunks": self.chunks, + }), + ); + } + let value = serde_json::json!({ + "type": "error", + "error": { + "type": "api_error", + "message": "OpenCode Go Messages stream is invalid" + } + }); + if let Some(capture) = self.stream_capture.as_mut() { + capture.downstream_event("error", value.clone()); + } + self.finish_capture(false); + Bytes::from(encode_sse_event(Some("error"), &value.to_string())) + } + + 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-messages-stream-summary", + ); + } + } +} + +impl Drop for MessagesStreamState { + 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-messages-stream-summary", + ); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[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_request(&body, "minimax-m3").unwrap(); + assert_eq!(translated["thinking"], json!({"type": "adaptive"})); + assert_eq!(translated["max_tokens"], DEFAULT_MAX_TOKENS); + + let body: MessagesRequest = serde_json::from_value(json!({ + "model": "minimax-m3", + "max_tokens": 2048, + "messages": [{"role": "user", "content": "hello"}], + "thinking": {"type": "enabled", "budget_tokens": 2048} + })) + .unwrap(); + let translated = prepare_request(&body, "minimax-m3").unwrap(); + assert_eq!( + translated["thinking"], + json!({"type": "enabled", "budget_tokens": 2048}) + ); + assert_eq!(translated["max_tokens"], 2048); + } + + #[tokio::test] + async fn live_stream_rejects_an_incomplete_frame_after_message_stop() { + let upstream = + futures_util::stream::iter([Ok::(Bytes::from_static( + b"event: message_stop\ndata: {\"type\":\"message_stop\"}\n\ndata: {", + ))]); + let mut state = MessagesStreamState { + upstream, + decoder: SseDecoder::default(), + terminal: false, + error_sent: false, + monitor: None, + req_id: "req".into(), + bytes: 0, + chunks: 0, + stream_capture: None, + traffic: None, + }; + let output = state.next_output().await.expect("error event"); + assert!( + String::from_utf8_lossy(&output).contains("OpenCode Go Messages stream is invalid") + ); + } + + #[tokio::test] + async fn live_stream_rejects_a_complete_event_after_message_stop() { + let upstream = + futures_util::stream::iter([Ok::(Bytes::from_static( + concat!( + "event: message_stop\n", + "data: {\"type\":\"message_stop\"}\n\n", + "event: content_block_delta\n", + "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"late\"}}\n\n" + ) + .as_bytes(), + ))]); + let mut state = MessagesStreamState { + upstream, + decoder: SseDecoder::default(), + terminal: false, + error_sent: false, + monitor: None, + req_id: "req".into(), + bytes: 0, + chunks: 0, + stream_capture: None, + traffic: None, + }; + let output = state.next_output().await.expect("error event"); + assert!( + String::from_utf8_lossy(&output).contains("OpenCode Go Messages stream is invalid") + ); + } +} diff --git a/src/providers/opencode/mod.rs b/src/providers/opencode/mod.rs new file mode 100644 index 0000000..f9fc869 --- /dev/null +++ b/src/providers/opencode/mod.rs @@ -0,0 +1,758 @@ +pub mod chat; +pub mod client; +pub mod messages; +pub mod model; +pub mod responses; + +use std::sync::Arc; + +use async_trait::async_trait; +use axum::{ + Json, + body::Body, + http::StatusCode, + response::{IntoResponse, Response}, +}; + +use crate::anthropic::{ + error::json_error, + schema::{CountTokensResponse, MessagesRequest}, +}; +use crate::provider::{ + CliHandlers, Generation, GenerationBody, Provider, ProviderError, ProviderErrorKind, + RequestContext, +}; +use crate::providers::{ + codex::translate::accumulate::accumulate_response as accumulate_responses_response, + kimi::count_tokens, +}; + +use self::client::{OpenCodeClient, OpenCodeError}; +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()), + } + } + + async fn buffered_messages_response( + &self, + body: MessagesRequest, + ctx: RequestContext, + ) -> Response { + let requested = body.model.as_deref().unwrap_or_default(); + let Some(spec) = model::resolve(requested) else { + return unsupported_model(requested); + }; + if let Some(monitor) = ctx.monitor.as_ref() { + monitor.model_resolved(&ctx.req_id, spec.id); + } + let client = match self.client() { + Ok(client) => client, + Err(error) => return invalid_configuration_response(error), + }; + let message_id = format!("msg_{}", uuid::Uuid::new_v4().simple()); + + let value = match spec.endpoint { + EndpointKind::ChatCompletions => { + let translated = match chat::prepare_request(&body, spec.id) { + Ok(translated) => translated, + Err(error) => return invalid_request_response(error), + }; + mark_upstream_started(&ctx); + let upstream = match client + .post(spec.endpoint, &translated, true, ctx.traffic.clone()) + .await + { + Ok(upstream) => upstream, + Err(error) => return map_error(error), + }; + let bytes = match upstream.into_bytes().await { + Ok(bytes) => bytes, + Err(error) => return map_error(error), + }; + capture_buffered_upstream(&ctx, &bytes, "sse"); + match chat::accumulate_response(&bytes, &message_id, requested) { + Ok(value) => value, + Err(error) => return invalid_upstream_response(error), + } + } + EndpointKind::Messages => { + let translated = match messages::prepare_request(&body, spec.id) { + Ok(translated) => translated, + Err(error) => return invalid_request_response(error), + }; + mark_upstream_started(&ctx); + let upstream = match client + .post(spec.endpoint, &translated, false, ctx.traffic.clone()) + .await + { + Ok(upstream) => upstream, + Err(error) => return map_error(error), + }; + let bytes = match upstream.into_bytes().await { + Ok(bytes) => bytes, + Err(error) => return map_error(error), + }; + capture_buffered_upstream(&ctx, &bytes, "json"); + match serde_json::from_slice::(&bytes) { + Ok(value) => value, + Err(error) => return invalid_upstream_response(error), + } + } + EndpointKind::Responses => { + let translated = + match responses::prepare_request(&body, spec.id, ctx.session_id.clone()) { + Ok(translated) => translated, + Err(error) => return invalid_request_response(error), + }; + mark_upstream_started(&ctx); + let upstream = match client + .post(spec.endpoint, &translated, true, ctx.traffic.clone()) + .await + { + Ok(upstream) => upstream, + Err(error) => return map_error(error), + }; + let bytes = match upstream.into_bytes().await { + Ok(bytes) => bytes, + Err(error) => return map_error(error), + }; + capture_buffered_upstream(&ctx, &bytes, "sse"); + match accumulate_responses_response(&bytes, &message_id, requested) { + Ok(value) => value, + Err(error) => return invalid_upstream_response(error), + } + } + }; + + if let Some(traffic) = ctx.traffic.as_ref() { + traffic.write_json("051-downstream-response", &value); + } + update_buffered_usage(&ctx, &value); + (StatusCode::OK, Json(value)).into_response() + } +} + +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 { + if !body.stream { + return self.buffered_messages_response(body, ctx).await; + } + match self.generate_anthropic_stream(body, ctx).await { + Ok(generation) => sse_response(generation.body), + Err(error) => map_provider_error(error), + } + } + + 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 unsupported_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() + } + + async fn generate_anthropic_stream( + &self, + mut body: MessagesRequest, + ctx: RequestContext, + ) -> Result { + body.stream = true; + let requested = body.model.as_deref().unwrap_or_default(); + let spec = model::resolve(requested).ok_or_else(|| { + ProviderError::new( + StatusCode::BAD_REQUEST, + ProviderErrorKind::InvalidRequest, + format!("Unsupported OpenCode Go model: {requested}"), + ) + })?; + let client = self.client().map_err(|error| { + ProviderError::new( + StatusCode::INTERNAL_SERVER_ERROR, + ProviderErrorKind::Api, + format!("Invalid OpenCode Go configuration: {error}"), + ) + })?; + + if let Some(monitor) = ctx.monitor.as_ref() { + monitor.model_resolved(&ctx.req_id, spec.id); + monitor.upstream_started(&ctx.req_id); + } + let message_id = format!("msg_{}", uuid::Uuid::new_v4().simple()); + let body = match spec.endpoint { + EndpointKind::ChatCompletions => { + let translated = chat::prepare_request(&body, spec.id) + .map_err(invalid_request_provider_error)?; + let upstream = client + .post(spec.endpoint, &translated, true, ctx.traffic.clone()) + .await + .map_err(opencode_provider_error)?; + chat::stream_body( + upstream, + message_id, + requested.to_string(), + ctx.monitor.clone(), + ctx.req_id.clone(), + ctx.traffic.clone(), + ) + } + EndpointKind::Messages => { + let translated = messages::prepare_request(&body, spec.id) + .map_err(invalid_request_provider_error)?; + let upstream = client + .post(spec.endpoint, &translated, true, ctx.traffic.clone()) + .await + .map_err(opencode_provider_error)?; + messages::stream_body( + upstream, + ctx.monitor.clone(), + ctx.req_id.clone(), + ctx.traffic.clone(), + ) + } + EndpointKind::Responses => { + let translated = responses::prepare_request(&body, spec.id, ctx.session_id.clone()) + .map_err(invalid_request_provider_error)?; + let upstream = client + .post(spec.endpoint, &translated, true, ctx.traffic.clone()) + .await + .map_err(opencode_provider_error)?; + responses::stream_body( + upstream, + message_id, + requested.to_string(), + ctx.monitor.clone(), + ctx.req_id.clone(), + ctx.traffic.clone(), + ) + } + }; + + Ok(Generation { + body: GenerationBody::LiveSse(body), + resolved_model: spec.id.to_string(), + }) + } +} + +fn invalid_request_provider_error(error: impl std::fmt::Display) -> ProviderError { + ProviderError::new( + StatusCode::BAD_REQUEST, + ProviderErrorKind::InvalidRequest, + error.to_string(), + ) +} + +fn opencode_provider_error(error: OpenCodeError) -> ProviderError { + let (status, kind) = match error.status { + StatusCode::UNAUTHORIZED => (StatusCode::UNAUTHORIZED, ProviderErrorKind::Authentication), + StatusCode::PAYMENT_REQUIRED | StatusCode::FORBIDDEN => { + (error.status, ProviderErrorKind::Permission) + } + StatusCode::TOO_MANY_REQUESTS => { + (StatusCode::TOO_MANY_REQUESTS, ProviderErrorKind::RateLimit) + } + status if status.is_client_error() => (status, ProviderErrorKind::InvalidRequest), + _ => (StatusCode::BAD_GATEWAY, ProviderErrorKind::Api), + }; + let mut mapped = ProviderError::new(status, kind, error.message); + mapped.retry_after = error.retry_after; + mapped +} + +fn map_error(error: OpenCodeError) -> Response { + map_provider_error(opencode_provider_error(error)) +} + +fn map_provider_error(error: ProviderError) -> Response { + let response = json_error(error.status, error.error_type(), error.message); + if let Some(retry_after) = error.retry_after { + ([(http::header::RETRY_AFTER, retry_after)], response).into_response() + } else { + response + } +} + +fn unsupported_model(requested: &str) -> Response { + json_error( + StatusCode::BAD_REQUEST, + "invalid_request_error", + format!("Unsupported OpenCode Go model: {requested}"), + ) +} + +fn invalid_configuration_response(error: impl std::fmt::Display) -> Response { + json_error( + StatusCode::INTERNAL_SERVER_ERROR, + "api_error", + format!("Invalid OpenCode Go configuration: {error}"), + ) +} + +fn invalid_request_response(error: impl std::fmt::Display) -> Response { + json_error( + StatusCode::BAD_REQUEST, + "invalid_request_error", + error.to_string(), + ) +} + +fn invalid_upstream_response(error: impl std::fmt::Display) -> Response { + json_error( + StatusCode::BAD_GATEWAY, + "api_error", + format!("OpenCode Go response translation failed: {error}"), + ) +} + +fn mark_upstream_started(ctx: &RequestContext) { + if let Some(monitor) = ctx.monitor.as_ref() { + monitor.upstream_started(&ctx.req_id); + } +} + +fn capture_buffered_upstream(ctx: &RequestContext, bytes: &[u8], extension: &str) { + if let Some(traffic) = ctx.traffic.as_ref() { + traffic.write_bytes(&format!("032-upstream-response-body.{extension}"), bytes); + } +} + +fn update_buffered_usage(ctx: &RequestContext, value: &serde_json::Value) { + if let Some(monitor) = ctx.monitor.as_ref() { + monitor.usage_updated( + &ctx.req_id, + value + .pointer("/usage/input_tokens") + .and_then(serde_json::Value::as_u64), + value + .pointer("/usage/output_tokens") + .and_then(serde_json::Value::as_u64), + ); + } +} + +fn sse_response(body: GenerationBody) -> Response { + let body = match body { + GenerationBody::BufferedSse(bytes) => Body::from(bytes), + GenerationBody::LiveSse(body) => body, + }; + ( + [ + (http::header::CONTENT_TYPE, "text/event-stream"), + (http::header::CACHE_CONTROL, "no-cache"), + (http::header::CONNECTION, "keep-alive"), + ], + body, + ) + .into_response() +} + +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 axum::{ + Json, Router, + body::Body, + extract::OriginalUri, + http::HeaderMap, + response::{IntoResponse, Response}, + routing::post, + }; + use bytes::Bytes; + use futures_util::StreamExt; + use serde_json::json; + use std::convert::Infallible; + + use super::*; + + fn context() -> RequestContext { + RequestContext { + req_id: "req_test".to_string(), + provider: "opencode".to_string(), + session_id: None, + session_seq: None, + monitor: None, + traffic: 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_all_protocol_families() { + let provider = OpenCodeProvider::with_client( + OpenCodeClient::new("https://example.com/v1".into(), None).unwrap(), + ); + for model in ["glm-5.2", "minimax-m3", "opencode-go/gpt-5.6-luna"] { + 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("/messages") { + assert_eq!( + headers + .get("x-api-key") + .and_then(|value| value.to_str().ok()), + Some("test-key") + ); + } else { + assert_eq!( + headers + .get(http::header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()), + Some("Bearer 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 uri.path().ends_with("/responses") { + return ( + [(http::header::CONTENT_TYPE, "text/event-stream")], + concat!( + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"message\",\"id\":\"msg_up\"}}\n\n", + "data: {\"type\":\"response.output_text.delta\",\"output_index\":0,\"delta\":\"hello from responses\"}\n\n", + "data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"type\":\"message\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\",\"status\":\"completed\",\"usage\":{\"input_tokens\":6,\"output_tokens\":3}}}\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 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)) + .route("/v1/responses", 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_chat_upstream() -> Response { + let stream = futures_util::stream::unfold(0u8, |step| async move { + match step { + 0 => Some(( + Ok::(Bytes::from_static( + b"data: {\"choices\":[{\"delta\":{\"content\":\"early\"}}]}\n\n", + )), + 1, + )), + 1 => { + tokio::time::sleep(std::time::Duration::from_secs(30)).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() + } + + #[tokio::test] + async fn generate_contract_streams_before_upstream_completion() { + let app = Router::new().route("/v1/chat/completions", post(delayed_chat_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 provider = OpenCodeProvider::with_client( + OpenCodeClient::new(format!("http://{address}/v1"), Some("test-key".to_string())) + .unwrap(), + ); + let body: MessagesRequest = serde_json::from_value(json!({ + "model": "glm-5.2", + "messages": [{"role": "user", "content": "hello"}] + })) + .unwrap(); + let generation = provider + .generate_anthropic_stream(body, context()) + .await + .unwrap(); + let GenerationBody::LiveSse(body) = generation.body else { + panic!("OpenCode Go generation must remain live"); + }; + let mut stream = 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(); + assert!(String::from_utf8_lossy(&first).contains("early")); + server.abort(); + } + + #[tokio::test] + async fn buffered_messages_cover_all_three_upstream_protocols() { + let (provider, server) = mock_provider().await; + for (model, expected) in [ + ("glm-5.2", "hello from chat"), + ("opencode-go/minimax-m3", "hello from messages"), + ("opencode-go/gpt-5.6-luna", "hello from responses"), + ] { + let body: MessagesRequest = serde_json::from_value(json!({ + "model": model, + "stream": false, + "messages": [{"role":"user","content":"hello"}] + })) + .unwrap(); + let response = provider.handle_messages(body, context()).await; + assert_eq!(response.status(), StatusCode::OK, "model {model}"); + 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"], expected, "model {model}"); + } + server.abort(); + } + + #[tokio::test] + async fn streaming_contract_covers_all_three_upstream_protocols() { + let (provider, server) = mock_provider().await; + for model in [ + "glm-5.2", + "opencode-go/minimax-m3", + "opencode-go/gpt-5.6-luna", + ] { + let body: MessagesRequest = serde_json::from_value(json!({ + "model": model, + "messages": [{"role":"user","content":"hello"}] + })) + .unwrap(); + let generation = provider + .generate_anthropic_stream(body, context()) + .await + .unwrap(); + assert_eq!(generation.resolved_model, model::resolve(model).unwrap().id); + let GenerationBody::LiveSse(body) = generation.body else { + panic!("OpenCode Go generation must remain live"); + }; + let bytes = axum::body::to_bytes(body, usize::MAX).await.unwrap(); + assert!( + String::from_utf8_lossy(&bytes).contains("message_stop"), + "model {model}" + ); + } + server.abort(); + } + + #[tokio::test] + async fn rate_limit_and_retry_after_are_preserved() { + let (provider, server) = mock_provider().await; + let body: MessagesRequest = serde_json::from_value(json!({ + "model": "qwen3.7-max", + "messages": [{"role":"user","content":"hello"}] + })) + .unwrap(); + let response = provider.handle_messages(body, 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(); + } + + #[test] + fn permission_errors_are_not_reported_as_authentication_failures() { + for status in [StatusCode::PAYMENT_REQUIRED, StatusCode::FORBIDDEN] { + let error = opencode_provider_error(OpenCodeError { + status, + retry_after: None, + message: "denied".into(), + }); + assert_eq!(error.status, status); + assert_eq!(error.kind, ProviderErrorKind::Permission); + } + } +} diff --git a/src/providers/opencode/model.rs b/src/providers/opencode/model.rs new file mode 100644 index 0000000..fa97a4f --- /dev/null +++ b/src/providers/opencode/model.rs @@ -0,0 +1,158 @@ +pub const MODEL_PREFIX: &str = "opencode-go/"; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum EndpointKind { + ChatCompletions, + Messages, + Responses, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ModelSpec { + pub id: &'static str, + pub endpoint: EndpointKind, +} + +pub const MODELS: &[ModelSpec] = &[ + ModelSpec { + id: "grok-4.5", + endpoint: EndpointKind::ChatCompletions, + }, + ModelSpec { + id: "gpt-5.6-luna", + endpoint: EndpointKind::Responses, + }, + ModelSpec { + id: "glm-5.2", + endpoint: EndpointKind::ChatCompletions, + }, + ModelSpec { + id: "glm-5.1", + endpoint: EndpointKind::ChatCompletions, + }, + ModelSpec { + id: "kimi-k3", + endpoint: EndpointKind::ChatCompletions, + }, + ModelSpec { + id: "kimi-k2.7-code", + endpoint: EndpointKind::ChatCompletions, + }, + ModelSpec { + id: "kimi-k2.6", + 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: "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, + }, + ModelSpec { + id: "hy3", + endpoint: EndpointKind::ChatCompletions, + }, +]; + +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); + for model in MODELS { + if !matches!( + model.id, + "gpt-5.6-luna" | "grok-4.5" | "kimi-k3" | "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(); + let responses = MODELS + .iter() + .filter(|model| model.endpoint == EndpointKind::Responses) + .count(); + assert_eq!(chat, 11); + assert_eq!(messages, 6); + assert_eq!(responses, 1); + } + + #[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.7-plus").unwrap().endpoint, + EndpointKind::Messages + ); + assert!(resolve("opencode-go/not-a-model").is_none()); + } + + #[test] + fn conflicting_provider_ids_are_only_advertised_with_prefix() { + let models = advertised_models(); + for id in ["gpt-5.6-luna", "grok-4.5", "kimi-k3", "kimi-k2.6"] { + assert!(!models.iter().any(|model| model == id)); + assert!( + models + .iter() + .any(|model| model == &format!("opencode-go/{id}")) + ); + } + } +} diff --git a/src/providers/opencode/responses.rs b/src/providers/opencode/responses.rs new file mode 100644 index 0000000..3cf84ea --- /dev/null +++ b/src/providers/opencode/responses.rs @@ -0,0 +1,326 @@ +use std::convert::Infallible; +use std::sync::Arc; + +use axum::body::Body; +use bytes::Bytes; +use futures_util::StreamExt; + +use crate::anthropic::schema::MessagesRequest; +use crate::monitor::{MonitorHandle, usage_from_anthropic_sse}; +use crate::providers::codex::translate::{ + live_stream::LiveStreamTranslator, request::translate_openai_compatible_request, +}; +use crate::providers::grok::translate::stream::SseDecoder; +use crate::traffic::{StreamTrafficCapture, TrafficCapture}; + +use super::client::{OpenCodeError, OpenCodeResponse}; + +pub fn prepare_request( + body: &MessagesRequest, + model: &str, + session_id: Option, +) -> anyhow::Result { + let translated = translate_openai_compatible_request(body, model.to_string(), session_id)?; + let mut value = serde_json::to_value(translated)?; + if let Some(max_tokens) = body.max_tokens.filter(|value| *value > 0) { + value["max_output_tokens"] = serde_json::json!(max_tokens); + } + Ok(value) +} + +pub fn stream_body( + upstream: OpenCodeResponse, + message_id: String, + model: String, + monitor: Option, + req_id: String, + traffic: Option>, +) -> Body { + let state = ResponsesStreamState { + upstream: upstream.into_stream(), + decoder: SseDecoder::default(), + translator: LiveStreamTranslator::new(message_id, model), + 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)) + }); + Body::from_stream(stream) +} + +struct ResponsesStreamState { + upstream: S, + decoder: SseDecoder, + translator: LiveStreamTranslator, + terminal: bool, + error_sent: bool, + monitor: Option, + req_id: String, + bytes: u64, + chunks: u64, + stream_capture: Option, + traffic: Option>, +} + +impl ResponsesStreamState +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 => { + if self.decoder.finish().is_err() || !self.translator.is_finished() { + return Some(self.fail_at("decoder", "incomplete_stream")); + } + self.terminal = true; + self.finish_capture(true); + return None; + } + }; + + 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); + + let events = match self.decoder.push(&chunk) { + Ok(events) => events, + Err(_) => return Some(self.fail_at("decoder", "malformed_sse")), + }; + let mut completion_seen = false; + let mut done_seen = false; + for event in &events { + let data = event.data.trim(); + if data == "[DONE]" { + if !completion_seen || done_seen { + return Some(self.fail_at("protocol", "premature_done")); + } + done_seen = true; + continue; + } + if completion_seen || done_seen { + return Some(self.fail_at("protocol", "event_after_completion")); + } + let value: serde_json::Value = match serde_json::from_str(data) { + Ok(value) => value, + Err(_) => return Some(self.fail_at("json", "malformed_event")), + }; + completion_seen = matches!( + value.get("type").and_then(serde_json::Value::as_str), + Some("response.completed" | "response.incomplete" | "response.done") + ); + } + if completion_seen && self.decoder.finish().is_err() { + return Some(self.fail_at("decoder", "trailing_incomplete_frame")); + } + + let mut output = Vec::new(); + for event in events { + let data = event.data.trim(); + if data == "[DONE]" { + if let Some(capture) = self.stream_capture.as_mut() { + capture + .upstream_event(event.event.as_deref(), &serde_json::json!("[DONE]")); + } + continue; + } + let value: serde_json::Value = match serde_json::from_str(data) { + Ok(value) => value, + Err(_) => return Some(self.fail_at("json", "malformed_event")), + }; + if let Some(capture) = self.stream_capture.as_mut() { + capture.upstream_event(event.event.as_deref(), &value); + } + let translated = match self.translator.accept(&value, self.traffic.as_deref()) { + Ok(translated) => translated, + Err(_) => return Some(self.fail_at("translation", "invalid_event")), + }; + output.extend(translated); + } + if self.translator.is_finished() { + self.terminal = true; + self.record_progress(&output); + self.capture_downstream(&output); + self.finish_capture(true); + return (!output.is_empty()).then_some(output); + } + if !output.is_empty() { + self.record_progress(&output); + self.capture_downstream(&output); + return Some(output); + } + } + } + + fn record_progress(&self, output: &[u8]) { + let Some(monitor) = self.monitor.as_ref() else { + return; + }; + let (input_tokens, output_tokens) = usage_from_anthropic_sse(output); + monitor.stream_progress( + &self.req_id, + output.len() as u64, + count_sse_events(output), + input_tokens, + output_tokens, + ); + } + + fn fail_at(&mut self, stage: &str, kind: &str) -> Vec { + self.error_sent = true; + if let Some(capture) = self.stream_capture.as_mut() { + capture.malformed(stage, kind); + } + if let Some(traffic) = self.traffic.as_ref() { + traffic.write_json( + "060-opencode-responses-stream-error", + &serde_json::json!({ + "stage": stage, + "kind": kind, + "bytes": self.bytes, + "chunks": self.chunks, + }), + ); + } + let output = self.translator.error_chunk( + "OpenCode Go Responses stream is invalid", + "api_error", + self.traffic.as_deref(), + ); + self.capture_downstream(&output); + self.finish_capture(false); + output + } + + 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 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-responses-stream-summary", + ); + } + } +} + +impl Drop for ResponsesStreamState { + 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-responses-stream-summary", + ); + } + } +} + +fn count_sse_events(bytes: &[u8]) -> u64 { + String::from_utf8_lossy(bytes).matches("event:").count() as u64 +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn request_uses_standard_responses_fields_without_codex_lane_metadata() { + let body: MessagesRequest = serde_json::from_value(json!({ + "model": "opencode-go/gpt-5.6-luna", + "max_tokens": 2048, + "output_config": {"effort": "xhigh"}, + "messages": [{"role": "user", "content": "hello"}] + })) + .unwrap(); + let translated = prepare_request(&body, "gpt-5.6-luna", Some("session-1".into())) + .expect("translate Responses request"); + + assert_eq!(translated["model"], "gpt-5.6-luna"); + assert_eq!(translated["stream"], true); + assert_eq!(translated["max_output_tokens"], 2048); + assert_eq!(translated["reasoning"]["effort"], "xhigh"); + assert_eq!(translated["reasoning"]["summary"], "auto"); + assert_eq!(translated["prompt_cache_key"], "session-1"); + assert!(translated.get("service_tier").is_none()); + assert!(translated.get("client_metadata").is_none()); + } + + #[tokio::test] + async fn live_stream_rejects_an_incomplete_frame_after_completion() { + let upstream = futures_util::stream::iter([Ok::( + Bytes::from_static( + b"data: {\"type\":\"response.completed\",\"response\":{\"usage\":{\"input_tokens\":1,\"output_tokens\":1}}}\n\ndata: {", + ), + )]); + let mut state = ResponsesStreamState { + upstream, + decoder: SseDecoder::default(), + translator: LiveStreamTranslator::new("msg_1", "gpt-5.6-luna"), + terminal: false, + error_sent: false, + monitor: None, + req_id: "req".into(), + bytes: 0, + chunks: 0, + stream_capture: None, + traffic: None, + }; + let output = state.next_output().await.expect("error event"); + assert!( + String::from_utf8_lossy(&output).contains("OpenCode Go Responses stream is invalid") + ); + } +} diff --git a/src/registry.rs b/src/registry.rs index b181200..928af76 100644 --- a/src/registry.rs +++ b/src/registry.rs @@ -74,6 +74,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 { @@ -82,6 +86,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); @@ -306,7 +311,6 @@ 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" }; - fn expand_codex_models() -> Vec { let mut set = HashSet::new(); let mut out = Vec::new(); @@ -398,4 +402,47 @@ mod tests { "cursor" ); } + + #[test] + fn opencode_models_route_without_stealing_existing_provider_ids() { + 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" + ); + for (model, owner) in [ + ("gpt-5.6-luna", "codex"), + ("grok-4.5", "grok"), + ("kimi-k3", "kimi"), + ] { + assert_eq!( + registry.provider_for_model(model, None).unwrap().name(), + owner + ); + assert_eq!( + registry + .provider_for_model(&format!("opencode-go/{model}"), None) + .unwrap() + .name(), + "opencode" + ); + } + } } diff --git a/src/traffic.rs b/src/traffic.rs index 571d7ee..67f895b 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 1cc3e79..8cc555d 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")?;