diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index 82c12ed9e5..ba11862eff 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -739,7 +739,7 @@ pub(super) async fn handle_pending_thread_resume_request( let instruction_sources = pending.instruction_sources; let active_permission_profile = thread_response_active_permission_profile(active_permission_profile); - let session_id = conversation.session_configured().session_id.to_string(); + let session_id = conversation.startup_metadata().session_id.to_string(); thread.session_id = session_id; let response = ThreadResumeResponse { diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 5e2476e590..d79fc35096 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2035,7 +2035,7 @@ impl ThreadRequestProcessor { &self.config.cwd, ); if let Ok(loaded_thread) = self.thread_manager.get_thread(thread_uuid).await { - thread.session_id = loaded_thread.session_configured().session_id.to_string(); + thread.session_id = loaded_thread.startup_metadata().session_id.to_string(); let config_snapshot = loaded_thread.config_snapshot().await; apply_live_thread_settings(&mut thread, &config_snapshot); } @@ -4396,7 +4396,7 @@ impl ThreadRequestProcessor { config_snapshot.model_provider_id.as_str(), /*include_turns*/ false, ); - thread_summary.session_id = existing_thread.session_configured().session_id.to_string(); + thread_summary.session_id = existing_thread.startup_metadata().session_id.to_string(); thread_summary.thread_source = config_snapshot.thread_source.clone().map(Into::into); apply_live_thread_settings(&mut thread_summary, &config_snapshot); thread_summary.can_accept_direct_input = Some(can_accept_direct_input( @@ -4686,7 +4686,7 @@ impl ThreadRequestProcessor { include_turns: bool, ) -> std::result::Result { let config_snapshot = thread.config_snapshot().await; - let session_id = thread.session_configured().session_id.to_string(); + let session_id = thread.startup_metadata().session_id.to_string(); let can_accept_direct_input = can_accept_direct_input( thread.multi_agent_version(), &config_snapshot.session_source, @@ -6323,7 +6323,7 @@ fn build_thread_from_loaded_snapshot( ) -> Thread { build_thread_from_snapshot( thread_id, - loaded_thread.session_configured().session_id.to_string(), + loaded_thread.startup_metadata().session_id.to_string(), loaded_thread.multi_agent_version(), config_snapshot, loaded_thread.rollout_path(), diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index d004fb4fe6..37aca5dfe9 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -1473,7 +1473,7 @@ impl TurnRequestProcessor { } = self .agent_runner .start( - parent_thread.session_configured().thread_id, + parent_thread.startup_metadata().thread_id, AgentInvocation { config, prompt: prompt.to_string(), @@ -1504,7 +1504,7 @@ impl TurnRequestProcessor { if let Some(mut thread) = stored_thread { let config_snapshot = review_thread.config_snapshot().await; apply_live_thread_settings(&mut thread, &config_snapshot); - thread.session_id = review_thread.session_configured().session_id.to_string(); + thread.session_id = review_thread.startup_metadata().session_id.to_string(); self.thread_watch_manager .upsert_thread_silently(&thread.id) .await; diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 15f36cb12a..a7cd6cf38f 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -8,6 +8,7 @@ use crate::session::SessionSettingsUpdate; use crate::session::new_submission_id; use crate::session::session::Session; use crate::session::step_settings::StepSettingsUpdate; +use crate::thread_startup_metadata::ThreadStartupMetadata; use codex_diagnostics::Gauge; use codex_diagnostics::GaugeGuard; use codex_exec_server::SelectedCapabilityRootsStatus; @@ -39,7 +40,6 @@ use codex_protocol::protocol::Event; use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::Op; use codex_protocol::protocol::SandboxPolicy; -use codex_protocol::protocol::SessionConfiguredEvent; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::Submission; use codex_protocol::protocol::ThreadHistoryMode; @@ -184,7 +184,7 @@ pub struct CodexThread { // Registration source controls live access and lifecycle hooks. Managed Guardian // reviewers keep their existing subagent identity inside the session. pub(crate) session_source: SessionSource, - session_configured: SessionConfiguredEvent, + startup_metadata: ThreadStartupMetadata, rollout_path: Option, out_of_band_elicitations: Mutex, _diagnostics_guard: GaugeGuard, @@ -210,7 +210,7 @@ impl CodexThread { pub(crate) fn new( session: Arc, io: SessionIo, - session_configured: SessionConfiguredEvent, + startup_metadata: ThreadStartupMetadata, rollout_path: Option, session_source: SessionSource, ) -> Self { @@ -218,7 +218,7 @@ impl CodexThread { session, io, session_source, - session_configured, + startup_metadata, rollout_path, out_of_band_elicitations: Mutex::new(OutOfBandElicitations::default()), _diagnostics_guard: LIVE_THREADS.track(), @@ -707,8 +707,9 @@ impl CodexThread { self.rollout_path.clone() } - pub fn session_configured(&self) -> SessionConfiguredEvent { - self.session_configured.clone() + /// Returns startup metadata without the one-time initial message replay. + pub fn startup_metadata(&self) -> &ThreadStartupMetadata { + &self.startup_metadata } pub(crate) fn is_running(&self) -> bool { diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index ea8e1b29f5..b5d16bb817 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -34,6 +34,7 @@ mod compact_model_fallback; mod compact_remote_history; mod compact_remote_v2; mod compact_token_budget; +mod thread_startup_metadata; pub use codex_network_proxy::EnvironmentNetworkPolicy; pub use codex_network_proxy::NetworkDomainPermission; pub use codex_network_proxy::NetworkDomainPermissionEntry; @@ -50,6 +51,7 @@ pub use codex_thread::GuardianRootMessage; pub use codex_thread::GuardianRootSnapshot; pub use codex_thread::ThreadConfigSnapshot; pub use session::turn_context::TurnContext; +pub use thread_startup_metadata::ThreadStartupMetadata; mod agent; mod agent_communication; mod attestation; diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 50807e628c..4e4e37ecbc 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -21,6 +21,7 @@ use crate::session::resolve_multi_agent_version; use crate::session::session::Session; use crate::tasks::InterruptedTurnHistoryMarker; use crate::tasks::interrupted_turn_history_marker; +use crate::thread_startup_metadata::ThreadStartupMetadata; use codex_agent_graph_store::AgentGraphStore; use codex_agent_graph_store::LocalAgentGraphStore; use codex_analytics::AnalyticsEventsClient; @@ -2061,9 +2062,13 @@ impl ThreadManagerState { resumed.conversation_id ))); } + drop(threads); + let session_configured = thread + .startup_metadata() + .to_session_configured_event(initial_history.get_event_msgs()); return Ok(NewThread { thread_id: resumed.conversation_id, - session_configured: thread.session_configured(), + session_configured, thread, }); } @@ -2278,7 +2283,7 @@ impl ThreadManagerState { let thread = Arc::new(CodexThread::new( session, io, - session_configured.clone(), + ThreadStartupMetadata::from(&session_configured), session_configured.rollout_path.clone(), session_source, )); diff --git a/codex-rs/core/src/thread_startup_metadata.rs b/codex-rs/core/src/thread_startup_metadata.rs new file mode 100644 index 0000000000..a7f47a9f26 --- /dev/null +++ b/codex-rs/core/src/thread_startup_metadata.rs @@ -0,0 +1,110 @@ +//! Retained startup metadata, separate from the replay sent in startup responses. + +use codex_protocol::SessionId; +use codex_protocol::ThreadId; +use codex_protocol::config_types::ApprovalsReviewer; +use codex_protocol::models::ActivePermissionProfile; +use codex_protocol::models::PermissionProfile; +use codex_protocol::openai_models::ReasoningEffort; +use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::SessionConfiguredEvent; +use codex_protocol::protocol::SessionNetworkProxyRuntime; +use codex_protocol::protocol::ThreadSource; +use codex_utils_absolute_path::AbsolutePathBuf; +use std::path::PathBuf; + +/// The configuration reported when a thread started, without its replay history. +/// Current settings are available through [`crate::CodexThread::config_snapshot`]. +#[derive(Debug)] +pub struct ThreadStartupMetadata { + pub session_id: SessionId, + pub thread_id: ThreadId, + forked_from_id: Option, + parent_thread_id: Option, + thread_source: Option, + thread_name: Option, + model: String, + model_provider_id: String, + service_tier: Option, + approval_policy: AskForApproval, + approvals_reviewer: ApprovalsReviewer, + permission_profile: PermissionProfile, + active_permission_profile: Option, + cwd: AbsolutePathBuf, + reasoning_effort: Option, + network_proxy: Option, + rollout_path: Option, +} + +impl From<&SessionConfiguredEvent> for ThreadStartupMetadata { + fn from(event: &SessionConfiguredEvent) -> Self { + let SessionConfiguredEvent { + session_id, + thread_id, + forked_from_id, + parent_thread_id, + thread_source, + thread_name, + model, + model_provider_id, + service_tier, + approval_policy, + approvals_reviewer, + permission_profile, + active_permission_profile, + cwd, + reasoning_effort, + initial_messages: _, + network_proxy, + rollout_path, + } = event; + Self { + session_id: *session_id, + thread_id: *thread_id, + forked_from_id: *forked_from_id, + parent_thread_id: *parent_thread_id, + thread_source: thread_source.clone(), + thread_name: thread_name.clone(), + model: model.clone(), + model_provider_id: model_provider_id.clone(), + service_tier: service_tier.clone(), + approval_policy: *approval_policy, + approvals_reviewer: *approvals_reviewer, + permission_profile: permission_profile.clone(), + active_permission_profile: active_permission_profile.clone(), + cwd: cwd.clone(), + reasoning_effort: reasoning_effort.clone(), + network_proxy: network_proxy.clone(), + rollout_path: rollout_path.clone(), + } + } +} + +impl ThreadStartupMetadata { + pub(crate) fn to_session_configured_event( + &self, + initial_messages: Option>, + ) -> SessionConfiguredEvent { + SessionConfiguredEvent { + session_id: self.session_id, + thread_id: self.thread_id, + forked_from_id: self.forked_from_id, + parent_thread_id: self.parent_thread_id, + thread_source: self.thread_source.clone(), + thread_name: self.thread_name.clone(), + model: self.model.clone(), + model_provider_id: self.model_provider_id.clone(), + service_tier: self.service_tier.clone(), + approval_policy: self.approval_policy, + approvals_reviewer: self.approvals_reviewer, + permission_profile: self.permission_profile.clone(), + active_permission_profile: self.active_permission_profile.clone(), + cwd: self.cwd.clone(), + reasoning_effort: self.reasoning_effort.clone(), + initial_messages, + network_proxy: self.network_proxy.clone(), + rollout_path: self.rollout_path.clone(), + } + } +} diff --git a/codex-rs/core/tests/suite/guardian_checkpoint_migration_tests.rs b/codex-rs/core/tests/suite/guardian_checkpoint_migration_tests.rs index 5595389a54..93fb49d04f 100644 --- a/codex-rs/core/tests/suite/guardian_checkpoint_migration_tests.rs +++ b/codex-rs/core/tests/suite/guardian_checkpoint_migration_tests.rs @@ -44,7 +44,7 @@ async fn saved_history(test: &TestCodex, thread: &CodexThread) -> Result, ) -> Result> { - let thread_id = thread.session_configured().thread_id; + let thread_id = thread.startup_metadata().thread_id; let environments = thread.environment_selections().await; let model = thread.config_snapshot().await.model; thread.shutdown_and_wait().await?; diff --git a/codex-rs/core/tests/suite/guardian_retained_context.rs b/codex-rs/core/tests/suite/guardian_retained_context.rs index d1c073b971..32c502096f 100644 --- a/codex-rs/core/tests/suite/guardian_retained_context.rs +++ b/codex-rs/core/tests/suite/guardian_retained_context.rs @@ -185,7 +185,7 @@ async fn load_context(test: &TestCodex, thread: &CodexThread) -> Result Result Result> { - let thread_id = thread.session_configured().thread_id; + let thread_id = thread.startup_metadata().thread_id; thread.shutdown_and_wait().await?; test.thread_manager.remove_thread(&thread_id).await; let saved = load_context(test, thread).await?; @@ -729,7 +729,7 @@ async fn disabled_capture_stays_incomplete_after_compaction_and_enabled_resume() assert!(!checkpoint.user_messages_complete()); assert_eq!(checkpoint.ordered_entries().count(), 0); - let thread_id = test.codex.session_configured().thread_id; + let thread_id = test.codex.startup_metadata().thread_id; test.codex.shutdown_and_wait().await?; test.thread_manager.remove_thread(&thread_id).await; let items: Vec = serde_json::from_value(serde_json::to_value( @@ -1057,7 +1057,7 @@ async fn standalone_fork_retains_inherited_user_instructions( let prepared = test .thread_store .prepare_fork(PrepareForkParams { - thread_id: worker.session_configured().thread_id, + thread_id: worker.startup_metadata().thread_id, boundary: ForkBoundary::Latest, }) .await?; @@ -1073,7 +1073,7 @@ async fn standalone_fork_retains_inherited_user_instructions( ForkSnapshot::Interrupted, codex_core::StartThreadOptions::new(test.config.clone()), InitialHistory::Resumed(ResumedHistory { - conversation_id: worker.session_configured().thread_id, + conversation_id: worker.startup_metadata().thread_id, history: Arc::new(load_context(&test, &worker).await?), rollout_path: None, }), @@ -1260,7 +1260,7 @@ async fn forked_parent_instructions_do_not_become_local_authorization( tokio::time::timeout(Duration::from_secs(30), child_gate.entered.notified()) .await .context("child did not reach the paused tool call")?; - let child_id = child.session_configured().thread_id.to_string(); + let child_id = child.startup_metadata().thread_id.to_string(); let requests = child_requests.requests(); let child_request = requests .iter() diff --git a/codex-rs/core/tests/suite/resume.rs b/codex-rs/core/tests/suite/resume.rs index e1ae4fadb1..978bdf9f06 100644 --- a/codex-rs/core/tests/suite/resume.rs +++ b/codex-rs/core/tests/suite/resume.rs @@ -1,6 +1,7 @@ use anyhow::Result; use codex_core::TurnInputRequest; use codex_protocol::config_types::WindowsSandboxLevel; +use codex_protocol::mcp::ClientMcpExtensions; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::user_input::ByteRange; @@ -81,10 +82,11 @@ async fn resume_includes_initial_messages_from_rollout_events() -> Result<()> { .await?; wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await; - let resumed = builder.restart(&server, &initial).await?; + let mut resumed = builder.restart(&server, &initial).await?; let initial_messages = resumed .session_configured .initial_messages + .take() .expect("expected initial messages to be present for resumed session"); match initial_messages.as_slice() { [ @@ -106,6 +108,32 @@ async fn resume_includes_initial_messages_from_rollout_events() -> Result<()> { other => panic!("unexpected initial messages after resume: {other:#?}"), } + resumed.codex.flush_rollout().await?; + let mut rejoined = resumed + .thread_manager + .resume_thread_from_rollout( + resumed.config.clone(), + resumed.codex.rollout_path().expect("resumed rollout path"), + resumed.thread_manager.auth_manager(), + /*parent_trace*/ None, + ClientMcpExtensions::default(), + ) + .await?; + assert!(Arc::ptr_eq(&rejoined.thread, &resumed.codex)); + let rejoined_messages = rejoined + .session_configured + .initial_messages + .take() + .expect("rejoining a loaded thread must still provide replay messages"); + assert_eq!( + serde_json::to_value(&rejoined.session_configured)?, + serde_json::to_value(&resumed.session_configured)?, + ); + assert_eq!( + serde_json::to_value(&rejoined_messages[..initial_messages.len()])?, + serde_json::to_value(&initial_messages)?, + ); + Ok(()) } diff --git a/codex-rs/ext/queue/src/service.rs b/codex-rs/ext/queue/src/service.rs index e72f0c7cb7..06aeb63b4c 100644 --- a/codex-rs/ext/queue/src/service.rs +++ b/codex-rs/ext/queue/src/service.rs @@ -371,7 +371,7 @@ impl QueuedItemService { queued_item_id: Option, trace: Option, ) -> Result { - let thread_id = thread.session_configured().thread_id; + let thread_id = thread.startup_metadata().thread_id; let _dispatch_guard = self.dispatch_guard(thread_id).await; let item = self .list(thread_id)