beads_rust 0.7.0

//! Sync command implementation.
//!
//! Provides explicit JSONL sync actions without git operations.
//! Supports `--flush-only` (export) and `--import-only` (import).

use crate::cli::{DEFAULT_WITNESS_PARALLELISM, SyncArgs};
use crate::config;
use crate::error::{BeadsError, Result};
use crate::format::sanitize_terminal_inline;
use crate::health::{AnomalyClass, ReliabilityAuditRecord, WorkspaceClassification};
use crate::output::{OutputContext, record_pending_exit_code};
use crate::sync::history::HistoryConfig;
use crate::sync::witness::{
    JsonlMerkleWitness, JsonlWitnessComparison, JsonlWitnessParallelWorkPlan,
    JsonlWitnessReuseMaterialization, JsonlWitnessReusePlan, build_jsonl_merkle_witness_parallel,
    compare_jsonl_merkle_witnesses, materialize_jsonl_witness_reuse_plan,
    plan_jsonl_witness_parallel_work, plan_jsonl_witness_reuse,
};
use crate::sync::{
    AdditiveReconcileReceipt, AdditiveReconcileStatus, ConflictResolution, ExpectedJsonlSourceRef,
    ExpectedStagedExport, ExportConfig, ExportEntityType, ExportError, ExportErrorPolicy,
    ImportConfig, ImportResult, JsonlSalvageReceipt, JsonlSourceSnapshot, JsonlSourceStateWitness,
    METADATA_JSONL_CONTENT_HASH, METADATA_LAST_EXPORT_TIME, METADATA_LAST_IMPORT_TIME,
    MergeContext, OrphanMode, ReconcileActionKind, ReconcileApplyOutcome, ReconcilePlan,
    ReviewedAdditiveReconcilePlanRequest, ReviewedAdditiveReconcileRequest,
    SYNC_RECONCILE_SCHEMA_VERSION, SyncMergeIntent, SyncMergeNoteWitness, SyncMergePendingPhase,
    SyncMergePendingReceipt, analyze_jsonl_snapshot,
    apply_reviewed_additive_reconcile_under_authority, apply_sync_reconcile,
    canonical_source_repo_path, canonical_sync_path_sha256, capture_sync_database_witness,
    compute_jsonl_snapshot_content_hash, compute_staleness, database_write_authority_sha256,
    ensure_no_conflict_markers_snapshot, export_temp_path,
    export_to_jsonl_with_policy_expected_under_authorities,
    export_to_jsonl_with_policy_expected_under_authority, finalize_export_under_authority,
    get_issue_ids_from_jsonl_snapshot, id_matches_expected_prefix, import_from_jsonl_snapshot,
    load_base_snapshot_from_source, plan_reviewed_additive_reconcile, plan_sync_reconcile,
    read_issues_from_jsonl_snapshot, refresh_base_snapshot_from_flushed_jsonl_snapshot,
    refresh_base_snapshot_from_flushed_jsonl_snapshot_under_authority,
    require_safe_sync_overwrite_path, require_valid_sync_path, restore_tombstones_after_rebuild,
    salvage_invalid_jsonl_records_under_authority, scan_jsonl_snapshot_for_tombstone_filter,
    snapshot_tombstones, three_way_merge, tombstones_missing_from_jsonl_tombstones,
    validate_no_git_path, validate_sync_path_with_external,
};
use crate::util::id::split_prefix_remainder;
use rich_rust::prelude::*;
use serde::Serialize;
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use std::fs::{self, File, OpenOptions};
use std::io::{BufReader, IsTerminal};
use std::path::{Component, Path, PathBuf};
use std::sync::Arc;
use tracing::{debug, info, warn};

fn human_witness_value_digest(value: Option<&str>) -> (usize, String) {
    match value {
        Some(value) => (
            value.len(),
            crate::util::hex_encode(&Sha256::digest(value.as_bytes())),
        ),
        None => (0, String::from("none")),
    }
}

#[allow(clippy::too_many_lines)]
fn additive_conflict_human_lines(
    receipt: &AdditiveReconcileReceipt,
    witness_limit: usize,
) -> Vec<String> {
    if receipt.conflict_issue_ids.is_empty() {
        return Vec::new();
    }

    let shown_conflict_ids = receipt
        .conflict_issue_ids
        .iter()
        .take(witness_limit)
        .map(|issue_id| sanitize_terminal_inline(issue_id).into_owned())
        .collect::<Vec<_>>();
    let conflict_ids_truncated = receipt.conflict_issue_ids.len() > witness_limit;
    let mut lines = vec![
        format!(
            "conflict issue IDs: total={} shown={} manifest_sha256={} truncated={}: {}",
            receipt.conflict_issue_ids.len(),
            shown_conflict_ids.len(),
            receipt.conflict_issue_ids_sha256,
            conflict_ids_truncated,
            shown_conflict_ids.join(", "),
        ),
        format!(
            "conflict reasons: {}",
            receipt
                .conflict_reasons
                .iter()
                .map(|(reason, count)| format!("{reason}={count}"))
                .collect::<Vec<_>>()
                .join(", ")
        ),
    ];
    for witness in receipt.conflict_witnesses.iter().take(witness_limit) {
        lines.push(format!(
            "conflict witness: issue={} reasons={}",
            sanitize_terminal_inline(&witness.issue_id),
            witness.reasons.join(","),
        ));
    }
    let total_conflict_details = receipt
        .conflict_witnesses
        .iter()
        .map(|witness| witness.details.len())
        .sum::<usize>();
    for (issue_id, detail) in receipt
        .conflict_witnesses
        .iter()
        .flat_map(|witness| {
            witness
                .details
                .iter()
                .map(move |detail| (witness.issue_id.as_str(), detail))
        })
        .take(witness_limit)
    {
        lines.push(format!(
            "conflict detail: issue={} reason={} kind={} ordinal={:?} validation_subcodes={} related_value_sha256={} value_sha256={} detail_sha256={}",
            sanitize_terminal_inline(issue_id),
            detail.reason,
            detail.detail_kind,
            detail.ordinal,
            detail.validation_subcodes.join(","),
            detail.related_value_sha256.join(","),
            detail.value_sha256.as_deref().unwrap_or("none"),
            detail.detail_sha256,
        ));
    }
    for witness in receipt.conflict_scalar_diffs.iter().take(witness_limit) {
        lines.push(format!(
            "conflict scalar witness: issue={} fields={} diff_sha256={} before_sha256={} after_sha256={}",
            sanitize_terminal_inline(&witness.issue_id),
            witness.changed_fields.join(","),
            witness.diff_sha256,
            witness.before_payload_sha256,
            witness.after_payload_sha256,
        ));
    }
    for witness in receipt.conflict_relation_diffs.iter().take(witness_limit) {
        lines.push(format!(
            "conflict relation witness: issue={} classes={} before_counts={:?} after_counts={:?} added={} removed={} diff_sha256={} before_sha256={} after_sha256={}",
            sanitize_terminal_inline(&witness.issue_id),
            witness.changed_relation_classes.join(","),
            witness.before_counts,
            witness.after_counts,
            witness.added_element_sha256.len(),
            witness.removed_element_sha256.len(),
            witness.diff_sha256,
            witness.before_payload_sha256,
            witness.after_payload_sha256,
        ));
    }
    if conflict_ids_truncated
        || receipt.conflict_witnesses.len() > witness_limit
        || total_conflict_details > witness_limit
        || receipt.conflict_scalar_diffs.len() > witness_limit
        || receipt.conflict_relation_diffs.len() > witness_limit
    {
        lines.push(format!(
            "conflict witness output: issue_ids={}/{} witnesses={}/{} details={}/{} scalar_diffs={}/{} relation_diffs={}/{} witness_manifest_sha256={} scalar_diff_manifest_sha256={} relation_diff_manifest_sha256={} truncated=true; use --robot for complete manifests",
            shown_conflict_ids.len(),
            receipt.conflict_issue_ids.len(),
            receipt.conflict_witnesses.len().min(witness_limit),
            receipt.conflict_witnesses.len(),
            total_conflict_details.min(witness_limit),
            total_conflict_details,
            receipt.conflict_scalar_diffs.len().min(witness_limit),
            receipt.conflict_scalar_diffs.len(),
            receipt.conflict_relation_diffs.len().min(witness_limit),
            receipt.conflict_relation_diffs.len(),
            receipt.conflict_witnesses_sha256,
            receipt.conflict_scalar_diffs_sha256,
            receipt.conflict_relation_diffs_sha256,
        ));
    }
    if receipt.conflict_reasons.keys().any(|reason| {
        matches!(
            reason.as_str(),
            "equal_timestamp_shared_scalar_drift"
                | "database_newer_shared_scalar_drift"
                | "source_newer_scalar_drift_requires_resolution"
        )
    }) {
        lines.push(
            "review hint: inspect each scalar diff, then re-plan with a separate --resolve-source-id <ISSUE_ID> for every source-authoritative resolution"
                .to_string(),
        );
    }
    lines
}

/// Result of a flush (export) operation.
#[derive(Debug, Serialize)]
pub struct FlushResult {
    pub exported_issues: usize,
    pub exported_dependencies: usize,
    pub exported_labels: usize,
    pub exported_comments: usize,
    pub content_hash: String,
    pub cleared_dirty: usize,
    pub policy: ExportErrorPolicy,
    pub success_rate: f64,
    pub errors: Vec<ExportError>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub manifest_path: Option<String>,
    /// Present only when the JSONL could not be published through the atomic
    /// flagged-rename protocol (#419): the filesystem (WSL2 9p/DrvFS, for
    /// one) refused it, so the staged file was installed with a
    /// witness-checked plain rename under the held write authority. Absent
    /// means the publication was atomic.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub publication_atomicity: Option<String>,
}

/// Result of an import operation.
#[derive(Debug, Serialize)]
pub struct ImportResultOutput {
    pub created: usize,
    pub updated: usize,
    pub skipped: usize,
    pub tombstone_skipped: usize,
    pub orphans_removed: usize,
    pub blocked_cache_rebuilt: bool,
    /// Byte-identical repeated comment objects removed during import. A
    /// same-ID comment with different content remains a hard error.
    #[serde(skip_serializing_if = "is_zero_usize")]
    pub exact_duplicate_comments_deduplicated: usize,
    /// Old-id -> new-id receipt emitted when `--rename-prefix` rewrote ids.
    /// Omitted entirely when no rename happened (legacy JSON shape).
    #[serde(skip_serializing_if = "Vec::is_empty")]
    pub prefix_renames: Vec<crate::sync::ImportPrefixRename>,
    /// Present when --skip-invalid-records published a recovered JSONL source.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub salvage: Option<JsonlSalvageReceipt>,
}

// Serde's `skip_serializing_if` callback receives `&T`, including for Copy
// scalar fields, so this signature cannot take `usize` by value.
#[allow(clippy::trivially_copy_pass_by_ref)]
const fn is_zero_usize(value: &usize) -> bool {
    *value == 0
}

const SOURCE_REPO_PATH_MIGRATION_SCHEMA: &str = "br.sync.source-repo-path-migration.v1";

#[derive(Debug, Clone, Serialize, schemars::JsonSchema)]
pub struct SourceRepoPathMigrationReceipt {
    schema: &'static str,
    mode: &'static str,
    applied: bool,
    no_op: bool,
    plan_sha256: String,
    target_path: String,
    target_path_sha256: String,
    source_records: usize,
    database_records: usize,
    source_only_created: usize,
    source_newer_updated: usize,
    database_newer_preserved: usize,
    equal_records: usize,
    tombstones_preserved: usize,
    ephemeral_source_records_skipped: usize,
    paths_normalized: usize,
    changed_issue_ids: Vec<String>,
    jsonl_rewrite_required: bool,
    source_repo_preserved: bool,
    vcs_status: &'static str,
    #[serde(skip_serializing_if = "Vec::is_empty")]
    warnings: Vec<crate::close_policy::WorkflowCapacityWarning>,
    #[serde(skip_serializing_if = "Option::is_none")]
    receipt_id: Option<String>,
}

struct SourceRepoPathMigrationPlan {
    receipt: SourceRepoPathMigrationReceipt,
    changed_kept: Vec<crate::model::Issue>,
    database_before: crate::sync::AdditiveDatabaseWitness,
    source_before: JsonlSourceStateWitness,
    source_before_content_sha256: Option<String>,
}

#[derive(Serialize)]
struct SourceRepoPathMigrationPlanDigest<'a> {
    schema: &'static str,
    target_path: &'a str,
    source_before: &'a JsonlSourceStateWitness,
    source_before_content_sha256: Option<&'a str>,
    database_before: &'a crate::sync::AdditiveDatabaseWitness,
    changed_issue_witnesses: &'a [crate::sync::SyncMergeKeptIssueWitness],
    source_records: usize,
    source_only_created: usize,
    source_newer_updated: usize,
    database_newer_preserved: usize,
    equal_records: usize,
    tombstones_preserved: usize,
    ephemeral_source_records_skipped: usize,
    paths_normalized: usize,
    jsonl_rewrite_required: bool,
}

/// Maximum witness rows printed per list in human-readable sync output;
/// machine modes always carry the complete manifests.
const HUMAN_WITNESS_LIMIT: usize = 32;

/// Maximum ids serialized per preview list in a reconcile receipt.
///
/// Mirrors the doctor `IdDelta` preview cap: operators want the first few
/// divergent ids to grep for; counts always reflect the true totals.
const RECONCILE_PREVIEW_LIMIT: usize = 50;

/// Versioned receipt for `br sync --reconcile` (`br.sync.reconcile.v1`).
///
/// Emitted by both `--dry-run` (plan only, `applied: false`) and apply
/// (`applied: true`, with the `apply` block present). Deletion is impossible
/// in this mode, so `plan.deleted` is a constant `0`.
#[derive(Debug, Serialize, schemars::JsonSchema)]
pub struct SyncReconcileReceipt {
    /// Receipt schema identifier (`br.sync.reconcile.v1`).
    pub schema_version: &'static str,
    /// `"dry_run"` or `"apply"`.
    pub mode: &'static str,
    /// True when the plan was applied to the database.
    pub applied: bool,
    /// Resolved JSONL path the plan was computed against.
    pub jsonl_path: String,
    /// Source (JSONL) witness the plan is bound to.
    pub source: ReconcileSourceWitness,
    /// Target (database) witness the plan is bound to.
    pub target: ReconcileTargetWitness,
    /// Row classification counts.
    pub plan: ReconcilePlanCounts,
    /// Bounded id previews for the classified sets.
    pub previews: ReconcileIdPreviews,
    /// Relation rows carried by planned create/update rows.
    pub relations: ReconcileRelationCounts,
    /// Event rows before the operation.
    pub events_before: u64,
    /// Event rows after the operation (always equals `events_before`; the
    /// apply transaction rolls back otherwise).
    pub events_after: u64,
    /// Apply-only details; absent in dry-run receipts.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub apply: Option<ReconcileApplyReceipt>,
}

/// JSONL-side witness in a reconcile receipt.
#[derive(Debug, Serialize, schemars::JsonSchema)]
pub struct ReconcileSourceWitness {
    /// Non-empty JSONL rows parsed (including ephemeral rows).
    pub record_count: usize,
    /// Ephemeral (`-wisp-`) rows excluded from planning.
    pub ephemeral_skipped: usize,
    /// Whitespace-normalized SHA-256 of the JSONL content.
    pub content_hash: String,
    /// RFC3339 mtime of the JSONL file at plan time.
    pub mtime: String,
    /// Byte size of the JSONL file at plan time.
    pub size_bytes: u64,
}

/// Database-side witness in a reconcile receipt.
#[derive(Debug, Serialize, schemars::JsonSchema)]
pub struct ReconcileTargetWitness {
    /// Total issue rows at plan time (including tombstones).
    pub db_issue_count: usize,
    /// Whether the stored `jsonl_content_hash` metadata already matched the
    /// file at plan time. True alongside nonzero `created`/`updated` counts
    /// is the false-equal state this mode repairs.
    pub stored_hash_matches_jsonl: bool,
}

/// Row classification counts in a reconcile receipt.
#[derive(Debug, Serialize, schemars::JsonSchema)]
pub struct ReconcilePlanCounts {
    /// JSONL rows with no DB counterpart (inserted on apply).
    pub created: usize,
    /// JSONL rows strictly newer than their DB counterpart (updated on apply).
    pub updated: usize,
    /// Rows skipped because timestamps are equal.
    pub skipped_equal: usize,
    /// Rows skipped because the DB copy is strictly newer.
    pub skipped_older: usize,
    /// Rows skipped by tombstone protection.
    pub skipped_tombstone: usize,
    /// Always 0: additive reconciliation cannot delete.
    pub deleted: usize,
    /// Exportable DB issues absent from the JSONL row set (never touched).
    pub db_only: usize,
}

/// Bounded id previews in a reconcile receipt.
#[derive(Debug, Serialize, schemars::JsonSchema)]
pub struct ReconcileIdPreviews {
    /// First ids (sorted) that would be / were created.
    pub created_ids: Vec<String>,
    /// First ids (sorted) that would be / were updated.
    pub updated_ids: Vec<String>,
    /// First exportable DB-only ids (sorted).
    pub db_only_ids: Vec<String>,
    /// Per-list truncation cap; counts in `plan` reflect true totals.
    pub preview_limit: usize,
}

/// Relation rows carried by planned create/update rows.
#[derive(Debug, Serialize, schemars::JsonSchema)]
pub struct ReconcileRelationCounts {
    /// Label rows.
    pub labels: usize,
    /// Dependency rows.
    pub dependencies: usize,
    /// Comment rows.
    pub comments: usize,
}

/// Apply-only block of a reconcile receipt.
#[derive(Debug, Serialize, schemars::JsonSchema)]
pub struct ReconcileApplyReceipt {
    /// Export-hash rows recorded for rows whose DB copy now matches JSONL.
    pub export_hashes_recorded: usize,
    /// Skipped rows whose DB copy still differs from JSONL (local wins that
    /// need a future flush).
    pub uncertified_local_wins: usize,
    /// Dangling dependency rows removed from just-written issues.
    pub orphan_dependencies_cleaned: usize,
    /// Blocked-cache rows after rebuild (0 when no row changed).
    pub blocked_cache_entries: usize,
    /// Child-counter rows after rebuild (0 when no row changed).
    pub child_counter_entries: usize,
    /// Whether `needs_flush` was set because local state still diverges from
    /// JSONL (db-only rows or uncertified local wins).
    pub needs_flush_set: bool,
    /// Whether import metadata (content hash + stat witness + import time)
    /// was repaired in the apply transaction.
    pub metadata_repaired: bool,
}

/// Sync status information.
#[derive(Debug, Serialize)]
pub struct SyncStatus {
    pub dirty_count: usize,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_export_time: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_import_time: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub jsonl_content_hash: Option<String>,
    pub jsonl_exists: bool,
    pub jsonl_newer: bool,
    pub db_newer: bool,
    /// Workspace health classification (`healthy` / `degraded` /
    /// `recoverable` / `unsafe`) computed from the cheap signals this
    /// command already evaluates: file-state probes plus the
    /// DB↔JSONL drift booleans above. Same write-gate vocabulary as
    /// `br doctor --json` (beads_rust#334; docs/reliability/HEALTH_CONTRACT.md).
    pub workspace_health: String,
    /// Anomaly evidence backing `workspace_health`, in the same shape
    /// doctor emits (`anomalies[].code` / `severity` / `message`).
    pub reliability_audit: ReliabilityAuditRecord,
    /// DB↔JSONL coverage probe (`beads_rust-jdmh`). Absent when the JSONL
    /// is missing or unreadable.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub coverage: Option<SyncCoverageProbe>,
    /// True when `coverage` shows the DB and JSONL hold different issue
    /// sets even though the timestamp/hash signals report "in sync" —
    /// recover with `br sync --reconcile` (lossless) or
    /// `--import-only --rebuild` (JSONL-authoritative).
    pub coverage_drift: bool,
    /// Stable VCS-observation slot for the canonical JSONL export.
    ///
    /// Sync deliberately never probes Git. The object therefore reports
    /// `reason: "not_probed"` and points to the explicit `br vcs-status`
    /// diagnostic that owns the separate, user-requested VCS capability.
    pub git_export: GitExportStatus,
}

/// VCS-observation state carried by `br sync --status`.
///
/// The legacy fields remain optional so existing machine consumers keep a
/// stable shape, but sync itself has no process or VCS authority and never
/// populates them. Users explicitly request the observation with the command
/// named in `diagnostic_command`.
#[derive(Debug, Serialize)]
pub struct GitExportStatus {
    /// Always false in sync output because VCS state was not requested.
    pub available: bool,
    /// Why the optional observation is absent.
    pub reason: &'static str,
    /// Explicit, separately hardened diagnostic for obtaining the observation.
    pub diagnostic_command: &'static str,
    /// True when the JSONL is tracked in the git index
    /// (`git ls-files --error-unmatch`).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub tracked: Option<bool>,
    /// True when the worktree copy matches the index (untracked files
    /// report `false`: their content is invisible to git entirely).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub worktree_clean: Option<bool>,
    /// True when the index matches HEAD for this path.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub index_clean: Option<bool>,
    /// Blob hash of the committed copy (`git rev-parse HEAD:<relpath>`);
    /// `None` when the file is absent from HEAD (or no commits exist).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub head_hash: Option<String>,
    /// Blob hash of the on-disk copy (`git hash-object <jsonl>`);
    /// `None` when the file is missing.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub worktree_hash: Option<String>,
}

impl GitExportStatus {
    fn not_probed() -> Self {
        Self {
            available: false,
            reason: "not_probed",
            diagnostic_command: "br vcs-status --json",
            tracked: None,
            worktree_clean: None,
            index_clean: None,
            head_hash: None,
            worktree_hash: None,
        }
    }
}

/// JSONL witness command output.
#[derive(Debug, Serialize)]
pub struct SyncWitnessResult {
    pub jsonl_path: String,
    pub witness: JsonlMerkleWitness,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub base_jsonl_path: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub base_comparison: Option<JsonlWitnessComparison>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub base_reuse_plan: Option<JsonlWitnessReusePlan>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub base_parallel_work_plan: Option<JsonlWitnessParallelWorkPlan>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub base_reuse_materialization: Option<JsonlWitnessReuseMaterialization>,
}

struct BaseWitnessArtifacts {
    jsonl_path: Option<String>,
    comparison: Option<JsonlWitnessComparison>,
    reuse_plan: Option<JsonlWitnessReusePlan>,
    parallel_work_plan: Option<JsonlWitnessParallelWorkPlan>,
    reuse_materialization: Option<JsonlWitnessReuseMaterialization>,
}

#[derive(Debug)]
#[allow(dead_code)] // Fields may be used in future sync enhancements
struct SyncPathPolicy {
    jsonl_path: PathBuf,
    jsonl_temp_path: PathBuf,
    manifest_path: PathBuf,
    beads_dir: PathBuf,
    is_external: bool,
    allow_external_jsonl: bool,
}

struct SyncStartupState {
    beads_dir: PathBuf,
    path_policy: SyncPathPolicy,
    open_result: config::OpenStorageResult,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum SyncOperation {
    Status,
    Witness,
    ReconcileAdditive,
    MigrateSourceRepoPath,
    Flush,
    Merge,
    Import,
    Reconcile,
    Unspecified,
}

struct SyncDispatchOptions {
    db_path: PathBuf,
    retention_days: Option<u64>,
    use_json: bool,
    show_progress: bool,
    history_config: HistoryConfig,
}

struct PendingSyncMergeCompletion {
    receipt: SyncMergePendingReceipt,
    base_authority: crate::sync::JsonlFamilyWriteLock,
}

struct ReconciledPendingSyncMerge {
    published_source: Arc<JsonlSourceSnapshot>,
    terminal_receipt: SyncMergePendingReceipt,
}

enum DeferredSyncOutput {
    Merge {
        report: crate::sync::MergeReport,
        resolution: String,
        capacity_warnings: Vec<crate::close_policy::WorkflowCapacityWarning>,
        use_json: bool,
    },
    ResumedMerge {
        receipt_id: String,
        phase_before: SyncMergePendingPhase,
        capacity_warnings: Vec<crate::close_policy::WorkflowCapacityWarning>,
        use_json: bool,
    },
    SourceRepoPathMigration {
        receipt: SourceRepoPathMigrationReceipt,
        use_json: bool,
    },
}

#[derive(Default)]
struct SyncDispatchCompletion {
    published_source: Option<Arc<JsonlSourceSnapshot>>,
    /// JSONL-family authority acquired by the operation when startup did not
    /// already retain one. Finalization transfers this lease into
    /// `OpenStorageResult` before adopting the published source.
    owned_jsonl_authority: Option<crate::sync::JsonlFamilyWriteLock>,
    pending_merge: Option<PendingSyncMergeCompletion>,
    deferred_output: Option<DeferredSyncOutput>,
}

impl SyncDispatchCompletion {
    fn published(source: Option<Arc<JsonlSourceSnapshot>>) -> Self {
        Self {
            published_source: source,
            ..Self::default()
        }
    }
}

/// Execute the sync command.
///
/// # Errors
///
/// Returns an error if the database cannot be opened or the sync operation fails.
pub fn execute(
    args: &SyncArgs,
    _json: bool,
    cli: &config::CliOverrides,
    ctx: &OutputContext,
    startup_write_lock_held: bool,
) -> Result<()> {
    validate_sync_mode_args(args)?;

    if args.witness {
        let (_, _, path_policy) = resolve_sync_startup_paths(args, cli)?;
        return execute_witness(&path_policy, args, ctx.is_json() || args.robot, ctx);
    }

    if args.reconcile_additive && !args.apply {
        let beads_dir = config::discover_beads_dir_with_cli(cli)?;
        let plan = plan_reviewed_additive_reconcile(&ReviewedAdditiveReconcilePlanRequest {
            beads_dir,
            db_override: cli.db.clone(),
            source_path_override: None,
            allow_external_jsonl: args.allow_external_jsonl,
            source_authoritative_ids: args.resolve_source_ids.iter().cloned().collect(),
        })?;
        render_additive_reconcile_receipt(plan.receipt(), ctx, ctx.is_json() || args.robot);
        if plan.has_conflicts() {
            record_pending_exit_code(6);
        }
        return Ok(());
    }

    if args.reconcile_additive {
        let beads_dir = config::discover_beads_dir_with_cli(cli)?;
        let expected_plan_sha256 =
            args.expect_plan_sha256
                .clone()
                .ok_or_else(|| BeadsError::Validation {
                    field: "expect_plan_sha256".to_string(),
                    reason: "--apply requires --expect-plan-sha256".to_string(),
                })?;
        let receipt = apply_reviewed_additive_reconcile_under_authority(
            &ReviewedAdditiveReconcileRequest {
                beads_dir,
                db_override: cli.db.clone(),
                source_path_override: None,
                allow_external_jsonl: args.allow_external_jsonl,
                source_authoritative_ids: args.resolve_source_ids.iter().cloned().collect(),
                expected_plan_sha256,
                lock_timeout_ms: cli.lock_timeout,
            },
            cli.held_write_authority.as_ref(),
        )?;
        render_additive_reconcile_receipt(&receipt, ctx, ctx.is_json() || args.robot);
        if matches!(
            receipt.status,
            AdditiveReconcileStatus::CommittedWithPostconditionFailures
        ) {
            record_pending_exit_code(6);
        }
        return Ok(());
    }

    let mut startup = prepare_sync_startup(args, cli, startup_write_lock_held)?;

    let command_result = maybe_delegate_rebuild(args, &mut startup.open_result)
        .and_then(|()| startup.open_result.verify_retained_jsonl_source_current())
        .and_then(|()| {
            dispatch_sync_subcommand(
                args,
                cli,
                ctx,
                &startup.beads_dir,
                &startup.path_policy,
                &mut startup.open_result,
            )
        })
        .and_then(|completion| {
            finalize_sync_dispatch_completion(completion, &mut startup.open_result, ctx)
        });

    finalize_sync_result(command_result, &mut startup.open_result)
}

/// Resolve path policy and open storage before dispatch. Keeping this separate
/// from `execute` makes the command's startup phase distinct from the
/// status/export/import/merge operation handlers below.
fn prepare_sync_startup(
    args: &SyncArgs,
    cli: &config::CliOverrides,
    _startup_write_lock_held: bool,
) -> Result<SyncStartupState> {
    let (beads_dir, startup, path_policy) = resolve_sync_startup_paths(args, cli)?;
    let allow_external_jsonl = path_policy.allow_external_jsonl;

    let open_result = config::open_storage_with_startup_config_and_jsonl_policy(
        startup,
        cli,
        should_defer_jsonl_recovery(args),
        allow_external_jsonl,
    )?;

    Ok(SyncStartupState {
        beads_dir,
        path_policy,
        open_result,
    })
}

fn resolve_sync_startup_paths(
    args: &SyncArgs,
    cli: &config::CliOverrides,
) -> Result<(PathBuf, config::StartupConfig, SyncPathPolicy)> {
    let beads_dir = config::discover_beads_dir_with_cli(cli)?;
    let startup = config::load_startup_config_with_paths(&beads_dir, cli.db.as_ref())?;
    let allow_external_jsonl = args.allow_external_jsonl
        || config::implicit_external_jsonl_allowed(
            &startup.paths.beads_dir,
            &startup.paths.db_path,
            &startup.paths.jsonl_path,
        );
    let path_policy =
        validate_sync_paths(&beads_dir, &startup.paths.jsonl_path, allow_external_jsonl)?;
    let jsonl_log_path = if path_policy.is_external {
        "<external-source>".to_string()
    } else {
        path_policy.jsonl_path.display().to_string()
    };
    let manifest_log_path = if path_policy.is_external {
        "<external-manifest>".to_string()
    } else {
        path_policy.manifest_path.display().to_string()
    };
    debug!(
        jsonl_path = jsonl_log_path,
        manifest_path = manifest_log_path,
        external_jsonl = path_policy.is_external,
        allow_external_jsonl = path_policy.allow_external_jsonl,
        "Resolved sync path policy"
    );

    Ok((beads_dir, startup, path_policy))
}

/// For `--rename-prefix` imports, defer any implicit JSONL recovery until the
/// explicit import path below so the command's import semantics (ID rewrites and
/// duplicate external_ref cleanup) are applied in the same invocation instead
/// of being skipped by open-time recovery.
fn should_defer_jsonl_recovery(args: &SyncArgs) -> bool {
    args.reconcile_additive
        || args.migrate_source_repo_path
        || (args.import_only && (args.rename_prefix || args.skip_invalid_records))
}

/// Reject argument combinations that must fail BEFORE opening storage or
/// triggering any rebuild side effect. A `--flush-only --rebuild` or
/// `--merge --rebuild` combination must return an error without having
/// touched the DB family — otherwise the validation message arrives after
/// `recover_database_from_jsonl` has already moved the existing DB aside.
#[allow(clippy::too_many_lines)]
pub fn validate_sync_mode_args(args: &SyncArgs) -> Result<()> {
    if args.skip_invalid_records && !args.import_only {
        return Err(BeadsError::Validation {
            field: "skip_invalid_records".to_string(),
            reason: "--skip-invalid-records can only be used with --import-only".to_string(),
        });
    }
    if args.skip_invalid_records && (args.force || args.rebuild || args.rename_prefix) {
        return Err(BeadsError::Validation {
            field: "skip_invalid_records".to_string(),
            reason: "--skip-invalid-records is an additive recovery mode and cannot be combined with --force, --rebuild, or --rename-prefix"
                .to_string(),
        });
    }
    if args.dry_run && !(args.reconcile || args.reconcile_additive || args.migrate_source_repo_path)
    {
        return Err(BeadsError::Validation {
            field: "dry_run".to_string(),
            reason: "--dry-run can only be used with --reconcile, --reconcile-additive, or --migrate-source-repo-path"
                .to_string(),
        });
    }
    if args.apply && !(args.reconcile_additive || args.migrate_source_repo_path) {
        return Err(BeadsError::Validation {
            field: "apply".to_string(),
            reason:
                "--apply can only be used with --reconcile-additive or --migrate-source-repo-path"
                    .to_string(),
        });
    }
    if args.apply && args.expect_plan_sha256.is_none() {
        return Err(BeadsError::Validation {
            field: "expect_plan_sha256".to_string(),
            reason: "--apply requires --expect-plan-sha256 from the reviewed dry-run receipt"
                .to_string(),
        });
    }
    if let Some(plan_sha256) = &args.expect_plan_sha256
        && (plan_sha256.len() != 64
            || !plan_sha256
                .bytes()
                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)))
    {
        return Err(BeadsError::Validation {
            field: "expect_plan_sha256".to_string(),
            reason: "--expect-plan-sha256 must be exactly 64 lowercase hexadecimal characters"
                .to_string(),
        });
    }
    if args.reconcile_additive {
        let irrelevant = [
            (args.export_parallelism.is_some(), "--export-parallelism"),
            (args.force, "--force"),
            (args.force_db, "--force-db"),
            (args.force_jsonl, "--force-jsonl"),
            (args.manifest, "--manifest"),
            (args.error_policy.is_some(), "--error-policy"),
            (args.orphans.is_some(), "--orphans"),
            (args.rename_prefix, "--rename-prefix"),
            (args.rebuild, "--rebuild"),
            (args.witness_parallelism.is_some(), "--witness-parallelism"),
        ];
        if let Some((_, flag)) = irrelevant.into_iter().find(|(present, _)| *present) {
            return Err(BeadsError::Validation {
                field: "reconcile_additive".to_string(),
                reason: format!("{flag} is not used by --reconcile-additive"),
            });
        }
        let resolution_ids = args
            .resolve_source_ids
            .iter()
            .map(String::as_str)
            .collect::<BTreeSet<_>>();
        if resolution_ids.len() != args.resolve_source_ids.len()
            || resolution_ids
                .iter()
                .any(|issue_id| issue_id.trim().is_empty() || issue_id.trim() != *issue_id)
        {
            return Err(BeadsError::Validation {
                field: "resolve_source_ids".to_string(),
                reason: "--resolve-source-id values must be unique, nonblank, and trimmed"
                    .to_string(),
            });
        }
    }

    if args.migrate_source_repo_path {
        let irrelevant = [
            (args.force, "--force"),
            (args.force_db, "--force-db"),
            (args.force_jsonl, "--force-jsonl"),
            (args.manifest, "--manifest"),
            (args.error_policy.is_some(), "--error-policy"),
            (args.orphans.is_some(), "--orphans"),
            (args.rename_prefix, "--rename-prefix"),
            (args.rebuild, "--rebuild"),
            (args.skip_invalid_records, "--skip-invalid-records"),
            (!args.resolve_source_ids.is_empty(), "--resolve-source-id"),
        ];
        if let Some((_, flag)) = irrelevant.into_iter().find(|(present, _)| *present) {
            return Err(BeadsError::Validation {
                field: "migrate_source_repo_path".to_string(),
                reason: format!("{flag} is not used by --migrate-source-repo-path"),
            });
        }
    }

    let mode_count = u8::from(args.status)
        + u8::from(args.flush_only)
        + u8::from(args.import_only)
        + u8::from(args.merge)
        + u8::from(args.reconcile)
        + u8::from(args.witness)
        + u8::from(args.reconcile_additive)
        + u8::from(args.migrate_source_repo_path);
    if mode_count > 1 {
        return Err(BeadsError::Validation {
            field: "mode".to_string(),
            reason:
                "Must specify exactly one of --flush-only, --import-only, --merge, --reconcile, --reconcile-additive, --migrate-source-repo-path, --status, or --witness"
                    .to_string(),
        });
    }
    if mode_count == 0 {
        return Err(BeadsError::Validation {
            field: "mode".to_string(),
            reason:
                "Must specify one of --flush-only, --import-only, --merge, --reconcile, --reconcile-additive, --migrate-source-repo-path, --status, or --witness"
                    .to_string(),
        });
    }

    if args.reconcile {
        if args.force {
            return Err(BeadsError::Validation {
                field: "force".to_string(),
                reason: "--force cannot be used with --reconcile; reconcile is always additive \
                         and guarded (use --import-only --force for a destructive import)"
                    .to_string(),
            });
        }
        if args.rename_prefix {
            return Err(BeadsError::Validation {
                field: "rename_prefix".to_string(),
                reason: "--rename-prefix cannot be used with --reconcile; reconcile never \
                         rewrites issue ids"
                    .to_string(),
            });
        }
        if args.orphans.is_some() {
            return Err(BeadsError::Validation {
                field: "orphans".to_string(),
                reason: "--orphans cannot be used with --reconcile; reconcile only removes \
                         dangling dependency references from rows it just wrote"
                    .to_string(),
            });
        }
    }

    if args.witness && args.witness_chunk_lines == 0 {
        return Err(BeadsError::Validation {
            field: "witness_chunk_lines".to_string(),
            reason: "--witness-chunk-lines must be greater than zero".to_string(),
        });
    }

    if args.witness_parallelism == Some(0) {
        return Err(BeadsError::Validation {
            field: "witness_parallelism".to_string(),
            reason: "--witness-parallelism must be greater than zero".to_string(),
        });
    }
    if args.export_parallelism == Some(0) {
        return Err(BeadsError::Validation {
            field: "export_parallelism".to_string(),
            reason: "--export-parallelism must be greater than zero".to_string(),
        });
    }
    // --rebuild only makes sense with explicit import mode.
    if args.rebuild && !args.import_only {
        return Err(BeadsError::Validation {
            field: "rebuild".to_string(),
            reason: "--rebuild can only be used with --import-only".to_string(),
        });
    }

    if (args.force_db || args.force_jsonl) && !args.merge {
        return Err(BeadsError::Validation {
            field: "merge-resolution".to_string(),
            reason: "--force-db and --force-jsonl can only be used with --merge".to_string(),
        });
    }

    if args.force_db && args.force_jsonl {
        return Err(BeadsError::Validation {
            field: "merge-resolution".to_string(),
            reason: "--force-db conflicts with --force-jsonl; choose one merge winner".to_string(),
        });
    }

    if args.force && (args.force_db || args.force_jsonl) {
        return Err(BeadsError::Validation {
            field: "force".to_string(),
            reason: "--force conflicts with --force-db and --force-jsonl for --merge; choose one conflict resolution policy".to_string(),
        });
    }
    Ok(())
}

