use super::*; use crate::agent::control::SpawnAgentOptions; use crate::config::test_config; use crate::init_state_db; use crate::installation_id::INSTALLATION_ID_FILENAME; use crate::mcp::McpEnvironmentScope; use crate::mcp::McpThreadIdentity; use crate::rollout::RolloutRecorder; use crate::session::session::SessionSettingsUpdate; use crate::session::tests::build_world_state_from_turn_context; use crate::session::tests::make_session_and_context; use crate::tasks::InterruptedTurnHistoryMarker; use crate::tasks::interrupted_turn_history_marker; use crate::windows_sandbox::WindowsSandboxLevelExt; use codex_extension_api::empty_extension_registry; use codex_history::InitialHistory; use codex_history::ResumedHistory; use codex_models_manager::manager::RefreshStrategy; use codex_protocol::ResponseItemId; use codex_protocol::capabilities::CapabilityRootLocation; use codex_protocol::capabilities::SelectedCapabilityRoot; use codex_protocol::config_types::WindowsSandboxLevel; use codex_protocol::mcp::ClientMcpExtensions; use codex_protocol::mcp::MCP_APP_UI_EXTENSION_ID; use codex_protocol::mcp::OPENAI_FORM_EXTENSION_ID; use codex_protocol::models::ContentItem; use codex_protocol::models::ContentItemKind; use codex_protocol::models::InternalChatMessageMetadataPassthrough; use codex_protocol::models::ReasoningItemReasoningSummary; use codex_protocol::models::ResponseItem; use codex_protocol::openai_models::ModelsResponse; use codex_protocol::protocol::AgentMessageEvent; use codex_protocol::protocol::EnvironmentConfigState; use codex_protocol::protocol::InternalSessionSource; use codex_protocol::protocol::SessionMeta; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::ThreadSource; use codex_protocol::protocol::TurnStartedEvent; use codex_protocol::protocol::UserMessageEvent; use codex_protocol::user_input::UserInput; use codex_utils_path_uri::PathUri; use core_test_support::PathBufExt; use core_test_support::PathExt; use core_test_support::responses::mount_models_once; use core_test_support::responses::strip_response_item_ids_from_json; use pretty_assertions::assert_eq; use std::time::Duration; use tempfile::tempdir; use wiremock::MockServer; const TEST_INSTALLATION_ID: &str = "11111111-1111-4111-8111-111111111111"; /// Controls without a custom allocation policy still produce distinct thread identifiers. #[test] fn thread_id_generator_defaults_to_standard_ids() { let agent_control = AgentControl::default(); assert_ne!( agent_control.generate_thread_id(), agent_control.generate_thread_id() ); } #[tokio::test] async fn reserved_thread_id_is_used_without_changing_normal_id_generation() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let generated_ids = [ ThreadId::from_u128(/*value*/ 0x018f_0000_0000_7000_8000_0000_0000_0001), ThreadId::from_u128(/*value*/ 0x018f_0000_0000_7000_8000_0000_0000_0002), ThreadId::from_u128(/*value*/ 0x018f_0000_0000_7000_8000_0000_0000_0003), ]; let next_id = std::sync::atomic::AtomicUsize::new(0); let manager = ThreadManager::with_models_provider_and_home_for_tests( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), config.codex_home.to_path_buf(), Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ) .with_thread_id_generator(move || generated_ids[next_id.fetch_add(1, Ordering::Relaxed)]); let reserved_id = manager.reserve_thread_id(); let mut reserved_options = StartThreadOptions::new(config.clone()); reserved_options.reserved_thread_id = Some(reserved_id); let reserved = manager .start_thread(reserved_options) .await .expect("start reserved thread"); let mut resumed_options = StartThreadOptions::new(config.clone()); resumed_options.initial_history = InitialHistory::Resumed(ResumedHistory { conversation_id: reserved.thread_id, history: Arc::new(Vec::new()), rollout_path: None, }); let resumed_id = manager.reserve_thread_id(); resumed_options.reserved_thread_id = Some(resumed_id); let resume_error = manager .start_thread(resumed_options) .await .err() .expect("reject reserved ID for resume"); let generated = manager .start_thread(StartThreadOptions::new(config.clone())) .await .expect("start generated thread"); assert_eq!(reserved.thread_id, generated_ids[0]); assert!(matches!( resume_error.details(), codex_protocol::error::CodexErrorDetails::InvalidRequest(message) if message == "reserved thread ID cannot be used when resuming a thread" )); assert_eq!(generated.thread_id, generated_ids[2]); } /// One custom ID factory supplies identifiers for roots, actual child agents, and forks. #[tokio::test] async fn thread_id_generator_applies_to_roots_children_and_forks() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let generated_ids = [ ThreadId::from_u128(/*value*/ 0x018f_0000_0000_7000_8000_0000_0000_0001), ThreadId::from_u128(/*value*/ 0x018f_0000_0000_7000_8000_0000_0000_0002), ThreadId::from_u128(/*value*/ 0x018f_0000_0000_7000_8000_0000_0000_0003), ]; let next_id = std::sync::atomic::AtomicUsize::new(0); let manager = ThreadManager::with_models_provider_and_home_for_tests( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), config.codex_home.to_path_buf(), Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ) .with_thread_id_generator(move || generated_ids[next_id.fetch_add(1, Ordering::Relaxed)]); let root = manager .start_thread(StartThreadOptions::new(config.clone())) .await .expect("start root thread"); let child = root .thread .session .services .agent_control .spawn_agent_with_metadata( config.clone(), vec![UserInput::Text { text: "child task".to_string(), text_elements: Vec::new(), }], Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: root.thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: None, })), SpawnAgentOptions { parent_thread_id: Some(root.thread_id), ..Default::default() }, ) .await .expect("spawn actual child agent"); let fork = manager .spawn_subagent(root.thread_id, StartThreadOptions::new(config)) .await .expect("fork root thread"); assert_eq!( [root.thread_id, child.thread_id, fork.thread_id], generated_ids ); let report = manager .shutdown_all_threads_bounded(Duration::from_secs(10)) .await; assert_eq!(report.completed.len(), 3); } /// Resuming a thread preserves its stored ID instead of invoking the new manager's factory. #[tokio::test] async fn thread_id_generator_does_not_replace_resumed_thread_id() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let original_thread_id = ThreadId::from_u128(/*value*/ 0x018f_0000_0000_7000_8000_0000_0000_0001); let original_manager = ThreadManager::with_models_provider_and_home_for_tests( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), config.codex_home.to_path_buf(), Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ) .with_thread_id_generator(move || original_thread_id); let original = original_manager .start_thread(StartThreadOptions::new(config.clone())) .await .expect("start source thread"); original.thread.ensure_rollout_materialized().await; original .thread .flush_rollout() .await .expect("flush source rollout"); let rollout_path = original .thread .rollout_path() .expect("source rollout path should exist"); assert_eq!(original.thread_id, original_thread_id); original .thread .shutdown_and_wait() .await .expect("shut down source thread"); let _ = original_manager.remove_thread(&original_thread_id).await; let resumed_manager = ThreadManager::with_models_provider_and_home_for_tests( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), config.codex_home.to_path_buf(), Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ) .with_thread_id_generator(|| panic!("resuming must not allocate a new thread ID")); let resumed = resumed_manager .resume_thread_from_rollout( config, rollout_path, Arc::clone(&resumed_manager.state.auth_manager), /*parent_trace*/ None, ClientMcpExtensions::default(), ) .await .expect("resume existing source thread"); assert_eq!(resumed.thread_id, original_thread_id); resumed .thread .shutdown_and_wait() .await .expect("shut down resumed thread"); } #[tokio::test] async fn child_session_inherits_client_mcp_extensions() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let manager = ThreadManager::with_models_provider_and_home_for_tests( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), config.codex_home.to_path_buf(), Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ); let parent = manager .start_thread(StartThreadOptions { client_mcp_extensions: ClientMcpExtensions::new(HashMap::from([ (OPENAI_FORM_EXTENSION_ID.to_string(), serde_json::json!({})), ( MCP_APP_UI_EXTENSION_ID.to_string(), serde_json::json!({ "mimeTypes": ["text/html;profile=mcp-app"], }), ), ])), ..StartThreadOptions::new(config) }) .await .expect("start parent thread"); assert_eq!( manager .state .client_mcp_extensions_for_child(Some(parent.thread_id)) .await, ClientMcpExtensions::new(HashMap::from([ (OPENAI_FORM_EXTENSION_ID.to_string(), serde_json::json!({})), ( MCP_APP_UI_EXTENSION_ID.to_string(), serde_json::json!({ "mimeTypes": ["text/html;profile=mcp-app"], }), ), ])) ); } struct FakeAgentGraphStore { root_thread_id: ThreadId, descendant_thread_ids: Vec, } impl codex_agent_graph_store::AgentGraphStore for FakeAgentGraphStore { fn upsert_thread_spawn_edge( &self, _parent_thread_id: ThreadId, _child_thread_id: ThreadId, _status: codex_agent_graph_store::ThreadSpawnEdgeStatus, ) -> codex_agent_graph_store::AgentGraphStoreFuture<'_, ()> { Box::pin(async { panic!("unexpected graph upsert") }) } fn set_thread_spawn_edge_status( &self, _child_thread_id: ThreadId, _status: codex_agent_graph_store::ThreadSpawnEdgeStatus, ) -> codex_agent_graph_store::AgentGraphStoreFuture<'_, ()> { Box::pin(async { panic!("unexpected graph status update") }) } fn list_thread_spawn_children( &self, _parent_thread_id: ThreadId, _status_filter: Option, ) -> codex_agent_graph_store::AgentGraphStoreFuture<'_, Vec> { Box::pin(async { panic!("unexpected direct-child listing") }) } fn list_thread_spawn_descendants( &self, root_thread_id: ThreadId, status_filter: Option, ) -> codex_agent_graph_store::AgentGraphStoreFuture<'_, Vec> { assert_eq!(root_thread_id, self.root_thread_id); assert_eq!(status_filter, None); let descendant_thread_ids = self.descendant_thread_ids.clone(); Box::pin(async move { Ok(descendant_thread_ids) }) } } fn user_msg(text: &str) -> ResponseItem { ResponseItem::Message { id: None, role: "user".to_string(), content: vec![ContentItem::OutputText { text: text.to_string(), }], phase: None, internal_chat_message_metadata_passthrough: None, } } fn assistant_msg(text: &str) -> ResponseItem { ResponseItem::Message { id: None, role: "assistant".to_string(), content: vec![ContentItem::OutputText { text: text.to_string(), }], phase: None, internal_chat_message_metadata_passthrough: None, } } fn contextual_user_interrupted_marker() -> ResponseItem { interrupted_turn_history_marker(InterruptedTurnHistoryMarker::ContextualUser) .expect("contextual-user interrupted marker should be enabled") } fn developer_interrupted_marker() -> ResponseItem { interrupted_turn_history_marker(InterruptedTurnHistoryMarker::Developer) .expect("developer interrupted marker should be enabled") } #[test] fn effective_originator_prefers_thread_scoped_sources_before_env_originator() { for (metrics_service_name, persisted_originator, inherited_originator, expected_originator) in [ ( Some("codex_work_desktop"), Some("persisted_originator"), Some("inherited_originator"), "codex_work_desktop", ), ( Some("codex_work_web"), Some("persisted_originator"), Some("inherited_originator"), "codex_work_web", ), ( Some("codex_work_mobile"), Some("persisted_originator"), Some("inherited_originator"), "codex_work_mobile", ), ( Some("codex_work_cca"), Some("persisted_originator"), Some("inherited_originator"), "codex_work_cca", ), ( Some("chatgpt_cca"), Some("persisted_originator"), Some("inherited_originator"), "chatgpt_cca", ), ( Some("chatgpt_cca_extra"), Some("persisted_originator"), Some("inherited_originator"), "persisted_originator", ), ( None, Some("persisted_originator"), Some("inherited_originator"), "persisted_originator", ), ( None, None, Some("inherited_originator"), "inherited_originator", ), ] { assert_eq!( effective_originator_value( metrics_service_name, Some("Codex Desktop".to_string()), persisted_originator.map(str::to_string), inherited_originator.map(str::to_string), "codex_cli_rs".to_string(), ), expected_originator ); } } #[test] fn truncates_before_requested_user_message() { let items = [ user_msg("u1"), assistant_msg("a1"), assistant_msg("a2"), user_msg("u2"), assistant_msg("a3"), ResponseItem::Reasoning { id: Some(ResponseItemId::with_suffix("rs", "1")), summary: vec![ReasoningItemReasoningSummary::SummaryText { text: "s".to_string(), }], content: None, encrypted_content: None, internal_chat_message_metadata_passthrough: None, }, ResponseItem::FunctionCall { id: None, call_id: "c1".to_string(), name: "tool".to_string(), namespace: None, arguments: "{}".to_string(), encrypted_function_args: None, internal_chat_message_metadata_passthrough: None, }, assistant_msg("a4"), ]; let initial: Vec = items .iter() .cloned() .map(|item| RolloutItem::ResponseItem(item.into())) .collect(); let truncated = truncate_before_nth_user_message( InitialHistory::Forked(initial), /*n*/ 1, &SnapshotTurnState { ends_mid_turn: false, active_turn_id: None, active_turn_started_at: None, active_turn_start_index: None, }, ); let got_items = truncated.get_rollout_items(); let expected_items = vec![ RolloutItem::ResponseItem(items[0].clone().into()), RolloutItem::ResponseItem(items[1].clone().into()), RolloutItem::ResponseItem(items[2].clone().into()), ]; assert_eq!( serde_json::to_value(got_items).unwrap(), serde_json::to_value(&expected_items).unwrap() ); let initial2: Vec = items .iter() .cloned() .map(|item| RolloutItem::ResponseItem(item.into())) .collect(); let truncated2 = truncate_before_nth_user_message( InitialHistory::Forked(initial2.clone()), /*n*/ 2, &SnapshotTurnState { ends_mid_turn: false, active_turn_id: None, active_turn_started_at: None, active_turn_start_index: None, }, ); assert_eq!( serde_json::to_value(truncated2.get_rollout_items()).unwrap(), serde_json::to_value(initial2).unwrap() ); } #[test] fn out_of_range_truncation_drops_only_unfinished_suffix_mid_turn() { let items = vec![ RolloutItem::ResponseItem(user_msg("u1").into()), RolloutItem::ResponseItem(assistant_msg("a1").into()), RolloutItem::ResponseItem(user_msg("u2").into()), RolloutItem::ResponseItem(assistant_msg("partial").into()), ]; let truncated = truncate_before_nth_user_message( InitialHistory::Forked(items.clone()), usize::MAX, &SnapshotTurnState { ends_mid_turn: true, active_turn_id: None, active_turn_started_at: None, active_turn_start_index: None, }, ); assert_eq!( serde_json::to_value(truncated.get_rollout_items()).unwrap(), serde_json::to_value(items[..2].to_vec()).unwrap() ); } #[test] fn fork_thread_accepts_legacy_usize_snapshot_argument() { fn assert_legacy_snapshot_callsite( manager: &ThreadManager, config: Config, path: std::path::PathBuf, ) { let _future = manager.fork_thread( usize::MAX, config, path, /*thread_source*/ None, /*parent_trace*/ None, ); } let _: fn(&ThreadManager, Config, std::path::PathBuf) = assert_legacy_snapshot_callsite; } #[test] fn out_of_range_truncation_drops_pre_user_active_turn_prefix() { let items = vec![ RolloutItem::ResponseItem(user_msg("u1").into()), RolloutItem::ResponseItem(assistant_msg("a1").into()), RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-2".to_string(), trace_id: None, started_at: None, model_context_window: None, collaboration_mode_kind: Default::default(), })), RolloutItem::ResponseItem(user_msg("u2").into()), RolloutItem::ResponseItem(assistant_msg("partial").into()), ]; let snapshot_state = snapshot_turn_state(&InitialHistory::Forked(items.clone())); assert_eq!( snapshot_state, SnapshotTurnState { ends_mid_turn: true, active_turn_id: Some("turn-2".to_string()), active_turn_started_at: None, active_turn_start_index: Some(2), }, ); let truncated = truncate_before_nth_user_message( InitialHistory::Forked(items.clone()), usize::MAX, &snapshot_state, ); assert_eq!( serde_json::to_value(truncated.get_rollout_items()).unwrap(), serde_json::to_value(items[..2].to_vec()).unwrap() ); } #[tokio::test] async fn ignores_session_prefix_messages_when_truncating() { let (session, turn_context) = make_session_and_context().await; let turn_context = Arc::new(turn_context); let world_state = build_world_state_from_turn_context(&session, &turn_context).await; let mut items = session .build_initial_context_with_world_state(&turn_context, &world_state) .await; items.push(user_msg("feature request")); items.push(assistant_msg("ack")); items.push(user_msg("second question")); items.push(assistant_msg("answer")); let rollout_items: Vec = items .iter() .cloned() .map(|item| RolloutItem::ResponseItem(item.into())) .collect(); let truncated = truncate_before_nth_user_message( InitialHistory::Forked(rollout_items), /*n*/ 1, &SnapshotTurnState { ends_mid_turn: false, active_turn_id: None, active_turn_started_at: None, active_turn_start_index: None, }, ); let got_items = truncated.get_rollout_items(); let expected: Vec = vec![ RolloutItem::ResponseItem(items[0].clone().into()), RolloutItem::ResponseItem(items[1].clone().into()), RolloutItem::ResponseItem(items[2].clone().into()), RolloutItem::ResponseItem(items[3].clone().into()), ]; assert_eq!( serde_json::to_value(got_items).unwrap(), serde_json::to_value(&expected).unwrap() ); } #[tokio::test] async fn shutdown_all_threads_bounded_submits_shutdown_to_every_thread() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let manager = ThreadManager::with_models_provider_and_home_for_tests( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), config.codex_home.to_path_buf(), Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ); let thread_1 = manager .start_thread(StartThreadOptions::new(config.clone())) .await .expect("start first thread") .thread_id; let thread_2 = manager .start_thread(StartThreadOptions::new(config.clone())) .await .expect("start second thread") .thread_id; let report = manager .shutdown_all_threads_bounded(Duration::from_secs(10)) .await; let mut expected_completed = vec![thread_1, thread_2]; expected_completed.sort_by_key(std::string::ToString::to_string); assert_eq!(report.completed, expected_completed); assert!(report.submit_failed.is_empty()); assert!(report.timed_out.is_empty()); assert!(manager.list_thread_ids().await.is_empty()); } #[tokio::test] async fn code_mode_session_provider_is_shared_across_threads() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let provider: Arc = Arc::new(DisabledCodeModeSessionProvider); let manager = ThreadManager::with_models_provider_and_home_for_tests( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), config.codex_home.to_path_buf(), Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ) .with_code_mode_session_provider(Arc::clone(&provider)); let first = manager .start_thread(StartThreadOptions::new(config.clone())) .await .expect("start first thread"); let second = manager .start_thread(StartThreadOptions::new(config)) .await .expect("start second thread"); let first_provider = first .thread .session .services .code_mode_service .session_provider(); let second_provider = second .thread .session .services .code_mode_service .session_provider(); assert!(Arc::ptr_eq(&first_provider, &second_provider)); assert!(Arc::ptr_eq(&first_provider, &provider)); assert!(Arc::ptr_eq( &first_provider, &manager.state.code_mode_session_provider )); let mut completed = vec![first.thread_id, second.thread_id]; completed.sort_by_key(std::string::ToString::to_string); let report = manager .shutdown_all_threads_bounded(Duration::from_secs(10)) .await; assert_eq!( report, ThreadShutdownReport { completed, submit_failed: Vec::new(), timed_out: Vec::new(), } ); } #[tokio::test] async fn mcp_invalidation_refreshes_threads_that_are_still_starting() { struct BlockingThreadStartup { entered: tokio::sync::Notify, release: tokio::sync::Notify, refreshed: tokio::sync::Notify, projections: std::sync::atomic::AtomicUsize, } impl codex_extension_api::ThreadLifecycleContributor for BlockingThreadStartup { fn on_thread_start<'a>( &'a self, _input: codex_extension_api::ThreadStartInput<'a, Config>, ) -> codex_extension_api::ExtensionFuture<'a, ()> { Box::pin(async move { self.entered.notify_one(); self.release.notified().await; }) } } impl codex_extension_api::McpServerContributor for BlockingThreadStartup { fn id(&self) -> &'static str { "starting_mcp_runtime_refresh_test" } fn contribute<'a>( &'a self, _context: codex_extension_api::McpServerContributionContext<'a, Config>, ) -> codex_extension_api::ExtensionFuture<'a, Vec> { Box::pin(async move { if self.projections.fetch_add(1, Ordering::AcqRel) != 0 { self.refreshed.notify_one(); } Vec::new() }) } } let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let observer = Arc::new(BlockingThreadStartup { entered: tokio::sync::Notify::new(), release: tokio::sync::Notify::new(), refreshed: tokio::sync::Notify::new(), projections: std::sync::atomic::AtomicUsize::new(0), }); let mut extensions = codex_extension_api::ExtensionRegistryBuilder::new(); extensions.thread_lifecycle_contributor(observer.clone()); extensions.mcp_server_contributor(observer.clone()); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("dummy")); let manager = Arc::new(ThreadManager::new( &config, Arc::clone(&auth_manager), build_models_manager(&config, auth_manager), crate::CodexAppsToolsCache::default(), SessionSource::Exec, Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), Arc::new(extensions.build()), Arc::new(crate::test_support::EmptyUserInstructionsProvider), /*analytics_events_client*/ None, thread_store_from_config(&config, /*state_db*/ None), /*agent_graph_store*/ None, TEST_INSTALLATION_ID.to_string(), /*attestation_provider*/ None, /*external_time_provider*/ None, )); let starting = tokio::spawn({ let manager = Arc::clone(&manager); async move { manager.start_thread(StartThreadOptions::new(config)).await } }); tokio::time::timeout(Duration::from_secs(5), observer.entered.notified()) .await .expect("thread should enter its startup lifecycle"); assert!(manager.list_thread_ids().await.is_empty()); manager.invalidate_mcp_runtimes().await; observer.release.notify_one(); starting .await .expect("thread startup task should finish") .expect("thread should start"); tokio::time::timeout(Duration::from_secs(5), observer.refreshed.notified()) .await .expect("invalidation during startup should refresh the newly published thread"); let shutdown = manager .shutdown_all_threads_bounded(Duration::from_secs(5)) .await; assert!(shutdown.timed_out.is_empty()); } #[tokio::test] async fn start_thread_keeps_internal_threads_hidden_from_normal_lookups() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let manager = ThreadManager::with_models_provider_and_home_for_tests( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), config.codex_home.to_path_buf(), Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ); let thread = manager .start_thread(StartThreadOptions { session_source: Some(SessionSource::Internal( InternalSessionSource::MemoryConsolidation, )), environments: Some(Vec::new()), ..StartThreadOptions::new(config) }) .await .expect("internal thread should start"); assert_eq!(manager.list_thread_ids().await, Vec::new()); assert!(manager.get_thread(thread.thread_id).await.is_err()); assert!( codex_diagnostics::snapshot() .gauges .iter() .any(|gauge| gauge.name == "core.threads.live" && gauge.value > 0) ); let report = manager .shutdown_all_threads_bounded(Duration::from_secs(10)) .await; assert_eq!(report.completed, vec![thread.thread_id]); assert!(report.submit_failed.is_empty()); assert!(report.timed_out.is_empty()); assert!(manager.list_thread_ids().await.is_empty()); } #[tokio::test] async fn spawn_internal_guardian_session_preserves_windows_sandbox_proxy_settings() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let manager = ThreadManager::with_models_provider_and_home_for_tests( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), config.codex_home.to_path_buf(), Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ); let parent = manager .start_thread(StartThreadOptions::new(config.clone())) .await .expect("start parent thread"); let reviewer = manager .spawn_internal_session( parent.thread_id, StartThreadOptions { session_source: Some(SessionSource::Internal(InternalSessionSource::Guardian)), ..StartThreadOptions::new(config) }, ) .await .expect("start internal reviewer"); assert_eq!( ( parent.thread.session.windows_sandbox_proxy_settings_mode, reviewer.thread.session.windows_sandbox_proxy_settings_mode, ), ( codex_sandboxing::WindowsSandboxProxySettingsMode::Reconcile, codex_sandboxing::WindowsSandboxProxySettingsMode::Preserve, ) ); manager .shutdown_all_threads_bounded(Duration::from_secs(10)) .await; } #[tokio::test] async fn spawn_internal_session_preserves_parent_lineage_without_forking_history() { struct ParentLifecycleContributor { observed_mcp_sources: Arc>>, } impl codex_extension_api::ThreadLifecycleContributor for ParentLifecycleContributor {} impl codex_extension_api::McpServerContributor for ParentLifecycleContributor { fn id(&self) -> &'static str { "parent_mcp_contributor" } fn contribute<'a>( &'a self, context: codex_extension_api::McpServerContributionContext<'a, Config>, ) -> codex_extension_api::ExtensionFuture<'a, Vec> { Box::pin(async move { if let Some(session_source) = context.session_source() { self.observed_mcp_sources .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .push(session_source.clone()); } Vec::new() }) } } struct ParentInstructionsProvider(codex_extension_api::Instructions); impl codex_extension_api::UserInstructionsProvider for ParentInstructionsProvider { fn load_user_instructions(&self) -> codex_extension_api::LoadUserInstructionsFuture<'_> { Box::pin(async move { codex_extension_api::LoadedUserInstructions { instructions: Some(self.0.clone()), warnings: Vec::new(), } }) } } let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let mut managed_exec_policy = codex_execpolicy::Policy::empty(); managed_exec_policy .add_prefix_rule(&["rm".to_string()], codex_execpolicy::Decision::Forbidden) .expect("add managed execution restriction"); let mut requirements = config.config_layer_stack.requirements().clone(); requirements.exec_policy = Some(codex_config::Sourced::new( codex_execpolicy::RequirementsExecPolicy::new(managed_exec_policy), codex_config::RequirementSource::Unknown, )); requirements.additional_developer_instructions = Some(codex_config::Sourced::new( "managed instructions must not shape the reviewer".to_string(), codex_config::RequirementSource::Unknown, )); let mut requirements_toml = config.config_layer_stack.requirements_toml().clone(); requirements_toml.additional_developer_instructions = Some("managed instructions must not shape the reviewer".to_string()); config.config_layer_stack = codex_config::ConfigLayerStack::new( config .config_layer_stack .all_layers_low_to_high() .cloned() .collect(), requirements, requirements_toml, ) .expect("managed requirements stack"); let parent_instructions = codex_extension_api::Instructions { text: "parent user instructions must not be inherited".to_string(), source: config.codex_home.join("AGENTS.md"), }; let mut manager = ThreadManager::with_models_provider_and_home_for_tests( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), config.codex_home.to_path_buf(), Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ); let observed_mcp_sources = Arc::new(std::sync::Mutex::new(Vec::new())); let parent_contributor = Arc::new(ParentLifecycleContributor { observed_mcp_sources: Arc::clone(&observed_mcp_sources), }); let mut extensions = codex_extension_api::ExtensionRegistryBuilder::new(); extensions.thread_lifecycle_contributor(parent_contributor.clone()); extensions.mcp_server_contributor(parent_contributor); let manager_state = Arc::get_mut(&mut manager.state).expect("unshared thread manager state"); manager_state.extensions = Arc::new(extensions.build()); manager_state.mcp_manager = Arc::new(McpManager::new_with_extensions( Arc::clone(&manager_state.plugins_manager), Arc::clone(&manager_state.extensions), manager_state.mcp_manager.codex_apps_tools_cache(), )); manager_state.user_instructions_provider = Arc::new(ParentInstructionsProvider(parent_instructions.clone())); let parent = manager .start_thread(StartThreadOptions { metrics_service_name: Some("codex_work_desktop".to_string()), ..StartThreadOptions::new(config.clone()) }) .await .expect("start parent thread"); parent .thread .session .set_multi_agent_version_if_unset(MultiAgentVersion::V2); assert_eq!( parent.thread.session.user_instructions().await, Some(parent_instructions) ); assert_eq!( parent .thread .session .services .extensions .thread_lifecycle_contributors() .len(), 1 ); let mut reviewer_environments = parent .thread .session .services .turn_environments .selections(); let reviewer_environment = reviewer_environments .first_mut() .expect("parent should have an environment"); reviewer_environment.config = EnvironmentConfigState::Ready(codex_protocol::protocol::EnvironmentConfig { allow_login_shell: true, workspace_roots: reviewer_environment.workspace_roots.clone(), permission_profile: config.permissions.permission_profile_state().snapshot(), shell_environment_policy: Default::default(), windows_sandbox_level: WindowsSandboxLevel::from_config(&config), windows_sandbox_private_desktop: config.permissions.windows_sandbox_private_desktop, use_legacy_landlock: config.features.use_legacy_landlock(), exec_policy: Some(codex_execpolicy::RequirementsExecPolicy::new( codex_execpolicy::Policy::empty(), )), mcp_policy: None, network_policy: None, selected_capability_roots: Vec::new(), }); let reviewer = manager .spawn_internal_session( parent.thread_id, StartThreadOptions { session_source: Some(SessionSource::Internal(InternalSessionSource::Guardian)), initial_history: InitialHistory::Forked(vec![RolloutItem::ResponseItem( user_msg("parent history must not be inherited").into(), )]), environments: Some(reviewer_environments), ..StartThreadOptions::new(config) }, ) .await .expect("start internal reviewer"); let reviewer_config = reviewer.thread.config_snapshot().await; assert_eq!( reviewer.session_configured.session_id, parent.session_configured.session_id ); assert!(std::ptr::eq( reviewer .thread .session .services .agent_control .rollout_budget(), parent .thread .session .services .agent_control .rollout_budget(), )); assert_eq!(reviewer_config.parent_thread_id, Some(parent.thread_id)); assert_eq!(reviewer_config.forked_from_thread_id, None); assert_eq!(reviewer_config.originator, "codex_work_desktop"); assert_eq!( reviewer.thread.multi_agent_version(), Some(MultiAgentVersion::Disabled) ); assert_eq!( reviewer.session_configured.parent_thread_id, Some(parent.thread_id) ); assert_eq!(reviewer.session_configured.forked_from_id, None); assert!(reviewer.thread.session.user_instructions().await.is_none()); assert!( reviewer .thread .session .services .exec_policy .current() .rules() .contains_key("rm") ); let reviewer_turn = reviewer.thread.session.new_default_turn().await; let reviewer_world_state = build_world_state_from_turn_context(&reviewer.thread.session, &reviewer_turn).await; let reviewer_context = reviewer .thread .session .build_initial_context_with_world_state(&reviewer_turn, &reviewer_world_state) .await; assert!( !serde_json::to_string(&reviewer_context) .expect("reviewer context should serialize") .contains("managed instructions must not shape the reviewer") ); let reviewer_environment = reviewer .thread .environment_selections() .await .into_iter() .next() .expect("reviewer should retain its selected environment"); assert!(matches!( reviewer_environment.config, EnvironmentConfigState::Ready(config) if config.exec_policy.is_some() )); assert!( reviewer .thread .session .services .extensions .thread_lifecycle_contributors() .is_empty() ); { let observed_mcp_sources = observed_mcp_sources .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); assert!(!observed_mcp_sources.is_empty()); assert!( observed_mcp_sources .iter() .all(|source| !source.is_internal()) ); } assert_eq!(manager.list_thread_ids().await, vec![parent.thread_id]); assert!(manager.get_thread(reviewer.thread_id).await.is_err()); assert!( reviewer .thread .session .clone_history() .await .raw_items() .next() .is_none() ); manager .shutdown_all_threads_bounded(Duration::from_secs(10)) .await; } #[tokio::test] async fn start_thread_seeds_extension_data_for_mcp_and_lifecycle_contributors() { struct InitialDataRecorder { lifecycle_observed: Arc>>, mcp_observed: Arc>>, } impl codex_extension_api::ThreadLifecycleContributor for InitialDataRecorder { fn on_thread_start<'a>( &'a self, input: codex_extension_api::ThreadStartInput<'a, Config>, ) -> codex_extension_api::ExtensionFuture<'a, ()> { Box::pin(async move { let selected_root = input .thread_store .get::>() .and_then(|roots| roots.first().cloned()) .expect("selected root should be available"); self.lifecycle_observed .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .push((input.thread_store.level_id().to_string(), selected_root.id)); input .thread_store .insert(Vec::::new()); }) } } impl codex_extension_api::McpServerContributor for InitialDataRecorder { fn id(&self) -> &'static str { "selected_root_test" } fn contribute<'a>( &'a self, context: codex_extension_api::McpServerContributionContext<'a, Config>, ) -> codex_extension_api::ExtensionFuture<'a, Vec> { Box::pin(async move { let thread_init = context .thread_init() .expect("initial MCP resolution should be thread-scoped"); let selected_root = thread_init .get::>() .and_then(|roots| roots.first().cloned()) .expect("selected root should be available"); self.mcp_observed .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .push(( selected_root.id.clone(), context .session_source() .expect("thread-scoped MCP resolution should identify its source") .clone(), )); let mut server = codex_mcp::codex_apps_mcp_server_config( "https://selected.invalid", /*apps_mcp_product_sku*/ None, /*originator*/ None, ); let CapabilityRootLocation::Environment { environment_id, .. } = &selected_root.location; server.environment_id = environment_id.clone(); server.enabled = false; let plugin_id = selected_root.id; vec![codex_extension_api::McpServerContribution::SelectedPlugin { name: plugin_id.clone(), plugin_display_name: plugin_id.clone(), plugin_id, selection_order: 0, config: Box::new(server), }] }) } } let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); config .features .enable(Feature::Apps) .expect("test config should allow apps"); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let lifecycle_observed = Arc::new(std::sync::Mutex::new(Vec::new())); let mcp_observed = Arc::new(std::sync::Mutex::new(Vec::new())); let recorder = Arc::new(InitialDataRecorder { lifecycle_observed: Arc::clone(&lifecycle_observed), mcp_observed: Arc::clone(&mcp_observed), }); let mut extensions = codex_extension_api::ExtensionRegistryBuilder::new(); extensions.thread_lifecycle_contributor(recorder.clone()); extensions.mcp_server_contributor(recorder); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); let manager = ThreadManager::new( &config, auth_manager.clone(), build_models_manager(&config, auth_manager), crate::CodexAppsToolsCache::default(), SessionSource::Exec, Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), Arc::new(extensions.build()), Arc::new(crate::test_support::EmptyUserInstructionsProvider), /*analytics_events_client*/ None, thread_store_from_config(&config, /*state_db*/ None), /*agent_graph_store*/ None, TEST_INSTALLATION_ID.to_string(), /*attestation_provider*/ None, /*external_time_provider*/ None, ); let selected_root_init = |id: &str, environment_id: &str| { let mut init = codex_extension_api::ExtensionDataInit::new(); init.insert(vec![SelectedCapabilityRoot { id: id.to_string(), location: CapabilityRootLocation::Environment { environment_id: environment_id.to_string(), path: PathUri::parse(&format!("file:///plugins/{id}")).expect("plugin root URI"), }, }]); init }; let first_thread = manager .start_thread(StartThreadOptions { metrics_service_name: Some("codex_work_desktop".to_string()), environments: Some(Vec::new()), thread_extension_init: selected_root_init("selected-a", "env-a"), ..StartThreadOptions::new(config.clone()) }) .await .expect("start first thread"); let second_session_source = SessionSource::SubAgent(SubAgentSource::Review); let second_thread = manager .start_thread(StartThreadOptions { environments: Some(Vec::new()), session_source: Some(second_session_source.clone()), thread_extension_init: selected_root_init("selected-b", "env-b"), ..StartThreadOptions::new(config.clone()) }) .await .expect("start second thread"); let first_session = &first_thread.thread.session; let first_originator = first_session.originator().await; let first_resolved = first_session .services .mcp_manager .runtime_config_for_step( &config, &first_session.services.mcp_thread_init, &first_session.services.thread_extension_data, McpThreadIdentity { session_source: &SessionSource::Exec, originator: &first_originator, environments: McpEnvironmentScope::Live(&first_session.services.turn_environments), }, /*ready_selected_capability_roots*/ &[], /*executor_capability_discovery*/ None, ) .await; let second_session = &second_thread.thread.session; let second_originator = second_session.originator().await; let second_resolved = second_session .services .mcp_manager .runtime_config_for_step( &config, &second_session.services.mcp_thread_init, &second_session.services.thread_extension_data, McpThreadIdentity { session_source: &second_session_source, originator: &second_originator, environments: McpEnvironmentScope::Live(&second_session.services.turn_environments), }, /*ready_selected_capability_roots*/ &[], /*executor_capability_discovery*/ None, ) .await; assert_eq!( *lifecycle_observed .lock() .unwrap_or_else(std::sync::PoisonError::into_inner), vec![ (first_thread.thread_id.to_string(), "selected-a".to_string()), ( second_thread.thread_id.to_string(), "selected-b".to_string() ), ] ); assert_eq!( *mcp_observed .lock() .unwrap_or_else(std::sync::PoisonError::into_inner), vec![ ("selected-a".to_string(), SessionSource::Exec), ("selected-b".to_string(), second_session_source.clone()), ("selected-a".to_string(), SessionSource::Exec), ("selected-b".to_string(), second_session_source), ] ); let selected_servers = |config: &codex_mcp::McpConfig| { codex_mcp::configured_mcp_servers(config) .into_iter() .filter(|(name, _)| name.starts_with("selected-")) .map(|(name, server)| (name, server.environment_id)) .collect::>() }; assert_eq!( selected_servers(&first_resolved.config), std::collections::BTreeMap::from([("selected-a".to_string(), "env-a".to_string())]) ); assert_eq!( selected_servers(&second_resolved.config), std::collections::BTreeMap::from([("selected-b".to_string(), "env-b".to_string())]) ); let codex_apps_server = codex_mcp::configured_mcp_servers(&first_resolved.config) .remove(codex_mcp::CODEX_APPS_MCP_SERVER_NAME) .expect("Codex Apps server should be configured"); let codex_apps_headers = match codex_apps_server.transport { codex_config::McpServerTransportConfig::StreamableHttp { http_headers, .. } => http_headers, codex_config::McpServerTransportConfig::Stdio { .. } => { panic!("Codex Apps server should use streamable HTTP") } }; assert_eq!( codex_apps_headers .expect("Codex Apps headers should be configured") .get("originator"), Some(&"codex_work_desktop".to_string()) ); } #[tokio::test] async fn selected_capability_roots_round_trip_through_fork() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let manager = ThreadManager::with_models_provider_and_home_for_tests( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), config.codex_home.to_path_buf(), Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ); let selected_roots = vec![SelectedCapabilityRoot { id: "demo@1".to_string(), location: CapabilityRootLocation::Environment { environment_id: "build".to_string(), path: PathUri::parse("file:///plugins/demo").expect("plugin root URI"), }, }]; let inherited = manager .start_thread(StartThreadOptions { initial_history: InitialHistory::Forked(vec![RolloutItem::SessionMeta( SessionMetaLine { meta: SessionMeta { selected_capability_roots: selected_roots.clone(), ..SessionMeta::default() }, git: None, }, )]), environments: Some(Vec::new()), ..StartThreadOptions::new(config) }) .await .expect("start inherited fork"); inherited.thread.ensure_rollout_materialized().await; inherited .thread .flush_rollout() .await .expect("flush inherited fork"); let inherited_history = RolloutRecorder::get_rollout_history( &inherited .thread .rollout_path() .expect("inherited fork rollout path"), ) .await .expect("read inherited fork rollout"); assert_eq!( inherited_history.get_selected_capability_roots(), selected_roots ); } #[tokio::test] async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); let manager = ThreadManager::new( &config, auth_manager.clone(), build_models_manager(&config, auth_manager.clone()), crate::CodexAppsToolsCache::default(), SessionSource::Exec, Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), empty_extension_registry(), Arc::new(crate::test_support::EmptyUserInstructionsProvider), /*analytics_events_client*/ None, thread_store_from_config(&config, /*state_db*/ None), /*agent_graph_store*/ None, TEST_INSTALLATION_ID.to_string(), /*attestation_provider*/ None, /*external_time_provider*/ None, ); let selected_cwd = AbsolutePathBuf::try_from(config.cwd.as_path().join("selected")).expect("absolute path"); std::fs::create_dir_all(&selected_cwd).expect("create selected cwd"); let environments = vec![TurnEnvironmentSelection { environment_id: "local".to_string(), cwd: PathUri::from_abs_path(&selected_cwd), workspace_roots: Vec::new(), config: EnvironmentConfigState::FromThread, }]; let default_cwd = config.cwd.clone(); let mut source_config = config.clone(); source_config.cwd = selected_cwd.clone(); let source = manager .start_thread(StartThreadOptions { environments: Some(environments.clone()), ..StartThreadOptions::new(source_config) }) .await .expect("start source thread"); source.thread.ensure_rollout_materialized().await; source .thread .flush_rollout() .await .expect("flush source rollout"); let rollout_path = source .thread .rollout_path() .expect("source rollout path should exist"); source .thread .shutdown_and_wait() .await .expect("shutdown source thread before resume"); let _ = manager.remove_thread(&source.thread_id).await; let resumed = manager .resume_thread_from_rollout( config.clone(), rollout_path.clone(), auth_manager, /*parent_trace*/ None, ClientMcpExtensions::default(), ) .await .expect("resume source thread"); let (prepared_turn, _) = resumed .thread .session .new_turn_with_sub_id( "resume-turn".to_string(), SessionSettingsUpdate::default(), Default::default(), ) .await .expect("build resumed turn context"); let resumed_turn = prepared_turn; assert_eq!(resumed_turn.environments.turn_environments().count(), 1); assert_eq!( resumed_turn .environments .primary() .expect("primary environment") .cwd(), &PathUri::from_abs_path(&default_cwd) ); assert_ne!( resumed_turn .environments .primary() .expect("primary environment") .cwd(), &PathUri::from_abs_path(&selected_cwd) ); let forked = manager .fork_thread( ForkSnapshot::Interrupted, config, rollout_path, /*thread_source*/ None, /*parent_trace*/ None, ) .await .expect("fork source thread"); let (prepared_turn, _) = forked .thread .session .new_turn_with_sub_id( "fork-turn".to_string(), SessionSettingsUpdate::default(), Default::default(), ) .await .expect("build forked turn context"); let forked_turn = prepared_turn; assert_eq!(forked_turn.environments.turn_environments().count(), 1); assert_eq!( forked_turn .environments .primary() .expect("primary environment") .cwd(), &PathUri::from_abs_path(&default_cwd) ); assert_ne!( forked_turn .environments .primary() .expect("primary environment") .cwd(), &PathUri::from_abs_path(&selected_cwd) ); } #[tokio::test] async fn explicit_installation_id_skips_codex_home_file() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); let installation_id = uuid::Uuid::new_v4().to_string(); let state_db = init_state_db(&config).await; let thread_store = thread_store_from_config(&config, state_db.clone()); let manager = ThreadManager::new( &config, auth_manager.clone(), build_models_manager(&config, auth_manager), crate::CodexAppsToolsCache::default(), SessionSource::Exec, Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), empty_extension_registry(), Arc::new(crate::test_support::EmptyUserInstructionsProvider), /*analytics_events_client*/ None, thread_store, local_agent_graph_store_from_state_db(state_db.as_ref()), installation_id.clone(), /*attestation_provider*/ None, /*external_time_provider*/ None, ); let thread = manager .start_thread(StartThreadOptions::new(config.clone())) .await .expect("start thread with explicit installation id"); assert!(!config.codex_home.join(INSTALLATION_ID_FILENAME).exists()); assert_eq!(thread.thread.session.installation_id, installation_id); thread .thread .shutdown_and_wait() .await .expect("shutdown thread"); let _ = manager.remove_thread(&thread.thread_id).await; } #[tokio::test] async fn resume_active_thread_from_rollout_returns_running_thread() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); let manager = ThreadManager::new( &config, auth_manager.clone(), build_models_manager(&config, auth_manager.clone()), crate::CodexAppsToolsCache::default(), SessionSource::Exec, Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), empty_extension_registry(), Arc::new(crate::test_support::EmptyUserInstructionsProvider), /*analytics_events_client*/ None, thread_store_from_config(&config, /*state_db*/ None), /*agent_graph_store*/ None, TEST_INSTALLATION_ID.to_string(), /*attestation_provider*/ None, /*external_time_provider*/ None, ); let source = manager .start_thread(StartThreadOptions::new(config.clone())) .await .expect("start source thread"); source.thread.ensure_rollout_materialized().await; source .thread .flush_rollout() .await .expect("flush source rollout"); let rollout_path = source .thread .rollout_path() .expect("source rollout path should exist"); let resumed = manager .resume_thread_from_rollout( config, rollout_path, auth_manager, /*parent_trace*/ None, ClientMcpExtensions::default(), ) .await .expect("resume active source thread"); assert_eq!(resumed.thread_id, source.thread_id); assert!(Arc::ptr_eq(&resumed.thread, &source.thread)); source .thread .shutdown_and_wait() .await .expect("shutdown source thread"); } #[tokio::test] async fn resume_stopped_thread_from_rollout_spawns_new_thread() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); let manager = ThreadManager::new( &config, auth_manager.clone(), build_models_manager(&config, auth_manager.clone()), crate::CodexAppsToolsCache::default(), SessionSource::Exec, Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), empty_extension_registry(), Arc::new(crate::test_support::EmptyUserInstructionsProvider), /*analytics_events_client*/ None, thread_store_from_config(&config, /*state_db*/ None), /*agent_graph_store*/ None, TEST_INSTALLATION_ID.to_string(), /*attestation_provider*/ None, /*external_time_provider*/ None, ); let source = manager .start_thread(StartThreadOptions::new(config.clone())) .await .expect("start source thread"); source.thread.ensure_rollout_materialized().await; source .thread .flush_rollout() .await .expect("flush source rollout"); let rollout_path = source .thread .rollout_path() .expect("source rollout path should exist"); source .thread .shutdown_and_wait() .await .expect("shutdown source thread"); let resumed = manager .resume_thread_from_rollout( config, rollout_path, auth_manager, /*parent_trace*/ None, ClientMcpExtensions::default(), ) .await .expect("resume stopped source thread"); assert_eq!(resumed.thread_id, source.thread_id); assert!(!Arc::ptr_eq(&resumed.thread, &source.thread)); resumed .thread .shutdown_and_wait() .await .expect("shutdown resumed thread"); } #[tokio::test] async fn resume_stopped_thread_from_rollout_preserves_thread_source() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); let state_db = init_state_db(&config).await; let thread_store = thread_store_from_config(&config, state_db.clone()); let manager = ThreadManager::new( &config, auth_manager.clone(), build_models_manager(&config, auth_manager.clone()), crate::CodexAppsToolsCache::default(), SessionSource::Exec, Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), empty_extension_registry(), Arc::new(crate::test_support::EmptyUserInstructionsProvider), /*analytics_events_client*/ None, thread_store, local_agent_graph_store_from_state_db(state_db.as_ref()), TEST_INSTALLATION_ID.to_string(), /*attestation_provider*/ None, /*external_time_provider*/ None, ); let source = manager .start_thread(StartThreadOptions { thread_source: Some(ThreadSource::User), environments: Some(Vec::new()), ..StartThreadOptions::new(config.clone()) }) .await .expect("start source thread"); source.thread.ensure_rollout_materialized().await; source .thread .flush_rollout() .await .expect("flush source rollout"); let rollout_path = source .thread .rollout_path() .expect("source rollout path should exist"); source .thread .shutdown_and_wait() .await .expect("shutdown source thread before resume"); let _ = manager.remove_thread(&source.thread_id).await; let resumed = manager .resume_thread_from_rollout( config, rollout_path, auth_manager, /*parent_trace*/ None, ClientMcpExtensions::default(), ) .await .expect("resume source thread"); assert_eq!( resumed .thread .config_snapshot() .await .thread_source .as_ref(), Some(&ThreadSource::User) ); resumed .thread .shutdown_and_wait() .await .expect("shutdown resumed thread"); } #[tokio::test] async fn subtree_listing_uses_injected_graph_store_without_state_db() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let root_thread_id = ThreadId::new(); let descendant_thread_ids = vec![ThreadId::new(), ThreadId::new()]; let agent_graph_store = Arc::new(FakeAgentGraphStore { root_thread_id, descendant_thread_ids: descendant_thread_ids.clone(), }); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); let manager = ThreadManager::new( &config, auth_manager.clone(), build_models_manager(&config, auth_manager), crate::CodexAppsToolsCache::default(), SessionSource::Exec, Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), empty_extension_registry(), Arc::new(crate::test_support::EmptyUserInstructionsProvider), /*analytics_events_client*/ None, thread_store_from_config(&config, /*state_db*/ None), Some(agent_graph_store), TEST_INSTALLATION_ID.to_string(), /*attestation_provider*/ None, /*external_time_provider*/ None, ); let mut expected_thread_ids = vec![root_thread_id]; expected_thread_ids.extend(descendant_thread_ids); assert_eq!( manager .list_agent_subtree_thread_ids(root_thread_id) .await .expect("subtree should load from injected graph store"), expected_thread_ids ); } #[tokio::test] async fn rollout_path_resume_and_fork_read_history_through_thread_store() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); config.experimental_thread_store = ThreadStoreConfig::InMemory { id: format!("thread-manager-{}", uuid::Uuid::new_v4()), }; std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); let state_db = init_state_db(&config).await; let thread_store = thread_store_from_config(&config, state_db.clone()); let in_memory_store = thread_store .as_any() .downcast_ref::() .expect("configured in-memory store"); let manager = ThreadManager::new( &config, auth_manager.clone(), build_models_manager(&config, auth_manager.clone()), crate::CodexAppsToolsCache::default(), SessionSource::Exec, Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), empty_extension_registry(), Arc::new(crate::test_support::EmptyUserInstructionsProvider), /*analytics_events_client*/ None, thread_store.clone(), local_agent_graph_store_from_state_db(state_db.as_ref()), TEST_INSTALLATION_ID.to_string(), /*attestation_provider*/ None, /*external_time_provider*/ None, ); let source = manager .start_thread(StartThreadOptions::new(config.clone())) .await .expect("start source thread"); source .thread .shutdown_and_wait() .await .expect("shutdown source thread"); let _ = manager.remove_thread(&source.thread_id).await; let rollout_path = config .codex_home .join("rollouts/source.jsonl") .to_path_buf(); let resumed = manager .resume_thread_with_history( config.clone(), InitialHistory::Resumed(ResumedHistory { conversation_id: source.thread_id, history: Arc::new(vec![RolloutItem::ResponseItem(user_msg("hello").into())]), rollout_path: Some(rollout_path.clone()), }), auth_manager.clone(), /*parent_trace*/ None, ClientMcpExtensions::default(), ) .await .expect("seed rollout path in store"); resumed .thread .shutdown_and_wait() .await .expect("shutdown seeded resumed thread"); let _ = manager.remove_thread(&resumed.thread_id).await; let resumed_from_path = manager .resume_thread_from_rollout( config.clone(), rollout_path.clone(), auth_manager, /*parent_trace*/ None, ClientMcpExtensions::default(), ) .await .expect("resume from rollout path"); assert_eq!(resumed_from_path.thread_id, resumed.thread_id); let forked = manager .fork_thread( ForkSnapshot::Interrupted, config, rollout_path, /*thread_source*/ None, /*parent_trace*/ None, ) .await .expect("fork from rollout path"); assert_ne!(forked.thread_id, resumed.thread_id); let calls = in_memory_store.calls().await; assert_eq!(calls.read_thread_by_rollout_path, 2); resumed_from_path .thread .shutdown_and_wait() .await .expect("shutdown path-resumed thread"); forked .thread .shutdown_and_wait() .await .expect("shutdown forked thread"); } #[tokio::test] async fn metadata_update_without_result_reads_only_when_the_caller_needs_the_thread() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); config.experimental_thread_store = ThreadStoreConfig::InMemory { id: format!("metadata-update-none-{}", uuid::Uuid::new_v4()), }; std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); let thread_store = thread_store_from_config(&config, /*state_db*/ None); let in_memory_store = thread_store .as_any() .downcast_ref::() .expect("configured in-memory store"); let manager = ThreadManager::new( &config, auth_manager.clone(), build_models_manager(&config, auth_manager), crate::CodexAppsToolsCache::default(), SessionSource::Exec, Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), empty_extension_registry(), Arc::new(crate::test_support::EmptyUserInstructionsProvider), /*analytics_events_client*/ None, thread_store.clone(), /*agent_graph_store*/ None, TEST_INSTALLATION_ID.to_string(), /*attestation_provider*/ None, /*external_time_provider*/ None, ); let started = manager .start_thread(StartThreadOptions::new(config)) .await .expect("start thread"); started .thread .flush_rollout() .await .expect("flush initial metadata"); manager .update_thread_metadata( started.thread_id, ThreadMetadataPatch { name: Some(Some("initial name".to_string())), ..Default::default() }, /*include_archived*/ false, ) .await .expect("flush pending live metadata before measuring calls"); in_memory_store.omit_metadata_update_result_for_testing(); let before_loaded_update = in_memory_store.calls().await; let loaded = manager .update_thread_metadata( started.thread_id, ThreadMetadataPatch { name: Some(Some("loaded name".to_string())), ..Default::default() }, /*include_archived*/ false, ) .await .expect("update loaded thread metadata"); assert_eq!(loaded.name.as_deref(), Some("loaded name")); let after_loaded_update = in_memory_store.calls().await; assert_eq!( after_loaded_update.update_thread_metadata, before_loaded_update.update_thread_metadata + 1 ); assert_eq!( after_loaded_update.read_thread, before_loaded_update.read_thread + 1 ); started .thread .append_rollout_items(&[RolloutItem::EventMsg(EventMsg::UserMessage( UserMessageEvent { message: "completion-only metadata".to_string(), ..Default::default() }, ))]) .await .expect("append item with derived metadata"); let after_completion_only_update = in_memory_store.calls().await; assert_eq!( after_completion_only_update.update_thread_metadata, after_loaded_update.update_thread_metadata + 1 ); assert_eq!( after_completion_only_update.read_thread, after_loaded_update.read_thread ); started .thread .shutdown_and_wait() .await .expect("shutdown loaded thread"); let _ = manager.remove_thread(&started.thread_id).await; let before_cold_update = in_memory_store.calls().await; let cold = manager .update_thread_metadata( started.thread_id, ThreadMetadataPatch { name: Some(Some("cold name".to_string())), ..Default::default() }, /*include_archived*/ false, ) .await .expect("update cold thread metadata"); assert_eq!(cold.name.as_deref(), Some("cold name")); let after_cold_update = in_memory_store.calls().await; assert_eq!( after_cold_update.update_thread_metadata, before_cold_update.update_thread_metadata + 1 ); assert_eq!( after_cold_update.read_thread, before_cold_update.read_thread + 1 ); } #[tokio::test] async fn new_uses_active_provider_for_model_refresh() { let server = MockServer::start().await; let models_mock = mount_models_once(&server, ModelsResponse { models: vec![] }).await; let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); config.model_catalog = None; config.model_provider.base_url = Some(server.uri()); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); let manager = ThreadManager::new( &config, auth_manager.clone(), build_models_manager(&config, auth_manager), crate::CodexAppsToolsCache::default(), SessionSource::Exec, Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), empty_extension_registry(), Arc::new(crate::test_support::EmptyUserInstructionsProvider), /*analytics_events_client*/ None, thread_store_from_config(&config, /*state_db*/ None), /*agent_graph_store*/ None, TEST_INSTALLATION_ID.to_string(), /*attestation_provider*/ None, /*external_time_provider*/ None, ); let _ = manager .list_models( RefreshStrategy::Online, crate::test_support::default_http_client_factory(), ) .await; assert_eq!(models_mock.requests().len(), 1); } #[tokio::test] async fn injected_models_manager_controls_refresh_policy() { let server = MockServer::start().await; let _ = mount_models_once(&server, ModelsResponse { models: vec![] }).await; let _ = mount_models_once(&server, ModelsResponse { models: vec![] }).await; let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); config.model_catalog = None; config.model_provider.base_url = Some(server.uri()); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); let provider = create_model_provider( config.model_provider.clone(), Some(Arc::clone(&auth_manager)), ); let models_manager = provider.models_manager_without_cache(config.model_catalog.clone()); let manager = ThreadManager::new( &config, auth_manager, models_manager, crate::CodexAppsToolsCache::default(), SessionSource::Custom("test-embedder".to_string()), Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), empty_extension_registry(), Arc::new(crate::test_support::EmptyUserInstructionsProvider), /*analytics_events_client*/ None, thread_store_from_config(&config, /*state_db*/ None), /*agent_graph_store*/ None, TEST_INSTALLATION_ID.to_string(), /*attestation_provider*/ None, /*external_time_provider*/ None, ); let http_client_factory = crate::test_support::default_http_client_factory(); let _ = manager .list_models( RefreshStrategy::OnlineIfUncached, http_client_factory.clone(), ) .await; let _ = manager .list_models(RefreshStrategy::OnlineIfUncached, http_client_factory) .await; assert_eq!( server.received_requests().await.unwrap_or_default().len(), 2 ); assert!(!config.codex_home.join("models_cache.json").exists()); } #[test] fn interrupted_fork_snapshot_appends_interrupt_boundary() { let committed_history = InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("hello").into())]); assert_eq!( serde_json::to_value( append_interrupted_boundary( committed_history, /*turn_id*/ None, /*started_at*/ None, InterruptedTurnHistoryMarker::ContextualUser, ) .get_rollout_items() ) .expect("serialize interrupted fork history"), serde_json::to_value(vec![ RolloutItem::ResponseItem(user_msg("hello").into()), RolloutItem::ResponseItem(contextual_user_interrupted_marker().into()), RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent { turn_id: None, started_at: None, reason: TurnAbortReason::Interrupted, completed_at: None, duration_ms: None, })), ]) .expect("serialize expected interrupted fork history"), ); assert_eq!( serde_json::to_value( append_interrupted_boundary( InitialHistory::New, /*turn_id*/ None, /*started_at*/ None, InterruptedTurnHistoryMarker::ContextualUser, ) .get_rollout_items() ) .expect("serialize interrupted empty fork history"), serde_json::to_value(vec![ RolloutItem::ResponseItem(contextual_user_interrupted_marker().into()), RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent { turn_id: None, started_at: None, reason: TurnAbortReason::Interrupted, completed_at: None, duration_ms: None, })), ]) .expect("serialize expected interrupted empty history"), ); } #[test] fn disabled_interrupted_fork_snapshot_appends_only_interrupt_event() { let committed_history = InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("hello").into())]); assert_eq!( serde_json::to_value( append_interrupted_boundary( committed_history, /*turn_id*/ None, /*started_at*/ None, InterruptedTurnHistoryMarker::Disabled, ) .get_rollout_items() ) .expect("serialize disabled interrupted fork history"), serde_json::to_value(vec![ RolloutItem::ResponseItem(user_msg("hello").into()), RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent { turn_id: None, started_at: None, reason: TurnAbortReason::Interrupted, completed_at: None, duration_ms: None, })), ]) .expect("serialize expected disabled interrupted fork history"), ); assert_eq!( serde_json::to_value( append_interrupted_boundary( InitialHistory::New, /*turn_id*/ None, /*started_at*/ None, InterruptedTurnHistoryMarker::Disabled, ) .get_rollout_items() ) .expect("serialize disabled interrupted empty fork history"), serde_json::to_value(vec![RolloutItem::EventMsg(EventMsg::TurnAborted( TurnAbortedEvent { turn_id: None, started_at: None, reason: TurnAbortReason::Interrupted, completed_at: None, duration_ms: None, }, ))]) .expect("serialize expected disabled interrupted empty fork history"), ); } #[test] fn interrupted_snapshot_is_not_mid_turn() { let interrupted_history = InitialHistory::Forked(vec![ RolloutItem::ResponseItem(user_msg("hello").into()), RolloutItem::ResponseItem(assistant_msg("partial").into()), RolloutItem::ResponseItem(contextual_user_interrupted_marker().into()), RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent { turn_id: Some("turn-1".to_string()), started_at: None, reason: TurnAbortReason::Interrupted, completed_at: None, duration_ms: None, })), ]); assert_eq!( snapshot_turn_state(&interrupted_history), SnapshotTurnState { ends_mid_turn: false, active_turn_id: None, active_turn_started_at: None, active_turn_start_index: None, }, ); } #[test] fn multi_agent_v2_interrupted_marker_uses_developer_input_message() { assert_eq!( developer_interrupted_marker(), ResponseItem::Message { id: None, role: "developer".to_string(), content: vec![ContentItem::InputText { text: format!( "\n{}\n", crate::context::TurnAborted::INTERRUPTED_DEVELOPER_GUIDANCE ), }], phase: None, internal_chat_message_metadata_passthrough: Some( InternalChatMessageMetadataPassthrough { content_item_kinds: Some(vec![ContentItemKind( "generic.turn_aborted".to_string() )]), ..Default::default() } ), } ); } #[test] fn completed_legacy_event_history_is_not_mid_turn() { let completed_history = InitialHistory::Forked(vec![ RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { client_id: None, message: "hello".to_string(), images: None, text_elements: Vec::new(), local_images: Vec::new(), ..Default::default() })), RolloutItem::EventMsg(EventMsg::AgentMessage(AgentMessageEvent { message: "done".to_string(), phase: None, memory_citation: None, delivery: None, })), ]); assert_eq!( snapshot_turn_state(&completed_history), SnapshotTurnState { ends_mid_turn: false, active_turn_id: None, active_turn_started_at: None, active_turn_start_index: None, }, ); } #[test] fn mixed_response_and_legacy_user_event_history_is_mid_turn() { let mixed_history = InitialHistory::Forked(vec![ RolloutItem::ResponseItem(user_msg("hello").into()), RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { client_id: None, message: "hello".to_string(), images: None, text_elements: Vec::new(), local_images: Vec::new(), ..Default::default() })), ]); assert_eq!( snapshot_turn_state(&mixed_history), SnapshotTurnState { ends_mid_turn: true, active_turn_id: None, active_turn_started_at: None, active_turn_start_index: None, }, ); } #[tokio::test] async fn interrupted_fork_snapshot_does_not_synthesize_turn_id_for_legacy_history() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); let state_db = init_state_db(&config).await; let manager = ThreadManager::new( &config, auth_manager.clone(), build_models_manager(&config, auth_manager.clone()), crate::CodexAppsToolsCache::default(), SessionSource::Exec, Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), empty_extension_registry(), Arc::new(crate::test_support::EmptyUserInstructionsProvider), /*analytics_events_client*/ None, thread_store_from_config(&config, state_db.clone()), local_agent_graph_store_from_state_db(state_db.as_ref()), TEST_INSTALLATION_ID.to_string(), /*attestation_provider*/ None, /*external_time_provider*/ None, ); let source = manager .resume_thread_with_history( config.clone(), InitialHistory::Forked(vec![ RolloutItem::ResponseItem(user_msg("hello").into()), RolloutItem::ResponseItem(assistant_msg("partial").into()), ]), auth_manager, /*parent_trace*/ None, ClientMcpExtensions::default(), ) .await .expect("create source thread from completed history"); let source_path = source .thread .rollout_path() .expect("source rollout path should exist"); let source_history = RolloutRecorder::get_rollout_history(&source_path) .await .expect("read source rollout history"); let source_snapshot_state = snapshot_turn_state(&source_history); assert!(source_snapshot_state.ends_mid_turn); let expected_turn_id = source_snapshot_state.active_turn_id.clone(); assert_eq!(expected_turn_id, None); let forked = manager .fork_thread( ForkSnapshot::Interrupted, config.clone(), source_path, /*thread_source*/ None, /*parent_trace*/ None, ) .await .expect("fork interrupted snapshot"); let forked_path = forked .thread .rollout_path() .expect("forked rollout path should exist"); let history = RolloutRecorder::get_rollout_history(&forked_path) .await .expect("read forked rollout history"); assert!(!snapshot_turn_state(&history).ends_mid_turn); let rollout_items: Vec<_> = history .get_rollout_items() .iter() .filter(|item| !matches!(item, RolloutItem::SessionMeta(_))) .collect(); let interrupted_marker_json = serde_json::to_value(RolloutItem::ResponseItem( contextual_user_interrupted_marker().into(), )) .expect("serialize interrupted marker"); let interrupted_abort_json = serde_json::to_value(RolloutItem::EventMsg( EventMsg::TurnAborted(TurnAbortedEvent { turn_id: expected_turn_id, started_at: None, reason: TurnAbortReason::Interrupted, completed_at: None, duration_ms: None, }), )) .expect("serialize interrupted abort event"); assert_eq!( rollout_items .iter() .filter(|item| { strip_response_item_ids_from_json( serde_json::to_value(item).expect("serialize rollout item"), ) == interrupted_marker_json }) .count(), 1, ); assert_eq!( rollout_items .iter() .filter(|item| { serde_json::to_value(item).expect("serialize rollout item") == interrupted_abort_json }) .count(), 1, ); } #[tokio::test] async fn interrupted_fork_snapshot_preserves_explicit_turn_id() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); let state_db = init_state_db(&config).await; let manager = ThreadManager::new( &config, auth_manager.clone(), build_models_manager(&config, auth_manager.clone()), crate::CodexAppsToolsCache::default(), SessionSource::Exec, Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), empty_extension_registry(), Arc::new(crate::test_support::EmptyUserInstructionsProvider), /*analytics_events_client*/ None, thread_store_from_config(&config, state_db.clone()), local_agent_graph_store_from_state_db(state_db.as_ref()), TEST_INSTALLATION_ID.to_string(), /*attestation_provider*/ None, /*external_time_provider*/ None, ); let source = manager .resume_thread_with_history( config.clone(), InitialHistory::Forked(vec![ RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-explicit".to_string(), trace_id: None, started_at: None, model_context_window: None, collaboration_mode_kind: Default::default(), })), RolloutItem::ResponseItem(user_msg("hello").into()), RolloutItem::ResponseItem(assistant_msg("partial").into()), ]), auth_manager, /*parent_trace*/ None, ClientMcpExtensions::default(), ) .await .expect("create source thread from explicit partial history"); let source_path = source .thread .rollout_path() .expect("source rollout path should exist"); let source_history = RolloutRecorder::get_rollout_history(&source_path) .await .expect("read source rollout history"); let source_snapshot_state = snapshot_turn_state(&source_history); assert_eq!( source_snapshot_state, SnapshotTurnState { ends_mid_turn: true, active_turn_id: Some("turn-explicit".to_string()), active_turn_started_at: None, active_turn_start_index: Some(1), }, ); let forked = manager .fork_thread( ForkSnapshot::Interrupted, config.clone(), source_path, /*thread_source*/ None, /*parent_trace*/ None, ) .await .expect("fork interrupted snapshot"); let forked_path = forked .thread .rollout_path() .expect("forked rollout path should exist"); let history = RolloutRecorder::get_rollout_history(&forked_path) .await .expect("read forked rollout history"); let rollout_items: Vec<_> = history .get_rollout_items() .iter() .filter(|item| !matches!(item, RolloutItem::SessionMeta(_))) .collect(); assert!(rollout_items.iter().any(|item| { matches!( item, RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent { turn_id: Some(turn_id), started_at: None, reason: TurnAbortReason::Interrupted, completed_at: None, duration_ms: None, })) if turn_id == "turn-explicit" ) })); } #[tokio::test] async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_source() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config().await; config.codex_home = temp_dir.path().join("codex-home").abs(); config.cwd = config.codex_home.abs(); std::fs::create_dir_all(&config.codex_home).expect("create codex home"); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); let state_db = init_state_db(&config).await; let manager = ThreadManager::new( &config, auth_manager.clone(), build_models_manager(&config, auth_manager.clone()), crate::CodexAppsToolsCache::default(), SessionSource::Exec, Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), empty_extension_registry(), Arc::new(crate::test_support::EmptyUserInstructionsProvider), /*analytics_events_client*/ None, thread_store_from_config(&config, state_db.clone()), local_agent_graph_store_from_state_db(state_db.as_ref()), TEST_INSTALLATION_ID.to_string(), /*attestation_provider*/ None, /*external_time_provider*/ None, ); let source = manager .resume_thread_with_history( config.clone(), InitialHistory::Forked(vec![ RolloutItem::ResponseItem(user_msg("hello").into()), RolloutItem::ResponseItem(assistant_msg("partial").into()), ]), auth_manager, /*parent_trace*/ None, ClientMcpExtensions::default(), ) .await .expect("create source thread from partial history"); let source_path = source .thread .rollout_path() .expect("source rollout path should exist"); let source_history = RolloutRecorder::get_rollout_history(&source_path) .await .expect("read source rollout history"); assert!(snapshot_turn_state(&source_history).ends_mid_turn); manager.remove_thread(&source.thread_id).await; let forked = manager .fork_thread( ForkSnapshot::Interrupted, config.clone(), source_path, /*thread_source*/ None, /*parent_trace*/ None, ) .await .expect("fork interrupted snapshot"); let forked_path = forked .thread .rollout_path() .expect("forked rollout path should exist"); let history = RolloutRecorder::get_rollout_history(&forked_path) .await .expect("read forked rollout history"); assert!(!snapshot_turn_state(&history).ends_mid_turn); let forked_rollout_items: Vec<_> = history .get_rollout_items() .iter() .filter(|item| !matches!(item, RolloutItem::SessionMeta(_))) .collect(); let interrupted_marker_json = serde_json::to_value(RolloutItem::ResponseItem( contextual_user_interrupted_marker().into(), )) .expect("serialize interrupted marker"); assert_eq!( forked_rollout_items .iter() .filter(|item| { strip_response_item_ids_from_json( serde_json::to_value(item).expect("serialize forked rollout item"), ) == interrupted_marker_json }) .count(), 1, ); manager.remove_thread(&forked.thread_id).await; let reforked = manager .fork_thread( ForkSnapshot::Interrupted, config.clone(), forked_path, /*thread_source*/ None, /*parent_trace*/ None, ) .await .expect("re-fork interrupted snapshot"); let reforked_path = reforked .thread .rollout_path() .expect("re-forked rollout path should exist"); let reforked_history = RolloutRecorder::get_rollout_history(&reforked_path) .await .expect("read re-forked rollout history"); let reforked_rollout_items: Vec<_> = reforked_history .get_rollout_items() .iter() .filter(|item| !matches!(item, RolloutItem::SessionMeta(_))) .collect(); assert_eq!( reforked_rollout_items .iter() .filter(|item| { strip_response_item_ids_from_json( serde_json::to_value(item).expect("serialize re-forked rollout item"), ) == interrupted_marker_json }) .count(), 1, ); assert_eq!( reforked_rollout_items .iter() .filter(|item| { matches!( item, RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent { reason: TurnAbortReason::Interrupted, .. })) ) }) .count(), 1, ); }