diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index bbf9b26c98..24f4fb889d 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -101,6 +101,7 @@ use crate::tools::spec::create_tools_json_for_responses_api; pub const OPENAI_BETA_HEADER: &str = "OpenAI-Beta"; pub const OPENAI_BETA_RESPONSES_WEBSOCKETS: &str = "responses_websockets=2026-02-04"; pub const X_CODEX_TURN_STATE_HEADER: &str = "x-codex-turn-state"; +pub const X_CODEX_TURN_ID_HEADER: &str = "x-codex-turn-id"; pub const X_CODEX_TURN_METADATA_HEADER: &str = "x-codex-turn-metadata"; pub const X_RESPONSESAPI_INCLUDE_TIMING_METRICS_HEADER: &str = "x-responsesapi-include-timing-metrics"; @@ -393,9 +394,11 @@ impl ModelClient { api_provider: codex_api::Provider, api_auth: CoreAuthProvider, turn_state: Option>>, + turn_id_header: Option<&str>, turn_metadata_header: Option<&str>, ) -> std::result::Result { - let headers = self.build_websocket_headers(turn_state.as_ref(), turn_metadata_header); + let headers = + self.build_websocket_headers(turn_state.as_ref(), turn_id_header, turn_metadata_header); let websocket_telemetry = ModelClientSession::build_websocket_telemetry(otel_manager); ApiWebSocketResponsesClient::new(api_provider, api_auth) .connect( @@ -414,12 +417,15 @@ impl ModelClient { fn build_websocket_headers( &self, turn_state: Option<&Arc>>, + turn_id_header: Option<&str>, turn_metadata_header: Option<&str>, ) -> ApiHeaderMap { + let turn_id_header = parse_turn_id_header(turn_id_header); let turn_metadata_header = parse_turn_metadata_header(turn_metadata_header); let mut headers = build_responses_headers( self.state.beta_features_header.as_deref(), turn_state, + turn_id_header.as_ref(), turn_metadata_header.as_ref(), ); headers.extend(build_conversation_headers(Some( @@ -523,9 +529,11 @@ impl ModelClientSession { /// regardless of transport choice. fn build_responses_options( &self, + turn_id_header: Option<&str>, turn_metadata_header: Option<&str>, compression: Compression, ) -> ApiResponsesOptions { + let turn_id_header = parse_turn_id_header(turn_id_header); let turn_metadata_header = parse_turn_metadata_header(turn_metadata_header); let conversation_id = self.client.state.conversation_id.to_string(); @@ -535,6 +543,7 @@ impl ModelClientSession { extra_headers: build_responses_headers( self.client.state.beta_features_header.as_deref(), Some(&self.turn_state), + turn_id_header.as_ref(), turn_metadata_header.as_ref(), ), compression, @@ -630,6 +639,7 @@ impl ModelClientSession { &mut self, otel_manager: &OtelManager, model_info: &ModelInfo, + turn_id_header: Option<&str>, turn_metadata_header: Option<&str>, ) -> std::result::Result<(), ApiError> { if !self.client.responses_websocket_enabled(model_info) || self.client.websockets_disabled() @@ -653,6 +663,7 @@ impl ModelClientSession { client_setup.api_provider, client_setup.api_auth, Some(Arc::clone(&self.turn_state)), + turn_id_header, turn_metadata_header, ) .await?; @@ -666,6 +677,7 @@ impl ModelClientSession { otel_manager: &OtelManager, api_provider: codex_api::Provider, api_auth: CoreAuthProvider, + turn_id_header: Option<&str>, turn_metadata_header: Option<&str>, options: &ApiResponsesOptions, ) -> std::result::Result<&ApiWebSocketConnection, ApiError> { @@ -688,6 +700,7 @@ impl ModelClientSession { api_provider, api_auth, Some(turn_state), + turn_id_header, turn_metadata_header, ) .await?; @@ -722,6 +735,7 @@ impl ModelClientSession { otel_manager: &OtelManager, effort: Option, summary: ReasoningSummaryConfig, + turn_id_header: Option<&str>, turn_metadata_header: Option<&str>, ) -> Result { if let Some(path) = &*CODEX_RS_SSE_FIXTURE { @@ -744,7 +758,8 @@ impl ModelClientSession { let transport = ReqwestTransport::new(build_reqwest_client()); let (request_telemetry, sse_telemetry) = Self::build_streaming_telemetry(otel_manager); let compression = self.responses_request_compression(client_setup.auth.as_ref()); - let options = self.build_responses_options(turn_metadata_header, compression); + let options = + self.build_responses_options(turn_id_header, turn_metadata_header, compression); let request = self.build_responses_request( &client_setup.api_provider, @@ -786,6 +801,7 @@ impl ModelClientSession { otel_manager: &OtelManager, effort: Option, summary: ReasoningSummaryConfig, + turn_id_header: Option<&str>, turn_metadata_header: Option<&str>, ) -> Result { let auth_manager = self.client.state.auth_manager.clone(); @@ -797,7 +813,8 @@ impl ModelClientSession { let client_setup = self.client.current_client_setup().await?; let compression = self.responses_request_compression(client_setup.auth.as_ref()); - let options = self.build_responses_options(turn_metadata_header, compression); + let options = + self.build_responses_options(turn_id_header, turn_metadata_header, compression); let request = self.build_responses_request( &client_setup.api_provider, prompt, @@ -812,6 +829,7 @@ impl ModelClientSession { otel_manager, client_setup.api_provider, client_setup.api_auth, + turn_id_header, turn_metadata_header, &options, ) @@ -885,6 +903,7 @@ impl ModelClientSession { otel_manager: &OtelManager, effort: Option, summary: ReasoningSummaryConfig, + turn_id_header: Option<&str>, turn_metadata_header: Option<&str>, ) -> Result { let wire_api = self.client.state.provider.wire_api; @@ -901,6 +920,7 @@ impl ModelClientSession { otel_manager, effort, summary, + turn_id_header, turn_metadata_header, ) .await? @@ -918,6 +938,7 @@ impl ModelClientSession { otel_manager, effort, summary, + turn_id_header, turn_metadata_header, ) .await @@ -954,10 +975,10 @@ impl ModelClientSession { } } -/// Parses per-turn metadata into an HTTP header value. -/// -/// Invalid values are treated as absent so callers can compare and propagate -/// metadata with the same sanitization path used when constructing headers. +fn parse_turn_id_header(turn_id_header: Option<&str>) -> Option { + turn_id_header.and_then(|value| HeaderValue::from_str(value).ok()) +} + fn parse_turn_metadata_header(turn_metadata_header: Option<&str>) -> Option { turn_metadata_header.and_then(|value| HeaderValue::from_str(value).ok()) } @@ -968,10 +989,12 @@ fn parse_turn_metadata_header(turn_metadata_header: Option<&str>) -> Option, turn_state: Option<&Arc>>, + turn_id_header: Option<&HeaderValue>, turn_metadata_header: Option<&HeaderValue>, ) -> ApiHeaderMap { let mut headers = ApiHeaderMap::new(); @@ -987,6 +1010,9 @@ fn build_responses_headers( { headers.insert(X_CODEX_TURN_STATE_HEADER, header_value); } + if let Some(turn_id_header) = turn_id_header { + headers.insert(X_CODEX_TURN_ID_HEADER, turn_id_header.clone()); + } if let Some(header_value) = turn_metadata_header { headers.insert(X_CODEX_TURN_METADATA_HEADER, header_value.clone()); } diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 75a2551f79..c25b1ffd7a 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -37,6 +37,8 @@ use crate::stream_events_utils::handle_output_item_done; use crate::stream_events_utils::last_assistant_message_from_item; use crate::terminal; use crate::truncate::TruncationPolicy; +use crate::turn_metadata::TurnMetadataHeaderJob; +use crate::turn_metadata::TurnMetadataPoll; use crate::turn_metadata::build_turn_metadata_header; use crate::turn_metadata::resolve_turn_metadata_header_with_timeout; use crate::util::error_or_panic; @@ -90,7 +92,6 @@ use rmcp::model::RequestId; use serde_json; use serde_json::Value; use tokio::sync::Mutex; -use tokio::sync::OnceCell; use tokio::sync::RwLock; use tokio::sync::oneshot; use tokio::sync::watch; @@ -102,7 +103,6 @@ use tracing::field; use tracing::info; use tracing::info_span; use tracing::instrument; -use tracing::trace; use tracing::trace_span; use tracing::warn; use uuid::Uuid; @@ -557,7 +557,7 @@ pub(crate) struct TurnContext { pub(crate) truncation_policy: TruncationPolicy, pub(crate) js_repl: Arc, pub(crate) dynamic_tools: Vec, - turn_metadata_header: OnceCell>, + turn_metadata_job: Arc, } impl TurnContext { pub(crate) fn model_context_window(&self) -> Option { @@ -637,7 +637,7 @@ impl TurnContext { truncation_policy, js_repl: Arc::clone(&self.js_repl), dynamic_tools: self.dynamic_tools.clone(), - turn_metadata_header: self.turn_metadata_header.clone(), + turn_metadata_job: self.turn_metadata_job.clone(), } } @@ -653,39 +653,23 @@ impl TurnContext { .unwrap_or(compact::SUMMARIZATION_PROMPT) } - async fn build_turn_metadata_header(&self) -> Option { - let sandbox = sandbox_tag(&self.sandbox_policy, self.windows_sandbox_level); - self.turn_metadata_header - .get_or_init(|| async { - build_turn_metadata_header(self.cwd.as_path(), Some(sandbox)).await - }) - .await - .clone() - } - - /// Resolves the per-turn metadata header under a shared timeout policy. - /// - /// This uses the same timeout helper as websocket startup prewarm so both turn execution and - /// background prewarm observe identical "timeout means best-effort fallback" behavior. - pub async fn resolve_turn_metadata_header(&self) -> Option { - resolve_turn_metadata_header_with_timeout( - self.build_turn_metadata_header(), - self.turn_metadata_header.get().cloned().flatten(), - ) - .await + /// Returns turn metadata if ready, without waiting on metadata computation. + pub(crate) fn poll_turn_metadata_header(&self) -> TurnMetadataPoll { + self.turn_metadata_job.poll() } /// Starts best-effort background computation of turn metadata. /// - /// This warms the cached value used by [`TurnContext::resolve_turn_metadata_header`] so turns - /// and websocket prewarm are less likely to pay metadata construction latency on demand. + /// Requests do not await this task. Callers should poll via + /// [`TurnContext::poll_turn_metadata_header`] and attach metadata only when ready. pub fn spawn_turn_metadata_header_task(self: &Arc) { - let context = Arc::clone(self); - tokio::spawn(async move { - trace!("Spawning turn metadata calculation task"); - context.build_turn_metadata_header().await; - trace!("Turn metadata calculation task completed"); - }); + let sandbox = sandbox_tag(&self.sandbox_policy, self.windows_sandbox_level).to_string(); + self.turn_metadata_job + .spawn(self.cwd.clone(), Some(sandbox)); + } + + pub(crate) fn cancel_turn_metadata_header_task(&self) { + self.turn_metadata_job.cancel(); } } @@ -942,7 +926,7 @@ impl Session { truncation_policy: model_info.truncation_policy.into(), js_repl, dynamic_tools: session_configuration.dynamic_tools.clone(), - turn_metadata_header: OnceCell::new(), + turn_metadata_job: Arc::new(TurnMetadataHeaderJob::default()), } } @@ -3977,7 +3961,7 @@ async fn spawn_review_thread( js_repl: Arc::clone(&sess.js_repl), dynamic_tools: parent_turn_context.dynamic_tools.clone(), truncation_policy: model_info.truncation_policy.into(), - turn_metadata_header: parent_turn_context.turn_metadata_header.clone(), + turn_metadata_job: Arc::new(TurnMetadataHeaderJob::default()), }; // Seed the child task with the review prompt as the initial user message. @@ -3987,6 +3971,7 @@ async fn spawn_review_thread( text_elements: Vec::new(), }]; let tc = Arc::new(review_turn_context); + tc.spawn_turn_metadata_header_task(); sess.spawn_task(tc.clone(), input, ReviewTask::new()).await; // Announce entering review mode so UIs can switch modes. @@ -4065,7 +4050,14 @@ fn errors_to_info(errors: &[SkillError]) -> Vec { /// back to the model in the next sampling request. /// - If the model sends only an assistant message, we record it in the /// conversation history and consider the turn complete. -/// +struct TurnMetadataJobCancellation<'a>(&'a TurnContext); + +impl Drop for TurnMetadataJobCancellation<'_> { + fn drop(&mut self) { + self.0.cancel_turn_metadata_header_task(); + } +} + pub(crate) async fn run_turn( sess: Arc, turn_context: Arc, @@ -4073,6 +4065,8 @@ pub(crate) async fn run_turn( prewarmed_client_session: Option, cancellation_token: CancellationToken, ) -> Option { + let _turn_metadata_job_cancellation = TurnMetadataJobCancellation(turn_context.as_ref()); + if input.is_empty() { return None; } @@ -4194,7 +4188,6 @@ pub(crate) async fn run_turn( // many turns, from the perspective of the user, it is a single turn. let turn_diff_tracker = Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new())); - let turn_metadata_header = turn_context.resolve_turn_metadata_header().await; // `ModelClientSession` is turn-scoped and caches WebSocket + sticky routing state, so we reuse // one instance across retries within this turn. let mut client_session = @@ -4246,6 +4239,10 @@ pub(crate) async fn run_turn( }) .map(|user_message| user_message.message()) .collect::>(); + let turn_metadata_header = match turn_context.poll_turn_metadata_header() { + TurnMetadataPoll::Ready(header) => header, + TurnMetadataPoll::Pending => None, + }; match run_sampling_request( Arc::clone(&sess), Arc::clone(&turn_context), @@ -5204,6 +5201,7 @@ async fn try_run_sampling_request( &turn_context.otel_manager, turn_context.reasoning_effort, turn_context.reasoning_summary, + Some(turn_context.sub_id.as_str()), turn_metadata_header, ) .instrument(trace_span!("stream_request")) diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index f7de93dfe5..a9c1fce25e 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -17,6 +17,7 @@ use crate::protocol::WarningEvent; use crate::truncate::TruncationPolicy; use crate::truncate::approx_token_count; use crate::truncate::truncate_text; +use crate::turn_metadata::TurnMetadataPoll; use crate::util::backoff; use codex_protocol::items::ContextCompactionItem; use codex_protocol::items::TurnItem; @@ -85,7 +86,6 @@ async fn run_compact_task_inner( let max_retries = turn_context.provider.stream_max_retries(); let mut retries = 0; - let turn_metadata_header = turn_context.resolve_turn_metadata_header().await; let mut client_session = sess.services.model_client.new_session(); // Reuse one client session so turn-scoped state (sticky routing, websocket append tracking) // survives retries within this compact turn. @@ -124,6 +124,10 @@ async fn run_compact_task_inner( personality: turn_context.personality, ..Default::default() }; + let turn_metadata_header = match turn_context.poll_turn_metadata_header() { + TurnMetadataPoll::Ready(header) => header, + TurnMetadataPoll::Pending => None, + }; let attempt_result = drain_to_completed( &sess, turn_context.as_ref(), @@ -388,6 +392,7 @@ async fn drain_to_completed( &turn_context.otel_manager, turn_context.reasoning_effort, turn_context.reasoning_summary, + Some(turn_context.sub_id.as_str()), turn_metadata_header, ) .await?; diff --git a/codex-rs/core/src/git_info.rs b/codex-rs/core/src/git_info.rs index dcabc3a341..df59559939 100644 --- a/codex-rs/core/src/git_info.rs +++ b/codex-rs/core/src/git_info.rs @@ -150,6 +150,18 @@ pub async fn get_head_commit_hash(cwd: &Path) -> Option { } } +/// Return whether the repository has tracked or untracked modifications. +/// +/// Uses `git status --porcelain` where empty output means the working tree is clean. +pub async fn get_has_changes(cwd: &Path) -> Option { + let output = run_git_command_with_timeout(&["status", "--porcelain"], cwd).await?; + if !output.status.success() { + return None; + } + + Some(!output.stdout.is_empty()) +} + fn parse_git_remote_urls(stdout: &str) -> Option> { let mut remotes = BTreeMap::new(); for line in stdout.lines() { @@ -254,7 +266,11 @@ pub async fn git_diff_to_remote(cwd: &Path) -> Option { /// Run a git command with a timeout to prevent blocking on large repositories async fn run_git_command_with_timeout(args: &[&str], cwd: &Path) -> Option { let mut command = Command::new("git"); - command.args(args).current_dir(cwd).kill_on_drop(true); + command + .arg("--no-optional-locks") + .args(args) + .current_dir(cwd) + .kill_on_drop(true); let result = timeout(GIT_COMMAND_TIMEOUT, command.output()).await; match result { @@ -962,6 +978,43 @@ mod tests { assert_eq!(git_info.branch, Some("feature-branch".to_string())); } + #[tokio::test] + async fn test_get_has_changes_clean_repo_returns_false() { + let temp_dir = TempDir::new().expect("Failed to create temp dir"); + let repo_path = create_test_git_repo(&temp_dir).await; + + let has_changes = get_has_changes(&repo_path) + .await + .expect("git status should succeed"); + assert!(!has_changes); + } + + #[tokio::test] + async fn test_get_has_changes_returns_true_for_tracked_modifications() { + let temp_dir = TempDir::new().expect("Failed to create temp dir"); + let repo_path = create_test_git_repo(&temp_dir).await; + + fs::write(repo_path.join("test.txt"), "modified tracked file").expect("write tracked file"); + + let has_changes = get_has_changes(&repo_path) + .await + .expect("git status should succeed"); + assert!(has_changes); + } + + #[tokio::test] + async fn test_get_has_changes_returns_true_for_untracked_files() { + let temp_dir = TempDir::new().expect("Failed to create temp dir"); + let repo_path = create_test_git_repo(&temp_dir).await; + + fs::write(repo_path.join("new-file.txt"), "new untracked file").expect("write new file"); + + let has_changes = get_has_changes(&repo_path) + .await + .expect("git status should succeed"); + assert!(has_changes); + } + #[tokio::test] async fn test_get_git_working_tree_state_clean_repo() { let temp_dir = TempDir::new().expect("Failed to create temp dir"); diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 764fc4fbba..6bf70cd153 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -140,6 +140,7 @@ pub use codex_shell_command::parse_command; pub use codex_shell_command::powershell; pub use apply_patch::CODEX_APPLY_PATCH_ARG1; +pub use client::X_CODEX_TURN_ID_HEADER; pub use client::X_CODEX_TURN_METADATA_HEADER; pub use exec_policy::ExecPolicyError; pub use exec_policy::check_execpolicy_for_warnings; diff --git a/codex-rs/core/src/memories/phase1.rs b/codex-rs/core/src/memories/phase1.rs index a7c171d3e0..c4bfc227b3 100644 --- a/codex-rs/core/src/memories/phase1.rs +++ b/codex-rs/core/src/memories/phase1.rs @@ -8,6 +8,7 @@ use crate::memories::phase_one; use crate::memories::prompts::build_stage_one_input_message; use crate::rollout::INTERACTIVE_SESSION_SOURCES; use crate::rollout::policy::should_persist_response_item_for_memories; +use crate::turn_metadata::TurnMetadataPoll; use codex_api::ResponseEvent; use codex_otel::OtelManager; use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig; @@ -29,6 +30,7 @@ use tracing::warn; #[derive(Clone, Debug)] pub(in crate::memories) struct Phase1RequestContext { + pub(in crate::memories) turn_id: String, pub(in crate::memories) model_info: ModelInfo, pub(in crate::memories) otel_manager: OtelManager, pub(in crate::memories) reasoning_effort: Option, @@ -124,6 +126,7 @@ impl Phase1RequestContext { turn_metadata_header: Option, ) -> Self { Self { + turn_id: turn_context.sub_id.clone(), model_info: turn_context.model_info.clone(), otel_manager: turn_context.otel_manager.clone(), reasoning_effort: turn_context.reasoning_effort, @@ -174,10 +177,11 @@ async fn claim_startup_jobs(session: &Arc) -> Option) -> Phase1RequestContext { let turn_context = session.new_default_turn().await; - Phase1RequestContext::from_turn_context( - turn_context.as_ref(), - turn_context.resolve_turn_metadata_header().await, - ) + let turn_metadata_header = match turn_context.poll_turn_metadata_header() { + TurnMetadataPoll::Ready(header) => header, + TurnMetadataPoll::Pending => None, + }; + Phase1RequestContext::from_turn_context(turn_context.as_ref(), turn_metadata_header) } async fn run_jobs( @@ -283,6 +287,7 @@ mod job { &stage_one_context.otel_manager, stage_one_context.reasoning_effort, stage_one_context.reasoning_summary, + Some(stage_one_context.turn_id.as_str()), stage_one_context.turn_metadata_header.as_deref(), ) .await?; diff --git a/codex-rs/core/src/tasks/regular.rs b/codex-rs/core/src/tasks/regular.rs index 8782d7cedb..05dd8232f3 100644 --- a/codex-rs/core/src/tasks/regular.rs +++ b/codex-rs/core/src/tasks/regular.rs @@ -45,7 +45,12 @@ impl RegularTask { let mut client_session = model_client.new_session(); let turn_metadata_header = turn_metadata_header.await; match client_session - .prewarm_websocket(&otel_manager, &model_info, turn_metadata_header.as_deref()) + .prewarm_websocket( + &otel_manager, + &model_info, + None, + turn_metadata_header.as_deref(), + ) .await { Ok(()) => Some(client_session), diff --git a/codex-rs/core/src/turn_metadata.rs b/codex-rs/core/src/turn_metadata.rs index e17a78804a..e0b3c60a01 100644 --- a/codex-rs/core/src/turn_metadata.rs +++ b/codex-rs/core/src/turn_metadata.rs @@ -1,19 +1,25 @@ -//! Helpers for computing and resolving optional per-turn metadata headers. +//! Helpers for computing optional per-turn metadata headers. //! //! This module owns both metadata construction and the shared timeout policy used by -//! turn execution and startup websocket prewarm. Keeping timeout behavior centralized -//! ensures both call sites treat timeout as the same best-effort fallback condition. +//! startup websocket prewarm. Turn-time request attachment is handled via a non-blocking +//! background job (`TurnMetadataHeaderJob`) so request send paths never await metadata. use std::collections::BTreeMap; use std::future::Future; use std::path::Path; +use std::path::PathBuf; +use std::sync::Mutex; use std::time::Duration; use serde::Serialize; +use tokio::sync::oneshot; +use tokio::sync::oneshot::error::TryRecvError; +use tokio::task::JoinHandle; use tracing::warn; use crate::git_info::get_git_remote_urls_assume_git_repo; use crate::git_info::get_git_repo_root; +use crate::git_info::get_has_changes; use crate::git_info::get_head_commit_hash; pub(crate) const TURN_METADATA_HEADER_TIMEOUT: Duration = Duration::from_millis(250); @@ -22,8 +28,8 @@ pub(crate) const TURN_METADATA_HEADER_TIMEOUT: Duration = Duration::from_millis( /// /// On timeout, this logs a warning and returns the provided fallback header. /// -/// Keeping this helper centralized avoids drift between turn-time metadata resolution and startup -/// websocket prewarm, both of which need identical timeout semantics. +/// Keeping this helper centralized avoids drift between startup websocket prewarm and any other +/// timeout-bounded one-shot metadata call sites. pub(crate) async fn resolve_turn_metadata_header_with_timeout( build_header: F, fallback_on_timeout: Option, @@ -48,6 +54,8 @@ struct TurnMetadataWorkspace { #[serde(default, skip_serializing_if = "Option::is_none")] associated_remote_urls: Option>, #[serde(default, skip_serializing_if = "Option::is_none")] + has_changes: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] latest_git_commit_hash: Option, } @@ -62,11 +70,16 @@ struct TurnMetadata { pub async fn build_turn_metadata_header(cwd: &Path, sandbox: Option<&str>) -> Option { let repo_root = get_git_repo_root(cwd); - let (latest_git_commit_hash, associated_remote_urls) = tokio::join!( + let (latest_git_commit_hash, associated_remote_urls, has_changes) = tokio::join!( get_head_commit_hash(cwd), - get_git_remote_urls_assume_git_repo(cwd) + get_git_remote_urls_assume_git_repo(cwd), + get_has_changes(cwd) ); - if latest_git_commit_hash.is_none() && associated_remote_urls.is_none() && sandbox.is_none() { + if latest_git_commit_hash.is_none() + && associated_remote_urls.is_none() + && has_changes.is_none() + && sandbox.is_none() + { return None; } @@ -76,6 +89,7 @@ pub async fn build_turn_metadata_header(cwd: &Path, sandbox: Option<&str>) -> Op repo_root.to_string_lossy().into_owned(), TurnMetadataWorkspace { associated_remote_urls, + has_changes, latest_git_commit_hash, }, ); @@ -86,3 +100,93 @@ pub async fn build_turn_metadata_header(cwd: &Path, sandbox: Option<&str>) -> Op }) .ok() } + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum TurnMetadataPoll { + Pending, + Ready(Option), +} + +#[derive(Debug, Default)] +enum TurnMetadataHeaderJobState { + #[default] + NotStarted, + Pending { + receiver: oneshot::Receiver>, + task: JoinHandle<()>, + }, + Ready(Option), +} + +#[derive(Debug, Default)] +pub(crate) struct TurnMetadataHeaderJob { + state: Mutex, +} + +impl TurnMetadataHeaderJob { + pub(crate) fn spawn(&self, cwd: PathBuf, sandbox: Option) { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if !matches!(*state, TurnMetadataHeaderJobState::NotStarted) { + return; + } + + let Ok(handle) = tokio::runtime::Handle::try_current() else { + *state = TurnMetadataHeaderJobState::Ready(None); + return; + }; + + let (tx, rx) = oneshot::channel::>(); + let task = handle.spawn(async move { + let header = build_turn_metadata_header(cwd.as_path(), sandbox.as_deref()).await; + let _ = tx.send(header); + }); + *state = TurnMetadataHeaderJobState::Pending { receiver: rx, task }; + } + + pub(crate) fn poll(&self) -> TurnMetadataPoll { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + match &mut *state { + TurnMetadataHeaderJobState::NotStarted => TurnMetadataPoll::Pending, + TurnMetadataHeaderJobState::Ready(header) => TurnMetadataPoll::Ready(header.clone()), + TurnMetadataHeaderJobState::Pending { receiver, .. } => match receiver.try_recv() { + Ok(header) => { + *state = TurnMetadataHeaderJobState::Ready(header.clone()); + TurnMetadataPoll::Ready(header) + } + Err(TryRecvError::Empty) => TurnMetadataPoll::Pending, + Err(TryRecvError::Closed) => { + *state = TurnMetadataHeaderJobState::Ready(None); + TurnMetadataPoll::Ready(None) + } + }, + } + } + + pub(crate) fn cancel(&self) { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let TurnMetadataHeaderJobState::Pending { task, .. } = &mut *state { + task.abort(); + } + *state = TurnMetadataHeaderJobState::Ready(None); + } +} + +impl Drop for TurnMetadataHeaderJob { + fn drop(&mut self) { + let Ok(state) = self.state.get_mut() else { + return; + }; + if let TurnMetadataHeaderJobState::Pending { task, .. } = state { + task.abort(); + } + } +} diff --git a/codex-rs/core/tests/responses_headers.rs b/codex-rs/core/tests/responses_headers.rs index d22e5c66e4..737bcbbcc6 100644 --- a/codex-rs/core/tests/responses_headers.rs +++ b/codex-rs/core/tests/responses_headers.rs @@ -111,7 +111,15 @@ async fn responses_stream_includes_subagent_header_on_review() { }]; let mut stream = client_session - .stream(&prompt, &model_info, &otel_manager, effort, summary, None) + .stream( + &prompt, + &model_info, + &otel_manager, + effort, + summary, + None, + None, + ) .await .expect("stream failed"); while let Some(event) = stream.next().await { @@ -217,7 +225,15 @@ async fn responses_stream_includes_subagent_header_on_other() { }]; let mut stream = client_session - .stream(&prompt, &model_info, &otel_manager, effort, summary, None) + .stream( + &prompt, + &model_info, + &otel_manager, + effort, + summary, + None, + None, + ) .await .expect("stream failed"); while let Some(event) = stream.next().await { @@ -322,7 +338,15 @@ async fn responses_respects_model_info_overrides_from_config() { }]; let mut stream = client_session - .stream(&prompt, &model_info, &otel_manager, effort, summary, None) + .stream( + &prompt, + &model_info, + &otel_manager, + effort, + summary, + None, + None, + ) .await .expect("stream failed"); while let Some(event) = stream.next().await { @@ -357,31 +381,20 @@ async fn responses_stream_includes_turn_metadata_header_for_git_workspace_e2e() core_test_support::skip_if_no_network!(); let server = responses::start_mock_server().await; - let response_body = responses::sse(vec![ + let first_response = responses::sse(vec![ responses::ev_response_created("resp-1"), + responses::ev_shell_command_call("call-1", "sleep 1"), responses::ev_completed("resp-1"), ]); + let second_response = responses::sse(vec![ + responses::ev_response_created("resp-2"), + responses::ev_assistant_message("msg-1", "done"), + responses::ev_completed("resp-2"), + ]); let test = test_codex().build(&server).await.expect("build test codex"); let cwd = test.cwd_path(); - let first_request = responses::mount_sse_once(&server, response_body.clone()).await; - test.submit_turn("hello") - .await - .expect("submit first turn prompt"); - let initial_header = first_request - .single_request() - .header("x-codex-turn-metadata") - .expect("x-codex-turn-metadata header should be present"); - let initial_parsed: serde_json::Value = - serde_json::from_str(&initial_header).expect("x-codex-turn-metadata should be valid JSON"); - assert_eq!( - initial_parsed - .get("sandbox") - .and_then(serde_json::Value::as_str), - Some("none") - ); - let git_config_global = cwd.join("empty-git-config"); std::fs::write(&git_config_global, "").expect("write empty git config"); let run_git = |args: &[&str]| { @@ -423,57 +436,111 @@ async fn responses_stream_includes_turn_metadata_header_for_git_workspace_e2e() .expect("git remote get-url output should be valid UTF-8") .trim() .to_string(); + let expected_repo_root = std::fs::canonicalize(cwd) + .unwrap_or_else(|_| cwd.to_path_buf()) + .to_string_lossy() + .into_owned(); - let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5); - loop { - let request_recorder = responses::mount_sse_once(&server, response_body.clone()).await; - tokio::time::sleep(std::time::Duration::from_millis(50)).await; - test.submit_turn("hello") - .await - .expect("submit post-git turn prompt"); + let clean_turn_recorder = responses::mount_response_sequence( + &server, + vec![ + responses::sse_response(first_response.clone()), + responses::sse_response(second_response.clone()), + ], + ) + .await; + test.submit_turn("run a shell command") + .await + .expect("submit clean turn prompt"); - let maybe_metadata = request_recorder - .single_request() - .header("x-codex-turn-metadata") - .and_then(|header_value| { - let parsed: serde_json::Value = serde_json::from_str(&header_value).ok()?; - let workspace = parsed - .get("workspaces") - .and_then(serde_json::Value::as_object) - .and_then(|workspaces| workspaces.values().next()) - .cloned()?; - Some((parsed, workspace)) - }); - let Some((parsed, workspace)) = maybe_metadata else { - if tokio::time::Instant::now() >= deadline { - break; - } - tokio::time::sleep(std::time::Duration::from_millis(25)).await; - continue; - }; + let clean_requests = clean_turn_recorder.requests(); + assert_eq!(clean_requests.len(), 2); + let clean_turn_id_initial = clean_requests[0] + .header("x-codex-turn-id") + .expect("x-codex-turn-id should be present on initial request"); + let clean_turn_id_follow_up = clean_requests[1] + .header("x-codex-turn-id") + .expect("x-codex-turn-id should be present on follow-up request"); + assert_eq!(clean_turn_id_initial, clean_turn_id_follow_up); - assert_eq!( - parsed.get("sandbox").and_then(serde_json::Value::as_str), - Some("none") - ); - assert_eq!( - workspace - .get("latest_git_commit_hash") - .and_then(serde_json::Value::as_str), - Some(expected_head.as_str()) - ); - assert_eq!( - workspace - .get("associated_remote_urls") - .and_then(serde_json::Value::as_object) - .and_then(|remotes| remotes.get("origin")) - .and_then(serde_json::Value::as_str), - Some(expected_origin.as_str()) - ); - return; - } + let clean_metadata_header = clean_requests[1] + .header("x-codex-turn-metadata") + .expect("follow-up request should include x-codex-turn-metadata"); + let clean_parsed: serde_json::Value = serde_json::from_str(&clean_metadata_header) + .expect("x-codex-turn-metadata should be valid JSON"); + let clean_workspace = clean_parsed + .get("workspaces") + .and_then(serde_json::Value::as_object) + .and_then(|workspaces| { + workspaces + .get(&expected_repo_root) + .or_else(|| workspaces.values().next()) + }) + .expect("metadata should include expected repository root"); + assert_eq!( + clean_workspace + .get("latest_git_commit_hash") + .and_then(serde_json::Value::as_str), + Some(expected_head.as_str()) + ); + assert_eq!( + clean_workspace + .get("associated_remote_urls") + .and_then(serde_json::Value::as_object) + .and_then(|remotes| remotes.get("origin")) + .and_then(serde_json::Value::as_str), + Some(expected_origin.as_str()) + ); + assert_eq!( + clean_workspace + .get("has_changes") + .and_then(serde_json::Value::as_bool), + Some(false) + ); - panic!( - "x-codex-turn-metadata with git workspace info was never observed within 5s after git setup" + std::fs::write(cwd.join("untracked.txt"), "new file").expect("write untracked file"); + + let dirty_turn_recorder = responses::mount_response_sequence( + &server, + vec![ + responses::sse_response(first_response), + responses::sse_response(second_response), + ], + ) + .await; + test.submit_turn("run a shell command") + .await + .expect("submit dirty turn prompt"); + + let dirty_requests = dirty_turn_recorder.requests(); + assert_eq!(dirty_requests.len(), 2); + let dirty_turn_id_initial = dirty_requests[0] + .header("x-codex-turn-id") + .expect("x-codex-turn-id should be present on initial dirty request"); + let dirty_turn_id_follow_up = dirty_requests[1] + .header("x-codex-turn-id") + .expect("x-codex-turn-id should be present on follow-up dirty request"); + assert_eq!(dirty_turn_id_initial, dirty_turn_id_follow_up); + assert_ne!(clean_turn_id_initial, dirty_turn_id_initial); + + let dirty_metadata_header = dirty_requests[1] + .header("x-codex-turn-metadata") + .expect("dirty follow-up request should include x-codex-turn-metadata"); + let dirty_parsed: serde_json::Value = serde_json::from_str(&dirty_metadata_header) + .expect("x-codex-turn-metadata should be valid JSON"); + let dirty_workspace = dirty_parsed + .get("workspaces") + .and_then(serde_json::Value::as_object) + .and_then(|workspaces| { + workspaces + .get(&expected_repo_root) + .or_else(|| workspaces.values().next()) + }) + .expect("metadata should include expected repository root"); + assert_eq!( + dirty_workspace + .get("has_changes") + .and_then(serde_json::Value::as_bool), + Some(true) ); } diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index 7c6799df92..76f35dffb5 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -1416,7 +1416,15 @@ async fn azure_responses_request_includes_store_and_reasoning_ids() { }); let mut stream = client_session - .stream(&prompt, &model_info, &otel_manager, effort, summary, None) + .stream( + &prompt, + &model_info, + &otel_manager, + effort, + summary, + None, + None, + ) .await .expect("responses stream to start"); diff --git a/codex-rs/core/tests/suite/client_websockets.rs b/codex-rs/core/tests/suite/client_websockets.rs index d436cac599..60b629dcdc 100755 --- a/codex-rs/core/tests/suite/client_websockets.rs +++ b/codex-rs/core/tests/suite/client_websockets.rs @@ -106,7 +106,7 @@ async fn responses_websocket_preconnect_reuses_connection() { let harness = websocket_harness(&server).await; let mut client_session = harness.client.new_session(); client_session - .prewarm_websocket(&harness.otel_manager, &harness.model_info, None) + .prewarm_websocket(&harness.otel_manager, &harness.model_info, None, None) .await .expect("websocket prewarm failed"); let prompt = prompt_with_input(vec![message_item("hello")]); @@ -131,7 +131,7 @@ async fn responses_websocket_preconnect_is_reused_even_with_header_changes() { let harness = websocket_harness(&server).await; let mut client_session = harness.client.new_session(); client_session - .prewarm_websocket(&harness.otel_manager, &harness.model_info, None) + .prewarm_websocket(&harness.otel_manager, &harness.model_info, None, None) .await .expect("websocket prewarm failed"); let prompt = prompt_with_input(vec![message_item("hello")]); @@ -143,6 +143,7 @@ async fn responses_websocket_preconnect_is_reused_even_with_header_changes() { harness.effort, harness.summary, None, + None, ) .await .expect("websocket stream failed"); @@ -172,7 +173,7 @@ async fn responses_websocket_prewarm_uses_model_preference_when_feature_disabled let harness = websocket_harness_with_options(&server, false, false, false, true).await; let mut client_session = harness.client.new_session(); client_session - .prewarm_websocket(&harness.otel_manager, &harness.model_info, None) + .prewarm_websocket(&harness.otel_manager, &harness.model_info, None, None) .await .expect("websocket prewarm failed"); @@ -318,6 +319,7 @@ async fn responses_websocket_emits_reasoning_included_event() { harness.effort, harness.summary, None, + None, ) .await .expect("websocket stream failed"); @@ -389,6 +391,7 @@ async fn responses_websocket_emits_rate_limit_events() { harness.effort, harness.summary, None, + None, ) .await .expect("websocket stream failed"); @@ -840,6 +843,7 @@ async fn responses_websocket_v2_after_error_uses_full_create_without_previous_re harness.effort, harness.summary, None, + None, ) .await .expect("websocket stream failed"); @@ -1080,6 +1084,7 @@ async fn stream_until_complete( harness.effort, harness.summary, None, + None, ) .await .expect("websocket stream failed"); diff --git a/codex-rs/core/tests/suite/turn_state.rs b/codex-rs/core/tests/suite/turn_state.rs index 833c2aad9b..94f47b987a 100644 --- a/codex-rs/core/tests/suite/turn_state.rs +++ b/codex-rs/core/tests/suite/turn_state.rs @@ -18,6 +18,7 @@ use core_test_support::test_codex::test_codex; use pretty_assertions::assert_eq; const TURN_STATE_HEADER: &str = "x-codex-turn-state"; +const TURN_ID_HEADER: &str = "x-codex-turn-id"; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn responses_turn_state_persists_within_turn_and_resets_after() -> Result<()> { @@ -65,6 +66,18 @@ async fn responses_turn_state_persists_within_turn_and_resets_after() -> Result< ); assert_eq!(requests[2].header(TURN_STATE_HEADER), None); + let first_turn_request_id = requests[0] + .header(TURN_ID_HEADER) + .expect("first request should include x-codex-turn-id"); + let first_turn_follow_up_id = requests[1] + .header(TURN_ID_HEADER) + .expect("follow-up request should include x-codex-turn-id"); + let second_turn_id = requests[2] + .header(TURN_ID_HEADER) + .expect("second turn request should include x-codex-turn-id"); + assert_eq!(first_turn_request_id, first_turn_follow_up_id); + assert_ne!(first_turn_request_id, second_turn_id); + Ok(()) } @@ -120,6 +133,16 @@ async fn websocket_turn_state_persists_within_turn_and_resets_after() -> Result< ); assert_eq!(handshakes[2].header(TURN_STATE_HEADER), None); + // First handshake is startup preconnect; no turn id is available yet. + assert_eq!(handshakes[0].header(TURN_ID_HEADER), None); + let first_turn_follow_up_id = handshakes[1] + .header(TURN_ID_HEADER) + .expect("turn-time reconnect should include x-codex-turn-id"); + let second_turn_id = handshakes[2] + .header(TURN_ID_HEADER) + .expect("second turn handshake should include x-codex-turn-id"); + assert_ne!(first_turn_follow_up_id, second_turn_id); + server.shutdown().await; Ok(()) }