fn merge_conflict_resolution(args: &SyncArgs) -> ConflictResolution {
    if args.force_db {
        ConflictResolution::PreferLocal
    } else if args.force_jsonl {
        ConflictResolution::PreferExternal
    } else if args.force {
        ConflictResolution::PreferNewer
    } else {
        ConflictResolution::Manual
    }
}

fn merge_conflict_resolution_label(strategy: ConflictResolution) -> &'static str {
    match strategy {
        ConflictResolution::PreferLocal => "force-db",
        ConflictResolution::PreferExternal => "force-jsonl",
        ConflictResolution::PreferNewer => "force-newer",
        ConflictResolution::Manual => "manual",
    }
}

/// When `--rebuild` is requested against an existing (non-auto-rebuilt)
/// DB, delegate the actual rebuild to the same proven path that auto-
/// recovery uses: backup the DB family, open a fresh connection, import
/// JSONL, checkpoint, VACUUM/REINDEX. The in-place
/// `reset_data_tables`+`import_from_jsonl` code path inside
/// `execute_import` is fragile on fsqlite — it trips stale-pager/MVCC
/// bugs that leave "never used" pages and partial-index mismatches that
/// VACUUM can't always reclaim. Using `recover_database_from_jsonl`
/// sidesteps all of that, and `execute_import` then sees
/// `auto_rebuilt == true` and short-circuits.
///
/// Only fire this for the request that will actually go through
/// `execute_import`: `--rebuild --import-only`. All other mode pairings
/// are rejected before storage opens. Also require the JSONL to exist —
/// `recover_database_from_jsonl` runs a preflight that fails hard if the
/// file is missing, whereas `execute_import` already handles a missing
/// JSONL gracefully, so leave that case to the normal path.
///
/// Skip the delegation when the caller asked for behavior that the
/// auto-recovery path does not replicate: `--rename-prefix` rewrites
/// imported IDs into the configured prefix, while
/// `repair_database_from_jsonl` always runs with
/// `rename_on_import = false`. That means the delegation would silently
/// skip the requested rename behavior.
///
/// `--orphans` is intentionally *not* part of this guard today. The
/// current import engine parses `orphan_mode` into `ImportConfig`, but it
/// does not consult that field during import, so delegating does not
/// change effective behavior. If orphan-mode semantics become active in
/// the future, revisit this guard and the auto-rebuild conflict
/// detection below.
fn maybe_delegate_rebuild(
    args: &SyncArgs,
    open_result: &mut config::OpenStorageResult,
) -> Result<()> {
    let delegation_would_drop_user_flags = args.rename_prefix || args.skip_invalid_records;
    let should_delegate = args.rebuild
        && args.import_only
        && !open_result.no_db
        && !open_result.auto_rebuilt
        && open_result.paths.jsonl_path.is_file()
        && !delegation_would_drop_user_flags;
    if !should_delegate {
        return Ok(());
    }

    info!(
        db_path = %open_result.paths.db_path.display(),
        jsonl_path = %open_result.paths.jsonl_path.display(),
        "--rebuild requested on existing DB: delegating to auto-recovery rebuild path"
    );
    // Recovery itself now captures the JSONL once before inspecting
    // tombstones or replacing the database. It preserves unflushed
    // tombstones (and dirty live issues, GitHub #394) against that same
    // immutable generation, then imports the identical bytes. Keeping this
    // at the recovery boundary prevents a path replacement from mixing
    // filter state from A with rows from B.
    open_result.recover_database_from_jsonl()?;
    Ok(())
}

/// Dispatch to the appropriate sync-subcommand implementation based on
/// the flag pattern (`--status` / `--flush-only` / `--merge` /
/// `--import-only`). Read-only branches avoid mutation; operations that can
/// modify state hold a `&mut` borrow on `open_result.storage` for the
/// duration of their execution. Any `Err` propagates back to
/// `finalize_sync_result`, which is the single place that decides how to
/// handle recovery-backup rollback.
fn dispatch_sync_subcommand(
    args: &SyncArgs,
    cli: &config::CliOverrides,
    ctx: &OutputContext,
    beads_dir: &Path,
    path_policy: &SyncPathPolicy,
    open_result: &mut config::OpenStorageResult,
) -> Result<SyncDispatchCompletion> {
    let options = sync_dispatch_options(args, cli, ctx, open_result);
    let operation = sync_operation(args);

    match operation {
        SyncOperation::Status => execute_status(
            &open_result.storage,
            path_policy,
            &options.db_path,
            options.use_json,
            ctx,
        )
        .map(|()| SyncDispatchCompletion::default()),
        SyncOperation::Witness => {
            execute_witness(path_policy, args, options.use_json, ctx)
                .map(|()| SyncDispatchCompletion::default())
        }
        SyncOperation::Reconcile => {
            execute_reconcile(&mut open_result.storage, path_policy, args, options.use_json, ctx)
                .map(|()| SyncDispatchCompletion::default())
        }
        SyncOperation::ReconcileAdditive => Err(BeadsError::Internal {
            message: "reviewed additive reconciliation bypassed its sole lock-owning command path"
                .to_string(),
        }),
        SyncOperation::MigrateSourceRepoPath
        | SyncOperation::Flush
        | SyncOperation::Merge
        | SyncOperation::Import => dispatch_publishing_sync_subcommand(
            args,
            cli,
            ctx,
            beads_dir,
            path_policy,
            open_result,
            &options,
        ),
        SyncOperation::Unspecified => Err(BeadsError::Validation {
            field: "mode".to_string(),
            reason:
                "Must specify one of --flush-only, --import-only, --merge, --reconcile, --reconcile-additive, --migrate-source-repo-path, --status, or --witness"
                    .to_string(),
        }),
    }
}

fn dispatch_publishing_sync_subcommand(
    args: &SyncArgs,
    cli: &config::CliOverrides,
    ctx: &OutputContext,
    beads_dir: &Path,
    path_policy: &SyncPathPolicy,
    open_result: &mut config::OpenStorageResult,
    options: &SyncDispatchOptions,
) -> Result<SyncDispatchCompletion> {
    match sync_operation(args) {
        SyncOperation::MigrateSourceRepoPath => {
            let no_db = open_result.no_db;
            let (storage, retained_source, expected_source, jsonl_authority) =
                open_result.jsonl_write_context();
            execute_source_repo_path_migration(
                storage,
                path_policy,
                args,
                options.use_json,
                options.show_progress,
                options.retention_days,
                options.history_config.clone(),
                retained_source,
                expected_source,
                jsonl_authority,
                cli,
                &options.db_path,
                no_db,
                ctx,
            )
        }
        SyncOperation::Flush => {
            let (storage, retained_source, expected_source, jsonl_authority) =
                open_result.jsonl_write_context();
            execute_flush(
                storage,
                beads_dir,
                path_policy,
                args,
                options.use_json,
                options.show_progress,
                options.retention_days,
                options.history_config.clone(),
                retained_source,
                expected_source,
                jsonl_authority,
                ctx,
            )
            .map(SyncDispatchCompletion::published)
        }
        SyncOperation::Merge => {
            let no_db = open_result.no_db;
            let (storage, retained_source, expected_source, jsonl_authority) =
                open_result.jsonl_write_context();
            execute_merge(
                storage,
                path_policy,
                args,
                options.use_json,
                options.show_progress,
                options.retention_days,
                options.history_config.clone(),
                retained_source,
                expected_source,
                jsonl_authority,
                cli,
                &options.db_path,
                no_db,
                ctx,
            )
        }
        SyncOperation::Import => {
            let auto_rebuilt = open_result.auto_rebuilt;
            let (storage, retained_source, retained_authority) = open_result.import_context();
            execute_import(
                storage,
                beads_dir,
                cli,
                path_policy,
                args,
                options.use_json,
                options.show_progress,
                auto_rebuilt,
                retained_source,
                retained_authority,
                &options.db_path,
                ctx,
            )
            .map(SyncDispatchCompletion::published)
        }
        SyncOperation::Status
        | SyncOperation::Witness
        | SyncOperation::Reconcile
        | SyncOperation::ReconcileAdditive
        | SyncOperation::Unspecified => unreachable!("non-publishing operation was pre-dispatched"),
    }
}

fn finalize_sync_dispatch_completion(
    completion: SyncDispatchCompletion,
    open_result: &mut config::OpenStorageResult,
    ctx: &OutputContext,
) -> Result<()> {
    let SyncDispatchCompletion {
        published_source,
        owned_jsonl_authority,
        pending_merge,
        deferred_output,
    } = completion;
    if let Some(source) = published_source.as_ref() {
        open_result.adopt_published_jsonl_source(Arc::clone(source), owned_jsonl_authority)?;
    } else if owned_jsonl_authority.is_some() {
        return Err(BeadsError::Internal {
            message: "sync completion retained JSONL authority without a published source"
                .to_string(),
        });
    }
    open_result.verify_retained_jsonl_source_current()?;

    if let Some(pending_merge) = pending_merge {
        let published_source = published_source
            .as_ref()
            .ok_or_else(|| BeadsError::Internal {
                message: "pending sync merge completion did not retain its published JSONL witness"
                    .to_string(),
            })?;
        open_result.verify_retained_jsonl_authority(
            &pending_merge.receipt.intent.jsonl_authority_sha256,
        )?;
        finalize_pending_sync_merge_after_adoption(
            &mut open_result.storage,
            &open_result.paths.db_path,
            published_source,
            &pending_merge,
            open_result.no_db,
        )
        .map_err(|source| BeadsError::CommittedStateUnwitnessed {
            operation: "terminal sync merge adoption and receipt cleanup".to_string(),
            source: Box::new(source),
        })?;
    }

    if let Some(output) = deferred_output {
        render_deferred_sync_output(output, ctx);
    }
    Ok(())
}

fn finalize_pending_sync_merge_after_adoption(
    storage: &mut crate::storage::SqliteStorage,
    db_path: &Path,
    published_source: &JsonlSourceSnapshot,
    pending: &PendingSyncMergeCompletion,
    no_db: bool,
) -> Result<()> {
    let receipt = &pending.receipt;
    receipt.validate()?;
    if receipt.phase != SyncMergePendingPhase::ExportFinalized {
        return Err(BeadsError::SyncConflict {
            message: "Cannot clear a pending merge receipt before export finalization".to_string(),
        });
    }
    if receipt.jsonl_after.as_ref() != Some(&published_source.state_witness())
        || published_source.raw_sha256() != receipt.jsonl_after_raw_sha256
        || published_source.content_sha256() != receipt.jsonl_after_content_sha256
    {
        return Err(BeadsError::SyncConflict {
            message: "Adopted JSONL source does not match the pending merge receipt".to_string(),
        });
    }

    let configured_database_authority = database_write_authority_sha256(db_path)?;
    if configured_database_authority != receipt.intent.database_authority_sha256 {
        return Err(BeadsError::SyncConflict {
            message: "Pending merge database authority does not match the committed merge intent"
                .to_string(),
        });
    }
    if !no_db {
        let database_authority = storage.attached_write_authority();
        let database_authority = database_authority.ok_or_else(|| BeadsError::SyncConflict {
            message:
                "Pending merge terminal verification requires retained database-family authority"
                    .to_string(),
        })?;
        database_authority.verify_database_authority()?;
        if database_authority.authority_path_sha256() != receipt.intent.database_authority_sha256 {
            return Err(BeadsError::SyncConflict {
                message:
                    "Retained database-family authority differs from the committed merge intent"
                        .to_string(),
            });
        }
    }

    pending.base_authority.verify_jsonl_authority()?;
    if pending.base_authority.authority_path_sha256() != receipt.intent.base_authority_sha256 {
        return Err(BeadsError::SyncConflict {
            message: "Pending merge retained base authority differs from the committed intent"
                .to_string(),
        });
    }
    let terminal_base = pending.base_authority.capture_target()?;
    if terminal_base.raw_sha256() != published_source.raw_sha256()
        || terminal_base.content_sha256() != published_source.content_sha256()
        || terminal_base.size() != published_source.size()
    {
        return Err(BeadsError::SyncConflict {
            message: "Merge base changed before outer completion adopted the JSONL".to_string(),
        });
    }
    if storage.with_read_transaction(crate::sync::capture_sync_merge_core_witness)?
        != receipt.database_after
    {
        return Err(BeadsError::SyncConflict {
            message: "Database changed before outer merge completion verification".to_string(),
        });
    }
    if storage.pending_sync_merge_receipt()?.as_ref() != Some(receipt) {
        return Err(BeadsError::SyncConflict {
            message: "Pending merge receipt changed before terminal cleanup".to_string(),
        });
    }
    storage.compare_and_clear_pending_sync_merge_receipt(receipt)
}

fn render_deferred_sync_output(output: DeferredSyncOutput, ctx: &OutputContext) {
    match output {
        DeferredSyncOutput::Merge {
            report,
            resolution,
            capacity_warnings,
            use_json,
        } => {
            if use_json {
                ctx.json_pretty(&serde_json::json!({
                    "status": "success",
                    "merged_issues": report.kept.len(),
                    "deleted_issues": report.deleted.len(),
                    "conflicts": report.conflicts.len(),
                    "resolution": resolution,
                    "notes": report.notes,
                    "id_collisions": report.id_collisions,
                    "warnings": capacity_warnings,
                }));
            } else if should_render_human_sync_output(ctx, use_json) {
                for collision in &report.id_collisions {
                    ctx.warning(&id_collision_warning(collision));
                }
                if ctx.is_rich() {
                    render_merge_result_rich(&report, ctx);
                } else {
                    println!("Merge complete:");
                    println!("  Kept/Updated: {} issues", report.kept.len());
                    println!("  Deleted: {} issues", report.deleted.len());
                    if !report.notes.is_empty() {
                        println!("  Notes:");
                        for (id, note) in &report.notes {
                            println!("    - {id}: {note}");
                        }
                    }
                    println!("  Base snapshot updated.");
                    println!("  JSONL exported.");
                }
                for warning in &capacity_warnings {
                    ctx.warning(&warning.to_string());
                }
            }
        }
        DeferredSyncOutput::ResumedMerge {
            receipt_id,
            phase_before,
            capacity_warnings,
            use_json,
        } => {
            if use_json {
                ctx.json_pretty(&serde_json::json!({
                    "status": "resumed",
                    "receipt_id": receipt_id,
                    "phase_before": phase_before,
                    "base_updated": true,
                    "warnings": capacity_warnings,
                }));
            } else if should_render_human_sync_output(ctx, use_json) {
                println!(
                    "Resumed committed merge {} from phase {:?}; JSONL and base are verified.",
                    receipt_id, phase_before
                );
                for warning in &capacity_warnings {
                    ctx.warning(&warning.to_string());
                }
            }
        }
        DeferredSyncOutput::SourceRepoPathMigration { receipt, use_json } => {
            render_source_repo_path_migration_receipt(&receipt, ctx, use_json);
        }
    }
}

fn sync_dispatch_options(
    args: &SyncArgs,
    cli: &config::CliOverrides,
    ctx: &OutputContext,
    open_result: &config::OpenStorageResult,
) -> SyncDispatchOptions {
    let use_json = ctx.is_json() || args.robot;
    let quiet = cli.quiet.unwrap_or(false);
    SyncDispatchOptions {
        db_path: open_result.paths.db_path.clone(),
        retention_days: open_result.paths.metadata.deletions_retention_days,
        use_json,
        show_progress: should_show_progress(use_json, quiet),
        history_config: open_result.resolved_history_config(),
    }
}

fn sync_operation(args: &SyncArgs) -> SyncOperation {
    if args.witness {
        SyncOperation::Witness
    } else if args.reconcile_additive {
        SyncOperation::ReconcileAdditive
    } else if args.migrate_source_repo_path {
        SyncOperation::MigrateSourceRepoPath
    } else if args.status {
        SyncOperation::Status
    } else if args.flush_only {
        SyncOperation::Flush
    } else if args.merge {
        SyncOperation::Merge
    } else if args.import_only {
        SyncOperation::Import
    } else if args.reconcile {
        SyncOperation::Reconcile
    } else {
        SyncOperation::Unspecified
    }
}

#[allow(clippy::too_many_lines)]
fn render_additive_reconcile_receipt(
    receipt: &AdditiveReconcileReceipt,
    ctx: &OutputContext,
    machine_readable: bool,
) {
    if machine_readable {
        ctx.json_pretty(receipt);
        return;
    }

    println!(
        "additive reconciliation: status={} plan_sha256={} source={} target_before={} created={} updated={} equal={} synchronized={} conflicted_issues={} conflict_observations={} deleted=0",
        receipt.status.as_str(),
        receipt.plan_sha256,
        receipt.source_issues,
        receipt.target_before.issues,
        receipt.created,
        receipt.updated,
        receipt.skipped_equal,
        receipt.synchronized,
        receipt.conflicted,
        receipt.conflict_occurrences,
    );
    println!(
        "sync bookkeeping: export_hashes={}/{} dirty_markers={}/{} metadata_update_planned={} metadata_changed={}",
        receipt.export_hashes_updated,
        receipt.export_hash_updates_planned,
        receipt.dirty_markers_cleared,
        receipt.dirty_markers_clear_planned,
        receipt.metadata_update_planned,
        receipt.metadata_changed
    );
    println!(
        "audit events: {}/{} preserved; database-only issues preserved={}; jsonl_written=false",
        receipt.events_before, receipt.events_after, receipt.db_only_preserved
    );
    println!(
        "blocking cycles: preexisting={} projected={} new={}",
        receipt.preexisting_blocking_cycles,
        receipt.projected_blocking_cycles,
        receipt.new_blocking_cycles
    );
    println!(
        "authority witnesses: workspace_path_sha256={} workspace_identity_sha256={} source_path_sha256={} source_identity_sha256={} database_path_sha256={} database_identity_sha256={} write_lock_authority_sha256={} schema_version={}",
        receipt.workspace_path_sha256,
        receipt.workspace_identity_sha256,
        receipt.source_path_sha256,
        receipt.source_identity_sha256,
        receipt.database_path_sha256,
        receipt
            .database_identity_sha256
            .as_deref()
            .unwrap_or("none"),
        receipt.write_lock_authority_sha256,
        receipt.database_user_version,
    );
    println!(
        "source witnesses: raw_sha256={} content_sha256={} storage_projection_sha256={} size={} mtime={}",
        receipt.source_raw_sha256,
        receipt.source_content_sha256,
        receipt.source_storage_projection_sha256,
        receipt.source_size,
        receipt.source_mtime,
    );
    println!(
        "relation proof: before={:?} after={:?} planned={:?} applied={:?}",
        receipt.relations_before,
        receipt.relations_after,
        receipt.relation_rows_planned,
        receipt.relation_rows_applied,
    );
    println!(
        "expected poststate digests: issue_raw={} issue_semantic={} issue_content_hash={} export_hash={} dirty={} metadata={} blocked_cache={} child_counter={} sqlite_sequence={}",
        receipt.expected_issue_raw_payload_sha256,
        receipt.expected_issue_semantic_payload_sha256,
        receipt.expected_issue_content_hash_payload_sha256,
        receipt.expected_export_hash_payload_sha256,
        receipt.expected_dirty_payload_sha256,
        receipt.expected_metadata_payload_sha256,
        receipt.expected_blocked_cache_payload_sha256,
        receipt.expected_child_counter_payload_sha256,
        receipt.expected_sqlite_sequence_payload_sha256,
    );
    println!(
        "manifest digests: created={} updated={} equal={} db_only={} scalar_updates={} content_hash_repairs={} comment_remaps={} conflict_issue_ids={} conflict_witnesses={} conflict_scalar_diffs={} conflict_relation_diffs={}",
        receipt.created_issue_ids_sha256,
        receipt.updated_issue_ids_sha256,
        receipt.equal_issue_ids_sha256,
        receipt.db_only_issue_ids_sha256,
        receipt.scalar_updates_sha256,
        receipt.content_hash_repairs_sha256,
        receipt.comment_id_remaps_sha256,
        receipt.conflict_issue_ids_sha256,
        receipt.conflict_witnesses_sha256,
        receipt.conflict_scalar_diffs_sha256,
        receipt.conflict_relation_diffs_sha256,
    );
    if let Some(health_after) = &receipt.health_after {
        println!(
            "health: before_integrity={} before_fk_violations={} after_integrity={} after_fk_violations={}",
            receipt.health_before.integrity_messages.len(),
            receipt.health_before.foreign_key_violations.len(),
            health_after.integrity_messages.len(),
            health_after.foreign_key_violations.len(),
        );
    } else {
        println!(
            "health: before_integrity={} before_fk_violations={} after=not_checked",
            receipt.health_before.integrity_messages.len(),
            receipt.health_before.foreign_key_violations.len(),
        );
    }
    println!(
        "postcommit checks: database_authority={:?} database_poststate={:?} workspace_authority={:?} source={:?} foreign_keys={:?} failures={:?}",
        receipt.database_authority_preserved_after_commit,
        receipt.database_poststate_preserved_after_commit,
        receipt.workspace_authority_preserved_after_commit,
        receipt.source_preserved_after_commit,
        receipt.foreign_keys_restored_after_commit,
        receipt.postcommit_failures,
    );
    if !receipt.scalar_updates.is_empty() {
        for update in receipt.scalar_updates.iter().take(HUMAN_WITNESS_LIMIT) {
            println!(
                "scalar witness: issue={} resolution={} fields={} diff_sha256={} before_sha256={} after_sha256={} relations_sha256={}",
                update.issue_id,
                update.resolution.as_str(),
                update.changed_fields.join(","),
                update.diff_sha256,
                update.before_payload_sha256,
                update.after_payload_sha256,
                update.relation_payload_sha256,
            );
        }
        if receipt.scalar_updates.len() > HUMAN_WITNESS_LIMIT {
            println!(
                "scalar witness output: total={} shown={} manifest_sha256={} truncated=true; use --robot for the complete manifest",
                receipt.scalar_updates.len(),
                HUMAN_WITNESS_LIMIT,
                receipt.scalar_updates_sha256,
            );
        }
    }
    for repair in receipt
        .content_hash_repairs
        .iter()
        .take(HUMAN_WITNESS_LIMIT)
    {
        let (before_len, before_sha256) = human_witness_value_digest(repair.before.as_deref());
        let (after_len, after_sha256) = human_witness_value_digest(Some(&repair.after));
        println!(
            "content-hash repair witness: issue={} before_len={} before_sha256={} after_len={} after_sha256={}",
            repair.issue_id, before_len, before_sha256, after_len, after_sha256,
        );
    }
    if receipt.content_hash_repairs.len() > HUMAN_WITNESS_LIMIT {
        println!(
            "content-hash repair output: total={} shown={} manifest_sha256={} truncated=true; use --robot for the complete manifest",
            receipt.content_hash_repairs.len(),
            HUMAN_WITNESS_LIMIT,
            receipt.content_hash_repairs_sha256,
        );
    }
    for remap in receipt.comment_id_remaps.iter().take(HUMAN_WITNESS_LIMIT) {
        println!(
            "comment remap witness: issue={} old_id={} new_id={} logical_payload_sha256={}",
            remap.issue_id, remap.old_id, remap.new_id, remap.logical_payload_sha256,
        );
    }
    if receipt.comment_id_remaps.len() > HUMAN_WITNESS_LIMIT {
        println!(
            "comment remap output: total={} shown={} manifest_sha256={} truncated=true; use --robot for the complete manifest",
            receipt.comment_id_remaps.len(),
            HUMAN_WITNESS_LIMIT,
            receipt.comment_id_remaps_sha256,
        );
    }
    for line in additive_conflict_human_lines(receipt, HUMAN_WITNESS_LIMIT) {
        println!("{line}");
    }
}

/// Fold the subcommand result into the final command outcome, restoring
/// the pre-recovery backup on error (deferred-recovery paths only) and
/// discarding it on success.
fn finalize_sync_result(
    command_result: Result<()>,
    open_result: &mut config::OpenStorageResult,
) -> Result<()> {
    match command_result {
        Ok(()) => {
            open_result.discard_pending_recovery_backup()?;
            Ok(())
        }
        Err(command_err) => {
            if command_err.primary_mutation_committed() {
                if let Err(finalize_err) = open_result.discard_pending_recovery_backup() {
                    return Err(BeadsError::WithContext {
                        context: format!(
                            "sync primary state committed ({command_err}); refusing rollback, but deferred recovery finalization also failed: {finalize_err}"
                        ),
                        source: Box::new(command_err),
                    });
                }
                return Err(command_err);
            }
            let recovery_dir = open_result.pending_recovery_dir().map(PathBuf::from);
            if let Err(restore_err) = open_result.restore_pending_recovery_backup() {
                let context = recovery_dir.map_or_else(
                    || {
                        format!(
                            "sync command failed after deferred database recovery ({command_err}); original database restore also failed"
                        )
                    },
                    |dir| {
                        format!(
                            "sync command failed after deferred database recovery ({command_err}); original database restore from '{}' also failed",
                            dir.display()
                        )
                    },
                );
                return Err(BeadsError::WithContext {
                    context,
                    source: Box::new(restore_err),
                });
            }
            Err(command_err)
        }
    }
}

fn should_render_human_sync_output(ctx: &OutputContext, use_json: bool) -> bool {
    // Keep JSON/robot output paths alive even when quiet suppresses human text.
    !ctx.is_quiet() || use_json
}

