deciduous 1.0.12

//! SQLite database with Diesel ORM
//!
//! Stores decision graphs and command logs for AI-assisted development.
//! Uses embedded migrations for schema management.

use crate::records::{EdgeRecord, NodeRecord, RecordStore, TagRecord, ThemeRecord};
use crate::schema::*;
use diesel::prelude::*;
use diesel::r2d2::{ConnectionManager, Pool, PooledConnection};
use diesel::sqlite::SqliteConnection;
use serde_json::json;
use std::path::Path;
#[cfg(feature = "ts-rs")]
use ts_rs::TS;
use uuid::Uuid;

/// Build metadata JSON from optional fields (confidence, commit, prompt, files, branch)
pub fn build_metadata_json(
    confidence: Option<u8>,
    commit: Option<&str>,
    prompt: Option<&str>,
    files: Option<&str>,
    branch: Option<&str>,
) -> Option<String> {
    // Only create JSON if at least one field is present
    if confidence.is_none()
        && commit.is_none()
        && prompt.is_none()
        && files.is_none()
        && branch.is_none()
    {
        return None;
    }

    let mut obj = serde_json::Map::new();

    if let Some(c) = confidence {
        obj.insert("confidence".to_string(), json!(c.min(100)));
    }
    if let Some(h) = commit {
        obj.insert("commit".to_string(), json!(h));
    }
    if let Some(p) = prompt {
        obj.insert("prompt".to_string(), json!(p));
    }
    if let Some(f) = files {
        // Split comma-separated files into array
        let file_list: Vec<&str> = f.split(',').map(|s| s.trim()).collect();
        obj.insert("files".to_string(), json!(file_list));
    }
    if let Some(b) = branch {
        obj.insert("branch".to_string(), json!(b));
    }

    Some(serde_json::Value::Object(obj).to_string())
}

/// Get current git branch name
pub fn get_current_git_branch() -> Option<String> {
    std::process::Command::new("git")
        .args(["rev-parse", "--abbrev-ref", "HEAD"])
        .output()
        .ok()
        .and_then(|output| {
            if output.status.success() {
                String::from_utf8(output.stdout)
                    .ok()
                    .map(|s| s.trim().to_string())
                    .filter(|s| !s.is_empty() && s != "HEAD")
            } else {
                None
            }
        })
}

/// Get the current HEAD commit hash (short form, 7 chars)
pub fn get_current_git_commit() -> Option<String> {
    std::process::Command::new("git")
        .args(["rev-parse", "--short", "HEAD"])
        .output()
        .ok()
        .and_then(|output| {
            if output.status.success() {
                String::from_utf8(output.stdout)
                    .ok()
                    .map(|s| s.trim().to_string())
                    .filter(|s| !s.is_empty())
            } else {
                None
            }
        })
}

/// Every kind of node reference the CLI takes, for errors that have to say.
pub const NODE_REF_KINDS: &str = "A node is named by its local id (12, or #12 when \
    digits could also be a change_id prefix), its change_id or a prefix of it (the CHANGE \
    column of `deciduous nodes`, at least 4 hex characters), or, in a project with a \
    [remote], the server's id for it (the `id` MCP tools return) or a prefix of that.";

/// The full hash of the commit `rev` names in the current directory's
/// repository: `HEAD`, a branch, a tag, `origin/main`, `HEAD~2`, a hash or a
/// prefix of one. Anything git cannot resolve to a commit is an error naming
/// `rev`, never the string stored as if it were a commit: `--commit
/// origin/main` used to store the literal "origin/main".
pub fn resolve_git_commit(rev: &str) -> std::result::Result<String, String> {
    let rev = rev.trim();
    if rev.is_empty() || rev.starts_with('-') {
        return Err(format!("{rev:?} is not a git revision"));
    }
    let out = std::process::Command::new("git")
        .args(["rev-parse", "--verify", "--quiet", "--end-of-options"])
        .arg(format!("{rev}^{{commit}}"))
        .output()
        .map_err(|e| format!("could not run git to resolve {rev:?}: {e}"))?;
    let hash = String::from_utf8_lossy(&out.stdout).trim().to_string();
    if out.status.success() && !hash.is_empty() {
        return Ok(hash);
    }
    let in_repo = std::process::Command::new("git")
        .args(["rev-parse", "--git-dir"])
        .output()
        .is_ok_and(|o| o.status.success());
    Err(if in_repo {
        format!("{rev:?} is not a commit in this repository (git rev-parse could not resolve it)")
    } else {
        format!("{rev:?} cannot be resolved: this directory is not in a git repository")
    })
}

/// Walk up directory tree to find .deciduous folder (like git finds .git)
/// Can be overridden with DECIDUOUS_DB_PATH env var
fn get_db_path() -> std::path::PathBuf {
    // Check env var first - always takes priority
    // Made absolute: a bare `deciduous.db` has an empty parent, and every
    // file found beside the database (the server log, graph.json) went
    // missing with it.
    if let Ok(path) = std::env::var("DECIDUOUS_DB_PATH") {
        let path = std::path::PathBuf::from(path);
        return std::path::absolute(&path).unwrap_or(path);
    }

    // Walk up directory tree to find .deciduous folder
    if let Ok(current_dir) = std::env::current_dir() {
        let mut dir = current_dir.as_path();
        loop {
            let deciduous_dir = dir.join(".deciduous");
            if deciduous_dir.exists() && deciduous_dir.is_dir() {
                return deciduous_dir.join("deciduous.db");
            }
            // Move to parent directory
            match dir.parent() {
                Some(parent) => dir = parent,
                None => break, // Reached filesystem root
            }
        }
    }

    // No .deciduous found - default to current directory
    // (deciduous init will create it here)
    std::path::PathBuf::from(".deciduous/deciduous.db")
}

/// Node types a new node may have.
pub const NODE_TYPES: &[&str] = &[
    "goal",
    "decision",
    "option",
    "action",
    "outcome",
    "observation",
    "revisit",
];

/// Statuses a node may be set to.
pub const NODE_STATUSES: &[&str] = &[
    "pending",
    "active",
    "completed",
    "rejected",
    "superseded",
    "abandoned",
];

/// Edge types a new edge may have.
pub const EDGE_TYPES: &[&str] = &[
    "leads_to",
    "requires",
    "chosen",
    "rejected",
    "blocks",
    "enables",
    "took_from",
];

/// Node types a record read from graph.json or the server may carry: every
/// type a new node may have, plus `feedback`, which exists in graphs on disk
/// and which the server accepts. Kept equal to the server's list
/// (deciduous_mcp/lib/deciduous_mcp/schema/node.ex), so a pull is never
/// refused here for a node the server holds.
pub const IMPORTABLE_NODE_TYPES: &[&str] = &[
    "goal",
    "decision",
    "option",
    "action",
    "outcome",
    "observation",
    "revisit",
    "feedback",
];

/// Statuses a record may carry: every settable status plus `done`, for the
/// same reason as [`IMPORTABLE_NODE_TYPES`].
pub const IMPORTABLE_STATUSES: &[&str] = &[
    "pending",
    "active",
    "completed",
    "rejected",
    "superseded",
    "abandoned",
    "done",
];

fn one_of(what: &str, value: &str, allowed: &[&str]) -> Result<()> {
    if allowed.contains(&value) {
        Ok(())
    } else {
        Err(DbError::Validation(format!(
            "unknown {what} '{value}'; expected one of: {}",
            allowed.join(", ")
        )))
    }
}

/// Current schema version for deciduous
pub const CURRENT_SCHEMA: DecisionSchema = DecisionSchema {
    major: 1,
    minor: 1,
    patch: 0,
    name: "decision-graph",
    features: &[
        "decision_nodes",
        "decision_edges",
        "decision_context",
        "decision_sessions",
        "command_log",
        "node_documents",
        "themes",
        "node_themes",
    ],
};

/// Describes the version and capabilities of the schema
#[derive(Debug, Clone)]
pub struct DecisionSchema {
    pub major: u32,
    pub minor: u32,
    pub patch: u32,
    pub name: &'static str,
    pub features: &'static [&'static str],
}

impl DecisionSchema {
    pub fn version_string(&self) -> String {
        format!("{}.{}.{}", self.major, self.minor, self.patch)
    }

    pub fn is_compatible_with(&self, other: &DecisionSchema) -> bool {
        self.major == other.major
    }

    pub fn is_newer_than(&self, other: &DecisionSchema) -> bool {
        (self.major, self.minor, self.patch) > (other.major, other.minor, other.patch)
    }

    pub fn has_feature(&self, feature: &str) -> bool {
        self.features.contains(&feature)
    }
}

impl std::fmt::Display for DecisionSchema {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "v{} ({})", self.version_string(), self.name)
    }
}

// ============================================================================
// Diesel Models
// ============================================================================

/// Insertable schema version
#[derive(Insertable)]
#[diesel(table_name = schema_versions)]
pub struct NewSchemaVersion<'a> {
    pub version: &'a str,
    pub name: &'a str,
    pub features: &'a str,
    pub introduced_at: &'a str,
}

/// Queryable schema version
#[derive(Queryable, Selectable, Debug, Clone, serde::Serialize)]
#[diesel(table_name = schema_versions)]
pub struct StoredSchema {
    pub id: i32,
    pub version: String,
    pub name: String,
    pub features: String,
    pub introduced_at: String,
}

// ============================================================================
// Decision Graph Models
// ============================================================================

/// Insertable decision node
#[derive(Insertable)]
#[diesel(table_name = decision_nodes)]
pub struct NewDecisionNode<'a> {
    pub change_id: &'a str,
    pub node_type: &'a str,
    pub title: &'a str,
    pub description: Option<&'a str>,
    pub status: &'a str,
    pub created_at: &'a str,
    pub updated_at: &'a str,
    pub metadata_json: Option<&'a str>,
}

/// Queryable decision node
#[derive(Queryable, Selectable, Debug, Clone, serde::Serialize)]
#[cfg_attr(feature = "ts-rs", derive(TS))]
#[cfg_attr(feature = "ts-rs", ts(export))]
#[diesel(table_name = decision_nodes)]
pub struct DecisionNode {
    pub id: i32,
    pub change_id: String,
    pub node_type: String,
    pub title: String,
    pub description: Option<String>,
    pub status: String,
    pub created_at: String,
    pub updated_at: String,
    pub metadata_json: Option<String>,
}

/// Summary of what was deleted by delete_node
#[derive(Debug)]
pub struct DeleteSummary {
    pub node_title: String,
    pub edges_deleted: usize,
}

/// Insertable decision edge
#[derive(Insertable)]
#[diesel(table_name = decision_edges)]
pub struct NewDecisionEdge<'a> {
    pub from_node_id: i32,
    pub to_node_id: i32,
    pub from_change_id: Option<&'a str>,
    pub to_change_id: Option<&'a str>,
    pub edge_type: &'a str,
    pub weight: Option<f64>,
    pub rationale: Option<&'a str>,
    pub created_at: &'a str,
}

/// Queryable decision edge
#[derive(Queryable, Selectable, Debug, Clone, serde::Serialize)]
#[cfg_attr(feature = "ts-rs", derive(TS))]
#[cfg_attr(feature = "ts-rs", ts(export))]
#[diesel(table_name = decision_edges)]
pub struct DecisionEdge {
    pub id: i32,
    pub from_node_id: i32,
    pub to_node_id: i32,
    pub from_change_id: Option<String>,
    pub to_change_id: Option<String>,
    pub edge_type: String,
    pub weight: Option<f64>,
    pub rationale: Option<String>,
    pub created_at: String,
}

/// Insertable decision context
#[derive(Insertable)]
#[diesel(table_name = decision_context)]
pub struct NewDecisionContext<'a> {
    pub node_id: i32,
    pub context_type: &'a str,
    pub content_json: &'a str,
    pub captured_at: &'a str,
}

/// Queryable decision context
#[derive(Queryable, Selectable, Debug, Clone, serde::Serialize)]
#[cfg_attr(feature = "ts-rs", derive(TS))]
#[cfg_attr(feature = "ts-rs", ts(export))]
#[diesel(table_name = decision_context)]
pub struct DecisionContext {
    pub id: i32,
    pub node_id: i32,
    pub context_type: String,
    pub content_json: String,
    pub captured_at: String,
}

/// Insertable session
#[derive(Insertable)]
#[diesel(table_name = decision_sessions)]
pub struct NewDecisionSession<'a> {
    pub name: Option<&'a str>,
    pub started_at: &'a str,
    pub ended_at: Option<&'a str>,
    pub root_node_id: Option<i32>,
    pub summary: Option<&'a str>,
}

/// Queryable session
#[derive(Queryable, Selectable, Debug, Clone, serde::Serialize)]
#[cfg_attr(feature = "ts-rs", derive(TS))]
#[cfg_attr(feature = "ts-rs", ts(export))]
#[diesel(table_name = decision_sessions)]
pub struct DecisionSession {
    pub id: i32,
    pub name: Option<String>,
    pub started_at: String,
    pub ended_at: Option<String>,
    pub root_node_id: Option<i32>,
    pub summary: Option<String>,
}

// ============================================================================
// Command Log Models
// ============================================================================

/// Insertable command log entry
#[derive(Insertable)]
#[diesel(table_name = command_log)]
pub struct NewCommandLog<'a> {
    pub command: &'a str,
    pub description: Option<&'a str>,
    pub working_dir: Option<&'a str>,
    pub exit_code: Option<i32>,
    pub stdout: Option<&'a str>,
    pub stderr: Option<&'a str>,
    pub started_at: &'a str,
    pub completed_at: Option<&'a str>,
    pub duration_ms: Option<i32>,
    pub decision_node_id: Option<i32>,
}

/// Queryable command log entry
#[derive(Queryable, Selectable, Debug, Clone, serde::Serialize)]
#[cfg_attr(feature = "ts-rs", derive(TS))]
#[cfg_attr(feature = "ts-rs", ts(export))]
#[diesel(table_name = command_log)]
pub struct CommandLog {
    pub id: i32,
    pub command: String,
    pub description: Option<String>,
    pub working_dir: Option<String>,
    pub exit_code: Option<i32>,
    pub stdout: Option<String>,
    pub stderr: Option<String>,
    pub started_at: String,
    pub completed_at: Option<String>,
    pub duration_ms: Option<i32>,
    pub decision_node_id: Option<i32>,
}

// ============================================================================
// Roadmap Board Models
// ============================================================================

/// Checkbox state enum for type safety
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[cfg_attr(feature = "ts-rs", derive(TS))]
#[cfg_attr(feature = "ts-rs", ts(export))]
pub enum CheckboxState {
    /// Section header or item without checkbox
    None,
    /// Unchecked checkbox: - [ ]
    Unchecked,
    /// Checked checkbox: - [x]
    Checked,
}

impl CheckboxState {
    /// Convert to database string representation
    pub fn as_str(&self) -> &'static str {
        match self {
            CheckboxState::None => "none",
            CheckboxState::Unchecked => "unchecked",
            CheckboxState::Checked => "checked",
        }
    }

    /// Parse from database string representation
    pub fn parse(s: &str) -> Self {
        match s.to_lowercase().as_str() {
            "checked" => CheckboxState::Checked,
            "unchecked" => CheckboxState::Unchecked,
            _ => CheckboxState::None,
        }
    }

    /// Convert from boolean (for checkbox items)
    pub fn from_bool(checked: bool) -> Self {
        if checked {
            CheckboxState::Checked
        } else {
            CheckboxState::Unchecked
        }
    }

    /// Check if this represents a checked state
    pub fn is_checked(&self) -> bool {
        matches!(self, CheckboxState::Checked)
    }
}

impl std::fmt::Display for CheckboxState {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.as_str())
    }
}

/// Insertable roadmap item
#[derive(Insertable)]
#[diesel(table_name = roadmap_items)]
pub struct NewRoadmapItem<'a> {
    pub change_id: &'a str,
    pub title: &'a str,
    pub description: Option<&'a str>,
    pub section: Option<&'a str>,
    pub parent_id: Option<i32>,
    pub checkbox_state: &'a str,
    pub github_issue_number: Option<i32>,
    pub github_issue_state: Option<&'a str>,
    pub outcome_node_id: Option<i32>,
    pub outcome_change_id: Option<&'a str>,
    pub markdown_line_start: Option<i32>,
    pub markdown_line_end: Option<i32>,
    pub content_hash: Option<&'a str>,
    pub created_at: &'a str,
    pub updated_at: &'a str,
    pub last_synced_at: Option<&'a str>,
}

/// Queryable roadmap item
#[derive(Queryable, Selectable, Debug, Clone, serde::Serialize, serde::Deserialize)]
#[cfg_attr(feature = "ts-rs", derive(TS))]
#[cfg_attr(feature = "ts-rs", ts(export))]
#[diesel(table_name = roadmap_items)]
pub struct RoadmapItem {
    pub id: i32,
    pub change_id: String,
    pub title: String,
    pub description: Option<String>,
    pub section: Option<String>,
    pub parent_id: Option<i32>,
    pub checkbox_state: String,
    pub github_issue_number: Option<i32>,
    pub github_issue_state: Option<String>,
    pub outcome_node_id: Option<i32>,
    pub outcome_change_id: Option<String>,
    pub markdown_line_start: Option<i32>,
    pub markdown_line_end: Option<i32>,
    pub content_hash: Option<String>,
    pub created_at: String,
    pub updated_at: String,
    pub last_synced_at: Option<String>,
}

impl RoadmapItem {
    /// Get the checkbox state as a typed enum
    pub fn checkbox(&self) -> CheckboxState {
        CheckboxState::parse(&self.checkbox_state)
    }

    /// Check if this item is completed (checkbox checked)
    pub fn is_checked(&self) -> bool {
        self.checkbox().is_checked()
    }
}

/// Insertable roadmap sync state
#[derive(Insertable)]
#[diesel(table_name = roadmap_sync_state)]
pub struct NewRoadmapSyncState<'a> {
    pub roadmap_path: &'a str,
    pub roadmap_content_hash: Option<&'a str>,
    pub github_repo: Option<&'a str>,
    pub last_github_sync: Option<&'a str>,
    pub last_markdown_parse: Option<&'a str>,
    pub conflict_count: i32,
}

/// Queryable roadmap sync state
#[derive(Queryable, Selectable, Debug, Clone, serde::Serialize)]
#[cfg_attr(feature = "ts-rs", derive(TS))]
#[cfg_attr(feature = "ts-rs", ts(export))]
#[diesel(table_name = roadmap_sync_state)]
pub struct RoadmapSyncState {
    pub id: i32,
    pub roadmap_path: String,
    pub roadmap_content_hash: Option<String>,
    pub github_repo: Option<String>,
    pub last_github_sync: Option<String>,
    pub last_markdown_parse: Option<String>,
    pub conflict_count: i32,
}

/// Insertable roadmap conflict
#[derive(Insertable)]
#[diesel(table_name = roadmap_conflicts)]
pub struct NewRoadmapConflict<'a> {
    pub item_change_id: &'a str,
    pub conflict_type: &'a str,
    pub local_value: Option<&'a str>,
    pub remote_value: Option<&'a str>,
    pub resolution: Option<&'a str>,
    pub detected_at: &'a str,
    pub resolved_at: Option<&'a str>,
}

/// Queryable roadmap conflict
#[derive(Queryable, Selectable, Debug, Clone, serde::Serialize)]
#[cfg_attr(feature = "ts-rs", derive(TS))]
#[cfg_attr(feature = "ts-rs", ts(export))]
#[diesel(table_name = roadmap_conflicts)]
pub struct RoadmapConflict {
    pub id: i32,
    pub item_change_id: String,
    pub conflict_type: String,
    pub local_value: Option<String>,
    pub remote_value: Option<String>,
    pub resolution: Option<String>,
    pub detected_at: String,
    pub resolved_at: Option<String>,
}

// ============================================================================
// GitHub Issue Cache
// ============================================================================

/// Insertable GitHub issue cache entry
#[derive(Insertable, Debug)]
#[diesel(table_name = github_issue_cache)]
pub struct NewGitHubIssueCache<'a> {
    pub issue_number: i32,
    pub repo: &'a str,
    pub title: &'a str,
    pub body: Option<&'a str>,
    pub state: &'a str,
    pub html_url: &'a str,
    pub created_at: &'a str,
    pub updated_at: &'a str,
    pub cached_at: &'a str,
}

/// Queryable GitHub issue cache entry
#[derive(Queryable, Selectable, Debug, Clone, serde::Serialize)]
#[cfg_attr(feature = "ts-rs", derive(TS))]
#[cfg_attr(feature = "ts-rs", ts(export))]
#[diesel(table_name = github_issue_cache)]
pub struct GitHubIssueCache {
    pub id: i32,
    pub issue_number: i32,
    pub repo: String,
    pub title: String,
    pub body: Option<String>,
    pub state: String,
    pub html_url: String,
    pub created_at: String,
    pub updated_at: String,
    pub cached_at: String,
}

// ============================================================================
// Helper structs for raw SQL queries
// ============================================================================

/// Helper for PRAGMA table_info queries
#[derive(QueryableByName, Debug)]
struct PragmaTableInfo {
    #[diesel(sql_type = diesel::sql_types::Integer)]
    #[allow(dead_code)]
    cid: i32,
    #[diesel(sql_type = diesel::sql_types::Text)]
    name: String,
    #[diesel(sql_type = diesel::sql_types::Text)]
    #[allow(dead_code)]
    r#type: String,
    #[diesel(sql_type = diesel::sql_types::Integer)]
    #[allow(dead_code)]
    notnull: i32,
    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
    #[allow(dead_code)]
    dflt_value: Option<String>,
    #[diesel(sql_type = diesel::sql_types::Integer)]
    #[allow(dead_code)]
    pk: i32,
}

/// Helper for node ID queries
#[derive(QueryableByName, Debug)]
struct NodeIdOnly {
    #[diesel(sql_type = diesel::sql_types::Integer)]
    id: i32,
}

/// Helper for sqlite_master table queries
#[derive(QueryableByName, Debug)]
#[allow(dead_code)]
struct TableInfo {
    #[diesel(sql_type = diesel::sql_types::Text)]
    name: String,
}

/// Helper for FTS5 search results
#[derive(QueryableByName, Debug)]
struct FtsSearchRow {
    #[diesel(sql_type = diesel::sql_types::Integer)]
    id: i32,
    #[diesel(sql_type = diesel::sql_types::Text)]
    user_prompt: String,
    #[diesel(sql_type = diesel::sql_types::Text)]
    total_prompt: String,
    #[diesel(sql_type = diesel::sql_types::Text)]
    response: String,
    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
    context_json: Option<String>,
    #[diesel(sql_type = diesel::sql_types::Text)]
    inserted_at: String,
    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
    deleted_at: Option<String>,
    #[diesel(sql_type = diesel::sql_types::Double)]
    rank: f64,
    #[diesel(sql_type = diesel::sql_types::Text)]
    snippet_prompt: String,
    #[diesel(sql_type = diesel::sql_types::Text)]
    snippet_response: String,
}

// ============================================================================
// Database Connection
// ============================================================================

type DbPool = Pool<ConnectionManager<SqliteConnection>>;

/// How long a statement waits for another process's write lock before it
/// gives up with "database is locked". A CLI command, two MCP servers and the
/// API daemon routinely write one file at once; a single write holds the lock
/// for milliseconds, so ten seconds is only ever reached by a stuck process.
pub const BUSY_TIMEOUT_MS: u32 = 10_000;

/// Set on every pooled connection before first use.
///
/// `busy_timeout` is per connection and defaults to 0, which is what made
/// half of all concurrent MCP writes fail immediately. `journal_mode=WAL` is
/// stored in the file, so the first connection converts the database once;
/// after that readers never block the writer and the writer never blocks
/// readers. The timeout goes first so the conversion itself waits for a lock
/// rather than failing on one.
#[derive(Debug)]
struct SqlitePragmas;

impl diesel::r2d2::CustomizeConnection<SqliteConnection, diesel::r2d2::Error> for SqlitePragmas {
    fn on_acquire(
        &self,
        conn: &mut SqliteConnection,
    ) -> std::result::Result<(), diesel::r2d2::Error> {
        use diesel::connection::SimpleConnection;
        conn.batch_execute(&format!(
            "PRAGMA busy_timeout = {BUSY_TIMEOUT_MS}; PRAGMA journal_mode = WAL;"
        ))
        .map_err(diesel::r2d2::Error::QueryError)
    }
}
type DbConn = PooledConnection<ConnectionManager<SqliteConnection>>;

/// Database connection wrapper with connection pool.
///
/// When a graph file is attached (automatically, if `graph.json` sits
/// sits next to the database file), every graph mutation is mirrored into
/// it so teammates receive it through git. See [`crate::records`].
pub struct Database {
    pool: DbPool,
    /// The database file, absolute. Everything else that belongs to the
    /// project (documents, the active-session file) is found next to it,
    /// never relative to the working directory: an MCP server started in
    /// `repo/src` opens `repo/.deciduous/deciduous.db` by walking up, and a
    /// cwd-relative `.deciduous/documents` there created a second
    /// `.deciduous/` that every later command in `src` then opened instead.
    path: std::path::PathBuf,
    /// Attached graph file, if any. Behind a lock so `deciduous sync` can
    /// attach a store it just created without a mutable handle.
    store: std::sync::RwLock<Option<RecordStore>>,
    /// Log of writes bound for the shared server, attached when the
    /// project has a `[remote]`. See [`crate::oplog`].
    oplog: std::sync::RwLock<Option<crate::oplog::OpLog>>,
    /// Whether to attach the graph file next to the database once it
    /// appears. A long-lived handle (the API daemon's per-graph cache, an
    /// MCP server) that opened before `deciduous sync` created graph.json
    /// otherwise never wrote to it: every record after that was missing
    /// until someone ran sync again. `set_store(None)` turns it off.
    auto_attach: std::sync::atomic::AtomicBool,
    /// Author for a graph file attached later (see `set_store_author`).
    store_author: std::sync::RwLock<Option<String>>,
    /// The HEAD file of the repository holding the graph file, found once:
    /// reading it is how every write-through asks, cheaply, whether HEAD is
    /// on a branch (see [`Self::write_store`]).
    head_file: std::sync::OnceLock<Option<std::path::PathBuf>>,
    /// Set while the database is being made to follow graph.json after git
    /// changed it (a checkout, reset or pull). What that changes is local
    /// only: a node the checked-out branch does not have is not deleted on
    /// the shared server, and an older version brought back by a reset is
    /// not an edit to send. See [`Self::with_ops_suppressed`].
    ops_suppressed: std::sync::atomic::AtomicBool,
    /// Set by a local write (not one git brought) made while there is no
    /// server log, and taken by [`Self::to_log`] once it is committed, so
    /// the command can say its write stayed here.
    local_only_pending: std::sync::atomic::AtomicBool,
    /// The follow lock this process holds, and how many writes/syncs in it
    /// are inside it (see [`Self::hold_mirror`]).
    mirror_hold: std::sync::Mutex<(usize, Option<std::fs::File>)>,
}

