beads_rust 0.7.6

//! Self-healing for tracker databases left behind on an older schema.
//!
//! Ordinary commands refuse a database whose schema predates this binary.
//! That refusal exists to protect data that lives only in the database: an
//! edit or creation that was never flushed to `issues.jsonl`. For the common
//! case — a stale cache whose every issue is already represented in the
//! git-tracked JSONL — the refusal protects nothing and blocks every command,
//! including the reviewed migration's own prerequisite (`plan` refuses the
//! unversioned schema-0 databases written by pre-versioning `br` and Go `bd`).
//!
//! This module separates the two cases:
//!
//! 1. [`audit_stale_database`] reads the old database schema-agnostically
//!    (plain `SELECT`s over whatever columns exist) and compares every issue
//!    row against the JSONL snapshot. An issue is *represented* when the JSONL
//!    carries it as a tombstone, carries equivalent content, or carries a
//!    copy at least as new while the database never marked the row dirty.
//!    Anything else — absent from the JSONL, an unflushed (dirty) edit that
//!    differs, a row newer than its JSONL copy, or a row that cannot be read —
//!    is database-only data.
//! 2. [`heal_stale_schema`] first attempts the reviewed in-place migration for
//!    supported source schemas (lossless, recovery bundle, undo). This preserves
//!    raw SQL values without requiring that they can be encoded in JSONL. Only
//!    a JSONL-rebuild fallback needs the strict audit: with no database-only data
//!    it rebuilds from the audited snapshot with the old family retained in
//!    `.br_recovery`. With database-only data, automatic rebuilding refuses and
//!    names the explicit `br doctor migrate-schema heal`, which does the same
//!    upgrade but re-adds the database-only issues as unflushed changes so the
//!    next flush exports them. A malformed issue or relation is never silently
//!    omitted from that audit. A rebuild that cannot restore an unreadable
//!    payload requires explicit `--discard-db-only` authorization; a retained
//!    backup alone is not permission to remove data from the live tracker.
//!    That authorization applies only to unrestorable payloads and orphan
//!    relations: recoverable database-only issues are always re-added.

mod rebuild_preflight;

use std::collections::{HashMap, HashSet};
use std::io::BufRead;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use chrono::{DateTime, NaiveDateTime, TimeZone, Utc};
use fsqlite_types::SqliteValue;
use serde::Serialize;
use serde_json::{Map, Value};

use crate::config;
use crate::error::{BeadsError, Result};
use crate::franken_sync::Connection;
use crate::franken_sync::compat::{OpenFlags, open_with_flags};
use crate::model::{Comment, Dependency, Issue, Status};
use crate::storage::schema::{CURRENT_SCHEMA_VERSION, REVIEWED_MIGRATION_SOURCE_VERSIONS};
use crate::sync::{DatabaseFamilyWriteLock, JsonlSourceSnapshot, PreservedIssue};

/// Most database-only issue ids named in a refusal before eliding the rest.
const DB_ONLY_IDS_SHOWN: usize = 10;

/// The explicit command every refusal names.
pub const HEAL_COMMAND: &str = "br doctor migrate-schema heal";

/// Why an issue row counts as database-only.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DbOnlyReason {
    /// The JSONL has no record with this id.
    AbsentFromJsonl,
    /// The row is marked dirty (never flushed) and differs from the JSONL.
    UnflushedEdit,
    /// The row is not marked dirty but is newer than, and differs from, the JSONL.
    NewerThanJsonl,
    /// The row could not be decoded, so its representation cannot be proven.
    Unreadable,
}

impl DbOnlyReason {
    const fn describe(self) -> &'static str {
        match self {
            Self::AbsentFromJsonl => "absent from JSONL",
            Self::UnflushedEdit => "unflushed edit",
            Self::NewerThanJsonl => "newer than JSONL",
            Self::Unreadable => "unreadable row",
        }
    }
}

/// One database-only issue found by the audit.
#[derive(Debug, Clone, Serialize)]
pub struct DbOnlyIssue {
    pub id: String,
    pub reason: DbOnlyReason,
    /// Whether the explicit heal re-adds this row after the rebuild. Unflushed
    /// edits that a newer JSONL edit already superseded stay in the backup
    /// only, matching the automatic-rebuild rule of GitHub #394.
    pub restorable: bool,
}

/// Relation rows (`labels`, `dependencies`, `comments`) whose owning issue row
/// is gone, typically left by deletes that older releases ran without
/// foreign-key enforcement.
#[derive(Debug, Clone, Serialize)]
pub struct OrphanRelationRows {
    pub table: &'static str,
    pub rows: usize,
}

/// Result of comparing a stale database with the JSONL snapshot.
#[derive(Debug, Clone, Serialize)]
pub struct StaleSchemaAudit {
    pub from_version: u32,
    pub to_version: u32,
    pub db_issue_count: usize,
    pub jsonl_issue_count: usize,
    pub db_only: Vec<DbOnlyIssue>,
    /// Well-formed relation rows without an owning issue row. A reviewed
    /// in-place migration keeps them; a JSONL rebuild cannot carry them, so it
    /// requires explicit `--discard-db-only`.
    pub orphan_relations: Vec<OrphanRelationRows>,
    /// A reason the whole database cannot be audited (not a beads tracker,
    /// no JSONL, unparseable JSONL). Always blocks a JSONL rebuild, but cannot
    /// prevent a reviewed lossless migration that never imports JSONL.
    pub unprovable: Option<String>,
    #[serde(skip)]
    restorable: Vec<PreservedIssue>,
}

impl StaleSchemaAudit {
    /// True when every database issue is provably represented in the JSONL.
    #[must_use]
    pub fn is_clean(&self) -> bool {
        self.unprovable.is_none() && self.db_only.is_empty()
    }

    /// True when the database holds relation rows without an owning issue.
    #[must_use]
    pub fn has_orphan_relations(&self) -> bool {
        !self.orphan_relations.is_empty()
    }

    /// Whether an automatic heal may be attempted. Reviewed migrations have
    /// their own raw-row preservation checks and do not need a JSONL audit.
    /// This is not permission to rebuild after a migration fails.
    #[must_use]
    pub fn automatic_heal_allowed(&self) -> bool {
        REVIEWED_MIGRATION_SOURCE_VERSIONS.contains(&self.from_version)
            || self.automatic_rebuild_allowed()
    }

    /// A rebuild, unlike a reviewed migration, must not drop database-only
    /// issues or orphan relations without an explicit operator decision.
    fn automatic_rebuild_allowed(&self) -> bool {
        self.is_clean() && !self.has_orphan_relations()
    }

    fn orphan_summary(&self) -> String {
        self.orphan_relations
            .iter()
            .map(|orphans| format!("{} {} row(s)", orphans.rows, orphans.table))
            .collect::<Vec<_>>()
            .join(", ")
    }

    fn db_only_summary(&self) -> String {
        let mut shown: Vec<String> = self
            .db_only
            .iter()
            .take(DB_ONLY_IDS_SHOWN)
            .map(|issue| format!("{} ({})", issue.id, issue.reason.describe()))
            .collect();
        if self.db_only.len() > DB_ONLY_IDS_SHOWN {
            shown.push(format!(
                "and {} more",
                self.db_only.len() - DB_ONLY_IDS_SHOWN
            ));
        }
        shown.join(", ")
    }

    /// Explain which live issues a rebuild will restore, independently of
    /// permission to leave unreadable payloads and orphans in the backup.
    fn restoration_notice(&self) -> String {
        if self.restorable.is_empty() {
            return "No recoverable database-only issues were found to re-add".to_owned();
        }
        let mut ids: Vec<&str> = self
            .restorable
            .iter()
            .map(|entry| entry.issue.id.as_str())
            .collect();
        ids.sort_unstable();
        let shown = ids
            .iter()
            .take(DB_ONLY_IDS_SHOWN)
            .copied()
            .collect::<Vec<_>>()
            .join(", ");
        let remaining = if ids.len() > DB_ONLY_IDS_SHOWN {
            format!(", and {} more", ids.len() - DB_ONLY_IDS_SHOWN)
        } else {
            String::new()
        };
        format!(
            "{} recoverable database-only issue(s) ({shown}{remaining}) will still be \
             re-added as unflushed changes, including with --discard-db-only",
            ids.len()
        )
    }
}

/// How [`heal_stale_schema`] was invoked.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HealMode {
    /// Ordinary command startup: prefer a reviewed lossless migration;
    /// rebuild only when the audit is clean and no orphan relations are lost.
    Automatic,
    /// `br doctor migrate-schema heal`: always re-add restorable database-only
    /// issues. `discard_db_only` authorizes backup-only retention of unreadable
    /// payloads and orphan relations, not omission of recoverable issues.
    Explicit { discard_db_only: bool },
}

/// What the heal did.
#[derive(Debug, Clone, Serialize)]
pub struct HealOutcome {
    pub from_version: u32,
    pub to_version: u32,
    /// `migrated` (reviewed in-place migration) or `rebuilt` (from JSONL).
    pub action: &'static str,
    /// Where the pre-heal database family was preserved.
    pub backup: String,
    /// Database-only issues re-added as unflushed changes.
    pub restored_db_only: Vec<String>,
    /// Database-only issues left only in the backup.
    pub backup_only_db_only: Vec<String>,
    /// Relation rows without an owning issue that a rebuild kept only in the
    /// backup (`--discard-db-only`).
    #[serde(skip_serializing_if = "Vec::is_empty")]
    pub backup_only_orphan_relations: Vec<OrphanRelationRows>,
    /// Why a reviewed migration was not used, when the rebuild ran instead.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub migration_fallback_reason: Option<String>,
}

impl HealOutcome {
    /// One-line human notice for stderr.
    #[must_use]
    pub fn notice(&self) -> String {
        let how = if self.action == "migrated" {
            "migrated it in place"
        } else {
            "rebuilt it from the JSONL"
        };
        let mut line = format!(
            "br: tracker database was on schema {} (this br uses {}); {how} — backup at {}",
            self.from_version, self.to_version, self.backup
        );
        if !self.restored_db_only.is_empty() {
            line.push_str(&format!(
                "; re-added {} database-only issue(s) as unflushed changes",
                self.restored_db_only.len()
            ));
        }
        if !self.backup_only_db_only.is_empty() {
            line.push_str(&format!(
                "; {} database-only issue(s) not restored; kept only in the backup",
                self.backup_only_db_only.len()
            ));
        }
        if !self.backup_only_orphan_relations.is_empty() {
            let rows: usize = self
                .backup_only_orphan_relations
                .iter()
                .map(|orphans| orphans.rows)
                .sum();
            line.push_str(&format!(
                "; {rows} relation row(s) without an owning issue kept only in the backup"
            ));
        }
        line
    }
}