fn validate_sync_paths(
    beads_dir: &Path,
    jsonl_path: &Path,
    allow_external_jsonl: bool,
) -> Result<SyncPathPolicy> {
    debug!(
        beads_dir = %beads_dir.display(),
        jsonl_path = %jsonl_path.display(),
        allow_external_jsonl,
        "Validating sync paths"
    );
    validate_operator_requested_sync_path(beads_dir, jsonl_path)?;

    let canonical_beads = dunce::canonicalize(beads_dir).map_err(|e| {
        BeadsError::Config(format!(
            "Failed to resolve .beads directory {}: {e}",
            beads_dir.display()
        ))
    })?;

    // Resolve the requested path to an absolute operator-facing location without
    // collapsing the final component. Raw-path validation must inspect the
    // actual path the operator asked sync to touch so symlink and `.git`
    // invariants cannot be bypassed by early canonicalization.
    let jsonl_path = resolve_requested_sync_path(jsonl_path)?;

    let extension = jsonl_path
        .extension()
        .and_then(|ext| ext.to_str())
        .map(str::to_ascii_lowercase);
    if extension.as_deref() != Some("jsonl") {
        return Err(BeadsError::Config(format!(
            "JSONL path must end with .jsonl: {}",
            jsonl_path.display()
        )));
    }

    let is_external = !jsonl_path.starts_with(&canonical_beads);
    if is_external && !allow_external_jsonl {
        warn!(
            path = %jsonl_path.display(),
            "Rejected JSONL path outside .beads"
        );
        return Err(BeadsError::Config(format!(
            "Refusing to use JSONL path outside .beads: {}.\n\
             Hint: pass --allow-external-jsonl if this is intentional.",
            jsonl_path.display()
        )));
    }

    let manifest_path = canonical_beads.join(".manifest.json");
    let jsonl_temp_path = export_temp_path(&jsonl_path);

    if contains_git_dir(&jsonl_path) {
        warn!(
            path = %jsonl_path.display(),
            "Rejected JSONL path inside .git directory"
        );
        return Err(BeadsError::Config(format!(
            "Refusing to use JSONL path inside .git directory: {}.\n\
            Move the JSONL path outside .git to proceed.",
            jsonl_path.display()
        )));
    }

    validate_sync_path_with_external(&jsonl_path, &canonical_beads, allow_external_jsonl)?;

    debug!(
        jsonl_path = %jsonl_path.display(),
        jsonl_temp_path = %jsonl_temp_path.display(),
        manifest_path = %manifest_path.display(),
        is_external,
        "Sync path validation complete"
    );

    Ok(SyncPathPolicy {
        jsonl_path,
        jsonl_temp_path,
        manifest_path,
        beads_dir: canonical_beads,
        is_external,
        allow_external_jsonl,
    })
}

fn validate_operator_requested_sync_path(beads_dir: &Path, jsonl_path: &Path) -> Result<()> {
    let git_check = validate_no_git_path(jsonl_path);
    if !git_check.is_allowed() {
        return Err(BeadsError::Config(
            git_check
                .rejection_reason()
                .unwrap_or_else(|| "Git path access denied".to_string()),
        ));
    }

    let canonical_beads = dunce::canonicalize(beads_dir).map_err(|e| {
        BeadsError::Config(format!(
            "Failed to resolve .beads directory {}: {e}",
            beads_dir.display()
        ))
    })?;

    let operator_path = if jsonl_path.is_absolute() {
        jsonl_path.to_path_buf()
    } else {
        std::env::current_dir()
            .map(|cwd| cwd.join(jsonl_path))
            .map_err(|e| {
                BeadsError::Config(format!(
                    "Failed to resolve current directory for JSONL path {}: {e}",
                    jsonl_path.display()
                ))
            })?
    };

    if !operator_path.starts_with(beads_dir) && !operator_path.starts_with(&canonical_beads) {
        return Ok(());
    }

    let mut candidate = PathBuf::new();
    for component in operator_path.components() {
        candidate.push(component.as_os_str());
        let Ok(metadata) = fs::symlink_metadata(&candidate) else {
            continue;
        };
        if !metadata.file_type().is_symlink() {
            continue;
        }

        let target = fs::read_link(&candidate).map_err(|e| {
            BeadsError::Config(format!(
                "Failed to inspect symlinked JSONL path component {}: {e}",
                candidate.display()
            ))
        })?;
        let absolute_target = if target.is_absolute() {
            target
        } else {
            candidate
                .parent()
                .unwrap_or_else(|| Path::new(""))
                .join(target)
        };
        let canonical_target =
            dunce::canonicalize(&absolute_target).unwrap_or_else(|_| absolute_target.clone());
        if !crate::sync::path::path_within(&canonical_target, &canonical_beads) {
            return Err(BeadsError::Config(format!(
                "Refusing to use JSONL path through symlink escaping .beads: {} -> {}",
                candidate.display(),
                canonical_target.display()
            )));
        }
    }

    Ok(())
}

fn resolve_requested_sync_path(jsonl_path: &Path) -> Result<PathBuf> {
    if jsonl_path.is_absolute() {
        return Ok(jsonl_path.to_path_buf());
    }

    let file_name = jsonl_path
        .file_name()
        .ok_or_else(|| BeadsError::Config("JSONL path must include a filename".to_string()))?;
    let jsonl_parent = jsonl_path
        .parent()
        .filter(|parent| !parent.as_os_str().is_empty())
        .unwrap_or_else(|| Path::new("."));

    Ok(resolve_sync_parent_path(jsonl_parent)?.join(file_name))
}

fn resolve_sync_parent_path(jsonl_parent: &Path) -> Result<PathBuf> {
    if jsonl_parent.exists() {
        return dunce::canonicalize(jsonl_parent).map_err(|e| {
            BeadsError::Config(format!(
                "JSONL directory is not accessible: {} ({e})",
                jsonl_parent.display()
            ))
        });
    }

    if jsonl_parent.is_absolute() {
        return Ok(jsonl_parent.to_path_buf());
    }

    let cwd = std::env::current_dir().map_err(|e| {
        BeadsError::Config(format!(
            "Failed to resolve current directory for JSONL path {}: {e}",
            jsonl_parent.display()
        ))
    })?;
    Ok(cwd.join(jsonl_parent))
}

fn contains_git_dir(path: &Path) -> bool {
    path.components().any(|component| match component {
        Component::Normal(name) => name == ".git",
        _ => false,
    })
}

/// Classify workspace health from the cheap signals available in
/// sync-status context (beads_rust#334): the file-state probes shared
/// with doctor (`classify_file_state`: DB header, sidecars, conflict
/// markers, orphaned locks) plus the DB↔JSONL drift booleans the
/// command already computed. This intentionally does NOT run the full
/// doctor checklist — only anomaly codes actually evaluated here are
/// emitted, so the audit record stays honest.
fn classify_sync_status_workspace(
    db_path: &Path,
    jsonl_path: &Path,
    jsonl_newer: bool,
    db_newer: bool,
) -> WorkspaceClassification {
    let mut anomalies = crate::health::classify_file_state(db_path, jsonl_path);
    if jsonl_newer {
        anomalies.push(AnomalyClass::JsonlNewer);
    }
    if db_newer {
        anomalies.push(AnomalyClass::DbNewer);
    }
    WorkspaceClassification::from_anomalies(anomalies)
}

/// Cheap DB↔JSONL coverage probe (`beads_rust-jdmh`).
///
/// The stored-hash shortcut proves the JSONL bytes are unchanged since the
/// last *recorded* import — not that this database ever ingested them. Stored
/// metadata that lies about a partial or lost import makes `--status` and the
/// `--import-only` shortcut assert health over a DB that is missing rows the
/// JSONL holds (GH escalation from jeffreys-skills.md, 2026-07-26 incident:
/// 101 missing issues under "Status: In sync"). Comparing the exportable DB
/// issue count against the JSONL's unique id count catches that state for
/// the cost of one COUNT(*) and one line scan.
#[derive(Debug, Clone, Serialize)]
pub struct SyncCoverageProbe {
    /// Issues the DB would export (tombstones included; ephemerals/wisps excluded).
    pub db_exportable_issues: usize,
    /// Unique `id` values among the JSONL's parseable lines.
    pub jsonl_unique_ids: usize,
}

impl SyncCoverageProbe {
    #[must_use]
    pub const fn drifted(&self) -> bool {
        self.db_exportable_issues != self.jsonl_unique_ids
    }
}

/// Count unique issue ids in a readable JSONL stream. Best-effort: returns
/// `None` on read failures or lines without a string `id`, so callers degrade
/// to legacy behavior instead of failing a read-only diagnostic.
fn jsonl_unique_id_count<R: std::io::BufRead>(reader: R) -> Option<usize> {
    let mut ids: HashSet<String> = HashSet::new();
    for line in reader.lines() {
        let line = line.ok()?;
        let trimmed = line.trim();
        if trimmed.is_empty() {
            continue;
        }
        let value: serde_json::Value = serde_json::from_str(trimmed).ok()?;
        ids.insert(value.get("id")?.as_str()?.to_string());
    }
    Some(ids.len())
}

/// Build the coverage probe for `--status` from the JSONL path on disk.
/// Best-effort; `None` when the JSONL is absent or unreadable.
fn compute_status_coverage_probe(
    storage: &crate::storage::SqliteStorage,
    jsonl_path: &Path,
) -> Option<SyncCoverageProbe> {
    let file = File::open(jsonl_path).ok()?;
    let jsonl_unique_ids = jsonl_unique_id_count(BufReader::new(file))?;
    let db_exportable_issues = storage.count_exportable_issues().ok()?;
    Some(SyncCoverageProbe {
        db_exportable_issues,
        jsonl_unique_ids,
    })
}

/// Compute the `--status` coverage probe and drift flag, logging drift.
fn status_coverage(
    storage: &crate::storage::SqliteStorage,
    jsonl_path: &Path,
    jsonl_exists: bool,
) -> (Option<SyncCoverageProbe>, bool) {
    let coverage = if jsonl_exists {
        compute_status_coverage_probe(storage, jsonl_path)
    } else {
        None
    };
    let coverage_drift = coverage.as_ref().is_some_and(SyncCoverageProbe::drifted);
    if coverage_drift && let Some(probe) = coverage.as_ref() {
        warn!(
            db_exportable_issues = probe.db_exportable_issues,
            jsonl_unique_ids = probe.jsonl_unique_ids,
            "Coverage drift: DB and JSONL hold different issue sets despite hash/timestamp signals"
        );
    }
    (coverage, coverage_drift)
}

/// Execute the --status subcommand.
fn execute_status(
    storage: &crate::storage::SqliteStorage,
    path_policy: &SyncPathPolicy,
    db_path: &Path,
    use_json: bool,
    ctx: &OutputContext,
) -> Result<()> {
    let last_export_time = storage.get_metadata(METADATA_LAST_EXPORT_TIME)?;
    let last_import_time = storage.get_metadata(METADATA_LAST_IMPORT_TIME)?;
    let jsonl_content_hash = storage.get_metadata(METADATA_JSONL_CONTENT_HASH)?;

    let jsonl_path = &path_policy.jsonl_path;
    let staleness = compute_staleness(storage, jsonl_path)?;
    let dirty_count = staleness.dirty_count;
    let jsonl_exists = staleness.jsonl_exists;
    debug!(
        jsonl_path = %jsonl_path.display(),
        jsonl_exists,
        dirty_count,
        "Computed sync status inputs"
    );

    let classification = classify_sync_status_workspace(
        db_path,
        jsonl_path,
        staleness.jsonl_newer,
        staleness.db_newer,
    );
    let reliability_audit = classification.audit_record("sync.status");
    reliability_audit.emit_tracing(
        "status",
        if classification.anomalies.is_empty() {
            "ok"
        } else {
            "findings"
        },
    );

    let (coverage, coverage_drift) = status_coverage(storage, jsonl_path, jsonl_exists);

    let status = SyncStatus {
        dirty_count,
        last_export_time,
        last_import_time,
        jsonl_content_hash,
        jsonl_exists,
        jsonl_newer: staleness.jsonl_newer,
        db_newer: staleness.db_newer,
        workspace_health: classification.health.to_string(),
        reliability_audit,
        coverage,
        coverage_drift,
        git_export: GitExportStatus::not_probed(),
    };
    debug!(
        jsonl_newer = staleness.jsonl_newer,
        db_newer = staleness.db_newer,
        workspace_health = %status.workspace_health,
        "Computed sync staleness"
    );

    if !should_render_human_sync_output(ctx, use_json) {
        return Ok(());
    }

    if use_json {
        // Print JSON directly so --robot works even if OutputContext is non-JSON.
        println!("{}", serde_json::to_string_pretty(&status)?);
    } else if ctx.is_rich() {
        render_status_rich(&status, ctx);
    } else {
        println!("Sync Status:");
        println!("  Dirty issues: {}", status.dirty_count);
        if let Some(ref t) = status.last_export_time {
            println!("  Last export: {t}");
        }
        if let Some(ref t) = status.last_import_time {
            println!("  Last import: {t}");
        }
        println!("  JSONL exists: {}", status.jsonl_exists);
        println!(
            "  VCS status: not probed (run {})",
            status.git_export.diagnostic_command
        );
        if status.jsonl_newer {
            println!("  Status: JSONL is newer (import recommended)");
        } else if status.db_newer {
            println!("  Status: Database is newer (export recommended)");
        } else if status.coverage_drift {
            if let Some(probe) = &status.coverage {
                println!(
                    "  Status: COVERAGE DRIFT — JSONL has {} unique ids but the database holds {} exportable issues",
                    probe.jsonl_unique_ids, probe.db_exportable_issues
                );
            } else {
                println!("  Status: COVERAGE DRIFT — DB and JSONL hold different issue sets");
            }
            println!(
                "  Recover: `br sync --reconcile --dry-run` (lossless preview) or `br sync --import-only --rebuild` (JSONL-authoritative)"
            );
        } else {
            println!("  Status: In sync");
        }
    }

    Ok(())
}

/// Render sync status with rich formatting.
fn render_status_rich(status: &SyncStatus, ctx: &OutputContext) {
    let _console = Console::default();
    let theme = ctx.theme();

    // Determine sync state and color
    let (state_icon, state_text, state_style) = if status.jsonl_newer {
        (
            "⬇",
            "JSONL is newer (import recommended)",
            theme.info.clone(),
        )
    } else if status.db_newer {
        (
            "⬆",
            "Database is newer (export recommended)",
            theme.warning.clone(),
        )
    } else if status.coverage_drift {
        (
            "✗",
            "Coverage drift: DB and JSONL hold different issue sets (see `br sync --reconcile --dry-run`)",
            theme.error.clone(),
        )
    } else {
        ("✓", "In sync", theme.success.clone())
    };

    // Build status content
    let mut text = Text::new("");

    // State line
    text.append_styled(state_icon, state_style.clone());
    text.append(" ");
    text.append_styled(state_text, state_style);
    text.append("\n\n");

    // Dirty count
    text.append_styled("Dirty issues: ", theme.dimmed.clone());
    if status.dirty_count > 0 {
        text.append_styled(&status.dirty_count.to_string(), theme.warning.clone());
    } else {
        text.append_styled("0", theme.success.clone());
    }
    text.append("\n");

    text.append_styled("VCS status:   ", theme.dimmed.clone());
    text.append_styled("not probed", theme.muted.clone());
    text.append(" (run ");
    text.append_styled(status.git_export.diagnostic_command, theme.accent.clone());
    text.append(")\n");

    // JSONL exists
    text.append_styled("JSONL exists: ", theme.dimmed.clone());
    text.append_styled(
        if status.jsonl_exists { "yes" } else { "no" },
        if status.jsonl_exists {
            theme.success.clone()
        } else {
            theme.muted.clone()
        },
    );
    text.append("\n");

    // Last export time
    if let Some(ref t) = status.last_export_time {
        text.append_styled("Last export:  ", theme.dimmed.clone());
        text.append_styled(t, theme.timestamp.clone());
        text.append("\n");
    }

    // Last import time
    if let Some(ref t) = status.last_import_time {
        text.append_styled("Last import:  ", theme.dimmed.clone());
        text.append_styled(t, theme.timestamp.clone());
        text.append("\n");
    }

    // Content hash (truncated)
    if let Some(ref hash) = status.jsonl_content_hash {
        text.append_styled("Content hash: ", theme.dimmed.clone());
        let display_hash = if hash.len() > 12 {
            format!("{}…", &hash[..12])
        } else {
            hash.clone()
        };
        text.append_styled(&display_hash, theme.muted.clone());
    }

    let panel = Panel::from_rich_text(&text, ctx.width())
        .title(Text::new("Sync Status"))
        .box_style(theme.box_style);
    ctx.render(&panel);
}

/// Execute the --witness operation.
fn execute_witness(
    path_policy: &SyncPathPolicy,
    args: &SyncArgs,
    use_json: bool,
    ctx: &OutputContext,
) -> Result<()> {
    let jsonl_path = &path_policy.jsonl_path;
    if !jsonl_path.is_file() {
        return Err(BeadsError::Config(format!(
            "JSONL file not found: {}",
            jsonl_path.display()
        )));
    }

    let witness_parallelism = effective_witness_parallelism(args);
    let witness =
        build_witness_for_path(jsonl_path, args.witness_chunk_lines, witness_parallelism)?;
    let base_artifacts = build_base_witness_artifacts(
        path_policy,
        args.witness_chunk_lines,
        witness_parallelism,
        &witness,
    )?;
    let result = SyncWitnessResult {
        jsonl_path: jsonl_path.display().to_string(),
        witness,
        base_jsonl_path: base_artifacts.jsonl_path,
        base_comparison: base_artifacts.comparison,
        base_reuse_plan: base_artifacts.reuse_plan,
        base_parallel_work_plan: base_artifacts.parallel_work_plan,
        base_reuse_materialization: base_artifacts.reuse_materialization,
    };

    if !should_render_human_sync_output(ctx, use_json) {
        return Ok(());
    }

    if use_json {
        ctx.json_pretty(&result);
    } else {
        render_witness_text(&result);
    }

    Ok(())
}

fn effective_witness_parallelism(args: &SyncArgs) -> usize {
    args.witness_parallelism
        .unwrap_or(DEFAULT_WITNESS_PARALLELISM)
}

fn build_witness_for_path(
    jsonl_path: &Path,
    chunk_size_lines: usize,
    max_parallelism: usize,
) -> Result<JsonlMerkleWitness> {
    crate::sync::ensure_no_conflict_markers(jsonl_path)?;
    let file = File::open(jsonl_path).map_err(|err| {
        BeadsError::Config(format!(
            "Failed to open JSONL file for witness {}: {err}",
            jsonl_path.display()
        ))
    })?;
    build_jsonl_merkle_witness_parallel(BufReader::new(file), chunk_size_lines, max_parallelism)
        .map_err(|err| {
            BeadsError::Config(format!(
                "Failed to build JSONL witness for {}: {err}",
                jsonl_path.display()
            ))
        })
}

fn build_base_witness_artifacts(
    path_policy: &SyncPathPolicy,
    chunk_size_lines: usize,
    max_parallelism: usize,
    current_witness: &JsonlMerkleWitness,
) -> Result<BaseWitnessArtifacts> {
    let base_jsonl_path = path_policy.beads_dir.join("beads.base.jsonl");
    match fs::symlink_metadata(&base_jsonl_path) {
        Ok(metadata) if metadata.file_type().is_symlink() => {
            return Err(BeadsError::Config(format!(
                "Base JSONL snapshot '{}' must not be a symlink",
                base_jsonl_path.display()
            )));
        }
        Ok(metadata) if !metadata.is_file() => {
            return Err(BeadsError::Config(format!(
                "Base JSONL snapshot '{}' must be a regular file",
                base_jsonl_path.display()
            )));
        }
        Ok(_) => {}
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
            return Ok(BaseWitnessArtifacts {
                jsonl_path: None,
                comparison: None,
                reuse_plan: None,
                parallel_work_plan: None,
                reuse_materialization: None,
            });
        }
        Err(err) => {
            return Err(BeadsError::Config(format!(
                "Failed to inspect base JSONL snapshot '{}': {err}",
                base_jsonl_path.display()
            )));
        }
    }

    require_valid_sync_path(&base_jsonl_path, &path_policy.beads_dir)?;

    let base_witness = build_witness_for_path(&base_jsonl_path, chunk_size_lines, max_parallelism)?;
    let comparison = compare_jsonl_merkle_witnesses(&base_witness, current_witness);
    let reuse_plan = plan_jsonl_witness_reuse(&base_witness, current_witness);
    let parallel_work_plan = plan_jsonl_witness_parallel_work(&reuse_plan, max_parallelism)
        .map_err(|err| BeadsError::Config(format!("Failed to plan JSONL witness work: {err}")))?;
    let reuse_materialization = materialize_base_reuse_plan(
        &base_jsonl_path,
        &path_policy.jsonl_path,
        &base_witness,
        current_witness,
        &reuse_plan,
    )?;

    Ok(BaseWitnessArtifacts {
        jsonl_path: Some(base_jsonl_path.display().to_string()),
        comparison: Some(comparison),
        reuse_plan: Some(reuse_plan),
        parallel_work_plan: Some(parallel_work_plan),
        reuse_materialization: Some(reuse_materialization),
    })
}

fn materialize_base_reuse_plan(
    base_jsonl_path: &Path,
    current_jsonl_path: &Path,
    base_witness: &JsonlMerkleWitness,
    current_witness: &JsonlMerkleWitness,
    reuse_plan: &JsonlWitnessReusePlan,
) -> Result<JsonlWitnessReuseMaterialization> {
    let mut base_file = File::open(base_jsonl_path).map_err(|err| {
        BeadsError::Config(format!(
            "Failed to open base JSONL file for reuse materialization {}: {err}",
            base_jsonl_path.display()
        ))
    })?;
    let mut current_file = File::open(current_jsonl_path).map_err(|err| {
        BeadsError::Config(format!(
            "Failed to open current JSONL file for reuse materialization {}: {err}",
            current_jsonl_path.display()
        ))
    })?;
    let mut sink = std::io::sink();

    materialize_jsonl_witness_reuse_plan(
        &mut base_file,
        &mut current_file,
        &mut sink,
        base_witness,
        current_witness,
        reuse_plan,
    )
    .map_err(|err| BeadsError::Config(format!("Failed to materialize JSONL reuse plan: {err}")))
}

fn render_witness_text(result: &SyncWitnessResult) {
    let witness = &result.witness;
    println!("JSONL Witness:");
    println!("  Path: {}", result.jsonl_path);
    println!("  Schema: {}", witness.schema_version);
    println!("  Lines: {}", witness.line_count);
    println!("  Bytes: {}", witness.byte_count);
    println!("  Chunk size: {} lines", witness.chunk_size_lines);
    println!("  Chunks: {}", witness.chunks.len());
    println!("  Root hash: {}", witness.root_hash);

    if let Some(comparison) = &result.base_comparison {
        if let Some(base_path) = &result.base_jsonl_path {
            println!("  Base path: {base_path}");
        }
        println!(
            "  Base comparison: drift={}, unchanged_chunks={}, changed_chunks={}, added_chunks={}, removed_chunks={}, safe_prefix_chunks={}",
            comparison.drift_detected,
            comparison.unchanged_chunks,
            comparison.changed_chunks,
            comparison.added_chunks,
            comparison.removed_chunks,
            comparison.safe_reuse_prefix_chunks
        );
        if let Some(index) = comparison.first_changed_chunk_index {
            println!("  First changed chunk: {index}");
        }
    }

    if let Some(plan) = &result.base_reuse_plan {
        println!("  Reuse plan actions: {}", plan.actions.len());
    }
    if let Some(plan) = &result.base_parallel_work_plan {
        println!(
            "  Parallel work batches: {} (max_parallelism={})",
            plan.total_batches, plan.max_parallelism
        );
    }
    if let Some(materialization) = &result.base_reuse_materialization {
        println!(
            "  Reuse materialization: output_bytes={}, reused_chunks={}, rebuilt_chunks={}, read_added_chunks={}, dropped_chunks={}",
            materialization.output_byte_count,
            materialization.reused_chunks,
            materialization.rebuilt_chunks,
            materialization.read_added_chunks,
            materialization.dropped_chunks
        );
    }
}

/// Execute the --flush-only (export) operation.
#[allow(clippy::too_many_lines, clippy::too_many_arguments)]
fn execute_flush(
    storage: &mut crate::storage::SqliteStorage,
    _beads_dir: &Path,
    path_policy: &SyncPathPolicy,
    args: &SyncArgs,
    use_json: bool,
    show_progress: bool,
    retention_days: Option<u64>,
    history_config: HistoryConfig,
    retained_source: config::RetainedJsonlSourceRef<'_>,
    expected_source: &JsonlSourceStateWitness,
    retained_authority: Option<&crate::sync::JsonlFamilyWriteLock>,
    ctx: &OutputContext,
) -> Result<Option<Arc<JsonlSourceSnapshot>>> {
    info!("Starting JSONL export");
    let export_policy = parse_export_policy(args)?;
    let jsonl_path = &path_policy.jsonl_path;
    debug!(
        jsonl_path = %jsonl_path.display(),
        external_jsonl = path_policy.is_external,
        export_policy = %export_policy,
        force = args.force,
        ?retention_days,
        "Export configuration resolved"
    );

    let owned_jsonl_authority = retained_authority
        .is_none()
        .then(|| crate::sync::blocking_jsonl_family_write_lock_with_timeout(jsonl_path, None))
        .transpose()?;
    let jsonl_authority = retained_authority
        .or(owned_jsonl_authority.as_ref())
        .expect("JSONL authority is retained or acquired");
    jsonl_authority.verify_jsonl_authority()?;
    let captured_source = jsonl_authority.capture_optional_target()?;
    let source = captured_source.as_ref();
    let observed_source = source.map_or(
        JsonlSourceStateWitness::Missing,
        JsonlSourceSnapshot::state_witness,
    );
    if !matches!(retained_source, config::RetainedJsonlSourceRef::Uncaptured)
        && &observed_source != expected_source
    {
        return Err(BeadsError::SyncConflict {
            message: "Retained JSONL source does not match its startup witness".to_string(),
        });
    }

    // Check for dirty issues
    let dirty_ids = storage.get_dirty_issue_ids()?;
    let needs_flush = storage.get_metadata("needs_flush")?.as_deref() == Some("true");
    let jsonl_exists = source.is_some();
    let db_issue_count = storage.count_issues()?;
    debug!(dirty_count = dirty_ids.len(), "Found dirty issues");

    // Refuse to overwrite a JSONL that still holds unresolved merge-conflict
    // markers. The main flush path below would blow away the `<<<<<<<` /
    // `=======` / `>>>>>>>` regions along with whatever remote side of the
    // merge they contain, silently resolving the conflict in favor of the
    // local DB. Detect the markers up-front so the operator can resolve the
    // merge (or pass `--force` if they actually intend the DB to win).
    if let Some(source) = source
        && !args.force
    {
        ensure_no_conflict_markers_snapshot(source)?;
    }

    // If no dirty issues and no force, report nothing to do
    if dirty_ids.is_empty() && !needs_flush && jsonl_exists && !args.force {
        // `ensure_no_conflict_markers` ran above before we got here, so
        // `analyze_jsonl_snapshot` below won't trip over unresolved `<<<<<<<` /
        // `=======` / `>>>>>>>` lines.

        // Guard against stale DB state without parsing the JSONL twice for count
        // and IDs.
        let (existing_count, jsonl_ids) =
            analyze_jsonl_snapshot(source.expect("existing JSONL has an immutable snapshot"))?;
        if existing_count > 0 && db_issue_count == 0 {
            warn!(
                jsonl_count = existing_count,
                "Refusing export of empty DB over non-empty JSONL"
            );
            return Err(BeadsError::Config(format!(
                "Refusing to export empty database over non-empty JSONL file.\n\
                     Database has 0 issues, JSONL has {existing_count} issues.\n\
                     This would result in data loss!\n\
                     Hint: Use --force to override this safety check."
            )));
        }

        if !jsonl_ids.is_empty() {
            let db_ids: HashSet<String> = storage.get_all_ids()?.into_iter().collect();
            let purged_ids = storage.get_purged_ids_pending_export()?;
            let mut missing_list = jsonl_ids
                .difference(&db_ids)
                .filter(|id| !purged_ids.contains(id.as_str()))
                .cloned()
                .collect::<Vec<_>>();

            if !missing_list.is_empty() {
                missing_list.sort();
                let display_count = missing_list.len().min(10);
                let preview = missing_list
                    .iter()
                    .take(display_count)
                    .map(String::as_str)
                    .collect::<Vec<_>>()
                    .join(", ");
                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.",
                    db_issue_count,
                    jsonl_ids.len(),
                    missing_list.len(),
                    preview,
                    more
                )));
            }
        }

        // Even with nothing to export, maintain the merge anchor from the
        // clean JSONL (issues #378/#394): a missing anchor is materialized
        // and a stale anchor is replaced, making `br sync --flush-only` an
        // idempotent recovery command for the doctor's
        // `base_jsonl.missing_post_flush` / stale-anchor findings.
        //
        // IDs/counts alone cannot prove that a same-ID JSONL record still
        // matches the database. Immediately before touching the merge
        // anchor, require the last certified semantic hash and compare it
        // directly with the captured snapshot. Missing or mismatched
        // evidence fails closed; --force deliberately takes the real-export
        // path instead.
        let noop_source = source.expect("existing JSONL has an immutable snapshot");
        match storage
            .get_metadata(METADATA_JSONL_CONTENT_HASH)?
            .filter(|hash| !hash.trim().is_empty())
        {
            Some(stored_content_hash) => {
                let observed_content_hash = noop_source.content_sha256();
                if observed_content_hash != stored_content_hash {
                    return Err(BeadsError::Config(format!(
                        "Refusing a no-op flush because the JSONL changed since its last certified \
                         sync (stored hash {stored_content_hash}, observed hash \
                         {observed_content_hash}). The merge anchor was not changed.\n\
                         Inspect/reconcile with `br sync --merge`, accept the JSONL intentionally with \
                         `br sync --import-only --force`, or replace it from the database with \
                         `br sync --flush-only --force`."
                    )));
                }
            }
            // A fresh workspace has no cached content hash yet. When BOTH
            // sides are globally empty (zero JSONL records, zero DB issues)
            // there is nothing a stored hash could disagree about, so the
            // no-op flush is certifiable directly (beads_rust-a6kl /
            // GitHub #472). Any non-empty state without a cached hash stays
            // fail-closed exactly as before.
            None if existing_count == 0 && db_issue_count == 0 => {}
            None => {
                return Err(BeadsError::Config(
                    "Cannot certify a no-op flush because the stored JSONL content hash is \
                     missing. The merge anchor was not changed.\n\
                     Inspect/reconcile with `br sync --merge`, accept the JSONL intentionally \
                     with `br sync --import-only --force`, or replace it from the database with \
                     `br sync --flush-only --force`."
                        .to_string(),
                ));
            }
        }

        // Certified: ensure the anchor holds the exact snapshot bytes (also
        // covers the missing-anchor case). A byte-identical regular-file
        // anchor is left untouched so an idempotent no-op flush keeps its
        // inode; anything else (missing, symlinked, byte-divergent — even
        // whitespace-only drift the content hash cannot see) is replaced
        // with the exact snapshot bytes.
        let anchor_path = path_policy.beads_dir.join("beads.base.jsonl");
        let snapshot_bytes = {
            let mut bytes = Vec::with_capacity(usize::try_from(noop_source.size()).unwrap_or(0));
            std::io::copy(&mut noop_source.reader(), &mut bytes).map_err(BeadsError::Io)?;
            bytes
        };
        let anchor_is_exact = fs::symlink_metadata(&anchor_path)
            .map(|meta| meta.is_file())
            .unwrap_or(false)
            && fs::read(&anchor_path).is_ok_and(|bytes| bytes == snapshot_bytes);
        if !anchor_is_exact {
            refresh_base_snapshot_from_flushed_jsonl_snapshot(noop_source, &path_policy.beads_dir)?;
        }

        if use_json {
            let result = FlushResult {
                exported_issues: 0,
                exported_dependencies: 0,
                exported_labels: 0,
                exported_comments: 0,
                content_hash: String::new(),
                cleared_dirty: 0,
                policy: export_policy,
                success_rate: 1.0,
                errors: Vec::new(),
                manifest_path: None,
                publication_atomicity: None,
            };
            ctx.json_pretty(&result);
        } else if should_render_human_sync_output(ctx, use_json) {
            println!("Nothing to export (no dirty issues)");
        }
        return Ok(None);
    }

    // Configure export. `needs_flush` must NOT be conflated with the user's
    // explicit `--force`: its only job is to bypass the nothing-to-do early
    // return above so a re-export happens when the DB holds canonical
    // content. Passing it as `force` disabled the exporter's data-loss
    // guards, letting a post-merge flush silently destroy JSONL issues the
    // DB had never imported (#405). Intentional purges are excluded from the
    // guard via the purged-pending-export marker instead.
    let export_config = ExportConfig {
        force: args.force,
        is_default_path: true,
        error_policy: export_policy,
        retention_days,
        export_as_of: None,
        beads_dir: Some(path_policy.beads_dir.clone()),
        allow_external_jsonl: path_policy.allow_external_jsonl,
        show_progress,
        history: history_config,
        max_parallel_workers: args.export_parallelism.unwrap_or(0),
        expected_staged_output: None,
    };

    // Execute export
    info!(path = %jsonl_path.display(), "Writing issues.jsonl");
    let (export_result, report) = export_to_jsonl_with_policy_expected_under_authority(
        storage,
        jsonl_path,
        &export_config,
        source.map_or(
            ExpectedJsonlSourceRef::Missing,
            ExpectedJsonlSourceRef::Present,
        ),
        jsonl_authority,
    )?;
    let published_source = export_result.published_source_arc()?;
    debug!(
        issues_exported = report.issues_exported,
        dependencies_exported = report.dependencies_exported,
        labels_exported = report.labels_exported,
        comments_exported = report.comments_exported,
        errors = report.errors.len(),
        "Export completed"
    );

    debug!(
        issues = export_result.exported_count,
        "Exported issues to JSONL"
    );

    // A clean flush leaves DB == JSONL, so the JSONL that just reached disk
    // is the new common state future 3-way merges should diff against.
    // Refresh the merge anchor to match (issue #378): historically only the
    // merge path wrote `beads.base.jsonl`, leaving flush-only workspaces
    // permanently anchor-less and tripping the doctor's
    // `base_jsonl.missing_post_flush` warning while `br sync --status`
    // reported "In sync". Skip when the export had per-record errors — a
    // partial export must not become the merge base. Publish the anchor
    // BEFORE clearing dirty/export metadata so an anchor publication failure
    // keeps the workspace dirty and remains recoverable by a later explicit
    // flush: certifying "In sync" over a stale anchor would hand future
    // 3-way merges the wrong ancestor. (This fail-closed ordering shipped in
    // 5414143b alongside the failure-injection coverage; the 77ae88ff
    // tree-preference merge silently reverted it to the older best-effort
    // wording while keeping the test.)
    if !report.has_errors() {
        refresh_base_snapshot_from_flushed_jsonl_snapshot(
            export_result.published_source()?,
            &path_policy.beads_dir,
        )
        .map_err(|source| BeadsError::WithContext {
            context: format!(
                "Failed to publish the merge anchor {} for the flushed JSONL; \
                 dirty/export metadata was retained so this flush can be retried",
                path_policy.beads_dir.join("beads.base.jsonl").display()
            ),
            source: Box::new(source),
        })?;
    }

    // Finalize export (clear dirty flags, update metadata) only after a
    // clean export's anchor is durable.
    finalize_export_under_authority(
        storage,
        &export_result,
        Some(&export_result.issue_hashes),
        jsonl_path,
        jsonl_authority,
    )
    .map_err(|source| BeadsError::CommittedStateUnwitnessed {
        operation: "flush JSONL export finalization".to_string(),
        source: Box::new(source),
    })?;
    info!("Export complete, cleared dirty flags");

    // Write manifest if requested (atomic: temp + fsync + durable_rename)
    let manifest_path = if args.manifest {
        let manifest = serde_json::json!({
            "export_time": chrono::Utc::now().to_rfc3339(),
            "issues_count": export_result.exported_count,
            "content_hash": export_result.content_hash,
            "exported_ids": export_result.exported_ids,
            "policy": report.policy_used,
            "errors": &report.errors,
        });
        let manifest_file = path_policy.manifest_path.clone();
        require_safe_sync_overwrite_path(
            &manifest_file,
            &path_policy.beads_dir,
            path_policy.allow_external_jsonl,
            "write manifest",
        )?;
        let manifest_publication =
            write_manifest_atomically(&manifest_file, &manifest).map_err(|source| {
                BeadsError::CommittedArtifactFailure {
                    operation: "flush".to_string(),
                    primary_path: jsonl_path.clone(),
                    artifact_path: manifest_file.clone(),
                    source: Box::new(source),
                }
            })?;
        if !manifest_publication.cleanup_durable() {
            warn!(
                manifest_path = %manifest_file.display(),
                recovery_path = manifest_publication.retained_recovery_path(),
                "Manifest reached its verified destination, but displaced-generation cleanup was not certified durable"
            );
        }
        Some(manifest_file.to_string_lossy().to_string())
    } else {
        None
    };

    // Output result
    let cleared_dirty = export_result.exported_marked_at.len();
    // Only a downgraded publication is worth a field: the atomic protocol is
    // the documented default, and keeping the field absent leaves every
    // existing `--json` consumer and golden untouched (#419).
    let publication_atomicity = export_result
        .publication
        .as_ref()
        .map(crate::sync::ExportPublicationReceipt::atomicity)
        .filter(|atomicity| atomicity.is_downgraded())
        .map(|atomicity| atomicity.as_str().to_string());
    let result = FlushResult {
        exported_issues: report.issues_exported,
        exported_dependencies: report.dependencies_exported,
        exported_labels: report.labels_exported,
        exported_comments: report.comments_exported,
        content_hash: export_result.content_hash,
        cleared_dirty,
        policy: report.policy_used,
        success_rate: report.success_rate(),
        errors: report.errors.clone(),
        manifest_path,
        publication_atomicity,
    };

    if use_json {
        ctx.json_pretty(&result);
    } else if !should_render_human_sync_output(ctx, use_json) {
        return Ok(Some(published_source));
    } else if ctx.is_rich() {
        render_flush_result_rich(&result, &report.errors, ctx);
    } else {
        if report.policy_used != ExportErrorPolicy::Strict || report.has_errors() {
            println!("Export completed with policy: {}", report.policy_used);
        }
        println!("Exported:");
        println!(
            "  {} issue{}",
            result.exported_issues,
            if result.exported_issues == 1 { "" } else { "s" }
        );
        println!(
            "  {} dependenc{}{}",
            result.exported_dependencies,
            if result.exported_dependencies == 1 {
                "y"
            } else {
                "ies"
            },
            format_error_suffix(&report.errors, ExportEntityType::Dependency)
        );
        println!(
            "  {} label{}{}",
            result.exported_labels,
            if result.exported_labels == 1 { "" } else { "s" },
            format_error_suffix(&report.errors, ExportEntityType::Label)
        );
        println!(
            "  {} comment{}{}",
            result.exported_comments,
            if result.exported_comments == 1 {
                ""
            } else {
                "s"
            },
            format_error_suffix(&report.errors, ExportEntityType::Comment)
        );

        if result.cleared_dirty > 0 {
            println!(
                "Cleared dirty flag for {} issue{}",
                result.cleared_dirty,
                if result.cleared_dirty == 1 { "" } else { "s" }
            );
        }
        if let Some(ref path) = result.manifest_path {
            println!("Wrote manifest to {path}");
        }
        if let Some(ref atomicity) = result.publication_atomicity {
            println!(
                "Publication downgraded to {atomicity}: this filesystem does not support \
                 flagged rename, so the JSONL was installed with a witness-checked plain \
                 rename under the write lock"
            );
        }
        if report.has_errors() {
            println!();
            println!("Errors ({}):", report.errors.len());
            for err in &report.errors {
                println!("  {}", err.summary());
            }
        }
    }

    Ok(Some(published_source))
}

