diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index ad0e627d1..b76e079dc 100644 --- a/codex-rs/Cargo.lock +++ b/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" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index 866d343fd..e4985e458 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -87,6 +87,7 @@ members = [ "state", "terminal-detection", "test-binary-support", + "thread-store", "codex-experimental-api-macros", "plugin", ] diff --git a/codex-rs/thread-store/BUILD.bazel b/codex-rs/thread-store/BUILD.bazel new file mode 100644 index 000000000..5d800f51f --- /dev/null +++ b/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", +) diff --git a/codex-rs/thread-store/Cargo.toml b/codex-rs/thread-store/Cargo.toml new file mode 100644 index 000000000..a8f22fbfe --- /dev/null +++ b/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 } diff --git a/codex-rs/thread-store/src/error.rs b/codex-rs/thread-store/src/error.rs new file mode 100644 index 000000000..c5cee9a8b --- /dev/null +++ b/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 = Result; + +/// 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, + }, +} diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs new file mode 100644 index 000000000..5525db139 --- /dev/null +++ b/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; diff --git a/codex-rs/thread-store/src/recorder.rs b/codex-rs/thread-store/src/recorder.rs new file mode 100644 index 000000000..03b02e80c --- /dev/null +++ b/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<()>; +} diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs new file mode 100644 index 000000000..e75c93df0 --- /dev/null +++ b/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>; + + /// Reopens a live recorder for an existing thread. + async fn resume_thread_recorder( + &self, + params: ResumeThreadRecorderParams, + ) -> ThreadStoreResult>; + + /// 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; + + /// Reads a thread summary and optionally its persisted history. + async fn read_thread(&self, params: ReadThreadParams) -> ThreadStoreResult; + + /// Lists stored threads matching the supplied filters. + async fn list_threads(&self, params: ListThreadsParams) -> ThreadStoreResult; + + /// 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; + + /// 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; +} diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs new file mode 100644 index 000000000..73f983a46 --- /dev/null +++ b/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, + /// 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, + /// 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, +} + +/// 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, +} + +/// 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, + /// Sort order requested by the caller. + pub sort_key: ThreadSortKey, + /// Allowed session sources. Empty means implementation default. + pub allowed_sources: Vec, + /// Optional model provider filter. `None` means implementation default, while an empty vector + /// means all providers. + pub model_providers: Option>, + /// 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, +} + +/// A page of stored thread records. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct ThreadPage { + /// Threads returned for this page. + pub items: Vec, + /// Opaque cursor to continue listing. + pub next_cursor: Option, +} + +/// 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, + /// Best available user-facing preview, usually the first user message. + pub preview: String, + /// Optional user-facing thread name/title. + pub name: Option, + /// Model provider id associated with the thread. + pub model_provider: String, + /// Latest observed model, if known. + pub model: Option, + /// Latest observed reasoning effort, if known. + pub reasoning_effort: Option, + /// Thread creation timestamp. + pub created_at: DateTime, + /// Thread last-update timestamp. + pub updated_at: DateTime, + /// Thread archive timestamp, if archived. + pub archived_at: Option>, + /// 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, + /// Optional role for thread-spawn sub-agents. + pub agent_role: Option, + /// Optional canonical path for thread-spawn sub-agents. + pub agent_path: Option, + /// Optional Git metadata captured for the thread. + pub git_info: Option, + /// 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, + /// First user message observed for this thread, if any. + pub first_user_message: Option, + /// Persisted history, populated only when requested. + pub history: Option, +} + +/// 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>; + +/// 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, +} + +/// 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, +}