tokmesh-core 0.1.5

//! OpenCode session parser
//!
//! Parses messages from:
//! - SQLite database (OpenCode 1.2+): ~/.local/share/opencode/opencode.db
//! - Legacy JSON files: ~/.local/share/opencode/storage/message/
//!
//! The SQLite message schema — and the driver that reads it — is shared with
//! the other clients that adopted it; see `super::opencode_schema`. This
//! module keeps OpenCode's own legacy JSON file parser and its JSON-to-SQLite
//! migration cache.

// The message payload type and the SQLite driver are shared with every other
// client that adopted OpenCode's schema.
use super::opencode_schema::{
    parse_opencode_schema_sqlite, reported_cost, rescan_opencode_schema_sqlite,
    scan_opencode_schema_sqlite, set_workspace_from_root, OpenCodeIncrementalState,
    OpenCodeSchemaConfig, OpenCodeSchemaMessage as OpenCodeMessage, OpenCodeSchemaScan,
};
use super::utils::read_file_or_none;
use super::{normalize_opencode_agent_name, UnifiedMessage};
use crate::{provider_identity, TokenBreakdown};
#[cfg(test)]
use rusqlite::Connection;
use serde::{Deserialize, Serialize};
use std::path::Path;

pub fn parse_opencode_sqlite(db_path: &Path) -> Vec<UnifiedMessage> {
    parse_opencode_schema_sqlite(db_path, OpenCodeSchemaConfig::opencode())
}

/// Full scan that also records where a later scan can resume from.
pub(crate) fn scan_opencode_sqlite(db_path: &Path) -> OpenCodeSchemaScan {
    scan_opencode_schema_sqlite(db_path, OpenCodeSchemaConfig::opencode())
}

/// Resume from `cached_state`, reading only the rows that changed. `None`
/// means the caller has to fall back to [`scan_opencode_sqlite`].
pub(crate) fn rescan_opencode_sqlite(
    db_path: &Path,
    cached_state: &OpenCodeIncrementalState,
    cached_messages: Vec<UnifiedMessage>,
) -> Option<OpenCodeSchemaScan> {
    rescan_opencode_schema_sqlite(
        db_path,
        OpenCodeSchemaConfig::opencode(),
        cached_state,
        cached_messages,
    )
}

pub fn parse_opencode_file(path: &Path) -> Option<UnifiedMessage> {
    let data = read_file_or_none(path)?;
    let mut bytes = data;

    let msg: OpenCodeMessage = simd_json::from_slice(&mut bytes).ok()?;

    // OpenCode JSON files (v1) always carry an explicit role, so require it to
    // be "assistant" here. Missing-role acceptance (is_assistant) is reserved
    // for the v2 `session_message` SQLite path, whose SQL already filters
    // `type = 'assistant'`; applying it to files would count a role-less or
    // malformed file as assistant usage (previously it was skipped when the
    // required `role` field failed to deserialize).
    if msg.role.as_deref() != Some("assistant") {
        return None;
    }

    let workspace_root = msg
        .path
        .as_ref()
        .and_then(|path| path.root.as_deref())
        .map(str::to_string);
    // Resolve model + provider before moving any fields out of `msg`, since
    // both borrow the whole struct to fall back onto the nested `model` object.
    let model_id = msg.resolve_model_id()?;
    let provider_id = msg
        .resolve_provider_id()
        .unwrap_or_else(|| "unknown".to_string());
    let provider_id = provider_identity::canonical_provider(&provider_id).unwrap_or(provider_id);

    let tokens = msg.tokens?;
    // Legacy JSON files carry a complete `cache` object; a missing or partial
    // one has always dropped the message rather than counting it as zero.
    let cache = tokens.cache?;
    let (cache_read, cache_write) = (cache.read?, cache.write?);
    let time = msg.time?;
    let agent_or_mode = msg.mode.or(msg.agent);
    let agent = agent_or_mode.map(|a| normalize_opencode_agent_name(&a));

    let session_id = msg.session_id.unwrap_or_else(|| "unknown".to_string());

    // Embedded message ids are globally stable across the legacy JSON and
    // SQLite representations, so keep them unnamespaced for overlap and fork
    // dedup. A filename is only unique inside its session directory; make the
    // no-id fallback path-scoped so same-named files in separate sessions do
    // not silently collapse (#1198). Canonicalization also keeps one physical
    // file reached through two path spellings on one key. If a non-UTF-8 path
    // cannot be represented, leave the key absent: retaining both candidates
    // is safer than inventing a lossy identity that can undercount.
    let dedup_key = msg.id.or_else(|| legacy_json_path_dedup_key(path));
    let cost = reported_cost(msg.cost).unwrap_or(0.0);

    let mut unified = UnifiedMessage::new_with_agent(
        "opencode",
        model_id,
        provider_id,
        session_id,
        time.created as i64,
        TokenBreakdown {
            input: tokens.input.max(0),
            output: tokens.output.max(0),
            cache_read: cache_read.max(0),
            cache_write: cache_write.max(0),
            reasoning: tokens.reasoning.unwrap_or(0).max(0),
        },
        cost,
        agent,
    );
    unified.duration_ms = time.completed.and_then(|completed| {
        let duration = completed - time.created;
        (duration.is_finite() && duration > 0.0).then_some(duration as i64)
    });
    unified.dedup_key = dedup_key;
    set_workspace_from_root(&mut unified, workspace_root.as_deref());
    // OpenCode computes per-message cost at request time from its own pricing
    // data (models.dev), so a positive `cost` is authoritative and must survive
    // tokmesh's LiteLLM repricing pass. A zero cost usually means OpenCode
    // itself had no pricing for the model — leave it `Unknown` so
    // `apply_pricing_if_available` can still estimate.
    if unified.cost > 0.0 {
        unified.mark_provider_reported_cost();
    }
    Some(unified)
}

fn legacy_json_path_dedup_key(path: &Path) -> Option<String> {
    let canonical = std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf());
    canonical
        .to_str()
        .map(|path| format!("legacy-json-path:{path}"))
}

// =============================================================================
// Migration cache: skip redundant legacy JSON scanning after full migration
// =============================================================================

const MIGRATION_CACHE_FILENAME: &str = "opencode-migration.json";

/// Persisted migration status for OpenCode JSON → SQLite migration.
/// Stored at <config_dir>/cache/opencode-migration.json.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct OpenCodeMigrationCache {
    /// True when every legacy JSON message was already present in SQLite.
    pub migration_complete: bool,
    /// Number of JSON files in the message directory at detection time.
    pub json_file_count: u64,
    /// Modification time of the JSON directory (Unix seconds) at detection time.
    pub json_dir_mtime_secs: u64,
    /// When this entry was written (Unix seconds).
    pub checked_at_secs: u64,
}

fn migration_cache_dir() -> std::path::PathBuf {
    crate::paths::get_cache_dir()
}

fn migration_cache_path() -> std::path::PathBuf {
    migration_cache_dir().join(MIGRATION_CACHE_FILENAME)
}

fn legacy_migration_cache_paths() -> Vec<std::path::PathBuf> {
    Vec::new()
}

/// Load the migration cache from disk. Returns `None` if the file is missing or
/// unparseable.
pub fn load_opencode_migration_cache() -> Option<OpenCodeMigrationCache> {
    let canonical = migration_cache_path();
    match std::fs::read_to_string(&canonical) {
        Ok(content) => serde_json::from_str(&content).ok(),
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
            legacy_migration_cache_paths().into_iter().find_map(|path| {
                let content = std::fs::read_to_string(path).ok()?;
                serde_json::from_str(&content).ok()
            })
        }
        Err(_) => None,
    }
}

/// Persist the migration cache atomically (write to temp file, then rename).
pub fn save_opencode_migration_cache(cache: &OpenCodeMigrationCache) {
    use std::io::Write as _;

    let dir = migration_cache_dir();
    if std::fs::create_dir_all(&dir).is_err() {
        return;
    }

    let content = match serde_json::to_string(cache) {
        Ok(c) => c,
        Err(_) => return,
    };

    let final_path = migration_cache_path();
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::SystemTime::UNIX_EPOCH)
        .map(|d| d.as_nanos() as u64)
        .unwrap_or(0);
    let tmp_name = format!(".opencode-migration.{}.{:x}.tmp", std::process::id(), nanos);
    let tmp_path = dir.join(tmp_name);

    // INVARIANT: All cache writes use atomic temp-file rename. NEVER delete
    // the canonical cache file before writing — a partial save or process
    // crash between delete and rename would lose the cache. The temp-file
    // pattern makes corruption-on-crash impossible.
    let result = (|| -> std::io::Result<()> {
        let mut file = std::fs::File::create(&tmp_path)?;
        file.write_all(content.as_bytes())?;
        file.sync_all()?;
        crate::fs_atomic::replace_file(&tmp_path, &final_path)?;
        Ok(())
    })();

    if result.is_err() {
        let _ = std::fs::remove_file(&tmp_path);
    }
}

/// Return the modification time of `json_dir` as Unix seconds, or `None` on
/// error (directory absent, permissions, etc.).
pub fn get_json_dir_mtime(json_dir: &Path) -> Option<u64> {
    std::fs::metadata(json_dir)
        .ok()?
        .modified()
        .ok()?
        .duration_since(std::time::SystemTime::UNIX_EPOCH)
        .ok()
        .map(|d| d.as_secs())
}

/// Current Unix timestamp in seconds.
pub fn now_secs() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::SystemTime::UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs()
}