fn create_temp_manifest_file(manifest_path: &Path) -> Result<(PathBuf, File)> {
    let pid = std::process::id();

    for attempt in 0..64_u32 {
        let extension = if attempt == 0 {
            format!("json.{pid}.tmp")
        } else {
            format!("json.{pid}.{attempt}.tmp")
        };
        let temp_path = manifest_path.with_extension(extension);
        let mut options = OpenOptions::new();
        options.write(true).create_new(true);
        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt;
            options.mode(0o600);
        }
        match options.open(&temp_path) {
            Ok(file) => return Ok((temp_path, file)),
            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
                if fs::symlink_metadata(&temp_path)
                    .is_ok_and(|metadata| metadata.file_type().is_symlink())
                {
                    return Err(BeadsError::Config(format!(
                        "failed to create temp manifest file {}: {error}",
                        temp_path.display()
                    )));
                }
            }
            Err(error) => {
                return Err(BeadsError::Config(format!(
                    "failed to create temp manifest file {}: {error}",
                    temp_path.display()
                )));
            }
        }
    }

    Err(BeadsError::Config(format!(
        "failed to allocate temp manifest file for {}",
        manifest_path.display()
    )))
}

fn write_manifest_atomically(
    manifest_path: &Path,
    manifest: &serde_json::Value,
) -> Result<crate::sync::ExportPublicationReceipt> {
    use std::io::Write;

    let content = serde_json::to_string_pretty(manifest)?;

    let (temp_path, mut file) = create_temp_manifest_file(manifest_path)?;

    // Close the handle before attempting cleanup so Windows does not retain a
    // torn temp solely because the writer still had the file open.
    if let Err(error) = file.write_all(content.as_bytes()) {
        drop(file);
        let _ = fs::remove_file(&temp_path);
        return Err(BeadsError::Io(error));
    }
    if let Err(error) = file.sync_all() {
        drop(file);
        let _ = fs::remove_file(&temp_path);
        return Err(BeadsError::Io(error));
    }
    drop(file);

    crate::sync::publish_staged_file_conditionally(&temp_path, manifest_path)
}

/// Render flush (export) result with rich formatting.
fn render_flush_result_rich(result: &FlushResult, errors: &[ExportError], ctx: &OutputContext) {
    let _console = Console::default();
    let theme = ctx.theme();

    let mut text = Text::new("");

    // Success indicator
    if errors.is_empty() {
        text.append_styled("✓ ", theme.success.clone());
        text.append_styled("Export Complete", theme.success.clone());
    } else {
        text.append_styled("⚠ ", theme.warning.clone());
        text.append_styled("Export Complete (with errors)", theme.warning.clone());
    }
    text.append("\n\n");

    // Direction indicator
    text.append_styled("Direction     ", theme.dimmed.clone());
    text.append_styled("SQLite → JSONL", theme.info.clone());
    text.append("\n");

    // Exported counts
    text.append_styled("Issues        ", theme.dimmed.clone());
    text.append_styled(&result.exported_issues.to_string(), theme.accent.clone());
    text.append("\n");

    text.append_styled("Dependencies  ", theme.dimmed.clone());
    text.append(&result.exported_dependencies.to_string());
    text.append("\n");

    text.append_styled("Labels        ", theme.dimmed.clone());
    text.append(&result.exported_labels.to_string());
    text.append("\n");

    text.append_styled("Comments      ", theme.dimmed.clone());
    text.append(&result.exported_comments.to_string());
    text.append("\n");

    // Dirty flags cleared
    if result.cleared_dirty > 0 {
        text.append_styled("Dirty cleared ", theme.dimmed.clone());
        text.append_styled(&result.cleared_dirty.to_string(), theme.success.clone());
        text.append("\n");
    }

    // Content hash (truncated)
    if !result.content_hash.is_empty() {
        text.append("\n");
        text.append_styled("Content hash  ", theme.dimmed.clone());
        let display_hash = if result.content_hash.len() > 12 {
            format!("{}…", &result.content_hash[..12])
        } else {
            result.content_hash.clone()
        };
        text.append_styled(&display_hash, theme.muted.clone());
    }

    // Manifest path
    if let Some(ref path) = result.manifest_path {
        text.append("\n");
        text.append_styled("Manifest      ", theme.dimmed.clone());
        text.append_styled(path, theme.muted.clone());
    }

    // Non-atomic publication downgrade (#419)
    if let Some(ref atomicity) = result.publication_atomicity {
        text.append("\n");
        text.append_styled("Publication   ", theme.dimmed.clone());
        text.append_styled(
            &format!("{atomicity} (filesystem lacks flagged rename)"),
            theme.warning.clone(),
        );
    }

    let panel = Panel::from_rich_text(&text, ctx.width())
        .title(Text::new("Flush (Export)"))
        .box_style(theme.box_style);
    ctx.render(&panel);

    // Errors section if any
    if !errors.is_empty() {
        ctx.newline();
        render_errors_rich(errors, ctx);
    }
}

/// Render export errors with rich formatting.
fn render_errors_rich(errors: &[ExportError], ctx: &OutputContext) {
    let _console = Console::default();
    let theme = ctx.theme();

    let mut text = Text::new("");
    text.append_styled(
        &format!("{} error(s) during export:\n\n", errors.len()),
        theme.error.clone(),
    );

    for (i, err) in errors.iter().enumerate() {
        let prefix = if i == errors.len() - 1 {
            "└──"
        } else {
            "├──"
        };
        text.append_styled(prefix, theme.muted.clone());
        text.append(" ");
        text.append_styled(&err.summary(), theme.error.clone());
        text.append("\n");
    }

    let panel = Panel::from_rich_text(&text, ctx.width())
        .title(Text::new("⚠ Errors"))
        .box_style(theme.box_style);
    ctx.render(&panel);
}

fn parse_export_policy(args: &SyncArgs) -> Result<ExportErrorPolicy> {
    args.error_policy.as_deref().map_or_else(
        || Ok(ExportErrorPolicy::Strict),
        |value| {
            value.parse().map_err(|message| BeadsError::Validation {
                field: "error_policy".to_string(),
                reason: message,
            })
        },
    )
}

fn format_error_suffix(errors: &[ExportError], entity: ExportEntityType) -> String {
    let count = errors
        .iter()
        .filter(|err| err.entity_type == entity)
        .count();
    if count > 0 {
        format!(" ({count} error{})", if count == 1 { "" } else { "s" })
    } else {
        String::new()
    }
}

fn should_show_progress(json: bool, quiet: bool) -> bool {
    !json && !quiet && std::io::stdout().is_terminal()
}

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

fn push_cli_rerun_overrides(rerun: &mut Vec<String>, cli: &config::CliOverrides) {
    if cli.json == Some(true) {
        rerun.push("--json".to_string());
    }
    if cli.quiet == Some(true) {
        rerun.push("--quiet".to_string());
    }
    // Preserve `--no-color` so the re-run inherits the caller's output
    // preference; dropping it silently flips colorized output back on.
    if cli.display_color == Some(false) {
        rerun.push("--no-color".to_string());
    }
    // Preserve `--actor` so audit-log entries from the re-run carry the
    // same identity the operator originally specified.
    if let Some(actor) = &cli.actor {
        rerun.push("--actor".to_string());
        rerun.push(shell_quote(actor));
    }
    if cli.allow_stale == Some(true) {
        rerun.push("--allow-stale".to_string());
    }
    if cli.no_daemon == Some(true) {
        rerun.push("--no-daemon".to_string());
    }
    if cli.no_auto_import == Some(true) {
        rerun.push("--no-auto-import".to_string());
    }
    if cli.no_auto_flush == Some(true) {
        rerun.push("--no-auto-flush".to_string());
    }
    if let Some(timeout) = cli.lock_timeout {
        rerun.push("--lock-timeout".to_string());
        rerun.push(timeout.to_string());
    }
}

fn integrity_check_is_clean(messages: &[String]) -> bool {
    matches!(messages, [message] if message.trim().eq_ignore_ascii_case("ok"))
}

fn fresh_force_import_maintenance_gate_applies(
    args: &SyncArgs,
    force_import_target_was_empty: bool,
    import_rewrote_storage: bool,
) -> bool {
    args.force
        && !args.rebuild
        && !args.rename_prefix
        && force_import_target_was_empty
        && import_rewrote_storage
}

#[allow(clippy::too_many_arguments)]
fn replace_database_from_jsonl_snapshot(
    storage: &mut crate::storage::SqliteStorage,
    beads_dir: &Path,
    cli: &config::CliOverrides,
    source: &JsonlSourceSnapshot,
    jsonl_authority: &crate::sync::JsonlFamilyWriteLock,
    db_path: &Path,
    import_config: &ImportConfig,
    target_prefix: Option<&str>,
    preserved_tombstones: &[crate::sync::PreservedIssue],
) -> Result<ImportResult> {
    let startup = config::load_startup_config_with_paths(beads_dir, cli.db.as_ref())?;
    let mut bootstrap_layer = startup.merged_config;
    if let Some(prefix) = target_prefix {
        bootstrap_layer
            .runtime
            .insert("issue_prefix".to_string(), prefix.to_string());
    }

    // Close the old connection before the verified backup-and-replace path
    // opens a new database at the same location.
    let placeholder = crate::storage::SqliteStorage::open_memory()?;
    let previous_storage = std::mem::replace(storage, placeholder);
    drop(previous_storage);

    let recovery =
        if let Some(authority) = cli.database_family_write_authority_for(beads_dir, db_path) {
            config::repair_database_from_jsonl_snapshot_with_import_config_under_write_authority(
                beads_dir,
                db_path,
                cli.lock_timeout,
                &bootstrap_layer,
                import_config.clone(),
                source,
                jsonl_authority,
                authority,
            )
        } else {
            config::repair_database_from_jsonl_snapshot_with_import_config(
                beads_dir,
                db_path,
                cli.lock_timeout,
                &bootstrap_layer,
                import_config.clone(),
                source,
                jsonl_authority,
            )
        };

    let (mut rebuilt_storage, import_result, _) = match recovery {
        Ok(rebuilt) => rebuilt,
        Err(error) => {
            let reopened = cli
                .database_family_write_authority_for(beads_dir, db_path)
                .map_or_else(
                    || crate::storage::SqliteStorage::open_with_timeout(db_path, cli.lock_timeout),
                    |authority| {
                        crate::storage::SqliteStorage::open_with_timeout_under_write_authority(
                            db_path,
                            cli.lock_timeout,
                            authority,
                        )
                    },
                );
            if let Ok(reopened) = reopened {
                *storage = reopened;
            }
            return Err(error);
        }
    };
    restore_tombstones_after_rebuild(&mut rebuilt_storage, preserved_tombstones)?;
    *storage = rebuilt_storage;
    Ok(import_result)
}

fn repair_import_integrity_if_needed(
    storage: &mut crate::storage::SqliteStorage,
    beads_dir: &Path,
    cli: &config::CliOverrides,
    source: &JsonlSourceSnapshot,
    jsonl_authority: &crate::sync::JsonlFamilyWriteLock,
    db_path: &Path,
    import_config: &ImportConfig,
) -> Result<()> {
    let messages = storage.integrity_check_messages()?;
    if integrity_check_is_clean(&messages) {
        return Ok(());
    }

    warn!(
        db_path = %db_path.display(),
        integrity_messages = ?messages,
        "Post-import maintenance left SQLite integrity warnings; rebuilding DB from JSONL with original import semantics"
    );

    let jsonl_filter = scan_jsonl_snapshot_for_tombstone_filter(source)?;
    let preserved_tombstones =
        tombstones_missing_from_jsonl_tombstones(snapshot_tombstones(storage), &jsonl_filter);

    replace_database_from_jsonl_snapshot(
        storage,
        beads_dir,
        cli,
        source,
        jsonl_authority,
        db_path,
        import_config,
        None,
        &preserved_tombstones,
    )?;
    Ok(())
}

fn auto_rebuild_semantic_flag_conflict_reason(
    args: &SyncArgs,
    cli: &config::CliOverrides,
    db_path: Option<&Path>,
) -> Option<String> {
    if !args.rename_prefix {
        return None;
    }

    let mut rerun = vec!["br".to_string()];
    if let Some(path) = db_path {
        rerun.push("--db".to_string());
        rerun.push(shell_quote(&path.display().to_string()));
    }
    push_cli_rerun_overrides(&mut rerun, cli);
    rerun.push("sync".to_string());
    rerun.push("--import-only".to_string());
    if args.allow_external_jsonl {
        rerun.push("--allow-external-jsonl".to_string());
    }
    if args.force {
        rerun.push("--force".to_string());
    }
    if args.rebuild {
        rerun.push("--rebuild".to_string());
    }
    rerun.push("--rename-prefix".to_string());

    Some(format!(
        "Open-time recovery rebuilt the database before import, so the requested import semantics (`--rename-prefix`) were not applied. Re-run `{}` now that the DB is healthy.",
        rerun.join(" ")
    ))
}

fn auto_rebuild_semantic_conflict_field(args: &SyncArgs) -> &'static str {
    if args.rebuild {
        "rebuild"
    } else if args.force {
        "force"
    } else {
        "rename_prefix"
    }
}

fn jsonl_contains_prefix_mismatch(
    source: &JsonlSourceSnapshot,
    expected_prefix: &str,
) -> Result<bool> {
    for issue in read_issues_from_jsonl_snapshot(source)? {
        if issue.status == crate::model::Status::Tombstone {
            continue;
        }
        if !id_matches_expected_prefix(&issue.id, expected_prefix) {
            return Ok(true);
        }
    }
    Ok(false)
}

fn jsonl_contains_duplicate_external_refs(source: &JsonlSourceSnapshot) -> Result<bool> {
    let mut seen_external_refs = HashSet::new();
    for issue in read_issues_from_jsonl_snapshot(source)? {
        if let Some(external_ref) = issue.external_ref
            && !seen_external_refs.insert(external_ref)
        {
            return Ok(true);
        }
    }
    Ok(false)
}

fn emit_auto_rebuild_import_result(
    storage: &crate::storage::SqliteStorage,
    use_json: bool,
    ctx: &OutputContext,
) -> Result<()> {
    let created = storage.count_all_issues()?;
    let result = ImportResultOutput {
        created,
        updated: 0,
        skipped: 0,
        tombstone_skipped: 0,
        orphans_removed: 0,
        blocked_cache_rebuilt: true,
        exact_duplicate_comments_deduplicated: 0,
        prefix_renames: Vec::new(),
        salvage: None,
    };
    if use_json {
        ctx.json_pretty(&result);
    } else if should_render_human_sync_output(ctx, use_json) {
        if ctx.is_rich() {
            render_import_result_rich(&result, ctx);
        } else {
            println!("Imported from JSONL (via automatic recovery):");
            println!("  Created: {} issues", result.created);
            println!("  Rebuilt blocked cache");
        }
    }
    Ok(())
}

