openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-29 19:08:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Introduce rollout source groundwork

Browse Source 
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Charles Cunningham
2026-03-10 15:37:54 -07:00
co-authored by Codex
parent c7e28cffab
commit 61a0323a32
29 changed files with 935 additions and 661 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/bespoke_event_handling.rs
+1 -1
View File
@@ -1608,7 +1608,7 @@ pub(crate) async fn apply_bespoke_event_handling(
let Some(rollout_path) = conversation.rollout_path() else {
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "thread has no persisted rollout".to_string(),
message: "thread has no rollout".to_string(),
data: None,
};
outgoing.send_error(request_id, error).await;
codex-rs/app-server/src/codex_message_processor.rs
+25 -54
View File
@@ -182,7 +182,7 @@ use codex_core::CodexAuth;
use codex_core::CodexThread;
use codex_core::Cursor as RolloutCursor;
use codex_core::NewThread;
use codex_core::RolloutRecorder;
use codex_core::RolloutStore;
use codex_core::SessionMeta;
use codex_core::SteerInputError;
use codex_core::ThreadConfigSnapshot;
@@ -2456,17 +2456,18 @@ impl CodexMessageProcessor {
"ephemeral thread does not support metadata updates: {thread_uuid}"
)));
};
reconcile_rollout(
Some(state_db_ctx),
rollout_path.as_path(),
self.config.model_provider_id.as_str(),
None,
&[],
None,
None,
)
.await;
if thread.rollout_path().is_some() {
reconcile_rollout(
Some(state_db_ctx),
rollout_path.as_path(),
self.config.model_provider_id.as_str(),
None,
&[],
None,
None,
)
.await;
}
match state_db_ctx.get_thread(thread_uuid).await {
Ok(Some(_)) => return Ok(()),
@@ -3101,16 +3102,6 @@ impl CodexMessageProcessor {
Ok(items) => {
thread.turns = build_turns_from_rollout_items(&items);
}
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
self.send_invalid_request_error(
request_id,
format!(
"thread {thread_uuid} is not materialized yet; includeTurns is unavailable before first user message"
),
)
.await;
return;
}
Err(err) => {
self.send_internal_error(
request_id,
@@ -3398,34 +3389,7 @@ impl CodexMessageProcessor {
}
let rollout_path = if let Some(path) = existing_thread.rollout_path() {
if path.exists() {
path
} else {
match find_thread_path_by_id_str(
&self.config.codex_home,
&existing_thread_id.to_string(),
)
.await
{
Ok(Some(path)) => path,
Ok(None) => {
self.send_invalid_request_error(
request_id,
format!("no rollout found for thread id {existing_thread_id}"),
)
.await;
return true;
}
Err(err) => {
self.send_invalid_request_error(
request_id,
format!("failed to locate thread id {existing_thread_id}: {err}"),
)
.await;
return true;
}
}
}
path
} else {
match find_thread_path_by_id_str(
&self.config.codex_home,
@@ -3610,7 +3574,11 @@ impl CodexMessageProcessor {
}
};
match RolloutRecorder::get_rollout_history(&rollout_path).await {
// App-server should not need direct rollout-file access.
// TODO(ccunningham): make resumed startup stop requiring eager
// `InitialHistory::Resumed(Vec<RolloutItem>)` materialization so core can own rollout
// loading all the way through thread startup.
match RolloutStore::get_rollout_history(&rollout_path).await {
Ok(initial_history) => Some(initial_history),
Err(err) => {
self.send_invalid_request_error(
@@ -4048,7 +4016,7 @@ impl CodexMessageProcessor {
while remaining > 0 {
let page_size = remaining.min(THREAD_LIST_MAX_LIMIT);
let page = if archived {
RolloutRecorder::list_archived_threads(
RolloutStore::list_archived_threads(
&self.config,
page_size,
cursor_obj.as_ref(),
@@ -4065,7 +4033,7 @@ impl CodexMessageProcessor {
data: None,
})?
} else {
RolloutRecorder::list_threads(
RolloutStore::list_threads(
&self.config,
page_size,
cursor_obj.as_ref(),
@@ -7667,7 +7635,10 @@ pub(crate) async fn read_summary_from_rollout(
pub(crate) async fn read_rollout_items_from_rollout(
path: &Path,
) -> std::io::Result<Vec<RolloutItem>> {
let items = match RolloutRecorder::get_rollout_history(path).await? {
// This helper exists specifically for callers that still need owned rollout items after
// loading by path. When startup and downstream consumers become `RolloutSource`-backed, they
// should bypass this eager conversion.
let items = match RolloutStore::get_rollout_history(path).await? {
InitialHistory::New => Vec::new(),
InitialHistory::Forked(items) => items,
InitialHistory::Resumed(resumed) => resumed.history,
codex-rs/app-server/tests/suite/v2/thread_archive.rs
+5 -70
View File
@@ -2,7 +2,6 @@ use anyhow::Result;
use app_test_support::McpProcess;
use app_test_support::create_mock_responses_server_repeating_assistant;
use app_test_support::to_response;
use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ThreadArchiveParams;
@@ -28,7 +27,7 @@ use tokio::time::timeout;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
#[tokio::test]
async fn thread_archive_requires_materialized_rollout() -> Result<()> {
async fn thread_archive_works_before_first_user_message() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
@@ -52,72 +51,14 @@ async fn thread_archive_requires_materialized_rollout() -> Result<()> {
assert!(!thread.id.is_empty());
let rollout_path = thread.path.clone().expect("thread path");
assert!(
!rollout_path.exists(),
"fresh thread rollout should not exist yet at {}",
rollout_path.display()
);
assert!(rollout_path.exists(), "fresh thread rollout should exist");
assert!(
find_thread_path_by_id_str(codex_home.path(), &thread.id)
.await?
.is_none(),
"thread id should not be discoverable before rollout materialization"
.is_some(),
"thread id should be discoverable immediately after startup"
);
// Archive should fail before the rollout is materialized.
let archive_id = mcp
.send_thread_archive_request(ThreadArchiveParams {
thread_id: thread.id.clone(),
})
.await?;
let archive_err: JSONRPCError = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(archive_id)),
)
.await??;
assert!(
archive_err
.error
.message
.contains("no rollout found for thread id"),
"unexpected archive error: {}",
archive_err.error.message
);
// Materialize rollout via a real user turn and confirm archive succeeds.
let turn_start_id = mcp
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
input: vec![UserInput::Text {
text: "materialize".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let turn_start_response: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(turn_start_id)),
)
.await??;
let _: TurnStartResponse = to_response::<TurnStartResponse>(turn_start_response)?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
assert!(
rollout_path.exists(),
"expected rollout path {} to exist after first user message",
rollout_path.display()
);
let discovered_path = find_thread_path_by_id_str(codex_home.path(), &thread.id)
.await?
.expect("expected rollout path for thread id to exist after materialization");
assert_paths_match_on_disk(&discovered_path, &rollout_path)?;
let archive_id = mcp
.send_thread_archive_request(ThreadArchiveParams {
thread_id: thread.id.clone(),
@@ -129,6 +70,7 @@ async fn thread_archive_requires_materialized_rollout() -> Result<()> {
)
.await??;
let _: ThreadArchiveResponse = to_response::<ThreadArchiveResponse>(archive_resp)?;
let archive_notification = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("thread/archived"),
@@ -316,10 +258,3 @@ stream_max_retries = 0
"#
)
}
fn assert_paths_match_on_disk(actual: &Path, expected: &Path) -> std::io::Result<()> {
let actual = actual.canonicalize()?;
let expected = expected.canonicalize()?;
assert_eq!(actual, expected);
Ok(())
}
codex-rs/app-server/tests/suite/v2/thread_fork.rs
+11 -13
View File
@@ -3,7 +3,6 @@ use app_test_support::McpProcess;
use app_test_support::create_fake_rollout;
use app_test_support::create_mock_responses_server_repeating_assistant;
use app_test_support::to_response;
use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
@@ -164,7 +163,7 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
}
#[tokio::test]
async fn thread_fork_rejects_unmaterialized_thread() -> Result<()> {
async fn thread_fork_creates_empty_loaded_thread() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
@@ -185,25 +184,24 @@ async fn thread_fork_rejects_unmaterialized_thread() -> Result<()> {
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
let thread_id = thread.id.clone();
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: thread.id,
thread_id,
..Default::default()
})
.await?;
let fork_err: JSONRPCError = timeout(
let fork_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(fork_id)),
mcp.read_stream_until_response_message(RequestId::Integer(fork_id)),
)
.await??;
assert!(
fork_err
.error
.message
.contains("no rollout found for thread id"),
"unexpected fork error: {}",
fork_err.error.message
);
let ThreadForkResponse { thread: forked, .. } = to_response::<ThreadForkResponse>(fork_resp)?;
assert_ne!(forked.id, thread.id);
assert!(forked.preview.is_empty());
assert_eq!(forked.turns.len(), 0);
assert_eq!(forked.status, ThreadStatus::Idle);
Ok(())
}
codex-rs/app-server/tests/suite/v2/thread_read.rs
+13 -19
View File
@@ -3,7 +3,6 @@ use app_test_support::McpProcess;
use app_test_support::create_fake_rollout_with_text_elements;
use app_test_support::create_mock_responses_server_repeating_assistant;
use app_test_support::to_response;
use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::SessionSource;
@@ -153,7 +152,7 @@ async fn thread_read_can_include_turns() -> Result<()> {
}
#[tokio::test]
async fn thread_read_loaded_thread_returns_precomputed_path_before_materialization() -> Result<()> {
async fn thread_read_loaded_thread_returns_startup_rollout_path() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
@@ -175,8 +174,8 @@ async fn thread_read_loaded_thread_returns_precomputed_path_before_materializati
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
let thread_path = thread.path.clone().expect("thread path");
assert!(
!thread_path.exists(),
"fresh thread rollout should not be materialized yet"
thread_path.exists(),
"fresh thread rollout should be readable at startup"
);
let read_id = mcp
@@ -357,7 +356,7 @@ async fn thread_name_set_is_reflected_in_read_list_and_resume() -> Result<()> {
}
#[tokio::test]
async fn thread_read_include_turns_rejects_unmaterialized_loaded_thread() -> Result<()> {
async fn thread_read_include_turns_returns_empty_loaded_thread() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
@@ -378,10 +377,7 @@ async fn thread_read_include_turns_rejects_unmaterialized_loaded_thread() -> Res
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
let thread_path = thread.path.clone().expect("thread path");
assert!(
!thread_path.exists(),
"fresh thread rollout should not be materialized yet"
);
assert!(thread_path.exists(), "fresh thread rollout should exist");
let read_id = mcp
.send_thread_read_request(ThreadReadParams {
@@ -389,20 +385,18 @@ async fn thread_read_include_turns_rejects_unmaterialized_loaded_thread() -> Res
include_turns: true,
})
.await?;
let read_err: JSONRPCError = timeout(
let read_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(read_id)),
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadReadResponse { thread: read } = to_response::<ThreadReadResponse>(read_resp)?;
assert!(
read_err
.error
.message
.contains("includeTurns is unavailable before first user message"),
"unexpected error: {}",
read_err.error.message
);
assert_eq!(read.id, thread.id);
assert_eq!(read.path, Some(thread_path));
assert!(read.preview.is_empty());
assert_eq!(read.turns.len(), 0);
assert_eq!(read.status, ThreadStatus::Idle);
Ok(())
}
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+11 -12
View File
@@ -95,7 +95,7 @@ async fn wait_for_responses_request_count(
}
#[tokio::test]
async fn thread_resume_rejects_unmaterialized_thread() -> Result<()> {
async fn thread_resume_returns_empty_loaded_thread() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
@@ -117,26 +117,25 @@ async fn thread_resume_rejects_unmaterialized_thread() -> Result<()> {
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
// Resume should fail before the first user message materializes rollout storage.
let resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: thread.id.clone(),
..Default::default()
})
.await?;
let resume_err: JSONRPCError = timeout(
let resume_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(resume_id)),
mcp.read_stream_until_response_message(RequestId::Integer(resume_id)),
)
.await??;
assert!(
resume_err
.error
.message
.contains("no rollout found for thread id"),
"unexpected resume error: {}",
resume_err.error.message
);
let ThreadResumeResponse {
thread: resumed, ..
} = to_response::<ThreadResumeResponse>(resume_resp)?;
assert_eq!(resumed.id, thread.id);
assert!(resumed.preview.is_empty());
assert_eq!(resumed.turns.len(), 0);
assert_eq!(resumed.status, ThreadStatus::Idle);
Ok(())
}
codex-rs/app-server/tests/suite/v2/thread_start.rs
+2 -2
View File
@@ -73,8 +73,8 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> {
let thread_path = thread.path.clone().expect("thread path should be present");
assert!(thread_path.is_absolute(), "thread path should be absolute");
assert!(
!thread_path.exists(),
"fresh thread rollout should not be materialized until first user message"
thread_path.exists(),
"fresh thread rollout should be readable at startup"
);
// Wire contract: thread title field is `name`, serialized as null when unset.
codex-rs/core/src/agent/control.rs
+8 -21
View File
@@ -6,7 +6,7 @@ use crate::agent::status::is_final;
use crate::error::CodexErr;
use crate::error::Result as CodexResult;
use crate::find_thread_path_by_id_str;
use crate::rollout::RolloutRecorder;
use crate::rollout::RolloutStore;
use crate::session_prefix::format_subagent_context_line;
use crate::session_prefix::format_subagent_notification_message;
use crate::shell_snapshot::ShellSnapshot;
@@ -144,12 +144,7 @@ impl AgentControl {
let parent_thread = state.get_thread(parent_thread_id).await.ok();
if let Some(parent_thread) = parent_thread.as_ref() {
// `record_conversation_items` only queues rollout writes asynchronously.
// Flush/materialize the live parent before snapshotting JSONL for a fork.
parent_thread
.codex
.session
.ensure_rollout_materialized()
.await;
// Flush the live parent before snapshotting JSONL for a fork.
parent_thread.codex.session.flush_rollout().await;
}
let rollout_path = parent_thread
@@ -165,10 +160,12 @@ impl AgentControl {
"parent thread rollout unavailable for fork: {parent_thread_id}"
))
})?;
let mut forked_rollout_items =
RolloutRecorder::get_rollout_history(&rollout_path)
.await?
.get_rollout_items();
// TODO(ccunningham): delete this eager owned-history path once fork startup
// becomes `RolloutSource`-backed end-to-end; today agent forking still
// appends a fresh tool output onto the parent rollout items before startup.
let mut forked_rollout_items = RolloutStore::get_rollout_history(&rollout_path)
.await?
.get_rollout_items();
let mut output = FunctionCallOutputPayload::from_text(
FORKED_SPAWN_AGENT_OUTPUT_MESSAGE.to_string(),
);
@@ -886,11 +883,6 @@ mod tests {
.session
.record_conversation_items(turn_context.as_ref(), &[parent_spawn_call])
.await;
parent_thread
.codex
.session
.ensure_rollout_materialized()
.await;
parent_thread.codex.session.flush_rollout().await;
let child_thread_id = harness
@@ -968,11 +960,6 @@ mod tests {
.session
.record_conversation_items(turn_context.as_ref(), &[parent_spawn_call])
.await;
parent_thread
.codex
.session
.ensure_rollout_materialized()
.await;
parent_thread.codex.session.flush_rollout().await;
let child_thread_id = harness
codex-rs/core/src/codex.rs
+108 -103
View File
@@ -256,11 +256,12 @@ use crate::protocol::TokenUsage;
use crate::protocol::TokenUsageInfo;
use crate::protocol::TurnDiffEvent;
use crate::protocol::WarningEvent;
use crate::rollout::RolloutRecorder;
use crate::rollout::RolloutRecorderParams;
use crate::rollout::RolloutStore;
use crate::rollout::RolloutStoreParams;
use crate::rollout::map_session_init_error;
use crate::rollout::metadata;
use crate::rollout::policy::EventPersistenceMode;
use crate::rollout::recorder::InMemoryRolloutSource;
use crate::shell;
use crate::shell_snapshot::ShellSnapshot;
use crate::skills::SkillError;
@@ -1254,7 +1255,7 @@ impl Session {
exec_policy: ExecPolicyManager,
tx_event: Sender<Event>,
agent_status: watch::Sender<AgentStatus>,
initial_history: InitialHistory,
mut initial_history: InitialHistory,
session_source: SessionSource,
skills_manager: Arc<SkillsManager>,
plugins_manager: Arc<PluginsManager>,
@@ -1275,40 +1276,12 @@ impl Session {
}
let forked_from_id = initial_history.forked_from_id();
let (conversation_id, rollout_params) = match &initial_history {
InitialHistory::New | InitialHistory::Forked(_) => {
let conversation_id = ThreadId::default();
(
conversation_id,
RolloutRecorderParams::new(
conversation_id,
forked_from_id,
session_source,
BaseInstructions {
text: session_configuration.base_instructions.clone(),
},
session_configuration.dynamic_tools.clone(),
if session_configuration.persist_extended_history {
EventPersistenceMode::Extended
} else {
EventPersistenceMode::Limited
},
),
)
}
InitialHistory::Resumed(resumed_history) => (
resumed_history.conversation_id,
RolloutRecorderParams::resume(
resumed_history.rollout_path.clone(),
if session_configuration.persist_extended_history {
EventPersistenceMode::Extended
} else {
EventPersistenceMode::Limited
},
),
),
let event_persistence_mode = if session_configuration.persist_extended_history {
EventPersistenceMode::Extended
} else {
EventPersistenceMode::Limited
};
let initial_messages = initial_history.get_event_msgs();
let state_builder = match &initial_history {
InitialHistory::Resumed(resumed) => metadata::builder_from_items(
resumed.history.as_slice(),
@@ -1317,24 +1290,62 @@ impl Session {
InitialHistory::New | InitialHistory::Forked(_) => None,
};
let (conversation_id, rollout_params) = match &mut initial_history {
InitialHistory::New | InitialHistory::Forked(_) => {
let conversation_id = ThreadId::default();
(
conversation_id,
RolloutStoreParams::new(
conversation_id,
forked_from_id,
session_source,
BaseInstructions {
text: session_configuration.base_instructions.clone(),
},
session_configuration.dynamic_tools.clone(),
event_persistence_mode,
),
)
}
InitialHistory::Resumed(resumed_history) => (
resumed_history.conversation_id,
if config.ephemeral {
RolloutStoreParams::resume(
resumed_history.rollout_path.clone(),
event_persistence_mode,
)
} else {
let resumed_source =
InMemoryRolloutSource::new(std::mem::take(&mut resumed_history.history));
RolloutStoreParams::resume_with_source(
resumed_history.rollout_path.clone(),
event_persistence_mode,
resumed_source,
)
},
),
};
// Kick off independent async setup tasks in parallel to reduce startup latency.
//
// - initialize RolloutRecorder with new or resumed session info
// - initialize RolloutStore with new or resumed session info
// - perform default shell discovery
// - load history metadata (skipped for subagents)
let rollout_fut = async {
if config.ephemeral {
Ok::<_, anyhow::Error>((None, None))
} else {
let state_db_ctx = state_db::init(&config, None).await;
let rollout_store = RolloutStore::new(
let state_db_ctx = state_db::init(&config).await;
let rollout_recorder = RolloutRecorder::new(
let rollout_store = RolloutStore::new(
&config,
rollout_params,
state_db_ctx.clone(),
state_builder.clone(),
)
.await?;
Ok((Some(rollout_recorder), state_db_ctx))
Ok((Some(rollout_store), state_db_ctx))
}
};
@@ -1364,18 +1375,16 @@ impl Session {
// Join all independent futures.
let (
rollout_recorder_and_state_db,
rollout_store_and_state_db,
(history_log_id, history_entry_count),
(auth, mcp_servers, auth_statuses),
) = tokio::join!(rollout_fut, history_meta_fut, auth_and_mcp_fut);
let (rollout_recorder, state_db_ctx) = rollout_recorder_and_state_db.map_err(|e| {
error!("failed to initialize rollout recorder: {e:#}");
let (rollout_store, state_db_ctx) = rollout_store_and_state_db.map_err(|e| {
error!("failed to initialize rollout store: {e:#}");
e
})?;
let rollout_path = rollout_recorder
.as_ref()
.map(|rec| rec.rollout_path.clone());
let rollout_path = rollout_store.as_ref().map(|rec| rec.rollout_path.clone());
let mut post_session_configured_events = Vec::<Event>::new();
@@ -1672,7 +1681,6 @@ impl Session {
}
// Dispatch the SessionConfiguredEvent first and then report any errors.
// If resuming, include converted initial messages in the payload so UIs can render them immediately.
let initial_messages = initial_history.get_event_msgs();
let events = std::iter::once(Event {
id: INITIAL_SUBMIT_ID.to_owned(),
msg: EventMsg::SessionConfigured(SessionConfiguredEvent {
@@ -1808,18 +1816,6 @@ impl Session {
}
}
pub(crate) async fn ensure_rollout_materialized(&self) {
let recorder = {
let guard = self.services.rollout.lock().await;
guard.clone()
};
if let Some(rec) = recorder
&& let Err(e) = rec.persist().await
{
warn!("failed to materialize rollout recorder: {e}");
}
}
fn next_internal_sub_id(&self) -> String {
let id = self
.next_internal_sub_id
@@ -1942,13 +1938,20 @@ impl Session {
}
}
InitialHistory::Resumed(resumed_history) => {
let rollout_items = resumed_history.history;
let rollout_source = if resumed_history.history.is_empty() {
let rollout = self.services.rollout.lock().await;
match rollout.as_ref() {
Some(rollout) => rollout.source.lock().await.clone(),
None => InMemoryRolloutSource::new(Vec::new()),
}
} else {
InMemoryRolloutSource::new(resumed_history.history)
};
let restored_tool_selection =
Self::extract_mcp_tool_selection_from_rollout(&rollout_items);
Self::extract_mcp_tool_selection_from_rollout_source(&rollout_source);
let reconstructed_rollout = self
.reconstruct_history_from_rollout(&turn_context, &rollout_items)
.await;
let reconstructed_rollout =
self.reconstruct_history_from_rollout(&turn_context, &rollout_source);
let previous_turn_settings = reconstructed_rollout.previous_turn_settings.clone();
self.set_previous_turn_settings(previous_turn_settings.clone())
.await;
@@ -1986,7 +1989,7 @@ impl Session {
// Seed usage info from the recorded rollout so UIs can show token counts
// immediately on resume/fork.
if let Some(info) = Self::last_token_info_from_rollout(&rollout_items) {
if let Some(info) = Self::last_token_info_from_rollout_source(&rollout_source) {
let mut state = self.state.lock().await;
state.set_token_info(Some(info));
}
@@ -2001,12 +2004,12 @@ impl Session {
}
}
InitialHistory::Forked(rollout_items) => {
let rollout_source = InMemoryRolloutSource::new(rollout_items.clone());
let restored_tool_selection =
Self::extract_mcp_tool_selection_from_rollout(&rollout_items);
Self::extract_mcp_tool_selection_from_rollout_source(&rollout_source);
let reconstructed_rollout = self
.reconstruct_history_from_rollout(&turn_context, &rollout_items)
.await;
let reconstructed_rollout =
self.reconstruct_history_from_rollout(&turn_context, &rollout_source);
self.set_previous_turn_settings(
reconstructed_rollout.previous_turn_settings.clone(),
)
@@ -2027,7 +2030,7 @@ impl Session {
// Seed usage info from the recorded rollout so UIs can show token counts
// immediately on resume/fork.
if let Some(info) = Self::last_token_info_from_rollout(&rollout_items) {
if let Some(info) = Self::last_token_info_from_rollout_source(&rollout_source) {
let mut state = self.state.lock().await;
state.set_token_info(Some(info));
}
@@ -2049,9 +2052,6 @@ impl Session {
state.set_reference_context_item(Some(turn_context.to_turn_context_item()));
}
// Forked threads should remain file-backed immediately after startup.
self.ensure_rollout_materialized().await;
// Flush after seeding history and any persisted rollout copy.
if !is_subagent {
self.flush_rollout().await;
@@ -2060,20 +2060,26 @@ impl Session {
}
}
fn last_token_info_from_rollout(rollout_items: &[RolloutItem]) -> Option<TokenUsageInfo> {
rollout_items.iter().rev().find_map(|item| match item {
RolloutItem::EventMsg(EventMsg::TokenCount(ev)) => ev.info.clone(),
_ => None,
})
fn last_token_info_from_rollout_source(
rollout_source: &InMemoryRolloutSource,
) -> Option<TokenUsageInfo> {
rollout_source
.iter_reverse_from(rollout_source.exclusive_end_of_rollout_index())
.find_map(|(_, item)| match item {
RolloutItem::EventMsg(EventMsg::TokenCount(ev)) => ev.info.clone(),
_ => None,
})
}
fn extract_mcp_tool_selection_from_rollout(
rollout_items: &[RolloutItem],
fn extract_mcp_tool_selection_from_rollout_source(
rollout_source: &InMemoryRolloutSource,
) -> Option<Vec<String>> {
let mut search_call_ids = HashSet::new();
let mut active_selected_tools: Option<Vec<String>> = None;
for item in rollout_items {
for (_, item) in
rollout_source.iter_forward_from(rollout_source.inclusive_start_of_rollout_index())
{
let RolloutItem::ResponseItem(response_item) = item else {
continue;
};
@@ -3667,7 +3673,6 @@ impl Session {
let turn_item = TurnItem::UserMessage(UserMessageItem::new(input));
self.emit_turn_item_started(turn_context, &turn_item).await;
self.emit_turn_item_completed(turn_context, turn_item).await;
self.ensure_rollout_materialized().await;
}
pub(crate) async fn notify_background_event(
@@ -4275,7 +4280,7 @@ mod handlers {
use crate::mcp::auth::compute_auth_statuses;
use crate::mcp::collect_mcp_snapshot_from_manager;
use crate::review_prompts::resolve_review_request;
use crate::rollout::RolloutRecorder;
use crate::rollout::RolloutStore;
use crate::rollout::session_index;
use crate::tasks::CompactTask;
use crate::tasks::UndoTask;
@@ -4925,24 +4930,24 @@ mod handlers {
return;
}
let initial_history =
match RolloutRecorder::get_rollout_history(rollout_path.as_path()).await {
Ok(history) => history,
Err(err) => {
sess.send_event_raw(Event {
id: turn_context.sub_id.clone(),
msg: EventMsg::Error(ErrorEvent {
message: format!(
"failed to load rollout `{}` for rollback replay: {err}",
rollout_path.display()
),
codex_error_info: Some(CodexErrorInfo::ThreadRollbackFailed),
}),
})
.await;
return;
}
};
let initial_history = match RolloutStore::get_rollout_history(rollout_path.as_path()).await
{
Ok(history) => history,
Err(err) => {
sess.send_event_raw(Event {
id: turn_context.sub_id.clone(),
msg: EventMsg::Error(ErrorEvent {
message: format!(
"failed to load rollout `{}` for rollback replay: {err}",
rollout_path.display()
),
codex_error_info: Some(CodexErrorInfo::ThreadRollbackFailed),
}),
})
.await;
return;
}
};
let rollback_event = ThreadRolledBackEvent { num_turns };
let replay_items = initial_history
@@ -4952,10 +4957,10 @@ mod handlers {
EventMsg::ThreadRolledBack(rollback_event.clone()),
)))
.collect::<Vec<_>>();
let replay_source = crate::rollout::recorder::InMemoryRolloutSource::new(replay_items);
let reconstructed = sess
.reconstruct_history_from_rollout(turn_context.as_ref(), replay_items.as_slice())
.await;
let reconstructed =
sess.reconstruct_history_from_rollout(turn_context.as_ref(), &replay_source);
sess.replace_history(
reconstructed.history,
reconstructed.reference_context_item.clone(),
codex-rs/core/src/codex/rollout_reconstruction.rs
+13 -10
View File
@@ -1,4 +1,5 @@
use super::*;
use crate::rollout::recorder::InMemoryRolloutSource;
// Return value of `Session::reconstruct_history_from_rollout`, bundling the rebuilt history with
// the resume/fork hydration metadata derived from the same replay.
@@ -83,10 +84,10 @@ fn finalize_active_segment<'a>(
}
impl Session {
pub(super) async fn reconstruct_history_from_rollout(
pub(super) fn reconstruct_history_from_rollout(
&self,
turn_context: &TurnContext,
rollout_items: &[RolloutItem],
source: &InMemoryRolloutSource,
) -> RolloutReconstruction {
// Replay metadata should already match the shape of the future lazy reverse loader, even
// while history materialization still uses an eager bridge. Scan newest-to-oldest,
@@ -99,14 +100,15 @@ impl Session {
// Rollback is "drop the newest N user turns". While scanning in reverse, that becomes
// "skip the next N user-turn segments we finalize".
let mut pending_rollback_turns = 0usize;
// Borrowed suffix of rollout items newer than the newest surviving replacement-history
// checkpoint. If no such checkpoint exists, this remains the full rollout.
let mut rollout_suffix = rollout_items;
// Default to replaying from the beginning of the rollout. If reverse replay later finds a
// surviving compaction with `replacement_history`, it advances this start index to the
// first rollout row after that compaction checkpoint.
let mut rollout_suffix_start = source.inclusive_start_of_rollout_index();
// Reverse replay accumulates rollout items into the newest in-progress turn segment until
// we hit its matching `TurnStarted`, at which point the segment can be finalized.
let mut active_segment: Option<ActiveReplaySegment<'_>> = None;
for (index, item) in rollout_items.iter().enumerate().rev() {
for (index, item) in source.iter_reverse_from(source.exclusive_end_of_rollout_index()) {
match item {
RolloutItem::Compacted(compacted) => {
let active_segment =
@@ -123,7 +125,7 @@ impl Session {
&& let Some(replacement_history) = &compacted.replacement_history
{
active_segment.base_replacement_history = Some(replacement_history);
rollout_suffix = &rollout_items[index + 1..];
rollout_suffix_start = index.next_newer();
}
}
RolloutItem::EventMsg(EventMsg::ThreadRolledBack(rollback)) => {
@@ -235,7 +237,7 @@ impl Session {
// Materialize exact history semantics from the replay-derived suffix. The eventual lazy
// design should keep this same replay shape, but drive it from a resumable reverse source
// instead of an eagerly loaded `&[RolloutItem]`.
for item in rollout_suffix {
for (_, item) in source.iter_forward_from(rollout_suffix_start) {
match item {
RolloutItem::ResponseItem(response_item) => {
history.record_items(
@@ -256,8 +258,9 @@ impl Session {
// `reference_context_item`, reinject canonical context at the end of the
// resumed conversation, and accept the temporary out-of-distribution
// prompt shape.
// TODO(ccunningham): if we drop support for None replacement_history compaction items,
// we can get rid of this second loop entirely and just build `history` directly in the first loop.
// If we eventually drop support for legacy compaction items without
// `replacement_history`, this second pass can go away and the first replay
// loop can materialize history directly.
let user_messages = collect_user_messages(history.raw_items());
let rebuilt = compact::build_compacted_history(
Vec::new(),
codex-rs/core/src/codex/rollout_reconstruction_tests.rs
+25 -18
View File
@@ -3,6 +3,7 @@ use super::*;
use crate::protocol::CompactedItem;
use crate::protocol::InitialHistory;
use crate::protocol::ResumedHistory;
use crate::rollout::recorder::InMemoryRolloutSource;
use codex_protocol::ThreadId;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
@@ -219,9 +220,10 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_com
)),
];
let reconstructed = session
.reconstruct_history_from_rollout(&turn_context, &rollout_items)
.await;
let reconstructed = session.reconstruct_history_from_rollout(
&turn_context,
&InMemoryRolloutSource::new(rollout_items),
);
assert_eq!(
reconstructed.history,
@@ -301,9 +303,10 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_inc
)),
];
let reconstructed = session
.reconstruct_history_from_rollout(&turn_context, &rollout_items)
.await;
let reconstructed = session.reconstruct_history_from_rollout(
&turn_context,
&InMemoryRolloutSource::new(rollout_items),
);
assert_eq!(
reconstructed.history,
@@ -407,9 +410,10 @@ async fn reconstruct_history_rollback_skips_non_user_turns_for_history_and_metad
)),
];
let reconstructed = session
.reconstruct_history_from_rollout(&turn_context, &rollout_items)
.await;
let reconstructed = session.reconstruct_history_from_rollout(
&turn_context,
&InMemoryRolloutSource::new(rollout_items),
);
assert_eq!(
reconstructed.history,
@@ -468,9 +472,10 @@ async fn reconstruct_history_rollback_clears_history_and_metadata_when_exceeding
)),
];
let reconstructed = session
.reconstruct_history_from_rollout(&turn_context, &rollout_items)
.await;
let reconstructed = session.reconstruct_history_from_rollout(
&turn_context,
&InMemoryRolloutSource::new(rollout_items),
);
assert_eq!(reconstructed.history, Vec::new());
assert_eq!(reconstructed.previous_turn_settings, None);
@@ -674,9 +679,10 @@ async fn reconstruct_history_legacy_compaction_without_replacement_history_does_
}),
];
let reconstructed = session
.reconstruct_history_from_rollout(&turn_context, &rollout_items)
.await;
let reconstructed = session.reconstruct_history_from_rollout(
&turn_context,
&InMemoryRolloutSource::new(rollout_items),
);
assert_eq!(
reconstructed.history,
@@ -727,9 +733,10 @@ async fn reconstruct_history_legacy_compaction_without_replacement_history_clear
)),
];
let reconstructed = session
.reconstruct_history_from_rollout(&turn_context, &rollout_items)
.await;
let reconstructed = session.reconstruct_history_from_rollout(
&turn_context,
&InMemoryRolloutSource::new(rollout_items),
);
assert!(reconstructed.reference_context_item.is_none());
}
codex-rs/core/src/codex_tests.rs
+197 -22
View File
@@ -40,8 +40,9 @@ use crate::protocol::TokenUsageInfo;
use crate::protocol::TurnCompleteEvent;
use crate::protocol::UserMessageEvent;
use crate::rollout::policy::EventPersistenceMode;
use crate::rollout::recorder::RolloutRecorder;
use crate::rollout::recorder::RolloutRecorderParams;
use crate::rollout::recorder::InMemoryRolloutSource;
use crate::rollout::recorder::RolloutStore;
use crate::rollout::recorder::RolloutStoreParams;
use crate::state::TaskKind;
use crate::tasks::SessionTask;
use crate::tasks::SessionTaskContext;
@@ -746,11 +747,11 @@ fn extract_mcp_tool_selection_from_rollout_returns_none_without_valid_search_out
async fn reconstruct_history_matches_live_compactions() {
let (session, turn_context) = make_session_and_context().await;
let (rollout_items, expected) = sample_rollout(&session, &turn_context).await;
let rollout_source = InMemoryRolloutSource::new(rollout_items);
let reconstruction_turn = session.new_default_turn().await;
let reconstructed = session
.reconstruct_history_from_rollout(reconstruction_turn.as_ref(), &rollout_items)
.await;
let reconstructed =
session.reconstruct_history_from_rollout(reconstruction_turn.as_ref(), &rollout_source);
assert_eq!(expected, reconstructed.history);
}
@@ -768,7 +769,7 @@ async fn reconstruct_history_uses_replacement_history_verbatim() {
phase: None,
};
let replacement_history = vec![
summary_item.clone(),
summary_item,
ResponseItem::Message {
id: None,
role: "developer".to_string(),
@@ -783,10 +784,9 @@ async fn reconstruct_history_uses_replacement_history_verbatim() {
message: String::new(),
replacement_history: Some(replacement_history.clone()),
})];
let rollout_source = InMemoryRolloutSource::new(rollout_items);
let reconstructed = session
.reconstruct_history_from_rollout(&turn_context, &rollout_items)
.await;
let reconstructed = session.reconstruct_history_from_rollout(&turn_context, &rollout_source);
assert_eq!(reconstructed.history, replacement_history);
}
@@ -1114,7 +1114,7 @@ async fn thread_rollback_drops_last_turn_from_history() {
assert_eq!(sess.previous_turn_settings().await, None);
assert!(sess.reference_context_item().await.is_none());
let InitialHistory::Resumed(resumed) = RolloutRecorder::get_rollout_history(&rollout_path)
let InitialHistory::Resumed(resumed) = RolloutStore::get_rollout_history(&rollout_path)
.await
.expect("read rollout history")
else {
@@ -1366,7 +1366,7 @@ async fn thread_rollback_persists_marker_and_replays_cumulatively() {
]
);
let InitialHistory::Resumed(resumed) = RolloutRecorder::get_rollout_history(&rollout_path)
let InitialHistory::Resumed(resumed) = RolloutStore::get_rollout_history(&rollout_path)
.await
.expect("read rollout history")
else {
@@ -1800,9 +1800,9 @@ async fn wait_for_thread_rollback_failed(rx: &async_channel::Receiver<Event>) ->
async fn attach_rollout_recorder(session: &Arc<Session>) -> PathBuf {
let config = session.get_config().await;
let recorder = RolloutRecorder::new(
let recorder = RolloutStore::new(
config.as_ref(),
RolloutRecorderParams::new(
RolloutStoreParams::new(
ThreadId::default(),
None,
SessionSource::Exec,
@@ -1820,7 +1820,6 @@ async fn attach_rollout_recorder(session: &Arc<Session>) -> PathBuf {
let mut rollout = session.services.rollout.lock().await;
*rollout = Some(recorder);
}
session.ensure_rollout_materialized().await;
session.flush_rollout().await;
rollout_path
}
@@ -3367,9 +3366,9 @@ async fn record_context_updates_and_set_reference_context_item_persists_baseline
state.set_reference_context_item(Some(previous_context_item.clone()));
}
let config = session.get_config().await;
let recorder = RolloutRecorder::new(
let recorder = RolloutStore::new(
config.as_ref(),
RolloutRecorderParams::new(
RolloutStoreParams::new(
ThreadId::default(),
None,
SessionSource::Exec,
@@ -3407,10 +3406,9 @@ async fn record_context_updates_and_set_reference_context_item_persists_baseline
serde_json::to_value(Some(turn_context.to_turn_context_item()))
.expect("serialize expected context item")
);
session.ensure_rollout_materialized().await;
session.flush_rollout().await;
let InitialHistory::Resumed(resumed) = RolloutRecorder::get_rollout_history(&rollout_path)
let InitialHistory::Resumed(resumed) = RolloutStore::get_rollout_history(&rollout_path)
.await
.expect("read rollout history")
else {
@@ -3464,9 +3462,9 @@ async fn record_context_updates_and_set_reference_context_item_persists_full_rei
.with_model(next_model.to_string(), &session.services.models_manager)
.await;
let config = session.get_config().await;
let recorder = RolloutRecorder::new(
let recorder = RolloutStore::new(
config.as_ref(),
RolloutRecorderParams::new(
RolloutStoreParams::new(
ThreadId::default(),
None,
SessionSource::Exec,
@@ -3509,10 +3507,9 @@ async fn record_context_updates_and_set_reference_context_item_persists_full_rei
session
.record_context_updates_and_set_reference_context_item(&turn_context)
.await;
session.ensure_rollout_materialized().await;
session.flush_rollout().await;
let InitialHistory::Resumed(resumed) = RolloutRecorder::get_rollout_history(&rollout_path)
let InitialHistory::Resumed(resumed) = RolloutStore::get_rollout_history(&rollout_path)
.await
.expect("read rollout history")
else {
@@ -3531,6 +3528,184 @@ async fn record_context_updates_and_set_reference_context_item_persists_full_rei
);
}
#[tokio::test]
async fn session_new_ephemeral_resume_keeps_in_memory_history() {
let (seed_session, turn_context) = make_session_and_context().await;
let (rollout_items, expected) = sample_rollout(&seed_session, &turn_context).await;
let codex_home = tempfile::tempdir().expect("create temp dir");
let mut config = build_test_config(codex_home.path()).await;
config.ephemeral = true;
let config = Arc::new(config);
let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("Test API Key"));
let models_manager = Arc::new(ModelsManager::new(
config.codex_home.clone(),
auth_manager.clone(),
None,
CollaborationModesConfig::default(),
));
let model = ModelsManager::get_model_offline_for_tests(config.model.as_deref());
let model_info = ModelsManager::construct_model_info_offline_for_tests(model.as_str(), &config);
let collaboration_mode = CollaborationMode {
mode: ModeKind::Default,
settings: Settings {
model,
reasoning_effort: config.model_reasoning_effort,
developer_instructions: None,
},
};
let session_configuration = SessionConfiguration {
provider: config.model_provider.clone(),
collaboration_mode,
model_reasoning_summary: config.model_reasoning_summary,
developer_instructions: config.developer_instructions.clone(),
user_instructions: config.user_instructions.clone(),
personality: config.personality,
base_instructions: config
.base_instructions
.clone()
.unwrap_or_else(|| model_info.get_model_instructions(config.personality)),
compact_prompt: config.compact_prompt.clone(),
approval_policy: config.permissions.approval_policy.clone(),
sandbox_policy: config.permissions.sandbox_policy.clone(),
windows_sandbox_level: WindowsSandboxLevel::from_config(&config),
cwd: config.cwd.clone(),
codex_home: config.codex_home.clone(),
thread_name: None,
original_config_do_not_use: Arc::clone(&config),
service_tier: config.service_tier,
metrics_service_name: None,
app_server_client_name: None,
session_source: SessionSource::Exec,
dynamic_tools: Vec::new(),
persist_extended_history: false,
inherited_shell_snapshot: None,
};
let (tx_event, _rx_event) = async_channel::unbounded();
let (agent_status_tx, _agent_status_rx) = watch::channel(AgentStatus::PendingInit);
let plugins_manager = Arc::new(PluginsManager::new(config.codex_home.clone()));
let mcp_manager = Arc::new(McpManager::new(Arc::clone(&plugins_manager)));
let skills_manager = Arc::new(SkillsManager::new(
config.codex_home.clone(),
Arc::clone(&plugins_manager),
));
let session = Session::new(
session_configuration,
Arc::clone(&config),
auth_manager,
models_manager,
ExecPolicyManager::default(),
tx_event,
agent_status_tx,
InitialHistory::Resumed(ResumedHistory {
conversation_id: ThreadId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
}),
SessionSource::Exec,
skills_manager,
plugins_manager,
mcp_manager,
Arc::new(FileWatcher::noop()),
AgentControl::default(),
)
.await
.expect("create ephemeral resumed session");
assert_eq!(expected, session.clone_history().await.raw_items());
}
#[tokio::test]
async fn session_new_ephemeral_resume_with_empty_history_does_not_panic() {
let codex_home = tempfile::tempdir().expect("create temp dir");
let mut config = build_test_config(codex_home.path()).await;
config.ephemeral = true;
let config = Arc::new(config);
let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("Test API Key"));
let models_manager = Arc::new(ModelsManager::new(
config.codex_home.clone(),
auth_manager.clone(),
None,
CollaborationModesConfig::default(),
));
let model = ModelsManager::get_model_offline_for_tests(config.model.as_deref());
let model_info = ModelsManager::construct_model_info_offline_for_tests(model.as_str(), &config);
let collaboration_mode = CollaborationMode {
mode: ModeKind::Default,
settings: Settings {
model,
reasoning_effort: config.model_reasoning_effort,
developer_instructions: None,
},
};
let session_configuration = SessionConfiguration {
provider: config.model_provider.clone(),
collaboration_mode,
model_reasoning_summary: config.model_reasoning_summary,
developer_instructions: config.developer_instructions.clone(),
user_instructions: config.user_instructions.clone(),
personality: config.personality,
base_instructions: config
.base_instructions
.clone()
.unwrap_or_else(|| model_info.get_model_instructions(config.personality)),
compact_prompt: config.compact_prompt.clone(),
approval_policy: config.permissions.approval_policy.clone(),
sandbox_policy: config.permissions.sandbox_policy.clone(),
windows_sandbox_level: WindowsSandboxLevel::from_config(&config),
cwd: config.cwd.clone(),
codex_home: config.codex_home.clone(),
thread_name: None,
original_config_do_not_use: Arc::clone(&config),
service_tier: config.service_tier,
metrics_service_name: None,
app_server_client_name: None,
session_source: SessionSource::Exec,
dynamic_tools: Vec::new(),
persist_extended_history: false,
inherited_shell_snapshot: None,
};
let (tx_event, _rx_event) = async_channel::unbounded();
let (agent_status_tx, _agent_status_rx) = watch::channel(AgentStatus::PendingInit);
let plugins_manager = Arc::new(PluginsManager::new(config.codex_home.clone()));
let mcp_manager = Arc::new(McpManager::new(Arc::clone(&plugins_manager)));
let skills_manager = Arc::new(SkillsManager::new(
config.codex_home.clone(),
Arc::clone(&plugins_manager),
));
let session = Session::new(
session_configuration,
Arc::clone(&config),
auth_manager,
models_manager,
ExecPolicyManager::default(),
tx_event,
agent_status_tx,
InitialHistory::Resumed(ResumedHistory {
conversation_id: ThreadId::default(),
history: Vec::new(),
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
}),
SessionSource::Exec,
skills_manager,
plugins_manager,
mcp_manager,
Arc::new(FileWatcher::noop()),
AgentControl::default(),
)
.await
.expect("create ephemeral resumed session with empty history");
assert_eq!(
Vec::<ResponseItem>::new(),
session.clone_history().await.raw_items()
);
}
#[tokio::test]
async fn run_user_shell_command_does_not_set_reference_context_item() {
let (session, _turn_context, rx) = make_session_and_context_with_rx().await;
codex-rs/core/src/codex_thread.rs
+1
View File
@@ -132,6 +132,7 @@ impl CodexThread {
.await;
}
/// Return the rollout path for this thread, if any.
pub fn rollout_path(&self) -> Option<PathBuf> {
self.rollout_path.clone()
}
codex-rs/core/src/lib.rs
+2 -2
View File
@@ -118,8 +118,8 @@ mod turn_metadata;
mod turn_timing;
pub use rollout::ARCHIVED_SESSIONS_SUBDIR;
pub use rollout::INTERACTIVE_SESSION_SOURCES;
pub use rollout::RolloutRecorder;
pub use rollout::RolloutRecorderParams;
pub use rollout::RolloutStore;
pub use rollout::RolloutStoreParams;
pub use rollout::SESSIONS_SUBDIR;
pub use rollout::SessionMeta;
pub use rollout::append_thread_name;
codex-rs/core/src/memories/phase1.rs
+15 -6
View File
@@ -1,5 +1,5 @@
use crate::Prompt;
use crate::RolloutRecorder;
use crate::RolloutStore;
use crate::codex::Session;
use crate::codex::TurnContext;
use crate::config::Config;
@@ -315,8 +315,17 @@ mod job {
rollout_cwd: &Path,
stage_one_context: &RequestContext,
) -> anyhow::Result<(StageOneOutput, Option<TokenUsage>)> {
let (rollout_items, _, _) = RolloutRecorder::load_rollout_items(rollout_path).await?;
let rollout_contents = serialize_filtered_rollout_response_items(&rollout_items)?;
let loaded_rollout = RolloutStore::load_source(rollout_path).await?;
let source = loaded_rollout.source;
// TODO(ccunningham): avoid serializing the whole rollout here; once this path has a
// token-budget-aware head/tail collector, it should use `RolloutSource` forward/reverse
// iteration to build only the sections that will survive into the stage-1 sampling
// request.
let rollout_contents = serialize_filtered_rollout_response_items(
source
.iter_forward_from(source.inclusive_start_of_rollout_index())
.map(|(_, item)| item),
)?;
let prompt = Prompt {
input: vec![ResponseItem::Message {
@@ -463,11 +472,11 @@ mod job {
}
/// Serializes filtered stage-1 memory items for prompt inclusion.
fn serialize_filtered_rollout_response_items(
items: &[RolloutItem],
fn serialize_filtered_rollout_response_items<'a>(
items: impl IntoIterator<Item = &'a RolloutItem>,
) -> crate::error::Result<String> {
let filtered = items
.iter()
.into_iter()
.filter_map(|item| {
if let RolloutItem::ResponseItem(item) = item
&& should_persist_response_item_for_memories(item)
codex-rs/core/src/rollout/metadata.rs
+53 -19
View File
@@ -1,7 +1,7 @@
use crate::config::Config;
use crate::rollout;
use crate::rollout::RolloutStore;
use crate::rollout::list::parse_timestamp_uuid_from_filename;
use crate::rollout::recorder::RolloutRecorder;
use crate::state_db::normalize_cwd_for_state_db;
use chrono::DateTime;
use chrono::NaiveDateTime;
@@ -61,11 +61,11 @@ pub(crate) fn builder_from_session_meta(
Some(builder)
}
pub(crate) fn builder_from_items(
items: &[RolloutItem],
pub(crate) fn builder_from_items<'a>(
items: impl IntoIterator<Item = &'a RolloutItem>,
rollout_path: &Path,
) -> Option<ThreadMetadataBuilder> {
if let Some(session_meta) = items.iter().find_map(|item| match item {
if let Some(session_meta) = items.into_iter().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => Some(meta_line),
RolloutItem::ResponseItem(_)
| RolloutItem::Compacted(_)
@@ -96,22 +96,31 @@ pub(crate) async fn extract_metadata_from_rollout(
rollout_path: &Path,
default_provider: &str,
) -> anyhow::Result<ExtractionOutcome> {
let (items, _thread_id, parse_errors) =
RolloutRecorder::load_rollout_items(rollout_path).await?;
if items.is_empty() {
return Err(anyhow::anyhow!(
"empty session file: {}",
let loaded_rollout = RolloutStore::load_source(rollout_path).await?;
let parse_errors = loaded_rollout.parse_errors;
let source = loaded_rollout.source;
let rollout_start = source.inclusive_start_of_rollout_index();
let has_any_parsed_items = source.iter_forward_from(rollout_start).next().is_some();
if parse_errors > 0 && !has_any_parsed_items {
anyhow::bail!(
"rollout contains parse errors and no readable items: {}",
rollout_path.display()
));
);
}
let builder = builder_from_items(items.as_slice(), rollout_path).ok_or_else(|| {
let builder = builder_from_items(
source
.iter_forward_from(rollout_start)
.map(|(_, item)| item),
rollout_path,
)
.ok_or_else(|| {
anyhow::anyhow!(
"rollout missing metadata builder: {}",
rollout_path.display()
)
})?;
let mut metadata = builder.build(default_provider);
for item in &items {
for (_, item) in source.iter_forward_from(rollout_start) {
apply_rollout_item(&mut metadata, item, default_provider);
}
if let Some(updated_at) = file_modified_time_utc(rollout_path).await {
@@ -119,13 +128,15 @@ pub(crate) async fn extract_metadata_from_rollout(
}
Ok(ExtractionOutcome {
metadata,
memory_mode: items.iter().rev().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => meta_line.meta.memory_mode.clone(),
RolloutItem::ResponseItem(_)
| RolloutItem::Compacted(_)
| RolloutItem::TurnContext(_)
| RolloutItem::EventMsg(_) => None,
}),
memory_mode: source
.iter_reverse_from(source.exclusive_end_of_rollout_index())
.find_map(|(_, item)| match item {
RolloutItem::SessionMeta(meta_line) => meta_line.meta.memory_mode.clone(),
RolloutItem::ResponseItem(_)
| RolloutItem::Compacted(_)
| RolloutItem::TurnContext(_)
| RolloutItem::EventMsg(_) => None,
}),
parse_errors,
})
}
@@ -573,6 +584,29 @@ mod tests {
assert_eq!(outcome.memory_mode.as_deref(), Some("polluted"));
}
#[tokio::test]
async fn extract_metadata_from_rollout_rejects_unparseable_rollout_with_no_items() {
let dir = tempdir().expect("tempdir");
let uuid = Uuid::new_v4();
let path = dir
.path()
.join(format!("rollout-2026-01-27T12-34-56-{uuid}.jsonl"));
let mut file = File::create(&path).expect("create rollout");
writeln!(file, "{{").expect("write invalid rollout");
let err = extract_metadata_from_rollout(&path, "openai", None)
.await
.expect_err("unparseable rollout should fail metadata extraction");
assert_eq!(
err.to_string(),
format!(
"rollout contains parse errors and no readable items: {}",
path.display()
)
);
}
#[test]
fn builder_from_items_falls_back_to_filename() {
let dir = tempdir().expect("tempdir");
codex-rs/core/src/rollout/mod.rs
+3 -3
View File
@@ -11,7 +11,7 @@ pub(crate) mod error;
pub mod list;
pub(crate) mod metadata;
pub(crate) mod policy;
pub mod recorder;
pub(crate) mod recorder;
pub(crate) mod session_index;
pub(crate) mod truncation;
@@ -22,8 +22,8 @@ pub use list::find_thread_path_by_id_str;
#[deprecated(note = "use find_thread_path_by_id_str")]
pub use list::find_thread_path_by_id_str as find_conversation_path_by_id_str;
pub use list::rollout_date_parts;
pub use recorder::RolloutRecorder;
pub use recorder::RolloutRecorderParams;
pub use recorder::RolloutStore;
pub use recorder::RolloutStoreParams;
pub use session_index::append_thread_name;
pub use session_index::find_thread_name_by_id;
pub use session_index::find_thread_path_by_name_str;
codex-rs/core/src/rollout/recorder.rs
+401 -241
View File
@@ -5,6 +5,7 @@ use std::fs::{self};
use std::io::Error as IoError;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use chrono::SecondsFormat;
use chrono::Utc;
@@ -17,6 +18,7 @@ use time::format_description::FormatItem;
use time::format_description::well_known::Rfc3339;
use time::macros::format_description;
use tokio::io::AsyncWriteExt;
use tokio::sync::Mutex;
use tokio::sync::mpsc::Sender;
use tokio::sync::mpsc::{self};
use tokio::sync::oneshot;
@@ -41,7 +43,6 @@ use super::policy::EventPersistenceMode;
use super::policy::is_persisted_response_item;
use crate::config::Config;
use crate::default_client::originator;
use crate::git_info::collect_git_info;
use crate::path_utils;
use crate::state_db;
use crate::state_db::StateDbHandle;
@@ -58,8 +59,134 @@ use codex_protocol::protocol::SessionSource;
use codex_state::StateRuntime;
use codex_state::ThreadMetadataBuilder;
/// Records all [`ResponseItem`]s for a session and flushes them to disk after
/// every update.
/// Logical position in the rollout stream.
///
/// This typed wrapper keeps rollout positions distinct from raw slice offsets.
///
/// `-1` is the newest rollout row that already existed when this source was created. Older
/// persisted rows are more negative, and any rows appended after startup are `0`, `1`, `2`, and
/// so on.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) struct RolloutIndex(i64);
impl RolloutIndex {
/// Return the next newer logical rollout position.
pub(crate) fn next_newer(self) -> Self {
Self(self.0 + 1)
}
}
/// Parsed rollout source plus file-level metadata gathered while loading by path.
pub(crate) struct LoadedRolloutSource {
pub(crate) source: InMemoryRolloutSource,
pub(crate) thread_id: Option<ThreadId>,
pub(crate) parse_errors: usize,
}
/// In-memory `RolloutSource` for the current startup/replay paths that still materialize the
/// entire rollout before reconstructing history or extracting metadata.
///
/// The long-term plan is to make this source lazy over the rollout file so startup and replay can
/// support unbounded conversation history without loading the entire rollout into memory up front.
/// When that happens, the equivalent source should keep the same "load older items on demand"
/// contract, but page older rollout rows from disk instead of cloning them out of a fully loaded
/// `Vec<RolloutItem>`.
#[derive(Clone, Debug)]
pub struct InMemoryRolloutSource {
rollout_items: Vec<RolloutItem>,
startup_rollout_len: i64,
}
impl InMemoryRolloutSource {
/// Build an in-memory `RolloutSource` from a fully loaded rollout item stream.
pub(crate) fn new(rollout_items: Vec<RolloutItem>) -> Self {
let startup_rollout_len = i64::try_from(rollout_items.len()).unwrap_or(i64::MAX);
Self {
rollout_items,
startup_rollout_len,
}
}
/// Load a persisted rollout file into the eager in-memory `RolloutSource` used today.
pub(crate) async fn load_from_path(path: &Path) -> std::io::Result<LoadedRolloutSource> {
let (items, thread_id, parse_errors) = RolloutStore::load_rollout_items(path).await?;
Ok(LoadedRolloutSource {
source: Self::new(items),
thread_id,
parse_errors,
})
}
/// Append newly recorded rollout items after startup.
pub(crate) fn append_items(&mut self, items: Vec<RolloutItem>) {
self.rollout_items.extend(items);
}
/// Consume the source back into owned rollout items.
pub(crate) fn into_items(self) -> Vec<RolloutItem> {
self.rollout_items
}
/// Return the inclusive index of the start of the persisted rollout file.
pub(crate) fn inclusive_start_of_rollout_index(&self) -> RolloutIndex {
RolloutIndex(-self.startup_rollout_len)
}
/// Return the exclusive end index of the persisted rollout file plus any items
/// appended in memory after startup.
pub(crate) fn exclusive_end_of_rollout_index(&self) -> RolloutIndex {
let rollout_len = i64::try_from(self.rollout_items.len()).unwrap_or(i64::MAX);
RolloutIndex(rollout_len.saturating_sub(self.startup_rollout_len))
}
/// Iterate forward from the inclusive `start` position.
pub(crate) fn iter_forward_from(
&self,
start: RolloutIndex,
) -> impl Iterator<Item = (RolloutIndex, &RolloutItem)> + '_ {
let start = self.loaded_offset_from_rollout_index(start);
self.rollout_items[start..]
.iter()
.enumerate()
.map(move |(offset, item)| {
let offset = i64::try_from(offset).unwrap_or(i64::MAX);
(
RolloutIndex(start as i64 + offset - self.startup_rollout_len),
item,
)
})
}
/// Iterate backward over rows strictly older than the exclusive `end` position.
pub(crate) fn iter_reverse_from(
&self,
end: RolloutIndex,
) -> impl Iterator<Item = (RolloutIndex, &RolloutItem)> + '_ {
let end = self.loaded_offset_from_rollout_index(end);
self.rollout_items[..end]
.iter()
.enumerate()
.rev()
.map(move |(actual_index, item)| {
let actual_index = i64::try_from(actual_index).unwrap_or(i64::MAX);
(RolloutIndex(actual_index - self.startup_rollout_len), item)
})
}
fn loaded_offset_from_rollout_index(&self, rollout_index: RolloutIndex) -> usize {
let actual_index = rollout_index.0.saturating_add(self.startup_rollout_len);
if actual_index <= 0 {
return 0;
}
match usize::try_from(actual_index) {
Ok(actual_index) => actual_index.min(self.rollout_items.len()),
Err(_) => self.rollout_items.len(),
}
}
}
/// Manages the canonical rollout state for the current process and persists it to disk.
///
/// Rollouts are recorded as JSONL and can be inspected with tools such as:
///
@@ -68,15 +195,20 @@ use codex_state::ThreadMetadataBuilder;
/// $ fx ~/.codex/sessions/rollout-2025-05-07T17-24-21-5973b6c0-94b8-487b-a530-2aeb6098ae0e.jsonl
/// ```
#[derive(Clone)]
pub struct RolloutRecorder {
pub struct RolloutStore {
tx: Sender<RolloutCmd>,
pub(crate) rollout_path: PathBuf,
state_db: Option<StateDbHandle>,
event_persistence_mode: EventPersistenceMode,
// Canonical in-memory `RolloutSource` for this process. Startup loading and runtime writes
// both update this source so replay sees the same rollout items and session metadata that the
// writer is responsible for persisting.
pub(crate) source: Arc<Mutex<InMemoryRolloutSource>>,
}
/// Parameters for creating or resuming a rollout store.
#[derive(Clone)]
pub enum RolloutRecorderParams {
pub enum RolloutStoreParams {
Create {
conversation_id: ThreadId,
forked_from_id: Option<ThreadId>,
@@ -88,6 +220,7 @@ pub enum RolloutRecorderParams {
Resume {
path: PathBuf,
event_persistence_mode: EventPersistenceMode,
source: Option<InMemoryRolloutSource>,
},
}
@@ -105,7 +238,8 @@ enum RolloutCmd {
},
}
impl RolloutRecorderParams {
impl RolloutStoreParams {
/// Create rollout parameters for a brand-new session.
pub fn new(
conversation_id: ThreadId,
forked_from_id: Option<ThreadId>,
@@ -124,10 +258,26 @@ impl RolloutRecorderParams {
}
}
/// Create rollout parameters for resuming an existing persisted rollout.
pub fn resume(path: PathBuf, event_persistence_mode: EventPersistenceMode) -> Self {
Self::Resume {
path,
event_persistence_mode,
source: None,
}
}
/// Create rollout parameters for resume when startup already has a preloaded
/// in-memory rollout source and should avoid reading the rollout file again.
pub fn resume_with_source(
path: PathBuf,
event_persistence_mode: EventPersistenceMode,
source: InMemoryRolloutSource,
) -> Self {
Self::Resume {
path,
event_persistence_mode,
source: Some(source),
}
}
}
@@ -159,7 +309,12 @@ fn sanitize_rollout_item_for_persistence(
}
}
impl RolloutRecorder {
impl RolloutStore {
/// Load a persisted rollout file into the eager in-memory `RolloutSource` used today.
pub(crate) async fn load_source(path: &Path) -> std::io::Result<LoadedRolloutSource> {
InMemoryRolloutSource::load_from_path(path).await
}
/// List threads (rollout files) under the provided Codex home directory.
#[allow(clippy::too_many_arguments)]
pub async fn list_threads(
@@ -361,108 +516,124 @@ impl RolloutRecorder {
}
}
/// Attempt to create a new [`RolloutRecorder`].
/// Attempt to create a new [`RolloutStore`].
///
/// For newly created sessions, this precomputes path/metadata and defers
/// file creation/open until an explicit `persist()` call.
///
/// For resumed sessions, this immediately opens the existing rollout file.
/// For both new and resumed sessions, this opens the rollout file immediately so the path is
/// readable as soon as startup returns.
pub async fn new(
config: &Config,
params: RolloutRecorderParams,
params: RolloutStoreParams,
state_db_ctx: Option<StateDbHandle>,
state_builder: Option<ThreadMetadataBuilder>,
) -> std::io::Result<Self> {
let (file, deferred_log_file_info, rollout_path, meta, event_persistence_mode) =
match params {
RolloutRecorderParams::Create {
conversation_id,
let (file, rollout_path, source, meta, event_persistence_mode) = match params {
RolloutStoreParams::Create {
conversation_id,
forked_from_id,
source,
base_instructions,
dynamic_tools,
event_persistence_mode,
} => {
let log_file_info = precompute_log_file_info(config, conversation_id)?;
let path = log_file_info.path.clone();
let session_id = log_file_info.conversation_id;
let started_at = log_file_info.timestamp;
let timestamp_format: &[FormatItem] = format_description!(
"[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:3]Z"
);
let timestamp = started_at
.to_offset(time::UtcOffset::UTC)
.format(timestamp_format)
.map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?;
let session_meta = SessionMeta {
id: session_id,
forked_from_id,
timestamp,
cwd: config.cwd.clone(),
originator: originator().value,
cli_version: env!("CARGO_PKG_VERSION").to_string(),
agent_nickname: source.get_nickname(),
agent_role: source.get_agent_role(),
source,
base_instructions,
dynamic_tools,
event_persistence_mode,
} => {
let log_file_info = precompute_log_file_info(config, conversation_id)?;
let path = log_file_info.path.clone();
let session_id = log_file_info.conversation_id;
let started_at = log_file_info.timestamp;
model_provider: Some(config.model_provider_id.clone()),
base_instructions: Some(base_instructions),
dynamic_tools: if dynamic_tools.is_empty() {
None
} else {
Some(dynamic_tools)
},
memory_mode: (!config.memories.generate_memories)
.then_some("disabled".to_string()),
};
let session_meta_line = SessionMetaLine {
meta: session_meta,
git: None,
};
let timestamp_format: &[FormatItem] = format_description!(
"[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:3]Z"
);
let timestamp = started_at
.to_offset(time::UtcOffset::UTC)
.format(timestamp_format)
.map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?;
let session_meta = SessionMeta {
id: session_id,
forked_from_id,
timestamp,
cwd: config.cwd.clone(),
originator: originator().value,
cli_version: env!("CARGO_PKG_VERSION").to_string(),
agent_nickname: source.get_nickname(),
agent_role: source.get_agent_role(),
source,
model_provider: Some(config.model_provider_id.clone()),
base_instructions: Some(base_instructions),
dynamic_tools: if dynamic_tools.is_empty() {
None
} else {
Some(dynamic_tools)
},
memory_mode: (!config.memories.generate_memories)
.then_some("disabled".to_string()),
};
(
None,
Some(log_file_info),
path,
Some(session_meta),
event_persistence_mode,
)
}
RolloutRecorderParams::Resume {
(
tokio::fs::File::from_std(open_log_file(path.as_path())?),
path,
InMemoryRolloutSource::new(vec![RolloutItem::SessionMeta(
session_meta_line.clone(),
)]),
Some(session_meta_line),
event_persistence_mode,
} => (
Some(
tokio::fs::OpenOptions::new()
.append(true)
.open(&path)
.await?,
),
None,
)
}
RolloutStoreParams::Resume {
path,
event_persistence_mode,
source,
} => {
let source = match source {
Some(source) => source,
None => Self::load_source(path.as_path()).await?.source,
};
(
tokio::fs::OpenOptions::new()
.append(true)
.open(&path)
.await?,
path,
source,
None,
event_persistence_mode,
),
};
// Clone the cwd for the spawned task to collect git info asynchronously
let cwd = config.cwd.clone();
)
}
};
// A reasonably-sized bounded channel. If the buffer fills up the send
// future will yield, which is fine – we only need to ensure we do not
// perform *blocking* I/O on the caller's thread.
let (tx, rx) = mpsc::channel::<RolloutCmd>(256);
let source = Arc::new(Mutex::new(source));
let mut writer = JsonlWriter { file };
let mut state_builder = state_builder;
if let Some(session_meta_line) = meta {
write_session_meta(
&mut writer,
session_meta_line,
&rollout_path,
state_db_ctx.as_deref(),
&mut state_builder,
config.model_provider_id.as_str(),
)
.await?;
}
// Spawn a Tokio task that owns the file handle and performs async
// writes. Using `tokio::fs::File` keeps everything on the async I/O
// driver instead of blocking the runtime.
tokio::task::spawn(rollout_writer(
file,
deferred_log_file_info,
writer.file,
rx,
meta,
cwd,
rollout_path.clone(),
state_db_ctx.clone(),
state_builder,
config.model_provider_id.clone(),
config.memories.generate_memories,
));
Ok(Self {
@@ -470,13 +641,16 @@ impl RolloutRecorder {
rollout_path,
state_db: state_db_ctx,
event_persistence_mode,
source,
})
}
/// Return the path for this rollout file.
pub fn rollout_path(&self) -> &Path {
self.rollout_path.as_path()
}
/// Return the state DB handle used for rollout reconciliation, when enabled.
pub fn state_db(&self) -> Option<StateDbHandle> {
self.state_db.clone()
}
@@ -497,15 +671,16 @@ impl RolloutRecorder {
if filtered.is_empty() {
return Ok(());
}
let mut source = self.source.lock().await;
self.tx
.send(RolloutCmd::AddItems(filtered))
.send(RolloutCmd::AddItems(filtered.clone()))
.await
.map_err(|e| IoError::other(format!("failed to queue rollout items: {e}")))
.map_err(|e| IoError::other(format!("failed to queue rollout items: {e}")))?;
source.append_items(filtered);
Ok(())
}
/// Materialize the rollout file and persist all buffered items.
///
/// This is idempotent; after first materialization, repeated calls are no-ops.
/// Ensure all queued rollout writes have been persisted to disk.
pub async fn persist(&self) -> std::io::Result<()> {
let (tx, rx) = oneshot::channel();
self.tx
@@ -527,6 +702,8 @@ impl RolloutRecorder {
.map_err(|e| IoError::other(format!("failed waiting for rollout flush: {e}")))
}
// TODO(ccunningham): move this parser under the source implementation
// (not done in this PR to reduce diff)
pub(crate) async fn load_rollout_items(
path: &Path,
) -> std::io::Result<(Vec<RolloutItem>, Option<ThreadId>, usize)> {
@@ -552,29 +729,18 @@ impl RolloutRecorder {
}
};
// Parse the rollout line structure
match serde_json::from_value::<RolloutLine>(v.clone()) {
Ok(rollout_line) => match rollout_line.item {
RolloutItem::SessionMeta(session_meta_line) => {
// Use the FIRST SessionMeta encountered in the file as the canonical
// thread id and main session information. Keep all items intact.
if thread_id.is_none() {
thread_id = Some(session_meta_line.meta.id);
}
items.push(RolloutItem::SessionMeta(session_meta_line));
}
RolloutItem::ResponseItem(item) => {
items.push(RolloutItem::ResponseItem(item));
}
RolloutItem::Compacted(item) => {
items.push(RolloutItem::Compacted(item));
}
RolloutItem::TurnContext(item) => {
items.push(RolloutItem::TurnContext(item));
}
RolloutItem::EventMsg(_ev) => {
items.push(RolloutItem::EventMsg(_ev));
}
RolloutItem::ResponseItem(item) => items.push(RolloutItem::ResponseItem(item)),
RolloutItem::Compacted(item) => items.push(RolloutItem::Compacted(item)),
RolloutItem::TurnContext(item) => items.push(RolloutItem::TurnContext(item)),
RolloutItem::EventMsg(event) => items.push(RolloutItem::EventMsg(event)),
},
Err(e) => {
trace!("failed to parse rollout line: {e}");
@@ -592,10 +758,21 @@ impl RolloutRecorder {
Ok((items, thread_id, parse_errors))
}
/// Load a persisted rollout file into the eager startup history shape used today.
pub async fn get_rollout_history(path: &Path) -> std::io::Result<InitialHistory> {
let (items, thread_id, _parse_errors) = Self::load_rollout_items(path).await?;
let LoadedRolloutSource {
source,
thread_id,
parse_errors: _,
} = Self::load_source(path).await?;
let conversation_id = thread_id
.ok_or_else(|| IoError::other("failed to parse thread ID from rollout file"))?;
// `InitialHistory::Resumed` still carries an owned `Vec<RolloutItem>`, so this is the
// boundary where the store's canonical `RolloutSource` snapshot is flattened back into
// the protocol-owned rollout payload for resume/fork startup.
// TODO(ccunningham): when resume startup becomes lazy, replace this eager
// `InitialHistory::Resumed` materialization with a `RolloutSource`-backed history input.
let items = source.into_items();
if items.is_empty() {
return Ok(InitialHistory::New);
@@ -609,6 +786,7 @@ impl RolloutRecorder {
}))
}
/// Shut down the background writer after draining all previously queued work.
pub async fn shutdown(&self) -> std::io::Result<()> {
let (tx_done, rx_done) = oneshot::channel();
match self.tx.send(RolloutCmd::Shutdown { ack: tx_done }).await {
@@ -706,41 +884,18 @@ fn open_log_file(path: &Path) -> std::io::Result<File> {
#[allow(clippy::too_many_arguments)]
async fn rollout_writer(
file: Option<tokio::fs::File>,
mut deferred_log_file_info: Option<LogFileInfo>,
file: tokio::fs::File,
mut rx: mpsc::Receiver<RolloutCmd>,
mut meta: Option<SessionMeta>,
cwd: std::path::PathBuf,
rollout_path: PathBuf,
state_db_ctx: Option<StateDbHandle>,
mut state_builder: Option<ThreadMetadataBuilder>,
default_provider: String,
generate_memories: bool,
) -> std::io::Result<()> {
let mut writer = file.map(|file| JsonlWriter { file });
let mut buffered_items = Vec::<RolloutItem>::new();
let mut writer = JsonlWriter { file };
if let Some(builder) = state_builder.as_mut() {
builder.rollout_path = rollout_path.clone();
}
// Resumed sessions already have a file handle open, so session metadata can
// be written immediately if present.
if writer.is_some()
&& let Some(session_meta) = meta.take()
{
write_session_meta(
writer.as_mut(),
session_meta,
&cwd,
&rollout_path,
state_db_ctx.as_deref(),
&mut state_builder,
default_provider.as_str(),
generate_memories,
)
.await?;
}
// Process rollout commands
while let Some(cmd) = rx.recv().await {
match cmd {
@@ -749,13 +904,8 @@ async fn rollout_writer(
continue;
}
if writer.is_none() {
buffered_items.extend(items);
continue;
}
write_and_reconcile_items(
writer.as_mut(),
&mut writer,
items.as_slice(),
&rollout_path,
state_db_ctx.as_deref(),
@@ -765,61 +915,14 @@ async fn rollout_writer(
.await?;
}
RolloutCmd::Persist { ack } => {
if writer.is_none() {
let result = async {
let Some(log_file_info) = deferred_log_file_info.take() else {
return Err(IoError::other(
"deferred rollout recorder missing log file metadata",
));
};
let file = open_log_file(log_file_info.path.as_path())?;
writer = Some(JsonlWriter {
file: tokio::fs::File::from_std(file),
});
if let Some(session_meta) = meta.take() {
write_session_meta(
writer.as_mut(),
session_meta,
&cwd,
&rollout_path,
state_db_ctx.as_deref(),
&mut state_builder,
default_provider.as_str(),
generate_memories,
)
.await?;
}
if !buffered_items.is_empty() {
write_and_reconcile_items(
writer.as_mut(),
buffered_items.as_slice(),
&rollout_path,
state_db_ctx.as_deref(),
state_builder.as_ref(),
default_provider.as_str(),
)
.await?;
buffered_items.clear();
}
Ok(())
}
.await;
if let Err(err) = result {
let _ = ack.send(());
return Err(err);
}
if let Err(e) = writer.file.flush().await {
let _ = ack.send(());
return Err(e);
}
let _ = ack.send(());
}
RolloutCmd::Flush { ack } => {
// Deferred fresh threads may not have an initialized file yet.
if let Some(writer) = writer.as_mut()
&& let Err(e) = writer.file.flush().await
{
if let Err(e) = writer.file.flush().await {
let _ = ack.send(());
return Err(e);
}
@@ -836,52 +939,43 @@ async fn rollout_writer(
#[allow(clippy::too_many_arguments)]
async fn write_session_meta(
mut writer: Option<&mut JsonlWriter>,
session_meta: SessionMeta,
cwd: &Path,
writer: &mut JsonlWriter,
session_meta_line: SessionMetaLine,
rollout_path: &Path,
state_db_ctx: Option<&StateRuntime>,
state_builder: &mut Option<ThreadMetadataBuilder>,
default_provider: &str,
generate_memories: bool,
) -> std::io::Result<()> {
let git_info = collect_git_info(cwd).await;
let session_meta_line = SessionMetaLine {
meta: session_meta,
git: git_info,
};
let memory_mode = session_meta_line.meta.memory_mode.clone();
if state_db_ctx.is_some() {
*state_builder = metadata::builder_from_session_meta(&session_meta_line, rollout_path);
}
let rollout_item = RolloutItem::SessionMeta(session_meta_line);
if let Some(writer) = writer.as_mut() {
writer.write_rollout_item(&rollout_item).await?;
}
sync_thread_state_after_write(
writer.write_rollout_item(&rollout_item).await?;
state_db::reconcile_rollout(
state_db_ctx,
rollout_path,
default_provider,
state_builder.as_ref(),
std::slice::from_ref(&rollout_item),
default_provider,
(!generate_memories).then_some("disabled"),
None,
memory_mode.as_deref(),
)
.await;
Ok(())
}
async fn write_and_reconcile_items(
mut writer: Option<&mut JsonlWriter>,
writer: &mut JsonlWriter,
items: &[RolloutItem],
rollout_path: &Path,
state_db_ctx: Option<&StateRuntime>,
state_builder: Option<&ThreadMetadataBuilder>,
default_provider: &str,
) -> std::io::Result<()> {
if let Some(writer) = writer.as_mut() {
for item in items {
writer.write_rollout_item(item).await?;
}
for item in items {
writer.write_rollout_item(item).await?;
}
sync_thread_state_after_write(
state_db_ctx,
@@ -1049,14 +1143,20 @@ async fn resume_candidate_matches_cwd(
return true;
}
if let Ok((items, _, _)) = RolloutRecorder::load_rollout_items(rollout_path).await
&& let Some(latest_turn_context_cwd) = items.iter().rev().find_map(|item| match item {
RolloutItem::TurnContext(turn_context) => Some(turn_context.cwd.as_path()),
RolloutItem::SessionMeta(_)
| RolloutItem::ResponseItem(_)
| RolloutItem::Compacted(_)
| RolloutItem::EventMsg(_) => None,
})
if let Ok(LoadedRolloutSource {
source,
thread_id: _,
parse_errors: _,
}) = RolloutStore::load_source(rollout_path).await
&& let Some(latest_turn_context_cwd) = source
.iter_reverse_from(source.exclusive_end_of_rollout_index())
.find_map(|(_, item)| match item {
RolloutItem::TurnContext(turn_context) => Some(turn_context.cwd.as_path()),
RolloutItem::SessionMeta(_)
| RolloutItem::ResponseItem(_)
| RolloutItem::Compacted(_)
| RolloutItem::EventMsg(_) => None,
})
{
return cwd_matches(latest_turn_context_cwd, cwd);
}
@@ -1157,16 +1257,16 @@ mod tests {
}
#[tokio::test]
async fn recorder_materializes_only_after_explicit_persist() -> std::io::Result<()> {
async fn store_creates_rollout_immediately() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
let config = ConfigBuilder::default()
.codex_home(home.path().to_path_buf())
.build()
.await?;
let thread_id = ThreadId::new();
let recorder = RolloutRecorder::new(
let store = RolloutStore::new(
&config,
RolloutRecorderParams::new(
RolloutStoreParams::new(
thread_id,
None,
SessionSource::Exec,
@@ -1179,13 +1279,13 @@ mod tests {
)
.await?;
let rollout_path = recorder.rollout_path().to_path_buf();
let rollout_path = store.rollout_path().to_path_buf();
assert!(
!rollout_path.exists(),
"rollout file should not exist before first user message"
rollout_path.exists(),
"rollout file should exist at startup"
);
recorder
store
.record_items(&[RolloutItem::EventMsg(EventMsg::AgentMessage(
AgentMessageEvent {
message: "buffered-event".to_string(),
@@ -1193,13 +1293,9 @@ mod tests {
},
))])
.await?;
recorder.flush().await?;
assert!(
!rollout_path.exists(),
"rollout file should remain deferred before first user message"
);
store.flush().await?;
recorder
store
.record_items(&[RolloutItem::EventMsg(EventMsg::UserMessage(
UserMessageEvent {
message: "first-user-message".to_string(),
@@ -1209,16 +1305,12 @@ mod tests {
},
))])
.await?;
recorder.flush().await?;
assert!(
!rollout_path.exists(),
"user-message-like items should not materialize without explicit persist"
);
store.flush().await?;
recorder.persist().await?;
// Second call verifies `persist()` is idempotent after materialization.
recorder.persist().await?;
assert!(rollout_path.exists(), "rollout file should be materialized");
store.persist().await?;
// Second call verifies `persist()` is idempotent after startup materialization.
store.persist().await?;
assert!(rollout_path.exists(), "rollout file should exist");
let text = std::fs::read_to_string(&rollout_path)?;
assert!(
@@ -1238,7 +1330,75 @@ mod tests {
let text_after_second_persist = std::fs::read_to_string(&rollout_path)?;
assert_eq!(text_after_second_persist, text);
recorder.shutdown().await?;
store.shutdown().await?;
Ok(())
}
#[tokio::test]
async fn store_keeps_source_order_consistent_with_queue_order() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
let config = ConfigBuilder::default()
.codex_home(home.path().to_path_buf())
.build()
.await?;
let store = Arc::new(
RolloutStore::new(
&config,
RolloutStoreParams::new(
ThreadId::new(),
None,
SessionSource::Exec,
BaseInstructions::default(),
Vec::new(),
EventPersistenceMode::Limited,
),
None,
None,
)
.await?,
);
let user_message_a = RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
message: "A".to_string(),
images: None,
local_images: Vec::new(),
text_elements: Vec::new(),
}));
let user_message_b = RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
message: "B".to_string(),
images: None,
local_images: Vec::new(),
text_elements: Vec::new(),
}));
let source_guard = store.source.lock().await;
let store_a = Arc::clone(&store);
let user_message_a_for_task = user_message_a.clone();
let task_a =
tokio::spawn(async move { store_a.record_items(&[user_message_a_for_task]).await });
tokio::task::yield_now().await;
let store_b = Arc::clone(&store);
let user_message_b_for_task = user_message_b.clone();
let task_b =
tokio::spawn(async move { store_b.record_items(&[user_message_b_for_task]).await });
tokio::task::yield_now().await;
drop(source_guard);
task_a.await.expect("join task A")?;
task_b.await.expect("join task B")?;
let actual_items = store.source.lock().await.clone().into_items();
assert!(matches!(
actual_items.first(),
Some(RolloutItem::SessionMeta(_))
));
let actual_tail = serde_json::to_value(&actual_items[1..])?;
let expected_tail = serde_json::to_value(vec![user_message_a, user_message_b])?;
assert_eq!(actual_tail, expected_tail);
store.shutdown().await?;
Ok(())
}
@@ -1264,9 +1424,9 @@ mod tests {
.expect("backfill should be complete");
let thread_id = ThreadId::new();
let recorder = RolloutRecorder::new(
let recorder = RolloutStore::new(
&config,
RolloutRecorderParams::new(
RolloutStoreParams::new(
thread_id,
None,
SessionSource::Cli,
@@ -1399,7 +1559,7 @@ mod tests {
write_session_file(home.path(), "2025-01-01T12-00-00", Uuid::from_u128(9003))?;
let default_provider = config.model_provider_id.clone();
let page1 = RolloutRecorder::list_threads(
let page1 = RolloutStore::list_threads(
&config,
1,
None,
@@ -1414,7 +1574,7 @@ mod tests {
assert_eq!(page1.items[0].path, newest);
let cursor = page1.next_cursor.clone().expect("cursor should be present");
let page2 = RolloutRecorder::list_threads(
let page2 = RolloutStore::list_threads(
&config,
1,
Some(&cursor),
@@ -1478,7 +1638,7 @@ mod tests {
.expect("state db upsert should succeed");
let default_provider = config.model_provider_id.clone();
let page = RolloutRecorder::list_threads(
let page = RolloutStore::list_threads(
&config,
10,
None,
@@ -1547,7 +1707,7 @@ mod tests {
.expect("state db upsert should succeed");
let default_provider = config.model_provider_id.clone();
let page = RolloutRecorder::list_threads(
let page = RolloutStore::list_threads(
&config,
1,
None,
codex-rs/core/src/rollout/tests.rs
+3 -3
View File
@@ -109,7 +109,7 @@ async fn insert_state_db_thread(
// ));
// insert_state_db_thread(home, db_thread_id, db_rollout_path.as_path(), false).await;
//
// let page = RolloutRecorder::list_threads(
// let page = RolloutStore::list_threads(
// home,
// 10,
// None,
@@ -160,7 +160,7 @@ async fn insert_state_db_thread(
// )
// .await;
//
// let page = RolloutRecorder::list_threads(
// let page = RolloutStore::list_threads(
// home,
// 10,
// None,
@@ -190,7 +190,7 @@ async fn insert_state_db_thread(
// )
// .unwrap();
//
// let page = RolloutRecorder::list_threads(
// let page = RolloutStore::list_threads(
// home,
// 10,
// None,
codex-rs/core/src/state/service.rs
+2 -2
View File
@@ -2,7 +2,7 @@ use std::collections::HashMap;
use std::sync::Arc;
use crate::AuthManager;
use crate::RolloutRecorder;
use crate::RolloutStore;
use crate::agent::AgentControl;
use crate::analytics_client::AnalyticsEventsClient;
use crate::client::ModelClient;
@@ -38,7 +38,7 @@ pub(crate) struct SessionServices {
pub(crate) main_execve_wrapper_exe: Option<PathBuf>,
pub(crate) analytics_events_client: AnalyticsEventsClient,
pub(crate) hooks: Hooks,
pub(crate) rollout: Mutex<Option<RolloutRecorder>>,
pub(crate) rollout: Mutex<Option<RolloutStore>>,
pub(crate) user_shell: Arc<crate::shell::Shell>,
pub(crate) shell_snapshot_tx: watch::Sender<Option<Arc<crate::shell_snapshot::ShellSnapshot>>>,
pub(crate) show_raw_agent_reasoning: bool,
codex-rs/core/src/tasks/review.rs
+3 -4
View File
@@ -265,8 +265,7 @@ pub(crate) async fn exit_review_mode(
)
.await;
// Review turns can run before any regular user turn, so explicitly
// materialize rollout persistence. Do this after emitting review output so
// file creation + git metadata collection cannot delay client-facing items.
session.ensure_rollout_materialized().await;
// Review turns record rollout items asynchronously, so explicitly flush
// before any follow-up path snapshots the thread's JSONL.
session.flush_rollout().await;
}
codex-rs/core/src/thread_manager.rs
+13 -7
View File
@@ -18,7 +18,7 @@ use crate::plugins::PluginsManager;
use crate::protocol::Event;
use crate::protocol::EventMsg;
use crate::protocol::SessionConfiguredEvent;
use crate::rollout::RolloutRecorder;
use crate::rollout::RolloutStore;
use crate::rollout::truncation;
use crate::shell_snapshot::ShellSnapshot;
use crate::skills::SkillsManager;
@@ -358,8 +358,10 @@ impl ThreadManager {
rollout_path: PathBuf,
auth_manager: Arc<AuthManager>,
) -> CodexResult<NewThread> {
let initial_history = RolloutRecorder::get_rollout_history(&rollout_path).await?;
Box::pin(self.resume_thread_with_history(config, initial_history, auth_manager, false))
// Thread startup still expects an eager `InitialHistory`, so the rollout store
// materializes one here before resume.
let initial_history = RolloutStore::get_rollout_history(&rollout_path).await?;
self.resume_thread_with_history(config, initial_history, auth_manager, false)
.await
}
@@ -409,7 +411,9 @@ impl ThreadManager {
path: PathBuf,
persist_extended_history: bool,
) -> CodexResult<NewThread> {
let history = RolloutRecorder::get_rollout_history(&path).await?;
// Fork truncation still operates on an owned `InitialHistory`, so the rollout
// store materializes the rollout before trimming it by user-turn boundary.
let history = RolloutStore::get_rollout_history(&path).await?;
let history = truncate_before_nth_user_message(history, nth_user_message);
Box::pin(self.state.spawn_thread(
config,
@@ -515,8 +519,10 @@ impl ThreadManagerState {
session_source: SessionSource,
inherited_shell_snapshot: Option<Arc<ShellSnapshot>>,
) -> CodexResult<NewThread> {
let initial_history = RolloutRecorder::get_rollout_history(&rollout_path).await?;
Box::pin(self.spawn_thread_with_source(
// This follows the same eager startup boundary as `resume_thread_from_rollout`:
// the rollout store materializes an owned `InitialHistory` before thread startup.
let initial_history = RolloutStore::get_rollout_history(&rollout_path).await?;
self.spawn_thread_with_source(
config,
initial_history,
Arc::clone(&self.auth_manager),
@@ -526,7 +532,7 @@ impl ThreadManagerState {
false,
None,
inherited_shell_snapshot,
))
)
.await
}
codex-rs/core/tests/suite/cli_stream.rs
+1 -1
View File
@@ -70,7 +70,7 @@ async fn responses_mode_stream_cli() {
// TODO(jif) fix
// // Verify a new session rollout was created and is discoverable via list_conversations
// let provider_filter = vec!["mock".to_string()];
// let page = RolloutRecorder::list_threads(
// let page = RolloutStore::list_threads(
// home.path(),
// 10,
// None,
codex-rs/core/tests/suite/rollout_list_find.rs
+4 -4
View File
@@ -5,8 +5,8 @@ use std::path::PathBuf;
use chrono::Utc;
use codex_core::EventPersistenceMode;
use codex_core::RolloutRecorder;
use codex_core::RolloutRecorderParams;
use codex_core::RolloutStore;
use codex_core::RolloutStoreParams;
use codex_core::config::ConfigBuilder;
use codex_core::find_archived_thread_path_by_id_str;
use codex_core::find_thread_path_by_id_str;
@@ -164,9 +164,9 @@ async fn find_locates_rollout_file_written_by_recorder() -> std::io::Result<()>
.await?;
let thread_id = ThreadId::new();
let thread_name = "named thread";
let recorder = RolloutRecorder::new(
let recorder = RolloutStore::new(
&config,
RolloutRecorderParams::new(
RolloutStoreParams::new(
thread_id,
None,
SessionSource::Exec,
codex-rs/core/tests/suite/sqlite_state.rs
+2 -14
View File
@@ -60,18 +60,10 @@ async fn new_thread_is_recorded_in_state_db() -> Result<()> {
let db = test.codex.state_db().expect("state db enabled");
assert!(
!rollout_path.exists(),
"fresh thread rollout should not be materialized before first user message"
rollout_path.exists(),
"fresh thread rollout should exist at startup"
);
let initial_metadata = db.get_thread(thread_id).await?;
assert!(
initial_metadata.is_none(),
"fresh thread should not be recorded in state db before first user message"
);
test.submit_turn("materialize rollout").await?;
let mut metadata = None;
for _ in 0..100 {
metadata = db.get_thread(thread_id).await?;
@@ -84,10 +76,6 @@ async fn new_thread_is_recorded_in_state_db() -> Result<()> {
let metadata = metadata.expect("thread should exist in state db");
assert_eq!(metadata.id, thread_id);
assert_eq!(metadata.rollout_path, rollout_path);
assert!(
rollout_path.exists(),
"rollout should be materialized after first user message"
);
Ok(())
}
codex-rs/exec/src/lib.rs
+1 -1
View File
@@ -1382,7 +1382,7 @@ async fn resolve_resume_path(
} else {
Some(config.cwd.as_path())
};
match codex_core::RolloutRecorder::find_latest_thread_path(
match codex_core::RolloutStore::find_latest_thread_path(
config,
1,
None,
codex-rs/protocol/src/protocol.rs
+4 -1
View File
@@ -3011,7 +3011,10 @@ pub struct SessionConfiguredEvent {
#[ts(optional)]
pub network_proxy: Option<SessionNetworkProxyRuntime>,
/// Path in which the rollout is stored. Can be `None` for ephemeral threads
/// Path reserved for this rollout. Can be `None` for ephemeral threads.
///
/// Fresh non-ephemeral sessions may advertise this path before the rollout file has been
/// materialized on disk.
#[serde(skip_serializing_if = "Option::is_none")]
pub rollout_path: Option<PathBuf>,
}
codex-rs/tui/src/lib.rs
+3 -3
View File
@@ -11,7 +11,7 @@ use codex_cloud_requirements::cloud_requirements_loader;
use codex_core::AuthManager;
use codex_core::CodexAuth;
use codex_core::INTERACTIVE_SESSION_SOURCES;
use codex_core::RolloutRecorder;
use codex_core::RolloutStore;
use codex_core::ThreadSortKey;
use codex_core::auth::AuthMode;
use codex_core::auth::enforce_login_restrictions;
@@ -687,7 +687,7 @@ async fn run_ratatui_app(
}
} else if cli.fork_last {
let provider_filter = vec![config.model_provider_id.clone()];
match RolloutRecorder::list_threads(
match RolloutStore::list_threads(
&config,
1,
None,
@@ -782,7 +782,7 @@ async fn run_ratatui_app(
} else {
Some(config.cwd.as_path())
};
match RolloutRecorder::find_latest_thread_path(
match RolloutStore::find_latest_thread_path(
&config,
1,
None,
codex-rs/tui/src/resume_picker.rs
+5 -5
View File
@@ -14,7 +14,7 @@ use chrono::DateTime;
use chrono::Utc;
use codex_core::Cursor;
use codex_core::INTERACTIVE_SESSION_SOURCES;
use codex_core::RolloutRecorder;
use codex_core::RolloutStore;
use codex_core::ThreadItem;
use codex_core::ThreadSortKey;
use codex_core::ThreadsPage;
@@ -111,7 +111,7 @@ enum BackgroundEvent {
/// between sorting by creation time and last-updated time using the Tab key.
///
/// Sessions are loaded on-demand via cursor-based pagination. The backend
/// `RolloutRecorder::list_threads` returns pages ordered by the selected sort key,
/// `RolloutStore::list_threads` returns pages ordered by the selected sort key,
/// and the picker deduplicates across pages to handle overlapping windows when
/// new sessions appear during pagination.
///
@@ -159,7 +159,7 @@ async fn run_session_picker(
let config = config.clone();
tokio::spawn(async move {
let provider_filter = vec![request.default_provider.clone()];
let page = RolloutRecorder::list_threads(
let page = RolloutStore::list_threads(
&config,
PAGE_SIZE,
request.cursor.as_ref(),
@@ -1456,7 +1456,7 @@ mod tests {
// set_rollout_mtime(&path_a, now - Duration::minutes(10));
// set_rollout_mtime(&path_b, now - Duration::seconds(10));
//
// let page = RolloutRecorder::list_threads(
// let page = RolloutStore::list_threads(
// tempdir.path(),
// PAGE_SIZE,
// None,
@@ -1765,7 +1765,7 @@ mod tests {
// SessionPickerAction::Resume,
// );
//
// let page = RolloutRecorder::list_threads(
// let page = RolloutStore::list_threads(
// &state.codex_home,
// PAGE_SIZE,
// None,