roder_api/

thread.rs

use std::collections::BTreeSet;
use std::path::PathBuf;
use std::sync::Arc;

use serde::{Deserialize, Deserializer, Serialize};
use time::OffsetDateTime;

use crate::artifacts::ContextArtifactStore;
use crate::events::{EventEnvelope, ThreadId, TurnId};
pub use crate::extension::{CheckpointStoreId, ThreadStoreId};
use crate::extension_state::ExtensionStateRecord;
use crate::inference::{TokenUsage, cache_hit_rate};
use crate::inference_routing::ModelSelectionMode;
use crate::remote_runner::{RunnerDestination, RunnerSessionState, ThreadRunnerBinding};
use crate::transcript::{InputImage, TranscriptItem};

mod history;
mod projection;
pub use history::*;
pub use projection::{project_thread_item_events, project_turns_from_events};

#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ThreadListOptions {
    pub limit: Option<usize>,
    pub cursor: Option<String>,
}

#[derive(Debug, Clone, Default)]
pub struct ThreadListPage {
    pub threads: Vec<ThreadMetadata>,
    pub next_cursor: Option<String>,
    pub backwards_cursor: Option<String>,
}

/// Thread IDs reserved for system event streams that are not user-visible conversations.
pub const SYNTHETIC_EVENT_THREAD_IDS: &[&str] = &["app-server", "runtime", "thread-workflow"];

/// Returns true for production-emitted synthetic event streams.
pub fn is_synthetic_event_thread_id(thread_id: &str) -> bool {
    SYNTHETIC_EVENT_THREAD_IDS.contains(&thread_id)
}

#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]
pub struct ThreadUsageMetadata {
    #[serde(default)]
    pub prompt_tokens: u64,
    #[serde(default)]
    pub completion_tokens: u64,
    #[serde(default)]
    pub total_tokens: u64,
    #[serde(default)]
    pub cached_prompt_tokens: u64,
    /// Subset of `prompt_tokens` written to the provider prompt cache.
    #[serde(default)]
    pub cache_creation_prompt_tokens: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cache_hit_rate: Option<f64>,
    /// Terminal turns included in the cumulative counters, persisted with those counters.
    #[serde(default, skip_serializing_if = "BTreeSet::is_empty")]
    pub accounted_turn_ids: BTreeSet<TurnId>,
}

impl ThreadUsageMetadata {
    /// Adds a terminal turn's usage once, including after metadata is reloaded.
    pub fn add_turn_token_usage(&mut self, turn_id: &TurnId, usage: &TokenUsage) -> bool {
        if !self.accounted_turn_ids.insert(turn_id.clone()) {
            return false;
        }
        self.add_token_usage(usage);
        true
    }

    pub fn add_token_usage(&mut self, usage: &TokenUsage) {
        self.prompt_tokens = self
            .prompt_tokens
            .saturating_add(u64::from(usage.prompt_tokens));
        self.completion_tokens = self
            .completion_tokens
            .saturating_add(u64::from(usage.completion_tokens));
        self.total_tokens = self
            .total_tokens
            .saturating_add(u64::from(usage.total_tokens));
        self.cached_prompt_tokens = self
            .cached_prompt_tokens
            .saturating_add(u64::from(usage.cached_prompt_tokens));
        self.cache_creation_prompt_tokens = self
            .cache_creation_prompt_tokens
            .saturating_add(u64::from(usage.cache_creation_prompt_tokens));
        self.cache_hit_rate = if self.prompt_tokens == 0 {
            None
        } else if self.prompt_tokens > u64::from(u32::MAX) {
            Some(
                (self.cached_prompt_tokens.min(self.prompt_tokens) as f64)
                    / (self.prompt_tokens as f64),
            )
        } else {
            cache_hit_rate(self.prompt_tokens as u32, self.cached_prompt_tokens as u32)
        };
    }

