diff --git a/crates/adapter-zarvis/src/agent.rs b/crates/adapter-zarvis/src/agent.rs index 95769717..6476f4cd 100644 --- a/crates/adapter-zarvis/src/agent.rs +++ b/crates/adapter-zarvis/src/agent.rs @@ -692,6 +692,7 @@ impl ResolvedModel { provider::routing::Provider::OpenAI => "openai", provider::routing::Provider::Anthropic => "anthropic", provider::routing::Provider::Ollama => "ollama", + provider::routing::Provider::CodexOauth => "codex-oauth", } } } @@ -733,6 +734,9 @@ pub fn resolve_model_from_spec(spec_str: &str) -> Result { Box::new(provider::anthropic::Anthropic::from_env()?) } provider::routing::Provider::Ollama => Box::new(provider::ollama::Ollama::from_env()?), + provider::routing::Provider::CodexOauth => { + Box::new(provider::codex_oauth::CodexOauth::from_env()?) + } }; Ok(ResolvedModel { model: spec.model, diff --git a/crates/adapter-zarvis/src/context.rs b/crates/adapter-zarvis/src/context.rs index c6446176..ec498fb5 100644 --- a/crates/adapter-zarvis/src/context.rs +++ b/crates/adapter-zarvis/src/context.rs @@ -33,6 +33,13 @@ pub fn context_window_tokens(provider: &str, model: &str) -> usize { ("openai", _) => 32_000, ("anthropic", _) => 200_000, ("ollama", _) => 8_000, + // ChatGPT-subscription Codex backend. Same gpt-5* family, + // same advertised context window as the platform API — the + // billing pipe is what differs, not the model. Starting + // value; the runtime overflow-learn path in `model_limits.rs` + // will tighten if the subscription tier enforces something + // lower in practice. + ("codex-oauth", _) => 400_000, _ => 8_000, } } diff --git a/crates/adapter-zarvis/src/provider/codex_oauth.rs b/crates/adapter-zarvis/src/provider/codex_oauth.rs new file mode 100644 index 00000000..9b594499 --- /dev/null +++ b/crates/adapter-zarvis/src/provider/codex_oauth.rs @@ -0,0 +1,1082 @@ +//! Codex OAuth provider — bills against the user's ChatGPT subscription +//! by reading `~/.codex/auth.json` (the credential file the +//! `codex login` command writes) and routing requests through +//! `POST https://chatgpt.com/backend-api/codex/responses`. +//! +//! This is NOT the public platform API at `api.openai.com/v1/responses`: +//! +//! - Endpoint host + path differ (`chatgpt.com/backend-api/codex/...`). +//! - Auth is `Authorization: Bearer ` from +//! `auth.json`, plus a `ChatGPT-Account-ID` header carrying the +//! `account_id` claim from the OAuth `id_token` JWT. +//! - Required `OpenAI-Beta` + `originator: codex_cli_rs` + non-default +//! User-Agent — the default `reqwest/*` UA gets blocked by +//! Cloudflare (verified against `openai/codex` CLI source). +//! - The request body uses Responses-API shape (`instructions` + +//! `input` array), not Chat Completions. Models are Codex-specific +//! strings (`gpt-5`, `gpt-5-codex`, `gpt-5-codex-mini`). +//! +//! Status: scaffolding + auth.json loader / refresh. Request shaping +//! and SSE parsing land in subsequent commits; `complete()` still +//! returns an "implementation pending" error so the rest of the +//! workspace builds. + +use anyhow::{anyhow, Context as _, Result}; +use async_trait::async_trait; +use eventsource_stream::Eventsource; +use futures::StreamExt; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use std::path::PathBuf; +use std::sync::Arc; +use tokio::sync::Mutex; + +use super::{ + Content, LlmProvider, Message, ProviderTurn, Role, StopReason, TextSink, ToolCall, + ToolSpec, Usage, +}; + +/// Base URL for the OAuth-backed Codex backend. Not the public +/// platform `api.openai.com`. +const CODEX_RESPONSES_URL: &str = "https://chatgpt.com/backend-api/codex/responses"; + +/// Originator identifier that the codex CLI sends; Cloudflare uses +/// this + the User-Agent + cookies to classify CLI traffic. Don't +/// rebrand it — anonymous third-party identifiers get challenged. +const CODEX_ORIGINATOR: &str = "codex_cli_rs"; + +/// Required `OpenAI-Beta` header value (from openai/codex source). +const CODEX_OPENAI_BETA: &str = "responses=experimental"; + +/// Optional env var that overrides the `instructions` field. When +/// set, takes precedence over the agent's `system` prompt — useful +/// when an operator wants to mirror Codex CLI exactly by pasting the +/// upstream `gpt_5_codex_prompt.md`. When unset, we fall back to the +/// `system` arg the agent loop already builds (same shape every +/// other provider uses). The Codex backend rejects empty +/// `instructions`, so the final value must be non-empty either way. +const INSTRUCTIONS_ENV: &str = "AGENTD_ZARVIS_CODEX_INSTRUCTIONS"; + +/// Refresh tokens are good for ~30 days but the server-side window can +/// be tighter under load. We refresh when the access_token is within +/// this much of its expected lifetime to avoid edge races against +/// in-flight requests. Codex CLI uses a similar bias. +const ACCESS_TOKEN_REFRESH_LEEWAY_SECS: i64 = 5 * 60; + +/// OAuth `client_id` Codex CLI uses against `auth.openai.com`. Lifted +/// from `openai/codex` (`codex-rs/login/src/auth/manager.rs` — +/// `client_id = "app_EMoamEEZ73f0CkXaXp7hrann"`). Public per the +/// source; safe to embed. +const OPENAI_CLIENT_ID: &str = "app_EMoamEEZ73f0CkXaXp7hrann"; + +/// Token endpoint that mints/refreshes ChatGPT-subscription OAuth +/// tokens. Same host the `codex login` flow points to. +const OPENAI_TOKEN_URL: &str = "https://auth.openai.com/oauth/token"; + +/// On-disk shape of `~/.codex/auth.json`. Only the fields we read. +/// Every field is optional / defaulted so a slight schema drift in +/// future codex builds doesn't break us. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct AuthDotJson { + /// Some installations have this; we just preserve it on write. + #[serde( + rename = "OPENAI_API_KEY", + default, + skip_serializing_if = "Option::is_none" + )] + pub openai_api_key: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tokens: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_refresh: Option, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct Tokens { + #[serde(default, skip_serializing_if = "String::is_empty")] + pub id_token: String, + #[serde(default, skip_serializing_if = "String::is_empty")] + pub access_token: String, + #[serde(default, skip_serializing_if = "String::is_empty")] + pub refresh_token: String, + /// Codex stores this directly in newer builds; older ones leave it + /// blank and clients fall back to parsing it out of the `id_token` + /// JWT. JWT-parsing fallback is a follow-up. + #[serde(default, skip_serializing_if = "String::is_empty")] + pub account_id: String, +} + +/// Returns the path to `auth.json`. Honors `$CODEX_HOME` first, then +/// falls back to `$HOME/.codex/auth.json`. Mirrors what +/// `agentd-adapter-codex` already does to find rollouts so the two +/// crates agree on where codex stores its credential file. +pub fn auth_json_path() -> Result { + if let Ok(home) = std::env::var("CODEX_HOME") { + if !home.is_empty() { + return Ok(PathBuf::from(home).join("auth.json")); + } + } + let home = std::env::var("HOME") + .map_err(|_| anyhow!("$HOME is not set; cannot locate ~/.codex/auth.json"))?; + Ok(PathBuf::from(home).join(".codex").join("auth.json")) +} + +/// Read + parse `auth.json`. Returns an error if the file doesn't +/// exist, can't be read, or doesn't carry an OAuth `access_token` +/// (i.e. the user is in API-key mode, not the subscription mode this +/// provider serves). +pub fn load_auth_json(path: &std::path::Path) -> Result { + let bytes = std::fs::read(path) + .with_context(|| format!("read {}", path.display()))?; + let auth: AuthDotJson = serde_json::from_slice(&bytes) + .with_context(|| format!("parse {} as JSON", path.display()))?; + let tokens = auth.tokens.as_ref().ok_or_else(|| { + anyhow!( + "{} has no `tokens` field. Run `codex login` to authenticate \ + via your ChatGPT subscription before using the codex-oauth \ + provider.", + path.display() + ) + })?; + if tokens.access_token.is_empty() || tokens.refresh_token.is_empty() { + return Err(anyhow!( + "{} has empty access_token or refresh_token. Re-run \ + `codex login`.", + path.display() + )); + } + Ok(auth) +} + +/// Write `auth.json` atomically: write to a sibling `.tmp` file, fsync +/// it, then rename over the original. Without this, a crash between +/// truncate and write leaves the file empty and the user is logged +/// out. The same dance Codex CLI does in +/// `codex-rs/login/src/auth/manager.rs`. +pub fn save_auth_json_atomic( + path: &std::path::Path, + auth: &AuthDotJson, +) -> Result<()> { + use std::io::Write as _; + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent) + .with_context(|| format!("mkdir -p {}", parent.display()))?; + } + let tmp_path = path.with_extension("json.tmp"); + let json = serde_json::to_vec_pretty(auth).context("serialize auth.json")?; + { + let mut f = std::fs::OpenOptions::new() + .write(true) + .create(true) + .truncate(true) + .open(&tmp_path) + .with_context(|| format!("open {}", tmp_path.display()))?; + f.write_all(&json) + .with_context(|| format!("write {}", tmp_path.display()))?; + f.sync_all() + .with_context(|| format!("fsync {}", tmp_path.display()))?; + } + std::fs::rename(&tmp_path, path).with_context(|| { + format!("rename {} -> {}", tmp_path.display(), path.display()) + })?; + Ok(()) +} + +/// In-process refresh coordination. Multiple zarvis sessions in the +/// same adapter process share one of these so only one of them at a +/// time performs the rotation-prone refresh. Cross-process (multiple +/// adapter instances) races remain possible; documented as a known +/// limitation since adding a file lock has its own concurrency edges +/// and the refresh window is narrow. +struct AuthState { + path: PathBuf, + auth: AuthDotJson, +} + +pub struct CodexOauth { + state: Arc>, + http: reqwest::Client, +} + +impl CodexOauth { + /// Construct from on-disk credentials. Fails with a clear message + /// when `auth.json` is missing, in API-key mode, or unreadable. + pub fn from_env() -> Result { + let path = auth_json_path()?; + let auth = load_auth_json(&path)?; + let http = reqwest::Client::builder() + // Cloudflare in front of chatgpt.com rejects the default + // `reqwest/` User-Agent on the codex backend path. + // Use the same identity codex CLI uses so we look like a + // CLI client, not an anonymous Rust HTTP library. + .user_agent(format!( + "codex_cli_rs/{} (agentd zarvis)", + env!("CARGO_PKG_VERSION") + )) + // NOTE: a cookie jar would let cf_clearance challenge + // cookies stick across the auth-refresh ↔ responses-call + // sequence (matching codex CLI). Reqwest's `cookie_store` + // feature isn't enabled workspace-wide. If we hit + // Cloudflare challenges in practice, enable the + // `cookies` feature on reqwest in the workspace and add + // `.cookie_store(true)` here. + .build() + .context("build reqwest client")?; + Ok(Self { + state: Arc::new(Mutex::new(AuthState { path, auth })), + http, + }) + } + + /// Force a token refresh and persist the rotated refresh_token. + /// Caller is responsible for holding the auth-state mutex; this + /// fn assumes serialized callers (the LlmProvider impl uses the + /// mutex around the entire refresh-then-request sequence). + /// + /// Returns an actionable error when the refresh_token has been + /// invalidated server-side — the only fix is for the user to + /// re-run `codex login`. + async fn refresh_locked(&self, state: &mut AuthState) -> Result<()> { + let Some(tokens) = state.auth.tokens.as_ref() else { + return Err(anyhow!( + "auth.json lost its tokens between load and refresh; \ + re-run `codex login`" + )); + }; + if tokens.refresh_token.is_empty() { + return Err(anyhow!( + "auth.json has no refresh_token; re-run `codex login`" + )); + } + let body = serde_json::json!({ + "client_id": OPENAI_CLIENT_ID, + "grant_type": "refresh_token", + "refresh_token": tokens.refresh_token, + "scope": "openid profile email", + }); + let resp = self + .http + .post(OPENAI_TOKEN_URL) + .header("Content-Type", "application/json") + .json(&body) + .send() + .await + .context("POST auth.openai.com/oauth/token")?; + let status = resp.status(); + let bytes = resp.bytes().await.unwrap_or_default(); + if !status.is_success() { + let body_txt = String::from_utf8_lossy(&bytes).to_string(); + // `refresh_token_reused` / `refresh_token_invalidated` from + // auth.openai.com is fatal — the only recovery is + // `codex login` again. Surface it as such instead of + // letting it look like a transient network failure. + if body_txt.contains("refresh_token_reused") + || body_txt.contains("refresh_token_invalidated") + { + return Err(anyhow!( + "ChatGPT OAuth refresh token has been invalidated \ + ({status}). Re-run `codex login` to mint fresh \ + credentials. Server body: {body_txt}" + )); + } + return Err(anyhow!( + "ChatGPT OAuth refresh failed ({status}): {body_txt}" + )); + } + #[derive(Deserialize)] + struct RefreshResp { + #[serde(default)] + id_token: Option, + #[serde(default)] + access_token: Option, + #[serde(default)] + refresh_token: Option, + } + let r: RefreshResp = serde_json::from_slice(&bytes) + .context("parse refresh response from auth.openai.com")?; + // Mutate in place; persist atomically. Codex's `persist_tokens` + // semantics — only update fields the server returned. + let tokens = state.auth.tokens.get_or_insert_with(Tokens::default); + if let Some(t) = r.id_token { + tokens.id_token = t; + } + if let Some(t) = r.access_token { + tokens.access_token = t; + } + if let Some(t) = r.refresh_token { + tokens.refresh_token = t; + } + state.auth.last_refresh = Some(chrono::Utc::now().to_rfc3339()); + save_auth_json_atomic(&state.path, &state.auth)?; + Ok(()) + } + + /// Decide if a refresh is needed right now. v1 heuristic: refresh + /// if `last_refresh` is missing OR older than (24h - leeway). We + /// don't introspect the JWT exp claim yet because decoding JWTs + /// without a verification key just to read `exp` adds a dep, and + /// `last_refresh` is enough for the common case. Refining this is + /// a follow-up. + fn needs_refresh(auth: &AuthDotJson) -> bool { + let Some(last) = auth.last_refresh.as_deref() else { + return true; + }; + let Ok(last) = chrono::DateTime::parse_from_rfc3339(last) else { + return true; + }; + let elapsed = chrono::Utc::now().signed_duration_since(last); + elapsed.num_seconds() + ACCESS_TOKEN_REFRESH_LEEWAY_SECS + >= 24 * 60 * 60 + } +} + +/// Resolve the `instructions` field for the Codex backend. Order: +/// +/// 1. `AGENTD_ZARVIS_CODEX_INSTRUCTIONS` env var, if set and +/// non-empty — explicit operator override (e.g. to mirror Codex +/// CLI exactly with the upstream `gpt_5_codex_prompt.md`). +/// 2. The `system` argument the agent loop passes — same prompt +/// every other provider (openai / anthropic / ollama) uses for +/// the equivalent slot. This is the default, and the same +/// out-of-box experience the other providers give. +/// +/// Errors only if BOTH are empty — the Codex backend rejects an +/// empty `instructions` field with `400`, so failing fast here gives +/// a more actionable message than letting the request go out and +/// get refused. +fn resolve_instructions(system: &str) -> Result { + if let Ok(raw) = std::env::var(INSTRUCTIONS_ENV) { + if !raw.trim().is_empty() { + return Ok(raw); + } + } + if !system.trim().is_empty() { + return Ok(system.to_string()); + } + Err(anyhow!( + "codex-oauth: no `instructions` available. Either set \ + {INSTRUCTIONS_ENV} to a non-empty value, or run zarvis with \ + a non-empty `system` prompt (the agent loop populates this \ + from its system-prompt builder). Empty / unset → Codex \ + backend will reject the request with a 400." + )) +} + +/// Convert one of our `Message`s into Responses-API "input items". +/// One Message can produce multiple items: the AssistantToolCalls +/// variant fans out into one `message` item (for the prose, if any) +/// plus one `function_call` item per tool call. The ToolResult +/// variant produces a `function_call_output` item. +fn message_to_input_items(m: &Message) -> Vec { + match &m.content { + Content::Text { text } => { + let role = match m.role { + Role::System => "system", + Role::User => "user", + Role::Assistant => "assistant", + // Tool-role plain text shouldn't really happen (tool + // results use the dedicated variant), but be + // defensive — surface it as a `user` message rather + // than dropping it. + Role::Tool => "user", + }; + let typ = if matches!(m.role, Role::Assistant) { + "output_text" + } else { + "input_text" + }; + vec![json!({ + "type": "message", + "role": role, + "content": [{ "type": typ, "text": text }], + })] + } + Content::AssistantToolCalls { text, calls } => { + let mut out = Vec::with_capacity(calls.len() + 1); + if let Some(t) = text.as_deref().filter(|t| !t.is_empty()) { + out.push(json!({ + "type": "message", + "role": "assistant", + "content": [{ "type": "output_text", "text": t }], + })); + } + for c in calls { + out.push(json!({ + "type": "function_call", + "call_id": c.id, + "name": c.name, + "arguments": serde_json::to_string(&c.input) + .unwrap_or_else(|_| "{}".into()), + })); + } + out + } + Content::ToolResult { + call_id, + output, + is_error: _, + } => { + // Responses API doesn't have a dedicated error flag on + // function_call_output — error text just rides along in + // `output`. The model picks up on it from content. + vec![json!({ + "type": "function_call_output", + "call_id": call_id, + "output": output, + })] + } + } +} + +/// Convert a `ToolSpec` to Responses-API tool definition shape. +/// Responses uses a flatter shape than chat-completions: no +/// `{ "type": "function", "function": {...} }` wrapping. +fn tool_spec_to_value(t: &ToolSpec) -> Value { + json!({ + "type": "function", + "name": t.name, + "description": t.description, + "parameters": t.schema.clone(), + }) +} + +/// Build the request body for `POST /backend-api/codex/responses`. +/// Public for unit testing — keeps the shape pinned independently +/// of the live transport. +pub fn build_responses_body( + model: &str, + instructions: &str, + messages: &[Message], + tools: &[ToolSpec], +) -> Value { + let input: Vec = messages.iter().flat_map(message_to_input_items).collect(); + let mut body = json!({ + "model": model, + "instructions": instructions, + "input": input, + "stream": true, + // ChatGPT-account backend REQUIRES `store: false`. The + // public Responses API defaults to true (server-side + // conversation storage), but `chatgpt.com/backend-api/codex` + // refuses requests where store != false with + // `400 "Store must be set to false"`. Codex CLI sets this + // explicitly in `codex-rs/core/src/client.rs`. + "store": false, + // Asks the server to emit reasoning summary deltas; we don't + // surface them to the user today but it costs nothing to + // request and the SSE parser handles them gracefully. + "reasoning": { "summary": "auto" }, + }); + if !tools.is_empty() { + body["tools"] = Value::Array(tools.iter().map(tool_spec_to_value).collect()); + body["parallel_tool_calls"] = Value::Bool(true); + } + body +} + +/// Tool-call accumulator for the SSE stream. The Responses API +/// streams function_call items as item-added + arguments-delta +/// events; we collect them keyed by `item_id` here and flush when +/// `output_item.done` fires for that id. +#[derive(Default, Debug)] +struct FnCallAcc { + call_id: String, + name: String, + args: String, +} + +#[async_trait] +impl LlmProvider for CodexOauth { + fn name(&self) -> &str { + "codex-oauth" + } + + async fn complete( + &self, + model: &str, + system: &str, + messages: &[Message], + tools: &[ToolSpec], + sink: &mut dyn TextSink, + ) -> Result { + // Codex's Responses API uses `instructions` instead of an + // inline system message. We pass the agent's `system` arg + // through as-is, matching what every other provider does + // for the equivalent slot. `AGENTD_ZARVIS_CODEX_INSTRUCTIONS` + // is an optional operator override (handy for mirroring + // Codex CLI exactly with the upstream prompt). + let instructions = resolve_instructions(system)?; + let body = build_responses_body(model, &instructions, messages, tools); + + // Take the auth lock for the duration of the request so + // concurrent zarvis turns don't race on token rotation. + let mut state = self.state.lock().await; + if Self::needs_refresh(&state.auth) { + self.refresh_locked(&mut state).await?; + } + let (access_token, account_id) = { + let tokens = state + .auth + .tokens + .as_ref() + .ok_or_else(|| anyhow!("auth.json tokens missing after refresh"))?; + (tokens.access_token.clone(), tokens.account_id.clone()) + }; + // Release the lock for the duration of the HTTP call — we + // don't need exclusive access to the auth state while the + // request streams, and holding the lock would serialize all + // codex-oauth turns within a process. + drop(state); + + let mut req = self + .http + .post(CODEX_RESPONSES_URL) + .bearer_auth(&access_token) + .header("OpenAI-Beta", CODEX_OPENAI_BETA) + .header("originator", CODEX_ORIGINATOR) + .header("Content-Type", "application/json") + .header("Accept", "text/event-stream") + .json(&body); + if !account_id.is_empty() { + req = req.header("ChatGPT-Account-ID", account_id); + } + let resp = req + .send() + .await + .context("POST chatgpt.com/backend-api/codex/responses")?; + let status = resp.status(); + if !status.is_success() { + let raw = resp.text().await.unwrap_or_default(); + if status.as_u16() == 400 { + if let Some(extracted) = super::parse_overflow(&raw) { + return Err(anyhow::Error::new(super::ContextOverflow { + extracted, + raw, + })); + } + } + // Auth failures on this endpoint usually mean the access + // token expired between our refresh check and now (or + // the server invalidated it). Surface that distinctly so + // the agent loop can stop retrying — re-running `codex + // login` is the only fix. + if status.as_u16() == 401 { + return Err(anyhow!( + "codex-oauth: 401 from chatgpt.com (access token \ + rejected). Try re-running `codex login`. Body: {raw}" + )); + } + return Err(anyhow!("codex-oauth: {status} from chatgpt.com: {raw}")); + } + + let mut stream = resp.bytes_stream().eventsource(); + let mut assistant_text = String::new(); + let mut fn_calls: std::collections::HashMap = + std::collections::HashMap::new(); + let mut fn_call_order: Vec = Vec::new(); + let mut stop_reason = StopReason::EndTurn; + let mut usage = Usage::default(); + + while let Some(ev) = stream.next().await { + let ev = ev.context("codex-oauth SSE stream")?; + // The Codex Responses transport uses `event:` SSE names; + // `data:` carries the JSON body. eventsource_stream + // surfaces both as fields on the event. + let kind = ev.event.as_str(); + let chunk: Value = match serde_json::from_str(&ev.data) { + Ok(v) => v, + Err(_) => continue, + }; + match kind { + // Streaming assistant text. + "response.output_text.delta" => { + if let Some(text) = chunk.get("delta").and_then(|v| v.as_str()) { + if !text.is_empty() { + sink.delta(text); + assistant_text.push_str(text); + } + } + } + // A new item appeared. We care about function_call + // items — record the id/name so subsequent + // arguments-delta events can find the accumulator. + "response.output_item.added" => { + let Some(item) = chunk.get("item") else { continue }; + if item.get("type").and_then(|v| v.as_str()) != Some("function_call") { + continue; + } + let item_id = item + .get("id") + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(); + let call_id = item + .get("call_id") + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(); + let name = item + .get("name") + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(); + if !item_id.is_empty() { + fn_call_order.push(item_id.clone()); + fn_calls.insert( + item_id, + FnCallAcc { + call_id, + name, + args: String::new(), + }, + ); + } + } + // Tool-call arguments stream a piece at a time. + "response.function_call_arguments.delta" => { + let item_id = chunk + .get("item_id") + .and_then(|v| v.as_str()) + .unwrap_or_default(); + if let Some(acc) = fn_calls.get_mut(item_id) { + if let Some(delta) = + chunk.get("delta").and_then(|v| v.as_str()) + { + acc.args.push_str(delta); + } + } + } + // Final arguments value — some servers send this in + // addition to the deltas. Be defensive: overwrite if + // we got it, since `delta`s + final is a known + // shape. + "response.function_call_arguments.done" => { + let item_id = chunk + .get("item_id") + .and_then(|v| v.as_str()) + .unwrap_or_default(); + if let Some(acc) = fn_calls.get_mut(item_id) { + if let Some(args) = + chunk.get("arguments").and_then(|v| v.as_str()) + { + if !args.is_empty() { + acc.args = args.to_string(); + } + } + } + } + // End of stream. Pull usage + stop reason if the + // server reported them. + "response.completed" | "response.incomplete" | "response.failed" => { + if let Some(u) = chunk.pointer("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/response/usage") { + usage.input_tokens = u + .get("input_tokens") + .and_then(|n| n.as_u64()) + .unwrap_or(usage.input_tokens); + usage.output_tokens = u + .get("output_tokens") + .and_then(|n| n.as_u64()) + .unwrap_or(usage.output_tokens); + } + if let Some(reason) = chunk + .pointer("/response/incomplete_details/reason") + .and_then(|v| v.as_str()) + { + if reason == "max_output_tokens" { + stop_reason = StopReason::MaxTokens; + } + } + break; + } + _ => { + // Other events (reasoning summaries, status, + // etc.) are not surfaced to the agent in v1. + } + } + } + + // If the model emitted tool calls, that's the turn's stop + // reason regardless of what `response.completed` said. + let calls: Vec = fn_call_order + .iter() + .filter_map(|id| fn_calls.remove(id)) + .filter(|a| !a.name.is_empty()) + .map(|a| { + let input = if a.args.is_empty() { + json!({}) + } else { + serde_json::from_str::(&a.args).unwrap_or_else(|_| json!({})) + }; + ToolCall { + id: if a.call_id.is_empty() { + format!("tool_{}", short_hash(&a.name)) + } else { + a.call_id + }, + name: a.name, + input, + } + }) + .collect(); + if !calls.is_empty() { + stop_reason = StopReason::ToolUse; + } + + Ok(ProviderTurn { + text: if assistant_text.is_empty() { + None + } else { + Some(assistant_text) + }, + tool_calls: calls, + stop_reason, + usage, + }) + } +} + +fn short_hash(s: &str) -> String { + use std::collections::hash_map::DefaultHasher; + use std::hash::{Hash, Hasher}; + let mut h = DefaultHasher::new(); + s.hash(&mut h); + format!("{:x}", h.finish())[..8].to_string() +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Reading a well-formed auth.json with OAuth tokens succeeds. + #[test] + fn load_auth_json_round_trips_through_tmp_file() { + let tmp = std::env::temp_dir().join(format!( + "agentd-codex-oauth-load-{}.json", + std::process::id() + )); + let _ = std::fs::remove_file(&tmp); + std::fs::write( + &tmp, + r#"{ + "OPENAI_API_KEY": "sk-existing-platform-key", + "tokens": { + "id_token": "JWT.eyJh.SIG", + "access_token": "access-abc", + "refresh_token": "refresh-xyz", + "account_id": "acct_123" + }, + "last_refresh": "2026-05-18T00:00:00Z" + }"#, + ) + .unwrap(); + + let auth = load_auth_json(&tmp).expect("load"); + let tokens = auth.tokens.as_ref().expect("tokens present"); + assert_eq!(tokens.access_token, "access-abc"); + assert_eq!(tokens.refresh_token, "refresh-xyz"); + assert_eq!(tokens.account_id, "acct_123"); + assert_eq!( + auth.openai_api_key.as_deref(), + Some("sk-existing-platform-key") + ); + + let _ = std::fs::remove_file(&tmp); + } + + /// Missing file → clear actionable error mentioning the path. + #[test] + fn load_auth_json_missing_file_errors_with_path() { + let bogus = std::env::temp_dir().join(format!( + "agentd-codex-oauth-missing-{}", + std::process::id() + )); + let _ = std::fs::remove_file(&bogus); + let err = load_auth_json(&bogus).unwrap_err(); + let msg = format!("{err:#}"); + assert!(msg.contains(bogus.to_string_lossy().as_ref()), "{msg}"); + } + + /// `tokens` field absent → message points the user at + /// `codex login`. This is the API-key-only mode of `auth.json`. + #[test] + fn load_auth_json_without_tokens_errors_with_codex_login_hint() { + let tmp = std::env::temp_dir().join(format!( + "agentd-codex-oauth-noauth-{}.json", + std::process::id() + )); + std::fs::write(&tmp, r#"{ "OPENAI_API_KEY": "sk-platform-only" }"#) + .unwrap(); + let err = load_auth_json(&tmp).unwrap_err(); + let msg = format!("{err:#}"); + assert!(msg.contains("codex login"), "{msg}"); + let _ = std::fs::remove_file(&tmp); + } + + /// Empty access/refresh tokens are rejected loudly — having a + /// `tokens` block with blanks would otherwise silently flow into + /// a refresh attempt that fails with a confusing 4xx. + #[test] + fn load_auth_json_with_blank_tokens_errors() { + let tmp = std::env::temp_dir().join(format!( + "agentd-codex-oauth-blank-{}.json", + std::process::id() + )); + std::fs::write( + &tmp, + r#"{ "tokens": { "id_token": "x", "access_token": "", "refresh_token": "" } }"#, + ) + .unwrap(); + let err = load_auth_json(&tmp).unwrap_err(); + let msg = format!("{err:#}"); + assert!(msg.contains("codex login"), "{msg}"); + let _ = std::fs::remove_file(&tmp); + } + + /// Atomic write succeeds and the resulting file parses back to + /// the same fields. Doesn't (yet) crash-test the rename window — + /// that's an integration concern with a fault-injecting FS. + #[test] + fn save_auth_json_atomic_round_trips() { + let dir = std::env::temp_dir().join(format!( + "agentd-codex-oauth-save-{}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let target = dir.join("auth.json"); + + let auth = AuthDotJson { + openai_api_key: Some("sk-keep".into()), + tokens: Some(Tokens { + id_token: "ID".into(), + access_token: "A1".into(), + refresh_token: "R1".into(), + account_id: "acct".into(), + }), + last_refresh: Some("2026-05-18T01:02:03Z".into()), + }; + save_auth_json_atomic(&target, &auth).expect("save"); + + // The temp file we used during the dance must be gone. + assert!(!target.with_extension("json.tmp").exists()); + + let loaded = load_auth_json(&target).expect("load back"); + let t = loaded.tokens.unwrap(); + assert_eq!(t.access_token, "A1"); + assert_eq!(t.refresh_token, "R1"); + assert_eq!( + loaded.last_refresh.as_deref(), + Some("2026-05-18T01:02:03Z") + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + fn user(text: &str) -> Message { + Message { + role: Role::User, + content: Content::Text { + text: text.into(), + }, + } + } + + fn assistant(text: &str) -> Message { + Message { + role: Role::Assistant, + content: Content::Text { + text: text.into(), + }, + } + } + + /// Smoke test: plain user→assistant exchange maps to the + /// Responses-API `input` shape with `message` items, correct + /// roles, and `input_text` / `output_text` content kinds. + #[test] + fn build_body_emits_responses_message_shape() { + let body = build_responses_body( + "gpt-5-codex", + "system-prompt-here", + &[user("hi"), assistant("hello back")], + &[], + ); + assert_eq!(body["model"], "gpt-5-codex"); + assert_eq!(body["instructions"], "system-prompt-here"); + assert_eq!(body["stream"], true); + // ChatGPT-account backend rejects requests with `store: true` + // (the public-API default). Locking this in here so the + // field never silently gets dropped — re-adding it would + // produce a 400 "Store must be set to false" that's + // confusing without context. + assert_eq!(body["store"], false); + let input = body["input"].as_array().expect("input array"); + assert_eq!(input.len(), 2); + assert_eq!(input[0]["role"], "user"); + assert_eq!(input[0]["type"], "message"); + assert_eq!(input[0]["content"][0]["type"], "input_text"); + assert_eq!(input[0]["content"][0]["text"], "hi"); + assert_eq!(input[1]["role"], "assistant"); + assert_eq!(input[1]["content"][0]["type"], "output_text"); + assert_eq!(input[1]["content"][0]["text"], "hello back"); + // No tools were passed; the field should be absent rather + // than emitting `tools: []` (server-side has historically + // disliked empty arrays here). + assert!(body.get("tools").is_none()); + assert!(body.get("parallel_tool_calls").is_none()); + } + + /// Tools are emitted in the flatter Responses shape (no + /// `function` wrapper) and `parallel_tool_calls: true` is set. + #[test] + fn build_body_emits_responses_tool_shape() { + let tools = vec![ToolSpec { + name: "shell".into(), + description: "run a shell command".into(), + schema: json!({ + "type": "object", + "properties": { "cmd": { "type": "string" } }, + "required": ["cmd"], + }), + }]; + let body = build_responses_body("gpt-5-codex", "sys", &[user("ls")], &tools); + let tools_v = body["tools"].as_array().expect("tools present"); + assert_eq!(tools_v.len(), 1); + assert_eq!(tools_v[0]["type"], "function"); + assert_eq!(tools_v[0]["name"], "shell"); + assert_eq!(tools_v[0]["description"], "run a shell command"); + assert!(tools_v[0]["parameters"]["properties"]["cmd"].is_object()); + assert_eq!(body["parallel_tool_calls"], true); + } + + /// An AssistantToolCalls message fans out into one `message` + /// item (for the pre-tool prose, if any) plus one `function_call` + /// item per call. Tool results map to `function_call_output`. + /// This is the long pole on round-trip correctness: getting it + /// wrong means the model can't see its own prior tool calls and + /// goes into a re-call loop. + #[test] + fn build_body_round_trips_tool_call_history() { + let msgs = vec![ + user("list files"), + Message { + role: Role::Assistant, + content: Content::AssistantToolCalls { + text: Some("calling shell".into()), + calls: vec![ToolCall { + id: "call_abc".into(), + name: "shell".into(), + input: json!({"cmd": "ls"}), + }], + }, + }, + Message { + role: Role::Tool, + content: Content::ToolResult { + call_id: "call_abc".into(), + output: "Cargo.toml\nsrc".into(), + is_error: false, + }, + }, + ]; + let body = build_responses_body("gpt-5-codex", "sys", &msgs, &[]); + let input = body["input"].as_array().unwrap(); + assert_eq!(input.len(), 4, "user + assistant-prose + fn_call + fn_output"); + assert_eq!(input[0]["role"], "user"); + // Assistant prose item. + assert_eq!(input[1]["type"], "message"); + assert_eq!(input[1]["role"], "assistant"); + assert_eq!(input[1]["content"][0]["text"], "calling shell"); + // function_call item. + assert_eq!(input[2]["type"], "function_call"); + assert_eq!(input[2]["call_id"], "call_abc"); + assert_eq!(input[2]["name"], "shell"); + // Arguments must be a JSON-encoded STRING, not an object — + // that's the Responses API contract. + let args_str = input[2]["arguments"].as_str().unwrap(); + let parsed: Value = serde_json::from_str(args_str).unwrap(); + assert_eq!(parsed["cmd"], "ls"); + // function_call_output item. + assert_eq!(input[3]["type"], "function_call_output"); + assert_eq!(input[3]["call_id"], "call_abc"); + assert_eq!(input[3]["output"], "Cargo.toml\nsrc"); + } + + /// An AssistantToolCalls with empty `text` does NOT emit an + /// empty-text message — the Responses API treats `output_text: + /// ""` as a malformed item. + #[test] + fn build_body_skips_empty_assistant_prose_before_tool_call() { + let msgs = vec![Message { + role: Role::Assistant, + content: Content::AssistantToolCalls { + text: None, + calls: vec![ToolCall { + id: "x".into(), + name: "shell".into(), + input: json!({}), + }], + }, + }]; + let body = build_responses_body("gpt-5-codex", "sys", &msgs, &[]); + let input = body["input"].as_array().unwrap(); + assert_eq!(input.len(), 1); + assert_eq!(input[0]["type"], "function_call"); + } + + /// `instructions` resolution: env var wins when set, + /// `system` arg is the fallback, both empty errors out. Uses a + /// scoped env-guard to keep the global env clean for other + /// tests running in parallel. + #[test] + fn resolve_instructions_prefers_env_then_system() { + // Save/restore the env var around the test body so parallel + // tests don't see our state. + struct EnvGuard(Option); + impl Drop for EnvGuard { + fn drop(&mut self) { + match &self.0 { + Some(v) => std::env::set_var(INSTRUCTIONS_ENV, v), + None => std::env::remove_var(INSTRUCTIONS_ENV), + } + } + } + let _g = EnvGuard(std::env::var(INSTRUCTIONS_ENV).ok()); + + // env wins when set & non-empty. + std::env::set_var(INSTRUCTIONS_ENV, "from-env"); + assert_eq!(resolve_instructions("from-system").unwrap(), "from-env"); + + // env blank → falls back to system. + std::env::set_var(INSTRUCTIONS_ENV, " "); + assert_eq!(resolve_instructions("from-system").unwrap(), "from-system"); + + // env unset → system is used. + std::env::remove_var(INSTRUCTIONS_ENV); + assert_eq!(resolve_instructions("from-system").unwrap(), "from-system"); + + // Both empty → actionable error mentioning the env var. + let err = resolve_instructions("").unwrap_err(); + let msg = format!("{err:#}"); + assert!(msg.contains(INSTRUCTIONS_ENV), "{msg}"); + } + + /// `needs_refresh` semantics: no `last_refresh` → refresh now; + /// recent `last_refresh` → don't; >24h ago → refresh. + #[test] + fn needs_refresh_uses_last_refresh_age() { + let mut auth = AuthDotJson::default(); + // No timestamp → refresh. + assert!(CodexOauth::needs_refresh(&auth)); + // Recent timestamp → no refresh. + auth.last_refresh = Some(chrono::Utc::now().to_rfc3339()); + assert!(!CodexOauth::needs_refresh(&auth)); + // Old timestamp → refresh. + auth.last_refresh = + Some((chrono::Utc::now() - chrono::Duration::days(2)).to_rfc3339()); + assert!(CodexOauth::needs_refresh(&auth)); + // Malformed timestamp → conservatively refresh. + auth.last_refresh = Some("not-a-date".to_string()); + assert!(CodexOauth::needs_refresh(&auth)); + } +} diff --git a/crates/adapter-zarvis/src/provider/mod.rs b/crates/adapter-zarvis/src/provider/mod.rs index cd539581..b99205a6 100644 --- a/crates/adapter-zarvis/src/provider/mod.rs +++ b/crates/adapter-zarvis/src/provider/mod.rs @@ -8,6 +8,7 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; pub mod anthropic; +pub mod codex_oauth; pub mod ollama; pub mod openai; pub mod routing; diff --git a/crates/adapter-zarvis/src/provider/routing.rs b/crates/adapter-zarvis/src/provider/routing.rs index d8e5f0aa..a0bc0e81 100644 --- a/crates/adapter-zarvis/src/provider/routing.rs +++ b/crates/adapter-zarvis/src/provider/routing.rs @@ -1,18 +1,27 @@ //! Translate a model spec string into a (provider, bare model name). //! -//! Explicit prefixes (`openai:`, `anthropic:`, `ollama:`) always win. -//! Otherwise we sniff the bare name: +//! Explicit prefixes (`openai:`, `anthropic:`, `ollama:`, `codex-oauth:`) +//! always win. Otherwise we sniff the bare name: //! - starts with `gpt-` or `o[1-5]` → OpenAI //! - starts with `claude-` → Anthropic //! - anything else → Ollama (local fallback) //! //! Returning an enum keeps the dispatch table small and testable. +//! +//! `codex-oauth:` is the subscription-billed Codex backend +//! (`chatgpt.com/backend-api/codex/responses`), distinct from `openai:` +//! which hits the public platform API at `api.openai.com`. There is no +//! bare-name fallback for it — users must opt in explicitly because the +//! billing path is different. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Provider { OpenAI, Anthropic, Ollama, + /// OAuth-backed Codex backend; reads `~/.codex/auth.json`, bills + /// against the user's ChatGPT subscription. + CodexOauth, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -44,11 +53,20 @@ pub fn parse_model_spec(s: &str) -> Result { model: rest.to_string(), }); } + if let Some(rest) = s.strip_prefix("codex-oauth:") { + return Ok(ModelSpec { + provider: Provider::CodexOauth, + model: rest.to_string(), + }); + } if let Some(prefix) = s.split(':').next() { // Reject unknown explicit prefixes so typos don't silently fall through. - if s.contains(':') && !matches!(prefix, "openai" | "anthropic" | "ollama") { + if s.contains(':') + && !matches!(prefix, "openai" | "anthropic" | "ollama" | "codex-oauth") + { return Err(format!( - "unknown provider prefix `{prefix}:` (expected one of openai:, anthropic:, ollama:)" + "unknown provider prefix `{prefix}:` (expected one of \ + openai:, anthropic:, ollama:, codex-oauth:)" )); } } @@ -146,4 +164,31 @@ mod tests { assert!(parse_model_spec("").is_err()); assert!(parse_model_spec(" ").is_err()); } + + /// `codex-oauth:` routes to the OAuth-backed Codex backend + /// (`chatgpt.com/backend-api/codex/responses`). Distinct from + /// `openai:` to keep billing paths from getting silently swapped — + /// `openai:gpt-5` stays on platform pay-as-you-go, only + /// `codex-oauth:gpt-5` draws against ChatGPT subscription. + #[test] + fn codex_oauth_prefix_is_recognized() { + let s = parse("codex-oauth:gpt-5-codex"); + assert_eq!(s.provider, Provider::CodexOauth); + assert_eq!(s.model, "gpt-5-codex"); + + let s = parse("codex-oauth:gpt-5"); + assert_eq!(s.provider, Provider::CodexOauth); + assert_eq!(s.model, "gpt-5"); + } + + /// No bare-name fallback for codex-oauth: opting into a different + /// billing path requires explicit prefix, even for the + /// codex-specific model strings. + #[test] + fn gpt_5_codex_bare_does_not_route_to_codex_oauth() { + // The platform OpenAI API doesn't serve `gpt-5-codex`, but the + // router treats it as OpenAI based on the `gpt-` prefix; users + // must type the `codex-oauth:` prefix explicitly. + assert_eq!(parse("gpt-5-codex").provider, Provider::OpenAI); + } } diff --git a/crates/adapter-zarvis/src/title_mode.rs b/crates/adapter-zarvis/src/title_mode.rs index 873d3ab1..19209d90 100644 --- a/crates/adapter-zarvis/src/title_mode.rs +++ b/crates/adapter-zarvis/src/title_mode.rs @@ -50,6 +50,16 @@ fn provider_for(p: Provider) -> Result> { Provider::OpenAI => Box::new(provider::openai::OpenAi::from_env()?), Provider::Anthropic => Box::new(provider::anthropic::Anthropic::from_env()?), Provider::Ollama => Box::new(provider::ollama::Ollama::from_env()?), + // Title generation always uses one of the trio above; the + // user never picks `codex-oauth:` for title-gen since the + // selection comes from `pick_default_spec_str` which only + // emits openai/anthropic/ollama. Bail loudly if we ever get + // here so the contract doesn't drift silently. + Provider::CodexOauth => { + return Err(anyhow!( + "title-gen does not support codex-oauth provider" + )); + } }) }