/// Cheap startup probe: `Some(found)` when the database exists and its
/// effective schema is older than this binary's.
///
/// Reads the 100-byte header first and opens the engine only when the header
/// is behind, because a migration may still live in an uncheckpointed WAL.
///
/// # Errors
///
/// Returns an error if the header is unstable or the engine probe fails.
pub fn stale_schema_version(db_path: &Path) -> Result<Option<u32>> {
    let current = current_version();
    let Some(header) = crate::storage::sqlite::database_header_user_version(db_path) else {
        return Ok(None);
    };
    if header >= current {
        return Ok(None);
    }
    let effective =
        crate::storage::sqlite::effective_database_user_version(db_path)?.unwrap_or(header);
    Ok((effective < current).then_some(effective))
}

fn current_version() -> u32 {
    u32::try_from(CURRENT_SCHEMA_VERSION).unwrap_or(u32::MAX)
}

/// Refusal naming the one command that resolves a stale database holding
/// database-only data (or one that cannot be audited).
#[must_use]
pub fn refusal_error(audit: &StaleSchemaAudit) -> BeadsError {
    let found = i32::try_from(audit.from_version).unwrap_or(i32::MAX);
    let why = audit.unprovable.clone().unwrap_or_else(|| {
        let mut why = Vec::new();
        if !audit.db_only.is_empty() {
            why.push(format!(
                "{} issue(s) exist only in the database: {}",
                audit.db_only.len(),
                audit.db_only_summary()
            ));
        }
        if audit.has_orphan_relations() {
            why.push(format!(
                "the database holds relation rows without an owning issue ({}), which a \
                 rebuild from issues.jsonl cannot carry",
                audit.orphan_summary()
            ));
        }
        why.join("; ")
    });
    let what = if REVIEWED_MIGRATION_SOURCE_VERSIONS.contains(&audit.from_version) {
        "migrates the database in place without dropping any row, keeping a recovery bundle \
         and printing an undo command"
            .to_string()
    } else if audit
        .db_only
        .iter()
        .any(|issue| issue.reason == DbOnlyReason::Unreadable)
    {
        format!(
            "refuses a JSONL rebuild that cannot restore unreadable issues; repair those \
             original payloads first, or use `{HEAL_COMMAND} --discard-db-only` to keep \
             unreadable payloads and orphan relations only in the retained backup. {}",
            audit.restoration_notice()
        )
    } else if audit.has_orphan_relations() {
        format!(
            "refuses a JSONL rebuild that would drop those relation rows; `{HEAL_COMMAND} \
             --discard-db-only` is the explicit choice to rebuild and keep them only in the \
             retained .beads/.br_recovery backup. {}",
            audit.restoration_notice()
        )
    } else {
        "rebuilds the database from issues.jsonl, re-adds the database-only issues as \
         unflushed changes (exported by the next flush), and keeps the old database in \
         .beads/.br_recovery"
            .to_string()
    };
    BeadsError::WithContext {
        context: format!(
            "automatic schema upgrade refused because {why}. Run `{HEAL_COMMAND}`: it {what}. \
             `{HEAL_COMMAND} --dry-run` shows the audit without changing anything"
        ),
        source: Box::new(BeadsError::SchemaMismatch {
            expected: CURRENT_SCHEMA_VERSION,
            found,
        }),
    }
}

// ---------------------------------------------------------------------------
// Audit
// ---------------------------------------------------------------------------

/// Compare the stale database with the JSONL snapshot.
///
/// `jsonl` is `None` when the workspace has no JSONL file.
///
/// # Errors
///
/// Returns an error for engine access failures or records whose identity or
/// ownership cannot be read without loss. Undecodable issue payloads and
/// their relations are reported as unreadable through [`StaleSchemaAudit`].
pub(crate) fn audit_stale_database(
    db_path: &Path,
    jsonl: Option<&JsonlSourceSnapshot>,
) -> Result<StaleSchemaAudit> {
    let conn = open_with_flags(
        db_path.to_string_lossy().as_ref(),
        OpenFlags::SQLITE_OPEN_READ_ONLY,
    )
    .map_err(BeadsError::Database)?;
    let result = audit_connection(&conn, jsonl.map(JsonlSourceSnapshot::reader));
    let _ = conn.close();
    result
}

fn audit_connection(
    conn: &Connection,
    jsonl_reader: Option<impl BufRead>,
) -> Result<StaleSchemaAudit> {
    let from_version = query_user_version(conn)?;
    let mut audit = StaleSchemaAudit {
        from_version,
        to_version: current_version(),
        db_issue_count: 0,
        jsonl_issue_count: 0,
        db_only: Vec::new(),
        orphan_relations: Vec::new(),
        unprovable: None,
        restorable: Vec::new(),
    };

    let tables = table_names(conn)?;
    let issue_columns = if tables.contains("issues") {
        column_names(conn, "issues")?
    } else {
        Vec::new()
    };
    if !["id", "title", "status", "updated_at"]
        .iter()
        .all(|required| issue_columns.iter().any(|column| column == required))
    {
        audit.unprovable = Some(
            "the database has no recognizable beads `issues` table, so its contents cannot be \
             compared with the JSONL"
                .to_string(),
        );
        return Ok(audit);
    }

    let (db_issues, orphan_relations) = read_legacy_issues(conn, &tables, &issue_columns)?;
    audit.db_issue_count = db_issues.len();
    audit.orphan_relations = orphan_relations;
    let dirty: Option<HashSet<String>> = if tables.contains("dirty_issues") {
        Some(
            conn.query("SELECT issue_id FROM dirty_issues")?
                .iter()
                .map(|row| required_legacy_text(row.get(0), "dirty_issues", "issue_id"))
                .collect::<Result<HashSet<_>>>()?,
        )
    } else {
        // Without dirty tracking every differing row must be treated as
        // possibly unflushed.
        None
    };

    let jsonl_issues = match jsonl_reader {
        Some(reader) => match parse_jsonl(reader) {
            Ok(issues) => issues,
            Err(error) => {
                audit.unprovable = Some(format!("issues.jsonl could not be parsed ({error})"));
                return Ok(audit);
            }
        },
        None if db_issues.is_empty() => HashMap::new(),
        None => {
            audit.unprovable = Some(format!(
                "there is no issues.jsonl to compare the database's {} issue(s) against",
                db_issues.len()
            ));
            return Ok(audit);
        }
    };
    audit.jsonl_issue_count = jsonl_issues.len();

    for legacy in db_issues {
        let id = legacy.id.clone();
        let Some(preserved) = legacy.decoded else {
            audit.db_only.push(DbOnlyIssue {
                id,
                reason: DbOnlyReason::Unreadable,
                restorable: false,
            });
            continue;
        };
        let is_dirty = dirty.as_ref().is_none_or(|dirty| dirty.contains(&id));
        let verdict = classify(&preserved.issue, jsonl_issues.get(&id), is_dirty);
        if let Some((reason, restorable)) = verdict {
            audit.db_only.push(DbOnlyIssue {
                id,
                reason,
                restorable,
            });
            if restorable {
                audit.restorable.push(preserved);
            }
        }
    }
    Ok(audit)
}

/// `None` when represented; otherwise the reason and whether the explicit
/// heal re-adds the row.
fn classify(db: &Issue, jsonl: Option<&Issue>, is_dirty: bool) -> Option<(DbOnlyReason, bool)> {
    let Some(jsonl) = jsonl else {
        return Some((DbOnlyReason::AbsentFromJsonl, true));
    };
    // A flushed deletion wins over any local state (import tombstone guard).
    if jsonl.status == Status::Tombstone || issues_equivalent(db, jsonl) {
        return None;
    }
    let db_newer = micros(db.updated_at) > micros(jsonl.updated_at);
    if is_dirty {
        // Re-add only when the local edit is newer; an older unflushed edit
        // was superseded by the JSONL and stays in the backup (#394 rule).
        return Some((DbOnlyReason::UnflushedEdit, db_newer));
    }
    db_newer.then_some((DbOnlyReason::NewerThanJsonl, true))
}

fn micros(value: DateTime<Utc>) -> i64 {
    value.timestamp_micros()
}

fn opt_micros(value: Option<DateTime<Utc>>) -> Option<i64> {
    value.map(micros)
}

fn norm(value: Option<&String>) -> Option<&str> {
    value.map(String::as_str).filter(|text| !text.is_empty())
}