/// Held from the start of a graph write until it is in graph.json, and for
/// the whole of a sync or a follow. See [`Database::hold_mirror`].
pub struct MirrorGuard<'a>(Option<&'a Database>);

impl Drop for MirrorGuard<'_> {
    fn drop(&mut self) {
        if let Some(db) = self.0 {
            let mut hold = db.mirror_hold.lock().unwrap_or_else(|e| e.into_inner());
            hold.0 = hold.0.saturating_sub(1);
            if hold.0 == 0 {
                hold.1 = None; // closing the file releases the lock
            }
        }
    }
}

/// Error type for database operations
#[derive(Debug)]
pub enum DbError {
    Connection(String),
    Query(diesel::result::Error),
    Pool(diesel::r2d2::Error),
    Validation(String),
    /// A reference that names no node here: no local id, no change_id
    /// prefix. Kept apart from Validation so the CLI can go on to ask the
    /// server whether it is one of the server's ids.
    NoSuchNode(String),
}

impl std::fmt::Display for DbError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            DbError::Connection(msg) => write!(f, "Connection error: {}", msg),
            DbError::Query(e) => write!(f, "Query error: {}", e),
            DbError::Pool(e) => write!(f, "Pool error: {}", e),
            DbError::Validation(msg) | DbError::NoSuchNode(msg) => write!(f, "{}", msg),
        }
    }
}

impl std::error::Error for DbError {}

impl From<diesel::result::Error> for DbError {
    fn from(e: diesel::result::Error) -> Self {
        DbError::Query(e)
    }
}

impl From<diesel::r2d2::Error> for DbError {
    fn from(e: diesel::r2d2::Error) -> Self {
        DbError::Pool(e)
    }
}

pub type Result<T> = std::result::Result<T, DbError>;

/// An op written to `remote_outbox` in its write's transaction, on its way to
/// the log. See `Database::queue_in_tx`.
struct Queued {
    seq: i64,
    op: crate::oplog::Op,
}

/// A one-entry JSON object, for an op that changes a single field.
fn one_field(key: &str, value: &str) -> serde_json::Map<String, serde_json::Value> {
    let mut m = serde_json::Map::new();
    m.insert(
        key.to_string(),
        serde_json::Value::String(value.to_string()),
    );
    m
}

fn node_not_found(node_id: i32) -> DbError {
    DbError::Validation(format!(
        "Node {node_id} does not exist. Run 'deciduous nodes' to see existing nodes."
    ))
}

impl Database {
    /// Get the database path that will be used
    pub fn db_path() -> std::path::PathBuf {
        get_db_path()
    }

    /// Create a new database at a custom path
    pub fn new(path: &str) -> Result<Self> {
        Self::open_at(path)
    }

    /// Open database at default path (respects DECIDUOUS_DB_PATH env var)
    pub fn open() -> Result<Self> {
        let path = get_db_path();
        // Create parent directory if it doesn't exist
        if let Some(parent) = path.parent() {
            if !parent.exists() {
                std::fs::create_dir_all(parent).ok();
            }
        }
        Self::open_at(&path)
    }

    /// Open database at specified path
    pub fn open_at<P: AsRef<Path>>(path: P) -> Result<Self> {
        let path_str = path.as_ref().to_string_lossy().to_string();
        let manager = ConnectionManager::<SqliteConnection>::new(&path_str);
        let pool = Pool::builder()
            .max_size(5)
            .connection_customizer(Box::new(SqlitePragmas))
            .build(manager)
            .map_err(|e| DbError::Connection(e.to_string()))?;

        let db = Self {
            pool,
            path: std::path::absolute(path.as_ref())
                .unwrap_or_else(|_| path.as_ref().to_path_buf()),
            store: std::sync::RwLock::new(None),
            oplog: std::sync::RwLock::new(None),
            auto_attach: std::sync::atomic::AtomicBool::new(true),
            store_author: std::sync::RwLock::new(None),
            head_file: std::sync::OnceLock::new(),
            ops_suppressed: std::sync::atomic::AtomicBool::new(false),
            local_only_pending: std::sync::atomic::AtomicBool::new(false),
            mirror_hold: std::sync::Mutex::new((0, None)),
        };
        // Auto-migrate FIRST - add change_id columns to existing databases before init_schema creates new tables
        let _ = db.migrate_add_change_ids_raw();
        db.init_schema()?;
        if let Some(store) = RecordStore::path_for_db(path.as_ref()).and_then(RecordStore::open) {
            db.set_store(Some(store));
        }
        db.set_oplog(crate::oplog::OpLog::for_db(&db.path));
        db.drain_outbox_at_open();
        Ok(db)
    }

    /// Attach (or detach) the log that writes are queued in for the server.
    pub fn set_oplog(&self, log: Option<crate::oplog::OpLog>) {
        if let Ok(mut slot) = self.oplog.write() {
            *slot = log;
        }
    }

    /// The attached server log, if this project has a remote.
    pub fn oplog(&self) -> Option<crate::oplog::OpLog> {
        self.oplog.read().ok().and_then(|s| s.clone())
    }

    // ------------------------------------------------------------------
    // Server log write-through.
    //
    // A logged write puts its op in `remote_outbox`, in the same SQLite
    // transaction as the change itself, and moves it to remote-log.jsonl
    // after the commit. The op used to be built and appended only after the
    // commit: a process killed in between (an MCP client closing, SIGTERM
    // at shutdown) left a write here with no op, and a lost delete was then
    // undone by `remote pull`. The commit now carries its op; the move to
    // the log is redone by the next logged write, or the next process that
    // opens this database, if it did not happen.
    // ------------------------------------------------------------------

    /// A write that was made here and whose op is not in the log: said on
    /// stderr, and kept for the stdio MCP server to put in the tool result
    /// of the call that made it, which an agent does read.
    fn not_queued(&self, what: String) {
        eprintln!("Warning: the local write succeeded but was not queued for the server: {what}");
        crate::oplog::notice(
            crate::oplog::NoticeKind::Unqueued,
            format!(
                "The write WAS made in the local database, but it was NOT queued for the shared \
                 server: {what}\nDo not repeat it (that would write it twice here). \
                 `deciduous remote status` shows what differs from the server."
            ),
        );
    }

    /// Makes the ops for `bodies` and writes them to `remote_outbox` in the
    /// caller's transaction. Refuses (so the transaction, and the write,
    /// roll back) an op holding a NUL: the server can never store it, and a
    /// write it can never receive is the fork this log exists to prevent.
    fn queue_in_tx(
        &self,
        conn: &mut SqliteConnection,
        bodies: Vec<crate::oplog::OpBody>,
    ) -> Result<Vec<Queued>> {
        self.queue_in_tx_from(conn, bodies, None)
    }

    /// [`Self::queue_in_tx`] for what reconcile applies from graph.json.
    /// `made` is the time the record carries for the change (its
    /// `updated_at`, a tombstone's time, an edge's `created_at`): the op is
    /// dated by it, not by when this machine queued it.
    ///
    /// The server orders an op against a delete by `at`: an edit revives a
    /// deleted node only when made after the delete, and a delete's `at`
    /// dates the tombstone. Stamped at queue time, an edit made before a
    /// delete that reached this machine after it counted as made after it
    /// and revived the node; and a delete that came through git dated its
    /// tombstone after edits made since, which it then refused.
    ///
    /// The record's time is capped at now: a teammate's clock running ahead
    /// cannot date a change later than the moment it was seen here, which is
    /// how every such op was dated before. A time that does not parse is
    /// read as reconcile read it (`records::parse_ts`: the epoch), so the
    /// server orders the change the way this sync did.
    fn queue_git_in_tx(
        &self,
        conn: &mut SqliteConnection,
        bodies: Vec<crate::oplog::OpBody>,
        made: chrono::DateTime<chrono::Utc>,
    ) -> Result<Vec<Queued>> {
        let at = made.min(chrono::Utc::now()).to_rfc3339();
        self.queue_in_tx_from(conn, bodies, Some((crate::oplog::GIT, at.as_str())))
    }

    /// `origin` is `git` and the time to date the op by, for a change that
    /// came through git; None for a local write, dated now.
    fn queue_in_tx_from(
        &self,
        conn: &mut SqliteConnection,
        bodies: Vec<crate::oplog::OpBody>,
        origin: Option<(&str, &str)>,
    ) -> Result<Vec<Queued>> {
        if self
            .ops_suppressed
            .load(std::sync::atomic::Ordering::SeqCst)
        {
            return Ok(Vec::new());
        }
        if origin.is_none() && self.oplog().is_none() {
            self.local_only_pending
                .store(true, std::sync::atomic::Ordering::SeqCst);
        }
        let mut out = Vec::with_capacity(bodies.len());
        for body in bodies {
            let mut op = crate::oplog::OpLog::new_op(body);
            if let Some((origin, at)) = origin {
                op.origin = Some(origin.to_string());
                op.at = at.to_string();
            }
            if let Some(at) = crate::oplog::nul_path(&op) {
                return Err(DbError::Validation(format!(
                    "nothing was written: {at} holds a NUL character, which the shared server \
                     cannot store, so this write could never reach it ({}). Remove the NUL and \
                     write again.",
                    op.body.describe().replace('\0', "\\0")
                )));
            }
            let json = serde_json::to_string(&op)
                .map_err(|e| DbError::Validation(format!("serializing op: {e}")))?;
            diesel::sql_query("INSERT INTO remote_outbox (op_json) VALUES (?)")
                .bind::<diesel::sql_types::Text, _>(&json)
                .execute(conn)?;
            let seq: i64 = diesel::select(diesel::dsl::sql::<diesel::sql_types::BigInt>(
                "last_insert_rowid()",
            ))
            .first(conn)?;
            out.push(Queued { seq, op });
        }
        Ok(out)
    }

    /// After the commit, still holding the log's lock: moves a write's ops
    /// from `remote_outbox` to the log.
    fn to_log(&self, queued: Vec<Queued>) {
        if self
            .local_only_pending
            .swap(false, std::sync::atomic::Ordering::SeqCst)
        {
            crate::oplog::note_local_only(&self.path);
        }
        if queued.is_empty() {
            return;
        }
        let Some(log) = self.oplog() else { return };
        let ops: Vec<crate::oplog::Op> = queued.iter().map(|q| q.op.clone()).collect();
        match log.append_ops(&ops) {
            Ok(()) => {
                if let Err(e) = self.forget_queued(queued.last().map(|q| q.seq).unwrap_or(0)) {
                    // Harmless: the next drain finds the op in the log.
                    eprintln!("Warning: could not clear remote_outbox after queueing: {e}");
                }
            }
            Err(e) => self.not_queued(format!(
                "{e}\nIts op is kept in this database and is added to {} by the first write \
                 after that file can be written. Until then the server will not get: {}",
                log.path().display(),
                ops.iter()
                    .map(|o| o.body.describe())
                    .collect::<Vec<_>>()
                    .join("; ")
            )),
        }
    }

    fn forget_queued(&self, up_to: i64) -> Result<()> {
        let mut conn = self.get_conn()?;
        diesel::sql_query("DELETE FROM remote_outbox WHERE seq <= ?")
            .bind::<diesel::sql_types::BigInt, _>(up_to)
            .execute(&mut conn)?;
        Ok(())
    }

    /// Moves to the log every op a write committed and did not get there
    /// (its process stopped in between, or the log could not be written),
    /// in commit order. The caller holds the log's lock. An op already in
    /// the log (the process stopped after its append) is not added twice.
    /// Returns how many were added.
    fn drain_outbox(&self, log: &crate::oplog::OpLog) -> std::result::Result<usize, String> {
        #[derive(QueryableByName)]
        struct Row {
            #[diesel(sql_type = diesel::sql_types::BigInt)]
            seq: i64,
            #[diesel(sql_type = diesel::sql_types::Text)]
            op_json: String,
        }
        let mut conn = self.get_conn().map_err(|e| e.to_string())?;
        let rows: Vec<Row> =
            diesel::sql_query("SELECT seq, op_json FROM remote_outbox ORDER BY seq")
                .load(&mut conn)
                .map_err(|e| e.to_string())?;
        drop(conn);
        let Some(last) = rows.last().map(|r| r.seq) else {
            return Ok(0);
        };
        let ops = rows
            .iter()
            .map(|r| {
                serde_json::from_str::<crate::oplog::Op>(&r.op_json).map_err(|e| {
                    format!(
                        "remote_outbox row {} is not an op ({e}): {}",
                        r.seq, r.op_json
                    )
                })
            })
            .collect::<std::result::Result<Vec<_>, _>>()?;
        let have = log.op_ids()?;
        let missing: Vec<crate::oplog::Op> = ops
            .into_iter()
            .filter(|op| !have.contains(&op.op_id))
            .collect();
        log.append_ops(&missing)?;
        self.forget_queued(last).map_err(|e| e.to_string())?;
        // What reconcile applied from graph.json reaches the log this way
        // by design (one append per reconcile); only a write made here and
        // not queued is news.
        let lost: Vec<&crate::oplog::Op> = missing.iter().filter(|o| !o.from_git()).collect();
        if !lost.is_empty() {
            eprintln!(
                "Note: {} write(s) made here earlier had not reached {} (the process that made \
                 them stopped before queueing them, or the file could not be written); they are \
                 queued now: {}",
                lost.len(),
                log.path().display(),
                lost.iter()
                    .map(|o| o.body.describe())
                    .collect::<Vec<_>>()
                    .join("; ")
            );
        }
        Ok(missing.len())
    }

    /// At open: a write whose process stopped between its commit and its
    /// append is queued now, if no one else holds the log. (Otherwise the
    /// holder, a writer, does it.)
    fn drain_outbox_at_open(&self) {
        let Some(log) = self.oplog() else { return };
        let pending = self
            .get_conn()
            .ok()
            .and_then(|mut c| {
                diesel::select(diesel::dsl::sql::<diesel::sql_types::BigInt>(
                    "(SELECT count(*) FROM remote_outbox)",
                ))
                .first::<i64>(&mut c)
                .ok()
            })
            .unwrap_or(0);
        if pending == 0 {
            return;
        }
        if let Ok(Some(_held)) = log.try_lock() {
            if let Err(e) = self.drain_outbox(&log) {
                eprintln!(
                    "Warning: writes made here earlier could not be queued for the server: {e}"
                );
            }
        }
    }

    fn node_created_body(&self, new: &NewDecisionNode) -> Option<crate::oplog::OpBody> {
        self.oplog()?;
        let metadata = new
            .metadata_json
            .and_then(|m| serde_json::from_str::<serde_json::Value>(m).ok())
            .and_then(|v| v.as_object().cloned())
            .unwrap_or_default();
        Some(crate::oplog::OpBody::CreateNode {
            change_id: new.change_id.to_string(),
            node_type: new.node_type.to_string(),
            title: new.title.to_string(),
            description: new.description.map(str::to_string),
            status: new.status.to_string(),
            metadata,
            created_at: crate::records::wire_ts(new.created_at),
            updated_at: crate::records::wire_ts(new.updated_at),
        })
    }

    /// Holds the server log's lock from before a write changes the database
    /// until its op is appended; `None` when there is no log.
    ///
    /// Without it, two local writers (two MCP servers and the CLI on one
    /// project) could commit in one order and append in the other, and an
    /// update op's `was` could be a value read before another writer's
    /// commit. The server applies an update only over its `was`, so either
    /// made it refuse an edit "changed on the server after this edit was
    /// made" when nobody had touched the server (RUST-N5). Held across the
    /// write, the log's order is the commit order.
    ///
    /// Taking it also queues what an earlier write committed and did not
    /// append, before this write's op, so the order holds for those too.
    ///
    /// If the lock cannot be had, the write is refused before anything is
    /// written: a write made and then not queued is the silent fork this
    /// log exists to prevent.
    fn hold_log(&self) -> Result<Option<crate::oplog::LockGuard>> {
        let Some(log) = self.oplog() else {
            return Ok(None);
        };
        let guard = log.lock().map_err(|e| {
            DbError::Validation(format!(
                "nothing was written: this write would have to be queued for the server in {}, \
                 and its lock could not be taken: {e}",
                log.path().display()
            ))
        })?;
        if let Err(e) = self.drain_outbox(&log) {
            eprintln!("Warning: writes made here earlier could not be queued for the server: {e}");
        }
        Ok(Some(guard))
    }

    /// Reads the node inside a write's transaction, when there is a server
    /// log to record what the write replaced; `None` otherwise.
    fn replaced_in_tx(
        &self,
        conn: &mut SqliteConnection,
        node_id: i32,
    ) -> Result<Option<DecisionNode>> {
        if self.oplog().is_none() {
            return Ok(None);
        }
        Ok(decision_nodes::table
            .filter(decision_nodes::id.eq(node_id))
            .first::<DecisionNode>(conn)
            .optional()?)
    }

    /// The op for an edit of `before` (read in the edit's transaction):
    /// the fields it sets, and for each the value it replaced.
    fn node_update_body(
        before: Option<DecisionNode>,
        set: serde_json::Map<String, serde_json::Value>,
        metadata: serde_json::Map<String, serde_json::Value>,
    ) -> Vec<crate::oplog::OpBody> {
        // No log, so nothing was read.
        let Some(before) = before else {
            return Vec::new();
        };
        let old_meta: serde_json::Map<String, serde_json::Value> = before
            .metadata_json
            .as_deref()
            .and_then(|m| serde_json::from_str::<serde_json::Value>(m).ok())
            .and_then(|v| v.as_object().cloned())
            .unwrap_or_default();
        let was = set
            .keys()
            .map(|k| {
                let v = match k.as_str() {
                    "title" => serde_json::Value::String(before.title.clone()),
                    "status" => serde_json::Value::String(before.status.clone()),
                    "description" => before
                        .description
                        .clone()
                        .map(serde_json::Value::String)
                        .unwrap_or(serde_json::Value::Null),
                    other => unreachable!("update op sets an unknown column {other}"),
                };
                (k.clone(), v)
            })
            .collect();
        let was_metadata = metadata
            .keys()
            .map(|k| {
                (
                    k.clone(),
                    old_meta.get(k).cloned().unwrap_or(serde_json::Value::Null),
                )
            })
            .collect();
        vec![crate::oplog::OpBody::UpdateNode {
            change_id: before.change_id,
            set,
            metadata,
            was,
            was_metadata,
        }]
    }

    /// The op deleting `node`, carrying what it held (see
    /// [`crate::oplog::OpBody::DeleteNode`]).
    pub(crate) fn node_deleted_body(node: &DecisionNode) -> crate::oplog::OpBody {
        let opt = |v: &Option<String>| {
            v.clone()
                .map(serde_json::Value::String)
                .unwrap_or(serde_json::Value::Null)
        };
        let mut was = serde_json::Map::new();
        was.insert(
            "title".into(),
            serde_json::Value::String(node.title.clone()),
        );
        was.insert("description".into(), opt(&node.description));
        was.insert(
            "status".into(),
            serde_json::Value::String(node.status.clone()),
        );
        crate::oplog::OpBody::DeleteNode {
            change_id: node.change_id.clone(),
            node_type: Some(node.node_type.clone()),
            was,
            was_metadata: node
                .metadata_json
                .as_deref()
                .and_then(|m| serde_json::from_str::<serde_json::Value>(m).ok())
                .and_then(|v| v.as_object().cloned())
                .unwrap_or_default(),
        }
    }

    /// The ops removing `edges`, and a description of each edge that cannot
    /// be named to the server (an endpoint with no change_id).
    fn edges_deleted_bodies(
        &self,
        edges: &[DecisionEdge],
        change_ids: &std::collections::HashMap<i32, String>,
    ) -> (Vec<crate::oplog::OpBody>, Vec<String>) {
        let mut bodies = Vec::new();
        let mut unnamed = Vec::new();
        if self.oplog().is_none() {
            return (bodies, unnamed);
        }
        for e in edges {
            let from = change_ids
                .get(&e.from_node_id)
                .cloned()
                .or_else(|| e.from_change_id.clone());
            let to = change_ids
                .get(&e.to_node_id)
                .cloned()
                .or_else(|| e.to_change_id.clone());
            match (from, to) {
                (Some(from_change_id), Some(to_change_id)) => {
                    bodies.push(crate::oplog::OpBody::DeleteEdge {
                        from_change_id,
                        to_change_id,
                        edge_type: e.edge_type.clone(),
                        deleted_at: None,
                    })
                }
                _ => unnamed.push(format!(
                    "edge {} ({} -> {}) has no change_id for an endpoint, so its removal cannot be named to the server",
                    e.id, e.from_node_id, e.to_node_id
                )),
            }
        }
        (bodies, unnamed)
    }

    /// change_ids of the endpoints of `edges`, read before they are deleted.
    fn endpoint_change_ids(
        &self,
        edges: &[DecisionEdge],
    ) -> std::collections::HashMap<i32, String> {
        let mut out = std::collections::HashMap::new();
        if self.oplog().is_none() {
            return out;
        }
        for e in edges {
            for id in [e.from_node_id, e.to_node_id] {
                if let std::collections::hash_map::Entry::Vacant(v) = out.entry(id) {
                    if let Ok(Some(n)) = self.get_node(id) {
                        v.insert(n.change_id);
                    }
                }
            }
        }
        out
    }

    /// The database file this handle opened, as an absolute path.
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// The directory holding the database: `.deciduous/` in a project,
    /// `graphs/<id>/` under the API daemon.
    pub fn data_dir(&self) -> &Path {
        self.path.parent().unwrap_or(Path::new("."))
    }

    /// Where attached documents are stored.
    pub fn documents_dir(&self) -> std::path::PathBuf {
        self.data_dir().join("documents")
    }

    /// The project directory: the one containing `.deciduous/`. `None` when
    /// the database does not sit in a `.deciduous/` directory (the API
    /// daemon, a scratch database), which has no project around it.
    pub fn project_root(&self) -> Option<&Path> {
        let dir = self.data_dir();
        (dir.file_name()? == ".deciduous")
            .then(|| dir.parent())
            .flatten()
    }

    /// Attach (or detach) the graph file that mutations are mirrored into.
    /// Detaching also stops the graph file being attached when it appears.
    pub fn set_store(&self, store: Option<RecordStore>) {
        self.auto_attach
            .store(store.is_some(), std::sync::atomic::Ordering::SeqCst);
        let store = store.map(|s| self.mirrored(s));
        if let Ok(mut slot) = self.store.write() {
            *slot = store;
        }
    }

    /// `store` with its writes recorded as what this database mirrors: each
    /// write moves the recorded digest from the file it wrote over to the
    /// file it wrote, and only if the recorded digest was the file it wrote
    /// over (compare-and-set, under the graph file's lock). A file git
    /// changed in between is not claimed as mirrored, so it is followed.
    fn mirrored(&self, store: RecordStore) -> RecordStore {
        let pool = self.pool.clone();
        store.with_flush_hook(move |before: &str, after: &str| {
            if let Ok(mut conn) = pool.get() {
                let _ = diesel::sql_query(
                    "UPDATE graph_mirror SET digest = ? WHERE id = 1 AND digest = ?",
                )
                .bind::<diesel::sql_types::Text, _>(after)
                .bind::<diesel::sql_types::Text, _>(before)
                .execute(&mut conn);
            }
        })
    }

    /// The digest of the graph.json this database mirrors, if one was ever
    /// recorded (a database from before 1.0.11 has none).
    pub(crate) fn mirror_digest(&self) -> Result<Option<String>> {
        #[derive(diesel::QueryableByName)]
        struct Row {
            #[diesel(sql_type = diesel::sql_types::Text)]
            digest: String,
        }
        let mut conn = self.get_conn()?;
        Ok(
            diesel::sql_query("SELECT digest FROM graph_mirror WHERE id = 1")
                .load::<Row>(&mut conn)?
                .pop()
                .map(|r| r.digest),
        )
    }

    pub(crate) fn set_mirror_digest(&self, digest: &str) -> Result<()> {
        let mut conn = self.get_conn()?;
        diesel::sql_query(
            "INSERT INTO graph_mirror (id, digest) VALUES (1, ?)
             ON CONFLICT (id) DO UPDATE SET digest = excluded.digest",
        )
        .bind::<diesel::sql_types::Text, _>(digest)
        .execute(&mut conn)?;
        Ok(())
    }

    /// Record that a write of `kind` (node, edge, theme, tag) with this id
    /// is in the database but not in graph.json.
    pub(crate) fn mark_unpublished(&self, kind: &str, record_id: &str) {
        let result = self.get_conn().and_then(|mut conn| {
            diesel::sql_query("INSERT OR IGNORE INTO unpublished (kind, record_id) VALUES (?, ?)")
                .bind::<diesel::sql_types::Text, _>(kind)
                .bind::<diesel::sql_types::Text, _>(record_id)
                .execute(&mut conn)
                .map_err(DbError::from)
        });
        if let Err(e) = result {
            // Not recorded means the next checkout could take it out of the
            // database: say so, loudly, rather than hope.
            eprintln!(
                "Warning: could not record that {kind} {record_id} is not in graph.json yet ({e}). \
                 Run `deciduous sync` on a branch before switching branches."
            );
        }
    }

