use chrono::Utc;
use sqlx::SqlitePool;
use tiycore::types::Usage;

use crate::model::errors::AppError;
use crate::model::thread::{RunStatus, RunSummaryDto, RunUsageDto};

#[derive(sqlx::FromRow)]
struct RunRow {
    id: String,
    thread_id: String,
    run_mode: String,
    status: String,
    model_id: Option<String>,
    effective_model_plan_json: Option<String>,
    error_message: Option<String>,
    started_at: String,
    input_tokens: i64,
    output_tokens: i64,
    cache_read_tokens: i64,
    cache_write_tokens: i64,
    total_tokens: i64,
}

/// Full run record for insert.
pub struct RunInsert {
    pub id: String,
    pub thread_id: String,
    pub profile_id: Option<String>,
    pub run_mode: String,
    pub provider_id: Option<String>,
    pub model_id: Option<String>,
    pub effective_model_plan_json: Option<String>,
    pub status: String,
}

#[derive(Debug, Clone, sqlx::FromRow)]
pub struct RunStartContext {
    pub run_mode: String,
    pub effective_model_plan_json: Option<String>,
}

pub async fn insert(pool: &SqlitePool, r: &RunInsert) -> Result<(), AppError> {
    let now = Utc::now().to_rfc3339();
    sqlx::query(
        "INSERT INTO thread_runs (id, thread_id, profile_id, run_mode,
                provider_id, model_id, effective_model_plan_json, status, started_at)
         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
    )
    .bind(&r.id)
    .bind(&r.thread_id)
    .bind(&r.profile_id)
    .bind(&r.run_mode)
    .bind(&r.provider_id)
    .bind(&r.model_id)
    .bind(&r.effective_model_plan_json)
    .bind(&r.status)
    .bind(&now)
    .execute(pool)
    .await?;
    Ok(())
}

pub async fn update_status(pool: &SqlitePool, id: &str, status: RunStatus) -> Result<(), AppError> {
    let is_entering_running = status == RunStatus::Running;
    let is_terminal = status.is_terminal();

    if is_terminal {
        // Terminal: settle any open running segment, set finished_at.
        let now = Utc::now().to_rfc3339();
        sqlx::query(
            "UPDATE thread_runs SET \
                status = ?, \
                elapsed_running_secs = elapsed_running_secs + \
                    CASE WHEN running_since IS NOT NULL \
                         THEN CAST(strftime('%s', 'now') - strftime('%s', running_since) AS INTEGER) \
                         ELSE 0 END, \
                running_since = NULL, \
                finished_at = ? \
             WHERE id = ?",
        )
        .bind(status.as_str())
        .bind(&now)
        .bind(id)
        .execute(pool)
        .await?;
    } else if is_entering_running {
        // Entering Running: record running_since if not already set (idempotent).
        sqlx::query(
            "UPDATE thread_runs SET \
                status = ?, \
                running_since = CASE WHEN running_since IS NULL \
                                     THEN strftime('%Y-%m-%dT%H:%M:%fZ', 'now') \
                                     ELSE running_since END \
             WHERE id = ?",
        )
        .bind(status.as_str())
        .bind(id)
        .execute(pool)
        .await?;
    } else {
        // Leaving Running (→ WaitingApproval / NeedsReply / WaitingToolResult etc.)
        // or non-Running → non-Running: settle any open running segment.
        sqlx::query(
            "UPDATE thread_runs SET \
                status = ?, \
                elapsed_running_secs = elapsed_running_secs + \
                    CASE WHEN running_since IS NOT NULL \
                         THEN CAST(strftime('%s', 'now') - strftime('%s', running_since) AS INTEGER) \
                         ELSE 0 END, \
                running_since = NULL \
             WHERE id = ?",
        )
        .bind(status.as_str())
        .bind(id)
        .execute(pool)
        .await?;
    }

    Ok(())
}

pub async fn update_usage(pool: &SqlitePool, id: &str, usage: &Usage) -> Result<(), AppError> {
    sqlx::query(
        "UPDATE thread_runs
         SET input_tokens = ?, output_tokens = ?, cache_read_tokens = ?,
             cache_write_tokens = ?, total_tokens = ?
         WHERE id = ?",
    )
    .bind(i64::try_from(usage.input).unwrap_or(i64::MAX))
    .bind(i64::try_from(usage.output).unwrap_or(i64::MAX))
    .bind(i64::try_from(usage.cache_read).unwrap_or(i64::MAX))
    .bind(i64::try_from(usage.cache_write).unwrap_or(i64::MAX))
    .bind(i64::try_from(usage.total_tokens).unwrap_or(i64::MAX))
    .bind(id)
    .execute(pool)
    .await?;

    Ok(())
}

pub async fn set_error_message(
    pool: &SqlitePool,
    id: &str,
    error_message: &str,
) -> Result<(), AppError> {
    sqlx::query("UPDATE thread_runs SET error_message = ? WHERE id = ?")
        .bind(error_message)
        .bind(id)
        .execute(pool)
        .await?;
    Ok(())
}

pub async fn find_effective_model_plan_json(
    pool: &SqlitePool,
    id: &str,
) -> Result<Option<String>, AppError> {
    let value = sqlx::query_scalar::<_, Option<String>>(
        "SELECT effective_model_plan_json
         FROM thread_runs
         WHERE id = ?
         LIMIT 1",
    )
    .bind(id)
    .fetch_optional(pool)
    .await?
    .flatten();

    Ok(value)
}

