diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index a78ce21eb9..f55b252c93 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server/tests/common/rollout.rs b/codex-rs/app-server/tests/common/rollout.rs index 9aebb3f870..1c36d944aa 100644 --- a/codex-rs/app-server/tests/common/rollout.rs +++ b/codex-rs/app-server/tests/common/rollout.rs @@ -256,6 +256,7 @@ fn create_fake_rollout_with_source_and_parent_thread_id( session_id, id: conversation_id, forked_from_id: None, + forked_from_ordinal_exclusive: None, parent_thread_id, timestamp: meta_rfc3339.to_string(), cwd: test_path_buf("/"), @@ -348,6 +349,7 @@ pub fn create_fake_rollout_with_text_elements( session_id: conversation_id.into(), id: conversation_id, forked_from_id: None, + forked_from_ordinal_exclusive: None, parent_thread_id: None, timestamp: meta_rfc3339.to_string(), cwd: test_path_buf("/"), diff --git a/codex-rs/app-server/tests/suite/v2/client_metadata.rs b/codex-rs/app-server/tests/suite/v2/client_metadata.rs index 67b1412039..28f441fd4c 100644 --- a/codex-rs/app-server/tests/suite/v2/client_metadata.rs +++ b/codex-rs/app-server/tests/suite/v2/client_metadata.rs @@ -116,6 +116,7 @@ async fn turn_start_forwards_client_metadata_to_responses_request_v2() -> Result metadata["window_id"].as_str(), request.header("x-codex-window-id").as_deref() ); + assert_eq!(metadata["window_number"].as_u64(), Some(0)); assert!( metadata["context_window_id"] .as_str() @@ -193,6 +194,7 @@ async fn turn_start_sends_fork_lineage_in_turn_metadata_for_thread_fork_v2() -> metadata["forked_from_thread_id"].as_str(), Some(source_thread_id.as_str()) ); + assert!(metadata.get("forked_from_ordinal_exclusive").is_none()); assert_eq!(metadata["thread_id"].as_str(), Some(thread.id.as_str())); assert_eq!(metadata["turn_id"].as_str(), Some(turn.id.as_str())); diff --git a/codex-rs/app-server/tests/suite/v2/thread_fork.rs b/codex-rs/app-server/tests/suite/v2/thread_fork.rs index 5a47d56f1c..5d6729bfbd 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -1442,6 +1442,19 @@ async fn thread_fork_creates_reference_backed_paginated_thread() -> Result<()> { .find(|request| request.url.path().ends_with("/responses")) .expect("forked turn response request"); let request_body = response_request.body_json::()?; + let turn_metadata: Value = serde_json::from_str( + request_body["client_metadata"]["x-codex-turn-metadata"] + .as_str() + .expect("forked turn metadata"), + )?; + assert_eq!( + turn_metadata["forked_from_thread_id"].as_str(), + Some(conversation_id.as_str()) + ); + assert_eq!( + turn_metadata["forked_from_ordinal_exclusive"].as_u64(), + Some(history_base.end_ordinal_exclusive) + ); let model_input = request_body["input"] .as_array() .expect("response input array"); @@ -1702,6 +1715,11 @@ async fn assert_thread_fork_freezes_active_paginated_turn_as_interrupted( .await?; let forked_thread_id = forked_thread.id.clone(); let forked_path = forked_thread.path.expect("forked rollout path"); + let history_base = read_session_meta_line(forked_path.as_path()) + .await? + .meta + .history_base + .expect("fork history base"); let child_rollout = std::fs::read_to_string(forked_path.as_path())? .lines() .map(serde_json::from_str::) @@ -1933,6 +1951,19 @@ async fn assert_thread_fork_freezes_active_paginated_turn_as_interrupted( .find(|request| request.url.path().ends_with("/responses")) .expect("cold-resumed model request") .body_json::()?; + let turn_metadata: Value = serde_json::from_str( + request_body["client_metadata"]["x-codex-turn-metadata"] + .as_str() + .expect("cold-resumed turn metadata"), + )?; + assert_eq!( + turn_metadata["forked_from_thread_id"].as_str(), + Some(source_thread_id.as_str()) + ); + assert_eq!( + turn_metadata["forked_from_ordinal_exclusive"].as_u64(), + Some(history_base.end_ordinal_exclusive) + ); let model_input = request_body["input"].as_array().expect("model input"); assert!(model_input.iter().any(|item| { item["role"] == expected_marker_role diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index f0f5f75351..807705667f 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -3766,6 +3766,7 @@ async fn thread_resume_prefers_persisted_git_metadata_for_local_threads() -> Res session_id: conversation_id.into(), id: conversation_id, forked_from_id: None, + forked_from_ordinal_exclusive: None, parent_thread_id: None, timestamp: "2025-01-05T12:00:00Z".to_string(), cwd: repo_path.clone(), diff --git a/codex-rs/app-server/tests/suite/v2/thread_revert.rs b/codex-rs/app-server/tests/suite/v2/thread_revert.rs index 7287d2b2bf..5ce27011e1 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_revert.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_revert.rs @@ -13,6 +13,8 @@ use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::JSONRPCMessage; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::SortDirection; +use codex_app_server_protocol::ThreadForkParams; +use codex_app_server_protocol::ThreadForkResponse; use codex_app_server_protocol::ThreadHistoryMode; use codex_app_server_protocol::ThreadItemsListParams; use codex_app_server_protocol::ThreadItemsListResponse; @@ -34,12 +36,170 @@ use codex_protocol::config_types::CollaborationMode; use codex_protocol::config_types::ModeKind; use codex_protocol::config_types::Settings; use codex_protocol::openai_models::ReasoningEffort; +use codex_protocol::protocol::EventMsg; +use codex_rollout::RolloutItem; +use codex_rollout::RolloutLine; +use codex_rollout::read_session_meta_line; use pretty_assertions::assert_eq; +use serde_json::Value; use tempfile::TempDir; use tokio::time::timeout; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +#[tokio::test] +async fn thread_revert_preserves_fork_cutoff_after_cold_resume() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + MockResponsesConfig::new(&server.uri()).write(codex_home.path())?; + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build() + .await?; + initialize_experimental(&mut mcp).await?; + let ThreadStartResponse { thread: parent, .. } = mcp + .start_thread(ThreadStartParams { + history_mode: Some(ThreadHistoryMode::Paginated), + ..Default::default() + }) + .await?; + let mut parent_turns = Vec::new(); + for text in ["parent first", "parent second"] { + let completed = mcp + .start_turn_and_wait_for_completion(TurnStartParams { + thread_id: parent.id.clone(), + input: vec![UserInput::Text { + text: text.to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + parent_turns.push(completed.turn.id); + } + let ThreadForkResponse { thread: child, .. } = mcp + .request(|request_id| ClientRequest::ThreadFork { + request_id, + params: ThreadForkParams { + thread_id: parent.id.clone(), + ..Default::default() + }, + }) + .await?; + let child_meta = read_session_meta_line(child.path.as_ref().expect("child rollout")) + .await? + .meta; + let fork_cutoff = child_meta + .history_base + .expect("fork history base") + .end_ordinal_exclusive; + assert_eq!(child_meta.forked_from_ordinal_exclusive, Some(fork_cutoff)); + let inherited_revert_cutoff = + std::fs::read_to_string(parent.path.as_ref().expect("parent rollout"))? + .lines() + .map(serde_json::from_str::) + .collect::, _>>()? + .into_iter() + .find_map(|line| match line.item { + RolloutItem::EventMsg(EventMsg::TurnStarted(turn)) + if turn.turn_id == parent_turns[1] => + { + line.ordinal + } + _ => None, + }) + .expect("inherited turn start ordinal"); + let mut child_turns = Vec::new(); + for text in ["child first", "child second"] { + let completed = mcp + .start_turn_and_wait_for_completion(TurnStartParams { + thread_id: child.id.clone(), + input: vec![UserInput::Text { + text: text.to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + child_turns.push(completed.turn.id); + } + + // First revert within the child, then revert into its inherited parent history. + for (before_turn_id, expected_cutoff) in [ + (child_turns[1].clone(), fork_cutoff), + (parent_turns[1].clone(), inherited_revert_cutoff), + ] { + let ThreadRevertResponse { + thread: reverted, .. + } = mcp + .request(|request_id| ClientRequest::ThreadRevert { + request_id, + params: ThreadRevertParams { + thread_id: child.id.clone(), + before_turn_id, + }, + }) + .await?; + let meta = read_session_meta_line(reverted.path.as_ref().expect("reverted rollout")) + .await? + .meta; + assert_eq!(meta.forked_from_ordinal_exclusive, Some(expected_cutoff)); + if expected_cutoff == fork_cutoff { + assert!( + meta.history_base + .expect("child revert base") + .end_ordinal_exclusive + > fork_cutoff + ); + } + + mcp.shutdown_gracefully().await?; + mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build() + .await?; + initialize_experimental(&mut mcp).await?; + let _: ThreadResumeResponse = mcp + .request(|request_id| ClientRequest::ThreadResume { + request_id, + params: ThreadResumeParams { + thread_id: child.id.clone(), + ..Default::default() + }, + }) + .await?; + mcp.start_turn_and_wait_for_completion(TurnStartParams { + thread_id: child.id.clone(), + input: vec![UserInput::Text { + text: "continue after revert and cold resume".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + let requests = server.received_requests().await.expect("response requests"); + let body = requests + .iter() + .rev() + .find(|request| request.url.path().ends_with("/responses")) + .expect("resumed model request") + .body_json::()?; + let metadata: Value = serde_json::from_str( + body["client_metadata"]["x-codex-turn-metadata"] + .as_str() + .expect("turn metadata"), + )?; + assert_eq!( + ( + metadata["forked_from_thread_id"].as_str(), + metadata["forked_from_ordinal_exclusive"].as_u64() + ), + (Some(parent.id.as_str()), Some(expected_cutoff)) + ); + } + Ok(()) +} + #[tokio::test] async fn thread_revert_replaces_paginated_history_before_turn() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; diff --git a/codex-rs/core/src/responses_metadata.rs b/codex-rs/core/src/responses_metadata.rs index 61d67d3dfd..698e5a0ca2 100644 --- a/codex-rs/core/src/responses_metadata.rs +++ b/codex-rs/core/src/responses_metadata.rs @@ -30,6 +30,7 @@ pub(crate) const THREAD_ID_KEY: &str = "thread_id"; pub(crate) const AGENT_NAME_KEY: &str = "agent_name"; pub(crate) const TURN_ID_KEY: &str = "turn_id"; pub(crate) const WINDOW_ID_KEY: &str = "window_id"; +pub(crate) const WINDOW_NUMBER_KEY: &str = "window_number"; pub(crate) const CONTEXT_WINDOW_ID_KEY: &str = "context_window_id"; pub(crate) const REQUEST_KIND_KEY: &str = "request_kind"; pub(crate) const COMPACTION_KEY: &str = "compaction"; @@ -39,6 +40,7 @@ pub(crate) const TOOL_NAMESPACES_INFO_KEY: &str = "tool_namespaces_info"; pub(crate) const TURN_STARTED_AT_UNIX_MS_KEY: &str = "turn_started_at_unix_ms"; pub(crate) const FORKED_FROM_THREAD_ID_KEY: &str = "forked_from_thread_id"; +pub(crate) const FORKED_FROM_ORDINAL_EXCLUSIVE_KEY: &str = "forked_from_ordinal_exclusive"; pub(crate) const PARENT_THREAD_ID_KEY: &str = "parent_thread_id"; pub(crate) const PARENT_TURN_ID_KEY: &str = "parent_turn_id"; pub(crate) const ROOT_TURN_ID_KEY: &str = "root_turn_id"; @@ -62,6 +64,7 @@ const RESERVED_METADATA_KEYS: &[&str] = &[ AGENT_NAME_KEY, TURN_ID_KEY, WINDOW_ID_KEY, + WINDOW_NUMBER_KEY, CONTEXT_WINDOW_ID_KEY, X_CODEX_WINDOW_ID_HEADER, X_CODEX_TURN_METADATA_HEADER, @@ -73,6 +76,7 @@ const RESERVED_METADATA_KEYS: &[&str] = &[ TOOL_NAMESPACES_INFO_KEY, TURN_STARTED_AT_UNIX_MS_KEY, FORKED_FROM_THREAD_ID_KEY, + FORKED_FROM_ORDINAL_EXCLUSIVE_KEY, PARENT_THREAD_ID_KEY, PARENT_TURN_ID_KEY, ROOT_TURN_ID_KEY, @@ -86,6 +90,10 @@ const RESERVED_METADATA_KEYS: &[&str] = &[ NODE_REPL_DISABLED_KEY, WORKSPACES_KEY, ]; +// These keys were previously valid user configuration. Accept existing configs while filtering +// their values before constructing Core-owned request metadata. +const BACKWARD_COMPATIBLE_RESERVED_METADATA_KEYS: &[&str] = + &[WINDOW_NUMBER_KEY, FORKED_FROM_ORDINAL_EXCLUSIVE_KEY]; const MAX_EXTRA_METADATA_ENTRIES: usize = 16; const MAX_EXTRA_METADATA_KEY_BYTES: usize = 64; const MAX_EXTRA_METADATA_VALUE_BYTES: usize = 128; @@ -214,9 +222,11 @@ pub struct CodexResponsesMetadata { pub(crate) turn_id: Option, pub(crate) routing_hint: Option, pub(crate) window_id: String, + pub(crate) window_number: Option, pub(crate) context_window_id: Option, pub(crate) request_kind: Option, pub(crate) forked_from_thread_id: Option, + pub(crate) forked_from_ordinal_exclusive: Option, pub(crate) parent_thread_id: Option, pub(crate) parent_turn_id: Option, pub(crate) root_turn_id: Option, @@ -250,9 +260,11 @@ impl CodexResponsesMetadata { turn_id: None, routing_hint: None, window_id, + window_number: None, context_window_id: None, request_kind: None, forked_from_thread_id: None, + forked_from_ordinal_exclusive: None, parent_thread_id: None, parent_turn_id: None, root_turn_id: None, @@ -374,11 +386,13 @@ impl CodexResponsesMetadata { .then_some(self.turn_id.as_deref()) .flatten(), window_id: has_request_identity.then_some(self.window_id.as_str()), + window_number: has_request_identity.then_some(self.window_number).flatten(), context_window_id: has_request_identity .then_some(self.context_window_id) .flatten(), request_kind: request_kind_value, forked_from_thread_id: self.forked_from_thread_id, + forked_from_ordinal_exclusive: self.forked_from_ordinal_exclusive, parent_thread_id: self.parent_thread_id, parent_turn_id: self.parent_turn_id.as_deref(), root_turn_id: self.root_turn_id.as_deref(), @@ -452,7 +466,9 @@ pub(crate) fn validate_extra_metadata<'a>( if key.len() > MAX_EXTRA_METADATA_KEY_BYTES || !valid_extra_metadata_key(key) { return Err("responses_api_metadata keys must be short ASCII identifiers"); } - if RESERVED_METADATA_KEYS.contains(&key.as_str()) { + if RESERVED_METADATA_KEYS.contains(&key.as_str()) + && !BACKWARD_COMPATIBLE_RESERVED_METADATA_KEYS.contains(&key.as_str()) + { return Err("responses_api_metadata contains a reserved key"); } if value.len() > MAX_EXTRA_METADATA_VALUE_BYTES { @@ -498,12 +514,16 @@ struct CodexTurnMetadataPayload<'a> { #[serde(default, skip_serializing_if = "Option::is_none")] window_id: Option<&'a str>, #[serde(default, skip_serializing_if = "Option::is_none")] + window_number: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] context_window_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] request_kind: Option<&'static str>, #[serde(default, skip_serializing_if = "Option::is_none")] forked_from_thread_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] + forked_from_ordinal_exclusive: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] parent_thread_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] parent_turn_id: Option<&'a str>, diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index c48c016b61..04d47879e0 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -3888,12 +3888,16 @@ impl Session { self.current_window().await.0 } - pub(crate) async fn current_window(&self) -> (String, Uuid) { + pub(crate) async fn current_window(&self) -> (String, u64, Uuid) { let state = self.state.lock().await; let thread_id = self.thread_id; let window_number = state.auto_compact_window_number(); let context_window_id = state.auto_compact_window_ids().window_id; - (format!("{thread_id}:{window_number}"), context_window_id) + ( + format!("{thread_id}:{window_number}"), + window_number, + context_window_id, + ) } pub(crate) async fn advance_auto_compact_window(&self) -> (u64, AutoCompactWindowIds) { diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index ffbec36a73..e0250aee6d 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -68,6 +68,7 @@ pub(crate) struct Session { pub(crate) services: SessionServices, pub(super) git_enrichment_policy: GitEnrichmentPolicy, pub(super) fork_persistence: ForkPersistence, + pub(super) forked_from_ordinal_exclusive: Option, pub(super) next_internal_sub_id: AtomicU64, } @@ -599,14 +600,19 @@ impl Session { turn_context: &TurnContext, request_kind: CodexResponsesRequestKind, ) -> CodexResponsesMetadata { - let (window_id, context_window_id) = self.current_window().await; + let (window_id, window_number, context_window_id) = self.current_window().await; + let responses_metadata = turn_context.turn_metadata_state.to_responses_metadata( + self.installation_id.clone(), + window_id, + request_kind, + ); CodexResponsesMetadata { + window_number: Some(window_number), context_window_id: Some(context_window_id), - ..turn_context.turn_metadata_state.to_responses_metadata( - self.installation_id.clone(), - window_id, - request_kind, - ) + forked_from_ordinal_exclusive: self + .forked_from_ordinal_exclusive + .filter(|_| responses_metadata.forked_from_thread_id.is_some()), + ..responses_metadata } } @@ -680,6 +686,31 @@ impl Session { .forked_from_thread_id .or_else(|| initial_history.forked_from_id()); session_configuration.forked_from_thread_id = forked_from_id; + let forked_from_ordinal_exclusive = match &fork_persistence { + ForkPersistence::Referenced { history_base, .. } => { + history_base.map(|position| position.end_ordinal_exclusive) + } + ForkPersistence::Copied => match &initial_history { + InitialHistory::Resumed(resumed) => { + // Both local and CCA thread stores place the resumed thread's + // canonical SessionMeta first. Never inspect inherited metadata: + // an ancestor's history_base describes a different fork boundary. + resumed.history.first().and_then(|item| match item { + RolloutItem::SessionMeta(meta) + if meta.meta.id == resumed.conversation_id => + { + codex_rollout::forked_from_ordinal_exclusive( + &meta.meta, + resumed.rollout_path.as_deref(), + ) + } + _ => None, + }) + } + InitialHistory::New | InitialHistory::Cleared | InitialHistory::Forked(_) => None, + }, + } + .filter(|_| forked_from_id.is_some()); let parent_thread_id = session_configuration .parent_thread_id .or_else(|| initial_history.get_resumed_parent_thread_id()); @@ -1430,6 +1461,7 @@ impl Session { services, git_enrichment_policy, fork_persistence, + forked_from_ordinal_exclusive, next_internal_sub_id: AtomicU64::new(0), }); if let Some(network_policy_decider_session) = network_policy_decider_session { diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 88ddab0c66..3889978a0a 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -141,6 +141,7 @@ use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::ConversationAudioParams; use codex_protocol::protocol::CreditsSnapshot; use codex_protocol::protocol::GranularApprovalConfig; +use codex_protocol::protocol::HistoryPosition; use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::NetworkApprovalProtocol; @@ -6314,6 +6315,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { services, git_enrichment_policy: GitEnrichmentPolicy::Fresh, fork_persistence: ForkPersistence::Copied, + forked_from_ordinal_exclusive: None, next_internal_sub_id: AtomicU64::new(0), }; let per_turn_config = @@ -6709,6 +6711,52 @@ async fn resumed_subagent_session_restores_persisted_session_id() { assert_eq!(event.thread_id, thread_id); } +#[tokio::test] +async fn resumed_copied_fork_ignores_source_history_base() { + let ancestor_thread_id = ThreadId::new(); + let parent_thread_id = ThreadId::new(); + let thread_id = ThreadId::new(); + let history = vec![ + RolloutItem::SessionMeta(SessionMetaLine { + meta: SessionMeta { + id: thread_id, + session_id: SessionId::from(thread_id), + forked_from_id: Some(parent_thread_id), + ..SessionMeta::default() + }, + git: None, + }), + RolloutItem::SessionMeta(SessionMetaLine { + meta: SessionMeta { + id: parent_thread_id, + session_id: SessionId::from(parent_thread_id), + forked_from_id: Some(ancestor_thread_id), + history_base: Some(HistoryPosition { + thread_id: ancestor_thread_id, + end_ordinal_exclusive: 42, + end_byte_offset: 100, + }), + ..SessionMeta::default() + }, + git: None, + }), + ]; + let (session, _rx_event) = make_session_with_history_source_and_agent_control_and_rx( + InitialHistory::Resumed(ResumedHistory { + conversation_id: thread_id, + history: Arc::new(history), + rollout_path: None, + }), + SessionSource::Exec, + AgentControl::default(), + ) + .await + .expect("resume should succeed"); + + assert_eq!(session.thread_id(), thread_id); + assert_eq!(session.forked_from_ordinal_exclusive, None); +} + #[tokio::test] async fn notify_request_permissions_response_ignores_unmatched_call_id() { let (session, _turn_context) = make_session_and_context().await; @@ -8545,6 +8593,7 @@ where services, git_enrichment_policy: GitEnrichmentPolicy::Fresh, fork_persistence: ForkPersistence::Copied, + forked_from_ordinal_exclusive: None, next_internal_sub_id: AtomicU64::new(0), }); let per_turn_config = diff --git a/codex-rs/core/src/session_startup_prewarm.rs b/codex-rs/core/src/session_startup_prewarm.rs index 70a9efea2f..4b61d92b83 100644 --- a/codex-rs/core/src/session_startup_prewarm.rs +++ b/codex-rs/core/src/session_startup_prewarm.rs @@ -313,14 +313,9 @@ async fn schedule_startup_prewarm_inner( build_prompt_started_at.elapsed(), /*status*/ None, ); - let window_id = session.current_window_id().await; - let responses_metadata = startup_turn_context - .turn_metadata_state - .to_responses_metadata( - session.installation_id.clone(), - window_id, - CodexResponsesRequestKind::Prewarm, - ); + let responses_metadata = session + .responses_metadata(&startup_turn_context, CodexResponsesRequestKind::Prewarm) + .await; let mut client_session = session.services.model_client.new_session(); let websocket_warmup_started_at = Instant::now(); client_session diff --git a/codex-rs/core/src/turn_metadata.rs b/codex-rs/core/src/turn_metadata.rs index c8c71a5985..afae92f22f 100644 --- a/codex-rs/core/src/turn_metadata.rs +++ b/codex-rs/core/src/turn_metadata.rs @@ -357,7 +357,8 @@ impl TurnMetadataState { *self .responses_api_metadata .write() - .unwrap_or_else(std::sync::PoisonError::into_inner) = responses_api_metadata; + .unwrap_or_else(std::sync::PoisonError::into_inner) = + filter_extra_metadata(responses_api_metadata); } pub(crate) fn workspace_kind(&self) -> Option { @@ -370,6 +371,9 @@ impl TurnMetadataState { fn responses_metadata_template(&self) -> CodexResponsesMetadata { let mut metadata = self.mcp_metadata_template(); + if metadata.parent_thread_id.is_some() { + metadata.forked_from_thread_id = None; + } metadata.extra.extend( self.responses_api_metadata .read() diff --git a/codex-rs/core/src/turn_metadata_tests.rs b/codex-rs/core/src/turn_metadata_tests.rs index 3d27bd8956..132bd644be 100644 --- a/codex-rs/core/src/turn_metadata_tests.rs +++ b/codex-rs/core/src/turn_metadata_tests.rs @@ -4,6 +4,7 @@ use crate::responses_metadata::AUTO_REVIEW_ENABLED_KEY; use crate::responses_metadata::CONTEXT_WINDOW_ID_KEY; use crate::responses_metadata::CodexResponsesRequestKind; use crate::responses_metadata::CompactionTurnMetadata; +use crate::responses_metadata::FORKED_FROM_ORDINAL_EXCLUSIVE_KEY; use crate::responses_metadata::INSTALLATION_ID_KEY; use crate::responses_metadata::LEGACY_CODE_MODE_TOOL_NAMES_KEY; use crate::responses_metadata::NODE_REPL_AUTO_REVIEW_REQUIRED_KEY; @@ -17,6 +18,7 @@ use crate::responses_metadata::TurnToolFunctionInfo; use crate::responses_metadata::TurnToolNamespaceInfo; use crate::responses_metadata::TurnToolSource; use crate::responses_metadata::WINDOW_ID_KEY; +use crate::responses_metadata::WINDOW_NUMBER_KEY; use crate::responses_metadata::validate_extra_metadata; use crate::sandbox_tags::permission_profile_sandbox_tag; use codex_analytics::CompactionImplementation; @@ -342,7 +344,7 @@ fn turn_metadata_state_includes_thread_spawn_subagent_parent_without_fork() { } #[test] -fn turn_metadata_state_includes_forked_thread_spawn_subagent_lineage() { +fn turn_metadata_state_omits_fork_lineage_for_context_inheriting_subagent() { let temp_dir = TempDir::new().expect("temp dir"); let cwd = temp_dir.path().abs(); let permission_profile = PermissionProfile::read_only(); @@ -374,10 +376,8 @@ fn turn_metadata_state_includes_forked_thread_spawn_subagent_lineage() { let header = test_turn_metadata_header(&state); let json: Value = serde_json::from_str(&header).expect("json"); - assert_eq!( - json["forked_from_thread_id"].as_str(), - Some("33333333-3333-4333-8333-333333333333") - ); + assert!(json.get("forked_from_thread_id").is_none()); + assert!(json.get(FORKED_FROM_ORDINAL_EXCLUSIVE_KEY).is_none()); assert_eq!( json["parent_thread_id"].as_str(), Some("33333333-3333-4333-8333-333333333333") @@ -386,6 +386,14 @@ fn turn_metadata_state_includes_forked_thread_spawn_subagent_lineage() { // V1 subagents have no canonical agent path and are intentionally unsupported by // agent-name-addressed history and notes; their metadata falls back to the root agent. assert_eq!(json["agent_name"].as_str(), Some("/root")); + + let mcp_metadata = state + .current_meta_value_for_mcp_request(test_mcp_turn_metadata_context()) + .expect("MCP request metadata"); + assert_eq!( + mcp_metadata["forked_from_thread_id"].as_str(), + Some("33333333-3333-4333-8333-333333333333") + ); } #[test] @@ -679,10 +687,17 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields( /*auto_review_enabled*/ false, &model_info_from_slug("gpt-5.4"), ); - state.set_responses_api_metadata(BTreeMap::from([( - "codex_security_surface".to_string(), - "sdk".to_string(), - )])); + state.set_responses_api_metadata(BTreeMap::from([ + ("codex_security_surface".to_string(), "sdk".to_string()), + ( + WINDOW_NUMBER_KEY.to_string(), + "configured-value".to_string(), + ), + ( + FORKED_FROM_ORDINAL_EXCLUSIVE_KEY.to_string(), + "configured-value".to_string(), + ), + ])); state.set_parent_turn_id("parent-turn-a".to_string()); state.set_root_turn_id("root-turn-a".to_string()); state.set_turn_trigger("goal".to_string()); @@ -736,10 +751,15 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields( ), ("turn_id".to_string(), "client-supplied".to_string()), (WINDOW_ID_KEY.to_string(), "client-supplied".to_string()), + (WINDOW_NUMBER_KEY.to_string(), "client-supplied".to_string()), ( CONTEXT_WINDOW_ID_KEY.to_string(), "client-supplied".to_string(), ), + ( + FORKED_FROM_ORDINAL_EXCLUSIVE_KEY.to_string(), + "client-supplied".to_string(), + ), ("thread_source".to_string(), "client-supplied".to_string()), (TURN_TRIGGER_KEY.to_string(), "client-supplied".to_string()), ("request_kind".to_string(), "client-supplied".to_string()), @@ -807,10 +827,8 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields( assert!(json.get("x-codex-installation-id").is_none()); assert!(json.get("x-codex-parent-thread-id").is_none()); assert!(json.get("x-openai-subagent").is_none()); - assert_eq!( - json["forked_from_thread_id"].as_str(), - Some("44444444-4444-4444-8444-444444444444") - ); + assert!(json.get("forked_from_thread_id").is_none()); + assert!(json.get(FORKED_FROM_ORDINAL_EXCLUSIVE_KEY).is_none()); assert_eq!( json["parent_thread_id"].as_str(), Some("55555555-5555-4555-8555-555555555555") @@ -823,6 +841,7 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields( assert_eq!(json["turn_id"].as_str(), Some("turn-a")); assert!(json.get("request_kind").is_none()); assert!(json.get(WINDOW_ID_KEY).is_none()); + assert!(json.get(WINDOW_NUMBER_KEY).is_none()); assert!(json.get(CONTEXT_WINDOW_ID_KEY).is_none()); assert_eq!( json["turn_started_at_unix_ms"].as_i64(), @@ -974,8 +993,20 @@ fn responses_api_metadata_rejects_reserved_keys() { } } +#[test] +fn responses_api_metadata_accepts_previously_valid_rollout_position_keys() { + for legacy_key in [WINDOW_NUMBER_KEY, FORKED_FROM_ORDINAL_EXCLUSIVE_KEY] { + assert_eq!( + validate_extra_metadata( + BTreeMap::from([(legacy_key.to_string(), "legacy-value".to_string())]).iter() + ), + Ok(()) + ); + } +} + #[tokio::test] -async fn turn_metadata_state_preserves_lineage_after_git_enrichment() { +async fn turn_metadata_state_preserves_subagent_parent_after_git_enrichment() { let (_temp_dir, repo_path) = create_clean_git_repo("repo").await; let permission_profile = PermissionProfile::read_only(); @@ -1006,10 +1037,7 @@ async fn turn_metadata_state_preserves_lineage_after_git_enrichment() { state.spawn_git_enrichment_task(); let json = wait_for_git_enrichment(&state).await; - assert_eq!( - json["forked_from_thread_id"].as_str(), - Some("66666666-6666-4666-8666-666666666666") - ); + assert!(json.get("forked_from_thread_id").is_none()); assert_eq!( json["parent_thread_id"].as_str(), Some("66666666-6666-4666-8666-666666666666") diff --git a/codex-rs/core/tests/suite/agent_websocket.rs b/codex-rs/core/tests/suite/agent_websocket.rs index 3b3b8afa00..cad03435cf 100644 --- a/codex-rs/core/tests/suite/agent_websocket.rs +++ b/codex-rs/core/tests/suite/agent_websocket.rs @@ -217,6 +217,16 @@ async fn websocket_first_turn_uses_startup_prewarm_and_create() -> Result<()> { .expect("turn metadata"), )?; assert_eq!(turn_metadata["request_kind"].as_str(), Some("turn")); + assert_eq!(warmup_metadata["window_number"].as_u64(), Some(0)); + assert_eq!( + warmup_metadata["window_number"], + turn_metadata["window_number"] + ); + assert!(warmup_metadata["context_window_id"].is_string()); + assert_eq!( + warmup_metadata["context_window_id"], + turn_metadata["context_window_id"] + ); server.shutdown().await; Ok(()) diff --git a/codex-rs/core/tests/suite/compact_remote.rs b/codex-rs/core/tests/suite/compact_remote.rs index f0f921cf06..c19f60ea91 100644 --- a/codex-rs/core/tests/suite/compact_remote.rs +++ b/codex-rs/core/tests/suite/compact_remote.rs @@ -51,6 +51,7 @@ use codex_protocol::protocol::Op; use codex_protocol::protocol::RealtimeConversationRealtimeEvent; use codex_protocol::protocol::RealtimeEvent; use codex_protocol::protocol::RealtimeOutputModality; +use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::user_input::UserInput; use core_test_support::PathBufExt; @@ -770,7 +771,9 @@ async fn remote_compact_replaces_history_for_followups() -> Result<()> { skip_if_no_network!(Ok(())); let harness = TestCodexHarness::with_builder( - test_codex().with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()), + test_codex() + .with_history_mode(ThreadHistoryMode::Paginated) + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()), ) .await?; let codex = harness.test().codex.clone(); @@ -865,6 +868,7 @@ async fn remote_compact_replaces_history_for_followups() -> Result<()> { compact_metadata["window_id"].as_str(), compact_request.header("x-codex-window-id").as_deref() ); + assert_eq!(compact_metadata["window_number"].as_u64(), Some(0)); assert!(compact_metadata["context_window_id"].as_str().is_some()); assert_eq!( compact_metadata["compaction"], @@ -1828,6 +1832,7 @@ async fn remote_compact_v2_retries_failures_with_stream_retry_budget() -> Result let harness = TestCodexHarness::with_builder( test_codex() + .with_history_mode(ThreadHistoryMode::Paginated) .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) .with_config(|config| { let _ = config.features.enable(Feature::RemoteCompactionV2); @@ -1896,7 +1901,6 @@ async fn remote_compact_v2_retries_failures_with_stream_retry_budget() -> Result response_requests.len(), "expected initial turn, failed open, failed stream, compact retry, and follow-up turn" ); - for compact_request in &response_requests[1..=3] { assert_eq!("/v1/responses", compact_request.path()); let compact_metadata: Value = serde_json::from_str( @@ -1904,6 +1908,7 @@ async fn remote_compact_v2_retries_failures_with_stream_retry_budget() -> Result .header("x-codex-turn-metadata") .expect("v2 compact request should include turn metadata"), )?; + assert_eq!(compact_metadata["window_number"].as_u64(), Some(0)); assert!(compact_metadata["context_window_id"].as_str().is_some()); assert!( compact_request diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_api_auth_prompt_cache_key_request_diff.snap b/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_api_auth_prompt_cache_key_request_diff.snap index 2dad0171e8..c1da677d74 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_api_auth_prompt_cache_key_request_diff.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_api_auth_prompt_cache_key_request_diff.snap @@ -12,7 +12,7 @@ Scenario: After five varied API-key-auth turns, remote manual compaction omits s - "thread_id": "", - "turn_id": "", - "x-codex-installation-id": "", -- "x-codex-turn-metadata": "{\"installation_id\":\"\",\"session_id\":\"\",\"thread_id\":\"\",\"agent_name\":\"/root\",\"turn_id\":\"\",\"window_id\":\":0\",\"context_window_id\":\"\",\"request_kind\":\"turn\",\"root_turn_id\":\"\",\"sandbox\":\"\",\"sandbox_mode\":\"danger-full-access\",\"auto_review_enabled\":false,\"node_repl_auto_review_required\":false,\"node_repl_disabled\":false,\"turn_started_at_unix_ms\":}", +- "x-codex-turn-metadata": "{\"installation_id\":\"\",\"session_id\":\"\",\"thread_id\":\"\",\"agent_name\":\"/root\",\"turn_id\":\"\",\"window_id\":\":0\",\"window_number\":0,\"context_window_id\":\"\",\"request_kind\":\"turn\",\"root_turn_id\":\"\",\"sandbox\":\"\",\"sandbox_mode\":\"danger-full-access\",\"auto_review_enabled\":false,\"node_repl_auto_review_required\":false,\"node_repl_disabled\":false,\"turn_started_at_unix_ms\":}", - "x-codex-window-id": ":0" - }, - "include": [ diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_chatgpt_auth_service_tier_prompt_cache_key_request_diff.snap b/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_chatgpt_auth_service_tier_prompt_cache_key_request_diff.snap index c4f8bc6830..82a1b699b9 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_chatgpt_auth_service_tier_prompt_cache_key_request_diff.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_chatgpt_auth_service_tier_prompt_cache_key_request_diff.snap @@ -12,7 +12,7 @@ Scenario: After five varied ChatGPT-auth turns, remote manual compaction reuses - "thread_id": "", - "turn_id": "", - "x-codex-installation-id": "", -- "x-codex-turn-metadata": "{\"installation_id\":\"\",\"session_id\":\"\",\"thread_id\":\"\",\"agent_name\":\"/root\",\"turn_id\":\"\",\"window_id\":\":0\",\"context_window_id\":\"\",\"request_kind\":\"turn\",\"root_turn_id\":\"\",\"sandbox\":\"\",\"sandbox_mode\":\"danger-full-access\",\"auto_review_enabled\":false,\"node_repl_auto_review_required\":false,\"node_repl_disabled\":false,\"turn_started_at_unix_ms\":}", +- "x-codex-turn-metadata": "{\"installation_id\":\"\",\"session_id\":\"\",\"thread_id\":\"\",\"agent_name\":\"/root\",\"turn_id\":\"\",\"window_id\":\":0\",\"window_number\":0,\"context_window_id\":\"\",\"request_kind\":\"turn\",\"root_turn_id\":\"\",\"sandbox\":\"\",\"sandbox_mode\":\"danger-full-access\",\"auto_review_enabled\":false,\"node_repl_auto_review_required\":false,\"node_repl_disabled\":false,\"turn_started_at_unix_ms\":}", - "x-codex-window-id": ":0" - }, - "include": [ diff --git a/codex-rs/core/tests/suite/sqlite_state.rs b/codex-rs/core/tests/suite/sqlite_state.rs index a5c8c08923..f28dd835d7 100644 --- a/codex-rs/core/tests/suite/sqlite_state.rs +++ b/codex-rs/core/tests/suite/sqlite_state.rs @@ -417,6 +417,7 @@ async fn backfill_scans_existing_rollouts() -> Result<()> { session_id: thread_id.into(), id: thread_id, forked_from_id: None, + forked_from_ordinal_exclusive: None, parent_thread_id: None, timestamp: "2026-01-27T12:00:00Z".to_string(), cwd: codex_home.to_path_buf(), diff --git a/codex-rs/core/tests/suite/window_headers.rs b/codex-rs/core/tests/suite/window_headers.rs index a0a2312e93..22bbd2d8e0 100644 --- a/codex-rs/core/tests/suite/window_headers.rs +++ b/codex-rs/core/tests/suite/window_headers.rs @@ -129,6 +129,13 @@ async fn window_id_advances_after_compact_persists_on_resume_and_resets_on_fork( metadata[2]["context_window_id"], metadata[3]["context_window_id"] ); + assert_eq!( + metadata + .iter() + .map(|metadata| metadata["window_number"].as_u64()) + .collect::>(), + vec![Some(0), Some(0), Some(1), Some(1), Some(0)] + ); Ok(()) } diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 3d09b2f548..0ab7c8d6df 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -2976,6 +2976,10 @@ pub struct SessionMeta { pub id: ThreadId, #[serde(skip_serializing_if = "Option::is_none")] pub forked_from_id: Option, + /// Exclusive ordinal inherited from the logical fork parent, independent of `history_base`. + /// Revert may replace the physical history base while retaining this fork boundary. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub forked_from_ordinal_exclusive: Option, #[serde(skip_serializing_if = "Option::is_none")] pub parent_thread_id: Option, pub timestamp: String, @@ -3037,6 +3041,7 @@ impl Default for SessionMeta { session_id: id.into(), id, forked_from_id: None, + forked_from_ordinal_exclusive: None, parent_thread_id: None, timestamp: String::new(), cwd: PathBuf::new(), @@ -5885,7 +5890,9 @@ mod tests { assert_eq!(session_meta.history_mode, ThreadHistoryMode::Legacy); assert_eq!(session_meta.history_base, None); + assert_eq!(session_meta.forked_from_ordinal_exclusive, None); let serialized = serde_json::to_value(&session_meta)?; + assert!(serialized.get("forked_from_ordinal_exclusive").is_none()); assert_eq!(serialized["history_mode"], json!("legacy")); let mut unknown = serialized; unknown["history_mode"] = json!("future"); diff --git a/codex-rs/rollout/src/compression_tests.rs b/codex-rs/rollout/src/compression_tests.rs index 00cba45b42..48a862ae07 100644 --- a/codex-rs/rollout/src/compression_tests.rs +++ b/codex-rs/rollout/src/compression_tests.rs @@ -697,6 +697,7 @@ fn write_rollout(path: &std::path::Path, thread_id: ThreadId, message: &str) -> session_id: thread_id.into(), id: thread_id, forked_from_id: None, + forked_from_ordinal_exclusive: None, parent_thread_id: None, timestamp: "2025-01-03T12:00:00Z".to_string(), cwd: parent.to_path_buf(), diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index 9f38cd569a..04733d33bf 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -113,6 +113,7 @@ pub use list::rollout_date_parts; pub use maintenance::RolloutMaintenanceGuard; pub use maintenance::try_acquire_rollout_maintenance_lock; pub use metadata::builder_from_items; +pub use metadata::forked_from_ordinal_exclusive; pub use metadata::rollout_id_from_path; pub use model_context::ModelContextScan; pub use model_context::ModelContextScanProgress; diff --git a/codex-rs/rollout/src/metadata.rs b/codex-rs/rollout/src/metadata.rs index 62407790ce..3ac5a744cb 100644 --- a/codex-rs/rollout/src/metadata.rs +++ b/codex-rs/rollout/src/metadata.rs @@ -12,6 +12,7 @@ use chrono::Utc; use codex_protocol::RolloutId; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::SandboxPolicy; +use codex_protocol::protocol::SessionMeta; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::ThreadHistoryMode; @@ -109,6 +110,26 @@ pub fn rollout_id_from_path(rollout_path: &Path) -> Option { Some(RolloutFileName::parse(file_name)?.rollout_id()) } +/// Reads the logical fork cutoff without mistaking a revert's history base for its parent. +/// +/// Older rollouts lack the explicit cutoff. Their history base is safe to use only when it +/// names the logical parent directly or the current file is the thread's original rollout. +/// An ambiguous legacy revert omits the cutoff rather than reporting another thread's boundary. +pub fn forked_from_ordinal_exclusive( + meta: &SessionMeta, + rollout_path: Option<&Path>, +) -> Option { + let parent_id = meta.forked_from_id?; + meta.forked_from_ordinal_exclusive.or_else(|| { + meta.history_base + .filter(|base| { + base.thread_id == parent_id + || rollout_path.and_then(rollout_id_from_path) == Some(meta.id) + }) + .map(|base| base.end_ordinal_exclusive) + }) +} + pub async fn extract_metadata_from_rollout( rollout_path: &Path, default_provider: &str, diff --git a/codex-rs/rollout/src/metadata_tests.rs b/codex-rs/rollout/src/metadata_tests.rs index afea81849f..9ab474b13a 100644 --- a/codex-rs/rollout/src/metadata_tests.rs +++ b/codex-rs/rollout/src/metadata_tests.rs @@ -11,6 +11,7 @@ use chrono::Utc; use codex_protocol::SanitizedGitUrl; use codex_protocol::ThreadId; use codex_protocol::protocol::GitInfo; +use codex_protocol::protocol::HistoryPosition; use codex_protocol::protocol::SessionMeta; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; @@ -26,6 +27,86 @@ use std::path::PathBuf; use tempfile::tempdir; use uuid::Uuid; +#[test] +fn fork_cutoff_distinguishes_logical_parent_from_reverted_rollout() { + let parent_id = ThreadId::new(); + let thread_id = ThreadId::new(); + let physical_id = ThreadId::new(); + let replacement_id = ThreadId::new(); + let original_path = PathBuf::from(format!("rollout-2026-01-27T12-34-56-{thread_id}.jsonl")); + let reverted_path = PathBuf::from(format!( + "rollout-2026-01-27T12-34-56-{thread_id}_{replacement_id}.jsonl" + )); + + for (name, parent, cutoff, base_id, path, expected) in [ + ( + "persisted revert", + Some(parent_id), + Some(20), + thread_id, + Some(reverted_path.as_path()), + Some(20), + ), + ( + "legacy direct fork", + Some(parent_id), + None, + parent_id, + None, + Some(40), + ), + ( + "legacy fork of reverted parent", + Some(parent_id), + None, + physical_id, + Some(original_path.as_path()), + Some(40), + ), + ( + "legacy child revert", + Some(parent_id), + None, + thread_id, + Some(reverted_path.as_path()), + None, + ), + ( + "legacy repeated revert", + Some(parent_id), + None, + physical_id, + Some(reverted_path.as_path()), + None, + ), + ( + "missing parent", + None, + Some(20), + parent_id, + Some(original_path.as_path()), + None, + ), + ] { + let meta = SessionMeta { + id: thread_id, + forked_from_id: parent, + forked_from_ordinal_exclusive: cutoff, + history_base: Some(HistoryPosition { + thread_id: base_id, + end_ordinal_exclusive: 40, + end_byte_offset: 100, + }), + ..SessionMeta::default() + }; + assert_eq!( + forked_from_ordinal_exclusive(&meta, path), + expected, + "{name}" + ); + } +} + #[tokio::test] async fn extract_metadata_from_rollout_uses_session_meta() { let dir = tempdir().expect("tempdir"); @@ -39,6 +120,7 @@ async fn extract_metadata_from_rollout_uses_session_meta() { session_id: id.into(), id, forked_from_id: None, + forked_from_ordinal_exclusive: None, parent_thread_id: None, timestamp: "2026-01-27T12:34:56Z".to_string(), cwd: dir.path().to_path_buf(), @@ -137,6 +219,7 @@ async fn extract_metadata_from_rollout_returns_latest_memory_mode() { session_id: id.into(), id, forked_from_id: None, + forked_from_ordinal_exclusive: None, parent_thread_id: None, timestamp: "2026-01-27T12:34:56Z".to_string(), cwd: dir.path().to_path_buf(), @@ -477,6 +560,7 @@ fn write_rollout_in_sessions_with_cwd( session_id: id.into(), id, forked_from_id: None, + forked_from_ordinal_exclusive: None, parent_thread_id: None, timestamp: event_ts.to_string(), cwd, diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index 394a6d3e9d..1ca019356e 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -103,6 +103,7 @@ pub enum RolloutRecorderParams { /// thread ID stable while creating a new immutable rollout file. rollout_id_override: Option, forked_from_id: Option, + forked_from_ordinal_exclusive: Option, parent_thread_id: Option, source: Box, thread_source: Option, @@ -199,6 +200,7 @@ impl RolloutRecorderParams { conversation_id, rollout_id_override: None, forked_from_id, + forked_from_ordinal_exclusive: None, parent_thread_id, source: Box::new(source), thread_source, @@ -283,6 +285,18 @@ impl RolloutRecorderParams { self } + /// Set the logical fork boundary independently of the physical history base. + pub fn with_forked_from_ordinal_exclusive(mut self, cutoff: Option) -> Self { + if let Self::Create { + forked_from_ordinal_exclusive, + .. + } = &mut self + { + *forked_from_ordinal_exclusive = cutoff; + } + self + } + pub fn with_subagent_history_start_ordinal( mut self, subagent_history_start_ordinal: Option, @@ -830,6 +844,7 @@ impl RolloutRecorder { conversation_id, rollout_id_override, forked_from_id, + forked_from_ordinal_exclusive, parent_thread_id, source, thread_source, @@ -860,6 +875,8 @@ impl RolloutRecorder { session_id, id: conversation_id, forked_from_id, + forked_from_ordinal_exclusive: forked_from_ordinal_exclusive + .filter(|_| forked_from_id.is_some()), parent_thread_id, timestamp, cwd: cwd.clone(), diff --git a/codex-rs/rollout/src/recorder_tests.rs b/codex-rs/rollout/src/recorder_tests.rs index 8ccad419b4..0056d0ef77 100644 --- a/codex-rs/rollout/src/recorder_tests.rs +++ b/codex-rs/rollout/src/recorder_tests.rs @@ -188,6 +188,7 @@ async fn state_db_init_backfills_before_returning() -> anyhow::Result<()> { session_id: thread_id.into(), id: thread_id, forked_from_id: None, + forked_from_ordinal_exclusive: None, parent_thread_id: None, timestamp: "2026-01-27T12:34:56Z".to_string(), cwd: home.path().to_path_buf(), @@ -728,7 +729,7 @@ async fn referenced_paginated_rollout_starts_at_history_cutoff_and_resumes() -> &config, RolloutRecorderParams::new( ThreadId::new(), - /*forked_from_id*/ None, + Some(history_base.thread_id), /*parent_thread_id*/ None, SessionSource::Exec, /*thread_source*/ None, @@ -737,13 +738,20 @@ async fn referenced_paginated_rollout_starts_at_history_cutoff_and_resumes() -> Vec::new(), ) .with_history_mode(ThreadHistoryMode::Paginated) - .with_history_base(Some(history_base)), + .with_history_base(Some(history_base)) + .with_forked_from_ordinal_exclusive(Some(history_base.end_ordinal_exclusive)), ) .await?; let rollout_path = recorder.rollout_path().to_path_buf(); recorder.persist().await?; recorder.shutdown().await?; + let meta = crate::read_session_meta_line(&rollout_path).await?.meta; + assert_eq!( + meta.forked_from_ordinal_exclusive, + Some(history_base.end_ordinal_exclusive) + ); + let resumed = RolloutRecorder::new(&config, RolloutRecorderParams::resume(rollout_path.clone())).await?; resumed diff --git a/codex-rs/rollout/src/session_index_tests.rs b/codex-rs/rollout/src/session_index_tests.rs index 419701e537..47c0eaff37 100644 --- a/codex-rs/rollout/src/session_index_tests.rs +++ b/codex-rs/rollout/src/session_index_tests.rs @@ -38,6 +38,7 @@ fn write_rollout_with_source_and_provider( session_id: thread_id.into(), id: thread_id, forked_from_id: None, + forked_from_ordinal_exclusive: None, parent_thread_id: None, timestamp, cwd: ".".into(), diff --git a/codex-rs/rollout/src/state_db_tests.rs b/codex-rs/rollout/src/state_db_tests.rs index 6fd9427f65..ad38541c29 100644 --- a/codex-rs/rollout/src/state_db_tests.rs +++ b/codex-rs/rollout/src/state_db_tests.rs @@ -328,6 +328,7 @@ fn write_rollout_with_user_message( session_id: thread_id.into(), id: thread_id, forked_from_id: None, + forked_from_ordinal_exclusive: None, parent_thread_id: None, timestamp: "2026-06-01T14:26:25Z".to_string(), cwd: home.to_path_buf(), diff --git a/codex-rs/rollout/src/tests.rs b/codex-rs/rollout/src/tests.rs index 0e6180a41a..30ba0bd8e8 100644 --- a/codex-rs/rollout/src/tests.rs +++ b/codex-rs/rollout/src/tests.rs @@ -1454,6 +1454,7 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> { session_id: conversation_id.into(), id: conversation_id, forked_from_id: None, + forked_from_ordinal_exclusive: None, parent_thread_id: None, timestamp: ts.to_string(), cwd: ".".into(), diff --git a/codex-rs/state/src/extract.rs b/codex-rs/state/src/extract.rs index b5cbd93de6..af4f291577 100644 --- a/codex-rs/state/src/extract.rs +++ b/codex-rs/state/src/extract.rs @@ -394,6 +394,7 @@ mod tests { forked_from_id: Some( ThreadId::from_string(&Uuid::now_v7().to_string()).expect("thread id"), ), + forked_from_ordinal_exclusive: None, parent_thread_id: None, timestamp: "2026-02-26T00:00:00.000Z".to_string(), cwd: PathBuf::from("/child/worktree"), @@ -651,6 +652,7 @@ mod tests { session_id: thread_id.into(), id: thread_id, forked_from_id: None, + forked_from_ordinal_exclusive: None, parent_thread_id: None, timestamp: "2026-02-26T00:00:00.000Z".to_string(), cwd: PathBuf::from("/workspace"), diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index 86200c59ca..3d2e0d923c 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -2568,6 +2568,7 @@ mod tests { session_id: thread_id.into(), id: thread_id, forked_from_id: None, + forked_from_ordinal_exclusive: None, parent_thread_id: None, timestamp: metadata.created_at.to_rfc3339(), cwd: PathBuf::new(), @@ -2638,6 +2639,7 @@ mod tests { session_id: thread_id.into(), id: thread_id, forked_from_id: None, + forked_from_ordinal_exclusive: None, parent_thread_id: None, timestamp: created_at, cwd: PathBuf::new(), diff --git a/codex-rs/thread-store/src/local/create_thread.rs b/codex-rs/thread-store/src/local/create_thread.rs index 3fab5ea761..3be4c808bb 100644 --- a/codex-rs/thread-store/src/local/create_thread.rs +++ b/codex-rs/thread-store/src/local/create_thread.rs @@ -42,6 +42,12 @@ pub(super) async fn create_thread( .with_multi_agent_version(params.multi_agent_version) .with_history_mode(params.history_mode) .with_history_base(params.history_base) + .with_forked_from_ordinal_exclusive( + params + .forked_from_id + .and(params.history_base) + .map(|base| base.end_ordinal_exclusive), + ) .with_subagent_history_start_ordinal(params.subagent_history_start_ordinal) .with_initial_window_id(params.initial_window_id), ) diff --git a/codex-rs/thread-store/src/local/revert_thread.rs b/codex-rs/thread-store/src/local/revert_thread.rs index 1bb40be400..e2418f7845 100644 --- a/codex-rs/thread-store/src/local/revert_thread.rs +++ b/codex-rs/thread-store/src/local/revert_thread.rs @@ -96,9 +96,22 @@ pub(super) async fn revert( ) .await?; + let forked_from_ordinal_exclusive = + codex_rollout::forked_from_ordinal_exclusive(&source_meta, Some(source_path.as_path())) + .map(|cutoff| { + // Reverting into inherited history can shrink, but never grow, the parent prefix. + cutoff.min(history_base.map_or(0, |base| base.end_ordinal_exclusive)) + }); + let rollout_id = ThreadId::new(); - let recorder = - create_replacement_recorder(store, source_meta, rollout_id, history_base).await?; + let recorder = create_replacement_recorder( + store, + source_meta, + rollout_id, + history_base, + forked_from_ordinal_exclusive, + ) + .await?; let replacement_path = recorder.rollout_path().to_path_buf(); recorder.persist().await.map_err(thread_store_io_error)?; recorder.shutdown().await.map_err(thread_store_io_error)?; @@ -127,6 +140,7 @@ async fn create_replacement_recorder( source_meta: codex_rollout::SessionMeta, rollout_id: ThreadId, history_base: Option, + forked_from_ordinal_exclusive: Option, ) -> ThreadStoreResult { let config = RolloutConfig { codex_home: store.config.codex_home.clone(), @@ -154,6 +168,7 @@ async fn create_replacement_recorder( .with_multi_agent_version(source_meta.multi_agent_version) .with_history_mode(ThreadHistoryMode::Paginated) .with_history_base(history_base) + .with_forked_from_ordinal_exclusive(forked_from_ordinal_exclusive) .with_subagent_history_start_ordinal(source_meta.subagent_history_start_ordinal); if let Some(context_window) = source_meta.context_window { params = params.with_initial_window_id(context_window.window_id);