hanzoai/dev
2
0
Fork 0
mirror of https://github.com/hanzoai/dev.git synced 2026-10-25 03:32:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(resume): restore full history replay

Browse Source
This commit is contained in:
James Peter
2025-09-18 11:56:22 +10:00
parent 736293a90a
commit 6d1bfdd0b2
13 changed files with 720 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/codex.rs
+60 -11
View File
@@ -480,6 +480,7 @@ use crate::protocol::InputItem;
use crate::protocol::Op;
use crate::protocol::PatchApplyBeginEvent;
use crate::protocol::PatchApplyEndEvent;
use crate::protocol::RecordedEvent;
use crate::protocol::ReviewDecision;
use crate::protocol::SandboxPolicy;
use crate::protocol::SessionConfiguredEvent;
@@ -821,8 +822,39 @@ impl Session {
}
}
/// Persist an event into the rollout log if appropriate.
fn persist_event(&self, event: &Event) {
if !crate::rollout::policy::should_persist_event_msg(&event.msg) {
return;
}
let Some(msg) = crate::protocol::event_msg_to_protocol(&event.msg) else {
return;
};
let recorder = {
let guard = self.rollout.lock().unwrap();
guard.as_ref().cloned()
};
if let Some(rec) = recorder {
let order = event
.order
.as_ref()
.map(crate::protocol::order_meta_to_protocol);
let protocol_event = codex_protocol::protocol::RecordedEvent {
id: event.id.clone(),
event_seq: event.event_seq,
order,
msg,
};
tokio::spawn(async move {
if let Err(e) = rec.record_events(&[protocol_event]).await {
warn!("failed to persist rollout event: {e}");
}
});
}
}
/// Create a stamped Event with a per-turn sequence number.
pub(crate) fn make_event(&self, sub_id: &str, msg: EventMsg) -> Event {
fn stamp_event(&self, sub_id: &str, msg: EventMsg) -> Event {
let mut state = self.state.lock().unwrap();
let seq = match msg {
EventMsg::TaskStarted => {
@@ -845,7 +877,18 @@ impl Session {
*e
}
};
Event { id: sub_id.to_string(), event_seq: seq, msg, order: None }
Event {
id: sub_id.to_string(),
event_seq: seq,
msg,
order: None,
}
}
pub(crate) fn make_event(&self, sub_id: &str, msg: EventMsg) -> Event {
let event = self.stamp_event(sub_id, msg);
self.persist_event(&event);
event
}
/// Same as make_event but allows supplying a provider sequence_number
@@ -854,10 +897,9 @@ impl Session {
/// and local to our runtime. Provider ordering is carried via
/// `OrderMeta` when applicable.
fn make_event_with_hint(&self, sub_id: &str, msg: EventMsg, _seq_hint: Option<u64>) -> Event {
// Preserve the monotonic invariant of event_seq by delegating to make_event.
// Any ordering hints from the provider should be conveyed through
// OrderMeta (see make_event_with_order) rather than event_seq.
self.make_event(sub_id, msg)
let event = self.stamp_event(sub_id, msg);
self.persist_event(&event);
event
}
fn make_event_with_order(
@@ -865,10 +907,11 @@ impl Session {
sub_id: &str,
msg: EventMsg,
order: crate::protocol::OrderMeta,
seq_hint: Option<u64>,
_seq_hint: Option<u64>,
) -> Event {
let mut ev = self.make_event_with_hint(sub_id, msg, seq_hint);
let mut ev = self.stamp_event(sub_id, msg);
ev.order = Some(order);
self.persist_event(&ev);
ev
}
@@ -1102,7 +1145,7 @@ impl Session {
if let Err(e) = rec.record_state(snapshot).await {
error!("failed to record rollout state: {e:#}");
}
if let Err(e) = rec.record_items(items).await {
if let Err(e) = rec.record_response_items(items).await {
error!("failed to record rollout items: {e:#}");
}
}
@@ -1720,6 +1763,7 @@ async fn submission_loop(
// Optionally resume an existing rollout.
let mut restored_items: Option<Vec<ResponseItem>> = None;
let mut restored_events: Option<Vec<RecordedEvent>> = None;
let rollout_recorder: Option<RolloutRecorder> =
if let Some(path) = resume_path.as_ref() {
match RolloutRecorder::resume(&config, path).await {
@@ -1728,6 +1772,9 @@ async fn submission_loop(
if !saved.items.is_empty() {
restored_items = Some(saved.items);
}
if !saved.events.is_empty() {
restored_events = Some(saved.events);
}
Some(rec)
}
Err(e) => {
@@ -1896,10 +1943,12 @@ async fn submission_loop(
}
}
// If we resumed from a rollout, replay the prior transcript into the UI.
if let Some(items) = restored_items {
if restored_items.is_some() || restored_events.is_some() {
let items = restored_items.unwrap_or_default();
let events = restored_events.unwrap_or_default();
let event = sess_arc.make_event(
&sub.id,
EventMsg::ReplayHistory(crate::protocol::ReplayHistoryEvent { items }),
EventMsg::ReplayHistory(crate::protocol::ReplayHistoryEvent { items, events }),
);
if let Err(e) = tx_event.send(event).await {
warn!("failed to send ReplayHistory event: {e}");
codex-rs/core/src/conversation_manager.rs
+2 -2
View File
@@ -183,9 +183,9 @@ impl ConversationManager {
.collect();
if !response_items.is_empty() {
recorder
.record_items(&response_items)
.record_response_items(&response_items)
.await
.map_err(|e| CodexErr::Io(e))?;
.map_err(CodexErr::Io)?;
}
// Ensure data is flushed to disk before resuming.
recorder.shutdown().await.map_err(|e| CodexErr::Io(e))?;
codex-rs/core/src/lib.rs
+1
View File
@@ -72,6 +72,7 @@ pub mod terminal;
mod tool_apply_patch;
pub mod turn_diff_tracker;
pub use rollout::list::ConversationsPage;
pub use rollout::list::Cursor;
pub use rollout::list::find_conversation_path_by_id_str;
pub use rollout::recorder::RolloutRecorder;
mod user_notification;
codex-rs/core/src/protocol.rs
+93
View File
@@ -11,6 +11,7 @@ use std::str::FromStr;
use std::time::Duration;
use mcp_types::CallToolResult;
use serde::de::DeserializeOwned;
use serde::Deserialize;
use serde::Serialize;
use serde_bytes::ByteBuf;
@@ -391,6 +392,95 @@ pub struct Event {
pub order: Option<OrderMeta>,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct RecordedEvent {
pub id: String,
pub event_seq: u64,
pub order: Option<OrderMeta>,
pub msg: EventMsg,
}
pub fn event_msg_to_protocol(msg: &EventMsg) -> Option<codex_protocol::protocol::EventMsg> {
if matches!(msg, EventMsg::ReplayHistory(_)) {
return None;
}
convert_value(msg)
}
pub fn event_msg_from_protocol(msg: &codex_protocol::protocol::EventMsg) -> Option<EventMsg> {
let Some(converted) = convert_value(msg) else {
return None;
};
if matches!(converted, EventMsg::ReplayHistory(_)) {
return None;
}
Some(converted)
}
pub fn order_meta_to_protocol(
order: &OrderMeta,
) -> codex_protocol::protocol::OrderMeta {
codex_protocol::protocol::OrderMeta {
request_ordinal: order.request_ordinal,
output_index: order.output_index,
sequence_number: order.sequence_number,
}
}
pub fn order_meta_from_protocol(
order: &codex_protocol::protocol::OrderMeta,
) -> OrderMeta {
OrderMeta {
request_ordinal: order.request_ordinal,
output_index: order.output_index,
sequence_number: order.sequence_number,
}
}
pub fn recorded_event_to_protocol(
event: &RecordedEvent,
) -> Option<codex_protocol::protocol::RecordedEvent> {
let msg = event_msg_to_protocol(&event.msg)?;
let order = event
.order
.as_ref()
.map(order_meta_to_protocol);
Some(codex_protocol::protocol::RecordedEvent {
id: event.id.clone(),
event_seq: event.event_seq,
order,
msg,
})
}
pub fn recorded_event_from_protocol(
src: codex_protocol::protocol::RecordedEvent,
) -> Option<RecordedEvent> {
let msg = event_msg_from_protocol(&src.msg)?;
let order = src.order.as_ref().map(order_meta_from_protocol);
Some(RecordedEvent {
id: src.id,
event_seq: src.event_seq,
order,
msg,
})
}
fn convert_value<T, U>(value: &T) -> Option<U>
where
T: Serialize,
U: DeserializeOwned,
{
let Ok(json) = serde_json::to_value(value) else {
return None;
};
serde_json::from_value(json).ok()
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct OrderMeta {
/// 1-based ordinal of this request/turn in the session
@@ -575,6 +665,9 @@ pub struct ReplayHistoryEvent {
/// Items to render in order. Front-ends should render these as static
/// history without triggering any tool execution.
pub items: Vec<codex_protocol::models::ResponseItem>,
/// Previously emitted events to replay for UI restoration.
#[serde(default)]
pub events: Vec<RecordedEvent>,
}
impl From<TokenUsage> for FinalOutput {
codex-rs/core/src/rollout/list.rs
+6 -4
View File
@@ -15,7 +15,6 @@ use time::macros::format_description;
use uuid::Uuid;
use super::SESSIONS_SUBDIR;
use codex_protocol::protocol::EventMsg as ProtocolEventMsg;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
@@ -323,9 +322,11 @@ async fn read_head_and_flags(
head.push(val);
}
}
RolloutItem::EventMsg(ev) => {
if matches!(ev, ProtocolEventMsg::AgentMessage(_)) {
saw_user_event = true;
RolloutItem::Event(event) => {
if let Some(msg) = crate::protocol::event_msg_from_protocol(&event.msg) {
if matches!(msg, crate::protocol::EventMsg::AgentMessage(_)) {
saw_user_event = true;
}
}
}
// Skip variants not displayed in list summaries.
@@ -336,6 +337,7 @@ async fn read_head_and_flags(
Ok((head, saw_session_meta, saw_user_event))
}
/// Locate a recorded conversation rollout file by its UUID string using the existing
/// paginated listing implementation. Returns `Ok(Some(path))` if found, `Ok(None)` if not present
/// or the id is invalid.
codex-rs/core/src/rollout/policy.rs
+10 -48
View File
@@ -1,16 +1,13 @@
use crate::protocol::EventMsg;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::RolloutItem;
/// Whether a rollout `item` should be persisted in rollout files.
#[inline]
pub(crate) fn is_persisted_response_item(item: &RolloutItem) -> bool {
pub(crate) fn should_persist_rollout_item(item: &RolloutItem) -> bool {
match item {
RolloutItem::ResponseItem(item) => should_persist_response_item(item),
RolloutItem::EventMsg(_ev) => {
// We do not persist protocol EventMsg entries in this fork.
false
}
RolloutItem::Event(_) => true,
// Always persist session meta
RolloutItem::SessionMeta(_) => true,
// Do not persist variants not used by this fork.
@@ -33,48 +30,13 @@ pub(crate) fn should_persist_response_item(item: &ResponseItem) -> bool {
}
}
/// Whether an `EventMsg` should be persisted in rollout files.
/// Whether an [`EventMsg`] should be persisted.
#[inline]
#[allow(dead_code)]
pub(crate) fn should_persist_event_msg(ev: &EventMsg) -> bool {
match ev {
EventMsg::UserMessage(_)
| EventMsg::AgentMessage(_)
| EventMsg::AgentReasoning(_)
| EventMsg::AgentReasoningRawContent(_)
| EventMsg::TokenCount(_)
| EventMsg::EnteredReviewMode(_)
| EventMsg::ExitedReviewMode(_)
| EventMsg::TurnAborted(_) => true,
EventMsg::Error(_)
| EventMsg::TaskStarted
| EventMsg::TaskComplete(_)
| EventMsg::AgentMessageDelta(_)
| EventMsg::AgentReasoningDelta(_)
| EventMsg::AgentReasoningRawContentDelta(_)
| EventMsg::AgentReasoningSectionBreak(_)
| EventMsg::SessionConfigured(_)
| EventMsg::McpToolCallBegin(_)
| EventMsg::McpToolCallEnd(_)
| EventMsg::WebSearchBegin(_)
| EventMsg::WebSearchComplete(_)
| EventMsg::CustomToolCallBegin(_)
| EventMsg::CustomToolCallEnd(_)
| EventMsg::ExecCommandBegin(_)
| EventMsg::ExecCommandOutputDelta(_)
| EventMsg::ExecCommandEnd(_)
| EventMsg::ExecApprovalRequest(_)
| EventMsg::ApplyPatchApprovalRequest(_)
| EventMsg::BackgroundEvent(_)
| EventMsg::PatchApplyBegin(_)
| EventMsg::PatchApplyEnd(_)
| EventMsg::TurnDiff(_)
| EventMsg::GetHistoryEntryResponse(_)
| EventMsg::PlanUpdate(_)
| EventMsg::BrowserScreenshotUpdate(_)
| EventMsg::AgentStatusUpdate(_)
| EventMsg::ShutdownComplete
| EventMsg::ConversationPath(_)
| EventMsg::ReplayHistory(_) => false,
}
!matches!(
ev,
EventMsg::AgentMessageDelta(_)
| EventMsg::AgentReasoningDelta(_)
| EventMsg::AgentReasoningRawContentDelta(_)
)
}
codex-rs/core/src/rollout/recorder.rs
+48 -16
View File
@@ -24,7 +24,7 @@ use super::SESSIONS_SUBDIR;
use super::list::ConversationsPage;
use super::list::Cursor;
use super::list::get_conversations;
use super::policy::is_persisted_response_item;
use super::policy::{should_persist_response_item, should_persist_rollout_item};
use crate::config::Config;
use crate::default_client::DEFAULT_ORIGINATOR;
use crate::git_info::collect_git_info;
@@ -45,6 +45,8 @@ pub struct SavedSession {
#[serde(default)]
pub items: Vec<ResponseItem>,
#[serde(default)]
pub events: Vec<crate::protocol::RecordedEvent>,
#[serde(default)]
pub state: SessionStateSnapshot,
pub session_id: uuid::Uuid,
}
@@ -166,13 +168,13 @@ impl RolloutRecorder {
Ok(Self { tx, rollout_path })
}
pub(crate) async fn record_items(&self, items: &[ResponseItem]) -> std::io::Result<()> {
pub(crate) async fn record_response_items(
&self,
items: &[ResponseItem],
) -> std::io::Result<()> {
let mut filtered: Vec<RolloutItem> = Vec::new();
for item in items {
// Note that function calls may look a bit strange if they are
// "fully qualified MCP tool calls," so we could consider
// reformatting them in that case.
if super::policy::should_persist_response_item(item) {
if should_persist_response_item(item) {
filtered.push(RolloutItem::ResponseItem(item.clone()));
}
}
@@ -185,6 +187,24 @@ impl RolloutRecorder {
.map_err(|e| IoError::other(format!("failed to queue rollout items: {e}")))
}
pub(crate) async fn record_events(
&self,
events: &[codex_protocol::protocol::RecordedEvent],
) -> std::io::Result<()> {
if events.is_empty() {
return Ok(());
}
let filtered = events
.iter()
.cloned()
.map(RolloutItem::Event)
.collect::<Vec<_>>();
self.tx
.send(RolloutCmd::AddItems(filtered))
.await
.map_err(|e| IoError::other(format!("failed to queue rollout events: {e}")))
}
/// No-op compatibility shim for older APIs expecting a state snapshot.
pub async fn record_state(&self, _snapshot: SessionStateSnapshot) -> std::io::Result<()> {
Ok(())
@@ -194,17 +214,29 @@ impl RolloutRecorder {
pub async fn resume(config: &Config, path: &Path) -> std::io::Result<(Self, SavedSession)> {
let recorder = Self::new(config, RolloutRecorderParams::Resume { path: path.to_path_buf() }).await?;
let history = Self::get_rollout_history(path).await?;
let (session_id, items) = match history {
InitialHistory::Resumed(resumed) => (resumed.conversation_id.0, resumed
.history
.into_iter()
.filter_map(|ri| match ri { RolloutItem::ResponseItem(it) => Some(it), _ => None })
.collect::<Vec<ResponseItem>>()),
_ => (uuid::Uuid::new_v4(), Vec::new()),
let (session_id, items, events) = match history {
InitialHistory::Resumed(resumed) => {
let mut responses = Vec::new();
let mut recorded_events = Vec::new();
for entry in resumed.history {
match entry {
RolloutItem::ResponseItem(it) => responses.push(it),
RolloutItem::Event(ev) => {
if let Some(rec) = crate::protocol::recorded_event_from_protocol(ev) {
recorded_events.push(rec);
}
}
_ => {}
}
}
(resumed.conversation_id.0, responses, recorded_events)
}
_ => (uuid::Uuid::new_v4(), Vec::new(), Vec::new()),
};
let saved = SavedSession {
session: SessionMeta::default(),
items,
events,
state: SessionStateSnapshot::default(),
session_id,
};
@@ -246,8 +278,8 @@ impl RolloutRecorder {
RolloutItem::ResponseItem(item) => {
items.push(RolloutItem::ResponseItem(item));
}
RolloutItem::EventMsg(_ev) => {
items.push(RolloutItem::EventMsg(_ev));
RolloutItem::Event(ev) => {
items.push(RolloutItem::Event(ev));
}
// Ignore variants not used by this fork when resuming.
RolloutItem::Compacted(_)
@@ -376,7 +408,7 @@ async fn rollout_writer(
match cmd {
RolloutCmd::AddItems(items) => {
for item in items {
if is_persisted_response_item(&item) {
if should_persist_rollout_item(&item) {
writer.write_rollout_item(item).await?;
}
}
codex-rs/core/tests/suite/mod.rs
+1
View File
@@ -12,6 +12,7 @@ mod model_overrides;
mod prompt_caching;
mod review;
mod rollout_list_find;
mod rollout_resume;
mod seatbelt;
mod stream_error_allows_next_turn;
mod stream_no_completed;
codex-rs/core/tests/suite/rollout_resume.rs
+62
View File
@@ -0,0 +1,62 @@
use codex_core::protocol::{event_msg_to_protocol, AgentMessageEvent, EventMsg, RecordedEvent};
use codex_core::rollout::RolloutRecorder;
use codex_core::rollout::recorder::RolloutRecorderParams;
use codex_protocol::mcp_protocol::ConversationId;
use codex_protocol::models::{ContentItem, ResponseItem};
use tempfile::TempDir;
use uuid::Uuid;
use crate::common::load_default_config_for_test;
#[tokio::test]
async fn resume_restores_recorded_events() {
let codex_home = TempDir::new().expect("tempdir");
let mut config = load_default_config_for_test(&codex_home);
config.cwd = codex_home.path().to_path_buf();
let conversation_id = ConversationId(Uuid::new_v4());
let recorder = RolloutRecorder::new(&config, RolloutRecorderParams::new(conversation_id, None))
.await
.expect("create recorder");
let response_item = ResponseItem::Message {
id: None,
role: "assistant".to_string(),
content: vec![ContentItem::OutputText {
text: "hello from resume".to_string(),
}],
};
recorder
.record_response_items(std::slice::from_ref(&response_item))
.await
.expect("record response");
let recorded_event = RecordedEvent {
id: "turn-1".to_string(),
event_seq: 1,
order: None,
msg: EventMsg::AgentMessage(AgentMessageEvent {
message: "assistant says hi".to_string(),
}),
};
let protocol_event = codex_protocol::protocol::RecordedEvent {
id: recorded_event.id.clone(),
event_seq: recorded_event.event_seq,
order: None,
msg: event_msg_to_protocol(&recorded_event.msg).expect("convert event"),
};
recorder
.record_events(std::slice::from_ref(&protocol_event))
.await
.expect("record event");
let rollout_path = recorder.rollout_path.clone();
recorder.shutdown().await.expect("shutdown recorder");
let (_rec, saved) = RolloutRecorder::resume(&config, &rollout_path)
.await
.expect("resume recorder");
assert_eq!(saved.items, vec![response_item]);
assert_eq!(saved.events, vec![recorded_event]);
}
codex-rs/protocol/src/protocol.rs
+42 -4
View File
@@ -405,12 +405,40 @@ pub enum InputItem {
}
/// Event Queue Entry - events from agent
#[derive(Debug, Clone, Deserialize, Serialize)]
#[derive(Debug, Clone, Deserialize, Serialize, TS)]
pub struct Event {
/// Submission `id` that this event is correlated with.
pub id: String,
/// Monotonic, per-turn sequence for ordering within a submission id.
/// Resets to 0 at TaskStarted and increments for each subsequent event.
pub event_seq: u64,
/// Payload
pub msg: EventMsg,
/// Optional model-provided ordering metadata (when applicable)
#[serde(skip_serializing_if = "Option::is_none")]
pub order: Option<OrderMeta>,
}
/// Lightweight representation of an event suitable for persistence and replay.
#[derive(Debug, Clone, Deserialize, Serialize, TS)]
pub struct RecordedEvent {
pub id: String,
pub event_seq: u64,
#[serde(skip_serializing_if = "Option::is_none")]
pub order: Option<OrderMeta>,
pub msg: EventMsg,
}
#[derive(Debug, Clone, Deserialize, Serialize, TS)]
pub struct OrderMeta {
/// 1-based ordinal of this request/turn in the session
pub request_ordinal: u64,
/// Model-provided output_index for the top-level item
#[serde(skip_serializing_if = "Option::is_none")]
pub output_index: Option<u32>,
/// Model-provided sequence_number within the output_index stream
#[serde(skip_serializing_if = "Option::is_none")]
pub sequence_number: Option<u64>,
}
/// Response event from the agent
@@ -591,6 +619,14 @@ pub struct TokenCountEvent {
pub info: Option<TokenUsageInfo>,
}
/// Payload for `ReplayHistory` containing prior `ResponseItem`s and optional events.
#[derive(Debug, Clone, Deserialize, Serialize, TS)]
pub struct ReplayHistoryEvent {
pub items: Vec<crate::models::ResponseItem>,
#[serde(default)]
pub events: Vec<RecordedEvent>,
}
// Includes prompts, tools and space to call compact.
const BASELINE_TOKENS: u64 = 12000;
@@ -878,7 +914,7 @@ impl InitialHistory {
.history
.iter()
.filter_map(|ri| match ri {
RolloutItem::EventMsg(ev) => Some(ev.clone()),
RolloutItem::Event(ev) => Some(ev.msg.clone()),
_ => None,
})
.collect(),
@@ -887,7 +923,7 @@ impl InitialHistory {
items
.iter()
.filter_map(|ri| match ri {
RolloutItem::EventMsg(ev) => Some(ev.clone()),
RolloutItem::Event(ev) => Some(ev.msg.clone()),
_ => None,
})
.collect(),
@@ -921,7 +957,7 @@ pub enum RolloutItem {
ResponseItem(ResponseItem),
Compacted(CompactedItem),
TurnContext(TurnContextItem),
EventMsg(EventMsg),
Event(RecordedEvent),
}
#[derive(Serialize, Deserialize, Clone, Debug, TS)]
@@ -1256,6 +1292,7 @@ mod tests {
let rollout_file = NamedTempFile::new().unwrap();
let event = Event {
id: "1234".to_string(),
event_seq: 1,
msg: EventMsg::SessionConfigured(SessionConfiguredEvent {
session_id: conversation_id,
model: "codex-mini-latest".to_string(),
@@ -1265,6 +1302,7 @@ mod tests {
initial_messages: None,
rollout_path: rollout_file.path().to_path_buf(),
}),
order: None,
};
let expected = json!({
codex-rs/tui/src/app.rs
+4 -1
View File
@@ -1279,7 +1279,10 @@ impl App<'_> {
id: "fork".to_string(),
event_seq: 0,
msg: codex_core::protocol::EventMsg::ReplayHistory(
codex_core::protocol::ReplayHistoryEvent { items: prefix_items }
codex_core::protocol::ReplayHistoryEvent {
items: prefix_items,
events: Vec::new(),
}
),
order: None,
};
codex-rs/tui/src/chatwidget.rs
+130 -43
View File
@@ -78,6 +78,7 @@ use ratatui::widgets::WidgetRef;
use ratatui_image::picker::Picker;
use std::cell::RefCell;
use tokio::sync::mpsc::UnboundedSender;
use serde_json::Value as JsonValue;
use tokio::sync::mpsc::unbounded_channel;
use tracing::info;
// use image::GenericImageView;
@@ -1066,37 +1067,30 @@ impl ChatWidget<'_> {
_ => {}
}
}
// Show internal system status messages (rendered with markdown) so
// code blocks and formatting are consistent with assistant output.
if text.contains("== System Status ==") {
use ratatui::text::Line as RLine;
let mut lines: Vec<RLine<'static>> = Vec::new();
crate::markdown::append_markdown(&text, &mut lines, &self.config);
let key = self.next_internal_key();
let _ = self.history_insert_with_key_global(
Box::new(crate::history_cell::PlainHistoryCell {
lines,
kind: crate::history_cell::HistoryCellType::Notice,
}),
key,
);
let text = text.trim();
if text.is_empty() {
return;
}
if text.starts_with("== System Status ==") {
return;
}
if role == "assistant" {
let mut lines: Vec<ratatui::text::Line<'static>> = Vec::new();
crate::markdown::append_markdown(text, &mut lines, &self.config);
self.insert_final_answer_with_id(None, lines, text.to_string());
return;
}
if role == "user" {
let key = self.next_internal_key();
let _ = self.history_insert_with_key_global(
Box::new(crate::history_cell::new_user_prompt(text)),
Box::new(crate::history_cell::new_user_prompt(text.to_string())),
key,
);
} else {
// Build a PlainHistoryCell with Assistant kind; header line hidden by renderer
use crate::history_cell::HistoryCellType;
use crate::history_cell::PlainHistoryCell;
let mut lines = Vec::new();
lines.push(ratatui::text::Line::from("assistant"));
for l in text.lines() {
lines.push(ratatui::text::Line::from(l.to_string()));
}
crate::markdown::append_markdown(text, &mut lines, &self.config);
let key = self.next_internal_key();
let _ = self.history_insert_with_key_global(
Box::new(PlainHistoryCell {
@@ -1107,6 +1101,24 @@ impl ChatWidget<'_> {
);
}
}
ResponseItem::FunctionCall { name, arguments, call_id, .. } => {
let pretty_args = serde_json::from_str::<JsonValue>(&arguments)
.and_then(|v| serde_json::to_string_pretty(&v))
.unwrap_or_else(|_| arguments.clone());
let mut message = format!("🔧 Tool call: {}", name);
if !pretty_args.trim().is_empty() {
message.push_str("\n");
message.push_str(&pretty_args);
}
if !call_id.is_empty() {
message.push_str(&format!("\ncall_id: {}", call_id));
}
let key = self.next_internal_key();
let _ = self.history_insert_with_key_global(
Box::new(crate::history_cell::new_background_event(message)),
key,
);
}
ResponseItem::Reasoning { summary, .. } => {
for s in summary {
let codex_protocol::models::ReasoningItemReasoningSummary::SummaryText { text } =
@@ -1122,17 +1134,50 @@ impl ChatWidget<'_> {
.finalize(crate::streaming::StreamKind::Reasoning, true, &sink);
}
}
ResponseItem::FunctionCallOutput { output, .. } => {
// Try to unwrap common JSON wrapper {"output": "...", ...}
let mut content = output.content;
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&content) {
ResponseItem::FunctionCallOutput { output, call_id, .. } => {
let mut content = output.content.clone();
let mut metadata_summary = String::new();
if let Ok(v) = serde_json::from_str::<JsonValue>(&content) {
if let Some(s) = v.get("output").and_then(|x| x.as_str()) {
content = s.to_string();
}
if let Some(meta) = v.get("metadata").and_then(|m| m.as_object()) {
let mut parts = Vec::new();
if let Some(code) = meta.get("exit_code").and_then(|x| x.as_i64()) {
parts.push(format!("exit_code={}", code));
}
if let Some(duration) =
meta.get("duration_seconds").and_then(|x| x.as_f64())
{
parts.push(format!("duration={:.2}s", duration));
}
if !parts.is_empty() {
metadata_summary = parts.join(", ");
}
}
}
let mut message = String::new();
if !content.trim().is_empty() {
message.push_str(content.trim_end());
}
if !metadata_summary.is_empty() {
if !message.is_empty() {
message.push_str("\n\n");
}
message.push_str(&format!("({})", metadata_summary));
}
if !call_id.is_empty() {
if !message.is_empty() {
message.push_str("\n");
}
message.push_str(&format!("call_id: {}", call_id));
}
if message.trim().is_empty() {
return;
}
let key = self.next_internal_key();
let _ = self.history_insert_with_key_global(
Box::new(crate::history_cell::new_background_event(content)),
Box::new(crate::history_cell::new_background_event(message)),
key,
);
}
@@ -1702,15 +1747,25 @@ impl ChatWidget<'_> {
let config_for_agent_loop = config.clone();
tokio::spawn(async move {
// Use ConversationManager with an AuthManager (API key by default)
let conversation_manager = ConversationManager::new(AuthManager::shared(
let auth_manager = AuthManager::shared(
config_for_agent_loop.codex_home.clone(),
AuthMode::ApiKey,
config_for_agent_loop.responses_originator_header.clone(),
));
let new_conversation = match conversation_manager
.new_conversation(config_for_agent_loop)
.await
{
);
let conversation_manager = ConversationManager::new(auth_manager.clone());
let resume_path = config_for_agent_loop.experimental_resume.clone();
let new_conversation = match resume_path {
Some(path) => conversation_manager
.resume_conversation_from_rollout(
config_for_agent_loop,
path,
auth_manager,
)
.await,
None => conversation_manager.new_conversation(config_for_agent_loop).await,
};
let new_conversation = match new_conversation {
Ok(conv) => conv,
Err(e) => {
tracing::error!("failed to initialize conversation: {e}");
@@ -1891,16 +1946,20 @@ impl ChatWidget<'_> {
// don't wait for MCP initialization to finish.
let mut w = new_widget;
w.standard_terminal_mode = !config.tui.alternate_screen;
w.history_push_top_next_req(history_cell::new_animated_welcome()); // tag: prelude
let connecting_mcp = !w.config.mcp_servers.is_empty();
w.history_push_top_next_req(history_cell::new_popular_commands_notice(false)); // tag: prelude
if connecting_mcp {
// Render connecting status as a separate cell with standard gutter and spacing
w.history_push_top_next_req(history_cell::new_connecting_mcp_status());
if config.experimental_resume.is_none() {
w.history_push_top_next_req(history_cell::new_animated_welcome()); // tag: prelude
let connecting_mcp = !w.config.mcp_servers.is_empty();
w.history_push_top_next_req(history_cell::new_popular_commands_notice(false)); // tag: prelude
if connecting_mcp {
// Render connecting status as a separate cell with standard gutter and spacing
w.history_push_top_next_req(history_cell::new_connecting_mcp_status());
}
// Mark welcome as shown to avoid duplicating the Popular commands section
// when SessionConfigured arrives shortly after.
w.welcome_shown = true;
} else {
w.welcome_shown = true;
}
// Mark welcome as shown to avoid duplicating the Popular commands section
// when SessionConfigured arrives shortly after.
w.welcome_shown = true;
w
}
@@ -3318,9 +3377,37 @@ impl ChatWidget<'_> {
self.maybe_hide_spinner();
}
EventMsg::ReplayHistory(ev) => {
// Render prior transcript items statically without executing tools
for item in ev.items {
self.render_replay_item(item);
let codex_core::protocol::ReplayHistoryEvent { items, events } = ev;
let mut max_req = self.last_seen_request_index;
if events.is_empty() {
for item in &items {
self.render_replay_item(item.clone());
}
} else {
for recorded in events {
if matches!(recorded.msg, EventMsg::ReplayHistory(_)) {
continue;
}
if let Some(order) = recorded.order.as_ref() {
max_req = max_req.max(order.request_ordinal);
}
let event = Event {
id: recorded.id,
event_seq: recorded.event_seq,
msg: recorded.msg,
order: recorded.order,
};
self.handle_codex_event(event);
}
}
if !items.is_empty() {
// History items were inserted using synthetic keys; promote current request
// index so subsequent messages append to the end instead of the top.
self.last_seen_request_index = self.last_seen_request_index.max(self.current_request_index);
}
if max_req > 0 {
self.last_seen_request_index = self.last_seen_request_index.max(max_req);
self.current_request_index = self.last_seen_request_index;
}
self.request_redraw();
}
codex-rs/tui/src/resume/discovery.rs
+261 -6
View File
@@ -1,7 +1,14 @@
use codex_core::protocol::EventMsg;
use codex_protocol::models::{ContentItem, ResponseItem};
use codex_protocol::protocol::{RolloutItem, RolloutLine};
use codex_core::Cursor;
use codex_core::RolloutRecorder;
use serde::Deserialize;
use std::fs;
use std::io::{BufRead, BufReader};
use std::path::{Path, PathBuf};
use std::thread;
use tokio::runtime::{Builder, Handle};
/// One candidate session for the picker
pub struct ResumeCandidate {
@@ -20,14 +27,28 @@ pub struct ResumeCandidate {
/// Return rollout files under ~/.codex/sessions matching the provided cwd.
/// Reads only the first line of each file to avoid heavy IO.
pub fn list_sessions_for_cwd(cwd: &Path, codex_home: &Path) -> Vec<ResumeCandidate> {
// First: try per-directory index
if let Some(mut v) = read_dir_index(codex_home, cwd) {
v.sort_by(|a, b| b.sort_key.cmp(&a.sort_key));
return v.into_iter().take(200).collect();
use std::collections::HashMap;
const MAX_RESULTS: usize = 200;
let mut by_path: HashMap<PathBuf, ResumeCandidate> = HashMap::new();
if let Some(index_rows) = read_dir_index(codex_home, cwd) {
for row in index_rows {
by_path.entry(row.path.clone()).or_insert(row);
}
}
// No index found for this directory
Vec::new()
for row in fallback_scan_sessions_for_cwd(cwd, codex_home) {
by_path.entry(row.path.clone()).or_insert(row);
}
let mut results: Vec<ResumeCandidate> = by_path.into_values().collect();
results.sort_by(|a, b| b.sort_key.cmp(&a.sort_key));
if results.len() > MAX_RESULTS {
results.truncate(MAX_RESULTS);
}
results
}
#[derive(Deserialize)]
@@ -109,3 +130,237 @@ fn super_sanitize_dir_index_path(codex_home: &Path, cwd: &Path) -> PathBuf {
}
// Removed fallback slow scan; the fast per-directory index is authoritative.
fn fallback_scan_sessions_for_cwd(cwd: &Path, codex_home: &Path) -> Vec<ResumeCandidate> {
const MAX_RESULTS: usize = 200;
let codex_home = codex_home.to_path_buf();
let cwd = cwd.to_path_buf();
let fetch = async move {
let mut collected: Vec<ResumeCandidate> = Vec::new();
let mut cursor: Option<Cursor> = None;
while collected.len() < MAX_RESULTS {
let page = match RolloutRecorder::list_conversations(codex_home.as_path(), 256, cursor.as_ref()).await {
Ok(page) => page,
Err(err) => {
tracing::warn!("failed to list conversations for resume fallback: {err}");
break;
}
};
if page.items.is_empty() {
break;
}
for item in page.items {
if let Some(candidate) = parse_rollout_candidate(&item.path, &cwd) {
collected.push(candidate);
if collected.len() >= MAX_RESULTS {
break;
}
}
}
if collected.len() >= MAX_RESULTS || page.next_cursor.is_none() {
break;
}
cursor = page.next_cursor;
}
collected
};
// Execute the async fetch, reusing an existing runtime when available.
let mut sessions = match Handle::try_current() {
Ok(handle) => {
let handle = handle.clone();
match thread::spawn(move || handle.block_on(fetch)).join() {
Ok(result) => result,
Err(_) => {
tracing::warn!("resume fallback thread panicked while listing conversations");
Vec::new()
}
}
}
Err(_) => match Builder::new_current_thread().enable_all().build() {
Ok(rt) => rt.block_on(fetch),
Err(err) => {
tracing::warn!("failed to build tokio runtime for resume fallback: {err}");
Vec::new()
}
},
};
sessions.sort_by(|a, b| b.sort_key.cmp(&a.sort_key));
if sessions.len() > MAX_RESULTS {
sessions.truncate(MAX_RESULTS);
}
sessions
}
fn parse_rollout_candidate(path: &Path, target_cwd: &Path) -> Option<ResumeCandidate> {
let file = fs::File::open(path).ok()?;
let reader = BufReader::new(file);
let mut created_ts: Option<String> = None;
let mut modified_ts: Option<String> = None;
let mut message_count: usize = 0;
let mut branch: Option<String> = None;
let mut last_user_snippet: Option<String> = None;
let mut instructions: Option<String> = None;
let mut path_checked = false;
for line in reader.lines() {
let Ok(line) = line else { continue };
let trimmed = line.trim();
if trimmed.is_empty() {
continue;
}
let parsed: RolloutLine = match serde_json::from_str(trimmed) {
Ok(v) => v,
Err(_) => continue,
};
modified_ts = Some(parsed.timestamp.clone());
match parsed.item {
RolloutItem::SessionMeta(meta_line) => {
if !path_checked {
path_checked = true;
if !paths_match(&meta_line.meta.cwd, target_cwd) {
return None;
}
}
if created_ts.is_none() {
created_ts = Some(meta_line.meta.timestamp.clone());
}
instructions = instructions.or_else(|| meta_line.meta.instructions.clone());
if branch.is_none() {
if let Some(git) = &meta_line.git {
if let Some(b) = &git.branch {
if !b.is_empty() {
branch = Some(b.clone());
}
}
}
}
}
RolloutItem::ResponseItem(response_item) => {
message_count = message_count.saturating_add(1);
if let Some(snippet) = extract_user_snippet_from_response(&response_item) {
last_user_snippet = Some(snippet);
}
}
RolloutItem::Event(recorded) => {
if let Some(event) = codex_core::protocol::event_msg_from_protocol(&recorded.msg) {
if let Some(snippet) = extract_user_snippet_from_event(&event) {
last_user_snippet = Some(snippet);
}
}
}
RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) => {}
}
}
if !path_checked {
return None;
}
if message_count == 0 {
return None;
}
let modified_ts = modified_ts.or_else(|| created_ts.clone());
let snippet = last_user_snippet.or_else(|| instructions.clone());
Some(ResumeCandidate {
path: path.to_path_buf(),
subtitle: snippet.clone(),
sort_key: modified_ts.clone().unwrap_or_default(),
created_ts,
modified_ts,
message_count,
branch,
snippet,
})
}
fn extract_user_snippet_from_response(item: &ResponseItem) -> Option<String> {
match item {
ResponseItem::Message { role, content, .. } if role == "user" => content
.iter()
.find_map(|c| match c {
ContentItem::InputText { text } | ContentItem::OutputText { text } => {
let trimmed = text.trim();
if trimmed.is_empty() || trimmed.contains("== System Status ==") {
None
} else {
Some(trimmed.to_string())
}
}
_ => None,
}),
_ => None,
}
}
fn extract_user_snippet_from_event(event: &EventMsg) -> Option<String> {
match event {
EventMsg::UserMessage(ev) => {
let trimmed = ev.message.trim();
if trimmed.is_empty() || trimmed.contains("== System Status ==") {
None
} else {
Some(trimmed.to_string())
}
}
_ => None,
}
}
fn paths_match(meta: &Path, target: &Path) -> bool {
if meta == target {
return true;
}
let meta_norm = normalize_path(meta);
let target_norm = normalize_path(target);
if meta_norm == target_norm || meta_norm.eq_ignore_ascii_case(&target_norm) {
return true;
}
if let (Ok(meta_real), Ok(target_real)) = (fs::canonicalize(meta), fs::canonicalize(target)) {
let meta_real_norm = normalize_path(&meta_real);
let target_real_norm = normalize_path(&target_real);
if meta_real_norm == target_real_norm
|| meta_real_norm.eq_ignore_ascii_case(&target_real_norm)
{
return true;
}
// If both paths belong to the same Git project (including worktrees) consider them equivalent.
if let (Some(meta_root), Some(target_root)) = (
codex_core::git_info::resolve_root_git_project_for_trust(&meta_real)
.and_then(|p| fs::canonicalize(p).ok()),
codex_core::git_info::resolve_root_git_project_for_trust(&target_real)
.and_then(|p| fs::canonicalize(p).ok()),
) {
let meta_root_norm = normalize_path(&meta_root);
let target_root_norm = normalize_path(&target_root);
if meta_root_norm == target_root_norm
|| meta_root_norm.eq_ignore_ascii_case(&target_root_norm)
{
return true;
}
}
}
false
}
fn normalize_path(path: &Path) -> String {
let mut s = path.to_string_lossy().replace('\\', "/");
while s.ends_with('/') && s.len() > 1 {
s.pop();
}
s
}