diff --git a/codex-rs/api-client/src/chat.rs b/codex-rs/api-client/src/chat.rs index 1f24b61b5e..cc6e43869f 100644 --- a/codex-rs/api-client/src/chat.rs +++ b/codex-rs/api-client/src/chat.rs @@ -140,7 +140,7 @@ impl ApiClient for ChatCompletionsApiClient { .get(reqwest::header::RETRY_AFTER) .and_then(|v| v.to_str().ok()) .and_then(|s| s.parse::().ok()) - .map(|s| Duration::from_secs(s)); + .map(Duration::from_secs); tokio::time::sleep(retry_after.unwrap_or_else(|| backoff(attempt))).await; } Err(error) => { @@ -624,7 +624,7 @@ async fn process_chat_sse( .and_then(|v| v.as_array()) { for call in tool_calls { - if let Some(index) = call.get("index").and_then(|i| i.as_u64()) + if let Some(index) = call.get("index").and_then(serde_json::Value::as_u64) && index == 0 && let Some(function) = call.get("function") { @@ -641,8 +641,7 @@ async fn process_chat_sse( if let Some(finish) = choice.get("finish_reason").and_then(|f| f.as_str()) && finish == "tool_calls" - { - if let Some(name) = function_call_state.name.take() { + && let Some(name) = function_call_state.name.take() { let call_id = function_call_state.call_id.take().unwrap_or_default(); let arguments = std::mem::take(&mut function_call_state.arguments); @@ -655,7 +654,6 @@ async fn process_chat_sse( let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await; } - } } } } diff --git a/codex-rs/api-client/src/model_provider.rs b/codex-rs/api-client/src/model_provider.rs index 88fea05c0f..82eb8ed504 100644 --- a/codex-rs/api-client/src/model_provider.rs +++ b/codex-rs/api-client/src/model_provider.rs @@ -120,11 +120,10 @@ impl ModelProviderInfo { let url = self.get_full_url(effective_auth.as_ref()); let mut builder = client.post(url); - if let Some(context) = effective_auth.as_ref() { - if let Some(token) = context.bearer_token.as_ref() { + if let Some(context) = effective_auth.as_ref() + && let Some(token) = context.bearer_token.as_ref() { builder = builder.bearer_auth(token); } - } Ok(self.apply_http_headers(builder)) } diff --git a/codex-rs/api-client/src/responses.rs b/codex-rs/api-client/src/responses.rs index c3a4ca8c89..cdfe621e1c 100644 --- a/codex-rs/api-client/src/responses.rs +++ b/codex-rs/api-client/src/responses.rs @@ -74,13 +74,12 @@ impl ApiClient for ResponsesApiClient { let mut payload_json = self.build_payload(&prompt)?; - if self.config.provider.is_azure_responses_endpoint() { - if let Some(input_value) = payload_json.get_mut("input") + if self.config.provider.is_azure_responses_endpoint() + && let Some(input_value) = payload_json.get_mut("input") && let Some(array) = input_value.as_array_mut() { attach_item_ids_array(array, &prompt.input); } - } let max_attempts = self.config.provider.request_max_retries(); for attempt in 0..=max_attempts { @@ -218,7 +217,7 @@ impl ResponsesApiClient { .headers() .get("cf-ray") .and_then(|v| v.to_str().ok()) - .map(|s| s.to_string()); + .map(std::string::ToString::to_string); } match res { @@ -260,8 +259,8 @@ impl ResponsesApiClient { .and_then(|s| s.parse::().ok()); let retry_after = retry_after_secs.map(|s| Duration::from_millis(s * 1_000)); - if status == StatusCode::UNAUTHORIZED { - if let Some(provider) = self.config.auth_provider.as_ref() + if status == StatusCode::UNAUTHORIZED + && let Some(provider) = self.config.auth_provider.as_ref() && let Some(ctx) = auth.as_ref() && ctx.mode == AuthMode::ChatGPT { @@ -270,7 +269,6 @@ impl ResponsesApiClient { .await .map_err(|err| StreamAttemptError::Fatal(Error::Auth(err)))?; } - } if !(status == StatusCode::TOO_MANY_REQUESTS || status == StatusCode::UNAUTHORIZED @@ -718,13 +716,13 @@ pub async fn stream_from_fixture( let (tx_event, rx_event) = mpsc::channel::>(1600); let display_path = path.as_ref().display().to_string(); let file = std::fs::File::open(path.as_ref()) - .map_err(|e| Error::Other(format!("failed to open fixture {}: {}", display_path, e)))?; + .map_err(|e| Error::Other(format!("failed to open fixture {display_path}: {e}")))?; let lines = std::io::BufReader::new(file).lines(); let mut content = String::new(); for line in lines { let line = line - .map_err(|e| Error::Other(format!("failed to read fixture {}: {}", display_path, e)))?; + .map_err(|e| Error::Other(format!("failed to read fixture {display_path}: {e}")))?; content.push_str(&line); content.push_str("\n\n"); } diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index d3318153b2..90fa35d02b 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -101,7 +101,7 @@ impl ResponsesBackend { self.client .stream(prompt) .await - .map(|stream| stream.boxed()) + .map(futures::StreamExt::boxed) } } @@ -181,8 +181,8 @@ impl ModelClient { pub async fn stream(&self, payload: &StreamPayload) -> Result { let mut prompt = payload.prompt.clone(); self.populate_prompt(&mut prompt); - if self.provider.wire_api == WireApi::Responses { - if let Some(path) = &*CODEX_RS_SSE_FIXTURE { + if self.provider.wire_api == WireApi::Responses + && let Some(path) = &*CODEX_RS_SSE_FIXTURE { warn!(path, "Streaming from fixture"); let stream = stream_from_fixture( path, @@ -194,7 +194,6 @@ impl ModelClient { .boxed(); return Ok(wrap_stream(stream)); } - } let backend = self .backend diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 86fc429bab..02371c7161 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -303,7 +303,7 @@ pub(crate) struct SessionConfiguration { provider: ModelProviderInfo, /// If not specified, server will use its default model. - model: String, + pub(crate) model: String, model_reasoning_effort: Option, model_reasoning_summary: ReasoningSummaryConfig, @@ -315,7 +315,7 @@ pub(crate) struct SessionConfiguration { user_instructions: Option, /// Base instructions override. - base_instructions: Option, + pub(crate) base_instructions: Option, /// Compact prompt override. compact_prompt: Option, @@ -335,7 +335,7 @@ pub(crate) struct SessionConfiguration { cwd: PathBuf, /// Set of feature flags for this session - features: Features, + pub(crate) features: Features, // TODO(pakrym): Remove config from here original_config_do_not_use: Arc, @@ -588,8 +588,9 @@ impl Session { config.active_profile.clone(), ); - // Create the mutable state for the Session. - let state = SessionState::new(session_configuration.clone()); + let model_family = find_family_for_model(&session_configuration.model) + .unwrap_or_else(|| config.model_family.clone()); + let state = SessionState::new(session_configuration.clone(), model_family); let services = SessionServices { mcp_connection_manager, @@ -696,7 +697,6 @@ impl Session { pub(crate) async fn update_settings(&self, updates: SessionSettingsUpdate) { let mut state = self.state.lock().await; - state.session_configuration = state.session_configuration.apply(&updates); } @@ -942,12 +942,6 @@ impl Session { self.send_raw_response_items(turn_context, items).await; } - async fn prompt_for_turn(&self, turn_context: &TurnContext) -> Prompt { - let supports_chain = turn_context.client.supports_responses_api_chaining(); - let mut state = self.state.lock().await; - state.prompt_for_turn(supports_chain) - } - fn reconstruct_history_from_rollout( &self, turn_context: &TurnContext, @@ -1801,7 +1795,8 @@ pub(crate) async fn run_task( // Construct the input that we will send to the model. sess.record_conversation_items(&turn_context, &pending_input) .await; - let prompt = sess.prompt_for_turn(&turn_context).await; + let mut state = sess.state.lock().await; + let prompt = state.prompt_for_turn(); let turn_input_messages: Vec = { prompt @@ -2594,7 +2589,9 @@ mod tests { session_source: SessionSource::Exec, }; - let state = SessionState::new(session_configuration.clone()); + let model_family = find_family_for_model(&session_configuration.model) + .unwrap_or_else(|| config.model_family.clone()); + let state = SessionState::new(session_configuration.clone(), model_family); let services = SessionServices { mcp_connection_manager: McpConnectionManager::default(), @@ -2670,7 +2667,9 @@ mod tests { session_source: SessionSource::Exec, }; - let state = SessionState::new(session_configuration.clone()); + let model_family = find_family_for_model(&session_configuration.model) + .unwrap_or_else(|| config.model_family.clone()); + let state = SessionState::new(session_configuration.clone(), model_family); let services = SessionServices { mcp_connection_manager: McpConnectionManager::default(), diff --git a/codex-rs/core/src/state/session.rs b/codex-rs/core/src/state/session.rs index b79166db3f..fedb6452f5 100644 --- a/codex-rs/core/src/state/session.rs +++ b/codex-rs/core/src/state/session.rs @@ -3,27 +3,39 @@ use codex_protocol::models::ResponseItem; use crate::client_common::Prompt; +use crate::client_common::compute_full_instructions; use crate::codex::SessionConfiguration; use crate::conversation_history::ConversationHistory; use crate::conversation_history::ResponsesApiChainState; +use crate::conversation_history::format_prompt_items; +use crate::model_family::ModelFamily; use crate::protocol::RateLimitSnapshot; use crate::protocol::TokenUsage; use crate::protocol::TokenUsageInfo; +use crate::tools::spec::ToolsConfig; +use crate::tools::spec::ToolsConfigParams; +use crate::tools::spec::build_specs; +use crate::tools::spec::tools_metadata_for_prompt; /// Persistent, session-scoped state previously stored directly on `Session`. pub(crate) struct SessionState { pub(crate) session_configuration: SessionConfiguration, pub(crate) history: ConversationHistory, pub(crate) latest_rate_limits: Option, + pub(crate) model_family: ModelFamily, } impl SessionState { /// Create a new session state mirroring previous `State::default()` semantics. - pub(crate) fn new(session_configuration: SessionConfiguration) -> Self { + pub(crate) fn new( + session_configuration: SessionConfiguration, + model_family: ModelFamily, + ) -> Self { Self { session_configuration, history: ConversationHistory::new(), latest_rate_limits: None, + model_family, } } @@ -79,13 +91,36 @@ impl SessionState { self.history.set_token_usage_full(context_window); } - pub(crate) fn prompt_for_turn(&mut self, supports_responses_api_chaining: bool) -> Prompt { + pub(crate) fn prompt_for_turn(&mut self) -> Prompt { + let tools_config = ToolsConfig::new(&ToolsConfigParams { + model_family: &self.model_family, + features: &self.session_configuration.features, + }); + let (tool_specs, _registry) = build_specs(&tools_config, None).build(); + let tool_specs = tool_specs.into_iter().map(|c| c.spec).collect::>(); + let prompt_items = self.history.get_history_for_prompt(); let chain_state = self.history.responses_api_chain(); - let (prompt, reset_chain) = build_prompt_from_items(prompt_items, chain_state.as_ref()); + let (mut prompt, reset_chain) = build_prompt_from_items(prompt_items, chain_state.as_ref()); if reset_chain { self.reset_responses_api_chain(); } + + // Populate prompt fields that depend only on session state. + let (tools_json, has_freeform_apply_patch) = + tools_metadata_for_prompt(&tool_specs).expect("tool specs serialization"); + format_prompt_items(&mut prompt.input, has_freeform_apply_patch); + + let apply_patch_present = tool_specs.iter().any(|spec| spec.name() == "apply_patch"); + let base_override = self.session_configuration.base_instructions.as_deref(); + let instructions = + compute_full_instructions(base_override, &self.model_family, apply_patch_present) + .into_owned(); + + prompt.instructions = instructions; + prompt.tools = tools_json; + prompt.parallel_tool_calls = self.model_family.supports_parallel_tool_calls; + prompt } }