beads_rust 0.1.45

//! JSONL import/export for `beads_rust`.
//!
//! This module handles:
//! - Export: `SQLite` -> JSONL (for git tracking)
//! - Import: JSONL -> `SQLite` (for git clone/pull)
//! - Dirty tracking for incremental exports
//! - Collision detection during imports
//! - Path validation and allowlist enforcement

pub mod history;
pub mod path;

pub use path::{
    ALLOWED_EXACT_NAMES, ALLOWED_EXTENSIONS, PathValidation, is_sync_path_allowed,
    require_safe_sync_overwrite_path, require_valid_sync_path, validate_no_git_path,
    validate_sync_path, validate_sync_path_with_external, validate_temp_file_path,
};

use crate::error::{BeadsError, Result};
use crate::model::{Comment, Dependency, Issue};
use crate::storage::SqliteStorage;
use crate::sync::history::HistoryConfig;
use crate::util::id::parse_id;
use crate::util::progress::{create_progress_bar, create_spinner};
use crate::validation::IssueValidator;
use chrono::Utc;
use fsqlite_types::SqliteValue;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, HashMap, HashSet, hash_map::RandomState};
use std::fmt::Write as FmtWrite;
use std::fs::{self, File, OpenOptions};
use std::io::{BufRead, BufReader, BufWriter, Write};
use std::path::{Path, PathBuf};

/// Acquire a blocking exclusive lock on `.beads/.write.lock`.
///
/// This serializes all mutating operations across processes, preventing
/// concurrent-write deadlocks in the underlying SQLite engine. Uses a fast-path
/// `try_lock()` for the uncontended case, then falls back to a true blocking
/// `lock()` that waits indefinitely until the holder releases. The lock is
/// held until the returned `File` drops.
#[allow(clippy::incompatible_msrv)]
#[must_use]
pub fn blocking_write_lock(beads_dir: &Path) -> Option<File> {
    let lock_path = beads_dir.join(".write.lock");
    let file = OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(&lock_path)
        .ok()?;

    // Fast path: non-blocking try for the common uncontended case.
    if file.try_lock().is_ok() {
        return Some(file);
    }

    // Contended: block until the current holder releases.
    tracing::debug!(".write.lock is held by another process; waiting for release");
    match file.lock() {
        Ok(()) => Some(file),
        Err(e) => {
            tracing::debug!("failed to acquire .write.lock: {e}; proceeding without lock");
            None
        }
    }
}

/// Try to acquire an exclusive advisory lock on `.beads/.sync.lock`.
///
/// Returns the lock file on success. The lock is held until the returned
/// `File` is dropped. If another process already holds the lock, returns
/// `None` (non-blocking).
#[allow(clippy::incompatible_msrv)]
#[must_use]
pub fn try_sync_lock(beads_dir: &Path) -> Option<File> {
    let lock_path = beads_dir.join(".sync.lock");
    let file = OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(&lock_path)
        .ok()?;
    match file.try_lock() {
        Ok(()) => Some(file),
        Err(_) => None,
    }
}

struct TempFileGuard {
    path: PathBuf,
    persist: bool,
}

impl TempFileGuard {
    fn new(path: PathBuf) -> Self {
        Self {
            path,
            persist: false,
        }
    }

    fn persist(&mut self) {
        self.persist = true;
    }
}

impl Drop for TempFileGuard {
    fn drop(&mut self) {
        if !self.persist {
            let _ = fs::remove_file(&self.path);
        }
    }
}

pub(crate) fn export_temp_path(output_path: &Path) -> PathBuf {
    output_path.with_extension(format!("jsonl.{}.tmp", std::process::id()))
}

/// Configuration for JSONL export.
#[derive(Debug, Clone, Default)]
#[allow(clippy::struct_excessive_bools)]
pub struct ExportConfig {
    /// Force export even if database is empty and JSONL has issues.
    pub force: bool,
    /// Whether this is an export to the default JSONL path (affects dirty flag clearing).
    pub is_default_path: bool,
    /// Error handling policy for export.
    pub error_policy: ExportErrorPolicy,
    /// Retention period for tombstones in days (None = keep forever).
    pub retention_days: Option<u64>,
    /// The `.beads` directory path for path validation.
    /// If None, path validation is skipped (for backwards compatibility).
    pub beads_dir: Option<PathBuf>,
    /// Allow JSONL path outside `.beads/` directory (requires explicit opt-in).
    /// Even with this flag, git paths are ALWAYS rejected.
    pub allow_external_jsonl: bool,
    /// Show progress indicators for long-running operations.
    pub show_progress: bool,
    /// Configuration for history backups.
    pub history: HistoryConfig,
}

/// Export error handling policy.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Default)]
#[serde(rename_all = "kebab-case")]
pub enum ExportErrorPolicy {
    /// Abort export on any error (default).
    #[default]
    Strict,
    /// Skip problematic records, export what we can.
    BestEffort,
    /// Export valid records, report failures.
    Partial,
    /// Only export core issues; non-core errors are tolerated.
    RequiredCore,
}

impl std::fmt::Display for ExportErrorPolicy {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let value = match self {
            Self::Strict => "strict",
            Self::BestEffort => "best-effort",
            Self::Partial => "partial",
            Self::RequiredCore => "required-core",
        };
        write!(f, "{value}")
    }
}

impl std::str::FromStr for ExportErrorPolicy {
    type Err = String;

    fn from_str(input: &str) -> std::result::Result<Self, Self::Err> {
        match input.to_ascii_lowercase().as_str() {
            "strict" => Ok(Self::Strict),
            "best-effort" | "best_effort" | "best" => Ok(Self::BestEffort),
            "partial" => Ok(Self::Partial),
            "required-core" | "required_core" | "core" => Ok(Self::RequiredCore),
            other => Err(format!(
                "Invalid error policy: {other}. Must be one of: strict, best-effort, partial, required-core"
            )),
        }
    }
}

/// Export entity types for error reporting.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum ExportEntityType {
    Issue,
    Dependency,
    Label,
    Comment,
}

/// Export error record.
#[derive(Debug, Clone, Serialize)]
pub struct ExportError {
    pub entity_type: ExportEntityType,
    pub entity_id: String,
    pub message: String,
}

impl ExportError {
    fn new(
        entity_type: ExportEntityType,
        entity_id: impl Into<String>,
        message: impl Into<String>,
    ) -> Self {
        Self {
            entity_type,
            entity_id: entity_id.into(),
            message: message.into(),
        }
    }

    #[must_use]
    pub fn summary(&self) -> String {
        let id = if self.entity_id.is_empty() {
            "<unknown>"
        } else {
            self.entity_id.as_str()
        };
        format!("{:?} {id}: {}", self.entity_type, self.message)
    }
}

/// Export report with error details and counts.
#[derive(Debug, Clone, Serialize)]
pub struct ExportReport {
    pub issues_exported: usize,
    pub dependencies_exported: usize,
    pub labels_exported: usize,
    pub comments_exported: usize,
    pub errors: Vec<ExportError>,
    pub policy_used: ExportErrorPolicy,
}

impl ExportReport {
    const fn new(policy: ExportErrorPolicy) -> Self {
        Self {
            issues_exported: 0,
            dependencies_exported: 0,
            labels_exported: 0,
            comments_exported: 0,
            errors: Vec::new(),
            policy_used: policy,
        }
    }

    /// True if any errors were recorded.
    #[must_use]
    pub fn has_errors(&self) -> bool {
        !self.errors.is_empty()
    }

    /// Success rate for exported entities.
    #[must_use]
    #[allow(clippy::cast_precision_loss)]
    pub fn success_rate(&self) -> f64 {
        let total = self.issues_exported
            + self.dependencies_exported
            + self.labels_exported
            + self.comments_exported;
        let failed = self.errors.len();
        if total + failed == 0 {
            1.0
        } else {
            total as f64 / (total + failed) as f64
        }
    }
}

struct ExportContext {
    policy: ExportErrorPolicy,
    errors: Vec<ExportError>,
}

impl ExportContext {
    const fn new(policy: ExportErrorPolicy) -> Self {
        Self {
            policy,
            errors: Vec::new(),
        }
    }

    fn handle_error(&mut self, err: ExportError) -> Result<()> {
        match self.policy {
            ExportErrorPolicy::Strict => Err(BeadsError::Config(format!(
                "Export error: {}",
                err.summary()
            ))),
            ExportErrorPolicy::BestEffort | ExportErrorPolicy::Partial => {
                self.errors.push(err);
                Ok(())
            }
            ExportErrorPolicy::RequiredCore => {
                if err.entity_type == ExportEntityType::Issue {
                    Err(BeadsError::Config(format!(
                        "Export error: {}",
                        err.summary()
                    )))
                } else {
                    self.errors.push(err);
                    Ok(())
                }
            }
        }
    }
}

/// Result of a JSONL export operation.
#[derive(Debug, Clone, Default)]
pub struct ExportResult {
    /// Number of issues exported.
    pub exported_count: usize,
    /// IDs of exported issues.
    pub exported_ids: Vec<String>,
    /// IDs and timestamps of dirty issues that were cleared.
    pub exported_marked_at: Vec<(String, String)>,
    /// IDs skipped due to expired tombstone retention (still clear dirty flags).
    pub skipped_tombstone_ids: Vec<String>,
    /// SHA256 hash of the exported JSONL content.
    pub content_hash: String,
    /// Output file path (None if stdout).
    pub output_path: Option<String>,
    /// Per-issue content hashes (`issue_id`, `content_hash`) for incremental export tracking.
    pub issue_hashes: Vec<(String, String)>,
}

/// Configuration for JSONL import.
#[derive(Debug, Clone)]
#[allow(clippy::struct_excessive_bools)]
pub struct ImportConfig {
    /// Skip prefix validation when importing.
    pub skip_prefix_validation: bool,
    /// Rewrite IDs and references on prefix mismatch.
    pub rename_on_import: bool,
    /// Clear duplicate external refs instead of erroring.
    pub clear_duplicate_external_refs: bool,
    /// How to handle orphaned issues during import.
    pub orphan_mode: OrphanMode,
    /// Force upsert even if timestamps are equal or older.
    pub force_upsert: bool,
    /// The `.beads` directory path for path validation.
    /// If None, path validation is skipped (for backwards compatibility).
    pub beads_dir: Option<PathBuf>,
    /// Allow JSONL path outside `.beads/` directory (requires explicit opt-in).
    /// Even with this flag, git paths are ALWAYS rejected.
    pub allow_external_jsonl: bool,
    /// Show progress indicators for long-running operations.
    pub show_progress: bool,
}

impl Default for ImportConfig {
    fn default() -> Self {
        Self {
            skip_prefix_validation: false,
            rename_on_import: false,
            clear_duplicate_external_refs: false,
            orphan_mode: OrphanMode::Strict,
            force_upsert: false,
            beads_dir: None,
            allow_external_jsonl: false,
            show_progress: false,
        }
    }
}

/// Orphan handling behavior for import.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OrphanMode {
    /// Fail if any issue references a missing parent.
    Strict,
    /// Attempt to resurrect missing parents if found.
    Resurrect,
    /// Skip orphaned issues.
    Skip,
    /// Allow orphans (no parent validation).
    Allow,
}

/// Result of a JSONL import.
#[derive(Debug, Clone, Default)]
pub struct ImportResult {
    /// Number of issues imported (created or updated).
    pub imported_count: usize,
    /// Number of issues created during import.
    pub created_count: usize,
    /// Number of issues updated during import.
    pub updated_count: usize,
    /// Number of issues skipped.
    pub skipped_count: usize,
    /// Number of tombstones skipped.
    pub tombstone_skipped: usize,
    /// Conflict markers detected (if any).
    pub conflict_markers: Vec<ConflictMarker>,
    /// Number of orphaned DB entries removed during --rebuild.
    pub orphans_removed: usize,
    /// Number of orphaned FK rows cleaned after deferred-FK import.
    pub orphan_cleaned_count: usize,
}

// ============================================================================
// PREFLIGHT CHECKS (beads_rust-0v1.2.7)
// ============================================================================

/// Status of a preflight check.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PreflightCheckStatus {
    /// Check passed.
    Pass,
    /// Check passed with warnings.
    Warn,
    /// Check failed.
    Fail,
}

/// A single preflight check result.
#[derive(Debug, Clone)]
pub struct PreflightCheck {
    /// Name of the check (e.g., "`path_validation`").
    pub name: String,
    /// Human-readable description of what was checked.
    pub description: String,
    /// Status of the check.
    pub status: PreflightCheckStatus,
    /// Detailed message (error/warning reason, or success confirmation).
    pub message: String,
    /// Actionable remediation hint (if status is Fail or Warn).
    pub remediation: Option<String>,
}

impl PreflightCheck {
    fn pass(
        name: impl Into<String>,
        description: impl Into<String>,
        message: impl Into<String>,
    ) -> Self {
        Self {
            name: name.into(),
            description: description.into(),
            status: PreflightCheckStatus::Pass,
            message: message.into(),
            remediation: None,
        }
    }

    fn warn(
        name: impl Into<String>,
        description: impl Into<String>,
        message: impl Into<String>,
        remediation: impl Into<String>,
    ) -> Self {
        Self {
            name: name.into(),
            description: description.into(),
            status: PreflightCheckStatus::Warn,
            message: message.into(),
            remediation: Some(remediation.into()),
        }
    }

    fn fail(
        name: impl Into<String>,
        description: impl Into<String>,
        message: impl Into<String>,
        remediation: impl Into<String>,
    ) -> Self {
        Self {
            name: name.into(),
            description: description.into(),
            status: PreflightCheckStatus::Fail,
            message: message.into(),
            remediation: Some(remediation.into()),
        }
    }
}

/// Result of running all preflight checks.
#[derive(Debug, Clone)]
pub struct PreflightResult {
    /// All checks that were run.
    pub checks: Vec<PreflightCheck>,
    /// Overall status (Fail if any check failed, Warn if any warned, Pass otherwise).
    pub overall_status: PreflightCheckStatus,
}

impl PreflightResult {
    const fn new() -> Self {
        Self {
            checks: Vec::new(),
            overall_status: PreflightCheckStatus::Pass,
        }
    }

    fn add(&mut self, check: PreflightCheck) {
        // Update overall status (Fail > Warn > Pass)
        match check.status {
            PreflightCheckStatus::Fail => self.overall_status = PreflightCheckStatus::Fail,
            PreflightCheckStatus::Warn if self.overall_status != PreflightCheckStatus::Fail => {
                self.overall_status = PreflightCheckStatus::Warn;
            }
            _ => {}
        }
        self.checks.push(check);
    }

    /// Returns true if all checks passed (no failures or warnings).
    #[must_use]
    pub fn is_ok(&self) -> bool {
        self.overall_status == PreflightCheckStatus::Pass
    }

    /// Returns true if there are no failures (warnings are acceptable).
    #[must_use]
    pub fn has_no_failures(&self) -> bool {
        self.overall_status != PreflightCheckStatus::Fail
    }

    /// Get all failed checks.
    #[must_use]
    pub fn failures(&self) -> Vec<&PreflightCheck> {
        self.checks
            .iter()
            .filter(|c| c.status == PreflightCheckStatus::Fail)
            .collect()
    }

    /// Get all warnings.
    #[must_use]
    pub fn warnings(&self) -> Vec<&PreflightCheck> {
        self.checks
            .iter()
            .filter(|c| c.status == PreflightCheckStatus::Warn)
            .collect()
    }

    /// Convert to an error if there are failures.
    ///
    /// # Errors
    ///
    /// Returns an error if there are failed checks.
    pub fn into_result(self) -> Result<Self> {
        if self.overall_status == PreflightCheckStatus::Fail {
            let mut msg = String::from("Preflight checks failed:\n");
            for check in self.failures() {
                use std::fmt::Write;
                let _ = writeln!(msg, "  - {}: {}", check.name, check.message);
                if let Some(ref rem) = check.remediation {
                    let _ = writeln!(msg, "    Hint: {rem}");
                }
            }
            Err(BeadsError::Config(msg))
        } else {
            Ok(self)
        }
    }
}

const JSONL_VALIDATION_PREVIEW_LIMIT: usize = 10;

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct JsonlIssueValidationFailure {
    pub line: usize,
    pub message: String,
}

#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub(crate) struct JsonlIssueValidationSummary {
    pub record_count: usize,
    pub invalid_count: usize,
    pub failures: Vec<JsonlIssueValidationFailure>,
}

impl JsonlIssueValidationSummary {
    fn push_failure(&mut self, line: usize, message: impl Into<String>) {
        self.invalid_count += 1;
        if self.failures.len() < JSONL_VALIDATION_PREVIEW_LIMIT {
            self.failures.push(JsonlIssueValidationFailure {
                line,
                message: message.into(),
            });
        }
    }

    pub(crate) fn preview_messages(&self) -> Vec<String> {
        self.failures
            .iter()
            .map(|failure| format!("line {}: {}", failure.line, failure.message))
            .collect()
    }
}

pub(crate) fn validate_jsonl_issue_records(path: &Path) -> Result<JsonlIssueValidationSummary> {
    let file = File::open(path)?;
    let reader = BufReader::with_capacity(2 * 1024 * 1024, file);
    let mut summary = JsonlIssueValidationSummary::default();
    let mut seen_ids = HashSet::new();

    for (line_num, line_result) in reader.lines().enumerate() {
        let line = line_result?;
        let trimmed = line.trim();
        if trimmed.is_empty() {
            continue;
        }

        summary.record_count += 1;
        match serde_json::from_str::<Issue>(trimmed) {
            Ok(mut issue) => {
                normalize_issue(&mut issue);
                if !seen_ids.insert(issue.id.clone()) {
                    summary
                        .push_failure(line_num + 1, format!("Duplicate issue id '{}'", issue.id));
                    continue;
                }
                if let Err(errors) = IssueValidator::validate(&issue) {
                    summary.push_failure(
                        line_num + 1,
                        errors
                            .iter()
                            .map(ToString::to_string)
                            .collect::<Vec<_>>()
                            .join(", "),
                    );
                }
            }
            Err(err) => summary.push_failure(line_num + 1, err.to_string()),
        }
    }

    Ok(summary)
}

/// Run preflight checks for export operation.
///
/// This function is read-only and validates:
/// - Beads directory exists
/// - Output path is within allowlist (not in .git, within `beads_dir`)
/// - Database is accessible
/// - Export won't cause data loss (empty db over non-empty JSONL, stale db)
///
/// # Arguments
///
/// * `storage` - Database connection for validation
/// * `output_path` - Target JSONL path
/// * `config` - Export configuration
///
/// # Returns
///
/// `PreflightResult` with all check results. Use `.into_result()` to convert
/// failures to an error.
///
/// # Errors
///
/// Returns an error if the preflight checks fail.
#[allow(clippy::too_many_lines)]
pub fn preflight_export(
    storage: &SqliteStorage,
    output_path: &Path,
    config: &ExportConfig,
) -> Result<PreflightResult> {
    let mut result = PreflightResult::new();

    tracing::debug!(
        output_path = %output_path.display(),
        beads_dir = ?config.beads_dir,
        "Running export preflight checks"
    );

    // Check 1: Beads directory exists
    if let Some(ref beads_dir) = config.beads_dir {
        if beads_dir.is_dir() {
            result.add(PreflightCheck::pass(
                "beads_dir_exists",
                "Beads directory exists",
                format!("Found: {}", beads_dir.display()),
            ));
            tracing::debug!(beads_dir = %beads_dir.display(), "Beads directory check: PASS");
        } else {
            result.add(PreflightCheck::fail(
                "beads_dir_exists",
                "Beads directory exists",
                format!("Not found: {}", beads_dir.display()),
                "Run 'br init' to initialize the beads directory.",
            ));
            tracing::debug!(beads_dir = %beads_dir.display(), "Beads directory check: FAIL");
        }
    }

    // Check 2: Output path validation (PC-1, PC-2, PC-3, NGI-3)
    if let Some(ref beads_dir) = config.beads_dir {
        // Determine if the path is external (outside .beads/)
        let canonical_beads = dunce::canonicalize(beads_dir).unwrap_or_else(|_| beads_dir.clone());
        let is_external =
            !output_path.starts_with(beads_dir) && !output_path.starts_with(&canonical_beads);

        match validate_sync_path_with_external(output_path, beads_dir, config.allow_external_jsonl)
        {
            Ok(()) => {
                let msg = format!(
                    "Path {} validated (external={})",
                    output_path.display(),
                    is_external
                );
                if is_external && config.allow_external_jsonl {
                    result.add(PreflightCheck::warn(
                        "path_validation",
                        "Output path is within allowlist",
                        msg,
                        "Consider moving JSONL to .beads/ directory for better safety.",
                    ));
                } else {
                    result.add(PreflightCheck::pass(
                        "path_validation",
                        "Output path is within allowlist",
                        msg,
                    ));
                }
                tracing::debug!(path = %output_path.display(), is_external = is_external, "Path validation: PASS");
            }
            Err(e) => {
                result.add(PreflightCheck::fail(
                    "path_validation",
                    "Output path is within allowlist",
                    format!("Path rejected: {e}"),
                    "Use a path within .beads/ directory or set --allow-external-jsonl.",
                ));
                tracing::debug!(path = %output_path.display(), error = %e, "Path validation: FAIL");
            }
        }
    }

    // Check 3: Database is accessible
    match storage.count_issues() {
        Ok(count) => {
            result.add(PreflightCheck::pass(
                "database_accessible",
                "Database is accessible",
                format!("Database contains {count} issue(s)"),
            ));
            tracing::debug!(issue_count = count, "Database access check: PASS");

            // Check 4: Empty database safety (would overwrite non-empty JSONL)
            if count == 0 && !config.force && output_path.exists() {
                match count_issues_in_jsonl(output_path) {
                    Ok(jsonl_count) if jsonl_count > 0 => {
                        result.add(PreflightCheck::fail(
                            "empty_database_safety",
                            "Export won't cause data loss",
                            format!(
                                "Database has 0 issues, JSONL has {jsonl_count} issues. Export would cause data loss.",
                            ),
                            "Import the JSONL first, or use --force to override.",
                        ));
                        tracing::debug!(
                            db_count = 0,
                            jsonl_count = jsonl_count,
                            "Empty database safety check: FAIL"
                        );
                    }
                    Ok(_) => {
                        result.add(PreflightCheck::pass(
                            "empty_database_safety",
                            "Export won't cause data loss",
                            "Database is empty, no existing JSONL to overwrite.",
                        ));
                    }
                    Err(e) => {
                        result.add(PreflightCheck::warn(
                            "empty_database_safety",
                            "Export won't cause data loss",
                            format!("Could not read existing JSONL: {e}"),
                            "Verify JSONL file is readable.",
                        ));
                    }
                }
            } else if count == 0 && !config.force {
                result.add(PreflightCheck::pass(
                    "empty_database_safety",
                    "Export won't cause data loss",
                    "Database is empty, no existing JSONL to overwrite.",
                ));
            }

            // Check 5: Stale database safety (would lose issues from JSONL)
            if count > 0 && !config.force && output_path.exists() {
                match get_issue_ids_from_jsonl(output_path) {
                    Ok(jsonl_ids) if !jsonl_ids.is_empty() => {
                        let db_ids: HashSet<String> = storage.get_all_ids()?.into_iter().collect();
                        let missing: Vec<_> = jsonl_ids.difference(&db_ids).take(5).collect();
                        if missing.is_empty() {
                            result.add(PreflightCheck::pass(
                                "stale_database_safety",
                                "Export won't lose JSONL issues",
                                "All JSONL issues are present in database.",
                            ));
                        } else {
                            let total_missing = jsonl_ids.difference(&db_ids).count();
                            result.add(PreflightCheck::fail(
                                "stale_database_safety",
                                "Export won't lose JSONL issues",
                                format!(
                                    "Database is missing {total_missing} issue(s) from JSONL: {}{}",
                                    missing
                                        .iter()
                                        .map(|s| s.as_str())
                                        .collect::<Vec<_>>()
                                        .join(", "),
                                    if total_missing > 5 { " ..." } else { "" }
                                ),
                                "Import the JSONL first to sync, or use --force to override.",
                            ));
                            tracing::debug!(
                                missing_count = total_missing,
                                sample = ?missing,
                                "Stale database safety check: FAIL"
                            );
                        }
                    }
                    Ok(_) => {
                        result.add(PreflightCheck::pass(
                            "stale_database_safety",
                            "Export won't lose JSONL issues",
                            "JSONL is empty or doesn't exist.",
                        ));
                    }
                    Err(e) => {
                        result.add(PreflightCheck::warn(
                            "stale_database_safety",
                            "Export won't lose JSONL issues",
                            format!("Could not read existing JSONL: {e}"),
                            "Verify JSONL file is readable.",
                        ));
                    }
                }
            }
        }
        Err(e) => {
            result.add(PreflightCheck::fail(
                "database_accessible",
                "Database is accessible",
                format!("Database error: {e}"),
                "Check database file permissions and integrity.",
            ));
            tracing::debug!(error = %e, "Database access check: FAIL");
        }
    }

    tracing::debug!(
        overall_status = ?result.overall_status,
        check_count = result.checks.len(),
        failure_count = result.failures().len(),
        "Export preflight complete"
    );

    Ok(result)
}