pub async fn find_start_context_by_id(
    pool: &SqlitePool,
    id: &str,
) -> Result<Option<RunStartContext>, AppError> {
    let row = sqlx::query_as::<_, RunStartContext>(
        "SELECT run_mode, effective_model_plan_json
         FROM thread_runs
         WHERE id = ?
         LIMIT 1",
    )
    .bind(id)
    .fetch_optional(pool)
    .await?;

    Ok(row)
}

pub async fn find_latest_evaluable_run_id_by_thread(
    pool: &SqlitePool,
    thread_id: &str,
) -> Result<Option<String>, AppError> {
    let value = sqlx::query_scalar::<_, String>(
        "SELECT id
         FROM thread_runs
         WHERE thread_id = ?
           AND status IN ('completed', 'interrupted')
           AND finished_at IS NOT NULL
         ORDER BY started_at DESC
         LIMIT 1",
    )
    .bind(thread_id)
    .fetch_optional(pool)
    .await?;

    Ok(value)
}

/// Find the currently active (non-terminal) run for a thread.
pub async fn find_active_by_thread(
    pool: &SqlitePool,
    thread_id: &str,
) -> Result<Option<RunSummaryDto>, AppError> {
    let row = sqlx::query_as::<_, RunRow>(
        // Excludes all non-progressing statuses — see RunStatus::non_progressing_sql_in_clause()
        "SELECT id, thread_id, run_mode, status, model_id, effective_model_plan_json,
                error_message, started_at, input_tokens, output_tokens,
                cache_read_tokens, cache_write_tokens, total_tokens
         FROM thread_runs
         WHERE thread_id = ?
           AND status NOT IN ('completed','failed','denied','interrupted','cancelled','limit_reached','waiting_approval','needs_reply')
         ORDER BY started_at DESC
         LIMIT 1",
    )
    .bind(thread_id)
    .fetch_optional(pool)
    .await?;

    Ok(row.map(map_run_summary))
}

/// Find the latest run for a thread (any status), used for ThreadStatus derivation.
pub async fn find_latest_by_thread(
    pool: &SqlitePool,
    thread_id: &str,
) -> Result<Option<RunSummaryDto>, AppError> {
    let row = sqlx::query_as::<_, RunRow>(
        "SELECT id, thread_id, run_mode, status, model_id, effective_model_plan_json,
                error_message, started_at, input_tokens, output_tokens,
                cache_read_tokens, cache_write_tokens, total_tokens
         FROM thread_runs
         WHERE thread_id = ?
         ORDER BY started_at DESC
         LIMIT 1",
    )
    .bind(thread_id)
    .fetch_optional(pool)
    .await?;

    Ok(row.map(map_run_summary))
}

/// Find the latest historical run for a thread that has non-zero prompt usage
/// (`input_tokens + cache_read_tokens`), excluding the currently created run.
/// This is used to seed conservative context-token calibration for the next
/// run on the same thread.
pub async fn find_latest_with_prompt_usage_by_thread_excluding_run(
    pool: &SqlitePool,
    thread_id: &str,
    excluded_run_id: &str,
) -> Result<Option<RunSummaryDto>, AppError> {
    let row = sqlx::query_as::<_, RunRow>(
        "SELECT id, thread_id, run_mode, status, model_id, effective_model_plan_json,
                error_message, started_at, input_tokens, output_tokens,
                cache_read_tokens, cache_write_tokens, total_tokens
         FROM thread_runs
         WHERE thread_id = ?
           AND id != ?
           AND (input_tokens > 0 OR cache_read_tokens > 0)
         ORDER BY started_at DESC
         LIMIT 1",
    )
    .bind(thread_id)
    .bind(excluded_run_id)
    .fetch_optional(pool)
    .await?;

    Ok(row.map(map_run_summary))
}

pub async fn list_thread_ids_with_active_runs(pool: &SqlitePool) -> Result<Vec<String>, AppError> {
    let rows = sqlx::query_scalar::<_, String>(
        // Excludes all non-progressing statuses — see RunStatus::non_progressing_sql_in_clause()
        "SELECT DISTINCT thread_id
         FROM thread_runs
         WHERE status NOT IN ('completed','failed','denied','interrupted','cancelled','limit_reached','waiting_approval','needs_reply')
           AND finished_at IS NULL",
    )
    .fetch_all(pool)
    .await?;

    Ok(rows)
}

/// Cancel all `waiting_approval` runs for a thread, setting them to `cancelled`
/// with a `finished_at` timestamp.  Called when a pending plan approval is
/// superseded by a new run so the old run does not linger as a zombie.
pub async fn cancel_waiting_approval_by_thread(
    pool: &SqlitePool,
    thread_id: &str,
) -> Result<u64, AppError> {
    let result = sqlx::query(
        "UPDATE thread_runs
         SET status = 'cancelled',
             finished_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
         WHERE thread_id = ?
           AND status = 'waiting_approval'
           AND finished_at IS NULL",
    )
    .bind(thread_id)
    .execute(pool)
    .await?;

    Ok(result.rows_affected())
}