    pub(crate) fn unpublished(&self) -> Result<std::collections::HashSet<(String, String)>> {
        #[derive(diesel::QueryableByName)]
        struct Row {
            #[diesel(sql_type = diesel::sql_types::Text)]
            kind: String,
            #[diesel(sql_type = diesel::sql_types::Text)]
            record_id: String,
        }
        let mut conn = self.get_conn()?;
        Ok(diesel::sql_query("SELECT kind, record_id FROM unpublished")
            .load::<Row>(&mut conn)?
            .into_iter()
            .map(|r| (r.kind, r.record_id))
            .collect())
    }

    pub(crate) fn clear_unpublished(&self) -> Result<()> {
        let mut conn = self.get_conn()?;
        diesel::sql_query("DELETE FROM unpublished").execute(&mut conn)?;
        Ok(())
    }

    /// One graph write, from its database commit to its line in graph.json,
    /// or one sync or follow, at a time across every process on this
    /// project. Without it a follow in one process could see another's
    /// committed row before that row reached the file, take it for one git
    /// took out, and forget it; the write then reached the file under a
    /// mirror stamp that claimed the database had it (seen in the 5-writer
    /// stress test: two acknowledged nodes missing from the database).
    /// Reentrant within a process: a write that syncs holds it once.
    pub fn hold_mirror(&self) -> Result<MirrorGuard<'_>> {
        let Some(store) = self.store() else {
            return Ok(MirrorGuard(None));
        };
        let mut hold = self.mirror_hold.lock().unwrap_or_else(|e| e.into_inner());
        if hold.0 == 0 {
            let path = store.path().with_extension("json.follow.lock");
            let file = std::fs::OpenOptions::new()
                .create(true)
                .truncate(false)
                .write(true)
                .open(&path)
                .and_then(|f| f.lock().map(|_| f))
                .map_err(|e| {
                    DbError::Validation(format!(
                        "nothing was changed: could not take the graph lock {}: {e}",
                        path.display()
                    ))
                })?;
            hold.1 = Some(file);
        }
        hold.0 += 1;
        Ok(MirrorGuard(Some(self)))
    }

    /// Run `f` with nothing queued for the shared server. For following a
    /// graph file git changed: what that does to the database is local.
    pub(crate) fn with_ops_suppressed<T>(&self, f: impl FnOnce() -> T) -> T {
        use std::sync::atomic::Ordering;
        let was = self.ops_suppressed.swap(true, Ordering::SeqCst);
        let out = f();
        self.ops_suppressed.store(was, Ordering::SeqCst);
        out
    }

    /// Take a node out of the local database because the graph file no
    /// longer has it (git took it out: a checkout, a reset, a pull). Not a
    /// deletion: nothing is queued for the server and nothing is written to
    /// the file. Its edges and tags go with it (the file has them only with
    /// it); its documents, sessions and context stay, under the id
    /// node_aliases keeps, for when a later checkout brings it back.
    pub(crate) fn forget_node_local(&self, node_id: i32) -> Result<()> {
        let mut conn = self.get_conn()?;
        // Foreign keys are enforced, and the rows this keeps (documents,
        // context, session links) reference the node: they point at an id
        // that is absent until a checkout brings the node back under it.
        // The pragma cannot change inside a transaction, so it is turned
        // off around this one, on this connection, and always back on.
        diesel::sql_query("PRAGMA foreign_keys = OFF").execute(&mut conn)?;
        let result = conn.immediate_transaction(|conn| {
            diesel::delete(
                decision_edges::table.filter(
                    decision_edges::from_node_id
                        .eq(node_id)
                        .or(decision_edges::to_node_id.eq(node_id)),
                ),
            )
            .execute(conn)?;
            diesel::delete(node_themes::table.filter(node_themes::node_id.eq(node_id)))
                .execute(conn)?;
            diesel::delete(decision_nodes::table.filter(decision_nodes::id.eq(node_id)))
                .execute(conn)?;
            Ok::<_, DbError>(())
        });
        let restored = diesel::sql_query("PRAGMA foreign_keys = ON").execute(&mut conn);
        result?;
        restored?;
        Ok(())
    }

    /// Refuse a deletion while HEAD is detached at a commit being looked
    /// at. A write there stays in the database and is published on the next
    /// branch; a deletion cannot wait like that (the commit's file still has
    /// the record, so following it would bring the record back), and
    /// writing it into that commit's file blocks `git checkout` (G7).
    fn require_branch_for_delete(&self, what: &str) -> Result<()> {
        let Some(store) = self.store() else {
            return Ok(());
        };
        if self.write_store().is_some() {
            return Ok(());
        }
        let at = crate::records::viewing_history(store.path()).unwrap_or_default();
        Err(DbError::Validation(format!(
            "nothing was changed: HEAD is detached at {at}, a commit being looked at, and a \
             deletion has to be recorded in a branch's graph.json. Check out a branch \
             (`git switch <branch>`) and {what} there."
        )))
    }

    /// Attribute records in the graph file to `author`, including a graph
    /// file that is only attached later. Setting it on the store alone was
    /// lost when a store was attached after the fact.
    pub fn set_store_author(&self, author: &str) {
        if let Ok(mut a) = self.store_author.write() {
            *a = Some(author.to_string());
        }
        if let Ok(mut slot) = self.store.write() {
            if let Some(store) = slot.take() {
                *slot = Some(store.with_author(author));
            }
        }
    }

    /// The attached graph file, if sync is enabled for this database. If
    /// none is attached yet but one now exists next to the database, it is
    /// attached here: one `stat` per write while there is none.
    pub fn store(&self) -> Option<RecordStore> {
        if let Some(store) = self.store.read().ok().and_then(|s| s.clone()) {
            return Some(store);
        }
        if !self.auto_attach.load(std::sync::atomic::Ordering::SeqCst) {
            return None;
        }
        let found = RecordStore::path_for_db(&self.path).and_then(RecordStore::open)?;
        let found = self.mirrored(found);
        let mut slot = self.store.write().ok()?;
        if slot.is_none() {
            let author = self.store_author.read().ok().and_then(|a| a.clone());
            *slot = Some(match author {
                Some(a) => found.with_author(a),
                None => found,
            });
        }
        slot.clone()
    }

    /// The graph file that writes are mirrored into: [`Self::store`], except
    /// while HEAD is detached at a commit being looked at (G7), where the
    /// file is that commit's and writing a new node into it made
    /// `git checkout main` fail with "Your local changes ... would be
    /// overwritten". Such a write stays in the database, and the next sync
    /// on a branch exports it, as `deciduous sync` there already says.
    ///
    /// On a branch this costs one read of HEAD; git runs only when it is
    /// detached, so a rebase (detached too, and whose file is being
    /// merged on purpose) still writes through.
    fn write_store(&self) -> Option<RecordStore> {
        let store = self.store()?;
        let head = self.head_file.get_or_init(|| {
            let dir = store.path().parent()?;
            let out = std::process::Command::new("git")
                .arg("-C")
                .arg(dir)
                .args(["rev-parse", "--path-format=absolute", "--git-path", "HEAD"])
                .stderr(std::process::Stdio::null())
                .output()
                .ok()?;
            out.status
                .success()
                .then(|| std::path::PathBuf::from(String::from_utf8_lossy(&out.stdout).trim()))
        });
        let Some(head) = head else {
            return Some(store);
        };
        match std::fs::read_to_string(head) {
            Ok(h) if h.starts_with("ref:") => Some(store),
            _ if crate::records::viewing_history(store.path()).is_some() => None,
            _ => Some(store),
        }
    }

    // ------------------------------------------------------------------
    // Record store write-through. A file that cannot be read refuses the
    // operation before the database is touched (require_readable_store).
    // Any other failure does not fail it: the database write already
    // happened, and the next `deciduous sync` will export what is missing. They are reported on stderr so MCP/stdio
    // transports stay clean.
    // ------------------------------------------------------------------

    /// Refuse a graph mutation while the attached graph file cannot be read
    /// (conflict markers, a bad hand edit). The write-through below would
    /// fail with only a warning, the database would be ahead of the file,
    /// and the sync that repairs the file would then compare the two
    /// without this write ever having been stamped against the file's
    /// versions: a record stamped later than this clock wins, and the edit
    /// the CLI reported as done is reverted.
    fn require_readable_store(&self) -> Result<MirrorGuard<'_>> {
        let guard = self.hold_mirror()?;
        let Some(store) = self.write_store() else {
            return Ok(guard);
        };
        store
            .read_doc_all()
            .map(|_| guard)
            .map_err(|e| DbError::Validation(format!("nothing was changed: {}", e)))
    }

    fn store_warn(what: &str, e: impl std::fmt::Display) {
        eprintln!("Warning: graph file: {} ({})", what, e);
    }

    /// A write graph.json did not get (HEAD detached at a commit being
    /// looked at, or the file write failed): record it, so following the
    /// file does not take it out of the database, and the next sync on a
    /// branch publishes it.
    fn withheld(&self, kind: &str, record_id: &str) {
        if self.store().is_some() {
            self.mark_unpublished(kind, record_id);
        }
    }

    fn edge_key(edge: &DecisionEdge) -> Option<String> {
        match (&edge.from_change_id, &edge.to_change_id) {
            (Some(f), Some(t)) => Some(crate::records::edge_id(f, t, &edge.edge_type)),
            _ => None,
        }
    }

    fn publish_node_by_id(&self, node_id: i32) {
        self.publish_node_edit(node_id, None);
    }

    /// Publish a node this process just changed. `before` is the row as it
    /// was before the change, so the file can keep fields a teammate changed
    /// that this write did not touch.
    fn publish_node_edit(&self, node_id: i32, before: Option<DecisionNode>) {
        let Some(store) = self.write_store() else {
            if let Ok(Some(node)) = self.get_node(node_id) {
                self.withheld("node", &node.change_id);
            }
            return;
        };
        match self.get_node(node_id) {
            Ok(Some(node)) => match store.publish_node_edit(before.as_ref(), &node) {
                Ok((_, Some(stamp))) => {
                    // The file held a version stamped after this clock; the
                    // write went in just after it, and the row must agree or
                    // the next sync would re-import the file's copy.
                    if let Err(e) = self.set_node_updated_at(node_id, &stamp) {
                        Self::store_warn("could not restamp node", e);
                    }
                }
                Ok((_, None)) => {}
                Err(e) => {
                    Self::store_warn("could not write node record", e);
                    self.withheld("node", &node.change_id);
                }
            },
            Ok(None) => {}
            Err(e) => Self::store_warn("could not read node for publishing", e),
        }
    }

    /// The row as it is now, for [`Self::publish_node_edit`]. Only read when
    /// a graph file is attached.
    fn node_before_edit(&self, node_id: i32) -> Option<DecisionNode> {
        self.write_store()?;
        self.get_node(node_id).ok().flatten()
    }

    fn set_node_updated_at(&self, node_id: i32, updated_at: &str) -> Result<()> {
        let mut conn = self.get_conn()?;
        diesel::update(decision_nodes::table.filter(decision_nodes::id.eq(node_id)))
            .set(decision_nodes::updated_at.eq(updated_at))
            .execute(&mut conn)?;
        Ok(())
    }

    fn publish_edge_by_id(&self, edge_id: i32) {
        let Some(store) = self.write_store() else {
            if let Some(key) = self
                .get_edge(edge_id)
                .ok()
                .flatten()
                .as_ref()
                .and_then(Self::edge_key)
            {
                self.withheld("edge", &key);
            }
            return;
        };
        match self.get_edge(edge_id) {
            Ok(Some(edge)) => {
                if let Err(e) = store.publish_edge(&edge) {
                    Self::store_warn("could not write edge record", e);
                    if let Some(key) = Self::edge_key(&edge) {
                        self.withheld("edge", &key);
                    }
                }
            }
            Ok(None) => {}
            Err(e) => Self::store_warn("could not read edge for publishing", e),
        }
    }

    fn tombstone_edges(&self, edges: &[DecisionEdge]) {
        self.tombstone_edges_of(edges, None);
    }

    /// Tombstone edges; `with_node` names the node whose deletion took
    /// them, so they come back if that node does.
    fn tombstone_edges_of(&self, edges: &[DecisionEdge], with_node: Option<&str>) {
        let Some(store) = self.write_store() else {
            return;
        };
        for edge in edges {
            if let Err(e) = store.tombstone_edge_of(edge, with_node) {
                Self::store_warn("could not write edge tombstone", e);
            }
        }
    }

    fn publish_theme_by_id(&self, theme_id: i32) {
        let Some(store) = self.write_store() else {
            if let Ok(Some(theme)) = self.get_theme_by_id(theme_id) {
                self.withheld("theme", &theme.change_id);
            }
            return;
        };
        match self.get_theme_by_id(theme_id) {
            Ok(Some(theme)) => match store.publish_theme_edit(&theme) {
                Ok((_, Some(stamp))) => {
                    let result = self.get_conn().and_then(|mut conn| {
                        diesel::update(themes::table.filter(themes::id.eq(theme_id)))
                            .set(themes::updated_at.eq(&stamp))
                            .execute(&mut conn)
                            .map_err(DbError::from)
                    });
                    if let Err(e) = result {
                        Self::store_warn("could not restamp theme", e);
                    }
                }
                Ok((_, None)) => {}
                Err(e) => {
                    Self::store_warn("could not write theme record", e);
                    self.withheld("theme", &theme.change_id);
                }
            },
            Ok(None) => {}
            Err(e) => Self::store_warn("could not read theme for publishing", e),
        }
    }

    fn publish_tag(&self, node_id: i32, theme_id: i32) {
        let node = self.get_node(node_id).ok().flatten();
        let theme = self.get_theme_by_id(theme_id).ok().flatten();
        let key = match (&node, &theme) {
            (Some(n), Some(t)) => Some(crate::records::tag_id(&n.change_id, &t.change_id)),
            _ => None,
        };
        let Some(store) = self.write_store() else {
            if let Some(key) = &key {
                self.withheld("tag", key);
            }
            return;
        };
        let tag = self.get_tag(node_id, theme_id).ok().flatten();
        if let (Some(node), Some(theme), Some(tag)) = (node, theme, tag) {
            if let Err(e) = store.publish_tag(
                &node.change_id,
                &theme.change_id,
                &tag.source,
                &tag.created_at,
            ) {
                Self::store_warn("could not write tag record", e);
                if let Some(key) = &key {
                    self.withheld("tag", key);
                }
            }
        }
    }

    fn tombstone_tag(&self, node_change_id: &str, theme_change_id: &str) {
        let Some(store) = self.write_store() else {
            return;
        };
        if let Err(e) = store.tombstone_tag(node_change_id, theme_change_id, false) {
            Self::store_warn("could not write tag tombstone", e);
        }
    }

    /// Raw SQL migration that runs before Diesel ORM is used
    fn migrate_add_change_ids_raw(&self) -> Result<bool> {
        let mut conn = self.get_conn()?;

        // Check if decision_nodes table exists
        let tables: Vec<TableInfo> = diesel::sql_query(
            "SELECT name FROM sqlite_master WHERE type='table' AND name='decision_nodes'",
        )
        .load::<TableInfo>(&mut conn)
        .unwrap_or_default();

        if tables.is_empty() {
            return Ok(false); // Table doesn't exist yet, init_schema will create it
        }

        // Check if change_id column exists in decision_nodes
        let columns: Vec<PragmaTableInfo> = diesel::sql_query("PRAGMA table_info(decision_nodes)")
            .load(&mut conn)
            .unwrap_or_default();

        let has_change_id = columns.iter().any(|c| c.name == "change_id");

        if !has_change_id {
            // Add change_id column to decision_nodes
            diesel::sql_query("ALTER TABLE decision_nodes ADD COLUMN change_id TEXT")
                .execute(&mut conn)?;
        }

        // Always backfill any NULL change_ids (handles both new columns and stragglers)
        let nodes: Vec<NodeIdOnly> =
            diesel::sql_query("SELECT id FROM decision_nodes WHERE change_id IS NULL")
                .load(&mut conn)
                .unwrap_or_default();

        if nodes.is_empty() && has_change_id {
            return Ok(false); // Already fully migrated
        }

        for node in nodes {
            let change_id = Uuid::new_v4().to_string();
            diesel::sql_query(format!(
                "UPDATE decision_nodes SET change_id = '{}' WHERE id = {}",
                change_id, node.id
            ))
            .execute(&mut conn)?;
        }

        // Check if edge columns need migration
        let edge_columns: Vec<PragmaTableInfo> =
            diesel::sql_query("PRAGMA table_info(decision_edges)")
                .load(&mut conn)
                .unwrap_or_default();

        let has_from_change_id = edge_columns.iter().any(|c| c.name == "from_change_id");

        if !has_from_change_id {
            diesel::sql_query("ALTER TABLE decision_edges ADD COLUMN from_change_id TEXT")
                .execute(&mut conn)?;
            diesel::sql_query("ALTER TABLE decision_edges ADD COLUMN to_change_id TEXT")
                .execute(&mut conn)?;

            // Backfill edge change_ids
            diesel::sql_query(
                "UPDATE decision_edges SET
                    from_change_id = (SELECT change_id FROM decision_nodes WHERE id = decision_edges.from_node_id),
                    to_change_id = (SELECT change_id FROM decision_nodes WHERE id = decision_edges.to_node_id)"
            )
            .execute(&mut conn)?;
        }

        Ok(true) // Migration performed
    }

    /// Write a consistent, self-contained copy of the database to `dest`,
    /// including writes still sitting in the WAL. Refuses to overwrite.
    pub fn backup_to(&self, dest: &Path) -> Result<()> {
        if dest.exists() {
            return Err(DbError::Validation(format!(
                "{} already exists; refusing to overwrite it",
                dest.display()
            )));
        }
        let target = dest.to_str().ok_or_else(|| {
            DbError::Validation(format!("{} is not a UTF-8 path", dest.display()))
        })?;
        let mut conn = self.get_conn()?;
        diesel::sql_query("VACUUM INTO ?")
            .bind::<diesel::sql_types::Text, _>(target)
            .execute(&mut conn)?;
        Ok(())
    }

    fn get_conn(&self) -> Result<DbConn> {
        self.pool
            .get()
            .map_err(|e| DbError::Connection(e.to_string()))
    }

    fn init_schema(&self) -> Result<()> {
        let mut conn = self.get_conn()?;

        // Ops of writes on their way to remote-log.jsonl: written in the
        // write's own transaction, so a write is never committed without its
        // op (see `Database::queue_in_tx`). Empty between writes.
        diesel::sql_query(
            "CREATE TABLE IF NOT EXISTS remote_outbox (
                seq INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
                op_json TEXT NOT NULL
            )",
        )
        .execute(&mut conn)?;

        // The server's copy of each node as of the last `remote pull`: what
        // `remote push --repair` checks before it sends this copy's fields
        // over the server's. See `Database::remote_base`.
        diesel::sql_query(
            "CREATE TABLE IF NOT EXISTS remote_base (
                change_id TEXT PRIMARY KEY NOT NULL,
                node_json TEXT NOT NULL
            )",
        )
        .execute(&mut conn)?;

        // Run raw SQL to create tables if they don't exist
        diesel::sql_query(
            r#"
            CREATE TABLE IF NOT EXISTS schema_versions (
                id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
                version TEXT NOT NULL UNIQUE,
                name TEXT NOT NULL,
                features TEXT NOT NULL,
                introduced_at TEXT NOT NULL
            )
        "#,
        )
        .execute(&mut conn)?;

        diesel::sql_query(
            r#"
            CREATE TABLE IF NOT EXISTS decision_nodes (
                id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
                change_id TEXT NOT NULL UNIQUE,
                node_type TEXT NOT NULL,
                title TEXT NOT NULL,
                description TEXT,
                status TEXT NOT NULL DEFAULT 'pending',
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                metadata_json TEXT
            )
        "#,
        )
        .execute(&mut conn)?;

        diesel::sql_query(
            r#"
            CREATE TABLE IF NOT EXISTS decision_edges (
                id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
                from_node_id INTEGER NOT NULL,
                to_node_id INTEGER NOT NULL,
                from_change_id TEXT,
                to_change_id TEXT,
                edge_type TEXT NOT NULL,
                weight REAL DEFAULT 1.0,
                rationale TEXT,
                created_at TEXT NOT NULL,
                FOREIGN KEY (from_node_id) REFERENCES decision_nodes(id),
                FOREIGN KEY (to_node_id) REFERENCES decision_nodes(id),
                UNIQUE(from_node_id, to_node_id, edge_type)
            )
        "#,
        )
        .execute(&mut conn)?;

        diesel::sql_query(
            r#"
            CREATE TABLE IF NOT EXISTS decision_context (
                id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
                node_id INTEGER NOT NULL,
                context_type TEXT NOT NULL,
                content_json TEXT NOT NULL,
                captured_at TEXT NOT NULL,
                FOREIGN KEY (node_id) REFERENCES decision_nodes(id)
            )
        "#,
        )
        .execute(&mut conn)?;

        diesel::sql_query(
            r#"
            CREATE TABLE IF NOT EXISTS decision_sessions (
                id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
                name TEXT,
                started_at TEXT NOT NULL,
                ended_at TEXT,
                root_node_id INTEGER,
                summary TEXT,
                FOREIGN KEY (root_node_id) REFERENCES decision_nodes(id)
            )
        "#,
        )
        .execute(&mut conn)?;

        diesel::sql_query(
            r#"
            CREATE TABLE IF NOT EXISTS session_nodes (
                session_id INTEGER NOT NULL,
                node_id INTEGER NOT NULL,
                added_at TEXT NOT NULL,
                PRIMARY KEY (session_id, node_id),
                FOREIGN KEY (session_id) REFERENCES decision_sessions(id),
                FOREIGN KEY (node_id) REFERENCES decision_nodes(id)
            )
        "#,
        )
        .execute(&mut conn)?;

        diesel::sql_query(
            r#"
            CREATE TABLE IF NOT EXISTS command_log (
                id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
                command TEXT NOT NULL,
                description TEXT,
                working_dir TEXT,
                exit_code INTEGER,
                stdout TEXT,
                stderr TEXT,
                started_at TEXT NOT NULL,
                completed_at TEXT,
                duration_ms INTEGER,
                decision_node_id INTEGER,
                FOREIGN KEY (decision_node_id) REFERENCES decision_nodes(id)
            )
        "#,
        )
        .execute(&mut conn)?;

        // Roadmap Board Tables
        diesel::sql_query(
            r#"
            CREATE TABLE IF NOT EXISTS roadmap_items (
                id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
                change_id TEXT NOT NULL UNIQUE,
                title TEXT NOT NULL,
                description TEXT,
                section TEXT,
                parent_id INTEGER,
                checkbox_state TEXT NOT NULL DEFAULT 'none',
                github_issue_number INTEGER,
                github_issue_state TEXT,
                outcome_node_id INTEGER,
                outcome_change_id TEXT,
                markdown_line_start INTEGER,
                markdown_line_end INTEGER,
                content_hash TEXT,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                last_synced_at TEXT,
                FOREIGN KEY (parent_id) REFERENCES roadmap_items(id),
                FOREIGN KEY (outcome_node_id) REFERENCES decision_nodes(id)
            )
        "#,
        )
        .execute(&mut conn)?;

        diesel::sql_query(
            r#"
            CREATE TABLE IF NOT EXISTS roadmap_sync_state (
                id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
                roadmap_path TEXT NOT NULL,
                roadmap_content_hash TEXT,
                github_repo TEXT,
                last_github_sync TEXT,
                last_markdown_parse TEXT,
                conflict_count INTEGER NOT NULL DEFAULT 0
            )
        "#,
        )
        .execute(&mut conn)?;

        diesel::sql_query(
            r#"
            CREATE TABLE IF NOT EXISTS roadmap_conflicts (
                id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
                item_change_id TEXT NOT NULL,
                conflict_type TEXT NOT NULL,
                local_value TEXT,
                remote_value TEXT,
                resolution TEXT,
                detected_at TEXT NOT NULL,
                resolved_at TEXT,
                FOREIGN KEY (item_change_id) REFERENCES roadmap_items(change_id)
            )
        "#,
        )
        .execute(&mut conn)?;

        // GitHub issue cache for Web display
        diesel::sql_query(
            r#"
            CREATE TABLE IF NOT EXISTS github_issue_cache (
                id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
                issue_number INTEGER NOT NULL,
                repo TEXT NOT NULL,
                title TEXT NOT NULL,
                body TEXT,
                state TEXT NOT NULL,
                html_url TEXT NOT NULL,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                cached_at TEXT NOT NULL,
                UNIQUE(repo, issue_number)
            )
        "#,
        )
        .execute(&mut conn)?;

        // Q&A interactions for chat history
        diesel::sql_query(
            r#"
            CREATE TABLE IF NOT EXISTS qa_interactions (
                id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
                user_prompt TEXT NOT NULL,
                total_prompt TEXT NOT NULL,
                response TEXT NOT NULL,
                context_json TEXT,
                inserted_at TEXT NOT NULL,
                deleted_at TEXT
            )
        "#,
        )
        .execute(&mut conn)?;

        // FTS5 virtual table for full-text search on Q&A interactions
        diesel::sql_query(
            r#"
            CREATE VIRTUAL TABLE IF NOT EXISTS qa_interactions_fts USING fts5(
                user_prompt,
                response,
                content='qa_interactions',
                content_rowid='id'
            )
        "#,
        )
        .execute(&mut conn)?;

        // Triggers to keep FTS index in sync
        diesel::sql_query(
            r#"
            CREATE TRIGGER IF NOT EXISTS qa_interactions_ai AFTER INSERT ON qa_interactions BEGIN
                INSERT INTO qa_interactions_fts(rowid, user_prompt, response)
                VALUES (new.id, new.user_prompt, new.response);
            END
        "#,
        )
        .execute(&mut conn)?;

        diesel::sql_query(
            r#"
            CREATE TRIGGER IF NOT EXISTS qa_interactions_ad AFTER DELETE ON qa_interactions BEGIN
                INSERT INTO qa_interactions_fts(qa_interactions_fts, rowid, user_prompt, response)
                VALUES ('delete', old.id, old.user_prompt, old.response);
            END
        "#,
        )
        .execute(&mut conn)?;

        diesel::sql_query(
            r#"
            CREATE TRIGGER IF NOT EXISTS qa_interactions_au AFTER UPDATE ON qa_interactions BEGIN
                INSERT INTO qa_interactions_fts(qa_interactions_fts, rowid, user_prompt, response)
                VALUES ('delete', old.id, old.user_prompt, old.response);
                INSERT INTO qa_interactions_fts(rowid, user_prompt, response)
                VALUES (new.id, new.user_prompt, new.response);
            END
        "#,
        )
        .execute(&mut conn)?;

        // Create indexes
        diesel::sql_query("CREATE INDEX IF NOT EXISTS idx_nodes_type ON decision_nodes(node_type)")
            .execute(&mut conn)?;
        diesel::sql_query("CREATE INDEX IF NOT EXISTS idx_nodes_status ON decision_nodes(status)")
            .execute(&mut conn)?;
        diesel::sql_query(
            "CREATE INDEX IF NOT EXISTS idx_nodes_change_id ON decision_nodes(change_id)",
        )
        .execute(&mut conn)?;
        diesel::sql_query(
            "CREATE INDEX IF NOT EXISTS idx_edges_from ON decision_edges(from_node_id)",
        )
        .execute(&mut conn)?;
        diesel::sql_query("CREATE INDEX IF NOT EXISTS idx_edges_to ON decision_edges(to_node_id)")
            .execute(&mut conn)?;
        diesel::sql_query(
            "CREATE INDEX IF NOT EXISTS idx_edges_from_change ON decision_edges(from_change_id)",
        )
        .execute(&mut conn)?;
        diesel::sql_query(
            "CREATE INDEX IF NOT EXISTS idx_edges_to_change ON decision_edges(to_change_id)",
        )
        .execute(&mut conn)?;
        diesel::sql_query(
            "CREATE INDEX IF NOT EXISTS idx_command_started_at ON command_log(started_at)",
        )
        .execute(&mut conn)?;

        // Roadmap indexes
        diesel::sql_query(
            "CREATE INDEX IF NOT EXISTS idx_roadmap_items_change_id ON roadmap_items(change_id)",
        )
        .execute(&mut conn)?;
        diesel::sql_query(
            "CREATE INDEX IF NOT EXISTS idx_roadmap_items_section ON roadmap_items(section)",
        )
        .execute(&mut conn)?;
        diesel::sql_query("CREATE INDEX IF NOT EXISTS idx_roadmap_items_github_issue ON roadmap_items(github_issue_number)").execute(&mut conn)?;
        diesel::sql_query("CREATE INDEX IF NOT EXISTS idx_roadmap_items_outcome ON roadmap_items(outcome_change_id)").execute(&mut conn)?;
        diesel::sql_query("CREATE INDEX IF NOT EXISTS idx_roadmap_conflicts_item ON roadmap_conflicts(item_change_id)").execute(&mut conn)?;
        diesel::sql_query("CREATE INDEX IF NOT EXISTS idx_github_issue_cache_repo ON github_issue_cache(repo, issue_number)").execute(&mut conn)?;

        // Q&A indexes
        diesel::sql_query("CREATE INDEX IF NOT EXISTS idx_qa_interactions_inserted_at ON qa_interactions(inserted_at)").execute(&mut conn)?;

        // Node Documents table
        diesel::sql_query(
            r#"
            CREATE TABLE IF NOT EXISTS node_documents (
                id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
                change_id TEXT NOT NULL UNIQUE,
                node_id INTEGER NOT NULL,
                node_change_id TEXT NOT NULL,
                content_hash TEXT NOT NULL,
                original_filename TEXT NOT NULL,
                storage_filename TEXT NOT NULL,
                mime_type TEXT NOT NULL,
                file_size INTEGER NOT NULL,
                description TEXT,
                description_source TEXT NOT NULL DEFAULT 'none',
                attached_at TEXT NOT NULL,
                attached_by TEXT,
                detached_at TEXT,
                FOREIGN KEY (node_id) REFERENCES decision_nodes(id)
            )
        "#,
        )
        .execute(&mut conn)?;

        // Document indexes
        diesel::sql_query("CREATE INDEX IF NOT EXISTS idx_docs_node_id ON node_documents(node_id)")
            .execute(&mut conn)?;
        diesel::sql_query(
            "CREATE INDEX IF NOT EXISTS idx_docs_content_hash ON node_documents(content_hash)",
        )
        .execute(&mut conn)?;
        diesel::sql_query(
            "CREATE INDEX IF NOT EXISTS idx_docs_change_id ON node_documents(change_id)",
        )
        .execute(&mut conn)?;

        // The graph file this database mirrors: the digest of graph.json as
        // deciduous last wrote or followed it. git never updates it, so a
        // checkout, reset or pull shows up as a digest that differs, and the
        // database follows the file (records::follow_graph_file).
        diesel::sql_query(
            "CREATE TABLE IF NOT EXISTS graph_mirror (
                id INTEGER PRIMARY KEY CHECK (id = 1),
                digest TEXT NOT NULL
            )",
        )
        .execute(&mut conn)?;

        // Local writes that are not in graph.json yet (made while HEAD was
        // detached, or whose write to the file failed). Following the file
        // keeps these and the next sync on a branch publishes them; every
        // other row the file does not have is one git took out of it.
        diesel::sql_query(
            "CREATE TABLE IF NOT EXISTS unpublished (
                kind TEXT NOT NULL,
                record_id TEXT NOT NULL,
                PRIMARY KEY (kind, record_id)
            )",
        )
        .execute(&mut conn)?;

        // A node's local integer id, kept after the node leaves the
        // database: a node that leaves on `git checkout` and returns on the
        // next one comes back under the id people typed, with the
        // documents, sessions and context that point at that id.
        diesel::sql_query(
            "CREATE TABLE IF NOT EXISTS node_aliases (
                change_id TEXT PRIMARY KEY NOT NULL,
                node_id INTEGER NOT NULL UNIQUE
            )",
        )
        .execute(&mut conn)?;
        diesel::sql_query(
            "INSERT OR IGNORE INTO node_aliases (change_id, node_id)
             SELECT change_id, id FROM decision_nodes",
        )
        .execute(&mut conn)?;
        diesel::sql_query(
            "CREATE TRIGGER IF NOT EXISTS node_alias_on_insert
             AFTER INSERT ON decision_nodes
             BEGIN
                 INSERT OR IGNORE INTO node_aliases (change_id, node_id)
                 VALUES (NEW.change_id, NEW.id);
             END",
        )
        .execute(&mut conn)?;

        // Themes table
        diesel::sql_query(
            r#"
            CREATE TABLE IF NOT EXISTS themes (
                id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
                change_id TEXT NOT NULL UNIQUE,
                name TEXT NOT NULL UNIQUE,
                color TEXT NOT NULL DEFAULT '#6b7280',
                description TEXT,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL
            )
        "#,
        )
        .execute(&mut conn)?;

        // Node-Themes junction table
        diesel::sql_query(
            r#"
            CREATE TABLE IF NOT EXISTS node_themes (
                node_id INTEGER NOT NULL,
                theme_id INTEGER NOT NULL,
                source TEXT NOT NULL DEFAULT 'manual',
                created_at TEXT NOT NULL,
                PRIMARY KEY (node_id, theme_id),
                FOREIGN KEY (node_id) REFERENCES decision_nodes(id) ON DELETE CASCADE,
                FOREIGN KEY (theme_id) REFERENCES themes(id) ON DELETE CASCADE
            )
        "#,
        )
        .execute(&mut conn)?;

        // Theme indexes
        diesel::sql_query("CREATE INDEX IF NOT EXISTS idx_themes_name ON themes(name)")
            .execute(&mut conn)?;
        diesel::sql_query("CREATE INDEX IF NOT EXISTS idx_themes_change_id ON themes(change_id)")
            .execute(&mut conn)?;
        diesel::sql_query(
            "CREATE INDEX IF NOT EXISTS idx_node_themes_node ON node_themes(node_id)",
        )
        .execute(&mut conn)?;
        diesel::sql_query(
            "CREATE INDEX IF NOT EXISTS idx_node_themes_theme ON node_themes(theme_id)",
        )
        .execute(&mut conn)?;

        // The agent message board (src/board.rs). Local coordination only:
        // nothing in records.rs or export.rs reads it, so it never reaches
        // graph.json, docs/graph-data.json or the server log.
        crate::board::create_table(&mut conn)?;

        // Register current schema
        self.register_schema(&CURRENT_SCHEMA)?;
        Ok(())
    }

    fn register_schema(&self, schema: &DecisionSchema) -> Result<()> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();
        let features_json = serde_json::to_string(&schema.features).unwrap_or_default();

        let new_schema = NewSchemaVersion {
            version: &schema.version_string(),
            name: schema.name,
            features: &features_json,
            introduced_at: &now,
        };

        diesel::insert_or_ignore_into(schema_versions::table)
            .values(&new_schema)
            .execute(&mut conn)?;

        Ok(())
    }

    /// Migrate existing database to add change_id columns if missing
    pub fn migrate_add_change_ids(&self) -> Result<bool> {
        let mut conn = self.get_conn()?;

        // Check if change_id column exists in decision_nodes
        let columns: Vec<(String,)> = diesel::sql_query("PRAGMA table_info(decision_nodes)")
            .load::<PragmaTableInfo>(&mut conn)
            .map(|rows| rows.into_iter().map(|r| (r.name,)).collect())
            .unwrap_or_default();

        let has_change_id = columns.iter().any(|(name,)| name == "change_id");

        if has_change_id {
            return Ok(false); // Already migrated
        }

        // Add change_id column to decision_nodes
        diesel::sql_query("ALTER TABLE decision_nodes ADD COLUMN change_id TEXT")
            .execute(&mut conn)?;

        // Backfill change_id with UUIDs for existing nodes
        let nodes: Vec<(i32,)> =
            diesel::sql_query("SELECT id FROM decision_nodes WHERE change_id IS NULL")
                .load::<NodeIdOnly>(&mut conn)
                .map(|rows| rows.into_iter().map(|r| (r.id,)).collect())
                .unwrap_or_default();

        for (node_id,) in nodes {
            let change_id = Uuid::new_v4().to_string();
            diesel::sql_query(format!(
                "UPDATE decision_nodes SET change_id = '{}' WHERE id = {}",
                change_id, node_id
            ))
            .execute(&mut conn)?;
        }

        // Create unique index on change_id
        diesel::sql_query("CREATE UNIQUE INDEX IF NOT EXISTS idx_nodes_change_id_unique ON decision_nodes(change_id)")
            .execute(&mut conn)?;

        // Add from_change_id and to_change_id columns to decision_edges
        let edge_columns: Vec<(String,)> = diesel::sql_query("PRAGMA table_info(decision_edges)")
            .load::<PragmaTableInfo>(&mut conn)
            .map(|rows| rows.into_iter().map(|r| (r.name,)).collect())
            .unwrap_or_default();

        let has_from_change_id = edge_columns.iter().any(|(name,)| name == "from_change_id");

        if !has_from_change_id {
            diesel::sql_query("ALTER TABLE decision_edges ADD COLUMN from_change_id TEXT")
                .execute(&mut conn)?;
            diesel::sql_query("ALTER TABLE decision_edges ADD COLUMN to_change_id TEXT")
                .execute(&mut conn)?;

            // Backfill edge change_ids from node change_ids
            diesel::sql_query(
                "UPDATE decision_edges SET
                    from_change_id = (SELECT change_id FROM decision_nodes WHERE id = decision_edges.from_node_id),
                    to_change_id = (SELECT change_id FROM decision_nodes WHERE id = decision_edges.to_node_id)"
            )
            .execute(&mut conn)?;

            // Create indexes
            diesel::sql_query("CREATE INDEX IF NOT EXISTS idx_edges_from_change ON decision_edges(from_change_id)")
                .execute(&mut conn)?;
            diesel::sql_query(
                "CREATE INDEX IF NOT EXISTS idx_edges_to_change ON decision_edges(to_change_id)",
            )
            .execute(&mut conn)?;
        }

        Ok(true) // Migration performed
    }

    // ========================================================================
    // Decision Graph Operations
    // ========================================================================

    /// Create a new decision node
    pub fn create_node(
        &self,
        node_type: &str,
        title: &str,
        description: Option<&str>,
        confidence: Option<u8>,
        commit: Option<&str>,
    ) -> Result<i32> {
        self.create_node_full(
            node_type,
            title,
            description,
            confidence,
            commit,
            None,
            None,
            None,
            None,
        )
    }

    /// Create a decision node with full metadata (including prompt and files)
    pub fn create_node_full(
        &self,
        node_type: &str,
        title: &str,
        description: Option<&str>,
        confidence: Option<u8>,
        commit: Option<&str>,
        prompt: Option<&str>,
        files: Option<&str>,
        branch: Option<&str>,
        created_at: Option<&str>,
    ) -> Result<i32> {
        one_of("node type", node_type, NODE_TYPES)?;
        if title.trim().is_empty() {
            return Err(DbError::Validation(
                "a node needs a non-empty title".to_string(),
            ));
        }
        let _mirror = self.require_readable_store()?;
        let _log = self.hold_log()?;
        let mut conn = self.get_conn()?;
        let now = created_at
            .map(|s| s.to_string())
            .unwrap_or_else(|| chrono::Local::now().to_rfc3339());
        let change_id = Uuid::new_v4().to_string();

        // Build metadata JSON with all optional fields
        let metadata = build_metadata_json(confidence, commit, prompt, files, branch);

        let new_node = NewDecisionNode {
            change_id: &change_id,
            node_type,
            title,
            description,
            status: "pending",
            created_at: &now,
            updated_at: &now,
            metadata_json: metadata.as_deref(),
        };

        // The row and its op commit together; see `queue_in_tx`.
        let body = self.node_created_body(&new_node);
        let (id, queued) = conn.immediate_transaction(|conn| {
            diesel::insert_into(decision_nodes::table)
                .values(&new_node)
                .execute(conn)?;
            let id: i32 = diesel::select(diesel::dsl::sql::<diesel::sql_types::Integer>(
                "last_insert_rowid()",
            ))
            .first(conn)?;
            let queued = self.queue_in_tx(conn, body.into_iter().collect())?;
            Ok::<_, DbError>((id, queued))
        })?;
        drop(conn);

        self.publish_node_by_id(id);
        self.to_log(queued);
        Ok(id)
    }

    /// Add a node (alias for create_node for doc examples)
    pub fn add_node(
        &self,
        node_type: &str,
        title: &str,
        description: Option<&str>,
        confidence: Option<u8>,
        commit: Option<&str>,
    ) -> Result<i32> {
        self.create_node(node_type, title, description, confidence, commit)
    }

    /// Create a node with a specific change_id (for patch application)
    pub fn create_node_with_change_id(
        &self,
        change_id: &str,
        node_type: &str,
        title: &str,
        description: Option<&str>,
        confidence: Option<u8>,
        commit: Option<&str>,
        prompt: Option<&str>,
        files: Option<&str>,
        branch: Option<&str>,
    ) -> Result<i32> {
        let _mirror = self.require_readable_store()?;
        let _log = self.hold_log()?;
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        // Build metadata JSON with all optional fields
        let metadata = build_metadata_json(confidence, commit, prompt, files, branch);

        let new_node = NewDecisionNode {
            change_id,
            node_type,
            title,
            description,
            status: "pending",
            created_at: &now,
            updated_at: &now,
            metadata_json: metadata.as_deref(),
        };

        // The row and its op commit together; see `queue_in_tx`.
        let body = self.node_created_body(&new_node);
        let (id, queued) = conn.immediate_transaction(|conn| {
            diesel::insert_into(decision_nodes::table)
                .values(&new_node)
                .execute(conn)?;
            let id: i32 = diesel::select(diesel::dsl::sql::<diesel::sql_types::Integer>(
                "last_insert_rowid()",
            ))
            .first(conn)?;
            let queued = self.queue_in_tx(conn, body.into_iter().collect())?;
            Ok::<_, DbError>((id, queued))
        })?;
        drop(conn);

        self.publish_node_by_id(id);
        self.to_log(queued);
        Ok(id)
    }

    // ========================================================================
    // Record store import (used by `records::reconcile`). These write the
    // database only; they never publish back to the store.
    // ========================================================================

    // What these apply came through git, and with a server log attached it
    // is queued for the server like a local write: an edit that arrived
    // through git is newer than anything the server holds from before it.
    // Without that, bob's `status C rejected`, made over the "completed"
    // alice's commit brought him, was refused (the server still had
    // "pending": alice's own op was still queued on her laptop), and
    // alice's op, replayed later, overwrote bob's edit on the server and
    // then, through `remote pull`, everywhere (round-2 NEW-1).
    //
    // The ops go into remote_outbox in the row's own transaction and reach
    // the log in one append when the reconcile ends (`flush_outbox`), not
    // one fsync per row: a fresh clone's first sync imports every node.
    // `remote pull` detaches the log while it applies the server's rows,
    // which the server does not need to hear about.

    /// Insert a node exactly as recorded (timestamps, status, metadata kept).
    pub(crate) fn import_node_record(&self, rec: &NodeRecord) -> Result<i32> {
        // The same vocabulary `add` enforces, widened only to what the
        // server holds. A hand-edited or corrupted graph.json used to import
        // any type and status here, while `add` refused them, and with a
        // remote the import was queued as an op the server then rejected.
        one_of("node type", &rec.node_type, IMPORTABLE_NODE_TYPES)?;
        one_of("status", &rec.status, IMPORTABLE_STATUSES)?;
        let mut conn = self.get_conn()?;
        let metadata = rec.metadata_json();
        let new_node = NewDecisionNode {
            change_id: &rec.change_id,
            node_type: &rec.node_type,
            title: &rec.title,
            description: rec.description.as_deref(),
            status: &rec.status,
            created_at: &rec.created_at,
            updated_at: &rec.updated_at,
            metadata_json: metadata.as_deref(),
        };
        let body = self.node_created_body(&new_node);
        conn.immediate_transaction(|conn| {
            // A node this database had before (a checkout took it out and
            // this one brings it back) returns under its old id.
            #[derive(diesel::QueryableByName)]
            struct Alias {
                #[diesel(sql_type = diesel::sql_types::Integer)]
                node_id: i32,
            }
            let alias = diesel::sql_query(
                "SELECT a.node_id FROM node_aliases a
                 WHERE a.change_id = ?
                   AND NOT EXISTS (SELECT 1 FROM decision_nodes n WHERE n.id = a.node_id)",
            )
            .bind::<diesel::sql_types::Text, _>(&rec.change_id)
            .load::<Alias>(conn)?
            .pop();
            let id = match alias {
                Some(Alias { node_id }) => {
                    diesel::insert_into(decision_nodes::table)
                        .values((decision_nodes::id.eq(node_id), &new_node))
                        .execute(conn)?;
                    node_id
                }
                None => {
                    diesel::insert_into(decision_nodes::table)
                        .values(&new_node)
                        .execute(conn)?;
                    diesel::select(diesel::dsl::sql::<diesel::sql_types::Integer>(
                        "last_insert_rowid()",
                    ))
                    .first(conn)?
                }
            };
            self.queue_git_in_tx(
                conn,
                body.into_iter().collect(),
                crate::records::parse_ts(&rec.updated_at),
            )?;
            Ok(id)
        })
    }

    /// Overwrite a node's fields exactly as recorded, and its `created_at`
    /// when the record's is earlier (never later: see
    /// `records::earlier_created`).
    pub(crate) fn update_node_record(&self, node_id: i32, rec: &NodeRecord) -> Result<()> {
        let mut conn = self.get_conn()?;
        let metadata = rec.metadata_json();
        conn.immediate_transaction(|conn| {
            let before = self.replaced_in_tx(conn, node_id)?;
            let had: Option<String> = decision_nodes::table
                .filter(decision_nodes::id.eq(node_id))
                .select(decision_nodes::created_at)
                .first(conn)
                .optional()?;
            let made = match &had {
                Some(h) => crate::records::earlier_created(h, &rec.created_at),
                None => &rec.created_at,
            };
            diesel::update(decision_nodes::table.filter(decision_nodes::id.eq(node_id)))
                .set((
                    decision_nodes::created_at.eq(made),
                    decision_nodes::node_type.eq(&rec.node_type),
                    decision_nodes::title.eq(&rec.title),
                    decision_nodes::description.eq(rec.description.as_deref()),
                    decision_nodes::status.eq(&rec.status),
                    decision_nodes::metadata_json.eq(metadata.as_deref()),
                    decision_nodes::updated_at.eq(&rec.updated_at),
                ))
                .execute(conn)?;
            let bodies = match before {
                Some(before) => Self::record_update_body(before, rec),
                None => Vec::new(),
            };
            self.queue_git_in_tx(conn, bodies, crate::records::parse_ts(&rec.updated_at))?;
            Ok(())
        })
    }

    /// Move a node's `created_at` to an earlier date a copy of it carries
    /// (graph.json, after a migration re-stamped this database's row).
    /// Refuses to move it later. Local only: no op changes a date on the
    /// server, whose import keeps the earlier one by the same rule.
    pub(crate) fn redate_node_record(&self, node_id: i32, created_at: &str) -> Result<bool> {
        let mut conn = self.get_conn()?;
        conn.immediate_transaction(|conn| {
            let had: String = decision_nodes::table
                .filter(decision_nodes::id.eq(node_id))
                .select(decision_nodes::created_at)
                .first(conn)?;
            if !crate::records::created_before(created_at, &had) {
                return Ok(false);
            }
            diesel::update(decision_nodes::table.filter(decision_nodes::id.eq(node_id)))
                .set(decision_nodes::created_at.eq(created_at))
                .execute(conn)?;
            Ok(true)
        })
    }

    /// The update op for applying `rec` over `before`: every column and
    /// metadata key whose value changes, with the value it replaces. A key
    /// `rec` lacks is left out (an op merges keys and cannot remove one),
    /// and so is the node type (not settable over /ops).
    fn record_update_body(before: DecisionNode, rec: &NodeRecord) -> Vec<crate::oplog::OpBody> {
        use serde_json::{Map, Value};
        let mut set = Map::new();
        if before.title != rec.title {
            set.insert("title".into(), Value::String(rec.title.clone()));
        }
        if before.status != rec.status {
            set.insert("status".into(), Value::String(rec.status.clone()));
        }
        if before.description.as_deref().unwrap_or("") != rec.description.as_deref().unwrap_or("") {
            set.insert(
                "description".into(),
                rec.description
                    .clone()
                    .map(Value::String)
                    .unwrap_or(Value::Null),
            );
        }
        let old: Map<String, Value> = before
            .metadata_json
            .as_deref()
            .and_then(|m| serde_json::from_str::<Value>(m).ok())
            .and_then(|v| v.as_object().cloned())
            .unwrap_or_default();
        let new: Map<String, Value> = rec
            .metadata
            .as_ref()
            .and_then(|m| m.as_object().cloned())
            .unwrap_or_default();
        let metadata: Map<String, Value> = new
            .into_iter()
            .filter(|(k, v)| old.get(k) != Some(v))
            .collect();
        if set.is_empty() && metadata.is_empty() {
            return Vec::new();
        }
        Self::node_update_body(Some(before), set, metadata)
    }

    /// The server's copy of every node as it was at the last `remote pull`,
    /// by change_id: `{"title", "description", "status", "metadata"}`.
    ///
    /// `remote push --repair` sends this copy's fields over the server's
    /// for edits that never became ops. Sent blind (with the server's
    /// current value as `was`), it overwrote every field an agent had
    /// changed since, because nothing said which side had moved (round-2
    /// BRIDGE-N1). A field whose server value still equals this base did
    /// not change there since the pull, so the difference is this copy's.
    pub fn remote_base(&self) -> Result<std::collections::HashMap<String, serde_json::Value>> {
        #[derive(QueryableByName)]
        struct Row {
            #[diesel(sql_type = diesel::sql_types::Text)]
            change_id: String,
            #[diesel(sql_type = diesel::sql_types::Text)]
            node_json: String,
        }
        let mut conn = self.get_conn()?;
        let rows: Vec<Row> =
            diesel::sql_query("SELECT change_id, node_json FROM remote_base").load(&mut conn)?;
        Ok(rows
            .into_iter()
            .filter_map(|r| Some((r.change_id, serde_json::from_str(&r.node_json).ok()?)))
            .collect())
    }

    /// [`Self::remote_base`] of the database at `path`, read without opening
    /// it as a [`Database`]: no schema migration, no graph.json attached, no
    /// outbox drained. For the git merge driver, which runs inside a merge
    /// and must not write anything but the file it merges. `Ok(None)` when
    /// there is no database there, or it has never recorded a pull (a clone
    /// without a remote, or one older than the table).
    pub fn remote_base_at(
        path: &Path,
    ) -> Result<Option<std::collections::HashMap<String, serde_json::Value>>> {
        #[derive(QueryableByName)]
        struct Row {
            #[diesel(sql_type = diesel::sql_types::Text)]
            change_id: String,
            #[diesel(sql_type = diesel::sql_types::Text)]
            node_json: String,
        }
        #[derive(QueryableByName)]
        struct Count {
            #[diesel(sql_type = diesel::sql_types::BigInt)]
            n: i64,
        }
        if !path.is_file() {
            return Ok(None);
        }
        let mut conn = SqliteConnection::establish(&path.to_string_lossy())
            .map_err(|e| DbError::Connection(e.to_string()))?;
        diesel::sql_query("PRAGMA busy_timeout = 5000").execute(&mut conn)?;
        let table: Vec<Count> = diesel::sql_query(
            "SELECT COUNT(*) AS n FROM sqlite_master WHERE type = 'table' AND name = 'remote_base'",
        )
        .load(&mut conn)?;
        if table.first().map(|c| c.n).unwrap_or(0) == 0 {
            return Ok(None);
        }
        let rows: Vec<Row> =
            diesel::sql_query("SELECT change_id, node_json FROM remote_base").load(&mut conn)?;
        if rows.is_empty() {
            return Ok(None);
        }
        Ok(Some(
            rows.into_iter()
                .filter_map(|r| Some((r.change_id, serde_json::from_str(&r.node_json).ok()?)))
                .collect(),
        ))
    }

    /// Replaces the base with the server's live nodes, as a pull saw them.
    pub fn set_remote_base(&self, nodes: &[(String, serde_json::Value)]) -> Result<()> {
        let mut conn = self.get_conn()?;
        conn.immediate_transaction(|conn| {
            diesel::sql_query("DELETE FROM remote_base").execute(conn)?;
            for (cid, v) in nodes {
                diesel::sql_query("INSERT INTO remote_base (change_id, node_json) VALUES (?, ?)")
                    .bind::<diesel::sql_types::Text, _>(cid)
                    .bind::<diesel::sql_types::Text, _>(v.to_string())
                    .execute(conn)?;
            }
            Ok::<_, DbError>(())
        })
    }

    /// Moves what reconcile queued in remote_outbox to the log, in one
    /// append. A failure leaves them in the outbox, where the next logged
    /// write or the next process to open the database finds them.
    pub(crate) fn flush_outbox(&self) {
        if self.oplog().is_none() {
            return;
        }
        if let Err(e) = self.hold_log() {
            eprintln!(
                "Warning: edits applied from graph.json are kept in this database and queued for \
                 the server by the next write: {e}"
            );
        }
    }

    /// Delete a node (and its edges) without writing tombstones: a delete
    /// that came through git. With a log, the node's delete is queued (its
    /// edges go with it on the server, which hides a deleted node's edges),
    /// dated `deleted`, when the tombstone says the node was deleted.
    pub(crate) fn delete_node_local(
        &self,
        node_id: i32,
        deleted: chrono::DateTime<chrono::Utc>,
    ) -> Result<()> {
        self.delete_node_impl(node_id, false, Some(deleted))
            .map(|_| ())
    }

    /// Insert an edge exactly as recorded.
    pub(crate) fn import_edge_record(
        &self,
        from_id: i32,
        to_id: i32,
        rec: &EdgeRecord,
    ) -> Result<i32> {
        // What create_edge refuses, refused on import too: an unknown type,
        // a self-loop, and a 2-cycle (the server refuses both on every path,
        // so an imported one would sit in the op log as a rejected op).
        one_of("edge type", &rec.edge_type, EDGE_TYPES)?;
        if from_id == to_id {
            return Err(DbError::Validation(format!(
                "edge {} -> {} ({}) links a node to itself",
                rec.from_change_id, rec.to_change_id, rec.edge_type
            )));
        }
        let mut conn = self.get_conn()?;
        if rec.edge_type != "took_from" {
            let reverse = decision_edges::table
                .filter(decision_edges::from_node_id.eq(to_id))
                .filter(decision_edges::to_node_id.eq(from_id))
                .filter(decision_edges::edge_type.ne("took_from"))
                .first::<DecisionEdge>(&mut conn)
                .optional()?;
            if let Some(rev) = reverse {
                return Err(DbError::Validation(format!(
                    "edge {} -> {} ({}) would make the two nodes each other's parent; \
                     {} -> {} ({}) is already in the graph",
                    rec.from_change_id,
                    rec.to_change_id,
                    rec.edge_type,
                    rec.to_change_id,
                    rec.from_change_id,
                    rev.edge_type
                )));
            }
        }
        let new_edge = NewDecisionEdge {
            from_node_id: from_id,
            to_node_id: to_id,
            from_change_id: Some(&rec.from_change_id),
            to_change_id: Some(&rec.to_change_id),
            edge_type: &rec.edge_type,
            weight: rec.weight.or(Some(1.0)),
            rationale: rec.rationale.as_deref(),
            created_at: &rec.created_at,
        };
        let body = self.oplog().map(|_| crate::oplog::OpBody::CreateEdge {
            from_change_id: rec.from_change_id.clone(),
            to_change_id: rec.to_change_id.clone(),
            edge_type: rec.edge_type.clone(),
            rationale: rec.rationale.clone(),
            weight: new_edge.weight,
            created_at: crate::records::wire_ts(&rec.created_at),
        });
        conn.immediate_transaction(|conn| {
            diesel::insert_into(decision_edges::table)
                .values(&new_edge)
                .execute(conn)?;
            let id: i32 = diesel::select(diesel::dsl::sql::<diesel::sql_types::Integer>(
                "last_insert_rowid()",
            ))
            .first(conn)?;
            self.queue_git_in_tx(
                conn,
                body.into_iter().collect(),
                crate::records::parse_ts(&rec.created_at),
            )?;
            Ok(id)
        })
    }

    /// Take an edge's rationale and weight from its record, without
    /// publishing (sync is applying the file, not changing it).
    pub(crate) fn update_edge_record(&self, edge_id: i32, rec: &EdgeRecord) -> Result<()> {
        let mut conn = self.get_conn()?;
        diesel::update(decision_edges::table.filter(decision_edges::id.eq(edge_id)))
            .set((
                decision_edges::rationale.eq(rec.rationale.as_deref()),
                decision_edges::weight.eq(rec.weight.or(Some(1.0))),
            ))
            .execute(&mut conn)?;
        Ok(())
    }

    /// Delete one edge row by primary key without writing a tombstone.
    /// `deleted_at` is when the unlink in graph.json was made, which the
    /// server orders against a link of the same edge.
    pub(crate) fn delete_edge_local(&self, edge_id: i32, deleted_at: Option<&str>) -> Result<()> {
        let mut conn = self.get_conn()?;
        let logged = self.oplog().is_some();
        let unnamed = conn.immediate_transaction(|conn| {
            let row = decision_edges::table
                .filter(decision_edges::id.eq(edge_id))
                .first::<DecisionEdge>(conn)
                .optional()?;
            diesel::delete(decision_edges::table.filter(decision_edges::id.eq(edge_id)))
                .execute(conn)?;
            let Some(row) = row.filter(|_| logged) else {
                return Ok(None);
            };
            match (row.from_change_id.clone(), row.to_change_id.clone()) {
                (Some(from_change_id), Some(to_change_id)) => {
                    self.queue_git_in_tx(
                        conn,
                        vec![crate::oplog::OpBody::DeleteEdge {
                            from_change_id,
                            to_change_id,
                            edge_type: row.edge_type.clone(),
                            deleted_at: deleted_at.map(str::to_string),
                        }],
                        deleted_at.map_or_else(chrono::Utc::now, crate::records::parse_ts),
                    )?;
                    Ok(None)
                }
                _ => Ok::<_, DbError>(Some(format!(
                    "edge {} ({} -> {}) has no change_id for an endpoint, so its removal cannot be named to the server",
                    row.id, row.from_node_id, row.to_node_id
                ))),
            }
        })?;
        if let Some(u) = unnamed {
            self.not_queued(u);
        }
        Ok(())
    }

    /// Insert a theme exactly as recorded.
    pub(crate) fn import_theme_record(&self, rec: &ThemeRecord) -> Result<i32> {
        let mut conn = self.get_conn()?;
        let new_theme = NewTheme {
            change_id: &rec.change_id,
            name: &rec.name,
            color: &rec.color,
            description: rec.description.as_deref(),
            created_at: &rec.created_at,
            updated_at: &rec.updated_at,
        };
        diesel::insert_into(themes::table)
            .values(&new_theme)
            .execute(&mut conn)?;
        let id: i32 = diesel::select(diesel::dsl::sql::<diesel::sql_types::Integer>(
            "last_insert_rowid()",
        ))
        .first(&mut conn)?;
        Ok(id)
    }

    /// Overwrite a theme's fields exactly as recorded.
    pub(crate) fn update_theme_record(&self, theme_id: i32, rec: &ThemeRecord) -> Result<()> {
        let mut conn = self.get_conn()?;
        diesel::update(themes::table.filter(themes::id.eq(theme_id)))
            .set((
                themes::name.eq(&rec.name),
                themes::color.eq(&rec.color),
                themes::description.eq(rec.description.as_deref()),
                themes::updated_at.eq(&rec.updated_at),
            ))
            .execute(&mut conn)?;
        Ok(())
    }

    /// Give a theme a new change_id (used when two machines created the
    /// same theme name independently and one id becomes canonical).
    pub(crate) fn update_theme_change_id(&self, theme_id: i32, change_id: &str) -> Result<()> {
        let mut conn = self.get_conn()?;
        diesel::update(themes::table.filter(themes::id.eq(theme_id)))
            .set(themes::change_id.eq(change_id))
            .execute(&mut conn)?;
        Ok(())
    }

    /// Delete a theme and its tags without writing tombstones.
    pub(crate) fn delete_theme_local(&self, theme_id: i32) -> Result<()> {
        let mut conn = self.get_conn()?;
        diesel::delete(node_themes::table.filter(node_themes::theme_id.eq(theme_id)))
            .execute(&mut conn)?;
        diesel::delete(themes::table.filter(themes::id.eq(theme_id))).execute(&mut conn)?;
        Ok(())
    }

    /// Insert a tag exactly as recorded (no-op if already present).
    pub(crate) fn import_tag_record(
        &self,
        node_id: i32,
        theme_id: i32,
        rec: &TagRecord,
    ) -> Result<()> {
        let mut conn = self.get_conn()?;
        let new_nt = NewNodeTheme {
            node_id,
            theme_id,
            source: rec.source.clone(),
            created_at: rec.created_at.clone(),
        };
        diesel::insert_or_ignore_into(node_themes::table)
            .values(&new_nt)
            .execute(&mut conn)?;
        Ok(())
    }

    /// Remove a tag without writing a tombstone.
    pub(crate) fn delete_tag_local(&self, node_id: i32, theme_id: i32) -> Result<()> {
        let mut conn = self.get_conn()?;
        diesel::delete(
            node_themes::table
                .filter(node_themes::node_id.eq(node_id))
                .filter(node_themes::theme_id.eq(theme_id)),
        )
        .execute(&mut conn)?;
        Ok(())
    }

    /// Resolve a node reference from the command line or an MCP call.
    ///
    /// Accepts a local integer id (`42`, `#42`) or a `change_id` prefix of
    /// at least four characters (`a1b2c3d4`). Local ids differ between
    /// machines; the prefix is how you point at a teammate's node.
    ///
    /// `#42` is always a local id. A bare number of four or more digits is
    /// also a valid change_id prefix (about one change_id in seven starts
    /// with four digits), so it is looked up both ways: if it names a local
    /// node *and* prefixes a change_id, that is an error listing both,
    /// never a guess, because a guess aims `delete` at the wrong node and
    /// the tombstone reaches everyone. A full change_id wins over a longer
    /// one it is a prefix of.
    pub fn resolve_node_ref(&self, reference: &str) -> Result<i32> {
        let trimmed = reference.trim();
        if let Some(explicit) = trimmed.strip_prefix('#') {
            return explicit.parse::<i32>().map_err(|_| {
                DbError::Validation(format!(
                    "'{}' is not a local node id (after '#' comes an integer)",
                    reference
                ))
            });
        }
        let r = trimmed;
        let looks_like_prefix =
            r.len() >= 4 && r.chars().all(|c| c.is_ascii_hexdigit() || c == '-');
        let as_id = r.parse::<i32>().ok().filter(|id| *id >= 0);
        if !looks_like_prefix {
            return as_id.ok_or_else(|| {
                DbError::Validation(format!(
                    "'{}' is not a node id, a change_id prefix or a server id. {}",
                    reference, NODE_REF_KINDS
                ))
            });
        }
        let Some(id) = as_id else {
            return self.resolve_change_id_prefix(r);
        };
        // Digits only, four or more: an id, a change_id prefix, or both.
        let local = self.get_node(id)?;
        let by_prefix = self.change_id_matches(r)?;
        let pending = self.unsynced_matches(r)?;
        if !pending.is_empty() {
            return Err(Self::unsynced_error(
                r,
                local.as_ref(),
                &by_prefix,
                &pending,
            ));
        }
        match (local, by_prefix.as_slice()) {
            // Nothing prints a local id zero-padded, so "0012" is far more
            // likely the start of a change_id or server id than local #12,
            // and reading it as #12 named some other node without a word.
            (Some(node), []) if r.starts_with('0') => Err(DbError::NoSuchNode(format!(
                "'{}' is zero-padded, so it is not read as local id {} (#{} {}; write {} or #{} for that), and no change_id starts with it.",
                r, id, id, node.title, id, id
            ))),
            (Some(_), []) => Ok(id),
            // Neither a local node nor a change_id prefix. Returning the id
            // anyway (as this did) made "no such node" the caller's error,
            // so a server id whose prefix is all digits (89346034-83ce...,
            // 2.3% of them at 8 characters) was never looked up as one:
            // `dx show 89346034` said "Node #89346034 not found".
            (None, []) => Err(DbError::NoSuchNode(format!(
                "No node has local id {} or a change_id starting with '{}'.",
                id, r
            ))),
            (None, _) => self.resolve_change_id_prefix(r),
            (Some(node), matches) => {
                let mut list = vec![format!("local id #{} ({})", node.id, node.title)];
                list.extend(matches.iter().take(5).map(Self::describe_match));
                Err(DbError::Validation(format!(
                    "'{}' is both a local id and a change_id prefix: {}. Use '#{}' for the local id, or more change_id characters",
                    r,
                    list.join("; "),
                    id
                )))
            }
        }
    }

    fn describe_match(n: &DecisionNode) -> String {
        format!(
            "#{} {} ({})",
            n.id,
            n.change_id.chars().take(12).collect::<String>(),
            n.title
        )
    }

    /// Nodes whose change_id starts with `r` (at most six), with an exact
    /// match, if any, first and alone.
    fn change_id_matches(&self, r: &str) -> Result<Vec<DecisionNode>> {
        let mut conn = self.get_conn()?;
        let exact: Option<DecisionNode> = decision_nodes::table
            .filter(decision_nodes::change_id.eq(r))
            .first(&mut conn)
            .optional()?;
        if let Some(node) = exact {
            return Ok(vec![node]);
        }
        let pattern = format!("{}%", r);
        Ok(decision_nodes::table
            .filter(decision_nodes::change_id.like(&pattern))
            .order(decision_nodes::id.asc())
            .limit(6)
            .load(&mut conn)?)
    }

    /// Live nodes in the graph file whose change_id starts with `r` and
    /// that the database does not hold yet: pulled, not synced. Resolving
    /// against the database alone aimed `delete 0002` at local #2 when the
    /// node meant was a teammate's that the next sync would import.
    /// (At most six, like [`Self::change_id_matches`].)
    fn unsynced_matches(&self, r: &str) -> Result<Vec<(String, String)>> {
        let Some(store) = self.store() else {
            return Ok(Vec::new());
        };
        // An unreadable file is reported by whatever reads or writes it
        // next; resolving a reference is not the place to fail on it.
        let Ok(doc) = store.read_doc_all() else {
            return Ok(Vec::new());
        };
        let mut conn = self.get_conn()?;
        let mut out = Vec::new();
        for rec in doc
            .nodes
            .values()
            .filter(|n| !n.is_tombstone() && n.change_id.starts_with(r))
        {
            let known: i64 = decision_nodes::table
                .filter(decision_nodes::change_id.eq(&rec.change_id))
                .count()
                .get_result(&mut conn)?;
            if known == 0 {
                out.push((rec.change_id.clone(), rec.title.clone()));
                if out.len() == 6 {
                    break;
                }
            }
        }
        Ok(out)
    }

    fn unsynced_error(
        r: &str,
        local: Option<&DecisionNode>,
        in_db: &[DecisionNode],
        pending: &[(String, String)],
    ) -> DbError {
        let mut list: Vec<String> = local
            .map(|n| format!("local id #{} ({})", n.id, n.title))
            .into_iter()
            .collect();
        list.extend(in_db.iter().take(5).map(Self::describe_match));
        list.extend(pending.iter().take(5).map(|(cid, title)| {
            format!(
                "{} ({}, in graph.json, not synced yet)",
                cid.chars().take(12).collect::<String>(),
                title
            )
        }));
        DbError::Validation(format!(
            "'{}' matches a node that is in graph.json but not in this database yet: {}. Run `deciduous sync` first, then use more change_id characters{}",
            r,
            list.join("; "),
            local.map(|n| format!(" (or '#{}' for the local id)", n.id)).unwrap_or_default()
        ))
    }

    fn resolve_change_id_prefix(&self, r: &str) -> Result<i32> {
        let matches = self.change_id_matches(r)?;
        let exact = matches.len() == 1 && matches[0].change_id == r;
        if !exact {
            let pending = self.unsynced_matches(r)?;
            if !pending.is_empty() {
                return Err(Self::unsynced_error(r, None, &matches, &pending));
            }
        }
        match matches.len() {
            0 => Err(DbError::NoSuchNode(format!(
                "No node has a change_id starting with '{}'. Run 'deciduous sync' if a teammate created it.",
                r
            ))),
            1 => Ok(matches[0].id),
            _ => {
                let list: Vec<String> = matches.iter().take(5).map(Self::describe_match).collect();
                Err(DbError::Validation(format!(
                    "'{}' matches several nodes; use more characters: {}",
                    r,
                    list.join("; ")
                )))
            }
        }
    }

    /// Get a single edge by primary key.
    pub fn get_edge(&self, edge_id: i32) -> Result<Option<DecisionEdge>> {
        let mut conn = self.get_conn()?;
        let edge = decision_edges::table
            .filter(decision_edges::id.eq(edge_id))
            .first::<DecisionEdge>(&mut conn)
            .optional()?;
        Ok(edge)
    }

    /// Get a theme by primary key.
    pub fn get_theme_by_id(&self, theme_id: i32) -> Result<Option<Theme>> {
        let mut conn = self.get_conn()?;
        let theme = themes::table
            .filter(themes::id.eq(theme_id))
            .first::<Theme>(&mut conn)
            .optional()?;
        Ok(theme)
    }

    /// Get one node/theme association.
    pub fn get_tag(&self, node_id: i32, theme_id: i32) -> Result<Option<NodeTheme>> {
        let mut conn = self.get_conn()?;
        let tag = node_themes::table
            .filter(node_themes::node_id.eq(node_id))
            .filter(node_themes::theme_id.eq(theme_id))
            .first::<NodeTheme>(&mut conn)
            .optional()?;
        Ok(tag)
    }

    /// Create an edge between nodes
    pub fn create_edge(
        &self,
        from_id: i32,
        to_id: i32,
        edge_type: &str,
        rationale: Option<&str>,
    ) -> Result<i32> {
        one_of("edge type", edge_type, EDGE_TYPES)?;
        if from_id == to_id {
            return Err(DbError::Validation(format!(
                "cannot link node {from_id} to itself"
            )));
        }
        let _mirror = self.require_readable_store()?;
        let _log = self.hold_log()?;
        let mut conn = self.get_conn()?;

        // Validate both nodes exist and get their change_ids
        let from_node = decision_nodes::table
            .filter(decision_nodes::id.eq(from_id))
            .first::<DecisionNode>(&mut conn)
            .ok();
        let to_node = decision_nodes::table
            .filter(decision_nodes::id.eq(to_id))
            .first::<DecisionNode>(&mut conn)
            .ok();

        let from_change_id = from_node.as_ref().map(|n| n.change_id.clone());
        let to_change_id = to_node.as_ref().map(|n| n.change_id.clone());

        if from_node.is_none() && to_node.is_none() {
            return Err(DbError::Validation(format!(
                "Both nodes {} and {} do not exist. Run 'deciduous nodes' to see existing nodes.",
                from_id, to_id
            )));
        } else if from_node.is_none() {
            return Err(DbError::Validation(format!(
                "Source node {} does not exist. Run 'deciduous nodes' to see existing nodes.",
                from_id
            )));
        } else if to_node.is_none() {
            return Err(DbError::Validation(format!(
                "Target node {} does not exist. Run 'deciduous nodes' to see existing nodes.",
                to_id
            )));
        }

        // A 2-cycle: the two nodes would be each other's parent. The server
        // refuses it on every path (MCP add_edge, POST /ops), so an edge
        // written here would be an op it rejects, kept in the log until
        // someone drops it. Refused here first, where it can be said
        // plainly. took_from, a borrow between branches, is exempt on both
        // sides, as it is on the server.
        if edge_type != "took_from" {
            let reverse = decision_edges::table
                .filter(decision_edges::from_node_id.eq(to_id))
                .filter(decision_edges::to_node_id.eq(from_id))
                .filter(decision_edges::edge_type.ne("took_from"))
                .first::<DecisionEdge>(&mut conn)
                .optional()?;
            if let Some(rev) = reverse {
                return Err(DbError::Validation(format!(
                    "{to_id} -> {from_id} ({}) already exists; {from_id} -> {to_id} would make \
                     the two nodes each other's parent. A decision's options hang under it \
                     (decision -> option); to put a decision under its goal, link goal -> decision.",
                    rev.edge_type
                )));
            }
        }

        let now = chrono::Local::now().to_rfc3339();

        let new_edge = NewDecisionEdge {
            from_node_id: from_id,
            to_node_id: to_id,
            from_change_id: from_change_id.as_deref(),
            to_change_id: to_change_id.as_deref(),
            edge_type,
            weight: Some(1.0),
            rationale,
            created_at: &now,
        };

        let body = match (self.oplog(), &from_change_id, &to_change_id) {
            (Some(_), Some(f), Some(t)) => vec![crate::oplog::OpBody::CreateEdge {
                from_change_id: f.clone(),
                to_change_id: t.clone(),
                edge_type: edge_type.to_string(),
                rationale: rationale.map(str::to_string),
                weight: new_edge.weight,
                created_at: now.clone(),
            }],
            _ => Vec::new(),
        };
        let (id, queued) = conn.immediate_transaction(|conn| {
            let inserted = diesel::insert_into(decision_edges::table)
                .values(&new_edge)
                .execute(conn);
            if let Err(diesel::result::Error::DatabaseError(
                diesel::result::DatabaseErrorKind::UniqueViolation,
                _,
            )) = inserted
            {
                // SQLite's text ("UNIQUE constraint failed: decision_edges...")
                // names columns, not the edge, and not what to do instead.
                let existing = decision_edges::table
                    .filter(decision_edges::from_node_id.eq(from_id))
                    .filter(decision_edges::to_node_id.eq(to_id))
                    .filter(decision_edges::edge_type.eq(edge_type))
                    .first::<DecisionEdge>(conn)
                    .ok();
                let rationale = existing
                    .as_ref()
                    .and_then(|e| e.rationale.as_deref())
                    .map(|r| format!(" (rationale: {r:?})"))
                    .unwrap_or_default();
                return Err(DbError::Validation(format!(
                    "node {from_id} already {edge_type} node {to_id}{rationale}; nothing was changed. \
                     To give it a new rationale, unlink it first (`deciduous unlink {from_id} {to_id} -t {edge_type}`, \
                     or unlink_nodes with edge_type {edge_type:?}), then link again"
                )));
            }
            inserted?;
            let id: i32 = diesel::select(diesel::dsl::sql::<diesel::sql_types::Integer>(
                "last_insert_rowid()",
            ))
            .first(conn)?;
            let queued = self.queue_in_tx(conn, body)?;
            Ok::<_, DbError>((id, queued))
        })?;
        drop(conn);

        self.publish_edge_by_id(id);
        self.to_log(queued);
        Ok(id)
    }

    /// Add an edge (alias for create_edge for doc examples)
    pub fn add_edge(
        &self,
        from_id: i32,
        to_id: i32,
        edge_type: &str,
        rationale: Option<&str>,
    ) -> Result<i32> {
        self.create_edge(from_id, to_id, edge_type, rationale)
    }

    /// Delete an edge between two nodes
    /// Remove the edge from `from_id` to `to_id`, of `edge_type` when given,
    /// and return what was removed.
    ///
    /// Without a type it removes the pair's edge only when there is exactly
    /// one. It used to remove every edge between the two, whatever its type,
    /// and print "Removed edge (5 -> 6)": the `unlink 5 6` that the
    /// duplicate-link refusal advised for a `chosen` edge silently took the
    /// pair's `leads_to` edge with it.
    pub fn delete_edge(
        &self,
        from_id: i32,
        to_id: i32,
        edge_type: Option<&str>,
    ) -> Result<Vec<DecisionEdge>> {
        let _mirror = self.require_readable_store()?;
        self.require_branch_for_delete("unlink")?;
        let _log = self.hold_log()?;
        let mut conn = self.get_conn()?;

        let between: Vec<DecisionEdge> = decision_edges::table
            .filter(decision_edges::from_node_id.eq(from_id))
            .filter(decision_edges::to_node_id.eq(to_id))
            .order(decision_edges::id.asc())
            .load(&mut conn)?;

        if between.is_empty() {
            // Get outgoing edges from source node to provide helpful error
            let outgoing: Vec<DecisionEdge> = decision_edges::table
                .filter(decision_edges::from_node_id.eq(from_id))
                .load(&mut conn)?;

            if outgoing.is_empty() {
                return Err(DbError::Validation(format!(
                    "No edge from node {} to node {}. Node {} has no outgoing edges.",
                    from_id, to_id, from_id
                )));
            } else {
                let targets: Vec<String> =
                    outgoing.iter().map(|e| e.to_node_id.to_string()).collect();
                return Err(DbError::Validation(format!(
                    "No edge from node {} to node {}. Node {} has edges to: {}",
                    from_id,
                    to_id,
                    from_id,
                    targets.join(", ")
                )));
            }
        }

        let describe = |edges: &[DecisionEdge]| {
            edges
                .iter()
                .map(|e| match &e.rationale {
                    Some(r) => format!("{} (rationale: {r:?})", e.edge_type),
                    None => e.edge_type.clone(),
                })
                .collect::<Vec<_>>()
                .join(", ")
        };
        let doomed: Vec<DecisionEdge> = match edge_type {
            Some(t) => {
                let picked: Vec<DecisionEdge> = between
                    .iter()
                    .filter(|e| e.edge_type == t)
                    .cloned()
                    .collect();
                if picked.is_empty() {
                    return Err(DbError::Validation(format!(
                        "No {t} edge from node {from_id} to node {to_id}; the edges between them are: {}. Nothing was removed.",
                        describe(&between)
                    )));
                }
                picked
            }
            None if between.len() > 1 => {
                return Err(DbError::Validation(format!(
                    "Node {from_id} has {} edges to node {to_id}: {}. Nothing was removed; name the one to remove with its type (`deciduous unlink {from_id} {to_id} -t {}`, or unlink_nodes with edge_type).",
                    between.len(),
                    describe(&between),
                    between[0].edge_type
                )));
            }
            None => between,
        };

        drop(conn);
        let endpoints = self.endpoint_change_ids(&doomed);
        let (bodies, unnamed) = self.edges_deleted_bodies(&doomed, &endpoints);
        let mut conn = self.get_conn()?;

        let ids: Vec<i32> = doomed.iter().map(|e| e.id).collect();
        let queued = conn.immediate_transaction(|conn| {
            diesel::delete(decision_edges::table.filter(decision_edges::id.eq_any(&ids)))
                .execute(conn)?;
            self.queue_in_tx(conn, bodies)
        })?;
        drop(conn);

        self.tombstone_edges(&doomed);
        self.to_log(queued);
        for u in unnamed {
            self.not_queued(u);
        }
        Ok(doomed)
    }

    /// Delete a node and all its connected edges
    pub fn delete_node(&self, node_id: i32, dry_run: bool) -> Result<DeleteSummary> {
        self.delete_node_impl(node_id, dry_run, None)
    }

    /// `from_git` is when a delete that came through graph.json was made;
    /// None for a local delete, which is published and dated now.
    fn delete_node_impl(
        &self,
        node_id: i32,
        dry_run: bool,
        from_git: Option<chrono::DateTime<chrono::Utc>>,
    ) -> Result<DeleteSummary> {
        let publish = from_git.is_none();
        let _mirror = if !dry_run && publish {
            let guard = self.require_readable_store()?;
            self.require_branch_for_delete("delete it")?;
            Some(guard)
        } else {
            None
        };
        let _log = if dry_run { None } else { self.hold_log()? };
        let mut conn = self.get_conn()?;

        // Check if node exists
        let node = decision_nodes::table
            .filter(decision_nodes::id.eq(node_id))
            .first::<DecisionNode>(&mut conn)
            .optional()?;

        let node = match node {
            Some(n) => n,
            None => {
                return Err(DbError::Validation(format!(
                    "Node {} does not exist. Run 'deciduous nodes' to see existing nodes.",
                    node_id
                )));
            }
        };

        // Count edges that will be deleted
        let edges_count: i64 = decision_edges::table
            .filter(
                decision_edges::from_node_id
                    .eq(node_id)
                    .or(decision_edges::to_node_id.eq(node_id)),
            )
            .count()
            .get_result(&mut conn)?;

        let summary = DeleteSummary {
            node_title: node.title.clone(),
            edges_deleted: edges_count as usize,
        };

        if dry_run {
            return Ok(summary);
        }

        let doomed_edges: Vec<DecisionEdge> = decision_edges::table
            .filter(
                decision_edges::from_node_id
                    .eq(node_id)
                    .or(decision_edges::to_node_id.eq(node_id)),
            )
            .load(&mut conn)?;
        drop(conn);
        let endpoints = if publish {
            self.endpoint_change_ids(&doomed_edges)
        } else {
            Default::default()
        };
        // Edges first, then the node: the server keeps a deleted node's
        // edges unless told, and the local graph just lost them.
        let (mut bodies, unnamed) = if publish {
            self.edges_deleted_bodies(&doomed_edges, &endpoints)
        } else {
            Default::default()
        };
        if self.oplog().is_some() {
            bodies.push(Self::node_deleted_body(&node));
        }
        let mut conn = self.get_conn()?;

        // One transaction: the node, everything pointing at it, and the ops
        // that tell the server, or none of them.
        let (doomed_tags, queued) = conn.immediate_transaction(|conn| {
            // Delete in order to respect foreign key constraints:

            // 1. Delete edges connected to this node
            diesel::delete(
                decision_edges::table.filter(
                    decision_edges::from_node_id
                        .eq(node_id)
                        .or(decision_edges::to_node_id.eq(node_id)),
                ),
            )
            .execute(conn)?;

            // 2. Delete context entries for this node
            diesel::delete(decision_context::table.filter(decision_context::node_id.eq(node_id)))
                .execute(conn)?;

            // 3. Delete session_nodes entries for this node
            diesel::delete(session_nodes::table.filter(session_nodes::node_id.eq(node_id)))
                .execute(conn)?;

            // 3b. Tags on this node (foreign keys are declared but not enforced)
            let doomed_tags: Vec<NodeTheme> = node_themes::table
                .filter(node_themes::node_id.eq(node_id))
                .load(conn)?;
            diesel::delete(node_themes::table.filter(node_themes::node_id.eq(node_id)))
                .execute(conn)?;

            // 4. Set nullable foreign keys to NULL
            diesel::update(
                decision_sessions::table.filter(decision_sessions::root_node_id.eq(node_id)),
            )
            .set(decision_sessions::root_node_id.eq::<Option<i32>>(None))
            .execute(conn)?;

            diesel::update(command_log::table.filter(command_log::decision_node_id.eq(node_id)))
                .set(command_log::decision_node_id.eq::<Option<i32>>(None))
                .execute(conn)?;

            diesel::update(roadmap_items::table.filter(roadmap_items::outcome_node_id.eq(node_id)))
                .set(roadmap_items::outcome_node_id.eq::<Option<i32>>(None))
                .execute(conn)?;

            // 5. Finally delete the node itself
            diesel::delete(decision_nodes::table.filter(decision_nodes::id.eq(node_id)))
                .execute(conn)?;
            let queued = match from_git {
                None => self.queue_in_tx(conn, bodies)?,
                Some(made) => self.queue_git_in_tx(conn, bodies, made)?,
            };
            Ok::<_, DbError>((doomed_tags, queued))
        })?;
        drop(conn);

        if publish {
            if let Some(store) = self.write_store() {
                if let Err(e) = store.tombstone_node(&node) {
                    Self::store_warn("could not write node tombstone", e);
                }
            }
            self.tombstone_edges_of(&doomed_edges, Some(&node.change_id));
            self.to_log(queued);
            for u in unnamed {
                self.not_queued(u);
            }
            if let Some(store) = self.write_store() {
                for tag in &doomed_tags {
                    if let Ok(Some(theme)) = self.get_theme_by_id(tag.theme_id) {
                        if let Err(e) = store.tombstone_tag(&node.change_id, &theme.change_id, true)
                        {
                            Self::store_warn("could not write tag tombstone", e);
                        }
                    }
                }
            }
        }

        Ok(summary)
    }

    /// Update node status
    pub fn update_node_status(&self, node_id: i32, status: &str) -> Result<()> {
        one_of("status", status, NODE_STATUSES)?;
        let _mirror = self.require_readable_store()?;
        let _log = self.hold_log()?;
        let before = self.node_before_edit(node_id);
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        // `was` is read in the transaction that writes, so it is the value
        // this write replaced, whatever wrote in between.
        let queued = conn.immediate_transaction(|conn| {
            let replaced = self.replaced_in_tx(conn, node_id)?;
            let updated =
                diesel::update(decision_nodes::table.filter(decision_nodes::id.eq(node_id)))
                    .set((
                        decision_nodes::status.eq(status),
                        decision_nodes::updated_at.eq(&now),
                    ))
                    .execute(conn)?;
            if updated == 0 {
                return Err(node_not_found(node_id));
            }
            let body =
                Self::node_update_body(replaced, one_field("status", status), Default::default());
            self.queue_in_tx(conn, body)
        })?;
        drop(conn);

        self.publish_node_edit(node_id, before);
        self.to_log(queued);
        Ok(())
    }

    /// Update a node's commit hash in metadata_json
    pub fn update_node_commit(&self, node_id: i32, commit_hash: &str) -> Result<()> {
        let _mirror = self.require_readable_store()?;
        let _log = self.hold_log()?;
        let before = self.node_before_edit(node_id);
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        // Read, merge and write in one transaction: the map read is the map
        // replaced, so neither a concurrent key nor `was` goes stale.
        let queued = conn.immediate_transaction(|conn| {
            let replaced = self.replaced_in_tx(conn, node_id)?;
            let current_meta: Option<String> = decision_nodes::table
                .filter(decision_nodes::id.eq(node_id))
                .select(decision_nodes::metadata_json)
                .first(conn)
                .optional()?
                .ok_or_else(|| node_not_found(node_id))?;

            // Parse existing metadata or create new
            let mut meta: serde_json::Value = current_meta
                .as_ref()
                .and_then(|m| serde_json::from_str(m).ok())
                .unwrap_or_else(|| serde_json::json!({}));

            // Add/update commit field
            if let Some(obj) = meta.as_object_mut() {
                obj.insert("commit".to_string(), serde_json::json!(commit_hash));
            }

            let new_meta = serde_json::to_string(&meta)
                .map_err(|e| DbError::Validation(format!("JSON serialization error: {}", e)))?;

            diesel::update(decision_nodes::table.filter(decision_nodes::id.eq(node_id)))
                .set((
                    decision_nodes::metadata_json.eq(Some(new_meta)),
                    decision_nodes::updated_at.eq(&now),
                ))
                .execute(conn)?;
            let body = Self::node_update_body(
                replaced,
                Default::default(),
                one_field("commit", commit_hash),
            );
            self.queue_in_tx(conn, body)
        })?;
        drop(conn);

        self.publish_node_edit(node_id, before);
        self.to_log(queued);
        Ok(())
    }

    /// Update a node's prompt in metadata_json
    pub fn update_node_prompt(&self, node_id: i32, prompt: &str) -> Result<()> {
        let _mirror = self.require_readable_store()?;
        let _log = self.hold_log()?;
        let before = self.node_before_edit(node_id);
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        // Read, merge and write in one transaction: the map read is the map
        // replaced, so neither a concurrent key nor `was` goes stale.
        let queued = conn.immediate_transaction(|conn| {
            let replaced = self.replaced_in_tx(conn, node_id)?;
            let current_meta: Option<String> = decision_nodes::table
                .filter(decision_nodes::id.eq(node_id))
                .select(decision_nodes::metadata_json)
                .first(conn)
                .optional()?
                .ok_or_else(|| node_not_found(node_id))?;

            // Parse existing metadata or create new
            let mut meta: serde_json::Value = current_meta
                .as_ref()
                .and_then(|m| serde_json::from_str(m).ok())
                .unwrap_or_else(|| serde_json::json!({}));

            // Add/update prompt field
            if let Some(obj) = meta.as_object_mut() {
                obj.insert("prompt".to_string(), serde_json::json!(prompt));
            }

            let new_meta = serde_json::to_string(&meta)
                .map_err(|e| DbError::Validation(format!("JSON serialization error: {}", e)))?;

            diesel::update(decision_nodes::table.filter(decision_nodes::id.eq(node_id)))
                .set((
                    decision_nodes::metadata_json.eq(Some(new_meta)),
                    decision_nodes::updated_at.eq(&now),
                ))
                .execute(conn)?;
            let body =
                Self::node_update_body(replaced, Default::default(), one_field("prompt", prompt));
            self.queue_in_tx(conn, body)
        })?;
        drop(conn);

        self.publish_node_edit(node_id, before);
        self.to_log(queued);
        Ok(())
    }

    /// Get all nodes
    pub fn get_all_nodes(&self) -> Result<Vec<DecisionNode>> {
        let mut conn = self.get_conn()?;
        let nodes = decision_nodes::table
            .order(decision_nodes::created_at.asc())
            .load::<DecisionNode>(&mut conn)?;
        Ok(nodes)
    }

    /// Get all edges
    pub fn get_all_edges(&self) -> Result<Vec<DecisionEdge>> {
        let mut conn = self.get_conn()?;
        let edges = decision_edges::table
            .order(decision_edges::created_at.asc())
            .load::<DecisionEdge>(&mut conn)?;
        Ok(edges)
    }

    /// Get a single node by ID
    pub fn get_node(&self, node_id: i32) -> Result<Option<DecisionNode>> {
        let mut conn = self.get_conn()?;
        let node = decision_nodes::table
            .filter(decision_nodes::id.eq(node_id))
            .first::<DecisionNode>(&mut conn)
            .optional()?;
        Ok(node)
    }

    /// Get children of a node (outgoing edges)
    pub fn get_node_children(&self, node_id: i32) -> Result<Vec<DecisionNode>> {
        let mut conn = self.get_conn()?;

        let child_ids: Vec<i32> = decision_edges::table
            .filter(decision_edges::from_node_id.eq(node_id))
            .select(decision_edges::to_node_id)
            .load(&mut conn)?;

        let children = decision_nodes::table
            .filter(decision_nodes::id.eq_any(child_ids))
            .load::<DecisionNode>(&mut conn)?;

        Ok(children)
    }

    /// Get parents of a node (incoming edges)
    pub fn get_node_parents(&self, node_id: i32) -> Result<Vec<DecisionNode>> {
        let mut conn = self.get_conn()?;

        let parent_ids: Vec<i32> = decision_edges::table
            .filter(decision_edges::to_node_id.eq(node_id))
            .select(decision_edges::from_node_id)
            .load(&mut conn)?;

        let parents = decision_nodes::table
            .filter(decision_nodes::id.eq_any(parent_ids))
            .load::<DecisionNode>(&mut conn)?;

        Ok(parents)
    }

    /// Get full graph as JSON-serializable structure
    pub fn get_graph(&self) -> Result<DecisionGraph> {
        let nodes = self.get_all_nodes()?;
        let edges = self.get_all_edges()?;
        let themes = self.get_all_themes().unwrap_or_default();
        let node_themes = self.get_all_node_themes().unwrap_or_default();
        let documents = self.get_node_documents(None, false).unwrap_or_default();
        Ok(DecisionGraph {
            nodes,
            edges,
            config: None,
            themes,
            node_themes,
            documents,
        })
    }

    /// Get full graph with config included (for export)
    pub fn get_graph_with_config(
        &self,
        config: Option<crate::config::Config>,
    ) -> Result<DecisionGraph> {
        let nodes = self.get_all_nodes()?;
        let edges = self.get_all_edges()?;
        let themes = self.get_all_themes().unwrap_or_default();
        let node_themes = self.get_all_node_themes().unwrap_or_default();
        let documents = self.get_node_documents(None, false).unwrap_or_default();
        Ok(DecisionGraph {
            nodes,
            edges,
            config,
            themes,
            node_themes,
            documents,
        })
    }

    // ========================================================================
    // Command Log Operations
    // ========================================================================

    /// Log a command execution
    pub fn log_command(
        &self,
        command: &str,
        description: Option<&str>,
        working_dir: Option<&str>,
    ) -> Result<i32> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        let new_log = NewCommandLog {
            command,
            description,
            working_dir,
            exit_code: None,
            stdout: None,
            stderr: None,
            started_at: &now,
            completed_at: None,
            duration_ms: None,
            decision_node_id: None,
        };

        diesel::insert_into(command_log::table)
            .values(&new_log)
            .execute(&mut conn)?;

        let id: i32 = diesel::select(diesel::dsl::sql::<diesel::sql_types::Integer>(
            "last_insert_rowid()",
        ))
        .first(&mut conn)?;

        Ok(id)
    }

    /// Complete a command log entry
    pub fn complete_command(
        &self,
        log_id: i32,
        exit_code: i32,
        stdout: Option<&str>,
        stderr: Option<&str>,
        duration_ms: i32,
    ) -> Result<()> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        diesel::update(command_log::table.filter(command_log::id.eq(log_id)))
            .set((
                command_log::exit_code.eq(Some(exit_code)),
                command_log::stdout.eq(stdout),
                command_log::stderr.eq(stderr),
                command_log::completed_at.eq(Some(&now)),
                command_log::duration_ms.eq(Some(duration_ms)),
            ))
            .execute(&mut conn)?;

        Ok(())
    }

    /// Get recent commands
    pub fn get_recent_commands(&self, limit: i64) -> Result<Vec<CommandLog>> {
        let mut conn = self.get_conn()?;
        let commands = command_log::table
            .order(command_log::started_at.desc())
            .limit(limit)
            .load::<CommandLog>(&mut conn)?;
        Ok(commands)
    }

    // ========================================================================
    // Roadmap Board Operations
    // ========================================================================

    /// Create a new roadmap item
    pub fn create_roadmap_item(
        &self,
        title: &str,
        description: Option<&str>,
        section: Option<&str>,
        parent_id: Option<i32>,
        checkbox_state: &str,
    ) -> Result<i32> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();
        let change_id = Uuid::new_v4().to_string();

        let new_item = NewRoadmapItem {
            change_id: &change_id,
            title,
            description,
            section,
            parent_id,
            checkbox_state,
            github_issue_number: None,
            github_issue_state: None,
            outcome_node_id: None,
            outcome_change_id: None,
            markdown_line_start: None,
            markdown_line_end: None,
            content_hash: None,
            created_at: &now,
            updated_at: &now,
            last_synced_at: None,
        };

        diesel::insert_into(roadmap_items::table)
            .values(&new_item)
            .execute(&mut conn)?;

        let id: i32 = diesel::select(diesel::dsl::sql::<diesel::sql_types::Integer>(
            "last_insert_rowid()",
        ))
        .first(&mut conn)?;

        Ok(id)
    }

    /// Create a roadmap item with full metadata (for sync operations)
    pub fn create_roadmap_item_full(
        &self,
        change_id: &str,
        title: &str,
        description: Option<&str>,
        section: Option<&str>,
        parent_id: Option<i32>,
        checkbox_state: &str,
        github_issue_number: Option<i32>,
        github_issue_state: Option<&str>,
        outcome_node_id: Option<i32>,
        outcome_change_id: Option<&str>,
        markdown_line_start: Option<i32>,
        markdown_line_end: Option<i32>,
        content_hash: Option<&str>,
    ) -> Result<i32> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        let new_item = NewRoadmapItem {
            change_id,
            title,
            description,
            section,
            parent_id,
            checkbox_state,
            github_issue_number,
            github_issue_state,
            outcome_node_id,
            outcome_change_id,
            markdown_line_start,
            markdown_line_end,
            content_hash,
            created_at: &now,
            updated_at: &now,
            last_synced_at: None,
        };

        diesel::insert_into(roadmap_items::table)
            .values(&new_item)
            .execute(&mut conn)?;

        let id: i32 = diesel::select(diesel::dsl::sql::<diesel::sql_types::Integer>(
            "last_insert_rowid()",
        ))
        .first(&mut conn)?;

        Ok(id)
    }

    /// Get all roadmap items
    pub fn get_all_roadmap_items(&self) -> Result<Vec<RoadmapItem>> {
        let mut conn = self.get_conn()?;
        let items = roadmap_items::table
            .order(roadmap_items::created_at.asc())
            .load::<RoadmapItem>(&mut conn)?;
        Ok(items)
    }

    /// Clear all roadmap items (for refresh)
    pub fn clear_roadmap_items(&self) -> Result<usize> {
        let mut conn = self.get_conn()?;
        let deleted = diesel::delete(roadmap_items::table).execute(&mut conn)?;
        Ok(deleted)
    }

    /// Get roadmap items by section
    pub fn get_roadmap_items_by_section(&self, section: &str) -> Result<Vec<RoadmapItem>> {
        let mut conn = self.get_conn()?;
        let items = roadmap_items::table
            .filter(roadmap_items::section.eq(section))
            .order(roadmap_items::created_at.asc())
            .load::<RoadmapItem>(&mut conn)?;
        Ok(items)
    }

    /// Get a roadmap item by change_id
    pub fn get_roadmap_item_by_change_id(&self, change_id: &str) -> Result<Option<RoadmapItem>> {
        let mut conn = self.get_conn()?;
        let item = roadmap_items::table
            .filter(roadmap_items::change_id.eq(change_id))
            .first::<RoadmapItem>(&mut conn)
            .optional()?;
        Ok(item)
    }

    /// Update a roadmap item's GitHub issue info
    pub fn update_roadmap_item_github(
        &self,
        item_id: i32,
        issue_number: Option<i32>,
        issue_state: Option<&str>,
    ) -> Result<()> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        diesel::update(roadmap_items::table.filter(roadmap_items::id.eq(item_id)))
            .set((
                roadmap_items::github_issue_number.eq(issue_number),
                roadmap_items::github_issue_state.eq(issue_state),
                roadmap_items::updated_at.eq(&now),
            ))
            .execute(&mut conn)?;

        Ok(())
    }

    /// Update a roadmap item's GitHub issue info by finding it by title (first match)
    pub fn update_roadmap_item_github_by_title(
        &self,
        title: &str,
        issue_number: i32,
        issue_state: &str,
    ) -> Result<()> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        let affected = diesel::update(roadmap_items::table.filter(roadmap_items::title.eq(title)))
            .set((
                roadmap_items::github_issue_number.eq(Some(issue_number)),
                roadmap_items::github_issue_state.eq(Some(issue_state)),
                roadmap_items::updated_at.eq(&now),
            ))
            .execute(&mut conn)?;

        if affected == 0 {
            return Err(DbError::Validation(format!(
                "No roadmap item found with title: {}",
                title
            )));
        }

        Ok(())
    }

    /// Update a roadmap item's GitHub issue info by change_id (unique key)
    pub fn update_roadmap_item_github_by_change_id(
        &self,
        change_id: &str,
        issue_number: i32,
        issue_state: &str,
    ) -> Result<()> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        let affected =
            diesel::update(roadmap_items::table.filter(roadmap_items::change_id.eq(change_id)))
                .set((
                    roadmap_items::github_issue_number.eq(Some(issue_number)),
                    roadmap_items::github_issue_state.eq(Some(issue_state)),
                    roadmap_items::updated_at.eq(&now),
                ))
                .execute(&mut conn)?;

        if affected == 0 {
            return Err(DbError::Validation(format!(
                "No roadmap item found with change_id: {}",
                change_id
            )));
        }

        Ok(())
    }

    /// Link a roadmap item to a decision graph outcome node
    pub fn link_roadmap_to_outcome(
        &self,
        item_id: i32,
        outcome_node_id: i32,
        outcome_change_id: &str,
    ) -> Result<()> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        diesel::update(roadmap_items::table.filter(roadmap_items::id.eq(item_id)))
            .set((
                roadmap_items::outcome_node_id.eq(Some(outcome_node_id)),
                roadmap_items::outcome_change_id.eq(Some(outcome_change_id)),
                roadmap_items::updated_at.eq(&now),
            ))
            .execute(&mut conn)?;

        Ok(())
    }

    /// Unlink a roadmap item from its outcome node
    pub fn unlink_roadmap_from_outcome(&self, item_id: i32) -> Result<()> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        diesel::update(roadmap_items::table.filter(roadmap_items::id.eq(item_id)))
            .set((
                roadmap_items::outcome_node_id.eq(None::<i32>),
                roadmap_items::outcome_change_id.eq(None::<String>),
                roadmap_items::updated_at.eq(&now),
            ))
            .execute(&mut conn)?;

        Ok(())
    }

    /// Update a roadmap item's checkbox state
    pub fn update_roadmap_item_checkbox(&self, item_id: i32, checkbox_state: &str) -> Result<()> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        diesel::update(roadmap_items::table.filter(roadmap_items::id.eq(item_id)))
            .set((
                roadmap_items::checkbox_state.eq(checkbox_state),
                roadmap_items::updated_at.eq(&now),
            ))
            .execute(&mut conn)?;

        Ok(())
    }

    /// Update last synced timestamp for a roadmap item
    pub fn update_roadmap_item_synced(&self, item_id: i32) -> Result<()> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        diesel::update(roadmap_items::table.filter(roadmap_items::id.eq(item_id)))
            .set((
                roadmap_items::last_synced_at.eq(Some(&now)),
                roadmap_items::updated_at.eq(&now),
            ))
            .execute(&mut conn)?;

        Ok(())
    }

    /// Get roadmap sync state (returns None if not initialized)
    pub fn get_roadmap_sync_state(&self, roadmap_path: &str) -> Result<Option<RoadmapSyncState>> {
        let mut conn = self.get_conn()?;
        let state = roadmap_sync_state::table
            .filter(roadmap_sync_state::roadmap_path.eq(roadmap_path))
            .first::<RoadmapSyncState>(&mut conn)
            .optional()?;
        Ok(state)
    }

    /// Get or create roadmap sync state
    pub fn get_or_create_sync_state(&self, roadmap_path: &str) -> Result<RoadmapSyncState> {
        let mut conn = self.get_conn()?;

        // Try to find existing state
        let existing = roadmap_sync_state::table
            .filter(roadmap_sync_state::roadmap_path.eq(roadmap_path))
            .first::<RoadmapSyncState>(&mut conn)
            .optional()?;

        if let Some(state) = existing {
            return Ok(state);
        }

        // Create new state
        let new_state = NewRoadmapSyncState {
            roadmap_path,
            roadmap_content_hash: None,
            github_repo: None,
            last_github_sync: None,
            last_markdown_parse: None,
            conflict_count: 0,
        };

        diesel::insert_into(roadmap_sync_state::table)
            .values(&new_state)
            .execute(&mut conn)?;

        roadmap_sync_state::table
            .filter(roadmap_sync_state::roadmap_path.eq(roadmap_path))
            .first::<RoadmapSyncState>(&mut conn)
            .map_err(|e| e.into())
    }

    /// Update sync state after a sync operation
    pub fn update_sync_state(
        &self,
        state_id: i32,
        content_hash: Option<&str>,
        github_repo: Option<&str>,
        github_synced: bool,
        markdown_parsed: bool,
        conflict_count: i32,
    ) -> Result<()> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        let last_github = if github_synced {
            Some(now.clone())
        } else {
            None
        };
        let last_parse = if markdown_parsed { Some(now) } else { None };

        diesel::update(roadmap_sync_state::table.filter(roadmap_sync_state::id.eq(state_id)))
            .set((
                roadmap_sync_state::roadmap_content_hash.eq(content_hash),
                roadmap_sync_state::github_repo.eq(github_repo),
                roadmap_sync_state::last_github_sync.eq(last_github),
                roadmap_sync_state::last_markdown_parse.eq(last_parse),
                roadmap_sync_state::conflict_count.eq(conflict_count),
            ))
            .execute(&mut conn)?;

        Ok(())
    }

    /// Create a conflict record
    pub fn create_roadmap_conflict(
        &self,
        item_change_id: &str,
        conflict_type: &str,
        local_value: Option<&str>,
        remote_value: Option<&str>,
    ) -> Result<i32> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        let new_conflict = NewRoadmapConflict {
            item_change_id,
            conflict_type,
            local_value,
            remote_value,
            resolution: None,
            detected_at: &now,
            resolved_at: None,
        };

        diesel::insert_into(roadmap_conflicts::table)
            .values(&new_conflict)
            .execute(&mut conn)?;

        let id: i32 = diesel::select(diesel::dsl::sql::<diesel::sql_types::Integer>(
            "last_insert_rowid()",
        ))
        .first(&mut conn)?;

        Ok(id)
    }

    /// Get all unresolved conflicts
    pub fn get_unresolved_conflicts(&self) -> Result<Vec<RoadmapConflict>> {
        let mut conn = self.get_conn()?;
        let conflicts = roadmap_conflicts::table
            .filter(roadmap_conflicts::resolution.is_null())
            .order(roadmap_conflicts::detected_at.desc())
            .load::<RoadmapConflict>(&mut conn)?;
        Ok(conflicts)
    }

    /// Resolve a conflict
    pub fn resolve_roadmap_conflict(&self, conflict_id: i32, resolution: &str) -> Result<()> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        diesel::update(roadmap_conflicts::table.filter(roadmap_conflicts::id.eq(conflict_id)))
            .set((
                roadmap_conflicts::resolution.eq(Some(resolution)),
                roadmap_conflicts::resolved_at.eq(Some(&now)),
            ))
            .execute(&mut conn)?;

        Ok(())
    }

    /// Delete a roadmap item by ID
    pub fn delete_roadmap_item(&self, item_id: i32) -> Result<()> {
        let mut conn = self.get_conn()?;
        diesel::delete(roadmap_items::table.filter(roadmap_items::id.eq(item_id)))
            .execute(&mut conn)?;
        Ok(())
    }

    /// Check if a roadmap item is complete (has outcome AND issue closed)
    pub fn check_roadmap_item_completion(&self, item_id: i32) -> Result<(bool, bool, bool)> {
        let mut conn = self.get_conn()?;

        let item = roadmap_items::table
            .filter(roadmap_items::id.eq(item_id))
            .first::<RoadmapItem>(&mut conn)?;

        let has_outcome = item.outcome_change_id.is_some();
        let issue_closed = item.github_issue_state.as_deref() == Some("closed");
        let is_complete = has_outcome && issue_closed;

        Ok((is_complete, has_outcome, issue_closed))
    }

    // ========================================================================
    // GitHub Issue Cache Methods
    // ========================================================================

    /// Cache a GitHub issue for local display in web viewer
    pub fn cache_github_issue(
        &self,
        issue_number: i32,
        repo: &str,
        title: &str,
        body: Option<&str>,
        state: &str,
        html_url: &str,
        created_at: &str,
        updated_at: &str,
    ) -> Result<()> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        // Upsert: delete existing then insert
        diesel::delete(
            github_issue_cache::table
                .filter(github_issue_cache::repo.eq(repo))
                .filter(github_issue_cache::issue_number.eq(issue_number)),
        )
        .execute(&mut conn)?;

        let new_cache = NewGitHubIssueCache {
            issue_number,
            repo,
            title,
            body,
            state,
            html_url,
            created_at,
            updated_at,
            cached_at: &now,
        };

        diesel::insert_into(github_issue_cache::table)
            .values(&new_cache)
            .execute(&mut conn)?;

        Ok(())
    }

    /// Get a cached GitHub issue by repo and number
    pub fn get_cached_issue(
        &self,
        repo: &str,
        issue_number: i32,
    ) -> Result<Option<GitHubIssueCache>> {
        let mut conn = self.get_conn()?;

        let result = github_issue_cache::table
            .filter(github_issue_cache::repo.eq(repo))
            .filter(github_issue_cache::issue_number.eq(issue_number))
            .first::<GitHubIssueCache>(&mut conn)
            .optional()?;

        Ok(result)
    }

    /// Get all cached issues for a repo
    pub fn get_cached_issues_for_repo(&self, repo: &str) -> Result<Vec<GitHubIssueCache>> {
        let mut conn = self.get_conn()?;

        let issues = github_issue_cache::table
            .filter(github_issue_cache::repo.eq(repo))
            .order(github_issue_cache::issue_number.desc())
            .load::<GitHubIssueCache>(&mut conn)?;

        Ok(issues)
    }

    /// Get all cached issues
    pub fn get_all_cached_issues(&self) -> Result<Vec<GitHubIssueCache>> {
        let mut conn = self.get_conn()?;

        let issues = github_issue_cache::table
            .order(github_issue_cache::cached_at.desc())
            .load::<GitHubIssueCache>(&mut conn)?;

        Ok(issues)
    }

    /// Clear cached issues older than a specified duration
    pub fn clear_stale_cache(&self, max_age_hours: i64) -> Result<usize> {
        let mut conn = self.get_conn()?;
        let cutoff = chrono::Local::now() - chrono::Duration::hours(max_age_hours);
        let cutoff_str = cutoff.to_rfc3339();

        let deleted = diesel::delete(
            github_issue_cache::table.filter(github_issue_cache::cached_at.lt(&cutoff_str)),
        )
        .execute(&mut conn)?;

        Ok(deleted)
    }

    // ========================================================================
    // Document Attachment Operations
    // ========================================================================

    /// Attach a document to a node
    pub fn attach_document(
        &self,
        node_id: i32,
        content_hash: &str,
        original_filename: &str,
        storage_filename: &str,
        mime_type: &str,
        file_size: i32,
        description: Option<&str>,
        description_source: &str,
        attached_by: Option<&str>,
    ) -> Result<i32> {
        // Validate node exists and get its change_id
        let node = self
            .get_node(node_id)?
            .ok_or_else(|| DbError::Validation(format!("Node {} not found", node_id)))?;

        let _log = self.hold_log()?;
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();
        let change_id = Uuid::new_v4().to_string();

        let body = self.oplog().map(|_| crate::oplog::OpBody::AttachDocument {
            change_id: change_id.clone(),
            node_change_id: node.change_id.clone(),
            content_hash: content_hash.to_string(),
            original_filename: original_filename.to_string(),
            storage_filename: storage_filename.to_string(),
            mime_type: mime_type.to_string(),
            file_size: file_size as i64,
            description: description.map(str::to_string),
            description_source: server_description_source(description_source),
            attached_by: attached_by.map(str::to_string),
            attached_at: now.clone(),
        });

        let new_doc = NewNodeDocument {
            change_id: &change_id,
            node_id,
            node_change_id: &node.change_id,
            content_hash,
            original_filename,
            storage_filename,
            mime_type,
            file_size,
            description,
            description_source,
            attached_at: &now,
            attached_by,
            detached_at: None,
        };

        let (id, queued) = conn.immediate_transaction(|conn| {
            diesel::insert_into(node_documents::table)
                .values(&new_doc)
                .execute(conn)?;
            let id: i32 = diesel::select(diesel::dsl::sql::<diesel::sql_types::Integer>(
                "last_insert_rowid()",
            ))
            .first(conn)?;
            let queued = self.queue_in_tx(conn, body.into_iter().collect())?;
            Ok::<_, DbError>((id, queued))
        })?;
        drop(conn);
        self.to_log(queued);
        Ok(id)
    }

    /// List documents, optionally filtered by node_id
    pub fn get_node_documents(
        &self,
        node_id: Option<i32>,
        include_detached: bool,
    ) -> Result<Vec<NodeDocument>> {
        let mut conn = self.get_conn()?;

        let mut query = node_documents::table.into_boxed();

        if let Some(nid) = node_id {
            query = query.filter(node_documents::node_id.eq(nid));
        }

        if !include_detached {
            query = query.filter(node_documents::detached_at.is_null());
        }

        let results = query
            .order(node_documents::attached_at.desc())
            .load::<NodeDocument>(&mut conn)?;
        Ok(results)
    }

    /// Get a specific document by ID
    pub fn get_document(&self, doc_id: i32) -> Result<Option<NodeDocument>> {
        let mut conn = self.get_conn()?;
        let result = node_documents::table
            .filter(node_documents::id.eq(doc_id))
            .first::<NodeDocument>(&mut conn)
            .optional()?;
        Ok(result)
    }

    /// Update document description
    pub fn update_document_description(
        &self,
        doc_id: i32,
        description: &str,
        source: &str,
    ) -> Result<()> {
        let _log = self.hold_log()?;
        let mut conn = self.get_conn()?;
        let logged = self.oplog().is_some();
        let queued = conn.immediate_transaction(|conn| {
            let before = node_documents::table
                .filter(node_documents::id.eq(doc_id))
                .first::<NodeDocument>(conn)
                .optional()?;
            diesel::update(node_documents::table.filter(node_documents::id.eq(doc_id)))
                .set((
                    node_documents::description.eq(description),
                    node_documents::description_source.eq(source),
                ))
                .execute(conn)?;
            let body = before
                .filter(|_| logged)
                .map(|d| crate::oplog::OpBody::DescribeDocument {
                    change_id: d.change_id,
                    node_change_id: d.node_change_id,
                    description: Some(description.to_string()),
                    description_source: server_description_source(source),
                    was_description: d.description,
                });
            self.queue_in_tx(conn, body.into_iter().collect())
        })?;
        drop(conn);
        self.to_log(queued);
        Ok(())
    }

    /// Soft-delete (detach) a document
    pub fn detach_document(&self, doc_id: i32) -> Result<()> {
        let _log = self.hold_log()?;
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();
        let logged = self.oplog().is_some();
        let queued = conn.immediate_transaction(|conn| {
            let before = node_documents::table
                .filter(node_documents::id.eq(doc_id))
                .first::<NodeDocument>(conn)
                .optional()?;
            diesel::update(node_documents::table.filter(node_documents::id.eq(doc_id)))
                .set(node_documents::detached_at.eq(&now))
                .execute(conn)?;
            // Detaching twice tells the server nothing new.
            let body = before
                .filter(|d| logged && d.detached_at.is_none())
                .map(|d| crate::oplog::OpBody::DetachDocument {
                    change_id: d.change_id,
                    node_change_id: d.node_change_id,
                });
            self.queue_in_tx(conn, body.into_iter().collect())
        })?;
        drop(conn);
        self.to_log(queued);
        Ok(())
    }

    /// Get all active content hashes (for garbage collection)
    pub fn get_active_content_hashes(&self) -> Result<Vec<String>> {
        let mut conn = self.get_conn()?;
        let results = node_documents::table
            .filter(node_documents::detached_at.is_null())
            .select(node_documents::content_hash)
            .distinct()
            .load::<String>(&mut conn)?;
        Ok(results)
    }

    // ========================================================================
    // Theme Operations
    // ========================================================================

    /// Create a new theme
    pub fn create_theme(&self, name: &str, color: &str, description: Option<&str>) -> Result<i32> {
        if name.trim().is_empty() {
            return Err(DbError::Validation(
                "a theme needs a non-empty name".to_string(),
            ));
        }
        let _mirror = self.require_readable_store()?;
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();
        let change_id = Uuid::new_v4().to_string();
        let normalized_name = name.to_lowercase().replace(' ', "-");

        let new_theme = NewTheme {
            change_id: &change_id,
            name: &normalized_name,
            color,
            description,
            created_at: &now,
            updated_at: &now,
        };

        diesel::insert_into(themes::table)
            .values(&new_theme)
            .execute(&mut conn)?;

        let id: i32 = diesel::select(diesel::dsl::sql::<diesel::sql_types::Integer>(
            "last_insert_rowid()",
        ))
        .first(&mut conn)?;
        drop(conn);

        self.publish_theme_by_id(id);
        Ok(id)
    }

    /// Get a theme by name
    pub fn get_theme_by_name(&self, name: &str) -> Result<Option<Theme>> {
        let mut conn = self.get_conn()?;
        let normalized = name.to_lowercase().replace(' ', "-");

        let result = themes::table
            .filter(themes::name.eq(&normalized))
            .first::<Theme>(&mut conn)
            .optional()?;

        Ok(result)
    }

    /// Get all themes
    pub fn get_all_themes(&self) -> Result<Vec<Theme>> {
        let mut conn = self.get_conn()?;
        let results = themes::table
            .order(themes::name.asc())
            .load::<Theme>(&mut conn)?;
        Ok(results)
    }

    /// Delete a theme by name (also removes all node_themes associations)
    pub fn delete_theme(&self, name: &str) -> Result<bool> {
        let _mirror = self.require_readable_store()?;
        self.require_branch_for_delete("delete the theme")?;
        let mut conn = self.get_conn()?;
        let normalized = name.to_lowercase().replace(' ', "-");

        let theme = themes::table
            .filter(themes::name.eq(&normalized))
            .first::<Theme>(&mut conn)
            .optional()?;

        if let Some(theme) = theme {
            let tagged: Vec<NodeTheme> = node_themes::table
                .filter(node_themes::theme_id.eq(theme.id))
                .load(&mut conn)?;
            // Delete junction entries first
            diesel::delete(node_themes::table.filter(node_themes::theme_id.eq(theme.id)))
                .execute(&mut conn)?;
            // Delete theme
            diesel::delete(themes::table.filter(themes::id.eq(theme.id))).execute(&mut conn)?;
            drop(conn);

            if let Some(store) = self.write_store() {
                for t in &tagged {
                    if let Ok(Some(node)) = self.get_node(t.node_id) {
                        self.tombstone_tag(&node.change_id, &theme.change_id);
                    }
                }
                if let Err(e) = store.tombstone_theme(&theme) {
                    Self::store_warn("could not write theme tombstone", e);
                }
            }
            Ok(true)
        } else {
            Ok(false)
        }
    }

    /// Tag a node with a theme
    pub fn tag_node(&self, node_id: i32, theme_name: &str, source: &str) -> Result<()> {
        let _mirror = self.require_readable_store()?;
        let theme = self.get_theme_by_name(theme_name)?.ok_or_else(|| {
            DbError::Validation(format!(
                "Theme '{}' not found. Create it with: deciduous themes create {}",
                theme_name, theme_name
            ))
        })?;

        // Validate node exists
        let _ = self
            .get_node(node_id)?
            .ok_or_else(|| DbError::Validation(format!("Node {} not found", node_id)))?;

        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        let new_nt = NewNodeTheme {
            node_id,
            theme_id: theme.id,
            source: source.to_string(),
            created_at: now,
        };

        diesel::insert_or_ignore_into(node_themes::table)
            .values(&new_nt)
            .execute(&mut conn)?;
        drop(conn);

        self.publish_tag(node_id, theme.id);
        Ok(())
    }

    /// The theme named `theme_name` and the node `node_id`, or an error
    /// naming whichever is missing. Untagging either one that does not
    /// exist answered "was not tagged", as if the call had been meaningful.
    fn existing_tag_parts(&self, node_id: i32, theme_name: &str) -> Result<Theme> {
        let theme = self
            .get_theme_by_name(theme_name)?
            .ok_or_else(|| DbError::Validation(format!("Theme '{theme_name}' not found")))?;
        self.get_node(node_id)?
            .ok_or_else(|| DbError::Validation(format!("Node {node_id} not found")))?;
        Ok(theme)
    }

    /// Remove a theme from a node. `Ok(false)` only when both exist and the
    /// node simply was not tagged.
    pub fn untag_node(&self, node_id: i32, theme_name: &str) -> Result<bool> {
        let _mirror = self.require_readable_store()?;
        self.require_branch_for_delete("untag it")?;
        let theme = self.existing_tag_parts(node_id, theme_name)?;
        let mut conn = self.get_conn()?;
        let deleted = diesel::delete(
            node_themes::table
                .filter(node_themes::node_id.eq(node_id))
                .filter(node_themes::theme_id.eq(theme.id)),
        )
        .execute(&mut conn)?;
        drop(conn);
        if deleted > 0 {
            if let Ok(Some(node)) = self.get_node(node_id) {
                self.tombstone_tag(&node.change_id, &theme.change_id);
            }
        }
        Ok(deleted > 0)
    }

    /// Confirm a suggested tag (change source from "suggested" to "manual")
    pub fn confirm_tag(&self, node_id: i32, theme_name: &str) -> Result<bool> {
        let _mirror = self.require_readable_store()?;
        let theme = self.existing_tag_parts(node_id, theme_name)?;
        let mut conn = self.get_conn()?;
        let updated = diesel::update(
            node_themes::table
                .filter(node_themes::node_id.eq(node_id))
                .filter(node_themes::theme_id.eq(theme.id)),
        )
        .set(node_themes::source.eq("manual"))
        .execute(&mut conn)?;
        drop(conn);
        if updated > 0 {
            self.publish_tag(node_id, theme.id);
        }
        Ok(updated > 0)
    }

    /// Get themes for a specific node
    pub fn get_node_themes(&self, node_id: i32) -> Result<Vec<Theme>> {
        let mut conn = self.get_conn()?;

        let theme_ids: Vec<i32> = node_themes::table
            .filter(node_themes::node_id.eq(node_id))
            .select(node_themes::theme_id)
            .load(&mut conn)?;

        let results = themes::table
            .filter(themes::id.eq_any(theme_ids))
            .order(themes::name.asc())
            .load::<Theme>(&mut conn)?;

        Ok(results)
    }

    /// Get nodes that have a specific theme
    pub fn get_nodes_by_theme(&self, theme_name: &str) -> Result<Vec<DecisionNode>> {
        let theme = self.get_theme_by_name(theme_name)?;

        if let Some(theme) = theme {
            let mut conn = self.get_conn()?;

            let node_ids: Vec<i32> = node_themes::table
                .filter(node_themes::theme_id.eq(theme.id))
                .select(node_themes::node_id)
                .load(&mut conn)?;

            let results = decision_nodes::table
                .filter(decision_nodes::id.eq_any(node_ids))
                .load::<DecisionNode>(&mut conn)?;

            Ok(results)
        } else {
            Ok(Vec::new())
        }
    }

    /// Get all node-theme associations (for export)
    pub fn get_all_node_themes(&self) -> Result<Vec<NodeTheme>> {
        let mut conn = self.get_conn()?;
        let results = node_themes::table.load::<NodeTheme>(&mut conn)?;
        Ok(results)
    }

    // ========================================================================
    // Sessions — conversation-scoped decision trees
    // ========================================================================

    /// Create a new session with an optional name and root goal node.
    pub fn create_session(&self, name: Option<&str>, root_node_id: Option<i32>) -> Result<i32> {
        if name.is_some_and(|n| n.trim().is_empty()) {
            return Err(DbError::Validation(
                "a session name, when given, must not be empty".to_string(),
            ));
        }
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        let new_session = NewDecisionSession {
            name,
            started_at: &now,
            ended_at: None,
            root_node_id,
            summary: None,
        };

        diesel::insert_into(decision_sessions::table)
            .values(&new_session)
            .execute(&mut conn)?;

        let id: i32 = diesel::select(diesel::dsl::sql::<diesel::sql_types::Integer>(
            "last_insert_rowid()",
        ))
        .first(&mut conn)?;

        Ok(id)
    }

    /// End a session (set ended_at timestamp and optional summary).
    pub fn end_session(&self, session_id: i32, summary: Option<&str>) -> Result<()> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        diesel::update(decision_sessions::table.filter(decision_sessions::id.eq(session_id)))
            .set((
                decision_sessions::ended_at.eq(Some(&now)),
                decision_sessions::summary.eq(summary),
            ))
            .execute(&mut conn)?;

        Ok(())
    }

    /// Reopen an ended session: clear `ended_at`, keep its summary.
    pub fn reopen_session(&self, session_id: i32) -> Result<()> {
        let mut conn = self.get_conn()?;
        let n =
            diesel::update(decision_sessions::table.filter(decision_sessions::id.eq(session_id)))
                .set(decision_sessions::ended_at.eq(None::<String>))
                .execute(&mut conn)?;
        if n == 0 {
            return Err(DbError::Validation(format!(
                "Session {session_id} not found"
            )));
        }
        Ok(())
    }

    /// Associate a node with a session.
    pub fn add_node_to_session(&self, session_id: i32, node_id: i32) -> Result<()> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        diesel::sql_query(
            "INSERT OR IGNORE INTO session_nodes (session_id, node_id, added_at) VALUES (?, ?, ?)",
        )
        .bind::<diesel::sql_types::Integer, _>(session_id)
        .bind::<diesel::sql_types::Integer, _>(node_id)
        .bind::<diesel::sql_types::Text, _>(&now)
        .execute(&mut conn)?;

        Ok(())
    }

    /// Get all nodes in a session.
    pub fn get_session_nodes(&self, session_id: i32) -> Result<Vec<DecisionNode>> {
        let mut conn = self.get_conn()?;

        let node_ids: Vec<i32> = session_nodes::table
            .filter(session_nodes::session_id.eq(session_id))
            .select(session_nodes::node_id)
            .load(&mut conn)?;

        let nodes = decision_nodes::table
            .filter(decision_nodes::id.eq_any(node_ids))
            .order(decision_nodes::created_at.asc())
            .load::<DecisionNode>(&mut conn)?;

        Ok(nodes)
    }

    /// Get a session by ID.
    pub fn get_session(&self, session_id: i32) -> Result<Option<DecisionSession>> {
        let mut conn = self.get_conn()?;
        let session = decision_sessions::table
            .filter(decision_sessions::id.eq(session_id))
            .first::<DecisionSession>(&mut conn)
            .optional()?;
        Ok(session)
    }

    /// Get all sessions (optionally only active ones).
    pub fn get_sessions(&self, active_only: bool) -> Result<Vec<DecisionSession>> {
        let mut conn = self.get_conn()?;
        if active_only {
            let sessions = decision_sessions::table
                .filter(decision_sessions::ended_at.is_null())
                .order(decision_sessions::started_at.desc())
                .load::<DecisionSession>(&mut conn)?;
            Ok(sessions)
        } else {
            let sessions = decision_sessions::table
                .order(decision_sessions::started_at.desc())
                .load::<DecisionSession>(&mut conn)?;
            Ok(sessions)
        }
    }

    // ========================================================================
    // Q&A Interactions
    // ========================================================================

    /// Save a Q&A interaction (user question and Claude response)
    pub fn save_qa_interaction(
        &self,
        user_prompt: &str,
        total_prompt: &str,
        response: &str,
        context_json: Option<&str>,
    ) -> Result<i32> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        let new_interaction = NewQaInteraction {
            user_prompt,
            total_prompt,
            response,
            context_json,
            inserted_at: &now,
            deleted_at: None,
        };

        diesel::insert_into(qa_interactions::table)
            .values(&new_interaction)
            .execute(&mut conn)?;

        let id: i32 = diesel::select(diesel::dsl::sql::<diesel::sql_types::Integer>(
            "last_insert_rowid()",
        ))
        .first(&mut conn)?;

        Ok(id)
    }

    /// Get all Q&A interactions (excluding soft-deleted)
    pub fn get_qa_interactions(&self) -> Result<Vec<QaInteraction>> {
        let mut conn = self.get_conn()?;

        let interactions = qa_interactions::table
            .filter(qa_interactions::deleted_at.is_null())
            .order(qa_interactions::inserted_at.desc())
            .load::<QaInteraction>(&mut conn)?;

        Ok(interactions)
    }

    /// Soft delete a Q&A interaction
    pub fn soft_delete_qa_interaction(&self, id: i32) -> Result<()> {
        let mut conn = self.get_conn()?;
        let now = chrono::Local::now().to_rfc3339();

        diesel::update(qa_interactions::table.filter(qa_interactions::id.eq(id)))
            .set(qa_interactions::deleted_at.eq(Some(&now)))
            .execute(&mut conn)?;

        Ok(())
    }

    /// Search Q&A interactions using FTS5 full-text search
    /// Returns results ranked by relevance (BM25 algorithm)
    pub fn search_qa_interactions(&self, query: &str, limit: i32) -> Result<Vec<QaSearchResult>> {
        let mut conn = self.get_conn()?;

        // Escape single quotes in query for SQL safety
        let safe_query = query.replace('\'', "''");

        // FTS5 query with BM25 ranking and highlighted snippets
        // bm25() returns negative scores (closer to 0 = more relevant)
        // highlight() wraps matches in <mark> tags
        let sql = format!(
            r#"
            SELECT
                qa.id,
                qa.user_prompt,
                qa.total_prompt,
                qa.response,
                qa.context_json,
                qa.inserted_at,
                qa.deleted_at,
                bm25(qa_interactions_fts) as rank,
                highlight(qa_interactions_fts, 0, '<mark>', '</mark>') as snippet_prompt,
                highlight(qa_interactions_fts, 1, '<mark>', '</mark>') as snippet_response
            FROM qa_interactions_fts
            JOIN qa_interactions qa ON qa.id = qa_interactions_fts.rowid
            WHERE qa_interactions_fts MATCH '{}'
            AND qa.deleted_at IS NULL
            ORDER BY rank
            LIMIT {}
            "#,
            safe_query, limit
        );

        let rows: Vec<FtsSearchRow> = diesel::sql_query(sql).load(&mut conn).unwrap_or_default();

        Ok(rows
            .into_iter()
            .map(|row| QaSearchResult {
                interaction: QaInteraction {
                    id: row.id,
                    user_prompt: row.user_prompt,
                    total_prompt: row.total_prompt,
                    response: row.response,
                    context_json: row.context_json,
                    inserted_at: row.inserted_at,
                    deleted_at: row.deleted_at,
                },
                rank: row.rank,
                snippet_prompt: row.snippet_prompt,
                snippet_response: row.snippet_response,
            })
            .collect())
    }

    /// Get a single Q&A interaction by ID
    pub fn get_qa_interaction(&self, id: i32) -> Result<Option<QaInteraction>> {
        let mut conn = self.get_conn()?;

        let result = qa_interactions::table
            .filter(qa_interactions::id.eq(id))
            .filter(qa_interactions::deleted_at.is_null())
            .first::<QaInteraction>(&mut conn)
            .optional()?;

        Ok(result)
    }

    /// Get paginated Q&A interactions (for browsing without search)
    pub fn get_qa_interactions_paginated(
        &self,
        offset: i64,
        limit: i64,
    ) -> Result<Vec<QaInteraction>> {
        let mut conn = self.get_conn()?;

        let interactions = qa_interactions::table
            .filter(qa_interactions::deleted_at.is_null())
            .order(qa_interactions::inserted_at.desc())
            .offset(offset)
            .limit(limit)
            .load::<QaInteraction>(&mut conn)?;

        Ok(interactions)
    }

    /// Get total count of Q&A interactions (for pagination UI)
    pub fn count_qa_interactions(&self) -> Result<i64> {
        let mut conn = self.get_conn()?;

        let count: i64 = qa_interactions::table
            .filter(qa_interactions::deleted_at.is_null())
            .count()
            .get_result(&mut conn)?;

        Ok(count)
    }
}