/// Content equivalence over what the JSONL carries: the database copy must
/// hold nothing the JSONL copy lacks. Compare the complete synced scalar
/// payload, not just display fields. Machine-local source_repo_path and
/// derived/audit-only fields follow Issue::sync_equals and are not sync keys.
fn issues_equivalent(db: &Issue, jsonl: &Issue) -> bool {
    let scalar_fields_match = db.id == jsonl.id
        && db.title == jsonl.title
        && norm(db.description.as_ref()) == norm(jsonl.description.as_ref())
        && norm(db.design.as_ref()) == norm(jsonl.design.as_ref())
        && norm(db.acceptance_criteria.as_ref()) == norm(jsonl.acceptance_criteria.as_ref())
        && norm(db.prerequisites.as_ref()) == norm(jsonl.prerequisites.as_ref())
        && norm(db.notes.as_ref()) == norm(jsonl.notes.as_ref())
        && db.status == jsonl.status
        && db.priority == jsonl.priority
        && db.issue_type == jsonl.issue_type
        && norm(db.assignee.as_ref()) == norm(jsonl.assignee.as_ref())
        && norm(db.owner.as_ref()) == norm(jsonl.owner.as_ref())
        && db.estimated_minutes == jsonl.estimated_minutes
        && norm(db.created_by.as_ref()) == norm(jsonl.created_by.as_ref())
        && norm(db.close_reason.as_ref()) == norm(jsonl.close_reason.as_ref())
        && norm(db.closed_by_session.as_ref()) == norm(jsonl.closed_by_session.as_ref())
        && norm(db.external_ref.as_ref()) == norm(jsonl.external_ref.as_ref())
        && norm(db.source_system.as_ref()) == norm(jsonl.source_system.as_ref())
        && db.source_repo.as_deref().unwrap_or(".") == jsonl.source_repo.as_deref().unwrap_or(".")
        && norm(db.agent_context.as_ref()) == norm(jsonl.agent_context.as_ref())
        && norm(db.deleted_by.as_ref()) == norm(jsonl.deleted_by.as_ref())
        && norm(db.delete_reason.as_ref()) == norm(jsonl.delete_reason.as_ref())
        && norm(db.original_type.as_ref()) == norm(jsonl.original_type.as_ref())
        && db.compaction_level.unwrap_or(0) == jsonl.compaction_level.unwrap_or(0)
        && norm(db.compacted_at_commit.as_ref()) == norm(jsonl.compacted_at_commit.as_ref())
        && db.original_size == jsonl.original_size
        && norm(db.sender.as_ref()) == norm(jsonl.sender.as_ref())
        && db.ephemeral == jsonl.ephemeral
        && db.pinned == jsonl.pinned
        && db.is_template == jsonl.is_template
        && micros(db.created_at) == micros(jsonl.created_at)
        && micros(db.updated_at) == micros(jsonl.updated_at)
        && opt_micros(db.closed_at) == opt_micros(jsonl.closed_at)
        && opt_micros(db.due_at) == opt_micros(jsonl.due_at)
        && opt_micros(db.defer_until) == opt_micros(jsonl.defer_until)
        && opt_micros(db.deleted_at) == opt_micros(jsonl.deleted_at)
        && opt_micros(db.compacted_at) == opt_micros(jsonl.compacted_at);
    if !scalar_fields_match {
        return false;
    }
    let jsonl_labels: HashSet<&str> = jsonl.labels.iter().map(String::as_str).collect();
    db.labels
        .iter()
        .all(|label| jsonl_labels.contains(label.as_str()))
        && multiset_is_subset(
            db.dependencies.iter().map(dependency_payload_key),
            jsonl.dependencies.iter().map(dependency_payload_key),
        )
        && multiset_is_subset(
            db.comments.iter().map(Comment::sync_key),
            jsonl.comments.iter().map(Comment::sync_key),
        )
}

/// Defaulted relation fields have the same persisted meaning, but populated
/// metadata, thread ownership, creator and timestamp must all survive. The
/// old (target, type) key certified losses in those fields as equivalence.
fn dependency_payload_key(
    dependency: &Dependency,
) -> (
    &str,
    &str,
    &str,
    DateTime<Utc>,
    Option<&str>,
    &str,
    Option<&str>,
) {
    (
        &dependency.issue_id,
        &dependency.depends_on_id,
        dependency.dep_type.as_str(),
        dependency.created_at,
        norm(dependency.created_by.as_ref()),
        dependency
            .metadata
            .as_deref()
            .filter(|text| !text.trim().is_empty())
            .unwrap_or("{}"),
        norm(dependency.thread_id.as_ref()),
    )
}

/// A source superset is safe, but one serialized child cannot certify two
/// distinct stored occurrences. Comment rowids are deliberately not keys:
/// they are database-local; Comment::sync_key supplies the portable identity.
fn multiset_is_subset<K: Eq + std::hash::Hash>(
    mut required: impl Iterator<Item = K>,
    present: impl Iterator<Item = K>,
) -> bool {
    let mut counts = HashMap::<K, usize>::new();
    for key in present {
        *counts.entry(key).or_default() += 1;
    }
    required.all(|key| {
        let Some(count) = counts.get_mut(&key) else {
            return false;
        };
        if *count == 0 {
            return false;
        }
        *count -= 1;
        true
    })
}

fn parse_jsonl(mut reader: impl BufRead) -> Result<HashMap<String, Issue>> {
    let mut issues = HashMap::new();
    let mut line = String::new();
    let mut line_num = 0usize;
    loop {
        line.clear();
        if reader.read_line(&mut line)? == 0 {
            break;
        }
        line_num += 1;
        let trimmed = line.trim();
        if trimmed.is_empty() {
            continue;
        }
        let issue: Issue = serde_json::from_str(trimmed)
            .map_err(|error| BeadsError::Config(format!("line {line_num}: {error}")))?;
        if issue.id.trim().is_empty() {
            return Err(BeadsError::Config(format!(
                "line {line_num}: empty issue id cannot certify a schema-heal source"
            )));
        }
        if issues.insert(issue.id.clone(), issue).is_some() {
            return Err(BeadsError::Config(format!(
                "line {line_num}: duplicate issue id cannot certify a schema-heal source"
            )));
        }
    }
    Ok(issues)
}

// ---------------------------------------------------------------------------
// Schema-agnostic legacy row decoding
// ---------------------------------------------------------------------------

struct LegacyIssue {
    id: String,
    decoded: Option<PreservedIssue>,
}

const TIMESTAMP_FIELDS: &[&str] = &[
    "created_at",
    "updated_at",
    "closed_at",
    "due_at",
    "defer_until",
    "deleted_at",
    "compacted_at",
];
const BOOL_FIELDS: &[&str] = &["ephemeral", "pinned", "is_template", "bypassed_policy"];
const INT_FIELDS: &[&str] = &[
    "priority",
    "estimated_minutes",
    "compaction_level",
    "original_size",
];
const STRING_FIELDS: &[&str] = &[
    "id",
    "title",
    "description",
    "design",
    "acceptance_criteria",
    "prerequisites",
    "notes",
    "status",
    "issue_type",
    "assignee",
    "owner",
    "created_by",
    "close_reason",
    "closed_by_session",
    "bypass_reason",
    "external_ref",
    "source_system",
    "source_repo",
    "source_repo_path",
    "agent_context",
    "deleted_by",
    "delete_reason",
    "original_type",
    "compacted_at_commit",
    "sender",
];

fn read_legacy_issues(
    conn: &Connection,
    tables: &HashSet<String>,
    issue_columns: &[String],
) -> Result<(Vec<LegacyIssue>, Vec<OrphanRelationRows>)> {
    let mut labels: HashMap<String, Vec<String>> = HashMap::new();
    if tables.contains("labels") {
        for row in conn.query("SELECT issue_id, label FROM labels")? {
            let issue_id = required_legacy_text(row.get(0), "labels", "issue_id")?;
            let label = required_legacy_text(row.get(1), "labels", "label")?;
            labels.entry(issue_id).or_default().push(label);
        }
    }
    let mut dependencies = read_related_rows(conn, tables, "dependencies")?;
    let mut comments = read_related_rows(conn, tables, "comments")?;

    let select = issue_columns
        .iter()
        .map(|column| quote_identifier(column))
        .collect::<Vec<_>>()
        .join(", ");
    let rows = conn.query(&format!("SELECT {select} FROM issues"))?;
    let mut issues = Vec::with_capacity(rows.len());
    let mut ids = HashSet::new();
    for row in rows {
        let values: Vec<SqliteValue> = row.values().to_vec();
        let id = required_legacy_text(column_value(issue_columns, &values, "id"), "issues", "id")?;
        if !ids.insert(id.clone()) {
            return Err(BeadsError::Config(
                "duplicate legacy issue ids cannot be preserved by schema healing".to_string(),
            ));
        }
        let issue_dependencies = dependencies.remove(&id).unwrap_or_default();
        let issue_comments = comments.remove(&id).unwrap_or_default();
        let decoded = decode_issue(
            issue_columns,
            &values,
            labels.remove(&id).unwrap_or_default(),
            &issue_dependencies,
            &issue_comments,
        );
        issues.push(LegacyIssue { id, decoded });
    }
    // Well-formed relation rows whose issue row is gone (deletes that older
    // releases ran without foreign-key enforcement) have no issue to compare
    // or restore. Report them rather than refusing the whole audit: the
    // reviewed in-place migration keeps them, and a JSONL rebuild requires
    // explicit `--discard-db-only` (see `require_lossless_rebuild`).
    let orphan_relations = [
        ("labels", labels.values().map(Vec::len).sum::<usize>()),
        (
            "dependencies",
            dependencies.values().map(Vec::len).sum::<usize>(),
        ),
        ("comments", comments.values().map(Vec::len).sum::<usize>()),
    ]
    .into_iter()
    .filter(|&(_, rows)| rows > 0)
    .map(|(table, rows)| OrphanRelationRows { table, rows })
    .collect();
    Ok((issues, orphan_relations))
}

fn required_legacy_text(value: Option<&SqliteValue>, table: &str, column: &str) -> Result<String> {
    value
        .and_then(SqliteValue::as_text)
        .filter(|text| !text.trim().is_empty())
        .map(str::to_owned)
        .ok_or_else(|| {
            BeadsError::Config(format!(
                "legacy {table}.{column} must be nonempty text; refusing a lossy schema heal"
            ))
        })
}

fn column_value<'a>(
    columns: &[String],
    values: &'a [SqliteValue],
    name: &str,
) -> Option<&'a SqliteValue> {
    columns
        .iter()
        .position(|column| column == name)
        .and_then(|index| values.get(index))
}

/// Rows of a relation table keyed by `issue_id`, each as a column map.
fn read_related_rows(
    conn: &Connection,
    tables: &HashSet<String>,
    table: &str,
) -> Result<HashMap<String, Vec<Map<String, Value>>>> {
    let mut related: HashMap<String, Vec<Map<String, Value>>> = HashMap::new();
    if !tables.contains(table) {
        return Ok(related);
    }
    let columns = column_names(conn, table)?;
    let select = columns
        .iter()
        .map(|column| quote_identifier(column))
        .collect::<Vec<_>>()
        .join(", ");
    for row in conn.query(&format!("SELECT {select} FROM {}", quote_identifier(table)))? {
        let issue_id = required_legacy_text(
            column_value(&columns, row.values(), "issue_id"),
            table,
            "issue_id",
        )?;
        let mut object = Map::new();
        for (column, value) in columns.iter().zip(row.values()) {
            if let Some(json) = sqlite_to_json(value) {
                object.insert(column.clone(), json);
            } else if !matches!(value, SqliteValue::Null) {
                return Err(BeadsError::Config(format!(
                    "legacy {table}.{column} cannot be represented in JSON; \
                     refusing a lossy schema heal"
                )));
            }
        }
        related.entry(issue_id).or_default().push(object);
    }
    Ok(related)
}

