cctop 0.24.0

//! OpenCode session discovery and extraction from its current SQLite store.

use super::extract;
use super::{
    ActivityState, ContextUsage, Costs, FallbackRates, ModelBreakdown, Session, SessionData,
    Subagent, SubagentStatus, Tokens,
};
use crate::config;
use crate::pricing::Provider;
use crate::util;
use rusqlite::{Connection, OpenFlags, OptionalExtension, params};
use serde_json::Value;
use std::cell::RefCell;
use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};

/// The line delta behind OpenCode's built-in editing tools, and the patch to
/// show for it.
///
/// `edit` is handed the text it replaces and the text that takes its place, so
/// its diff is the literal before and after. `write` can replace a file
/// wholesale, so its removed lines cannot be known from the recorded input and
/// are deliberately not guessed. `apply_patch` carries a real patch.
///
/// Read by the table's extraction and by the conversation reader alike: what
/// counts as an edit is one fact about this harness, not one per caller.
pub fn tool_delta(name: &str, state: &Value) -> Option<super::Delta> {
    if state.get("status").and_then(Value::as_str) != Some("completed") {
        return None;
    }
    let input = state.get("input")?;
    match name {
        "edit" => extract::edit_delta(input),
        "apply_patch" => input
            .get("patch")
            .and_then(Value::as_str)
            .map(|patch| extract::parse_apply_patch(patch).1),
        _ => None,
    }
}

fn tool_duration_ms(state: &Value) -> Option<i64> {
    let time = state.get("time")?;
    let start = time.get("start")?.as_i64()?;
    let end = time.get("end")?.as_i64()?;
    Some((end - start).max(0))
}

fn readonly(path: &Path) -> rusqlite::Result<Connection> {
    Connection::open_with_flags(
        path,
        OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
    )
}

thread_local! {
    /// Read-only connections, kept open and reused.
    ///
    /// Every read of an OpenCode database goes through here — discovery,
    /// extraction, the activity dot, the context window, the last tool, and the
    /// transcript scan behind search. Opening per read came to several thousand
    /// opens of a 555 MB database per walk, and the cost was not the reading:
    /// 240,000 voluntary context switches and nine seconds of kernel time, spent
    /// almost entirely on setting up and tearing down file locks. Extraction
    /// alone — the one call a walk makes per session, and the heaviest of them by
    /// a factor of three — had been paying that price per session all along.
    ///
    /// Thread-local rather than shared, which is what makes it safe *and*
    /// fast. `SQLITE_OPEN_NO_MUTEX` means a connection may not be used from two
    /// threads at once, so one shared connection would have to be behind a
    /// mutex — and extraction fans out across every core, so that mutex would
    /// serialise the one path that most needs not to be. A connection per
    /// thread has neither problem: the serial tail pass opens one, and a cold
    /// extraction opens one per worker instead of one per session.
    ///
    /// ponytail: held for the life of the thread, never revalidated. SQLite
    /// writes in place and a read-only connection picks up another process's
    /// commits, so this stays correct while OpenCode is running; a database
    /// swapped out wholesale underneath it would be read from the old file
    /// until cctop restarts.
    static CONNECTIONS: RefCell<HashMap<PathBuf, Connection>> = RefCell::new(HashMap::new());
}

/// Release this thread's cached connections.
///
/// A cached connection keeps a handle on the database, so a test that writes a
/// scratch database, reads it, and then removes it has to say when it is done.
/// Nothing in the program needs this: the only deletion cctop performs is
/// [`delete`], which removes a *row* through its own connection.
#[cfg(test)]
fn close_databases() {
    CONNECTIONS.with(|cache| cache.borrow_mut().clear());
}

/// Run `query` against `path`'s database, opening it only the first time this
/// thread asks. `None` if it cannot be opened.
fn with_db<T>(path: &Path, query: impl FnOnce(&Connection) -> T) -> Option<T> {
    CONNECTIONS.with(|cache| {
        let mut cache = cache.borrow_mut();
        if !cache.contains_key(path) {
            cache.insert(path.to_path_buf(), readonly(path).ok()?);
        }
        cache.get(path).map(query)
    })
}

/// State of the newest OpenCode message. OpenCode keeps events in SQLite rather
/// than a transcript file, but its assistant messages record both completion
/// and provider errors.
pub fn extract_activity_state(path: &Path, session_id: &str) -> ActivityState {
    with_db(path, |db| activity_state(db, session_id)).unwrap_or(ActivityState::Working)
}

fn activity_state(db: &Connection, session_id: &str) -> ActivityState {
    let message = db
        .query_row(
            "SELECT id, data FROM message WHERE session_id = ?1 ORDER BY time_created DESC, id DESC LIMIT 1",
            params![session_id],
            |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
        )
        .optional()
        .ok()
        .flatten();
    let Some((message_id, raw)) = message else {
        return ActivityState::Working;
    };
    let Ok(message) = serde_json::from_str::<Value>(&raw) else {
        return ActivityState::Working;
    };
    if message.get("role").and_then(Value::as_str) != Some("assistant") {
        return ActivityState::Working;
    }
    if is_api_failure(&message) {
        ActivityState::ApiError
    } else if has_input_request(db, &message_id) {
        ActivityState::WaitingForInput
    } else {
        ActivityState::Working
    }
}

fn has_input_request(db: &Connection, message_id: &str) -> bool {
    let Ok(mut stmt) = db.prepare("SELECT data FROM part WHERE message_id = ?1") else {
        return false;
    };
    let Ok(rows) = stmt.query_map(params![message_id], |row| row.get::<_, String>(0)) else {
        return false;
    };
    rows.flatten().any(|raw| {
        let Ok(part) = serde_json::from_str::<Value>(&raw) else {
            return false;
        };
        matches!(
            part.get("type").and_then(Value::as_str),
            Some("tool" | "toolCall")
        ) && part
            .get("tool")
            .or_else(|| part.get("name"))
            .and_then(Value::as_str)
            .is_some_and(super::is_input_request_tool)
    })
}

fn is_api_failure(message: &Value) -> bool {
    let Some(error) = message.get("error") else {
        return false;
    };
    let text = [
        error.get("name").and_then(Value::as_str),
        error
            .get("data")
            .and_then(|data| data.get("message"))
            .and_then(Value::as_str),
        error.get("message").and_then(Value::as_str),
    ]
    .into_iter()
    .flatten()
    .collect::<Vec<_>>()
    .join(" ")
    .to_ascii_lowercase();
    [
        "timeout",
        "timed out",
        "rate limit",
        "overloaded",
        "connection",
        "network",
        "api error",
        "apierror",
        "internal server",
        "queue is full",
    ]
    .iter()
    .any(|needle| text.contains(needle))
}