/// Mark all non-terminal runs for a thread as interrupted (crash recovery).
pub async fn interrupt_active_runs(pool: &SqlitePool) -> Result<u64, AppError> {
    let result = sqlx::query(
        // Excludes all non-progressing statuses — see RunStatus::non_progressing_sql_in_clause()
        "UPDATE thread_runs
         SET status = 'interrupted',
             error_message = COALESCE(
                 error_message,
                 'The app closed or the run was terminated before completion. Restarted in interrupted state.'
             ),
             finished_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
         WHERE status NOT IN ('completed','failed','denied','interrupted','cancelled','limit_reached','waiting_approval','needs_reply')
           AND finished_at IS NULL",
    )
    .execute(pool)
    .await?;

    Ok(result.rows_affected())
}

/// Returns the status of a specific run by ID.
pub async fn find_status_by_id(pool: &SqlitePool, id: &str) -> Result<Option<String>, AppError> {
    let row = sqlx::query_scalar::<_, String>("SELECT status FROM thread_runs WHERE id = ?")
        .bind(id)
        .fetch_optional(pool)
        .await?;
    Ok(row)
}

fn map_run_summary(row: RunRow) -> RunSummaryDto {
    let (model_display_name, context_window) =
        extract_primary_model_details(row.effective_model_plan_json.as_deref());

    RunSummaryDto {
        id: row.id,
        thread_id: row.thread_id,
        run_mode: row.run_mode,
        status: row.status,
        model_id: row.model_id,
        model_display_name,
        context_window,
        error_message: row.error_message,
        started_at: row.started_at,
        usage: RunUsageDto {
            input_tokens: row.input_tokens.max(0) as u64,
            output_tokens: row.output_tokens.max(0) as u64,
            cache_read_tokens: row.cache_read_tokens.max(0) as u64,
            cache_write_tokens: row.cache_write_tokens.max(0) as u64,
            total_tokens: row.total_tokens.max(0) as u64,
            // Derive the cross-protocol unified context size on read —
            // see the doc-comment in the `RunUsageDto` declaration.
            // The DB schema stores the four per-bucket token columns;
            // `context_size` is a derived projection that we re-emit
            // on every read so the frontend always has a unified
            // "context occupancy" figure to display.
            context_size: (row.input_tokens
                + row.output_tokens
                + row.cache_read_tokens
                + row.cache_write_tokens)
                .max(0) as u64,
        },
    }
}

fn extract_primary_model_details(
    effective_model_plan_json: Option<&str>,
) -> (Option<String>, Option<String>) {
    let Some(raw) = effective_model_plan_json else {
        return (None, None);
    };

    let Ok(value) = serde_json::from_str::<serde_json::Value>(raw) else {
        return (None, None);
    };

    let primary = value.get("primary").and_then(|entry| entry.as_object());
    let model_display_name = primary
        .and_then(|entry| {
            entry
                .get("modelDisplayName")
                .and_then(|value| value.as_str())
        })
        .or_else(|| primary.and_then(|entry| entry.get("model").and_then(|value| value.as_str())))
        .map(str::to_string);
    let context_window = primary
        .and_then(|entry| entry.get("contextWindow").and_then(|value| value.as_str()))
        .map(str::to_string);

    (model_display_name, context_window)
}

#[cfg(test)]
mod tests {
    use super::*;
    use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
    use sqlx::Row;
    use std::str::FromStr;

