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

make fork snapshots deletion safe

Browse Source
This commit is contained in:
anais
2026-06-21 15:01:31 -07:00
parent a95cf32abb
commit 28345fe367
7 changed files with 163 additions and 637 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/request_processors/thread_processor.rs
+96 -14
View File
@@ -3399,14 +3399,30 @@ impl ThreadRequestProcessor {
let Some(rollout_path) = stored_thread.rollout_path.clone() else {
return Ok(None);
};
let Some(compacted_history) = try_read_compacted_resume_rollout_items_from_rollout_path(
let Some(snapshot) = try_create_fork_parent_rollout_snapshot(
self.config.codex_home.as_path(),
rollout_path.as_path(),
stored_thread.thread_id,
)
.await?
.await
else {
return Ok(None);
};
let compacted_history = try_read_compacted_resume_rollout_items_from_rollout_path(
snapshot.absolute_path.as_path(),
stored_thread.thread_id,
)
.await;
let Some(mut compacted_history) = (match compacted_history {
Ok(compacted_history) => compacted_history,
Err(error) => {
remove_fork_parent_rollout_snapshot(snapshot.absolute_path.as_path()).await;
return Err(error);
}
}) else {
remove_fork_parent_rollout_snapshot(snapshot.absolute_path.as_path()).await;
return Ok(None);
};
compacted_history.parent_ref.path = snapshot.reference_path;
stored_thread.history = Some(codex_thread_store::StoredThreadHistory {
thread_id: stored_thread.thread_id,
items: compacted_history.items,
@@ -3728,6 +3744,16 @@ impl ThreadRequestProcessor {
history: history_items,
rollout_path: source_thread.rollout_path.clone(),
});
let initial_rollout_snapshot_path = initial_rollout_copy.as_ref().map(|reference| {
if reference.path.is_absolute() {
reference.path.clone()
} else {
self.config
.codex_home
.join(reference.path.as_path())
.to_path_buf()
}
});
let parent_trace = self.request_trace_context(&request_id).await;
let thread_source = thread_source.map(Into::into);
let fork_result = if let Some(initial_rollout_copy) = initial_rollout_copy {
@@ -3755,18 +3781,25 @@ impl ThreadRequestProcessor {
.await
};
let NewThread {
thread_id,
thread: forked_thread,
session_configured,
..
} = fork_result.map_err(|err| match err {
let fork_result = fork_result.map_err(|err| match err {
CodexErr::Io(_) | CodexErr::Json(_) => {
invalid_request(format!("failed to load thread {source_thread_id}: {err}"))
}
CodexErr::InvalidRequest(message) => invalid_request(message),
err => internal_error(format!("error forking thread: {err}")),
})?;
});
if fork_result.is_err()
&& let Some(snapshot_path) = initial_rollout_snapshot_path.as_deref()
{
remove_fork_parent_rollout_snapshot(snapshot_path).await;
}
let NewThread {
thread_id,
thread: forked_thread,
session_configured,
..
} = fork_result?;
Self::set_app_server_client_info(
forked_thread.as_ref(),
@@ -4136,6 +4169,56 @@ struct CompactedResumeRollout {
parent_ref: codex_rollout::ForkParentRolloutRef,
}
struct ForkParentRolloutSnapshot {
absolute_path: PathBuf,
reference_path: PathBuf,
}
async fn try_create_fork_parent_rollout_snapshot(
codex_home: &Path,
rollout_path: &Path,
) -> Option<ForkParentRolloutSnapshot> {
let existing_path = codex_rollout::existing_rollout_path(rollout_path).await?;
if !is_plain_jsonl_rollout_path(existing_path.as_path()) {
return None;
}
let parent_directory = existing_path.parent()?;
let snapshot_name = format!(
"{}{}.jsonl",
codex_rollout::FORK_PARENT_ROLLOUT_SNAPSHOT_PREFIX,
ThreadId::default()
);
let absolute_path = parent_directory.join(snapshot_name);
if tokio::fs::hard_link(existing_path.as_path(), absolute_path.as_path())
.await
.is_err()
{
return None;
}
let reference_path = match absolute_path.strip_prefix(codex_home) {
Ok(path) => path.to_path_buf(),
Err(_) => {
remove_fork_parent_rollout_snapshot(absolute_path.as_path()).await;
return None;
}
};
Some(ForkParentRolloutSnapshot {
absolute_path,
reference_path,
})
}
async fn remove_fork_parent_rollout_snapshot(path: &Path) {
if let Err(error) = tokio::fs::remove_file(path).await
&& error.kind() != std::io::ErrorKind::NotFound
{
warn!(
"failed to remove fork parent rollout snapshot {}: {error}",
path.display()
);
}
}
fn thread_resume_initial_page_is_recent_not_loaded(
params: &ThreadResumeInitialTurnsPageParams,
) -> bool {
@@ -4382,15 +4465,14 @@ fn read_compacted_resume_suffix_items(
&rollout_line.item,
RolloutItem::EventMsg(EventMsg::UserMessage(_))
);
if is_turn_context {
found_checkpoint_turn_context = true;
}
suffix_items.push(rollout_line.item);
if is_checkpoint {
found_checkpoint = true;
return Ok(true);
}
if found_checkpoint && is_turn_context {
found_checkpoint_turn_context = true;
return Ok(true);
}
if found_checkpoint && is_user_boundary {
found_checkpoint_user_boundary = true;
return Ok(false);
codex-rs/app-server/src/request_processors/thread_processor_tests.rs
-407
View File
@@ -324,413 +324,6 @@ mod thread_processor_behavior_tests {
assert_eq!(turns.last(), Some(&active_turn));
}
fn compacted_resume_test_line(item: RolloutItem) -> codex_protocol::protocol::RolloutLine {
codex_protocol::protocol::RolloutLine {
timestamp: "2025-09-05T16:53:11.850Z".to_string(),
item,
}
}
fn compacted_resume_test_session(thread_id: ThreadId) -> RolloutItem {
RolloutItem::SessionMeta(codex_protocol::protocol::SessionMetaLine {
meta: codex_protocol::protocol::SessionMeta {
id: thread_id,
timestamp: "2025-09-05T16:53:11.850Z".to_string(),
..codex_protocol::protocol::SessionMeta::default()
},
git: None,
})
}
fn compacted_resume_test_user_event(text: &str) -> RolloutItem {
RolloutItem::EventMsg(EventMsg::UserMessage(
codex_protocol::protocol::UserMessageEvent {
message: text.to_string(),
..Default::default()
},
))
}
fn compacted_resume_test_agent_event(text: &str) -> RolloutItem {
RolloutItem::EventMsg(EventMsg::AgentMessage(
codex_protocol::protocol::AgentMessageEvent {
message: text.to_string(),
phase: None,
memory_citation: None,
},
))
}
fn compacted_resume_test_turn_context() -> RolloutItem {
RolloutItem::TurnContext(codex_protocol::protocol::TurnContextItem {
turn_id: Some("turn-1".to_string()),
cwd: test_path_buf("/tmp/codex-compacted-resume").abs(),
workspace_roots: None,
current_date: None,
timezone: None,
approval_policy: AskForApproval::Never,
sandbox_policy: codex_protocol::protocol::SandboxPolicy::new_read_only_policy(),
permission_profile: None,
network: None,
file_system_sandbox_policy: None,
model: "test-model".to_string(),
comp_hash: None,
personality: None,
collaboration_mode: None,
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: None,
effort: None,
summary: codex_protocol::config_types::ReasoningSummary::Auto,
})
}
fn compacted_resume_test_message(
role: &str,
text: &str,
) -> codex_protocol::models::ResponseItem {
let content = if role == "user" {
vec![codex_protocol::models::ContentItem::InputText {
text: text.to_string(),
}]
} else {
vec![codex_protocol::models::ContentItem::OutputText {
text: text.to_string(),
}]
};
codex_protocol::models::ResponseItem::Message {
id: None,
role: role.to_string(),
content,
phase: None,
metadata: None,
}
}
fn compacted_resume_test_compaction(message: &str, window_id: u64) -> RolloutItem {
RolloutItem::Compacted(codex_protocol::protocol::CompactedItem {
message: message.to_string(),
replacement_history: Some(vec![compacted_resume_test_message("assistant", message)]),
window_number: Some(window_id),
window_id: None,
})
}
fn compacted_resume_test_token_count() -> RolloutItem {
RolloutItem::EventMsg(EventMsg::TokenCount(
codex_protocol::protocol::TokenCountEvent {
info: None,
rate_limits: None,
},
))
}
fn compacted_resume_test_goal_update(thread_id: ThreadId) -> RolloutItem {
RolloutItem::EventMsg(EventMsg::ThreadGoalUpdated(
codex_protocol::protocol::ThreadGoalUpdatedEvent {
thread_id,
turn_id: None,
goal: codex_protocol::protocol::ThreadGoal {
thread_id,
objective: "finish the task".to_string(),
status: codex_protocol::protocol::ThreadGoalStatus::Active,
token_budget: None,
tokens_used: 0,
time_used_seconds: 0,
created_at: 1,
updated_at: 1,
},
},
))
}
fn compacted_resume_test_rollout_contents(items: Vec<RolloutItem>) -> Result<String> {
let mut contents = String::new();
for item in items {
contents.push_str(&serde_json::to_string(&compacted_resume_test_line(item))?);
contents.push('\n');
}
Ok(contents)
}
fn compacted_resume_test_replacement_text(item: &RolloutItem) -> Option<&str> {
let RolloutItem::Compacted(compacted) = item else {
return None;
};
let [codex_protocol::models::ResponseItem::Message { content, .. }] =
compacted.replacement_history.as_deref()?
else {
return None;
};
let [codex_protocol::models::ContentItem::OutputText { text }] = content.as_slice() else {
return None;
};
Some(text.as_str())
}
#[test]
fn compacted_resume_rollout_items_reads_latest_checkpoint_suffix_across_chunks() -> Result<()> {
let temp_dir = TempDir::new()?;
let path = temp_dir.path().join("rollout.jsonl");
let thread_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?;
let large_tail_message = "tail".repeat(20_000);
let contents = compacted_resume_test_rollout_contents(vec![
compacted_resume_test_session(thread_id),
compacted_resume_test_user_event("first user event"),
RolloutItem::ResponseItem(compacted_resume_test_message("user", "first user response")),
compacted_resume_test_compaction("older checkpoint", /*window_id*/ 1),
compacted_resume_test_user_event("middle user event"),
compacted_resume_test_turn_context(),
compacted_resume_test_compaction("latest checkpoint", /*window_id*/ 2),
compacted_resume_test_agent_event(&large_tail_message),
])?;
std::fs::write(&path, contents)?;
let byte_len = std::fs::metadata(path.as_path())?.len();
let items = read_compacted_resume_rollout_items(path.as_path(), thread_id, byte_len)?
.expect("compacted resume items");
assert_eq!(items.len(), 7);
assert!(matches!(items[0], RolloutItem::SessionMeta(_)));
assert!(matches!(
items[1],
RolloutItem::EventMsg(EventMsg::UserMessage(_))
));
assert!(matches!(items[2], RolloutItem::ResponseItem(_)));
assert!(matches!(
items[3],
RolloutItem::EventMsg(EventMsg::UserMessage(_))
));
assert!(matches!(items[4], RolloutItem::TurnContext(_)));
assert_eq!(
compacted_resume_test_replacement_text(&items[5]),
Some("latest checkpoint")
);
assert!(matches!(
&items[6],
RolloutItem::EventMsg(EventMsg::AgentMessage(event))
if event.message == large_tail_message
));
assert_eq!(
items
.iter()
.filter(|item| matches!(item, RolloutItem::Compacted(_)))
.count(),
1
);
Ok(())
}
#[test]
fn compacted_resume_rollout_items_carries_metadata_before_checkpoint() -> Result<()> {
let temp_dir = TempDir::new()?;
let path = temp_dir.path().join("rollout.jsonl");
let thread_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?;
let contents = compacted_resume_test_rollout_contents(vec![
compacted_resume_test_session(thread_id),
compacted_resume_test_user_event("first user event"),
RolloutItem::ResponseItem(compacted_resume_test_message("user", "first user response")),
compacted_resume_test_token_count(),
compacted_resume_test_goal_update(thread_id),
compacted_resume_test_user_event("middle user event"),
compacted_resume_test_turn_context(),
compacted_resume_test_compaction("latest checkpoint", /*window_id*/ 2),
compacted_resume_test_agent_event("tail"),
])?;
std::fs::write(&path, contents)?;
let byte_len = std::fs::metadata(path.as_path())?.len();
let items = read_compacted_resume_rollout_items(path.as_path(), thread_id, byte_len)?
.expect("compacted resume items");
assert!(
items
.iter()
.any(|item| matches!(item, RolloutItem::EventMsg(EventMsg::TokenCount(_))))
);
assert!(
items
.iter()
.any(|item| matches!(item, RolloutItem::EventMsg(EventMsg::ThreadGoalUpdated(_))))
);
Ok(())
}
#[test]
fn compacted_resume_rollout_items_falls_back_on_rollback_before_checkpoint() -> Result<()> {
let temp_dir = TempDir::new()?;
let path = temp_dir.path().join("rollout.jsonl");
let thread_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?;
let contents = compacted_resume_test_rollout_contents(vec![
compacted_resume_test_session(thread_id),
compacted_resume_test_user_event("first user event"),
RolloutItem::ResponseItem(compacted_resume_test_message("user", "first user response")),
compacted_resume_test_user_event("middle user event"),
compacted_resume_test_turn_context(),
RolloutItem::EventMsg(EventMsg::ThreadRolledBack(
codex_protocol::protocol::ThreadRolledBackEvent { num_turns: 1 },
)),
compacted_resume_test_compaction("rolled back checkpoint", /*window_id*/ 2),
compacted_resume_test_agent_event("tail"),
])?;
std::fs::write(&path, contents)?;
let byte_len = std::fs::metadata(path.as_path())?.len();
assert!(
read_compacted_resume_rollout_items(path.as_path(), thread_id, byte_len)?.is_none()
);
Ok(())
}
#[test]
fn compacted_resume_rollout_items_falls_back_on_malformed_tail_line() -> Result<()> {
let temp_dir = TempDir::new()?;
let path = temp_dir.path().join("rollout.jsonl");
let thread_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?;
let mut contents = compacted_resume_test_rollout_contents(vec![
compacted_resume_test_session(thread_id),
compacted_resume_test_user_event("first user event"),
RolloutItem::ResponseItem(compacted_resume_test_message("user", "first user response")),
compacted_resume_test_compaction("latest checkpoint", /*window_id*/ 2),
])?;
contents.push_str("{not json}\n");
std::fs::write(&path, contents)?;
let byte_len = std::fs::metadata(path.as_path())?.len();
assert!(
read_compacted_resume_rollout_items(path.as_path(), thread_id, byte_len)?.is_none()
);
Ok(())
}
#[test]
fn recent_not_loaded_rollout_items_keeps_lifecycle_events() -> Result<()> {
let temp_dir = TempDir::new()?;
let path = temp_dir.path().join("rollout.jsonl");
let thread_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?;
let contents = compacted_resume_test_rollout_contents(vec![
compacted_resume_test_session(thread_id),
RolloutItem::EventMsg(EventMsg::TurnStarted(
codex_protocol::protocol::TurnStartedEvent {
turn_id: "turn-a".to_string(),
trace_id: None,
started_at: None,
model_context_window: None,
collaboration_mode_kind: Default::default(),
},
)),
compacted_resume_test_user_event("first user event"),
compacted_resume_test_agent_event("first agent event"),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
turn_id: "turn-a".to_string(),
last_agent_message: None,
completed_at: None,
duration_ms: None,
time_to_first_token_ms: None,
},
)),
RolloutItem::EventMsg(EventMsg::TurnStarted(
codex_protocol::protocol::TurnStartedEvent {
turn_id: "turn-b".to_string(),
trace_id: None,
started_at: None,
model_context_window: None,
collaboration_mode_kind: Default::default(),
},
)),
compacted_resume_test_user_event("second user event"),
compacted_resume_test_agent_event("second agent event"),
RolloutItem::EventMsg(EventMsg::TurnComplete(
codex_protocol::protocol::TurnCompleteEvent {
turn_id: "turn-b".to_string(),
last_agent_message: None,
completed_at: None,
duration_ms: None,
time_to_first_token_ms: None,
},
)),
])?;
std::fs::write(&path, contents)?;
let recent = read_recent_not_loaded_rollout_items(path.as_path(), /*page_size*/ 1)?
.expect("recent notLoaded suffix");
assert!(
recent
.items
.iter()
.any(|item| { matches!(item, RolloutItem::EventMsg(EventMsg::TurnStarted(_))) })
);
assert!(
recent
.items
.iter()
.any(|item| { matches!(item, RolloutItem::EventMsg(EventMsg::TurnComplete(_))) })
);
let page = build_recent_not_loaded_turns_page_response(
recent,
codex_app_server_protocol::ThreadStatus::NotLoaded,
Some(1),
)
.expect("recent notLoaded page should build");
assert_eq!(page.data.len(), 1);
assert_eq!(page.data[0].id, "turn-b");
assert_eq!(
page.data[0].items_view,
codex_app_server_protocol::TurnItemsView::NotLoaded
);
assert_eq!(
page.data[0].status,
codex_app_server_protocol::TurnStatus::Completed
);
Ok(())
}
#[test]
fn recent_not_loaded_rollout_items_falls_back_without_explicit_turn_starts() -> Result<()> {
let temp_dir = TempDir::new()?;
let path = temp_dir.path().join("rollout.jsonl");
let thread_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?;
let contents = compacted_resume_test_rollout_contents(vec![
compacted_resume_test_session(thread_id),
compacted_resume_test_user_event("first user event"),
compacted_resume_test_agent_event("first agent event"),
])?;
std::fs::write(&path, contents)?;
assert!(read_recent_not_loaded_rollout_items(path.as_path(), /*page_size*/ 1)?.is_none());
Ok(())
}
#[test]
fn recent_not_loaded_rollout_items_falls_back_on_non_persisted_suffix_item() -> Result<()> {
let temp_dir = TempDir::new()?;
let path = temp_dir.path().join("rollout.jsonl");
let thread_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?;
let contents = compacted_resume_test_rollout_contents(vec![
compacted_resume_test_session(thread_id),
RolloutItem::EventMsg(EventMsg::TurnStarted(
codex_protocol::protocol::TurnStartedEvent {
turn_id: "turn-a".to_string(),
trace_id: None,
started_at: None,
model_context_window: None,
collaboration_mode_kind: Default::default(),
},
)),
compacted_resume_test_user_event("first user event"),
RolloutItem::ResponseItem(codex_protocol::models::ResponseItem::CompactionTrigger {
metadata: None,
}),
compacted_resume_test_agent_event("first agent event"),
])?;
std::fs::write(&path, contents)?;
assert!(read_recent_not_loaded_rollout_items(path.as_path(), /*page_size*/ 1)?.is_none());
Ok(())
}
#[test]
fn validate_dynamic_tools_rejects_empty_namespace() {
let tools = vec![dynamic_tool(
codex-rs/core/tests/suite/compact_resume_fork.rs
-153
View File
@@ -14,16 +14,11 @@ use codex_core::ThreadManager;
use codex_core::compact::SUMMARIZATION_PROMPT;
use codex_core::config::Config;
use codex_core::spawn::CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR;
use codex_login::CodexAuth;
use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::Settings;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::InitialHistory;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::ResumedHistory;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::WarningEvent;
use codex_protocol::user_input::UserInput;
use core_test_support::context_snapshot;
@@ -284,73 +279,6 @@ async fn compact_resume_and_fork_preserve_model_history_view() {
assert_eq!(requests.len(), 5);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
/// Scenario: a compacted resume history slice should reconstruct the same
/// model-visible context when the next user turn runs.
async fn compacted_resume_history_slice_preserves_next_turn_context() -> Result<()> {
if network_disabled() {
println!("Skipping test because network is disabled in this sandbox");
return Ok(());
}
const AFTER_COMPACTED_SLICE_RESUME: &str = "AFTER_COMPACTED_SLICE_RESUME";
let server = MockServer::start().await;
let request_log = mount_sse_sequence(
&server,
vec![
sse(vec![
ev_assistant_message("m1", FIRST_REPLY),
ev_completed("r1"),
]),
sse(vec![
ev_assistant_message("m2", SUMMARY_TEXT),
ev_completed("r2"),
]),
sse(vec![ev_completed("r3")]),
sse(vec![ev_completed("r4")]),
],
)
.await;
let (_home, config, manager, base) = start_test_conversation(&server, /*model*/ None).await;
user_turn(&base, "hello world").await;
compact_conversation(&base).await;
user_turn(&base, "AFTER_COMPACT").await;
let rollout_path = fetch_conversation_path(&base);
let compacted_history = compacted_resume_history_slice_from_rollout(&rollout_path).await?;
shutdown_conversation(&base).await;
let resumed = resume_conversation_with_history(&manager, &config, compacted_history).await;
user_turn(&resumed, AFTER_COMPACTED_SLICE_RESUME).await;
let mut requests = request_log
.requests()
.into_iter()
.map(|request| request.body_json())
.collect::<Vec<_>>();
requests.iter_mut().for_each(normalize_line_endings);
normalize_compact_prompts(&mut requests);
assert_eq!(requests.len(), 4);
let summary_after_compact = extract_summary_user_text(&requests[2], SUMMARY_TEXT);
let summary_after_resume = extract_summary_user_text(&requests[3], SUMMARY_TEXT);
assert_eq!(summary_after_resume, summary_after_compact);
let after_compact_user_texts = json_message_input_texts(&requests[2], "user");
let resumed_user_texts = json_message_input_texts(&requests[3], "user");
let (resumed_last, resumed_prefix) = resumed_user_texts
.split_last()
.expect("compacted slice resume request missing user messages");
assert_eq!(resumed_last, AFTER_COMPACTED_SLICE_RESUME);
assert!(
resumed_prefix.starts_with(&after_compact_user_texts),
"compacted resume history should preserve the model-visible user history prefix",
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
/// Scenario: after the forked branch is compacted, resuming again should reuse
/// the compacted history and only append the new user message.
@@ -909,87 +837,6 @@ async fn resume_conversation(
.thread
}
async fn resume_conversation_with_history(
manager: &ThreadManager,
config: &Config,
history: InitialHistory,
) -> Arc<CodexThread> {
let auth_manager =
codex_core::test_support::auth_manager_from_auth(CodexAuth::from_api_key("dummy"));
Box::pin(manager.resume_thread_with_history(
config.clone(),
history,
auth_manager,
/*parent_trace*/ None,
/*supports_openai_form_elicitation*/ false,
))
.await
.expect("resume conversation from compacted history")
.thread
}
async fn compacted_resume_history_slice_from_rollout(
path: &std::path::Path,
) -> Result<InitialHistory> {
let (items, thread_id, _parse_errors) =
codex_rollout::RolloutRecorder::load_rollout_items(path).await?;
let conversation_id = thread_id.expect("rollout thread id");
let mut history = compacted_resume_prefix_items(&items);
let suffix_start_index = compacted_resume_suffix_start_index(&items);
history.extend(items[suffix_start_index..].iter().cloned());
Ok(InitialHistory::Resumed(ResumedHistory {
conversation_id,
history,
rollout_path: Some(path.to_path_buf()),
}))
}
fn compacted_resume_prefix_items(items: &[RolloutItem]) -> Vec<RolloutItem> {
let mut history = Vec::new();
let mut saw_session_meta = false;
let mut saw_user_event = false;
let mut saw_user_response_item = false;
for item in items {
match item {
RolloutItem::SessionMeta(_) if !saw_session_meta => {
saw_session_meta = true;
history.push(item.clone());
}
RolloutItem::EventMsg(EventMsg::UserMessage(_)) if !saw_user_event => {
saw_user_event = true;
history.push(item.clone());
}
RolloutItem::ResponseItem(ResponseItem::Message { role, .. })
if role == "user" && !saw_user_response_item =>
{
saw_user_response_item = true;
history.push(item.clone());
}
_ => {}
}
if saw_session_meta && saw_user_event && saw_user_response_item {
break;
}
}
history
}
fn compacted_resume_suffix_start_index(items: &[RolloutItem]) -> usize {
let compacted_index = items
.iter()
.rposition(|item| {
matches!(item, RolloutItem::Compacted(compacted) if compacted.replacement_history.is_some())
})
.expect("replacement-history compaction item");
items[..compacted_index]
.iter()
.rposition(|item| matches!(item, RolloutItem::TurnContext(_)))
.unwrap_or(compacted_index)
}
#[cfg(test)]
async fn fork_thread(
manager: &ThreadManager,
codex-rs/rollout/src/lib.rs
+2
View File
@@ -23,6 +23,7 @@ pub(crate) use codex_protocol::protocol;
pub const SESSIONS_SUBDIR: &str = "sessions";
pub const ARCHIVED_SESSIONS_SUBDIR: &str = "archived_sessions";
pub const FORK_PARENT_ROLLOUT_SNAPSHOT_PREFIX: &str = ".fork-parent-snapshot-";
pub static INTERACTIVE_SESSION_SOURCES: LazyLock<Vec<SessionSource>> = LazyLock::new(|| {
vec![
SessionSource::Cli,
@@ -67,6 +68,7 @@ pub use recorder::ForkParentRolloutRef;
pub use recorder::RolloutRecorder;
pub use recorder::RolloutRecorderParams;
pub use recorder::append_rollout_item_to_path;
pub use recorder::read_fork_parent_rollout_ref;
pub use search::first_rollout_content_match_snippet;
pub use search::search_rollout_matches;
pub use search::search_rollout_paths;
codex-rs/rollout/src/recorder.rs
+22 -4
View File
@@ -1105,10 +1105,6 @@ async fn resolve_fork_parent_rollout_path(
parent_ref: &ForkParentRolloutRef,
parent_thread_id: ThreadId,
) -> std::io::Result<PathBuf> {
if let Some(path) = compression::existing_rollout_path(parent_ref.path.as_path()).await {
return Ok(path);
}
let codex_home = child_path
.ancestors()
.find(|ancestor| {
@@ -1123,6 +1119,15 @@ async fn resolve_fork_parent_rollout_path(
parent_ref.path.display()
))
})?;
let referenced_path = if parent_ref.path.is_absolute() {
parent_ref.path.clone()
} else {
codex_home.join(parent_ref.path.as_path())
};
if let Some(path) = compression::existing_rollout_path(referenced_path.as_path()).await {
return Ok(path);
}
let parent_thread_id = parent_thread_id.to_string();
if let Some(path) = find_thread_path_by_id_str_in_subdir_from_filesystem(
codex_home,
@@ -1183,6 +1188,19 @@ pub fn parse_fork_parent_rollout_ref_from_line(line: &str) -> Option<ForkParentR
})
}
pub async fn read_fork_parent_rollout_ref(
path: &Path,
) -> std::io::Result<Option<ForkParentRolloutRef>> {
let mut reader = compression::open_rollout_line_reader(path).await?;
while let Some(line) = reader.next_line().await? {
let trimmed = line.trim();
if !trimmed.is_empty() {
return Ok(parse_fork_parent_rollout_ref_from_line(trimmed));
}
}
Ok(None)
}
fn push_loaded_rollout_item(
items: &mut Vec<RolloutItem>,
thread_id: &mut Option<ThreadId>,
codex-rs/rollout/src/tests.rs
-59
View File
@@ -26,7 +26,6 @@ use crate::list::ThreadSortKey;
use crate::list::ThreadsPage;
use crate::list::get_threads;
use crate::list::read_head_for_summary;
use crate::list::read_thread_item_from_rollout;
use crate::rollout_date_parts;
use anyhow::Result;
use codex_protocol::ThreadId;
@@ -1349,64 +1348,6 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn read_thread_item_from_cow_child_uses_parent_preview() -> Result<()> {
let temp = TempDir::new().unwrap();
let home = temp.path();
let parent_ts = "2025-06-01T08-00-00";
let parent_uuid = Uuid::from_u128(42);
write_session_file(
home,
parent_ts,
parent_uuid,
/*num_records*/ 0,
Some(SessionSource::VSCode),
)?;
let parent_path = home
.join("sessions")
.join("2025")
.join("06")
.join("01")
.join(format!("rollout-{parent_ts}-{parent_uuid}.jsonl"));
let parent_byte_len = fs::metadata(&parent_path)?.len();
let child_ts = "2025-06-01T09-00-00";
let child_uuid = Uuid::from_u128(43);
let child_dir = home.join("sessions").join("2025").join("06").join("01");
fs::create_dir_all(&child_dir)?;
let child_path = child_dir.join(format!("rollout-{child_ts}-{child_uuid}.jsonl"));
let child_meta = serde_json::json!({
"timestamp": child_ts,
"type": "session_meta",
"payload": {
"id": child_uuid,
"forked_from_id": parent_uuid,
"timestamp": child_ts,
"cwd": ".",
"originator": "test_originator",
"cli_version": "test_version",
"base_instructions": null,
"source": "vscode",
"model_provider": TEST_PROVIDER,
"fork_parent_rollout_path": parent_path.to_string_lossy(),
"fork_parent_rollout_byte_len": parent_byte_len,
}
});
fs::write(&child_path, format!("{child_meta}\n"))?;
let item = read_thread_item_from_rollout(child_path.clone())
.await
.expect("child summary should expand parent preview");
assert_eq!(item.path, child_path);
assert_eq!(item.thread_id, Some(thread_id_from_uuid(child_uuid)));
assert_eq!(item.preview.as_deref(), Some("Hello from user"));
assert_eq!(item.first_user_message.as_deref(), Some("Hello from user"));
assert_eq!(item.model_provider.as_deref(), Some(TEST_PROVIDER));
Ok(())
}
#[tokio::test]
async fn test_timestamp_only_cursor_skips_same_second_filesystem_ties() {
let temp = TempDir::new().unwrap();
codex-rs/thread-store/src/local/delete_thread.rs
+43
View File
@@ -11,7 +11,9 @@ use codex_rollout::ARCHIVED_SESSIONS_SUBDIR;
use codex_rollout::SESSIONS_SUBDIR;
use codex_rollout::find_archived_thread_path_by_id_str;
use codex_rollout::find_thread_path_by_id_str;
use codex_rollout::read_fork_parent_rollout_ref;
use codex_rollout::remove_thread_name_entries;
use tracing::warn;
use super::LocalThreadStore;
use super::helpers::matching_rollout_file_name;
@@ -66,9 +68,27 @@ pub(super) async fn delete_thread(
}
let found_rollout_path = !rollout_paths.is_empty();
let mut fork_parent_snapshots = Vec::new();
for rollout_path in rollout_paths {
if let Ok(Some(reference)) = read_fork_parent_rollout_ref(rollout_path.as_path()).await
&& let Some(snapshot_path) =
owned_fork_parent_snapshot_path(store, reference.path.as_path())
&& !fork_parent_snapshots.contains(&snapshot_path)
{
fork_parent_snapshots.push(snapshot_path);
}
delete_rollout_file(store, rollout_path.as_path(), thread_id)?;
}
for snapshot_path in fork_parent_snapshots {
if let Err(err) = std::fs::remove_file(snapshot_path.as_path())
&& err.kind() != ErrorKind::NotFound
{
warn!(
"failed to remove fork parent rollout snapshot `{}`: {err}",
snapshot_path.display()
);
}
}
remove_thread_name_entries(store.config.codex_home.as_path(), thread_id)
.await
.map_err(|err| ThreadStoreError::Internal {
@@ -84,6 +104,29 @@ pub(super) async fn delete_thread(
Ok(())
}
fn owned_fork_parent_snapshot_path(
store: &LocalThreadStore,
reference_path: &Path,
) -> Option<std::path::PathBuf> {
let path = if reference_path.is_absolute() {
reference_path.to_path_buf()
} else {
store.config.codex_home.join(reference_path)
};
let file_name = path.file_name()?.to_str()?;
if !file_name.starts_with(codex_rollout::FORK_PARENT_ROLLOUT_SNAPSHOT_PREFIX) {
return None;
}
let canonical_path = std::fs::canonicalize(path).ok()?;
[SESSIONS_SUBDIR, ARCHIVED_SESSIONS_SUBDIR]
.into_iter()
.filter_map(|subdirectory| {
std::fs::canonicalize(store.config.codex_home.join(subdirectory)).ok()
})
.any(|root| canonical_path.starts_with(root))
.then_some(canonical_path)
}
fn delete_rollout_file(
store: &LocalThreadStore,
rollout_path: &Path,