ThreadStore interface (#17659)

Introduce a ThreadStore interface for mediating access to the filesystem
(rollout jsonl files + sqlite db) based thread storage.

In later PRs we'll move the existing fs code behind a "local"
implementation of this ThreadStore interface.

This PR should be a no-op behaviorally, it only introduces the
interface.

codex-rs/Cargo.lock

@@ -2824,6 +2824,17 @@ dependencies = [
  "tempfile",
 ]
 
+[[package]]
+name = "codex-thread-store"
+version = "0.0.0"
+dependencies = [
+ "async-trait",
+ "chrono",
+ "codex-protocol",
+ "serde",
+ "thiserror 2.0.18",
+]
+
 [[package]]
 name = "codex-tools"
 version = "0.0.0"

codex-rs/Cargo.toml

@@ -87,6 +87,7 @@ members = [
     "state",
     "terminal-detection",
     "test-binary-support",
+    "thread-store",
     "codex-experimental-api-macros",
     "plugin",
 ]

codex-rs/thread-store/BUILD.bazel

@@ -0,0 +1,6 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "thread-store",
+    crate_name = "codex_thread_store",
+)

codex-rs/thread-store/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+edition.workspace = true
+license.workspace = true
+name = "codex-thread-store"
+version.workspace = true
+
+[lib]
+name = "codex_thread_store"
+path = "src/lib.rs"
+
+[lints]
+workspace = true
+
+[dependencies]
+async-trait = { workspace = true }
+chrono = { workspace = true, features = ["serde"] }
+codex-protocol = { workspace = true }
+serde = { workspace = true, features = ["derive"] }
+thiserror = { workspace = true }

codex-rs/thread-store/src/error.rs

@@ -0,0 +1,36 @@
+use codex_protocol::ThreadId;
+
+/// Result type returned by thread-store operations.
+pub type ThreadStoreResult<T> = Result<T, ThreadStoreError>;
+
+/// Error type shared by thread-store implementations.
+#[derive(Debug, thiserror::Error)]
+pub enum ThreadStoreError {
+    /// The requested thread does not exist in this store.
+    #[error("thread {thread_id} not found")]
+    ThreadNotFound {
+        /// Thread id requested by the caller.
+        thread_id: ThreadId,
+    },
+
+    /// The caller supplied invalid request data.
+    #[error("invalid thread-store request: {message}")]
+    InvalidRequest {
+        /// User-facing explanation of the invalid request.
+        message: String,
+    },
+
+    /// The operation conflicted with current store state.
+    #[error("thread-store conflict: {message}")]
+    Conflict {
+        /// User-facing explanation of the conflict.
+        message: String,
+    },
+
+    /// Catch-all for implementation failures that do not fit a more specific category.
+    #[error("thread-store internal error: {message}")]
+    Internal {
+        /// User-facing explanation of the implementation failure.
+        message: String,
+    },
+}

codex-rs/thread-store/src/lib.rs

@@ -0,0 +1,32 @@
+//! Storage-neutral thread persistence interfaces.
+//!
+//! Application code should treat [`codex_protocol::ThreadId`] as the only durable thread handle.
+//! Implementations are responsible for resolving that id to local rollout files, RPC requests, or
+//! any other backing store.
+
+mod error;
+mod recorder;
+mod store;
+mod types;
+
+pub use error::ThreadStoreError;
+pub use error::ThreadStoreResult;
+pub use recorder::ThreadRecorder;
+pub use store::ThreadStore;
+pub use types::AppendThreadItemsParams;
+pub use types::ArchiveThreadParams;
+pub use types::CreateThreadParams;
+pub use types::GitInfoPatch;
+pub use types::ListThreadsParams;
+pub use types::LoadThreadHistoryParams;
+pub use types::OptionalStringPatch;
+pub use types::ReadThreadParams;
+pub use types::ResumeThreadRecorderParams;
+pub use types::SetThreadNameParams;
+pub use types::StoredThread;
+pub use types::StoredThreadHistory;
+pub use types::ThreadEventPersistenceMode;
+pub use types::ThreadMetadataPatch;
+pub use types::ThreadPage;
+pub use types::ThreadSortKey;
+pub use types::UpdateThreadMetadataParams;