fn sqlite_to_json(value: &SqliteValue) -> Option<Value> {
    match value {
        SqliteValue::Null => None,
        SqliteValue::Integer(integer) => Some(Value::from(*integer)),
        SqliteValue::Float(float) => serde_json::Number::from_f64(*float).map(Value::Number),
        other => other.as_text().map(|text| Value::String(text.to_owned())),
    }
}

/// Decode one legacy issue row into the current model, or `None` when a
/// populated field or relation cannot be represented. Returning a partial
/// issue would let an omitted child masquerade as data already in JSONL.
fn decode_issue(
    columns: &[String],
    values: &[SqliteValue],
    labels: Vec<String>,
    dependencies: &[Map<String, Value>],
    comments: &[Map<String, Value>],
) -> Option<PreservedIssue> {
    let mut object = Map::new();
    for (column, value) in columns.iter().zip(values) {
        let name = column.as_str();
        if matches!(value, SqliteValue::Null) {
            continue;
        }
        if TIMESTAMP_FIELDS.contains(&name) {
            match legacy_timestamp(value) {
                Some(timestamp) => {
                    object.insert(name.to_owned(), Value::String(timestamp.to_rfc3339()));
                }
                // Legacy optional timestamps may be empty, but a populated
                // value that cannot be decoded is data, not absence.
                None if !matches!(name, "created_at" | "updated_at")
                    && value.as_text().is_some_and(|text| text.trim().is_empty()) => {}
                None => return None,
            }
        } else if BOOL_FIELDS.contains(&name) {
            let flag = match value {
                SqliteValue::Integer(integer) => *integer != 0,
                other => match other.as_text()?.trim() {
                    "1" => true,
                    "0" | "" => false,
                    text if text.eq_ignore_ascii_case("true") => true,
                    text if text.eq_ignore_ascii_case("false") => false,
                    _ => return None,
                },
            };
            object.insert(name.to_owned(), Value::Bool(flag));
        } else if INT_FIELDS.contains(&name) {
            if name != "priority" && value.as_text().is_some_and(|text| text.trim().is_empty()) {
                continue;
            }
            let integer = match value {
                SqliteValue::Integer(integer) => Some(*integer),
                other => other.as_text().and_then(|text| text.trim().parse().ok()),
            }?;
            object.insert(name.to_owned(), Value::from(integer));
        } else if STRING_FIELDS.contains(&name) {
            let text = value.as_text()?;
            if !text.is_empty() || matches!(name, "id" | "title") {
                object.insert(name.to_owned(), Value::String(text.to_owned()));
            }
        }
    }
    let created_at = object.get("created_at").cloned();
    let dependencies: Vec<Dependency> = dependencies
        .iter()
        .map(|row| decode_related(row, created_at.as_ref()))
        .collect::<Option<_>>()?;
    let comments: Vec<Comment> = comments
        .iter()
        .map(|row| {
            let mut row = row.clone();
            if let Some(Value::String(text)) = row.remove("text").or_else(|| row.remove("body")) {
                row.insert("text".to_owned(), Value::String(text));
            }
            decode_related(&row, created_at.as_ref())
        })
        .collect::<Option<_>>()?;
    object.insert(
        "labels".to_owned(),
        Value::Array(labels.iter().cloned().map(Value::String).collect()),
    );
    let mut issue: Issue = serde_json::from_value(Value::Object(object)).ok()?;
    issue.dependencies.clone_from(&dependencies);
    issue.comments.clone_from(&comments);
    Some(PreservedIssue {
        labels: Some(labels),
        dependencies: Some(dependencies),
        comments: Some(comments),
        issue,
    })
}

/// Decode a dependency or comment row, normalizing its timestamp and
/// defaulting a missing one to the owning issue's `created_at`.
fn decode_related<T: serde::de::DeserializeOwned>(
    row: &Map<String, Value>,
    fallback_created_at: Option<&Value>,
) -> Option<T> {
    let mut row = row.clone();
    let created_at = match row.get("created_at") {
        None | Some(Value::Null) => fallback_created_at.cloned()?,
        Some(Value::String(text)) if text.trim().is_empty() => fallback_created_at.cloned()?,
        Some(value) => {
            let timestamp = match value {
                Value::String(text) => legacy_timestamp_text(text),
                Value::Number(number) => number.as_i64().and_then(legacy_epoch),
                _ => None,
            }?;
            Value::String(timestamp.to_rfc3339())
        }
    };
    row.insert("created_at".to_owned(), created_at);
    serde_json::from_value(Value::Object(row)).ok()
}

fn legacy_timestamp(value: &SqliteValue) -> Option<DateTime<Utc>> {
    match value {
        SqliteValue::Integer(integer) => legacy_epoch(*integer),
        other => other.as_text().and_then(legacy_timestamp_text),
    }
}

fn legacy_timestamp_text(text: &str) -> Option<DateTime<Utc>> {
    let text = text.trim();
    if text.is_empty() {
        return None;
    }
    if let Ok(parsed) = DateTime::parse_from_rfc3339(text) {
        return Some(parsed.with_timezone(&Utc));
    }
    for format in ["%Y-%m-%d %H:%M:%S%.f", "%Y-%m-%dT%H:%M:%S%.f"] {
        if let Ok(naive) = NaiveDateTime::parse_from_str(text, format) {
            return Some(Utc.from_utc_datetime(&naive));
        }
    }
    // Go's time.Time String() form, as some bd builds stored it:
    // "2026-01-03 14:28:28.085091 -0500 EST".
    let mut parts = text.split_whitespace();
    if let (Some(date), Some(time), Some(offset)) = (parts.next(), parts.next(), parts.next())
        && let Ok(parsed) = DateTime::parse_from_str(
            &format!("{date} {time} {offset}"),
            "%Y-%m-%d %H:%M:%S%.f %z",
        )
    {
        return Some(parsed.with_timezone(&Utc));
    }
    text.parse::<i64>().ok().and_then(legacy_epoch)
}

/// Integer datetimes: seconds, milliseconds, microseconds, or nanoseconds
/// since the epoch, distinguished by magnitude.
fn legacy_epoch(value: i64) -> Option<DateTime<Utc>> {
    let magnitude = value.unsigned_abs();
    if magnitude < 100_000_000_000 {
        Utc.timestamp_opt(value, 0).single()
    } else if magnitude < 100_000_000_000_000 {
        Utc.timestamp_millis_opt(value).single()
    } else if magnitude < 100_000_000_000_000_000 {
        Some(DateTime::from_timestamp_micros(value)?.with_timezone(&Utc))
    } else {
        Some(DateTime::from_timestamp_nanos(value))
    }
}

fn query_user_version(conn: &Connection) -> Result<u32> {
    let row = conn.query_row("PRAGMA user_version")?;
    Ok(row
        .get(0)
        .and_then(SqliteValue::as_integer)
        .and_then(|value| u32::try_from(value).ok())
        .unwrap_or(0))
}

fn table_names(conn: &Connection) -> Result<HashSet<String>> {
    Ok(conn
        .query("SELECT name FROM sqlite_master WHERE type='table'")?
        .iter()
        .filter_map(|row| row.get(0).and_then(SqliteValue::as_text).map(str::to_owned))
        .collect())
}

fn column_names(conn: &Connection, table: &str) -> Result<Vec<String>> {
    Ok(conn
        .query(&format!("PRAGMA table_info({})", quote_identifier(table)))?
        .iter()
        .filter_map(|row| row.get(1).and_then(SqliteValue::as_text).map(str::to_owned))
        .collect())
}

fn quote_identifier(name: &str) -> String {
    format!("\"{}\"", name.replace('"', "\"\""))
}

// ---------------------------------------------------------------------------
// Heal
// ---------------------------------------------------------------------------

/// Inputs shared by both heal paths.
pub struct HealContext<'a> {
    pub beads_dir: &'a Path,
    pub cli: &'a config::CliOverrides,
    pub write_authority: &'a Arc<DatabaseFamilyWriteLock>,
}

/// Result of a heal attempt.
pub enum HealResult {
    /// The database was not on an older schema.
    NotNeeded,
    /// The database is current now.
    Healed(HealOutcome),
    /// Automatic mode refused; the audit explains why.
    Refused(Box<StaleSchemaAudit>),
}