    async fn setup_test_pool() -> SqlitePool {
        let options = SqliteConnectOptions::from_str("sqlite::memory:")
            .expect("invalid sqlite options")
            .foreign_keys(true);

        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect_with(options)
            .await
            .expect("failed to create in-memory pool");

        crate::persistence::sqlite::run_migrations(&pool)
            .await
            .expect("migrations failed");

        sqlx::query(
            "INSERT INTO workspaces (id, name, path, canonical_path, display_path,
                    is_default, is_git, auto_work_tree, status, created_at, updated_at)
             VALUES ('ws-1', 'ws', '/tmp', '/tmp', '/tmp', 0, 0, 0, 'ready',
                     strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),
                     strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))",
        )
        .execute(&pool)
        .await
        .expect("seed workspace");

        sqlx::query(
            "INSERT INTO threads (id, workspace_id, title, status, created_at, updated_at, last_active_at)
             VALUES ('t1', 'ws-1', 't', 'idle',
                     strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),
                     strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),
                     strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))",
        )
        .execute(&pool)
        .await
        .expect("seed thread t1");

        sqlx::query(
            "INSERT INTO threads (id, workspace_id, title, status, created_at, updated_at, last_active_at)
             VALUES ('t2', 'ws-1', 't2', 'idle',
                     strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),
                     strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),
                     strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))",
        )
        .execute(&pool)
        .await
        .expect("seed thread t2");

        pool
    }

    async fn insert_run_with_started_at(
        pool: &SqlitePool,
        id: &str,
        thread_id: &str,
        started_at: &str,
        input_tokens: i64,
    ) {
        insert_run_with_usage(pool, id, thread_id, started_at, input_tokens, 0).await;
    }

    async fn insert_run_with_usage(
        pool: &SqlitePool,
        id: &str,
        thread_id: &str,
        started_at: &str,
        input_tokens: i64,
        cache_read_tokens: i64,
    ) {
        sqlx::query(
            "INSERT INTO thread_runs (
                id, thread_id, profile_id, run_mode, provider_id, model_id,
                effective_model_plan_json, status, started_at, input_tokens,
                output_tokens, cache_read_tokens, cache_write_tokens, total_tokens
             )
             VALUES (?, ?, NULL, 'default', NULL, NULL, NULL, 'completed', ?, ?, 0, ?, 0, ?)",
        )
        .bind(id)
        .bind(thread_id)
        .bind(started_at)
        .bind(input_tokens)
        .bind(cache_read_tokens)
        .bind(input_tokens.saturating_add(cache_read_tokens))
        .execute(pool)
        .await
        .expect("seed run");
    }

    async fn set_run_finished_at(pool: &SqlitePool, id: &str, finished_at: &str) {
        sqlx::query("UPDATE thread_runs SET finished_at = ? WHERE id = ?")
            .bind(finished_at)
            .bind(id)
            .execute(pool)
            .await
            .expect("set finished_at");
    }

    async fn set_run_elapsed_tracking(
        pool: &SqlitePool,
        id: &str,
        elapsed_running_secs: i64,
        running_since: Option<&str>,
    ) {
        sqlx::query(
            "UPDATE thread_runs
             SET elapsed_running_secs = ?, running_since = ?
             WHERE id = ?",
        )
        .bind(elapsed_running_secs)
        .bind(running_since)
        .bind(id)
        .execute(pool)
        .await
        .expect("set elapsed tracking");
    }

    #[tokio::test]
    async fn find_latest_with_prompt_usage_returns_none_when_no_matching_history_exists() {
        let pool = setup_test_pool().await;

        assert!(
            find_latest_with_prompt_usage_by_thread_excluding_run(&pool, "t1", "run-x")
                .await
                .unwrap()
                .is_none()
        );

        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 0).await;

        assert!(
            find_latest_with_prompt_usage_by_thread_excluding_run(&pool, "t1", "run-x")
                .await
                .unwrap()
                .is_none()
        );
    }

    #[tokio::test]
    async fn find_latest_with_prompt_usage_filters_thread_zero_usage_and_excluded_run() {
        let pool = setup_test_pool().await;

        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 10).await;
        insert_run_with_started_at(&pool, "run-2", "t1", "2026-04-22T09:05:00Z", 0).await;
        insert_run_with_started_at(&pool, "run-3", "t1", "2026-04-22T09:10:00Z", 50).await;
        insert_run_with_started_at(&pool, "run-4", "t2", "2026-04-22T09:20:00Z", 999).await;

        let latest = find_latest_with_prompt_usage_by_thread_excluding_run(&pool, "t1", "run-x")
            .await
            .unwrap()
            .expect("latest matching run for t1");
        assert_eq!(latest.id, "run-3");
        assert_eq!(latest.thread_id, "t1");
        assert_eq!(latest.usage.input_tokens, 50);

        let excluded = find_latest_with_prompt_usage_by_thread_excluding_run(&pool, "t1", "run-3")
            .await
            .unwrap()
            .expect("previous matching run after exclusion");
        assert_eq!(excluded.id, "run-1");
        assert_eq!(excluded.usage.input_tokens, 10);
    }

    #[tokio::test]
    async fn find_latest_with_prompt_usage_includes_cache_read_only_runs() {
        let pool = setup_test_pool().await;

        insert_run_with_usage(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 10, 0).await;
        insert_run_with_usage(&pool, "run-2", "t1", "2026-04-22T09:05:00Z", 0, 64).await;
        insert_run_with_usage(&pool, "run-3", "t1", "2026-04-22T09:10:00Z", 0, 0).await;

        let latest = find_latest_with_prompt_usage_by_thread_excluding_run(&pool, "t1", "run-x")
            .await
            .unwrap()
            .expect("cache-read-only run should qualify as prompt usage");
        assert_eq!(latest.id, "run-2");
        assert_eq!(latest.usage.input_tokens, 0);
        assert_eq!(latest.usage.cache_read_tokens, 64);
    }

    #[tokio::test]
    async fn insert_run_persists_fields() {
        let pool = setup_test_pool().await;
        let r = RunInsert {
            id: "run-1".into(),
            thread_id: "t1".into(),
            profile_id: Some("prof-1".into()),
            run_mode: "default".into(),
            provider_id: Some("prov-1".into()),
            model_id: Some("gpt-4".into()),
            effective_model_plan_json: Some("{}".into()),
            status: "running".into(),
        };
        insert(&pool, &r).await.unwrap();

        let found = find_active_by_thread(&pool, "t1")
            .await
            .unwrap()
            .expect("should exist");
        assert_eq!(found.id, "run-1");
        assert_eq!(found.status, "running");
    }

    #[tokio::test]
    async fn update_status_sets_finished_at_for_terminal() {
        let pool = setup_test_pool().await;
        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 0).await;

        update_status(&pool, "run-1", RunStatus::Completed)
            .await
            .unwrap();

        let found = find_latest_by_thread(&pool, "t1")
            .await
            .unwrap()
            .expect("should exist");
        assert_eq!(found.status, "completed");
    }

    #[tokio::test]
    async fn update_status_for_non_terminal_does_not_set_finished_at() {
        let pool = setup_test_pool().await;
        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 0).await;

        update_status(&pool, "run-1", RunStatus::WaitingApproval)
            .await
            .unwrap();

        let found = find_latest_by_thread(&pool, "t1")
            .await
            .unwrap()
            .expect("should exist");
        assert_eq!(found.status, "waiting_approval");
    }

    // ── Elapsed-time transition tests ──────────────────────────────────

    #[tokio::test]
    async fn update_status_entering_running_sets_running_since_when_null() {
        let pool = setup_test_pool().await;
        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 0).await;
        set_run_elapsed_tracking(&pool, "run-1", 0, None).await;

        update_status(&pool, "run-1", RunStatus::Running)
            .await
            .unwrap();

        let row = sqlx::query(
            "SELECT elapsed_running_secs, running_since FROM thread_runs WHERE id = 'run-1'",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        assert_eq!(row.get::<i64, _>("elapsed_running_secs"), 0);
        assert!(
            row.get::<Option<String>, _>("running_since").is_some(),
            "running_since should be set when entering Running from idle"
        );
    }

    #[tokio::test]
    async fn update_status_entering_running_is_idempotent_for_running_since() {
        let pool = setup_test_pool().await;
        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 0).await;
        set_run_elapsed_tracking(&pool, "run-1", 10, Some("2026-01-01T00:00:00Z")).await;

        update_status(&pool, "run-1", RunStatus::Running)
            .await
            .unwrap();

        let row = sqlx::query(
            "SELECT elapsed_running_secs, running_since FROM thread_runs WHERE id = 'run-1'",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        assert_eq!(row.get::<i64, _>("elapsed_running_secs"), 10);
        assert_eq!(
            row.get::<String, _>("running_since"),
            "2026-01-01T00:00:00Z",
            "running_since should not be overwritten when entering Running again"
        );
    }

    #[tokio::test]
    async fn update_status_leaving_running_accumulates_elapsed_and_clears_running_since() {
        let pool = setup_test_pool().await;
        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 0).await;
        set_run_elapsed_tracking(&pool, "run-1", 15, Some("2026-01-01T00:00:00Z")).await;

        update_status(&pool, "run-1", RunStatus::WaitingApproval)
            .await
            .unwrap();

        let row = sqlx::query(
            "SELECT elapsed_running_secs, running_since FROM thread_runs WHERE id = 'run-1'",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        assert!(
            row.get::<i64, _>("elapsed_running_secs") > 15,
            "elapsed should accumulate the open running_since segment"
        );
        assert!(
            row.get::<Option<String>, _>("running_since").is_none(),
            "running_since should be cleared when leaving Running"
        );
    }

    #[tokio::test]
    async fn update_status_non_running_to_terminal_sets_finished_at_and_preserves_elapsed() {
        let pool = setup_test_pool().await;
        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 0).await;
        set_run_elapsed_tracking(&pool, "run-1", 20, None).await;

        update_status(&pool, "run-1", RunStatus::Completed)
            .await
            .unwrap();

        let row = sqlx::query(
            "SELECT elapsed_running_secs, running_since, finished_at FROM thread_runs WHERE id = 'run-1'",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        assert_eq!(
            row.get::<i64, _>("elapsed_running_secs"),
            20,
            "elapsed should not change when no running segment is open"
        );
        assert!(row.get::<Option<String>, _>("running_since").is_none());
        assert!(
            row.get::<Option<String>, _>("finished_at").is_some(),
            "finished_at should be set for terminal status"
        );
    }

    #[tokio::test]
    async fn update_status_running_to_terminal_accumulates_elapsed_and_sets_finished_at() {
        let pool = setup_test_pool().await;
        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 0).await;
        set_run_elapsed_tracking(&pool, "run-1", 10, Some("2026-01-01T00:00:00Z")).await;

        update_status(&pool, "run-1", RunStatus::Failed)
            .await
            .unwrap();

        let row = sqlx::query(
            "SELECT elapsed_running_secs, running_since, finished_at, status FROM thread_runs WHERE id = 'run-1'",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        assert!(
            row.get::<i64, _>("elapsed_running_secs") > 10,
            "elapsed should include the open running_since segment"
        );
        assert!(row.get::<Option<String>, _>("running_since").is_none());
        assert!(row.get::<Option<String>, _>("finished_at").is_some());
        assert_eq!(row.get::<String, _>("status"), "failed");
    }

    #[tokio::test]
    async fn update_usage_persists_token_counts() {
        let pool = setup_test_pool().await;
        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 0).await;

        let usage = tiycore::types::Usage {
            input: 500,
            output: 300,
            cache_read: 100,
            cache_write: 50,
            total_tokens: 950,
            cost: tiycore::types::UsageCost::default(),
        };
        update_usage(&pool, "run-1", &usage).await.unwrap();

        let found = find_latest_by_thread(&pool, "t1")
            .await
            .unwrap()
            .expect("should exist");
        assert_eq!(found.usage.input_tokens, 500);
        assert_eq!(found.usage.total_tokens, 950);
    }

    #[tokio::test]
    async fn set_error_message_persists() {
        let pool = setup_test_pool().await;
        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 0).await;

        set_error_message(&pool, "run-1", "oops").await.unwrap();

        let found = find_latest_by_thread(&pool, "t1")
            .await
            .unwrap()
            .expect("should exist");
        assert_eq!(found.error_message, Some("oops".into()));
    }

    #[tokio::test]
    async fn find_effective_model_plan_json_returns_json() {
        let pool = setup_test_pool().await;
        let r = RunInsert {
            id: "run-1".into(),
            thread_id: "t1".into(),
            profile_id: None,
            run_mode: "default".into(),
            provider_id: None,
            model_id: None,
            effective_model_plan_json: Some(r#"{"primary":{"modelDisplayName":"GPT-4"}}"#.into()),
            status: "completed".into(),
        };
        insert(&pool, &r).await.unwrap();

        let json = find_effective_model_plan_json(&pool, "run-1")
            .await
            .unwrap()
            .expect("should exist");
        assert!(json.contains("GPT-4"));
    }

    #[tokio::test]
    async fn find_effective_model_plan_json_returns_none_for_missing() {
        let pool = setup_test_pool().await;
        let json = find_effective_model_plan_json(&pool, "nope").await.unwrap();
        assert!(json.is_none());
    }

    #[tokio::test]
    async fn find_start_context_by_id_returns_run_mode_and_model_plan() {
        let pool = setup_test_pool().await;
        let r = RunInsert {
            id: "run-ctx".into(),
            thread_id: "t1".into(),
            profile_id: Some("profile-1".into()),
            run_mode: "plan".into(),
            provider_id: Some("provider-1".into()),
            model_id: Some("model-1".into()),
            effective_model_plan_json: Some(r#"{"profileId":"profile-1"}"#.into()),
            status: "completed".into(),
        };
        insert(&pool, &r).await.unwrap();

        let ctx = find_start_context_by_id(&pool, "run-ctx")
            .await
            .unwrap()
            .expect("start context should exist");
        assert_eq!(ctx.run_mode, "plan");
        assert_eq!(
            ctx.effective_model_plan_json.as_deref(),
            Some(r#"{"profileId":"profile-1"}"#)
        );
    }

    #[tokio::test]
    async fn get_run_duration_returns_specific_finished_run_duration() {
        let pool = setup_test_pool().await;
        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 0).await;
        set_run_finished_at(&pool, "run-1", "2026-04-22T09:00:42Z").await;

        let duration = get_run_duration(&pool, "run-1").await.unwrap();
        assert_eq!(duration, Some(42));
        assert!(get_run_duration(&pool, "missing").await.unwrap().is_none());
    }

    #[tokio::test]
    async fn find_latest_evaluable_run_id_by_thread_returns_latest_completed_or_interrupted() {
        let pool = setup_test_pool().await;
        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 0).await;
        set_run_finished_at(&pool, "run-1", "2026-04-22T09:00:10Z").await;

        let r = RunInsert {
            id: "run-2".into(),
            thread_id: "t1".into(),
            profile_id: None,
            run_mode: "default".into(),
            provider_id: None,
            model_id: None,
            effective_model_plan_json: None,
            status: RunStatus::Interrupted.as_str().into(),
        };
        insert(&pool, &r).await.unwrap();
        set_run_finished_at(&pool, "run-2", "2026-04-22T09:05:10Z").await;

        let latest = find_latest_evaluable_run_id_by_thread(&pool, "t1")
            .await
            .unwrap();
        assert_eq!(latest.as_deref(), Some("run-2"));
    }

    #[tokio::test]
    async fn find_active_by_thread_skips_terminal_and_waiting_approval() {
        let pool = setup_test_pool().await;
        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 0).await;
        // run-1 is 'completed' status → not active
        assert!(find_active_by_thread(&pool, "t1").await.unwrap().is_none());

        let r = RunInsert {
            id: "run-2".into(),
            thread_id: "t1".into(),
            profile_id: None,
            run_mode: "default".into(),
            provider_id: None,
            model_id: None,
            effective_model_plan_json: None,
            status: "running".into(),
        };
        insert(&pool, &r).await.unwrap();
        assert!(find_active_by_thread(&pool, "t1").await.unwrap().is_some());
    }

    #[tokio::test]
    async fn needs_reply_runs_are_non_progressing_for_active_lookup_and_recovery() {
        let pool = setup_test_pool().await;
        let r = RunInsert {
            id: "run-needs-reply".into(),
            thread_id: "t1".into(),
            profile_id: None,
            run_mode: "default".into(),
            provider_id: None,
            model_id: None,
            effective_model_plan_json: None,
            status: RunStatus::NeedsReply.as_str().into(),
        };
        insert(&pool, &r).await.unwrap();

        assert!(find_active_by_thread(&pool, "t1").await.unwrap().is_none());
        assert!(list_thread_ids_with_active_runs(&pool)
            .await
            .unwrap()
            .is_empty());

        let affected = interrupt_active_runs(&pool).await.unwrap();
        assert_eq!(affected, 0);

        let found = find_latest_by_thread(&pool, "t1")
            .await
            .unwrap()
            .expect("needs_reply run should still exist");
        assert_eq!(found.status, RunStatus::NeedsReply.as_str());
        assert!(found.error_message.is_none());
    }

    #[tokio::test]
    async fn list_thread_ids_with_active_runs_finds_running() {
        let pool = setup_test_pool().await;
        let r = RunInsert {
            id: "run-1".into(),
            thread_id: "t1".into(),
            profile_id: None,
            run_mode: "default".into(),
            provider_id: None,
            model_id: None,
            effective_model_plan_json: None,
            status: "running".into(),
        };
        insert(&pool, &r).await.unwrap();

        let ids = list_thread_ids_with_active_runs(&pool).await.unwrap();
        assert!(ids.contains(&"t1".to_string()));
    }

    #[tokio::test]
    async fn interrupt_active_runs_marks_non_terminal() {
        let pool = setup_test_pool().await;
        let r = RunInsert {
            id: "run-1".into(),
            thread_id: "t1".into(),
            profile_id: None,
            run_mode: "default".into(),
            provider_id: None,
            model_id: None,
            effective_model_plan_json: None,
            status: "running".into(),
        };
        insert(&pool, &r).await.unwrap();

        let affected = interrupt_active_runs(&pool).await.unwrap();
        assert_eq!(affected, 1);

        let found = find_latest_by_thread(&pool, "t1")
            .await
            .unwrap()
            .expect("should exist");
        assert_eq!(found.status, "interrupted");
    }

    #[tokio::test]
    async fn cancel_waiting_approval_by_thread_terminates_zombie_run() {
        let pool = setup_test_pool().await;
        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 0).await;

        // Move run-1 to waiting_approval (non-terminal, no finished_at)
        update_status(&pool, "run-1", RunStatus::WaitingApproval)
            .await
            .unwrap();
        let before = find_latest_by_thread(&pool, "t1")
            .await
            .unwrap()
            .expect("should exist");
        assert_eq!(before.status, "waiting_approval");

        let affected = cancel_waiting_approval_by_thread(&pool, "t1")
            .await
            .unwrap();
        assert_eq!(affected, 1);

        let after = find_latest_by_thread(&pool, "t1")
            .await
            .unwrap()
            .expect("should exist");
        assert_eq!(after.status, "cancelled");
    }

    #[tokio::test]
    async fn cancel_waiting_approval_by_thread_ignores_other_statuses() {
        let pool = setup_test_pool().await;
        let r = RunInsert {
            id: "run-1".into(),
            thread_id: "t1".into(),
            profile_id: None,
            run_mode: "default".into(),
            provider_id: None,
            model_id: None,
            effective_model_plan_json: None,
            status: "running".into(),
        };
        insert(&pool, &r).await.unwrap();

        let affected = cancel_waiting_approval_by_thread(&pool, "t1")
            .await
            .unwrap();
        assert_eq!(affected, 0);

        let found = find_latest_by_thread(&pool, "t1")
            .await
            .unwrap()
            .expect("should exist");
        assert_eq!(found.status, "running");
    }

    // ── #5 / #11: Duration query tests ──

    #[tokio::test]
    async fn get_run_duration_returns_seconds() {
        let pool = setup_test_pool().await;
        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 0).await;
        set_run_finished_at(&pool, "run-1", "2026-04-22T09:00:42Z").await;

        let duration = super::get_run_duration(&pool, "run-1")
            .await
            .unwrap()
            .expect("should return duration");
        assert_eq!(duration, 42);
    }

    #[tokio::test]
    async fn get_last_completed_run_duration_returns_most_recent() {
        let pool = setup_test_pool().await;

        // Insert two completed runs for same thread.
        // IMPORTANT: set_run_finished_at must be called AFTER update_status(Completed)
        // because update_status for terminal statuses overwrites finished_at to now().
        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 0).await;
        super::update_status(&pool, "run-1", RunStatus::Completed)
            .await
            .unwrap();
        set_run_finished_at(&pool, "run-1", "2026-04-22T09:00:42Z").await;

        insert_run_with_started_at(&pool, "run-2", "t1", "2026-04-22T10:00:00Z", 0).await;
        super::update_status(&pool, "run-2", RunStatus::Completed)
            .await
            .unwrap();
        set_run_finished_at(&pool, "run-2", "2026-04-22T10:00:30Z").await;

        // Should return the most recent completed run's duration (run-2 = 30s)
        let duration = super::get_last_completed_run_duration(&pool, "t1")
            .await
            .unwrap()
            .expect("should return duration");
        assert_eq!(duration, 30);
    }

    #[tokio::test]
    async fn get_last_completed_run_duration_returns_none_when_no_completed() {
        let pool = setup_test_pool().await;
        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 0).await;
        // Run is still in default status (not completed)

        let duration = super::get_last_completed_run_duration(&pool, "t1")
            .await
            .unwrap();
        assert!(duration.is_none());
    }

    #[tokio::test]
    async fn list_thread_elapsed_running_seconds_includes_terminal_runs() {
        let pool = setup_test_pool().await;
        insert_run_with_started_at(&pool, "run-completed", "t1", "2026-04-22T09:00:00Z", 0).await;
        set_run_elapsed_tracking(&pool, "run-completed", 37, None).await;

        let values =
            super::list_thread_elapsed_running_seconds_by_threads(&pool, &["t1".to_string()])
                .await
                .unwrap();

        assert_eq!(values.get("t1"), Some(&37));
    }

    #[tokio::test]
    async fn list_thread_elapsed_running_seconds_sums_multiple_runs() {
        let pool = setup_test_pool().await;
        insert_run_with_started_at(&pool, "run-1", "t1", "2026-04-22T09:00:00Z", 0).await;
        insert_run_with_started_at(&pool, "run-2", "t1", "2026-04-22T10:00:00Z", 0).await;
        insert_run_with_started_at(&pool, "run-other", "t2", "2026-04-22T11:00:00Z", 0).await;
        set_run_elapsed_tracking(&pool, "run-1", 12, None).await;
        set_run_elapsed_tracking(&pool, "run-2", 30, None).await;
        set_run_elapsed_tracking(&pool, "run-other", 999, None).await;

        let values = super::list_thread_elapsed_running_seconds_by_threads(
            &pool,
            &["t1".to_string(), "t2".to_string()],
        )
        .await
        .unwrap();

        assert_eq!(values.get("t1"), Some(&42));
        assert_eq!(values.get("t2"), Some(&999));
    }

    #[tokio::test]
    async fn list_thread_elapsed_running_seconds_adds_open_running_segment() {
        let pool = setup_test_pool().await;
        insert_run_with_started_at(&pool, "run-running", "t1", "2026-04-22T09:00:00Z", 0).await;
        super::update_status(&pool, "run-running", RunStatus::Running)
            .await
            .unwrap();
        set_run_elapsed_tracking(&pool, "run-running", 20, Some("2026-04-22T09:00:00Z")).await;

        let values =
            super::list_thread_elapsed_running_seconds_by_threads(&pool, &["t1".to_string()])
                .await
                .unwrap();
        let elapsed = values.get("t1").copied().expect("thread elapsed");

        assert!(
            elapsed > 20,
            "expected running segment to be added, got {elapsed}"
        );
    }
}