// ============================================================================
// Q&A Interactions Models
// ============================================================================

/// Insertable Q&A interaction
#[derive(Insertable)]
#[diesel(table_name = qa_interactions)]
pub struct NewQaInteraction<'a> {
    pub user_prompt: &'a str,
    pub total_prompt: &'a str,
    pub response: &'a str,
    pub context_json: Option<&'a str>,
    pub inserted_at: &'a str,
    pub deleted_at: Option<&'a str>,
}

/// Queryable Q&A interaction
#[derive(Queryable, Selectable, Debug, Clone, serde::Serialize)]
#[cfg_attr(feature = "ts-rs", derive(TS))]
#[cfg_attr(feature = "ts-rs", ts(export))]
#[diesel(table_name = qa_interactions)]
pub struct QaInteraction {
    pub id: i32,
    pub user_prompt: String,
    pub total_prompt: String,
    pub response: String,
    pub context_json: Option<String>,
    pub inserted_at: String,
    pub deleted_at: Option<String>,
}

/// Q&A search result with FTS5 ranking and highlighted snippets
#[derive(Debug, Clone, serde::Serialize)]
#[cfg_attr(feature = "ts-rs", derive(TS))]
#[cfg_attr(feature = "ts-rs", ts(export))]
pub struct QaSearchResult {
    pub interaction: QaInteraction,
    pub rank: f64,
    pub snippet_prompt: String,
    pub snippet_response: String,
}