/// Stable releases use `opencode.db`; nonstandard channels suffix the filename.
fn database_paths() -> Vec<PathBuf> {
    config::opencode_data_roots()
        .into_iter()
        .flat_map(|root| {
            config::list_dir(&root)
                .into_iter()
                .filter(|name| name.starts_with("opencode") && name.ends_with(".db"))
                .map(move |name| root.join(name))
        })
        .collect()
}

/// Every message one session recorded, oldest first, with its own parts.
///
/// `message` holds the envelope and `part` the words and the calls, so the two
/// are read as one ordered pass: the join is the only place that knows which
/// part belongs to which message, and a reader needs them together.
///
/// Ponytail: the whole session is read, not the newest of it. A subagent's
/// messages are not in it — a subagent is its own session, so a parent's
/// conversation cannot pick one up by accident — but a parent that delegated
/// reads as though it did not.
pub fn for_each_message(path: &Path, session_id: &str, mut f: impl FnMut(&Value, i64, &[Value])) {
    with_db(path, |db| {
        let Ok(mut stmt) = db.prepare(
            "SELECT m.id, m.time_created, m.data, p.data FROM message m \
             LEFT JOIN part p ON p.message_id = m.id \
             WHERE m.session_id = ?1 ORDER BY m.time_created, m.id, p.id",
        ) else {
            return;
        };
        let Ok(mut rows) = stmt.query(params![session_id]) else {
            return;
        };
        let mut current = String::new();
        let mut created = 0i64;
        let mut message = Value::Null;
        let mut parts: Vec<Value> = Vec::new();
        while let Ok(Some(row)) = rows.next() {
            let Ok(id) = row.get::<_, String>(0) else {
                continue;
            };
            if id != current {
                if !current.is_empty() && !message.is_null() {
                    f(&message, created, &parts);
                }
                current = id.clone();
                created = row
                    .get::<_, Option<i64>>(1)
                    .ok()
                    .flatten()
                    .unwrap_or_default();
                message = row
                    .get::<_, Option<String>>(2)
                    .ok()
                    .flatten()
                    .and_then(|raw| serde_json::from_str(&raw).ok())
                    .unwrap_or(Value::Null);
                parts.clear();
            }
            // A message with no parts at all is a `LEFT JOIN` that found nothing,
            // and one whose part will not parse is the same as not being there.
            if let Ok(Some(raw)) = row.get::<_, Option<String>>(3)
                && let Ok(part) = serde_json::from_str::<Value>(&raw)
            {
                parts.push(part);
            }
        }
        if !current.is_empty() && !message.is_null() {
            f(&message, created, &parts);
        }
    });
}

/// Feed every stored message and part of one session to `feed`, stopping early
/// once it says the session is found.
///
/// This exists so that searching shares the connection pool rather than opening
/// the database per candidate: search runs as the user types, and the pool is
/// what keeps that from costing an open of a database that may be half a
/// gigabyte, once per session, per keystroke.
///
/// `part` carries the text of a message; `message` carries the envelope, and the
/// tool calls that are worth finding a session by. Both are per-session in this
/// schema, so neither can leak another session's text into this hit. The `part`
/// table is absent in older databases, which is a failed prepare rather than an
/// empty result — hence trying each in turn.
pub fn scan_session(path: &Path, session_id: &str, mut feed: impl FnMut(&str) -> bool) -> bool {
    with_db(path, |db| {
        for sql in [
            "SELECT data FROM part WHERE session_id = ?1 ORDER BY id",
            "SELECT data FROM message WHERE session_id = ?1 ORDER BY time_created, id",
        ] {
            let Ok(mut stmt) = db.prepare(sql) else {
                continue;
            };
            let Ok(rows) = stmt.query_map(params![session_id], |row| row.get::<_, String>(0))
            else {
                continue;
            };
            for raw in rows.flatten() {
                if feed(&raw) {
                    return true;
                }
            }
        }
        false
    })
    .unwrap_or(false)
}

/// When a message was written, in RFC-3339.
///
/// The column is the fallback because `data.time.created` is the envelope's own
/// idea of when it was spoken, which is not always when the row landed.
pub fn message_time(message: &Value, column: i64) -> String {
    let ms = message
        .pointer("/time/created")
        .and_then(Value::as_i64)
        .unwrap_or(column);
    util::ms_to_rfc3339(ms)
}

/// Discover sessions from every OpenCode channel database, preferring the most
/// recently updated copy if the same session was migrated between channels.
pub fn list_sessions() -> Vec<Session> {
    let mut sessions = Vec::new();
    for path in database_paths() {
        // Pooled per database rather than opened per walk: this is the walk's
        // first and broadest read, and it opens one connection per channel
        // either way — the table below is then reused by every extract that
        // follows on the same thread.
        let Some(found) = with_db(&path, |db| sessions_in(db, &path)) else {
            continue;
        };
        sessions.extend(found);
    }

    sessions.sort_by(|a, b| b.last_active.cmp(&a.last_active));
    let mut seen = HashSet::new();
    sessions.retain(|s| seen.insert(s.session_id.clone()));
    sessions
}

fn sessions_in(db: &Connection, path: &Path) -> Vec<Session> {
    let mut sessions = Vec::new();
    // A subagent is a session of its own, with `parent_id` naming the one that
    // spawned it; listed, each stands alone as a conversation. They are hidden
    // here and folded into their parent by `extract`, the way Claude's subagents
    // are. A database from before the column existed has no subagents to hide,
    // so it falls back to the unfiltered query.
    const COLUMNS: &str =
        "SELECT id, directory, title, model, time_created, time_updated FROM session";
    let Ok(mut stmt) = db
        .prepare(&format!("{COLUMNS} WHERE parent_id IS NULL"))
        .or_else(|_| db.prepare(COLUMNS))
    else {
        return sessions;
    };
    let Ok(rows) = stmt.query_map([], |row| {
        Ok((
            row.get::<_, String>(0)?,
            row.get::<_, String>(1)?,
            row.get::<_, String>(2)?,
            row.get::<_, Option<String>>(3)?,
            row.get::<_, i64>(4)?,
            row.get::<_, i64>(5)?,
        ))
    }) else {
        return sessions;
    };
    for row in rows.flatten() {
        let (id, directory, title, model_json, created, updated) = row;
        let model = model_json
            .as_deref()
            .and_then(|raw| serde_json::from_str::<Value>(raw).ok())
            .and_then(|v| v.get("id").and_then(Value::as_str).map(str::to_string))
            .unwrap_or_default();
        let mut session = Session::new(Provider::OpenCode, id);
        session.started_at = util::ms_to_rfc3339(created);
        session.last_active = util::ms_to_rfc3339(updated);
        session.label_source = directory;
        session.title = (!title.is_empty()).then_some(title);
        session.model = model;
        session.data_file = Some(path.to_path_buf());
        sessions.push(session);
    }
    sessions
}