/// Run preflight checks for import operation.
///
/// This function is read-only and validates:
/// - Beads directory exists
/// - Input path is within allowlist (not in .git, within `beads_dir`)
/// - Input file exists and is readable
/// - No merge conflict markers in input file
/// - JSONL is parseable (basic syntax check)
/// - Issue ID prefixes match expected prefix (unless explicitly skipped)
///
/// # Arguments
///
/// * `input_path` - Source JSONL path
/// * `config` - Import configuration
/// * `expected_prefix` - Expected issue ID prefix (e.g., "bd") for mismatch guardrails
///
/// # Returns
///
/// `PreflightResult` with all check results. Use `.into_result()` to convert
/// failures to an error.
///
/// # Errors
///
/// Returns an error if the preflight checks fail.
#[allow(clippy::too_many_lines)]
pub fn preflight_import(
    input_path: &Path,
    config: &ImportConfig,
    expected_prefix: Option<&str>,
) -> Result<PreflightResult> {
    let mut result = PreflightResult::new();

    tracing::debug!(
        input_path = %input_path.display(),
        beads_dir = ?config.beads_dir,
        "Running import preflight checks"
    );

    // Check 1: Beads directory exists
    if let Some(ref beads_dir) = config.beads_dir {
        if beads_dir.is_dir() {
            result.add(PreflightCheck::pass(
                "beads_dir_exists",
                "Beads directory exists",
                format!("Found: {}", beads_dir.display()),
            ));
            tracing::debug!(beads_dir = %beads_dir.display(), "Beads directory check: PASS");
        } else {
            result.add(PreflightCheck::fail(
                "beads_dir_exists",
                "Beads directory exists",
                format!("Not found: {}", beads_dir.display()),
                "Run 'br init' to initialize the beads directory.",
            ));
            tracing::debug!(beads_dir = %beads_dir.display(), "Beads directory check: FAIL");
        }
    }

    // Check 2: Input path validation (PC-1, PC-2, PC-3, NGI-3)
    if let Some(ref beads_dir) = config.beads_dir {
        // Determine if the path is external (outside .beads/)
        let canonical_beads = dunce::canonicalize(beads_dir).unwrap_or_else(|_| beads_dir.clone());
        let is_external =
            !input_path.starts_with(beads_dir) && !input_path.starts_with(&canonical_beads);

        match validate_sync_path_with_external(input_path, beads_dir, config.allow_external_jsonl) {
            Ok(()) => {
                let msg = format!(
                    "Path {} validated (external={})",
                    input_path.display(),
                    is_external
                );
                if is_external && config.allow_external_jsonl {
                    result.add(PreflightCheck::warn(
                        "path_validation",
                        "Input path is within allowlist",
                        msg,
                        "Consider using JSONL from .beads/ directory for better safety.",
                    ));
                } else {
                    result.add(PreflightCheck::pass(
                        "path_validation",
                        "Input path is within allowlist",
                        msg,
                    ));
                }
                tracing::debug!(path = %input_path.display(), is_external = is_external, "Path validation: PASS");
            }
            Err(e) => {
                result.add(PreflightCheck::fail(
                    "path_validation",
                    "Input path is within allowlist",
                    format!("Path rejected: {e}"),
                    "Use a path within .beads/ directory or set --allow-external-jsonl.",
                ));
                tracing::debug!(path = %input_path.display(), error = %e, "Path validation: FAIL");
            }
        }
    }

    // Check 3: Input file exists and is readable
    if input_path.exists() {
        match File::open(input_path) {
            Ok(_) => {
                result.add(PreflightCheck::pass(
                    "file_readable",
                    "Input file exists and is readable",
                    format!("File accessible: {}", input_path.display()),
                ));
                tracing::debug!(path = %input_path.display(), "File readable check: PASS");
            }
            Err(e) => {
                result.add(PreflightCheck::fail(
                    "file_readable",
                    "Input file exists and is readable",
                    format!("Cannot read file: {e}"),
                    "Check file permissions.",
                ));
                tracing::debug!(path = %input_path.display(), error = %e, "File readable check: FAIL");
            }
        }
    } else {
        result.add(PreflightCheck::fail(
            "file_readable",
            "Input file exists and is readable",
            format!("File not found: {}", input_path.display()),
            "Verify the path is correct or run export first.",
        ));
        tracing::debug!(path = %input_path.display(), "File readable check: FAIL (not found)");
        // Return early since we can't do further checks without the file
        return Ok(result);
    }

    // Check 4: No merge conflict markers
    match scan_conflict_markers(input_path) {
        Ok(markers) if markers.is_empty() => {
            result.add(PreflightCheck::pass(
                "no_conflict_markers",
                "No merge conflict markers",
                "File is clean of conflict markers.",
            ));
            tracing::debug!(path = %input_path.display(), "Conflict marker check: PASS");
        }
        Ok(markers) => {
            let preview: Vec<String> = markers
                .iter()
                .take(3)
                .map(|m| {
                    format!(
                        "line {}: {:?}{}",
                        m.line,
                        m.marker_type,
                        m.branch
                            .as_ref()
                            .map_or(String::new(), |b| format!(" ({b})"))
                    )
                })
                .collect();
            result.add(PreflightCheck::fail(
                "no_conflict_markers",
                "No merge conflict markers",
                format!(
                    "Found {} conflict marker(s): {}{}",
                    markers.len(),
                    preview.join("; "),
                    if markers.len() > 3 { " ..." } else { "" }
                ),
                "Resolve git merge conflicts before importing.",
            ));
            tracing::debug!(
                path = %input_path.display(),
                marker_count = markers.len(),
                "Conflict marker check: FAIL"
            );
        }
        Err(e) => {
            result.add(PreflightCheck::warn(
                "no_conflict_markers",
                "No merge conflict markers",
                format!("Could not scan for markers: {e}"),
                "Verify file is readable and not corrupted.",
            ));
            tracing::debug!(path = %input_path.display(), error = %e, "Conflict marker check: WARN");
        }
    }

    // Check 5: Per-line issue-record validation
    match validate_jsonl_issue_records(input_path) {
        Ok(summary) if summary.invalid_count == 0 => {
            result.add(PreflightCheck::pass(
                "json_valid",
                "All JSONL lines are valid issue records",
                format!("Validated {} issue record(s).", summary.record_count),
            ));
            tracing::debug!(path = %input_path.display(), record_count = summary.record_count, "JSONL issue validation check: PASS");
        }
        Ok(summary) => {
            let preview = summary.preview_messages();
            result.add(PreflightCheck::fail(
                "json_valid",
                "All JSONL lines are valid issue records",
                format!(
                    "Found {} invalid issue record(s): {}{}",
                    summary.invalid_count,
                    preview.join("; "),
                    if summary.invalid_count > preview.len() {
                        " ..."
                    } else {
                        ""
                    }
                ),
                "Fix or remove malformed issue records before importing.",
            ));
            tracing::debug!(
                path = %input_path.display(),
                invalid_count = summary.invalid_count,
                "JSONL issue validation check: FAIL"
            );
        }
        Err(err) => {
            result.add(PreflightCheck::warn(
                "json_valid",
                "All JSONL lines are valid issue records",
                format!("Could not open file for JSONL validation: {err}"),
                "Verify file is readable.",
            ));
        }
    }

    // Check 6: Prefix mismatch guard
    if !config.skip_prefix_validation
        && let Some(prefix) = expected_prefix
    {
        let file = File::open(input_path);
        match file {
            Ok(f) => {
                let reader = BufReader::new(f);
                let mut mismatched_ids: Vec<String> = Vec::new();
                for line_result in reader.lines() {
                    let Ok(line) = line_result else { continue };
                    let trimmed = line.trim();
                    if trimmed.is_empty() {
                        continue;
                    }
                    if let Ok(partial) = serde_json::from_str::<PartialId>(trimmed) {
                        // Skip tombstones — they may retain a foreign prefix legitimately
                        #[derive(Deserialize)]
                        struct StatusProbe {
                            status: Option<String>,
                        }
                        let is_tombstone = serde_json::from_str::<StatusProbe>(trimmed)
                            .ok()
                            .and_then(|p| p.status)
                            .is_some_and(|s| s == "tombstone");
                        if is_tombstone {
                            continue;
                        }
                        if !id_matches_expected_prefix(&partial.id, prefix) {
                            mismatched_ids.push(partial.id);
                        }
                    }
                }
                if mismatched_ids.is_empty() {
                    result.add(PreflightCheck::pass(
                        "prefix_match",
                        "Issue IDs match expected prefix",
                        format!("All issue IDs start with '{prefix}'."),
                    ));
                    tracing::debug!(prefix = prefix, "Prefix match check: PASS");
                } else {
                    let preview: Vec<String> = mismatched_ids.iter().take(5).cloned().collect();
                    result.add(PreflightCheck::fail(
                        "prefix_match",
                        "Issue IDs match expected prefix",
                        format!(
                            "Expected prefix '{}', found {} mismatched ID(s): {}{}",
                            prefix,
                            mismatched_ids.len(),
                            preview.join(", "),
                            if mismatched_ids.len() > 5 { " ..." } else { "" }
                        ),
                        "Use --force to skip prefix validation or --rename-prefix to remap IDs.",
                    ));
                    tracing::debug!(
                        prefix = prefix,
                        mismatch_count = mismatched_ids.len(),
                        "Prefix match check: FAIL"
                    );
                }
            }
            Err(e) => {
                result.add(PreflightCheck::warn(
                    "prefix_match",
                    "Issue IDs match expected prefix",
                    format!("Could not open file for prefix validation: {e}"),
                    "Verify file is readable.",
                ));
            }
        }
    }

    tracing::debug!(
        overall_status = ?result.overall_status,
        check_count = result.checks.len(),
        failure_count = result.failures().len(),
        "Import preflight complete"
    );

    Ok(result)
}

/// Conflict marker kind.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConflictMarkerType {
    Start,
    Separator,
    End,
}

/// A detected merge conflict marker within an import file.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConflictMarker {
    pub path: PathBuf,
    pub line: usize,
    pub marker_type: ConflictMarkerType,
    pub branch: Option<String>,
}

const CONFLICT_START: &str = "<<<<<<<";
const CONFLICT_SEPARATOR: &str = "=======";
const CONFLICT_END: &str = ">>>>>>>";

/// Scan a file for merge conflict markers.
///
/// # Errors
///
/// Returns an error if the file cannot be read.
pub fn scan_conflict_markers(path: &Path) -> Result<Vec<ConflictMarker>> {
    let file = File::open(path)?;
    let reader = BufReader::with_capacity(2 * 1024 * 1024, file);
    let mut markers = Vec::new();

    for (line_num, line) in reader.lines().enumerate() {
        let line = line?;
        if let Some((marker_type, branch)) = detect_conflict_marker(&line) {
            markers.push(ConflictMarker {
                path: path.to_path_buf(),
                line: line_num + 1,
                marker_type,
                branch,
            });
        }
    }

    Ok(markers)
}

fn detect_conflict_marker(line: &str) -> Option<(ConflictMarkerType, Option<String>)> {
    if let Some(branch) = line.strip_prefix(CONFLICT_START) {
        return Some((ConflictMarkerType::Start, Some(branch.trim().to_string())));
    }
    if line.starts_with(CONFLICT_SEPARATOR) {
        return Some((ConflictMarkerType::Separator, None));
    }
    if let Some(branch) = line.strip_prefix(CONFLICT_END) {
        return Some((ConflictMarkerType::End, Some(branch.trim().to_string())));
    }
    None
}

/// Fail if a file contains merge conflict markers.
///
/// # Errors
///
/// Returns a config error describing the first few markers found.
pub fn ensure_no_conflict_markers(path: &Path) -> Result<()> {
    let markers = scan_conflict_markers(path)?;
    if markers.is_empty() {
        return Ok(());
    }

    let mut preview = String::new();
    for marker in markers.iter().take(5) {
        let _ = writeln!(
            preview,
            "{}:{} {:?}{}",
            marker.path.display(),
            marker.line,
            marker.marker_type,
            marker
                .branch
                .as_ref()
                .map_or(String::new(), |b| format!(" ({b})"))
        );
    }

    Err(BeadsError::Config(format!(
        "Merge conflict markers detected in {}.\n{}Resolve conflicts before importing.",
        path.display(),
        preview
    )))
}

#[derive(Deserialize)]
struct PartialId {
    id: String,
}

/// Analyze JSONL to get line count and unique issue IDs efficiently.
///
/// # Errors
///
/// Returns an error if the file cannot be read or contains invalid JSON.
pub fn analyze_jsonl(path: &Path) -> Result<(usize, HashSet<String>)> {
    let file = match File::open(path) {
        Ok(f) => f,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok((0, HashSet::new())),
        Err(e) => return Err(BeadsError::Io(e)),
    };

    let mut reader = BufReader::new(file);
    let mut count = 0;
    let mut ids = HashSet::new();
    let mut line_buf = String::new();
    let mut line_num = 0;

    loop {
        line_buf.clear();
        let bytes = reader.read_line(&mut line_buf)?;
        if bytes == 0 {
            break;
        }

        line_num += 1;
        let trimmed = line_buf.trim_end_matches(['\n', '\r']);
        if trimmed.trim().is_empty() {
            continue;
        }

        let partial: PartialId = serde_json::from_str(trimmed)
            .map_err(|e| BeadsError::Config(format!("Invalid JSON at line {}: {}", line_num, e)))?;

        if !ids.insert(partial.id.clone()) {
            return Err(BeadsError::Config(format!(
                "Duplicate issue id '{}' in {} at line {}",
                partial.id,
                path.display(),
                line_num
            )));
        }
        count += 1;
    }

    Ok((count, ids))
}

/// Count issues in an existing JSONL file.
///
/// # Errors
///
/// Returns an error if the file cannot be read or contains invalid JSON.
pub fn count_issues_in_jsonl(path: &Path) -> Result<usize> {
    Ok(analyze_jsonl(path)?.0)
}

/// Get issue IDs from an existing JSONL file.
///
/// # Errors
///
/// Returns an error if the file cannot be read or contains invalid JSON.
pub fn get_issue_ids_from_jsonl(path: &Path) -> Result<HashSet<String>> {
    Ok(analyze_jsonl(path)?.1)
}

fn read_jsonl_lines_by_id(path: &Path) -> Result<BTreeMap<String, String>> {
    let file = File::open(path)?;
    let mut reader = BufReader::new(file);
    let mut lines_by_id = BTreeMap::new();
    let mut line_buf = String::new();
    let mut line_num = 0;

    loop {
        line_buf.clear();
        let bytes = reader.read_line(&mut line_buf)?;
        if bytes == 0 {
            break;
        }

        line_num += 1;
        let trimmed = line_buf.trim_end_matches(['\n', '\r']);
        if trimmed.trim().is_empty() {
            continue;
        }

        let partial: PartialId = serde_json::from_str(trimmed)
            .map_err(|e| BeadsError::Config(format!("Invalid JSON at line {}: {}", line_num, e)))?;

        if lines_by_id
            .insert(partial.id.clone(), trimmed.to_string())
            .is_some()
        {
            return Err(BeadsError::Config(format!(
                "Duplicate issue id '{}' in {} at line {}",
                partial.id,
                path.display(),
                line_num
            )));
        }
    }

    Ok(lines_by_id)
}

/// Export issues from `SQLite` to JSONL format.
///
/// This implements the classic beads export semantics:
/// - Include tombstones (for sync propagation)
/// - Exclude ephemerals/wisps
/// - Sort by ID for deterministic output
/// - Populate dependencies and labels for each issue
/// - Atomic write (temp file -> rename)
/// - Safety guard against empty DB overwriting non-empty JSONL
///
/// # Errors
///
/// Returns an error if:
/// - Database read fails
/// - Safety guard is violated (empty DB, non-empty JSONL, no force)
/// - File write fails
#[allow(clippy::too_many_lines)]
pub fn export_to_jsonl(
    storage: &SqliteStorage,
    output_path: &Path,
    config: &ExportConfig,
) -> Result<ExportResult> {
    let (result, _report) = export_to_jsonl_with_policy(storage, output_path, config)?;
    Ok(result)
}

/// Export issues with configurable error policy, returning a report.
///
/// # Errors
///
/// Returns an error if:
/// - Path validation fails (git path, outside `beads_dir` without opt-in)
/// - Database queries fail and the policy requires strict handling
/// - Safety guards are violated (empty/stale export without `force`)
/// - File I/O fails
#[allow(clippy::too_many_lines)]
pub fn export_to_jsonl_with_policy(
    storage: &SqliteStorage,
    output_path: &Path,
    config: &ExportConfig,
) -> Result<(ExportResult, ExportReport)> {
    // Path validation (PC-1, PC-2, PC-3, NGI-3)
    if let Some(ref beads_dir) = config.beads_dir {
        validate_sync_path_with_external(output_path, beads_dir, config.allow_external_jsonl)?;
        tracing::debug!(
            output_path = %output_path.display(),
            beads_dir = %beads_dir.display(),
            allow_external = config.allow_external_jsonl,
            "Export path validated"
        );

        // Perform backup before overwriting (if enabled and we have a beads_dir).
        // We backup any JSONL file that has been validated as safe for sync,
        // even if it's outside the .beads/ directory (e.g., in repo root).
        let output_abs = if output_path.is_absolute() {
            output_path.to_path_buf()
        } else if let Ok(cwd) = std::env::current_dir() {
            cwd.join(output_path)
        } else {
            output_path.to_path_buf()
        };

        history::backup_before_export(beads_dir, &config.history, &output_abs)?;
    }

    // Get all issues for export (sorted by ID, excludes ephemerals/wisps)
    let mut issues = storage.get_all_issues_for_export()?;

    // Fetch dirty metadata for safe clearing later
    let dirty_metadata = storage.get_dirty_issue_metadata()?;

    // Safety checks
    if !config.force && output_path.exists() {
        let (jsonl_count, jsonl_ids) = analyze_jsonl(output_path)?;

        // Check 1: prevent exporting empty database over non-empty JSONL
        if issues.is_empty() && jsonl_count > 0 {
            return Err(BeadsError::Config(format!(
                "Refusing to export empty database over non-empty JSONL file.\n\
                 Database has 0 issues, JSONL has {jsonl_count} lines.\n\
                 This would result in data loss!\n\
                 Hint: Use --force to override this safety check."
            )));
        }

        // Check 2: prevent exporting stale database that would lose issues
        if !jsonl_ids.is_empty() {
            let db_ids: HashSet<String> = issues.iter().map(|i| i.id.clone()).collect();
            let missing: Vec<_> = jsonl_ids.difference(&db_ids).collect();

            if !missing.is_empty() {
                let mut missing_list = missing.into_iter().cloned().collect::<Vec<_>>();
                missing_list.sort();
                let display_count = missing_list.len().min(10);
                let preview: Vec<_> = missing_list.iter().take(display_count).collect();
                let more = if missing_list.len() > 10 {
                    format!(" ... and {} more", missing_list.len() - 10)
                } else {
                    String::new()
                };

                return Err(BeadsError::Config(format!(
                    "Refusing to export stale database that would lose issues.\n\
                     Database has {} issues, JSONL has {} unique issues.\n\
                     Export would lose {} issue(s): {}{}\n\
                     Hint: Run import first, or use --force to override.",
                    issues.len(),
                    jsonl_ids.len(),
                    missing_list.len(),
                    preview
                        .iter()
                        .map(|s| s.as_str())
                        .collect::<Vec<_>>()
                        .join(", "),
                    more
                )));
            }
        }
    }

    let mut ctx = ExportContext::new(config.error_policy);
    let mut report = ExportReport::new(config.error_policy);

    let progress = create_progress_bar(
        issues.len() as u64,
        "Exporting issues",
        config.show_progress,
    );

    // Populate dependencies and labels for all issues (batch queries to avoid N+1)
    let all_deps = match storage.get_all_dependency_records() {
        Ok(map) => Some(map),
        Err(err) => {
            ctx.handle_error(ExportError::new(
                ExportEntityType::Dependency,
                "all",
                err.to_string(),
            ))?;
            None
        }
    };
    let all_labels = match storage.get_all_labels() {
        Ok(map) => Some(map),
        Err(err) => {
            ctx.handle_error(ExportError::new(
                ExportEntityType::Label,
                "all",
                err.to_string(),
            ))?;
            None
        }
    };
    let all_comments = match storage.get_all_comments() {
        Ok(map) => Some(map),
        Err(err) => {
            ctx.handle_error(ExportError::new(
                ExportEntityType::Comment,
                "all",
                err.to_string(),
            ))?;
            None
        }
    };

    for issue in &mut issues {
        // Dependencies
        if let Some(ref map) = all_deps {
            if let Some(deps) = map.get(&issue.id) {
                issue.dependencies = deps.clone();
            }
        } else if ctx.policy != ExportErrorPolicy::RequiredCore {
            // Bulk failed, but we are in best-effort/partial mode — try individual query
            if let Ok(deps) = storage.get_dependencies_full(&issue.id) {
                issue.dependencies = deps;
            }
        }

        // Labels
        if let Some(ref map) = all_labels {
            if let Some(labels) = map.get(&issue.id) {
                issue.labels = labels.clone();
            }
        } else if ctx.policy != ExportErrorPolicy::RequiredCore
            && let Ok(labels) = storage.get_labels(&issue.id)
        {
            issue.labels = labels;
        }

        // Normalize for consistent round-trip hashing (matches import behavior)
        normalize_issue_for_export(issue);

        // Comments
        if let Some(ref map) = all_comments {
            if let Some(comments) = map.get(&issue.id) {
                issue.comments = comments.clone();
            }
        } else if ctx.policy != ExportErrorPolicy::RequiredCore
            && let Ok(comments) = storage.get_comments(&issue.id)
        {
            issue.comments = comments;
        }
    }

    // Write to temp file for atomic rename
    let parent_dir = output_path.parent().ok_or_else(|| {
        BeadsError::Config(format!("Invalid output path: {}", output_path.display()))
    })?;

    // Ensure parent directory exists
    fs::create_dir_all(parent_dir)?;

    let temp_path = export_temp_path(output_path);

    // Validate temp file path (PC-4: temp files must be in same directory as target)
    if let Some(ref beads_dir) = config.beads_dir {
        validate_temp_file_path(
            &temp_path,
            output_path,
            beads_dir,
            config.allow_external_jsonl,
        )?;
        tracing::debug!(
            temp_path = %temp_path.display(),
            target_path = %output_path.display(),
            "Temp file path validated"
        );
    }

    let temp_file = OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(&temp_path)
        .map_err(|err| {
            if err.kind() == std::io::ErrorKind::AlreadyExists {
                BeadsError::Config(format!(
                    "Temporary export file already exists: {}",
                    temp_path.display()
                ))
            } else {
                err.into()
            }
        })?;
    let mut temp_guard = TempFileGuard::new(temp_path.clone());
    let mut writer = BufWriter::new(temp_file);

    // Write JSONL and compute hash
    let mut hasher = Sha256::new();
    let issues_len = issues.len();
    let mut exported_ids = Vec::with_capacity(issues_len);
    let mut skipped_tombstone_ids = Vec::new(); // Usually small
    let mut issue_hashes = Vec::with_capacity(issues_len);
    let mut buffer = Vec::with_capacity(1024);

    for issue in &issues {
        // Skip expired tombstones
        if issue.is_expired_tombstone(config.retention_days) {
            skipped_tombstone_ids.push(issue.id.clone());
            progress.inc(1);
            continue;
        }

        buffer.clear();
        if let Err(err) = serde_json::to_writer(&mut buffer, issue) {
            ctx.handle_error(ExportError::new(
                ExportEntityType::Issue,
                issue.id.clone(),
                err.to_string(),
            ))?;
            progress.inc(1);
            continue;
        }

        if let Err(err) = writer
            .write_all(&buffer)
            .and_then(|()| writer.write_all(b"\n"))
        {
            ctx.handle_error(ExportError::new(
                ExportEntityType::Issue,
                issue.id.clone(),
                err.to_string(),
            ))?;
            progress.inc(1);
            continue;
        }

        hasher.update(&buffer);
        hasher.update(b"\n");

        exported_ids.push(issue.id.clone());
        issue_hashes.push((
            issue.id.clone(),
            issue
                .content_hash
                .clone()
                .unwrap_or_else(|| crate::util::content_hash(issue)),
        ));
        report.issues_exported += 1;
        report.dependencies_exported += issue.dependencies.len();
        report.labels_exported += issue.labels.len();
        report.comments_exported += issue.comments.len();
        progress.inc(1);
    }

    progress.finish_with_message("Export complete");

    // Flush and sync
    writer.flush()?;
    writer
        .into_inner()
        .map_err(|e| BeadsError::Io(e.into_error()))?
        .sync_all()?;

    // Compute final hash
    let content_hash = format!("{:x}", hasher.finalize());

    // Verify staged export integrity before replacing the live JSONL.
    let actual_count = count_issues_in_jsonl(&temp_path)?;
    if actual_count != exported_ids.len() {
        return Err(BeadsError::Config(format!(
            "Export verification failed: expected {} issues, JSONL has {} lines",
            exported_ids.len(),
            actual_count
        )));
    }

    if let Some(ref beads_dir) = config.beads_dir {
        require_safe_sync_overwrite_path(
            &temp_path,
            beads_dir,
            config.allow_external_jsonl,
            "rename temp file",
        )?;
        require_safe_sync_overwrite_path(
            output_path,
            beads_dir,
            config.allow_external_jsonl,
            "overwrite JSONL output",
        )?;
    }

    // Atomic rename
    fs::rename(&temp_path, output_path)?;
    temp_guard.persist();

    // Set file permissions (0600)
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        let perms = std::fs::Permissions::from_mode(0o600);
        if let Err(e) = fs::set_permissions(output_path, perms) {
            tracing::warn!(
                path = %output_path.display(),
                error = %e,
                "Failed to set restrictive permissions on exported JSONL file"
            );
        }
    }

    let result = ExportResult {
        exported_count: exported_ids.len(),
        exported_marked_at: filter_dirty_metadata_for_export(
            &dirty_metadata,
            &exported_ids,
            &skipped_tombstone_ids,
        ),
        exported_ids,
        skipped_tombstone_ids,
        content_hash,
        output_path: Some(output_path.to_string_lossy().to_string()),
        issue_hashes,
    };

    report.errors = ctx.errors;

    Ok((result, report))
}

/// Export issues to a writer (e.g., stdout).
///
/// # Errors
///
/// Returns an error if serialization or writing fails.
pub fn export_to_writer<W: Write>(storage: &SqliteStorage, writer: &mut W) -> Result<ExportResult> {
    let (result, _report) =
        export_to_writer_with_policy(storage, writer, ExportErrorPolicy::Strict)?;
    Ok(result)
}

/// Export issues to a writer with configurable error policy.
///
/// # Errors
///
/// Returns an error if serialization or writing fails under a strict policy.
#[allow(clippy::too_many_lines)]
pub fn export_to_writer_with_policy<W: Write>(
    storage: &SqliteStorage,
    writer: &mut W,
    policy: ExportErrorPolicy,
) -> Result<(ExportResult, ExportReport)> {
    let mut issues = storage.get_all_issues_for_export()?;

    // Populate dependencies and labels
    let mut ctx = ExportContext::new(policy);
    let mut report = ExportReport::new(policy);
    let all_deps = match storage.get_all_dependency_records() {
        Ok(map) => Some(map),
        Err(err) => {
            ctx.handle_error(ExportError::new(
                ExportEntityType::Dependency,
                "all",
                err.to_string(),
            ))?;
            None
        }
    };
    let all_labels = match storage.get_all_labels() {
        Ok(map) => Some(map),
        Err(err) => {
            ctx.handle_error(ExportError::new(
                ExportEntityType::Label,
                "all",
                err.to_string(),
            ))?;
            None
        }
    };
    let all_comments = match storage.get_all_comments() {
        Ok(map) => Some(map),
        Err(err) => {
            ctx.handle_error(ExportError::new(
                ExportEntityType::Comment,
                "all",
                err.to_string(),
            ))?;
            None
        }
    };

    for issue in &mut issues {
        // Dependencies
        if let Some(ref map) = all_deps {
            if let Some(deps) = map.get(&issue.id) {
                issue.dependencies = deps.clone();
            }
        } else if ctx.policy != ExportErrorPolicy::RequiredCore {
            // Bulk failed, but we are in best-effort/partial mode — try individual query
            if let Ok(deps) = storage.get_dependencies_full(&issue.id) {
                issue.dependencies = deps;
            }
        }

        // Labels
        if let Some(ref map) = all_labels {
            if let Some(labels) = map.get(&issue.id) {
                issue.labels = labels.clone();
            }
        } else if ctx.policy != ExportErrorPolicy::RequiredCore
            && let Ok(labels) = storage.get_labels(&issue.id)
        {
            issue.labels = labels;
        }

        // Normalize for consistent round-trip hashing (matches import behavior)
        normalize_issue_for_export(issue);

        // Comments
        if let Some(ref map) = all_comments {
            if let Some(comments) = map.get(&issue.id) {
                issue.comments = comments.clone();
            }
        } else if ctx.policy != ExportErrorPolicy::RequiredCore
            && let Ok(comments) = storage.get_comments(&issue.id)
        {
            issue.comments = comments;
        }
    }

    let mut hasher = Sha256::new();
    let issues_len = issues.len();
    let mut exported_ids = Vec::with_capacity(issues_len);
    let skipped_tombstone_ids = Vec::new();
    let mut issue_hashes = Vec::with_capacity(issues_len);

    for issue in &issues {
        let json = match serde_json::to_string(issue) {
            Ok(json) => json,
            Err(err) => {
                ctx.handle_error(ExportError::new(
                    ExportEntityType::Issue,
                    issue.id.clone(),
                    err.to_string(),
                ))?;
                continue;
            }
        };
        if let Err(err) = writeln!(writer, "{json}") {
            ctx.handle_error(ExportError::new(
                ExportEntityType::Issue,
                issue.id.clone(),
                err.to_string(),
            ))?;
            continue;
        }
        hasher.update(json.as_bytes());
        hasher.update(b"\n");

        exported_ids.push(issue.id.clone());
        issue_hashes.push((
            issue.id.clone(),
            issue
                .content_hash
                .clone()
                .unwrap_or_else(|| crate::util::content_hash(issue)),
        ));
        report.issues_exported += 1;
        report.dependencies_exported += issue.dependencies.len();
        report.labels_exported += issue.labels.len();
        report.comments_exported += issue.comments.len();
    }

    let content_hash = format!("{:x}", hasher.finalize());

    let result = ExportResult {
        exported_count: exported_ids.len(),
        exported_ids,
        exported_marked_at: Vec::new(),
        skipped_tombstone_ids,
        content_hash,
        output_path: None,
        issue_hashes,
    };

    report.errors = ctx.errors;

    Ok((result, report))
}

/// Metadata key for the JSONL content hash.
pub const METADATA_JSONL_CONTENT_HASH: &str = "jsonl_content_hash";
/// Metadata key for the exact observed JSONL mtime at the last successful sync.
pub const METADATA_JSONL_MTIME: &str = "jsonl_mtime";
/// Metadata key for the exact observed JSONL size at the last successful sync.
pub const METADATA_JSONL_SIZE: &str = "jsonl_size";
/// Metadata key for the last export time.
pub const METADATA_LAST_EXPORT_TIME: &str = "last_export_time";
/// Metadata key for the last import time.
pub const METADATA_LAST_IMPORT_TIME: &str = "last_import_time";

#[derive(Debug, Clone)]
struct JsonlWitness {
    mtime: std::time::SystemTime,
    mtime_witness: String,
    size: u64,
}

/// Result of a staleness check between JSONL and DB.
#[derive(Debug, Clone, Copy)]
pub struct StalenessCheck {
    pub dirty_count: usize,
    pub jsonl_exists: bool,
    pub jsonl_mtime: Option<std::time::SystemTime>,
    pub jsonl_newer: bool,
    pub db_newer: bool,
}

fn pending_export_state(
    storage: &SqliteStorage,
    jsonl_exists: bool,
) -> Result<(usize, bool, bool)> {
    let dirty_count = storage.get_dirty_issue_count()?;
    let needs_flush = storage.get_metadata("needs_flush")?.as_deref() == Some("true");
    let missing_jsonl_with_data = !jsonl_exists && storage.count_issues()? > 0;
    Ok((
        dirty_count,
        needs_flush,
        dirty_count > 0 || needs_flush || missing_jsonl_with_data,
    ))
}

/// Compute staleness based on JSONL mtime + content hash and DB dirty state.
///
/// Uses Lstat (`symlink_metadata`) for JSONL mtime to match classic bd behavior.
///
/// # Errors
///
/// Returns an error if reading dirty state, metadata, JSONL mtime, or hashing fails.
pub fn compute_staleness(storage: &SqliteStorage, jsonl_path: &Path) -> Result<StalenessCheck> {
    let (staleness, _) = compute_staleness_impl(storage, jsonl_path)?;
    Ok(staleness)
}

/// Compute staleness and opportunistically persist refreshed JSONL witnesses.
///
/// When the stored content hash still matches but the cached mtime/size witness
/// is stale or incomplete, this updates the metadata so later commands can skip
/// re-hashing an unchanged JSONL file.
///
/// # Errors
///
/// Returns an error if reading dirty state, metadata, JSONL metadata, or
/// hashing fails. Opportunistic witness refresh failures are logged and
/// ignored so startup freshness probes do not fail on metadata backfill races.
pub fn compute_staleness_refreshing_witnesses(
    storage: &mut SqliteStorage,
    jsonl_path: &Path,
) -> Result<StalenessCheck> {
    let (staleness, refresh_witness) = compute_staleness_impl(storage, jsonl_path)?;
    if let Some(observed) = refresh_witness {
        refresh_jsonl_witness_best_effort(storage, jsonl_path, &observed);
    }
    Ok(staleness)
}