// ============================================================================
// Node Document Models
// ============================================================================

/// Insertable node document attachment
#[derive(Insertable)]
#[diesel(table_name = node_documents)]
pub struct NewNodeDocument<'a> {
    pub change_id: &'a str,
    pub node_id: i32,
    pub node_change_id: &'a str,
    pub content_hash: &'a str,
    pub original_filename: &'a str,
    pub storage_filename: &'a str,
    pub mime_type: &'a str,
    pub file_size: i32,
    pub description: Option<&'a str>,
    pub description_source: &'a str,
    pub attached_at: &'a str,
    pub attached_by: Option<&'a str>,
    pub detached_at: Option<&'a str>,
}

/// The server's name for how a description was written (`none`, `user`,
/// `ai`): the CLI stores "manual" for a description typed in.
fn server_description_source(source: &str) -> String {
    match source {
        "none" | "user" | "ai" => source.to_string(),
        _ => "user".to_string(),
    }
}

/// Queryable node document attachment
#[derive(Queryable, Selectable, Debug, Clone, serde::Serialize)]
#[cfg_attr(feature = "ts-rs", derive(TS))]
#[cfg_attr(feature = "ts-rs", ts(export))]
#[diesel(table_name = node_documents)]
pub struct NodeDocument {
    pub id: i32,
    pub change_id: String,
    pub node_id: i32,
    pub node_change_id: String,
    pub content_hash: String,
    pub original_filename: String,
    pub storage_filename: String,
    pub mime_type: String,
    pub file_size: i32,
    pub description: Option<String>,
    pub description_source: String,
    pub attached_at: String,
    pub attached_by: Option<String>,
    pub detached_at: Option<String>,
}