/// Execute the --import-only operation.
#[allow(clippy::too_many_lines, clippy::too_many_arguments)]
fn execute_import(
    storage: &mut crate::storage::SqliteStorage,
    beads_dir: &std::path::Path,
    cli: &config::CliOverrides,
    path_policy: &SyncPathPolicy,
    args: &SyncArgs,
    use_json: bool,
    show_progress: bool,
    auto_rebuilt: bool,
    retained_source: config::RetainedJsonlSourceRef<'_>,
    retained_authority: Option<&crate::sync::JsonlFamilyWriteLock>,
    db_path: &std::path::Path,
    ctx: &OutputContext,
) -> Result<Option<Arc<JsonlSourceSnapshot>>> {
    info!("Starting JSONL import");
    let jsonl_path = &path_policy.jsonl_path;
    debug!(
        jsonl_path = %jsonl_path.display(),
        external_jsonl = path_policy.is_external,
        force = args.force,
        auto_rebuilt,
        "Import configuration resolved"
    );

    // A missing import source is a no-op, but every present source is captured
    // exactly once before prefix inference, validation, tombstone filtering, or
    // any SQLite mutation. The retained JSONL-family authority serializes
    // cooperating writers for the full import; the immutable bytes also make an
    // out-of-band path replacement harmless to internal pass consistency.
    let owned_jsonl_authority = retained_authority
        .is_none()
        .then(|| {
            crate::sync::blocking_jsonl_family_write_lock_with_timeout(jsonl_path, cli.lock_timeout)
        })
        .transpose()?;
    let jsonl_authority = retained_authority
        .or(owned_jsonl_authority.as_ref())
        .expect("JSONL authority is retained or acquired");
    jsonl_authority.verify_jsonl_authority()?;
    let captured_source = jsonl_authority.capture_optional_target()?;
    let source = captured_source.as_ref();
    match retained_source {
        config::RetainedJsonlSourceRef::Present(retained)
            if source.map(JsonlSourceSnapshot::state_witness) != Some(retained.state_witness()) =>
        {
            return Err(BeadsError::SyncConflict {
                message:
                    "Authority-pinned import source differs from its retained startup snapshot"
                        .to_string(),
            });
        }
        config::RetainedJsonlSourceRef::Missing if source.is_some() => {
            return Err(BeadsError::SyncConflict {
                message: "JSONL appeared after startup captured a missing import source"
                    .to_string(),
            });
        }
        _ => {}
    }
    let Some(source) = source else {
        jsonl_authority.verify_jsonl_authority()?;
        if jsonl_authority.capture_optional_target()?.is_some() {
            return Err(BeadsError::SyncConflict {
                message: "JSONL appeared after import captured a missing source".to_string(),
            });
        }
        warn!(path = %jsonl_path.display(), "JSONL path missing, skipping import");
        if use_json {
            let result = ImportResultOutput {
                created: 0,
                updated: 0,
                skipped: 0,
                tombstone_skipped: 0,
                orphans_removed: 0,
                blocked_cache_rebuilt: false,
                exact_duplicate_comments_deduplicated: 0,
                prefix_renames: Vec::new(),
                salvage: None,
            };
            ctx.json_pretty(&result);
        } else if should_render_human_sync_output(ctx, use_json) {
            println!("No JSONL file found at {}", jsonl_path.display());
        }
        return Ok(None);
    };
    let mut published_salvage_source = None;
    let mut salvage_receipt = None;
    if args.skip_invalid_records
        && let Some(salvage) = salvage_invalid_jsonl_records_under_authority(
            source,
            jsonl_path,
            &path_policy.beads_dir,
            path_policy.allow_external_jsonl,
            jsonl_authority,
        )?
    {
        salvage_receipt = Some(salvage.receipt);
        published_salvage_source = Some(salvage.source);
    }
    let source = published_salvage_source.as_deref().unwrap_or(source);
    let source_content_hash = compute_jsonl_snapshot_content_hash(source)?;

    // If the storage was just rebuilt from JSONL during the open sequence
    // (either the DB file did not exist or a recoverable anomaly triggered
    // `rebuild_database_from_jsonl`), the DB is already a clean import of the
    // JSONL. Re-running `--rebuild`/`--force` here would redo the import and
    // trigger fsqlite's stale-pager OpenRead bug ("could not open storage
    // cursor on root page N") because `reset_data_tables` + bulk INSERT within
    // the fresh connection exercises exactly the code path that just ran.
    // Prefix is a default for newly generated IDs, not a project-wide import
    // invariant. Only compute an expected prefix when the caller explicitly
    // asked to rename imported IDs into the configured prefix.
    let target_prefix = if args.rename_prefix {
        let layer = config::load_config_with_external_jsonl_policy_snapshot(
            beads_dir,
            Some(storage),
            cli,
            path_policy.allow_external_jsonl,
            source,
        )?;
        let id_cfg = config::id_config_from_layer(&layer);
        Some(if id_cfg.prefix == "br" {
            // Prefix is still the default — check if we should auto-detect from JSONL
            let db_prefix = storage.get_config("issue_prefix")?;
            if let Some(p) = db_prefix {
                p
            } else if let Some(detected) = detect_prefix_from_jsonl(source)? {
                info!(detected_prefix = %detected, "Auto-detected prefix from JSONL (no prefix configured)");
                // Persist the detected prefix to config for future operations
                storage.set_config("issue_prefix", &detected)?;
                detected
            } else {
                "br".to_string()
            }
        } else {
            // Config layer resolved a non-default prefix — use it
            id_cfg.prefix
        })
    } else {
        None
    };

    // When the caller requested semantics that auto-recovery could not honor
    // (`--rename-prefix`) *and* the JSONL actually contains mismatched IDs
    // that would have been renamed, fail explicitly so the operator can re-run
    // on the now-healthy DB. If the flag would have been a no-op, preserve the
    // happy-path short-circuit because the rebuild is already done. Skip the
    // whole check when there is no rename request (`target_prefix.is_none()`)
    // so we avoid the disk-touching `resolve_paths` call on the common path.
    let auto_rebuilt_source_is_current = auto_rebuilt
        && (matches!(retained_source, config::RetainedJsonlSourceRef::Present(_))
            || storage
                .get_metadata(METADATA_JSONL_CONTENT_HASH)?
                .as_deref()
                == Some(source_content_hash.as_str()));
    if auto_rebuilt && !auto_rebuilt_source_is_current {
        warn!(
            path = %jsonl_path.display(),
            "JSONL changed after automatic recovery; continuing with an explicit import of the newly captured source"
        );
    }
    let rename_semantics_were_skipped = auto_rebuilt_source_is_current
        && target_prefix.as_deref().is_some_and(|prefix| {
            jsonl_contains_prefix_mismatch(source, prefix).unwrap_or(true)
                || jsonl_contains_duplicate_external_refs(source).unwrap_or(true)
        });
    if rename_semantics_were_skipped {
        let rerun_db_path = config::resolve_paths(beads_dir, None)
            .ok()
            .filter(|paths| paths.db_path != *db_path)
            .map(|_| db_path);
        if let Some(reason) = auto_rebuild_semantic_flag_conflict_reason(args, cli, rerun_db_path) {
            return Err(BeadsError::Validation {
                field: auto_rebuild_semantic_conflict_field(args).to_string(),
                reason,
            });
        }
    }

    if auto_rebuilt_source_is_current {
        info!(
            force = args.force,
            rebuild = args.rebuild,
            "Skipping import body: database was rebuilt from JSONL during open"
        );
        crate::sync::verify_jsonl_source_snapshot_current(source, jsonl_authority)?;
        emit_auto_rebuild_import_result(storage, use_json, ctx)?;
        return Ok(published_salvage_source);
    }

    // Check staleness (unless --force or --rebuild)
    if !args.force && !args.rebuild {
        let last_import_time = storage.get_metadata(METADATA_LAST_IMPORT_TIME)?;
        let stored_hash = storage.get_metadata(METADATA_JSONL_CONTENT_HASH)?;

        if let (Some(import_time), Some(stored)) = (last_import_time, stored_hash) {
            // Check if JSONL content hash matches
            let current_hash = source_content_hash.clone();
            let coverage_probe = if current_hash == stored {
                // Coverage invariant (`beads_rust-jdmh`): a matching stored
                // hash proves the JSONL bytes are unchanged since the last
                // *recorded* import, not that this DB ingested them. If the
                // exportable DB issue count disagrees with the JSONL's
                // unique id count, the shortcut would assert health over a
                // partial/lost import — fall through to the real (additive,
                // never-deleting) import body instead.
                jsonl_unique_id_count(source.reader()).map(|jsonl_unique_ids| {
                    storage
                        .count_exportable_issues()
                        .map(|db_exportable_issues| SyncCoverageProbe {
                            db_exportable_issues,
                            jsonl_unique_ids,
                        })
                })
            } else {
                None
            };
            let coverage_drift = matches!(
                coverage_probe,
                Some(Ok(ref probe)) if probe.drifted()
            );
            if coverage_drift && let Some(Ok(probe)) = &coverage_probe {
                warn!(
                    db_exportable_issues = probe.db_exportable_issues,
                    jsonl_unique_ids = probe.jsonl_unique_ids,
                    "Stored-hash shortcut rejected: DB does not cover the JSONL issue set; running full import"
                );
            }
            if current_hash == stored && !coverage_drift {
                debug!(
                    path = %jsonl_path.display(),
                    last_import = %import_time,
                    "JSONL is current, skipping import"
                );

                crate::sync::verify_jsonl_source_snapshot_current(source, jsonl_authority)?;
                if use_json {
                    let result = ImportResultOutput {
                        created: 0,
                        updated: 0,
                        skipped: 0,
                        tombstone_skipped: 0,
                        orphans_removed: 0,
                        blocked_cache_rebuilt: false,
                        exact_duplicate_comments_deduplicated: 0,
                        prefix_renames: Vec::new(),
                        salvage: salvage_receipt.clone(),
                    };
                    ctx.json_pretty(&result);
                } else if should_render_human_sync_output(ctx, use_json) {
                    println!("JSONL is current (hash unchanged since last import)");
                }
                return Ok(published_salvage_source);
            }
        }
    }

    // Parse orphan mode
    let orphan_mode = match args.orphans.as_deref() {
        Some("strict") | None => OrphanMode::Strict,
        Some("resurrect") => OrphanMode::Resurrect,
        Some("skip") => OrphanMode::Skip,
        Some("allow") => OrphanMode::Allow,
        Some(other) => {
            return Err(BeadsError::Validation {
                field: "orphans".to_string(),
                reason: format!(
                    "Invalid orphan mode: {other}. Must be one of: strict, resurrect, skip, allow"
                ),
            });
        }
    };
    debug!(orphan_mode = ?orphan_mode, "Import orphan handling configured");

    // Configure import
    let import_config = ImportConfig {
        // Keep prefix validation when explicitly renaming prefixes.
        skip_prefix_validation: args.force && !args.rename_prefix,
        rename_on_import: args.rename_prefix,
        clear_duplicate_external_refs: args.rename_prefix,
        orphan_mode,
        force_upsert: args.force,
        beads_dir: Some(path_policy.beads_dir.clone()),
        allow_external_jsonl: path_policy.allow_external_jsonl,
        show_progress,
    };

    // Force/rebuild prepasses and the import itself all consume the same
    // captured bytes. Run the marker scan first so malformed merge content
    // retains its actionable error class, then derive IDs and tombstone state
    // before any destructive table reset.
    if args.force || args.rebuild {
        ensure_no_conflict_markers_snapshot(source)?;
    }
    let jsonl_issue_ids = if args.force || args.rebuild {
        Some(get_issue_ids_from_jsonl_snapshot(source)?)
    } else {
        None
    };
    let jsonl_filter = if args.force || args.rebuild {
        Some(scan_jsonl_snapshot_for_tombstone_filter(source)?)
    } else {
        None
    };

    let preserved_tombstones = if args.force || args.rebuild {
        tombstones_missing_from_jsonl_tombstones(
            snapshot_tombstones(storage),
            jsonl_filter
                .as_ref()
                .expect("force/rebuild imports should precompute JSONL tombstone filter"),
        )
    } else {
        Vec::new()
    };
    let preserved_resurrection_attempts = jsonl_filter.as_ref().map_or(0, |filter| {
        preserved_tombstones
            .iter()
            .filter(|tombstone| {
                filter
                    .non_tombstone_updated_at
                    .contains_key(&tombstone.issue.id)
            })
            .count()
    });

    // Force/rebuild imports replace the database through the verified
    // backup-and-restore recovery path. This avoids fsqlite's in-place
    // DROP/CREATE pager hazards and, critically, guarantees that any late
    // validation or storage failure restores the complete previous DB family.
    let import_used_backup_rebuild = args.force || args.rebuild;
    let force_import_target_was_empty = false;
    info!(path = %jsonl_path.display(), "Importing from JSONL");
    let mut import_result = if import_used_backup_rebuild {
        replace_database_from_jsonl_snapshot(
            storage,
            beads_dir,
            cli,
            source,
            jsonl_authority,
            db_path,
            &import_config,
            target_prefix.as_deref(),
            &preserved_tombstones,
        )?
    } else {
        import_from_jsonl_snapshot(storage, source, &import_config, target_prefix.as_deref())?
    };

    if let Some(salvage) = salvage_receipt.as_mut() {
        let exportable_records = storage.count_exportable_issues()?;
        let records_requiring_export =
            exportable_records.saturating_sub(import_result.export_hashes_recorded);
        salvage.database_records_requiring_export = records_requiring_export;
        if records_requiring_export > 0 {
            storage.set_metadata("needs_flush", "true")?;
            salvage.needs_flush_set = true;
            warn!(
                records_requiring_export,
                "JSONL salvage preserved database records absent from the recovered source; a full export is required"
            );
        }
    }

    info!(
        created_or_updated = import_result.imported_count,
        skipped = import_result.skipped_count,
        tombstone_skipped = import_result.tombstone_skipped,
        "Import complete"
    );

    // --rebuild: remove DB entries not present in JSONL.
    //
    // Skip this entirely when `--rename-prefix` is also set: the import just
    // rewrote every JSONL ID into the configured prefix, so `db_ids` are
    // post-rename (e.g. "newpref-xre") while `jsonl_ids` are pre-rename
    // (e.g. "oldpref-001"). The set-difference would classify every
    // newly-imported issue as an orphan and wipe the DB — exactly the
    // opposite of what the user asked for. With `reset_data_tables` having
    // cleared everything beforehand, the post-import DB contents already
    // mirror the JSONL (modulo the prefix rewrite), so the orphan pass has
    // nothing legitimate to remove anyway.
    //
    // Tombstones preserved across `reset_data_tables` via `snapshot_tombstones`
    // are NOT orphans — the whole point of preserving them was to keep
    // deletion-retention state alive across the rebuild. If the user has not
    // flushed to JSONL since deleting an issue, the tombstone is in the DB
    // but not in the JSONL, and a naïve set-difference would wipe it. Union
    // their IDs into the "acceptable" set so they survive the cleanup.
    if args.rebuild && !args.rename_prefix {
        let jsonl_ids = jsonl_issue_ids
            .as_ref()
            .expect("--rebuild should precompute JSONL issue IDs");
        let preserved_ids: HashSet<String> = preserved_tombstones
            .iter()
            .map(|t| t.issue.id.clone())
            .collect();
        let db_ids: HashSet<String> = storage.get_all_ids()?.into_iter().collect();
        let orphan_ids: Vec<String> = db_ids
            .iter()
            .filter(|id| !jsonl_ids.contains(*id) && !preserved_ids.contains(*id))
            .cloned()
            .collect();

        if !orphan_ids.is_empty() {
            info!(
                count = orphan_ids.len(),
                "Removing orphaned DB entries not present in JSONL"
            );
            for id in &orphan_ids {
                debug!(id = %id, "Removing orphaned issue");
                storage.delete_issue(id, "br-rebuild", "rebuild: not in JSONL", None)?;
            }
            import_result.orphans_removed = orphan_ids.len();
            // Rebuild blocked cache again after removals
            storage.rebuild_blocked_cache(true)?;
            info!(
                removed = orphan_ids.len(),
                "Rebuild orphan cleanup complete"
            );
        }
    } else if args.rebuild {
        debug!(
            "Skipping --rebuild orphan cleanup: --rename-prefix rewrote IDs, so JSONL IDs no longer match DB IDs and the set-difference would be incorrect"
        );
    }

    if args.force || args.rebuild {
        if !import_used_backup_rebuild {
            restore_tombstones_after_rebuild(storage, &preserved_tombstones)?;
        }
        import_result.tombstone_skipped += preserved_resurrection_attempts;
    }

    // Update the source JSONL content hash before post-import maintenance.
    // Metadata table/index writes are part of the same B-tree surface that
    // triggered frankentorch-dbp, so compaction must be the final storage
    // mutation in this path.
    if !import_used_backup_rebuild {
        storage.set_metadata(METADATA_JSONL_CONTENT_HASH, &source_content_hash)?;
    }

    // Post-import VACUUM + REINDEX to eliminate B-tree/index corruption
    // artifacts that frankensqlite's bulk-insert and metadata-update paths
    // can leave behind.  This mirrors what `rebuild_database_family` (used
    // by `br doctor --repair` and auto recovery) does at the equivalent
    // chokepoint.
    //
    // Without this, large `br sync --import-only` runs can produce a DB
    // where C sqlite3's `PRAGMA integrity_check` reports free-space or
    // index-entry corruption.  Force/rebuild imports hit this through
    // `reset_data_tables()` + bulk import (issue #248); the FrankenTorch
    // current-JSONL reproducer hit the plain import path through metadata
    // table/index churn after importing hundreds of rows.
    let import_rewrote_storage = import_result.imported_count > 0
        || import_result.blocked_cache_entries > 0
        || import_result.child_counter_entries > 0;
    let skip_heavy_import_maintenance = if fresh_force_import_maintenance_gate_applies(
        args,
        force_import_target_was_empty,
        import_rewrote_storage,
    ) {
        let messages = storage.integrity_check_messages()?;
        let clean = integrity_check_is_clean(&messages);
        if clean {
            debug!(
                db_path = %db_path.display(),
                "Skipping post-import VACUUM/REINDEX: fresh force import already passed integrity_check"
            );
        } else {
            warn!(
                db_path = %db_path.display(),
                integrity_messages = ?messages,
                "Fresh force import reported integrity warnings; running full post-import maintenance"
            );
        }
        clean
    } else {
        false
    };

    if !import_used_backup_rebuild
        && (args.force || args.rebuild || import_rewrote_storage)
        && !skip_heavy_import_maintenance
    {
        // Final maintenance is isolated from the live tracker. The compaction
        // helper checkpoints once, copies the main database into a private
        // recovery-directory source, and runs VACUUM/REINDEX/VACUUM INTO
        // there. Only the attested candidate crosses the existing atomic
        // installation boundary, so an interrupted source VACUUM cannot
        // poison the live WAL-index (#507). Reopen failures still abort this
        // import rather than letting metadata updates use a placeholder.
        let placeholder = crate::storage::SqliteStorage::open_memory()?;
        let original_storage = std::mem::replace(storage, placeholder);
        match config::compact_database_via_vacuum_into_in_place(
            original_storage,
            db_path,
            cli.lock_timeout,
        ) {
            Ok(compacted_storage) => *storage = compacted_storage,
            Err(err) => {
                let reopened = cli
                    .database_family_write_authority_for(beads_dir, db_path)
                    .map_or_else(
                        || {
                            crate::storage::SqliteStorage::open_with_timeout(
                                db_path,
                                cli.lock_timeout,
                            )
                        },
                        |authority| {
                            crate::storage::SqliteStorage::open_with_timeout_under_write_authority(
                                db_path,
                                cli.lock_timeout,
                                authority,
                            )
                        },
                    );
                if let Ok(reopened) = reopened {
                    *storage = reopened;
                }
                return Err(err);
            }
        }
        repair_import_integrity_if_needed(
            storage,
            beads_dir,
            cli,
            source,
            jsonl_authority,
            db_path,
            &import_config,
        )?;
    }

    crate::sync::verify_jsonl_source_snapshot_current(source, jsonl_authority)?;

    // Output result
    let result = ImportResultOutput {
        created: import_result.created_count,
        updated: import_result.updated_count,
        skipped: import_result.skipped_count,
        tombstone_skipped: import_result.tombstone_skipped,
        orphans_removed: import_result.orphans_removed,
        blocked_cache_rebuilt: true,
        exact_duplicate_comments_deduplicated: import_result.exact_duplicate_comments_deduplicated,
        prefix_renames: import_result.prefix_renames.clone(),
        salvage: salvage_receipt.clone(),
    };

    if use_json {
        ctx.json_pretty(&result);
    } else if !should_render_human_sync_output(ctx, use_json) {
        return Ok(published_salvage_source);
    } else if ctx.is_rich() {
        render_import_result_rich(&result, ctx);
        if result.exact_duplicate_comments_deduplicated > 0 {
            ctx.warning(&format!(
                "Deduplicated {} byte-identical repeated comment object(s); conflicting duplicate IDs are still rejected",
                result.exact_duplicate_comments_deduplicated
            ));
        }
        if let Some(salvage) = &result.salvage {
            ctx.warning(&format!(
                "JSONL salvage retained {} valid record(s), rejected {} invalid record(s); exact source backup: {}",
                salvage.valid_records,
                salvage.rejected_records.len(),
                salvage.backup_path
            ));
            for rejected in salvage.rejected_records.iter().take(HUMAN_WITNESS_LIMIT) {
                ctx.warning(&format!(
                    "Rejected JSONL line {}: {}",
                    rejected.line, rejected.error
                ));
            }
            if salvage.needs_flush_set {
                ctx.warning(&format!(
                    "{} preserved database record(s) require `br sync --flush-only` to restore JSONL coverage",
                    salvage.database_records_requiring_export
                ));
            }
        }
    } else {
        let processed = import_result.imported_count
            + import_result.skipped_count
            + import_result.tombstone_skipped;
        println!("Imported from JSONL:");
        println!("  Processed: {processed} issues");
        println!("  Created: {} issues", result.created);
        println!("  Updated: {} issues", result.updated);
        if result.skipped > 0 {
            println!("  Skipped: {} issues (up-to-date)", result.skipped);
        }
        if result.tombstone_skipped > 0 {
            println!("  Tombstone protected: {} issues", result.tombstone_skipped);
        }
        if result.orphans_removed > 0 {
            println!(
                "  Orphans removed: {} issues (not in JSONL)",
                result.orphans_removed
            );
        }
        if result.exact_duplicate_comments_deduplicated > 0 {
            println!(
                "  Warning: deduplicated {} byte-identical repeated comment object(s); conflicting duplicate IDs are still rejected",
                result.exact_duplicate_comments_deduplicated
            );
        }
        if !result.prefix_renames.is_empty() {
            println!("  Prefix renames: {} issues", result.prefix_renames.len());
            for rename in result.prefix_renames.iter().take(HUMAN_WITNESS_LIMIT) {
                match rename.fallback.as_deref() {
                    Some(reason) => println!(
                        "    {} -> {} (fallback: {reason})",
                        rename.old_id, rename.new_id
                    ),
                    None => println!("    {} -> {}", rename.old_id, rename.new_id),
                }
            }
            if result.prefix_renames.len() > HUMAN_WITNESS_LIMIT {
                println!(
                    "    ... {} more; use --json for the complete mapping",
                    result.prefix_renames.len() - HUMAN_WITNESS_LIMIT
                );
            }
        }
        if let Some(salvage) = &result.salvage {
            println!(
                "  JSONL salvage: retained {} valid record(s), rejected {} invalid record(s)",
                salvage.valid_records,
                salvage.rejected_records.len()
            );
            println!(
                "  Exact source backup: {}",
                sanitize_terminal_inline(&salvage.backup_path)
            );
            for rejected in salvage.rejected_records.iter().take(HUMAN_WITNESS_LIMIT) {
                println!(
                    "    Rejected line {}: {}",
                    rejected.line,
                    sanitize_terminal_inline(&rejected.error)
                );
            }
            if salvage.rejected_records.len() > HUMAN_WITNESS_LIMIT {
                println!(
                    "    ... {} more; use --json for the complete rejection receipt",
                    salvage.rejected_records.len() - HUMAN_WITNESS_LIMIT
                );
            }
            if salvage.needs_flush_set {
                println!(
                    "  Follow-up: {} preserved database record(s) require `br sync --flush-only`",
                    salvage.database_records_requiring_export
                );
            }
        }
        println!("  Rebuilt blocked cache");
    }

    Ok(published_salvage_source)
}

/// Render import result with rich formatting.
fn render_import_result_rich(result: &ImportResultOutput, ctx: &OutputContext) {
    let _console = Console::default();
    let theme = ctx.theme();

    let mut text = Text::new("");

    // Success indicator
    text.append_styled("✓ ", theme.success.clone());
    text.append_styled("Import Complete", theme.success.clone());
    text.append("\n\n");

    // Direction indicator
    text.append_styled("Direction          ", theme.dimmed.clone());
    text.append_styled("JSONL → SQLite", theme.info.clone());
    text.append("\n");

    // Created count
    text.append_styled("Created            ", theme.dimmed.clone());
    text.append_styled(&result.created.to_string(), theme.accent.clone());
    text.append_styled(" issues", theme.dimmed.clone());
    text.append("\n");

    // Updated count
    text.append_styled("Updated            ", theme.dimmed.clone());
    text.append_styled(&result.updated.to_string(), theme.accent.clone());
    text.append_styled(" issues", theme.dimmed.clone());
    text.append("\n");

    // Skipped count
    if result.skipped > 0 {
        text.append_styled("Skipped            ", theme.dimmed.clone());
        text.append(&result.skipped.to_string());
        text.append_styled(" (up-to-date)", theme.muted.clone());
        text.append("\n");
    }

    // Tombstone protected
    if result.tombstone_skipped > 0 {
        text.append_styled("Tombstone protected ", theme.dimmed.clone());
        text.append(&result.tombstone_skipped.to_string());
        text.append("\n");
    }

    // Orphans removed
    if result.orphans_removed > 0 {
        text.append_styled("Orphans removed    ", theme.dimmed.clone());
        text.append_styled(&result.orphans_removed.to_string(), theme.warning.clone());
        text.append_styled(" (not in JSONL)", theme.muted.clone());
        text.append("\n");
    }

    // Prefix renames (--rename-prefix receipt)
    if !result.prefix_renames.is_empty() {
        text.append_styled("Prefix renames     ", theme.dimmed.clone());
        text.append_styled(
            &result.prefix_renames.len().to_string(),
            theme.accent.clone(),
        );
        text.append_styled(" issues", theme.dimmed.clone());
        text.append("\n");
        for rename in result.prefix_renames.iter().take(HUMAN_WITNESS_LIMIT) {
            text.append_styled("  ", theme.dimmed.clone());
            text.append(&rename.old_id);
            text.append_styled(" -> ", theme.dimmed.clone());
            text.append_styled(&rename.new_id, theme.info.clone());
            if let Some(reason) = rename.fallback.as_deref() {
                text.append_styled(&format!(" (fallback: {reason})"), theme.warning.clone());
            }
            text.append("\n");
        }
        if result.prefix_renames.len() > HUMAN_WITNESS_LIMIT {
            text.append_styled(
                &format!(
                    "  ... {} more; use --json for the complete mapping\n",
                    result.prefix_renames.len() - HUMAN_WITNESS_LIMIT
                ),
                theme.muted.clone(),
            );
        }
    }

    // Cache rebuilt
    text.append("\n");
    text.append_styled("✓ ", theme.success.clone());
    text.append_styled("Blocked cache rebuilt", theme.muted.clone());

    let panel = Panel::from_rich_text(&text, ctx.width())
        .title(Text::new("Import"))
        .box_style(theme.box_style);
    ctx.render(&panel);
}

/// Detect the issue ID prefix from the first non-tombstone issue in a JSONL file.
///
/// Returns `None` if the file is empty or contains no issues with a recognizable prefix.
/// Supports hyphenated prefixes such as `document-intelligence-0sa`.
fn detect_prefix_from_jsonl(source: &JsonlSourceSnapshot) -> Result<Option<String>> {
    let issues = read_issues_from_jsonl_snapshot(source)?;

    for issue in issues {
        if issue.status == crate::model::Status::Tombstone {
            continue;
        }

        if let Some((prefix, _)) = split_prefix_remainder(&issue.id) {
            return Ok(Some(prefix.to_string()));
        }
    }

    Ok(None)
}

/// Execute the --reconcile operation (additive JSONL→DB reconciliation).
///
/// Plans read-only, then either emits the plan receipt (`--dry-run`) or
/// applies it through `apply_additive_reconcile`, which re-verifies the
/// plan's witnesses inside a single write transaction and rolls back on any
/// mismatch. Deletion is impossible in this mode and no JSONL, base
/// snapshot, or event rows are ever written.
fn execute_reconcile(
    storage: &mut crate::storage::SqliteStorage,
    path_policy: &SyncPathPolicy,
    args: &SyncArgs,
    use_json: bool,
    ctx: &OutputContext,
) -> Result<()> {
    let jsonl_path = &path_policy.jsonl_path;
    info!(
        jsonl_path = %jsonl_path.display(),
        dry_run = args.dry_run,
        "Starting additive reconcile"
    );

    let import_config = ImportConfig {
        // Mixed prefixes are supported and reconcile never rewrites ids.
        skip_prefix_validation: true,
        rename_on_import: false,
        clear_duplicate_external_refs: false,
        orphan_mode: OrphanMode::Strict,
        force_upsert: false,
        beads_dir: Some(path_policy.beads_dir.clone()),
        allow_external_jsonl: path_policy.allow_external_jsonl,
        show_progress: false,
    };

    let plan = plan_sync_reconcile(storage, jsonl_path, &import_config)?;
    debug!(
        records = plan.record_count,
        creates = plan.count_kind(ReconcileActionKind::Create),
        updates = plan.count_kind(ReconcileActionKind::Update),
        db_only = plan.db_only_ids.len(),
        stored_hash_matches_jsonl = plan.stored_hash_matches_jsonl,
        "Reconcile plan computed"
    );

    let outcome = if args.dry_run {
        None
    } else {
        Some(apply_sync_reconcile(
            storage,
            jsonl_path,
            &import_config,
            &plan,
        )?)
    };

    let receipt = build_reconcile_receipt(jsonl_path, &plan, outcome.as_ref());

    if use_json {
        ctx.json_pretty(&receipt);
    } else if should_render_human_sync_output(ctx, use_json) {
        render_reconcile_receipt_text(&receipt);
    }
    Ok(())
}

fn build_reconcile_receipt(
    jsonl_path: &Path,
    plan: &ReconcilePlan,
    outcome: Option<&ReconcileApplyOutcome>,
) -> SyncReconcileReceipt {
    let truncate = |mut ids: Vec<String>| {
        ids.truncate(RECONCILE_PREVIEW_LIMIT);
        ids
    };
    let mut db_only_preview = plan.db_only_ids.clone();
    db_only_preview.truncate(RECONCILE_PREVIEW_LIMIT);

    SyncReconcileReceipt {
        schema_version: SYNC_RECONCILE_SCHEMA_VERSION,
        mode: if outcome.is_some() {
            "apply"
        } else {
            "dry_run"
        },
        applied: outcome.is_some(),
        jsonl_path: jsonl_path.display().to_string(),
        source: ReconcileSourceWitness {
            record_count: plan.record_count,
            ephemeral_skipped: plan.ephemeral_skipped,
            content_hash: plan.witness.jsonl_content_hash.clone(),
            mtime: plan.witness.jsonl_mtime_witness.clone(),
            size_bytes: plan.witness.jsonl_size,
        },
        target: ReconcileTargetWitness {
            db_issue_count: plan.witness.db_issue_count,
            stored_hash_matches_jsonl: plan.stored_hash_matches_jsonl,
        },
        plan: ReconcilePlanCounts {
            created: plan.count_kind(ReconcileActionKind::Create),
            updated: plan.count_kind(ReconcileActionKind::Update),
            skipped_equal: plan.count_kind(ReconcileActionKind::SkipEqual),
            skipped_older: plan.count_kind(ReconcileActionKind::SkipOlder),
            skipped_tombstone: plan.count_kind(ReconcileActionKind::SkipTombstone),
            deleted: 0,
            db_only: plan.db_only_ids.len(),
        },
        previews: ReconcileIdPreviews {
            created_ids: truncate(plan.target_ids_for_kind(ReconcileActionKind::Create)),
            updated_ids: truncate(plan.target_ids_for_kind(ReconcileActionKind::Update)),
            db_only_ids: db_only_preview,
            preview_limit: RECONCILE_PREVIEW_LIMIT,
        },
        relations: ReconcileRelationCounts {
            labels: plan.labels_planned,
            dependencies: plan.dependencies_planned,
            comments: plan.comments_planned,
        },
        events_before: plan.witness.events_count,
        events_after: outcome.map_or(plan.witness.events_count, |o| o.events_after),
        apply: outcome.map(|o| ReconcileApplyReceipt {
            export_hashes_recorded: o.export_hashes_recorded,
            uncertified_local_wins: o.uncertified_local_wins,
            orphan_dependencies_cleaned: o.orphan_dependencies_cleaned,
            blocked_cache_entries: o.blocked_cache_entries,
            child_counter_entries: o.child_counter_entries,
            needs_flush_set: o.needs_flush_set,
            metadata_repaired: o.metadata_repaired,
        }),
    }
}

fn render_reconcile_receipt_text(receipt: &SyncReconcileReceipt) {
    if receipt.applied {
        println!("Reconciled JSONL into database (additive):");
    } else {
        println!("Reconcile plan (dry run, nothing changed):");
    }
    println!(
        "  JSONL records: {} ({} ephemeral skipped)",
        receipt.source.record_count, receipt.source.ephemeral_skipped
    );
    println!(
        "  Create: {}  Update: {}  Delete: {} (structurally impossible)",
        receipt.plan.created, receipt.plan.updated, receipt.plan.deleted
    );
    println!(
        "  Skipped: {} equal, {} older-in-JSONL, {} tombstone-protected",
        receipt.plan.skipped_equal, receipt.plan.skipped_older, receipt.plan.skipped_tombstone
    );
    if receipt.plan.db_only > 0 {
        println!(
            "  DB-only issues (absent from JSONL, untouched): {}",
            receipt.plan.db_only
        );
    }
    println!(
        "  Events: {} before, {} after (preserved)",
        receipt.events_before, receipt.events_after
    );
    if receipt.target.stored_hash_matches_jsonl
        && (receipt.plan.created > 0 || receipt.plan.updated > 0)
    {
        println!(
            "  Note: stored content hash matched the JSONL while rows diverged (false-equal state)"
        );
    }
    if let Some(apply) = &receipt.apply {
        println!(
            "  Relations written: {} labels, {} dependencies, {} comments",
            receipt.relations.labels, receipt.relations.dependencies, receipt.relations.comments
        );
        if apply.orphan_dependencies_cleaned > 0 {
            println!(
                "  Dangling dependency rows removed: {}",
                apply.orphan_dependencies_cleaned
            );
        }
        if apply.metadata_repaired {
            println!("  Sync metadata repaired (content hash + witness recorded)");
        }
        if apply.needs_flush_set {
            println!(
                "  Local state still diverges from JSONL; database marked for flush (run: br sync --flush-only)"
            );
        }
    } else {
        println!("  Run without --dry-run to apply.");
    }
}

fn optional_source_matches(
    source: Option<&JsonlSourceSnapshot>,
    expected_state: &JsonlSourceStateWitness,
    expected_content_sha256: Option<&str>,
) -> bool {
    match (source, expected_state) {
        (None, JsonlSourceStateWitness::Missing) => expected_content_sha256.is_none(),
        (Some(source), JsonlSourceStateWitness::Present { .. }) => {
            source.state_witness() == *expected_state
                && expected_content_sha256
                    .is_none_or(|expected| source.content_sha256() == expected)
        }
        _ => false,
    }
}

fn source_matches_pending_merge_output(
    source: Option<&JsonlSourceSnapshot>,
    receipt: &SyncMergePendingReceipt,
) -> Result<bool> {
    let Some(source) = source else {
        return Ok(false);
    };
    if source.raw_sha256() != receipt.jsonl_after_raw_sha256
        || source.content_sha256() != receipt.jsonl_after_content_sha256
    {
        return Ok(false);
    }
    let (count, _) = analyze_jsonl_snapshot(source)?;
    Ok(count == receipt.jsonl_after_issue_count)
}

#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
fn reconcile_pending_sync_merge_artifacts(
    storage: &mut crate::storage::SqliteStorage,
    db_path: &Path,
    path_policy: &SyncPathPolicy,
    args: &SyncArgs,
    show_progress: bool,
    history_config: HistoryConfig,
    current_source: Option<&JsonlSourceSnapshot>,
    jsonl_authority: &crate::sync::JsonlFamilyWriteLock,
    current_base: Option<&JsonlSourceSnapshot>,
    base_authority: &crate::sync::JsonlFamilyWriteLock,
    mut receipt: SyncMergePendingReceipt,
    no_db: bool,
) -> Result<ReconciledPendingSyncMerge> {
    receipt.validate()?;
    if database_write_authority_sha256(db_path)? != receipt.intent.database_authority_sha256 {
        return Err(BeadsError::SyncConflict {
            message:
                "Pending sync merge database authority does not match the current configured path"
                    .to_string(),
        });
    }
    if receipt.intent.jsonl_authority_sha256 != jsonl_authority.authority_path_sha256()
        || receipt.intent.jsonl_path_sha256
            != canonical_sync_path_sha256(jsonl_authority.canonical_jsonl_path())
    {
        return Err(BeadsError::SyncConflict {
            message:
                "Pending sync merge JSONL authority does not match the current configured path"
                    .to_string(),
        });
    }
    if receipt.intent.base_authority_sha256 != base_authority.authority_path_sha256() {
        return Err(BeadsError::SyncConflict {
            message: "Pending sync merge base authority does not match the current configured path"
                .to_string(),
        });
    }
    let database_authority =
        if no_db {
            None
        } else {
            let authority = storage.attached_write_authority().ok_or_else(|| {
                BeadsError::SyncConflict {
                message:
                    "Persistent pending merge recovery has no retained database-family authority"
                        .to_string(),
            }
            })?;
            if authority.authority_path_sha256() != receipt.intent.database_authority_sha256 {
                return Err(BeadsError::SyncConflict {
                message:
                    "Retained database-family authority does not match the pending merge receipt"
                        .to_string(),
            });
            }
            authority.verify_database_authority()?;
            Some(authority)
        };
    if (args.force || args.force_db || args.force_jsonl)
        && merge_conflict_resolution_label(merge_conflict_resolution(args))
            != receipt.intent.resolution
    {
        return Err(BeadsError::SyncConflict {
            message: "Resume flags select a different merge resolution than the committed receipt"
                .to_string(),
        });
    }
    if storage.with_read_transaction(crate::sync::capture_sync_merge_core_witness)?
        != receipt.database_after
    {
        return Err(BeadsError::SyncConflict {
            message: "Database merge-authoritative state drifted after the pending merge committed"
                .to_string(),
        });
    }

    let jsonl_is_before = optional_source_matches(
        current_source,
        &receipt.intent.jsonl_before,
        receipt.intent.jsonl_before_content_sha256.as_deref(),
    );
    let jsonl_is_after = source_matches_pending_merge_output(current_source, &receipt)?;
    let jsonl_path = &path_policy.jsonl_path;
    let published_source = match receipt.phase {
        SyncMergePendingPhase::DatabaseCommitted => {
            if !jsonl_is_before && !jsonl_is_after {
                return Err(BeadsError::SyncConflict {
                    message:
                        "JSONL is neither the exact pre-merge generation nor the committed merge output"
                            .to_string(),
                });
            }
            let export_config = ExportConfig {
                force: true,
                is_default_path: true,
                error_policy: ExportErrorPolicy::Strict,
                retention_days: receipt.intent.retention_days,
                export_as_of: Some(receipt.intent.export_as_of),
                beads_dir: Some(path_policy.beads_dir.clone()),
                allow_external_jsonl: path_policy.allow_external_jsonl,
                show_progress,
                history: history_config,
                max_parallel_workers: args.export_parallelism.unwrap_or(0),
                expected_staged_output: Some(ExpectedStagedExport {
                    raw_sha256: receipt.jsonl_after_raw_sha256.clone(),
                    issue_count: receipt.jsonl_after_issue_count,
                    issue_hashes: receipt.jsonl_after_issue_hashes.clone(),
                }),
            };
            let expected_source = current_source.map_or(
                ExpectedJsonlSourceRef::Missing,
                ExpectedJsonlSourceRef::Present,
            );
            let (export_result, _) = if let Some(database_authority) = database_authority.as_deref()
            {
                export_to_jsonl_with_policy_expected_under_authorities(
                    storage,
                    jsonl_path,
                    &export_config,
                    expected_source,
                    jsonl_authority,
                    database_authority,
                )?
            } else {
                export_to_jsonl_with_policy_expected_under_authority(
                    storage,
                    jsonl_path,
                    &export_config,
                    expected_source,
                    jsonl_authority,
                )?
            };
            if export_result.content_hash != receipt.jsonl_after_raw_sha256
                || export_result.exported_count != receipt.jsonl_after_issue_count
            {
                return Err(BeadsError::SyncConflict {
                    message: "Deterministic merge export no longer matches the committed receipt"
                        .to_string(),
                });
            }
            let published_source = export_result.published_source_arc()?;
            if published_source.raw_sha256() != receipt.jsonl_after_raw_sha256
                || published_source.content_sha256() != receipt.jsonl_after_content_sha256
            {
                return Err(BeadsError::SyncConflict {
                    message:
                        "Published merge JSONL does not match the receipt's exact output hashes"
                            .to_string(),
                });
            }
            finalize_export_under_authority(
                storage,
                &export_result,
                Some(&export_result.issue_hashes),
                jsonl_path,
                jsonl_authority,
            )?;
            let (database_core, export_finalization) =
                storage.with_read_transaction(|storage| {
                    Ok((
                        crate::sync::capture_sync_merge_core_witness(storage)?,
                        crate::sync::capture_sync_merge_export_finalization_witness(storage)?,
                    ))
                })?;
            if database_core != receipt.database_after {
                return Err(BeadsError::SyncConflict {
                    message:
                        "Merge-authoritative database state changed during export finalization"
                            .to_string(),
                });
            }
            let finalized = receipt.advance_to_export_finalized(
                published_source.state_witness(),
                export_finalization,
            )?;
            storage.compare_and_set_pending_sync_merge_receipt(&receipt, &finalized)?;
            receipt = finalized;
            published_source
        }
        SyncMergePendingPhase::ExportFinalized => {
            if !jsonl_is_after
                || receipt.jsonl_after.as_ref()
                    != current_source
                        .map(JsonlSourceSnapshot::state_witness)
                        .as_ref()
            {
                return Err(BeadsError::SyncConflict {
                    message:
                        "Finalized pending merge JSONL no longer matches its exact published witness"
                            .to_string(),
                });
            }
            let pinned_source = jsonl_authority.capture_target()?;
            if receipt.jsonl_after.as_ref() != Some(&pinned_source.state_witness())
                || !source_matches_pending_merge_output(Some(&pinned_source), &receipt)?
            {
                return Err(BeadsError::SyncConflict {
                    message:
                        "Authority-pinned finalized JSONL recapture does not match the pending receipt"
                            .to_string(),
                });
            }
            Arc::new(pinned_source)
        }
    };

    let base_is_before = optional_source_matches(
        current_base,
        &receipt.intent.base_before,
        receipt.intent.base_before_content_sha256.as_deref(),
    );
    let base_is_after = current_base.is_some_and(|base| {
        base.raw_sha256() == published_source.raw_sha256()
            && base.content_sha256() == published_source.content_sha256()
            && base.size() == published_source.size()
    });
    if !base_is_before && !base_is_after {
        return Err(BeadsError::SyncConflict {
            message:
                "Base snapshot is neither the exact merge ancestor nor the published merge output"
                    .to_string(),
        });
    }
    let current_base_state = current_base.map_or(
        JsonlSourceStateWitness::Missing,
        JsonlSourceSnapshot::state_witness,
    );
    if let Some(database_authority) = database_authority.as_deref() {
        database_authority.verify_database_authority()?;
    }
    crate::sync::verify_jsonl_source_snapshot_current(&published_source, jsonl_authority)?;
    let base_publication = refresh_base_snapshot_from_flushed_jsonl_snapshot_under_authority(
        &published_source,
        &path_policy.beads_dir,
        &current_base_state,
        base_authority,
    )?;
    if let Some(database_authority) = database_authority.as_deref() {
        database_authority.verify_database_authority()?;
    }
    crate::sync::verify_jsonl_source_snapshot_current(&published_source, jsonl_authority)?;
    if base_publication.content_sha256() != published_source.content_sha256() {
        return Err(BeadsError::SyncConflict {
            message: "Published merge base does not match the exact finalized JSONL generation"
                .to_string(),
        });
    }
    if !base_publication.cleanup_durable() {
        warn!(
            base_path = %path_policy.beads_dir.join("beads.base.jsonl").display(),
            recovery_path = base_publication.retained_recovery_path(),
            "Base snapshot is verified, but displaced-generation cleanup was not certified durable"
        );
    }

    let terminal_base = base_authority.capture_target()?;
    if terminal_base.raw_sha256() != published_source.raw_sha256()
        || terminal_base.content_sha256() != published_source.content_sha256()
        || terminal_base.size() != published_source.size()
    {
        return Err(BeadsError::SyncConflict {
            message: "Terminal merge base witness differs from the exact finalized JSONL"
                .to_string(),
        });
    }
    if receipt.phase != SyncMergePendingPhase::ExportFinalized {
        return Err(BeadsError::SyncConflict {
            message: "Pending merge reconciliation did not reach export-finalized phase"
                .to_string(),
        });
    }
    if storage.with_read_transaction(crate::sync::capture_sync_merge_core_witness)?
        != receipt.database_after
    {
        return Err(BeadsError::SyncConflict {
            message: "Database merge-authoritative state changed before outer receipt cleanup"
                .to_string(),
        });
    }
    Ok(ReconciledPendingSyncMerge {
        published_source,
        terminal_receipt: receipt,
    })
}