fn compute_staleness_impl(
    storage: &SqliteStorage,
    jsonl_path: &Path,
) -> Result<(StalenessCheck, Option<JsonlWitness>)> {
    let jsonl_exists = jsonl_path.exists();
    let (dirty_count, _needs_flush, db_newer) = pending_export_state(storage, jsonl_exists)?;
    let mut refresh_witness = None;

    let (jsonl_mtime, jsonl_newer) = if jsonl_exists {
        let observed = observed_jsonl_witness(jsonl_path)?;
        let stored_mtime = storage.get_metadata(METADATA_JSONL_MTIME)?;
        let stored_size = storage.get_metadata(METADATA_JSONL_SIZE)?;
        let stored_hash = storage.get_metadata(METADATA_JSONL_CONTENT_HASH)?;

        if stored_mtime.as_deref() == Some(observed.mtime_witness.as_str()) {
            let stored_size_matches =
                stored_size.as_deref().and_then(parse_jsonl_size_witness) == Some(observed.size);
            let jsonl_newer = if stored_size_matches {
                stored_hash.is_none()
            } else {
                stored_hash.as_ref().is_none_or(|hash| {
                    compute_jsonl_hash(jsonl_path)
                        .map_or(true, |current_hash| &current_hash != hash)
                })
            };

            if !jsonl_newer && stored_hash.is_some() && !stored_size_matches {
                refresh_witness = Some(observed.clone());
            }

            return Ok((
                StalenessCheck {
                    dirty_count,
                    jsonl_exists: true,
                    jsonl_mtime: Some(observed.mtime),
                    jsonl_newer,
                    db_newer,
                },
                refresh_witness,
            ));
        }

        let last_import_time = storage.get_metadata(METADATA_LAST_IMPORT_TIME)?;
        let last_export_time = storage.get_metadata(METADATA_LAST_EXPORT_TIME)?;

        // Get the latest known sync time (either import or export)
        let mut latest_sync_ts: Option<chrono::DateTime<Utc>> = None;

        if let Some(import_time) = &last_import_time
            && let Ok(ts) = chrono::DateTime::parse_from_rfc3339(import_time)
        {
            latest_sync_ts = Some(ts.with_timezone(&Utc));
        }

        if let Some(export_time) = &last_export_time
            && let Ok(ts) = chrono::DateTime::parse_from_rfc3339(export_time)
        {
            let ts_utc = ts.with_timezone(&Utc);
            if latest_sync_ts.is_none_or(|latest| ts_utc > latest) {
                latest_sync_ts = Some(ts_utc);
            }
        }

        // JSONL is newer if it was modified after the latest sync
        // If metadata is missing or invalid, assume JSONL is newer (safe default)
        let mtime_newer = latest_sync_ts.is_none_or(|sync_ts| {
            let sync_sys_time = std::time::SystemTime::from(sync_ts);
            observed.mtime > sync_sys_time
        });

        let jsonl_newer = if mtime_newer {
            stored_hash.as_ref().is_none_or(|stored_hash| {
                compute_jsonl_hash(jsonl_path)
                    .map_or(true, |current_hash| &current_hash != stored_hash)
            })
        } else {
            false
        };

        if !jsonl_newer && stored_hash.is_some() {
            let stored_size_matches =
                stored_size.as_deref().and_then(parse_jsonl_size_witness) == Some(observed.size);
            if stored_mtime.as_deref() != Some(observed.mtime_witness.as_str())
                || !stored_size_matches
            {
                refresh_witness = Some(observed.clone());
            }
        }

        (Some(observed.mtime), jsonl_newer)
    } else {
        (None, false)
    };

    Ok((
        StalenessCheck {
            dirty_count,
            jsonl_exists,
            jsonl_mtime,
            jsonl_newer,
            db_newer,
        },
        refresh_witness,
    ))
}

#[cfg(test)]
fn observed_jsonl_mtime(jsonl_path: &Path) -> Result<(std::time::SystemTime, String)> {
    let observed = observed_jsonl_witness(jsonl_path)?;
    Ok((observed.mtime, observed.mtime_witness))
}

fn observed_jsonl_witness(jsonl_path: &Path) -> Result<JsonlWitness> {
    let metadata = fs::symlink_metadata(jsonl_path)?;
    let jsonl_mtime = metadata.modified()?;
    Ok(JsonlWitness {
        mtime: jsonl_mtime,
        mtime_witness: chrono::DateTime::<Utc>::from(jsonl_mtime).to_rfc3339(),
        size: metadata.len(),
    })
}

fn parse_jsonl_size_witness(value: &str) -> Option<u64> {
    value.parse().ok()
}

fn record_jsonl_witness_in_tx(storage: &SqliteStorage, jsonl_path: &Path) -> Result<()> {
    let observed = observed_jsonl_witness(jsonl_path)?;
    record_observed_jsonl_witness_in_tx(storage, &observed)
}

fn record_observed_jsonl_witness_in_tx(
    storage: &SqliteStorage,
    observed: &JsonlWitness,
) -> Result<()> {
    storage.set_metadata_in_tx(METADATA_JSONL_MTIME, &observed.mtime_witness)?;
    storage.set_metadata_in_tx(METADATA_JSONL_SIZE, &observed.size.to_string())
}

fn maybe_refresh_jsonl_witness(
    storage: &mut SqliteStorage,
    jsonl_path: &Path,
    observed: &JsonlWitness,
) -> Result<()> {
    let current = observed_jsonl_witness(jsonl_path)?;
    if current.mtime != observed.mtime || current.size != observed.size {
        return Ok(());
    }

    storage.with_write_transaction(|storage| record_observed_jsonl_witness_in_tx(storage, &current))
}

fn refresh_jsonl_witness_best_effort(
    storage: &mut SqliteStorage,
    jsonl_path: &Path,
    observed: &JsonlWitness,
) {
    if let Err(error) = maybe_refresh_jsonl_witness(storage, jsonl_path, observed) {
        tracing::debug!(
            path = %jsonl_path.display(),
            error = %error,
            "Skipping opportunistic JSONL witness refresh"
        );
    }
}

/// Result of an auto-import attempt.
#[derive(Debug, Default)]
pub struct AutoImportResult {
    /// Whether an import was attempted.
    pub attempted: bool,
    /// Number of issues imported (created or updated).
    pub imported_count: usize,
}

/// Auto-import JSONL if it is newer than the DB.
///
/// Honors `--no-auto-import` and `--allow-stale` behavior.
/// Both flags short-circuit before any staleness probe so startup can skip the
/// JSONL stat/hash path entirely when the caller explicitly opted out.
///
/// # Errors
///
/// Returns an error if staleness checks, metadata reads, or import steps fail.
pub fn auto_import_if_stale(
    storage: &mut SqliteStorage,
    beads_dir: &Path,
    jsonl_path: &Path,
    expected_prefix: Option<&str>,
    allow_external_jsonl: bool,
    allow_stale: bool,
    no_auto_import: bool,
) -> Result<AutoImportResult> {
    if allow_stale || no_auto_import {
        tracing::debug!(
            allow_stale,
            no_auto_import,
            "Skipping auto-import staleness probe due to startup override"
        );
        return Ok(AutoImportResult::default());
    }

    let staleness = compute_staleness_refreshing_witnesses(storage, jsonl_path)?;
    if !staleness.jsonl_newer {
        return Ok(AutoImportResult::default());
    }

    // When both JSONL and DB have changed, skip the auto-import with a
    // warning instead of failing the command.  This prevents spurious
    // SyncConflict errors when ≥3 concurrent `br` processes race: one
    // process flushes JSONL while another has pending local writes,
    // causing both `jsonl_newer` and `db_newer` to be true.
    //
    // Explicit `br sync` still detects this as a hard conflict so the
    // user can reconcile manually.
    if staleness.db_newer && !allow_stale {
        tracing::warn!(
            dirty_count = staleness.dirty_count,
            jsonl_mtime = ?staleness.jsonl_mtime,
            "Skipping auto-import: JSONL changed externally while {} local change(s) are pending. \
             Run `br sync` to reconcile.",
            staleness.dirty_count,
        );
        return Ok(AutoImportResult::default());
    }

    let import_config = ImportConfig {
        // The configured prefix is the default for new IDs, not a project-wide
        // invariant. Auto-import should preserve mixed-prefix workspaces.
        skip_prefix_validation: true,
        beads_dir: Some(beads_dir.to_path_buf()),
        allow_external_jsonl,
        show_progress: false,
        ..Default::default()
    };

    let result = import_from_jsonl(storage, jsonl_path, &import_config, expected_prefix)?;

    tracing::debug!(
        imported_count = result.imported_count,
        jsonl_path = %jsonl_path.display(),
        "Auto-import completed"
    );

    Ok(AutoImportResult {
        attempted: true,
        imported_count: result.imported_count,
    })
}

/// Finalize an export by updating metadata, clearing dirty flags, and recording export hashes.
///
/// This should be called after a successful export to the default JSONL path.
/// It performs the following updates:
/// - Clears dirty flags for the exported issue IDs
/// - Records export hashes for each exported issue (for incremental export)
/// - Updates `jsonl_content_hash` metadata with the export hash
/// - Updates `last_export_time` metadata with the current timestamp
///
/// # Errors
///
/// Returns an error if database updates fail.
pub fn finalize_export(
    storage: &mut SqliteStorage,
    result: &ExportResult,
    issue_hashes: Option<&[(String, String)]>,
    jsonl_path: &Path,
) -> Result<()> {
    use chrono::Utc;

    storage.with_write_transaction(|storage| -> Result<()> {
        // Clear dirty flags for exported issues (safe version with timestamp validation)
        if !result.exported_marked_at.is_empty() {
            storage.clear_dirty_issues(&result.exported_marked_at)?;
        }

        // Record export hashes for each exported issue (for incremental export detection)
        if let Some(hashes) = issue_hashes {
            storage.set_export_hashes_in_tx(hashes)?;
        }

        // Update metadata
        storage.set_metadata_in_tx(METADATA_JSONL_CONTENT_HASH, &result.content_hash)?;
        storage.set_metadata_in_tx(METADATA_LAST_EXPORT_TIME, &Utc::now().to_rfc3339())?;
        record_jsonl_witness_in_tx(storage, jsonl_path)?;

        // Keep the row stable and clear the flag in place so ordinary export
        // cycles avoid delete+insert churn on the metadata B-tree.
        storage.set_metadata_in_tx("needs_flush", "false")?;

        Ok(())
    })?;

    Ok(())
}

fn normalize_issue_for_export(issue: &mut Issue) {
    if !issue.labels.is_empty() {
        issue.labels.sort_unstable();
        issue.labels.dedup();
    }

    if !issue.dependencies.is_empty() {
        issue.dependencies.sort_by(|left, right| {
            left.issue_id
                .cmp(&right.issue_id)
                .then_with(|| left.depends_on_id.cmp(&right.depends_on_id))
                .then_with(|| left.dep_type.as_str().cmp(right.dep_type.as_str()))
                .then_with(|| left.created_at.cmp(&right.created_at))
                .then_with(|| left.created_by.cmp(&right.created_by))
                .then_with(|| left.metadata.cmp(&right.metadata))
                .then_with(|| left.thread_id.cmp(&right.thread_id))
        });
    }

    if !issue.comments.is_empty() {
        issue.comments.sort_by(|left, right| {
            left.issue_id
                .cmp(&right.issue_id)
                .then_with(|| left.created_at.cmp(&right.created_at))
                .then_with(|| left.author.cmp(&right.author))
                .then_with(|| left.body.cmp(&right.body))
                .then_with(|| left.id.cmp(&right.id))
        });
    }
}

fn filter_dirty_metadata_for_export(
    dirty_metadata: &[(String, String)],
    exported_ids: &[String],
    skipped_tombstone_ids: &[String],
) -> Vec<(String, String)> {
    let dirty_by_id: HashMap<&str, &str> = dirty_metadata
        .iter()
        .map(|(issue_id, marked_at)| (issue_id.as_str(), marked_at.as_str()))
        .collect();

    exported_ids
        .iter()
        .chain(skipped_tombstone_ids.iter())
        .filter_map(|issue_id| {
            dirty_by_id
                .get(issue_id.as_str())
                .map(|marked_at| (issue_id.clone(), (*marked_at).to_string()))
        })
        .collect()
}

fn restore_foreign_keys_after_import(
    storage: &SqliteStorage,
    validate_integrity: bool,
) -> Result<()> {
    storage
        .execute_raw("PRAGMA foreign_keys = ON")
        .map_err(|source| BeadsError::WithContext {
            context: "Failed to re-enable foreign key enforcement after import".to_string(),
            source: Box::new(source),
        })?;

    let foreign_keys_enabled = storage
        .execute_raw_query("PRAGMA foreign_keys")
        .map_err(|source| BeadsError::WithContext {
            context: "Failed to verify foreign key enforcement state after import".to_string(),
            source: Box::new(source),
        })?
        .first()
        .and_then(|row| row.first())
        .and_then(SqliteValue::as_integer)
        .unwrap_or(0);

    if foreign_keys_enabled != 1 {
        return Err(BeadsError::Other(anyhow::anyhow!(
            "Import completed with foreign key enforcement still disabled"
        )));
    }

    if !validate_integrity {
        return Ok(());
    }

    if let Some((table, column)) = find_post_import_fk_violation(storage)? {
        return Err(BeadsError::Other(anyhow::anyhow!(
            "Import finished with orphaned rows in {table}.{column}"
        )));
    }

    Ok(())
}

fn find_post_import_fk_violation(storage: &SqliteStorage) -> Result<Option<(String, String)>> {
    let fk_backed_tables = [
        ("dependencies", "issue_id"),
        ("labels", "issue_id"),
        ("comments", "issue_id"),
        ("events", "issue_id"),
        ("dirty_issues", "issue_id"),
        ("export_hashes", "issue_id"),
        ("blocked_issues_cache", "issue_id"),
        ("child_counters", "parent_id"),
    ];

    for (table, column) in fk_backed_tables {
        let has_orphan = storage
            .has_missing_issue_reference(table, column)
            .map_err(|source| BeadsError::WithContext {
                context: format!(
                    "Failed to verify import integrity for foreign-key-backed table {table}.{column}"
                ),
                source: Box::new(source),
            })?;

        if has_orphan {
            return Ok(Some((table.to_string(), column.to_string())));
        }
    }

    Ok(None)
}

fn is_issue_exportable(issue: &Issue, retention_days: Option<u64>) -> bool {
    !issue.ephemeral && !issue.id.contains("-wisp-") && !issue.is_expired_tombstone(retention_days)
}

fn finalize_incremental_auto_flush(
    storage: &mut SqliteStorage,
    clear_dirty_metadata: &[(String, String)],
    removed_hash_ids: &[String],
    issue_hashes: &[(String, String)],
    content_hash: Option<&str>,
    jsonl_path: Option<&Path>,
) -> Result<()> {
    use chrono::Utc;
    storage.with_write_transaction(|storage| -> Result<()> {
        if !clear_dirty_metadata.is_empty() {
            storage.clear_dirty_issues(clear_dirty_metadata)?;
        }
        if !removed_hash_ids.is_empty() {
            storage.clear_export_hashes_in_tx(removed_hash_ids)?;
        }
        if !issue_hashes.is_empty() {
            storage.set_export_hashes_in_tx(issue_hashes)?;
        }
        if let Some(content_hash) = content_hash {
            storage.set_metadata_in_tx(METADATA_JSONL_CONTENT_HASH, content_hash)?;
            storage.set_metadata_in_tx(METADATA_LAST_EXPORT_TIME, &Utc::now().to_rfc3339())?;
            let jsonl_path = jsonl_path.ok_or_else(|| {
                BeadsError::Config(
                    "incremental auto-flush metadata update requires a JSONL path".to_string(),
                )
            })?;
            record_jsonl_witness_in_tx(storage, jsonl_path)?;
        }
        storage.set_metadata_in_tx("needs_flush", "false")?;
        Ok(())
    })?;

    Ok(())
}

fn write_jsonl_lines_atomically(
    lines_by_id: &BTreeMap<String, String>,
    output_path: &Path,
    config: &ExportConfig,
) -> Result<String> {
    if let Some(ref beads_dir) = config.beads_dir {
        validate_sync_path_with_external(output_path, beads_dir, config.allow_external_jsonl)?;

        let output_abs = if output_path.is_absolute() {
            output_path.to_path_buf()
        } else if let Ok(cwd) = std::env::current_dir() {
            cwd.join(output_path)
        } else {
            output_path.to_path_buf()
        };

        history::backup_before_export(beads_dir, &config.history, &output_abs)?;
    }

    let parent_dir = output_path.parent().ok_or_else(|| {
        BeadsError::Config(format!("Invalid output path: {}", output_path.display()))
    })?;
    fs::create_dir_all(parent_dir)?;

    let temp_path = export_temp_path(output_path);
    if let Some(ref beads_dir) = config.beads_dir {
        validate_temp_file_path(
            &temp_path,
            output_path,
            beads_dir,
            config.allow_external_jsonl,
        )?;
    }

    let temp_file = OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(&temp_path)
        .map_err(|err| {
            if err.kind() == std::io::ErrorKind::AlreadyExists {
                BeadsError::Config(format!(
                    "Temporary export file already exists: {}",
                    temp_path.display()
                ))
            } else {
                err.into()
            }
        })?;
    let mut temp_guard = TempFileGuard::new(temp_path.clone());
    let mut writer = BufWriter::new(temp_file);
    let mut hasher = Sha256::new();

    for line in lines_by_id.values() {
        writeln!(writer, "{line}")?;
        hasher.update(line.as_bytes());
        hasher.update(b"\n");
    }

    writer.flush()?;
    writer
        .into_inner()
        .map_err(|e| BeadsError::Io(e.into_error()))?
        .sync_all()?;

    let actual_count = count_issues_in_jsonl(&temp_path)?;
    if actual_count != lines_by_id.len() {
        return Err(BeadsError::Config(format!(
            "Export verification failed: expected {} issues, JSONL has {} lines",
            lines_by_id.len(),
            actual_count
        )));
    }

    if let Some(ref beads_dir) = config.beads_dir {
        require_safe_sync_overwrite_path(
            &temp_path,
            beads_dir,
            config.allow_external_jsonl,
            "rename temp file",
        )?;
        require_safe_sync_overwrite_path(
            output_path,
            beads_dir,
            config.allow_external_jsonl,
            "overwrite JSONL output",
        )?;
    }

    fs::rename(&temp_path, output_path)?;
    temp_guard.persist();

    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        let perms = std::fs::Permissions::from_mode(0o600);
        if let Err(e) = fs::set_permissions(output_path, perms) {
            tracing::warn!(
                path = %output_path.display(),
                error = %e,
                "Failed to set restrictive permissions on exported JSONL file"
            );
        }
    }

    Ok(format!("{:x}", hasher.finalize()))
}

fn try_incremental_auto_flush(
    storage: &mut SqliteStorage,
    beads_dir: &Path,
    jsonl_path: &Path,
    allow_external_jsonl: bool,
) -> Result<Option<AutoFlushResult>> {
    if !jsonl_path.exists() {
        return Ok(None);
    }

    let mut lines_by_id = read_jsonl_lines_by_id(jsonl_path)?;
    let dirty_metadata = storage.get_dirty_issue_metadata()?;
    if dirty_metadata.is_empty() {
        return Ok(Some(AutoFlushResult::default()));
    }

    let dirty_len = dirty_metadata.len();
    let mut removed_hash_ids = Vec::with_capacity(dirty_len);
    let mut issue_hashes = Vec::with_capacity(dirty_len);
    let mut changed = false;

    let dirty_ids: Vec<String> = dirty_metadata.iter().map(|(id, _)| id.clone()).collect();
    let batch_issues = storage.get_issues_for_export(&dirty_ids)?;
    let mut issues_by_id: std::collections::HashMap<String, crate::model::Issue> = batch_issues
        .into_iter()
        .map(|i| (i.id.clone(), i))
        .collect();

    for (issue_id, _) in &dirty_metadata {
        let maybe_issue = issues_by_id.remove(issue_id);
        match maybe_issue {
            Some(mut issue) if is_issue_exportable(&issue, None) => {
                normalize_issue_for_export(&mut issue);
                let json = serde_json::to_string(&issue).map_err(|err| {
                    BeadsError::Config(format!(
                        "Failed to serialize issue '{}' during auto-flush: {err}",
                        issue.id
                    ))
                })?;

                if lines_by_id.get(issue_id) != Some(&json) {
                    lines_by_id.insert(issue_id.clone(), json);
                    changed = true;
                }

                issue_hashes.push((
                    issue_id.clone(),
                    issue
                        .content_hash
                        .clone()
                        .unwrap_or_else(|| issue.compute_content_hash()),
                ));
            }
            Some(_) | None => {
                removed_hash_ids.push(issue_id.clone());
                changed |= lines_by_id.remove(issue_id).is_some();
            }
        }
    }

    if !changed {
        finalize_incremental_auto_flush(
            storage,
            &dirty_metadata,
            &removed_hash_ids,
            &issue_hashes,
            None,
            None,
        )?;
        return Ok(Some(AutoFlushResult::default()));
    }

    let export_config = ExportConfig {
        force: false,
        beads_dir: Some(beads_dir.to_path_buf()),
        allow_external_jsonl,
        ..Default::default()
    };
    let content_hash = write_jsonl_lines_atomically(&lines_by_id, jsonl_path, &export_config)?;
    finalize_incremental_auto_flush(
        storage,
        &dirty_metadata,
        &removed_hash_ids,
        &issue_hashes,
        Some(&content_hash),
        Some(jsonl_path),
    )?;

    Ok(Some(AutoFlushResult {
        flushed: true,
        exported_count: lines_by_id.len(),
        content_hash,
    }))
}

/// Result of an auto-flush operation.
#[derive(Debug, Default)]
pub struct AutoFlushResult {
    /// Whether the flush was performed (false if skipped due to no dirty issues).
    pub flushed: bool,
    /// Number of issues exported (0 if not flushed).
    pub exported_count: usize,
    /// Content hash of the exported JSONL (empty if not flushed).
    pub content_hash: String,
}

/// Perform an automatic flush of dirty issues to JSONL.
///
/// This is the auto-flush operation that runs at the end of mutating commands
/// (unless `--no-auto-flush` is set). It:
/// 1. Checks for dirty issues
/// 2. If any exist, exports them to the resolved JSONL path
/// 3. Clears dirty flags and updates metadata
///
/// Returns early (no-op) if there are no dirty issues.
///
/// # Arguments
///
/// * `storage` - Mutable reference to the `SQLite` storage
/// * `beads_dir` - Path to the .beads directory
/// * `jsonl_path` - Resolved JSONL export target for this workspace
///
/// # Errors
///
/// Returns an error if the export fails.
pub fn auto_flush(
    storage: &mut SqliteStorage,
    beads_dir: &Path,
    jsonl_path: &Path,
    allow_external_jsonl: bool,
) -> Result<AutoFlushResult> {
    // Check for dirty issues or forced flush first
    let jsonl_exists = jsonl_path.exists();
    let (dirty_count, needs_flush, db_newer) = pending_export_state(storage, jsonl_exists)?;

    if !db_newer {
        tracing::debug!("Auto-flush: no dirty issues, skipping");
        return Ok(AutoFlushResult::default());
    }

    // Refuse to auto-flush over a JSONL that still holds unresolved
    // merge-conflict markers. The downstream export path would otherwise
    // silently overwrite the `<<<<<<<` / `=======` / `>>>>>>>` regions
    // (along with the remote side of the merge the operator hadn't yet
    // looked at) every time a mutating CLI command returns. Explicit
    // `br sync --flush-only` already has a `--force` escape hatch for this
    // case; auto-flush has no such surface, so the only safe default is to
    // stop, log clearly, and let the next explicit sync surface the error.
    if jsonl_exists && let Err(err) = ensure_no_conflict_markers(jsonl_path) {
        tracing::warn!(
            jsonl_path = %jsonl_path.display(),
            error = %err,
            "Skipping auto-flush: JSONL contains merge-conflict markers. Resolve them (or run `br sync --flush-only --force` to override) before the next write.",
        );
        return Ok(AutoFlushResult::default());
    }

    tracing::debug!(
        dirty_count,
        needs_flush,
        "Auto-flush: exporting dirty issues"
    );

    if !needs_flush {
        match try_incremental_auto_flush(storage, beads_dir, jsonl_path, allow_external_jsonl) {
            Ok(Some(result)) => {
                tracing::info!(
                    flushed = result.flushed,
                    exported = result.exported_count,
                    "Auto-flush complete"
                );
                return Ok(result);
            }
            Ok(None) => {}
            Err(err) => {
                tracing::debug!(
                    ?err,
                    "Incremental auto-flush unavailable; falling back to full export"
                );
            }
        }
    }

    // Configure export with defaults, including beads_dir for path validation.
    // When needs_flush is set (e.g. after purge_issue), force must be true
    // even if there are also dirty issues from related mutations (like
    // dependency removal), so that the safety guard does not block export
    // of a DB that intentionally has fewer issues than the on-disk JSONL.
    let export_config = ExportConfig {
        force: needs_flush,
        beads_dir: Some(beads_dir.to_path_buf()),
        allow_external_jsonl,
        ..Default::default()
    };

    // Perform export
    let (export_result, _report) =
        export_to_jsonl_with_policy(storage, jsonl_path, &export_config)?;

    // Finalize export (clear dirty flags, update metadata)
    finalize_export(
        storage,
        &export_result,
        Some(&export_result.issue_hashes),
        jsonl_path,
    )?;

    tracing::info!(
        exported = export_result.exported_count,
        "Auto-flush complete"
    );

    Ok(AutoFlushResult {
        flushed: true,
        exported_count: export_result.exported_count,
        content_hash: export_result.content_hash,
    })
}

/// Read all issues from a JSONL file.
///
/// # Errors
///
/// Returns an error if the file cannot be read or contains invalid JSON.
pub fn read_issues_from_jsonl(path: &Path) -> Result<Vec<Issue>> {
    let file = File::open(path)?;
    let file_size = file.metadata().map_or(0, |m| m.len());
    let estimated_count = (file_size / 500) as usize;
    let mut reader = BufReader::new(file);
    let mut issues = Vec::with_capacity(estimated_count);
    let mut seen_ids = HashSet::with_capacity(estimated_count);
    let mut line = String::new();
    let mut line_num = 0;

    loop {
        line.clear();
        let bytes_read = reader.read_line(&mut line)?;
        if bytes_read == 0 {
            break;
        }

        let trimmed = line.trim();
        if trimmed.is_empty() {
            line_num += 1;
            continue;
        }

        let issue: Issue = serde_json::from_str(trimmed).map_err(|e| {
            BeadsError::Config(format!("Invalid JSON at line {}: {}", line_num + 1, e))
        })?;
        if !seen_ids.insert(issue.id.clone()) {
            return Err(BeadsError::Config(format!(
                "Duplicate issue id '{}' in {} at line {}",
                issue.id,
                path.display(),
                line_num + 1
            )));
        }
        issues.push(issue);
        line_num += 1;
    }

    Ok(issues)
}

// ===== 4-Phase Collision Detection =====

/// Match type from collision detection.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MatchType {
    /// Matched by external reference (e.g., JIRA-123).
    ExternalRef,
    /// Matched by content hash (deduplication).
    ContentHash,
    /// Matched by ID.
    Id,
}

/// Result of collision detection.
#[derive(Debug, Clone)]
pub enum CollisionResult {
    /// No match found - issue is new.
    NewIssue,
    /// Matched an existing issue.
    Match {
        /// The existing issue ID.
        existing_id: String,
        /// How the match was determined.
        match_type: MatchType,
        /// Which phase found the match (1-3).
        phase: u8,
    },
}

/// Action to take after collision detection.
#[derive(Debug, Clone)]
pub enum CollisionAction {
    /// Insert as a new issue.
    Insert,
    /// Update the existing issue.
    Update { existing_id: String },
    /// Skip this issue (existing is newer or it's a tombstone).
    Skip { reason: String },
}

/// Detect collision for an incoming issue using the 4-phase algorithm with preloaded metadata maps.
fn detect_collision(
    incoming: &Issue,
    id_by_ext_ref: &std::collections::HashMap<String, String>,
    id_by_hash: &std::collections::HashMap<String, String>,
    meta_by_id: &std::collections::HashMap<String, crate::storage::sqlite::IssueMetadata>,
    computed_hash: &str,
) -> CollisionResult {
    // Phase 1: External reference match
    if let Some(ref external_ref) = incoming.external_ref
        && let Some(existing_id) = id_by_ext_ref.get(external_ref)
    {
        return CollisionResult::Match {
            existing_id: existing_id.clone(),
            match_type: MatchType::ExternalRef,
            phase: 1,
        };
    }

    // Phase 2: ID match
    if meta_by_id.contains_key(&incoming.id) {
        return CollisionResult::Match {
            existing_id: incoming.id.clone(),
            match_type: MatchType::Id,
            phase: 2,
        };
    }

    // Phase 3: Content hash match
    if let Some(existing_id) = id_by_hash.get(computed_hash) {
        return CollisionResult::Match {
            existing_id: existing_id.clone(),
            match_type: MatchType::ContentHash,
            phase: 3,
        };
    }

    // Phase 4: No match
    CollisionResult::NewIssue
}