/// Upgrade a stale database, auditing only the JSONL-rebuild fallback.
///
/// The caller must hold the database-family write authority for the whole
/// call. A reviewed migration preserves raw rows without consulting JSONL.
/// Before any rebuild, JSONL family authority pins exactly the snapshot that
/// the strict audit compares and the rebuild imports.
///
/// # Errors
///
/// Returns an error when the audit or upgrade fails. Deterministic local
/// replay failures are rehearsed before live replacement. Reviewed migration
/// and JSONL installation retain their existing backup/rollback protocols;
/// late replay I/O failures still require the retained recovery backup.
#[allow(clippy::too_many_lines)]
pub fn heal_stale_schema(ctx: &HealContext<'_>, mode: HealMode) -> Result<HealResult> {
    let startup = config::load_startup_config_with_paths(ctx.beads_dir, ctx.cli.db.as_ref())?;
    let paths = startup.paths.clone();
    ctx.write_authority.verify_database_authority()?;
    let Some(from_version) = stale_schema_version(&paths.db_path)? else {
        return Ok(HealResult::NotNeeded);
    };

    // GH #551: the strict legacy decoder proves JSONL rebuildability, not
    // migration safety. Empty labels, NULL relation owners, BLOBs and legacy
    // scalar encodings can be kept by the reviewed raw-row migration even
    // when they cannot be represented by the current Issue/JSONL model.
    let mut migration_fallback_reason = None;
    if REVIEWED_MIGRATION_SOURCE_VERSIONS.contains(&from_version) {
        let migration = super::schema_migration::MigrationContext {
            beads_dir: ctx.beads_dir.to_path_buf(),
            db_path: paths.db_path.clone(),
            write_authority: Arc::clone(ctx.write_authority),
        };
        match super::schema_migration::plan_and_apply(&migration) {
            Ok(backup) => {
                return Ok(HealResult::Healed(HealOutcome {
                    from_version,
                    to_version: current_version(),
                    action: "migrated",
                    backup: backup.display().to_string(),
                    restored_db_only: Vec::new(),
                    backup_only_db_only: Vec::new(),
                    backup_only_orphan_relations: Vec::new(),
                    migration_fallback_reason: None,
                }));
            }
            // Consider fallback only when the original schema is still in
            // place. A failed migration is never permission to bypass the
            // independent audit below, or to rebuild a partially upgraded DB.
            Err(error) if stale_schema_version(&paths.db_path)? == Some(from_version) => {
                migration_fallback_reason = Some(error.to_string());
            }
            Err(error) => return Err(error),
        }
    }

    let (jsonl_authority, source) = capture_jsonl(&paths.jsonl_path, ctx.cli.lock_timeout)?;
    let mut audit = audit_stale_database(&paths.db_path, source.as_ref())?;
    audit.from_version = from_version;
    let discard_db_only = match mode {
        HealMode::Automatic if !audit.automatic_rebuild_allowed() => {
            // Do not reuse automatic_heal_allowed here: reviewed-range
            // eligibility was exhausted above and cannot authorize data loss.
            return Ok(HealResult::Refused(Box::new(audit)));
        }
        HealMode::Automatic => false,
        HealMode::Explicit { discard_db_only } => discard_db_only,
    };

    let (Some(jsonl_authority), Some(source)) = (jsonl_authority, source) else {
        let migration = migration_fallback_reason.as_ref().map_or_else(
            || format!("no reviewed migration exists for schema {from_version}"),
            |reason| format!("the reviewed migration did not complete ({reason})"),
        );
        return Err(BeadsError::Config(format!(
            "cannot heal the schema-{from_version} database: there is no issues.jsonl to rebuild \
             from and {migration}. Export it with the \
             br release that wrote it (`br sync --flush-only`), then rerun `{HEAL_COMMAND}`"
        )));
    };
    // A successful reviewed migration above preserves raw rows. A JSONL
    // rebuild cannot do that for unreadable payloads, even in explicit mode.
    // Enforce this before calling the routine that replaces the live family.
    require_lossless_rebuild(&audit, discard_db_only)?;
    let mut merged = startup.merged_config.clone();
    merged.merge_from(&ctx.cli.as_layer());
    let allow_external_jsonl =
        config::implicit_external_jsonl_allowed(ctx.beads_dir, &paths.db_path, &paths.jsonl_path);
    let lock_timeout = ctx
        .cli
        .lock_timeout
        .or_else(|| config::lock_timeout_from_layer(&merged))
        .or(Some(30000));
    // A readable legacy payload can still conflict with imported rows or
    // current constraints. Prove the complete replay on private state before
    // replacing the live family, then verify the real result against it.
    let rebuild_witness =
        rebuild_preflight::rehearse(&source, &audit.restorable, &merged, lock_timeout)?;
    let (mut storage, _import, backups) =
        config::repair_database_from_jsonl_snapshot_under_write_authority(
            ctx.beads_dir,
            &paths.db_path,
            lock_timeout,
            &merged,
            false,
            allow_external_jsonl,
            &source,
            &jsonl_authority,
            ctx.write_authority,
        )?;
    let backup = backups
        .iter()
        .find(|verified| Path::new(&verified.original) == paths.db_path)
        .or_else(|| backups.first())
        .map(|verified| {
            Path::new(&verified.backup)
                .parent()
                .map_or_else(|| PathBuf::from(&verified.backup), Path::to_path_buf)
        })
        .unwrap_or_else(|| config::recovery_dir_for_db_path(&paths.db_path, ctx.beads_dir));

    // GH #550: authorizing backup-only retention for an unreadable payload
    // or orphan relation must not also omit unrelated, recoverable work.
    // Superseded edits are already excluded by the audit's restore plan.
    let restore = audit.restorable.as_slice();
    crate::sync::restore_dirty_issues_after_rebuild(&mut storage, restore)?;
    if let Some(witness) = rebuild_witness {
        witness.verify(&storage)?;
    }
    drop(storage);

    let restored: HashSet<&str> = restore
        .iter()
        .map(|entry| entry.issue.id.as_str())
        .collect();
    let (restored_db_only, backup_only_db_only): (Vec<String>, Vec<String>) = audit
        .db_only
        .iter()
        .map(|issue| issue.id.clone())
        .partition(|id| restored.contains(id.as_str()));
    Ok(HealResult::Healed(HealOutcome {
        from_version,
        to_version: audit.to_version,
        action: "rebuilt",
        backup: backup.display().to_string(),
        restored_db_only,
        backup_only_db_only,
        // Reachable with orphans only under `--discard-db-only`
        // (`require_lossless_rebuild`): a rebuild never carries them.
        backup_only_orphan_relations: audit.orphan_relations.clone(),
        migration_fallback_reason,
    }))
}

fn capture_jsonl(
    jsonl_path: &Path,
    lock_timeout: Option<u64>,
) -> Result<(
    Option<Arc<crate::sync::JsonlFamilyWriteLock>>,
    Option<JsonlSourceSnapshot>,
)> {
    if !jsonl_path.is_file() {
        return Ok((None, None));
    }
    let authority = Arc::new(crate::sync::blocking_jsonl_family_write_lock_with_timeout(
        jsonl_path,
        lock_timeout,
    )?);
    authority.verify_jsonl_authority()?;
    let source = crate::sync::capture_jsonl_source_snapshot(jsonl_path)?;
    Ok((Some(authority), Some(source)))
}

fn require_lossless_rebuild(audit: &StaleSchemaAudit, discard_db_only: bool) -> Result<()> {
    if let Some(reason) = &audit.unprovable {
        return Err(BeadsError::Config(format!(
            "cannot certify a schema-heal rebuild: {reason}; the rebuild was not started"
        )));
    }
    if !discard_db_only
        && audit
            .db_only
            .iter()
            .any(|issue| issue.reason == DbOnlyReason::Unreadable)
    {
        return Err(BeadsError::Config(format!(
            "schema-heal rebuild refused: unreadable database-only payloads cannot be \
             restored ({}). The rebuild was not started. Repair the original payloads first; \
             `{HEAL_COMMAND} --discard-db-only` is an explicit choice to keep unreadable \
             payloads and orphan relations only in the retained backup, not in the rebuilt \
             tracker. {}",
            audit.db_only_summary(),
            audit.restoration_notice()
        )));
    }
    if !discard_db_only && audit.has_orphan_relations() {
        return Err(BeadsError::Config(format!(
            "schema-heal rebuild refused: the database holds relation rows without an owning \
             issue ({}), which a rebuild from issues.jsonl cannot carry. The rebuild was not \
             started. `{HEAL_COMMAND} --discard-db-only` is an explicit choice to keep them \
             only in the retained backup, not in the rebuilt tracker. {}",
            audit.orphan_summary(),
            audit.restoration_notice()
        )));
    }
    Ok(())
}

/// Planned action for a stale database, as `--dry-run` reports it.
fn planned_action(audit: &StaleSchemaAudit, discard_db_only: bool) -> String {
    if REVIEWED_MIGRATION_SOURCE_VERSIONS.contains(&audit.from_version) {
        return reviewed_migration_action(audit.from_version, audit.to_version);
    }
    if let Err(error) = require_lossless_rebuild(audit, discard_db_only) {
        return format!("refuse rebuild: {error}");
    }
    let restorable = audit
        .db_only
        .iter()
        .filter(|issue| issue.restorable)
        .count();
    let tail = if audit.db_only.is_empty() {
        String::new()
    } else {
        format!(
            "; re-add {restorable} database-only issue(s) as unflushed changes{}",
            if restorable < audit.db_only.len() {
                format!(
                    "; {} unreadable or superseded database-only issue(s) kept only in the backup",
                    audit.db_only.len() - restorable
                )
            } else {
                String::new()
            }
        )
    };
    // `require_lossless_rebuild` passed, so orphans here mean `--discard-db-only`.
    let orphans = if audit.has_orphan_relations() {
        format!(
            "; {} without an owning issue kept only in the backup",
            audit.orphan_summary()
        )
    } else {
        String::new()
    };
    format!(
        "rebuild schema {} -> {} from issues.jsonl, old database family moved to \
         .beads/.br_recovery{tail}{orphans}",
        audit.from_version, audit.to_version
    )
}

fn reviewed_migration_action(from_version: u32, to_version: u32) -> String {
    format!(
        "reviewed in-place migration {from_version} -> {to_version} (subject to the reviewed \
         migration's schema, integrity and raw-row preservation checks; recovery bundle and \
         undo command retained)"
    )
}

/// Failure to encode old rows as JSONL does not make a raw-row migration
/// unavailable. A dry run must report that distinction without running one or
/// inventing an empty/clean audit (GH #551).
fn emit_migration_dry_run_without_audit(
    from_version: u32,
    error: &BeadsError,
    json: bool,
) -> Result<()> {
    let action = reviewed_migration_action(from_version, current_version());
    let note = "No migration was run. The JSONL rebuild audit failed; a reviewed migration \
                can still be attempted without it. Rebuilding remains refused until a fresh \
                audit proves it safe. Use `br doctor migrate-schema plan` to inspect the \
                migration's own checks.";
    if json {
        println!(
            "{}",
            serde_json::to_string_pretty(&serde_json::json!({
                "dry_run": true,
                "audit": null,
                "audit_error": error.to_string(),
                "from_version": from_version,
                "to_version": current_version(),
                "automatic_heal_allowed": true,
                "automatic_rebuild_allowed": false,
                "planned_action": action,
                "note": note,
            }))?
        );
    } else {
        println!("JSONL rebuild audit failed: {error}");
        println!("Heal would attempt: {action}.");
        println!("{note}");
    }
    Ok(())
}