/// Get the duration in seconds of the last completed run for a thread.
/// Returns None if no completed run exists with a finished_at timestamp.
pub async fn get_last_completed_run_duration(
    pool: &SqlitePool,
    thread_id: &str,
) -> Result<Option<i64>, AppError> {
    let duration = sqlx::query_scalar::<_, Option<i64>>(
        "SELECT CAST(strftime('%s', finished_at) - strftime('%s', started_at) AS INTEGER)
         FROM thread_runs
         WHERE thread_id = ?
           AND status = 'completed'
           AND finished_at IS NOT NULL
         ORDER BY started_at DESC
         LIMIT 1",
    )
    .bind(thread_id)
    .fetch_optional(pool)
    .await?
    .flatten();
    Ok(duration)
}

/// Get the duration in seconds of a specific run.
/// Returns None if the run has no finished_at timestamp.
pub async fn get_run_duration(pool: &SqlitePool, run_id: &str) -> Result<Option<i64>, AppError> {
    let duration = sqlx::query_scalar::<_, Option<i64>>(
        "SELECT CAST(strftime('%s', finished_at) - strftime('%s', started_at) AS INTEGER)
         FROM thread_runs
         WHERE id = ?
           AND finished_at IS NOT NULL
         LIMIT 1",
    )
    .bind(run_id)
    .fetch_optional(pool)
    .await?
    .flatten();
    Ok(duration)
}