/// Determine the action to take based on collision result.
fn determine_action(
    collision: &CollisionResult,
    incoming: &Issue,
    meta_by_id: &std::collections::HashMap<String, crate::storage::sqlite::IssueMetadata>,
    force_upsert: bool,
) -> Result<CollisionAction> {
    match collision {
        CollisionResult::NewIssue => Ok(CollisionAction::Insert),
        CollisionResult::Match { existing_id, .. } => {
            let existing_meta =
                meta_by_id
                    .get(existing_id)
                    .ok_or_else(|| BeadsError::IssueNotFound {
                        id: existing_id.clone(),
                    })?;

            // Check for tombstone protection (even force doesn't override this)
            if existing_meta.status == crate::model::Status::Tombstone {
                return Ok(CollisionAction::Skip {
                    reason: format!("Tombstone protection: {existing_id}"),
                });
            }

            // If force_upsert is enabled, always update (skip timestamp comparison)
            if force_upsert {
                return Ok(CollisionAction::Update {
                    existing_id: existing_id.clone(),
                });
            }

            // Last-write-wins: compare updated_at
            match incoming.updated_at.cmp(&existing_meta.updated_at) {
                std::cmp::Ordering::Greater => Ok(CollisionAction::Update {
                    existing_id: existing_id.clone(),
                }),
                std::cmp::Ordering::Equal => Ok(CollisionAction::Skip {
                    reason: format!("Equal timestamps: {existing_id}"),
                }),
                std::cmp::Ordering::Less => Ok(CollisionAction::Skip {
                    reason: format!("Existing is newer: {existing_id}"),
                }),
            }
        }
    }
}

/// Normalize an issue for import.
///
/// - Recomputes `content_hash`
/// - Sets ephemeral=true if ID contains "-wisp-"
/// - Applies defaults and repairs `closed_at` invariant
fn normalize_issue(issue: &mut Issue) {
    use crate::util::content_hash;

    // Deduplicate labels
    if !issue.labels.is_empty() {
        issue.labels.sort();
        issue.labels.dedup();
    }

    // Normalize dependency types (fix legacy underscores)
    for dep in &mut issue.dependencies {
        if let crate::model::DependencyType::Custom(custom) = &dep.dep_type {
            let candidate = custom.replace('_', "-");
            if let Ok(normalized) = candidate.parse::<crate::model::DependencyType>()
                && !matches!(normalized, crate::model::DependencyType::Custom(_))
            {
                dep.dep_type = normalized;
            }
        }
    }

    // Deduplicate dependencies: for each (issue_id, depends_on_id, dep_type) triple,
    // keep only the most recent entry by created_at. This handles duplicate parent-child
    // entries from reparenting or migration artifacts (see issue #159).
    if issue.dependencies.len() > 1 {
        use std::collections::HashMap;
        // Build a map keyed by (issue_id, depends_on_id, dep_type), keeping the entry
        // with the latest created_at for each triple.
        let mut best: HashMap<(String, String, String), usize> = HashMap::new();
        for (i, dep) in issue.dependencies.iter().enumerate() {
            let key = (
                dep.issue_id.clone(),
                dep.depends_on_id.clone(),
                dep.dep_type.as_str().to_string(),
            );
            match best.get(&key) {
                Some(&prev_idx) if issue.dependencies[prev_idx].created_at >= dep.created_at => {
                    // existing entry is newer or equal, skip
                }
                _ => {
                    best.insert(key, i);
                }
            }
        }
        if best.len() < issue.dependencies.len() {
            let mut keep_indices: Vec<usize> = best.into_values().collect();
            keep_indices.sort_unstable();
            issue.dependencies = keep_indices
                .into_iter()
                .map(|i| issue.dependencies[i].clone())
                .collect();
        }
    }

    // Recompute content hash
    issue.content_hash = Some(content_hash(issue));

    // Wisp detection: if ID contains "-wisp-", mark as ephemeral
    if issue.id.contains("-wisp-") {
        issue.ephemeral = true;
    }

    // Repair closed_at invariant: if status is terminal (closed/tombstone), ensure closed_at is set
    if issue.status.is_terminal() && issue.closed_at.is_none() {
        issue.closed_at = Some(issue.updated_at);
    }

    // If status is not terminal, clear closed_at
    if !issue.status.is_terminal() {
        issue.closed_at = None;
    }

    // Normalize external_ref: empty string should be None to prevent UNIQUE constraint violations
    if let Some(ext_ref) = &issue.external_ref {
        if ext_ref.trim().is_empty() {
            issue.external_ref = None;
        } else {
            // Re-assign trimmed version just in case
            issue.external_ref = Some(ext_ref.trim().to_string());
        }
    }

    // Repair timestamps invariant: updated_at cannot be before created_at.
    // In distributed systems, clocks can be out of sync; we enforce the invariant
    // locally to keep the database consistent.
    if issue.updated_at < issue.created_at {
        issue.updated_at = issue.created_at;
    }
}

/// Import issues from a JSONL file.
///
/// Implements classic bd import semantics:
/// 0. Path validation - reject git paths and outside-beads paths without opt-in
/// 1. Conflict marker scan - abort if found
/// 2. Parse JSONL with 2MB buffer
/// 3. Normalize issues (recompute `content_hash`, set defaults)
/// 4. Prefix validation (optional)
/// 5. 4-phase collision detection
/// 6. Tombstone protection
/// 7. Orphan handling
/// 8. Create/update issues
/// 9. Sync deps/labels/comments
/// 10. Refresh blocked cache
/// 11. Update metadata
///
/// # Errors
///
/// Returns an error if:
/// - Path validation fails (git path, outside `beads_dir` without opt-in)
/// - Conflict markers are detected
/// - File cannot be read
/// - Prefix validation fails
/// - Database operations fail
#[allow(clippy::too_many_lines)]
pub fn import_from_jsonl(
    storage: &mut SqliteStorage,
    input_path: &Path,
    config: &ImportConfig,
    expected_prefix: Option<&str>,
) -> Result<ImportResult> {
    use crate::util::content_hash;

    // Step 0: Path validation (PC-1, PC-2, PC-3, NGI-3) - BEFORE any file operations
    if let Some(ref beads_dir) = config.beads_dir {
        validate_sync_path_with_external(input_path, beads_dir, config.allow_external_jsonl)?;
        tracing::debug!(
            input_path = %input_path.display(),
            beads_dir = %beads_dir.display(),
            allow_external = config.allow_external_jsonl,
            "Import path validated"
        );
    }

    // Step 1: Conflict marker scan
    ensure_no_conflict_markers(input_path)?;

    // Step 2: Parse, Normalize, and Validate JSONL
    let spinner = create_spinner("Parsing and validating issues", config.show_progress);
    let file = File::open(input_path)?;
    let file_size = file.metadata().map_or(0, |m| m.len());
    // Estimate ~500 bytes per issue to pre-allocate vector capacity
    let estimated_count = (file_size / 500) as usize;
    let mut reader = BufReader::with_capacity(2 * 1024 * 1024, file);
    let mut issues = Vec::with_capacity(estimated_count);
    let mut seen_ids = HashSet::with_capacity(estimated_count);
    let mut mismatches = Vec::new();

    let mut line = String::new();
    let mut line_num = 0;
    while reader.read_line(&mut line)? > 0 {
        line_num += 1;
        let trimmed = line.trim();
        if trimmed.is_empty() {
            line.clear();
            continue;
        }
        let mut issue: Issue = serde_json::from_str(trimmed)
            .map_err(|e| BeadsError::Config(format!("Invalid JSON at line {}: {}", line_num, e)))?;

        // 1. Normalize (Step 3)
        normalize_issue(&mut issue);

        // 2. Validate (Step 3.5)
        if let Err(errors) = IssueValidator::validate(&issue) {
            let details = errors
                .iter()
                .map(ToString::to_string)
                .collect::<Vec<_>>()
                .join(", ");
            return Err(BeadsError::Config(format!(
                "Validation failed for issue {} at line {}: {}",
                issue.id, line_num, details
            )));
        }

        // 3. Prefix Check (Step 4)
        if !config.skip_prefix_validation
            && let Some(prefix) = expected_prefix
            && !id_matches_expected_prefix(&issue.id, prefix)
            && issue.status != crate::model::Status::Tombstone
        {
            if !config.rename_on_import {
                return Err(BeadsError::Config(format!(
                    "Prefix mismatch at line {}: expected '{}', found issue '{}'",
                    line_num, prefix, issue.id
                )));
            }
            mismatches.push(issue.id.clone());
        }

        // 4. Reject duplicate IDs instead of silently collapsing records.
        if !seen_ids.insert(issue.id.clone()) {
            return Err(BeadsError::Config(format!(
                "Duplicate issue id '{}' in {} at line {}",
                issue.id,
                input_path.display(),
                line_num
            )));
        }
        issues.push(issue);

        line.clear();
    }
    spinner.finish_with_message("Parsed and validated issues");

    let mut result = ImportResult::default();

    // Step 5: Handle renames if requested
    if config.rename_on_import
        && !mismatches.is_empty()
        && let Some(prefix) = expected_prefix
    {
        use crate::util::id::{IdConfig, IdGenerator};

        let mismatch_set: std::collections::HashSet<String> = mismatches.iter().cloned().collect();

        // Collect details to avoid borrowing issues during generation
        let to_rename: Vec<_> = issues
            .iter()
            .filter(|i| mismatch_set.contains(&i.id))
            .map(|i| {
                (
                    i.id.clone(),
                    i.title.clone(),
                    i.description.clone(),
                    i.created_by.clone(),
                    i.created_at,
                )
            })
            .collect();

        let generator = IdGenerator::new(IdConfig::with_prefix(prefix));
        let mut renames = std::collections::HashMap::new();
        let existing_ids: std::collections::HashSet<String> =
            storage.get_all_ids()?.into_iter().collect();

        // Collect all IDs that will NOT be renamed to avoid collisions
        let mut occupied_ids: std::collections::HashSet<String> = issues
            .iter()
            .filter(|i| !mismatch_set.contains(&i.id))
            .map(|i| i.id.clone())
            .collect();

        // Add existing IDs from storage to occupied set
        occupied_ids.extend(existing_ids);

        let mut generated_ids = std::collections::HashSet::new();

        for (old_id, title, desc, creator, created_at) in to_rename {
            let new_id = generator.generate(
                &title,
                desc.as_deref(),
                creator.as_deref(),
                created_at,
                issues.len(),
                |candidate| occupied_ids.contains(candidate) || generated_ids.contains(candidate),
            );
            generated_ids.insert(new_id.clone());
            renames.insert(old_id, new_id);
        }

        // Apply renames
        for issue in &mut issues {
            if let Some(new_id) = renames.get(&issue.id) {
                // Preserve old ID in external_ref if empty
                if issue.external_ref.is_none() {
                    issue.external_ref = Some(issue.id.clone());
                }
                issue.id = new_id.clone();
                // Recompute content hash since ID/external_ref changed
                issue.content_hash = Some(content_hash(issue));
            }
            // Update dependencies
            for dep in &mut issue.dependencies {
                if let Some(new_target) = renames.get(&dep.depends_on_id) {
                    dep.depends_on_id = new_target.clone();
                }
                if let Some(new_source) = renames.get(&dep.issue_id) {
                    dep.issue_id = new_source.clone();
                }
            }
            // Update comments
            for comment in &mut issue.comments {
                if let Some(new_source) = renames.get(&comment.issue_id) {
                    comment.issue_id = new_source.clone();
                }
            }
        }
    }

    // Preload all metadata for O(1) collision detection (avoiding N+1 queries)
    let all_meta = storage.get_all_issues_metadata()?;
    let meta_len = all_meta.len();
    let mut meta_by_id = std::collections::HashMap::with_capacity(meta_len);
    let mut id_by_ext_ref = std::collections::HashMap::with_capacity(meta_len);
    let mut id_by_hash = std::collections::HashMap::with_capacity(meta_len);

    for m in all_meta {
        let issue_id = m.id.clone();
        if let Some(ext) = m.external_ref.as_ref() {
            id_by_ext_ref
                .entry(ext.clone())
                .or_insert_with(|| issue_id.clone());
        }
        if let Some(hash) = m.content_hash.as_ref() {
            // Preserve the first matching issue to mirror the old query_row
            // collision path when multiple issues share the same content hash.
            id_by_hash
                .entry(hash.clone())
                .or_insert_with(|| issue_id.clone());
        }
        meta_by_id.insert(issue_id, m);
    }

    // Phase 1: Scan and Resolve IDs
    let mut seen_external_refs: HashSet<String> = HashSet::new();
    let mut renames: std::collections::HashMap<String, String> = std::collections::HashMap::new();
    let issues_len = issues.len();
    let mut import_ops = Vec::with_capacity(issues_len);
    let mut new_export_hashes = Vec::with_capacity(issues_len);

    let progress =
        create_progress_bar(issues.len() as u64, "Scanning issues", config.show_progress);

    for mut issue in issues {
        // Skip ephemerals during import (they shouldn't be in JSONL anyway)
        if issue.ephemeral {
            result.skipped_count += 1;
            progress.inc(1);
            continue;
        }

        // Handle external ref duplicates before collision detection
        if let Some(ref ext_ref) = issue.external_ref {
            if seen_external_refs.contains(ext_ref) {
                if config.clear_duplicate_external_refs {
                    issue.external_ref = None;
                    issue.content_hash = Some(content_hash(&issue));
                } else {
                    progress.inc(1);
                    return Err(BeadsError::Config(format!(
                        "Duplicate external_ref: {ext_ref}"
                    )));
                }
            } else {
                seen_external_refs.insert(ext_ref.clone());
            }
        }

        // Compute content hash for collision detection
        let computed_hash = content_hash(&issue);

        // Detect collision
        let collision = detect_collision(
            &issue,
            &id_by_ext_ref,
            &id_by_hash,
            &meta_by_id,
            &computed_hash,
        );

        // Determine action
        let action = determine_action(&collision, &issue, &meta_by_id, config.force_upsert)?;

        // Determine target ID and record mapping
        let target_id = match &collision {
            CollisionResult::Match { existing_id, .. } => existing_id.clone(),
            CollisionResult::NewIssue => {
                // Keep distinct rows from the same JSONL snapshot distinct during the scan
                // phase. Only metadata-backed collisions (already in storage) should remap
                // IDs; otherwise `--no-db` imports can collapse valid duplicate records or
                // try to update rows that do not exist yet.
                issue.id.clone()
            }
        };

        if target_id != issue.id {
            renames.insert(issue.id.clone(), target_id.clone());
        }

        // Collect hash for export_hashes table
        new_export_hashes.push((target_id, computed_hash));

        import_ops.push((issue, action));
        progress.inc(1);
    }
    progress.finish_with_message("Scan complete");

    let jsonl_hash = compute_jsonl_hash(input_path)?;

    // Phase 2: Remap Dependencies
    if !renames.is_empty() {
        for (issue, _) in &mut import_ops {
            // Update issue ID if it was remapped (e.g. collision with existing issue)
            if let Some(new_id) = renames.get(&issue.id) {
                issue.id = new_id.clone();
            }

            // Remap dependencies to point to the resolved IDs
            for dep in &mut issue.dependencies {
                if let Some(new_target) = renames.get(&dep.depends_on_id) {
                    dep.depends_on_id = new_target.clone();
                }
                if let Some(new_source) = renames.get(&dep.issue_id) {
                    dep.issue_id = new_source.clone();
                }
            }

            // Remap comments to point to the resolved ID
            for comment in &mut issue.comments {
                if let Some(new_source) = renames.get(&comment.issue_id) {
                    comment.issue_id = new_source.clone();
                }
            }
        }
    }

    // Phase 3: Execute Actions
    //
    // Disable FK constraints during bulk import so that issues can reference
    // other issues (in dependencies/comments) that haven't been inserted yet.
    // FK integrity is restored and validated after all data is loaded.
    storage
        .execute_raw("PRAGMA foreign_keys = OFF")
        .map_err(|source| BeadsError::WithContext {
            context: "Failed to disable foreign key enforcement before import".to_string(),
            source: Box::new(source),
        })?;

    let progress = create_progress_bar(
        import_ops.len() as u64,
        "Importing issues",
        config.show_progress,
    );

    let apply_result = storage.with_write_transaction(|storage| -> Result<ImportResult> {
        let mut tx_result = result.clone();
        progress.set_position(0);
        // Keep export-hash state transactional so failed imports do not
        // erase incremental export bookkeeping.
        storage.clear_all_export_hashes_in_tx()?;

        for (issue, action) in &import_ops {
            process_import_action(storage, action, issue, &mut tx_result)?;
            progress.inc(1);
        }

        // Clean up any orphaned rows left by FK-deferred import
        // (e.g., dependencies referencing issues not in the JSONL)
        let orphan_tables = &[
            ("dependencies", "issue_id"),
            ("dependencies", "depends_on_id"),
            ("labels", "issue_id"),
            ("comments", "issue_id"),
            ("events", "issue_id"),
            ("dirty_issues", "issue_id"),
            ("blocked_issues_cache", "issue_id"),
            ("child_counters", "parent_id"),
        ];
        let mut orphans_cleaned = 0usize;
        for (table, col) in orphan_tables {
            let sql = if *table == "dependencies" && *col == "depends_on_id" {
                format!(
                    "DELETE FROM {table} WHERE {col} NOT IN (SELECT id FROM issues) AND {col} NOT LIKE 'external:%'"
                )
            } else {
                format!("DELETE FROM {table} WHERE {col} NOT IN (SELECT id FROM issues)")
            };
            orphans_cleaned += storage.execute_raw_count(&sql)?;
        }
        if orphans_cleaned > 0 {
            tracing::info!(
                count = orphans_cleaned,
                "Cleaned orphaned FK rows after import"
            );
            tx_result.orphan_cleaned_count = orphans_cleaned;
        }

        if !new_export_hashes.is_empty() {
            storage.set_export_hashes_in_tx(&new_export_hashes)?;
        }

        storage.rebuild_blocked_cache_in_tx()?;
        storage.rebuild_child_counters_in_tx()?;
        storage
            .set_metadata_in_tx(METADATA_LAST_IMPORT_TIME, &chrono::Utc::now().to_rfc3339())?;
        storage.set_metadata_in_tx(METADATA_JSONL_CONTENT_HASH, &jsonl_hash)?;
        record_jsonl_witness_in_tx(storage, input_path)?;

        Ok(tx_result)
    });

    let validate_foreign_keys = apply_result.is_ok();
    let fk_restore_result = restore_foreign_keys_after_import(storage, validate_foreign_keys);

    match (apply_result, fk_restore_result) {
        (Ok(import_result), Ok(())) => {
            progress.finish_with_message("Import complete");
            Ok(import_result)
        }
        (Ok(_), Err(fk_err)) => {
            progress.finish_and_clear();
            Err(fk_err)
        }
        (Err(import_err), Ok(())) => {
            progress.finish_and_clear();
            Err(import_err)
        }
        (Err(import_err), Err(fk_err)) => {
            tracing::error!(
                error = %fk_err,
                "Failed to restore foreign key enforcement after failed import"
            );
            progress.finish_and_clear();
            Err(import_err)
        }
    }
}

fn id_matches_expected_prefix(id: &str, expected_prefix: &str) -> bool {
    let normalized_prefix = expected_prefix.trim_end_matches('-');
    parse_id(id).is_ok_and(|parsed| parsed.prefix == normalized_prefix)
}

/// Process a single import action.
fn process_import_action(
    storage: &SqliteStorage,
    action: &CollisionAction,
    issue: &Issue,
    result: &mut ImportResult,
) -> Result<()> {
    match action {
        CollisionAction::Insert => {
            storage.upsert_issue_for_import(issue)?;
            sync_issue_relations(storage, issue)?;
            result.imported_count += 1;
            result.created_count += 1;
        }
        CollisionAction::Update { existing_id } => {
            // When updating by external_ref or content_hash, the incoming issue may have
            // a different ID than the existing one. We need to update using the existing ID.
            if existing_id == &issue.id {
                storage.upsert_issue_for_import(issue)?;
                sync_issue_relations(storage, issue)?;
            } else {
                let mut updated_issue = issue.clone();
                updated_issue.id.clone_from(existing_id);
                storage.upsert_issue_for_import(&updated_issue)?;
                sync_issue_relations(storage, &updated_issue)?;
            }
            result.imported_count += 1;
            result.updated_count += 1;
        }
        CollisionAction::Skip { reason } => {
            tracing::debug!(id = %issue.id, reason = %reason, "Skipping issue");
            if reason.starts_with("Tombstone") {
                result.tombstone_skipped += 1;
            } else {
                result.skipped_count += 1;
            }
        }
    }
    Ok(())
}

/// Sync labels, dependencies, and comments for an imported issue.
fn sync_issue_relations(storage: &SqliteStorage, issue: &Issue) -> Result<()> {
    // Sync labels
    storage.sync_labels_for_import(&issue.id, &issue.labels)?;

    // Sync dependencies
    storage.sync_dependencies_for_import(&issue.id, &issue.dependencies)?;

    // Sync comments
    storage.sync_comments_for_import(&issue.id, &issue.comments)?;

    Ok(())
}

/// Finalize an import by computing the content hash of the imported file.
///
/// # Errors
///
/// Returns an error if the file cannot be read.
pub fn compute_jsonl_hash(path: &Path) -> Result<String> {
    use std::io::BufRead;
    let file = std::fs::File::open(path)?;
    let mut reader = std::io::BufReader::new(file);
    let mut hasher = Sha256::new();
    let mut line_buf = Vec::with_capacity(4096);

    loop {
        line_buf.clear();
        let bytes_read = reader.read_until(b'\n', &mut line_buf)?;
        if bytes_read == 0 {
            break;
        }

        // Efficiently skip empty or whitespace-only lines without UTF-8 validation.
        // trim_ascii() is a fast byte-based trim.
        let trimmed = line_buf.trim_ascii();
        if !trimmed.is_empty() {
            hasher.update(trimmed);
            hasher.update(b"\n");
        }
    }

    Ok(format!("{:x}", hasher.finalize()))
}

// ============================================================================
// 3-Way Merge Types and Functions
// ============================================================================

/// Types of conflicts that can occur during 3-way merge.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConflictType {
    /// Issue was modified locally but deleted externally (or vice versa).
    DeleteVsModify,
    /// Issue was created in both local and external with different content.
    ConvergentCreation,
}

/// Result of merging a single issue across base, left (local), and right (external).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MergeResult {
    /// No action needed (e.g., issue doesn't exist in any source).
    NoAction,
    /// Keep the specified issue.
    Keep(Issue),
    /// Keep the specified issue with a note about the merge decision.
    KeepWithNote(Issue, String),
    /// Delete the issue.
    Delete,
    /// A conflict was detected that requires manual resolution.
    Conflict(ConflictType),
}

/// Context for performing a 3-way merge operation.
#[derive(Debug, Default)]
pub struct MergeContext {
    /// Base state (last known common state).
    pub base: std::collections::HashMap<String, Issue>,
    /// Left state (current SQLite/local changes).
    pub left: std::collections::HashMap<String, Issue>,
    /// Right state (current JSONL/external changes).
    pub right: std::collections::HashMap<String, Issue>,
}

impl MergeContext {
    /// Create a new merge context from the three states.
    #[must_use]
    #[allow(clippy::missing_const_for_fn)]
    pub fn new(
        base: std::collections::HashMap<String, Issue>,
        left: std::collections::HashMap<String, Issue>,
        right: std::collections::HashMap<String, Issue>,
    ) -> Self {
        Self { base, left, right }
    }

    /// Get all unique issue IDs across all three states.
    #[must_use]
    pub fn all_issue_ids(&self) -> std::collections::HashSet<String> {
        let mut ids = std::collections::HashSet::new();
        ids.extend(self.base.keys().cloned());
        ids.extend(self.left.keys().cloned());
        ids.extend(self.right.keys().cloned());
        ids
    }
}

/// Report of a 3-way merge operation.
#[derive(Debug, Default)]
pub struct MergeReport {
    /// Issues that were kept (created or updated).
    pub kept: Vec<Issue>,
    /// Issues that were deleted.
    pub deleted: Vec<String>,
    /// Conflicts that were detected.
    pub conflicts: Vec<(String, ConflictType)>,
    /// Issues that were skipped due to tombstone protection.
    pub tombstone_protected: Vec<String>,
    /// Notes about merge decisions.
    pub notes: Vec<(String, String)>,
}

impl MergeReport {
    /// Returns true if there were any conflicts.
    #[must_use]
    pub fn has_conflicts(&self) -> bool {
        !self.conflicts.is_empty()
    }

    /// Total number of actions taken.
    #[must_use]
    pub fn total_actions(&self) -> usize {
        self.kept.len() + self.deleted.len()
    }
}

/// Strategy for resolving conflicts during merge.
#[derive(
    Debug,
    Clone,
    Copy,
    PartialEq,
    Eq,
    Default,
    clap::ValueEnum,
    serde::Serialize,
    serde::Deserialize,
)]
pub enum ConflictResolution {
    /// Always keep the local (`SQLite`) version.
    #[default]
    PreferLocal,
    /// Always keep the external (`JSONL`) version.
    PreferExternal,
    /// Use `updated_at` timestamp to determine winner (or specified strategy)
    PreferNewer,
    /// Report conflict without auto-resolving.
    Manual,
}

/// Merge a single issue given its state in base, left (local), and right (external).
///
/// This implements the core 3-way merge logic for a single issue:
/// - New local issues are kept
/// - New external issues are imported
/// - Deletions are handled based on whether the other side modified
/// - Both-modified uses `updated_at` as tiebreaker (or specified strategy)
///
/// # Arguments
/// * `base` - The issue in the base (common ancestor) state, if it existed
/// * `left` - The issue in the local (`SQLite`) state, if it exists
/// * `right` - The issue in the external (JSONL) state, if it exists
/// * `strategy` - How to resolve conflicts when both sides modified
#[must_use]
#[allow(clippy::too_many_lines)]
pub fn merge_issue(
    base: Option<&Issue>,
    left: Option<&Issue>,
    right: Option<&Issue>,
    strategy: ConflictResolution,
) -> MergeResult {
    match (base, left, right) {
        // Case 1: Only in base (deleted in both local and external) -> no action
        (Some(_), None, None) => MergeResult::Delete,

        // Case 2: Only in left (new local) -> keep
        (None, Some(l), None) => MergeResult::Keep(l.clone()),

        // Case 3: Only in right (new external) -> keep
        (None, None, Some(r)) => MergeResult::Keep(r.clone()),

        // Case 4: In base and left only (deleted in right/external)
        (Some(b), Some(l), None) => {
            // Was it modified locally after base?
            if l.sync_equals(b) {
                // Local unchanged since base, external deleted -> delete
                MergeResult::Delete
            } else {
                // Local modified but external deleted - conflict
                match strategy {
                    ConflictResolution::PreferLocal => MergeResult::KeepWithNote(
                        l.clone(),
                        "Local modified, external deleted - kept local".to_string(),
                    ),
                    ConflictResolution::PreferExternal => MergeResult::Delete,
                    ConflictResolution::PreferNewer => {
                        // Keep local since it was modified more recently than base
                        MergeResult::KeepWithNote(
                            l.clone(),
                            "Local modified after base, external deleted - kept local".to_string(),
                        )
                    }
                    ConflictResolution::Manual => {
                        MergeResult::Conflict(ConflictType::DeleteVsModify)
                    }
                }
            }
        }

        // Case 5: In base and right only (deleted locally)
        (Some(b), None, Some(r)) => {
            // Was it modified externally after base?
            if r.sync_equals(b) {
                // External unchanged since base, local deleted -> delete
                MergeResult::Delete
            } else {
                // External modified but local deleted - conflict
                match strategy {
                    ConflictResolution::PreferLocal => MergeResult::Delete,
                    ConflictResolution::PreferExternal => MergeResult::KeepWithNote(
                        r.clone(),
                        "External modified, local deleted - kept external".to_string(),
                    ),
                    ConflictResolution::PreferNewer => {
                        // Keep external since it was modified more recently than base
                        MergeResult::KeepWithNote(
                            r.clone(),
                            "External modified after base, local deleted - kept external"
                                .to_string(),
                        )
                    }
                    ConflictResolution::Manual => {
                        MergeResult::Conflict(ConflictType::DeleteVsModify)
                    }
                }
            }
        }

        // Case 6: In all three (potentially modified in one or both)
        (Some(b), Some(l), Some(r)) => {
            if l.sync_equals(r) {
                return MergeResult::Keep(l.clone());
            }

            let left_changed = !l.sync_equals(b);
            let right_changed = !r.sync_equals(b);

            match (left_changed, right_changed) {
                // Neither changed OR only left changed - keep left
                (false | true, false) => MergeResult::Keep(l.clone()),
                // Only right changed - keep right
                (false, true) => MergeResult::Keep(r.clone()),
                // Both changed - use strategy
                (true, true) => match strategy {
                    ConflictResolution::PreferLocal => MergeResult::KeepWithNote(
                        l.clone(),
                        "Both modified - kept local".to_string(),
                    ),
                    ConflictResolution::PreferExternal => MergeResult::KeepWithNote(
                        r.clone(),
                        "Both modified - kept external".to_string(),
                    ),
                    ConflictResolution::PreferNewer => {
                        if l.updated_at >= r.updated_at {
                            MergeResult::KeepWithNote(
                                l.clone(),
                                "Both modified - kept local (newer)".to_string(),
                            )
                        } else {
                            MergeResult::KeepWithNote(
                                r.clone(),
                                "Both modified - kept external (newer)".to_string(),
                            )
                        }
                    }
                    ConflictResolution::Manual => {
                        // For manual, we still need to pick one, so use newer but mark as note
                        if l.updated_at >= r.updated_at {
                            MergeResult::KeepWithNote(
                                l.clone(),
                                "Both modified - kept local (newer), review recommended"
                                    .to_string(),
                            )
                        } else {
                            MergeResult::KeepWithNote(
                                r.clone(),
                                "Both modified - kept external (newer), review recommended"
                                    .to_string(),
                            )
                        }
                    }
                },
            }
        }

        // Case 7: In left and right but not base (convergent creation)
        (None, Some(l), Some(r)) => {
            // Same content? Keep one (use left by convention)
            if l.sync_equals(r) {
                MergeResult::Keep(l.clone())
            } else {
                // Different content - both created independently
                match strategy {
                    ConflictResolution::PreferLocal => MergeResult::KeepWithNote(
                        l.clone(),
                        "Convergent creation - kept local".to_string(),
                    ),
                    ConflictResolution::PreferExternal => MergeResult::KeepWithNote(
                        r.clone(),
                        "Convergent creation - kept external".to_string(),
                    ),
                    ConflictResolution::PreferNewer | ConflictResolution::Manual => {
                        if l.updated_at >= r.updated_at {
                            MergeResult::KeepWithNote(
                                l.clone(),
                                "Convergent creation - kept local (newer)".to_string(),
                            )
                        } else {
                            MergeResult::KeepWithNote(
                                r.clone(),
                                "Convergent creation - kept external (newer)".to_string(),
                            )
                        }
                    }
                }
            }
        }

        // Case 8: Not in any (impossible in practice, but handle gracefully)
        (None, None, None) => MergeResult::NoAction,
    }
}

