Separate thread startup metadata from replay history (#46294)

## Why

`CodexThread` retained the full `SessionConfiguredEvent`, including initial replay messages, and cloned it even for callers that only needed thread or session IDs.

## What changed

Store `ThreadStartupMetadata` without replay history and expose it by reference through `startup_metadata()`. Update callers to read IDs from this metadata. When rejoining a running thread, rebuild the startup response using the supplied resume history for `initial_messages`.

## Testing

Extend the resume test to verify that rejoining reuses the loaded thread, preserves startup metadata, and still returns the original replay messages as a prefix.

GitOrigin-RevId: 9d656063088c28f091d29e5862e4b9d894b7642c
This commit is contained in:
Charlie Marsh
2026-09-17 19:55:20 +00:00
committed by copyberry
parent 47fc8d661e
commit 17baabd01b
11 changed files with 171 additions and 25 deletions
@@ -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 {
@@ -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<Thread, String> {
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(),
@@ -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;
+7 -6
View File
@@ -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<PathBuf>,
out_of_band_elicitations: Mutex<OutOfBandElicitations>,
_diagnostics_guard: GaugeGuard,
@@ -210,7 +210,7 @@ impl CodexThread {
pub(crate) fn new(
session: Arc<Session>,
io: SessionIo,
session_configured: SessionConfiguredEvent,
startup_metadata: ThreadStartupMetadata,
rollout_path: Option<PathBuf>,
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 {
+2
View File
@@ -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;
+7 -2
View File
@@ -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,
));
@@ -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<ThreadId>,
parent_thread_id: Option<ThreadId>,
thread_source: Option<ThreadSource>,
thread_name: Option<String>,
model: String,
model_provider_id: String,
service_tier: Option<String>,
approval_policy: AskForApproval,
approvals_reviewer: ApprovalsReviewer,
permission_profile: PermissionProfile,
active_permission_profile: Option<ActivePermissionProfile>,
cwd: AbsolutePathBuf,
reasoning_effort: Option<ReasoningEffort>,
network_proxy: Option<SessionNetworkProxyRuntime>,
rollout_path: Option<PathBuf>,
}
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<Vec<EventMsg>>,
) -> 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(),
}
}
}
@@ -44,7 +44,7 @@ async fn saved_history(test: &TestCodex, thread: &CodexThread) -> Result<Vec<Rol
Ok(test
.thread_store
.load_latest_model_context(LoadThreadHistoryParams {
thread_id: thread.session_configured().thread_id,
thread_id: thread.startup_metadata().thread_id,
include_archived: false,
})
.await?
@@ -56,7 +56,7 @@ pub(super) async fn resume(
thread: &CodexThread,
history: Vec<RolloutItem>,
) -> Result<Arc<CodexThread>> {
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?;
@@ -185,7 +185,7 @@ async fn load_context(test: &TestCodex, thread: &CodexThread) -> Result<Vec<Roll
Ok(test
.thread_store
.load_latest_model_context(LoadThreadHistoryParams {
thread_id: thread.session_configured().thread_id,
thread_id: thread.startup_metadata().thread_id,
include_archived: false,
})
.await?
@@ -193,7 +193,7 @@ async fn load_context(test: &TestCodex, thread: &CodexThread) -> Result<Vec<Roll
}
async fn resume(test: &TestCodex, thread: &CodexThread) -> Result<Arc<CodexThread>> {
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<RolloutItem> = 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()
+29 -1
View File
@@ -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(())
}
+1 -1
View File
@@ -371,7 +371,7 @@ impl QueuedItemService {
queued_item_id: Option<String>,
trace: Option<W3cTraceContext>,
) -> Result<StartIfIdleSubmission, QueueServiceError> {
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)