agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

codex: migrate (more) app-server thread history reads to ThreadStore (#20575)

Browse Source 
Migrate token usage replay, rollback responses, and detached review
setup (a special case of forking) to be served from ThreadStore reads
rather direct rollout files.

- replay restored token usage from already-loaded `RolloutItem` history
instead of reopening `Thread.path`
- rebuild rollback responses from loaded `ThreadStore` snapshots and
history
- start detached reviews from store-backed parent history and stored
review-thread metadata
- remove obsolete app-server rollout-summary helper code that became
dead after the store-backed migration
- preserve response/notification ordering for resume, fork, rollback,
and detached review flows
- add integration test coverage for the affected paths
This commit is contained in:
Tom
2026-05-04 21:16:50 -07:00
committed by GitHub
Unverified
parent 7e71d02610
commit 33d24b0df5
11 changed files with 276 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/bespoke_event_handling.rs
+122 -63
View File
@@ -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<tokio::sync::Semaphore>,
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<ThreadRollbackResponse, String> {
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<Mutex<ThreadState>>,
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<Mutex<ThreadState>>,
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
codex-rs/app-server/src/message_processor.rs
-1
View File
@@ -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.
codex-rs/app-server/src/request_processors.rs
+8 -2
View File
@@ -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;
codex-rs/app-server/src/request_processors/thread_lifecycle.rs
+1 -2
View File
@@ -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>,
codex-rs/app-server/src/request_processors/thread_processor.rs
+5 -18
View File
@@ -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,
codex-rs/app-server/src/request_processors/thread_summary.rs
+4 -12
View File
@@ -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<Vec<RolloutItem>> {
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<DateTime<Utc>> {
})
}
#[cfg(test)]
async fn read_updated_at(path: &Path, created_at: Option<&str>) -> Option<String> {
let updated_at = tokio::fs::metadata(path)
.await
codex-rs/app-server/src/request_processors/token_usage_replay.rs
+59
View File
@@ -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<RolloutItem> {
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(),
})),
]
}
}
codex-rs/app-server/src/request_processors/turn_processor.rs
+40 -49
View File
@@ -13,7 +13,6 @@ pub(crate) struct TurnRequestProcessor {
thread_state_manager: ThreadStateManager,
thread_watch_manager: ThreadWatchManager,
thread_list_state_permit: Arc<Semaphore>,
state_db: Option<StateDbHandle>,
}
impl TurnRequestProcessor {
@@ -30,7 +29,6 @@ impl TurnRequestProcessor {
thread_state_manager: ThreadStateManager,
thread_watch_manager: ThreadWatchManager,
thread_list_state_permit: Arc<Semaphore>,
state_db: Option<StateDbHandle>,
) -> 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
codex-rs/core/src/codex_thread.rs
+17
View File
@@ -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<StoredThread> {
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,
codex-rs/thread-store/src/live_thread.rs
+15
View File
@@ -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<StoredThread> {
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,
codex-rs/thread-store/src/local/read_thread.rs
+5
View File
@@ -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)
}