From 61edaf10704bbf46487d90ad61b40f198f1f0c4e Mon Sep 17 00:00:00 2001 From: adrian Date: Mon, 20 Apr 2026 12:09:29 -0700 Subject: [PATCH] Revert "Persist and prewarm agent tasks per thread (#17978)" This reverts commit e5b52a3caa7a050c4df81570f5698d74793ef942. --- .../src/protocol/thread_history.rs | 4 +- .../app-server/tests/suite/v2/initialize.rs | 4 +- codex-rs/core/src/agent/control.rs | 1 - codex-rs/core/src/agent_identity.rs | 13 +- .../src/agent_identity/task_registration.rs | 42 +- .../core/src/session/agent_task_lifecycle.rs | 186 -------- codex-rs/core/src/session/mod.rs | 92 +++- .../src/session/rollout_reconstruction.rs | 5 +- codex-rs/core/src/session/session.rs | 4 +- codex-rs/core/src/session/tests.rs | 402 +----------------- codex-rs/core/src/state/session.rs | 8 +- codex-rs/core/src/state/session_tests.rs | 14 +- codex-rs/core/tests/suite/otel.rs | 2 - codex-rs/protocol/src/protocol.rs | 21 - codex-rs/rollout/src/list.rs | 4 - codex-rs/rollout/src/metadata.rs | 6 +- codex-rs/rollout/src/policy.rs | 7 +- codex-rs/rollout/src/recorder.rs | 4 - codex-rs/state/src/extract.rs | 8 +- codex-rs/state/src/runtime/threads.rs | 6 +- 20 files changed, 149 insertions(+), 684 deletions(-) delete mode 100644 codex-rs/core/src/session/agent_task_lifecycle.rs diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index 886d9c2557..ef6593d09c 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -232,9 +232,7 @@ impl ThreadHistoryBuilder { RolloutItem::EventMsg(event) => self.handle_event(event), RolloutItem::Compacted(payload) => self.handle_compacted(payload), RolloutItem::ResponseItem(item) => self.handle_response_item(item), - RolloutItem::TurnContext(_) - | RolloutItem::SessionMeta(_) - | RolloutItem::SessionState(_) => {} + RolloutItem::TurnContext(_) | RolloutItem::SessionMeta(_) => {} } } diff --git a/codex-rs/app-server/tests/suite/v2/initialize.rs b/codex-rs/app-server/tests/suite/v2/initialize.rs index 4fe5f2fbe7..165160468f 100644 --- a/codex-rs/app-server/tests/suite/v2/initialize.rs +++ b/codex-rs/app-server/tests/suite/v2/initialize.rs @@ -24,9 +24,7 @@ use std::time::Duration; use tempfile::TempDir; use tokio::time::timeout; -// This covers debug app-server process startup before the first JSON-RPC response, -// not expected steady-state request latency. -const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20); +const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); #[tokio::test] async fn initialize_uses_client_info_name_as_originator() -> Result<()> { diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 8004f6e6ed..d27f51d57d 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -116,7 +116,6 @@ fn keep_forked_rollout_item(item: &RolloutItem) -> bool { | ResponseItem::Compaction { .. } | ResponseItem::Other, ) => false, - RolloutItem::SessionState(_) => false, RolloutItem::Compacted(_) | RolloutItem::EventMsg(_) | RolloutItem::SessionMeta(_) diff --git a/codex-rs/core/src/agent_identity.rs b/codex-rs/core/src/agent_identity.rs index e7a087461e..8edcb7bb10 100644 --- a/codex-rs/core/src/agent_identity.rs +++ b/codex-rs/core/src/agent_identity.rs @@ -157,16 +157,14 @@ impl AgentIdentityManager { Ok(stored_identity) } - pub(crate) async fn task_matches_current_identity(&self, task: &RegisteredAgentTask) -> bool { + pub(crate) async fn task_matches_current_binding(&self, task: &RegisteredAgentTask) -> bool { if !self.feature_enabled { return false; } - self.current_stored_identity() + self.current_auth_binding() .await - .is_some_and(|stored_identity| { - stored_identity.agent_runtime_id == task.agent_runtime_id - }) + .is_some_and(|(_, binding)| task.matches_binding(&binding)) } async fn current_auth_binding(&self) -> Option<(CodexAuth, AgentIdentityBinding)> { @@ -183,11 +181,6 @@ impl AgentIdentityManager { binding.map(|binding| (auth, binding)) } - async fn current_stored_identity(&self) -> Option { - let (auth, binding) = self.current_auth_binding().await?; - self.load_stored_identity(&auth, &binding).ok().flatten() - } - async fn register_agent_identity( &self, binding: &AgentIdentityBinding, diff --git a/codex-rs/core/src/agent_identity/task_registration.rs b/codex-rs/core/src/agent_identity/task_registration.rs index 53bb272cbc..4fc5d51282 100644 --- a/codex-rs/core/src/agent_identity/task_registration.rs +++ b/codex-rs/core/src/agent_identity/task_registration.rs @@ -2,7 +2,6 @@ use std::time::Duration; use anyhow::Context; use anyhow::Result; -use codex_protocol::protocol::SessionAgentTask; use crypto_box::SecretKey as Curve25519SecretKey; use ed25519_dalek::Signer as _; use serde::Deserialize; @@ -17,6 +16,9 @@ const AGENT_TASK_REGISTRATION_TIMEOUT: Duration = Duration::from_secs(15); #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) struct RegisteredAgentTask { + pub(crate) binding_id: String, + pub(crate) chatgpt_account_id: String, + pub(crate) chatgpt_user_id: Option, pub(crate) agent_runtime_id: String, pub(crate) task_id: String, pub(crate) registered_at: String, @@ -80,6 +82,9 @@ impl AgentIdentityManager { .await .with_context(|| format!("failed to parse agent task response from {url}"))?; let registered_task = RegisteredAgentTask { + binding_id: stored_identity.binding_id.clone(), + chatgpt_account_id: stored_identity.chatgpt_account_id.clone(), + chatgpt_user_id: stored_identity.chatgpt_user_id.clone(), agent_runtime_id: stored_identity.agent_runtime_id.clone(), task_id: decrypt_task_id_response( &stored_identity, @@ -102,20 +107,18 @@ impl AgentIdentityManager { } impl RegisteredAgentTask { - pub(crate) fn to_session_agent_task(&self) -> SessionAgentTask { - SessionAgentTask { - agent_runtime_id: self.agent_runtime_id.clone(), - task_id: self.task_id.clone(), - registered_at: self.registered_at.clone(), - } + pub(super) fn matches_binding(&self, binding: &AgentIdentityBinding) -> bool { + binding.matches_parts( + &self.binding_id, + &self.chatgpt_account_id, + self.chatgpt_user_id.as_deref(), + ) } - pub(crate) fn from_session_agent_task(task: SessionAgentTask) -> Self { - Self { - agent_runtime_id: task.agent_runtime_id, - task_id: task.task_id, - registered_at: task.registered_at, - } + pub(crate) fn has_same_binding(&self, other: &Self) -> bool { + self.binding_id == other.binding_id + && self.chatgpt_account_id == other.chatgpt_account_id + && self.chatgpt_user_id == other.chatgpt_user_id } } @@ -239,6 +242,9 @@ mod tests { assert_eq!( task, RegisteredAgentTask { + binding_id: "chatgpt-account-account-123".to_string(), + chatgpt_account_id: "account-123".to_string(), + chatgpt_user_id: Some("user-123".to_string()), agent_runtime_id: "agent-123".to_string(), task_id: "task_123".to_string(), registered_at: task.registered_at.clone(), @@ -325,6 +331,9 @@ mod tests { assert_eq!( task, RegisteredAgentTask { + binding_id: "chatgpt-account-account-123".to_string(), + chatgpt_account_id: "account-123".to_string(), + chatgpt_user_id: Some("user-123".to_string()), agent_runtime_id: "agent-123".to_string(), task_id: "task_123".to_string(), registered_at: task.registered_at.clone(), @@ -333,7 +342,7 @@ mod tests { } #[tokio::test] - async fn task_matches_current_identity_rejects_stale_registered_identity() { + async fn task_matches_current_binding_rejects_stale_auth_binding() { let auth_manager = AuthManager::from_auth_for_testing(make_chatgpt_auth("account-456", Some("user-456"))); let manager = AgentIdentityManager::new_for_tests( @@ -343,12 +352,15 @@ mod tests { SessionSource::Cli, ); let task = RegisteredAgentTask { + binding_id: "chatgpt-account-account-123".to_string(), + chatgpt_account_id: "account-123".to_string(), + chatgpt_user_id: Some("user-123".to_string()), agent_runtime_id: "agent-123".to_string(), task_id: "task_123".to_string(), registered_at: "2026-03-23T12:00:00Z".to_string(), }; - assert!(!manager.task_matches_current_identity(&task).await); + assert!(!manager.task_matches_current_binding(&task).await); } async fn mount_human_biscuit( diff --git a/codex-rs/core/src/session/agent_task_lifecycle.rs b/codex-rs/core/src/session/agent_task_lifecycle.rs deleted file mode 100644 index f5c1c4c6e1..0000000000 --- a/codex-rs/core/src/session/agent_task_lifecycle.rs +++ /dev/null @@ -1,186 +0,0 @@ -use crate::agent_identity::RegisteredAgentTask; -use crate::session::session::Session; -use codex_protocol::protocol::RolloutItem; -use codex_protocol::protocol::SessionAgentTask; -use codex_protocol::protocol::SessionStateUpdate; -use tracing::debug; -use tracing::info; -use tracing::warn; - -impl Session { - pub(super) async fn maybe_prewarm_agent_task_registration(&self) { - // Startup task registration is best-effort: regular turns already retry on demand, and - // a prewarm failure should not shut down the session or block unrelated work. - if let Err(error) = self.ensure_agent_task_registered().await { - warn!( - error = %error, - "startup agent task prewarm failed; regular turns will retry registration" - ); - } - } - - fn latest_persisted_agent_task( - rollout_items: &[RolloutItem], - ) -> Option> { - rollout_items.iter().rev().find_map(|item| match item { - RolloutItem::SessionState(update) => Some(update.agent_task.clone()), - _ => None, - }) - } - - pub(super) async fn restore_persisted_agent_task(&self, rollout_items: &[RolloutItem]) { - let Some(agent_task_update) = Self::latest_persisted_agent_task(rollout_items) else { - return; - }; - - match agent_task_update { - Some(agent_task) => { - let registered_task = - RegisteredAgentTask::from_session_agent_task(agent_task.clone()); - if self - .services - .agent_identity_manager - .task_matches_current_identity(®istered_task) - .await - { - let mut state = self.state.lock().await; - state.set_agent_task(agent_task); - } else { - debug!( - agent_runtime_id = %registered_task.agent_runtime_id, - task_id = %registered_task.task_id, - "discarding persisted agent task because it does not match the registered agent identity" - ); - let mut state = self.state.lock().await; - state.clear_agent_task(); - } - } - None => { - let mut state = self.state.lock().await; - state.clear_agent_task(); - } - } - } - - async fn persist_agent_task_update(&self, agent_task: Option<&RegisteredAgentTask>) { - self.persist_rollout_items(&[RolloutItem::SessionState(SessionStateUpdate { - agent_task: agent_task.map(RegisteredAgentTask::to_session_agent_task), - })]) - .await; - } - - async fn clear_cached_agent_task(&self, agent_task: &RegisteredAgentTask) { - let cleared = { - let mut state = self.state.lock().await; - if state.agent_task().as_ref() == Some(&agent_task.to_session_agent_task()) { - state.clear_agent_task(); - true - } else { - false - } - }; - if cleared { - self.persist_agent_task_update(/*agent_task*/ None).await; - } - } - - async fn cache_agent_task(&self, agent_task: RegisteredAgentTask) -> RegisteredAgentTask { - let session_agent_task = agent_task.to_session_agent_task(); - let changed = { - let mut state = self.state.lock().await; - if state.agent_task().as_ref() == Some(&session_agent_task) { - false - } else { - state.set_agent_task(session_agent_task); - true - } - }; - if changed { - self.persist_agent_task_update(Some(&agent_task)).await; - } - agent_task - } - - pub(super) async fn cached_agent_task_for_current_identity( - &self, - ) -> Option { - let agent_task = { - let state = self.state.lock().await; - state - .agent_task() - .map(RegisteredAgentTask::from_session_agent_task) - }?; - - if self - .services - .agent_identity_manager - .task_matches_current_identity(&agent_task) - .await - { - debug!( - agent_runtime_id = %agent_task.agent_runtime_id, - task_id = %agent_task.task_id, - "reusing cached agent task" - ); - return Some(agent_task); - } - - debug!( - agent_runtime_id = %agent_task.agent_runtime_id, - task_id = %agent_task.task_id, - "discarding cached agent task because the registered agent identity changed" - ); - self.clear_cached_agent_task(&agent_task).await; - None - } - - pub(super) async fn ensure_agent_task_registered( - &self, - ) -> anyhow::Result> { - if let Some(agent_task) = self.cached_agent_task_for_current_identity().await { - return Ok(Some(agent_task)); - } - - let _guard = self - .agent_task_registration_lock - .acquire() - .await - .map_err(|_| anyhow::anyhow!("agent task registration semaphore closed"))?; - if let Some(agent_task) = self.cached_agent_task_for_current_identity().await { - return Ok(Some(agent_task)); - } - - for _ in 0..2 { - let Some(agent_task) = self.services.agent_identity_manager.register_task().await? - else { - return Ok(None); - }; - - if !self - .services - .agent_identity_manager - .task_matches_current_identity(&agent_task) - .await - { - debug!( - agent_runtime_id = %agent_task.agent_runtime_id, - task_id = %agent_task.task_id, - "discarding newly registered agent task because the registered agent identity changed" - ); - continue; - } - - let agent_task = self.cache_agent_task(agent_task).await; - - info!( - thread_id = %self.conversation_id, - agent_runtime_id = %agent_task.agent_runtime_id, - task_id = %agent_task.task_id, - "registered agent task for thread" - ); - return Ok(Some(agent_task)); - } - - Ok(None) - } -} diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index e90b9f831b..62c2b41243 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -14,6 +14,7 @@ use crate::agent::MailboxReceiver; use crate::agent::agent_status_from_event; use crate::agent::status::is_final; use crate::agent_identity::AgentIdentityManager; +use crate::agent_identity::RegisteredAgentTask; use crate::apps::render_apps_section; use crate::commit_attribution::commit_message_trailer_instruction; use crate::compact; @@ -164,7 +165,6 @@ use codex_protocol::error::Result as CodexResult; #[cfg(test)] use codex_protocol::exec_output::StreamOutput; -mod agent_task_lifecycle; mod handlers; mod mcp; mod review; @@ -992,10 +992,7 @@ impl Session { .ensure_registered_identity() .await { - Ok(Some(_)) => { - sess.maybe_prewarm_agent_task_registration().await; - return; - } + Ok(Some(_)) => return, Ok(None) => { drop(sess); if auth_state_rx.changed().await.is_err() { @@ -1026,6 +1023,90 @@ impl Session { .await; } + async fn cached_agent_task_for_current_binding(&self) -> Option { + let agent_task = { + let state = self.state.lock().await; + state.agent_task() + }?; + + if self + .services + .agent_identity_manager + .task_matches_current_binding(&agent_task) + .await + { + debug!( + agent_runtime_id = %agent_task.agent_runtime_id, + task_id = %agent_task.task_id, + "reusing cached agent task" + ); + return Some(agent_task); + } + + debug!( + agent_runtime_id = %agent_task.agent_runtime_id, + task_id = %agent_task.task_id, + "discarding cached agent task because auth binding changed" + ); + let mut state = self.state.lock().await; + if state.agent_task().as_ref() == Some(&agent_task) { + state.clear_agent_task(); + } + None + } + + async fn ensure_agent_task_registered(&self) -> anyhow::Result> { + if let Some(agent_task) = self.cached_agent_task_for_current_binding().await { + return Ok(Some(agent_task)); + } + + for _ in 0..2 { + let Some(agent_task) = self.services.agent_identity_manager.register_task().await? + else { + return Ok(None); + }; + + if !self + .services + .agent_identity_manager + .task_matches_current_binding(&agent_task) + .await + { + debug!( + agent_runtime_id = %agent_task.agent_runtime_id, + task_id = %agent_task.task_id, + "discarding newly registered agent task because auth binding changed" + ); + continue; + } + + { + let mut state = self.state.lock().await; + if let Some(existing_agent_task) = state.agent_task() { + if existing_agent_task.has_same_binding(&agent_task) { + return Ok(Some(existing_agent_task)); + } + debug!( + agent_runtime_id = %existing_agent_task.agent_runtime_id, + task_id = %existing_agent_task.task_id, + "replacing cached agent task because auth binding changed" + ); + } + state.set_agent_task(agent_task.clone()); + } + + info!( + thread_id = %self.conversation_id, + agent_runtime_id = %agent_task.agent_runtime_id, + task_id = %agent_task.task_id, + "registered agent task for thread" + ); + return Ok(Some(agent_task)); + } + + Ok(None) + } + pub(crate) fn get_tx_event(&self) -> Sender { self.tx_event.clone() } @@ -1173,7 +1254,6 @@ impl Session { } InitialHistory::Resumed(resumed_history) => { let rollout_items = resumed_history.history; - self.restore_persisted_agent_task(&rollout_items).await; let previous_turn_settings = self .apply_rollout_reconstruction(&turn_context, &rollout_items) .await; diff --git a/codex-rs/core/src/session/rollout_reconstruction.rs b/codex-rs/core/src/session/rollout_reconstruction.rs index 3e407c4cd7..a4c042af0c 100644 --- a/codex-rs/core/src/session/rollout_reconstruction.rs +++ b/codex-rs/core/src/session/rollout_reconstruction.rs @@ -207,9 +207,7 @@ impl Session { active_segment.get_or_insert_with(ActiveReplaySegment::default); active_segment.counts_as_user_turn |= is_user_turn_boundary(response_item); } - RolloutItem::EventMsg(_) - | RolloutItem::SessionMeta(_) - | RolloutItem::SessionState(_) => {} + RolloutItem::EventMsg(_) | RolloutItem::SessionMeta(_) => {} } if base_replacement_history.is_some() @@ -277,7 +275,6 @@ impl Session { history.drop_last_n_user_turns(rollback.num_turns); } RolloutItem::EventMsg(_) - | RolloutItem::SessionState(_) | RolloutItem::TurnContext(_) | RolloutItem::SessionMeta(_) => {} } diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 3bbb97dbcb..1f87373820 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -26,7 +26,6 @@ pub(crate) struct Session { pub(crate) services: SessionServices, pub(super) js_repl: Arc, pub(super) next_internal_sub_id: AtomicU64, - pub(super) agent_task_registration_lock: Semaphore, } #[derive(Clone)] @@ -709,7 +708,6 @@ impl Session { services, js_repl, next_internal_sub_id: AtomicU64::new(0), - agent_task_registration_lock: Semaphore::new(/*permits*/ 1), }); if let Some(network_policy_decider_session) = network_policy_decider_session { let mut guard = network_policy_decider_session.write().await; @@ -750,6 +748,7 @@ impl Session { // Start the watcher after SessionConfigured so it cannot emit earlier events. sess.start_skills_watcher_listener(); + sess.start_agent_identity_registration(); let mut required_mcp_servers: Vec = mcp_servers .iter() .filter(|(_, server)| server.enabled && server.required) @@ -836,7 +835,6 @@ impl Session { // record_initial_history can emit events. We record only after the SessionConfiguredEvent is emitted. sess.record_initial_history(initial_history).await; - sess.start_agent_identity_registration(); { let mut state = sess.state.lock().await; state.set_pending_session_start_source(Some(session_start_source)); diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index d5a3cbd37a..02ba035f5b 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -1,6 +1,4 @@ use super::*; -use crate::agent_identity::RegisteredAgentTask; -use crate::agent_identity::StoredAgentIdentity; use crate::config::ConfigBuilder; use crate::config::test_config; use crate::config_loader::ConfigLayerStack; @@ -17,20 +15,10 @@ use crate::shell::default_user_shell; use crate::skills::SkillRenderSideEffects; use crate::skills::render::SkillMetadataBudget; use crate::tools::format_exec_output_str; -use base64::Engine as _; -use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; -use base64::engine::general_purpose::URL_SAFE_NO_PAD; -use chrono::SecondsFormat; -use chrono::Utc; + use codex_features::Feature; use codex_features::Features; -use codex_login::AgentIdentityAuthRecord; -use codex_login::AuthCredentialsStoreMode; -use codex_login::AuthDotJson; use codex_login::CodexAuth; -use codex_login::save_auth; -use codex_login::token_data::IdTokenInfo; -use codex_login::token_data::TokenData; use codex_model_provider_info::ModelProviderInfo; use codex_models_manager::bundled_models_response; use codex_models_manager::model_info; @@ -129,9 +117,6 @@ use core_test_support::test_codex::test_codex; use core_test_support::test_path_buf; use core_test_support::tracing::install_test_tracing; use core_test_support::wait_for_event; -use crypto_box::SecretKey as Curve25519SecretKey; -use ed25519_dalek::SigningKey; -use ed25519_dalek::pkcs8::EncodePrivateKey; use opentelemetry::trace::TraceContextExt; use opentelemetry::trace::TraceId; use opentelemetry_sdk::metrics::InMemoryMetricExporter; @@ -139,20 +124,12 @@ use opentelemetry_sdk::metrics::data::AggregatedMetrics; use opentelemetry_sdk::metrics::data::Metric; use opentelemetry_sdk::metrics::data::MetricData; use opentelemetry_sdk::metrics::data::ResourceMetrics; -use sha2::Digest as _; -use sha2::Sha512; use std::path::Path; use std::time::Duration; use tokio::sync::Semaphore; use tokio::time::sleep; use tokio::time::timeout; use tracing_opentelemetry::OpenTelemetrySpanExt; -use wiremock::Mock; -use wiremock::MockServer; -use wiremock::ResponseTemplate; -use wiremock::matchers::header; -use wiremock::matchers::method; -use wiremock::matchers::path; use codex_protocol::mcp::CallToolResult as McpCallToolResult; use pretty_assertions::assert_eq; @@ -1291,120 +1268,6 @@ async fn record_initial_history_reconstructs_resumed_transcript() { assert_eq!(expected, history.raw_items()); } -#[tokio::test] -async fn record_initial_history_restores_latest_persisted_agent_task() { - let auth = make_chatgpt_auth("account-123", Some("user-123")); - seed_stored_identity(&auth, "agent-123", "account-123"); - let (session, _turn_context, _rx_event) = make_agent_identity_session_and_context_with_rx( - auth, - "https://chatgpt.com/backend-api".to_string(), - ) - .await; - let expected = RegisteredAgentTask { - agent_runtime_id: "agent-123".to_string(), - task_id: "task-123".to_string(), - registered_at: "2026-03-23T12:00:00Z".to_string(), - }; - let rollout_items = vec![ - RolloutItem::SessionState(codex_protocol::protocol::SessionStateUpdate { - agent_task: Some(expected.to_session_agent_task()), - }), - RolloutItem::SessionState(codex_protocol::protocol::SessionStateUpdate { - agent_task: None, - }), - RolloutItem::SessionState(codex_protocol::protocol::SessionStateUpdate { - agent_task: Some(expected.to_session_agent_task()), - }), - ]; - - session - .record_initial_history(InitialHistory::Resumed(ResumedHistory { - conversation_id: ThreadId::default(), - history: rollout_items, - rollout_path: PathBuf::from("/tmp/resume.jsonl"), - })) - .await; - - assert_eq!( - session.state.lock().await.agent_task(), - Some(expected.to_session_agent_task()) - ); -} - -#[tokio::test] -async fn record_initial_history_discards_persisted_agent_task_for_different_identity() { - let auth = make_chatgpt_auth("account-123", Some("user-123")); - seed_stored_identity(&auth, "agent-123", "account-123"); - let (session, _turn_context, _rx_event) = make_agent_identity_session_and_context_with_rx( - auth, - "https://chatgpt.com/backend-api".to_string(), - ) - .await; - let rollout_items = vec![RolloutItem::SessionState( - codex_protocol::protocol::SessionStateUpdate { - agent_task: Some( - RegisteredAgentTask { - agent_runtime_id: "agent-other".to_string(), - task_id: "task-other".to_string(), - registered_at: "2026-03-23T12:00:00Z".to_string(), - } - .to_session_agent_task(), - ), - }, - )]; - - session - .record_initial_history(InitialHistory::Resumed(ResumedHistory { - conversation_id: ThreadId::default(), - history: rollout_items, - rollout_path: PathBuf::from("/tmp/resume.jsonl"), - })) - .await; - - assert_eq!(session.state.lock().await.agent_task(), None); -} - -#[tokio::test] -async fn record_initial_history_honors_cleared_persisted_agent_task() { - let (session, _turn_context) = make_session_and_context().await; - { - let mut state = session.state.lock().await; - state.set_agent_task( - RegisteredAgentTask { - agent_runtime_id: "agent-fresh".to_string(), - task_id: "task-fresh".to_string(), - registered_at: "2026-03-23T12:01:00Z".to_string(), - } - .to_session_agent_task(), - ); - } - let rollout_items = vec![ - RolloutItem::SessionState(codex_protocol::protocol::SessionStateUpdate { - agent_task: Some( - RegisteredAgentTask { - agent_runtime_id: "agent-123".to_string(), - task_id: "task-123".to_string(), - registered_at: "2026-03-23T12:00:00Z".to_string(), - } - .to_session_agent_task(), - ), - }), - RolloutItem::SessionState(codex_protocol::protocol::SessionStateUpdate { - agent_task: None, - }), - ]; - - session - .record_initial_history(InitialHistory::Resumed(ResumedHistory { - conversation_id: ThreadId::default(), - history: rollout_items, - rollout_path: PathBuf::from("/tmp/resume.jsonl"), - })) - .await; - - assert_eq!(session.state.lock().await.agent_task(), None); -} - #[tokio::test] async fn record_initial_history_new_defers_initial_context_until_first_turn() { let (session, _turn_context) = make_session_and_context().await; @@ -3306,7 +3169,6 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { services, js_repl, next_internal_sub_id: AtomicU64::new(0), - agent_task_registration_lock: Semaphore::new(/*permits*/ 1), }; (session, turn_context) @@ -4056,25 +3918,19 @@ async fn shutdown_and_wait_shuts_down_tracked_ephemeral_guardian_review() { .expect("ephemeral guardian review should receive a shutdown op"); } -async fn make_session_and_context_with_auth_and_config_and_rx( - auth: CodexAuth, +pub(crate) async fn make_session_and_context_with_dynamic_tools_and_rx( dynamic_tools: Vec, - configure_config: F, ) -> ( Arc, Arc, async_channel::Receiver, -) -where - F: FnOnce(&mut Config), -{ +) { let (tx_event, rx_event) = async_channel::unbounded(); let codex_home = tempfile::tempdir().expect("create temp dir"); - let mut config = build_test_config(codex_home.path()).await; - configure_config(&mut config); + let config = build_test_config(codex_home.path()).await; let config = Arc::new(config); let conversation_id = ThreadId::default(); - let auth_manager = AuthManager::from_auth_for_testing(auth); + let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("Test API Key")); let models_manager = Arc::new(ModelsManager::new( config.codex_home.to_path_buf(), auth_manager.clone(), @@ -4276,45 +4132,11 @@ where services, js_repl, next_internal_sub_id: AtomicU64::new(0), - agent_task_registration_lock: Semaphore::new(/*permits*/ 1), }); (session, turn_context, rx_event) } -pub(crate) async fn make_session_and_context_with_dynamic_tools_and_rx( - dynamic_tools: Vec, -) -> ( - Arc, - Arc, - async_channel::Receiver, -) { - make_session_and_context_with_auth_and_config_and_rx( - CodexAuth::from_api_key("Test API Key"), - dynamic_tools, - |_config| {}, - ) - .await -} - -async fn make_agent_identity_session_and_context_with_rx( - auth: CodexAuth, - chatgpt_base_url: String, -) -> ( - Arc, - Arc, - async_channel::Receiver, -) { - make_session_and_context_with_auth_and_config_and_rx(auth, Vec::new(), move |config| { - config.chatgpt_base_url = chatgpt_base_url; - config - .features - .enable(Feature::UseAgentIdentity) - .expect("test config should allow use_agent_identity"); - }) - .await -} - // Like make_session_and_context, but returns Arc and the event receiver // so tests can assert on emitted events. pub(crate) async fn make_session_and_context_with_rx() -> ( @@ -4354,220 +4176,6 @@ async fn fail_agent_identity_registration_emits_error_without_shutdown() { assert!(rx_event.try_recv().is_err()); } -#[tokio::test] -async fn startup_agent_task_prewarm_caches_registered_task() { - let server = MockServer::start().await; - let chatgpt_base_url = server.uri(); - let auth = make_chatgpt_auth("account-123", Some("user-123")); - let stored_identity = seed_stored_identity(&auth, "agent-123", "account-123"); - let encrypted_task_id = - encrypt_task_id_for_identity(&stored_identity, "task_123").expect("task ciphertext"); - mount_human_biscuit(&server, &chatgpt_base_url, "agent-123").await; - Mock::given(method("POST")) - .and(path("/v1/agent/agent-123/task/register")) - .and(header("x-openai-authorization", "human-biscuit")) - .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ - "encrypted_task_id": encrypted_task_id, - }))) - .expect(1) - .mount(&server) - .await; - - let (session, _turn_context, rx_event) = - make_agent_identity_session_and_context_with_rx(auth, chatgpt_base_url).await; - - session.maybe_prewarm_agent_task_registration().await; - - let cached_task = session - .state - .lock() - .await - .agent_task() - .expect("task should be cached"); - assert_eq!(cached_task.agent_runtime_id, "agent-123"); - assert_eq!(cached_task.task_id, "task_123"); - assert!(rx_event.try_recv().is_err()); -} - -#[tokio::test] -async fn startup_agent_task_prewarm_failure_does_not_emit_error() { - let server = MockServer::start().await; - let chatgpt_base_url = server.uri(); - let auth = make_chatgpt_auth("account-123", Some("user-123")); - seed_stored_identity(&auth, "agent-123", "account-123"); - mount_human_biscuit(&server, &chatgpt_base_url, "agent-123").await; - Mock::given(method("POST")) - .and(path("/v1/agent/agent-123/task/register")) - .and(header("x-openai-authorization", "human-biscuit")) - .respond_with(ResponseTemplate::new(500)) - .expect(1) - .mount(&server) - .await; - - let (session, _turn_context, rx_event) = - make_agent_identity_session_and_context_with_rx(auth, chatgpt_base_url).await; - - session.maybe_prewarm_agent_task_registration().await; - - assert_eq!(session.state.lock().await.agent_task(), None); - assert!(rx_event.try_recv().is_err()); -} - -#[tokio::test] -async fn cached_agent_task_for_current_identity_clears_stale_task() { - let auth = make_chatgpt_auth("account-123", Some("user-123")); - seed_stored_identity(&auth, "agent-123", "account-123"); - let (session, _turn_context, _rx_event) = make_agent_identity_session_and_context_with_rx( - auth, - "https://chatgpt.com/backend-api".to_string(), - ) - .await; - { - let mut state = session.state.lock().await; - state.set_agent_task( - RegisteredAgentTask { - agent_runtime_id: "agent-old".to_string(), - task_id: "task-old".to_string(), - registered_at: "2026-04-15T00:00:00Z".to_string(), - } - .to_session_agent_task(), - ); - } - - assert_eq!(session.cached_agent_task_for_current_identity().await, None); - assert_eq!(session.state.lock().await.agent_task(), None); -} - -fn seed_stored_identity( - auth: &CodexAuth, - agent_runtime_id: &str, - account_id: &str, -) -> StoredAgentIdentity { - let signing_key = generate_test_signing_key(); - let private_key_pkcs8 = signing_key - .to_pkcs8_der() - .expect("encode test signing key as PKCS#8"); - let stored_identity = StoredAgentIdentity { - binding_id: format!("chatgpt-account-{account_id}"), - chatgpt_account_id: account_id.to_string(), - chatgpt_user_id: Some("user-123".to_string()), - agent_runtime_id: agent_runtime_id.to_string(), - private_key_pkcs8_base64: BASE64_STANDARD.encode(private_key_pkcs8.as_bytes()), - public_key_ssh: "ssh-ed25519 test".to_string(), - registered_at: Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true), - abom: crate::agent_identity::AgentBillOfMaterials { - agent_version: env!("CARGO_PKG_VERSION").to_string(), - agent_harness_id: "codex-cli".to_string(), - running_location: format!("{}-{}", SessionSource::Exec, std::env::consts::OS), - }, - }; - - auth.set_agent_identity(AgentIdentityAuthRecord { - workspace_id: account_id.to_string(), - chatgpt_user_id: stored_identity.chatgpt_user_id.clone(), - agent_runtime_id: stored_identity.agent_runtime_id.clone(), - agent_private_key: stored_identity.private_key_pkcs8_base64.clone(), - registered_at: stored_identity.registered_at.clone(), - }) - .expect("store identity"); - - stored_identity -} - -fn encrypt_task_id_for_identity( - stored_identity: &StoredAgentIdentity, - task_id: &str, -) -> anyhow::Result { - let signing_key = stored_identity.signing_key()?; - let mut rng = crypto_box::aead::OsRng; - let public_key = curve25519_secret_key_from_signing_key_for_tests(&signing_key).public_key(); - let ciphertext = public_key - .seal(&mut rng, task_id.as_bytes()) - .map_err(|_| anyhow::anyhow!("failed to encrypt test task id"))?; - Ok(BASE64_STANDARD.encode(ciphertext)) -} - -fn curve25519_secret_key_from_signing_key_for_tests( - signing_key: &SigningKey, -) -> Curve25519SecretKey { - let digest = Sha512::digest(signing_key.to_bytes()); - let mut secret_key = [0u8; 32]; - secret_key.copy_from_slice(&digest[..32]); - secret_key[0] &= 248; - secret_key[31] &= 127; - secret_key[31] |= 64; - Curve25519SecretKey::from(secret_key) -} - -fn generate_test_signing_key() -> SigningKey { - SigningKey::from_bytes(&[7u8; 32]) -} - -async fn mount_human_biscuit(server: &MockServer, chatgpt_base_url: &str, agent_runtime_id: &str) { - let biscuit_url = format!( - "{}/authenticate_app_v2", - chatgpt_base_url.trim_end_matches('/') - ); - let biscuit_path = reqwest::Url::parse(&biscuit_url) - .expect("biscuit URL parses") - .path() - .to_string(); - let target_url = format!( - "{}/v1/agent/{agent_runtime_id}/task/register", - chatgpt_base_url.trim_end_matches('/') - ); - Mock::given(method("GET")) - .and(path(biscuit_path)) - .and(header("authorization", "Bearer access-token-account-123")) - .and(header("x-original-method", "POST")) - .and(header("x-original-url", target_url)) - .respond_with( - ResponseTemplate::new(200).insert_header("x-openai-authorization", "human-biscuit"), - ) - .expect(1) - .mount(server) - .await; -} - -fn make_chatgpt_auth(account_id: &str, user_id: Option<&str>) -> CodexAuth { - let tempdir = tempfile::tempdir().expect("tempdir"); - let auth_json = AuthDotJson { - auth_mode: Some(codex_app_server_protocol::AuthMode::Chatgpt), - openai_api_key: None, - tokens: Some(TokenData { - id_token: IdTokenInfo { - email: None, - chatgpt_plan_type: None, - chatgpt_user_id: user_id.map(ToOwned::to_owned), - chatgpt_account_id: Some(account_id.to_string()), - chatgpt_account_is_fedramp: false, - raw_jwt: fake_id_token(account_id, user_id), - }, - access_token: format!("access-token-{account_id}"), - refresh_token: "refresh-token".to_string(), - account_id: Some(account_id.to_string()), - }), - last_refresh: Some(Utc::now()), - agent_identity: None, - }; - save_auth(tempdir.path(), &auth_json, AuthCredentialsStoreMode::File).expect("save auth"); - CodexAuth::from_auth_storage(tempdir.path(), AuthCredentialsStoreMode::File) - .expect("load auth") - .expect("auth") -} - -fn fake_id_token(account_id: &str, user_id: Option<&str>) -> String { - let header = URL_SAFE_NO_PAD.encode(r#"{"alg":"none","typ":"JWT"}"#); - let payload = serde_json::json!({ - "https://api.openai.com/auth": { - "chatgpt_user_id": user_id, - "chatgpt_account_id": account_id, - } - }); - let payload = URL_SAFE_NO_PAD.encode(payload.to_string()); - format!("{header}.{payload}.signature") -} - #[tokio::test] async fn refresh_mcp_servers_is_deferred_until_next_turn() { let (session, turn_context) = make_session_and_context().await; diff --git a/codex-rs/core/src/state/session.rs b/codex-rs/core/src/state/session.rs index 4ac8c4d530..bf6e06d8a9 100644 --- a/codex-rs/core/src/state/session.rs +++ b/codex-rs/core/src/state/session.rs @@ -6,12 +6,12 @@ use codex_sandboxing::policy_transforms::merge_permission_profiles; use std::collections::HashMap; use std::collections::HashSet; +use crate::agent_identity::RegisteredAgentTask; use crate::context_manager::ContextManager; use crate::session::PreviousTurnSettings; use crate::session::session::SessionConfiguration; use crate::session_startup_prewarm::SessionStartupPrewarmHandle; use codex_protocol::protocol::RateLimitSnapshot; -use codex_protocol::protocol::SessionAgentTask; use codex_protocol::protocol::TokenUsage; use codex_protocol::protocol::TokenUsageInfo; use codex_protocol::protocol::TurnContextItem; @@ -31,7 +31,7 @@ pub(crate) struct SessionState { previous_turn_settings: Option, /// Startup prewarmed session prepared during session initialization. pub(crate) startup_prewarm: Option, - pub(crate) agent_task: Option, + pub(crate) agent_task: Option, pub(crate) active_connector_selection: HashSet, pub(crate) pending_session_start_source: Option, granted_permissions: Option, @@ -189,11 +189,11 @@ impl SessionState { self.startup_prewarm.take() } - pub(crate) fn agent_task(&self) -> Option { + pub(crate) fn agent_task(&self) -> Option { self.agent_task.clone() } - pub(crate) fn set_agent_task(&mut self, agent_task: SessionAgentTask) { + pub(crate) fn set_agent_task(&mut self, agent_task: RegisteredAgentTask) { self.agent_task = Some(agent_task); } diff --git a/codex-rs/core/src/state/session_tests.rs b/codex-rs/core/src/state/session_tests.rs index 12229ee2f2..927c1fbb70 100644 --- a/codex-rs/core/src/state/session_tests.rs +++ b/codex-rs/core/src/state/session_tests.rs @@ -1,8 +1,8 @@ use super::*; +use crate::agent_identity::RegisteredAgentTask; use crate::session::tests::make_session_configuration_for_tests; use codex_protocol::protocol::CreditsSnapshot; use codex_protocol::protocol::RateLimitWindow; -use codex_protocol::protocol::SessionAgentTask; use pretty_assertions::assert_eq; #[tokio::test] @@ -38,8 +38,11 @@ async fn clear_connector_selection_removes_entries() { async fn set_agent_task_persists_plaintext_task_for_session_reuse() { let session_configuration = make_session_configuration_for_tests().await; let mut state = SessionState::new(session_configuration); - let agent_task = SessionAgentTask { - agent_runtime_id: "agent-123".to_string(), + let agent_task = RegisteredAgentTask { + binding_id: "chatgpt-account-account-123".to_string(), + chatgpt_account_id: "account-123".to_string(), + chatgpt_user_id: Some("user-123".to_string()), + agent_runtime_id: "agent_123".to_string(), task_id: "task_123".to_string(), registered_at: "2026-03-23T12:00:00Z".to_string(), }; @@ -53,7 +56,10 @@ async fn set_agent_task_persists_plaintext_task_for_session_reuse() { async fn clear_agent_task_removes_cached_task() { let session_configuration = make_session_configuration_for_tests().await; let mut state = SessionState::new(session_configuration); - let agent_task = SessionAgentTask { + let agent_task = RegisteredAgentTask { + binding_id: "chatgpt-account-account-123".to_string(), + chatgpt_account_id: "account-123".to_string(), + chatgpt_user_id: Some("user-123".to_string()), agent_runtime_id: "agent_123".to_string(), task_id: "task_123".to_string(), registered_at: "2026-03-23T12:00:00Z".to_string(), diff --git a/codex-rs/core/tests/suite/otel.rs b/codex-rs/core/tests/suite/otel.rs index f93945e78f..d771ed19ac 100644 --- a/codex-rs/core/tests/suite/otel.rs +++ b/codex-rs/core/tests/suite/otel.rs @@ -228,7 +228,6 @@ async fn process_sse_records_failed_event_when_stream_closes_without_completed() .features .disable(Feature::GhostCommit) .expect("test config should allow feature update"); - config.model_provider.stream_max_retries = Some(0); }) .build(&server) .await @@ -407,7 +406,6 @@ async fn process_sse_failed_event_logs_missing_error() { .features .disable(Feature::GhostCommit) .expect("test config should allow feature update"); - config.model_provider.stream_max_retries = Some(0); }) .build(&server) .await diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index ec1fc847a2..4c50c48459 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -2768,26 +2768,6 @@ impl fmt::Display for SubAgentSource { } } -/// Persisted agent-task details that let a resumed thread keep using the same backend task. -/// -/// `agent_runtime_id` is validation metadata for the globally registered agent identity, not a -/// separate session-scoped identity. Resume only restores this task after confirming that runtime -/// id still matches the globally registered identity; otherwise the cached task is discarded and a -/// fresh task can be registered. -#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema, TS)] -pub struct SessionAgentTask { - pub agent_runtime_id: String, - pub task_id: String, - pub registered_at: String, -} - -/// Session-scoped state updates that can be appended after the canonical SessionMeta line. -#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema, TS, Default)] -pub struct SessionStateUpdate { - #[serde(default)] - pub agent_task: Option, -} - /// SessionMeta contains session-level data that doesn't correspond to a specific turn. /// /// NOTE: There used to be an `instructions` field here, which stored user_instructions, but we @@ -2857,7 +2837,6 @@ pub struct SessionMetaLine { #[serde(tag = "type", content = "payload", rename_all = "snake_case")] pub enum RolloutItem { SessionMeta(SessionMetaLine), - SessionState(SessionStateUpdate), ResponseItem(ResponseItem), Compacted(CompactedItem), TurnContext(TurnContextItem), diff --git a/codex-rs/rollout/src/list.rs b/codex-rs/rollout/src/list.rs index 0df1b9c9f5..bf09d15e86 100644 --- a/codex-rs/rollout/src/list.rs +++ b/codex-rs/rollout/src/list.rs @@ -1082,9 +1082,6 @@ async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result { // Not included in `head`; skip. } - RolloutItem::SessionState(_) => { - // Not included in `head`; skip. - } RolloutItem::EventMsg(ev) => { if let EventMsg::UserMessage(user) = ev { summary.saw_user_event = true; @@ -1137,7 +1134,6 @@ pub async fn read_head_for_summary(path: &Path) -> io::Result {} } diff --git a/codex-rs/rollout/src/metadata.rs b/codex-rs/rollout/src/metadata.rs index 65e347a323..58d55a887d 100644 --- a/codex-rs/rollout/src/metadata.rs +++ b/codex-rs/rollout/src/metadata.rs @@ -70,8 +70,7 @@ pub fn builder_from_items( ) -> Option { if let Some(session_meta) = items.iter().find_map(|item| match item { RolloutItem::SessionMeta(meta_line) => Some(meta_line), - RolloutItem::SessionState(_) - | RolloutItem::ResponseItem(_) + RolloutItem::ResponseItem(_) | RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::EventMsg(_) => None, @@ -125,8 +124,7 @@ pub async fn extract_metadata_from_rollout( metadata, memory_mode: items.iter().rev().find_map(|item| match item { RolloutItem::SessionMeta(meta_line) => meta_line.meta.memory_mode.clone(), - RolloutItem::SessionState(_) - | RolloutItem::ResponseItem(_) + RolloutItem::ResponseItem(_) | RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::EventMsg(_) => None, diff --git a/codex-rs/rollout/src/policy.rs b/codex-rs/rollout/src/policy.rs index 86fc950a24..df41d2c766 100644 --- a/codex-rs/rollout/src/policy.rs +++ b/codex-rs/rollout/src/policy.rs @@ -16,10 +16,9 @@ pub fn is_persisted_response_item(item: &RolloutItem, mode: EventPersistenceMode RolloutItem::ResponseItem(item) => should_persist_response_item(item), RolloutItem::EventMsg(ev) => should_persist_event_msg(ev, mode), // Persist Codex executive markers so we can analyze flows (e.g., compaction, API turns). - RolloutItem::Compacted(_) - | RolloutItem::TurnContext(_) - | RolloutItem::SessionMeta(_) - | RolloutItem::SessionState(_) => true, + RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::SessionMeta(_) => { + true + } } } diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index f6e15a973e..871d53141b 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -707,9 +707,6 @@ impl RolloutRecorder { RolloutItem::Compacted(item) => { items.push(RolloutItem::Compacted(item)); } - RolloutItem::SessionState(update) => { - items.push(RolloutItem::SessionState(update)); - } RolloutItem::TurnContext(item) => { items.push(RolloutItem::TurnContext(item)); } @@ -1566,7 +1563,6 @@ async fn resume_candidate_matches_cwd( && let Some(latest_turn_context_cwd) = items.iter().rev().find_map(|item| match item { RolloutItem::TurnContext(turn_context) => Some(turn_context.cwd.as_path()), RolloutItem::SessionMeta(_) - | RolloutItem::SessionState(_) | RolloutItem::ResponseItem(_) | RolloutItem::Compacted(_) | RolloutItem::EventMsg(_) => None, diff --git a/codex-rs/state/src/extract.rs b/codex-rs/state/src/extract.rs index 64d7f5bbb7..6d36f211a4 100644 --- a/codex-rs/state/src/extract.rs +++ b/codex-rs/state/src/extract.rs @@ -19,7 +19,6 @@ pub fn apply_rollout_item( ) { match item { RolloutItem::SessionMeta(meta_line) => apply_session_meta_from_item(metadata, meta_line), - RolloutItem::SessionState(_) => {} RolloutItem::TurnContext(turn_ctx) => apply_turn_context(metadata, turn_ctx), RolloutItem::EventMsg(event) => apply_event_msg(metadata, event), RolloutItem::ResponseItem(item) => apply_response_item(metadata, item), @@ -37,10 +36,9 @@ pub fn rollout_item_affects_thread_metadata(item: &RolloutItem) -> bool { RolloutItem::EventMsg( EventMsg::TokenCount(_) | EventMsg::UserMessage(_) | EventMsg::ThreadNameUpdated(_), ) => true, - RolloutItem::SessionState(_) - | RolloutItem::EventMsg(_) - | RolloutItem::ResponseItem(_) - | RolloutItem::Compacted(_) => false, + RolloutItem::EventMsg(_) | RolloutItem::ResponseItem(_) | RolloutItem::Compacted(_) => { + false + } } } diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index 5592ca54df..eb7494a037 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -954,8 +954,7 @@ SELECT pub(super) fn extract_dynamic_tools(items: &[RolloutItem]) -> Option>> { items.iter().find_map(|item| match item { RolloutItem::SessionMeta(meta_line) => Some(meta_line.meta.dynamic_tools.clone()), - RolloutItem::SessionState(_) - | RolloutItem::ResponseItem(_) + RolloutItem::ResponseItem(_) | RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::EventMsg(_) => None, @@ -965,8 +964,7 @@ pub(super) fn extract_dynamic_tools(items: &[RolloutItem]) -> Option Option { items.iter().rev().find_map(|item| match item { RolloutItem::SessionMeta(meta_line) => meta_line.meta.memory_mode.clone(), - RolloutItem::SessionState(_) - | RolloutItem::ResponseItem(_) + RolloutItem::ResponseItem(_) | RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::EventMsg(_) => None,