#[allow(clippy::too_many_arguments)]
fn resume_pending_sync_merge(
    storage: &mut crate::storage::SqliteStorage,
    db_path: &Path,
    path_policy: &SyncPathPolicy,
    args: &SyncArgs,
    use_json: bool,
    show_progress: bool,
    history_config: HistoryConfig,
    current_source: Option<&JsonlSourceSnapshot>,
    jsonl_authority: &crate::sync::JsonlFamilyWriteLock,
    current_base: Option<&JsonlSourceSnapshot>,
    base_authority: &crate::sync::JsonlFamilyWriteLock,
    receipt: SyncMergePendingReceipt,
    no_db: bool,
) -> Result<(ReconciledPendingSyncMerge, DeferredSyncOutput)> {
    let receipt_id = receipt.receipt_id.clone();
    let phase = receipt.phase;
    let capacity_warnings = receipt.capacity_warnings.clone();
    let reconciled = reconcile_pending_sync_merge_artifacts(
        storage,
        db_path,
        path_policy,
        args,
        show_progress,
        history_config,
        current_source,
        jsonl_authority,
        current_base,
        base_authority,
        receipt,
        no_db,
    )
    .map_err(|source| {
        if no_db {
            source
        } else {
            BeadsError::CommittedStateUnwitnessed {
                operation: format!("resume committed sync merge {receipt_id}"),
                source: Box::new(source),
            }
        }
    })?;
    Ok((
        reconciled,
        DeferredSyncOutput::ResumedMerge {
            receipt_id,
            phase_before: phase,
            capacity_warnings,
            use_json,
        },
    ))
}

fn hydrate_exportable_database_issues(
    storage: &crate::storage::SqliteStorage,
) -> Result<BTreeMap<String, crate::model::Issue>> {
    let mut issues = storage.get_all_issues_for_export()?;
    let dependencies = storage.get_all_dependency_records()?;
    let labels = storage.get_all_labels()?;
    let comments = storage.get_all_comments()?;

    for issue in &mut issues {
        if let Some(issue_dependencies) = dependencies.get(&issue.id) {
            issue.dependencies.clone_from(issue_dependencies);
        }
        if let Some(issue_labels) = labels.get(&issue.id) {
            issue.labels.clone_from(issue_labels);
        }
        if let Some(issue_comments) = comments.get(&issue.id) {
            issue.comments.clone_from(issue_comments);
        }
    }

    Ok(issues
        .into_iter()
        .map(|issue| (issue.id.clone(), issue))
        .collect())
}

fn normalize_issue_source_repo_path(issue: &mut crate::model::Issue, target_path: &str) -> bool {
    let changed = issue.source_repo_path.as_deref() != Some(target_path);
    issue.source_repo_path = Some(target_path.to_string());
    issue.content_hash = Some(issue.compute_content_hash());
    changed
}

#[allow(clippy::too_many_lines)]
fn build_source_repo_path_migration_plan(
    storage: &crate::storage::SqliteStorage,
    beads_dir: &Path,
    source: Option<&JsonlSourceSnapshot>,
) -> Result<SourceRepoPathMigrationPlan> {
    let target_path = canonical_source_repo_path(beads_dir).ok_or_else(|| {
        BeadsError::Config(format!(
            "Cannot resolve the canonical workspace path above {}",
            beads_dir.display()
        ))
    })?;
    if !Path::new(&target_path).is_absolute() {
        return Err(BeadsError::SyncConflict {
            message: "Canonical source_repo_path migration target is not absolute".to_string(),
        });
    }

    let database_before = capture_sync_database_witness(storage)?;
    let original_database = hydrate_exportable_database_issues(storage)?;
    if capture_sync_database_witness(storage)? != database_before {
        return Err(BeadsError::SyncConflict {
            message:
                "Database changed while the source_repo_path migration plan was being hydrated"
                    .to_string(),
        });
    }

    let source_before = source.map_or(
        JsonlSourceStateWitness::Missing,
        JsonlSourceSnapshot::state_witness,
    );
    let source_before_content_sha256 = source
        .map(JsonlSourceSnapshot::content_sha256)
        .map(str::to_string);
    let source_issues = if let Some(source) = source {
        ensure_no_conflict_markers_snapshot(source)?;
        let validation = crate::sync::validate_jsonl_snapshot_issue_records(source)?;
        if validation.invalid_count != 0 {
            return Err(BeadsError::Config(format!(
                "Cannot migrate source_repo_path: JSONL contains {} invalid record(s): {}",
                validation.invalid_count,
                validation.preview_messages().join("; ")
            )));
        }
        read_issues_from_jsonl_snapshot(source)?
    } else {
        Vec::new()
    };
    let source_records = source_issues.len();

    let mut normalized_issue_ids = BTreeSet::new();
    let mut final_by_id = original_database.clone();
    for issue in final_by_id.values_mut() {
        if normalize_issue_source_repo_path(issue, &target_path) {
            normalized_issue_ids.insert(issue.id.clone());
        }
    }

    let mut normalized_source = BTreeMap::new();
    let mut source_only_created = 0usize;
    let mut source_newer_updated = 0usize;
    let mut database_newer_preserved = 0usize;
    let mut equal_records = 0usize;
    let mut tombstones_preserved = 0usize;
    let mut ephemeral_source_records_skipped = 0usize;

    for mut incoming in source_issues {
        if incoming.ephemeral || incoming.id.contains("-wisp-") {
            ephemeral_source_records_skipped += 1;
            continue;
        }
        if normalize_issue_source_repo_path(&mut incoming, &target_path) {
            normalized_issue_ids.insert(incoming.id.clone());
        }
        normalized_source.insert(incoming.id.clone(), incoming.clone());

        match final_by_id.get(&incoming.id) {
            None => {
                source_only_created += 1;
                final_by_id.insert(incoming.id.clone(), incoming);
            }
            Some(existing) if existing.status == crate::model::Status::Tombstone => {
                tombstones_preserved += 1;
            }
            Some(existing) => match incoming.updated_at.cmp(&existing.updated_at) {
                std::cmp::Ordering::Greater => {
                    source_newer_updated += 1;
                    final_by_id.insert(incoming.id.clone(), incoming);
                }
                std::cmp::Ordering::Less => {
                    database_newer_preserved += 1;
                }
                std::cmp::Ordering::Equal if existing.sync_equals(&incoming) => {
                    equal_records += 1;
                }
                std::cmp::Ordering::Equal => {
                    return Err(BeadsError::SyncConflict {
                        message: format!(
                            "Issue {} has equal timestamps but divergent DB/JSONL payloads; resolve it before source_repo_path migration",
                            incoming.id
                        ),
                    });
                }
            },
        }
    }

    let mut external_ref_owners = HashMap::<String, String>::new();
    for issue in final_by_id.values() {
        if let Some(external_ref) = issue.external_ref.as_ref()
            && let Some(existing_id) =
                external_ref_owners.insert(external_ref.clone(), issue.id.clone())
            && existing_id != issue.id
        {
            return Err(BeadsError::SyncConflict {
                message: format!(
                    "source_repo_path migration projection contains duplicate external_ref {external_ref:?} on {existing_id} and {}",
                    issue.id
                ),
            });
        }
    }

    let jsonl_rewrite_required = normalized_source.len() != final_by_id.len()
        || final_by_id.iter().any(|(issue_id, issue)| {
            normalized_source
                .get(issue_id)
                .is_none_or(|source_issue| !source_issue.sync_equals(issue))
        });
    let mut changed_kept = final_by_id
        .values()
        .filter(|issue| {
            original_database
                .get(&issue.id)
                .is_none_or(|before| !before.sync_equals(issue))
        })
        .cloned()
        .collect::<Vec<_>>();
    changed_kept.sort_by(|left, right| left.id.cmp(&right.id));
    let changed_issue_ids = changed_kept
        .iter()
        .map(|issue| issue.id.clone())
        .collect::<Vec<_>>();
    let changed_issue_witnesses = crate::sync::sync_merge_kept_issue_witnesses(&changed_kept)?;
    let digest_input = SourceRepoPathMigrationPlanDigest {
        schema: SOURCE_REPO_PATH_MIGRATION_SCHEMA,
        target_path: &target_path,
        source_before: &source_before,
        source_before_content_sha256: source_before_content_sha256.as_deref(),
        database_before: &database_before,
        changed_issue_witnesses: &changed_issue_witnesses,
        source_records,
        source_only_created,
        source_newer_updated,
        database_newer_preserved,
        equal_records,
        tombstones_preserved,
        ephemeral_source_records_skipped,
        paths_normalized: normalized_issue_ids.len(),
        jsonl_rewrite_required,
    };
    let plan_bytes = serde_json::to_vec(&digest_input)?;
    let plan_sha256 = crate::util::hex_encode(&Sha256::digest(plan_bytes));
    let target_path_sha256 = crate::util::hex_encode(&Sha256::digest(target_path.as_bytes()));
    let no_op = changed_kept.is_empty() && !jsonl_rewrite_required;

    Ok(SourceRepoPathMigrationPlan {
        receipt: SourceRepoPathMigrationReceipt {
            schema: SOURCE_REPO_PATH_MIGRATION_SCHEMA,
            mode: "dry_run",
            applied: false,
            no_op,
            plan_sha256,
            target_path,
            target_path_sha256,
            source_records,
            database_records: original_database.len(),
            source_only_created,
            source_newer_updated,
            database_newer_preserved,
            equal_records,
            tombstones_preserved,
            ephemeral_source_records_skipped,
            paths_normalized: normalized_issue_ids.len(),
            changed_issue_ids,
            jsonl_rewrite_required,
            source_repo_preserved: true,
            vcs_status: "not_probed",
            warnings: Vec::new(),
            receipt_id: None,
        },
        changed_kept,
        database_before,
        source_before,
        source_before_content_sha256,
    })
}

fn render_source_repo_path_migration_receipt(
    receipt: &SourceRepoPathMigrationReceipt,
    ctx: &OutputContext,
    use_json: bool,
) {
    if use_json {
        ctx.json_pretty(receipt);
        return;
    }
    if !should_render_human_sync_output(ctx, use_json) {
        return;
    }
    println!(
        "source_repo_path migration {}: plan_sha256={}",
        if receipt.applied { "complete" } else { "plan" },
        receipt.plan_sha256
    );
    println!(
        "  Target: {}",
        sanitize_terminal_inline(&receipt.target_path)
    );
    println!(
        "  Source-only: {}  source-newer: {}  DB-newer preserved: {}  equal: {}",
        receipt.source_only_created,
        receipt.source_newer_updated,
        receipt.database_newer_preserved,
        receipt.equal_records
    );
    println!(
        "  Paths normalized: {}  changed DB rows: {}  JSONL rewrite: {}",
        receipt.paths_normalized,
        receipt.changed_issue_ids.len(),
        receipt.jsonl_rewrite_required
    );
    println!("  source_repo preserved: yes; VCS status: not probed");
    if receipt.no_op {
        println!("  Result: already normalized; no mutation required");
    } else if !receipt.applied {
        println!(
            "  Apply: br sync --migrate-source-repo-path --apply --expect-plan-sha256 {}",
            receipt.plan_sha256
        );
    }
    for warning in &receipt.warnings {
        ctx.warning(&warning.to_string());
    }
}

#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
fn execute_source_repo_path_migration(
    storage: &mut crate::storage::SqliteStorage,
    path_policy: &SyncPathPolicy,
    args: &SyncArgs,
    use_json: bool,
    show_progress: bool,
    retention_days: Option<u64>,
    history_config: HistoryConfig,
    retained_source: config::RetainedJsonlSourceRef<'_>,
    expected_source: &JsonlSourceStateWitness,
    retained_authority: Option<&crate::sync::JsonlFamilyWriteLock>,
    cli: &config::CliOverrides,
    db_path: &Path,
    no_db: bool,
    ctx: &OutputContext,
) -> Result<SyncDispatchCompletion> {
    let jsonl_path = &path_policy.jsonl_path;
    let owned_jsonl_authority = retained_authority
        .is_none()
        .then(|| {
            crate::sync::blocking_jsonl_family_write_lock_with_timeout(jsonl_path, cli.lock_timeout)
        })
        .transpose()?;
    let jsonl_authority = retained_authority
        .or(owned_jsonl_authority.as_ref())
        .expect("JSONL authority is retained or acquired");
    jsonl_authority.verify_jsonl_authority()?;
    let captured_source = jsonl_authority.capture_optional_target()?;
    let source = captured_source.as_ref();
    let observed_source = source.map_or(
        JsonlSourceStateWitness::Missing,
        JsonlSourceSnapshot::state_witness,
    );
    if !matches!(retained_source, config::RetainedJsonlSourceRef::Uncaptured)
        && &observed_source != expected_source
    {
        return Err(BeadsError::SyncConflict {
            message: "Retained JSONL source does not match its startup witness".to_string(),
        });
    }

    let base_path = path_policy.beads_dir.join("beads.base.jsonl");
    if let Some(receipt) = storage.pending_sync_merge_receipt()? {
        let base_authority = crate::sync::blocking_jsonl_family_write_lock_with_timeout(
            &base_path,
            cli.lock_timeout,
        )?;
        base_authority.verify_jsonl_authority()?;
        let base_source = base_authority.capture_optional_target()?;
        let (reconciled, deferred_output) = resume_pending_sync_merge(
            storage,
            db_path,
            path_policy,
            args,
            use_json,
            show_progress,
            history_config,
            source,
            jsonl_authority,
            base_source.as_ref(),
            &base_authority,
            receipt,
            no_db,
        )?;
        return Ok(SyncDispatchCompletion {
            published_source: Some(reconciled.published_source),
            owned_jsonl_authority,
            pending_merge: Some(PendingSyncMergeCompletion {
                receipt: reconciled.terminal_receipt,
                base_authority,
            }),
            deferred_output: Some(deferred_output),
        });
    }

    let mut plan = build_source_repo_path_migration_plan(storage, &path_policy.beads_dir, source)?;
    if !args.apply {
        render_source_repo_path_migration_receipt(&plan.receipt, ctx, use_json);
        return Ok(SyncDispatchCompletion::default());
    }
    let expected_plan_sha256 =
        args.expect_plan_sha256
            .as_deref()
            .ok_or_else(|| BeadsError::Validation {
                field: "expect_plan_sha256".to_string(),
                reason: "--apply requires the exact reviewed migration plan token".to_string(),
            })?;
    if expected_plan_sha256 != plan.receipt.plan_sha256 {
        return Err(BeadsError::SyncConflict {
            message: format!(
                "source_repo_path migration plan changed: expected {expected_plan_sha256}, current {}",
                plan.receipt.plan_sha256
            ),
        });
    }
    plan.receipt.mode = "apply";
    plan.receipt.applied = true;
    if plan.receipt.no_op {
        render_source_repo_path_migration_receipt(&plan.receipt, ctx, use_json);
        return Ok(SyncDispatchCompletion::default());
    }

    jsonl_authority.verify_jsonl_authority()?;
    let recaptured_source = jsonl_authority.capture_optional_target()?;
    let recaptured_state = recaptured_source.as_ref().map_or(
        JsonlSourceStateWitness::Missing,
        JsonlSourceSnapshot::state_witness,
    );
    if recaptured_state != plan.source_before
        || recaptured_source
            .as_ref()
            .map(JsonlSourceSnapshot::content_sha256)
            != plan.source_before_content_sha256.as_deref()
    {
        return Err(BeadsError::SyncConflict {
            message: "JSONL changed after the source_repo_path migration plan was reviewed"
                .to_string(),
        });
    }

    let base_authority =
        crate::sync::blocking_jsonl_family_write_lock_with_timeout(&base_path, cli.lock_timeout)?;
    base_authority.verify_jsonl_authority()?;
    let base_source = base_authority.capture_optional_target()?;
    let base_state = base_source.as_ref().map_or(
        JsonlSourceStateWitness::Missing,
        JsonlSourceSnapshot::state_witness,
    );
    let changed_kept_ids = plan
        .changed_kept
        .iter()
        .map(|issue| issue.id.clone())
        .collect::<Vec<_>>();
    let kept_issue_witnesses = crate::sync::sync_merge_kept_issue_witnesses(&plan.changed_kept)?;
    let migration_as_of = chrono::Utc::now();
    let actor = cli.actor.as_deref().unwrap_or("br");
    let intent = SyncMergeIntent {
        schema_version: 2,
        database_authority_sha256: database_write_authority_sha256(db_path)?,
        jsonl_authority_sha256: jsonl_authority.authority_path_sha256().to_string(),
        jsonl_path_sha256: canonical_sync_path_sha256(jsonl_authority.canonical_jsonl_path()),
        jsonl_before: plan.source_before.clone(),
        jsonl_before_content_sha256: plan.source_before_content_sha256.clone(),
        base_authority_sha256: base_authority.authority_path_sha256().to_string(),
        base_before: base_state,
        base_before_content_sha256: base_source
            .as_ref()
            .map(JsonlSourceSnapshot::content_sha256)
            .map(str::to_string),
        resolution: "source-repo-path-migration".to_string(),
        actor: actor.to_string(),
        event_attribution: storage.pending_event_attribution_for_review(),
        capacity_policy: storage.workflow_capacity_policy_for_review(),
        retention_days,
        export_as_of: migration_as_of,
        changed_kept_issue_ids: changed_kept_ids,
        kept_issue_witnesses,
        deleted_issue_ids: Vec::new(),
        note_witnesses: Vec::new(),
        database_before: plan.database_before,
    };
    let pending_receipt =
        storage.apply_sync_merge_atomically(&plan.changed_kept, &[], &[], &intent)?;
    plan.receipt
        .warnings
        .clone_from(&pending_receipt.capacity_warnings);
    let _ = storage.take_capacity_warnings();

    let reconciled = reconcile_pending_sync_merge_artifacts(
        storage,
        db_path,
        path_policy,
        args,
        show_progress,
        history_config,
        source,
        jsonl_authority,
        base_source.as_ref(),
        &base_authority,
        pending_receipt,
        no_db,
    )
    .map_err(|source| {
        if no_db {
            source
        } else {
            BeadsError::CommittedStateUnwitnessed {
                operation: "source_repo_path migration artifact reconciliation".to_string(),
                source: Box::new(source),
            }
        }
    })?;
    plan.receipt.receipt_id = Some(reconciled.terminal_receipt.receipt_id.clone());

    Ok(SyncDispatchCompletion {
        published_source: Some(reconciled.published_source),
        owned_jsonl_authority,
        pending_merge: Some(PendingSyncMergeCompletion {
            receipt: reconciled.terminal_receipt,
            base_authority,
        }),
        deferred_output: Some(DeferredSyncOutput::SourceRepoPathMigration {
            receipt: plan.receipt,
            use_json,
        }),
    })
}

/// Execute the --merge operation.
#[allow(clippy::too_many_lines, clippy::too_many_arguments)]
fn execute_merge(
    storage: &mut crate::storage::SqliteStorage,
    path_policy: &SyncPathPolicy,
    args: &SyncArgs,
    use_json: bool,
    show_progress: bool,
    retention_days: Option<u64>,
    history_config: HistoryConfig,
    retained_source: config::RetainedJsonlSourceRef<'_>,
    expected_source: &JsonlSourceStateWitness,
    retained_authority: Option<&crate::sync::JsonlFamilyWriteLock>,
    cli: &config::CliOverrides,
    db_path: &Path,
    no_db: bool,
    ctx: &OutputContext,
) -> Result<SyncDispatchCompletion> {
    info!("Starting 3-way merge");
    let beads_dir = &path_policy.beads_dir;
    let jsonl_path = &path_policy.jsonl_path;

    let owned_jsonl_authority = retained_authority
        .is_none()
        .then(|| {
            crate::sync::blocking_jsonl_family_write_lock_with_timeout(jsonl_path, cli.lock_timeout)
        })
        .transpose()?;
    let jsonl_authority = retained_authority
        .or(owned_jsonl_authority.as_ref())
        .expect("JSONL authority is retained or acquired");
    jsonl_authority.verify_jsonl_authority()?;
    let captured_source = jsonl_authority.capture_optional_target()?;
    let source = captured_source.as_ref();
    let observed_source = source.map_or(
        JsonlSourceStateWitness::Missing,
        JsonlSourceSnapshot::state_witness,
    );
    if !matches!(retained_source, config::RetainedJsonlSourceRef::Uncaptured)
        && &observed_source != expected_source
    {
        return Err(BeadsError::SyncConflict {
            message: "Retained JSONL source does not match its startup witness".to_string(),
        });
    }

    let base_path = beads_dir.join("beads.base.jsonl");
    let base_authority =
        crate::sync::blocking_jsonl_family_write_lock_with_timeout(&base_path, cli.lock_timeout)?;
    base_authority.verify_jsonl_authority()?;
    let base_source = base_authority.capture_optional_target()?;
    let base_state = base_source.as_ref().map_or(
        JsonlSourceStateWitness::Missing,
        JsonlSourceSnapshot::state_witness,
    );

    if let Some(receipt) = storage.pending_sync_merge_receipt()? {
        let (reconciled, deferred_output) = resume_pending_sync_merge(
            storage,
            db_path,
            path_policy,
            args,
            use_json,
            show_progress,
            history_config,
            source,
            jsonl_authority,
            base_source.as_ref(),
            &base_authority,
            receipt,
            no_db,
        )?;
        return Ok(SyncDispatchCompletion {
            published_source: Some(reconciled.published_source),
            owned_jsonl_authority,
            pending_merge: Some(PendingSyncMergeCompletion {
                receipt: reconciled.terminal_receipt,
                base_authority,
            }),
            deferred_output: Some(deferred_output),
        });
    }

    let database_before = capture_sync_database_witness(storage)?;

    // 1. Load Base State (ancestor) from the exact retained generation.
    let base = load_base_snapshot_from_source(base_source.as_ref())?;
    debug!(base_count = base.len(), "Loaded base snapshot");

    // 2. Load Left State (local DB)
    let mut left_issues = storage.get_all_issues_for_export()?;
    let all_deps = storage.get_all_dependency_records()?;
    let all_labels = storage.get_all_labels()?;
    let all_comments = storage.get_all_comments()?;

    for issue in &mut left_issues {
        if let Some(deps) = all_deps.get(&issue.id) {
            issue.dependencies = deps.clone();
        }
        if let Some(labels) = all_labels.get(&issue.id) {
            issue.labels = labels.clone();
        }
        if let Some(comments) = all_comments.get(&issue.id) {
            issue.comments = comments.clone();
        }
    }

    let mut left = HashMap::new();
    for issue in left_issues {
        left.insert(issue.id.clone(), issue);
    }
    debug!(left_count = left.len(), "Loaded local state (DB)");
    if capture_sync_database_witness(storage)? != database_before {
        return Err(BeadsError::SyncConflict {
            message:
                "Database changed while the sync merge plan was being hydrated; retry from a stable generation"
                    .to_string(),
        });
    }

    // 3. Load Right State (external JSONL)
    let mut right = HashMap::new();
    if let Some(source) = source {
        // The JSONL parser yields a generic
        // generic "Invalid JSON at line 1" error when the JSONL still
        // contains unresolved merge-conflict markers from a botched
        // `git merge` / `git pull`. A three-way merge on top of that state
        // would be nonsense, so scan for markers first and surface the
        // helpful error before we try to parse.
        ensure_no_conflict_markers_snapshot(source)?;
        for issue in read_issues_from_jsonl_snapshot(source)? {
            right.insert(issue.id.clone(), issue);
        }
    }
    debug!(right_count = right.len(), "Loaded external state (JSONL)");

    if source.is_none()
        && (!base.is_empty() || !left.is_empty())
        && !args.force_db
        && !args.force_jsonl
    {
        return Err(BeadsError::SyncConflict {
            message:
                "issues.jsonl is missing, which is not equivalent to an intentionally empty source; use --force-db to recreate it or --force-jsonl to explicitly accept deletion"
                    .to_string(),
        });
    }
    // A JSONL with no issues at all (a truncated file, an empty checkout)
    // reads as "the other side deleted everything" and would delete every
    // issue the database still shares with the last sync, under every
    // strategy (an unchanged issue deleted on one side is not a conflict).
    // Deletions normally leave tombstones, so refuse unless the deletion is
    // explicitly accepted, as the export guard does for an empty database
    // over a non-empty JSONL.
    if source.is_some()
        && right.is_empty()
        && !args.force_jsonl
        && base.keys().any(|id| left.contains_key(id))
    {
        return Err(BeadsError::SyncConflict {
            message:
                "issues.jsonl contains no issues, so merging it would delete every issue the database shares with the last sync. Restore issues.jsonl (for example from git) or rewrite it from the database with `br sync --flush-only --force`; pass --force-jsonl to accept deleting them"
                    .to_string(),
        });
    }
    if base_source.is_none() && left != right && !args.force_db && !args.force_jsonl {
        return Err(BeadsError::SyncConflict {
            message:
                "beads.base.jsonl is missing and the database differs from JSONL; choose --force-db or --force-jsonl explicitly instead of guessing a merge ancestor"
                    .to_string(),
        });
    }

    // 4. Perform Merge
    let context = MergeContext::new(base, left, right);
    let strategy = merge_conflict_resolution(args);
    let resolution = merge_conflict_resolution_label(strategy);
    let local_tombstones: HashSet<String> = context
        .left
        .values()
        .filter(|issue| issue.status == crate::model::Status::Tombstone)
        .map(|issue| issue.id.clone())
        .collect();
    let tombstones = if local_tombstones.is_empty() {
        None
    } else {
        Some(&local_tombstones)
    };

    let report = three_way_merge(&context, strategy, tombstones);

    // 5. Apply Changes to DB
    info!(
        kept = report.kept.len(),
        deleted = report.deleted.len(),
        conflicts = report.conflicts.len(),
        resolution,
        "Merge calculated"
    );

    if report.has_conflicts() {
        // Require an explicit merge winner instead of guessing when both sides changed.
        if ctx.is_rich() {
            render_merge_conflicts_rich(&report.conflicts, ctx);
        }
        let mut msg = String::from("Merge conflicts detected:\n");
        for (id, kind) in &report.conflicts {
            use std::fmt::Write;
            let _ = writeln!(msg, "  - {id}: {kind:?}");
        }
        msg.push_str("\nUse --force-db to keep local DB changes, --force-jsonl to keep JSONL changes, or --force to keep the newer timestamp.");
        return Err(BeadsError::Config(msg));
    }

    let actor = cli.actor.as_deref().unwrap_or("br");
    let note_target_ids = report
        .notes
        .iter()
        .map(|(issue_id, _)| issue_id.as_str())
        .collect::<HashSet<_>>();
    let mut changed_kept = report
        .kept
        .iter()
        .filter(|issue| {
            context.left.get(&issue.id) != Some(*issue)
                || note_target_ids.contains(issue.id.as_str())
        })
        .cloned()
        .collect::<Vec<_>>();
    changed_kept.sort_by(|left, right| left.id.cmp(&right.id));
    let changed_kept_ids = changed_kept
        .iter()
        .map(|issue| issue.id.clone())
        .collect::<Vec<_>>();
    let kept_issue_witnesses = crate::sync::sync_merge_kept_issue_witnesses(&changed_kept)?;
    let mut deleted_ids = report.deleted.clone();
    deleted_ids.sort();
    let mut note_witnesses = report
        .notes
        .iter()
        .map(|(issue_id, note)| SyncMergeNoteWitness {
            issue_id: issue_id.clone(),
            note_sha256: crate::util::hex_encode(&Sha256::digest(note.as_bytes())),
        })
        .collect::<Vec<_>>();
    note_witnesses.sort_by(|left, right| left.issue_id.cmp(&right.issue_id));
    let merge_as_of = chrono::Utc::now();
    let intent = SyncMergeIntent {
        schema_version: 2,
        database_authority_sha256: database_write_authority_sha256(db_path)?,
        jsonl_authority_sha256: jsonl_authority.authority_path_sha256().to_string(),
        jsonl_path_sha256: canonical_sync_path_sha256(jsonl_authority.canonical_jsonl_path()),
        jsonl_before: observed_source,
        jsonl_before_content_sha256: source
            .map(JsonlSourceSnapshot::content_sha256)
            .map(str::to_string),
        base_authority_sha256: base_authority.authority_path_sha256().to_string(),
        base_before: base_state,
        base_before_content_sha256: base_source
            .as_ref()
            .map(JsonlSourceSnapshot::content_sha256)
            .map(str::to_string),
        resolution: resolution.to_string(),
        actor: actor.to_string(),
        event_attribution: storage.pending_event_attribution_for_review(),
        capacity_policy: storage.workflow_capacity_policy_for_review(),
        retention_days,
        export_as_of: merge_as_of,
        changed_kept_issue_ids: changed_kept_ids,
        kept_issue_witnesses,
        deleted_issue_ids: deleted_ids,
        note_witnesses,
        database_before,
    };
    let pending_receipt = storage.apply_sync_merge_atomically(
        &changed_kept,
        &report.deleted,
        &report.notes,
        &intent,
    )?;
    let capacity_warnings = pending_receipt.capacity_warnings.clone();
    let _ = storage.take_capacity_warnings();

    let reconciled = reconcile_pending_sync_merge_artifacts(
        storage,
        db_path,
        path_policy,
        args,
        show_progress,
        history_config,
        source,
        jsonl_authority,
        base_source.as_ref(),
        &base_authority,
        pending_receipt,
        no_db,
    )
    .map_err(|source| {
        if no_db {
            source
        } else {
            BeadsError::CommittedStateUnwitnessed {
                operation: "sync merge artifact reconciliation".to_string(),
                source: Box::new(source),
            }
        }
    })?;
    Ok(SyncDispatchCompletion {
        published_source: Some(reconciled.published_source),
        owned_jsonl_authority,
        pending_merge: Some(PendingSyncMergeCompletion {
            receipt: reconciled.terminal_receipt,
            base_authority,
        }),
        deferred_output: Some(DeferredSyncOutput::Merge {
            report,
            resolution: resolution.to_string(),
            capacity_warnings,
            use_json,
        }),
    })
}

