feat: sqlite 1 (#10004)

Add a `.sqlite` database to be used to store rollout metatdata (and
later logs)
This PR is phase 1:
* Add the database and the required infrastructure
* Add a backfill of the database
* Persist the newly created rollout both in files and in the DB
* When we need to get metadata or a rollout, consider the `JSONL` as the
source of truth but compare the results with the DB and show any errors

codex-rs/state/src/extract.rs

@@ -0,0 +1,182 @@
+use crate::model::ThreadMetadata;
+use codex_protocol::models::ContentItem;
+use codex_protocol::models::ResponseItem;
+use codex_protocol::models::is_local_image_close_tag_text;
+use codex_protocol::models::is_local_image_open_tag_text;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::RolloutItem;
+use codex_protocol::protocol::SessionMetaLine;
+use codex_protocol::protocol::TurnContextItem;
+use codex_protocol::protocol::USER_MESSAGE_BEGIN;
+use serde::Serialize;
+use serde_json::Value;
+
+/// Apply a rollout item to the metadata structure.
+pub fn apply_rollout_item(
+    metadata: &mut ThreadMetadata,
+    item: &RolloutItem,
+    default_provider: &str,
+) {
+    match item {
+        RolloutItem::SessionMeta(meta_line) => apply_session_meta_from_item(metadata, meta_line),
+        RolloutItem::TurnContext(turn_ctx) => apply_turn_context(metadata, turn_ctx),
+        RolloutItem::EventMsg(event) => apply_event_msg(metadata, event),
+        RolloutItem::ResponseItem(item) => apply_response_item(metadata, item),
+        RolloutItem::Compacted(_) => {}
+    }
+    if metadata.model_provider.is_empty() {
+        metadata.model_provider = default_provider.to_string();
+    }
+}
+
+fn apply_session_meta_from_item(metadata: &mut ThreadMetadata, meta_line: &SessionMetaLine) {
+    metadata.id = meta_line.meta.id;
+    metadata.source = enum_to_string(&meta_line.meta.source);
+    if let Some(provider) = meta_line.meta.model_provider.as_deref() {
+        metadata.model_provider = provider.to_string();
+    }
+    if !meta_line.meta.cwd.as_os_str().is_empty() {
+        metadata.cwd = meta_line.meta.cwd.clone();
+    }
+    if let Some(git) = meta_line.git.as_ref() {
+        metadata.git_sha = git.commit_hash.clone();
+        metadata.git_branch = git.branch.clone();
+        metadata.git_origin_url = git.repository_url.clone();
+    }
+}
+
+fn apply_turn_context(metadata: &mut ThreadMetadata, turn_ctx: &TurnContextItem) {
+    metadata.cwd = turn_ctx.cwd.clone();
+    metadata.sandbox_policy = enum_to_string(&turn_ctx.sandbox_policy);
+    metadata.approval_mode = enum_to_string(&turn_ctx.approval_policy);
+}
+
+fn apply_event_msg(metadata: &mut ThreadMetadata, event: &EventMsg) {
+    match event {
+        EventMsg::TokenCount(token_count) => {
+            if let Some(info) = token_count.info.as_ref() {
+                metadata.tokens_used = info.total_token_usage.total_tokens.max(0);
+            }
+        }
+        EventMsg::UserMessage(user) => {
+            metadata.has_user_event = true;
+            if metadata.title.is_empty() {
+                metadata.title = strip_user_message_prefix(user.message.as_str()).to_string();
+            }
+        }
+        _ => {}
+    }
+}
+
+fn apply_response_item(metadata: &mut ThreadMetadata, item: &ResponseItem) {
+    if let Some(text) = extract_user_message_text(item) {
+        metadata.has_user_event = true;
+        if metadata.title.is_empty() {
+            metadata.title = text;
+        }
+    }
+}
+
+fn extract_user_message_text(item: &ResponseItem) -> Option<String> {
+    let ResponseItem::Message { role, content, .. } = item else {
+        return None;
+    };
+    if role != "user" {
+        return None;
+    }
+    let texts: Vec<&str> = content
+        .iter()
+        .filter_map(|content_item| match content_item {
+            ContentItem::InputText { text } => Some(text.as_str()),
+            ContentItem::InputImage { .. } | ContentItem::OutputText { .. } => None,
+        })
+        .filter(|text| !is_local_image_open_tag_text(text) && !is_local_image_close_tag_text(text))
+        .collect();
+    if texts.is_empty() {
+        return None;
+    }
+    let joined = texts.join("\n");
+    Some(
+        strip_user_message_prefix(joined.as_str())
+            .trim()
+            .to_string(),
+    )
+}
+
+fn strip_user_message_prefix(text: &str) -> &str {
+    match text.find(USER_MESSAGE_BEGIN) {
+        Some(idx) => text[idx + USER_MESSAGE_BEGIN.len()..].trim(),
+        None => text.trim(),
+    }
+}
+
+pub(crate) fn enum_to_string<T: Serialize>(value: &T) -> String {
+    match serde_json::to_value(value) {
+        Ok(Value::String(s)) => s,
+        Ok(other) => other.to_string(),
+        Err(_) => String::new(),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::extract_user_message_text;
+    use crate::model::ThreadMetadata;
+    use chrono::DateTime;
+    use chrono::Utc;
+    use codex_protocol::ThreadId;
+    use codex_protocol::models::ContentItem;
+    use codex_protocol::models::ResponseItem;
+    use codex_protocol::protocol::USER_MESSAGE_BEGIN;
+    use pretty_assertions::assert_eq;
+    use std::path::PathBuf;
+    use uuid::Uuid;
+
+    #[test]
+    fn extracts_user_message_text() {
+        let item = ResponseItem::Message {
+            id: None,
+            role: "user".to_string(),
+            content: vec![
+                ContentItem::InputText {
+                    text: format!("<prior context> {USER_MESSAGE_BEGIN}actual question"),
+                },
+                ContentItem::InputImage {
+                    image_url: "https://example.com/image.png".to_string(),
+                },
+            ],
+            end_turn: None,
+        };
+        let actual = extract_user_message_text(&item);
+        assert_eq!(actual.as_deref(), Some("actual question"));
+    }
+
+    #[test]
+    fn diff_fields_detects_changes() {
+        let id = ThreadId::from_string(&Uuid::now_v7().to_string()).expect("thread id");
+        let created_at = DateTime::<Utc>::from_timestamp(1_735_689_600, 0).expect("timestamp");
+        let base = ThreadMetadata {
+            id,
+            rollout_path: PathBuf::from("/tmp/a.jsonl"),
+            created_at,
+            updated_at: created_at,
+            source: "cli".to_string(),
+            model_provider: "openai".to_string(),
+            cwd: PathBuf::from("/tmp"),
+            title: "hello".to_string(),
+            sandbox_policy: "read-only".to_string(),
+            approval_mode: "on-request".to_string(),
+            tokens_used: 1,
+            has_user_event: false,
+            archived_at: None,
+            git_sha: None,
+            git_branch: None,
+            git_origin_url: None,
+        };
+        let mut other = base.clone();
+        other.tokens_used = 2;
+        other.title = "world".to_string();
+        let diffs = base.diff_fields(&other);
+        assert_eq!(diffs, vec!["title", "tokens_used"]);
+    }
+}

codex-rs/state/src/lib.rs

@@ -0,0 +1,34 @@
+//! SQLite-backed state for rollout metadata.
+//!
+//! This crate is intentionally small and focused: it extracts rollout metadata
+//! from JSONL rollouts and mirrors it into a local SQLite database. Backfill
+//! orchestration and rollout scanning live in `codex-core`.
+
+mod extract;
+mod migrations;
+mod model;
+mod paths;
+mod runtime;
+
+/// Preferred entrypoint: owns configuration and metrics.
+pub use runtime::StateRuntime;
+
+/// Low-level storage engine: useful for focused tests.
+///
+/// Most consumers should prefer [`StateRuntime`].
+pub use extract::apply_rollout_item;
+pub use model::Anchor;
+pub use model::BackfillStats;
+pub use model::ExtractionOutcome;
+pub use model::SortKey;
+pub use model::ThreadMetadata;
+pub use model::ThreadMetadataBuilder;
+pub use model::ThreadsPage;
+pub use runtime::STATE_DB_FILENAME;
+
+/// Errors encountered during DB operations. Tags: [stage]
+pub const DB_ERROR_METRIC: &str = "codex.db.error";
+/// Metrics on backfill process during first init of the db. Tags: [status]
+pub const DB_METRIC_BACKFILL: &str = "codex.db.backfill";
+/// Metrics on errors during comparison between DB and rollout file. Tags: [stage]
+pub const DB_METRIC_COMPARE_ERROR: &str = "codex.db.compare_error";

codex-rs/state/src/migrations.rs

@@ -0,0 +1,3 @@
+use sqlx::migrate::Migrator;
+
+pub(crate) static MIGRATOR: Migrator = sqlx::migrate!("./migrations");

codex-rs/state/src/model.rs

@@ -0,0 +1,352 @@
+use anyhow::Result;
+use chrono::DateTime;
+use chrono::Timelike;
+use chrono::Utc;
+use codex_protocol::ThreadId;
+use codex_protocol::protocol::AskForApproval;
+use codex_protocol::protocol::SandboxPolicy;
+use codex_protocol::protocol::SessionSource;
+use sqlx::Row;
+use sqlx::sqlite::SqliteRow;
+use std::path::PathBuf;
+use uuid::Uuid;
+
+/// The sort key to use when listing threads.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum SortKey {
+    /// Sort by the thread's creation timestamp.
+    CreatedAt,
+    /// Sort by the thread's last update timestamp.
+    UpdatedAt,
+}
+
+/// A pagination anchor used for keyset pagination.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Anchor {
+    /// The timestamp component of the anchor.
+    pub ts: DateTime<Utc>,
+    /// The UUID component of the anchor.
+    pub id: Uuid,
+}
+
+/// A single page of thread metadata results.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ThreadsPage {
+    /// The thread metadata items in this page.
+    pub items: Vec<ThreadMetadata>,
+    /// The next anchor to use for pagination, if any.
+    pub next_anchor: Option<Anchor>,
+    /// The number of rows scanned to produce this page.
+    pub num_scanned_rows: usize,
+}
+
+/// The outcome of extracting metadata from a rollout.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ExtractionOutcome {
+    /// The extracted thread metadata.
+    pub metadata: ThreadMetadata,
+    /// The number of rollout lines that failed to parse.
+    pub parse_errors: usize,
+}
+
+/// Canonical thread metadata derived from rollout files.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ThreadMetadata {
+    /// The thread identifier.
+    pub id: ThreadId,
+    /// The absolute rollout path on disk.
+    pub rollout_path: PathBuf,
+    /// The creation timestamp.
+    pub created_at: DateTime<Utc>,
+    /// The last update timestamp.
+    pub updated_at: DateTime<Utc>,
+    /// The session source (stringified enum).
+    pub source: String,
+    /// The model provider identifier.
+    pub model_provider: String,
+    /// The working directory for the thread.
+    pub cwd: PathBuf,
+    /// A best-effort thread title.
+    pub title: String,
+    /// The sandbox policy (stringified enum).
+    pub sandbox_policy: String,
+    /// The approval mode (stringified enum).
+    pub approval_mode: String,
+    /// The last observed token usage.
+    pub tokens_used: i64,
+    /// Whether the thread has observed a user message.
+    pub has_user_event: bool,
+    /// The archive timestamp, if the thread is archived.
+    pub archived_at: Option<DateTime<Utc>>,
+    /// The git commit SHA, if known.
+    pub git_sha: Option<String>,
+    /// The git branch name, if known.
+    pub git_branch: Option<String>,
+    /// The git origin URL, if known.
+    pub git_origin_url: Option<String>,
+}
+
+/// Builder data required to construct [`ThreadMetadata`] without parsing filenames.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ThreadMetadataBuilder {
+    /// The thread identifier.
+    pub id: ThreadId,
+    /// The absolute rollout path on disk.
+    pub rollout_path: PathBuf,
+    /// The creation timestamp.
+    pub created_at: DateTime<Utc>,
+    /// The last update timestamp, if known.
+    pub updated_at: Option<DateTime<Utc>>,
+    /// The session source.
+    pub source: SessionSource,
+    /// The model provider identifier, if known.
+    pub model_provider: Option<String>,
+    /// The working directory for the thread.
+    pub cwd: PathBuf,
+    /// The sandbox policy.
+    pub sandbox_policy: SandboxPolicy,
+    /// The approval mode.
+    pub approval_mode: AskForApproval,
+    /// The archive timestamp, if the thread is archived.
+    pub archived_at: Option<DateTime<Utc>>,
+    /// The git commit SHA, if known.
+    pub git_sha: Option<String>,
+    /// The git branch name, if known.
+    pub git_branch: Option<String>,
+    /// The git origin URL, if known.
+    pub git_origin_url: Option<String>,
+}
+
+impl ThreadMetadataBuilder {
+    /// Create a new builder with required fields and sensible defaults.
+    pub fn new(
+        id: ThreadId,
+        rollout_path: PathBuf,
+        created_at: DateTime<Utc>,
+        source: SessionSource,
+    ) -> Self {
+        Self {
+            id,
+            rollout_path,
+            created_at,
+            updated_at: None,
+            source,
+            model_provider: None,
+            cwd: PathBuf::new(),
+            sandbox_policy: SandboxPolicy::ReadOnly,
+            approval_mode: AskForApproval::OnRequest,
+            archived_at: None,
+            git_sha: None,
+            git_branch: None,
+            git_origin_url: None,
+        }
+    }
+
+    /// Build canonical thread metadata, filling missing values from defaults.
+    pub fn build(&self, default_provider: &str) -> ThreadMetadata {
+        let source = crate::extract::enum_to_string(&self.source);
+        let sandbox_policy = crate::extract::enum_to_string(&self.sandbox_policy);
+        let approval_mode = crate::extract::enum_to_string(&self.approval_mode);
+        let created_at = canonicalize_datetime(self.created_at);
+        let updated_at = self
+            .updated_at
+            .map(canonicalize_datetime)
+            .unwrap_or(created_at);
+        ThreadMetadata {
+            id: self.id,
+            rollout_path: self.rollout_path.clone(),
+            created_at,
+            updated_at,
+            source,
+            model_provider: self
+                .model_provider
+                .clone()
+                .unwrap_or_else(|| default_provider.to_string()),
+            cwd: self.cwd.clone(),
+            title: String::new(),
+            sandbox_policy,
+            approval_mode,
+            tokens_used: 0,
+            has_user_event: false,
+            archived_at: self.archived_at.map(canonicalize_datetime),
+            git_sha: self.git_sha.clone(),
+            git_branch: self.git_branch.clone(),
+            git_origin_url: self.git_origin_url.clone(),
+        }
+    }
+}
+
+impl ThreadMetadata {
+    /// Return the list of field names that differ between `self` and `other`.
+    pub fn diff_fields(&self, other: &Self) -> Vec<&'static str> {
+        let mut diffs = Vec::new();
+        if self.id != other.id {
+            diffs.push("id");
+        }
+        if self.rollout_path != other.rollout_path {
+            diffs.push("rollout_path");
+        }
+        if self.created_at != other.created_at {
+            diffs.push("created_at");
+        }
+        if self.updated_at != other.updated_at {
+            diffs.push("updated_at");
+        }
+        if self.source != other.source {
+            diffs.push("source");
+        }
+        if self.model_provider != other.model_provider {
+            diffs.push("model_provider");
+        }
+        if self.cwd != other.cwd {
+            diffs.push("cwd");
+        }
+        if self.title != other.title {
+            diffs.push("title");
+        }
+        if self.sandbox_policy != other.sandbox_policy {
+            diffs.push("sandbox_policy");
+        }
+        if self.approval_mode != other.approval_mode {
+            diffs.push("approval_mode");
+        }
+        if self.tokens_used != other.tokens_used {
+            diffs.push("tokens_used");
+        }
+        if self.has_user_event != other.has_user_event {
+            diffs.push("has_user_event");
+        }
+        if self.archived_at != other.archived_at {
+            diffs.push("archived_at");
+        }
+        if self.git_sha != other.git_sha {
+            diffs.push("git_sha");
+        }
+        if self.git_branch != other.git_branch {
+            diffs.push("git_branch");
+        }
+        if self.git_origin_url != other.git_origin_url {
+            diffs.push("git_origin_url");
+        }
+        diffs
+    }
+}
+
+fn canonicalize_datetime(dt: DateTime<Utc>) -> DateTime<Utc> {
+    dt.with_nanosecond(0).unwrap_or(dt)
+}
+
+#[derive(Debug)]
+pub(crate) struct ThreadRow {
+    id: String,
+    rollout_path: String,
+    created_at: i64,
+    updated_at: i64,
+    source: String,
+    model_provider: String,
+    cwd: String,
+    title: String,
+    sandbox_policy: String,
+    approval_mode: String,
+    tokens_used: i64,
+    has_user_event: bool,
+    archived_at: Option<i64>,
+    git_sha: Option<String>,
+    git_branch: Option<String>,
+    git_origin_url: Option<String>,
+}
+
+impl ThreadRow {
+    pub(crate) fn try_from_row(row: &SqliteRow) -> Result<Self> {
+        Ok(Self {
+            id: row.try_get("id")?,
+            rollout_path: row.try_get("rollout_path")?,
+            created_at: row.try_get("created_at")?,
+            updated_at: row.try_get("updated_at")?,
+            source: row.try_get("source")?,
+            model_provider: row.try_get("model_provider")?,
+            cwd: row.try_get("cwd")?,
+            title: row.try_get("title")?,
+            sandbox_policy: row.try_get("sandbox_policy")?,
+            approval_mode: row.try_get("approval_mode")?,
+            tokens_used: row.try_get("tokens_used")?,
+            has_user_event: row.try_get("has_user_event")?,
+            archived_at: row.try_get("archived_at")?,
+            git_sha: row.try_get("git_sha")?,
+            git_branch: row.try_get("git_branch")?,
+            git_origin_url: row.try_get("git_origin_url")?,
+        })
+    }
+}
+
+impl TryFrom<ThreadRow> for ThreadMetadata {
+    type Error = anyhow::Error;
+
+    fn try_from(row: ThreadRow) -> std::result::Result<Self, Self::Error> {
+        let ThreadRow {
+            id,
+            rollout_path,
+            created_at,
+            updated_at,
+            source,
+            model_provider,
+            cwd,
+            title,
+            sandbox_policy,
+            approval_mode,
+            tokens_used,
+            has_user_event,
+            archived_at,
+            git_sha,
+            git_branch,
+            git_origin_url,
+        } = row;
+        Ok(Self {
+            id: ThreadId::try_from(id)?,
+            rollout_path: PathBuf::from(rollout_path),
+            created_at: epoch_seconds_to_datetime(created_at)?,
+            updated_at: epoch_seconds_to_datetime(updated_at)?,
+            source,
+            model_provider,
+            cwd: PathBuf::from(cwd),
+            title,
+            sandbox_policy,
+            approval_mode,
+            tokens_used,
+            has_user_event,
+            archived_at: archived_at.map(epoch_seconds_to_datetime).transpose()?,
+            git_sha,
+            git_branch,
+            git_origin_url,
+        })
+    }
+}
+
+pub(crate) fn anchor_from_item(item: &ThreadMetadata, sort_key: SortKey) -> Option<Anchor> {
+    let id = Uuid::parse_str(&item.id.to_string()).ok()?;
+    let ts = match sort_key {
+        SortKey::CreatedAt => item.created_at,
+        SortKey::UpdatedAt => item.updated_at,
+    };
+    Some(Anchor { ts, id })
+}
+
+pub(crate) fn datetime_to_epoch_seconds(dt: DateTime<Utc>) -> i64 {
+    dt.timestamp()
+}
+
+pub(crate) fn epoch_seconds_to_datetime(secs: i64) -> Result<DateTime<Utc>> {
+    DateTime::<Utc>::from_timestamp(secs, 0)
+        .ok_or_else(|| anyhow::anyhow!("invalid unix timestamp: {secs}"))
+}
+
+/// Statistics about a backfill operation.
+#[derive(Debug, Clone)]
+pub struct BackfillStats {
+    /// The number of rollout files scanned.
+    pub scanned: usize,
+    /// The number of rows upserted successfully.
+    pub upserted: usize,
+    /// The number of rows that failed to upsert.
+    pub failed: usize,
+}