/// Perform a 3-way merge across all issues in the context.
///
/// This iterates through all unique issue IDs across base, left, and right,
/// and calls `merge_issue` for each to determine the appropriate action.
///
/// # Arguments
/// * `context` - The merge context containing base, left, and right states
/// * `strategy` - How to resolve conflicts when both sides modified
/// * `tombstones` - Optional set of issue IDs that should never be resurrected
///
/// # Returns
/// A `MergeReport` containing all actions taken and any conflicts detected.
#[must_use]
pub fn three_way_merge(
    context: &MergeContext,
    strategy: ConflictResolution,
    tombstones: Option<&HashSet<String, RandomState>>,
) -> MergeReport {
    let mut report = MergeReport::default();
    let empty_tombstones: HashSet<String, RandomState> = HashSet::new();
    let tombstones = tombstones.unwrap_or(&empty_tombstones);

    for id in context.all_issue_ids() {
        let base = context.base.get(&id);
        let left = context.left.get(&id);
        let right = context.right.get(&id);

        // Check tombstone protection: if issue is tombstoned and trying to resurrect
        if tombstones.contains(&id) {
            // Issue is tombstoned - only allow if it exists in local (left)
            if left.is_none() && right.is_some() {
                // Trying to resurrect from external - skip
                report.tombstone_protected.push(id.clone());
                continue;
            }
        }

        let result = merge_issue(base, left, right, strategy);

        match result {
            MergeResult::NoAction => {}
            MergeResult::Keep(issue) => {
                report.kept.push(issue);
            }
            MergeResult::KeepWithNote(issue, note) => {
                report.notes.push((issue.id.clone(), note));
                report.kept.push(issue);
            }
            MergeResult::Delete => {
                report.deleted.push(id.clone());
            }
            MergeResult::Conflict(conflict_type) => {
                report.conflicts.push((id.clone(), conflict_type));
            }
        }
    }

    report
}

/// Configuration for a 3-way merge operation.
#[derive(Debug, Clone, Default)]
pub struct MergeConfig {
    /// Strategy for resolving conflicts.
    pub strategy: ConflictResolution,
    /// Whether to skip tombstoned issues.
    pub respect_tombstones: bool,
}

/// Save the base snapshot to a file.
///
/// This is used after a successful merge to record the common state.
///
/// # Errors
///
/// Returns an error if the file cannot be written.
pub fn save_base_snapshot<S: ::std::hash::BuildHasher>(
    issues: &std::collections::HashMap<String, Issue, S>,
    jsonl_dir: &Path,
) -> Result<()> {
    let snapshot_path = jsonl_dir.join("beads.base.jsonl");
    let pid = std::process::id();
    let temp_path = snapshot_path.with_extension(format!("jsonl.{pid}.tmp"));
    validate_temp_file_path(&temp_path, &snapshot_path, jsonl_dir, false)?;

    let temp_file = OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(&temp_path)
        .map_err(|err| {
            if err.kind() == std::io::ErrorKind::AlreadyExists {
                BeadsError::Config(format!(
                    "Temporary base snapshot file already exists: {}",
                    temp_path.display()
                ))
            } else {
                err.into()
            }
        })?;
    let mut temp_guard = TempFileGuard::new(temp_path.clone());
    let mut writer = BufWriter::new(temp_file);

    let mut ordered_issues: Vec<_> = issues.values().collect();
    ordered_issues.sort_by(|left, right| left.id.cmp(&right.id));

    let mut buffer = Vec::new();
    for issue in ordered_issues {
        buffer.clear();
        serde_json::to_writer(&mut buffer, issue).map_err(|e| {
            BeadsError::Config(format!("Failed to serialize issue {}: {}", issue.id, e))
        })?;
        writer.write_all(&buffer).map_err(BeadsError::Io)?;
        writer.write_all(b"\n").map_err(BeadsError::Io)?;
    }
    writer.flush()?;
    writer
        .into_inner()
        .map_err(|e| BeadsError::Io(e.into_error()))?
        .sync_all()?;
    require_safe_sync_overwrite_path(&temp_path, jsonl_dir, false, "rename base snapshot")?;
    require_safe_sync_overwrite_path(&snapshot_path, jsonl_dir, false, "overwrite base snapshot")?;
    fs::rename(&temp_path, &snapshot_path)?;
    temp_guard.persist();
    Ok(())
}

/// Load the base snapshot from a file.
///
/// Returns an empty map if the snapshot does not exist.
///
/// # Errors
///
/// Returns an error if the file exists but cannot be read or parsed.
pub fn load_base_snapshot(jsonl_dir: &Path) -> Result<std::collections::HashMap<String, Issue>> {
    let snapshot_path = jsonl_dir.join("beads.base.jsonl");
    let mut base = std::collections::HashMap::new();

    if !snapshot_path.exists() {
        return Ok(base);
    }

    require_valid_sync_path(&snapshot_path, jsonl_dir)?;

    // `beads.base.jsonl` is normally .gitignore'd, but if a user's workspace
    // has it committed (or committed by accident once), a botched `git merge`
    // / `git pull` can leave `<<<<<<<` / `=======` / `>>>>>>>` markers in
    // this file just like in `issues.jsonl`. The serde parse below would
    // then fail with a generic "Invalid JSON in base snapshot at line N"
    // that buries the actual problem. Run the conflict-marker scan first so
    // the operator gets the same actionable "merge conflict markers
    // detected" diagnostic that the JSONL path surfaces.
    ensure_no_conflict_markers(&snapshot_path)?;

    let file = File::open(&snapshot_path)?;
    let reader = BufReader::new(file);

    for (line_num, line) in reader.lines().enumerate() {
        let line = line?;
        if line.trim().is_empty() {
            continue;
        }
        let issue: Issue = serde_json::from_str(&line).map_err(|e| {
            BeadsError::Config(format!(
                "Invalid JSON in base snapshot at line {}: {}",
                line_num + 1,
                e
            ))
        })?;
        base.insert(issue.id.clone(), issue);
    }

    Ok(base)
}

/// A tombstone row plus its related labels, dependencies, and comments,
/// snapshotted before a rebuild so it can be atomically restored afterwards.
///
/// The option wrappers on the relations let callers partially preserve a
/// tombstone whose relation fetches failed (a pattern the CLI layer already
/// uses): we keep the issue row and skip whatever relation set couldn't be
/// read, rather than losing the tombstone entirely.
#[derive(Clone, Debug)]
pub(crate) struct PreservedTombstone {
    pub(crate) issue: Issue,
    pub(crate) labels: Option<Vec<String>>,
    pub(crate) dependencies: Option<Vec<Dependency>>,
    pub(crate) comments: Option<Vec<Comment>>,
}

/// Snapshot every tombstoned issue in the database, including its labels,
/// dependencies, and comments, so a rebuild can restore deletion-retention
/// state that is not present in the JSONL export.
///
/// This is fully best-effort — the function never returns an error: if
/// the enumeration query fails outright we log and return an empty list
/// (the rebuild still proceeds without tombstone preservation), and
/// per-tombstone relation fetches also degrade gracefully to issue-row-
/// only preservation.
#[must_use]
pub(crate) fn snapshot_tombstones(storage: &SqliteStorage) -> Vec<PreservedTombstone> {
    let mut tombstones = Vec::new();
    let tombstone_ids = match storage.get_issue_ids_by_status(&crate::model::Status::Tombstone) {
        Ok(ids) => ids,
        Err(error) => {
            tracing::warn!(
                error = %error,
                "Failed to enumerate tombstones before rebuild; continuing without tombstone preservation"
            );
            return tombstones;
        }
    };

    for tombstone_id in tombstone_ids {
        let Some(issue) = (match storage.get_issue(&tombstone_id) {
            Ok(issue) => issue,
            Err(error) => {
                tracing::warn!(
                    issue_id = %tombstone_id,
                    error = %error,
                    "Skipping tombstone preservation for issue that could not be read before rebuild"
                );
                continue;
            }
        }) else {
            continue;
        };

        let labels = match storage.get_labels(&tombstone_id) {
            Ok(labels) => Some(labels),
            Err(error) => {
                tracing::warn!(
                    issue_id = %tombstone_id,
                    error = %error,
                    "Failed to snapshot tombstone labels before rebuild; preserving issue row only"
                );
                None
            }
        };
        let dependencies = match storage.get_dependencies_full(&tombstone_id) {
            Ok(dependencies) => Some(dependencies),
            Err(error) => {
                tracing::warn!(
                    issue_id = %tombstone_id,
                    error = %error,
                    "Failed to snapshot tombstone dependencies before rebuild; preserving issue row only"
                );
                None
            }
        };
        let comments = match storage.get_comments(&tombstone_id) {
            Ok(comments) => Some(comments),
            Err(error) => {
                tracing::warn!(
                    issue_id = %tombstone_id,
                    error = %error,
                    "Failed to snapshot tombstone comments before rebuild; preserving issue row only"
                );
                None
            }
        };
        tombstones.push(PreservedTombstone {
            issue,
            labels,
            dependencies,
            comments,
        });
    }
    tombstones
}

/// Restore preserved tombstones after a successful rebuild, wrapping any
/// failure with a message that makes clear the rebuild itself succeeded —
/// only the retention-state restoration step failed.
///
/// The rebuild has already moved the original database family into the
/// recovery directory and replaced it with a clean JSONL import at this
/// point, so on failure the live DB is *valid* (it mirrors the JSONL),
/// just missing whatever local unflushed tombstones we tried to preserve.
/// Without this wrapper, a transient lock-contention retry exhaustion
/// inside `restore_tombstones` would bubble up through callers that
/// otherwise describe the failure as "JSONL may be corrupt" or "database
/// recovery failed", both of which are actively misleading for this
/// specific post-rebuild failure mode. The wrapped message tells the
/// operator: re-running the command is idempotent and safe; the only
/// thing they've lost is local deletions that hadn't yet been flushed.
///
/// Callers should prefer this helper over calling `restore_tombstones`
/// directly when the restore follows an already-completed rebuild. Use
/// the bare `restore_tombstones` when the surrounding transaction is
/// still mid-rebuild and a rollback is still possible.
///
/// # Errors
///
/// Returns a `BeadsError::WithContext` whose source is the original
/// `restore_tombstones` error. Returns `Ok(())` when `tombstones` is
/// empty without calling into the write-transaction retry loop.
pub(crate) fn restore_tombstones_after_rebuild(
    storage: &mut SqliteStorage,
    tombstones: &[PreservedTombstone],
) -> Result<()> {
    if tombstones.is_empty() {
        return Ok(());
    }
    let count = tombstones.len();
    restore_tombstones(storage, tombstones).map_err(|err| BeadsError::WithContext {
        context: format!(
            "Rebuild from JSONL succeeded, but failed to restore {count} preserved \
             tombstone(s). The database now mirrors the JSONL exactly — any local \
             deletions that had not yet been flushed to the JSONL are gone. \
             Re-running the command is idempotent and safe (the rebuild itself \
             completed successfully). If the underlying cause is lock contention, \
             wait for other `br` processes to finish and try again."
        ),
        source: Box::new(err),
    })
}

/// Restore preserved tombstones (and their relations) atomically and mark
/// them dirty so the next flush re-exports them.
///
/// # Errors
///
/// Returns an error if the underlying write transaction fails; the entire
/// restore is rolled back on failure.
pub(crate) fn restore_tombstones(
    storage: &mut SqliteStorage,
    tombstones: &[PreservedTombstone],
) -> Result<()> {
    if tombstones.is_empty() {
        return Ok(());
    }

    let marked_at = Utc::now().to_rfc3339();
    storage.with_write_transaction(|storage| {
        for tombstone in tombstones {
            storage.upsert_issue_for_import(&tombstone.issue)?;
        }
        for tombstone in tombstones {
            if let Some(labels) = &tombstone.labels {
                storage.sync_labels_for_import(&tombstone.issue.id, labels)?;
            }
            if let Some(dependencies) = &tombstone.dependencies {
                storage.sync_dependencies_for_import(&tombstone.issue.id, dependencies)?;
            }
            if let Some(comments) = &tombstone.comments {
                storage.sync_comments_for_import(&tombstone.issue.id, comments)?;
            }
            storage.replace_dirty_issue_marker(&tombstone.issue.id, &marked_at)?;
        }
        Ok(())
    })?;

    tracing::debug!(
        count = tombstones.len(),
        "Restored tombstones atomically after rebuild and marked them dirty for export"
    );
    Ok(())
}

/// Per-ID view of the JSONL used to decide which preserved tombstones
/// should actually be restored after a rebuild. The rebuild imports
/// everything in the JSONL first, so tombstone preservation only needs to
/// fix up rows where the local DB and the JSONL disagree.
///
/// Two buckets:
///
/// - `tombstone_ids`: IDs whose JSONL record carries `status = tombstone`.
///   The deletion has already been flushed, so the rebuild will reimport it
///   as a tombstone on its own — we drop any local preserved tombstone for
///   these IDs.
///
/// - `non_tombstone_updated_at`: IDs whose JSONL record carries a *non*-
///   tombstone status, mapped to the record's `updated_at`. When the local
///   DB has one of these IDs as a tombstone, there is a disagreement: the
///   JSONL says the issue is alive, the DB says it's deleted. We resolve
///   it by timestamp — whichever side's last write is newer wins.
#[derive(Debug, Clone, Default)]
pub(crate) struct JsonlTombstoneFilter {
    pub(crate) tombstone_ids: HashSet<String>,
    pub(crate) non_tombstone_updated_at:
        std::collections::HashMap<String, chrono::DateTime<chrono::Utc>>,
}

/// Filter the preserved tombstone set down to those that should actually
/// be restored after the rebuild has reimported the JSONL. Three cases:
///
/// 1. JSONL has this ID as a tombstone: drop from preservation set — the
///    rebuild's own `import_from_jsonl` will reinstate the tombstone.
///
/// 2. JSONL has this ID as a non-tombstone: the user's local delete
///    competes with the JSONL's live record. Keep the local tombstone
///    only if `deleted_at > updated_at` (i.e. the local deletion actually
///    happened after the state the JSONL captured). Otherwise the JSONL
///    is newer and we let the rebuild's import win, same way a 3-way
///    merge with `--merge` would resolve it. Missing `deleted_at` (which
///    shouldn't happen for a well-formed tombstone) is treated
///    conservatively as "local is newer": preserve the tombstone rather
///    than silently lose a local deletion.
///
/// 3. JSONL doesn't have this ID at all: the deletion has never been
///    flushed anywhere. Always preserve — otherwise this path would
///    silently lose the local delete.
#[must_use]
pub(crate) fn tombstones_missing_from_jsonl_tombstones(
    tombstones: Vec<PreservedTombstone>,
    jsonl_filter: &JsonlTombstoneFilter,
) -> Vec<PreservedTombstone> {
    let original_count = tombstones.len();
    let mut skipped_already_flushed = 0usize;
    let mut skipped_jsonl_newer = 0usize;
    let preserved: Vec<PreservedTombstone> = tombstones
        .into_iter()
        .filter(|tombstone| {
            let id = &tombstone.issue.id;
            if jsonl_filter.tombstone_ids.contains(id) {
                skipped_already_flushed += 1;
                return false;
            }
            if let Some(jsonl_updated_at) = jsonl_filter.non_tombstone_updated_at.get(id) {
                let local_deleted_at = tombstone.issue.deleted_at;
                let local_is_newer =
                    local_deleted_at.is_none_or(|deleted_at| deleted_at > *jsonl_updated_at);
                if !local_is_newer {
                    skipped_jsonl_newer += 1;
                    return false;
                }
            }
            true
        })
        .collect();

    if skipped_already_flushed > 0 || skipped_jsonl_newer > 0 {
        tracing::debug!(
            preserved = preserved.len(),
            skipped_already_flushed,
            skipped_jsonl_newer,
            original = original_count,
            "Filtered preserved tombstones against JSONL state"
        );
    }

    preserved
}

