use anyhow::Result; use chrono::DateTime; use chrono::Utc; use codex_protocol::SanitizedGitUrl; use codex_protocol::ThreadId; use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadSource; use serde::Deserialize; use serde::Serialize; use sqlx::Row; use sqlx::sqlite::SqliteRow; use std::collections::HashMap; use std::path::PathBuf; /// 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, /// Sort by the thread's product recency timestamp. RecencyAt, /// Sort by the thread's stable position within its user-selected section. SectionPosition, } /// Sort direction to use when listing threads. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum SortDirection { Asc, Desc, } /// Thread relationship used to filter thread listings. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ThreadRelationFilter { /// Include attached threads alongside ordinary threads for search. IncludeAttached, /// Return only threads whose immediate parent is the given thread. DirectChildrenOf(ThreadId), /// Return every thread transitively descended from the given thread. DescendantsOf(ThreadId), /// Return only saved threads attached to this parent. AttachedThreadsOf(ThreadId), } /// A pagination anchor used for keyset pagination. #[derive(Debug, Clone, PartialEq, Eq)] pub struct Anchor { /// The timestamp component of the anchor. pub ts: DateTime, /// The thread ID component used to disambiguate equal recency timestamps. pub id: Option, } /// Visual presentation metadata owned by a thread section. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct ThreadSectionAppearance { pub icon: Option, pub color: Option, } /// An independently persisted thread section and its user-facing presentation. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct ThreadSection { /// Opaque UUIDv7 identifying the section independently of its name. pub id: String, /// User-facing section name. pub name: String, #[serde(default)] pub appearance: Option, } impl ThreadSection { pub(crate) fn from_row( (id, name, appearance): (String, String, Option), ) -> Result { Ok(Self { id, name, appearance: appearance .map(|appearance| serde_json::from_str(&appearance)) .transpose()?, }) } } /// A cursor-paginated page of independently persisted thread sections. #[derive(Debug, Clone, PartialEq, Eq)] pub struct ThreadSectionsPage { /// Sections in ascending identifier order. pub sections: Vec, /// Identifier after which the next page starts, if any. pub next_cursor: Option, } /// 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, /// Immediate parents for page items found through the persisted spawn graph. pub parent_thread_ids: HashMap, /// The next anchor to use for pagination, if any. pub next_anchor: Option, /// 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 explicit thread memory mode from rollout metadata, if present. pub memory_mode: Option, /// The number of rollout lines that failed to parse. pub parse_errors: usize, } /// Canonical persisted thread metadata. #[derive(Debug, Clone, PartialEq, Eq)] pub struct ThreadMetadata { /// The parent of an attached thread, derived from canonical rollout metadata. pub attached_thread_parent_id: Option, /// Originator recorded at creation, if available. pub originator: Option, /// ChatGPT user that created the thread, if known. pub creator_user_id: Option, /// ChatGPT account at creation, if known. pub creator_account_id: Option, /// The thread identifier. pub id: ThreadId, /// The absolute rollout path on disk. pub rollout_path: PathBuf, /// The creation timestamp. pub created_at: DateTime, /// The last update timestamp. pub updated_at: DateTime, /// The product recency timestamp. pub recency_at: DateTime, /// The session source (stringified enum). pub source: String, /// Persisted thread history contract selected when this thread was created. pub history_mode: ThreadHistoryMode, /// Optional analytics source classification for this thread. pub thread_source: Option, /// Optional random unique nickname assigned to an AgentControl-spawned sub-agent. pub agent_nickname: Option, /// Optional role (agent_role) assigned to an AgentControl-spawned sub-agent. pub agent_role: Option, /// Optional canonical agent path assigned to an AgentControl-spawned sub-agent. pub agent_path: Option, /// The model provider identifier. pub model_provider: String, /// The latest observed model for the thread. pub model: Option, /// The latest observed reasoning effort for the thread. pub reasoning_effort: Option, /// The working directory for the thread. pub cwd: PathBuf, /// Version of the CLI that created the thread. pub cli_version: String, /// A best-effort thread title. pub title: String, /// Explicit user-facing thread name, if one was set. pub name: Option, /// Best available user-facing preview for discovery and list display. pub preview: Option, /// Serialized permission profile or legacy sandbox value. pub sandbox_policy: String, /// The approval mode (stringified enum). pub approval_mode: String, /// The last observed token usage. pub tokens_used: i64, /// First user message observed for this thread, if any. pub first_user_message: Option, /// The archive timestamp, if the thread is archived. pub archived_at: Option>, /// The user-selected section for this thread, if any. pub section: Option, /// The stable sparse ordering rank within the user-selected section. pub section_position: Option, /// The time when the thread most recently entered its current section. pub section_entered_at: Option>, /// Canonical project assignment owned by app-server, if any. pub project_id: Option, /// User-selected Daybreak preference, absent until explicitly set. pub daybreak_enabled: Option, /// The git commit SHA, if known. pub git_sha: Option, /// The git branch name, if known. pub git_branch: Option, /// The git origin URL, if known. pub git_origin_url: Option, } impl ThreadMetadata { /// Preserve SQLite-owned Git fields when rollout-derived metadata is reconciled. pub fn prefer_existing_git_info(&mut self, existing: &Self) { if matches!(self.history_mode, ThreadHistoryMode::Paginated) && matches!(existing.history_mode, ThreadHistoryMode::Paginated) { // `self` was rebuilt from the rollout's initial SessionMeta. `existing` is the // current SQLite row. Once that row says paginated, metadata updates are SQLite-only, // so a NULL is an explicit clear, not missing data. Copy the whole tuple or the stale // rollout value would be written back during reconciliation. self.git_sha = existing.git_sha.clone(); self.git_branch = existing.git_branch.clone(); self.git_origin_url = existing.git_origin_url.clone(); return; } if existing.git_sha.is_some() { self.git_sha = existing.git_sha.clone(); } if existing.git_branch.is_some() { self.git_branch = existing.git_branch.clone(); } if existing.git_origin_url.is_some() { self.git_origin_url = existing.git_origin_url.clone(); } } /// Preserve an existing user-facing title when reconciling rollout-derived metadata. pub fn prefer_existing_explicit_title(&mut self, existing: &Self) { let existing_title = existing.title.trim(); if existing_title.is_empty() || existing.first_user_message.as_deref().map(str::trim) == Some(existing_title) { return; } let title = self.title.trim(); if title.is_empty() || self.first_user_message.as_deref().map(str::trim) == Some(title) || (title == crate::GUARDIAN_THREAD_TITLE && crate::extract::metadata_is_guardian_review(self)) { self.title = existing.title.clone(); } } /// 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.creator_user_id != other.creator_user_id { diffs.push("creator_user_id"); } if self.creator_account_id != other.creator_account_id { diffs.push("creator_account_id"); } if self.originator != other.originator { diffs.push("originator"); } if self.agent_nickname != other.agent_nickname { diffs.push("agent_nickname"); } if self.agent_role != other.agent_role { diffs.push("agent_role"); } if self.agent_path != other.agent_path { diffs.push("agent_path"); } if self.model_provider != other.model_provider { diffs.push("model_provider"); } if self.model != other.model { diffs.push("model"); } if self.reasoning_effort != other.reasoning_effort { diffs.push("reasoning_effort"); } if self.cwd != other.cwd { diffs.push("cwd"); } if self.cli_version != other.cli_version { diffs.push("cli_version"); } if self.title != other.title { diffs.push("title"); } if self.name != other.name { diffs.push("name"); } if self.preview != other.preview { diffs.push("preview"); } 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.first_user_message != other.first_user_message { diffs.push("first_user_message"); } if self.archived_at != other.archived_at { diffs.push("archived_at"); } if self.section != other.section { diffs.push("section"); } if self.section_position != other.section_position { diffs.push("section_position"); } if self.section_entered_at != other.section_entered_at { diffs.push("section_entered_at"); } if self.project_id != other.project_id { diffs.push("project_id"); } if self.daybreak_enabled != other.daybreak_enabled { diffs.push("daybreak_enabled"); } 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 } } pub(super) fn canonicalize_datetime(dt: DateTime) -> DateTime { epoch_millis_to_datetime(datetime_to_epoch_millis(dt)).unwrap_or(dt) } #[derive(Debug)] pub(crate) struct ThreadRow { attached_thread_parent_id: Option, originator: Option, creator_user_id: Option, creator_account_id: Option, id: String, rollout_path: String, created_at: i64, updated_at: i64, recency_at: i64, source: String, history_mode: String, thread_source: Option, agent_nickname: Option, agent_role: Option, agent_path: Option, model_provider: String, model: Option, reasoning_effort: Option, cwd: String, cli_version: String, title: String, name: Option, preview: String, sandbox_policy: String, approval_mode: String, tokens_used: i64, first_user_message: String, archived_at: Option, section: Option, section_name: Option, section_appearance: Option, section_position: Option, section_entered_at_ms: Option, project_id: Option, daybreak_enabled: Option, git_sha: Option, git_branch: Option, git_origin_url: Option, } impl ThreadRow { pub(crate) fn try_from_row(row: &SqliteRow) -> Result { Ok(Self { attached_thread_parent_id: row.try_get("attached_thread_parent_id")?, originator: row.try_get("originator")?, creator_user_id: row.try_get("creator_user_id")?, creator_account_id: row.try_get("creator_account_id")?, 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")?, recency_at: row.try_get("recency_at")?, source: row.try_get("source")?, history_mode: row.try_get("history_mode")?, thread_source: row.try_get("thread_source")?, agent_nickname: row.try_get("agent_nickname")?, agent_role: row.try_get("agent_role")?, agent_path: row.try_get("agent_path")?, model_provider: row.try_get("model_provider")?, model: row.try_get("model")?, reasoning_effort: row.try_get("reasoning_effort")?, cwd: row.try_get("cwd")?, cli_version: row.try_get("cli_version")?, title: row.try_get("title")?, name: row.try_get("name")?, preview: row.try_get("preview")?, sandbox_policy: row.try_get("sandbox_policy")?, approval_mode: row.try_get("approval_mode")?, tokens_used: row.try_get("tokens_used")?, first_user_message: row.try_get("first_user_message")?, archived_at: row.try_get("archived_at")?, section: row.try_get("section")?, section_name: row.try_get("section_name")?, section_appearance: row.try_get("section_appearance")?, section_position: row.try_get("section_position")?, section_entered_at_ms: row.try_get("section_entered_at_ms")?, project_id: row.try_get("project_id")?, daybreak_enabled: row.try_get("daybreak_enabled")?, 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 for ThreadMetadata { type Error = anyhow::Error; fn try_from(row: ThreadRow) -> std::result::Result { let ThreadRow { attached_thread_parent_id, originator, creator_user_id, creator_account_id, id, rollout_path, created_at, updated_at, recency_at, source, history_mode, thread_source, agent_nickname, agent_role, agent_path, model_provider, model, reasoning_effort, cwd, cli_version, title, name, preview, sandbox_policy, approval_mode, tokens_used, first_user_message, archived_at, section, section_name, section_appearance, section_position, section_entered_at_ms, project_id, daybreak_enabled, git_sha, git_branch, git_origin_url, } = row; let thread_source = thread_source .map(|thread_source| thread_source.parse()) .transpose() .map_err(anyhow::Error::msg)?; let history_mode = history_mode.parse().map_err(anyhow::Error::msg)?; let section = match (section, section_name) { (Some(id), Some(name)) => { Some(ThreadSection::from_row((id, name, section_appearance))?) } (None, None) => None, (Some(id), None) => { return Err(anyhow::anyhow!( "thread references an unknown section: {id}" )); } (None, Some(name)) => { return Err(anyhow::anyhow!( "thread has a section name without a section id: {name}" )); } }; Ok(Self { id: ThreadId::try_from(id)?, attached_thread_parent_id: attached_thread_parent_id .map(ThreadId::try_from) .transpose()?, originator, creator_user_id, creator_account_id, rollout_path: PathBuf::from(rollout_path), created_at: epoch_millis_to_datetime(created_at)?, updated_at: epoch_millis_to_datetime(updated_at)?, recency_at: epoch_millis_to_datetime(recency_at)?, source, history_mode, thread_source, agent_nickname, agent_role, agent_path, model_provider, model, reasoning_effort: reasoning_effort .and_then(|value| value.parse::().ok()), cwd: PathBuf::from(cwd), cli_version, title, name, preview: (!preview.is_empty()).then_some(preview), sandbox_policy, approval_mode, tokens_used, first_user_message: (!first_user_message.is_empty()).then_some(first_user_message), archived_at: archived_at.map(epoch_seconds_to_datetime).transpose()?, section, section_position, section_entered_at: section_entered_at_ms .map(epoch_millis_to_datetime) .transpose()?, project_id, daybreak_enabled, git_sha, git_branch, git_origin_url: git_origin_url .and_then(|origin_url| SanitizedGitUrl::try_from(origin_url).ok()), }) } } pub(crate) fn anchor_from_item(item: &ThreadMetadata, sort_key: SortKey) -> Option { let ts = match sort_key { SortKey::CreatedAt => item.created_at, SortKey::UpdatedAt => item.updated_at, SortKey::RecencyAt => item.recency_at, SortKey::SectionPosition => DateTime::::from_timestamp_millis(item.section_position?)?, }; Some(Anchor { ts, id: Some(item.id), }) } pub(crate) fn datetime_to_epoch_millis(dt: DateTime) -> i64 { dt.timestamp_millis() } pub(crate) fn datetime_to_epoch_seconds(dt: DateTime) -> i64 { dt.timestamp() } pub(crate) fn epoch_millis_to_datetime(value: i64) -> Result> { // Values older than 2020 if interpreted as milliseconds are legacy second-precision rows. // Convert them in memory so old state DBs keep ordering correctly after new writes use ms. const MIN_EPOCH_MILLIS: i64 = 1_577_836_800_000; let millis = if value < MIN_EPOCH_MILLIS { value.saturating_mul(1000) } else { value }; DateTime::::from_timestamp_millis(millis) .ok_or_else(|| anyhow::anyhow!("invalid unix timestamp millis: {value}")) } pub(crate) fn epoch_seconds_to_datetime(value: i64) -> Result> { DateTime::::from_timestamp(value, 0) .ok_or_else(|| anyhow::anyhow!("invalid unix timestamp seconds: {value}")) } /// 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, } #[cfg(test)] mod tests { use super::ThreadMetadata; use super::ThreadRow; use chrono::DateTime; use chrono::Utc; use codex_protocol::SanitizedGitUrl; use codex_protocol::ThreadId; use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::protocol::ThreadHistoryMode; use pretty_assertions::assert_eq; use std::path::PathBuf; fn thread_row(reasoning_effort: Option<&str>) -> ThreadRow { ThreadRow { attached_thread_parent_id: None, originator: None, creator_user_id: None, creator_account_id: None, id: "00000000-0000-0000-0000-000000000123".to_string(), rollout_path: "/tmp/rollout-123.jsonl".to_string(), created_at: 1_700_000_000, updated_at: 1_700_000_100, recency_at: 1_700_000_100, source: "cli".to_string(), history_mode: "legacy".to_string(), thread_source: None, agent_nickname: None, agent_role: None, agent_path: None, model_provider: "openai".to_string(), model: Some("gpt-5".to_string()), reasoning_effort: reasoning_effort.map(str::to_string), cwd: "/tmp/workspace".to_string(), cli_version: "0.0.0".to_string(), title: String::new(), name: None, preview: String::new(), sandbox_policy: "read-only".to_string(), approval_mode: "on-request".to_string(), tokens_used: 1, first_user_message: String::new(), archived_at: None, section: None, section_name: None, section_appearance: None, section_position: None, section_entered_at_ms: None, project_id: None, daybreak_enabled: None, git_sha: None, git_branch: None, git_origin_url: None, } } fn expected_thread_metadata(reasoning_effort: Option) -> ThreadMetadata { ThreadMetadata { attached_thread_parent_id: None, originator: None, creator_user_id: None, creator_account_id: None, id: ThreadId::from_string("00000000-0000-0000-0000-000000000123") .expect("valid thread id"), rollout_path: PathBuf::from("/tmp/rollout-123.jsonl"), created_at: DateTime::::from_timestamp(1_700_000_000, 0).expect("timestamp"), updated_at: DateTime::::from_timestamp(1_700_000_100, 0).expect("timestamp"), recency_at: DateTime::::from_timestamp(1_700_000_100, 0).expect("timestamp"), source: "cli".to_string(), history_mode: ThreadHistoryMode::Legacy, thread_source: None, agent_nickname: None, agent_role: None, agent_path: None, model_provider: "openai".to_string(), model: Some("gpt-5".to_string()), reasoning_effort, cwd: PathBuf::from("/tmp/workspace"), cli_version: "0.0.0".to_string(), title: String::new(), name: None, preview: None, sandbox_policy: "read-only".to_string(), approval_mode: "on-request".to_string(), tokens_used: 1, first_user_message: None, archived_at: None, section: None, section_position: None, section_entered_at: None, project_id: None, daybreak_enabled: None, git_sha: None, git_branch: None, git_origin_url: None, } } #[test] fn thread_row_parses_reasoning_effort() { let metadata = ThreadMetadata::try_from(thread_row(Some("high"))) .expect("thread metadata should parse"); assert_eq!( metadata, expected_thread_metadata(Some(ReasoningEffort::High)) ); } #[test] fn thread_row_preserves_model_defined_reasoning_effort_values() { let metadata = ThreadMetadata::try_from(thread_row(Some("future"))) .expect("thread metadata should parse"); assert_eq!( metadata, expected_thread_metadata(Some(ReasoningEffort::Custom("future".to_string()))) ); } #[test] fn thread_row_rejects_unknown_history_mode() { let mut row = thread_row(/*reasoning_effort*/ None); row.history_mode = "future".to_string(); assert!(ThreadMetadata::try_from(row).is_err()); } #[test] fn paginated_rollout_git_info_keeps_rollout_values_until_sqlite_mode_is_paginated() { let mut reconciled = expected_thread_metadata(/*reasoning_effort*/ None); reconciled.history_mode = ThreadHistoryMode::Paginated; reconciled.git_sha = Some("rollout-sha".to_string()); reconciled.git_branch = Some("rollout-branch".to_string()); reconciled.git_origin_url = Some( SanitizedGitUrl::try_from("https://example.com/rollout-origin") .expect("valid git remote URL"), ); let existing = expected_thread_metadata(/*reasoning_effort*/ None); let expected = reconciled.clone(); reconciled.prefer_existing_git_info(&existing); assert_eq!(reconciled, expected); } }