/// Bulk-fetch the Unix-millisecond start timestamp of the currently active
/// (non-terminal) run for each thread in `thread_ids`. Threads without an
/// active run are simply absent from the returned map. Used by the sidebar
/// query so the frontend can derive per-thread elapsed time from a source
/// of truth that survives app restarts (`thread_runs.started_at`).
pub async fn list_active_run_started_at_ms_by_threads(
    pool: &SqlitePool,
    thread_ids: &[String],
) -> Result<std::collections::HashMap<String, i64>, AppError> {
    if thread_ids.is_empty() {
        return Ok(std::collections::HashMap::new());
    }
    // `IN (?, ?, …)` with a single bind argument per id. Keeps the query
    // shape easy to read; thread_ids is bounded by the sidebar page size.
    let placeholders = std::iter::repeat("?")
        .take(thread_ids.len())
        .collect::<Vec<_>>()
        .join(",");
    let sql = format!(
        "SELECT thread_id, CAST(strftime('%s', started_at) AS INTEGER) * 1000 AS started_at_ms
         FROM (
             SELECT thread_id, MAX(started_at) AS started_at FROM thread_runs
             WHERE thread_id IN ({placeholders})
               AND status NOT IN ('completed','failed','denied','interrupted','cancelled','limit_reached')
             GROUP BY thread_id
         )",
    );
    let mut query = sqlx::query_as::<_, (String, i64)>(&sql);
    for id in thread_ids {
        query = query.bind(id);
    }
    let rows = query.fetch_all(pool).await?;
    Ok(rows.into_iter().collect())
}

