diff --git a/codex-rs/app-server-protocol/src/lib.rs b/codex-rs/app-server-protocol/src/lib.rs index 9c02ea924..06102083f 100644 --- a/codex-rs/app-server-protocol/src/lib.rs +++ b/codex-rs/app-server-protocol/src/lib.rs @@ -7,5 +7,6 @@ pub use export::generate_ts; pub use export::generate_types; pub use jsonrpc_lite::*; pub use protocol::common::*; +pub use protocol::thread_history::*; pub use protocol::v1::*; pub use protocol::v2::*; diff --git a/codex-rs/app-server-protocol/src/protocol/mod.rs b/codex-rs/app-server-protocol/src/protocol/mod.rs index 11edf04cc..8e2d63e06 100644 --- a/codex-rs/app-server-protocol/src/protocol/mod.rs +++ b/codex-rs/app-server-protocol/src/protocol/mod.rs @@ -2,5 +2,6 @@ // Exposes protocol pieces used by `lib.rs` via `pub use protocol::common::*;`. pub mod common; +pub mod thread_history; pub mod v1; pub mod v2; diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs new file mode 100644 index 000000000..04cd1190b --- /dev/null +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -0,0 +1,409 @@ +use crate::protocol::v2::ThreadItem; +use crate::protocol::v2::Turn; +use crate::protocol::v2::TurnStatus; +use crate::protocol::v2::UserInput; +use codex_protocol::protocol::AgentReasoningEvent; +use codex_protocol::protocol::AgentReasoningRawContentEvent; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::TurnAbortedEvent; +use codex_protocol::protocol::UserMessageEvent; + +/// Convert persisted [`EventMsg`] entries into a sequence of [`Turn`] values. +/// +/// The purpose of this is to convert the EventMsgs persisted in a rollout file +/// into a sequence of Turns and ThreadItems, which allows the client to render +/// the historical messages when resuming a thread. +pub fn build_turns_from_event_msgs(events: &[EventMsg]) -> Vec { + let mut builder = ThreadHistoryBuilder::new(); + for event in events { + builder.handle_event(event); + } + builder.finish() +} + +struct ThreadHistoryBuilder { + turns: Vec, + current_turn: Option, + next_turn_index: i64, + next_item_index: i64, +} + +impl ThreadHistoryBuilder { + fn new() -> Self { + Self { + turns: Vec::new(), + current_turn: None, + next_turn_index: 1, + next_item_index: 1, + } + } + + fn finish(mut self) -> Vec { + self.finish_current_turn(); + self.turns + } + + /// This function should handle all EventMsg variants that can be persisted in a rollout file. + /// See `should_persist_event_msg` in `codex-rs/core/rollout/policy.rs`. + fn handle_event(&mut self, event: &EventMsg) { + match event { + EventMsg::UserMessage(payload) => self.handle_user_message(payload), + EventMsg::AgentMessage(payload) => self.handle_agent_message(payload.message.clone()), + EventMsg::AgentReasoning(payload) => self.handle_agent_reasoning(payload), + EventMsg::AgentReasoningRawContent(payload) => { + self.handle_agent_reasoning_raw_content(payload) + } + EventMsg::TokenCount(_) => {} + EventMsg::EnteredReviewMode(_) => {} + EventMsg::ExitedReviewMode(_) => {} + EventMsg::UndoCompleted(_) => {} + EventMsg::TurnAborted(payload) => self.handle_turn_aborted(payload), + _ => {} + } + } + + fn handle_user_message(&mut self, payload: &UserMessageEvent) { + self.finish_current_turn(); + let mut turn = self.new_turn(); + let id = self.next_item_id(); + let content = self.build_user_inputs(payload); + turn.items.push(ThreadItem::UserMessage { id, content }); + self.current_turn = Some(turn); + } + + fn handle_agent_message(&mut self, text: String) { + if text.is_empty() { + return; + } + + let id = self.next_item_id(); + self.ensure_turn() + .items + .push(ThreadItem::AgentMessage { id, text }); + } + + fn handle_agent_reasoning(&mut self, payload: &AgentReasoningEvent) { + if payload.text.is_empty() { + return; + } + + // If the last item is a reasoning item, add the new text to the summary. + if let Some(ThreadItem::Reasoning { summary, .. }) = self.ensure_turn().items.last_mut() { + summary.push(payload.text.clone()); + return; + } + + // Otherwise, create a new reasoning item. + let id = self.next_item_id(); + self.ensure_turn().items.push(ThreadItem::Reasoning { + id, + summary: vec![payload.text.clone()], + content: Vec::new(), + }); + } + + fn handle_agent_reasoning_raw_content(&mut self, payload: &AgentReasoningRawContentEvent) { + if payload.text.is_empty() { + return; + } + + // If the last item is a reasoning item, add the new text to the content. + if let Some(ThreadItem::Reasoning { content, .. }) = self.ensure_turn().items.last_mut() { + content.push(payload.text.clone()); + return; + } + + // Otherwise, create a new reasoning item. + let id = self.next_item_id(); + self.ensure_turn().items.push(ThreadItem::Reasoning { + id, + summary: Vec::new(), + content: vec![payload.text.clone()], + }); + } + + fn handle_turn_aborted(&mut self, _payload: &TurnAbortedEvent) { + let Some(turn) = self.current_turn.as_mut() else { + return; + }; + turn.status = TurnStatus::Interrupted; + } + + fn finish_current_turn(&mut self) { + if let Some(turn) = self.current_turn.take() { + if turn.items.is_empty() { + return; + } + self.turns.push(turn.into()); + } + } + + fn new_turn(&mut self) -> PendingTurn { + PendingTurn { + id: self.next_turn_id(), + items: Vec::new(), + status: TurnStatus::Completed, + } + } + + fn ensure_turn(&mut self) -> &mut PendingTurn { + if self.current_turn.is_none() { + let turn = self.new_turn(); + return self.current_turn.insert(turn); + } + + if let Some(turn) = self.current_turn.as_mut() { + return turn; + } + + unreachable!("current turn must exist after initialization"); + } + + fn next_turn_id(&mut self) -> String { + let id = format!("turn-{}", self.next_turn_index); + self.next_turn_index += 1; + id + } + + fn next_item_id(&mut self) -> String { + let id = format!("item-{}", self.next_item_index); + self.next_item_index += 1; + id + } + + fn build_user_inputs(&self, payload: &UserMessageEvent) -> Vec { + let mut content = Vec::new(); + if !payload.message.trim().is_empty() { + content.push(UserInput::Text { + text: payload.message.clone(), + }); + } + if let Some(images) = &payload.images { + for image in images { + content.push(UserInput::Image { url: image.clone() }); + } + } + content + } +} + +struct PendingTurn { + id: String, + items: Vec, + status: TurnStatus, +} + +impl From for Turn { + fn from(value: PendingTurn) -> Self { + Self { + id: value.id, + items: value.items, + status: value.status, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use codex_protocol::protocol::AgentMessageEvent; + use codex_protocol::protocol::AgentReasoningEvent; + use codex_protocol::protocol::AgentReasoningRawContentEvent; + use codex_protocol::protocol::TurnAbortReason; + use codex_protocol::protocol::TurnAbortedEvent; + use codex_protocol::protocol::UserMessageEvent; + use pretty_assertions::assert_eq; + + #[test] + fn builds_multiple_turns_with_reasoning_items() { + let events = vec![ + EventMsg::UserMessage(UserMessageEvent { + message: "First turn".into(), + images: Some(vec!["https://example.com/one.png".into()]), + }), + EventMsg::AgentMessage(AgentMessageEvent { + message: "Hi there".into(), + }), + EventMsg::AgentReasoning(AgentReasoningEvent { + text: "thinking".into(), + }), + EventMsg::AgentReasoningRawContent(AgentReasoningRawContentEvent { + text: "full reasoning".into(), + }), + EventMsg::UserMessage(UserMessageEvent { + message: "Second turn".into(), + images: None, + }), + EventMsg::AgentMessage(AgentMessageEvent { + message: "Reply two".into(), + }), + ]; + + let turns = build_turns_from_event_msgs(&events); + assert_eq!(turns.len(), 2); + + let first = &turns[0]; + assert_eq!(first.id, "turn-1"); + assert_eq!(first.status, TurnStatus::Completed); + assert_eq!(first.items.len(), 3); + assert_eq!( + first.items[0], + ThreadItem::UserMessage { + id: "item-1".into(), + content: vec![ + UserInput::Text { + text: "First turn".into(), + }, + UserInput::Image { + url: "https://example.com/one.png".into(), + } + ], + } + ); + assert_eq!( + first.items[1], + ThreadItem::AgentMessage { + id: "item-2".into(), + text: "Hi there".into(), + } + ); + assert_eq!( + first.items[2], + ThreadItem::Reasoning { + id: "item-3".into(), + summary: vec!["thinking".into()], + content: vec!["full reasoning".into()], + } + ); + + let second = &turns[1]; + assert_eq!(second.id, "turn-2"); + assert_eq!(second.items.len(), 2); + assert_eq!( + second.items[0], + ThreadItem::UserMessage { + id: "item-4".into(), + content: vec![UserInput::Text { + text: "Second turn".into() + }], + } + ); + assert_eq!( + second.items[1], + ThreadItem::AgentMessage { + id: "item-5".into(), + text: "Reply two".into(), + } + ); + } + + #[test] + fn splits_reasoning_when_interleaved() { + let events = vec![ + EventMsg::UserMessage(UserMessageEvent { + message: "Turn start".into(), + images: None, + }), + EventMsg::AgentReasoning(AgentReasoningEvent { + text: "first summary".into(), + }), + EventMsg::AgentReasoningRawContent(AgentReasoningRawContentEvent { + text: "first content".into(), + }), + EventMsg::AgentMessage(AgentMessageEvent { + message: "interlude".into(), + }), + EventMsg::AgentReasoning(AgentReasoningEvent { + text: "second summary".into(), + }), + ]; + + let turns = build_turns_from_event_msgs(&events); + assert_eq!(turns.len(), 1); + let turn = &turns[0]; + assert_eq!(turn.items.len(), 4); + + assert_eq!( + turn.items[1], + ThreadItem::Reasoning { + id: "item-2".into(), + summary: vec!["first summary".into()], + content: vec!["first content".into()], + } + ); + assert_eq!( + turn.items[3], + ThreadItem::Reasoning { + id: "item-4".into(), + summary: vec!["second summary".into()], + content: Vec::new(), + } + ); + } + + #[test] + fn marks_turn_as_interrupted_when_aborted() { + let events = vec![ + EventMsg::UserMessage(UserMessageEvent { + message: "Please do the thing".into(), + images: None, + }), + EventMsg::AgentMessage(AgentMessageEvent { + message: "Working...".into(), + }), + EventMsg::TurnAborted(TurnAbortedEvent { + reason: TurnAbortReason::Replaced, + }), + EventMsg::UserMessage(UserMessageEvent { + message: "Let's try again".into(), + images: None, + }), + EventMsg::AgentMessage(AgentMessageEvent { + message: "Second attempt complete.".into(), + }), + ]; + + let turns = build_turns_from_event_msgs(&events); + assert_eq!(turns.len(), 2); + + let first_turn = &turns[0]; + assert_eq!(first_turn.status, TurnStatus::Interrupted); + assert_eq!(first_turn.items.len(), 2); + assert_eq!( + first_turn.items[0], + ThreadItem::UserMessage { + id: "item-1".into(), + content: vec![UserInput::Text { + text: "Please do the thing".into() + }], + } + ); + assert_eq!( + first_turn.items[1], + ThreadItem::AgentMessage { + id: "item-2".into(), + text: "Working...".into(), + } + ); + + let second_turn = &turns[1]; + assert_eq!(second_turn.status, TurnStatus::Completed); + assert_eq!(second_turn.items.len(), 2); + assert_eq!( + second_turn.items[0], + ThreadItem::UserMessage { + id: "item-3".into(), + content: vec![UserInput::Text { + text: "Let's try again".into() + }], + } + ); + assert_eq!( + second_turn.items[1], + ThreadItem::AgentMessage { + id: "item-4".into(), + text: "Second attempt complete.".into(), + } + ); + } +} diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index fa1037dd1..b49dd3e7e 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -517,6 +517,10 @@ pub struct Thread { pub created_at: i64, /// [UNSTABLE] Path to the thread on disk. pub path: PathBuf, + /// Only populated on a `thread/resume` response. + /// For all other responses and notifications returning a Thread, + /// the turns field will be an empty list. + pub turns: Vec, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] @@ -531,8 +535,9 @@ pub struct AccountUpdatedNotification { #[ts(export_to = "v2/")] pub struct Turn { pub id: String, - /// This is currently only populated for resumed threads. - /// TODO: properly populate items for all turns. + /// Only populated on a `thread/resume` response. + /// For all other responses and notifications returning a Turn, + /// the items field will be an empty list. pub items: Vec, #[serde(flatten)] pub status: TurnStatus, diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 1ccc9b129..036224cab 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -91,6 +91,7 @@ use codex_app_server_protocol::TurnStatus; use codex_app_server_protocol::UserInfoResponse; use codex_app_server_protocol::UserInput as V2UserInput; use codex_app_server_protocol::UserSavedConfig; +use codex_app_server_protocol::build_turns_from_event_msgs; use codex_backend_client::Client as BackendClient; use codex_core::AuthManager; use codex_core::CodexConversation; @@ -1652,6 +1653,11 @@ impl CodexMessageProcessor { session_configured, .. }) => { + let SessionConfiguredEvent { + rollout_path, + initial_messages, + .. + } = session_configured; // Auto-attach a conversation listener when resuming a thread. if let Err(err) = self .attach_conversation_listener(conversation_id, false, ApiVersion::V2) @@ -1664,8 +1670,8 @@ impl CodexMessageProcessor { ); } - let thread = match read_summary_from_rollout( - session_configured.rollout_path.as_path(), + let mut thread = match read_summary_from_rollout( + rollout_path.as_path(), fallback_model_provider.as_str(), ) .await @@ -1676,13 +1682,17 @@ impl CodexMessageProcessor { request_id, format!( "failed to load rollout `{}` for conversation {conversation_id}: {err}", - session_configured.rollout_path.display() + rollout_path.display() ), ) .await; return; } }; + thread.turns = initial_messages + .as_deref() + .map_or_else(Vec::new, build_turns_from_event_msgs); + let response = ThreadResumeResponse { thread, model: session_configured.model, @@ -1692,6 +1702,7 @@ impl CodexMessageProcessor { sandbox: session_configured.sandbox_policy.into(), reasoning_effort: session_configured.reasoning_effort, }; + self.outgoing.send_response(request_id, response).await; } Err(err) => { @@ -2983,6 +2994,7 @@ fn summary_to_thread(summary: ConversationSummary) -> Thread { model_provider, created_at: created_at.map(|dt| dt.timestamp()).unwrap_or(0), path, + turns: Vec::new(), } } diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index 1a9c76979..2ef04aac3 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -1,13 +1,17 @@ use anyhow::Result; use app_test_support::McpProcess; +use app_test_support::create_fake_rollout; use app_test_support::create_mock_chat_completions_server; use app_test_support::to_response; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TurnStatus; +use codex_app_server_protocol::UserInput; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; use tempfile::TempDir; @@ -58,6 +62,65 @@ async fn thread_resume_returns_original_thread() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_resume_returns_rollout_history() -> Result<()> { + let server = create_mock_chat_completions_server(vec![]).await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let preview = "Saved user message"; + let conversation_id = create_fake_rollout( + codex_home.path(), + "2025-01-05T12-00-00", + "2025-01-05T12:00:00Z", + preview, + Some("mock_provider"), + )?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let resume_id = mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id: conversation_id.clone(), + ..Default::default() + }) + .await?; + let resume_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(resume_id)), + ) + .await??; + let ThreadResumeResponse { thread, .. } = to_response::(resume_resp)?; + + assert_eq!(thread.id, conversation_id); + assert_eq!(thread.preview, preview); + assert_eq!(thread.model_provider, "mock_provider"); + assert!(thread.path.is_absolute()); + + assert_eq!( + thread.turns.len(), + 1, + "expected rollouts to include one turn" + ); + let turn = &thread.turns[0]; + assert_eq!(turn.status, TurnStatus::Completed); + assert_eq!(turn.items.len(), 1, "expected user message item"); + match &turn.items[0] { + ThreadItem::UserMessage { content, .. } => { + assert_eq!( + content, + &vec![UserInput::Text { + text: preview.to_string() + }] + ); + } + other => panic!("expected user message item, got {other:?}"), + } + + Ok(()) +} + #[tokio::test] async fn thread_resume_prefers_path_over_thread_id() -> Result<()> { let server = create_mock_chat_completions_server(vec![]).await;