    pub fn is_empty(&self) -> bool {
        self.prompt_tokens == 0
            && self.completion_tokens == 0
            && self.total_tokens == 0
            && self.cached_prompt_tokens == 0
            && self.cache_creation_prompt_tokens == 0
    }
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct ThreadMetadata {
    /// Default provider service tier for this thread.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub service_tier: Option<String>,
    pub thread_id: ThreadId,
    pub title: Option<String>,
    #[serde(deserialize_with = "deserialize_thread_workspace")]
    pub workspace: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub workspace_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub root_id: Option<String>,
    pub provider: Option<String>,
    pub model: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub selection_mode: Option<ModelSelectionMode>,
    /// Per-thread tool filter applied on top of the runtime allowlist. Empty = no filtering.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub tool_allowlist: Vec<String>,
    /// Host-supplied instructions added to the developer slot of every turn's inference request.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub developer_instructions: Option<String>,
    /**
     * Host-executed tool specs advertised to the model on every turn of this thread. Calls to
     * these tools pause on a `thread/toolExecutionRequested` notification until the host client
     * answers with `tools/resolve`.
     */
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub external_tools: Vec<crate::tools::ToolSpec>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub runner_destination: Option<RunnerDestination>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub runner_state: Option<RunnerSessionState>,
    /**
     * Set when the thread explicitly selected a remote runner at creation.
     * Native coding tools for the thread route through this runner workspace;
     * absent = local tool execution even when a runtime-level runner
     * destination is configured.
     */
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub runner_binding: Option<ThreadRunnerBinding>,
    #[serde(with = "time::serde::rfc3339")]
    pub created_at: OffsetDateTime,
    #[serde(with = "time::serde::rfc3339")]
    pub updated_at: OffsetDateTime,
    pub message_count: u32,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub usage: Option<ThreadUsageMetadata>,
    /// Parent thread this conversation was forked from. Absent for normal threads.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub parent_thread_id: Option<ThreadId>,
    /// Parent turn the fork branched at, when the fork targeted a specific turn.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub forked_from_turn_id: Option<TurnId>,
    /**
     * Provider-neutral workspace-fork provenance for threads whose
     * workspace is a fork of another workspace (roadmap phase 81; the
     * phase-90 Git-worktree MVP shape was folded into this canonical type).
     */
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub workspace_fork: Option<crate::forks::WorkspaceFork>,
}

pub fn validate_thread_workspace(workspace: &str) -> anyhow::Result<String> {
    let workspace = workspace.trim();
    anyhow::ensure!(!workspace.is_empty(), "thread workspace is required");
    anyhow::ensure!(
        std::path::Path::new(workspace).is_absolute(),
        "thread workspace must be an absolute path: {workspace}"
    );
    Ok(workspace.to_string())
}

fn deserialize_thread_workspace<'de, D>(deserializer: D) -> Result<String, D::Error>
where
    D: Deserializer<'de>,
{
    let workspace = String::deserialize(deserializer)?;
    validate_thread_workspace(&workspace).map_err(serde::de::Error::custom)
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct TurnRecord {
    pub thread_id: ThreadId,
    pub turn_id: TurnId,
    pub items: Vec<TranscriptItem>,
    #[serde(with = "time::serde::rfc3339")]
    pub created_at: OffsetDateTime,
    #[serde(with = "time::serde::rfc3339::option")]
    pub completed_at: Option<OffsetDateTime>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub usage: Option<TokenUsage>,
    /// Normalized stop reason from `TurnCompleted`; `None` for failed or interrupted turns.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub finish_reason: Option<String>,
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub enum ThreadItemStatus {
    InProgress,
    Completed,
    Failed,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(tag = "type", rename_all = "camelCase")]
pub enum ThreadItem {
    UserMessage {
        id: String,
        text: String,
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        images: Vec<InputImage>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        status: Option<ThreadItemStatus>,
    },
    AgentMessage {
        id: String,
        text: String,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        phase: Option<String>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        status: Option<ThreadItemStatus>,
    },
    Reasoning {
        id: String,
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        summary: Vec<String>,
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        content: Vec<String>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        status: Option<ThreadItemStatus>,
    },
    ToolExecution {
        id: String,
        #[serde(rename = "toolCallId")]
        tool_call_id: String,
        #[serde(rename = "toolName")]
        tool_name: String,
        status: ThreadItemStatus,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        input: Option<serde_json::Value>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        output: Option<String>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        error: Option<String>,
    },
    RoutingDecision {
        id: String,
        decision: crate::events::InferenceRoutingDecisionEvent,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        status: Option<ThreadItemStatus>,
    },
    Compaction {
        id: String,
        summary: String,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        status: Option<ThreadItemStatus>,
    },
    Error {
        id: String,
        message: String,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        status: Option<ThreadItemStatus>,
    },
    Raw {
        id: String,
        payload: serde_json::Value,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        status: Option<ThreadItemStatus>,
    },
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct ThreadItemTurnRecord {
    pub thread_id: ThreadId,
    pub turn_id: TurnId,
    #[serde(with = "time::serde::rfc3339")]
    pub created_at: OffsetDateTime,
    pub items: Vec<ThreadItem>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(tag = "type", rename_all = "camelCase")]
pub enum ThreadItemDelta {
    AgentMessageText {
        delta: String,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        phase: Option<String>,
    },
    ReasoningText {
        delta: String,
        #[serde(rename = "contentIndex")]
        content_index: usize,
    },
    ReasoningSummaryPartAdded {
        #[serde(rename = "summaryIndex")]
        summary_index: usize,
    },
    ReasoningSummaryText {
        delta: String,
        #[serde(rename = "summaryIndex")]
        summary_index: usize,
    },
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(tag = "type", rename_all = "camelCase")]
pub enum ThreadItemEventKind {
    ItemStarted {
        item: ThreadItem,
    },
    ItemDelta {
        #[serde(rename = "itemId")]
        item_id: String,
        delta: ThreadItemDelta,
    },
    ItemCompleted {
        item: ThreadItem,
    },
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct ThreadItemEvent {
    pub seq: u64,
    #[serde(rename = "eventId")]
    pub event_id: String,
    #[serde(rename = "threadId")]
    pub thread_id: ThreadId,
    #[serde(rename = "turnId")]
    pub turn_id: TurnId,
    #[serde(with = "time::serde::rfc3339")]
    pub timestamp: OffsetDateTime,
    pub event: ThreadItemEventKind,
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct ThreadSnapshot {
    pub metadata: Option<ThreadMetadata>,
    pub events: Vec<EventEnvelope>,
    pub turns: Vec<TurnRecord>,
    #[serde(default)]
    pub item_events: Vec<ThreadItemEvent>,
    pub extension_states: Vec<ExtensionStateRecord>,
}

impl ThreadItem {
    pub fn id(&self) -> &str {
        match self {
            ThreadItem::UserMessage { id, .. }
            | ThreadItem::AgentMessage { id, .. }
            | ThreadItem::Reasoning { id, .. }
            | ThreadItem::ToolExecution { id, .. }
            | ThreadItem::RoutingDecision { id, .. }
            | ThreadItem::Compaction { id, .. }
            | ThreadItem::Error { id, .. }
            | ThreadItem::Raw { id, .. } => id,
        }
    }
}

#[async_trait::async_trait]
pub trait ThreadStore: Send + Sync {
    fn id(&self) -> ThreadStoreId;

    fn local_thread_root(&self) -> Option<PathBuf> {
        None
    }

    fn context_artifact_store(&self) -> Option<ContextArtifactStore> {
        None
    }

    async fn create_thread(&self, metadata: ThreadMetadata) -> anyhow::Result<ThreadMetadata>;
    async fn update_thread_metadata(
        &self,
        metadata: ThreadMetadata,
    ) -> anyhow::Result<ThreadMetadata> {
        Ok(metadata)
    }
    async fn list_threads(&self) -> anyhow::Result<Vec<ThreadMetadata>>;
    async fn list_threads_page(
        &self,
        options: ThreadListOptions,
    ) -> anyhow::Result<ThreadListPage> {
        let mut threads = self.list_threads().await?;
        threads.sort_by_key(|thread| std::cmp::Reverse(thread.updated_at));
        let offset = options
            .cursor
            .as_deref()
            .and_then(|cursor| cursor.parse::<usize>().ok())
            .unwrap_or(0)
            .min(threads.len());
        let limit = options
            .limit
            .unwrap_or(threads.len().saturating_sub(offset));
        let next_offset = offset.saturating_add(limit).min(threads.len());
        let total = threads.len();
        let page_threads = threads
            .into_iter()
            .skip(offset)
            .take(limit)
            .collect::<Vec<_>>();
        Ok(ThreadListPage {
            threads: page_threads,
            next_cursor: (next_offset < total).then(|| next_offset.to_string()),
            backwards_cursor: (offset > 0).then(|| offset.saturating_sub(limit).to_string()),
        })
    }
    async fn load_thread_metadata(
        &self,
        thread_id: &ThreadId,
    ) -> anyhow::Result<Option<ThreadMetadata>> {
        Ok(self
            .load_thread(thread_id)
            .await?
            .and_then(|snapshot| snapshot.metadata))
    }
    async fn load_thread(&self, thread_id: &ThreadId) -> anyhow::Result<Option<ThreadSnapshot>>;

    /// Read bounded history without loading a complete thread snapshot.
    async fn read_history_page(
        &self,
        _thread_id: &ThreadId,
        _options: ThreadHistoryReadOptions,
    ) -> anyhow::Result<Option<ThreadHistoryPage>> {
        anyhow::bail!("thread store does not support bounded history pagination")
    }
    /// Loads extension state without requiring callers to project a full thread snapshot.
    ///
    /// Stores with dedicated extension-state backing should override this method. The default
    /// preserves compatibility for stores that only expose extension state through `load_thread`.
    async fn load_extension_states(
        &self,
        thread_id: &ThreadId,
    ) -> anyhow::Result<Vec<ExtensionStateRecord>> {
        Ok(self
            .load_thread(thread_id)
            .await?
            .map_or_else(Vec::new, |snapshot| snapshot.extension_states))
    }
    async fn archive_thread(&self, thread_id: &ThreadId) -> anyhow::Result<bool> {
        let _ = thread_id;
        anyhow::bail!("thread store {} does not support archive", self.id())
    }
    async fn append_event(
        &self,
        thread_id: &ThreadId,
        envelope: &EventEnvelope,
    ) -> anyhow::Result<()>;
    async fn append_item_event(
        &self,
        thread_id: &ThreadId,
        item_event: &ThreadItemEvent,
    ) -> anyhow::Result<()> {
        let _ = (thread_id, item_event);
        Ok(())
    }
    async fn append_extension_state(
        &self,
        thread_id: &ThreadId,
        record: &ExtensionStateRecord,
    ) -> anyhow::Result<()> {
        let _ = (thread_id, record);
        anyhow::bail!(
            "thread store {} does not support extension state",
            self.id()
        )
    }
}

pub trait ThreadStoreFactory: Send + Sync + 'static {
    fn id(&self) -> ThreadStoreId;
    fn create(&self) -> Arc<dyn ThreadStore>;
}

#[async_trait::async_trait]
pub trait CheckpointStore: Send + Sync {
    fn id(&self) -> CheckpointStoreId;
    async fn save_snapshot(&self, snapshot: ThreadSnapshot) -> anyhow::Result<()>;
    async fn load_snapshot(&self, thread_id: &ThreadId) -> anyhow::Result<Option<ThreadSnapshot>>;
}

pub trait CheckpointStoreFactory: Send + Sync + 'static {
    fn id(&self) -> CheckpointStoreId;
    fn create(&self) -> Arc<dyn CheckpointStore>;
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::inference::ModelSelection;

    #[test]
    fn synthetic_event_thread_ids_are_reserved_production_ids() {
        assert!(is_synthetic_event_thread_id("app-server"));
        assert!(is_synthetic_event_thread_id("runtime"));
        assert!(is_synthetic_event_thread_id("thread-workflow"));

        assert!(!is_synthetic_event_thread_id("thread-discovery"));
        assert!(!is_synthetic_event_thread_id("thread-plan"));
        assert!(!is_synthetic_event_thread_id("thread-process"));
        assert!(!is_synthetic_event_thread_id("thread-1"));
    }

    #[test]
    fn thread_fork_metadata_is_additive_for_legacy_records() {
        // A legacy thread record without any fork fields still deserializes.
        let legacy = serde_json::json!({
            "thread_id": "thread-old",
            "title": null,
            "workspace": "/workspace",
            "provider": null,
            "model": null,
            "created_at": "1970-01-01T00:00:00Z",
            "updated_at": "1970-01-01T00:00:00Z",
            "message_count": 3
        });
        let metadata: ThreadMetadata = serde_json::from_value(legacy).unwrap();
        assert!(metadata.parent_thread_id.is_none());
        assert!(metadata.forked_from_turn_id.is_none());
        assert!(metadata.workspace_fork.is_none());

        // Non-forked threads serialize without any fork keys.
        let value = serde_json::to_value(&metadata).unwrap();
        assert!(value.get("parentThreadId").is_none());
        assert!(value.get("parent_thread_id").is_none());
        assert!(value.get("workspace_fork").is_none());
    }

    #[test]
    fn thread_fork_metadata_round_trips_workspace_fork_provenance() {
        let fork = crate::forks::WorkspaceFork {
            ownership: Some(crate::forks::ForkOwnership {
                creator_thread_id: Some("parent-thread".into()),
                owner_thread_id: Some("fork-thread".into()),
            }),
            id: "/repo/.roder/worktrees/parser-experiment".to_string(),
            provider_id: "git-worktree".to_string(),
            source_workspace: std::path::PathBuf::from("/repo"),
            workspace: std::path::PathBuf::from("/repo/.roder/worktrees/parser-experiment"),
            status: crate::forks::ForkStatus::Active,
            provenance: crate::forks::ForkProvenance {
                branch: Some("roder/fork/parser-experiment".to_string()),
                source_branch: Some("main".to_string()),
                source_commit: Some("abc123".to_string()),
                snapshot_id: None,
                session_id: None,
                created_at: OffsetDateTime::UNIX_EPOCH,
            },
            cleanup: crate::forks::ForkCleanupPolicy::Explicit,
            metadata: serde_json::json!({}),
        };
        let value = serde_json::to_value(&fork).unwrap();
        assert_eq!(value["providerId"], "git-worktree");
        assert_eq!(value["status"], "active");
        assert_eq!(value["cleanup"], "explicit");
        assert_eq!(value["provenance"]["sourceCommit"], "abc123");

        let round_trip: crate::forks::WorkspaceFork = serde_json::from_value(value).unwrap();
        assert_eq!(round_trip, fork);

        // A detached-HEAD fork keeps clear provenance without a branch name.
        let detached = crate::forks::WorkspaceFork {
            provenance: crate::forks::ForkProvenance {
                source_branch: None,
                ..fork.provenance.clone()
            },
            ..fork
        };
        let value = serde_json::to_value(&detached).unwrap();
        assert!(value["provenance"].get("sourceBranch").is_none());
    }

    #[test]
    fn thread_metadata_timestamps_serialize_as_rfc3339_strings() {
        let value = serde_json::to_value(ThreadMetadata {
            thread_id: "thread-a".to_string(),
            title: None,
            workspace: "/workspace".to_string(),
            workspace_id: None,
            root_id: None,
            provider: None,
            model: None,
            selection_mode: None,
            tool_allowlist: Vec::new(),
            service_tier: None,
            developer_instructions: None,
            external_tools: Vec::new(),
            runner_destination: None,
            runner_state: None,
            runner_binding: None,
            parent_thread_id: None,
            forked_from_turn_id: None,
            workspace_fork: None,
            created_at: OffsetDateTime::UNIX_EPOCH,
            updated_at: OffsetDateTime::UNIX_EPOCH,
            message_count: 0,
            usage: None,
        })
        .unwrap();

        assert_eq!(value["created_at"], "1970-01-01T00:00:00Z");
        assert_eq!(value["updated_at"], "1970-01-01T00:00:00Z");
        assert_eq!(value["workspace"], "/workspace");
    }

    #[test]
    fn thread_metadata_deserializes_without_selection_mode() {
        let value = serde_json::json!({
            "thread_id": "thread-a",
            "title": null,
            "workspace": "/workspace",
            "provider": "codex",
            "model": "gpt-5.5",
            "created_at": "1970-01-01T00:00:00Z",
            "updated_at": "1970-01-01T00:00:00Z",
            "message_count": 0
        });

        let metadata = serde_json::from_value::<ThreadMetadata>(value).unwrap();

        assert_eq!(metadata.provider.as_deref(), Some("codex"));
        assert_eq!(metadata.model.as_deref(), Some("gpt-5.5"));
        assert_eq!(metadata.selection_mode, None);
    }

    #[test]
    fn thread_metadata_round_trips_auto_selection_mode() {
        let metadata = ThreadMetadata {
            thread_id: "thread-a".to_string(),
            title: None,
            workspace: "/workspace".to_string(),
            workspace_id: None,
            root_id: None,
            provider: Some("codex".to_string()),
            model: Some("gpt-5.5".to_string()),
            selection_mode: Some(ModelSelectionMode::auto(
                "local-router:coding",
                "local-router",
                "Auto: Coding",
                ModelSelection {
                    provider: "codex".to_string(),
                    model: "gpt-5.5".to_string(),
                },
                Some("coding".to_string()),
                Some("low".to_string()),
            )),
            tool_allowlist: Vec::new(),
            service_tier: None,
            developer_instructions: None,
            external_tools: Vec::new(),
            runner_destination: None,
            runner_state: None,
            runner_binding: None,
            parent_thread_id: None,
            forked_from_turn_id: None,
            workspace_fork: None,
            created_at: OffsetDateTime::UNIX_EPOCH,
            updated_at: OffsetDateTime::UNIX_EPOCH,
            message_count: 0,
            usage: None,
        };

        let value = serde_json::to_value(&metadata).unwrap();
        let round_trip = serde_json::from_value::<ThreadMetadata>(value).unwrap();

        assert_eq!(round_trip, metadata);
    }

    #[test]
    fn thread_metadata_requires_workspace_when_deserializing() {
        let value = serde_json::json!({
            "thread_id": "thread-a",
            "title": null,
            "provider": null,
            "model": null,
            "created_at": "1970-01-01T00:00:00Z",
            "updated_at": "1970-01-01T00:00:00Z",
            "message_count": 0
        });

        let result = serde_json::from_value::<ThreadMetadata>(value);

        assert!(result.is_err());
    }

    #[test]
    fn thread_metadata_rejects_blank_or_relative_workspace_when_deserializing() {
        for workspace in ["", "project"] {
            let value = serde_json::json!({
                "thread_id": "thread-a",
                "title": null,
                "workspace": workspace,
                "provider": null,
                "model": null,
                "created_at": "1970-01-01T00:00:00Z",
                "updated_at": "1970-01-01T00:00:00Z",
                "message_count": 0
            });

            let result = serde_json::from_value::<ThreadMetadata>(value);

            assert!(result.is_err(), "workspace {workspace:?} should fail");
        }
    }

    #[test]
    fn thread_usage_metadata_accumulates_cache_hit_rate() {
        let mut usage = ThreadUsageMetadata::default();

        usage.add_token_usage(
            &TokenUsage::new(100, 10, 110)
                .with_cached_prompt_tokens(92)
                .with_cache_creation_prompt_tokens(5),
        );
        usage.add_token_usage(
            &TokenUsage::new(50, 5, 55)
                .with_cached_prompt_tokens(43)
                .with_cache_creation_prompt_tokens(3),
        );

        assert_eq!(usage.prompt_tokens, 150);
        assert_eq!(usage.cached_prompt_tokens, 135);
        assert_eq!(usage.cache_creation_prompt_tokens, 8);
        assert!((usage.cache_hit_rate.unwrap() - 0.9).abs() < f64::EPSILON);
    }

    #[test]
    fn thread_item_events_replay_reasoning_and_final_answer_into_stable_items() {
        let timestamp = OffsetDateTime::UNIX_EPOCH;
        let events = vec![
            ThreadItemEvent {
                seq: 1,
                event_id: "event-1".to_string(),
                thread_id: "thread-1".to_string(),
                turn_id: "turn-1".to_string(),
                timestamp,
                event: ThreadItemEventKind::ItemStarted {
                    item: ThreadItem::Reasoning {
                        id: "turn-1-agent-reasoning".to_string(),
                        summary: Vec::new(),
                        content: vec![String::new()],
                        status: Some(ThreadItemStatus::InProgress),
                    },
                },
            },
            ThreadItemEvent {
                seq: 2,
                event_id: "event-2".to_string(),
                thread_id: "thread-1".to_string(),
                turn_id: "turn-1".to_string(),
                timestamp,
                event: ThreadItemEventKind::ItemDelta {
                    item_id: "turn-1-agent-reasoning".to_string(),
                    delta: ThreadItemDelta::ReasoningText {
                        delta: "Inspecting".to_string(),
                        content_index: 0,
                    },
                },
            },
            ThreadItemEvent {
                seq: 3,
                event_id: "event-3".to_string(),
                thread_id: "thread-1".to_string(),
                turn_id: "turn-1".to_string(),
                timestamp,
                event: ThreadItemEventKind::ItemDelta {
                    item_id: "turn-1-agent-final_answer".to_string(),
                    delta: ThreadItemDelta::AgentMessageText {
                        delta: "Done".to_string(),
                        phase: Some("final_answer".to_string()),
                    },
                },
            },
            ThreadItemEvent {
                seq: 4,
                event_id: "event-4".to_string(),
                thread_id: "thread-1".to_string(),
                turn_id: "turn-1".to_string(),
                timestamp,
                event: ThreadItemEventKind::ItemCompleted {
                    item: ThreadItem::AgentMessage {
                        id: "turn-1-agent-final_answer".to_string(),
                        text: "Done.".to_string(),
                        phase: Some("final_answer".to_string()),
                        status: Some(ThreadItemStatus::Completed),
                    },
                },
            },
        ];

        let turns = project_thread_item_events(&events);

        assert_eq!(turns.len(), 1);
        assert_eq!(turns[0].turn_id, "turn-1");
        assert_eq!(
            turns[0].items,
            vec![
                ThreadItem::Reasoning {
                    id: "turn-1-agent-reasoning".to_string(),
                    summary: Vec::new(),
                    content: vec!["Inspecting".to_string()],
                    status: Some(ThreadItemStatus::InProgress),
                },
                ThreadItem::AgentMessage {
                    id: "turn-1-agent-final_answer".to_string(),
                    text: "Done.".to_string(),
                    phase: Some("final_answer".to_string()),
                    status: Some(ThreadItemStatus::Completed),
                }
            ]
        );
    }
}