From b3ae531b3a6c4f9742dff6a7bd1b14796bde9c53 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Tue, 14 Apr 2026 17:00:18 +0100 Subject: [PATCH] feat: codex sampler (#17784) Add a pure sampler using the Codex auth and model config. To be used by other binary such as tape recorder --- codex-rs/Cargo.lock | 1 + codex-rs/cli/Cargo.toml | 1 + codex-rs/cli/src/main.rs | 24 ++++ codex-rs/cli/src/responses_cmd.rs | 219 ++++++++++++++++++++++++++++++ 4 files changed, 245 insertions(+) create mode 100644 codex-rs/cli/src/responses_cmd.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index fc5c70e6b..ad0e627d1 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1669,6 +1669,7 @@ dependencies = [ "assert_matches", "clap", "clap_complete", + "codex-api", "codex-app-server", "codex-app-server-protocol", "codex-app-server-test-client", diff --git a/codex-rs/cli/Cargo.toml b/codex-rs/cli/Cargo.toml index ab0d86de2..60caf888f 100644 --- a/codex-rs/cli/Cargo.toml +++ b/codex-rs/cli/Cargo.toml @@ -24,6 +24,7 @@ codex-app-server = { workspace = true } codex-app-server-protocol = { workspace = true } codex-app-server-test-client = { workspace = true } codex-arg0 = { workspace = true } +codex-api = { workspace = true } codex-chatgpt = { workspace = true } codex-cloud-tasks = { path = "../cloud-tasks" } codex-utils-cli = { workspace = true } diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index baa6684b4..563790c82 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -40,11 +40,14 @@ mod app_cmd; mod desktop_app; mod marketplace_cmd; mod mcp_cmd; +mod responses_cmd; #[cfg(not(windows))] mod wsl_paths; use crate::marketplace_cmd::MarketplaceCli; use crate::mcp_cmd::McpCli; +use crate::responses_cmd::ResponsesCommand; +use crate::responses_cmd::run_responses_command; use codex_core::config::Config; use codex_core::config::ConfigOverrides; @@ -151,6 +154,10 @@ enum Subcommand { #[clap(hide = true)] ResponsesApiProxy(ResponsesApiProxyArgs), + /// Internal: send one raw Responses API payload through Codex auth. + #[clap(hide = true)] + Responses(ResponsesCommand), + /// Internal: relay stdio to a Unix domain socket. #[clap(hide = true, name = "stdio-to-uds")] StdioToUds(StdioToUdsCommand), @@ -1015,6 +1022,14 @@ async fn cli_main(arg0_paths: Arg0DispatchPaths) -> anyhow::Result<()> { tokio::task::spawn_blocking(move || codex_responses_api_proxy::run_main(args)) .await??; } + Some(Subcommand::Responses(ResponsesCommand {})) => { + reject_remote_mode_for_subcommand( + root_remote.as_deref(), + root_remote_auth_token_env.as_deref(), + "responses", + )?; + run_responses_command(root_config_overrides).await?; + } Some(Subcommand::StdioToUds(cmd)) => { reject_remote_mode_for_subcommand( root_remote.as_deref(), @@ -1666,6 +1681,15 @@ mod tests { ); } + #[test] + fn responses_subcommand_is_hidden_from_help_but_parses() { + let help = MultitoolCli::command().render_help().to_string(); + assert!(!help.contains("responses")); + + let cli = MultitoolCli::try_parse_from(["codex", "responses"]).expect("parse"); + assert!(matches!(cli.subcommand, Some(Subcommand::Responses(_)))); + } + fn sample_exit_info(conversation_id: Option<&str>, thread_name: Option<&str>) -> AppExitInfo { let token_usage = TokenUsage { output_tokens: 2, diff --git a/codex-rs/cli/src/responses_cmd.rs b/codex-rs/cli/src/responses_cmd.rs new file mode 100644 index 000000000..044c7e0b2 --- /dev/null +++ b/codex-rs/cli/src/responses_cmd.rs @@ -0,0 +1,219 @@ +use clap::Parser; +use codex_core::config::Config; +use codex_utils_cli::CliConfigOverrides; +use serde_json::json; +use tokio::io::AsyncReadExt; + +#[derive(Debug, Parser)] +pub(crate) struct ResponsesCommand {} + +pub(crate) async fn run_responses_command( + root_config_overrides: CliConfigOverrides, +) -> anyhow::Result<()> { + let mut payload_text = String::new(); + tokio::io::stdin().read_to_string(&mut payload_text).await?; + if payload_text.trim().is_empty() { + anyhow::bail!("expected Responses API JSON payload on stdin"); + } + + let payload: serde_json::Value = serde_json::from_str(&payload_text) + .map_err(|err| anyhow::anyhow!("failed to parse Responses API JSON payload: {err}"))?; + if payload.get("stream").and_then(serde_json::Value::as_bool) != Some(true) { + anyhow::bail!("codex responses expects a streaming payload with `\"stream\": true`"); + } + + let cli_overrides = root_config_overrides + .parse_overrides() + .map_err(anyhow::Error::msg)?; + let config = Config::load_with_cli_overrides(cli_overrides).await?; + let base_auth_manager = codex_login::AuthManager::shared_from_config( + &config, /*enable_codex_api_key_env*/ true, + ); + let auth_manager = + codex_login::auth_manager_for_provider(Some(base_auth_manager), &config.model_provider); + let auth = match auth_manager { + Some(auth_manager) => auth_manager.auth().await, + None => None, + }; + let api_provider = config + .model_provider + .to_api_provider(auth.as_ref().map(codex_login::CodexAuth::auth_mode))?; + let api_auth = codex_login::auth_provider_from_auth(auth, &config.model_provider)?; + let client = codex_api::ResponsesClient::new( + codex_api::ReqwestTransport::new(codex_login::default_client::build_reqwest_client()), + api_provider, + api_auth, + ); + + let mut stream = client + .stream( + payload, + Default::default(), + codex_api::Compression::None, + /*turn_state*/ None, + ) + .await?; + while let Some(event) = stream.rx_event.recv().await { + let event = event?; + println!("{}", serde_json::to_string(&response_event_to_json(event))?); + } + + Ok(()) +} + +fn response_event_to_json(event: codex_api::ResponseEvent) -> serde_json::Value { + match event { + codex_api::ResponseEvent::Created => { + json!({ "type": "response.created", "response": {} }) + } + codex_api::ResponseEvent::OutputItemDone(item) => { + json!({ "type": "response.output_item.done", "item": item }) + } + codex_api::ResponseEvent::OutputItemAdded(item) => { + json!({ "type": "response.output_item.added", "item": item }) + } + codex_api::ResponseEvent::ServerModel(model) => { + json!({ "type": "response.server_model", "model": model }) + } + codex_api::ResponseEvent::ServerReasoningIncluded(included) => { + json!({ "type": "response.server_reasoning_included", "included": included }) + } + codex_api::ResponseEvent::Completed { + response_id, + token_usage, + } => { + let response = match token_usage { + Some(token_usage) => json!({ + "id": response_id, + "usage": { + "input_tokens": token_usage.input_tokens, + "input_tokens_details": { + "cached_tokens": token_usage.cached_input_tokens, + }, + "output_tokens": token_usage.output_tokens, + "output_tokens_details": { + "reasoning_tokens": token_usage.reasoning_output_tokens, + }, + "total_tokens": token_usage.total_tokens, + }, + }), + None => json!({ "id": response_id }), + }; + json!({ "type": "response.completed", "response": response }) + } + codex_api::ResponseEvent::OutputTextDelta(delta) => { + json!({ "type": "response.output_text.delta", "delta": delta }) + } + codex_api::ResponseEvent::ReasoningSummaryDelta { + delta, + summary_index, + } => json!({ + "type": "response.reasoning_summary_text.delta", + "delta": delta, + "summary_index": summary_index, + }), + codex_api::ResponseEvent::ReasoningContentDelta { + delta, + content_index, + } => json!({ + "type": "response.reasoning_text.delta", + "delta": delta, + "content_index": content_index, + }), + codex_api::ResponseEvent::ReasoningSummaryPartAdded { summary_index } => { + json!({ + "type": "response.reasoning_summary_part.added", + "summary_index": summary_index, + }) + } + codex_api::ResponseEvent::RateLimits(rate_limits) => { + json!({ "type": "response.rate_limits", "rate_limits": rate_limits }) + } + codex_api::ResponseEvent::ModelsEtag(etag) => { + json!({ "type": "response.models_etag", "etag": etag }) + } + } +} + +#[cfg(test)] +mod tests { + use super::response_event_to_json; + use codex_protocol::protocol::TokenUsage; + use pretty_assertions::assert_eq; + use serde_json::json; + + #[test] + fn response_events_keep_replayable_response_envelopes() { + let created = response_event_to_json(codex_api::ResponseEvent::Created); + assert_eq!(created, json!({"type": "response.created", "response": {}})); + + let completed = response_event_to_json(codex_api::ResponseEvent::Completed { + response_id: "resp-1".to_string(), + token_usage: Some(TokenUsage { + input_tokens: 10, + cached_input_tokens: 4, + output_tokens: 7, + reasoning_output_tokens: 3, + total_tokens: 17, + }), + }); + assert_eq!( + completed, + json!({ + "type": "response.completed", + "response": { + "id": "resp-1", + "usage": { + "input_tokens": 10, + "input_tokens_details": { + "cached_tokens": 4, + }, + "output_tokens": 7, + "output_tokens_details": { + "reasoning_tokens": 3, + }, + "total_tokens": 17, + }, + }, + }) + ); + + let completed_without_usage = response_event_to_json(codex_api::ResponseEvent::Completed { + response_id: "resp-2".to_string(), + token_usage: None, + }); + assert_eq!( + completed_without_usage, + json!({"type": "response.completed", "response": {"id": "resp-2"}}) + ); + } + + #[test] + fn reasoning_deltas_use_responses_event_names() { + let summary = response_event_to_json(codex_api::ResponseEvent::ReasoningSummaryDelta { + delta: "plan".to_string(), + summary_index: 1, + }); + assert_eq!( + summary, + json!({ + "type": "response.reasoning_summary_text.delta", + "delta": "plan", + "summary_index": 1, + }) + ); + + let content = response_event_to_json(codex_api::ResponseEvent::ReasoningContentDelta { + delta: "detail".to_string(), + content_index: 2, + }); + assert_eq!( + content, + json!({ + "type": "response.reasoning_text.delta", + "delta": "detail", + "content_index": 2, + }) + ); + } +}