/// Render merge conflicts with rich formatting.
fn render_merge_conflicts_rich(
    conflicts: &[(String, crate::sync::ConflictType)],
    ctx: &OutputContext,
) {
    let console = Console::default();
    let theme = ctx.theme();

    let mut text = Text::new("");
    text.append_styled("⚠ ", theme.error.clone());
    text.append_styled(
        &format!("{} merge conflict(s) detected:\n\n", conflicts.len()),
        theme.error.clone(),
    );

    for (i, (id, kind)) in conflicts.iter().enumerate() {
        let prefix = if i == conflicts.len() - 1 {
            "└──"
        } else {
            "├──"
        };
        text.append_styled(prefix, theme.muted.clone());
        text.append(" ");
        text.append_styled(id, theme.issue_id.clone());
        text.append(": ");
        text.append_styled(&format!("{kind:?}"), theme.error.clone());
        text.append("\n");
    }

    text.append("\n");
    text.append_styled("Hint: ", theme.dimmed.clone());
    text.append("Use --force-db to keep local DB changes, --force-jsonl to keep JSONL changes, or --force to keep the newer timestamp.");

    let panel = Panel::from_rich_text(&text, ctx.width())
        .title(Text::new("Merge Conflicts"))
        .box_style(theme.box_style);
    console.print_renderable(&panel);
}

/// One loud line per id that two clones minted for different issues (#512).
fn id_collision_warning(collision: &crate::sync::IdCollision) -> String {
    let side = |side: crate::sync::MergeSide| match side {
        crate::sync::MergeSide::Local => "local database",
        crate::sync::MergeSide::External => "JSONL",
    };
    let id = &collision.id;
    let kept = side(collision.kept_side);
    let moved = side(collision.relocated_side);
    let title = sanitize_terminal_inline(&collision.relocated_title);
    let new_id = &collision.relocated_id;
    if collision.already_relocated {
        format!(
            "ID collision: {id} named two different issues. Kept the {kept} issue as {id}; the {moved} issue \"{title}\" was already renumbered to {new_id} by another clone and was merged there."
        )
    } else {
        format!(
            "ID collision: {id} named two different issues. Kept the {kept} issue as {id} and renumbered the {moved} issue \"{title}\" to {new_id}; update any external references to {id} that meant it."
        )
    }
}

/// Render merge result with rich formatting.
fn render_merge_result_rich(report: &crate::sync::MergeReport, ctx: &OutputContext) {
    let console = Console::default();
    let theme = ctx.theme();

    let mut text = Text::new("");

    // Success indicator
    text.append_styled("✓ ", theme.success.clone());
    text.append_styled("3-Way Merge Complete", theme.success.clone());
    text.append("\n\n");

    // Kept/Updated count
    text.append_styled("Kept/Updated  ", theme.dimmed.clone());
    text.append_styled(&report.kept.len().to_string(), theme.accent.clone());
    text.append_styled(" issues", theme.dimmed.clone());
    text.append("\n");

    // Deleted count
    text.append_styled("Deleted       ", theme.dimmed.clone());
    if report.deleted.is_empty() {
        text.append("0");
    } else {
        text.append_styled(&report.deleted.len().to_string(), theme.warning.clone());
    }
    text.append_styled(" issues", theme.dimmed.clone());
    text.append("\n");

    // Notes section
    if !report.notes.is_empty() {
        text.append("\n");
        text.append_styled("Notes:\n", theme.dimmed.clone());
        for (i, (id, note)) in report.notes.iter().enumerate() {
            let prefix = if i == report.notes.len() - 1 {
                "└──"
            } else {
                "├──"
            };
            text.append_styled(prefix, theme.muted.clone());
            text.append(" ");
            text.append_styled(id, theme.issue_id.clone());
            text.append(": ");
            text.append_styled(note, theme.muted.clone());
            text.append("\n");
        }
    }

    // Final status
    text.append("\n");
    text.append_styled("✓ ", theme.success.clone());
    text.append_styled("Base snapshot updated\n", theme.muted.clone());
    text.append_styled("✓ ", theme.success.clone());
    text.append_styled("JSONL exported", theme.muted.clone());

    let panel = Panel::from_rich_text(&text, ctx.width())
        .title(Text::new("Merge"))
        .box_style(theme.box_style);
    console.print_renderable(&panel);
}

#[cfg(test)]
mod tests {
    use super::{
        GitExportStatus, SyncOperation, additive_conflict_human_lines,
        auto_rebuild_semantic_conflict_field, auto_rebuild_semantic_flag_conflict_reason,
        classify_sync_status_workspace, detect_prefix_from_jsonl,
        fresh_force_import_maintenance_gate_applies, jsonl_contains_duplicate_external_refs,
        jsonl_contains_prefix_mismatch, merge_conflict_resolution, prepare_sync_startup,
        should_defer_jsonl_recovery, should_render_human_sync_output, sync_operation,
        validate_operator_requested_sync_path, validate_sync_mode_args, validate_sync_paths,
        write_manifest_atomically,
    };
    use crate::cli::SyncArgs;
    use crate::config::{self, CliOverrides};
    use crate::error::BeadsError;
    use crate::model::{Dependency, DependencyType, Issue, IssueType, Priority, Status};
    use crate::output::OutputContext;
    use crate::storage::SqliteStorage;
    use crate::sync::{
        AdditiveReconcileConfig, ConflictResolution, PreservedIssue, capture_jsonl_source_snapshot,
        dirty_issues_missing_from_jsonl, plan_additive_reconcile, restore_preserved_issues,
        scan_jsonl_for_tombstone_filter, snapshot_dirty_live_issues, snapshot_tombstones,
        tombstones_missing_from_jsonl_tombstones,
    };
    use chrono::Utc;
    use std::collections::{BTreeSet, HashSet};
    use std::fs;
    use std::path::{Path, PathBuf};
    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,
            prerequisites: 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,
            bypassed_policy: None,
            bypass_reason: None,
            policy_gates_fired: None,
            due_at: None,
            defer_until: None,
            external_ref: None,
            source_system: None,
            source_repo: None,
            source_repo_path: None,
            agent_context: 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 additive_conflict_human_renderer_is_bounded_and_redacts_embedded_relation_values() {
        let temp = TempDir::new().unwrap();
        let jsonl_path = temp.path().join("issues.jsonl");
        let storage = SqliteStorage::open_memory().unwrap();
        let private_target = "external:/tmp/private-agent-path\n\u{1b}[31m";
        let mut incoming = Vec::new();

        for id in ["bd-safe-a", "bd-safe-b"] {
            let existing = make_test_issue(id, "Same scalar payload");
            storage.upsert_issue_for_import(&existing).unwrap();
            let mut source = existing.clone();
            source.dependencies.push(Dependency {
                issue_id: id.to_string(),
                depends_on_id: private_target.to_string(),
                dep_type: DependencyType::Blocks,
                created_at: source.created_at,
                created_by: Some("fixture".to_string()),
                metadata: Some("{\"private\":\"/tmp/private-agent-path\"}".to_string()),
                thread_id: Some("\u{1b}]8;;file:///tmp/private-agent-path\u{7}".to_string()),
            });
            incoming.push(source);
        }
        let mut bytes = Vec::new();
        for issue in &incoming {
            serde_json::to_writer(&mut bytes, issue).unwrap();
            bytes.push(b'\n');
        }
        fs::write(&jsonl_path, bytes).unwrap();
        let plan = plan_additive_reconcile(
            &storage,
            &jsonl_path,
            &AdditiveReconcileConfig {
                beads_dir: Some(temp.path().to_path_buf()),
                database_path: None,
                allow_external_jsonl: false,
                source_authoritative_ids: BTreeSet::new(),
            },
        )
        .unwrap();
        assert_eq!(
            plan.receipt().conflict_reasons.get("shared_relation_drift"),
            Some(&2)
        );

        let human = additive_conflict_human_lines(plan.receipt(), 1).join("\n");
        let robot = serde_json::to_string(plan.receipt()).unwrap();
        let tracing_payload = format!(
            "{:?}{:?}",
            plan.receipt().conflict_witnesses,
            plan.receipt().conflict_relation_diffs
        );
        for rendered in [&human, &robot, &tracing_payload] {
            assert!(!rendered.contains("/tmp/private-agent-path"), "{rendered}");
            assert!(!rendered.contains('\u{1b}'), "{rendered}");
            assert!(!rendered.contains('\u{7}'), "{rendered}");
        }
        assert!(
            human.contains("issue_ids=1/2")
                && human.contains("witnesses=1/2")
                && human.contains("truncated=true"),
            "{human}"
        );
        assert!(
            human.contains(&plan.receipt().conflict_issue_ids_sha256)
                && human.contains(&plan.receipt().conflict_relation_diffs_sha256),
            "{human}"
        );
    }

    #[cfg(unix)]
    #[test]
    fn test_write_manifest_atomically_rejects_existing_temp_symlink() {
        use std::os::unix::fs::symlink;

        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        fs::create_dir_all(&beads_dir).unwrap();

        let manifest_path = beads_dir.join(".manifest.json");
        let temp_path = manifest_path.with_extension(format!("json.{}.tmp", std::process::id()));
        let outside_target = temp.path().join("outside.json");
        fs::write(&outside_target, "preserve").unwrap();
        symlink(&outside_target, &temp_path).unwrap();

        let err = write_manifest_atomically(&manifest_path, &serde_json::json!({ "ok": true }))
            .unwrap_err();

        assert!(
            matches!(&err, BeadsError::Config(_)),
            "unexpected error: {err:?}"
        );
        let message = if let BeadsError::Config(message) = &err {
            message.as_str()
        } else {
            ""
        };
        assert!(
            message.contains("failed to create temp manifest file"),
            "unexpected message: {message}"
        );
        assert_eq!(
            fs::read_to_string(&outside_target).unwrap(),
            "preserve",
            "manifest temp symlink target must not receive manifest bytes"
        );
        assert!(
            !manifest_path.exists(),
            "failed manifest write must not install a manifest"
        );
        assert!(
            fs::symlink_metadata(&temp_path)
                .unwrap()
                .file_type()
                .is_symlink(),
            "rejected pre-existing temp symlink should be left untouched"
        );
    }

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

        let manifest_path = beads_dir.join(".manifest.json");
        let stale_temp_path =
            manifest_path.with_extension(format!("json.{}.tmp", std::process::id()));
        fs::write(&stale_temp_path, "stale temp").unwrap();

        write_manifest_atomically(&manifest_path, &serde_json::json!({ "ok": true })).unwrap();