fn u64_at(value: &Value, path: &[&str]) -> u64 {
    let mut current = value;
    for key in path {
        current = current.get(*key).unwrap_or(&Value::Null);
    }
    current.as_u64().unwrap_or(0)
}

fn assistant_usage(value: &Value, rates: &mut FallbackRates) -> (String, String, Tokens, Costs) {
    let model = value
        .get("modelID")
        .and_then(Value::as_str)
        .unwrap_or("unknown")
        .to_string();
    let ts = value
        .get("time")
        .and_then(|v| v.get("created"))
        .and_then(Value::as_i64)
        .map(util::ms_to_rfc3339)
        .unwrap_or_default();
    let input = u64_at(value, &["tokens", "input"]);
    let output = u64_at(value, &["tokens", "output"]);
    let cache_read = u64_at(value, &["tokens", "cache", "read"]);
    let cache_write = u64_at(value, &["tokens", "cache", "write"]);
    let reasoning = u64_at(value, &["tokens", "reasoning"]);
    let tokens = Tokens {
        input,
        output,
        cache_read,
        cache_write_5m: cache_write,
        reasoning_output: reasoning,
        total: input + output + cache_read + cache_write,
        ..Default::default()
    };
    // OpenCode writes `cost: 0` for providers it has no rates for, which is
    // indistinguishable in the record from a genuinely free model. A cost it did
    // compute is authoritative; a zero falls back to pricing the tokens.
    let reported = value.get("cost").and_then(Value::as_f64).unwrap_or(0.0);
    let total = if reported > 0.0 || tokens.total == 0 {
        reported
    } else {
        rates.costs(&model, &tokens).map_or(0.0, |c| c.total)
    };
    let costs = Costs {
        total,
        ..Default::default()
    };
    (model, ts, tokens, costs)
}

fn add_usage(target: &mut SessionData, tokens: &Tokens, costs: &Costs) {
    target.tokens.input += tokens.input;
    target.tokens.output += tokens.output;
    target.tokens.cache_read += tokens.cache_read;
    target.tokens.cache_write_5m += tokens.cache_write_5m;
    target.tokens.reasoning_output += tokens.reasoning_output;
    target.tokens.total += tokens.total;
    target.costs.total += costs.total;
}

/// Extract accounting and tools for one OpenCode session.
///
/// The database records a cost on every assistant message, and that figure wins
/// wherever OpenCode knows the provider's rates. It reports zero for providers
/// it has no pricing for, so a zero falls back to pricing the tokens against
/// LiteLLM rather than being taken at face value.
pub fn extract(path: &Path, session_id: &str) -> SessionData {
    // Pooled like every other read of this database, and for the same reason it
    // matters most here: this is the call a walk makes once per session, so the
    // one-per-session open this used to pay was the shape of a whole walk's cost.
    with_db(path, |db| extract_of(db, session_id)).unwrap_or_else(|| SessionData {
        error: Some(format!(
            "Could not open OpenCode database {}",
            path.display()
        )),
        ..Default::default()
    })
}

