diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 89b5d2657..045854c9b 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -2,10 +2,8 @@ use crate::error_code::internal_error; use crate::error_code::invalid_request; use crate::outgoing_message::ClientRequestResult; use crate::outgoing_message::ThreadScopedOutgoingMessageSender; -use crate::request_processors::build_api_turns_from_rollout_items; -use crate::request_processors::read_rollout_items_from_rollout; -use crate::request_processors::read_summary_from_rollout; -use crate::request_processors::summary_to_thread; +use crate::request_processors::populate_thread_turns_from_history; +use crate::request_processors::thread_from_stored_thread; use crate::server_request_error::is_turn_transition_server_request_error; use crate::thread_state::ThreadState; use crate::thread_state::TurnSummary; @@ -65,6 +63,7 @@ use codex_app_server_protocol::ThreadRealtimeStartedNotification; use codex_app_server_protocol::ThreadRealtimeTranscriptDeltaNotification; use codex_app_server_protocol::ThreadRealtimeTranscriptDoneNotification; use codex_app_server_protocol::ThreadRollbackResponse; +use codex_app_server_protocol::ThreadStatus; use codex_app_server_protocol::ThreadTokenUsage; use codex_app_server_protocol::ThreadTokenUsageUpdatedNotification; use codex_app_server_protocol::ToolRequestUserInputOption; @@ -86,7 +85,6 @@ use codex_app_server_protocol::guardian_auto_approval_review_notification; use codex_app_server_protocol::item_event_to_server_notification; use codex_core::CodexThread; use codex_core::ThreadManager; -use codex_core::find_thread_name_by_id; use codex_core::review_format::format_review_findings_block; use codex_core::review_prompts; use codex_protocol::ThreadId; @@ -114,14 +112,12 @@ use codex_sandboxing::policy_transforms::intersect_permission_profiles; use codex_shell_command::parse_command::shlex_join; use codex_utils_absolute_path::AbsolutePathBuf; use std::collections::HashMap; -use std::path::Path; use std::sync::Arc; use std::time::SystemTime; use std::time::UNIX_EPOCH; use tokio::sync::Mutex; use tokio::sync::oneshot; use tracing::error; -use tracing::warn; enum CommandExecutionApprovalPresentation { Network(V2NetworkApprovalContext), @@ -147,7 +143,6 @@ pub(crate) async fn apply_bespoke_event_handling( thread_watch_manager: ThreadWatchManager, thread_list_state_permit: Arc, fallback_model_provider: String, - codex_home: &Path, ) { let Event { id: event_turn_id, @@ -1166,69 +1161,43 @@ pub(crate) async fn apply_bespoke_event_handling( return; } }; - let Some(rollout_path) = conversation.rollout_path() else { - outgoing - .send_error( - request_id, - invalid_request("thread has no persisted rollout"), - ) - .await; - return; - }; - let response = match read_summary_from_rollout( - rollout_path.as_path(), - fallback_model_provider.as_str(), - ) - .await + let fallback_cwd = conversation.config_snapshot().await.cwd; + let stored_thread = match conversation + .read_thread( + /*include_archived*/ true, /*include_history*/ true, + ) + .await { - Ok(summary) => { - let fallback_cwd = conversation.config_snapshot().await.cwd; - let mut thread = summary_to_thread(summary, &fallback_cwd); - match read_rollout_items_from_rollout(rollout_path.as_path()).await { - Ok(items) => { - thread.turns = build_api_turns_from_rollout_items(&items); - thread.status = thread_watch_manager - .loaded_status_for_thread(&thread.id) - .await; - match find_thread_name_by_id(codex_home, &conversation_id).await { - Ok(name) => { - thread.name = name; - } - Err(err) => { - warn!( - "Failed to read thread name for {conversation_id}: {err}" - ); - } - } - ThreadRollbackResponse { thread } - } - Err(err) => { - outgoing - .send_error( - request_id.clone(), - internal_error(format!( - "failed to load rollout `{}`: {err}", - rollout_path.display() - )), - ) - .await; - return; - } - } - } + Ok(stored_thread) => stored_thread, Err(err) => { outgoing .send_error( request_id.clone(), internal_error(format!( - "failed to load rollout `{}`: {err}", - rollout_path.display() + "failed to read thread {conversation_id} after rollback: {err}" )), ) .await; return; } }; + let loaded_status = thread_watch_manager + .loaded_status_for_thread(&conversation_id.to_string()) + .await; + let response = match thread_rollback_response_from_stored_thread( + stored_thread, + fallback_model_provider.as_str(), + &fallback_cwd, + loaded_status, + ) { + Ok(response) => response, + Err(err) => { + outgoing + .send_error(request_id.clone(), internal_error(err)) + .await; + return; + } + }; outgoing.send_response(request_id, response).await; } @@ -1578,6 +1547,25 @@ async fn handle_thread_rollback_failed( } } +fn thread_rollback_response_from_stored_thread( + stored_thread: codex_thread_store::StoredThread, + fallback_model_provider: &str, + fallback_cwd: &AbsolutePathBuf, + loaded_status: ThreadStatus, +) -> std::result::Result { + let thread_id = stored_thread.thread_id; + let (mut thread, history) = + thread_from_stored_thread(stored_thread, fallback_model_provider, fallback_cwd); + let Some(history) = history else { + return Err(format!( + "thread {thread_id} did not include persisted history after rollback" + )); + }; + populate_thread_turns_from_history(&mut thread, &history.items, /*active_turn*/ None); + thread.status = loaded_status; + Ok(ThreadRollbackResponse { thread }) +} + async fn respond_to_pending_interrupts( thread_state: &Arc>, outgoing: &ThreadScopedOutgoingMessageSender, @@ -2105,6 +2093,7 @@ mod tests { use anyhow::Result; use anyhow::anyhow; use anyhow::bail; + use chrono::Utc; use codex_app_server_protocol::AutoReviewDecisionSource; use codex_app_server_protocol::GuardianApprovalReviewStatus; use codex_app_server_protocol::JSONRPCErrorError; @@ -2121,20 +2110,28 @@ mod tests { use codex_protocol::permissions::FileSystemSpecialPath; use codex_protocol::plan_tool::PlanItemArg; use codex_protocol::plan_tool::StepStatus; + use codex_protocol::protocol::AgentMessageEvent; + use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::CreditsSnapshot; + use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::GuardianAssessmentEvent; use codex_protocol::protocol::GuardianAssessmentStatus; use codex_protocol::protocol::RateLimitSnapshot; use codex_protocol::protocol::RateLimitWindow; + use codex_protocol::protocol::RolloutItem; + use codex_protocol::protocol::SandboxPolicy; + use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::TokenUsage; use codex_protocol::protocol::TokenUsageInfo; + use codex_protocol::protocol::UserMessageEvent; + use codex_thread_store::StoredThread; + use codex_thread_store::StoredThreadHistory; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_absolute_path::test_support::PathBufExt; use codex_utils_absolute_path::test_support::test_path_buf; use core_test_support::load_default_config_for_test; use pretty_assertions::assert_eq; use serde_json::json; - use std::path::PathBuf; use tempfile::TempDir; use tokio::sync::Mutex; use tokio::sync::mpsc; @@ -2159,6 +2156,71 @@ mod tests { } } + #[test] + fn rollback_response_rebuilds_pathless_thread_from_stored_history() -> Result<()> { + let thread_id = ThreadId::from_string("00000000-0000-0000-0000-000000000789")?; + let created_at = Utc::now(); + let history_items = vec![ + RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { + message: "before rollback".to_string(), + images: None, + local_images: Vec::new(), + text_elements: Vec::new(), + })), + RolloutItem::EventMsg(EventMsg::AgentMessage(AgentMessageEvent { + message: "after rollback".to_string(), + phase: None, + memory_citation: None, + })), + ]; + let stored_thread = StoredThread { + thread_id, + rollout_path: None, + forked_from_id: None, + preview: "fallback preview".to_string(), + name: Some("Rollback thread".to_string()), + model_provider: "openai".to_string(), + model: None, + reasoning_effort: None, + created_at, + updated_at: created_at, + archived_at: None, + cwd: test_path_buf("/tmp").abs().into(), + cli_version: "0.0.0".to_string(), + source: SessionSource::Cli, + agent_nickname: None, + agent_role: None, + agent_path: None, + git_info: None, + approval_mode: AskForApproval::OnRequest, + sandbox_policy: SandboxPolicy::new_read_only_policy(), + token_usage: None, + first_user_message: Some("before rollback".to_string()), + history: Some(StoredThreadHistory { + thread_id, + items: history_items, + }), + }; + let fallback_cwd = test_path_buf("/tmp").abs(); + + let response = thread_rollback_response_from_stored_thread( + stored_thread, + "fallback-provider", + &fallback_cwd, + ThreadStatus::NotLoaded, + ) + .expect("rollback response should rebuild from stored history"); + + assert_eq!(response.thread.id, thread_id.to_string()); + assert_eq!(response.thread.path, None); + assert_eq!(response.thread.preview, "before rollback"); + assert_eq!(response.thread.name.as_deref(), Some("Rollback thread")); + assert_eq!(response.thread.status, ThreadStatus::NotLoaded); + assert_eq!(response.thread.turns.len(), 1); + assert_eq!(response.thread.turns[0].items.len(), 2); + Ok(()) + } + fn turn_complete_event(turn_id: &str) -> TurnCompleteEvent { TurnCompleteEvent { turn_id: turn_id.to_string(), @@ -2241,7 +2303,6 @@ mod tests { thread_state: Arc>, thread_watch_manager: ThreadWatchManager, analytics_events_client: AnalyticsEventsClient, - codex_home: PathBuf, } impl GuardianAssessmentTestContext { @@ -2261,7 +2322,6 @@ mod tests { self.thread_watch_manager.clone(), Arc::new(tokio::sync::Semaphore::new(/*permits*/ 1)), "test-provider".to_string(), - &self.codex_home, ) .await; } @@ -2596,7 +2656,6 @@ mod tests { "http://localhost".to_string(), Some(false), ), - codex_home: codex_home.path().to_path_buf(), }; guardian_context diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 605421204..308fd8f1e 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -409,7 +409,6 @@ impl MessageProcessor { thread_state_manager, thread_watch_manager, thread_list_state_permit, - state_db.clone(), ); if matches!(plugin_startup_tasks, crate::PluginStartupTasks::Start) { // Keep plugin startup warmups aligned at app-server startup. diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index ff88a076e..cb713ac6e 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -248,7 +248,7 @@ use codex_core::CodexThread; use codex_core::CodexThreadTurnContextOverrides; use codex_core::ForkSnapshot; use codex_core::NewThread; -use codex_core::RolloutRecorder; +#[cfg(test)] use codex_core::SessionMeta; use codex_core::StartThreadOptions; use codex_core::SteerInputError; @@ -266,6 +266,7 @@ use codex_core::exec_env::create_env; use codex_core::find_thread_name_by_id; use codex_core::find_thread_path_by_id_str; use codex_core::path_utils; +#[cfg(test)] use codex_core::read_head_for_summary; use codex_core::sandboxing::SandboxPermissions; use codex_core::windows_sandbox::WindowsSandboxLevelExt; @@ -336,6 +337,7 @@ use codex_protocol::config_types::WindowsSandboxLevel; use codex_protocol::dynamic_tools::DynamicToolSpec as CoreDynamicToolSpec; use codex_protocol::error::CodexErr; use codex_protocol::error::Result as CodexResult; +#[cfg(test)] use codex_protocol::items::TurnItem; use codex_protocol::models::ResponseItem; use codex_protocol::permissions::FileSystemSandboxPolicy; @@ -345,6 +347,7 @@ use codex_protocol::protocol::ConversationStartParams; use codex_protocol::protocol::ConversationStartTransport; use codex_protocol::protocol::ConversationTextParams; use codex_protocol::protocol::EventMsg; +#[cfg(test)] use codex_protocol::protocol::GitInfo as CoreGitInfo; use codex_protocol::protocol::InitialHistory; use codex_protocol::protocol::McpAuthStatus as CoreMcpAuthStatus; @@ -358,6 +361,7 @@ use codex_protocol::protocol::ReviewRequest; use codex_protocol::protocol::ReviewTarget as CoreReviewTarget; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::SessionConfiguredEvent; +#[cfg(test)] use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::TurnEnvironmentSelection; use codex_protocol::protocol::USER_MESSAGE_BEGIN; @@ -477,7 +481,9 @@ use self::thread_goal_processor::api_thread_goal_from_state; use self::thread_lifecycle::*; use self::thread_summary::*; -pub(crate) use self::thread_summary::read_rollout_items_from_rollout; +pub(crate) use self::thread_lifecycle::populate_thread_turns_from_history; +pub(crate) use self::thread_processor::thread_from_stored_thread; +#[cfg(test)] pub(crate) use self::thread_summary::read_summary_from_rollout; pub(crate) use self::thread_summary::summary_to_thread; diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index 9dabdea44..fae6fa855 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -331,7 +331,6 @@ pub(super) async fn ensure_listener_task_running( thread_watch_manager.clone(), thread_list_state_permit.clone(), fallback_model_provider.clone(), - codex_home.as_path(), ) .await; } @@ -702,7 +701,7 @@ pub(super) async fn send_thread_goal_snapshot_notification( } } -pub(super) fn populate_thread_turns_from_history( +pub(crate) fn populate_thread_turns_from_history( thread: &mut Thread, items: &[RolloutItem], active_turn: Option<&Turn>, diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 75f79b808..3d8746092 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -3006,23 +3006,10 @@ impl ThreadRequestProcessor { // `excludeTurns` is the cheap fork path, so skip restored usage replay // instead of rebuilding history only to attribute a historical update. if let Some(token_usage_thread) = token_usage_thread { - let token_usage_turn_id = if let Some(rollout_path) = token_usage_thread.path.as_deref() - { - read_rollout_items_from_rollout(rollout_path) - .await - .ok() - .and_then(|rollout_items| { - latest_token_usage_turn_id_from_rollout_items( - &rollout_items, - token_usage_thread.turns.as_slice(), - ) - }) - } else { - latest_token_usage_turn_id_from_rollout_items( - &history_items, - token_usage_thread.turns.as_slice(), - ) - }; + let token_usage_turn_id = latest_token_usage_turn_id_from_rollout_items( + &history_items, + token_usage_thread.turns.as_slice(), + ); // Mirror the resume contract for forks: the new thread is usable as soon // as the response arrives, so restored usage must follow immediately. send_thread_token_usage_update_to_connection( @@ -3588,7 +3575,7 @@ fn set_thread_name_from_title(thread: &mut Thread, title: String) { thread.name = Some(title); } -fn thread_from_stored_thread( +pub(crate) fn thread_from_stored_thread( thread: StoredThread, fallback_provider: &str, fallback_cwd: &AbsolutePathBuf, diff --git a/codex-rs/app-server/src/request_processors/thread_summary.rs b/codex-rs/app-server/src/request_processors/thread_summary.rs index ed2360ed1..bb02affcc 100644 --- a/codex-rs/app-server/src/request_processors/thread_summary.rs +++ b/codex-rs/app-server/src/request_processors/thread_summary.rs @@ -1,5 +1,6 @@ use super::*; +#[cfg(test)] pub(crate) async fn read_summary_from_rollout( path: &Path, fallback_provider: &str, @@ -74,18 +75,7 @@ pub(crate) async fn read_summary_from_rollout( }) } -pub(crate) async fn read_rollout_items_from_rollout( - path: &Path, -) -> std::io::Result> { - let items = match RolloutRecorder::get_rollout_history(path).await? { - InitialHistory::New | InitialHistory::Cleared => Vec::new(), - InitialHistory::Forked(items) => items, - InitialHistory::Resumed(resumed) => resumed.history, - }; - - Ok(items) -} - +#[cfg(test)] fn extract_conversation_summary( path: PathBuf, head: &[serde_json::Value], @@ -134,6 +124,7 @@ fn extract_conversation_summary( }) } +#[cfg(test)] fn map_git_info(git_info: &CoreGitInfo) -> ConversationGitInfo { ConversationGitInfo { sha: git_info.commit_hash.as_ref().map(|sha| sha.0.clone()), @@ -220,6 +211,7 @@ fn parse_datetime(timestamp: Option<&str>) -> Option> { }) } +#[cfg(test)] async fn read_updated_at(path: &Path, created_at: Option<&str>) -> Option { let updated_at = tokio::fs::metadata(path) .await diff --git a/codex-rs/app-server/src/request_processors/token_usage_replay.rs b/codex-rs/app-server/src/request_processors/token_usage_replay.rs index 75b07d0e4..b19c4a61a 100644 --- a/codex-rs/app-server/src/request_processors/token_usage_replay.rs +++ b/codex-rs/app-server/src/request_processors/token_usage_replay.rs @@ -112,3 +112,62 @@ fn latest_token_usage_turn_id(thread: &Thread) -> String { .map(|turn| turn.id.clone()) .unwrap_or_default() } + +#[cfg(test)] +mod tests { + use super::*; + use codex_app_server_protocol::build_turns_from_rollout_items; + use codex_protocol::protocol::AgentMessageEvent; + use codex_protocol::protocol::TokenCountEvent; + use codex_protocol::protocol::UserMessageEvent; + use pretty_assertions::assert_eq; + + #[test] + fn replay_attribution_uses_already_loaded_history() { + let rollout_items = token_usage_history(); + let turns = build_turns_from_rollout_items(&rollout_items); + + assert_eq!( + latest_token_usage_turn_id_from_rollout_items(&rollout_items, turns.as_slice()), + Some(turns[0].id.clone()) + ); + } + + #[test] + fn replay_attribution_falls_back_to_rebuilt_turn_position() { + let rollout_items = token_usage_history(); + let mut turns = build_turns_from_rollout_items(&rollout_items); + turns[0].id = "rebuilt-turn-id".to_string(); + + assert_eq!( + latest_token_usage_turn_id_from_rollout_items(&rollout_items, turns.as_slice()), + Some("rebuilt-turn-id".to_string()) + ); + } + + fn token_usage_history() -> Vec { + vec![ + RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { + message: "first turn".to_string(), + images: None, + local_images: Vec::new(), + text_elements: Vec::new(), + })), + RolloutItem::EventMsg(EventMsg::AgentMessage(AgentMessageEvent { + message: "first answer".to_string(), + phase: None, + memory_citation: None, + })), + RolloutItem::EventMsg(EventMsg::TokenCount(TokenCountEvent { + info: None, + rate_limits: None, + })), + RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { + message: "second turn".to_string(), + images: None, + local_images: Vec::new(), + text_elements: Vec::new(), + })), + ] + } +} diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index c11b7166e..1629f25c2 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -13,7 +13,6 @@ pub(crate) struct TurnRequestProcessor { thread_state_manager: ThreadStateManager, thread_watch_manager: ThreadWatchManager, thread_list_state_permit: Arc, - state_db: Option, } impl TurnRequestProcessor { @@ -30,7 +29,6 @@ impl TurnRequestProcessor { thread_state_manager: ThreadStateManager, thread_watch_manager: ThreadWatchManager, thread_list_state_permit: Arc, - state_db: Option, ) -> Self { Self { auth_manager, @@ -44,7 +42,6 @@ impl TurnRequestProcessor { thread_state_manager, thread_watch_manager, thread_list_state_permit, - state_db, } } @@ -891,24 +888,20 @@ impl TurnRequestProcessor { review_request: ReviewRequest, display_text: &str, ) -> std::result::Result<(), JSONRPCErrorError> { - let rollout_path = if let Some(path) = parent_thread.rollout_path() { - path - } else { - find_thread_path_by_id_str( - &self.config.codex_home, - &parent_thread_id.to_string(), - self.state_db.as_deref(), - ) + parent_thread.ensure_rollout_materialized().await; + parent_thread.flush_rollout().await.map_err(|err| { + internal_error(format!( + "failed to flush parent thread {parent_thread_id}: {err}" + )) + })?; + let parent_history = parent_thread + .load_history(/*include_archived*/ true) .await .map_err(|err| { internal_error(format!( - "failed to locate thread id {parent_thread_id}: {err}" + "failed to load parent thread {parent_thread_id}: {err}" )) - })? - .ok_or_else(|| { - invalid_request(format!("no rollout found for thread id {parent_thread_id}")) - })? - }; + })?; let mut config = self.config.as_ref().clone(); if let Some(review_model) = &config.review_model { @@ -918,14 +911,17 @@ impl TurnRequestProcessor { let NewThread { thread_id, thread: review_thread, - session_configured, .. } = self .thread_manager - .fork_thread( + .fork_thread_from_history( ForkSnapshot::Interrupted, config.clone(), - rollout_path, + InitialHistory::Resumed(ResumedHistory { + conversation_id: parent_thread_id, + history: parent_history.items, + rollout_path: parent_thread.rollout_path(), + }), /*persist_extended_history*/ false, self.request_trace_context(request_id).await, ) @@ -947,37 +943,32 @@ impl TurnRequestProcessor { ); let fallback_provider = self.config.model_provider_id.as_str(); - if let Some(rollout_path) = review_thread.rollout_path() { - match read_summary_from_rollout(rollout_path.as_path(), fallback_provider).await { - Ok(summary) => { - let mut thread = summary_to_thread(summary, &self.config.cwd); + match review_thread + .read_thread( + /*include_archived*/ true, /*include_history*/ false, + ) + .await + { + Ok(stored_thread) => { + let (mut thread, _) = + thread_from_stored_thread(stored_thread, fallback_provider, &self.config.cwd); + self.thread_watch_manager + .upsert_thread_silently(thread.clone()) + .await; + thread.status = resolve_thread_status( self.thread_watch_manager - .upsert_thread_silently(thread.clone()) - .await; - thread.status = resolve_thread_status( - self.thread_watch_manager - .loaded_status_for_thread(&thread.id) - .await, - /*has_in_progress_turn*/ false, - ); - let notif = thread_started_notification(thread); - self.outgoing - .send_server_notification(ServerNotification::ThreadStarted(notif)) - .await; - } - Err(err) => { - tracing::warn!( - "failed to load summary for review thread {}: {}", - session_configured.session_id, - err - ); - } + .loaded_status_for_thread(&thread.id) + .await, + /*has_in_progress_turn*/ false, + ); + let notif = thread_started_notification(thread); + self.outgoing + .send_server_notification(ServerNotification::ThreadStarted(notif)) + .await; + } + Err(err) => { + tracing::warn!("failed to load summary for review thread {thread_id}: {err}"); } - } else { - tracing::warn!( - "review thread {} has no rollout path", - session_configured.session_id - ); } let turn_id = self diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 3065fcf4a..b4167a0ec 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -413,6 +413,23 @@ impl CodexThread { live_thread.load_history(include_archived).await } + pub async fn read_thread( + &self, + include_archived: bool, + include_history: bool, + ) -> ThreadStoreResult { + let live_thread = self + .codex + .session + .live_thread_for_persistence("read thread") + .map_err(|err| ThreadStoreError::Internal { + message: err.to_string(), + })?; + live_thread + .read_thread(include_archived, include_history) + .await + } + pub async fn update_thread_metadata( &self, patch: ThreadMetadataPatch, diff --git a/codex-rs/thread-store/src/live_thread.rs b/codex-rs/thread-store/src/live_thread.rs index 6cbb1881f..ffbe2a564 100644 --- a/codex-rs/thread-store/src/live_thread.rs +++ b/codex-rs/thread-store/src/live_thread.rs @@ -10,6 +10,7 @@ use crate::AppendThreadItemsParams; use crate::CreateThreadParams; use crate::LoadThreadHistoryParams; use crate::LocalThreadStore; +use crate::ReadThreadParams; use crate::ResumeThreadParams; use crate::StoredThread; use crate::StoredThreadHistory; @@ -140,6 +141,20 @@ impl LiveThread { .await } + pub async fn read_thread( + &self, + include_archived: bool, + include_history: bool, + ) -> ThreadStoreResult { + self.thread_store + .read_thread(ReadThreadParams { + thread_id: self.thread_id, + include_archived, + include_history, + }) + .await + } + pub async fn update_memory_mode( &self, mode: ThreadMemoryMode, diff --git a/codex-rs/thread-store/src/local/read_thread.rs b/codex-rs/thread-store/src/local/read_thread.rs index 097277705..c538d938a 100644 --- a/codex-rs/thread-store/src/local/read_thread.rs +++ b/codex-rs/thread-store/src/local/read_thread.rs @@ -70,6 +70,11 @@ pub(super) async fn read_thread( })?; let mut thread = read_thread_from_rollout_path(store, path).await?; + if !params.include_archived && thread.archived_at.is_some() { + return Err(ThreadStoreError::InvalidRequest { + message: format!("thread {} is archived", thread.thread_id), + }); + } attach_history_if_requested(&mut thread, params.include_history).await?; Ok(thread) }