/// Human-readable `heal --dry-run` report.
fn print_dry_run_audit(audit: &StaleSchemaAudit, action: &str) {
    println!(
        "Database schema {} (this br uses {}); {} issue(s) in the database, {} in the JSONL.",
        audit.from_version, audit.to_version, audit.db_issue_count, audit.jsonl_issue_count
    );
    if let Some(reason) = &audit.unprovable {
        println!("Audit incomplete: {reason}.");
    } else if audit.db_only.is_empty() {
        if audit.has_orphan_relations() {
            println!(
                "Relation rows without an owning issue: {}.",
                audit.orphan_summary()
            );
        }
        if audit.automatic_heal_allowed() {
            println!(
                "Nothing exists only in the database; ordinary commands heal it \
                 automatically."
            );
        } else {
            println!(
                "No issue exists only in the database, but a rebuild would drop those \
                 relation rows, so ordinary commands refuse."
            );
        }
    } else {
        println!(
            "{} issue(s) exist only in the database:",
            audit.db_only.len()
        );
        for issue in &audit.db_only {
            println!("  {} ({})", issue.id, issue.reason.describe());
        }
        if audit.has_orphan_relations() {
            println!(
                "Relation rows without an owning issue: {}.",
                audit.orphan_summary()
            );
        }
    }
    println!("Heal would: {action}.");
    if REVIEWED_MIGRATION_SOURCE_VERSIONS.contains(&audit.from_version) {
        println!(
            "The reviewed migration is attempted before this JSONL audit. If it fails, \
             automatic rebuilding is allowed only when that audit is clean and no orphan \
             relations would be lost."
        );
    }
}

/// Execute `br doctor migrate-schema heal`.
///
/// # Errors
///
/// Returns an error when the audit or the upgrade fails.
pub fn execute_heal(
    args: &crate::cli::DoctorMigrateSchemaHealArgs,
    cli: &config::CliOverrides,
    beads_dir: &Path,
    write_authority: &Arc<DatabaseFamilyWriteLock>,
) -> Result<()> {
    if args.dry_run {
        let paths = config::resolve_paths(beads_dir, cli.db.as_ref())?;
        let Some(from_version) = stale_schema_version(&paths.db_path)? else {
            return emit_not_needed(args.json);
        };
        let audited = (|| {
            let (_jsonl_authority, source) = capture_jsonl(&paths.jsonl_path, cli.lock_timeout)?;
            audit_stale_database(&paths.db_path, source.as_ref())
        })();
        let mut audit = match audited {
            Ok(audit) => audit,
            Err(error) if REVIEWED_MIGRATION_SOURCE_VERSIONS.contains(&from_version) => {
                return emit_migration_dry_run_without_audit(from_version, &error, args.json);
            }
            Err(error) => return Err(error),
        };
        audit.from_version = from_version;
        let action = planned_action(&audit, args.discard_db_only);
        if args.json {
            let payload = serde_json::json!({
                "dry_run": true,
                "audit": audit,
                "automatic_heal_allowed": audit.automatic_heal_allowed(),
                "automatic_rebuild_allowed": audit.automatic_rebuild_allowed(),
                "planned_action": action,
            });
            println!(
                "{}",
                serde_json::to_string_pretty(&payload).map_err(BeadsError::Json)?
            );
        } else {
            print_dry_run_audit(&audit, &action);
        }
        return Ok(());
    }

    let ctx = HealContext {
        beads_dir,
        cli,
        write_authority,
    };
    match heal_stale_schema(
        &ctx,
        HealMode::Explicit {
            discard_db_only: args.discard_db_only,
        },
    )? {
        HealResult::NotNeeded => emit_not_needed(args.json),
        HealResult::Refused(audit) => Err(refusal_error(&audit)),
        HealResult::Healed(outcome) => {
            if args.json {
                println!(
                    "{}",
                    serde_json::to_string_pretty(&outcome).map_err(BeadsError::Json)?
                );
            } else {
                println!("{}", outcome.notice().trim_start_matches("br: "));
                if let Some(reason) = &outcome.migration_fallback_reason {
                    println!("(reviewed migration was not possible: {reason})");
                }
                for id in &outcome.restored_db_only {
                    println!("  re-added {id}");
                }
                for id in &outcome.backup_only_db_only {
                    println!("  kept only in backup: {id}");
                }
                if !outcome.restored_db_only.is_empty() {
                    println!("Run `br sync --flush-only` to export the re-added issues.");
                }
            }
            Ok(())
        }
    }
}

