openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-18 17:18:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Track window and fork positions in turn metadata (#40987)

Browse Source 
## What changed

- Add the zero-based `window_number` to Responses turn metadata and include it in prewarm requests.
- Persist `forked_from_ordinal_exclusive` separately from the physical history base so fork lineage remains accurate after reverts and cold resumes.
- Report context-inheriting subagents through `parent_thread_id` without also emitting fork lineage.
- Reserve the new metadata keys while accepting and filtering previously valid configured values so Codex-owned values take precedence.

## Testing

- Cover window numbering across compaction, resume, and fork operations.
- Cover fork cutoff persistence across direct forks, legacy rollouts, reverts into inherited history, and cold resumes.

GitOrigin-RevId: 288f6cd632254814c1f166668b33828082d799a1
This commit is contained in:
Brandon Zhang
2026-08-26 22:37:59 +00:00
committed by copyberry
parent 102ae5e2e6
commit 0d654e6536
33 changed files with 562 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst
Generated
BIN
View File
Binary file not shown.
codex-rs/app-server/tests/common/rollout.rs
+2
View File
@@ -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("/"),
codex-rs/app-server/tests/suite/v2/client_metadata.rs
+2
View File
@@ -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()));
codex-rs/app-server/tests/suite/v2/thread_fork.rs
+31
View File
@@ -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::<Value>()?;
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::<RolloutLine>)
@@ -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::<Value>()?;
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
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+1
View File
@@ -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(),
codex-rs/app-server/tests/suite/v2/thread_revert.rs
+160
View File
@@ -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::<RolloutLine>)
.collect::<Result<Vec<_>, _>>()?
.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::<Value>()?;
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;
codex-rs/core/src/responses_metadata.rs
+21 -1
View File
@@ -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<String>,
pub(crate) routing_hint: Option<HeaderValue>,
pub(crate) window_id: String,
pub(crate) window_number: Option<u64>,
pub(crate) context_window_id: Option<Uuid>,
pub(crate) request_kind: Option<CodexResponsesRequestKind>,
pub(crate) forked_from_thread_id: Option<ThreadId>,
pub(crate) forked_from_ordinal_exclusive: Option<u64>,
pub(crate) parent_thread_id: Option<ThreadId>,
pub(crate) parent_turn_id: Option<String>,
pub(crate) root_turn_id: Option<String>,
@@ -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<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
context_window_id: Option<Uuid>,
#[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<ThreadId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
forked_from_ordinal_exclusive: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
parent_thread_id: Option<ThreadId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
parent_turn_id: Option<&'a str>,
codex-rs/core/src/session/mod.rs
+6 -2
View File
@@ -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) {
codex-rs/core/src/session/session.rs
+38 -6
View File
@@ -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<u64>,
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 {
codex-rs/core/src/session/tests.rs
+49
View File
@@ -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 =
codex-rs/core/src/session_startup_prewarm.rs
+3 -8
View File
@@ -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
codex-rs/core/src/turn_metadata.rs
+5 -1
View File
@@ -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<String> {
@@ -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()
codex-rs/core/src/turn_metadata_tests.rs
+46 -18
View File
@@ -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")
codex-rs/core/tests/suite/agent_websocket.rs
+10
View File
@@ -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(())
codex-rs/core/tests/suite/compact_remote.rs
+7 -2
View File
@@ -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
codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_api_auth_prompt_cache_key_request_diff.snap
+1 -1
View File
@@ -12,7 +12,7 @@ Scenario: After five varied API-key-auth turns, remote manual compaction omits s
- "thread_id": "<UUID>",
- "turn_id": "<UUID>",
- "x-codex-installation-id": "<UUID>",
- "x-codex-turn-metadata": "{\"installation_id\":\"<UUID>\",\"session_id\":\"<UUID>\",\"thread_id\":\"<UUID>\",\"agent_name\":\"/root\",\"turn_id\":\"<UUID>\",\"window_id\":\"<UUID>:0\",\"context_window_id\":\"<UUID>\",\"request_kind\":\"turn\",\"root_turn_id\":\"<UUID>\",\"sandbox\":\"<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\":<UNIX_MS>}",
- "x-codex-turn-metadata": "{\"installation_id\":\"<UUID>\",\"session_id\":\"<UUID>\",\"thread_id\":\"<UUID>\",\"agent_name\":\"/root\",\"turn_id\":\"<UUID>\",\"window_id\":\"<UUID>:0\",\"window_number\":0,\"context_window_id\":\"<UUID>\",\"request_kind\":\"turn\",\"root_turn_id\":\"<UUID>\",\"sandbox\":\"<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\":<UNIX_MS>}",
- "x-codex-window-id": "<UUID>:0"
- },
- "include": [
codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_manual_compact_chatgpt_auth_service_tier_prompt_cache_key_request_diff.snap
+1 -1
View File
@@ -12,7 +12,7 @@ Scenario: After five varied ChatGPT-auth turns, remote manual compaction reuses
- "thread_id": "<UUID>",
- "turn_id": "<UUID>",
- "x-codex-installation-id": "<UUID>",
- "x-codex-turn-metadata": "{\"installation_id\":\"<UUID>\",\"session_id\":\"<UUID>\",\"thread_id\":\"<UUID>\",\"agent_name\":\"/root\",\"turn_id\":\"<UUID>\",\"window_id\":\"<UUID>:0\",\"context_window_id\":\"<UUID>\",\"request_kind\":\"turn\",\"root_turn_id\":\"<UUID>\",\"sandbox\":\"<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\":<UNIX_MS>}",
- "x-codex-turn-metadata": "{\"installation_id\":\"<UUID>\",\"session_id\":\"<UUID>\",\"thread_id\":\"<UUID>\",\"agent_name\":\"/root\",\"turn_id\":\"<UUID>\",\"window_id\":\"<UUID>:0\",\"window_number\":0,\"context_window_id\":\"<UUID>\",\"request_kind\":\"turn\",\"root_turn_id\":\"<UUID>\",\"sandbox\":\"<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\":<UNIX_MS>}",
- "x-codex-window-id": "<UUID>:0"
- },
- "include": [
codex-rs/core/tests/suite/sqlite_state.rs
+1
View File
@@ -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(),
codex-rs/core/tests/suite/window_headers.rs
+7
View File
@@ -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<_>>(),
vec![Some(0), Some(0), Some(1), Some(1), Some(0)]
);
Ok(())
}
codex-rs/protocol/src/protocol.rs
+7
View File
@@ -2976,6 +2976,10 @@ pub struct SessionMeta {
pub id: ThreadId,
#[serde(skip_serializing_if = "Option::is_none")]
pub forked_from_id: Option<ThreadId>,
/// 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<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub parent_thread_id: Option<ThreadId>,
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");
codex-rs/rollout/src/compression_tests.rs
+1
View File
@@ -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(),
codex-rs/rollout/src/lib.rs
+1
View File
@@ -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;
codex-rs/rollout/src/metadata.rs
+21
View File
@@ -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<RolloutId> {
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<u64> {
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,
codex-rs/rollout/src/metadata_tests.rs
+84
View File
@@ -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,
codex-rs/rollout/src/recorder.rs
+17
View File
@@ -103,6 +103,7 @@ pub enum RolloutRecorderParams {
/// thread ID stable while creating a new immutable rollout file.
rollout_id_override: Option<RolloutId>,
forked_from_id: Option<ThreadId>,
forked_from_ordinal_exclusive: Option<u64>,
parent_thread_id: Option<ThreadId>,
source: Box<SessionSource>,
thread_source: Option<ThreadSource>,
@@ -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<u64>) -> 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<u64>,
@@ -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(),
codex-rs/rollout/src/recorder_tests.rs
+10 -2
View File
@@ -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
codex-rs/rollout/src/session_index_tests.rs
+1
View File
@@ -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(),
codex-rs/rollout/src/state_db_tests.rs
+1
View File
@@ -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(),
codex-rs/rollout/src/tests.rs
+1
View File
@@ -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(),
codex-rs/state/src/extract.rs
+2
View File
@@ -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"),
codex-rs/state/src/runtime/threads.rs
+2
View File
@@ -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(),
codex-rs/thread-store/src/local/create_thread.rs
+6
View File
@@ -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),
)
codex-rs/thread-store/src/local/revert_thread.rs
+17 -2
View File
@@ -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<codex_protocol::protocol::HistoryPosition>,
forked_from_ordinal_exclusive: Option<u64>,
) -> ThreadStoreResult<RolloutRecorder> {
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);