codex-rs/state/src/paths.rs

@@ -0,0 +1,10 @@
+use chrono::DateTime;
+use chrono::Timelike;
+use chrono::Utc;
+use std::path::Path;
+
+pub(crate) async fn file_modified_time_utc(path: &Path) -> Option<DateTime<Utc>> {
+    let modified = tokio::fs::metadata(path).await.ok()?.modified().ok()?;
+    let updated_at: DateTime<Utc> = modified.into();
+    Some(updated_at.with_nanosecond(0).unwrap_or(updated_at))
+}

codex-rs/state/src/runtime.rs

@@ -0,0 +1,458 @@
+use crate::DB_ERROR_METRIC;
+use crate::SortKey;
+use crate::ThreadMetadata;
+use crate::ThreadMetadataBuilder;
+use crate::ThreadsPage;
+use crate::apply_rollout_item;
+use crate::migrations::MIGRATOR;
+use crate::model::ThreadRow;
+use crate::model::anchor_from_item;
+use crate::model::datetime_to_epoch_seconds;
+use crate::paths::file_modified_time_utc;
+use chrono::DateTime;
+use chrono::Utc;
+use codex_otel::OtelManager;
+use codex_protocol::ThreadId;
+use codex_protocol::protocol::RolloutItem;
+use sqlx::QueryBuilder;
+use sqlx::Row;
+use sqlx::Sqlite;
+use sqlx::SqlitePool;
+use sqlx::sqlite::SqliteConnectOptions;
+use sqlx::sqlite::SqliteJournalMode;
+use sqlx::sqlite::SqlitePoolOptions;
+use sqlx::sqlite::SqliteSynchronous;
+use std::path::Path;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::Duration;
+use tracing::warn;
+
+pub const STATE_DB_FILENAME: &str = "state.sqlite";
+
+const METRIC_DB_INIT: &str = "codex.db.init";
+
+#[derive(Clone)]
+pub struct StateRuntime {
+    codex_home: PathBuf,
+    default_provider: String,
+    pool: Arc<sqlx::SqlitePool>,
+}
+
+impl StateRuntime {
+    /// Initialize the state runtime using the provided Codex home and default provider.
+    ///
+    /// This opens (and migrates) the SQLite database at `codex_home/state.sqlite`.
+    pub async fn init(
+        codex_home: PathBuf,
+        default_provider: String,
+        otel: Option<OtelManager>,
+    ) -> anyhow::Result<Arc<Self>> {
+        tokio::fs::create_dir_all(&codex_home).await?;
+        let state_path = codex_home.join(STATE_DB_FILENAME);
+        let existed = tokio::fs::try_exists(&state_path).await.unwrap_or(false);
+        let pool = match open_sqlite(&state_path).await {
+            Ok(db) => Arc::new(db),
+            Err(err) => {
+                warn!("failed to open state db at {}: {err}", state_path.display());
+                if let Some(otel) = otel.as_ref() {
+                    otel.counter(METRIC_DB_INIT, 1, &[("status", "open_error")]);
+                }
+                return Err(err);
+            }
+        };
+        if let Some(otel) = otel.as_ref() {
+            otel.counter(METRIC_DB_INIT, 1, &[("status", "opened")]);
+        }
+        let runtime = Arc::new(Self {
+            pool,
+            codex_home,
+            default_provider,
+        });
+        if !existed && let Some(otel) = otel.as_ref() {
+            otel.counter(METRIC_DB_INIT, 1, &[("status", "created")]);
+        }
+        Ok(runtime)
+    }
+
+    /// Return the configured Codex home directory for this runtime.
+    pub fn codex_home(&self) -> &Path {
+        self.codex_home.as_path()
+    }
+
+    /// Load thread metadata by id using the underlying database.
+    pub async fn get_thread(&self, id: ThreadId) -> anyhow::Result<Option<crate::ThreadMetadata>> {
+        let row = sqlx::query(
+            r#"
+SELECT
+    id,
+    rollout_path,
+    created_at,
+    updated_at,
+    source,
+    model_provider,
+    cwd,
+    title,
+    sandbox_policy,
+    approval_mode,
+    tokens_used,
+    has_user_event,
+    archived_at,
+    git_sha,
+    git_branch,
+    git_origin_url
+FROM threads
+WHERE id = ?
+            "#,
+        )
+        .bind(id.to_string())
+        .fetch_optional(self.pool.as_ref())
+        .await?;
+        row.map(|row| ThreadRow::try_from_row(&row).and_then(ThreadMetadata::try_from))
+            .transpose()
+    }
+
+    /// Find a rollout path by thread id using the underlying database.
+    pub async fn find_rollout_path_by_id(
+        &self,
+        id: ThreadId,
+        archived_only: Option<bool>,
+    ) -> anyhow::Result<Option<PathBuf>> {
+        let mut builder =
+            QueryBuilder::<Sqlite>::new("SELECT rollout_path FROM threads WHERE id = ");
+        builder.push_bind(id.to_string());
+        match archived_only {
+            Some(true) => {
+                builder.push(" AND archived = 1");
+            }
+            Some(false) => {
+                builder.push(" AND archived = 0");
+            }
+            None => {}
+        }
+        let row = builder.build().fetch_optional(self.pool.as_ref()).await?;
+        Ok(row
+            .and_then(|r| r.try_get::<String, _>("rollout_path").ok())
+            .map(PathBuf::from))
+    }
+
+    /// List threads using the underlying database.
+    pub async fn list_threads(
+        &self,
+        page_size: usize,
+        anchor: Option<&crate::Anchor>,
+        sort_key: crate::SortKey,
+        allowed_sources: &[String],
+        model_providers: Option<&[String]>,
+        archived_only: bool,
+    ) -> anyhow::Result<crate::ThreadsPage> {
+        let limit = page_size.saturating_add(1);
+
+        let mut builder = QueryBuilder::<Sqlite>::new(
+            r#"
+SELECT
+    id,
+    rollout_path,
+    created_at,
+    updated_at,
+    source,
+    model_provider,
+    cwd,
+    title,
+    sandbox_policy,
+    approval_mode,
+    tokens_used,
+    has_user_event,
+    archived_at,
+    git_sha,
+    git_branch,
+    git_origin_url
+FROM threads
+            "#,
+        );
+        push_thread_filters(
+            &mut builder,
+            archived_only,
+            allowed_sources,
+            model_providers,
+            anchor,
+            sort_key,
+        );
+        push_thread_order_and_limit(&mut builder, sort_key, limit);
+
+        let rows = builder.build().fetch_all(self.pool.as_ref()).await?;
+        let mut items = rows
+            .into_iter()
+            .map(|row| ThreadRow::try_from_row(&row).and_then(ThreadMetadata::try_from))
+            .collect::<Result<Vec<_>, _>>()?;
+        let num_scanned_rows = items.len();
+        let next_anchor = if items.len() > page_size {
+            items.pop();
+            items
+                .last()
+                .and_then(|item| anchor_from_item(item, sort_key))
+        } else {
+            None
+        };
+        Ok(ThreadsPage {
+            items,
+            next_anchor,
+            num_scanned_rows,
+        })
+    }
+
+    /// List thread ids using the underlying database (no rollout scanning).
+    pub async fn list_thread_ids(
+        &self,
+        limit: usize,
+        anchor: Option<&crate::Anchor>,
+        sort_key: crate::SortKey,
+        allowed_sources: &[String],
+        model_providers: Option<&[String]>,
+        archived_only: bool,
+    ) -> anyhow::Result<Vec<ThreadId>> {
+        let mut builder = QueryBuilder::<Sqlite>::new("SELECT id FROM threads");
+        push_thread_filters(
+            &mut builder,
+            archived_only,
+            allowed_sources,
+            model_providers,
+            anchor,
+            sort_key,
+        );
+        push_thread_order_and_limit(&mut builder, sort_key, limit);
+
+        let rows = builder.build().fetch_all(self.pool.as_ref()).await?;
+        rows.into_iter()
+            .map(|row| {
+                let id: String = row.try_get("id")?;
+                Ok(ThreadId::try_from(id)?)
+            })
+            .collect()
+    }
+
+    /// Insert or replace thread metadata directly.
+    pub async fn upsert_thread(&self, metadata: &crate::ThreadMetadata) -> anyhow::Result<()> {
+        sqlx::query(
+            r#"
+INSERT INTO threads (
+    id,
+    rollout_path,
+    created_at,
+    updated_at,
+    source,
+    model_provider,
+    cwd,
+    title,
+    sandbox_policy,
+    approval_mode,
+    tokens_used,
+    has_user_event,
+    archived,
+    archived_at,
+    git_sha,
+    git_branch,
+    git_origin_url
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+ON CONFLICT(id) DO UPDATE SET
+    rollout_path = excluded.rollout_path,
+    created_at = excluded.created_at,
+    updated_at = excluded.updated_at,
+    source = excluded.source,
+    model_provider = excluded.model_provider,
+    cwd = excluded.cwd,
+    title = excluded.title,
+    sandbox_policy = excluded.sandbox_policy,
+    approval_mode = excluded.approval_mode,
+    tokens_used = excluded.tokens_used,
+    has_user_event = excluded.has_user_event,
+    archived = excluded.archived,
+    archived_at = excluded.archived_at,
+    git_sha = excluded.git_sha,
+    git_branch = excluded.git_branch,
+    git_origin_url = excluded.git_origin_url
+            "#,
+        )
+        .bind(metadata.id.to_string())
+        .bind(metadata.rollout_path.display().to_string())
+        .bind(datetime_to_epoch_seconds(metadata.created_at))
+        .bind(datetime_to_epoch_seconds(metadata.updated_at))
+        .bind(metadata.source.as_str())
+        .bind(metadata.model_provider.as_str())
+        .bind(metadata.cwd.display().to_string())
+        .bind(metadata.title.as_str())
+        .bind(metadata.sandbox_policy.as_str())
+        .bind(metadata.approval_mode.as_str())
+        .bind(metadata.tokens_used)
+        .bind(metadata.has_user_event)
+        .bind(metadata.archived_at.is_some())
+        .bind(metadata.archived_at.map(datetime_to_epoch_seconds))
+        .bind(metadata.git_sha.as_deref())
+        .bind(metadata.git_branch.as_deref())
+        .bind(metadata.git_origin_url.as_deref())
+        .execute(self.pool.as_ref())
+        .await?;
+        Ok(())
+    }
+
+    /// Apply rollout items incrementally using the underlying database.
+    pub async fn apply_rollout_items(
+        &self,
+        builder: &ThreadMetadataBuilder,
+        items: &[RolloutItem],
+        otel: Option<&OtelManager>,
+    ) -> anyhow::Result<()> {
+        if items.is_empty() {
+            return Ok(());
+        }
+        let mut metadata = self
+            .get_thread(builder.id)
+            .await?
+            .unwrap_or_else(|| builder.build(&self.default_provider));
+        metadata.rollout_path = builder.rollout_path.clone();
+        for item in items {
+            apply_rollout_item(&mut metadata, item, &self.default_provider);
+        }
+        if let Some(updated_at) = file_modified_time_utc(builder.rollout_path.as_path()).await {
+            metadata.updated_at = updated_at;
+        }
+        if let Err(err) = self.upsert_thread(&metadata).await {
+            if let Some(otel) = otel {
+                otel.counter(DB_ERROR_METRIC, 1, &[("stage", "apply_rollout_items")]);
+            }
+            return Err(err);
+        }
+        Ok(())
+    }
+
+    /// Mark a thread as archived using the underlying database.
+    pub async fn mark_archived(
+        &self,
+        thread_id: ThreadId,
+        rollout_path: &Path,
+        archived_at: DateTime<Utc>,
+    ) -> anyhow::Result<()> {
+        let Some(mut metadata) = self.get_thread(thread_id).await? else {
+            return Ok(());
+        };
+        metadata.archived_at = Some(archived_at);
+        metadata.rollout_path = rollout_path.to_path_buf();
+        if let Some(updated_at) = file_modified_time_utc(rollout_path).await {
+            metadata.updated_at = updated_at;
+        }
+        if metadata.id != thread_id {
+            warn!(
+                "thread id mismatch during archive: expected {thread_id}, got {}",
+                metadata.id
+            );
+        }
+        self.upsert_thread(&metadata).await
+    }
+
+    /// Mark a thread as unarchived using the underlying database.
+    pub async fn mark_unarchived(
+        &self,
+        thread_id: ThreadId,
+        rollout_path: &Path,
+    ) -> anyhow::Result<()> {
+        let Some(mut metadata) = self.get_thread(thread_id).await? else {
+            return Ok(());
+        };
+        metadata.archived_at = None;
+        metadata.rollout_path = rollout_path.to_path_buf();
+        if let Some(updated_at) = file_modified_time_utc(rollout_path).await {
+            metadata.updated_at = updated_at;
+        }
+        if metadata.id != thread_id {
+            warn!(
+                "thread id mismatch during unarchive: expected {thread_id}, got {}",
+                metadata.id
+            );
+        }
+        self.upsert_thread(&metadata).await
+    }
+}
+
+async fn open_sqlite(path: &Path) -> anyhow::Result<SqlitePool> {
+    let options = SqliteConnectOptions::new()
+        .filename(path)
+        .create_if_missing(true)
+        .journal_mode(SqliteJournalMode::Wal)
+        .synchronous(SqliteSynchronous::Normal)
+        .busy_timeout(Duration::from_secs(5));
+    let pool = SqlitePoolOptions::new()
+        .max_connections(5)
+        .connect_with(options)
+        .await?;
+    MIGRATOR.run(&pool).await?;
+    Ok(pool)
+}
+
+fn push_thread_filters<'a>(
+    builder: &mut QueryBuilder<'a, Sqlite>,
+    archived_only: bool,
+    allowed_sources: &'a [String],
+    model_providers: Option<&'a [String]>,
+    anchor: Option<&crate::Anchor>,
+    sort_key: SortKey,
+) {
+    builder.push(" WHERE 1 = 1");
+    if archived_only {
+        builder.push(" AND archived = 1");
+    } else {
+        builder.push(" AND archived = 0");
+    }
+    builder.push(" AND has_user_event = 1");
+    if !allowed_sources.is_empty() {
+        builder.push(" AND source IN (");
+        let mut separated = builder.separated(", ");
+        for source in allowed_sources {
+            separated.push_bind(source);
+        }
+        separated.push_unseparated(")");
+    }
+    if let Some(model_providers) = model_providers
+        && !model_providers.is_empty()
+    {
+        builder.push(" AND model_provider IN (");
+        let mut separated = builder.separated(", ");
+        for provider in model_providers {
+            separated.push_bind(provider);
+        }
+        separated.push_unseparated(")");
+    }
+    if let Some(anchor) = anchor {
+        let anchor_ts = datetime_to_epoch_seconds(anchor.ts);
+        let column = match sort_key {
+            SortKey::CreatedAt => "created_at",
+            SortKey::UpdatedAt => "updated_at",
+        };
+        builder.push(" AND (");
+        builder.push(column);
+        builder.push(" < ");
+        builder.push_bind(anchor_ts);
+        builder.push(" OR (");
+        builder.push(column);
+        builder.push(" = ");
+        builder.push_bind(anchor_ts);
+        builder.push(" AND id < ");
+        builder.push_bind(anchor.id.to_string());
+        builder.push("))");
+    }
+}
+
+fn push_thread_order_and_limit(
+    builder: &mut QueryBuilder<'_, Sqlite>,
+    sort_key: SortKey,
+    limit: usize,
+) {
+    let order_column = match sort_key {
+        SortKey::CreatedAt => "created_at",
+        SortKey::UpdatedAt => "updated_at",
+    };
+    builder.push(" ORDER BY ");
+    builder.push(order_column);
+    builder.push(" DESC, id DESC");
+    builder.push(" LIMIT ");
+    builder.push_bind(limit as i64);
+}