codex-rs/thread-store/src/recorder.rs

@@ -0,0 +1,28 @@
+use async_trait::async_trait;
+use codex_protocol::ThreadId;
+use codex_protocol::protocol::RolloutItem;
+
+use crate::ThreadStoreResult;
+
+/// Live append handle for a thread.
+///
+/// This is the storage-neutral version of the existing rollout recorder API. The local
+/// implementation is expected to wrap `codex_rollout::RolloutRecorder` and preserve its lazy
+/// materialization, filtering, flush, and shutdown behavior.
+#[async_trait]
+pub trait ThreadRecorder: Send + Sync {
+    /// Returns the thread id this recorder appends to.
+    fn thread_id(&self) -> ThreadId;
+
+    /// Queues items for persistence according to this recorder's filtering policy.
+    async fn record_items(&self, items: &[RolloutItem]) -> ThreadStoreResult<()>;
+
+    /// Materializes the thread if persistence is lazy, then persists all queued items.
+    async fn persist(&self) -> ThreadStoreResult<()>;
+
+    /// Flushes all queued items and returns once they are durable/readable.
+    async fn flush(&self) -> ThreadStoreResult<()>;
+
+    /// Flushes pending items and closes the recorder.
+    async fn shutdown(&self) -> ThreadStoreResult<()>;
+}

codex-rs/thread-store/src/store.rs

@@ -0,0 +1,65 @@
+use async_trait::async_trait;
+
+use crate::AppendThreadItemsParams;
+use crate::ArchiveThreadParams;
+use crate::CreateThreadParams;
+use crate::ListThreadsParams;
+use crate::LoadThreadHistoryParams;
+use crate::ReadThreadParams;
+use crate::ResumeThreadRecorderParams;
+use crate::SetThreadNameParams;
+use crate::StoredThread;
+use crate::StoredThreadHistory;
+use crate::ThreadPage;
+use crate::ThreadRecorder;
+use crate::ThreadStoreResult;
+use crate::UpdateThreadMetadataParams;
+
+/// Storage-neutral thread persistence boundary.
+#[async_trait]
+pub trait ThreadStore: Send + Sync {
+    /// Creates a new thread and returns a live recorder for future appends.
+    async fn create_thread(
+        &self,
+        params: CreateThreadParams,
+    ) -> ThreadStoreResult<Box<dyn ThreadRecorder>>;
+
+    /// Reopens a live recorder for an existing thread.
+    async fn resume_thread_recorder(
+        &self,
+        params: ResumeThreadRecorderParams,
+    ) -> ThreadStoreResult<Box<dyn ThreadRecorder>>;
+
+    /// Appends items to a stored thread outside the live-recorder path.
+    async fn append_items(&self, params: AppendThreadItemsParams) -> ThreadStoreResult<()>;
+
+    /// Loads persisted history for resume, fork, rollback, and memory jobs.
+    async fn load_history(
+        &self,
+        params: LoadThreadHistoryParams,
+    ) -> ThreadStoreResult<StoredThreadHistory>;
+
+    /// Reads a thread summary and optionally its persisted history.
+    async fn read_thread(&self, params: ReadThreadParams) -> ThreadStoreResult<StoredThread>;
+
+    /// Lists stored threads matching the supplied filters.
+    async fn list_threads(&self, params: ListThreadsParams) -> ThreadStoreResult<ThreadPage>;
+
+    /// Sets a user-facing thread name.
+    async fn set_thread_name(&self, params: SetThreadNameParams) -> ThreadStoreResult<()>;
+
+    /// Applies a mutable metadata patch and returns the updated thread.
+    async fn update_thread_metadata(
+        &self,
+        params: UpdateThreadMetadataParams,
+    ) -> ThreadStoreResult<StoredThread>;
+
+    /// Archives a thread.
+    async fn archive_thread(&self, params: ArchiveThreadParams) -> ThreadStoreResult<()>;
+
+    /// Unarchives a thread and returns its updated metadata.
+    async fn unarchive_thread(
+        &self,
+        params: ArchiveThreadParams,
+    ) -> ThreadStoreResult<StoredThread>;
+}