/// Batch-query thread-level cumulative active running seconds (excluding
/// pauses) for each requested thread. The total includes all historical runs'
/// `elapsed_running_secs`; if any run is currently running, its open
/// `running_since` segment is included as `now() - running_since`.
pub async fn list_thread_elapsed_running_seconds_by_threads(
    pool: &SqlitePool,
    thread_ids: &[String],
) -> Result<std::collections::HashMap<String, i64>, AppError> {
    if thread_ids.is_empty() {
        return Ok(std::collections::HashMap::new());
    }
    let placeholders = std::iter::repeat("?")
        .take(thread_ids.len())
        .collect::<Vec<_>>()
        .join(",");
    let sql = format!(
        "SELECT thread_id,
                SUM(
                    elapsed_running_secs +
                    CASE WHEN running_since IS NOT NULL
                         THEN CAST(strftime('%s', 'now') - strftime('%s', running_since) AS INTEGER)
                         ELSE 0
                    END
                ) AS elapsed_secs
         FROM thread_runs
         WHERE thread_id IN ({placeholders})
         GROUP BY thread_id",
    );
    let mut query = sqlx::query_as::<_, (String, i64)>(&sql);
    for id in thread_ids {
        query = query.bind(id);
    }
    let rows = query.fetch_all(pool).await?;
    Ok(rows.into_iter().collect())
}

/// Batch-query the cumulative **active running seconds** (excluding pauses)
/// used by the thread header timer. This is intentionally thread-level rather
/// than active-run-only so completed/interrupted history survives restarts and
/// later runs continue from the prior thread total.
#[deprecated(
    note = "use list_thread_elapsed_running_seconds_by_threads for thread header timer seeds"
)]
pub async fn list_active_run_elapsed_seconds_by_threads(
    pool: &SqlitePool,
    thread_ids: &[String],
) -> Result<std::collections::HashMap<String, i64>, AppError> {
    list_thread_elapsed_running_seconds_by_threads(pool, thread_ids).await
}