fn extract_of(db: &Connection, session_id: &str) -> SessionData {
    let mut data = SessionData::default();
    let mut breakdown: HashMap<String, (Tokens, Costs)> = HashMap::new();
    let mut saw_usage = false;
    let mut rates = FallbackRates::default();

    let aggregate = db
        .query_row(
            "SELECT title, model, cost, tokens_input, tokens_output, tokens_reasoning, \
             tokens_cache_read, tokens_cache_write FROM session WHERE id = ?1",
            params![session_id],
            |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, Option<String>>(1)?,
                    row.get::<_, f64>(2)?,
                    row.get::<_, u64>(3)?,
                    row.get::<_, u64>(4)?,
                    row.get::<_, u64>(5)?,
                    row.get::<_, u64>(6)?,
                    row.get::<_, u64>(7)?,
                ))
            },
        )
        .optional()
        .ok()
        .flatten();
    if let Some((title, model_json, ..)) = &aggregate {
        data.title = (!title.is_empty()).then(|| title.clone());
        data.last_model = model_json
            .as_deref()
            .and_then(|raw| serde_json::from_str::<Value>(raw).ok())
            .and_then(|v| v.get("id").and_then(Value::as_str).map(str::to_string))
            .unwrap_or_default();
    }

    if let Ok(mut stmt) =
        db.prepare("SELECT data FROM message WHERE session_id = ?1 ORDER BY time_created, id")
        && let Ok(rows) = stmt.query_map(params![session_id], |row| row.get::<_, String>(0))
    {
        for raw in rows.flatten() {
            let Ok(message) = serde_json::from_str::<Value>(&raw) else {
                continue;
            };
            if message.get("role").and_then(Value::as_str) != Some("assistant") {
                continue;
            }
            let (model, ts, tokens, costs) = assistant_usage(&message, &mut rates);
            saw_usage = true;
            data.last_model = model.clone();
            add_usage(&mut data, &tokens, &costs);
            let (mt, mc) = breakdown.entry(model.clone()).or_default();
            mt.input += tokens.input;
            mt.output += tokens.output;
            mt.cache_read += tokens.cache_read;
            mt.cache_write_5m += tokens.cache_write_5m;
            mt.reasoning_output += tokens.reasoning_output;
            mt.total += tokens.total;
            mc.total += costs.total;
            if let Some(dt) = util::parse_ts(&ts) {
                data.record_cost(&dt, &model, costs.total);
                // Reasoning is already inside `output` here — billing it again
                // would double-count, the same rule `FallbackRates` applies.
                let billed = tokens.all_input() + tokens.output;
                *data
                    .tokens_by_day
                    .entry(util::local_date_key(&dt))
                    .or_default()
                    .entry(model.clone())
                    .or_insert(0) += billed;
                *data
                    .tokens_by_hour
                    .entry(util::local_hour_key(&dt))
                    .or_default()
                    .entry(model)
                    .or_insert(0) += billed;
            }
        }
    }

    if !saw_usage
        && let Some((_, _, cost, input, output, reasoning, cache_read, cache_write)) = aggregate
    {
        data.tokens = Tokens {
            input,
            output,
            cache_read,
            cache_write_5m: cache_write,
            reasoning_output: reasoning,
            total: input + output + cache_read + cache_write,
            ..Default::default()
        };
        data.costs.total = if cost > 0.0 || data.tokens.total == 0 {
            cost
        } else {
            rates
                .costs(&data.last_model, &data.tokens)
                .map_or(0.0, |c| c.total)
        };
    }

    if let Ok(mut stmt) =
        db.prepare("SELECT data FROM part WHERE session_id = ?1 ORDER BY time_created, id")
        && let Ok(rows) = stmt.query_map(params![session_id], |row| row.get::<_, String>(0))
    {
        for raw in rows.flatten() {
            let Ok(part) = serde_json::from_str::<Value>(&raw) else {
                continue;
            };
            if part.get("type").and_then(Value::as_str) != Some("tool") {
                continue;
            }
            let name = part
                .get("tool")
                .and_then(Value::as_str)
                .unwrap_or("unknown");
            let state = part.get("state").unwrap_or(&Value::Null);
            let args = state.get("input").unwrap_or(&Value::Null);
            let (short, full) = extract::tool_detail(name, args);
            let delta = tool_delta(name, state);
            let duration_ms = tool_duration_ms(state);
            let ts = state
                .get("time")
                .and_then(|v| v.get("start"))
                .and_then(Value::as_i64)
                .map(util::ms_to_rfc3339)
                .unwrap_or_default();
            extract::push_tool_detail(
                &mut data.metrics.tool_details,
                name,
                short,
                full,
                ts,
                part.get("callID")
                    .and_then(Value::as_str)
                    .map(str::to_string),
                None,
            );
            *data.metrics.tools.entry(name.to_string()).or_insert(0) += 1;
            data.metrics.tool_count += 1;
            let failed = state.get("status").and_then(Value::as_str) == Some("error");
            data.metrics.tool_errors += u64::from(failed);
            if let Some(detail) = data
                .metrics
                .tool_details
                .get_mut(name)
                .and_then(|details| details.last_mut())
            {
                detail.delta = delta;
                detail.dur_ms = duration_ms;
                detail.failed = failed;
            }
        }
    }

    fold_subagents(db, session_id, &mut data, &mut breakdown, &mut rates);

    let mut models: Vec<_> = breakdown.into_iter().collect();
    models.sort_by(|a, b| a.0.cmp(&b.0));
    data.models = models.iter().map(|(m, _)| m.clone()).collect();
    data.model_breakdown = models
        .into_iter()
        .map(|(model, (tokens, costs))| ModelBreakdown {
            model,
            total: costs.total,
            tokens,
            costs,
        })
        .collect();
    data
}

/// Fold a session's subagents into it.
///
/// OpenCode records each subagent as a session of its own, linked by
/// `parent_id`, where Claude keeps its subagents inside one transcript. So the
/// spend that Claude's parser sums for free has to be summed here by hand: each
/// child's tokens and cost are added to the parent's totals — otherwise hiding
/// the child rows would hide their spend — and each child is listed under the
/// parent rather than standing alone.
///
/// The subagent's own metrics come straight off its session row, which already
/// carries the aggregate OpenCode keeps; only the tool count is a query. The
/// per-message context window and tool history are left out, as its numbers on
/// the parent's Subagents pane do not need them.
fn fold_subagents(
    db: &Connection,
    parent_id: &str,
    data: &mut SessionData,
    breakdown: &mut HashMap<String, (Tokens, Costs)>,
    rates: &mut FallbackRates,
) {
    let Ok(mut stmt) = db.prepare(
        "SELECT id, agent, title, model, cost, tokens_input, tokens_output, tokens_reasoning, \
         tokens_cache_read, tokens_cache_write, time_created, time_updated \
         FROM session WHERE parent_id = ?1 ORDER BY time_created, id",
    ) else {
        return;
    };
    let Ok(rows) = stmt.query_map(params![parent_id], |row| {
        Ok((
            row.get::<_, String>(0)?,
            row.get::<_, Option<String>>(1)?,
            row.get::<_, String>(2)?,
            row.get::<_, Option<String>>(3)?,
            row.get::<_, f64>(4)?,
            row.get::<_, u64>(5)?,
            row.get::<_, u64>(6)?,
            row.get::<_, u64>(7)?,
            row.get::<_, u64>(8)?,
            row.get::<_, u64>(9)?,
            row.get::<_, i64>(10)?,
            row.get::<_, i64>(11)?,
        ))
    }) else {
        return;
    };

    let now_ms = chrono::Utc::now().timestamp_millis();
    for row in rows.flatten() {
        let (
            id,
            agent,
            title,
            model_json,
            cost,
            input,
            output,
            reasoning,
            read,
            write,
            created,
            updated,
        ) = row;
        let model = model_json
            .as_deref()
            .and_then(|raw| serde_json::from_str::<Value>(raw).ok())
            .and_then(|v| v.get("id").and_then(Value::as_str).map(str::to_string))
            .unwrap_or_default();
        let tokens = Tokens {
            input,
            output,
            cache_read: read,
            cache_write_5m: write,
            reasoning_output: reasoning,
            total: input + output + read + write,
            ..Default::default()
        };
        // The same zero-cost fallback the parent uses: a provider OpenCode has
        // no rates for reports nothing, and its tokens are priced from LiteLLM.
        let cost = if cost > 0.0 || tokens.total == 0 {
            cost
        } else {
            rates.costs(&model, &tokens).map_or(0.0, |c| c.total)
        };

        add_usage(
            data,
            &tokens,
            &Costs {
                total: cost,
                ..Default::default()
            },
        );
        let (mt, mc) = breakdown.entry(model.clone()).or_default();
        mt.input += tokens.input;
        mt.output += tokens.output;
        mt.cache_read += tokens.cache_read;
        mt.cache_write_5m += tokens.cache_write_5m;
        mt.reasoning_output += tokens.reasoning_output;
        mt.total += tokens.total;
        mc.total += cost;
        // Attribute the subagent's spend to when it last ran, so the burn
        // charts do not lose it. ponytail: one point per subagent, not per
        // message — a per-minute curve only if a subagent's timeline matters.
        if let Some(dt) = util::parse_ts(&util::ms_to_rfc3339(updated)) {
            data.record_cost(&dt, &model, cost);
        }

        let tool_count = db
            .query_row(
                "SELECT count(*) FROM part WHERE session_id = ?1 \
                 AND json_extract(data, '$.type') = 'tool'",
                params![id],
                |row| row.get::<_, u64>(0),
            )
            .unwrap_or(0);

        data.subagents.push(Subagent {
            agent_id: id,
            agent_type: agent.unwrap_or_default(),
            description: title,
            model,
            started_at: Some(util::ms_to_rfc3339(created)),
            last_active: Some(util::ms_to_rfc3339(updated)),
            duration_ms: (updated - created).max(0),
            // A subagent still writing is one whose parent has not moved on; the
            // same quiet window the Claude parser uses tells the two apart.
            status: if now_ms - updated >= 30_000 {
                SubagentStatus::Done
            } else {
                SubagentStatus::Running
            },
            cost,
            tool_count,
            tool_use_id: None,
            context: None,
            ghost: false,
        });
    }
}