codex-rs/thread-store/src/types.rs

@@ -0,0 +1,224 @@
+use std::path::PathBuf;
+
+use chrono::DateTime;
+use chrono::Utc;
+use codex_protocol::ThreadId;
+use codex_protocol::dynamic_tools::DynamicToolSpec;
+use codex_protocol::models::BaseInstructions;
+use codex_protocol::openai_models::ReasoningEffort;
+use codex_protocol::protocol::AskForApproval;
+use codex_protocol::protocol::GitInfo;
+use codex_protocol::protocol::RolloutItem;
+use codex_protocol::protocol::SandboxPolicy;
+use codex_protocol::protocol::SessionSource;
+use codex_protocol::protocol::TokenUsage;
+use serde::Deserialize;
+use serde::Serialize;
+
+/// Controls how many event variants should be persisted for future replay.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub enum ThreadEventPersistenceMode {
+    /// Persist only the legacy minimal replay surface.
+    #[default]
+    Limited,
+    /// Persist the richer event surface used by app-server history reconstruction.
+    Extended,
+}
+
+/// Parameters required to create a persisted thread and its recorder.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct CreateThreadParams {
+    /// Thread id generated by Codex before opening persistence.
+    pub thread_id: ThreadId,
+    /// Source thread id when this thread is created as a fork.
+    pub forked_from_id: Option<ThreadId>,
+    /// Runtime source for the thread.
+    pub source: SessionSource,
+    /// Base instructions persisted in session metadata.
+    pub base_instructions: BaseInstructions,
+    /// Dynamic tools available to the thread at startup.
+    pub dynamic_tools: Vec<DynamicToolSpec>,
+    /// Whether the recorder should persist the extended event surface.
+    pub event_persistence_mode: ThreadEventPersistenceMode,
+}
+
+/// Parameters required to reopen persistence for an existing thread.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ResumeThreadRecorderParams {
+    /// Existing thread id whose future items should be appended.
+    pub thread_id: ThreadId,
+    /// Whether archived threads may be reopened.
+    pub include_archived: bool,
+    /// Whether the recorder should persist the extended event surface.
+    pub event_persistence_mode: ThreadEventPersistenceMode,
+}
+
+/// Parameters for appending rollout items outside a live recorder.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct AppendThreadItemsParams {
+    /// Thread id to append to.
+    pub thread_id: ThreadId,
+    /// Items to append in order.
+    pub items: Vec<RolloutItem>,
+}
+
+/// Parameters for loading persisted history for resume, fork, rollback, and memory jobs.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct LoadThreadHistoryParams {
+    /// Thread id to load.
+    pub thread_id: ThreadId,
+    /// Whether archived threads are eligible.
+    pub include_archived: bool,
+}
+
+/// Persisted rollout history for a thread, without any filesystem path requirement.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct StoredThreadHistory {
+    /// Thread id represented by the history.
+    pub thread_id: ThreadId,
+    /// Persisted rollout items in replay order.
+    pub items: Vec<RolloutItem>,
+}
+
+/// Parameters for reading a thread summary and optionally its replay history.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ReadThreadParams {
+    /// Thread id to read.
+    pub thread_id: ThreadId,
+    /// Whether archived threads are eligible.
+    pub include_archived: bool,
+    /// Whether persisted rollout items should be included in the response.
+    pub include_history: bool,
+}
+
+/// The sort key to use when listing stored threads.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub enum ThreadSortKey {
+    /// Sort by the thread creation timestamp.
+    #[default]
+    CreatedAt,
+    /// Sort by the thread last-update timestamp.
+    UpdatedAt,
+}
+
+/// Parameters for listing threads.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ListThreadsParams {
+    /// Maximum number of threads to return.
+    pub page_size: usize,
+    /// Opaque cursor returned by a previous list call.
+    pub cursor: Option<String>,
+    /// Sort order requested by the caller.
+    pub sort_key: ThreadSortKey,
+    /// Allowed session sources. Empty means implementation default.
+    pub allowed_sources: Vec<SessionSource>,
+    /// Optional model provider filter. `None` means implementation default, while an empty vector
+    /// means all providers.
+    pub model_providers: Option<Vec<String>>,
+    /// Whether archived threads should be listed instead of active threads.
+    pub archived: bool,
+    /// Optional substring/full-text search term for thread title/preview.
+    pub search_term: Option<String>,
+}
+
+/// A page of stored thread records.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct ThreadPage {
+    /// Threads returned for this page.
+    pub items: Vec<StoredThread>,
+    /// Opaque cursor to continue listing.
+    pub next_cursor: Option<String>,
+}
+
+/// Store-owned thread metadata used by list/read/resume responses.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct StoredThread {
+    /// Thread id.
+    pub thread_id: ThreadId,
+    /// Source thread id when this thread was forked from another thread.
+    pub forked_from_id: Option<ThreadId>,
+    /// Best available user-facing preview, usually the first user message.
+    pub preview: String,
+    /// Optional user-facing thread name/title.
+    pub name: Option<String>,
+    /// Model provider id associated with the thread.
+    pub model_provider: String,
+    /// Latest observed model, if known.
+    pub model: Option<String>,
+    /// Latest observed reasoning effort, if known.
+    pub reasoning_effort: Option<ReasoningEffort>,
+    /// Thread creation timestamp.
+    pub created_at: DateTime<Utc>,
+    /// Thread last-update timestamp.
+    pub updated_at: DateTime<Utc>,
+    /// Thread archive timestamp, if archived.
+    pub archived_at: Option<DateTime<Utc>>,
+    /// Working directory captured for the thread.
+    pub cwd: PathBuf,
+    /// Runtime source for the thread.
+    pub source: SessionSource,
+    /// Optional random nickname for thread-spawn sub-agents.
+    pub agent_nickname: Option<String>,
+    /// Optional role for thread-spawn sub-agents.
+    pub agent_role: Option<String>,
+    /// Optional canonical path for thread-spawn sub-agents.
+    pub agent_path: Option<String>,
+    /// Optional Git metadata captured for the thread.
+    pub git_info: Option<GitInfo>,
+    /// Approval mode captured for the thread.
+    pub approval_mode: AskForApproval,
+    /// Sandbox policy captured for the thread.
+    pub sandbox_policy: SandboxPolicy,
+    /// Last observed token usage.
+    pub token_usage: Option<TokenUsage>,
+    /// First user message observed for this thread, if any.
+    pub first_user_message: Option<String>,
+    /// Persisted history, populated only when requested.
+    pub history: Option<StoredThreadHistory>,
+}
+
+/// Parameters for setting a user-facing thread name.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct SetThreadNameParams {
+    /// Thread id to update.
+    pub thread_id: ThreadId,
+    /// Normalized thread name.
+    pub name: String,
+}
+
+/// Optional field patch where omission leaves a value unchanged and `Some(None)` clears it.
+pub type OptionalStringPatch = Option<Option<String>>;
+
+/// Patch for thread Git metadata.
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct GitInfoPatch {
+    /// Replacement commit SHA, clear request, or no-op.
+    pub sha: OptionalStringPatch,
+    /// Replacement branch name, clear request, or no-op.
+    pub branch: OptionalStringPatch,
+    /// Replacement origin URL, clear request, or no-op.
+    pub origin_url: OptionalStringPatch,
+}
+
+/// Patch for mutable thread metadata.
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ThreadMetadataPatch {
+    /// Optional Git metadata patch.
+    pub git_info: Option<GitInfoPatch>,
+}
+
+/// Parameters for patching mutable thread metadata.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct UpdateThreadMetadataParams {
+    /// Thread id to update.
+    pub thread_id: ThreadId,
+    /// Patch to apply.
+    pub patch: ThreadMetadataPatch,
+}
+
+/// Parameters for archiving or unarchiving a thread.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ArchiveThreadParams {
+    /// Thread id to archive or unarchive.
+    pub thread_id: ThreadId,
+}