#[cfg(test)]
mod tests {
    use super::*;
    fn create_opencode_sqlite_db(db_path: &Path) -> Connection {
        let conn = Connection::open(db_path).unwrap();
        conn.execute_batch(
            "CREATE TABLE message (
                id TEXT PRIMARY KEY,
                session_id TEXT NOT NULL,
                data TEXT NOT NULL
            );",
        )
        .unwrap();
        conn
    }

    /// Build a database shaped like OpenCode v2 (`opencode-next.db`): an empty
    /// `message` table plus the `session_message` + `session` tables that hold
    /// the real per-message data. Mirrors the columns tokmesh actually reads.
    fn create_opencode_v2_sqlite_db(db_path: &Path) -> Connection {
        let conn = Connection::open(db_path).unwrap();
        conn.execute_batch(
            "CREATE TABLE message (
                id TEXT PRIMARY KEY,
                session_id TEXT NOT NULL,
                data TEXT NOT NULL
            );
            CREATE TABLE session (
                id TEXT PRIMARY KEY,
                directory TEXT NOT NULL,
                title TEXT
            );
            CREATE TABLE session_message (
                id TEXT PRIMARY KEY,
                session_id TEXT NOT NULL,
                type TEXT NOT NULL,
                data TEXT NOT NULL
            );",
        )
        .unwrap();
        conn
    }

    /// Build a database shaped like current OpenCode v2: `session_v2` carries
    /// metadata while `session_message` holds the assistant usage payloads.
    fn create_opencode_session_v2_sqlite_db(db_path: &Path) -> Connection {
        let conn = Connection::open(db_path).unwrap();
        conn.execute_batch(
            "CREATE TABLE session_v2 (
                id TEXT PRIMARY KEY,
                directory TEXT NOT NULL,
                title TEXT
            );
            CREATE TABLE session_message (
                id TEXT PRIMARY KEY,
                session_id TEXT NOT NULL,
                type TEXT NOT NULL,
                data TEXT NOT NULL
            );",
        )
        .unwrap();
        conn
    }

    /// A representative v2 assistant payload: no `role` field, model + provider
    /// nested under `$.model`, integer timestamps.
    const V2_ASSISTANT_DATA: &str = r#"{
        "time": { "created": 1783882279705, "completed": 1783882279943 },
        "agent": "build",
        "model": { "id": "claude-sonnet-4", "providerID": "anthropic", "variant": "default" },
        "content": [],
        "finish": "stop",
        "cost": 0.0123,
        "tokens": {
            "input": 5519,
            "output": 20,
            "reasoning": 23,
            "cache": { "read": 100, "write": 50 }
        }
    }"#;

    #[test]
    fn test_deserialize_v2_message_resolves_nested_model() {
        let mut bytes = V2_ASSISTANT_DATA.as_bytes().to_vec();
        let msg: OpenCodeMessage = simd_json::from_slice(&mut bytes).unwrap();

        assert_eq!(msg.role, None, "v2 payloads carry no role field");
        assert!(msg.is_assistant(), "missing role defaults to assistant");
        assert_eq!(msg.resolve_model_id().as_deref(), Some("claude-sonnet-4"));
        assert_eq!(msg.resolve_provider_id().as_deref(), Some("anthropic"));
        assert_eq!(msg.agent.as_deref(), Some("build"));
    }

    #[test]
    fn test_top_level_model_id_takes_precedence_over_nested() {
        let json = r#"{
            "role": "assistant",
            "modelID": "top-level-model",
            "providerID": "top-level-provider",
            "model": { "id": "nested-model", "providerID": "nested-provider" },
            "tokens": { "input": 1, "output": 1, "reasoning": 0, "cache": { "read": 0, "write": 0 } },
            "time": { "created": 1700000000000.0 }
        }"#;
        let mut bytes = json.as_bytes().to_vec();
        let msg: OpenCodeMessage = simd_json::from_slice(&mut bytes).unwrap();

        assert_eq!(msg.resolve_model_id().as_deref(), Some("top-level-model"));
        assert_eq!(
            msg.resolve_provider_id().as_deref(),
            Some("top-level-provider")
        );
    }

    #[test]
    fn test_parse_v2_session_message_reads_tokens_and_workspace() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode-next.db");

        let conn = create_opencode_v2_sqlite_db(&db_path);
        conn.execute(
            "INSERT INTO session (id, directory) VALUES (?1, ?2)",
            rusqlite::params!["ses_v2", "/Users/alice/opencode-v2-repo"],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO session_message (id, session_id, type, data) VALUES (?1, ?2, ?3, ?4)",
            rusqlite::params!["msg_v2_001", "ses_v2", "assistant", V2_ASSISTANT_DATA],
        )
        .unwrap();
        drop(conn);

        let messages = parse_opencode_sqlite(&db_path);
        assert_eq!(messages.len(), 1, "v2 assistant row should be parsed");
        let msg = &messages[0];
        assert_eq!(msg.model_id, "claude-sonnet-4");
        assert_eq!(msg.provider_id, "anthropic");
        assert_eq!(msg.tokens.input, 5519);
        assert_eq!(msg.tokens.output, 20);
        assert_eq!(msg.tokens.reasoning, 23);
        assert_eq!(msg.tokens.cache_read, 100);
        assert_eq!(msg.tokens.cache_write, 50);
        assert_eq!(msg.duration_ms, Some(238));
        assert_eq!(
            msg.workspace_key.as_deref(),
            Some("/Users/alice/opencode-v2-repo"),
            "workspace should come from session.directory"
        );
        assert_eq!(msg.workspace_label.as_deref(), Some("opencode-v2-repo"));
        assert_eq!(
            msg.dedup_key.as_deref(),
            Some("msg_v2_001"),
            "v2 dedup_key falls back to the session_message row id"
        );
        assert_eq!(
            msg.cost_source,
            crate::sessions::CostSource::ProviderReported
        );
    }
    #[test]
    fn test_parse_v2_session_v2_message_reads_tokens_and_metadata() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode.db");
        let conn = create_opencode_session_v2_sqlite_db(&db_path);
        conn.execute(
            "INSERT INTO session_v2 (id, directory, title) VALUES (?1, ?2, ?3)",
            rusqlite::params![
                "ses_current_v2",
                "/Users/alice/current-opencode-repo",
                "Current OpenCode session"
            ],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO session_message (id, session_id, type, data) VALUES (?1, ?2, ?3, ?4)",
            rusqlite::params![
                "msg_current_v2",
                "ses_current_v2",
                "assistant",
                V2_ASSISTANT_DATA
            ],
        )
        .unwrap();
        drop(conn);

        let messages = parse_opencode_sqlite(&db_path);
        assert_eq!(messages.len(), 1, "current session_v2 rows should parse");
        let msg = &messages[0];
        assert_eq!(
            msg.workspace_key.as_deref(),
            Some("/Users/alice/current-opencode-repo")
        );
        assert_eq!(
            msg.workspace_label.as_deref(),
            Some("current-opencode-repo")
        );
        assert_eq!(
            msg.session_title.as_deref(),
            Some("Current OpenCode session")
        );
        assert_eq!(msg.dedup_key.as_deref(), Some("msg_current_v2"));
    }

    #[test]
    fn test_parse_v2_session_message_without_metadata_table() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode.db");
        let conn = Connection::open(&db_path).unwrap();
        conn.execute_batch(
            "CREATE TABLE session_message (
                id TEXT PRIMARY KEY,
                session_id TEXT NOT NULL,
                type TEXT NOT NULL,
                data TEXT NOT NULL
            );",
        )
        .unwrap();
        conn.execute(
            "INSERT INTO session_message (id, session_id, type, data) VALUES (?1, ?2, ?3, ?4)",
            rusqlite::params![
                "msg_without_metadata",
                "ses_without_metadata",
                "assistant",
                V2_ASSISTANT_DATA
            ],
        )
        .unwrap();
        drop(conn);

        let messages = parse_opencode_sqlite(&db_path);
        assert_eq!(
            messages.len(),
            1,
            "usage should parse without session metadata"
        );
        assert_eq!(messages[0].workspace_key, None);
        assert_eq!(messages[0].session_title, None);
        assert_eq!(
            messages[0].dedup_key.as_deref(),
            Some("msg_without_metadata")
        );
    }

    #[test]
    fn test_parse_v2_skips_non_assistant_and_tokenless_rows() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode-next.db");

        let conn = create_opencode_v2_sqlite_db(&db_path);
        let user_data = r#"{ "time": { "created": 1783882279705 }, "content": [] }"#;
        let tokenless = r#"{ "time": { "created": 1783882279705 }, "model": { "id": "m", "providerID": "p" } }"#;
        conn.execute(
            "INSERT INTO session_message (id, session_id, type, data) VALUES (?1, ?2, ?3, ?4)",
            rusqlite::params!["msg_ok", "ses_v2", "assistant", V2_ASSISTANT_DATA],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO session_message (id, session_id, type, data) VALUES (?1, ?2, ?3, ?4)",
            rusqlite::params!["msg_user", "ses_v2", "user", user_data],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO session_message (id, session_id, type, data) VALUES (?1, ?2, ?3, ?4)",
            rusqlite::params!["msg_synthetic", "ses_v2", "synthetic", V2_ASSISTANT_DATA],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO session_message (id, session_id, type, data) VALUES (?1, ?2, ?3, ?4)",
            rusqlite::params!["msg_no_tokens", "ses_v2", "assistant", tokenless],
        )
        .unwrap();
        drop(conn);

        let messages = parse_opencode_sqlite(&db_path);
        assert_eq!(
            messages.len(),
            1,
            "only the assistant row with tokens should parse"
        );
        assert_eq!(messages[0].dedup_key.as_deref(), Some("msg_ok"));
    }

    #[test]
    fn test_parse_v2_negative_tokens_clamped() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode-next.db");

        let conn = create_opencode_v2_sqlite_db(&db_path);
        let negative = r#"{
            "time": { "created": 1783882279705 },
            "model": { "id": "claude-sonnet-4", "providerID": "anthropic" },
            "cost": -1.0,
            "tokens": { "input": -100, "output": -50, "reasoning": -25, "cache": { "read": -200, "write": -10 } }
        }"#;
        conn.execute(
            "INSERT INTO session_message (id, session_id, type, data) VALUES (?1, ?2, ?3, ?4)",
            rusqlite::params!["msg_neg", "ses_v2", "assistant", negative],
        )
        .unwrap();
        drop(conn);

        let messages = parse_opencode_sqlite(&db_path);
        assert_eq!(messages.len(), 1);
        let msg = &messages[0];
        assert_eq!(msg.tokens.input, 0);
        assert_eq!(msg.tokens.output, 0);
        assert_eq!(msg.tokens.reasoning, 0);
        assert_eq!(msg.tokens.cache_read, 0);
        assert_eq!(msg.tokens.cache_write, 0);
        assert!(msg.cost >= 0.0);
    }

    #[test]
    fn test_parse_v2_deduplicates_forked_session_message_history() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode-next.db");

        let conn = create_opencode_v2_sqlite_db(&db_path);
        // Same payload copied into a forked session must collapse to one entry.
        conn.execute(
            "INSERT INTO session_message (id, session_id, type, data) VALUES (?1, ?2, ?3, ?4)",
            rusqlite::params!["root_row", "root_session", "assistant", V2_ASSISTANT_DATA],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO session_message (id, session_id, type, data) VALUES (?1, ?2, ?3, ?4)",
            rusqlite::params!["fork_row", "fork_session", "assistant", V2_ASSISTANT_DATA],
        )
        .unwrap();
        drop(conn);

        let messages = parse_opencode_sqlite(&db_path);
        assert_eq!(
            messages.len(),
            1,
            "forked copies of the same assistant turn collapse inside v2 parsing"
        );
    }

    #[test]
    fn test_distinct_embedded_ids_are_not_merged_despite_fingerprint_collision() {
        // Two genuinely different assistant messages can share every fingerprint
        // field (timestamp, model, tokens, cost, agent). When both carry an
        // embedded `$.id` and the ids DIFFER, they are distinct messages -- not
        // fork copies -- and must be kept separate rather than collapsed.
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode-next.db");
        let conn = create_opencode_v2_sqlite_db(&db_path);

        let payload = |id: &str| {
            format!(
                r#"{{
                    "id": "{id}",
                    "time": {{ "created": 1783882279705, "completed": 1783882279943 }},
                    "agent": "build",
                    "model": {{ "id": "claude-sonnet-4", "providerID": "anthropic" }},
                    "cost": 0.0123,
                    "tokens": {{ "input": 10, "output": 5, "reasoning": 0, "cache": {{ "read": 0, "write": 0 }} }}
                }}"#
            )
        };

        conn.execute(
            "INSERT INTO session_message (id, session_id, type, data) VALUES (?1, ?2, ?3, ?4)",
            rusqlite::params!["row_a", "ses_v2", "assistant", payload("msg_a")],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO session_message (id, session_id, type, data) VALUES (?1, ?2, ?3, ?4)",
            rusqlite::params!["row_b", "ses_v2", "assistant", payload("msg_b")],
        )
        .unwrap();
        // A true fork of msg_a (same embedded id, different session/row) must
        // still collapse into msg_a rather than becoming a third entry.
        conn.execute(
            "INSERT INTO session_message (id, session_id, type, data) VALUES (?1, ?2, ?3, ?4)",
            rusqlite::params!["row_a_fork", "fork_session", "assistant", payload("msg_a")],
        )
        .unwrap();
        drop(conn);

        let mut dedup_keys: Vec<String> = parse_opencode_sqlite(&db_path)
            .into_iter()
            .filter_map(|m| m.dedup_key)
            .collect();
        dedup_keys.sort();
        assert_eq!(
            dedup_keys,
            vec!["msg_a".to_string(), "msg_b".to_string()],
            "distinct embedded ids stay separate; a same-id fork collapses"
        );
    }

    #[test]
    fn test_parse_opencode_structure() {
        let json = r#"{
            "id": "msg_123",
            "sessionID": "ses_456",
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "cost": 0.05,
            "tokens": {
                "input": 1000,
                "output": 500,
                "reasoning": 100,
                "cache": { "read": 200, "write": 50 }
            },
            "time": { "created": 1700000000000.0 }
        }"#;

        let mut bytes = json.as_bytes().to_vec();
        let msg: OpenCodeMessage = simd_json::from_slice(&mut bytes).unwrap();

        assert_eq!(msg.model_id, Some("claude-sonnet-4".to_string()));
        assert_eq!(msg.tokens.unwrap().input, 1000);
        assert_eq!(msg.agent, None);
    }

    #[test]
    fn test_parse_opencode_with_agent() {
        let json = r#"{
            "id": "msg_123",
            "sessionID": "ses_456",
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "agent": "OmO",
            "cost": 0.05,
            "tokens": {
                "input": 1000,
                "output": 500,
                "reasoning": 100,
                "cache": { "read": 200, "write": 50 }
            },
            "time": { "created": 1700000000000.0 }
        }"#;

        let mut bytes = json.as_bytes().to_vec();
        let msg: OpenCodeMessage = simd_json::from_slice(&mut bytes).unwrap();

        assert_eq!(msg.agent, Some("OmO".to_string()));
    }

    /// Verify negative token values are clamped to 0 (defense-in-depth for PR #147)
    #[test]
    fn test_negative_values_clamped_to_zero() {
        use std::io::Write;

        let json = r#"{
            "id": "msg_negative",
            "sessionID": "ses_negative",
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "cost": -0.05,
            "tokens": {
                "input": -100,
                "output": -50,
                "reasoning": -25,
                "cache": { "read": -200, "write": -10 }
            },
            "time": { "created": 1700000000000.0 }
        }"#;

        let mut temp_file = tempfile::Builder::new().suffix(".json").tempfile().unwrap();
        temp_file.write_all(json.as_bytes()).unwrap();

        let result = parse_opencode_file(temp_file.path());
        assert!(result.is_some(), "Should parse file with negative values");

        let msg = result.unwrap();
        assert_eq!(msg.tokens.input, 0, "Negative input should be clamped to 0");
        assert_eq!(
            msg.tokens.output, 0,
            "Negative output should be clamped to 0"
        );
        assert_eq!(
            msg.tokens.cache_read, 0,
            "Negative cache_read should be clamped to 0"
        );
        assert_eq!(
            msg.tokens.cache_write, 0,
            "Negative cache_write should be clamped to 0"
        );
        assert_eq!(
            msg.tokens.reasoning, 0,
            "Negative reasoning should be clamped to 0"
        );
        assert!(
            msg.cost >= 0.0,
            "Negative cost should be clamped to 0.0, got {}",
            msg.cost
        );
    }

    #[test]
    fn test_parse_opencode_file_requires_explicit_assistant_role() {
        use std::io::Write;
        // Regression: making `role` optional for the v2 SQLite path must NOT
        // loosen file parsing. A file without a `role` (or a non-assistant one)
        // is not assistant usage and must be skipped -- the missing-role =>
        // assistant shortcut applies only to the type-filtered session_message
        // SQLite query, never to JSON files.
        let role_less = r#"{
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "tokens": { "input": 10, "output": 5, "reasoning": 0, "cache": { "read": 0, "write": 0 } },
            "time": { "created": 1700000000000.0 }
        }"#;
        let mut f1 = tempfile::Builder::new().suffix(".json").tempfile().unwrap();
        f1.write_all(role_less.as_bytes()).unwrap();
        assert!(
            parse_opencode_file(f1.path()).is_none(),
            "a role-less OpenCode JSON file must not be counted as assistant usage"
        );

        let user_role = r#"{
            "role": "user",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "tokens": { "input": 10, "output": 5, "reasoning": 0, "cache": { "read": 0, "write": 0 } },
            "time": { "created": 1700000000000.0 }
        }"#;
        let mut f2 = tempfile::Builder::new().suffix(".json").tempfile().unwrap();
        f2.write_all(user_role.as_bytes()).unwrap();
        assert!(
            parse_opencode_file(f2.path()).is_none(),
            "a non-assistant OpenCode JSON file must be skipped"
        );
    }

    /// JSON dedup_key uses msg.id when present
    #[test]
    fn test_dedup_key_from_json_message_id() {
        use std::io::Write;

        let json = r#"{
            "id": "msg_dedup_001",
            "sessionID": "ses_001",
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "cost": 0.01,
            "tokens": {
                "input": 100,
                "output": 50,
                "reasoning": 0,
                "cache": { "read": 0, "write": 0 }
            },
            "time": { "created": 1700000000000.0 }
        }"#;

        let mut temp_file = tempfile::Builder::new().suffix(".json").tempfile().unwrap();
        temp_file.write_all(json.as_bytes()).unwrap();

        let msg = parse_opencode_file(temp_file.path()).expect("Should parse");
        assert_eq!(
            msg.dedup_key,
            Some("msg_dedup_001".to_string()),
            "dedup_key should use msg.id from JSON"
        );
    }

    #[test]
    fn test_parse_opencode_file_sets_duration_from_completed_time() {
        use std::io::Write;

        let json = r#"{
            "id": "msg_timed",
            "sessionID": "ses_001",
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "cost": 0.01,
            "tokens": {
                "input": 100,
                "output": 50,
                "reasoning": 0,
                "cache": { "read": 0, "write": 0 }
            },
            "time": { "created": 1700000000000.0, "completed": 1700000001234.0 }
        }"#;

        let mut temp_file = tempfile::Builder::new().suffix(".json").tempfile().unwrap();
        temp_file.write_all(json.as_bytes()).unwrap();

        let msg = parse_opencode_file(temp_file.path()).expect("Should parse");
        assert_eq!(msg.duration_ms, Some(1234));
    }

    /// JSON dedup_key falls back to a path-scoped identity when msg.id is absent.
    #[test]
    fn test_dedup_key_falls_back_to_canonical_file_path() {
        let json = r#"{
            "sessionID": "ses_001",
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "cost": 0.01,
            "tokens": {
                "input": 100,
                "output": 50,
                "reasoning": 0,
                "cache": { "read": 0, "write": 0 }
            },
            "time": { "created": 1700000000000.0 }
        }"#;

        let dir = tempfile::tempdir().unwrap();
        let file_path = dir.path().join("msg_fallback_999.json");
        std::fs::write(&file_path, json).unwrap();

        let msg = parse_opencode_file(&file_path).expect("Should parse");
        assert_eq!(
            msg.dedup_key,
            legacy_json_path_dedup_key(&file_path),
            "an id-less message must use the file's canonical location"
        );
        assert!(msg
            .dedup_key
            .as_deref()
            .is_some_and(|key| key.starts_with("legacy-json-path:")));
    }

    #[test]
    fn same_named_idless_files_in_different_sessions_have_distinct_keys() {
        let json = r#"{
            "sessionID": "embedded-session-is-not-the-fallback",
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "tokens": {
                "input": 100,
                "output": 50,
                "reasoning": 0,
                "cache": { "read": 0, "write": 0 }
            },
            "time": { "created": 1700000000000.0 }
        }"#;
        let root = tempfile::tempdir().unwrap();
        let first = root.path().join("session-a/same-name.json");
        let second = root.path().join("session-b/same-name.json");
        std::fs::create_dir_all(first.parent().unwrap()).unwrap();
        std::fs::create_dir_all(second.parent().unwrap()).unwrap();
        std::fs::write(&first, json).unwrap();
        std::fs::write(&second, json).unwrap();

        let first = parse_opencode_file(&first).unwrap();
        let second = parse_opencode_file(&second).unwrap();

        assert_ne!(first.dedup_key, second.dedup_key);
        assert!(first.dedup_key.is_some());
        assert!(second.dedup_key.is_some());
    }

    /// Non-assistant messages are skipped (no dedup_key produced)
    #[test]
    fn test_dedup_key_skips_non_assistant() {
        let json = r#"{
            "id": "msg_user_001",
            "sessionID": "ses_001",
            "role": "user",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "tokens": {
                "input": 100,
                "output": 50,
                "reasoning": 0,
                "cache": { "read": 0, "write": 0 }
            },
            "time": { "created": 1700000000000.0 }
        }"#;

        let dir = tempfile::tempdir().unwrap();
        let file_path = dir.path().join("msg_user_001.json");
        std::fs::write(&file_path, json).unwrap();

        let result = parse_opencode_file(&file_path);
        assert!(result.is_none(), "User messages should be skipped");
    }

    /// SQLite dedup_key falls back to the database row id when the message has no embedded id.
    #[test]
    fn test_sqlite_dedup_key_falls_back_to_row_id() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test_opencode.db");

        let conn = create_opencode_sqlite_db(&db_path);

        let data_json = r#"{
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "cost": 0.05,
            "tokens": {
                "input": 1000,
                "output": 500,
                "reasoning": 0,
                "cache": { "read": 200, "write": 50 }
            },
            "time": { "created": 1700000000000.0 }
        }"#;

        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
            rusqlite::params!["msg_sqlite_001", "ses_001", data_json],
        )
        .unwrap();
        drop(conn);

        let messages = parse_opencode_sqlite(&db_path);
        assert_eq!(messages.len(), 1);
        assert_eq!(
            messages[0].dedup_key,
            Some("msg_sqlite_001".to_string()),
            "SQLite dedup_key should fall back to the row id when no embedded id exists"
        );
        assert_eq!(messages[0].model_id, "claude-sonnet-4");
        assert_eq!(messages[0].tokens.input, 1000);
    }

    #[test]
    fn test_parse_opencode_file_marks_positive_cost_as_provider_reported() {
        use std::io::Write;
        let json = r#"{
            "id": "msg_cost_001",
            "sessionID": "ses_cost",
            "role": "assistant",
            "modelID": "z-ai/glm-4.6",
            "providerID": "openrouter",
            "cost": 0.0025158,
            "tokens": {
                "input": 2675,
                "output": 28,
                "reasoning": 1,
                "cache": { "read": 7700, "write": 0 }
            },
            "time": { "created": 1765915142201.0 }
        }"#;

        let mut temp_file = tempfile::Builder::new().suffix(".json").tempfile().unwrap();
        temp_file.write_all(json.as_bytes()).unwrap();

        let msg = parse_opencode_file(temp_file.path()).expect("Should parse");
        assert_eq!(
            msg.cost_source,
            crate::sessions::CostSource::ProviderReported,
            "positive embedded cost must survive the LiteLLM repricing pass"
        );
        assert!((msg.cost - 0.0025158).abs() < 1e-12);
    }

    #[test]
    fn test_parse_opencode_file_keeps_zero_cost_unknown_for_estimation() {
        use std::io::Write;
        let json = r#"{
            "id": "msg_cost_002",
            "sessionID": "ses_cost",
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "cost": 0.0,
            "tokens": {
                "input": 1000,
                "output": 500,
                "reasoning": 0,
                "cache": { "read": 0, "write": 0 }
            },
            "time": { "created": 1700000000000.0 }
        }"#;

        let mut temp_file = tempfile::Builder::new().suffix(".json").tempfile().unwrap();
        temp_file.write_all(json.as_bytes()).unwrap();

        let msg = parse_opencode_file(temp_file.path()).expect("Should parse");
        assert_eq!(
            msg.cost_source,
            crate::sessions::CostSource::Unknown,
            "zero cost means OpenCode had no pricing — leave repricing enabled"
        );
    }

    #[test]
    fn test_parse_opencode_sqlite_marks_positive_cost_as_provider_reported() {
        let dir = tempfile::TempDir::new().unwrap();
        let db_path = dir.path().join("test_opencode_cost.db");
        let conn = create_opencode_sqlite_db(&db_path);

        let costed = r#"{
            "role": "assistant",
            "modelID": "z-ai/glm-4.6",
            "providerID": "openrouter",
            "cost": 0.0025158,
            "tokens": {
                "input": 2675,
                "output": 28,
                "reasoning": 1,
                "cache": { "read": 7700, "write": 0 }
            },
            "time": { "created": 1765915142201.0 }
        }"#;
        let free = r#"{
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "cost": 0.0,
            "tokens": {
                "input": 1000,
                "output": 500,
                "reasoning": 0,
                "cache": { "read": 0, "write": 0 }
            },
            "time": { "created": 1700000000000.0 }
        }"#;

        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
            rusqlite::params!["msg_costed", "ses_cost", costed],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
            rusqlite::params!["msg_free", "ses_cost", free],
        )
        .unwrap();
        drop(conn);

        let messages = parse_opencode_sqlite(&db_path);
        assert_eq!(messages.len(), 2);

        let costed_msg = messages
            .iter()
            .find(|m| m.dedup_key.as_deref() == Some("msg_costed"))
            .unwrap();
        assert_eq!(
            costed_msg.cost_source,
            crate::sessions::CostSource::ProviderReported
        );

        let free_msg = messages
            .iter()
            .find(|m| m.dedup_key.as_deref() == Some("msg_free"))
            .unwrap();
        assert_eq!(free_msg.cost_source, crate::sessions::CostSource::Unknown);
    }

    #[test]
    fn test_parse_opencode_file_uses_explicit_path_root_as_workspace() {
        let json = r#"{
            "id": "msg_workspace_001",
            "sessionID": "ses_001",
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "cost": 0.01,
            "tokens": {
                "input": 100,
                "output": 50,
                "reasoning": 0,
                "cache": { "read": 0, "write": 0 }
            },
            "time": { "created": 1700000000000.0 },
            "path": { "root": "/Users/alice/opencode-json-repo" }
        }"#;

        let dir = tempfile::tempdir().unwrap();
        let file_path = dir.path().join("msg_workspace_001.json");
        std::fs::write(&file_path, json).unwrap();

        let msg = parse_opencode_file(&file_path).expect("Should parse");
        assert_eq!(
            msg.workspace_key.as_deref(),
            Some("/Users/alice/opencode-json-repo")
        );
        assert_eq!(msg.workspace_label.as_deref(), Some("opencode-json-repo"));
    }

    #[test]
    fn test_parse_opencode_file_ignores_non_object_path_without_rejecting_message() {
        let json = r#"{
            "id": "msg_path_string_001",
            "sessionID": "ses_001",
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "cost": 0.01,
            "tokens": {
                "input": 100,
                "output": 50,
                "reasoning": 0,
                "cache": { "read": 0, "write": 0 }
            },
            "time": { "created": 1700000000000.0 },
            "path": "/Users/alice/not-object"
        }"#;

        let dir = tempfile::tempdir().unwrap();
        let file_path = dir.path().join("msg_path_string_001.json");
        std::fs::write(&file_path, json).unwrap();

        let msg = parse_opencode_file(&file_path).expect("Should parse");
        assert_eq!(msg.workspace_key, None);
        assert_eq!(msg.workspace_label, None);
    }

    #[test]
    fn test_parse_opencode_sqlite_uses_session_directory_as_workspace() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test_opencode.db");

        let conn = create_opencode_sqlite_db(&db_path);
        conn.execute_batch(
            "CREATE TABLE session (
                id TEXT PRIMARY KEY,
                directory TEXT NOT NULL,
                title TEXT
            );",
        )
        .unwrap();
        conn.execute(
            "INSERT INTO session (id, directory) VALUES (?1, ?2)",
            rusqlite::params!["ses_001", "/Users/alice/opencode-sqlite-repo"],
        )
        .unwrap();

        let data_json = r#"{
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "cost": 0.05,
            "tokens": {
                "input": 1000,
                "output": 500,
                "reasoning": 0,
                "cache": { "read": 200, "write": 50 }
            },
            "time": { "created": 1700000000000.0 }
        }"#;

        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
            rusqlite::params!["msg_sqlite_workspace", "ses_001", data_json],
        )
        .unwrap();
        drop(conn);

        let messages = parse_opencode_sqlite(&db_path);
        assert_eq!(messages.len(), 1);
        assert_eq!(
            messages[0].workspace_key.as_deref(),
            Some("/Users/alice/opencode-sqlite-repo")
        );
        assert_eq!(
            messages[0].workspace_label.as_deref(),
            Some("opencode-sqlite-repo")
        );
    }

    #[test]
    fn test_parse_opencode_sqlite_legacy_fallback_uses_path_root_when_session_table_missing() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test_opencode.db");

        let conn = create_opencode_sqlite_db(&db_path);

        let data_json = r#"{
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "cost": 0.05,
            "tokens": {
                "input": 1000,
                "output": 500,
                "reasoning": 0,
                "cache": { "read": 200, "write": 50 }
            },
            "time": { "created": 1700000000000.0 },
            "path": { "root": "/Users/alice/legacy-fallback-repo" }
        }"#;

        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
            rusqlite::params!["msg_sqlite_legacy_workspace", "ses_001", data_json],
        )
        .unwrap();
        drop(conn);

        let messages = parse_opencode_sqlite(&db_path);
        assert_eq!(messages.len(), 1);
        assert_eq!(
            messages[0].workspace_key.as_deref(),
            Some("/Users/alice/legacy-fallback-repo")
        );
        assert_eq!(
            messages[0].workspace_label.as_deref(),
            Some("legacy-fallback-repo")
        );
        assert_eq!(messages[0].tokens.input, 1000);
    }

    #[test]
    fn test_parse_opencode_sqlite_duplicate_workspace_conflict_is_unknown() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test_opencode.db");

        let conn = create_opencode_sqlite_db(&db_path);
        conn.execute_batch(
            "CREATE TABLE session (
                id TEXT PRIMARY KEY,
                directory TEXT NOT NULL,
                title TEXT
            );",
        )
        .unwrap();
        conn.execute(
            "INSERT INTO session (id, directory) VALUES (?1, ?2)",
            rusqlite::params!["ses_root", "/Users/alice/root-workspace"],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO session (id, directory) VALUES (?1, ?2)",
            rusqlite::params!["ses_fork", "/Users/alice/fork-workspace"],
        )
        .unwrap();

        let data_json = r#"{
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "cost": 0.05,
            "tokens": {
                "input": 1000,
                "output": 500,
                "reasoning": 0,
                "cache": { "read": 200, "write": 50 }
            },
            "time": { "created": 1700000000000.0, "completed": 1700000000500.0 },
            "mode": "build"
        }"#;

        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
            rusqlite::params!["z_root_copy", "ses_root", data_json],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
            rusqlite::params!["a_fork_copy", "ses_fork", data_json],
        )
        .unwrap();
        drop(conn);

        let messages = parse_opencode_sqlite(&db_path);
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].workspace_key, None);
        assert_eq!(messages[0].workspace_label, None);
        assert_eq!(messages[0].tokens.input, 1000);
    }

    /// SQLite prefers the embedded message id when present so JSON/SQLite overlap keeps deduplicating.
    #[test]
    fn test_sqlite_dedup_key_prefers_embedded_message_id() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test_opencode.db");

        let conn = create_opencode_sqlite_db(&db_path);

        let valid = r#"{
            "id": "embedded_msg_001",
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "tokens": { "input": 100, "output": 50, "reasoning": 0, "cache": { "read": 0, "write": 0 } },
            "time": { "created": 1700000000000.0 }
        }"#;

        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
            rusqlite::params!["row_msg_001", "ses_001", valid],
        )
        .unwrap();
        drop(conn);

        let messages = parse_opencode_sqlite(&db_path);
        assert_eq!(messages.len(), 1);
        assert_eq!(
            messages[0].dedup_key,
            Some("embedded_msg_001".to_string()),
            "SQLite dedup_key should prefer the embedded message id for cross-source overlap"
        );
    }

    /// SQLite skips rows without tokens or with non-assistant role
    #[test]
    fn test_sqlite_skips_invalid_rows() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test_opencode.db");

        let conn = create_opencode_sqlite_db(&db_path);

        let valid = r#"{
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "tokens": { "input": 100, "output": 50, "reasoning": 0, "cache": { "read": 0, "write": 0 } },
            "time": { "created": 1700000000000.0 }
        }"#;

        let user_msg = r#"{
            "role": "user",
            "modelID": "claude-sonnet-4",
            "time": { "created": 1700000000000.0 }
        }"#;

        let no_tokens = r#"{
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "time": { "created": 1700000000000.0 }
        }"#;

        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
            rusqlite::params!["msg_valid", "ses_001", valid],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
            rusqlite::params!["msg_user", "ses_001", user_msg],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
            rusqlite::params!["msg_no_tokens", "ses_001", no_tokens],
        )
        .unwrap();
        drop(conn);

        let messages = parse_opencode_sqlite(&db_path);
        assert_eq!(
            messages.len(),
            1,
            "Should only parse valid assistant message"
        );
        assert_eq!(messages[0].dedup_key, Some("msg_valid".to_string()));
    }

    /// Forked SQLite sessions should not count copied history more than once.
    #[test]
    fn test_sqlite_deduplicates_forked_history_rows() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test_opencode.db");
        let conn = create_opencode_sqlite_db(&db_path);

        let root_msg = r#"{
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "cost": 0.05,
            "tokens": {
                "input": 1000,
                "output": 500,
                "reasoning": 25,
                "cache": { "read": 200, "write": 50 }
            },
            "time": { "created": 1700000000000.0, "completed": 1700000000500.0 },
            "mode": "build"
        }"#;

        let new_msg = r#"{
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "cost": 0.08,
            "tokens": {
                "input": 1300,
                "output": 650,
                "reasoning": 40,
                "cache": { "read": 100, "write": 0 }
            },
            "time": { "created": 1700000001000.0, "completed": 1700000001500.0 },
            "mode": "build"
        }"#;

        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
            rusqlite::params!["root_row", "root_session", root_msg],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
            rusqlite::params!["fork_copy_row", "fork_session", root_msg],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
            rusqlite::params!["fork_new_row", "fork_session", new_msg],
        )
        .unwrap();
        drop(conn);

        let messages = parse_opencode_sqlite(&db_path);
        assert_eq!(
            messages.len(),
            2,
            "Forked copies of the same assistant history should collapse inside SQLite parsing"
        );
        assert_eq!(messages[0].tokens.input, 1000);
        assert_eq!(messages[1].tokens.input, 1300);
    }

    /// Same-timestamp messages with different payloads should remain distinct.
    #[test]
    fn test_sqlite_same_timestamp_distinct_payloads_survive() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test_opencode.db");
        let conn = create_opencode_sqlite_db(&db_path);

        let first = r#"{
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "cost": 0.05,
            "tokens": {
                "input": 1000,
                "output": 500,
                "reasoning": 0,
                "cache": { "read": 0, "write": 0 }
            },
            "time": { "created": 1700000000000.0, "completed": 1700000000100.0 },
            "mode": "build"
        }"#;

        let second = r#"{
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "cost": 0.05,
            "tokens": {
                "input": 1500,
                "output": 750,
                "reasoning": 0,
                "cache": { "read": 0, "write": 0 }
            },
            "time": { "created": 1700000000000.0, "completed": 1700000000100.0 },
            "mode": "build"
        }"#;

        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
            rusqlite::params!["row_one", "session_one", first],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
            rusqlite::params!["row_two", "session_two", second],
        )
        .unwrap();
        drop(conn);

        let messages = parse_opencode_sqlite(&db_path);
        assert_eq!(
            messages.len(),
            2,
            "Distinct assistant calls should survive even when they share the same creation timestamp"
        );
    }

    /// Cross-source dedup: matching IDs between SQLite and JSON should deduplicate
    #[test]
    fn test_cross_source_dedup_by_message_id() {
        use std::collections::HashSet;

        let dir = tempfile::tempdir().unwrap();

        // --- SQLite source ---
        let db_path = dir.path().join("opencode.db");
        let conn = Connection::open(&db_path).unwrap();
        conn.execute_batch(
            "CREATE TABLE message (
                id TEXT PRIMARY KEY,
                session_id TEXT NOT NULL,
                data TEXT NOT NULL
            );",
        )
        .unwrap();

        let shared_data_json = r#"{
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "tokens": { "input": 500, "output": 200, "reasoning": 0, "cache": { "read": 0, "write": 0 } },
            "time": { "created": 1700000000000.0 }
        }"#;
        let sqlite_only_data_json = r#"{
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "tokens": { "input": 700, "output": 250, "reasoning": 0, "cache": { "read": 0, "write": 0 } },
            "time": { "created": 1700000001000.0 }
        }"#;

        // Insert two messages into SQLite
        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
            rusqlite::params!["msg_shared_001", "ses_001", shared_data_json],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, ?2, ?3)",
            rusqlite::params!["msg_sqlite_only", "ses_001", sqlite_only_data_json],
        )
        .unwrap();
        drop(conn);

        // --- JSON source ---
        let json_dir = dir.path().join("json");
        std::fs::create_dir_all(&json_dir).unwrap();

        // Duplicate of SQLite msg_shared_001
        let json_shared = r#"{
            "id": "msg_shared_001",
            "sessionID": "ses_001",
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "tokens": { "input": 500, "output": 200, "reasoning": 0, "cache": { "read": 0, "write": 0 } },
            "time": { "created": 1700000000000.0 }
        }"#;
        std::fs::write(json_dir.join("msg_shared_001.json"), json_shared).unwrap();

        // JSON-only message (not in SQLite)
        let json_only = r#"{
            "id": "msg_json_only",
            "sessionID": "ses_001",
            "role": "assistant",
            "modelID": "claude-sonnet-4",
            "providerID": "anthropic",
            "tokens": { "input": 100, "output": 50, "reasoning": 0, "cache": { "read": 0, "write": 0 } },
            "time": { "created": 1700000000000.0 }
        }"#;
        std::fs::write(json_dir.join("msg_json_only.json"), json_only).unwrap();

        // --- Simulate the dedup logic from lib.rs ---
        let sqlite_messages = parse_opencode_sqlite(&db_path);
        assert_eq!(sqlite_messages.len(), 2);

        // Build seen set from SQLite (same as lib.rs)
        let mut seen: HashSet<String> = HashSet::new();
        for msg in &sqlite_messages {
            if let Some(ref key) = msg.dedup_key {
                seen.insert(key.clone());
            }
        }
        assert_eq!(seen.len(), 2);

        // Parse JSON files
        let json_msg_shared = parse_opencode_file(&json_dir.join("msg_shared_001.json")).unwrap();
        let json_msg_only = parse_opencode_file(&json_dir.join("msg_json_only.json")).unwrap();

        // Filter JSON through seen set (same logic as lib.rs)
        let json_messages = vec![json_msg_shared, json_msg_only];
        let deduped: Vec<UnifiedMessage> = json_messages
            .into_iter()
            .filter(|msg| {
                msg.dedup_key
                    .as_ref()
                    .is_none_or(|key| seen.insert(key.clone()))
            })
            .collect();

        // msg_shared_001 should be filtered (duplicate), msg_json_only should survive
        assert_eq!(
            deduped.len(),
            1,
            "Only the JSON-only message should survive dedup"
        );
        assert_eq!(
            deduped[0].dedup_key,
            Some("msg_json_only".to_string()),
            "Surviving message should be the JSON-only one"
        );

        // Total unique messages = 2 from SQLite + 1 from JSON
        let total = sqlite_messages.len() + deduped.len();
        assert_eq!(total, 3, "Should have 3 unique messages total");
    }

    // -------------------------------------------------------------------------
    // Migration cache tests
    // -------------------------------------------------------------------------

    /// Round-trip: save then load returns identical data.
    #[test]
    fn test_migration_cache_round_trip() {
        let dir = tempfile::tempdir().unwrap();
        // Point the cache at a temp dir by overriding via a temporary env var is
        // impractical here; instead we test the structs and serde directly.
        let cache = OpenCodeMigrationCache {
            migration_complete: true,
            json_file_count: 42,
            json_dir_mtime_secs: 1_700_000_000,
            checked_at_secs: 1_700_100_000,
        };

        let json = serde_json::to_string(&cache).unwrap();
        let loaded: OpenCodeMigrationCache = serde_json::from_str(&json).unwrap();
        assert_eq!(loaded, cache);

        // Ensure the JSON contains all expected keys
        assert!(json.contains("migration_complete"));
        assert!(json.contains("json_file_count"));
        assert!(json.contains("json_dir_mtime_secs"));
        assert!(json.contains("checked_at_secs"));

        drop(dir);
    }

    /// Cache is valid when file count and mtime are unchanged.
    #[test]
    fn test_migration_cache_valid_when_unchanged() {
        let dir = tempfile::tempdir().unwrap();
        let json_dir = dir.path().join("message");
        std::fs::create_dir_all(&json_dir).unwrap();

        // Write a dummy file so the directory exists and has a stable mtime
        std::fs::write(json_dir.join("msg.json"), b"{}").unwrap();

        let current_mtime = get_json_dir_mtime(&json_dir).expect("should stat dir");
        let current_file_count = 1u64;

        let cache = OpenCodeMigrationCache {
            migration_complete: true,
            json_file_count: current_file_count,
            json_dir_mtime_secs: current_mtime, // same mtime
            checked_at_secs: now_secs(),
        };

        // Simulate the validity check from lib.rs
        let is_valid = cache.migration_complete
            && current_file_count == cache.json_file_count
            && get_json_dir_mtime(&json_dir).is_some_and(|m| m <= cache.json_dir_mtime_secs);

        assert!(is_valid, "Cache should be valid when count and mtime match");
    }

    /// Cache is invalid when file count has changed.
    #[test]
    fn test_migration_cache_invalid_when_file_count_changes() {
        let dir = tempfile::tempdir().unwrap();
        let json_dir = dir.path().join("message");
        std::fs::create_dir_all(&json_dir).unwrap();
        std::fs::write(json_dir.join("msg1.json"), b"{}").unwrap();

        let current_mtime = get_json_dir_mtime(&json_dir).unwrap();

        let cache = OpenCodeMigrationCache {
            migration_complete: true,
            json_file_count: 1,
            json_dir_mtime_secs: current_mtime,
            checked_at_secs: now_secs(),
        };

        // Simulate: a new file was added → current_file_count = 2
        let current_file_count = 2u64; // changed
        let is_valid = cache.migration_complete
            && current_file_count == cache.json_file_count
            && get_json_dir_mtime(&json_dir).is_some_and(|m| m <= cache.json_dir_mtime_secs);

        assert!(!is_valid, "Cache should be invalid when file count changes");
    }

    /// Cache is invalid when directory mtime is newer than cached value.
    #[test]
    fn test_migration_cache_invalid_when_mtime_newer() {
        let dir = tempfile::tempdir().unwrap();
        let json_dir = dir.path().join("message");
        std::fs::create_dir_all(&json_dir).unwrap();
        std::fs::write(json_dir.join("msg.json"), b"{}").unwrap();

        let current_mtime = get_json_dir_mtime(&json_dir).unwrap();

        // Simulate: cache recorded an older mtime → directory is now newer
        let stale_mtime = current_mtime.saturating_sub(1);
        let cache = OpenCodeMigrationCache {
            migration_complete: true,
            json_file_count: 1,
            json_dir_mtime_secs: stale_mtime, // older than current
            checked_at_secs: now_secs(),
        };

        let is_valid = cache.migration_complete
            && 1u64 == cache.json_file_count
            && get_json_dir_mtime(&json_dir).is_some_and(|m| m <= cache.json_dir_mtime_secs);

        assert!(
            !is_valid,
            "Cache should be invalid when directory mtime is newer than cached value"
        );
    }

    /// Cache is not loaded when the file is missing (load returns None).
    #[test]
    fn test_migration_cache_missing_returns_none() {
        // load_opencode_migration_cache reads from ~/.cache/tokmesh/opencode-migration.json
        // We can't easily override the path in a unit test, but we can verify that
        // serde_json::from_str returns None for invalid input (simulating missing file).
        let result: Option<OpenCodeMigrationCache> = serde_json::from_str("").ok();
        assert!(
            result.is_none(),
            "Empty/missing content should produce None"
        );
    }

    /// migration_complete=false disables the cache even if count/mtime match.
    #[test]
    fn test_migration_cache_not_skipped_when_incomplete() {
        let dir = tempfile::tempdir().unwrap();
        let json_dir = dir.path().join("message");
        std::fs::create_dir_all(&json_dir).unwrap();
        std::fs::write(json_dir.join("msg.json"), b"{}").unwrap();

        let current_mtime = get_json_dir_mtime(&json_dir).unwrap();

        let cache = OpenCodeMigrationCache {
            migration_complete: false, // migration not complete
            json_file_count: 1,
            json_dir_mtime_secs: current_mtime,
            checked_at_secs: now_secs(),
        };

        let is_valid = cache.migration_complete
            && 1u64 == cache.json_file_count
            && get_json_dir_mtime(&json_dir).is_some_and(|m| m <= cache.json_dir_mtime_secs);

        assert!(
            !is_valid,
            "Cache should not allow skipping when migration_complete=false"
        );
    }

    // =========================================================================
    // Incremental SQLite scan
    // =========================================================================

    /// A `message` table with the columns a modern OpenCode database actually
    /// has. The existing fixtures above deliberately keep the minimal column
    /// set, which is itself a case the incremental lane has to refuse.
    fn create_timed_v1_db(db_path: &Path) -> Connection {
        let conn = Connection::open(db_path).unwrap();
        conn.execute_batch(
            "CREATE TABLE session (
                id TEXT PRIMARY KEY,
                directory TEXT NOT NULL,
                title TEXT,
                time_updated INTEGER NOT NULL
            );
            CREATE TABLE message (
                id TEXT PRIMARY KEY,
                session_id TEXT NOT NULL,
                time_created INTEGER NOT NULL,
                time_updated INTEGER NOT NULL,
                data TEXT NOT NULL
            );
            INSERT INTO session (id, directory, title, time_updated)
            VALUES ('ses_1', '/tmp/project', 'A session', 500);",
        )
        .unwrap();
        conn
    }

    fn timed_v1_payload(id: &str, output: i64) -> String {
        format!(
            r#"{{
                "id": "{id}",
                "role": "assistant",
                "sessionID": "ses_1",
                "modelID": "claude-sonnet-4",
                "providerID": "anthropic",
                "cost": 0.5,
                "tokens": {{
                    "input": 10,
                    "output": {output},
                    "reasoning": 0,
                    "cache": {{ "read": 0, "write": 0 }}
                }},
                "time": {{ "created": 1783882279705, "completed": 1783882279943 }}
            }}"#
        )
    }

    fn timed_v1_payload_without_id(output: i64) -> String {
        format!(
            r#"{{
                "role": "assistant",
                "sessionID": "ses_1",
                "modelID": "claude-sonnet-4",
                "providerID": "anthropic",
                "cost": 0.5,
                "tokens": {{
                    "input": 10,
                    "output": {output},
                    "reasoning": 0,
                    "cache": {{ "read": 0, "write": 0 }}
                }},
                "time": {{ "created": 1783882279705, "completed": 1783882279943 }}
            }}"#
        )
    }

    fn insert_timed_v1_message(conn: &Connection, id: &str, created: i64, output: i64) {
        conn.execute(
            "INSERT INTO message (id, session_id, time_created, time_updated, data)
             VALUES (?1, 'ses_1', ?2, ?2, ?3)",
            rusqlite::params![id, created, timed_v1_payload(id, output)],
        )
        .unwrap();
    }

    /// Rewrite a row's payload and stamp it as updated at `updated`, which is
    /// what OpenCode does to a message long after inserting it.
    fn touch_timed_v1_message(conn: &Connection, id: &str, updated: i64, output: i64) {
        let changed = conn
            .execute(
                "UPDATE message SET data = ?2, time_updated = ?3 WHERE id = ?1",
                rusqlite::params![id, timed_v1_payload(id, output), updated],
            )
            .unwrap();
        assert_eq!(changed, 1, "fixture row {id} should exist");
    }

    fn by_dedup_key(messages: &[UnifiedMessage]) -> Vec<UnifiedMessage> {
        let mut sorted = messages.to_vec();
        sorted.sort_by(|left, right| left.dedup_key.cmp(&right.dedup_key));
        sorted
    }

    fn output_tokens(messages: &[UnifiedMessage], dedup_key: &str) -> i64 {
        messages
            .iter()
            .find(|message| message.dedup_key.as_deref() == Some(dedup_key))
            .unwrap_or_else(|| panic!("{dedup_key} should be present"))
            .tokens
            .output
    }

    /// SplitMix64 keeps this stress test deterministic without adding a
    /// production dependency just to choose fixture mutations.
    fn next_mutation_word(state: &mut u64) -> u64 {
        *state = state.wrapping_add(0x9e37_79b9_7f4a_7c15);
        let mut word = *state;
        word = (word ^ (word >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
        word = (word ^ (word >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb);
        word ^ (word >> 31)
    }

    fn mutation_index(state: &mut u64, len: usize) -> usize {
        (next_mutation_word(state) % len as u64) as usize
    }

    struct RandomizedFixtureRow {
        row_id: String,
        dedup_key: String,
        qualified: bool,
    }

    #[test]
    #[serial_test::serial]
    fn test_randomized_incremental_mutations_match_a_full_parse() {
        const SEEDS: usize = 48;
        const MUTATIONS_PER_SEED: usize = 4;
        const MUTATION_NAMES: [&str; 5] = ["insert", "delete", "rewrite", "re-key", "disqualify"];

        let mut mutation_counts = [0_usize; MUTATION_NAMES.len()];
        let mut resumed = 0_usize;
        let mut fell_back = 0_usize;

        for seed in 0..SEEDS {
            let dir = tempfile::tempdir().unwrap();
            let db_path = dir.path().join("opencode.db");
            let conn = create_timed_v1_db(&db_path);
            let mut rows = Vec::new();
            for row_index in 0..6 {
                let row_id = format!("row_{row_index}");
                let dedup_key = format!("msg_{row_index}");
                conn.execute(
                    "INSERT INTO message (id, session_id, time_created, time_updated, data)
                     VALUES (?1, 'ses_1', ?2, ?2, ?3)",
                    rusqlite::params![
                        row_id,
                        1_000 + row_index,
                        timed_v1_payload(&dedup_key, 10 + row_index)
                    ],
                )
                .unwrap();
                rows.push(RandomizedFixtureRow {
                    row_id,
                    dedup_key,
                    qualified: true,
                });
            }
            drop(conn);

            let baseline = scan_opencode_sqlite(&db_path);
            let state = baseline
                .incremental
                .clone()
                .expect("the timed fixture must produce an incremental mark");
            let mut random = (seed as u64 + 1).wrapping_mul(0xd134_2543_de82_ef95);
            let conn = Connection::open(&db_path).unwrap();

            for step in 0..MUTATIONS_PER_SEED {
                let mutation = mutation_index(&mut random, MUTATION_NAMES.len());
                mutation_counts[mutation] += 1;
                let updated = 10_000 + step as i64;

                match mutation {
                    0 => {
                        let row_id = format!("inserted_row_{seed}_{step}");
                        let dedup_key = format!("inserted_msg_{seed}_{step}");
                        let output = (next_mutation_word(&mut random) % 900 + 1) as i64;
                        conn.execute(
                            "INSERT INTO message (id, session_id, time_created, time_updated, data)
                             VALUES (?1, 'ses_1', ?2, ?2, ?3)",
                            rusqlite::params![
                                row_id,
                                updated,
                                timed_v1_payload(&dedup_key, output)
                            ],
                        )
                        .unwrap();
                        rows.push(RandomizedFixtureRow {
                            row_id,
                            dedup_key,
                            qualified: true,
                        });
                    }
                    1 => {
                        let row = rows.remove(mutation_index(&mut random, rows.len()));
                        let changed = conn
                            .execute(
                                "DELETE FROM message WHERE id = ?1",
                                rusqlite::params![row.row_id],
                            )
                            .unwrap();
                        assert_eq!(changed, 1, "seed {seed}: delete target should exist");
                    }
                    2 => {
                        let row_index = mutation_index(&mut random, rows.len());
                        let output = (next_mutation_word(&mut random) % 900 + 1) as i64;
                        let row = &mut rows[row_index];
                        let changed = conn
                            .execute(
                                "UPDATE message SET data = ?2, time_updated = ?3 WHERE id = ?1",
                                rusqlite::params![
                                    row.row_id,
                                    timed_v1_payload(&row.dedup_key, output),
                                    updated
                                ],
                            )
                            .unwrap();
                        assert_eq!(changed, 1, "seed {seed}: rewrite target should exist");
                        row.qualified = true;
                    }
                    3 => {
                        let row_index = mutation_index(&mut random, rows.len());
                        let collision_keys: Vec<String> = rows
                            .iter()
                            .enumerate()
                            .filter(|(index, row)| *index != row_index && row.qualified)
                            .map(|(_, row)| row.dedup_key.clone())
                            .collect();
                        let use_existing_key =
                            next_mutation_word(&mut random) & 3 == 0 && !collision_keys.is_empty();
                        let dedup_key = if use_existing_key {
                            collision_keys[mutation_index(&mut random, collision_keys.len())]
                                .clone()
                        } else {
                            format!("rekeyed_msg_{seed}_{step}")
                        };
                        let output = (next_mutation_word(&mut random) % 900 + 1) as i64;
                        let row = &mut rows[row_index];
                        let changed = conn
                            .execute(
                                "UPDATE message SET data = ?2, time_updated = ?3 WHERE id = ?1",
                                rusqlite::params![
                                    row.row_id,
                                    timed_v1_payload(&dedup_key, output),
                                    updated
                                ],
                            )
                            .unwrap();
                        assert_eq!(changed, 1, "seed {seed}: re-key target should exist");
                        row.dedup_key = dedup_key;
                        row.qualified = true;
                    }
                    4 => {
                        let row_index = mutation_index(&mut random, rows.len());
                        let row = &mut rows[row_index];
                        let payload = format!(
                            r#"{{"id":"{}","sessionID":"ses_1","role":"user"}}"#,
                            row.dedup_key
                        );
                        disqualify_timed_v1_message(&conn, &row.row_id, updated, &payload);
                        row.qualified = false;
                    }
                    _ => unreachable!(),
                }
            }
            drop(conn);

            let effective_warm = match rescan_opencode_sqlite(&db_path, &state, baseline.messages) {
                Some(warm) => {
                    resumed += 1;
                    warm.messages
                }
                None => {
                    fell_back += 1;
                    scan_opencode_sqlite(&db_path).messages
                }
            };
            let full = scan_opencode_sqlite(&db_path);
            assert_eq!(
                by_dedup_key(&effective_warm),
                by_dedup_key(&full.messages),
                "seed {seed}: a warm rescan or its conservative fallback must match a full parse"
            );
        }

        for (name, count) in MUTATION_NAMES.into_iter().zip(mutation_counts) {
            assert!(count > 0, "the deterministic corpus must exercise {name}");
        }
        assert!(
            resumed >= SEEDS * 3 / 4,
            "the optimization must remain useful across mixed mutations: resumed {resumed}/{SEEDS}"
        );
        assert!(
            fell_back > 0,
            "the corpus must also exercise a conservative full-scan fallback"
        );
    }

    #[test]
    fn test_incremental_rescan_matches_a_full_parse_of_the_same_state() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode.db");
        let conn = create_timed_v1_db(&db_path);
        insert_timed_v1_message(&conn, "msg_a", 1_000, 11);
        insert_timed_v1_message(&conn, "msg_b", 2_000, 22);
        drop(conn);

        let cold = scan_opencode_sqlite(&db_path);
        assert_eq!(cold.messages.len(), 2);
        let state = cold
            .incremental
            .clone()
            .expect("a table with the time columns is resumable");

        let conn = Connection::open(&db_path).unwrap();
        touch_timed_v1_message(&conn, "msg_a", 9_000, 111);
        conn.execute(
            "INSERT INTO message (id, session_id, time_created, time_updated, data)
             VALUES ('msg_c', 'ses_1', 500, 9500, ?1)",
            rusqlite::params![timed_v1_payload("msg_c", 33)],
        )
        .unwrap();
        drop(conn);

        let warm = rescan_opencode_sqlite(&db_path, &state, cold.messages)
            .expect("an insert-only delta stays incremental");
        let full = scan_opencode_sqlite(&db_path);

        assert_eq!(
            by_dedup_key(&warm.messages),
            by_dedup_key(&full.messages),
            "a warm incremental scan and a cold full parse must agree"
        );
        assert_eq!(warm.messages.len(), 3);
        assert_eq!(output_tokens(&warm.messages, "msg_a"), 111);
        assert_eq!(output_tokens(&warm.messages, "msg_c"), 33);
    }

    #[test]
    fn same_millisecond_rewrite_matches_full_scan() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("opencode.db");
        let conn = create_timed_v1_db(&path);
        insert_timed_v1_message(&conn, "msg_a", 2_000, 11);
        let cold = scan_opencode_sqlite(&path);
        let state = cold.incremental.clone().unwrap();
        touch_timed_v1_message(&conn, "msg_a", 2_000, 999);
        let warm = rescan_opencode_sqlite(&path, &state, cold.messages)
            .unwrap_or_else(|| scan_opencode_sqlite(&path));
        let full = scan_opencode_sqlite(&path);
        assert_eq!(output_tokens(&full.messages, "msg_a"), 999);
        assert_eq!(output_tokens(&warm.messages, "msg_a"), 999);
    }

    #[test]
    fn test_incremental_rescan_reads_a_row_rewritten_long_after_it_was_inserted() {
        // The row that changes is the *oldest* one and sorts first by id, so a
        // mark keyed on the row id -- the Codex `consumed_offset` analogue --
        // would skip it. On a real database 99.98% of rows are rewritten after
        // insert, so that mark would under-report nearly everything.
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode.db");
        let conn = create_timed_v1_db(&db_path);
        insert_timed_v1_message(&conn, "msg_aaa", 1_000, 11);
        insert_timed_v1_message(&conn, "msg_zzz", 2_000, 22);
        drop(conn);

        let cold = scan_opencode_sqlite(&db_path);
        let state = cold.incremental.clone().unwrap();
        assert_eq!(output_tokens(&cold.messages, "msg_aaa"), 11);

        let conn = Connection::open(&db_path).unwrap();
        touch_timed_v1_message(&conn, "msg_aaa", 9_000, 999);
        drop(conn);

        let warm = rescan_opencode_sqlite(&db_path, &state, cold.messages)
            .expect("an in-place rewrite is not a deletion");

        assert_eq!(
            output_tokens(&warm.messages, "msg_aaa"),
            999,
            "a row rewritten after insert must reach the incremental scan"
        );
        assert_eq!(warm.messages.len(), 2, "a rewrite must not duplicate a row");
        assert_eq!(
            by_dedup_key(&warm.messages),
            by_dedup_key(&scan_opencode_sqlite(&db_path).messages)
        );
    }

    #[test]
    fn test_incremental_rescan_removes_a_deleted_row_by_provenance() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode.db");
        let conn = create_timed_v1_db(&db_path);
        insert_timed_v1_message(&conn, "msg_a", 1_000, 11);
        insert_timed_v1_message(&conn, "msg_b", 2_000, 22);
        drop(conn);

        let cold = scan_opencode_sqlite(&db_path);
        let state = cold.incremental.clone().unwrap();

        // OpenCode cascades a session delete onto its messages. The row
        // inventory makes that absence explicit even though no delta query can
        // return the deleted row.
        let conn = Connection::open(&db_path).unwrap();
        conn.execute("DELETE FROM message WHERE id = 'msg_b'", [])
            .unwrap();
        drop(conn);

        let warm = rescan_opencode_sqlite(&db_path, &state, cold.messages)
            .expect("an ordinary deletion is exact from row provenance");
        let full = scan_opencode_sqlite(&db_path);
        assert_eq!(by_dedup_key(&warm.messages), by_dedup_key(&full.messages));
        assert_eq!(full.messages.len(), 1);
        assert_eq!(full.messages[0].dedup_key.as_deref(), Some("msg_a"));
    }

    #[test]
    fn test_incremental_rescan_handles_a_delete_masked_by_an_insert() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode.db");
        let conn = create_timed_v1_db(&db_path);
        insert_timed_v1_message(&conn, "msg_a", 1_000, 11);
        insert_timed_v1_message(&conn, "msg_b", 2_000, 22);
        drop(conn);

        let cold = scan_opencode_sqlite(&db_path);
        let state = cold.incremental.clone().unwrap();

        let conn = Connection::open(&db_path).unwrap();
        conn.execute("DELETE FROM message WHERE id = 'msg_b'", [])
            .unwrap();
        // The replacement's creation time moves backwards, so the old
        // count/high-water inference sees neither an insert nor a deletion.
        // Its own update marker is current, and row provenance identifies both
        // physical changes directly.
        conn.execute(
            "INSERT INTO message (id, session_id, time_created, time_updated, data)
             VALUES ('msg_c', 'ses_1', 500, 9500, ?1)",
            rusqlite::params![timed_v1_payload("msg_c", 33)],
        )
        .unwrap();
        drop(conn);

        let warm = rescan_opencode_sqlite(&db_path, &state, cold.messages)
            .expect("row identity distinguishes the deletion from the insert");
        let full = scan_opencode_sqlite(&db_path);
        assert_eq!(by_dedup_key(&warm.messages), by_dedup_key(&full.messages));
    }

    /// Rewrite a row so the usage queries stop selecting it, without changing
    /// the row count. `payload` replaces the whole `data` object.
    fn disqualify_timed_v1_message(conn: &Connection, id: &str, updated: i64, payload: &str) {
        let changed = conn
            .execute(
                "UPDATE message SET data = ?2, time_updated = ?3 WHERE id = ?1",
                rusqlite::params![id, payload, updated],
            )
            .unwrap();
        assert_eq!(changed, 1, "fixture row {id} should exist");
    }

    /// A rename touches the session row and nothing else, so the message
    /// high-water does not move and the incremental scan reads no rows. Without
    /// a metadata refresh the cached messages keep the old title forever, and a
    /// cold parse disagrees with the cache indefinitely.
    #[test]
    fn test_incremental_rescan_picks_up_a_session_renamed_without_new_messages() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode.db");
        let conn = create_timed_v1_db(&db_path);
        insert_timed_v1_message(&conn, "msg_a", 1_000, 11);
        insert_timed_v1_message(&conn, "msg_b", 2_000, 22);
        drop(conn);

        let cold = scan_opencode_sqlite(&db_path);
        let state = cold.incremental.clone().unwrap();
        assert_eq!(cold.messages[0].session_title.as_deref(), Some("A session"));

        // Rename the session and move its directory. No message row changes.
        let conn = Connection::open(&db_path).unwrap();
        conn.execute(
            "UPDATE session SET title = 'Renamed session', directory = '/tmp/moved',
                    time_updated = 9000 WHERE id = 'ses_1'",
            [],
        )
        .unwrap();
        drop(conn);

        let warm = rescan_opencode_sqlite(&db_path, &state, cold.messages)
            .expect("a rename must not cost a full re-parse");
        let full = scan_opencode_sqlite(&db_path);

        assert_eq!(
            by_dedup_key(&warm.messages),
            by_dedup_key(&full.messages),
            "a warm rescan must agree with a cold parse after a rename"
        );
        for message in &warm.messages {
            assert_eq!(message.session_title.as_deref(), Some("Renamed session"));
        }
    }

    /// Both generations scan into one message list and a cached message does
    /// not record which produced it, so a session id present in both metadata
    /// tables cannot be re-stamped without one generation overwriting the
    /// other's title and workspace. A full scan is the correct answer there.
    #[test]
    fn test_incremental_rescan_refuses_a_session_id_shared_by_both_schemas() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode.db");
        let conn = create_timed_v1_db(&db_path);
        insert_timed_v1_message(&conn, "msg_a", 1_000, 11);
        // A half-migrated database: the same session id in the v2 table too.
        conn.execute_batch(
            "CREATE TABLE session_v2 (
                id TEXT PRIMARY KEY,
                directory TEXT NOT NULL,
                title TEXT,
                time_updated INTEGER NOT NULL
            );
            CREATE TABLE session_message (
                id TEXT PRIMARY KEY,
                session_id TEXT NOT NULL,
                type TEXT NOT NULL,
                time_created INTEGER NOT NULL,
                time_updated INTEGER NOT NULL,
                data TEXT NOT NULL
            );
            INSERT INTO session_v2 (id, directory, title, time_updated)
            VALUES ('ses_1', '/tmp/v2', 'V2 title', 500);",
        )
        .unwrap();
        drop(conn);

        let cold = scan_opencode_sqlite(&db_path);
        let Some(state) = cold.incremental.clone() else {
            // Refusing to mark at all is also a safe answer here.
            return;
        };

        let conn = Connection::open(&db_path).unwrap();
        conn.execute(
            "UPDATE session_v2 SET title = 'V2 renamed', time_updated = 9000 WHERE id = 'ses_1'",
            [],
        )
        .unwrap();
        conn.execute(
            "UPDATE session SET title = 'V1 renamed', time_updated = 9000 WHERE id = 'ses_1'",
            [],
        )
        .unwrap();
        drop(conn);

        assert!(
            rescan_opencode_sqlite(&db_path, &state, cold.messages).is_none(),
            "a session id in both metadata tables must force a full re-parse"
        );
    }

    /// A row whose embedded `$.id` changes has a new dedup key. Row provenance
    /// links both identities to the same physical source, so the old message is
    /// removed and the new one replaces it rather than being appended beside it.
    ///
    /// Two shapes, because the merge's content digest only catches one of them:
    /// an identity-only rewrite has the same digest as the cached message, but
    /// one that also changes usage does not.
    #[test]
    fn test_incremental_rescan_replaces_a_row_whose_embedded_id_changes() {
        for (label, output) in [("identity only", 11), ("identity and usage", 99)] {
            let dir = tempfile::tempdir().unwrap();
            let db_path = dir.path().join("opencode.db");
            let conn = create_timed_v1_db(&db_path);
            insert_timed_v1_message(&conn, "msg_a", 1_000, 11);
            drop(conn);

            let cold = scan_opencode_sqlite(&db_path);
            let state = cold.incremental.clone().unwrap();
            assert_eq!(cold.messages.len(), 1);
            assert_eq!(cold.messages[0].dedup_key.as_deref(), Some("msg_a"));

            // The row keeps its SQLite id but starts carrying its own id, which
            // is the key the parser prefers.
            let conn = Connection::open(&db_path).unwrap();
            conn.execute(
                "UPDATE message SET data = ?1, time_updated = 9000 WHERE id = 'msg_a'",
                rusqlite::params![timed_v1_payload("embedded_a", output)],
            )
            .unwrap();
            drop(conn);

            let full = scan_opencode_sqlite(&db_path);
            assert_eq!(full.messages.len(), 1, "{label}: a cold parse sees one row");

            let warm = rescan_opencode_sqlite(&db_path, &state, cold.messages)
                .unwrap_or_else(|| panic!("{label}: a key change should stay incremental"));
            assert_eq!(
                by_dedup_key(&warm.messages),
                by_dedup_key(&full.messages),
                "{label}: a warm scan must replace the old physical row"
            );
        }
    }

    #[test]
    fn test_incremental_rescan_replaces_an_embedded_id_with_the_row_id() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode.db");
        let conn = create_timed_v1_db(&db_path);
        conn.execute(
            "INSERT INTO message (id, session_id, time_created, time_updated, data)
             VALUES ('row_a', 'ses_1', 1000, 1000, ?1)",
            rusqlite::params![timed_v1_payload("embedded_a", 11)],
        )
        .unwrap();
        drop(conn);

        let cold = scan_opencode_sqlite(&db_path);
        let state = cold.incremental.clone().unwrap();
        assert_eq!(cold.messages[0].dedup_key.as_deref(), Some("embedded_a"));

        let conn = Connection::open(&db_path).unwrap();
        conn.execute(
            "UPDATE message SET data = ?1, time_updated = 9000 WHERE id = 'row_a'",
            rusqlite::params![timed_v1_payload_without_id(22)],
        )
        .unwrap();
        drop(conn);

        let warm = rescan_opencode_sqlite(&db_path, &state, cold.messages)
            .expect("losing an embedded id should stay incremental");
        let full = scan_opencode_sqlite(&db_path);
        assert_eq!(by_dedup_key(&warm.messages), by_dedup_key(&full.messages));
        assert_eq!(warm.messages[0].dedup_key.as_deref(), Some("row_a"));
        assert_eq!(warm.messages[0].tokens.output, 22);
    }

    #[test]
    fn test_incremental_rescan_removes_a_row_that_lost_its_tokens() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode.db");
        let conn = create_timed_v1_db(&db_path);
        insert_timed_v1_message(&conn, "msg_a", 1_000, 11);
        insert_timed_v1_message(&conn, "msg_b", 2_000, 22);
        drop(conn);

        let cold = scan_opencode_sqlite(&db_path);
        let state = cold.incremental.clone().unwrap();
        assert_eq!(cold.messages.len(), 2);

        let conn = Connection::open(&db_path).unwrap();
        disqualify_timed_v1_message(
            &conn,
            "msg_b",
            9_000,
            r#"{"id": "msg_b", "sessionID": "ses_1", "role": "assistant"}"#,
        );
        drop(conn);

        let warm = rescan_opencode_sqlite(&db_path, &state, cold.messages)
            .expect("the parser's rejected outcome removes the old message");
        let full = scan_opencode_sqlite(&db_path);
        assert_eq!(by_dedup_key(&warm.messages), by_dedup_key(&full.messages));
        assert_eq!(full.messages.len(), 1);
        assert_eq!(full.messages[0].dedup_key.as_deref(), Some("msg_a"));
    }

    #[test]
    fn test_incremental_rescan_removes_a_row_that_stopped_being_an_assistant_turn() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode.db");
        let conn = create_timed_v1_db(&db_path);
        insert_timed_v1_message(&conn, "msg_a", 1_000, 11);
        insert_timed_v1_message(&conn, "msg_b", 2_000, 22);
        drop(conn);

        let cold = scan_opencode_sqlite(&db_path);
        let state = cold.incremental.clone().unwrap();

        let conn = Connection::open(&db_path).unwrap();
        disqualify_timed_v1_message(
            &conn,
            "msg_b",
            9_000,
            r#"{"id": "msg_b", "sessionID": "ses_1", "role": "user",
                "tokens": {"input": 1, "output": 2, "cache": {"read": 0, "write": 0}}}"#,
        );
        drop(conn);

        let warm = rescan_opencode_sqlite(&db_path, &state, cold.messages)
            .expect("the changed-row parser owns qualification semantics");
        let full = scan_opencode_sqlite(&db_path);
        assert_eq!(by_dedup_key(&warm.messages), by_dedup_key(&full.messages));
    }

    #[test]
    fn test_incremental_rescan_uses_parser_qualification_for_changed_rows() {
        // SQL still sees an assistant row with a tokens object. The parser
        // rejects it because the model identity disappeared. A duplicated SQL
        // qualification predicate cannot express that rule without drifting.
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode.db");
        let conn = create_timed_v1_db(&db_path);
        insert_timed_v1_message(&conn, "msg_a", 1_000, 11);
        drop(conn);

        let cold = scan_opencode_sqlite(&db_path);
        let state = cold.incremental.clone().unwrap();

        let conn = Connection::open(&db_path).unwrap();
        disqualify_timed_v1_message(
            &conn,
            "msg_a",
            9_000,
            r#"{"id":"msg_a","sessionID":"ses_1","role":"assistant",
                "tokens":{"input":1,"output":2,"cache":{"read":0,"write":0}},
                "time":{"created":1783882279705}}"#,
        );
        drop(conn);

        let warm = rescan_opencode_sqlite(&db_path, &state, cold.messages)
            .expect("a parser rejection should be an explicit row outcome");
        assert!(warm.messages.is_empty());
        assert!(scan_opencode_sqlite(&db_path).messages.is_empty());
    }

    #[test]
    fn test_incremental_rescan_keeps_going_when_a_never_counted_row_changes() {
        // The guard above must not fire on ordinary traffic. User turns are
        // rewritten constantly -- 92,002 of the 92,028 non-assistant rows on a
        // real 14 GB database carry `time_updated > time_created` -- and none
        // of them ever backed a cached message, so none of them is evidence
        // that the cache went stale.
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode.db");
        let conn = create_timed_v1_db(&db_path);
        insert_timed_v1_message(&conn, "msg_a", 1_000, 11);
        conn.execute(
            "INSERT INTO message (id, session_id, time_created, time_updated, data)
             VALUES ('msg_user', 'ses_1', 2_000, 2_000, ?1)",
            rusqlite::params![r#"{"id": "msg_user", "sessionID": "ses_1", "role": "user"}"#],
        )
        .unwrap();
        drop(conn);

        let cold = scan_opencode_sqlite(&db_path);
        let state = cold.incremental.clone().unwrap();
        assert_eq!(cold.messages.len(), 1, "only the assistant turn is usage");

        let conn = Connection::open(&db_path).unwrap();
        disqualify_timed_v1_message(
            &conn,
            "msg_user",
            9_000,
            r#"{"id": "msg_user", "sessionID": "ses_1", "role": "user", "text": "edited"}"#,
        );
        drop(conn);

        let warm = rescan_opencode_sqlite(&db_path, &state, cold.messages)
            .expect("a rewritten user turn must not cost a full re-parse");
        assert_eq!(
            by_dedup_key(&warm.messages),
            by_dedup_key(&scan_opencode_sqlite(&db_path).messages)
        );
    }

    #[test]
    fn test_a_table_without_the_time_columns_records_no_mark() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode.db");
        let conn = create_opencode_sqlite_db(&db_path);
        conn.execute(
            "INSERT INTO message (id, session_id, data) VALUES (?1, 'ses_1', ?2)",
            rusqlite::params!["msg_a", timed_v1_payload("msg_a", 11)],
        )
        .unwrap();
        drop(conn);

        let cold = scan_opencode_sqlite(&db_path);
        assert_eq!(cold.messages.len(), 1);
        assert!(
            cold.incremental.is_none(),
            "rows that cannot be rescanned incrementally must not be marked as if they could"
        );
    }

    #[test]
    fn test_incremental_rescan_refuses_a_row_that_took_part_in_a_merge() {
        // Forked history puts one message id on two rows, and the fingerprint
        // dedup collapses them into a single entry. That entry is the only
        // trace the second row left, so re-reading either side cannot
        // reconstruct what the other contributed -- the scan has to re-parse.
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode.db");
        let conn = create_timed_v1_db(&db_path);
        for (row_id, created) in [("msg_fork_a", 1_000_i64), ("msg_fork_b", 1_500)] {
            conn.execute(
                "INSERT INTO message (id, session_id, time_created, time_updated, data)
                 VALUES (?1, 'ses_1', ?2, ?2, ?3)",
                rusqlite::params![row_id, created, timed_v1_payload("msg_shared", 11)],
            )
            .unwrap();
        }
        drop(conn);

        let cold = scan_opencode_sqlite(&db_path);
        assert_eq!(
            cold.messages.len(),
            1,
            "forked copies collapse to one entry"
        );
        let state = cold.incremental.clone().unwrap();
        assert!(
            state.merged_dedup_keys.contains(&"msg_shared".to_string()),
            "the collapse has to be recorded for the next scan to notice it"
        );

        let conn = Connection::open(&db_path).unwrap();
        conn.execute(
            "UPDATE message SET data = ?2, time_updated = ?3 WHERE id = ?1",
            rusqlite::params!["msg_fork_b", timed_v1_payload("msg_shared", 77), 9_000],
        )
        .unwrap();
        drop(conn);

        assert!(
            rescan_opencode_sqlite(&db_path, &state, cold.messages).is_none(),
            "a rewritten fork copy must re-parse rather than guess at the collapse"
        );
        assert_eq!(scan_opencode_sqlite(&db_path).messages.len(), 2);
    }

    #[test]
    fn test_incremental_rescan_refuses_a_new_row_that_would_collapse() {
        // A fork created after the mark copies completed turns verbatim. A full
        // scan collapses each copy into the original; appending them would
        // count every copied turn twice.
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode.db");
        let conn = create_timed_v1_db(&db_path);
        insert_timed_v1_message(&conn, "msg_a", 1_000, 11);
        drop(conn);

        let cold = scan_opencode_sqlite(&db_path);
        assert_eq!(cold.messages.len(), 1);
        let state = cold.incremental.clone().unwrap();
        assert!(state.merged_dedup_keys.is_empty());

        let conn = Connection::open(&db_path).unwrap();
        conn.execute(
            "INSERT INTO message (id, session_id, time_created, time_updated, data)
             VALUES ('msg_a_copy', 'ses_1', 9_500, 9_500, ?1)",
            rusqlite::params![timed_v1_payload("msg_a", 11)],
        )
        .unwrap();
        drop(conn);

        assert!(
            rescan_opencode_sqlite(&db_path, &state, cold.messages).is_none(),
            "a copied turn must re-parse rather than be appended beside its original"
        );
        assert_eq!(
            scan_opencode_sqlite(&db_path).messages.len(),
            1,
            "the full parse still collapses the copy"
        );
    }

    #[test]
    fn test_incremental_rescan_refuses_a_mark_from_a_different_schema_variant() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("opencode.db");
        let conn = Connection::open(&db_path).unwrap();
        conn.execute_batch(
            "CREATE TABLE message (
                id TEXT PRIMARY KEY,
                session_id TEXT NOT NULL,
                time_created INTEGER NOT NULL,
                time_updated INTEGER NOT NULL,
                data TEXT NOT NULL
            );",
        )
        .unwrap();
        insert_timed_v1_message(&conn, "msg_a", 1_000, 11);
        drop(conn);

        let cold = scan_opencode_sqlite(&db_path);
        let state = cold.incremental.clone().unwrap();
        assert_eq!(cold.messages[0].workspace_key, None);

        // The session metadata table appears, so a full scan would now pick the
        // joining variant and resolve a workspace the cached rows never had.
        let conn = Connection::open(&db_path).unwrap();
        conn.execute_batch(
            "CREATE TABLE session (
                id TEXT PRIMARY KEY,
                directory TEXT NOT NULL,
                title TEXT
            );
            INSERT INTO session (id, directory, title)
            VALUES ('ses_1', '/tmp/project', 'A session');",
        )
        .unwrap();
        drop(conn);

        assert!(
            rescan_opencode_sqlite(&db_path, &state, cold.messages).is_none(),
            "a variant change must invalidate the mark"
        );
        assert!(scan_opencode_sqlite(&db_path).messages[0]
            .workspace_key
            .is_some());
    }
}

#[cfg(test)]
mod integration_tests {
    use super::*;
    use std::path::PathBuf;

    #[test]
    #[ignore] // Run manually with: cargo test integration -- --ignored
    fn test_parse_real_sqlite_db() {
        let home = std::env::var("HOME").unwrap();
        let db_path = PathBuf::from(format!("{}/.local/share/opencode/opencode.db", home));

        if !db_path.exists() {
            println!("Skipping: OpenCode database not found at {:?}", db_path);
            return;
        }

        let messages = parse_opencode_sqlite(&db_path);
        println!("Parsed {} messages from SQLite", messages.len());

        if !messages.is_empty() {
            let first = &messages[0];
            println!(
                "First message: model={}, provider={}, tokens={:?}",
                first.model_id, first.provider_id, first.tokens
            );
        }

        assert!(
            !messages.is_empty(),
            "Expected to parse some messages from SQLite"
        );
    }
}