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

Rebase: resolve conflicts; use ConversationHistory entries; align fork_conversation API; update protocol and recorder; fix TUI app_backtrack; rustfmt

Browse Source
This commit is contained in:
Ahmed Ibrahim
2025-09-08 15:43:31 -07:00
parent 5f37b45158
commit c2c7778c38
5 changed files with 55 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/codex.rs
+2 -2
View File
@@ -204,8 +204,8 @@ impl Codex {
error!("Failed to create session: {e:#}");
CodexErr::InternalAgentDied
})?;
session
.record_initial_history(&turn_context, conversation_history)
let _ = session
.apply_initial_history(&turn_context, conversation_history)
.await;
let conversation_id = session.conversation_id;
codex-rs/core/src/conversation_manager.rs
+6 -9
View File
@@ -156,19 +156,16 @@ impl ConversationManager {
/// caller's `config`). The new conversation will have a fresh id.
pub async fn fork_conversation(
&self,
conversation_path: PathBuf,
conversation_id: Uuid,
conversation_history: Vec<ResponseItem>,
num_messages_to_drop: usize,
config: Config,
) -> CodexResult<NewConversation> {
// Compute the prefix up to the cut point.
let initial_history = RolloutRecorder::get_rollout_history(&conversation_path).await?;
let conversation_history = match initial_history {
InitialHistory::Resumed(items) => items,
InitialHistory::New => return Err(CodexErr::ConversationNotFound(conversation_id)),
};
let history =
truncate_after_dropping_last_messages(conversation_history, num_messages_to_drop);
let items: Vec<RolloutItem> = conversation_history
.into_iter()
.map(RolloutItem::from)
.collect();
let history = truncate_after_dropping_last_messages(items, num_messages_to_drop);
// Spawn a new conversation with the computed initial history.
let auth_manager = self.auth_manager.clone();
codex-rs/core/src/rollout/recorder.rs
+30 -54
View File
@@ -151,61 +151,37 @@ impl RolloutRecorder {
/// Attempt to create a new [`RolloutRecorder`]. If the sessions directory
/// cannot be created or the rollout file cannot be opened we return the
/// error so the caller can decide whether to disable persistence.
pub async fn new(config: &Config, params: RolloutRecorderParams) -> std::io::Result<Self> {
let (file, meta) = match params {
RolloutRecorderParams::Create {
conversation_id,
pub async fn new(
config: &Config,
conversation_id: ConversationId,
instructions: Option<String>,
) -> std::io::Result<Self> {
let LogFileInfo {
file,
conversation_id: session_id,
timestamp,
path,
} = create_log_file(config, conversation_id)?;
let timestamp_format: &[FormatItem] = format_description!(
"[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:3]Z"
);
let timestamp = timestamp
.to_offset(time::UtcOffset::UTC)
.format(timestamp_format)
.map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?;
(
tokio::fs::File::from_std(file),
Some(SessionMeta {
timestamp,
id: session_id,
cwd: config.cwd.to_string_lossy().to_string(),
originator: config.responses_originator_header.clone(),
cli_version: env!("CARGO_PKG_VERSION").to_string(),
instructions,
} => {
let LogFileInfo {
file,
conversation_id: session_id,
timestamp,
} = create_log_file(config, conversation_id)?;
let timestamp_format: &[FormatItem] = format_description!(
"[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:3]Z"
);
let timestamp = timestamp
.to_offset(time::UtcOffset::UTC)
.format(timestamp_format)
.map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?;
(
tokio::fs::File::from_std(file),
Some(SessionMeta {
timestamp,
id: session_id,
cwd: config.cwd.to_string_lossy().to_string(),
originator: config.responses_originator_header.clone(),
cli_version: env!("CARGO_PKG_VERSION").to_string(),
instructions,
}),
)
}
RolloutRecorderParams::Resume { path } => (
tokio::fs::OpenOptions::new()
.append(true)
.open(path)
.await?,
None,
),
};
// 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);
// 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, rx, meta, cwd));
Ok(Self { tx, path })
}),
)
}
pub(crate) async fn record_items(&self, item: RolloutItem) -> std::io::Result<()> {
codex-rs/core/tests/suite/fork_conversation.rs
+16 -53
View File
@@ -77,43 +77,13 @@ async fn fork_conversation_twice_drops_to_first_message() {
let base_history =
wait_for_event(&codex, |ev| matches!(ev, EventMsg::ConversationHistory(_))).await;
// Capture path/id from the base history and compute expected prefixes after each fork.
let (base_conv_id, base_path) = match &base_history {
EventMsg::ConversationHistory(ConversationPathResponseEvent {
conversation_id,
path,
}) => (*conversation_id, path.clone()),
// Capture entries from the base history and compute expected prefixes after each fork.
let entries_after_three: Vec<ResponseItem> = match &base_history {
EventMsg::ConversationHistory(ConversationPathResponseEvent { entries, .. }) => {
entries.clone()
}
_ => panic!("expected ConversationHistory event"),
};
// Read entries from rollout file.
async fn read_response_entries(path: &std::path::Path) -> Vec<ResponseItem> {
let text = tokio::fs::read_to_string(path).await.unwrap_or_default();
let mut out = Vec::new();
for line in text.lines() {
if line.trim().is_empty() {
continue;
}
if let Ok(item) = serde_json::from_str::<ResponseItem>(line) {
out.push(item);
}
}
out
}
async fn read_response_entries_with_retry(
path: &std::path::Path,
min_len: usize,
) -> Vec<ResponseItem> {
for _ in 0..50u32 {
let entries = read_response_entries(path).await;
if entries.len() >= min_len {
return entries;
}
tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
}
read_response_entries(path).await
}
let entries_after_three: Vec<ResponseItem> = read_response_entries(&base_path).await;
// History layout for this test:
// [0] user instructions,
// [1] environment context,
@@ -144,7 +114,7 @@ async fn fork_conversation_twice_drops_to_first_message() {
conversation: codex_fork1,
..
} = conversation_manager
.fork_conversation(base_path.clone(), base_conv_id, 1, config_for_fork.clone())
.fork_conversation(entries_after_three.clone(), 1, config_for_fork.clone())
.await
.expect("fork 1");
@@ -153,23 +123,20 @@ async fn fork_conversation_twice_drops_to_first_message() {
matches!(ev, EventMsg::ConversationHistory(_))
})
.await;
let (fork1_id, fork1_path) = match &fork1_history {
EventMsg::ConversationHistory(ConversationPathResponseEvent {
conversation_id,
path,
}) => (*conversation_id, path.clone()),
let entries_after_first_fork: Vec<ResponseItem> = match &fork1_history {
EventMsg::ConversationHistory(ConversationPathResponseEvent { entries, .. }) => {
assert_eq!(entries, &expected_after_first);
entries.clone()
}
_ => panic!("expected ConversationHistory event after first fork"),
};
let entries_after_first_fork: Vec<ResponseItem> =
read_response_entries_with_retry(&fork1_path, expected_after_first.len()).await;
assert_eq!(entries_after_first_fork, expected_after_first);
// Fork again with n=1 → drops the (new) last user message, leaving only the first.
let NewConversation {
conversation: codex_fork2,
..
} = conversation_manager
.fork_conversation(fork1_path.clone(), fork1_id, 1, config_for_fork.clone())
.fork_conversation(entries_after_first_fork.clone(), 1, config_for_fork.clone())
.await
.expect("fork 2");
@@ -178,14 +145,10 @@ async fn fork_conversation_twice_drops_to_first_message() {
matches!(ev, EventMsg::ConversationHistory(_))
})
.await;
let (_fork2_id, fork2_path) = match &fork2_history {
EventMsg::ConversationHistory(ConversationPathResponseEvent {
conversation_id,
path,
}) => (*conversation_id, path.clone()),
match &fork2_history {
EventMsg::ConversationHistory(ConversationPathResponseEvent { entries, .. }) => {
assert_eq!(entries, &expected_after_second);
}
_ => panic!("expected ConversationHistory event after second fork"),
};
let entries_after_second_fork: Vec<ResponseItem> =
read_response_entries_with_retry(&fork2_path, expected_after_second.len()).await;
assert_eq!(entries_after_second_fork, expected_after_second);
}
codex-rs/tui/src/app_backtrack.rs
+1 -1
View File
@@ -302,7 +302,7 @@ impl App {
/// Thin wrapper around ConversationManager::fork_conversation.
async fn perform_fork(
&self,
conversation_history: Vec<codex_core::protocol::ResponseItem>,
conversation_history: Vec<codex_protocol::models::ResponseItem>,
drop_count: usize,
cfg: codex_core::config::Config,
) -> codex_core::error::Result<codex_core::NewConversation> {