// ============================================================================
// Theme Models
// ============================================================================

/// Insertable theme
#[derive(Insertable)]
#[diesel(table_name = themes)]
pub struct NewTheme<'a> {
    pub change_id: &'a str,
    pub name: &'a str,
    pub color: &'a str,
    pub description: Option<&'a str>,
    pub created_at: &'a str,
    pub updated_at: &'a str,
}

/// Queryable theme
#[derive(Queryable, Selectable, Debug, Clone, serde::Serialize)]
#[cfg_attr(feature = "ts-rs", derive(TS))]
#[cfg_attr(feature = "ts-rs", ts(export))]
#[diesel(table_name = themes)]
pub struct Theme {
    pub id: i32,
    pub change_id: String,
    pub name: String,
    pub color: String,
    pub description: Option<String>,
    pub created_at: String,
    pub updated_at: String,
}

/// Insertable node-theme association
#[derive(Insertable)]
#[diesel(table_name = node_themes)]
pub struct NewNodeTheme {
    pub node_id: i32,
    pub theme_id: i32,
    pub source: String,
    pub created_at: String,
}

/// Queryable node-theme association
#[derive(Queryable, Selectable, Debug, Clone, serde::Serialize)]
#[cfg_attr(feature = "ts-rs", derive(TS))]
#[cfg_attr(feature = "ts-rs", ts(export))]
#[diesel(table_name = node_themes)]
pub struct NodeTheme {
    pub node_id: i32,
    pub theme_id: i32,
    pub source: String,
    pub created_at: String,
}

