diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index 1305ec00f..f0b900f08 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -124,6 +124,14 @@ impl ThreadHistoryBuilder { .or_else(|| self.turns.last().cloned()) } + pub fn turn_snapshot(&self, turn_id: &str) -> Option { + self.current_turn + .as_ref() + .filter(|turn| turn.id == turn_id) + .map(Turn::from) + .or_else(|| self.turns.iter().find(|turn| turn.id == turn_id).cloned()) + } + /// Returns the index of the active turn snapshot within the finished turn list. /// /// When a turn is still open, this is the index it will occupy after diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 2d14ceb5f..b8d6d7a1b 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -526,53 +526,47 @@ impl Session { } else { let live_thread = match &initial_history { InitialHistory::New | InitialHistory::Cleared | InitialHistory::Forked(_) => { - LiveThread::create( - Arc::clone(&thread_store), - CreateThreadParams { - thread_id, - extra_config: config.extra_config.clone(), - forked_from_id, - parent_thread_id, - source: session_source, - thread_source: session_configuration.thread_source.clone(), - base_instructions: BaseInstructions { - text: session_configuration.base_instructions.clone(), - }, - dynamic_tools: session_configuration.dynamic_tools.clone(), - multi_agent_version: initial_multi_agent_version, - metadata: ThreadPersistenceMetadata { - cwd: Some(config.cwd.to_path_buf()), - model_provider: config.model_provider_id.clone(), - memory_mode: if config.memories.generate_memories { - ThreadMemoryMode::Enabled - } else { - ThreadMemoryMode::Disabled - }, + let params = CreateThreadParams { + thread_id, + extra_config: config.extra_config.clone(), + forked_from_id, + parent_thread_id, + source: session_source, + thread_source: session_configuration.thread_source.clone(), + base_instructions: BaseInstructions { + text: session_configuration.base_instructions.clone(), + }, + dynamic_tools: session_configuration.dynamic_tools.clone(), + multi_agent_version: initial_multi_agent_version, + metadata: ThreadPersistenceMetadata { + cwd: Some(config.cwd.to_path_buf()), + model_provider: config.model_provider_id.clone(), + memory_mode: if config.memories.generate_memories { + ThreadMemoryMode::Enabled + } else { + ThreadMemoryMode::Disabled }, }, - ) - .await? + }; + LiveThread::create(Arc::clone(&thread_store), params).await? } InitialHistory::Resumed(resumed_history) => { - LiveThread::resume( - Arc::clone(&thread_store), - ResumeThreadParams { - thread_id: resumed_history.conversation_id, - rollout_path: resumed_history.rollout_path.clone(), - history: Some(resumed_history.history.clone()), - include_archived: true, - metadata: ThreadPersistenceMetadata { - cwd: Some(config.cwd.to_path_buf()), - model_provider: config.model_provider_id.clone(), - memory_mode: if config.memories.generate_memories { - ThreadMemoryMode::Enabled - } else { - ThreadMemoryMode::Disabled - }, + let params = ResumeThreadParams { + thread_id: resumed_history.conversation_id, + rollout_path: resumed_history.rollout_path.clone(), + history: Some(resumed_history.history.clone()), + include_archived: true, + metadata: ThreadPersistenceMetadata { + cwd: Some(config.cwd.to_path_buf()), + model_provider: config.model_provider_id.clone(), + memory_mode: if config.memories.generate_memories { + ThreadMemoryMode::Enabled + } else { + ThreadMemoryMode::Disabled }, }, - ) - .await? + }; + LiveThread::resume(Arc::clone(&thread_store), params).await? } }; Ok(Some(live_thread)) diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index 07156fe35..7983f16dd 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -14,6 +14,7 @@ use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::SessionMeta; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::ThreadMemoryMode; +use codex_rollout::persisted_rollout_items; use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; @@ -215,13 +216,17 @@ impl ThreadStore for InMemoryThreadStore { } async fn append_items(&self, params: AppendThreadItemsParams) -> ThreadStoreResult<()> { + let canonical_items = persisted_rollout_items(params.items.as_slice()); + if canonical_items.is_empty() { + return Ok(()); + } let mut state = self.state.lock().await; state.calls.append_items += 1; state .histories .entry(params.thread_id) .or_default() - .extend(params.items); + .extend(canonical_items); Ok(()) } diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index 284614bf1..5d590528a 100644 --- a/codex-rs/thread-store/src/lib.rs +++ b/codex-rs/thread-store/src/lib.rs @@ -40,6 +40,7 @@ pub use types::SearchThreadsParams; pub use types::SortDirection; pub use types::StoredThread; pub use types::StoredThreadHistory; +pub use types::StoredThreadItem; pub use types::StoredThreadSearchResult; pub use types::StoredTurn; pub use types::StoredTurnError; diff --git a/codex-rs/thread-store/src/live_thread.rs b/codex-rs/thread-store/src/live_thread.rs index d1487860c..88d347f57 100644 --- a/codex-rs/thread-store/src/live_thread.rs +++ b/codex-rs/thread-store/src/live_thread.rs @@ -116,7 +116,11 @@ impl LiveThread { { Ok(history) => params.history = Some(history.items), Err(err) => { - let _ = thread_store.discard_thread(thread_id).await; + if let Err(discard_err) = thread_store.discard_thread(thread_id).await { + warn!( + "failed to discard thread persistence after resume history load failed: {discard_err}" + ); + } return Err(err); } } @@ -131,15 +135,18 @@ impl LiveThread { pub async fn append_items(&self, items: &[RolloutItem]) -> ThreadStoreResult<()> { let canonical_items = persisted_rollout_items(items); - if canonical_items.is_empty() { + if items.is_empty() { return Ok(()); } self.thread_store .append_items(AppendThreadItemsParams { thread_id: self.thread_id, - items: canonical_items.clone(), + items: items.to_vec(), }) .await?; + if canonical_items.is_empty() { + return Ok(()); + } let update = self .metadata_sync .lock() diff --git a/codex-rs/thread-store/src/local/live_writer.rs b/codex-rs/thread-store/src/local/live_writer.rs index 0b31ae0ea..61c15a43d 100644 --- a/codex-rs/thread-store/src/local/live_writer.rs +++ b/codex-rs/thread-store/src/local/live_writer.rs @@ -5,6 +5,7 @@ use codex_protocol::protocol::ThreadMemoryMode; use codex_rollout::RolloutConfig; use codex_rollout::RolloutRecorder; use codex_rollout::RolloutRecorderParams; +use codex_rollout::persisted_rollout_items; use tracing::warn; use super::LocalThreadStore; @@ -77,9 +78,13 @@ pub(super) async fn append_items( store: &LocalThreadStore, params: AppendThreadItemsParams, ) -> ThreadStoreResult<()> { + let canonical_items = persisted_rollout_items(params.items.as_slice()); + if canonical_items.is_empty() { + return Ok(()); + } let recorder = store.live_recorder(params.thread_id).await?; recorder - .record_canonical_items(params.items.as_slice()) + .record_canonical_items(canonical_items.as_slice()) .await .map_err(thread_store_io_error)?; // LiveThread applies metadata immediately after append_items returns. Wait for the local diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs index 6373dd9b1..c0348365f 100644 --- a/codex-rs/thread-store/src/store.rs +++ b/codex-rs/thread-store/src/store.rs @@ -36,11 +36,10 @@ pub trait ThreadStore: Any + Send + Sync { /// Reopens an existing thread for live appends. async fn resume_thread(&self, params: ResumeThreadParams) -> ThreadStoreResult<()>; - /// Appends canonical rollout items to a live thread. + /// Appends raw rollout items to a live thread. /// - /// This is the raw history API. It does not infer metadata from item contents. Callers that - /// need metadata updates should call [`ThreadStore::update_thread_metadata`] with explicit - /// metadata facts prepared above the store. + /// Implementations should apply the shared rollout persistence policy before writing durable + /// replay history and before updating any implementation-owned projections. async fn append_items(&self, params: AppendThreadItemsParams) -> ThreadStoreResult<()>; /// Materializes the thread if persistence is lazy, then persists all queued items. diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index 853d866bc..741ca12bc 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -101,11 +101,14 @@ pub struct ResumeThreadParams { } /// Parameters for appending rollout items to a live thread. -#[derive(Clone, Debug, Serialize, Deserialize)] +#[derive(Clone, Debug)] pub struct AppendThreadItemsParams { /// Thread id to append to. pub thread_id: ThreadId, - /// Items to append in order. + /// Raw rollout items to append in order. + /// + /// Store implementations are responsible for applying the shared rollout persistence policy + /// before writing durable replay history or any implementation-owned projections. pub items: Vec, } @@ -297,6 +300,11 @@ pub struct StoredTurn { pub turn_id: String, /// Persisted rollout items associated with this turn, according to `items_view`. pub items: Vec, + /// Opaque serialized turn metadata supplied by a projected durable store. + pub metadata_json: Option>, + /// Semantic turn creation timestamp in milliseconds, when supplied by a projected durable + /// store. + pub turn_created_at_ms: Option, /// Amount of item detail included in `items`. pub items_view: StoredTurnItemsView, /// Store-owned status for API layer projection. @@ -339,11 +347,21 @@ pub struct ListItemsParams { pub sort_direction: SortDirection, } -/// A page of persisted rollout items within a turn. +/// A projected app-server `ThreadItem` snapshot within a turn. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct StoredThreadItem { + pub turn_id: Option, + pub item_key: String, + pub item_ordinal: u64, + pub item_created_at_ms: i64, + pub materialized_thread_item_json: Vec, +} + +/// A page of persisted items within a turn. #[derive(Clone, Debug, Serialize, Deserialize)] pub struct ItemPage { /// Items returned for this page. - pub items: Vec, + pub items: Vec, /// Opaque cursor to continue listing. pub next_cursor: Option, /// Opaque cursor for fetching in the opposite direction.