openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-14 11:57:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Simplify resumed rollout startup paths

Browse Source 
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Charles Cunningham
2026-03-07 15:34:51 -08:00
parent 34eb24848a
commit 50fe1fa89b
3 changed files with 27 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
codex-rs/core/src/codex.rs
View File

@@ -1948,44 +1948,26 @@ impl Session {
}
}
InitialHistory::Resumed(resumed_history) => {
let derive_from_rollout_source = |rollout_source: &InMemoryRolloutSource| {
(
self.reconstruct_history_from_rollout(&turn_context, rollout_source),
Self::extract_mcp_tool_selection_from_rollout_source(rollout_source),
Self::last_token_info_from_rollout_source(rollout_source),
)
};
let (reconstructed_rollout, restored_tool_selection, token_info) =
if resumed_history.history.is_empty() {
let rollout = self.services.rollout.lock().await;
match rollout.as_ref() {
Some(rollout) => {
let rollout_source = rollout.source.lock().await;
(
self.reconstruct_history_from_rollout(
&turn_context,
&rollout_source,
),
Self::extract_mcp_tool_selection_from_rollout_source(
&rollout_source,
),
Self::last_token_info_from_rollout_source(&rollout_source),
)
}
None => {
let rollout_source = InMemoryRolloutSource::new(Vec::new());
(
self.reconstruct_history_from_rollout(
&turn_context,
&rollout_source,
),
Self::extract_mcp_tool_selection_from_rollout_source(
&rollout_source,
),
Self::last_token_info_from_rollout_source(&rollout_source),
)
}
if let Some(rollout) = rollout.as_ref() {
let rollout_source = rollout.source.lock().await;
derive_from_rollout_source(&rollout_source)
} else {
derive_from_rollout_source(&InMemoryRolloutSource::new(Vec::new()))
}
} else {
let rollout_source = InMemoryRolloutSource::new(resumed_history.history);
(
self.reconstruct_history_from_rollout(&turn_context, &rollout_source),
Self::extract_mcp_tool_selection_from_rollout_source(&rollout_source),
Self::last_token_info_from_rollout_source(&rollout_source),
)
derive_from_rollout_source(&InMemoryRolloutSource::new(
resumed_history.history,
))
};
let previous_turn_settings = reconstructed_rollout.previous_turn_settings.clone();
self.set_previous_turn_settings(previous_turn_settings.clone())

14
codex-rs/core/src/rollout/metadata.rs
View File

@@ -100,20 +100,18 @@ pub(crate) async fn extract_metadata_from_rollout(
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();
let rollout_items = source
.iter_forward_from(rollout_start)
.map(|(_, item)| item)
.collect::<Vec<_>>();
let has_any_parsed_items = !rollout_items.is_empty();
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(
source
.iter_forward_from(rollout_start)
.map(|(_, item)| item),
rollout_path,
)
.ok_or_else(|| {
let builder = builder_from_items(rollout_items, rollout_path).ok_or_else(|| {
anyhow::anyhow!(
"rollout missing metadata builder: {}",
rollout_path.display()

6
codex-rs/core/src/rollout/recorder.rs
View File

@@ -700,6 +700,8 @@ impl RolloutStore {
/// Ensure all queued rollout writes have been persisted to disk.
pub async fn persist(&self) -> std::io::Result<()> {
let (tx, rx) = oneshot::channel();
// Control commands share the same ordering barrier as `record_items` so the writer sees
// them after any earlier in-memory source updates have been queued.
let queue_order_guard = self.queue_order.lock().await;
self.tx
.send(RolloutCmd::Persist { ack: tx })
@@ -713,6 +715,8 @@ impl RolloutStore {
/// Flush all queued writes and wait until they are committed by the writer task.
pub async fn flush(&self) -> std::io::Result<()> {
let (tx, rx) = oneshot::channel();
// `Flush` also participates in the ordering barrier so the writer drains any earlier
// `record_items` updates before acknowledging the flush.
let queue_order_guard = self.queue_order.lock().await;
self.tx
.send(RolloutCmd::Flush { ack: tx })
@@ -810,6 +814,8 @@ impl RolloutStore {
/// 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();
// `Shutdown` uses the same barrier so the writer observes it after any earlier in-memory
// source updates and queued control commands.
let queue_order_guard = self.queue_order.lock().await;
match self.tx.send(RolloutCmd::Shutdown { ack: tx_done }).await {
Ok(_) => {}