pub fn delete(session: &Session) -> rusqlite::Result<()> {
    let Some(path) = &session.data_file else {
        return Ok(());
    };
    let db = Connection::open(path)?;
    db.execute_batch("PRAGMA foreign_keys = ON")?;
    db.execute(
        "DELETE FROM session WHERE id = ?1",
        params![session.session_id],
    )?;
    Ok(())
}

/// Context usage from the newest assistant message that carries token counts.
///
/// OpenCode records no window size of its own, so the ceiling comes from
/// LiteLLM's listing for the model. Without one there is no denominator, and a
/// guessed window would misreport CTX% for every model that does not match it.
pub fn extract_context(session: &Session) -> Option<ContextUsage> {
    with_db(session.data_file.as_ref()?, |db| context_of(db, session)).flatten()
}

fn context_of(db: &Connection, session: &Session) -> Option<ContextUsage> {
    // A turn that was aborted or failed before the provider answered is still
    // recorded, with every count at zero. Those sit at the end of many sessions,
    // so walk back past them rather than reading the newest row and giving up.
    let mut stmt = db
        .prepare(
            "SELECT data FROM message WHERE session_id = ?1 \
             AND json_extract(data, '$.role') = 'assistant' \
             ORDER BY time_created DESC, id DESC LIMIT 40",
        )
        .ok()?;
    let rows = stmt
        .query_map(params![session.session_id], |row| row.get::<_, String>(0))
        .ok()?;

    for raw in rows.flatten() {
        let Ok(message) = serde_json::from_str::<Value>(&raw) else {
            continue;
        };
        // What the next turn has to resend: everything the model was given,
        // cached or not. Output is excluded — the next input count includes it.
        let used = u64_at(&message, &["tokens", "input"])
            + u64_at(&message, &["tokens", "cache", "read"])
            + u64_at(&message, &["tokens", "cache", "write"]);
        if used == 0 {
            continue;
        }
        let model = message
            .get("modelID")
            .and_then(Value::as_str)
            .filter(|m| !m.is_empty())
            .unwrap_or(&session.model);
        return Some(ContextUsage {
            used,
            max: crate::pricing::litellm_max_input_tokens(model)?,
            compacted: false,
        });
    }
    None
}

pub fn extract_last_tool(session: &Session) -> String {
    let Some(path) = &session.data_file else {
        return String::new();
    };
    with_db(path, |db| last_tool(db, &session.session_id)).unwrap_or_default()
}