// ============================================================================
// Additional Types
// ============================================================================

/// Summary statistics from the database (kept for compatibility)
#[derive(Debug, Clone, serde::Serialize)]
pub struct DbSummary {
    pub total_nodes: i32,
    pub total_edges: i32,
}

/// Alias for backwards compatibility
pub type DbRecord = DecisionNode;

/// Full decision graph for serialization
#[derive(Debug, Clone, serde::Serialize)]
pub struct DecisionGraph {
    pub nodes: Vec<DecisionNode>,
    pub edges: Vec<DecisionEdge>,
    /// Optional config from .deciduous/config.toml (for external repo links, etc.)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub config: Option<crate::config::Config>,
    /// Theme definitions
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub themes: Vec<Theme>,
    /// Node-theme associations
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub node_themes: Vec<NodeTheme>,
    /// Document attachments
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub documents: Vec<NodeDocument>,
}

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

    #[test]
    fn t6_link_refuses_an_edge_whose_reverse_exists() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");
        let db = Database::new(db_path.to_str().unwrap()).unwrap();
        let a = db.create_node("goal", "a", None, None, None).unwrap();
        let b = db.create_node("decision", "b", None, None, None).unwrap();
        db.create_edge(a, b, "leads_to", None).unwrap();
        let err = db
            .create_edge(b, a, "leads_to", None)
            .unwrap_err()
            .to_string();
        assert!(err.contains("each other's parent"), "{err}");
        // A borrow is not the tree.
        db.create_edge(b, a, "took_from", None).unwrap();
    }

    // === build_metadata_json Tests ===

    #[test]
    fn test_build_metadata_empty() {
        let result = build_metadata_json(None, None, None, None, None);
        assert!(result.is_none());
    }

    #[test]
    fn test_build_metadata_confidence_only() {
        let result = build_metadata_json(Some(85), None, None, None, None);
        assert!(result.is_some());
        let json: serde_json::Value = serde_json::from_str(&result.unwrap()).unwrap();
        assert_eq!(json.get("confidence").unwrap(), 85);
    }

    #[test]
    fn test_build_metadata_confidence_clamped() {
        let result = build_metadata_json(Some(150), None, None, None, None);
        let json: serde_json::Value = serde_json::from_str(&result.unwrap()).unwrap();
        // Should be clamped to 100
        assert_eq!(json.get("confidence").unwrap(), 100);
    }

    #[test]
    fn test_build_metadata_commit() {
        let result = build_metadata_json(None, Some("abc123"), None, None, None);
        let json: serde_json::Value = serde_json::from_str(&result.unwrap()).unwrap();
        assert_eq!(json.get("commit").unwrap(), "abc123");
    }

    #[test]
    fn test_build_metadata_prompt() {
        let result = build_metadata_json(None, None, Some("User asked: do X"), None, None);
        let json: serde_json::Value = serde_json::from_str(&result.unwrap()).unwrap();
        assert_eq!(json.get("prompt").unwrap(), "User asked: do X");
    }

    #[test]
    fn test_build_metadata_files() {
        let result = build_metadata_json(None, None, None, Some("a.rs, b.rs, c.rs"), None);
        let json: serde_json::Value = serde_json::from_str(&result.unwrap()).unwrap();
        let files = json.get("files").unwrap().as_array().unwrap();
        assert_eq!(files.len(), 3);
        assert_eq!(files[0], "a.rs");
        assert_eq!(files[1], "b.rs");
        assert_eq!(files[2], "c.rs");
    }

    #[test]
    fn test_build_metadata_branch() {
        let result = build_metadata_json(None, None, None, None, Some("feature-x"));
        let json: serde_json::Value = serde_json::from_str(&result.unwrap()).unwrap();
        assert_eq!(json.get("branch").unwrap(), "feature-x");
    }

    #[test]
    fn test_build_metadata_all_fields() {
        let result = build_metadata_json(
            Some(90),
            Some("def456"),
            Some("User prompt"),
            Some("x.rs"),
            Some("main"),
        );
        let json: serde_json::Value = serde_json::from_str(&result.unwrap()).unwrap();

        assert_eq!(json.get("confidence").unwrap(), 90);
        assert_eq!(json.get("commit").unwrap(), "def456");
        assert_eq!(json.get("prompt").unwrap(), "User prompt");
        assert_eq!(json.get("branch").unwrap(), "main");
        assert!(json.get("files").unwrap().as_array().is_some());
    }

    // === DecisionSchema Tests ===

    #[test]
    fn test_schema_version_string() {
        let schema = DecisionSchema {
            major: 1,
            minor: 2,
            patch: 3,
            name: "test",
            features: &[],
        };
        assert_eq!(schema.version_string(), "1.2.3");
    }

    #[test]
    fn test_schema_compatibility_same_major() {
        let schema1 = DecisionSchema {
            major: 1,
            minor: 0,
            patch: 0,
            name: "test",
            features: &[],
        };
        let schema2 = DecisionSchema {
            major: 1,
            minor: 5,
            patch: 3,
            name: "test",
            features: &[],
        };
        assert!(schema1.is_compatible_with(&schema2));
    }

    #[test]
    fn test_schema_incompatibility_different_major() {
        let schema1 = DecisionSchema {
            major: 1,
            minor: 0,
            patch: 0,
            name: "test",
            features: &[],
        };
        let schema2 = DecisionSchema {
            major: 2,
            minor: 0,
            patch: 0,
            name: "test",
            features: &[],
        };
        assert!(!schema1.is_compatible_with(&schema2));
    }

    #[test]
    fn test_schema_is_newer_than() {
        let old = DecisionSchema {
            major: 1,
            minor: 0,
            patch: 0,
            name: "test",
            features: &[],
        };
        let new = DecisionSchema {
            major: 1,
            minor: 1,
            patch: 0,
            name: "test",
            features: &[],
        };
        assert!(new.is_newer_than(&old));
        assert!(!old.is_newer_than(&new));
        assert!(!old.is_newer_than(&old));
    }

    // === Current Schema Tests ===

    #[test]
    fn test_current_schema() {
        assert_eq!(CURRENT_SCHEMA.major, 1);
        assert_eq!(CURRENT_SCHEMA.name, "decision-graph");
        assert!(CURRENT_SCHEMA.features.contains(&"decision_nodes"));
        assert!(CURRENT_SCHEMA.features.contains(&"decision_edges"));
    }

    // === update_node_commit Tests ===

    #[test]
    fn test_update_node_commit_new_metadata() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");
        let db = Database::new(db_path.to_str().unwrap()).unwrap();

        // Create a node without metadata
        let node_id = db
            .create_node("action", "Test action", None, None, None)
            .unwrap();

        // Update with commit
        db.update_node_commit(node_id, "abc123def456").unwrap();

        // Verify
        let nodes = db.get_all_nodes().unwrap();
        let node = nodes.iter().find(|n| n.id == node_id).unwrap();
        let meta: serde_json::Value =
            serde_json::from_str(node.metadata_json.as_ref().unwrap()).unwrap();
        assert_eq!(meta.get("commit").unwrap(), "abc123def456");
    }

    #[test]
    fn test_update_node_commit_preserves_existing_metadata() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");
        let db = Database::new(db_path.to_str().unwrap()).unwrap();

        // Create a node with existing metadata (confidence and branch)
        let node_id = db
            .create_node_full(
                "action",
                "Test action",
                None,
                Some(85),
                None,
                None,
                None,
                Some("feature-x"),
                None,
            )
            .unwrap();

        // Update with commit
        db.update_node_commit(node_id, "def789").unwrap();

        // Verify commit was added and other fields preserved
        let nodes = db.get_all_nodes().unwrap();
        let node = nodes.iter().find(|n| n.id == node_id).unwrap();
        let meta: serde_json::Value =
            serde_json::from_str(node.metadata_json.as_ref().unwrap()).unwrap();

        assert_eq!(meta.get("commit").unwrap(), "def789");
        assert_eq!(meta.get("confidence").unwrap(), 85);
        assert_eq!(meta.get("branch").unwrap(), "feature-x");
    }

    #[test]
    fn test_update_node_commit_overwrites_existing_commit() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");
        let db = Database::new(db_path.to_str().unwrap()).unwrap();

        // Create a node with an existing commit
        let node_id = db
            .create_node_full(
                "outcome",
                "Test outcome",
                None,
                None,
                Some("old_commit_hash"),
                None,
                None,
                None,
                None,
            )
            .unwrap();

        // Update with new commit
        db.update_node_commit(node_id, "new_commit_hash").unwrap();

        // Verify commit was overwritten
        let nodes = db.get_all_nodes().unwrap();
        let node = nodes.iter().find(|n| n.id == node_id).unwrap();
        let meta: serde_json::Value =
            serde_json::from_str(node.metadata_json.as_ref().unwrap()).unwrap();

        assert_eq!(meta.get("commit").unwrap(), "new_commit_hash");
    }

    // === Theme Tests ===

    #[test]
    fn test_create_and_get_theme() {
        let dir = tempfile::tempdir().unwrap();
        let db = Database::new(dir.path().join("test.db").to_str().unwrap()).unwrap();

        let id = db
            .create_theme("security", "#ef4444", Some("Security concerns"))
            .unwrap();
        assert!(id > 0);

        let theme = db.get_theme_by_name("security").unwrap().unwrap();
        assert_eq!(theme.name, "security");
        assert_eq!(theme.color, "#ef4444");
        assert_eq!(theme.description.as_deref(), Some("Security concerns"));
    }

    #[test]
    fn test_theme_name_normalization() {
        let dir = tempfile::tempdir().unwrap();
        let db = Database::new(dir.path().join("test.db").to_str().unwrap()).unwrap();

        db.create_theme("Technical Debt", "#666", None).unwrap();
        let theme = db.get_theme_by_name("technical-debt").unwrap();
        assert!(theme.is_some());
        assert_eq!(theme.unwrap().name, "technical-debt");
    }

    #[test]
    fn test_list_themes() {
        let dir = tempfile::tempdir().unwrap();
        let db = Database::new(dir.path().join("test.db").to_str().unwrap()).unwrap();

        db.create_theme("alpha", "#111", None).unwrap();
        db.create_theme("beta", "#222", None).unwrap();
        db.create_theme("gamma", "#333", None).unwrap();

        let themes = db.get_all_themes().unwrap();
        assert_eq!(themes.len(), 3);
        // Should be sorted by name
        assert_eq!(themes[0].name, "alpha");
        assert_eq!(themes[1].name, "beta");
        assert_eq!(themes[2].name, "gamma");
    }

    #[test]
    fn test_delete_theme() {
        let dir = tempfile::tempdir().unwrap();
        let db = Database::new(dir.path().join("test.db").to_str().unwrap()).unwrap();

        db.create_theme("temp", "#000", None).unwrap();
        assert!(db.delete_theme("temp").unwrap());
        assert!(db.get_theme_by_name("temp").unwrap().is_none());
        // Deleting non-existent returns false
        assert!(!db.delete_theme("temp").unwrap());
    }

    #[test]
    fn test_tag_and_untag_node() {
        let dir = tempfile::tempdir().unwrap();
        let db = Database::new(dir.path().join("test.db").to_str().unwrap()).unwrap();

        let node_id = db
            .create_node("goal", "Improve perf", None, None, None)
            .unwrap();
        db.create_theme("performance", "#10b981", None).unwrap();

        db.tag_node(node_id, "performance", "manual").unwrap();

        let themes = db.get_node_themes(node_id).unwrap();
        assert_eq!(themes.len(), 1);
        assert_eq!(themes[0].name, "performance");

        let nodes = db.get_nodes_by_theme("performance").unwrap();
        assert_eq!(nodes.len(), 1);
        assert_eq!(nodes[0].id, node_id);

        assert!(db.untag_node(node_id, "performance").unwrap());
        let themes = db.get_node_themes(node_id).unwrap();
        assert!(themes.is_empty());
    }

    #[test]
    fn test_delete_theme_cascades_to_tags() {
        let dir = tempfile::tempdir().unwrap();
        let db = Database::new(dir.path().join("test.db").to_str().unwrap()).unwrap();

        let node_id = db.create_node("goal", "Test", None, None, None).unwrap();
        db.create_theme("temp", "#000", None).unwrap();
        db.tag_node(node_id, "temp", "manual").unwrap();

        db.delete_theme("temp").unwrap();
        let themes = db.get_node_themes(node_id).unwrap();
        assert!(themes.is_empty());
    }

    #[test]
    fn test_confirm_tag() {
        let dir = tempfile::tempdir().unwrap();
        let db = Database::new(dir.path().join("test.db").to_str().unwrap()).unwrap();

        let node_id = db.create_node("goal", "Test", None, None, None).unwrap();
        db.create_theme("ux", "#3b82f6", None).unwrap();
        db.tag_node(node_id, "ux", "suggested").unwrap();

        // Check source is "suggested"
        let all_nt = db.get_all_node_themes().unwrap();
        assert_eq!(all_nt[0].source, "suggested");

        // Confirm it
        assert!(db.confirm_tag(node_id, "ux").unwrap());

        let all_nt = db.get_all_node_themes().unwrap();
        assert_eq!(all_nt[0].source, "manual");
    }

    #[test]
    fn test_theme_in_graph_export() {
        let dir = tempfile::tempdir().unwrap();
        let db = Database::new(dir.path().join("test.db").to_str().unwrap()).unwrap();

        let node_id = db.create_node("goal", "Test", None, None, None).unwrap();
        db.create_theme("ux", "#3b82f6", None).unwrap();
        db.tag_node(node_id, "ux", "manual").unwrap();

        let graph = db.get_graph().unwrap();
        assert_eq!(graph.themes.len(), 1);
        assert_eq!(graph.node_themes.len(), 1);

        // Verify serialization includes themes
        let json = serde_json::to_string(&graph).unwrap();
        assert!(json.contains("\"themes\""));
        assert!(json.contains("\"ux\""));
    }

    // === Document Tests ===

    #[test]
    fn test_attach_and_list_documents() {
        let dir = tempfile::tempdir().unwrap();
        let db = Database::new(dir.path().join("test.db").to_str().unwrap()).unwrap();

        let node_id = db
            .create_node("goal", "Test goal", None, None, None)
            .unwrap();

        let doc_id = db
            .attach_document(
                node_id,
                "abc123hash",
                "report.pdf",
                "report.pdf.abc12345",
                "application/pdf",
                1024,
                Some("A test report"),
                "manual",
                None,
            )
            .unwrap();
        assert!(doc_id > 0);

        let docs = db.get_node_documents(Some(node_id), false).unwrap();
        assert_eq!(docs.len(), 1);
        assert_eq!(docs[0].original_filename, "report.pdf");
        assert_eq!(docs[0].description.as_deref(), Some("A test report"));
    }

    #[test]
    fn test_detach_document() {
        let dir = tempfile::tempdir().unwrap();
        let db = Database::new(dir.path().join("test.db").to_str().unwrap()).unwrap();

        let node_id = db.create_node("goal", "Test", None, None, None).unwrap();
        let doc_id = db
            .attach_document(
                node_id,
                "hash1",
                "file.txt",
                "file.txt.hash1234",
                "text/plain",
                100,
                None,
                "none",
                None,
            )
            .unwrap();

        // Detach
        db.detach_document(doc_id).unwrap();

        // Should not appear in default list
        let docs = db.get_node_documents(Some(node_id), false).unwrap();
        assert!(docs.is_empty());

        // Should appear with include_detached
        let docs = db.get_node_documents(Some(node_id), true).unwrap();
        assert_eq!(docs.len(), 1);
        assert!(docs[0].detached_at.is_some());
    }

    #[test]
    fn test_update_document_description() {
        let dir = tempfile::tempdir().unwrap();
        let db = Database::new(dir.path().join("test.db").to_str().unwrap()).unwrap();

        let node_id = db.create_node("goal", "Test", None, None, None).unwrap();
        let doc_id = db
            .attach_document(
                node_id,
                "hash1",
                "file.txt",
                "file.txt.hash1234",
                "text/plain",
                100,
                None,
                "none",
                None,
            )
            .unwrap();

        db.update_document_description(doc_id, "Updated desc", "manual")
            .unwrap();

        let doc = db.get_document(doc_id).unwrap().unwrap();
        assert_eq!(doc.description.as_deref(), Some("Updated desc"));
        assert_eq!(doc.description_source, "manual");
    }

    #[test]
    fn test_documents_in_graph_export() {
        let dir = tempfile::tempdir().unwrap();
        let db = Database::new(dir.path().join("test.db").to_str().unwrap()).unwrap();

        let node_id = db.create_node("goal", "Test", None, None, None).unwrap();
        db.attach_document(
            node_id,
            "hash1",
            "file.txt",
            "file.txt.hash1234",
            "text/plain",
            100,
            Some("A file"),
            "manual",
            None,
        )
        .unwrap();

        let graph = db.get_graph().unwrap();
        assert_eq!(graph.documents.len(), 1);

        let json = serde_json::to_string(&graph).unwrap();
        assert!(json.contains("\"documents\""));
        assert!(json.contains("file.txt"));
    }

    // ------------------------------------------------------------------
    // resolve_node_ref
    // ------------------------------------------------------------------

    #[test]
    fn test_resolve_node_ref_accepts_ids_and_change_id_prefixes() {
        let dir = tempfile::tempdir().unwrap();
        let db = Database::new(dir.path().join("test.db").to_str().unwrap()).unwrap();
        // Fixed change_ids: a random UUID's first 8 hex characters are all
        // digits about 2% of the time, which this test does not want to
        // exercise by accident (the git_sync integration test does it on
        // purpose).
        let a = db
            .create_node_with_change_id(
                "a1b2c3d4-1111-4111-8111-111111111111",
                "goal",
                "A",
                None,
                None,
                None,
                None,
                None,
                None,
            )
            .unwrap();
        let b = db
            .create_node_with_change_id(
                "f9e8d7c6-2222-4222-8222-222222222222",
                "goal",
                "B",
                None,
                None,
                None,
                None,
                None,
                None,
            )
            .unwrap();
        let a_cid = db.get_node(a).unwrap().unwrap().change_id;

        assert_eq!(db.resolve_node_ref(&a.to_string()).unwrap(), a);
        assert_eq!(db.resolve_node_ref(&format!("#{}", b)).unwrap(), b);
        assert_eq!(db.resolve_node_ref(&a_cid).unwrap(), a);
        assert_eq!(db.resolve_node_ref(&a_cid[..8]).unwrap(), a);

        let err = db.resolve_node_ref("zz").unwrap_err().to_string();
        assert!(err.contains("not a node id"), "{err}");
        // Four or more digits that are neither a local id nor a change_id
        // prefix are no node, so the CLI can go on to try them as a server
        // id; a zero-padded one is not silently read as a local id.
        assert!(matches!(
            db.resolve_node_ref("99999"),
            Err(DbError::NoSuchNode(_))
        ));
        assert!(matches!(
            db.resolve_node_ref(&format!("000{a}")),
            Err(DbError::NoSuchNode(_))
        ));
        let err = db
            .resolve_node_ref("ffffffff-0000")
            .unwrap_err()
            .to_string();
        assert!(err.contains("No node"), "{err}");
    }

    #[test]
    fn test_resolve_node_ref_reports_ambiguity() {
        let dir = tempfile::tempdir().unwrap();
        let db = Database::new(dir.path().join("test.db").to_str().unwrap()).unwrap();
        db.create_node_with_change_id(
            "abcd1111-x",
            "goal",
            "One",
            None,
            None,
            None,
            None,
            None,
            None,
        )
        .unwrap();
        db.create_node_with_change_id(
            "abcd2222-y",
            "goal",
            "Two",
            None,
            None,
            None,
            None,
            None,
            None,
        )
        .unwrap();
        let err = db.resolve_node_ref("abcd").unwrap_err().to_string();
        assert!(err.contains("several"), "{err}");
        assert!(err.contains("One") && err.contains("Two"), "{err}");
        assert_eq!(db.resolve_node_ref("abcd1").unwrap(), 1);
    }
}