/// Scan the JSONL once and build a `JsonlTombstoneFilter` we can use to
/// decide which preserved tombstones to restore after a rebuild.
///
/// # Errors
///
/// Returns an error if the JSONL cannot be read, contains invalid JSON, or
/// has duplicate IDs across lines.
pub(crate) fn scan_jsonl_for_tombstone_filter(path: &Path) -> Result<JsonlTombstoneFilter> {
    let file = File::open(path)?;
    let mut reader = BufReader::new(file);
    let mut line_buf = String::new();
    let mut line_num = 0;
    let mut seen_ids = HashSet::new();
    let mut filter = JsonlTombstoneFilter::default();

    loop {
        line_buf.clear();
        let bytes = reader.read_line(&mut line_buf)?;
        if bytes == 0 {
            break;
        }

        line_num += 1;
        let trimmed = line_buf.trim_end_matches(['\n', '\r']);
        if trimmed.trim().is_empty() {
            continue;
        }

        let issue: Issue = serde_json::from_str(trimmed)
            .map_err(|e| BeadsError::Config(format!("Invalid JSON at line {}: {}", line_num, e)))?;

        if !seen_ids.insert(issue.id.clone()) {
            return Err(BeadsError::Config(format!(
                "Duplicate issue id '{}' in {} at line {}",
                issue.id,
                path.display(),
                line_num
            )));
        }

        if issue.status == crate::model::Status::Tombstone {
            filter.tombstone_ids.insert(issue.id);
        } else {
            filter
                .non_tombstone_updated_at
                .insert(issue.id, issue.updated_at);
        }
    }

    Ok(filter)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::model::{Comment, Issue, IssueType, Priority, Status};
    use chrono::Utc;
    use fsqlite_types::SqliteValue;
    use indicatif::{ProgressBar, ProgressStyle};
    use std::collections::HashMap;
    use std::io::{self, Write};
    #[cfg(unix)]
    use std::os::unix::fs::symlink;
    use tempfile::TempDir;

    fn make_test_issue(id: &str, title: &str) -> Issue {
        Issue {
            id: id.to_string(),
            content_hash: None,
            title: title.to_string(),
            description: None,
            design: None,
            acceptance_criteria: None,
            notes: None,
            status: Status::Open,
            priority: Priority::MEDIUM,
            issue_type: IssueType::Task,
            assignee: None,
            owner: None,
            estimated_minutes: None,
            created_at: Utc::now(),
            created_by: None,
            updated_at: Utc::now(),
            closed_at: None,
            close_reason: None,
            closed_by_session: None,
            due_at: None,
            defer_until: None,
            external_ref: None,
            source_system: None,
            source_repo: None,
            deleted_at: None,
            deleted_by: None,
            delete_reason: None,
            original_type: None,
            compaction_level: None,
            compacted_at: None,
            compacted_at_commit: None,
            original_size: None,
            sender: None,
            ephemeral: false,
            pinned: false,
            is_template: false,
            labels: vec![],
            dependencies: vec![],
            comments: vec![],
        }
    }

    #[test]
    fn export_temp_path_is_pid_scoped_and_sibling_to_target() {
        let target = Path::new("/tmp/issues.jsonl");
        let temp = export_temp_path(target);

        assert_eq!(temp.parent(), target.parent());
        assert_ne!(temp, target);
        assert!(
            temp.display()
                .to_string()
                .contains(&std::process::id().to_string())
        );
        assert!(temp.extension().is_some_and(|ext| ext == "tmp"));
    }

    fn make_issue_at(id: &str, title: &str, updated_at: chrono::DateTime<Utc>) -> Issue {
        let created_at = updated_at - chrono::Duration::seconds(60);
        Issue {
            id: id.to_string(),
            content_hash: None,
            title: title.to_string(),
            description: None,
            design: None,
            acceptance_criteria: None,
            notes: None,
            status: Status::Open,
            priority: Priority::MEDIUM,
            issue_type: IssueType::Task,
            assignee: None,
            owner: None,
            estimated_minutes: None,
            created_at,
            created_by: None,
            updated_at,
            closed_at: None,
            close_reason: None,
            closed_by_session: None,
            due_at: None,
            defer_until: None,
            external_ref: None,
            source_system: None,
            source_repo: None,
            deleted_at: None,
            deleted_by: None,
            delete_reason: None,
            original_type: None,
            compaction_level: None,
            compacted_at: None,
            compacted_at_commit: None,
            original_size: None,
            sender: None,
            ephemeral: false,
            pinned: false,
            is_template: false,
            labels: vec![],
            dependencies: vec![],
            comments: vec![],
        }
    }

    fn set_content_hash(issue: &mut Issue) {
        issue.content_hash = Some(crate::util::content_hash(issue));
    }

    fn fixed_time(secs: i64) -> chrono::DateTime<Utc> {
        chrono::DateTime::from_timestamp(secs, 0).expect("timestamp")
    }

    fn build_collision_maps(
        storage: &SqliteStorage,
    ) -> (
        HashMap<String, String>,
        HashMap<String, String>,
        HashMap<String, crate::storage::sqlite::IssueMetadata>,
    ) {
        let all_meta = storage.get_all_issues_metadata().unwrap();
        let mut meta_by_id = HashMap::new();
        let mut id_by_ext_ref = HashMap::new();
        let mut id_by_hash = HashMap::new();

        for meta in all_meta {
            let issue_id = meta.id.clone();
            if let Some(ext) = meta.external_ref.as_ref() {
                id_by_ext_ref
                    .entry(ext.clone())
                    .or_insert_with(|| issue_id.clone());
            }
            if let Some(hash) = meta.content_hash.as_ref() {
                id_by_hash
                    .entry(hash.clone())
                    .or_insert_with(|| issue_id.clone());
            }
            meta_by_id.insert(issue_id, meta);
        }

        (id_by_ext_ref, id_by_hash, meta_by_id)
    }

    struct LineFailWriter {
        buffer: Vec<u8>,
        current: Vec<u8>,
        fail_on: String,
        failed: bool,
    }

    impl LineFailWriter {
        fn new(fail_on: &str) -> Self {
            Self {
                buffer: Vec::new(),
                current: Vec::new(),
                fail_on: fail_on.to_string(),
                failed: false,
            }
        }

        fn into_string(self) -> String {
            String::from_utf8(self.buffer).unwrap_or_default()
        }
    }

    impl Write for LineFailWriter {
        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
            self.current.extend_from_slice(buf);
            while let Some(pos) = self.current.iter().position(|b| *b == b'\n') {
                let line: Vec<u8> = self.current.drain(..=pos).collect();
                let line_str = String::from_utf8_lossy(&line);
                if !self.failed && line_str.contains(&self.fail_on) {
                    self.failed = true;
                    return Err(io::Error::other("intentional failure"));
                }
                self.buffer.extend_from_slice(&line);
            }
            Ok(buf.len())
        }

        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
    }

    #[test]
    fn test_scan_conflict_markers_detects_all_kinds() {
        let temp = TempDir::new().expect("tempdir");
        let path = temp.path().join("issues.jsonl");
        let contents = "\
{\"id\":\"bd-1\",\"title\":\"ok\"}
<<<<<<< HEAD
{\"id\":\"bd-2\",\"title\":\"conflict\"}
=======
{\"id\":\"bd-2\",\"title\":\"other\"}
>>>>>>> feature-branch
";
        fs::write(&path, contents).expect("write");

        let markers = scan_conflict_markers(&path).expect("scan");
        assert_eq!(markers.len(), 3);
        assert_eq!(markers[0].marker_type, ConflictMarkerType::Start);
        assert_eq!(markers[1].marker_type, ConflictMarkerType::Separator);
        assert_eq!(markers[2].marker_type, ConflictMarkerType::End);
        assert_eq!(markers[0].branch.as_deref(), Some("HEAD"));
        assert_eq!(markers[2].branch.as_deref(), Some("feature-branch"));
    }

    #[test]
    fn test_ensure_no_conflict_markers_errors() {
        let temp = TempDir::new().unwrap();
        let path = temp.path().join("issues.jsonl");
        fs::write(&path, "<<<<<<< HEAD\n").expect("write");

        let err = ensure_no_conflict_markers(&path).unwrap_err();
        let message = err.to_string();
        assert!(message.contains("Merge conflict markers detected"));
    }

    #[test]
    fn test_export_empty_database() {
        let storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let output_path = temp_dir.path().join("issues.jsonl");

        let config = ExportConfig::default();
        let result = export_to_jsonl(&storage, &output_path, &config).unwrap();

        assert_eq!(result.exported_count, 0);
        assert!(result.exported_ids.is_empty());
        assert!(output_path.exists());
    }

    #[cfg(unix)]
    #[test]
    fn test_save_base_snapshot_rejects_existing_temp_symlink() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        let outside_dir = temp.path().join("outside");
        fs::create_dir_all(&beads_dir).unwrap();
        fs::create_dir_all(&outside_dir).unwrap();

        let snapshot_path = beads_dir.join("beads.base.jsonl");
        fs::write(&snapshot_path, "old-snapshot\n").unwrap();

        let temp_target = outside_dir.join("captured.txt");
        fs::write(&temp_target, "do-not-touch").unwrap();
        let pid = std::process::id();
        symlink(
            &temp_target,
            beads_dir.join(format!("beads.base.jsonl.{pid}.tmp")),
        )
        .unwrap();

        let mut issues = HashMap::new();
        issues.insert(
            "bd-base".to_string(),
            Issue {
                id: "bd-base".to_string(),
                title: "New base snapshot".to_string(),
                ..Issue::default()
            },
        );

        let err = save_base_snapshot(&issues, &beads_dir).unwrap_err();
        let message = err.to_string();
        assert!(
            message.contains("regular file")
                || message.contains("Temporary base snapshot file")
                || message.contains("Symlink")
                || message.contains("Path"),
            "unexpected error: {message}"
        );
        assert_eq!(
            fs::read_to_string(&snapshot_path).unwrap(),
            "old-snapshot\n",
            "existing base snapshot should remain unchanged on failure"
        );
        assert_eq!(
            fs::read_to_string(&temp_target).unwrap(),
            "do-not-touch",
            "symlink target should not be overwritten"
        );
    }

    #[test]
    fn test_save_base_snapshot_sorts_issues_deterministically() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        fs::create_dir_all(&beads_dir).unwrap();

        let mut issues = HashMap::new();
        issues.insert(
            "bd-z".to_string(),
            Issue {
                id: "bd-z".to_string(),
                title: "Last".to_string(),
                ..Issue::default()
            },
        );
        issues.insert(
            "bd-a".to_string(),
            Issue {
                id: "bd-a".to_string(),
                title: "First".to_string(),
                ..Issue::default()
            },
        );

        save_base_snapshot(&issues, &beads_dir).unwrap();

        let lines: Vec<_> = fs::read_to_string(beads_dir.join("beads.base.jsonl"))
            .unwrap()
            .lines()
            .map(str::to_string)
            .collect();
        assert_eq!(lines.len(), 2);

        let first: Issue = serde_json::from_str(&lines[0]).unwrap();
        let second: Issue = serde_json::from_str(&lines[1]).unwrap();
        assert_eq!(first.id, "bd-a");
        assert_eq!(second.id, "bd-z");
    }

    #[cfg(unix)]
    #[test]
    fn test_load_base_snapshot_rejects_symlink_escape() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        let outside_dir = temp.path().join("outside");
        fs::create_dir_all(&beads_dir).unwrap();
        fs::create_dir_all(&outside_dir).unwrap();

        let outside_snapshot = outside_dir.join("beads.base.jsonl");
        fs::write(&outside_snapshot, "{\"id\":\"bd-outside\"}\n").unwrap();
        symlink(&outside_snapshot, beads_dir.join("beads.base.jsonl")).unwrap();

        let err = load_base_snapshot(&beads_dir).unwrap_err();
        let message = err.to_string();
        assert!(
            message.contains("symlink") || message.contains("Symlink") || message.contains("Path"),
            "unexpected error: {message}"
        );
    }

    #[test]
    fn test_export_with_issues() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let output_path = temp_dir.path().join("issues.jsonl");

        // Create test issues
        let issue1 = make_test_issue("bd-001", "First issue");
        let issue2 = make_test_issue("bd-002", "Second issue");

        storage.create_issue(&issue1, "test").unwrap();
        storage.create_issue(&issue2, "test").unwrap();

        let config = ExportConfig::default();
        let result = export_to_jsonl(&storage, &output_path, &config).unwrap();

        assert_eq!(result.exported_count, 2);
        assert!(result.exported_ids.contains(&"bd-001".to_string()));
        assert!(result.exported_ids.contains(&"bd-002".to_string()));

        // Verify content
        let read_back = read_issues_from_jsonl(&output_path).unwrap();
        assert_eq!(read_back.len(), 2);
        assert_eq!(read_back[0].id, "bd-001");
        assert_eq!(read_back[1].id, "bd-002");
    }

    #[test]
    fn test_safety_guard_empty_over_nonempty() {
        let storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let output_path = temp_dir.path().join("issues.jsonl");

        // Create existing JSONL with issues
        let issue = make_test_issue("bd-existing", "Existing issue");
        let json = serde_json::to_string(&issue).unwrap();
        fs::write(&output_path, format!("{json}\n")).unwrap();

        // Try to export empty database (should fail)
        let config = ExportConfig {
            force: false,
            ..Default::default()
        };
        let result = export_to_jsonl(&storage, &output_path, &config);

        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(err.contains("empty database"));
    }

    #[test]
    fn test_safety_guard_with_force() {
        let storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let output_path = temp_dir.path().join("issues.jsonl");

        // Create existing JSONL with issues
        let issue = make_test_issue("bd-existing", "Existing issue");
        let json = serde_json::to_string(&issue).unwrap();
        fs::write(&output_path, format!("{json}\n")).unwrap();

        // Export with force (should succeed)
        let config = ExportConfig {
            force: true,
            ..Default::default()
        };
        let result = export_to_jsonl(&storage, &output_path, &config);
        assert!(result.is_ok());
    }

    #[test]
    fn test_count_issues_in_jsonl() {
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("test.jsonl");

        // Empty file
        fs::write(&path, "").unwrap();
        assert_eq!(count_issues_in_jsonl(&path).unwrap(), 0);

        // Two issues
        let issue1 = make_test_issue("bd-001", "One");
        let issue2 = make_test_issue("bd-002", "Two");
        let content = format!(
            "{}\n{}\n",
            serde_json::to_string(&issue1).unwrap(),
            serde_json::to_string(&issue2).unwrap()
        );
        fs::write(&path, content).unwrap();
        assert_eq!(count_issues_in_jsonl(&path).unwrap(), 2);
    }

    #[test]
    fn test_get_issue_ids_from_jsonl() {
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("test.jsonl");

        let issue1 = make_test_issue("bd-001", "One");
        let issue2 = make_test_issue("bd-002", "Two");
        let content = format!(
            "{}\n{}\n",
            serde_json::to_string(&issue1).unwrap(),
            serde_json::to_string(&issue2).unwrap()
        );
        fs::write(&path, content).unwrap();

        let ids = get_issue_ids_from_jsonl(&path).unwrap();
        assert_eq!(ids.len(), 2);
        assert!(ids.contains("bd-001"));
        assert!(ids.contains("bd-002"));
    }

    #[test]
    fn test_analyze_jsonl_rejects_duplicate_issue_ids() {
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("duplicate-ids.jsonl");

        let issue1 = make_test_issue("bd-dup", "Original");
        let issue2 = make_test_issue("bd-dup", "Duplicate");
        let content = format!(
            "{}\n{}\n",
            serde_json::to_string(&issue1).unwrap(),
            serde_json::to_string(&issue2).unwrap()
        );
        fs::write(&path, content).unwrap();

        let err = analyze_jsonl(&path).unwrap_err();
        match err {
            BeadsError::Config(message) => {
                assert!(
                    message.contains("Duplicate issue id 'bd-dup'"),
                    "unexpected duplicate-id error: {message}"
                );
            }
            other => panic!("unexpected error: {other:?}"),
        }
    }

    #[test]
    fn test_export_excludes_ephemerals() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let output_path = temp_dir.path().join("issues.jsonl");

        // Create regular and ephemeral issues
        let regular = make_test_issue("bd-regular", "Regular issue");
        let mut ephemeral = make_test_issue("bd-ephemeral", "Ephemeral issue");
        ephemeral.ephemeral = true;

        storage.create_issue(&regular, "test").unwrap();
        storage.create_issue(&ephemeral, "test").unwrap();

        let config = ExportConfig::default();
        let result = export_to_jsonl(&storage, &output_path, &config).unwrap();

        // Only regular issue should be exported
        assert_eq!(result.exported_count, 1);
        assert!(result.exported_ids.contains(&"bd-regular".to_string()));
        assert!(!result.exported_ids.contains(&"bd-ephemeral".to_string()));
    }

    #[test]
    fn test_stale_database_guard_prevents_losing_issues() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let output_path = temp_dir.path().join("issues.jsonl");

        // Create a JSONL with two issues
        let issue1 = make_test_issue("bd-001", "First");
        let issue2 = make_test_issue("bd-002", "Second");
        let content = format!(
            "{}\n{}\n",
            serde_json::to_string(&issue1).unwrap(),
            serde_json::to_string(&issue2).unwrap()
        );
        fs::write(&output_path, content).unwrap();

        // Only create one issue in DB (missing bd-002)
        storage.create_issue(&issue1, "test").unwrap();

        // Export should fail because it would lose bd-002
        let config = ExportConfig::default();
        let result = export_to_jsonl(&storage, &output_path, &config);

        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(err.contains("stale database") || err.contains("lose"));
    }

    #[test]
    fn test_stale_database_guard_with_force_succeeds() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let output_path = temp_dir.path().join("issues.jsonl");

        // Create a JSONL with two issues
        let issue1 = make_test_issue("bd-001", "First");
        let issue2 = make_test_issue("bd-002", "Second");
        let content = format!(
            "{}\n{}\n",
            serde_json::to_string(&issue1).unwrap(),
            serde_json::to_string(&issue2).unwrap()
        );
        fs::write(&output_path, content).unwrap();

        // Only create one issue in DB
        storage.create_issue(&issue1, "test").unwrap();

        // Export with force should succeed
        let config = ExportConfig {
            force: true,
            ..Default::default()
        };
        let result = export_to_jsonl(&storage, &output_path, &config);
        assert!(result.is_ok());
    }

    #[test]
    fn test_auto_import_if_stale_skips_probe_for_allow_stale() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let beads_dir = temp_dir.path().join(".beads");
        let jsonl_path = beads_dir.join("issues.jsonl");

        fs::create_dir_all(&beads_dir).unwrap();
        fs::write(&jsonl_path, [0xFF_u8, b'\n']).unwrap();
        storage
            .set_metadata(METADATA_JSONL_CONTENT_HASH, "stale-hash")
            .unwrap();

        let result = auto_import_if_stale(
            &mut storage,
            &beads_dir,
            &jsonl_path,
            None,
            false,
            true,
            false,
        )
        .unwrap();
        assert!(!result.attempted);
        assert_eq!(result.imported_count, 0);
    }

    #[test]
    fn test_auto_import_if_stale_skips_probe_for_no_auto_import() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let beads_dir = temp_dir.path().join(".beads");
        let jsonl_path = beads_dir.join("issues.jsonl");

        fs::create_dir_all(&beads_dir).unwrap();
        fs::write(&jsonl_path, [0xFF_u8, b'\n']).unwrap();
        storage
            .set_metadata(METADATA_JSONL_CONTENT_HASH, "stale-hash")
            .unwrap();

        let result = auto_import_if_stale(
            &mut storage,
            &beads_dir,
            &jsonl_path,
            None,
            false,
            false,
            true,
        )
        .unwrap();
        assert!(!result.attempted);
        assert_eq!(result.imported_count, 0);
    }

    #[test]
    fn test_compute_staleness_uses_matching_jsonl_mtime_witness() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let jsonl_path = temp_dir.path().join("issues.jsonl");

        fs::write(&jsonl_path, "{\"id\":\"bd-1\"}\n").unwrap();
        let (_, jsonl_mtime_witness) = observed_jsonl_mtime(&jsonl_path).unwrap();
        let current_hash = compute_jsonl_hash(&jsonl_path).unwrap();

        storage
            .set_metadata(METADATA_JSONL_CONTENT_HASH, &current_hash)
            .unwrap();
        storage
            .set_metadata(METADATA_JSONL_MTIME, &jsonl_mtime_witness)
            .unwrap();

        let staleness = compute_staleness(&storage, &jsonl_path).unwrap();
        assert!(staleness.jsonl_exists);
        assert!(!staleness.jsonl_newer);
        assert!(staleness.jsonl_mtime.is_some());
    }

    #[test]
    fn test_compute_staleness_does_not_trust_matching_mtime_without_hash_match() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let jsonl_path = temp_dir.path().join("issues.jsonl");

        fs::write(&jsonl_path, "{\"id\":\"bd-1\"}\n").unwrap();
        let (_, jsonl_mtime_witness) = observed_jsonl_mtime(&jsonl_path).unwrap();

        storage
            .set_metadata(METADATA_JSONL_CONTENT_HASH, "stale-hash")
            .unwrap();
        storage
            .set_metadata(METADATA_JSONL_MTIME, &jsonl_mtime_witness)
            .unwrap();

        let staleness = compute_staleness(&storage, &jsonl_path).unwrap();
        assert!(staleness.jsonl_exists);
        assert!(staleness.jsonl_newer);
        assert!(staleness.jsonl_mtime.is_some());
    }

    #[test]
    fn test_compute_staleness_refreshing_witnesses_backfills_jsonl_size() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let jsonl_path = temp_dir.path().join("issues.jsonl");

        fs::write(&jsonl_path, "{\"id\":\"bd-1\"}\n").unwrap();
        let (_, jsonl_mtime_witness) = observed_jsonl_mtime(&jsonl_path).unwrap();
        let current_hash = compute_jsonl_hash(&jsonl_path).unwrap();
        let jsonl_size = fs::metadata(&jsonl_path).unwrap().len().to_string();

        storage
            .set_metadata(METADATA_JSONL_CONTENT_HASH, &current_hash)
            .unwrap();
        storage
            .set_metadata(METADATA_JSONL_MTIME, &jsonl_mtime_witness)
            .unwrap();

        let staleness = compute_staleness_refreshing_witnesses(&mut storage, &jsonl_path).unwrap();
        assert!(!staleness.jsonl_newer);
        assert_eq!(
            storage.get_metadata(METADATA_JSONL_SIZE).unwrap(),
            Some(jsonl_size)
        );
    }

    #[test]
    fn test_refresh_jsonl_witness_best_effort_ignores_missing_jsonl() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let jsonl_path = temp_dir.path().join("issues.jsonl");

        fs::write(&jsonl_path, "{\"id\":\"bd-1\"}\n").unwrap();
        let observed = observed_jsonl_witness(&jsonl_path).unwrap();
        fs::remove_file(&jsonl_path).unwrap();

        refresh_jsonl_witness_best_effort(&mut storage, &jsonl_path, &observed);

        assert_eq!(storage.get_metadata(METADATA_JSONL_MTIME).unwrap(), None);
        assert_eq!(storage.get_metadata(METADATA_JSONL_SIZE).unwrap(), None);
    }

    #[test]
    fn test_compute_staleness_marks_db_newer_when_force_flush_is_pending() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let jsonl_path = temp_dir.path().join("issues.jsonl");

        storage.set_metadata("needs_flush", "true").unwrap();

        let staleness = compute_staleness(&storage, &jsonl_path).unwrap();
        assert!(staleness.db_newer);
        assert_eq!(staleness.dirty_count, 0);
    }

    #[test]
    fn test_compute_staleness_marks_db_newer_when_jsonl_is_missing_but_db_has_issues() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let jsonl_path = temp_dir.path().join("issues.jsonl");
        let issue = make_test_issue("bd-missing-jsonl", "DB only");
        storage.create_issue(&issue, "tester").unwrap();

        let staleness = compute_staleness(&storage, &jsonl_path).unwrap();
        assert!(staleness.db_newer);
        assert!(!staleness.jsonl_exists);
    }

    #[test]
    fn test_import_records_matching_jsonl_mtime_witness() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let jsonl_path = temp_dir.path().join("issues.jsonl");

        let issue = make_test_issue("bd-import", "Imported issue");
        let json = serde_json::to_string(&issue).unwrap();
        fs::write(&jsonl_path, format!("{json}\n")).unwrap();

        import_from_jsonl(
            &mut storage,
            &jsonl_path,
            &ImportConfig::default(),
            Some("bd-"),
        )
        .unwrap();

        let (_, jsonl_mtime_witness) = observed_jsonl_mtime(&jsonl_path).unwrap();
        let jsonl_size = fs::metadata(&jsonl_path).unwrap().len().to_string();
        assert_eq!(
            storage.get_metadata(METADATA_JSONL_MTIME).unwrap(),
            Some(jsonl_mtime_witness)
        );
        assert_eq!(
            storage.get_metadata(METADATA_JSONL_SIZE).unwrap(),
            Some(jsonl_size)
        );
    }

    #[test]
    fn test_import_skips_child_counters_for_missing_parents() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let jsonl_path = temp_dir.path().join("issues.jsonl");

        let orphan_child = make_test_issue("bd-orphan.6", "Recovered orphan child");
        let json = serde_json::to_string(&orphan_child).unwrap();
        fs::write(&jsonl_path, format!("{json}\n")).unwrap();

        import_from_jsonl(
            &mut storage,
            &jsonl_path,
            &ImportConfig::default(),
            Some("bd-"),
        )
        .unwrap();

        let child_counters = storage
            .execute_raw_query("SELECT parent_id FROM child_counters")
            .unwrap();
        assert!(
            child_counters.is_empty(),
            "orphan child IDs should not rebuild counters for missing parents"
        );
        assert!(
            !storage
                .has_missing_issue_reference("child_counters", "parent_id")
                .unwrap(),
            "child counters must remain free of FK orphans after import"
        );
    }

    #[test]
    fn test_import_rebuilds_nested_child_counters_only_for_existing_parents() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let jsonl_path = temp_dir.path().join("issues.jsonl");

        let orphan_child = make_test_issue("bd-orphan.6", "Recovered orphan child");
        let nested_child = make_test_issue("bd-orphan.6.1", "Recovered nested child");
        let orphan_json = serde_json::to_string(&orphan_child).unwrap();
        let nested_json = serde_json::to_string(&nested_child).unwrap();
        fs::write(&jsonl_path, format!("{orphan_json}\n{nested_json}\n")).unwrap();

        import_from_jsonl(
            &mut storage,
            &jsonl_path,
            &ImportConfig::default(),
            Some("bd-"),
        )
        .unwrap();

        let child_counters = storage
            .execute_raw_query(
                "SELECT parent_id, last_child FROM child_counters ORDER BY parent_id",
            )
            .unwrap();
        assert_eq!(
            child_counters.len(),
            1,
            "only the existing intermediate parent should get a counter"
        );
        assert_eq!(
            child_counters[0]
                .first()
                .and_then(SqliteValue::as_text)
                .unwrap_or(""),
            "bd-orphan.6"
        );
        assert_eq!(
            child_counters[0]
                .get(1)
                .and_then(SqliteValue::as_integer)
                .unwrap_or_default(),
            1
        );
        assert!(
            !storage
                .has_missing_issue_reference("child_counters", "parent_id")
                .unwrap(),
            "nested rebuild should not recreate orphan counters for missing roots"
        );
    }

    #[test]
    fn test_normalize_issue_wisp_detection() {
        let mut issue = make_test_issue("bd-wisp-123", "Wisp issue");
        assert!(!issue.ephemeral);

        normalize_issue(&mut issue);

        // Issue ID containing "-wisp-" should be marked ephemeral
        assert!(issue.ephemeral);
    }

    #[test]
    fn test_normalize_issue_closed_at_repair() {
        let mut issue = make_test_issue("bd-001", "Closed issue");
        issue.status = Status::Closed;
        issue.closed_at = None;

        normalize_issue(&mut issue);

        // closed_at should be set to updated_at for closed issues
        assert!(issue.closed_at.is_some());
        assert_eq!(issue.closed_at, Some(issue.updated_at));
    }

    #[test]
    fn test_normalize_issue_clears_closed_at_for_open() {
        let mut issue = make_test_issue("bd-001", "Open issue");
        issue.status = Status::Open;
        issue.closed_at = Some(Utc::now());

        normalize_issue(&mut issue);

        // closed_at should be cleared for open issues
        assert!(issue.closed_at.is_none());
    }

    #[test]
    fn test_normalize_issue_computes_content_hash() {
        let mut issue = make_test_issue("bd-001", "Test");
        issue.content_hash = None;

        normalize_issue(&mut issue);

        assert!(issue.content_hash.is_some());
        assert!(!issue.content_hash.as_ref().unwrap().is_empty());
    }

    #[test]
    fn test_normalize_issue_normalizes_legacy_standard_dependency_type_with_underscores() {
        let mut issue = make_test_issue("bd-001", "Legacy dependency");
        issue.dependencies.push(crate::model::Dependency {
            issue_id: issue.id.clone(),
            depends_on_id: "bd-002".to_string(),
            dep_type: crate::model::DependencyType::Custom("parent_child".to_string()),
            created_at: Utc::now(),
            created_by: None,
            metadata: None,
            thread_id: None,
        });

        normalize_issue(&mut issue);

        assert_eq!(
            issue.dependencies[0].dep_type,
            crate::model::DependencyType::ParentChild
        );
    }

    #[test]
    fn test_normalize_issue_preserves_custom_dependency_type_with_underscores() {
        let mut issue = make_test_issue("bd-001", "Custom dependency");
        issue.dependencies.push(crate::model::Dependency {
            issue_id: issue.id.clone(),
            depends_on_id: "bd-002".to_string(),
            dep_type: crate::model::DependencyType::Custom("review_needed".to_string()),
            created_at: Utc::now(),
            created_by: None,
            metadata: None,
            thread_id: None,
        });

        normalize_issue(&mut issue);

        assert_eq!(
            issue.dependencies[0].dep_type,
            crate::model::DependencyType::Custom("review_needed".to_string())
        );
    }

    #[test]
    fn test_import_collision_by_id_updates_newer() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("issues.jsonl");

        // Create existing issue in DB with older timestamp
        let mut existing = make_test_issue("test-001", "Old title");
        existing.updated_at = Utc::now() - chrono::Duration::hours(1);
        storage.create_issue(&existing, "test").unwrap();

        // Create JSONL with same ID but newer timestamp and new title
        let mut incoming = make_test_issue("test-001", "New title");
        incoming.updated_at = Utc::now();
        let json = serde_json::to_string(&incoming).unwrap();
        fs::write(&path, format!("{json}\n")).unwrap();

        // Import should update since incoming is newer
        let config = ImportConfig::default();
        let result = import_from_jsonl(&mut storage, &path, &config, Some("test-")).unwrap();
        assert_eq!(result.imported_count, 1);
        assert_eq!(result.created_count, 0);
        assert_eq!(result.updated_count, 1);

        // The existing issue should be updated
        let updated = storage.get_issue("test-001").unwrap().unwrap();
        assert_eq!(updated.title, "New title");
    }

    #[test]
    fn test_import_collision_by_id_skips_older() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("issues.jsonl");

        // Create existing issue in DB with newer timestamp
        let mut existing = make_test_issue("test-001", "Newer title");
        existing.updated_at = Utc::now();
        storage.create_issue(&existing, "test").unwrap();

        // Create JSONL with same ID but older timestamp
        let mut incoming = make_test_issue("test-001", "Older title");
        incoming.created_at = Utc::now() - chrono::Duration::hours(2); // Fix timestamp to be valid
        incoming.updated_at = Utc::now() - chrono::Duration::hours(1);
        let json = serde_json::to_string(&incoming).unwrap();
        fs::write(&path, format!("{json}\n")).unwrap();

        // Import should skip since existing is newer
        let config = ImportConfig::default();
        let result = import_from_jsonl(&mut storage, &path, &config, Some("test-")).unwrap();
        assert_eq!(result.skipped_count, 1);

        let unchanged = storage.get_issue("test-001").unwrap().unwrap();
        assert_eq!(unchanged.title, "Newer title");
    }

    #[test]
    fn test_import_collision_by_external_ref_same_id() {
        // Test collision detection by external_ref when IDs also match
        let storage = SqliteStorage::open_memory().unwrap();

        let mut ext_issue = make_issue_at("bd-ext", "External", fixed_time(100));
        ext_issue.external_ref = Some("JIRA-1".to_string());
        set_content_hash(&mut ext_issue);
        storage.upsert_issue_for_import(&ext_issue).unwrap();

        let mut hash_issue = make_issue_at("bd-hash", "Incoming", fixed_time(200));
        set_content_hash(&mut hash_issue);
        storage.upsert_issue_for_import(&hash_issue).unwrap();

        // Incoming has same external_ref as ext_issue - should match on external_ref
        // even though it has same title/content_hash as hash_issue
        let mut incoming = make_issue_at("bd-new", "Incoming", fixed_time(300));
        incoming.external_ref = Some("JIRA-1".to_string());
        let computed_hash = crate::util::content_hash(&incoming);

        let (id_by_ext_ref, id_by_hash, meta_by_id) = build_collision_maps(&storage);
        let collision = detect_collision(
            &incoming,
            &id_by_ext_ref,
            &id_by_hash,
            &meta_by_id,
            &computed_hash,
        );
        assert!(
            matches!(collision, CollisionResult::Match { .. }),
            "expected match"
        );
        if let CollisionResult::Match {
            existing_id,
            match_type,
            phase,
        } = collision
        {
            assert_eq!(existing_id, "bd-ext");
            assert_eq!(match_type, MatchType::ExternalRef);
            assert_eq!(phase, 1);
        }
    }

    #[test]
    fn test_import_tombstone_protection() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("issues.jsonl");

        // Create tombstone in DB
        let mut tombstone = make_issue_at("test-001", "Tombstone", fixed_time(100));
        tombstone.status = Status::Tombstone;
        tombstone.deleted_at = Some(Utc::now());
        storage.create_issue(&tombstone, "test").unwrap();

        // Create JSONL with same ID but trying to resurrect
        let mut incoming = make_issue_at("test-001", "Resurrected", fixed_time(200));
        incoming.status = Status::Open;
        let json = serde_json::to_string(&incoming).unwrap();
        fs::write(&path, format!("{json}\n")).unwrap();

        // Import should skip due to tombstone protection
        let config = ImportConfig::default();
        let result = import_from_jsonl(&mut storage, &path, &config, Some("test-")).unwrap();
        assert_eq!(result.tombstone_skipped, 1);

        let still_tombstone = storage.get_issue("test-001").unwrap().unwrap();
        assert_eq!(still_tombstone.status, Status::Tombstone);
    }

    #[test]
    fn test_import_new_issue_creates() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("issues.jsonl");

        // Create JSONL with new issue
        let new_issue = make_test_issue("test-new", "Brand new");
        let json = serde_json::to_string(&new_issue).unwrap();
        fs::write(&path, format!("{json}\n")).unwrap();

        let config = ImportConfig::default();
        let result = import_from_jsonl(&mut storage, &path, &config, Some("test-")).unwrap();

        // New issue should be imported
        assert_eq!(result.imported_count, 1);
        assert_eq!(result.created_count, 1);
        assert_eq!(result.updated_count, 0);
        assert_eq!(result.skipped_count, 0);
        assert!(storage.get_issue("test-new").unwrap().is_some());
    }

    #[test]
    fn test_get_issue_ids_missing_file_returns_empty() {
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("nonexistent.jsonl");

        let ids = get_issue_ids_from_jsonl(&path).unwrap();
        assert!(ids.is_empty());
    }

    #[test]
    fn test_count_issues_missing_file_returns_zero() {
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("nonexistent.jsonl");

        let count = count_issues_in_jsonl(&path).unwrap();
        assert_eq!(count, 0);
    }

    #[test]
    fn test_export_computes_content_hash() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let output_path = temp_dir.path().join("issues.jsonl");

        let issue = make_test_issue("bd-001", "Test");
        storage.create_issue(&issue, "test").unwrap();

        let config = ExportConfig::default();
        let result = export_to_jsonl(&storage, &output_path, &config).unwrap();

        // Result should include a non-empty content hash
        assert!(!result.content_hash.is_empty());
        // Hash should be hex (64 chars for SHA256)
        assert_eq!(result.content_hash.len(), 64);
    }

    #[test]
    fn test_export_deterministic_hash() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();

        let issue = make_test_issue("bd-001", "Deterministic");
        storage.create_issue(&issue, "test").unwrap();

        let config = ExportConfig::default();

        // Export twice to different files
        let path1 = temp_dir.path().join("export1.jsonl");
        let path2 = temp_dir.path().join("export2.jsonl");

        let result1 = export_to_jsonl(&storage, &path1, &config).unwrap();
        let result2 = export_to_jsonl(&storage, &path2, &config).unwrap();

        // Hashes should be identical for same content
        assert_eq!(result1.content_hash, result2.content_hash);
    }

    #[test]
    fn test_import_skips_ephemerals() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("issues.jsonl");

        // Create JSONL with ephemeral issue
        let mut ephemeral = make_test_issue("test-001", "Ephemeral");
        ephemeral.ephemeral = true;
        let json = serde_json::to_string(&ephemeral).unwrap();
        fs::write(&path, format!("{json}\n")).unwrap();

        let config = ImportConfig::default();
        let result = import_from_jsonl(&mut storage, &path, &config, Some("test-")).unwrap();
        assert_eq!(result.skipped_count, 1);
        assert_eq!(result.imported_count, 0);
        assert!(storage.get_issue("test-001").unwrap().is_none());
    }

    #[test]
    fn test_import_handles_empty_lines() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("issues.jsonl");

        // Create JSONL with empty lines
        let issue = make_test_issue("test-001", "Valid");
        let json = serde_json::to_string(&issue).unwrap();
        let content = format!("\n{json}\n\n\n");
        fs::write(&path, content).unwrap();

        let config = ImportConfig::default();
        let result = import_from_jsonl(&mut storage, &path, &config, Some("test-")).unwrap();
        assert_eq!(result.imported_count, 1);
    }

    #[test]
    fn test_import_keeps_distinct_ids_with_identical_content() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("issues.jsonl");

        let issue1 = make_test_issue("test-001", "Same content");
        let issue2 = make_test_issue("test-002", "Same content");
        let content = format!(
            "{}\n{}\n",
            serde_json::to_string(&issue1).unwrap(),
            serde_json::to_string(&issue2).unwrap()
        );
        fs::write(&path, content).unwrap();

        let result =
            import_from_jsonl(&mut storage, &path, &ImportConfig::default(), Some("test-"))
                .unwrap();
        assert_eq!(result.imported_count, 2);
        assert_eq!(result.skipped_count, 0);
        assert!(storage.get_issue("test-001").unwrap().is_some());
        assert!(storage.get_issue("test-002").unwrap().is_some());
    }

    #[test]
    fn test_import_restores_foreign_keys_after_relation_sync_failure() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("issues.jsonl");

        let issue = make_test_issue("test-001", "Broken relations");
        let json = serde_json::to_string(&issue).unwrap();
        fs::write(&path, format!("{json}\n")).unwrap();

        storage.execute_test_sql("DROP TABLE comments;").unwrap();

        let err = import_from_jsonl(&mut storage, &path, &ImportConfig::default(), Some("test-"))
            .unwrap_err();
        assert!(
            err.to_string().contains("comments"),
            "unexpected error: {err}"
        );

        let fk_enabled = storage
            .execute_raw_query("PRAGMA foreign_keys")
            .unwrap()
            .first()
            .and_then(|row| row.first())
            .and_then(SqliteValue::as_integer)
            .unwrap_or(0);
        assert_eq!(fk_enabled, 1, "foreign key enforcement should be restored");
    }

    #[test]
    fn test_restore_foreign_keys_after_import_errors_on_dangling_rows() {
        let storage = SqliteStorage::open_memory().unwrap();

        storage
            .execute_test_sql(
                "PRAGMA foreign_keys = OFF;
                 INSERT INTO comments (issue_id, author, text, created_at)
                 VALUES ('missing-issue', 'tester', 'dangling', '2026-01-01T00:00:00Z');",
            )
            .unwrap();

        let err = restore_foreign_keys_after_import(&storage, true).unwrap_err();
        assert!(
            err.to_string()
                .contains("orphaned rows in comments.issue_id"),
            "unexpected error: {err}"
        );

        let fk_enabled = storage
            .execute_raw_query("PRAGMA foreign_keys")
            .unwrap()
            .first()
            .and_then(|row| row.first())
            .and_then(SqliteValue::as_integer)
            .unwrap_or(0);
        assert_eq!(fk_enabled, 1, "foreign key enforcement should be restored");
    }

    #[test]
    fn test_import_rolls_back_partial_changes_after_relation_sync_failure() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("issues.jsonl");

        let existing = make_test_issue("test-existing", "Existing issue");
        storage.create_issue(&existing, "test").unwrap();
        storage
            .set_export_hashes(&[("test-existing".to_string(), "existing-hash".to_string())])
            .unwrap();

        let issue = make_test_issue("test-001", "Broken relations");
        let json = serde_json::to_string(&issue).unwrap();
        fs::write(&path, format!("{json}\n")).unwrap();

        storage.execute_test_sql("DROP TABLE comments;").unwrap();

        let err = import_from_jsonl(&mut storage, &path, &ImportConfig::default(), Some("test-"))
            .unwrap_err();
        assert!(
            err.to_string().contains("comments"),
            "unexpected error: {err}"
        );

        assert!(
            storage.get_issue("test-001").unwrap().is_none(),
            "failed import should not leave a partially inserted issue behind"
        );
        assert!(
            storage.get_issue("test-existing").unwrap().is_some(),
            "failed import should preserve pre-existing issues"
        );

        let export_hash_rows = storage
            .execute_raw_query("SELECT issue_id, content_hash FROM export_hashes")
            .unwrap();
        assert_eq!(export_hash_rows.len(), 1, "export hashes should roll back");
        assert_eq!(
            export_hash_rows[0]
                .first()
                .and_then(SqliteValue::as_text)
                .unwrap_or(""),
            "test-existing"
        );
    }

    #[test]
    fn test_detect_collision_external_ref_priority() {
        let storage = SqliteStorage::open_memory().unwrap();

        let mut ext_issue = make_issue_at("bd-ext", "External", fixed_time(100));
        ext_issue.external_ref = Some("JIRA-1".to_string());
        set_content_hash(&mut ext_issue);
        storage.upsert_issue_for_import(&ext_issue).unwrap();

        let mut hash_issue = make_issue_at("bd-hash", "Incoming", fixed_time(200));
        set_content_hash(&mut hash_issue);
        storage.upsert_issue_for_import(&hash_issue).unwrap();

        // Incoming has same external_ref as ext_issue - should match on external_ref
        // even though it has same title/content_hash as hash_issue
        let mut incoming = make_issue_at("bd-new", "Incoming", fixed_time(300));
        incoming.external_ref = Some("JIRA-1".to_string());
        let computed_hash = crate::util::content_hash(&incoming);

        let (id_by_ext_ref, id_by_hash, meta_by_id) = build_collision_maps(&storage);
        let collision = detect_collision(
            &incoming,
            &id_by_ext_ref,
            &id_by_hash,
            &meta_by_id,
            &computed_hash,
        );
        assert!(
            matches!(collision, CollisionResult::Match { .. }),
            "expected match"
        );
        if let CollisionResult::Match {
            existing_id,
            match_type,
            phase,
        } = collision
        {
            assert_eq!(existing_id, "bd-ext");
            assert_eq!(match_type, MatchType::ExternalRef);
            assert_eq!(phase, 1);
        }
    }

    #[test]
    fn test_detect_collision_id_preempts_content_hash() {
        let storage = SqliteStorage::open_memory().unwrap();

        let mut hash_issue = make_issue_at("bd-hash", "Same Content", fixed_time(100));
        set_content_hash(&mut hash_issue);
        storage.upsert_issue_for_import(&hash_issue).unwrap();

        let mut id_issue = make_issue_at("bd-same", "Different Content", fixed_time(100));
        set_content_hash(&mut id_issue);
        storage.upsert_issue_for_import(&id_issue).unwrap();

        let incoming = make_issue_at("bd-same", "Same Content", fixed_time(200));
        let computed_hash = crate::util::content_hash(&incoming);

        let (id_by_ext_ref, id_by_hash, meta_by_id) = build_collision_maps(&storage);
        let collision = detect_collision(
            &incoming,
            &id_by_ext_ref,
            &id_by_hash,
            &meta_by_id,
            &computed_hash,
        );
        assert!(
            matches!(collision, CollisionResult::Match { .. }),
            "expected match"
        );
        if let CollisionResult::Match {
            existing_id,
            match_type,
            phase,
        } = collision
        {
            assert_eq!(existing_id, "bd-same");
            assert_eq!(match_type, MatchType::Id);
            assert_eq!(phase, 2);
        }
    }

    #[test]
    fn test_detect_collision_duplicate_content_hash_keeps_first_match() {
        let storage = SqliteStorage::open_memory().unwrap();

        let mut first = make_issue_at("bd-first", "Same Content", fixed_time(100));
        set_content_hash(&mut first);
        storage.upsert_issue_for_import(&first).unwrap();

        let mut second = make_issue_at("bd-second", "Same Content", fixed_time(200));
        set_content_hash(&mut second);
        storage.upsert_issue_for_import(&second).unwrap();

        let incoming = make_issue_at("bd-new", "Same Content", fixed_time(300));
        let computed_hash = crate::util::content_hash(&incoming);

        let (id_by_ext_ref, id_by_hash, meta_by_id) = build_collision_maps(&storage);
        let collision = detect_collision(
            &incoming,
            &id_by_ext_ref,
            &id_by_hash,
            &meta_by_id,
            &computed_hash,
        );

        assert!(
            matches!(collision, CollisionResult::Match { .. }),
            "expected match"
        );
        if let CollisionResult::Match {
            existing_id,
            match_type,
            phase,
        } = collision
        {
            assert_eq!(existing_id, "bd-first");
            assert_eq!(match_type, MatchType::ContentHash);
            assert_eq!(phase, 3);
        }
    }

    #[test]
    fn test_detect_collision_id_match() {
        let mut storage = SqliteStorage::open_memory().unwrap();

        let existing = make_issue_at("bd-1", "Existing", fixed_time(100));
        storage.create_issue(&existing, "test").unwrap();

        let incoming = make_issue_at("bd-1", "Incoming", fixed_time(200));

        let computed_hash = crate::util::content_hash(&incoming);
        let (id_by_ext_ref, id_by_hash, meta_by_id) = build_collision_maps(&storage);
        let collision = detect_collision(
            &incoming,
            &id_by_ext_ref,
            &id_by_hash,
            &meta_by_id,
            &computed_hash,
        );

        assert!(
            matches!(collision, CollisionResult::Match { .. }),
            "expected match"
        );
        if let CollisionResult::Match {
            existing_id,
            match_type,
            phase,
        } = collision
        {
            assert_eq!(existing_id, "bd-1");
            assert_eq!(match_type, MatchType::Id);
            assert_eq!(phase, 2);
        }
    }

    #[test]
    fn test_determine_action_tombstone_skip() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let mut tombstone = make_issue_at("bd-1", "Tombstone", fixed_time(100));
        tombstone.status = Status::Tombstone;
        storage.create_issue(&tombstone, "test").unwrap();

        let incoming = make_issue_at("bd-1", "Incoming", fixed_time(200));
        let collision = CollisionResult::Match {
            existing_id: "bd-1".to_string(),
            match_type: MatchType::Id,
            phase: 3,
        };
        let (_, _, meta_by_id) = build_collision_maps(&storage);
        let action = determine_action(&collision, &incoming, &meta_by_id, false).unwrap();
        assert!(
            matches!(action, CollisionAction::Skip { .. }),
            "expected tombstone skip"
        );
        if let CollisionAction::Skip { reason } = action {
            assert!(reason.contains("Tombstone protection"));
        }
    }

    #[test]
    fn test_determine_action_timestamp_comparison() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let existing = make_issue_at("bd-1", "Existing", fixed_time(100));
        storage.create_issue(&existing, "test").unwrap();

        let collision = CollisionResult::Match {
            existing_id: "bd-1".to_string(),
            match_type: MatchType::Id,
            phase: 3,
        };
        let (_, _, meta_by_id) = build_collision_maps(&storage);

        let newer = make_issue_at("bd-1", "Incoming", fixed_time(200));
        let action = determine_action(&collision, &newer, &meta_by_id, false).unwrap();
        assert!(
            matches!(action, CollisionAction::Update { .. }),
            "expected update action"
        );

        let equal = make_issue_at("bd-1", "Incoming", fixed_time(100));
        let action = determine_action(&collision, &equal, &meta_by_id, false).unwrap();
        assert!(
            matches!(action, CollisionAction::Skip { .. }),
            "expected equal timestamp skip"
        );
        if let CollisionAction::Skip { reason } = action {
            assert!(reason.contains("Equal timestamps"));
        }

        let older = make_issue_at("bd-1", "Incoming", fixed_time(50));
        let action = determine_action(&collision, &older, &meta_by_id, false).unwrap();
        assert!(
            matches!(action, CollisionAction::Skip { .. }),
            "expected older timestamp skip"
        );
        if let CollisionAction::Skip { reason } = action {
            assert!(reason.contains("Existing is newer"));
        }
    }

    #[test]
    fn test_import_prefix_mismatch_error() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("issues.jsonl");

        let issue = make_issue_at("xx-001", "Bad prefix", fixed_time(100));
        let json = serde_json::to_string(&issue).unwrap();
        fs::write(&path, format!("{json}\n")).unwrap();

        let config = ImportConfig::default();
        let err = import_from_jsonl(&mut storage, &path, &config, Some("bd")).unwrap_err();
        assert!(err.to_string().contains("Prefix mismatch"));
    }

    #[test]
    fn test_import_prefix_mismatch_error_for_shared_prefix_superset() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("issues.jsonl");

        let issue = make_issue_at("bdx-001", "Looks similar but wrong prefix", fixed_time(100));
        let json = serde_json::to_string(&issue).unwrap();
        fs::write(&path, format!("{json}\n")).unwrap();

        let err = import_from_jsonl(&mut storage, &path, &ImportConfig::default(), Some("bd"))
            .unwrap_err();
        assert!(err.to_string().contains("Prefix mismatch"));
        assert!(err.to_string().contains("bdx-001"));
    }

    #[test]
    fn test_import_duplicate_external_ref_errors() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("issues.jsonl");

        let mut issue1 = make_issue_at("bd-001", "Issue 1", fixed_time(100));
        issue1.external_ref = Some("JIRA-1".to_string());
        let mut issue2 = make_issue_at("bd-002", "Issue 2", fixed_time(120));
        issue2.external_ref = Some("JIRA-1".to_string());

        let content = format!(
            "{}\n{}\n",
            serde_json::to_string(&issue1).unwrap(),
            serde_json::to_string(&issue2).unwrap()
        );
        fs::write(&path, content).unwrap();

        let config = ImportConfig::default();
        let err = import_from_jsonl(&mut storage, &path, &config, None).unwrap_err();
        assert!(err.to_string().contains("Duplicate external_ref"));
    }

    #[test]
    fn test_import_duplicate_issue_ids_error() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("issues.jsonl");

        let issue1 = make_issue_at("bd-001", "Issue 1", fixed_time(100));
        let issue2 = make_issue_at("bd-001", "Issue 2", fixed_time(120));

        let content = format!(
            "{}\n{}\n",
            serde_json::to_string(&issue1).unwrap(),
            serde_json::to_string(&issue2).unwrap()
        );
        fs::write(&path, content).unwrap();

        let err = import_from_jsonl(&mut storage, &path, &ImportConfig::default(), Some("bd"))
            .unwrap_err();
        assert!(err.to_string().contains("Duplicate issue id 'bd-001'"));
    }

    #[test]
    fn test_import_duplicate_external_ref_clears_and_inserts() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("issues.jsonl");

        let mut issue1 = make_issue_at("bd-001", "Issue 1", fixed_time(100));
        issue1.external_ref = Some("JIRA-1".to_string());
        let mut issue2 = make_issue_at("bd-002", "Issue 2", fixed_time(120));
        issue2.external_ref = Some("JIRA-1".to_string());

        let content = format!(
            "{}\n{}\n",
            serde_json::to_string(&issue1).unwrap(),
            serde_json::to_string(&issue2).unwrap()
        );
        fs::write(&path, content).unwrap();

        let config = ImportConfig {
            clear_duplicate_external_refs: true,
            ..Default::default()
        };
        let result = import_from_jsonl(&mut storage, &path, &config, None).unwrap();

        assert_eq!(result.imported_count, 2);
        assert_eq!(result.skipped_count, 0);
        let first = storage.get_issue("bd-001").unwrap().unwrap();
        let second = storage.get_issue("bd-002").unwrap().unwrap();
        assert_eq!(first.external_ref.as_deref(), Some("JIRA-1"));
        assert!(second.external_ref.is_none());
    }

    #[test]
    fn test_export_deterministic_order() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let output_path = temp_dir.path().join("issues.jsonl");

        let issue_a = make_test_issue("bd-z", "Zed");
        let issue_b = make_test_issue("bd-a", "Aye");
        let issue_c = make_test_issue("bd-m", "Em");

        storage.create_issue(&issue_a, "test").unwrap();
        storage.create_issue(&issue_b, "test").unwrap();
        storage.create_issue(&issue_c, "test").unwrap();

        let config = ExportConfig::default();
        export_to_jsonl(&storage, &output_path, &config).unwrap();

        let ids = read_issues_from_jsonl(&output_path)
            .unwrap()
            .into_iter()
            .map(|issue| issue.id)
            .collect::<Vec<_>>();
        assert_eq!(ids, vec!["bd-a", "bd-m", "bd-z"]);
    }

    #[test]
    fn test_normalize_issue_for_export_orders_identical_comments_by_id() {
        let timestamp = fixed_time(100);
        let mut issue = make_test_issue("bd-1", "Ordering");
        issue.comments = vec![
            Comment {
                id: 9,
                issue_id: issue.id.clone(),
                author: "tester".to_string(),
                body: "same".to_string(),
                created_at: timestamp,
            },
            Comment {
                id: 2,
                issue_id: issue.id.clone(),
                author: "tester".to_string(),
                body: "same".to_string(),
                created_at: timestamp,
            },
        ];

        normalize_issue_for_export(&mut issue);

        let ids = issue
            .comments
            .into_iter()
            .map(|comment| comment.id)
            .collect::<Vec<_>>();
        assert_eq!(ids, vec![2, 9]);
    }

    #[test]
    fn test_finalize_export_updates_metadata_and_clears_dirty() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let output_path = temp_dir.path().join("issues.jsonl");

        let issue = make_test_issue("bd-1", "Issue");
        storage.create_issue(&issue, "test").unwrap();
        assert_eq!(storage.get_dirty_issue_ids().unwrap().len(), 1);

        let config = ExportConfig::default();
        let result = export_to_jsonl(&storage, &output_path, &config).unwrap();
        finalize_export(
            &mut storage,
            &result,
            Some(&result.issue_hashes),
            &output_path,
        )
        .unwrap();

        assert!(storage.get_dirty_issue_ids().unwrap().is_empty());
        assert!(
            storage
                .get_metadata(METADATA_JSONL_CONTENT_HASH)
                .unwrap()
                .is_some()
        );
        assert!(
            storage
                .get_metadata(METADATA_LAST_EXPORT_TIME)
                .unwrap()
                .is_some()
        );
        assert!(
            storage
                .get_metadata(METADATA_JSONL_MTIME)
                .unwrap()
                .is_some()
        );
        assert!(storage.get_metadata(METADATA_JSONL_SIZE).unwrap().is_some());
    }

    #[test]
    fn test_filter_dirty_metadata_for_export_only_includes_exported_ids() {
        let dirty_metadata = vec![
            ("bd-1".to_string(), "t1".to_string()),
            ("bd-2".to_string(), "t2".to_string()),
            ("bd-3".to_string(), "t3".to_string()),
        ];
        let exported_ids = vec!["bd-1".to_string()];
        let skipped_tombstone_ids = vec!["bd-3".to_string()];

        let filtered = filter_dirty_metadata_for_export(
            &dirty_metadata,
            &exported_ids,
            &skipped_tombstone_ids,
        );

        assert_eq!(
            filtered,
            vec![
                ("bd-1".to_string(), "t1".to_string()),
                ("bd-3".to_string(), "t3".to_string()),
            ]
        );
    }

    #[test]
    fn test_finalize_export_rolls_back_on_failure() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let temp_dir = TempDir::new().unwrap();
        let output_path = temp_dir.path().join("issues.jsonl");

        let issue = make_test_issue("bd-finalize", "Issue");
        storage.create_issue(&issue, "test").unwrap();
        assert_eq!(storage.get_dirty_issue_ids().unwrap().len(), 1);

        let config = ExportConfig::default();
        let result = export_to_jsonl(&storage, &output_path, &config).unwrap();

        let invalid_issue_hashes = vec![("bd-missing".to_string(), "hash".to_string())];

        let err = finalize_export(
            &mut storage,
            &result,
            Some(&invalid_issue_hashes),
            &output_path,
        )
        .unwrap_err();
        match err {
            BeadsError::Database(_) => {}
            other => panic!("unexpected error: {other:?}"),
        }

        assert_eq!(
            storage.get_dirty_issue_ids().unwrap(),
            vec!["bd-finalize".to_string()]
        );
        assert!(storage.get_export_hash("bd-finalize").unwrap().is_none());
        assert!(
            storage
                .get_metadata(METADATA_JSONL_CONTENT_HASH)
                .unwrap()
                .is_none()
        );
        assert!(
            storage
                .get_metadata(METADATA_LAST_EXPORT_TIME)
                .unwrap()
                .is_none()
        );
        assert!(
            storage
                .get_metadata(METADATA_JSONL_MTIME)
                .unwrap()
                .is_none()
        );
        assert!(storage.get_metadata(METADATA_JSONL_SIZE).unwrap().is_none());
    }

    #[test]
    fn test_export_policy_strict_fails_on_write_error() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let issue1 = make_test_issue("bd-001", "First");
        let issue2 = make_test_issue("bd-002", "Second");
        storage.create_issue(&issue1, "test").unwrap();
        storage.create_issue(&issue2, "test").unwrap();

        let mut writer = LineFailWriter::new("bd-002");
        let result = export_to_writer_with_policy(&storage, &mut writer, ExportErrorPolicy::Strict);
        assert!(result.is_err());
    }

    #[test]
    fn test_export_policy_best_effort_skips_write_error() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let issue1 = make_test_issue("bd-001", "First");
        let issue2 = make_test_issue("bd-002", "Second");
        storage.create_issue(&issue1, "test").unwrap();
        storage.create_issue(&issue2, "test").unwrap();

        let mut writer = LineFailWriter::new("bd-002");
        let (result, report) =
            export_to_writer_with_policy(&storage, &mut writer, ExportErrorPolicy::BestEffort)
                .unwrap();
        assert_eq!(result.exported_count, 1);
        assert_eq!(report.errors.len(), 1);
        let output = writer.into_string();
        assert!(output.contains("bd-001"));
        assert!(!output.contains("bd-002"));
    }

    #[test]
    fn test_export_policy_partial_collects_write_error() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let issue1 = make_test_issue("bd-001", "First");
        let issue2 = make_test_issue("bd-002", "Second");
        storage.create_issue(&issue1, "test").unwrap();
        storage.create_issue(&issue2, "test").unwrap();

        let mut writer = LineFailWriter::new("bd-002");
        let (result, report) =
            export_to_writer_with_policy(&storage, &mut writer, ExportErrorPolicy::Partial)
                .unwrap();

        assert_eq!(result.exported_count, 1);
        assert_eq!(report.errors.len(), 1);
    }

    #[test]
    fn test_export_policy_required_core_fails_on_issue_error() {
        let mut storage = SqliteStorage::open_memory().unwrap();
        let issue1 = make_test_issue("bd-001", "First");
        let issue2 = make_test_issue("bd-002", "Second");
        storage.create_issue(&issue1, "test").unwrap();
        storage.create_issue(&issue2, "test").unwrap();

        let mut writer = LineFailWriter::new("bd-002");
        let result =
            export_to_writer_with_policy(&storage, &mut writer, ExportErrorPolicy::RequiredCore);
        assert!(result.is_err());
    }

    #[test]
    fn test_export_policy_required_core_allows_non_core_errors() {
        // This test verifies that RequiredCore policy exports all issues successfully
        // and would tolerate non-core errors (Label, Dependency, Comment) if they occurred.
        // The test doesn't generate non-core errors since the setup has no labels/deps.
        let mut storage = SqliteStorage::open_memory().unwrap();
        let issue1 = make_test_issue("bd-001", "First");
        let issue2 = make_test_issue("bd-002", "Second");
        storage.create_issue(&issue1, "test").unwrap();
        storage.create_issue(&issue2, "test").unwrap();

        let mut writer = Vec::new();
        let (result, report) =
            export_to_writer_with_policy(&storage, &mut writer, ExportErrorPolicy::RequiredCore)
                .unwrap();

        assert_eq!(result.exported_count, 2);
        // Any errors present should be non-core (Issue errors would cause failure above)
        for err in &report.errors {
            assert_ne!(
                err.entity_type,
                ExportEntityType::Issue,
                "Issue errors should fail RequiredCore policy"
            );
        }
    }

    // ============================================================================
    // PREFLIGHT TESTS (beads_rust-0v1.2.7)
    // ============================================================================

    #[test]
    fn test_preflight_check_status_ordering() {
        // Verify that PreflightCheckStatus can be used for comparison
        assert_ne!(PreflightCheckStatus::Pass, PreflightCheckStatus::Warn);
        assert_ne!(PreflightCheckStatus::Warn, PreflightCheckStatus::Fail);
        assert_ne!(PreflightCheckStatus::Pass, PreflightCheckStatus::Fail);
    }

    #[test]
    fn test_preflight_result_aggregates_status() {
        let mut result = PreflightResult::new();

        // Initial state is Pass
        assert_eq!(result.overall_status, PreflightCheckStatus::Pass);
        assert!(result.is_ok());
        assert!(result.has_no_failures());

        // Add a passing check
        result.add(PreflightCheck::pass("test1", "Test 1", "Passed"));
        assert_eq!(result.overall_status, PreflightCheckStatus::Pass);

        // Add a warning - overall becomes Warn
        result.add(PreflightCheck::warn("test2", "Test 2", "Warning", "Fix it"));
        assert_eq!(result.overall_status, PreflightCheckStatus::Warn);
        assert!(!result.is_ok());
        assert!(result.has_no_failures());

        // Add a failure - overall becomes Fail
        result.add(PreflightCheck::fail("test3", "Test 3", "Failed", "Fix it"));
        assert_eq!(result.overall_status, PreflightCheckStatus::Fail);
        assert!(!result.is_ok());
        assert!(!result.has_no_failures());

        // Check counts
        assert_eq!(result.failures().len(), 1);
        assert_eq!(result.warnings().len(), 1);
    }

    #[test]
    fn test_preflight_result_into_result_succeeds_on_pass() {
        let mut result = PreflightResult::new();
        result.add(PreflightCheck::pass("test", "Test", "OK"));

        let converted = result.into_result();
        assert!(converted.is_ok());
    }

    #[test]
    fn test_preflight_result_into_result_succeeds_on_warn() {
        let mut result = PreflightResult::new();
        result.add(PreflightCheck::warn("test", "Test", "Warning", "Fix"));

        let converted = result.into_result();
        assert!(converted.is_ok());
    }

    #[test]
    fn test_preflight_result_into_result_fails_on_fail() {
        let mut result = PreflightResult::new();
        result.add(PreflightCheck::fail("test", "Test", "Failed", "Fix it"));

        let converted = result.into_result();
        assert!(converted.is_err());

        let err_msg = converted.unwrap_err().to_string();
        assert!(err_msg.contains("Preflight checks failed"));
        assert!(err_msg.contains("test"));
        assert!(err_msg.contains("Failed"));
    }

    #[test]
    fn test_preflight_import_rejects_nonexistent_file() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("nonexistent.jsonl");

        let config = ImportConfig {
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        let result = preflight_import(&jsonl_path, &config, None).unwrap();

        assert_eq!(result.overall_status, PreflightCheckStatus::Fail);
        assert!(result.failures().iter().any(|c| c.name == "file_readable"));
    }

    #[test]
    fn test_preflight_import_rejects_conflict_markers() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("issues.jsonl");

        // Write a file with conflict markers
        let mut file = std::fs::File::create(&jsonl_path).unwrap();
        writeln!(file, "<<<<<<< HEAD").unwrap();
        file.write_all(br#"{"id":"bd-1","title":"Test"}"#).unwrap();
        writeln!(file).unwrap();
        writeln!(file, "=======").unwrap();
        file.write_all(br#"{"id":"bd-1","title":"Test Modified"}"#)
            .unwrap();
        writeln!(file).unwrap();
        writeln!(file, ">>>>>>> branch").unwrap();

        let config = ImportConfig {
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        let result = preflight_import(&jsonl_path, &config, None).unwrap();

        assert_eq!(result.overall_status, PreflightCheckStatus::Fail);
        assert!(
            result
                .failures()
                .iter()
                .any(|c| c.name == "no_conflict_markers")
        );
    }

    #[test]
    fn test_preflight_import_passes_valid_jsonl() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("issues.jsonl");

        // Write valid JSONL
        let issue = make_test_issue("bd-001", "Test Issue");
        let json = serde_json::to_string(&issue).unwrap();
        std::fs::write(&jsonl_path, format!("{json}\n")).unwrap();

        let config = ImportConfig {
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        let result = preflight_import(&jsonl_path, &config, None).unwrap();

        assert_eq!(result.overall_status, PreflightCheckStatus::Pass);
        assert!(result.failures().is_empty());
    }

    #[test]
    fn test_preflight_export_passes_with_valid_setup() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("issues.jsonl");

        let storage = SqliteStorage::open_memory().unwrap();
        let config = ExportConfig {
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        let result = preflight_export(&storage, &jsonl_path, &config).unwrap();

        assert_eq!(
            result.overall_status,
            PreflightCheckStatus::Pass,
            "Expected Pass, got {:?}. Failures: {:?}",
            result.overall_status,
            result.failures()
        );
        assert!(result.failures().is_empty());
    }

    // ========================================================================
    // Preflight Guardrail Tests (beads_rust-1quj)
    // ========================================================================

    #[test]
    fn test_preflight_import_rejects_invalid_json_lines() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("issues.jsonl");

        // Write JSONL with invalid lines
        let issue1 = make_test_issue("bd-001", "Good issue");
        let issue2 = make_test_issue("bd-002", "Another good issue");
        let good_json_1 = serde_json::to_string(&issue1).unwrap();
        let good_json_2 = serde_json::to_string(&issue2).unwrap();
        let content = format!("{good_json_1}\nNOT VALID JSON\n{good_json_2}\n{{\"broken: true}}\n");
        std::fs::write(&jsonl_path, content).unwrap();

        let config = ImportConfig {
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        let result = preflight_import(&jsonl_path, &config, None).unwrap();

        assert_eq!(result.overall_status, PreflightCheckStatus::Fail);
        let failures = result.failures();
        let json_check = failures.iter().find(|c| c.name == "json_valid");
        assert!(json_check.is_some(), "Expected json_valid failure");
        let msg = &json_check.unwrap().message;
        assert!(msg.contains("2 invalid issue record"), "Message was: {msg}");
        assert!(msg.contains("line 2"), "Should mention line 2: {msg}");
        assert!(msg.contains("line 4"), "Should mention line 4: {msg}");
    }

    #[test]
    fn test_preflight_import_passes_valid_json_lines() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("issues.jsonl");

        let issue1 = make_test_issue("bd-001", "First");
        let issue2 = make_test_issue("bd-002", "Second");
        let content = format!(
            "{}\n\n{}\n",
            serde_json::to_string(&issue1).unwrap(),
            serde_json::to_string(&issue2).unwrap()
        );
        std::fs::write(&jsonl_path, content).unwrap();

        let config = ImportConfig {
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        let result = preflight_import(&jsonl_path, &config, None).unwrap();

        // json_valid should pass
        let json_check = result.checks.iter().find(|c| c.name == "json_valid");
        assert!(json_check.is_some());
        assert_eq!(json_check.unwrap().status, PreflightCheckStatus::Pass);
    }

    #[test]
    fn test_validate_jsonl_issue_records_rejects_duplicate_issue_ids() {
        let temp = TempDir::new().unwrap();
        let jsonl_path = temp.path().join("issues.jsonl");

        let issue = make_test_issue("bd-dup", "Duplicate");
        let issue_json = serde_json::to_string(&issue).unwrap();
        std::fs::write(&jsonl_path, format!("{issue_json}\n{issue_json}\n")).unwrap();

        let summary = validate_jsonl_issue_records(&jsonl_path).unwrap();

        assert_eq!(summary.record_count, 2);
        assert_eq!(summary.invalid_count, 1);
        let preview = summary.preview_messages();
        assert!(
            preview
                .iter()
                .any(|message| message.contains("line 2: Duplicate issue id 'bd-dup'")),
            "expected duplicate-id validation failure in preview, got {preview:?}"
        );
    }

    #[test]
    fn test_preflight_import_rejects_duplicate_issue_ids_during_validation() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("issues.jsonl");

        let issue = make_test_issue("bd-dup", "Duplicate");
        let issue_json = serde_json::to_string(&issue).unwrap();
        std::fs::write(&jsonl_path, format!("{issue_json}\n{issue_json}\n")).unwrap();

        let config = ImportConfig {
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        let result = preflight_import(&jsonl_path, &config, None).unwrap();

        assert_eq!(result.overall_status, PreflightCheckStatus::Fail);
        let failures = result.failures();
        let json_check = failures
            .iter()
            .find(|c| c.name == "json_valid")
            .expect("expected json_valid failure");
        assert!(
            json_check.message.contains("Duplicate issue id 'bd-dup'"),
            "expected duplicate-id validation message, got {}",
            json_check.message
        );
    }

    #[test]
    fn test_preflight_import_rejects_semantically_invalid_issue_records() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("issues.jsonl");

        let mut invalid_issue = make_test_issue("bd-001", "");
        invalid_issue.title.clear();
        std::fs::write(
            &jsonl_path,
            format!("{}\n", serde_json::to_string(&invalid_issue).unwrap()),
        )
        .unwrap();

        let config = ImportConfig {
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        let result = preflight_import(&jsonl_path, &config, None).unwrap();

        assert_eq!(result.overall_status, PreflightCheckStatus::Fail);
        let failures = result.failures();
        let json_check = failures.iter().find(|c| c.name == "json_valid");
        assert!(json_check.is_some(), "Expected json_valid failure");
        assert!(
            json_check
                .expect("json_valid failure")
                .message
                .contains("title"),
            "Expected validation failure to mention the empty title"
        );
    }

    #[test]
    fn test_preflight_import_rejects_prefix_mismatch() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("issues.jsonl");

        // Write issues with wrong prefix
        let issue1 = make_test_issue("xx-001", "Wrong prefix 1");
        let issue2 = make_test_issue("xx-002", "Wrong prefix 2");
        let content = format!(
            "{}\n{}\n",
            serde_json::to_string(&issue1).unwrap(),
            serde_json::to_string(&issue2).unwrap()
        );
        std::fs::write(&jsonl_path, content).unwrap();

        let config = ImportConfig {
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        let result = preflight_import(&jsonl_path, &config, Some("bd")).unwrap();

        assert_eq!(result.overall_status, PreflightCheckStatus::Fail);
        let failures = result.failures();
        let prefix_check = failures.iter().find(|c| c.name == "prefix_match");
        assert!(prefix_check.is_some(), "Expected prefix_match failure");
        let msg = &prefix_check.unwrap().message;
        assert!(msg.contains("xx-001"), "Should list mismatched ID: {msg}");
        assert!(msg.contains("xx-002"), "Should list mismatched ID: {msg}");
        assert!(msg.contains("2 mismatched"), "Should show count: {msg}");
    }

    #[test]
    fn test_preflight_import_rejects_shared_prefix_superset() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("issues.jsonl");

        let issue = make_test_issue("bdx-001", "Wrong shared prefix");
        let json = serde_json::to_string(&issue).unwrap();
        std::fs::write(&jsonl_path, format!("{json}\n")).unwrap();

        let config = ImportConfig {
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        let result = preflight_import(&jsonl_path, &config, Some("bd")).unwrap();
        assert_eq!(result.overall_status, PreflightCheckStatus::Fail);
        let failures = result.failures();
        let prefix_check = failures.iter().find(|c| c.name == "prefix_match");
        assert!(prefix_check.is_some(), "Expected prefix_match failure");
        assert!(
            prefix_check.unwrap().message.contains("bdx-001"),
            "Should report the mismatched ID"
        );
    }

    #[test]
    fn test_preflight_import_prefix_check_skipped_when_override() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("issues.jsonl");

        // Write issues with wrong prefix
        let issue = make_test_issue("xx-001", "Wrong prefix");
        let json = serde_json::to_string(&issue).unwrap();
        std::fs::write(&jsonl_path, format!("{json}\n")).unwrap();

        let config = ImportConfig {
            skip_prefix_validation: true,
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        let result = preflight_import(&jsonl_path, &config, Some("bd")).unwrap();

        // prefix_match check should NOT be present when skip_prefix_validation is true
        let prefix_check = result.checks.iter().find(|c| c.name == "prefix_match");
        assert!(
            prefix_check.is_none(),
            "prefix_match check should be skipped when skip_prefix_validation is true"
        );
        // Overall should pass (or at least not fail on prefix)
        assert!(
            result.failures().iter().all(|c| c.name != "prefix_match"),
            "No prefix_match failures expected with override"
        );
    }

    #[test]
    fn test_preflight_import_prefix_passes_matching_prefix() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("issues.jsonl");

        let issue1 = make_test_issue("bd-001", "Correct prefix 1");
        let issue2 = make_test_issue("bd-002", "Correct prefix 2");
        let content = format!(
            "{}\n{}\n",
            serde_json::to_string(&issue1).unwrap(),
            serde_json::to_string(&issue2).unwrap()
        );
        std::fs::write(&jsonl_path, content).unwrap();

        let config = ImportConfig {
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        let result = preflight_import(&jsonl_path, &config, Some("bd")).unwrap();

        let prefix_check = result.checks.iter().find(|c| c.name == "prefix_match");
        assert!(
            prefix_check.is_some(),
            "prefix_match check should be present"
        );
        assert_eq!(
            prefix_check.unwrap().status,
            PreflightCheckStatus::Pass,
            "prefix_match should pass for matching prefix"
        );
    }

    #[test]
    fn test_preflight_import_prefix_no_check_without_expected() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("issues.jsonl");

        let issue = make_test_issue("xx-001", "Any prefix");
        let json = serde_json::to_string(&issue).unwrap();
        std::fs::write(&jsonl_path, format!("{json}\n")).unwrap();

        let config = ImportConfig {
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        // No expected_prefix passed — prefix check should not be added
        let result = preflight_import(&jsonl_path, &config, None).unwrap();

        let prefix_check = result.checks.iter().find(|c| c.name == "prefix_match");
        assert!(
            prefix_check.is_none(),
            "prefix_match check should not run without expected_prefix"
        );
    }

    #[test]
    fn test_preflight_import_conflict_markers_mixed_content() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("issues.jsonl");

        // Valid JSONL with embedded conflict markers
        let issue = make_test_issue("bd-001", "Good issue");
        let good_json = serde_json::to_string(&issue).unwrap();
        let content = format!(
            "{good_json}\n<<<<<<< HEAD\n{good_json}\n=======\n{good_json}\n>>>>>>> other\n"
        );
        std::fs::write(&jsonl_path, content).unwrap();

        let config = ImportConfig {
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        let result = preflight_import(&jsonl_path, &config, None).unwrap();

        assert_eq!(result.overall_status, PreflightCheckStatus::Fail);
        // Should have both conflict marker AND json validation failures
        assert!(
            result
                .failures()
                .iter()
                .any(|c| c.name == "no_conflict_markers"),
            "Should detect conflict markers"
        );
        assert!(
            result.failures().iter().any(|c| c.name == "json_valid"),
            "Conflict marker lines should also fail JSON validation"
        );
    }

    #[test]
    fn test_preflight_import_success_path_all_checks() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("issues.jsonl");

        // Write valid JSONL with correct prefix
        let issue1 = make_test_issue("bd-001", "Issue One");
        let issue2 = make_test_issue("bd-002", "Issue Two");
        let content = format!(
            "{}\n{}\n",
            serde_json::to_string(&issue1).unwrap(),
            serde_json::to_string(&issue2).unwrap()
        );
        std::fs::write(&jsonl_path, content).unwrap();

        let config = ImportConfig {
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        let result = preflight_import(&jsonl_path, &config, Some("bd")).unwrap();

        // All checks should pass
        assert_eq!(
            result.overall_status,
            PreflightCheckStatus::Pass,
            "All checks should pass. Failures: {:?}",
            result
                .failures()
                .iter()
                .map(|c| format!("{}: {}", c.name, c.message))
                .collect::<Vec<_>>()
        );
        assert!(result.failures().is_empty());

        // Verify all expected checks ran
        let check_names: Vec<&str> = result.checks.iter().map(|c| c.name.as_str()).collect();
        assert!(
            check_names.contains(&"beads_dir_exists"),
            "Should check beads dir: {check_names:?}"
        );
        assert!(
            check_names.contains(&"file_readable"),
            "Should check file readable: {check_names:?}"
        );
        assert!(
            check_names.contains(&"no_conflict_markers"),
            "Should check conflict markers: {check_names:?}"
        );
        assert!(
            check_names.contains(&"json_valid"),
            "Should check JSON validity: {check_names:?}"
        );
        assert!(
            check_names.contains(&"prefix_match"),
            "Should check prefix match: {check_names:?}"
        );
    }

    #[test]
    fn test_preflight_import_mixed_prefix_partial_mismatch() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("issues.jsonl");

        // Mix of correct and incorrect prefix
        let good_issue = make_test_issue("bd-001", "Good prefix");
        let bad_issue = make_test_issue("xx-002", "Bad prefix");
        let content = format!(
            "{}\n{}\n",
            serde_json::to_string(&good_issue).unwrap(),
            serde_json::to_string(&bad_issue).unwrap()
        );
        std::fs::write(&jsonl_path, content).unwrap();

        let config = ImportConfig {
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        let result = preflight_import(&jsonl_path, &config, Some("bd")).unwrap();

        assert_eq!(result.overall_status, PreflightCheckStatus::Fail);
        let failures = result.failures();
        let prefix_check = failures.iter().find(|c| c.name == "prefix_match");
        assert!(prefix_check.is_some());
        let msg = &prefix_check.unwrap().message;
        assert!(
            msg.contains("1 mismatched"),
            "Should show count of 1: {msg}"
        );
        assert!(msg.contains("xx-002"), "Should list the bad ID: {msg}");
    }

    #[test]
    fn test_preflight_import_prefix_skips_tombstones() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("issues.jsonl");

        // Create a tombstone with wrong prefix — should be silently ignored
        let mut tombstone = make_test_issue("xx-001", "Foreign tombstone");
        tombstone.status = Status::Tombstone;
        let good_issue = make_test_issue("bd-001", "Good issue");
        let content = format!(
            "{}\n{}\n",
            serde_json::to_string(&tombstone).unwrap(),
            serde_json::to_string(&good_issue).unwrap()
        );
        std::fs::write(&jsonl_path, content).unwrap();

        let config = ImportConfig {
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        let result = preflight_import(&jsonl_path, &config, Some("bd")).unwrap();

        // Tombstone with wrong prefix should not cause failure
        let prefix_check = result.checks.iter().find(|c| c.name == "prefix_match");
        assert!(prefix_check.is_some());
        assert_eq!(
            prefix_check.unwrap().status,
            PreflightCheckStatus::Pass,
            "Tombstones with wrong prefix should be ignored"
        );
    }

    #[test]
    fn test_preflight_import_empty_file_passes_json_check() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("issues.jsonl");

        // Empty file
        std::fs::write(&jsonl_path, "").unwrap();

        let config = ImportConfig {
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        let result = preflight_import(&jsonl_path, &config, None).unwrap();

        // An empty file should pass JSON validation (no invalid lines)
        let json_check = result.checks.iter().find(|c| c.name == "json_valid");
        assert!(json_check.is_some());
        assert_eq!(json_check.unwrap().status, PreflightCheckStatus::Pass);
    }

    #[test]
    fn test_preflight_import_only_blank_lines_passes_json_check() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        std::fs::create_dir_all(&beads_dir).unwrap();
        let jsonl_path = beads_dir.join("issues.jsonl");

        // Only whitespace/blank lines
        std::fs::write(&jsonl_path, "\n\n  \n\t\n").unwrap();

        let config = ImportConfig {
            beads_dir: Some(beads_dir),
            ..Default::default()
        };

        let result = preflight_import(&jsonl_path, &config, None).unwrap();

        let json_check = result.checks.iter().find(|c| c.name == "json_valid");
        assert!(json_check.is_some());
        assert_eq!(json_check.unwrap().status, PreflightCheckStatus::Pass);
    }

    // ========================================================================
    // 3-Way Merge Tests
    // ========================================================================

    fn fixed_time_merge(seconds: i64) -> chrono::DateTime<Utc> {
        chrono::DateTime::from_timestamp(seconds, 0).unwrap()
    }

    fn make_issue_with_hash(
        id: &str,
        title: &str,
        updated_at: chrono::DateTime<Utc>,
        hash: Option<&str>,
    ) -> Issue {
        let created_at = updated_at - chrono::Duration::seconds(60);
        Issue {
            id: id.to_string(),
            content_hash: hash.map(str::to_string),
            title: title.to_string(),
            description: None,
            design: None,
            acceptance_criteria: None,
            notes: None,
            status: Status::Open,
            priority: Priority::MEDIUM,
            issue_type: IssueType::Task,
            assignee: None,
            owner: None,
            estimated_minutes: None,
            created_at,
            created_by: None,
            updated_at,
            closed_at: None,
            close_reason: None,
            closed_by_session: None,
            due_at: None,
            defer_until: None,
            external_ref: None,
            source_system: None,
            source_repo: None,
            deleted_at: None,
            deleted_by: None,
            delete_reason: None,
            original_type: None,
            compaction_level: None,
            compacted_at: None,
            compacted_at_commit: None,
            original_size: None,
            sender: None,
            ephemeral: false,
            pinned: false,
            is_template: false,
            labels: vec![],
            dependencies: vec![],
            comments: vec![],
        }
    }

    #[test]
    fn test_merge_new_local_issue_kept() {
        // Issue only in left (new local) should be kept
        let local = make_issue_with_hash("bd-1", "New Local", fixed_time_merge(100), Some("hash1"));
        let result = merge_issue(None, Some(&local), None, ConflictResolution::PreferNewer);
        assert!(matches!(result, MergeResult::Keep(issue) if issue.id == "bd-1"));
    }

    #[test]
    fn test_merge_new_external_issue_kept() {
        // Issue only in right (new external) should be kept
        let external =
            make_issue_with_hash("bd-2", "New External", fixed_time_merge(100), Some("hash2"));
        let result = merge_issue(None, None, Some(&external), ConflictResolution::PreferNewer);
        assert!(matches!(result, MergeResult::Keep(issue) if issue.id == "bd-2"));
    }

    #[test]
    fn test_merge_deleted_both_sides() {
        // Issue in base but deleted in both local and external -> delete
        let base = make_issue_with_hash("bd-3", "Old", fixed_time_merge(100), Some("hash3"));
        let result = merge_issue(Some(&base), None, None, ConflictResolution::PreferNewer);
        assert!(matches!(result, MergeResult::Delete));
    }

    #[test]
    fn test_merge_deleted_external_unmodified_local() {
        // Issue in base and local (unmodified), deleted in external -> delete
        let base = make_issue_with_hash("bd-4", "Base", fixed_time_merge(100), Some("hash4"));
        let result = merge_issue(
            Some(&base),
            Some(&base),
            None,
            ConflictResolution::PreferNewer,
        );
        assert!(matches!(result, MergeResult::Delete));
    }

    #[test]
    fn test_merge_deleted_external_modified_local() {
        // Issue in base and local (modified), deleted in external -> conflict (or keep local with PreferNewer)
        let base = make_issue_with_hash("bd-5", "Base", fixed_time_merge(100), Some("hash5"));
        let local =
            make_issue_with_hash("bd-5", "Modified", fixed_time_merge(200), Some("hash5_mod")); // Modified after base

        let result = merge_issue(
            Some(&base),
            Some(&local),
            None,
            ConflictResolution::PreferNewer,
        );
        assert!(matches!(result, MergeResult::KeepWithNote(..)));
    }

    #[test]
    fn test_merge_deleted_local_modified_external() {
        // Issue in base and external (modified), deleted in local -> conflict (or keep external with PreferNewer)
        let base = make_issue_with_hash("bd-006", "Base", fixed_time_merge(100), Some("hash6"));
        let external = make_issue_with_hash(
            "bd-006",
            "Modified",
            fixed_time_merge(200),
            Some("hash6_ext"),
        );

        let result = merge_issue(
            Some(&base),
            None,
            Some(&external),
            ConflictResolution::PreferNewer,
        );
        assert!(matches!(result, MergeResult::KeepWithNote(issue, _) if issue.title == "Modified"));
    }

    #[test]
    fn test_merge_only_local_modified() {
        // Issue in all three, only local modified -> keep local
        let base = make_issue_with_hash("bd-007", "Base", fixed_time_merge(100), Some("hash7"));
        let local = make_issue_with_hash(
            "bd-007",
            "Modified",
            fixed_time_merge(200),
            Some("hash7_mod"),
        );
        let external = make_issue_with_hash("bd-007", "Base", fixed_time_merge(100), Some("hash7")); // Same as base

        let result = merge_issue(
            Some(&base),
            Some(&local),
            Some(&external),
            ConflictResolution::PreferNewer,
        );
        assert!(matches!(result, MergeResult::Keep(issue) if issue.title == "Modified"));
    }

    #[test]
    fn test_merge_only_external_modified() {
        // Issue in all three, only external modified -> keep external
        let base = make_issue_with_hash("bd-008", "Base", fixed_time_merge(100), Some("hash8"));
        let local = make_issue_with_hash("bd-008", "Base", fixed_time_merge(100), Some("hash8")); // Same as base
        let external = make_issue_with_hash(
            "bd-008",
            "Modified",
            fixed_time_merge(200),
            Some("hash8_ext"),
        );

        let result = merge_issue(
            Some(&base),
            Some(&local),
            Some(&external),
            ConflictResolution::PreferNewer,
        );
        assert!(matches!(result, MergeResult::Keep(issue) if issue.title == "Modified"));
    }

    #[test]
    fn test_merge_both_modified_prefer_newer() {
        // Issue in all three, both modified -> keep newer
        let base = make_issue_with_hash("bd-009", "Base", fixed_time_merge(100), Some("hash9"));
        let local = make_issue_with_hash(
            "bd-009",
            "Local Mod",
            fixed_time_merge(200),
            Some("hash9_local"),
        );
        let external = make_issue_with_hash(
            "bd-009",
            "External Mod",
            fixed_time_merge(300),
            Some("hash9_ext"),
        );

        let result = merge_issue(
            Some(&base),
            Some(&local),
            Some(&external),
            ConflictResolution::PreferNewer,
        );
        assert!(
            matches!(result, MergeResult::KeepWithNote(issue, _) if issue.title == "External Mod")
        );
    }

    #[test]
    fn test_merge_both_modified_prefer_local() {
        let base = make_issue_with_hash("bd-010", "Base", fixed_time_merge(100), Some("hash10"));
        let local = make_issue_with_hash(
            "bd-010",
            "Local Mod",
            fixed_time_merge(200),
            Some("hash10_local"),
        );
        let external = make_issue_with_hash(
            "bd-010",
            "External Mod",
            fixed_time_merge(300),
            Some("hash10_ext"),
        );

        let result = merge_issue(
            Some(&base),
            Some(&local),
            Some(&external),
            ConflictResolution::PreferLocal,
        );
        assert!(
            matches!(result, MergeResult::KeepWithNote(issue, _) if issue.title == "Local Mod")
        );
    }

    #[test]
    fn test_merge_convergent_creation_same_content() {
        // Both created independently with same content hash -> keep one
        let local = make_issue_with_hash("bd-011", "Same", fixed_time_merge(100), Some("hash11"));
        let external =
            make_issue_with_hash("bd-011", "Same", fixed_time_merge(100), Some("hash11"));

        let result = merge_issue(
            None,
            Some(&local),
            Some(&external),
            ConflictResolution::PreferNewer,
        );
        assert!(matches!(result, MergeResult::Keep(..)));
    }

    #[test]
    fn test_merge_convergent_creation_different_content() {
        // Both created independently with different content -> keep newer
        let local = make_issue_with_hash(
            "bd-012",
            "Local",
            fixed_time_merge(100),
            Some("hash12_local"),
        );
        let external = make_issue_with_hash(
            "bd-012",
            "External",
            fixed_time_merge(200),
            Some("hash12_ext"),
        );

        let result = merge_issue(
            None,
            Some(&local),
            Some(&external),
            ConflictResolution::PreferNewer,
        );
        assert!(matches!(result, MergeResult::KeepWithNote(issue, _) if issue.title == "External"));
    }

    #[test]
    fn test_merge_neither_changed() {
        // Issue in all three, neither changed -> keep (use left by convention)
        let base = make_issue_with_hash("bd-013", "Same", fixed_time_merge(100), Some("hash13"));
        let local = make_issue_with_hash("bd-013", "Same", fixed_time_merge(100), Some("hash13"));
        let external =
            make_issue_with_hash("bd-013", "Same", fixed_time_merge(100), Some("hash13"));

        let result = merge_issue(
            Some(&base),
            Some(&local),
            Some(&external),
            ConflictResolution::PreferNewer,
        );
        assert!(matches!(result, MergeResult::Keep(issue) if issue.id == "bd-013"));
    }

    #[test]
    fn test_merge_report_has_conflicts() {
        let mut report = MergeReport::default();
        assert!(!report.has_conflicts());

        report
            .conflicts
            .push(("bd-001".to_string(), ConflictType::DeleteVsModify));
        assert!(report.has_conflicts());
    }

    #[test]
    fn test_merge_report_total_actions() {
        let mut report = MergeReport::default();
        assert_eq!(report.total_actions(), 0);

        report.kept.push(make_test_issue("bd-001", "Kept"));
        report.kept.push(make_test_issue("bd-002", "Kept"));
        report.deleted.push("bd-003".to_string());
        assert_eq!(report.total_actions(), 3);
    }

    // ========================================================================
    // three_way_merge orchestration tests
    // ========================================================================

    #[test]
    fn test_three_way_merge_basic() {
        // Setup: one issue in each state
        let base_issue =
            make_issue_with_hash("bd-001", "Base", fixed_time_merge(100), Some("hash1"));
        let local_issue =
            make_issue_with_hash("bd-002", "Local Only", fixed_time_merge(200), Some("hash2"));
        let external_issue = make_issue_with_hash(
            "bd-003",
            "External Only",
            fixed_time_merge(300),
            Some("hash3"),
        );

        let mut base = std::collections::HashMap::new();
        base.insert("bd-001".to_string(), base_issue.clone());

        let mut left = std::collections::HashMap::new();
        left.insert("bd-001".to_string(), base_issue.clone());
        left.insert("bd-002".to_string(), local_issue);

        let mut right = std::collections::HashMap::new();
        right.insert("bd-001".to_string(), base_issue);
        right.insert("bd-003".to_string(), external_issue);

        let context = MergeContext::new(base, left, right);
        let report = three_way_merge(&context, ConflictResolution::PreferNewer, None);

        // Should keep bd-001 (in all three), bd-002 (local only), bd-003 (external only)
        assert_eq!(report.kept.len(), 3);
        assert!(report.conflicts.is_empty());
        assert!(report.deleted.is_empty());
    }

    #[test]
    fn test_three_way_merge_with_tombstone_protection() {
        // Setup: tombstoned issue trying to resurrect from external
        let external_issue = make_issue_with_hash(
            "bd-tomb",
            "Should Not Resurrect",
            fixed_time_merge(300),
            Some("hash1"),
        );

        let base = std::collections::HashMap::new();
        let left = std::collections::HashMap::new();
        let mut right = std::collections::HashMap::new();
        right.insert("bd-tomb".to_string(), external_issue);

        let context = MergeContext::new(base, left, right);

        // Create tombstones set
        let mut tombstones = std::collections::HashSet::new();
        tombstones.insert("bd-tomb".to_string());

        let report = three_way_merge(&context, ConflictResolution::PreferNewer, Some(&tombstones));

        // Should NOT keep the tombstoned issue
        assert!(report.kept.is_empty());
        assert_eq!(report.tombstone_protected.len(), 1);
        assert!(report.tombstone_protected.contains(&"bd-tomb".to_string()));
    }

    #[test]
    fn test_three_way_merge_tombstone_allows_local() {
        // Setup: tombstoned issue exists in local - should be allowed
        let local_issue = make_issue_with_hash(
            "bd-tomb",
            "Local Tombstoned",
            fixed_time_merge(200),
            Some("hash1"),
        );

        let base = std::collections::HashMap::new();
        let mut left = std::collections::HashMap::new();
        left.insert("bd-tomb".to_string(), local_issue);
        let right = std::collections::HashMap::new();

        let context = MergeContext::new(base, left, right);
        let mut tombstones = std::collections::HashSet::new();
        tombstones.insert("bd-tomb".to_string());

        let report = three_way_merge(&context, ConflictResolution::PreferNewer, Some(&tombstones));

        // Should keep local even if tombstoned
        assert_eq!(report.kept.len(), 1);
        assert!(report.tombstone_protected.is_empty());
    }

    #[test]
    fn test_three_way_merge_deletions() {
        // Setup: issue in base but deleted in both left and right
        let base_issue =
            make_issue_with_hash("bd-del", "To Delete", fixed_time_merge(100), Some("hash1"));

        let mut base = std::collections::HashMap::new();
        base.insert("bd-del".to_string(), base_issue);

        let left = std::collections::HashMap::new();
        let right = std::collections::HashMap::new();

        let context = MergeContext::new(base, left, right);
        let report = three_way_merge(&context, ConflictResolution::PreferNewer, None);

        assert!(report.kept.is_empty());
        assert_eq!(report.deleted.len(), 1);
        assert!(report.deleted.contains(&"bd-del".to_string()));
    }

    #[test]
    fn test_three_way_merge_empty_context() {
        let context = MergeContext::default();
        let report = three_way_merge(&context, ConflictResolution::PreferNewer, None);

        assert!(report.kept.is_empty());
        assert!(report.deleted.is_empty());
        assert!(report.conflicts.is_empty());
        assert!(report.tombstone_protected.is_empty());
        assert!(report.notes.is_empty());
        assert_eq!(report.total_actions(), 0);
    }

    #[test]
    fn test_merge_conflict_manual_strategy() {
        // Setup: issue deleted externally but modified locally with Manual strategy
        let base_issue =
            make_issue_with_hash("bd-001", "Base", fixed_time_merge(100), Some("base_hash"));
        let local_issue = make_issue_with_hash(
            "bd-001",
            "Modified",
            fixed_time_merge(200),
            Some("mod_hash"),
        );

        let mut base = std::collections::HashMap::new();
        base.insert("bd-001".to_string(), base_issue);
        let mut left = std::collections::HashMap::new();
        left.insert("bd-001".to_string(), local_issue);
        let right = std::collections::HashMap::new();

        let context = MergeContext::new(base, left, right);
        let report = three_way_merge(&context, ConflictResolution::Manual, None);

        // With Manual strategy, delete-vs-modify should be a conflict
        assert_eq!(report.conflicts.len(), 1);
        assert!(matches!(
            report.conflicts[0].1,
            ConflictType::DeleteVsModify
        ));
    }

    #[test]
    fn test_three_way_merge_with_notes() {
        // Setup: issue modified in both left and right
        let base_issue = make_issue_with_hash(
            "bd-001",
            "Base Title",
            fixed_time_merge(100),
            Some("base_hash"),
        );
        let local_issue = make_issue_with_hash(
            "bd-001",
            "Local Modified",
            fixed_time_merge(200),
            Some("mod_hash"),
        );
        let external_issue = make_issue_with_hash(
            "bd-001",
            "External Modified",
            fixed_time_merge(300),
            Some("external_hash"),
        );

        let mut base = std::collections::HashMap::new();
        base.insert("bd-001".to_string(), base_issue);
        let mut left = std::collections::HashMap::new();
        left.insert("bd-001".to_string(), local_issue);
        let mut right = std::collections::HashMap::new();
        right.insert("bd-001".to_string(), external_issue);

        let context = MergeContext::new(base, left, right);
        let report = three_way_merge(&context, ConflictResolution::PreferNewer, None);

        // Should have a note about the merge decision
        assert_eq!(report.kept.len(), 1);
        assert_eq!(report.notes.len(), 1);
        assert!(report.notes[0].1.contains("Both modified"));
    }

    /// Create a progress bar if enabled.
    #[allow(dead_code)]
    fn progress_bar(show: bool, len: u64, message: &str) -> ProgressBar {
        if !show {
            return ProgressBar::hidden();
        }
        let pb = ProgressBar::new(len);
        pb.set_style(
            ProgressStyle::default_bar()
                .template(
                    "{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {pos}/{len} {msg}",
                )
                .unwrap()
                .progress_chars("#>-"),
        );
        pb.set_message(message.to_string());
        pb
    }

    /// Create a progress spinner if enabled.
    #[allow(dead_code)]
    fn progress_spinner(show: bool, message: &str) -> ProgressBar {
        if !show {
            return ProgressBar::hidden();
        }
        let pb = ProgressBar::new_spinner();
        pb.set_style(
            ProgressStyle::default_spinner()
                .tick_chars("/|\\\\- ")
                .template("{spinner:.blue} {msg}")
                .unwrap(),
        );
        pb.set_message(message.to_string());
        pb.enable_steady_tick(std::time::Duration::from_millis(100));
        pb
    }

    #[test]
    fn test_compute_jsonl_hash_ignores_empty_lines_and_whitespace() {
        let temp_dir = TempDir::new().unwrap();
        let path1 = temp_dir.path().join("file1.jsonl");
        let path2 = temp_dir.path().join("file2.jsonl");

        let content1 = "{\"id\":\"bd-1\"}\n{\"id\":\"bd-2\"}\n";
        // content2 has extra empty lines, different line endings, and extra whitespace
        let content2 = "\n{\"id\":\"bd-1\"}\r\n  \n{\"id\":\"bd-2\"}  \n\n";

        fs::write(&path1, content1).unwrap();
        fs::write(&path2, content2).unwrap();

        let hash1 = compute_jsonl_hash(&path1).unwrap();
        let hash2 = compute_jsonl_hash(&path2).unwrap();

        assert_eq!(
            hash1, hash2,
            "Hashes should be identical regardless of empty lines or whitespace"
        );
    }
}