fn emit_not_needed(json: bool) -> Result<()> {
    if json {
        println!(
            "{}",
            serde_json::to_string_pretty(&serde_json::json!({
                "heal_needed": false,
                "schema_version": CURRENT_SCHEMA_VERSION,
            }))
            .map_err(BeadsError::Json)?
        );
    } else {
        println!("Database is not on an older schema; nothing to heal.");
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::model::{IssueType, Priority};

    #[test]
    fn reviewed_migration_eligibility_never_authorizes_an_unsafe_rebuild() {
        let mut audit = StaleSchemaAudit {
            from_version: REVIEWED_MIGRATION_SOURCE_VERSIONS[0],
            to_version: current_version(),
            db_issue_count: 1,
            jsonl_issue_count: 1,
            db_only: Vec::new(),
            orphan_relations: Vec::new(),
            unprovable: None,
            restorable: Vec::new(),
        };
        assert!(audit.automatic_heal_allowed());
        assert!(audit.automatic_rebuild_allowed());

        audit.db_only.push(DbOnlyIssue {
            id: "legacy-551".to_owned(),
            reason: DbOnlyReason::Unreadable,
            restorable: false,
        });
        assert!(audit.automatic_heal_allowed());
        assert!(!audit.automatic_rebuild_allowed());
        assert!(require_lossless_rebuild(&audit, false).is_err());

        audit.db_only.clear();
        audit.orphan_relations.push(OrphanRelationRows {
            table: "comments",
            rows: 1,
        });
        assert!(audit.automatic_heal_allowed());
        assert!(!audit.automatic_rebuild_allowed());
        assert!(require_lossless_rebuild(&audit, false).is_err());

        audit.orphan_relations.clear();
        audit.unprovable = Some("JSONL snapshot cannot be decoded".to_owned());
        assert!(audit.automatic_heal_allowed());
        assert!(!audit.automatic_rebuild_allowed());
        // Even --discard-db-only cannot invent a trustworthy rebuild source.
        assert!(require_lossless_rebuild(&audit, true).is_err());
        audit.from_version = 4;
        assert!(!audit.automatic_heal_allowed());
    }

    fn issue(id: &str, updated: &str) -> Issue {
        let at = DateTime::parse_from_rfc3339(updated)
            .unwrap()
            .with_timezone(&Utc);
        Issue {
            id: id.to_owned(),
            title: format!("title {id}"),
            status: Status::Open,
            priority: Priority::MEDIUM,
            issue_type: IssueType::Task,
            created_at: at,
            updated_at: at,
            ..Issue::default()
        }
    }

    #[test]
    fn equivalent_rows_are_represented_even_when_dirty() {
        let db = issue("a-1", "2026-01-01T00:00:00Z");
        let jsonl = db.clone();
        assert_eq!(classify(&db, Some(&jsonl), true), None);
    }

    #[test]
    fn absent_rows_are_db_only_and_restorable() {
        let db = issue("a-1", "2026-01-01T00:00:00Z");
        assert_eq!(
            classify(&db, None, false),
            Some((DbOnlyReason::AbsentFromJsonl, true))
        );
    }

    #[test]
    fn clean_row_superseded_by_newer_jsonl_is_represented() {
        let db = issue("a-1", "2026-01-01T00:00:00Z");
        let mut jsonl = issue("a-1", "2026-02-01T00:00:00Z");
        jsonl.title = "renamed upstream".to_owned();
        assert_eq!(classify(&db, Some(&jsonl), false), None);
    }

    #[test]
    fn clean_row_newer_than_jsonl_is_db_only() {
        let mut db = issue("a-1", "2026-03-01T00:00:00Z");
        db.title = "local rename".to_owned();
        let jsonl = issue("a-1", "2026-02-01T00:00:00Z");
        assert_eq!(
            classify(&db, Some(&jsonl), false),
            Some((DbOnlyReason::NewerThanJsonl, true))
        );
    }

    #[test]
    fn dirty_edit_is_db_only_but_restored_only_when_newer() {
        let mut older = issue("a-1", "2026-01-01T00:00:00Z");
        older.title = "unflushed".to_owned();
        let jsonl = issue("a-1", "2026-02-01T00:00:00Z");
        assert_eq!(
            classify(&older, Some(&jsonl), true),
            Some((DbOnlyReason::UnflushedEdit, false))
        );
        let mut newer = issue("a-1", "2026-03-01T00:00:00Z");
        newer.title = "unflushed".to_owned();
        assert_eq!(
            classify(&newer, Some(&jsonl), true),
            Some((DbOnlyReason::UnflushedEdit, true))
        );
    }

    #[test]
    fn flushed_tombstone_wins() {
        let db = issue("a-1", "2026-03-01T00:00:00Z");
        let mut jsonl = issue("a-1", "2026-01-01T00:00:00Z");
        jsonl.status = Status::Tombstone;
        assert_eq!(classify(&db, Some(&jsonl), true), None);
    }

    #[test]
    fn label_missing_from_jsonl_breaks_equivalence() {
        let mut db = issue("a-1", "2026-01-01T00:00:00Z");
        db.labels = vec!["local".to_owned()];
        let jsonl = issue("a-1", "2026-01-01T00:00:00Z");
        assert!(!issues_equivalent(&db, &jsonl));
        assert!(issues_equivalent(&jsonl, &db), "JSONL superset is fine");
    }

    #[test]
    fn legacy_timestamps_decode() {
        let expected = DateTime::parse_from_rfc3339("2026-01-03T19:28:28Z")
            .unwrap()
            .with_timezone(&Utc);
        for text in [
            "2026-01-03T14:28:28-05:00",
            "2026-01-03 19:28:28",
            "2026-01-03T19:28:28",
            "2026-01-03 14:28:28 -0500 EST",
            "1767468508",
        ] {
            assert_eq!(legacy_timestamp_text(text), Some(expected), "{text}");
        }
        assert_eq!(legacy_epoch(1_767_468_508_000), Some(expected));
        assert_eq!(legacy_timestamp_text("not a date"), None);
    }

    fn legacy_audit_fixture() -> (Connection, Vec<u8>) {
        let conn = Connection::open(":memory:").unwrap();
        conn.execute(
            "CREATE TABLE issues (id TEXT, title TEXT, status TEXT, priority INTEGER, \
             issue_type TEXT, created_at TEXT, updated_at TEXT)",
        )
        .unwrap();
        conn.execute(
            "INSERT INTO issues VALUES ('a-1', 'title a-1', 'open', 2, 'task', \
             '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z')",
        )
        .unwrap();
        let jsonl = serde_json::to_vec(&issue("a-1", "2026-01-01T00:00:00Z")).unwrap();
        (conn, jsonl)
    }

    #[test]
    fn malformed_relations_cannot_make_a_schema_audit_appear_clean() {
        for setup in [
            [
                "CREATE TABLE comments (id INTEGER, issue_id TEXT, author TEXT, text TEXT)",
                "INSERT INTO comments VALUES (1, 'a-1', 'agent', NULL)",
            ],
            [
                "CREATE TABLE dependencies (issue_id TEXT, depends_on_id TEXT, type TEXT)",
                "INSERT INTO dependencies VALUES ('a-1', NULL, 'blocks')",
            ],
        ] {
            let (conn, jsonl) = legacy_audit_fixture();
            for sql in setup {
                conn.execute(sql).unwrap();
            }
            // The old filter_map decoder discarded the only child and
            // incorrectly certified this exact issue as already in JSONL.
            let audit = audit_connection(&conn, Some(std::io::Cursor::new(jsonl))).unwrap();
            assert!(!audit.is_clean());
            assert_eq!(audit.db_issue_count, 1);
            assert_eq!(audit.db_only.len(), 1);
            assert_eq!(audit.db_only[0].id, "a-1");
            assert_eq!(audit.db_only[0].reason, DbOnlyReason::Unreadable);
            assert!(!audit.db_only[0].restorable);
            assert!(audit.restorable.is_empty());
            assert!(require_lossless_rebuild(&audit, false).is_err());
            assert!(require_lossless_rebuild(&audit, true).is_ok());
            assert!(planned_action(&audit, false).starts_with("refuse rebuild:"));
            assert!(planned_action(&audit, true).starts_with("rebuild schema"));
        }
    }

    #[test]
    fn schema_audit_refuses_unattributable_or_unrepresentable_records() {
        let cases: &[(&[&str], &str)] = &[
            (&["UPDATE issues SET id = NULL"], "issues.id"),
            (
                &["INSERT INTO issues SELECT * FROM issues"],
                "duplicate legacy issue ids",
            ),
            (
                &[
                    "CREATE TABLE labels (issue_id TEXT, label TEXT)",
                    "INSERT INTO labels VALUES (NULL, 'retained')",
                ],
                "labels.issue_id",
            ),
            (
                &[
                    "CREATE TABLE labels (issue_id TEXT, label BLOB)",
                    "INSERT INTO labels VALUES ('a-1', X'FEED')",
                ],
                "labels.label",
            ),
            (
                &[
                    "CREATE TABLE comments (issue_id TEXT, text TEXT)",
                    "INSERT INTO comments VALUES (NULL, 'retained')",
                ],
                "comments.issue_id",
            ),
            (
                &[
                    "CREATE TABLE dependencies (issue_id TEXT, metadata BLOB)",
                    "INSERT INTO dependencies VALUES ('a-1', X'FEED')",
                ],
                "dependencies.metadata",
            ),
            (
                &[
                    "CREATE TABLE dirty_issues (issue_id TEXT)",
                    "INSERT INTO dirty_issues VALUES (NULL)",
                ],
                "dirty_issues.issue_id",
            ),
        ];
        for &(setup, expected) in cases {
            let (conn, jsonl) = legacy_audit_fixture();
            for sql in setup {
                conn.execute(sql).unwrap();
            }
            let error = audit_connection(&conn, Some(std::io::Cursor::new(jsonl))).unwrap_err();
            assert!(error.to_string().contains(expected), "{expected}: {error}");
        }
    }

    /// Well-formed relation rows whose issue is gone (deletes that older releases
    /// ran without foreign-key enforcement) used to fail the whole audit, so every
    /// mutating command and even `heal --discard-db-only` refused such a database.
    /// They are reported instead: the reviewed in-place migration keeps them, and
    /// only a JSONL rebuild, which cannot carry them, needs `--discard-db-only`.
    #[test]
    fn orphan_relation_rows_are_reported_and_only_a_rebuild_needs_discard() {
        let (conn, jsonl) = legacy_audit_fixture();
        for sql in [
            "CREATE TABLE comments (issue_id TEXT, text TEXT)",
            "INSERT INTO comments VALUES ('missing-owner', 'retained')",
            "INSERT INTO comments VALUES ('missing-owner', 'second')",
            "CREATE TABLE labels (issue_id TEXT, label TEXT)",
            "INSERT INTO labels VALUES ('missing-owner', 'retained')",
        ] {
            conn.execute(sql).unwrap();
        }
        let mut audit = audit_connection(&conn, Some(std::io::Cursor::new(jsonl))).unwrap();
        assert!(audit.is_clean(), "orphans are not issues: {audit:?}");
        let orphans: Vec<(&str, usize)> = audit
            .orphan_relations
            .iter()
            .map(|orphans| (orphans.table, orphans.rows))
            .collect();
        assert_eq!(orphans, [("labels", 1), ("comments", 2)]);

        audit.from_version = REVIEWED_MIGRATION_SOURCE_VERSIONS[0];
        assert!(audit.automatic_heal_allowed(), "the migration keeps them");
        audit.from_version = 4;
        assert!(!audit.automatic_heal_allowed(), "a rebuild would drop them");
        assert!(
            refusal_error(&audit)
                .to_string()
                .contains("--discard-db-only")
        );

        let refused = require_lossless_rebuild(&audit, false).unwrap_err();
        assert!(
            refused.to_string().contains("without an owning issue"),
            "{refused}"
        );
        assert!(planned_action(&audit, false).starts_with("refuse rebuild"));
        require_lossless_rebuild(&audit, true).expect("explicit discard authorizes it");
        assert!(
            planned_action(&audit, true).contains("kept only in the backup"),
            "{}",
            planned_action(&audit, true)
        );
    }

    #[test]
    fn populated_invalid_optional_values_are_not_silently_defaulted() {
        for (column, value) in [
            ("due_at", SqliteValue::from("not a date")),
            ("pinned", SqliteValue::from("not a boolean")),
            ("estimated_minutes", SqliteValue::from("not an integer")),
            ("description", SqliteValue::Integer(42)),
        ] {
            let columns: Vec<String> = ["id", "title", "created_at", "updated_at", column]
                .into_iter()
                .map(str::to_owned)
                .collect();
            let values = [
                SqliteValue::from("a-1"),
                SqliteValue::from("title a-1"),
                SqliteValue::from("2026-01-01T00:00:00Z"),
                SqliteValue::from("2026-01-01T00:00:00Z"),
                value,
            ];
            assert!(
                decode_issue(&columns, &values, Vec::new(), &[], &[]).is_none(),
                "{column}"
            );
        }
    }

    #[test]
    fn related_timestamp_fallback_requires_absence_not_a_malformed_value() {
        let fallback = serde_json::json!("2026-01-01T00:00:00Z");
        let mut row = serde_json::json!({
            "id": 1, "issue_id": "a-1", "author": "agent", "text": "retained",
        })
        .as_object()
        .unwrap()
        .clone();
        assert!(decode_related::<Comment>(&row, Some(&fallback)).is_some());
        for absent in [Value::Null, serde_json::json!("")] {
            row.insert("created_at".to_string(), absent);
            assert!(decode_related::<Comment>(&row, Some(&fallback)).is_some());
        }
        for malformed in [
            serde_json::json!("not a date"),
            serde_json::json!({"bad": true}),
        ] {
            row.insert("created_at".to_string(), malformed);
            assert!(decode_related::<Comment>(&row, Some(&fallback)).is_none());
        }
    }

    #[test]
    fn valid_legacy_children_are_preserved_in_the_restore_plan() {
        let (conn, _) = legacy_audit_fixture();
        for sql in [
            "CREATE TABLE labels (issue_id TEXT, label TEXT)",
            "INSERT INTO labels VALUES ('a-1', 'local-only')",
            "CREATE TABLE comments (id INTEGER, issue_id TEXT, author TEXT, body TEXT)",
            "INSERT INTO comments VALUES (1, 'a-1', 'agent', 'unexported comment')",
            "CREATE TABLE dependencies (issue_id TEXT, depends_on_id TEXT, type TEXT, created_by TEXT)",
            "INSERT INTO dependencies VALUES ('a-1', 'external-target', 'blocks', 'agent')",
            "ALTER TABLE issues ADD COLUMN due_at TEXT",
            "UPDATE issues SET due_at = ''",
        ] {
            conn.execute(sql).unwrap();
        }
        let older = serde_json::to_vec(&issue("a-1", "2025-12-01T00:00:00Z")).unwrap();
        let audit = audit_connection(&conn, Some(std::io::Cursor::new(older))).unwrap();
        assert!(!audit.is_clean());
        assert_eq!(audit.restorable.len(), 1);
        let restored = &audit.restorable[0].issue;
        assert_eq!(restored.labels, vec!["local-only".to_string()]);
        assert_eq!(restored.comments.len(), 1);
        assert_eq!(restored.comments[0].body, "unexported comment");
        assert_eq!(restored.comments[0].created_at, restored.created_at);
        assert_eq!(restored.dependencies.len(), 1);
        assert_eq!(restored.dependencies[0].depends_on_id, "external-target");
        assert_eq!(restored.dependencies[0].created_at, restored.created_at);
        assert!(restored.due_at.is_none());
        assert!(require_lossless_rebuild(&audit, false).is_ok());
    }

    #[test]
    fn duplicate_or_unprovable_jsonl_never_authorizes_a_rebuild() {
        let (conn, jsonl) = legacy_audit_fixture();
        let mut duplicate = jsonl.clone();
        duplicate.push(b'\n');
        duplicate.extend_from_slice(&jsonl);
        let audit = audit_connection(&conn, Some(std::io::Cursor::new(duplicate))).unwrap();
        assert!(!audit.is_clean());
        assert!(
            audit
                .unprovable
                .as_deref()
                .unwrap()
                .contains("duplicate issue id")
        );
        for discard in [false, true] {
            assert!(require_lossless_rebuild(&audit, discard).is_err());
            assert!(planned_action(&audit, discard).starts_with("refuse rebuild:"));
        }
    }

    #[test]
    fn discard_plan_separates_recoverable_issues_from_unreadable_rows_and_orphans() {
        let (conn, _) = legacy_audit_fixture();
        for sql in [
            "INSERT INTO issues VALUES ('a-2', 'unreadable', 'open', 2, 'task', \
             '2026-01-01T00:00:00Z', 'not a timestamp')",
            "CREATE TABLE labels (issue_id TEXT, label TEXT)",
            "INSERT INTO labels VALUES ('missing-owner', 'orphan')",
        ] {
            conn.execute(sql).unwrap();
        }
        let audit = audit_connection(&conn, Some(std::io::Cursor::new(Vec::<u8>::new())))
            .expect("auditable identities with one unreadable payload");
        assert_eq!(audit.db_only.len(), 2);
        assert_eq!(audit.restorable.len(), 1);
        assert_eq!(audit.restorable[0].issue.id, "a-1");
        assert!(!audit.automatic_rebuild_allowed());
        let refusal = require_lossless_rebuild(&audit, false).unwrap_err();
        assert!(refusal.to_string().contains("unreadable"));
        assert!(
            refusal
                .to_string()
                .contains("1 recoverable database-only issue(s) (a-1)")
        );
        assert!(
            refusal_error(&audit)
                .to_string()
                .contains("--discard-db-only")
        );
        require_lossless_rebuild(&audit, true)
            .expect("explicit authorization for unrestorable data");
        let action = planned_action(&audit, true);
        assert!(
            action.contains("re-add 1 database-only issue(s)"),
            "{action}"
        );
        assert!(
            action.contains(
                "1 unreadable or superseded database-only issue(s) kept only in the backup"
            ),
            "{action}"
        );
        assert!(
            action.contains("1 labels row(s) without an owning issue"),
            "{action}"
        );
    }

    #[test]
    fn discard_plan_still_restores_issues_when_no_data_is_unreadable() {
        let (conn, _) = legacy_audit_fixture();
        let audit = audit_connection(&conn, Some(std::io::Cursor::new(Vec::<u8>::new())))
            .expect("recoverable database-only issue");
        require_lossless_rebuild(&audit, false).unwrap();
        require_lossless_rebuild(&audit, true).unwrap();
        assert_eq!(planned_action(&audit, false), planned_action(&audit, true));
        assert!(planned_action(&audit, true).contains("re-add 1 database-only issue(s)"));
    }

    #[test]
    fn backup_only_notice_does_not_call_unreadable_payloads_superseded() {
        let outcome = HealOutcome {
            from_version: 4,
            to_version: current_version(),
            action: "rebuilt",
            backup: "retained-backup".to_owned(),
            restored_db_only: vec!["a-1".to_owned()],
            backup_only_db_only: vec!["a-2".to_owned()],
            backup_only_orphan_relations: Vec::new(),
            migration_fallback_reason: None,
        };
        let notice = outcome.notice();
        assert!(notice.contains("re-added 1 database-only issue(s)"));
        assert!(notice.contains("1 database-only issue(s) not restored"));
        assert!(!notice.contains("superseded"));
    }

    #[test]
    fn legacy_governance_and_local_provenance_survive_decoding() {
        let (conn, _) = legacy_audit_fixture();
        for sql in [
            "ALTER TABLE issues ADD COLUMN agent_context TEXT",
            "ALTER TABLE issues ADD COLUMN source_repo_path TEXT",
            "ALTER TABLE issues ADD COLUMN prerequisites TEXT",
            "UPDATE issues SET agent_context = '{\"constraints\":[\"keep local work\"]}', \
             source_repo_path = '/original/workspace', prerequisites = '- [ ] review'",
        ] {
            conn.execute(sql).unwrap();
        }
        let audit = audit_connection(&conn, Some(std::io::Cursor::new(Vec::<u8>::new())))
            .expect("decode a database-only issue");
        assert_eq!(audit.restorable.len(), 1);
        let restored = &audit.restorable[0].issue;
        assert_eq!(
            restored.agent_context.as_deref(),
            Some("{\"constraints\":[\"keep local work\"]}")
        );
        assert_eq!(
            restored.source_repo_path.as_deref(),
            Some("/original/workspace")
        );
        assert_eq!(restored.prerequisites.as_deref(), Some("- [ ] review"));
    }

    #[test]
    fn governance_only_dirty_difference_is_not_certified_as_equivalent() {
        let (conn, jsonl) = legacy_audit_fixture();
        for sql in [
            "ALTER TABLE issues ADD COLUMN agent_context TEXT",
            "UPDATE issues SET agent_context = '{\"constraints\":[\"local\"]}'",
            "CREATE TABLE dirty_issues (issue_id TEXT)",
            "INSERT INTO dirty_issues VALUES ('a-1')",
        ] {
            conn.execute(sql).unwrap();
        }
        let audit = audit_connection(&conn, Some(std::io::Cursor::new(jsonl))).unwrap();
        assert!(!audit.is_clean());
        assert!(!audit.automatic_rebuild_allowed());
        assert_eq!(audit.db_only[0].reason, DbOnlyReason::UnflushedEdit);
    }

    #[test]
    fn nontext_governance_cannot_masquerade_as_an_absent_field() {
        for column in ["agent_context", "source_repo_path"] {
            let (conn, jsonl) = legacy_audit_fixture();
            conn.execute(&format!("ALTER TABLE issues ADD COLUMN {column} BLOB"))
                .unwrap();
            conn.execute(&format!("UPDATE issues SET {column} = X'FEED'"))
                .unwrap();
            let audit = audit_connection(&conn, Some(std::io::Cursor::new(jsonl))).unwrap();
            assert_eq!(
                audit.db_only[0].reason,
                DbOnlyReason::Unreadable,
                "{column}"
            );
            assert!(audit.restorable.is_empty());
            assert!(require_lossless_rebuild(&audit, false).is_err());
        }
    }

    #[test]
    fn synced_scalar_fields_do_not_disappear_from_equivalence() {
        let baseline = issue("a-1", "2026-01-01T00:00:00Z");
        for (field, changed) in [
            ("prerequisites", serde_json::json!("- [ ] review")),
            (
                "agent_context",
                serde_json::json!("{\"instructions\":[\"keep\"]}"),
            ),
            ("created_by", serde_json::json!("another agent")),
            ("closed_by_session", serde_json::json!("session-1")),
            ("source_system", serde_json::json!("upstream")),
            ("source_repo", serde_json::json!("repo")),
            ("deleted_by", serde_json::json!("agent")),
            ("delete_reason", serde_json::json!("retention decision")),
            ("original_type", serde_json::json!("feature")),
            ("compaction_level", serde_json::json!(2)),
            ("compacted_at_commit", serde_json::json!("abc123")),
            ("original_size", serde_json::json!(42)),
            ("sender", serde_json::json!("agent")),
            ("ephemeral", serde_json::json!(true)),
            ("pinned", serde_json::json!(true)),
            ("is_template", serde_json::json!(true)),
            ("created_at", serde_json::json!("2025-12-01T00:00:00Z")),
            ("deleted_at", serde_json::json!("2025-12-01T00:00:00Z")),
            ("compacted_at", serde_json::json!("2025-12-01T00:00:00Z")),
        ] {
            let mut value = serde_json::to_value(&baseline).unwrap();
            value[field] = changed;
            let local: Issue = serde_json::from_value(value).unwrap();
            assert!(!issues_equivalent(&local, &baseline), "lost field: {field}");
            assert_eq!(
                classify(&local, Some(&baseline), true),
                Some((DbOnlyReason::UnflushedEdit, false)),
                "timestamp precedence must remain unchanged for {field}"
            );
        }
        let mut persisted = baseline.clone();
        persisted.source_repo = Some(".".to_owned());
        persisted.compaction_level = Some(0);
        persisted.source_repo_path = Some("/local/path".to_owned());
        assert!(issues_equivalent(&persisted, &baseline));
    }

    #[test]
    fn comment_equivalence_uses_portable_identity_and_preserves_multiplicity() {
        let mut local = issue("a-1", "2026-01-01T00:00:00Z");
        local.comments.push(Comment {
            id: 1,
            issue_id: local.id.clone(),
            author: "agent".to_owned(),
            body: "same words at a different time".to_owned(),
            created_at: local.created_at,
        });
        let mut exported = local.clone();
        exported.comments[0].id = 99;
        assert!(issues_equivalent(&local, &exported), "rowids are local");
        exported.comments[0].created_at += chrono::Duration::seconds(1);
        assert!(!issues_equivalent(&local, &exported));
        exported.comments.push(local.comments[0].clone());
        assert!(issues_equivalent(&local, &exported), "a superset is safe");
        local.comments.push(local.comments[0].clone());
        assert!(
            !issues_equivalent(&local, &exported),
            "one match is not two"
        );
        exported.comments.push(local.comments[0].clone());
        assert!(issues_equivalent(&local, &exported));
    }

    #[test]
    fn dependency_equivalence_requires_the_complete_payload() {
        let mut local = issue("a-1", "2026-01-01T00:00:00Z");
        let dependency: Dependency = serde_json::from_value(serde_json::json!({
            "issue_id": "a-1", "depends_on_id": "a-2", "type": "blocks",
            "created_at": "2026-01-01T00:00:00Z"
        }))
        .unwrap();
        local.dependencies.push(dependency.clone());
        let mut exported = local.clone();
        exported.dependencies[0].metadata = Some("{}".to_owned());
        exported.dependencies[0].created_by = Some(String::new());
        exported.dependencies[0].thread_id = Some(String::new());
        assert!(issues_equivalent(&local, &exported), "persisted defaults");
        for (field, changed) in [
            ("metadata", serde_json::json!("{\"reason\":\"local\"}")),
            ("thread_id", serde_json::json!("thread-1")),
            ("created_by", serde_json::json!("another-agent")),
            ("created_at", serde_json::json!("2026-01-01T00:00:01Z")),
            ("issue_id", serde_json::json!("a-3")),
        ] {
            let mut value = serde_json::to_value(&dependency).unwrap();
            value[field] = changed;
            exported.dependencies = vec![serde_json::from_value(value).unwrap()];
            assert!(!issues_equivalent(&local, &exported), "lost field: {field}");
            exported.dependencies.push(dependency.clone());
            assert!(issues_equivalent(&local, &exported), "superset: {field}");
        }
    }
}