        let manifest = fs::read_to_string(&manifest_path).unwrap();
        assert!(
            manifest.contains("\"ok\": true"),
            "manifest should be written through a collision-free temp path"
        );
        assert_eq!(
            fs::read_to_string(&stale_temp_path).unwrap(),
            "stale temp",
            "stale temp collision should be left untouched"
        );
    }

    #[test]
    fn should_render_human_sync_output_preserves_quiet_json_semantics() {
        let quiet_ctx = OutputContext::from_flags(false, true, true);
        let plain_ctx = OutputContext::from_flags(false, false, true);

        assert!(!should_render_human_sync_output(&quiet_ctx, false));
        assert!(should_render_human_sync_output(&quiet_ctx, true));
        assert!(should_render_human_sync_output(&plain_ctx, false));
        assert!(should_render_human_sync_output(&plain_ctx, true));
    }

    #[test]
    fn fresh_force_import_maintenance_gate_only_applies_to_clean_empty_force_loads() {
        let force_import = SyncArgs {
            import_only: true,
            force: true,
            ..SyncArgs::default()
        };
        assert!(fresh_force_import_maintenance_gate_applies(
            &force_import,
            true,
            true
        ));
        assert!(!fresh_force_import_maintenance_gate_applies(
            &force_import,
            false,
            true
        ));
        assert!(!fresh_force_import_maintenance_gate_applies(
            &force_import,
            true,
            false
        ));

        let rebuild = SyncArgs {
            import_only: true,
            force: true,
            rebuild: true,
            ..SyncArgs::default()
        };
        assert!(!fresh_force_import_maintenance_gate_applies(
            &rebuild, true, true
        ));

        let rename_prefix = SyncArgs {
            import_only: true,
            force: true,
            rename_prefix: true,
            ..SyncArgs::default()
        };
        assert!(!fresh_force_import_maintenance_gate_applies(
            &rename_prefix,
            true,
            true
        ));
    }

    #[test]
    fn sync_operation_witness_is_explicit_read_only_mode() {
        let args = SyncArgs {
            witness: true,
            witness_chunk_lines: 2,
            ..SyncArgs::default()
        };

        assert_eq!(sync_operation(&args), SyncOperation::Witness);
        assert!(!should_defer_jsonl_recovery(&args));
    }

    #[cfg(unix)]
    #[test]
    fn build_base_witness_artifacts_rejects_symlinked_base_snapshot() {
        use super::{SyncPathPolicy, build_base_witness_artifacts};
        use std::os::unix::fs::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 current_jsonl_path = beads_dir.join("issues.jsonl");
        fs::write(&current_jsonl_path, "current\n").unwrap();
        let outside_base_path = outside_dir.join("beads.base.jsonl");
        fs::write(&outside_base_path, "outside\n").unwrap();
        symlink(&outside_base_path, beads_dir.join("beads.base.jsonl")).unwrap();

        let path_policy = SyncPathPolicy {
            jsonl_path: current_jsonl_path.clone(),
            jsonl_temp_path: current_jsonl_path.with_extension("jsonl.tmp"),
            manifest_path: beads_dir.join(".manifest.json"),
            beads_dir,
            is_external: false,
            allow_external_jsonl: false,
        };
        let current_witness = super::build_witness_for_path(&current_jsonl_path, 1, 1)
            .expect("current witness should build");

        let result = build_base_witness_artifacts(&path_policy, 1, 1, &current_witness);
        assert!(
            result.is_err(),
            "base witness should reject symlinked base snapshot"
        );
        let err = result.err().expect("checked error result");

        assert!(
            matches!(&err, BeadsError::Config(message) if message.contains("must not be a symlink")),
            "unexpected error: {err:?}"
        );
    }

    #[test]
    fn test_validate_sync_mode_args_rejects_mode_conflicts() {
        let status_conflict = SyncArgs {
            status: true,
            witness: true,
            witness_chunk_lines: 2,
            ..SyncArgs::default()
        };
        let err = validate_sync_mode_args(&status_conflict)
            .expect_err("status and witness should conflict");
        assert!(matches!(err, BeadsError::Validation { .. }));

        let status_flush_conflict = SyncArgs {
            status: true,
            flush_only: true,
            ..SyncArgs::default()
        };
        let err = validate_sync_mode_args(&status_flush_conflict)
            .expect_err("status and flush-only should conflict");
        assert!(matches!(err, BeadsError::Validation { .. }));

        let merge_conflict = SyncArgs {
            merge: true,
            witness: true,
            witness_chunk_lines: 2,
            ..SyncArgs::default()
        };
        let err = validate_sync_mode_args(&merge_conflict)
            .expect_err("merge and witness should conflict");
        assert!(matches!(err, BeadsError::Validation { .. }));

        let reconcile_conflict = SyncArgs {
            reconcile_additive: true,
            status: true,
            ..SyncArgs::default()
        };
        let err = validate_sync_mode_args(&reconcile_conflict)
            .expect_err("additive reconciliation and status should conflict");
        assert!(matches!(err, BeadsError::Validation { .. }));
    }

    #[test]
    fn test_validate_sync_mode_args_rejects_zero_witness_chunk_lines() {
        let args = SyncArgs {
            witness: true,
            witness_chunk_lines: 0,
            ..SyncArgs::default()
        };

        let err = validate_sync_mode_args(&args).expect_err("zero witness chunk size should fail");
        assert!(matches!(err, BeadsError::Validation { .. }));
    }

    #[test]
    fn test_validate_sync_mode_args_rejects_zero_witness_parallelism() {
        let args = SyncArgs {
            witness: true,
            witness_chunk_lines: 2,
            witness_parallelism: Some(0),
            ..SyncArgs::default()
        };

        let err = validate_sync_mode_args(&args).expect_err("zero witness parallelism should fail");
        assert!(matches!(err, BeadsError::Validation { .. }));
    }

    #[test]
    fn test_validate_sync_mode_args_rejects_zero_export_parallelism() {
        let args = SyncArgs {
            flush_only: true,
            export_parallelism: Some(0),
            ..SyncArgs::default()
        };

        let err = validate_sync_mode_args(&args).expect_err("zero export parallelism should fail");
        assert!(matches!(err, BeadsError::Validation { .. }));
    }

    #[test]
    fn test_validate_sync_mode_args_rebuild_requires_import_only() {
        let status_rebuild = SyncArgs {
            status: true,
            rebuild: true,
            ..SyncArgs::default()
        };
        let err = validate_sync_mode_args(&status_rebuild)
            .expect_err("status rebuild should be rejected");
        assert!(matches!(&err, BeadsError::Validation { field, .. } if field == "rebuild"));
        assert!(err.to_string().contains("--import-only"));

        let witness_rebuild = SyncArgs {
            witness: true,
            rebuild: true,
            witness_chunk_lines: 2,
            ..SyncArgs::default()
        };
        let err = validate_sync_mode_args(&witness_rebuild)
            .expect_err("witness rebuild should be rejected");
        assert!(matches!(&err, BeadsError::Validation { field, .. } if field == "rebuild"));
        assert!(err.to_string().contains("--import-only"));

        let import_rebuild = SyncArgs {
            import_only: true,
            rebuild: true,
            ..SyncArgs::default()
        };
        validate_sync_mode_args(&import_rebuild).unwrap();
    }

    #[test]
    fn test_validate_sync_mode_args_apply_requires_reviewed_operation() {
        let bare_apply = SyncArgs {
            apply: true,
            ..SyncArgs::default()
        };
        let err = validate_sync_mode_args(&bare_apply)
            .expect_err("apply without additive reconciliation should fail");
        assert!(matches!(&err, BeadsError::Validation { field, .. } if field == "apply"));

        let reconcile_apply = SyncArgs {
            reconcile_additive: true,
            apply: true,
            ..SyncArgs::default()
        };
        let err = validate_sync_mode_args(&reconcile_apply)
            .expect_err("additive apply without a reviewed token must fail");
        assert!(
            matches!(&err, BeadsError::Validation { field, .. } if field == "expect_plan_sha256")
        );

        let valid = SyncArgs {
            expect_plan_sha256: Some("a".repeat(64)),
            ..reconcile_apply.clone()
        };
        validate_sync_mode_args(&valid).unwrap();
        for malformed in [
            "a".to_string(),
            "A".repeat(64),
            "g".repeat(64),
            "a".repeat(65),
        ] {
            let args = SyncArgs {
                expect_plan_sha256: Some(malformed),
                ..reconcile_apply.clone()
            };
            let err = validate_sync_mode_args(&args)
                .expect_err("malformed reviewed tokens must fail preflight");
            assert!(
                matches!(&err, BeadsError::Validation { field, .. } if field == "expect_plan_sha256")
            );
        }
    }

    /// GitHub #473: the advertised additive dry-run invocation must be
    /// reachable. `--reconcile-additive --dry-run` is the same read-only plan
    /// mode as bare `--reconcile-additive`; it must validate instead of
    /// demanding `--reconcile`.
    #[test]
    fn test_validate_sync_mode_args_allows_dry_run_with_reviewed_modes() {
        validate_sync_mode_args(&SyncArgs {
            reconcile_additive: true,
            dry_run: true,
            ..SyncArgs::default()
        })
        .expect("--reconcile-additive --dry-run is the documented plan mode");

        validate_sync_mode_args(&SyncArgs {
            migrate_source_repo_path: true,
            dry_run: true,
            ..SyncArgs::default()
        })
        .expect("--migrate-source-repo-path --dry-run is the documented plan mode");

        // --dry-run still refuses modes where it has no meaning.
        let err = validate_sync_mode_args(&SyncArgs {
            flush_only: true,
            dry_run: true,
            ..SyncArgs::default()
        })
        .expect_err("--dry-run must not silently no-op with --flush-only");
        assert!(matches!(&err, BeadsError::Validation { field, .. } if field == "dry_run"));

        // And combining both reviewed modes is still exactly-one-mode.
        assert!(
            validate_sync_mode_args(&SyncArgs {
                reconcile: true,
                reconcile_additive: true,
                dry_run: true,
                ..SyncArgs::default()
            })
            .is_err()
        );
    }

    #[test]
    fn test_validate_sync_mode_args_accepts_source_repo_path_migration() {
        let dry_run = SyncArgs {
            migrate_source_repo_path: true,
            ..SyncArgs::default()
        };
        validate_sync_mode_args(&dry_run).unwrap();
        assert_eq!(
            sync_operation(&dry_run),
            SyncOperation::MigrateSourceRepoPath
        );
        assert!(should_defer_jsonl_recovery(&dry_run));

        let apply = SyncArgs {
            apply: true,
            expect_plan_sha256: Some("a".repeat(64)),
            ..dry_run.clone()
        };
        validate_sync_mode_args(&apply).unwrap();
    }

    #[test]
    fn test_validate_sync_mode_args_rejects_irrelevant_migration_flags() {
        for args in [
            SyncArgs {
                migrate_source_repo_path: true,
                force: true,
                ..SyncArgs::default()
            },
            SyncArgs {
                migrate_source_repo_path: true,
                rebuild: true,
                ..SyncArgs::default()
            },
            SyncArgs {
                migrate_source_repo_path: true,
                resolve_source_ids: vec!["br-1".to_string()],
                ..SyncArgs::default()
            },
        ] {
            let error = validate_sync_mode_args(&args)
                .expect_err("migration-only mode must reject unrelated mutation flags");
            assert!(matches!(error, BeadsError::Validation { .. }));
        }
    }

    #[test]
    fn test_merge_conflict_resolution_defaults_to_manual() {
        let args = SyncArgs {
            merge: true,
            ..SyncArgs::default()
        };

        assert_eq!(merge_conflict_resolution(&args), ConflictResolution::Manual);
    }

    #[test]
    fn test_merge_conflict_resolution_supports_explicit_winners() {
        let force_db = SyncArgs {
            merge: true,
            force_db: true,
            ..SyncArgs::default()
        };
        let force_jsonl = SyncArgs {
            merge: true,
            force_jsonl: true,
            ..SyncArgs::default()
        };
        let force_newer = SyncArgs {
            merge: true,
            force: true,
            ..SyncArgs::default()
        };

        assert_eq!(
            merge_conflict_resolution(&force_db),
            ConflictResolution::PreferLocal
        );
        assert_eq!(
            merge_conflict_resolution(&force_jsonl),
            ConflictResolution::PreferExternal
        );
        assert_eq!(
            merge_conflict_resolution(&force_newer),
            ConflictResolution::PreferNewer
        );
    }

    #[test]
    fn test_merge_resolution_flags_require_merge_mode() {
        let args = SyncArgs {
            force_db: true,
            ..SyncArgs::default()
        };

        let err = validate_sync_mode_args(&args).expect_err("force-db should require merge");
        assert!(matches!(err, BeadsError::Validation { .. }));
        assert!(err.to_string().contains("--merge"));
    }

    #[test]
    fn test_sync_operation_selects_unspecified_and_explicit_modes() {
        assert_eq!(
            sync_operation(&SyncArgs::default()),
            SyncOperation::Unspecified
        );
        let bare_err = validate_sync_mode_args(&SyncArgs::default())
            .expect_err("bare sync should require an explicit mode");
        assert!(matches!(bare_err, BeadsError::Validation { .. }));
        assert!(bare_err.to_string().contains("--flush-only"));

        let flush = SyncArgs {
            flush_only: true,
            ..SyncArgs::default()
        };
        assert_eq!(sync_operation(&flush), SyncOperation::Flush);

        let merge = SyncArgs {
            merge: true,
            ..SyncArgs::default()
        };
        assert_eq!(sync_operation(&merge), SyncOperation::Merge);

        let import = SyncArgs {
            import_only: true,
            ..SyncArgs::default()
        };
        assert_eq!(sync_operation(&import), SyncOperation::Import);

        let reconcile = SyncArgs {
            reconcile_additive: true,
            ..SyncArgs::default()
        };
        assert_eq!(sync_operation(&reconcile), SyncOperation::ReconcileAdditive);
    }

    #[test]
    fn test_sync_operation_status_is_explicit_read_only_mode() {
        let args = SyncArgs {
            status: true,
            ..SyncArgs::default()
        };

        validate_sync_mode_args(&args).unwrap();
        assert_eq!(sync_operation(&args), SyncOperation::Status);
    }

    #[test]
    fn test_should_defer_jsonl_recovery_for_operator_controlled_import_modes() {
        let rename_import = SyncArgs {
            import_only: true,
            rename_prefix: true,
            ..SyncArgs::default()
        };
        assert!(should_defer_jsonl_recovery(&rename_import));

        let salvage_import = SyncArgs {
            import_only: true,
            skip_invalid_records: true,
            ..SyncArgs::default()
        };
        validate_sync_mode_args(&salvage_import).unwrap();
        assert!(should_defer_jsonl_recovery(&salvage_import));

        let bare_salvage = SyncArgs {
            skip_invalid_records: true,
            ..SyncArgs::default()
        };
        let error = validate_sync_mode_args(&bare_salvage)
            .expect_err("salvage must require explicit import mode");
        assert!(
            matches!(&error, BeadsError::Validation { field, .. } if field == "skip_invalid_records")
        );
        assert!(!should_defer_jsonl_recovery(&bare_salvage));

        for destructive in [
            SyncArgs {
                import_only: true,
                skip_invalid_records: true,
                force: true,
                ..SyncArgs::default()
            },
            SyncArgs {
                import_only: true,
                skip_invalid_records: true,
                rebuild: true,
                ..SyncArgs::default()
            },
            SyncArgs {
                import_only: true,
                skip_invalid_records: true,
                rename_prefix: true,
                ..SyncArgs::default()
            },
        ] {
            let error = validate_sync_mode_args(&destructive)
                .expect_err("salvage must reject destructive import modifiers");
            assert!(
                matches!(&error, BeadsError::Validation { field, .. } if field == "skip_invalid_records")
            );
        }

        let bare_rename = SyncArgs {
            rename_prefix: true,
            ..SyncArgs::default()
        };
        assert!(!should_defer_jsonl_recovery(&bare_rename));

        let status = SyncArgs {
            status: true,
            rename_prefix: true,
            ..SyncArgs::default()
        };
        assert!(!should_defer_jsonl_recovery(&status));

        let flush = SyncArgs {
            flush_only: true,
            rename_prefix: true,
            ..SyncArgs::default()
        };
        assert!(!should_defer_jsonl_recovery(&flush));

        let merge = SyncArgs {
            merge: true,
            rename_prefix: true,
            ..SyncArgs::default()
        };
        assert!(!should_defer_jsonl_recovery(&merge));

        let reconcile_plan = SyncArgs {
            reconcile_additive: true,
            ..SyncArgs::default()
        };
        assert!(should_defer_jsonl_recovery(&reconcile_plan));

        let reconcile_apply = SyncArgs {
            reconcile_additive: true,
            apply: true,
            ..SyncArgs::default()
        };
        assert!(should_defer_jsonl_recovery(&reconcile_apply));
    }

    #[test]
    fn sync_status_fast_open_miss_reuses_caller_write_lock_for_rebuild() {
        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        fs::create_dir_all(&beads_dir).unwrap();
        let db_path = beads_dir.join("beads.db");
        let jsonl_path = beads_dir.join("issues.jsonl");
        let issue = make_test_issue("bd-sync-selflock", "Recovered while caller holds lock");
        fs::write(
            &jsonl_path,
            format!("{}\n", serde_json::to_string(&issue).unwrap()),
        )
        .unwrap();
        // main.rs takes the database-family authority before dispatch and
        // hands it to the command through the override marker; the startup
        // path must reuse that authority instead of contending for it.
        let held = std::sync::Arc::new(
            crate::sync::blocking_database_family_write_lock_with_timeout(
                &beads_dir,
                &db_path,
                Some(1_000),
            )
            .unwrap(),
        );
        let args = SyncArgs {
            status: true,
            ..SyncArgs::default()
        };
        let mut cli = CliOverrides {
            db: Some(db_path.clone()),
            lock_timeout: Some(1),
            read_only_fast_open: true,
            ..CliOverrides::default()
        };
        cli.mark_database_family_lock_held(&beads_dir, &held);

        let startup = prepare_sync_startup(&args, &cli, true)
            .expect("caller-held write lock should not be reacquired on fast-open miss");

        assert!(db_path.is_file(), "missing DB should rebuild from JSONL");
        assert!(
            startup
                .open_result
                .storage
                .id_exists("bd-sync-selflock")
                .unwrap()
        );
    }

    #[test]
    fn test_sync_git_export_status_is_stably_not_probed() {
        let status = GitExportStatus::not_probed();
        assert!(!status.available, "{status:?}");
        assert_eq!(status.reason, "not_probed");
        assert_eq!(status.diagnostic_command, "br vcs-status --json");
        assert!(status.tracked.is_none(), "{status:?}");
        assert!(status.worktree_clean.is_none(), "{status:?}");
        assert!(status.index_clean.is_none(), "{status:?}");
        assert!(status.head_hash.is_none(), "{status:?}");
        assert!(status.worktree_hash.is_none(), "{status:?}");

        let json = serde_json::to_value(&status).unwrap();
        assert_eq!(
            json,
            serde_json::json!({
                "available": false,
                "reason": "not_probed",
                "diagnostic_command": "br vcs-status --json"
            })
        );
    }

    #[test]
    fn test_classify_sync_status_workspace_maps_drift_to_anomaly_codes() {
        let temp = TempDir::new().unwrap();
        let db_path = temp.path().join("beads.db");
        let jsonl_path = temp.path().join("issues.jsonl");
        // Valid SQLite header so file-state probes stay quiet.
        let storage = SqliteStorage::open(&db_path).unwrap();
        drop(storage);
        fs::write(&jsonl_path, "{\"id\":\"bd-x\"}\n").unwrap();

        let healthy = classify_sync_status_workspace(&db_path, &jsonl_path, false, false);
        assert_eq!(healthy.health.as_str(), "healthy", "{healthy:?}");
        assert!(healthy.anomalies.is_empty(), "{healthy:?}");

        let pending_export = classify_sync_status_workspace(&db_path, &jsonl_path, false, true);
        assert_eq!(pending_export.health.as_str(), "degraded");
        let audit = pending_export.audit_record("sync.status");
        assert_eq!(audit.source, "sync.status");
        assert_eq!(audit.anomaly_codes_csv(), "db_newer");

        let diverged = classify_sync_status_workspace(&db_path, &jsonl_path, true, true);
        assert_eq!(diverged.health.as_str(), "degraded");
        assert_eq!(
            diverged.audit_record("sync.status").anomaly_codes_csv(),
            "jsonl_newer,db_newer"
        );
    }

    #[test]
    fn test_classify_sync_status_workspace_conflict_markers_are_unsafe() {
        let temp = TempDir::new().unwrap();
        let db_path = temp.path().join("beads.db");
        let jsonl_path = temp.path().join("issues.jsonl");
        let storage = SqliteStorage::open(&db_path).unwrap();
        drop(storage);
        fs::write(
            &jsonl_path,
            "<<<<<<< HEAD\n{\"id\":\"bd-a\"}\n=======\n{\"id\":\"bd-b\"}\n>>>>>>> theirs\n",
        )
        .unwrap();

        let classification = classify_sync_status_workspace(&db_path, &jsonl_path, false, false);
        assert_eq!(
            classification.health.as_str(),
            "unsafe",
            "{classification:?}"
        );
        assert!(
            classification
                .audit_record("sync.status")
                .anomaly_codes_csv()
                .contains("jsonl_conflict_markers"),
            "{classification:?}"
        );
    }

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

        // Execute status (would need to serialize manually for test)
        let dirty_ids = storage.get_dirty_issue_ids().unwrap();
        assert!(dirty_ids.is_empty());
    }

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

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

        let dirty_ids = storage.get_dirty_issue_ids().unwrap();
        assert!(!dirty_ids.is_empty());
    }

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

        let keep = make_test_issue("bd-keep", "Keep");
        let delete = make_test_issue("bd-delete", "Delete");
        storage.create_issue(&keep, "test").unwrap();
        storage.create_issue(&delete, "test").unwrap();
        storage.add_label("bd-delete", "urgent", "test").unwrap();
        storage
            .add_comment("bd-delete", "test", "preserve this comment")
            .unwrap();
        storage
            .add_dependency("bd-delete", "bd-keep", "blocks", "test")
            .unwrap();
        storage
            .delete_issue("bd-delete", "test", "deleted for rebuild", None)
            .unwrap();

        let tombstones = snapshot_tombstones(&storage);
        assert_eq!(tombstones.len(), 1);
        assert_eq!(tombstones[0].issue.id, "bd-delete");
        assert_eq!(
            tombstones[0].labels.as_ref().unwrap(),
            &vec!["urgent".to_string()]
        );
        assert_eq!(tombstones[0].comments.as_ref().unwrap().len(), 1);
        assert_eq!(tombstones[0].dependencies.as_ref().unwrap().len(), 1);
        assert_eq!(
            tombstones[0].dependencies.as_ref().unwrap()[0].depends_on_id,
            "bd-keep"
        );

        storage.reset_data_tables().unwrap();
        storage.upsert_issue_for_import(&keep).unwrap();
        restore_preserved_issues(&mut storage, &tombstones).unwrap();

        let restored = storage.get_issue("bd-delete").unwrap().unwrap();
        assert_eq!(restored.status, Status::Tombstone);
        assert_eq!(
            storage.get_labels("bd-delete").unwrap(),
            vec!["urgent".to_string()]
        );
        assert_eq!(storage.get_comments("bd-delete").unwrap().len(), 1);
        let dependencies = storage.get_dependencies_full("bd-delete").unwrap();
        assert_eq!(dependencies.len(), 1);
        assert_eq!(dependencies[0].depends_on_id, "bd-keep");

        let dirty_ids = storage.get_dirty_issue_ids().unwrap();
        assert_eq!(dirty_ids, vec!["bd-delete".to_string()]);
    }

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

        let keep = make_test_issue("bd-keep", "Keep");
        let issue = make_test_issue("bd-delete", "Delete");
        storage.create_issue(&keep, "test").unwrap();
        storage.create_issue(&issue, "test").unwrap();
        storage.add_label("bd-delete", "urgent", "test").unwrap();
        storage
            .add_comment("bd-delete", "test", "preserve this comment")
            .unwrap();
        storage
            .add_dependency("bd-delete", "bd-keep", "blocks", "test")
            .unwrap();
        storage
            .delete_issue("bd-delete", "test", "deleted for rebuild", None)
            .unwrap();

        let tombstones = snapshot_tombstones(&storage);

        storage.reset_data_tables().unwrap();
        storage.upsert_issue_for_import(&keep).unwrap();
        storage.execute_raw("DROP TABLE comments").unwrap();

        let err = restore_preserved_issues(&mut storage, &tombstones).unwrap_err();
        assert!(
            err.to_string().contains("comments"),
            "unexpected restore failure: {err}"
        );
        assert!(storage.get_issue("bd-delete").unwrap().is_none());
        assert!(storage.get_labels("bd-delete").unwrap().is_empty());
        assert!(
            storage
                .get_dependencies_full("bd-delete")
                .unwrap()
                .is_empty()
        );
        assert!(storage.get_dirty_issue_ids().unwrap().is_empty());
    }

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

        let first = make_test_issue("bd-first", "First");
        let second = make_test_issue("bd-second", "Second");
        storage.create_issue(&first, "test").unwrap();
        storage.create_issue(&second, "test").unwrap();
        storage
            .add_dependency("bd-first", "bd-second", "blocks", "test")
            .unwrap();
        storage
            .delete_issue("bd-first", "test", "deleted for rebuild", None)
            .unwrap();
        storage
            .delete_issue("bd-second", "test", "deleted for rebuild", None)
            .unwrap();

        let tombstones = snapshot_tombstones(&storage);

        storage.reset_data_tables().unwrap();
        restore_preserved_issues(&mut storage, &tombstones).unwrap();

        let dependencies = storage.get_dependencies_full("bd-first").unwrap();
        assert_eq!(dependencies.len(), 1);
        assert_eq!(dependencies[0].depends_on_id, "bd-second");
        let mut dirty_ids = storage.get_dirty_issue_ids().unwrap();
        dirty_ids.sort();
        assert_eq!(
            dirty_ids,
            vec!["bd-first".to_string(), "bd-second".to_string()]
        );
    }

    #[test]
    fn test_tombstones_missing_from_jsonl_tombstones_only_skips_already_flushed_deletions() {
        let in_jsonl = PreservedIssue {
            issue: make_test_issue("bd-in-jsonl", "in jsonl"),
            labels: Some(vec!["jsonl".to_string()]),
            dependencies: Some(Vec::new()),
            comments: Some(Vec::new()),
        };
        let missing = PreservedIssue {
            issue: make_test_issue("bd-missing", "missing"),
            labels: Some(vec!["local".to_string()]),
            dependencies: Some(Vec::new()),
            comments: Some(Vec::new()),
        };

        let filter = crate::sync::JsonlTombstoneFilter {
            tombstone_ids: HashSet::from(["bd-in-jsonl".to_string()]),
            non_tombstone_updated_at: std::collections::HashMap::new(),
        };
        let filtered =
            tombstones_missing_from_jsonl_tombstones(vec![in_jsonl, missing.clone()], &filter);

        assert_eq!(filtered.len(), 1);
        assert_eq!(filtered[0].issue.id, "bd-missing");
        assert_eq!(filtered[0].labels, missing.labels);
        assert_eq!(filtered[0].dependencies, missing.dependencies);
        assert_eq!(filtered[0].comments, missing.comments);
    }

    #[test]
    fn test_tombstones_missing_from_jsonl_tombstones_blocks_resurrection() {
        // Regression: when the JSONL has an ID as a *non*-tombstone, the
        // preserved tombstone must still overwrite the imported open row.
        // Timestamp ordering cannot resurrect a tombstone; that requires
        // an explicit reopen operation.
        use crate::model::Status;
        use chrono::{Duration, Utc};

        let jsonl_updated_at = Utc::now();
        let mut old_local_tombstone = make_test_issue("bd-contested-older", "older local delete");
        old_local_tombstone.status = Status::Tombstone;
        old_local_tombstone.deleted_at = Some(jsonl_updated_at - Duration::hours(1));
        let old_local_preserved = PreservedIssue {
            issue: old_local_tombstone,
            labels: None,
            dependencies: None,
            comments: None,
        };

        let mut new_local_tombstone = make_test_issue("bd-contested-newer", "newer local delete");
        new_local_tombstone.status = Status::Tombstone;
        new_local_tombstone.deleted_at = Some(jsonl_updated_at + Duration::hours(1));
        let new_local_preserved = PreservedIssue {
            issue: new_local_tombstone,
            labels: None,
            dependencies: None,
            comments: None,
        };

        let mut non_tombstone_map = std::collections::HashMap::new();
        non_tombstone_map.insert("bd-contested-older".to_string(), jsonl_updated_at);
        non_tombstone_map.insert("bd-contested-newer".to_string(), jsonl_updated_at);

        let filter = crate::sync::JsonlTombstoneFilter {
            tombstone_ids: HashSet::new(),
            non_tombstone_updated_at: non_tombstone_map,
        };

        let filtered = tombstones_missing_from_jsonl_tombstones(
            vec![old_local_preserved, new_local_preserved],
            &filter,
        );

        assert_eq!(filtered.len(), 2);
        let filtered_ids: HashSet<_> = filtered
            .iter()
            .map(|tombstone| tombstone.issue.id.as_str())
            .collect();
        assert!(filtered_ids.contains("bd-contested-older"));
        assert!(filtered_ids.contains("bd-contested-newer"));
    }

    #[test]
    fn test_dirty_issues_missing_from_jsonl_keeps_only_unreproducible_rows() {
        // GitHub #394: dirty live issues survive a rebuild only when the
        // JSONL cannot reproduce them — absent entirely, or strictly older.
        use chrono::{Duration, Utc};

        let jsonl_updated_at = Utc::now();

        // Never flushed anywhere: must be preserved.
        let never_flushed = PreservedIssue {
            issue: make_test_issue("bd-never-flushed", "db only"),
            labels: Some(vec!["local".to_string()]),
            dependencies: Some(Vec::new()),
            comments: Some(Vec::new()),
        };

        // JSONL copy is as new as the local row: rebuild's import restores
        // an identical row, no preservation needed.
        let mut jsonl_current = make_test_issue("bd-jsonl-current", "flushed");
        jsonl_current.updated_at = jsonl_updated_at;
        let jsonl_current_preserved = PreservedIssue {
            issue: jsonl_current,
            labels: None,
            dependencies: None,
            comments: None,
        };

        // Local edit newer than the JSONL copy: must be preserved.
        let mut locally_edited = make_test_issue("bd-locally-edited", "edited");
        locally_edited.updated_at = jsonl_updated_at + Duration::hours(1);
        let locally_edited_preserved = PreservedIssue {
            issue: locally_edited,
            labels: None,
            dependencies: None,
            comments: None,
        };

        // JSONL has the ID as a flushed tombstone: the deletion wins over
        // the unflushed local edit (mirrors the import tombstone guard).
        let flushed_delete = PreservedIssue {
            issue: make_test_issue("bd-flushed-delete", "deleted upstream"),
            labels: None,
            dependencies: None,
            comments: None,
        };

        let mut non_tombstone_map = std::collections::HashMap::new();
        non_tombstone_map.insert("bd-jsonl-current".to_string(), jsonl_updated_at);
        non_tombstone_map.insert("bd-locally-edited".to_string(), jsonl_updated_at);
        let filter = crate::sync::JsonlTombstoneFilter {
            tombstone_ids: HashSet::from(["bd-flushed-delete".to_string()]),
            non_tombstone_updated_at: non_tombstone_map,
        };

        let filtered = dirty_issues_missing_from_jsonl(
            vec![
                never_flushed,
                jsonl_current_preserved,
                locally_edited_preserved,
                flushed_delete,
            ],
            &filter,
        );

        let filtered_ids: HashSet<_> = filtered
            .iter()
            .map(|preserved| preserved.issue.id.as_str())
            .collect();
        assert_eq!(
            filtered_ids,
            HashSet::from(["bd-never-flushed", "bd-locally-edited"])
        );
    }

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

        let live = make_test_issue("bd-live", "Live dirty issue");
        let deleted = make_test_issue("bd-deleted", "Deleted issue");
        storage.create_issue(&live, "test").unwrap();
        storage.create_issue(&deleted, "test").unwrap();
        storage.add_label("bd-live", "urgent", "test").unwrap();
        storage
            .add_comment("bd-live", "test", "unflushed comment")
            .unwrap();
        storage
            .add_dependency("bd-live", "bd-deleted", "related", "test")
            .unwrap();
        storage
            .delete_issue("bd-deleted", "test", "gone", None)
            .unwrap();

        // Both rows are dirty (nothing has been flushed), but the tombstone
        // belongs to the `snapshot_tombstones` pass, not this one.
        let dirty = snapshot_dirty_live_issues(&storage);
        assert_eq!(dirty.len(), 1);
        assert_eq!(dirty[0].issue.id, "bd-live");
        assert_eq!(
            dirty[0].labels.as_ref().unwrap(),
            &vec!["urgent".to_string()]
        );
        assert_eq!(dirty[0].comments.as_ref().unwrap().len(), 1);
        assert_eq!(dirty[0].dependencies.as_ref().unwrap().len(), 1);

        // Round-trip: after a wipe (stand-in for the rebuild), restoring
        // the tombstone pass then the dirty pass — the order the rebuild
        // wiring uses — brings the live issue back and re-marks it dirty.
        let tombstones = snapshot_tombstones(&storage);
        storage.reset_data_tables().unwrap();
        restore_preserved_issues(&mut storage, &tombstones).unwrap();
        restore_preserved_issues(&mut storage, &dirty).unwrap();
        let restored = storage.get_issue("bd-live").unwrap().unwrap();
        assert_eq!(restored.title, "Live dirty issue");
        assert_eq!(
            storage.get_dependencies_full("bd-live").unwrap().len(),
            1,
            "dependency to the restored tombstone must survive"
        );
        assert!(
            storage
                .get_dirty_issue_ids()
                .unwrap()
                .contains(&"bd-live".to_string())
        );
    }

    #[test]
    fn test_scan_jsonl_for_tombstone_filter_rejects_duplicate_issue_ids() {
        let temp_dir = tempfile::tempdir().unwrap();
        let jsonl_path = temp_dir.path().join("duplicate-tombstones.jsonl");
        let mut first = make_test_issue("bd-dup", "first");
        first.status = Status::Tombstone;
        let second = make_test_issue("bd-dup", "second");
        let content = format!(
            "{}\n{}\n",
            serde_json::to_string(&first).unwrap(),
            serde_json::to_string(&second).unwrap()
        );
        std::fs::write(&jsonl_path, content).unwrap();

        let err = scan_jsonl_for_tombstone_filter(&jsonl_path).unwrap_err();
        assert!(
            matches!(
                &err,
                BeadsError::Config(message)
                    if message.contains("Duplicate issue id 'bd-dup'")
            ),
            "expected duplicate-id config error, got {err:?}"
        );
    }

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

        let issue = make_test_issue("bd-delete", "Delete");
        storage.create_issue(&issue, "test").unwrap();
        storage
            .delete_issue("bd-delete", "test", "deleted for rebuild", None)
            .unwrap();

        storage.execute_raw("DROP TABLE comments").unwrap();
        storage.execute_raw("DROP TABLE labels").unwrap();
        storage.execute_raw("DROP TABLE dependencies").unwrap();

        let tombstones = snapshot_tombstones(&storage);
        assert_eq!(tombstones.len(), 1);
        assert_eq!(tombstones[0].issue.id, "bd-delete");
        assert_eq!(tombstones[0].issue.status, Status::Tombstone);
        assert!(tombstones[0].labels.is_none());
        assert!(tombstones[0].dependencies.is_none());
        assert!(tombstones[0].comments.is_none());
    }

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

        let open_issue = make_test_issue("bd-open", "Open");
        let delete_issue = make_test_issue("bd-delete", "Delete");
        storage.create_issue(&open_issue, "test").unwrap();
        storage.create_issue(&delete_issue, "test").unwrap();
        storage
            .delete_issue("bd-delete", "test", "deleted for rebuild", None)
            .unwrap();

        storage
            .execute_raw("UPDATE issues SET updated_at = 'not-a-datetime' WHERE id = 'bd-open'")
            .unwrap();

        let tombstones = snapshot_tombstones(&storage);
        assert_eq!(tombstones.len(), 1);
        assert_eq!(tombstones[0].issue.id, "bd-delete");
        assert_eq!(tombstones[0].issue.status, Status::Tombstone);
    }

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

        let issue = make_test_issue("bd-delete", "Delete");
        storage.create_issue(&issue, "test").unwrap();
        storage
            .delete_issue("bd-delete", "test", "deleted for rebuild", None)
            .unwrap();

        storage.execute_raw("DROP TABLE issues").unwrap();

        let tombstones = snapshot_tombstones(&storage);
        assert!(tombstones.is_empty());
    }

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

        let jsonl_path = beads_dir.join("nested").join("issues.jsonl");
        let policy = validate_sync_paths(&beads_dir, &jsonl_path, false).expect("path policy");

        assert_eq!(policy.jsonl_path, jsonl_path);
        assert!(!policy.is_external);
        assert!(!policy.allow_external_jsonl);
    }

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

        let jsonl_path = temp
            .path()
            .join("external")
            .join("nested")
            .join("issues.jsonl");
        let policy = validate_sync_paths(&beads_dir, &jsonl_path, true).expect("path policy");

        assert_eq!(policy.jsonl_path, jsonl_path);
        assert!(policy.is_external);
        assert!(policy.allow_external_jsonl);
    }

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

        let db_path = external_dir.join("beads.db");
        let jsonl_path = external_dir.join("issues.jsonl");
        let allow_external_jsonl =
            config::implicit_external_jsonl_allowed(&beads_dir, &db_path, &jsonl_path);
        assert!(allow_external_jsonl);

        let policy = validate_sync_paths(&beads_dir, &jsonl_path, allow_external_jsonl)
            .expect("path policy");

        assert_eq!(policy.jsonl_path, jsonl_path);
        assert!(policy.is_external);
        assert!(policy.allow_external_jsonl);
    }

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

        let jsonl_path = external_dir.join("issues.jsonl");
        let err = validate_sync_paths(&beads_dir, &jsonl_path, false).unwrap_err();

        assert!(
            matches!(&err, BeadsError::Config(_)),
            "unexpected error: {err:?}"
        );
        let message = if let BeadsError::Config(message) = &err {
            message.as_str()
        } else {
            ""
        };
        assert!(
            message.contains("--allow-external-jsonl"),
            "unexpected message: {message}"
        );
    }

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

        let err =
            validate_operator_requested_sync_path(&beads_dir, Path::new(".git/../issues.jsonl"))
                .unwrap_err();

        assert!(
            matches!(&err, BeadsError::Config(_)),
            "unexpected error: {err:?}"
        );
        let message = if let BeadsError::Config(message) = &err {
            message.as_str()
        } else {
            ""
        };
        assert!(
            message.contains(".git") || message.contains("git"),
            "unexpected message: {message}"
        );
    }

    #[cfg(unix)]
    #[test]
    fn test_validate_sync_paths_rejects_internal_parent_symlink_escape_with_opt_in() {
        use std::os::unix::fs::symlink;

        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        let external_dir = temp.path().join("external");
        fs::create_dir_all(&beads_dir).unwrap();
        fs::create_dir_all(&external_dir).unwrap();

        let symlink_parent = beads_dir.join("external-link");
        symlink(&external_dir, &symlink_parent).unwrap();

        let jsonl_path = symlink_parent.join("issues.jsonl");
        let err = validate_sync_paths(&beads_dir, &jsonl_path, true).unwrap_err();

        assert!(
            matches!(&err, BeadsError::Config(_)),
            "unexpected error: {err:?}"
        );
        let message = if let BeadsError::Config(message) = &err {
            message.as_str()
        } else {
            ""
        };
        assert!(message.contains("symlink"), "unexpected message: {message}");
    }

    #[cfg(unix)]
    #[test]
    fn test_validate_sync_paths_rejects_symlinked_git_parent_with_opt_in() {
        use std::os::unix::fs::symlink;

        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        let git_dir = temp.path().join(".git");
        fs::create_dir_all(&beads_dir).unwrap();
        fs::create_dir_all(&git_dir).unwrap();

        let git_link = temp.path().join("git-link");
        symlink(&git_dir, &git_link).unwrap();

        let jsonl_path = git_link.join("issues.jsonl");
        let err = validate_sync_paths(&beads_dir, &jsonl_path, true).unwrap_err();

        assert!(
            matches!(&err, BeadsError::Config(_)),
            "unexpected error: {err:?}"
        );
        let message = if let BeadsError::Config(message) = &err {
            message.as_str()
        } else {
            ""
        };
        assert!(
            message.contains(".git") || message.contains("git"),
            "unexpected message: {message}"
        );
    }

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

        let traversal_path = PathBuf::from("../outside/issues.jsonl");
        let err = validate_sync_paths(&beads_dir, &traversal_path, true).unwrap_err();

        assert!(
            matches!(&err, BeadsError::Config(_)),
            "unexpected error: {err:?}"
        );
        let message = if let BeadsError::Config(message) = &err {
            message.as_str()
        } else {
            ""
        };
        assert!(
            message.contains("traversal"),
            "unexpected message: {message}"
        );
    }

    #[cfg(unix)]
    #[test]
    fn test_validate_sync_paths_rejects_symlinked_external_jsonl_with_opt_in() {
        use std::os::unix::fs::symlink;

        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        fs::create_dir_all(&beads_dir).unwrap();

        let outside_target = temp.path().join("outside.jsonl");
        fs::write(&outside_target, "{}\n").unwrap();

        let symlink_path = temp.path().join("linked.jsonl");
        symlink(&outside_target, &symlink_path).unwrap();

        let err = validate_sync_paths(&beads_dir, &symlink_path, true).unwrap_err();

        assert!(
            matches!(&err, BeadsError::Config(_)),
            "unexpected error: {err:?}"
        );
        let message = if let BeadsError::Config(message) = &err {
            message.as_str()
        } else {
            ""
        };
        assert!(message.contains("symlink"), "unexpected message: {message}");
    }

    #[cfg(unix)]
    #[test]
    fn test_validate_sync_paths_rejects_git_symlinked_jsonl_even_with_opt_in() {
        use std::os::unix::fs::symlink;

        let temp = TempDir::new().unwrap();
        let beads_dir = temp.path().join(".beads");
        let git_dir = temp.path().join(".git");
        fs::create_dir_all(&beads_dir).unwrap();
        fs::create_dir_all(&git_dir).unwrap();

        let outside_target = temp.path().join("outside.jsonl");
        fs::write(&outside_target, "{}\n").unwrap();

        let git_link = git_dir.join("linked.jsonl");
        symlink(&outside_target, &git_link).unwrap();

        let err = validate_sync_paths(&beads_dir, &git_link, true).unwrap_err();

        assert!(
            matches!(&err, BeadsError::Config(_)),
            "unexpected error: {err:?}"
        );
        let message = if let BeadsError::Config(message) = &err {
            message.as_str()
        } else {
            ""
        };
        assert!(
            message.contains(".git") || message.contains("git"),
            "unexpected message: {message}"
        );
    }

    #[test]
    fn test_detect_prefix_from_jsonl_supports_hyphenated_prefixes() {
        let temp = TempDir::new().unwrap();
        let jsonl_path = temp.path().join("issues.jsonl");
        let issue = make_test_issue("document-intelligence-0sa", "Hyphenated Prefix");
        fs::write(
            &jsonl_path,
            format!("{}\n", serde_json::to_string(&issue).unwrap()),
        )
        .unwrap();

        let source = capture_jsonl_source_snapshot(&jsonl_path).unwrap();
        assert_eq!(
            detect_prefix_from_jsonl(&source).unwrap(),
            Some("document-intelligence".to_string())
        );
    }

    #[test]
    fn test_detect_prefix_from_jsonl_rejects_malformed_before_prefix() {
        let temp = TempDir::new().unwrap();
        let jsonl_path = temp.path().join("issues.jsonl");
        let issue = make_test_issue("foreign-0sa", "Foreign Prefix");
        fs::write(
            &jsonl_path,
            format!("{{not-json\n{}\n", serde_json::to_string(&issue).unwrap()),
        )
        .unwrap();

        let source = capture_jsonl_source_snapshot(&jsonl_path).unwrap();
        let err = detect_prefix_from_jsonl(&source).unwrap_err();
        assert!(
            matches!(err, BeadsError::Config(ref message) if message.contains("Invalid JSON at line 1")),
            "unexpected error: {err:?}"
        );
    }

    #[test]
    fn test_detect_prefix_from_jsonl_validates_entire_file_before_returning_prefix() {
        let temp = TempDir::new().unwrap();
        let jsonl_path = temp.path().join("issues.jsonl");
        let issue = make_test_issue("foreign-0sa", "Foreign Prefix");
        fs::write(
            &jsonl_path,
            format!("{}\n{{not-json\n", serde_json::to_string(&issue).unwrap()),
        )
        .unwrap();

        let source = capture_jsonl_source_snapshot(&jsonl_path).unwrap();
        let err = detect_prefix_from_jsonl(&source).unwrap_err();
        assert!(
            matches!(err, BeadsError::Config(ref message) if message.contains("Invalid JSON at line 2")),
            "unexpected error: {err:?}"
        );
    }

    #[test]
    fn test_auto_rebuild_semantic_flag_conflict_reason_absent_for_default_import_semantics() {
        let args = SyncArgs::default();
        assert!(
            auto_rebuild_semantic_flag_conflict_reason(&args, &CliOverrides::default(), None)
                .is_none()
        );
    }

    #[test]
    fn test_auto_rebuild_semantic_flag_conflict_reason_mentions_rename_prefix_rerun() {
        let args = SyncArgs {
            force: true,
            rename_prefix: true,
            ..SyncArgs::default()
        };

        let reason =
            auto_rebuild_semantic_flag_conflict_reason(&args, &CliOverrides::default(), None)
                .expect("rename-prefix conflict");
        assert!(reason.contains("`--rename-prefix`"), "reason: {reason}");
        assert!(
            reason.contains("`br sync --import-only --force --rename-prefix`"),
            "reason: {reason}"
        );
    }

    #[test]
    fn test_auto_rebuild_semantic_flag_conflict_reason_ignores_orphans_only_request() {
        let args = SyncArgs {
            rebuild: true,
            orphans: Some("resurrect".to_string()),
            ..SyncArgs::default()
        };

        assert!(
            auto_rebuild_semantic_flag_conflict_reason(&args, &CliOverrides::default(), None)
                .is_none()
        );
    }

    #[test]
    fn test_auto_rebuild_semantic_flag_conflict_reason_mentions_both_flags() {
        let args = SyncArgs {
            force: true,
            rebuild: true,
            rename_prefix: true,
            orphans: Some("skip".to_string()),
            ..SyncArgs::default()
        };

        let reason =
            auto_rebuild_semantic_flag_conflict_reason(&args, &CliOverrides::default(), None)
                .expect("combined conflict");
        assert!(reason.contains("`--rename-prefix`"), "reason: {reason}");
        assert!(
            reason.contains("`br sync --import-only --force --rebuild --rename-prefix`"),
            "reason: {reason}"
        );
    }

    #[test]
    fn test_auto_rebuild_semantic_flag_conflict_reason_preserves_custom_db_override() {
        let args = SyncArgs {
            force: true,
            rename_prefix: true,
            ..SyncArgs::default()
        };

        let custom_db = Path::new("/tmp/custom db.sqlite");
        let reason = auto_rebuild_semantic_flag_conflict_reason(
            &args,
            &CliOverrides::default(),
            Some(custom_db),
        )
        .expect("rename-prefix conflict");
        assert!(
            reason.contains(
                "`br --db '/tmp/custom db.sqlite' sync --import-only --force --rename-prefix`"
            ),
            "reason: {reason}"
        );
    }

    #[test]
    fn test_auto_rebuild_semantic_flag_conflict_reason_preserves_external_jsonl_flag() {
        let args = SyncArgs {
            force: true,
            rename_prefix: true,
            allow_external_jsonl: true,
            ..SyncArgs::default()
        };

        let reason =
            auto_rebuild_semantic_flag_conflict_reason(&args, &CliOverrides::default(), None)
                .expect("rename-prefix conflict");
        assert!(
            reason
                .contains("`br sync --import-only --allow-external-jsonl --force --rename-prefix`"),
            "reason: {reason}"
        );
    }

    #[test]
    fn test_auto_rebuild_semantic_flag_conflict_reason_preserves_cli_startup_flags() {
        let args = SyncArgs {
            force: true,
            rename_prefix: true,
            ..SyncArgs::default()
        };
        let cli = CliOverrides {
            json: Some(true),
            allow_stale: Some(true),
            no_auto_import: Some(true),
            no_auto_flush: Some(true),
            lock_timeout: Some(17),
            ..CliOverrides::default()
        };

        let reason = auto_rebuild_semantic_flag_conflict_reason(&args, &cli, None)
            .expect("rename-prefix conflict");
        assert!(
            reason.contains(
                "`br --json --allow-stale --no-auto-import --no-auto-flush --lock-timeout 17 sync --import-only --force --rename-prefix`"
            ),
            "reason: {reason}"
        );
    }

    #[test]
    fn test_auto_rebuild_semantic_conflict_field_prefers_explicit_rebuild_then_force() {
        let plain = SyncArgs {
            rename_prefix: true,
            ..SyncArgs::default()
        };
        assert_eq!(
            auto_rebuild_semantic_conflict_field(&plain),
            "rename_prefix"
        );

        let force = SyncArgs {
            force: true,
            rename_prefix: true,
            ..SyncArgs::default()
        };
        assert_eq!(auto_rebuild_semantic_conflict_field(&force), "force");

        let rebuild = SyncArgs {
            force: true,
            rebuild: true,
            rename_prefix: true,
            ..SyncArgs::default()
        };
        assert_eq!(auto_rebuild_semantic_conflict_field(&rebuild), "rebuild");
    }

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

        let matching = make_test_issue("bd-alpha", "Matching");
        let mut tombstone = make_test_issue("other-beta", "Tombstone mismatch");
        tombstone.status = Status::Tombstone;

        fs::write(
            &jsonl_path,
            format!(
                "{}\n{}\n",
                serde_json::to_string(&matching).unwrap(),
                serde_json::to_string(&tombstone).unwrap()
            ),
        )
        .unwrap();

        let source = capture_jsonl_source_snapshot(&jsonl_path).unwrap();
        assert!(!jsonl_contains_prefix_mismatch(&source, "bd").unwrap());

        let slugged = make_test_issue("bd-survey-my-thing-abc123", "Slugged");
        fs::write(
            &jsonl_path,
            format!("{}\n", serde_json::to_string(&slugged).unwrap()),
        )
        .unwrap();

        let source = capture_jsonl_source_snapshot(&jsonl_path).unwrap();
        assert!(
            !jsonl_contains_prefix_mismatch(&source, "bd").unwrap(),
            "slugged IDs generated from prefix bd should not be treated as mismatches"
        );

        let mismatch = make_test_issue("other-gamma", "Mismatch");
        fs::write(
            &jsonl_path,
            format!("{}\n", serde_json::to_string(&mismatch).unwrap()),
        )
        .unwrap();

        let source = capture_jsonl_source_snapshot(&jsonl_path).unwrap();
        assert!(jsonl_contains_prefix_mismatch(&source, "bd").unwrap());
    }

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

        let mut first = make_test_issue("bd-alpha", "First");
        first.external_ref = Some("EXT-123".to_string());
        let mut second = make_test_issue("bd-beta", "Second");
        second.external_ref = Some("EXT-123".to_string());

        fs::write(
            &jsonl_path,
            format!(
                "{}\n{}\n",
                serde_json::to_string(&first).unwrap(),
                serde_json::to_string(&second).unwrap()
            ),
        )
        .unwrap();

        let source = capture_jsonl_source_snapshot(&jsonl_path).unwrap();
        assert!(jsonl_contains_duplicate_external_refs(&source).unwrap());

        second.external_ref = Some("EXT-456".to_string());
        fs::write(
            &jsonl_path,
            format!(
                "{}\n{}\n",
                serde_json::to_string(&first).unwrap(),
                serde_json::to_string(&second).unwrap()
            ),
        )
        .unwrap();

        let source = capture_jsonl_source_snapshot(&jsonl_path).unwrap();
        assert!(!jsonl_contains_duplicate_external_refs(&source).unwrap());
    }
}