fn last_tool(db: &Connection, session_id: &str) -> String {
    db.query_row(
        "SELECT json_extract(data, '$.tool') FROM part \
         WHERE session_id = ?1 AND json_extract(data, '$.type') = 'tool' \
         ORDER BY time_created DESC, id DESC LIMIT 1",
        params![session_id],
        |row| row.get(0),
    )
    .optional()
    .ok()
    .flatten()
    .unwrap_or_default()
}

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

    fn temp_db(tag: &str) -> PathBuf {
        std::env::temp_dir().join(format!(
            "cctop-opencode-{tag}-{}-{}.db",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ))
    }

    #[test]
    fn extracts_sqlite_accounting_and_tool_activity() {
        let path = std::env::temp_dir().join(format!("cctop-opencode-{}.db", std::process::id()));
        let db = Connection::open(&path).unwrap();
        db.execute_batch(
            "CREATE TABLE session (
                id TEXT PRIMARY KEY, title TEXT NOT NULL, model TEXT, cost REAL NOT NULL,
                tokens_input INTEGER NOT NULL, tokens_output INTEGER NOT NULL,
                tokens_reasoning INTEGER NOT NULL, tokens_cache_read INTEGER NOT NULL,
                tokens_cache_write INTEGER NOT NULL
             );
             CREATE TABLE message (id TEXT PRIMARY KEY, session_id TEXT, time_created INTEGER, data TEXT);
             CREATE TABLE part (id TEXT PRIMARY KEY, session_id TEXT, time_created INTEGER, data TEXT);",
        )
        .unwrap();
        db.execute(
            "INSERT INTO session VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
            params![
                "ses_1",
                "OpenCode work",
                r#"{"id":"claude-test","providerID":"anthropic"}"#,
                0.75,
                120u64,
                30u64,
                5u64,
                80u64,
                10u64
            ],
        )
        .unwrap();
        db.execute(
            "INSERT INTO part VALUES (?1, ?2, ?3, ?4)",
            params![
                "part_2",
                "ses_1",
                1_785_888_062_000i64,
                r#"{"type":"tool","tool":"edit","callID":"call_2","state":{"input":{"filePath":"src/main.rs","oldString":"old\nline","newString":"new\nlines\nhere"},"time":{"start":1785888062000,"end":1785888062750},"status":"completed","output":"Updated file"}}"#
            ],
        )
        .unwrap();
        db.execute(
            "INSERT INTO message VALUES (?1, ?2, ?3, ?4)",
            params![
                "msg_1",
                "ses_1",
                1_785_888_060_000i64,
                r#"{"role":"assistant","time":{"created":1785888060000},"modelID":"claude-test","providerID":"anthropic","cost":0.75,"tokens":{"input":120,"output":30,"reasoning":5,"cache":{"read":80,"write":10}}}"#
            ],
        )
        .unwrap();
        db.execute(
            "INSERT INTO part VALUES (?1, ?2, ?3, ?4)",
            params![
                "part_1",
                "ses_1",
                1_785_888_061_000i64,
                r#"{"type":"tool","tool":"bash","callID":"call_1","state":{"input":{"command":"pwd"},"time":{"start":1785888061000},"status":"completed"}}"#
            ],
        )
        .unwrap();
        db.execute(
            "INSERT INTO part VALUES (?1, ?2, ?3, ?4)",
            params![
                "part_3",
                "ses_1",
                1_785_888_063_000i64,
                r#"{"type":"tool","tool":"bash","callID":"call_3","state":{"input":{"command":"exit 1"},"time":{"start":1785888063000},"status":"error","output":"exit status 1"}}"#
            ],
        )
        .unwrap();
        drop(db);

        let data = extract(&path, "ses_1");
        assert_eq!(data.title.as_deref(), Some("OpenCode work"));
        assert_eq!(data.last_model, "claude-test");
        assert_eq!(data.tokens.total, 240);
        assert!((data.costs.total - 0.75).abs() < 1e-9);
        // Two bash calls: one that completed and one that errored.
        assert_eq!(data.metrics.tools.get("bash"), Some(&2));
        let edit = &data.metrics.tool_details["edit"][0];
        assert_eq!(edit.d, "src/main.rs");
        assert_eq!(edit.dur_ms, Some(750));
        let delta = edit.delta.as_ref().unwrap();
        assert_eq!((delta.added, delta.removed), (3, 2));
        assert!(!edit.failed);

        // `status: "error"` is the outcome OpenCode records for a failed call.
        let bash = &data.metrics.tool_details["bash"];
        assert_eq!(bash.len(), 2);
        assert!(!bash[0].failed, "a completed call must not be flagged");
        assert!(bash[1].failed, "status=error must be flagged");
        assert_eq!(data.metrics.tool_errors, 1);

        // Release the handle this thread holds before removing the file.
        close_databases();
        std::fs::remove_file(path).unwrap();
    }

    /// A subagent is a session of its own with `parent_id` set. Its spend must
    /// count towards the parent it ran under, and it must be listed under it
    /// rather than folded away or left standing alone as a conversation.
    #[test]
    fn subagents_fold_into_their_parent() {
        let path = temp_db("subagents");
        let db = Connection::open(&path).unwrap();
        db.execute_batch(
            "CREATE TABLE session (
                id TEXT PRIMARY KEY, parent_id TEXT, agent TEXT, directory TEXT NOT NULL,
                title TEXT NOT NULL, model TEXT, cost REAL NOT NULL,
                tokens_input INTEGER NOT NULL, tokens_output INTEGER NOT NULL,
                tokens_reasoning INTEGER NOT NULL, tokens_cache_read INTEGER NOT NULL,
                tokens_cache_write INTEGER NOT NULL,
                time_created INTEGER NOT NULL, time_updated INTEGER NOT NULL
             );
             CREATE TABLE message (id TEXT PRIMARY KEY, session_id TEXT, time_created INTEGER, data TEXT);
             CREATE TABLE part (id TEXT PRIMARY KEY, session_id TEXT, time_created INTEGER, data TEXT);",
        )
        .unwrap();
        let cols = "id,parent_id,agent,directory,title,model,cost,tokens_input,tokens_output,\
                    tokens_reasoning,tokens_cache_read,tokens_cache_write,time_created,time_updated";
        db.execute(
            &format!("INSERT INTO session ({cols}) VALUES ('par',NULL,NULL,'/w','parent',?1,0.10,100,20,0,0,0,1000,2000)"),
            params![r#"{"id":"claude-x"}"#],
        )
        .unwrap();
        db.execute(
            &format!("INSERT INTO session ({cols}) VALUES ('kid','par','general','/w','find the bug',?1,0.05,200,40,0,0,0,1500,1800)"),
            params![r#"{"id":"claude-x"}"#],
        )
        .unwrap();
        // A tool call the subagent made, so its tool count is not zero.
        db.execute(
            "INSERT INTO part VALUES ('prt','kid',1600,?1)",
            params![r#"{"type":"tool","tool":"bash"}"#],
        )
        .unwrap();
        drop(db);

        let data = extract(&path, "par");
        assert_eq!(
            data.subagents.len(),
            1,
            "the child is listed under the parent"
        );
        let sa = &data.subagents[0];
        assert_eq!(sa.agent_type, "general");
        assert_eq!(sa.description, "find the bug");
        assert_eq!(sa.tool_count, 1);
        assert!((sa.cost - 0.05).abs() < 1e-9);
        // The parent's totals now carry the subagent's tokens and spend.
        assert_eq!(data.tokens.input, 300);
        assert!((data.costs.total - 0.15).abs() < 1e-9);

        close_databases();
        std::fs::remove_file(path).unwrap();
    }

    /// Every other read of this database shares the pool's read-only handle,
    /// which is then open for the life of the thread. `delete` opens a read-write
    /// one of its own on top of it, and the two have to coexist: a rollback
    /// journal grants a writer nothing while a reader holds the database, so a
    /// handle that outlived its usefulness would turn a deletion into a failure
    /// and leave the row the user asked to remove on screen.
    ///
    /// Read again afterwards through the same pooled handle, which is the other
    /// half of it — a handle kept open across a write must not answer with the
    /// row that has just gone.
    #[test]
    fn a_delete_lands_under_the_handle_the_pool_is_holding() {
        let path = temp_db("delete");
        let db = Connection::open(&path).unwrap();
        db.execute_batch(
            // Rollback rather than WAL on purpose: WAL lets a writer past a
            // reader, so only this mode can fail for the reason above.
            "PRAGMA journal_mode = DELETE;
             CREATE TABLE session (
                id TEXT PRIMARY KEY, title TEXT NOT NULL, model TEXT, cost REAL NOT NULL,
                tokens_input INTEGER NOT NULL, tokens_output INTEGER NOT NULL,
                tokens_reasoning INTEGER NOT NULL, tokens_cache_read INTEGER NOT NULL,
                tokens_cache_write INTEGER NOT NULL
             );
             CREATE TABLE message (id TEXT PRIMARY KEY, session_id TEXT, time_created INTEGER, data TEXT,
                FOREIGN KEY (session_id) REFERENCES session(id) ON DELETE CASCADE);
             CREATE TABLE part (id TEXT PRIMARY KEY, session_id TEXT, time_created INTEGER, data TEXT,
                FOREIGN KEY (session_id) REFERENCES session(id) ON DELETE CASCADE);",
        )
        .unwrap();
        db.execute(
            "INSERT INTO session VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
            params![
                "ses_doomed",
                "Doomed",
                r#"{"id":"claude-test","providerID":"anthropic"}"#,
                0.5,
                120u64,
                30u64,
                5u64,
                80u64,
                10u64
            ],
        )
        .unwrap();
        db.execute(
            "INSERT INTO message VALUES (?1, ?2, ?3, ?4)",
            params![
                "msg_1",
                "ses_doomed",
                1_785_888_060_000i64,
                r#"{"role":"assistant","time":{"created":1785888060000},"modelID":"claude-test","cost":0.5,"tokens":{"input":120,"output":30,"reasoning":5,"cache":{"read":80,"write":10}}}"#
            ],
        )
        .unwrap();
        db.execute(
            "INSERT INTO part VALUES (?1, ?2, ?3, ?4)",
            params![
                "part_1",
                "ses_doomed",
                1_785_888_061_000i64,
                r#"{"type":"tool","tool":"bash","callID":"call_1","state":{"input":{"command":"ls"},"time":{"start":1785888061000},"status":"completed"}}"#
            ],
        )
        .unwrap();
        drop(db);

        let mut session = Session::new(Provider::OpenCode, "ses_doomed".to_string());
        session.data_file = Some(path.clone());

        // Open the pooled handle first, the way a walk has before anyone deletes.
        let before = extract(&path, "ses_doomed");
        assert_eq!(before.tokens.total, 240);
        assert_eq!(before.metrics.tool_count, 1);

        delete(&session).expect("a pooled reader must not lock the writer out");

        // The cascade is what takes the spend with it, so this also pins that
        // `foreign_keys` is on for the deleting connection — without it the
        // session's own row would go and its messages would stay behind,
        // still adding up.
        let after = extract(&path, "ses_doomed");
        assert_eq!(after.title, None, "a deleted session must not come back");
        assert_eq!(after.tokens.total, 0, "and neither must its spend");
        assert_eq!(after.metrics.tool_count, 0);

        close_databases();
        std::fs::remove_file(path).unwrap();
    }

    /// A custom provider (any OpenAI-compatible endpoint OpenCode has no rates
    /// for) reports `cost: 0` on every message. Reporting those sessions as free
    /// hid real spend, so the tokens are priced against LiteLLM instead — and the
    /// route prefix in the model name must not stop that lookup from landing.
    #[test]
    fn a_custom_provider_reporting_no_cost_is_priced_from_litellm() {
        let _rates = crate::pricing::install_test_table(&[(
            "zai/glm-5.1",
            serde_json::json!({
                "input_cost_per_token": 1.4e-6,
                "output_cost_per_token": 4.4e-6,
                "cache_read_input_token_cost": 2.6e-7,
                "cache_creation_input_token_cost": 0.0,
                "max_input_tokens": 200_000u64,
            }),
        )]);
        let path = temp_db("priced");
        let db = Connection::open(&path).unwrap();
        db.execute_batch(
            "CREATE TABLE session (
                id TEXT PRIMARY KEY, title TEXT NOT NULL, model TEXT, cost REAL NOT NULL,
                tokens_input INTEGER NOT NULL, tokens_output INTEGER NOT NULL,
                tokens_reasoning INTEGER NOT NULL, tokens_cache_read INTEGER NOT NULL,
                tokens_cache_write INTEGER NOT NULL
             );
             CREATE TABLE message (id TEXT PRIMARY KEY, session_id TEXT, time_created INTEGER, data TEXT);
             CREATE TABLE part (id TEXT PRIMARY KEY, session_id TEXT, time_created INTEGER, data TEXT);",
        )
        .unwrap();
        db.execute(
            "INSERT INTO session VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
            params![
                "ses_1",
                "proxied",
                r#"{"id":"canopywave/zai/glm-5.1","providerID":"myproxy"}"#,
                0.0,
                0u64,
                0u64,
                0u64,
                0u64,
                0u64
            ],
        )
        .unwrap();
        db.execute(
            "INSERT INTO message VALUES (?1, ?2, ?3, ?4)",
            params![
                "msg_1",
                "ses_1",
                1_785_888_060_000i64,
                r#"{"role":"assistant","time":{"created":1785888060000},"modelID":"canopywave/zai/glm-5.1","providerID":"myproxy","cost":0,"tokens":{"input":1000000,"output":1000000,"reasoning":500000,"cache":{"read":1000000,"write":1000000}}}"#
            ],
        )
        .unwrap();
        drop(db);

        let data = extract(&path, "ses_1");
        // 1.40 input + 4.40 output + 0.26 cache read + 0.00 cache write per Mtok.
        // Reasoning is part of output, not billed on top of it.
        assert!(
            (data.costs.total - 6.06).abs() < 1e-9,
            "priced at {}",
            data.costs.total
        );

        let mut session = Session::new(Provider::OpenCode, "ses_1".to_string());
        session.data_file = Some(path.clone());
        let ctx = extract_context(&session).expect("context from the LiteLLM window");
        assert_eq!(ctx.used, 3_000_000);
        assert_eq!(ctx.max, 200_000);

        // Release the handle this thread holds before removing the file.
        close_databases();
        std::fs::remove_file(path).unwrap();
    }

    /// A cost OpenCode did compute is authoritative — pricing over the top of it
    /// would replace a provider's real invoice with a list price.
    #[test]
    fn a_reported_cost_is_never_second_guessed() {
        let _rates = crate::pricing::install_test_table(&[(
            "zai/glm-5.1",
            serde_json::json!({"input_cost_per_token": 1.4e-6, "output_cost_per_token": 4.4e-6}),
        )]);
        let mut rates = FallbackRates::default();
        let message = serde_json::json!({
            "role": "assistant",
            "modelID": "canopywave/zai/glm-5.1",
            "cost": 0.5,
            "tokens": {"input": 1_000_000, "output": 0, "cache": {"read": 0, "write": 0}},
        });
        let (_, _, _, costs) = assistant_usage(&message, &mut rates);
        assert!((costs.total - 0.5).abs() < 1e-9);
    }

    /// A model LiteLLM does not list has no window, and a guessed one would make
    /// every CTX% wrong in a way the user cannot see.
    #[test]
    fn an_unlisted_model_reports_no_context_rather_than_a_guess() {
        let _rates = crate::pricing::install_test_table(&[]);
        let path = temp_db("noctx");
        let db = Connection::open(&path).unwrap();
        db.execute_batch(
            "CREATE TABLE message (id TEXT PRIMARY KEY, session_id TEXT, time_created INTEGER, data TEXT);",
        )
        .unwrap();
        db.execute(
            "INSERT INTO message VALUES (?1, ?2, ?3, ?4)",
            params![
                "msg_1",
                "ses_1",
                1i64,
                r#"{"role":"assistant","modelID":"local/some-finetune","tokens":{"input":500,"output":10,"cache":{"read":0,"write":0}}}"#
            ],
        )
        .unwrap();
        drop(db);

        let mut session = Session::new(Provider::OpenCode, "ses_1".to_string());
        session.data_file = Some(path.clone());
        assert!(extract_context(&session).is_none());

        // Release the handle this thread holds before removing the file.
        close_databases();
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn input_request_waits_and_api_timeout_is_red() {
        let path = std::env::temp_dir().join(format!(
            "cctop-opencode-state-{}-{}.db",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        let db = Connection::open(&path).unwrap();
        db.execute_batch(
            "CREATE TABLE message (id TEXT PRIMARY KEY, session_id TEXT, time_created INTEGER, data TEXT);
             CREATE TABLE part (id TEXT PRIMARY KEY, message_id TEXT, data TEXT);",
        )
        .unwrap();
        db.execute(
            "INSERT INTO part VALUES (?1, ?2, ?3)",
            params![
                "part_question",
                "msg_done",
                r#"{"type":"tool","tool":"question"}"#
            ],
        )
        .unwrap();
        db.execute(
            "INSERT INTO message VALUES (?1, ?2, ?3, ?4)",
            params![
                "msg_done",
                "ses_done",
                1i64,
                r#"{"role":"assistant","time":{"completed":1},"error":{"name":"MessageAbortedError"}}"#
            ],
        )
        .unwrap();
        db.execute(
            "INSERT INTO message VALUES (?1, ?2, ?3, ?4)",
            params![
                "msg_timeout",
                "ses_timeout",
                1i64,
                r#"{"role":"assistant","time":{"completed":1},"error":{"name":"APIError","data":{"message":"Request timed out"}}}"#
            ],
        )
        .unwrap();
        db.execute(
            "INSERT INTO message VALUES (?1, ?2, ?3, ?4)",
            params![
                "msg_queue_full",
                "ses_queue_full",
                1i64,
                r#"{"role":"assistant","time":{"completed":1},"error":{"name":"APIError","data":{"message":"Streaming response failed: [503] The request queue is full."}}}"#
            ],
        )
        .unwrap();
        drop(db);

        assert_eq!(
            extract_activity_state(&path, "ses_done"),
            ActivityState::WaitingForInput
        );
        assert_eq!(
            extract_activity_state(&path, "ses_timeout"),
            ActivityState::ApiError
        );
        assert_eq!(
            extract_activity_state(&path, "ses_queue_full"),
            ActivityState::ApiError
        );
        // Release the handle this thread holds before removing the file.
        close_databases();
        std::fs::remove_file(path).unwrap();
    }

    /// Search reads both tables, because the text of a message is in `part` and
    /// its envelope and tool calls are in `message` — a term naming either has
    /// to find the session. One database holds every session, so the scan is
    /// also the only thing keeping another session's text out of a hit.
    #[test]
    fn a_scan_reads_both_tables_and_stops_where_it_is_told_to() {
        let path = temp_db("scan");
        let db = Connection::open(&path).unwrap();
        db.execute_batch(
            "CREATE TABLE message (id TEXT PRIMARY KEY, session_id TEXT, time_created INTEGER, data TEXT);
             CREATE TABLE part (id TEXT PRIMARY KEY, session_id TEXT, time_created INTEGER, data TEXT);",
        )
        .unwrap();
        db.execute(
            "INSERT INTO message VALUES (?1, ?2, ?3, ?4)",
            params![
                "msg_1",
                "ses_mine",
                1i64,
                r#"{"role":"assistant","modelID":"claude-test"}"#
            ],
        )
        .unwrap();
        db.execute(
            "INSERT INTO part VALUES (?1, ?2, ?3, ?4)",
            params![
                "part_1",
                "ses_mine",
                1i64,
                r#"{"type":"text","text":"the needle is in a part"}"#
            ],
        )
        .unwrap();
        // Another session's row, in both tables.
        db.execute(
            "INSERT INTO message VALUES (?1, ?2, ?3, ?4)",
            params![
                "msg_2",
                "ses_theirs",
                1i64,
                r#"{"role":"assistant","modelID":"claude-test"}"#
            ],
        )
        .unwrap();
        db.execute(
            "INSERT INTO part VALUES (?1, ?2, ?3, ?4)",
            params![
                "part_2",
                "ses_theirs",
                1i64,
                r#"{"type":"text","text":"the needle is not ours"}"#
            ],
        )
        .unwrap();
        drop(db);

        let collect = |session_id: &str| {
            let mut seen = Vec::new();
            let found = scan_session(&path, session_id, |raw| {
                seen.push(raw.to_string());
                false
            });
            (found, seen)
        };

        let (found, seen) = collect("ses_mine");
        assert!(!found, "the feed said no, so the scan must read it all");
        assert_eq!(seen.len(), 2, "a message and a part, not one of them");
        assert!(seen.iter().any(|raw| raw.contains("in a part")));
        assert!(
            !seen.iter().any(|raw| raw.contains("not ours")),
            "another session's text must not reach this hit"
        );

        // The envelope carries no needle of its own here, so a term named in the
        // message alone is the other half of what makes this worth scanning both.
        assert!(scan_session(&path, "ses_mine", |raw| raw.contains("modelID")));
        assert!(
            !scan_session(&path, "ses_mine", |raw| raw.contains("absent")),
            "a term nothing holds must not"
        );

        close_databases();
        std::fs::remove_file(path).unwrap();
    }
}