tix/sync/

beads.rs

//! Beads (`bd`) interchange — the tix → `bd import` row mapping, its JSONL
//! encoding, and the `bd export` reader that backs the import leg.
//!
//! [beads](https://github.com/gastownhall/beads) is a Dolt-backed, dependency-graph
//! issue tracker. tix integrates with it **only through the `bd` CLI**: this module
//! produces the rows `bd import` consumes and parses what `bd export` emits. It
//! **never** opens the Dolt database, never reads `.beads/embeddeddolt/`, and never
//! treats `.beads/issues.jsonl` as a source of truth. beads owns its schema (it
//! refuses to open a database a newer binary migrated), its single-writer embedded
//! model, and its id assignment; staying on the CLI is how tix respects all three.
//! Writing SQL is a defect, not a shortcut.
//!
//! ## Wire format, pinned against beads 1.3.1
//!
//! A row is one JSON object; `bd export` emits one per line (`_type` is `"issue"`),
//! and `bd import` accepts arrays and JSONL alike. The fields this module writes are
//! the ones `bd import` documents (`title` is the only required one): `description`,
//! `status`, `priority`, `issue_type`, `assignee`, `labels`, `dependencies`,
//! `close_reason`.
//!
//! Dependencies are the part worth stating precisely, because beads stores them on
//! the **dependent** row as `{issue_id, depends_on_id, type}` and derives
//! blocked-ness (and the ready frontier) from them:
//!
//! | tix edge | beads row it lands on | record |
//! |---|---|---|
//! | `A blocks B` (`from` blocks `to`) | B's row | `{issue_id: B, depends_on_id: A, type: "blocks"}` |
//! | `A duplicates B` | A's row | `{issue_id: A, depends_on_id: B, type: "duplicates"}` |
//! | `A related B` | `min(A,B)`'s row | `{issue_id: A, depends_on_id: B, type: "relates-to"}` |
//! | `issue.parent = P` | the issue's own row | `{issue_id: issue, depends_on_id: P, type: "parent-child"}` |
//!
//! Confirmed by hand against `bd` 1.3.1 (`bd dep add B A` writes `blocks` on B with
//! `depends_on_id: A`; `bd dep relate` writes `relates-to`; `--parent` writes
//! `parent-child`), so the table is a measurement, not a guess.
//!
//! ## Identity
//!
//! A tix issue is exported under a **deterministic** id (`tix-<number>`), which is
//! what makes a re-run idempotent: `bd import` upserts on `id` ("a new ID inserts,
//! an existing ID is conditionally rewritten"), and it admits a row whose prefix
//! differs from the workspace's (`SkipPrefixValidation`), so no separate mapping
//! table is needed just to avoid duplicates.
//!
//! ## What beads cannot hold
//!
//! Local fields with no beads column — cycles, milestones, time entries, health,
//! series, projects, epics, files, reviewer, dates, estimate, and the Spec Kitty
//! pointers — do **not** transfer. They are reported as [`BeadsDiagnostic`]s rather
//! than dropped silently (the `unsupported_remote_field` precedent), and the local
//! value is never touched. Fields that are pure local bookkeeping — `filename`,
//! `history`, the `commit` a finished issue records, the GitLab/Linear ids — are
//! neither transferred nor reported: nothing about them is meaningful to beads, and
//! reporting them once per issue would bury the diagnostics that are.

use std::collections::{BTreeMap, HashSet};
use std::fmt;
use std::fmt::Write as _;
use std::path::{Path, PathBuf};

use serde::{Deserialize, Serialize};

use crate::config::Config;
use crate::error::{Result, TixError};
use crate::model::{Issue, Priority};
use crate::relationships::{RelationKind, Relationship};
use crate::storage::Store;
use crate::sync::{CommandOutcome, CommandRunner};
use crate::workflow::{self, CategoryResolution, WorkflowCategory};

/// The tool name every diagnostic from this module renders under. It is the
/// string `Tool::as_str` will return once beads joins the `src/sync` [`Tool`]
/// enum; keeping the spelling here now means the report does not change then.
pub const TOOL: &str = "beads";

/// The binary shelled out to for a push. Overridden per run once `[beads]`
/// configuration lands (#1709).
pub const DEFAULT_BINARY: &str = "bd";

/// The default id prefix for exported rows. The full beads id is
/// `"<prefix>-<number>"`; see the module docs for why it is deterministic.
pub const DEFAULT_ID_PREFIX: &str = "tix";

/// The beads `issue_type` vocabulary a tix `type:` may map onto. Anything else
/// falls back to `task`, because beads validates the field.
const BEADS_TYPES: [&str; 6] = ["bug", "feature", "task", "epic", "chore", "decision"];

/// The deterministic beads id for a tix issue number.
#[must_use]
pub fn beads_id(prefix: &str, number: u32) -> String {
    format!("{prefix}-{number}")
}

// ---------------------------------------------------------------------------
// The row
// ---------------------------------------------------------------------------

fn issue_record_type() -> String {
    "issue".to_string()
}

/// One dependency record, as beads stores it on the dependent's row.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct BeadsDependency {
    /// The dependent issue (the row this record belongs to).
    pub issue_id: String,
    /// The issue depended on.
    pub depends_on_id: String,
    /// The relation kind: `blocks`, `duplicates`, `relates-to`, or `parent-child`.
    #[serde(rename = "type")]
    pub kind: String,
}

/// One `bd import` row. Unknown keys present in a `bd export` line are ignored on
/// read (`deny_unknown_fields` is deliberately off, so beads may add fields), and
/// absent optional fields are omitted on write.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct BeadsRow {
    /// `bd export` writes `"_type": "issue"`; `bd import` does not require it.
    #[serde(rename = "_type", default = "issue_record_type")]
    pub record_type: String,
    pub id: String,
    pub title: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub status: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub priority: Option<u8>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub issue_type: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub assignee: Option<String>,
    /// `bd export` emits the assignee as `owner`; read either, write `assignee`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub owner: Option<String>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub labels: Vec<String>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub dependencies: Vec<BeadsDependency>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub close_reason: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub created_at: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub updated_at: Option<String>,
}

impl BeadsRow {
    /// The effective assignee of a row read back from beads (`assignee`, else the
    /// `owner` spelling `bd export` uses).
    #[must_use]
    pub fn effective_assignee(&self) -> Option<&str> {
        self.assignee.as_deref().or(self.owner.as_deref())
    }
}

// ---------------------------------------------------------------------------
// Diagnostics
// ---------------------------------------------------------------------------

/// One local field beads cannot hold. The shape — and the rendering — match
/// [`crate::sync::RemoteMappingDiagnostic`] so a reader learns one line format;
/// the type is separate only until beads joins the `Tool` enum (#1706).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BeadsDiagnostic {
    /// The local issue this is about, when it is about one. `None` for a
    /// diagnostic about a beads record that maps to no local issue — the import
    /// path can produce one, the export path never does.
    ///
    /// It exists so a caller can report only the diagnostics belonging to what it
    /// actually sent: a repeat export that sends nothing must not replay three
    /// hundred "cannot represent" lines about issues it did not touch.
    pub number: Option<u32>,
    /// What the diagnostic is about, e.g. `#1705 Investigate beads`.
    pub subject: String,
    /// The local field that cannot be represented.
    pub field: String,
    /// Human-readable reason, safe for CLI output and the TUI.
    pub message: String,
}

impl fmt::Display for BeadsDiagnostic {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "{TOOL}: {} cannot represent this local data ({}) — {}",
            self.field, self.subject, self.message
        )
    }
}

// ---------------------------------------------------------------------------
// The mapping
// ---------------------------------------------------------------------------

/// The banded `bd` status for a tix column, by the workflow **category** service
/// (never by comparing column names).
///
/// `completed` → `closed`, `active` → `in_progress`, everything else → `open`.
/// `blocked` is deliberately `open`: beads derives blocked-ness from the `blocks`
/// dependencies this module also sends, so mapping a tix `blocked` column onto
/// beads' own `blocked` status would set the two mechanisms against each other.
/// `canceled` → `closed` with `close_reason = "canceled"`, which is how the
/// distinction survives on a tracker with no canceled status.
#[must_use]
pub fn status_for(issue: &Issue, resolution: &CategoryResolution) -> &'static str {
    match workflow::category_of(issue, resolution) {
        Some(WorkflowCategory::Completed) => "closed",
        Some(WorkflowCategory::Canceled) => "closed",
        Some(WorkflowCategory::Active) => "in_progress",
        _ => "open",
    }
}

/// The `close_reason` to record, or `None`. Only a `canceled` issue carries one:
/// it is the one category whose distinction `status: closed` alone would erase.
#[must_use]
pub fn close_reason_for(issue: &Issue, resolution: &CategoryResolution) -> Option<String> {
    match workflow::category_of(issue, resolution) {
        Some(WorkflowCategory::Canceled) => Some("canceled".to_string()),
        _ => None,
    }
}

/// Map a tix priority onto beads' `0..=4` band (0 critical … 4 backlog), or `None`
/// to omit the field.
///
/// tix's vocabulary has no "critical", so `high` is `1` rather than `0` — claiming
/// urgency tix never expressed would be a guess. An explicitly unset priority
/// (`priority: ''`) omits the field; any other free-form value is exported as `2`
/// (medium) and reported.
#[must_use]
pub fn priority_for(priority: &Priority) -> (Option<u8>, Option<String>) {
    match priority {
        Priority::Low => (Some(3), None),
        Priority::Medium => (Some(2), None),
        Priority::High => (Some(1), None),
        Priority::Other(raw) if raw.trim().is_empty() => (None, None),
        Priority::Other(raw) => (
            Some(2),
            Some(format!(
                "priority `{raw}` has no beads band; exported as 2 (medium)"
            )),
        ),
    }
}

/// Map a tix `type:` onto a beads `issue_type`. Unknown values fall back to `task`
/// (beads validates the field against its own vocabulary) and are reported.
#[must_use]
pub fn issue_type_for(issue: &Issue) -> (String, Option<String>) {
    match issue.type_.as_deref() {
        None => ("task".to_string(), None),
        Some(t) if BEADS_TYPES.contains(&t) => (t.to_string(), None),
        Some(t) => (
            "task".to_string(),
            Some(format!(
                "type `{t}` is not a beads issue_type; exported as `task`"
            )),
        ),
    }
}

/// Every local field consumed by [`plan`], so a reader can see the whole mapping in
/// one place. Anything else on the issue is local bookkeeping with no remote
/// meaning and is not reported: `filename`, `history`, `created`/`updated`,
/// `gitlab_id`/`linear_id`, and the `commit` an issue records when it is moved to a
/// terminal column (a git SHA says nothing in a Dolt database, and it stays on the
/// local issue).
const UNMAPPED_FIELDS: [&str; 15] = [
    "estimate",
    "mission",
    "task",
    "related_issues",
    "series",
    "health",
    "files",
    "cycles",
    "milestones",
    "projects",
    "epics",
    "reviewer",
    "start_date",
    "due_date",
    "time_entries",
];

/// Whether a field named in [`UNMAPPED_FIELDS`] carries anything worth reporting.
fn field_is_set(issue: &Issue, field: &str) -> bool {
    match field {
        "estimate" => issue.estimate.is_some(),
        "mission" => issue.mission.is_some(),
        "task" => issue.task.is_some(),
        "related_issues" => !issue.related_issues.is_empty(),
        "series" => issue.series.is_some(),
        "health" => issue.health.is_some(),
        "files" => !issue.files.is_empty(),
        "cycles" => issue.cycles.as_ref().is_some_and(|v| !v.is_empty()),
        "milestones" => issue.milestones.as_ref().is_some_and(|v| !v.is_empty()),
        "projects" => issue.projects.as_ref().is_some_and(|v| !v.is_empty()),
        "epics" => issue.epics.as_ref().is_some_and(|v| !v.is_empty()),
        "reviewer" => issue.reviewer.is_some(),
        "start_date" => issue.start_date.is_some(),
        "due_date" => issue.due_date.is_some(),
        "time_entries" => !issue.time_entries.is_empty(),
        _ => false,
    }
}

/// The diagnostics for the local data beads cannot hold on one issue.
fn unmapped_diagnostics(issue: &Issue) -> Vec<BeadsDiagnostic> {
    let subject = format!("#{} {}", issue.number, issue.title);
    let mut out = Vec::new();
    for field in UNMAPPED_FIELDS {
        if field_is_set(issue, field) {
            out.push(BeadsDiagnostic {
                number: Some(issue.number),
                subject: subject.clone(),
                field: field.to_string(),
                message: "beads has no column for this; the local value is kept as-is".to_string(),
            });
        }
    }
    out
}

/// Which issue's row carries an edge's record, and which issue it names.
///
/// beads stores a dependency on the **dependent**: `blocks` lives on the blocked
/// side, `duplicates` on the duplicate, `parent-child` on the child. `related` is
/// symmetric, so its stored `(min, max)` order picks the carrier.
fn edge_endpoints(rel: &Relationship) -> (crate::id::IssueRef, crate::id::IssueRef) {
    let canonical = rel.canonical();
    match canonical.kind {
        RelationKind::Blocks => (canonical.to, canonical.from),
        RelationKind::Duplicates => (canonical.from, canonical.to),
        _ => (canonical.from, canonical.to),
    }
}

/// The beads relation name for a tix relation kind.
fn relation_name(kind: RelationKind) -> &'static str {
    match kind {
        RelationKind::Blocks => "blocks",
        RelationKind::Duplicates => "duplicates",
        _ => "relates-to",
    }
}

/// The dependencies to hang off each row, keyed by the tix issue number of the row,
/// plus the diagnostics for edges that point outside the export set.
fn dependencies_by_row(
    relationships: &[Relationship],
    prefix: &str,
    aliases: &crate::id::AliasIndex,
    exported: &HashSet<u32>,
) -> (BTreeMap<u32, Vec<BeadsDependency>>, Vec<BeadsDiagnostic>) {
    let mut by_row: BTreeMap<u32, Vec<BeadsDependency>> = BTreeMap::new();
    let mut diagnostics = Vec::new();
    let mut seen: HashSet<&str> = HashSet::new();
    for rel in relationships {
        // `relationships_for` is an adjacency lookup, so an edge arrives from both
        // endpoints; the stable id is the dedup key.
        if !seen.insert(rel.id.as_str()) {
            continue;
        }
        let (row_ref, other_ref) = edge_endpoints(rel);
        let resolve_number = |reference: crate::id::IssueRef| {
            reference.as_number().or_else(|| {
                aliases
                    .resolve(&reference)
                    .ok()
                    .and_then(|resolved| resolved.as_number())
            })
        };
        let (Some(row), Some(other)) = (resolve_number(row_ref), resolve_number(other_ref)) else {
            diagnostics.push(BeadsDiagnostic {
                number: None,
                subject: format!("{row_ref} -> {other_ref} ({})", rel.id),
                field: "relationships".to_string(),
                message: "an endpoint does not resolve to an exported issue".to_string(),
            });
            continue;
        };
        if !exported.contains(&row) || !exported.contains(&other) {
            let missing = if exported.contains(&row) { other } else { row };
            diagnostics.push(BeadsDiagnostic {
                number: Some(row),
                subject: format!("#{row} -> #{other} ({})", rel.id),
                field: "relationships".to_string(),
                message: format!(
                    "target #{missing} is not in the export set, so beads would skip this \
                     edge; export it too or the relation stays local"
                ),
            });
            continue;
        }
        by_row.entry(row).or_default().push(BeadsDependency {
            issue_id: beads_id(prefix, row),
            depends_on_id: beads_id(prefix, other),
            kind: relation_name(rel.canonical().kind).to_string(),
        });
    }
    (by_row, diagnostics)
}

/// The rows to hand to `bd`, in issue-number order, and everything local that
/// could not go with them.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct BeadsPlan {
    pub rows: Vec<BeadsRow>,
    pub diagnostics: Vec<BeadsDiagnostic>,
}

/// Build the `bd import` rows for a set of issues.
///
/// A *parent* becomes a `parent-child` dependency on the child's own row (beads
/// stores no `parent` field), and every typed edge lands per the module's table.
#[must_use]
pub fn plan(
    prefix: &str,
    resolution: &CategoryResolution,
    issues: &[Issue],
    relationships: &[Relationship],
) -> BeadsPlan {
    let exported: HashSet<u32> = issues.iter().map(|i| i.number).collect();
    let aliases = crate::id::AliasIndex::build(issues);
    let (mut deps_by_row, mut diagnostics) =
        dependencies_by_row(relationships, prefix, &aliases, &exported);

    let mut ordered: Vec<&Issue> = issues.iter().collect();
    ordered.sort_by_key(|i| i.number);

    let mut rows = Vec::with_capacity(ordered.len());
    for issue in ordered {
        let (priority, priority_note) = priority_for(&issue.priority);
        let (issue_type, type_note) = issue_type_for(issue);

        let mut deps = deps_by_row.remove(&issue.number).unwrap_or_default();
        if let Some(parent_ref) = issue.parent {
            let parent = aliases
                .resolve(&parent_ref)
                .ok()
                .and_then(|resolved| resolved.as_number())
                .or_else(|| parent_ref.as_number());
            if let Some(parent) = parent {
                if exported.contains(&parent) {
                    deps.push(BeadsDependency {
                        issue_id: beads_id(prefix, issue.number),
                        depends_on_id: beads_id(prefix, parent),
                        kind: "parent-child".to_string(),
                    });
                } else {
                    diagnostics.push(BeadsDiagnostic {
                        number: Some(issue.number),
                        subject: format!("#{} {}", issue.number, issue.title),
                        field: "parent".to_string(),
                        message: format!(
                            "parent #{parent} is not in the export set; beads would skip the \
                             parent-child edge"
                        ),
                    });
                }
            } else {
                diagnostics.push(BeadsDiagnostic {
                    number: Some(issue.number),
                    subject: format!("#{} {}", issue.number, issue.title),
                    field: "parent".to_string(),
                    message: format!("parent {parent_ref} does not resolve to an exported issue"),
                });
            }
        }
        deps.sort_by(|a, b| (&a.kind, &a.depends_on_id).cmp(&(&b.kind, &b.depends_on_id)));

        let subject = format!("#{} {}", issue.number, issue.title);
        for note in [priority_note, type_note].into_iter().flatten() {
            diagnostics.push(BeadsDiagnostic {
                number: Some(issue.number),
                subject: subject.clone(),
                field: "type/priority".to_string(),
                message: note,
            });
        }
        diagnostics.extend(unmapped_diagnostics(issue));

        rows.push(BeadsRow {
            record_type: issue_record_type(),
            id: beads_id(prefix, issue.number),
            title: issue.title.clone(),
            description: (!issue.body.trim().is_empty()).then(|| issue.body.clone()),
            status: Some(status_for(issue, resolution).to_string()),
            priority,
            issue_type: Some(issue_type),
            assignee: issue.assignee.clone(),
            owner: None,
            labels: issue.labels.clone(),
            dependencies: deps,
            close_reason: close_reason_for(issue, resolution),
            created_at: None,
            updated_at: None,
        });
    }
    BeadsPlan { rows, diagnostics }
}

// ---------------------------------------------------------------------------
// JSONL
// ---------------------------------------------------------------------------

/// One JSON object per line, `bd export`'s own encoding and `bd import`'s accepted
/// input. Trailing newline included; an empty plan is an empty string.
pub fn to_jsonl(rows: &[BeadsRow]) -> Result<String> {
    let mut out = String::new();
    for row in rows {
        let line = serde_json::to_string(row)
            .map_err(|e| TixError::InvalidInput(format!("could not encode a beads row: {e}")))?;
        writeln!(out, "{line}")
            .map_err(|e| TixError::InvalidInput(format!("could not encode a beads row: {e}")))?;
    }
    Ok(out)
}

/// Read rows back from `bd export` (JSONL) or `bd list --json` (an array),
/// unwrapping the `BD_JSON_ENVELOPE` `{"data": …}` when present.
///
/// Non-JSON lines are skipped rather than failing the read: `bd` prints a
/// one-time usage notice on stdout, and a caller that asked for JSON should get
/// the rows it asked for rather than an error about a banner.
pub fn parse_rows(raw: &str) -> Result<Vec<BeadsRow>> {
    let trimmed = raw.trim();
    if trimmed.is_empty() {
        return Ok(Vec::new());
    }
    if let Ok(value) = serde_json::from_str::<serde_json::Value>(trimmed) {
        let body = value.get("data").unwrap_or(&value);
        if body.is_array() {
            return serde_json::from_value(body.clone()).map_err(|e| {
                TixError::InvalidInput(format!("could not read bd JSON output: {e}"))
            });
        }
        if let Ok(row) = serde_json::from_value::<BeadsRow>(body.clone()) {
            return Ok(vec![row]);
        }
    }
    let mut rows = Vec::new();
    for line in trimmed.lines() {
        let line = line.trim();
        if line.is_empty() || !line.starts_with('{') {
            continue;
        }
        if let Ok(row) = serde_json::from_str::<BeadsRow>(line) {
            rows.push(row);
        }
    }
    Ok(rows)
}

// ---------------------------------------------------------------------------
// Push
// ---------------------------------------------------------------------------

/// Write the rows to `bd` via `bd import <file>`.
///
/// A file rather than stdout, because [`CommandRunner`] spawns without a stdin
/// pipe; `bd import` reads a path and a `-` alike, and the path is what the
/// runner can carry. Returns `bd`'s stdout so the caller can surface its
/// "Imported N issue(s)" line.
pub fn push(runner: &dyn CommandRunner, binary: &str, rows: &[BeadsRow]) -> Result<String> {
    let jsonl = to_jsonl(rows)?;
    let path = std::env::temp_dir().join(format!(
        "tix-beads-{}-{}.jsonl",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map_or(0, |d| d.as_nanos())
    ));
    std::fs::write(&path, jsonl.as_bytes()).map_err(|e| {
        TixError::External(format!("could not stage the beads import at {path:?}: {e}"))
    })?;
    let args = vec!["import".to_string(), path.display().to_string()];
    let outcome: Result<CommandOutcome> = runner.run(binary, &args, &[]);
    let _ = std::fs::remove_file(&path);
    let outcome = outcome?;
    if !outcome.success {
        return Err(TixError::External(format!(
            "`{binary} import` failed: {}",
            outcome.text().trim()
        )));
    }
    Ok(outcome.stdout)
}

// ---------------------------------------------------------------------------
// The `bd` CLI service
// ---------------------------------------------------------------------------

/// The environment every `bd` call runs with.
///
/// A configured workspace is exported as `BEADS_DIR` — beads' own override for
/// "which workspace" — rather than passed as a flag, because it is the one knob
/// `bd export`, `bd import` and `bd update` all honour.
#[must_use]
pub fn env_for(config: &Config) -> Vec<(String, String)> {
    config
        .beads
        .workspace
        .as_ref()
        .map(|dir| vec![("BEADS_DIR".to_string(), dir.display().to_string())])
        .unwrap_or_default()
}

/// Read every issue beads holds, through `bd export`.
///
/// `bd export` writes JSONL on stdout and excludes infrastructure beads (agents,
/// roles, messages) by default, which is what a view of the *work* wants.
pub fn fetch(runner: &dyn CommandRunner, config: &Config) -> Result<Vec<BeadsRow>> {
    let outcome = runner.run(
        &config.beads.binary,
        &["export".to_string()],
        &env_for(config),
    )?;
    if !outcome.success {
        return Err(TixError::External(format!(
            "`{} export` failed: {}",
            config.beads.binary,
            outcome.text().trim()
        )));
    }
    parse_rows(&outcome.stdout)
}

/// What a probe of `bd` and its workspace found.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Detection {
    /// The binary that was probed.
    pub binary: String,
    /// `bd version` answered.
    pub available: bool,
    /// `bd where` resolved an active workspace.
    pub workspace: bool,
    /// The version `bd version --json` reported.
    pub version: Option<String>,
    /// The resolved workspace path.
    pub path: Option<String>,
    /// The events-journal setting, when it was readable. A disabled journal makes
    /// a live mirror silently empty, so it is worth reporting.
    pub journal: Option<bool>,
    /// One line a human can act on.
    pub detail: String,
}

/// Probe for `bd` and its workspace. **Read-only**: it runs only `bd version`,
/// `bd where` and `bd config get`, never opens the database, and writes nothing.
#[must_use]
pub fn detect(runner: &dyn CommandRunner, config: &Config) -> Detection {
    let binary = config.beads.binary.clone();
    let env = env_for(config);
    let mut detection = Detection {
        binary: binary.clone(),
        available: false,
        workspace: false,
        version: None,
        path: None,
        journal: None,
        detail: String::new(),
    };

    match runner.run(
        &binary,
        &["version".to_string(), "--json".to_string()],
        &env,
    ) {
        Ok(outcome) if outcome.success => {
            detection.available = true;
            detection.version = serde_json::from_str::<serde_json::Value>(&outcome.stdout)
                .ok()
                .and_then(|v| {
                    v.get("version")
                        .and_then(|s| s.as_str())
                        .map(str::to_string)
                });
        }
        _ => {
            detection.detail =
                format!("`{binary}` is not on PATH — the beads integration stays off");
            return detection;
        }
    }

    if let Ok(outcome) = runner.run(&binary, &["where".to_string(), "--json".to_string()], &env) {
        if outcome.success {
            if let Ok(value) = serde_json::from_str::<serde_json::Value>(&outcome.stdout) {
                detection.path = value
                    .get("path")
                    .and_then(|s| s.as_str())
                    .map(str::to_string);
                detection.workspace = detection.path.is_some();
            }
        }
    }

    if detection.workspace {
        if let Ok(outcome) = runner.run(
            &binary,
            &[
                "config".to_string(),
                "get".to_string(),
                "events-journal".to_string(),
            ],
            &env,
        ) {
            detection.journal = match outcome.stdout.trim() {
                "true" => Some(true),
                "false" => Some(false),
                _ => None,
            };
        }
        let where_ = detection.path.as_deref().unwrap_or("somewhere");
        detection.detail = match (detection.version.as_deref(), detection.journal) {
            (Some(version), Some(false)) => format!(
                "beads {version} at {where_} — its events journal is OFF, so a live mirror \
                 would see nothing (enable it wherever beads writes: `bd config set \
                 events-journal true`)"
            ),
            (Some(version), _) => format!("beads {version} at {where_}"),
            (None, _) => format!("a beads workspace at {where_}"),
        };
    } else {
        detection.detail = format!(
            "`{binary}` is installed ({}) but no beads workspace is active here",
            detection.version.as_deref().unwrap_or("?")
        );
    }
    detection
}

/// The tix issue number one of *our* exported ids names (`<prefix>-<n>`).
#[must_use]
pub fn number_from_id(id: &str, prefix: &str) -> Option<u32> {
    id.strip_prefix(prefix)?.strip_prefix('-')?.parse().ok()
}

/// The tix issue number an `external_ref` names, when it carries ours (`tix#<n>`).
#[must_use]
pub fn number_from_external_ref(reference: &str) -> Option<u32> {
    reference.strip_prefix("tix#")?.parse().ok()
}

/// The `external_ref` tix writes back onto a bead it imported, so a later import
/// recognises it without depending on local sync state.
#[must_use]
pub fn external_ref_for(number: u32) -> String {
    format!("tix#{number}")
}

/// The local column a beads status maps onto, or `None` when no column does.
///
/// The inverse of [`status_for`], and it asks the **category** service the same
/// way: `closed` lands on the board's completion column, `in_progress` on the
/// first `active` one, and anything else on the board's first column. A
/// `close_reason` of `canceled` lands on a `canceled` column when the board has
/// one — the case [`status_for`] wrote the reason for in the first place.
#[must_use]
pub fn column_for_status(
    resolution: &CategoryResolution,
    status: &str,
    close_reason: Option<&str>,
) -> Option<String> {
    let column_with = |category: WorkflowCategory| {
        resolution
            .columns()
            .find(|c| resolution.resolve(c) == Some(category))
            .map(str::to_string)
    };
    if close_reason == Some("canceled") {
        if let Some(column) = column_with(WorkflowCategory::Canceled) {
            return Some(column);
        }
    }
    match status {
        "closed" => column_with(WorkflowCategory::Completed),
        "in_progress" => column_with(WorkflowCategory::Active),
        _ => resolution.columns().next().map(str::to_string),
    }
}

/// beads' `0..4` band maps onto tix's priority vocabulary: `0`/`1` are what tix
/// calls high, `2` medium, and `3`/`4` low.
#[must_use]
pub fn priority_from_beads(priority: Option<u8>) -> Priority {
    match priority {
        Some(0 | 1) => Priority::High,
        Some(3 | 4) => Priority::Low,
        _ => Priority::Medium,
    }
}

/// One row reduced to the local fields it can fill.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MappedRow {
    /// The tix issue the row names, when its id, `external_ref`, or the local
    /// sync state identifies one.
    pub number: Option<u32>,
    pub title: String,
    pub body: String,
    /// The local column, or `None` when the board has no column for the status.
    pub status: Option<String>,
    pub priority: Priority,
    pub type_: Option<String>,
    pub labels: Vec<String>,
    pub assignee: Option<String>,
    /// The bead's id, which is what identity is keyed on.
    pub beads_id: String,
    /// The bead's `updated_at`, for the newer-wins conflict rule.
    pub updated_at: Option<String>,
}

/// Map one row onto the local fields it can fill. Pure: no store, no subprocess.
///
/// `known` is the local sync state's id → number map, consulted for
/// beads-originated issues whose ids mean nothing to tix.
#[must_use]
pub fn map_row(
    row: &BeadsRow,
    config: &Config,
    resolution: &CategoryResolution,
    known: &BTreeMap<String, u32>,
) -> MappedRow {
    let prefix = &config.beads.id_prefix;
    let number = number_from_id(&row.id, prefix)
        .or_else(|| known.get(&row.id).copied())
        .or_else(|| {
            row.owner
                .as_deref()
                .and_then(number_from_external_ref)
                .or_else(|| number_from_external_ref(&row.id))
        });
    let accepted_types: Vec<&str> = match config.types.as_deref() {
        Some(types) => types.iter().map(String::as_str).collect(),
        None => crate::config::DEFAULT_TYPES.to_vec(),
    };
    let type_ = row
        .issue_type
        .as_deref()
        .filter(|t| accepted_types.contains(t))
        .map(str::to_string);
    MappedRow {
        number,
        title: row.title.clone(),
        body: row.description.clone().unwrap_or_default(),
        status: column_for_status(
            resolution,
            row.status.as_deref().unwrap_or("open"),
            row.close_reason.as_deref(),
        ),
        priority: priority_from_beads(row.priority),
        type_,
        labels: row.labels.clone(),
        assignee: row.effective_assignee().map(str::to_string),
        beads_id: row.id.clone(),
        updated_at: row.updated_at.clone(),
    }
}

/// The identity map tix keeps between a beads workspace and its store.
///
/// Only beads-originated issues need an entry: a bead tix exported has a
/// deterministic `<prefix>-<number>` id, which is its own mapping. The file is
/// small, and it is **identity** rather than cache — losing it would make the
/// next import create a second tix issue for a bead that already has one — so it
/// is written to the project root and is safe to commit.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct State {
    /// beads id → tix issue number, for beads-originated issues.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub numbers: BTreeMap<String, u32>,
    /// beads id → the `updated_at` last seen, the watermark that stops the two
    /// pipes echoing each other's change back.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub synced_at: BTreeMap<String, String>,
    /// beads id → [`fingerprint`] of the row last **exported** for it.
    ///
    /// This is the outbound watermark: a row that fingerprints the same as the
    /// one already sent has not changed, so a re-export can skip it. It is what
    /// makes a second export cheap on a repository with thousands of issues.
    ///
    /// Only meaningful together with [`Self::workspace`]: a fingerprint records
    /// that a row was exported **somewhere**, and skipping is only correct when
    /// "somewhere" is still the workspace in front of us.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub exported: BTreeMap<String, u64>,
    /// The workspace [`Self::exported`] was written against.
    ///
    /// `None` in a state file written before this was recorded, which is itself a
    /// reason not to trust the watermark: the export compares it and resets when it
    /// differs, so pointing tix at a different workspace re-exports everything
    /// instead of silently sending nothing.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub workspace: Option<String>,
}

/// The beads workspace an export is about to talk to, resolved as cheaply as it
/// can be.
///
/// The precedence is the one `bd` itself uses — `BEADS_DIR`, then the configured
/// workspace, then the nearest `.beads` walking up from the working directory —
/// and it is resolved **without a subprocess**, because it is checked on every
/// export and a repeat export is a few hundred milliseconds.
///
/// It exists to make [`State::exported`] safe. A fingerprint says "this row has
/// been exported"; that is only true of the workspace it was exported *to*, so
/// the two are recorded together and compared on every run.
#[must_use]
pub fn workspace_identity(config: &Config) -> Option<String> {
    let cwd = std::env::current_dir().ok()?;
    workspace_identity_from(
        std::env::var("BEADS_DIR").ok().as_deref(),
        config.beads.workspace.as_deref(),
        &cwd,
    )
}

/// [`workspace_identity`] with its three inputs passed in.
///
/// Split out so the precedence can be tested without a process-global
/// environment, which every other test in the suite shares.
fn workspace_identity_from(
    beads_dir: Option<&str>,
    configured: Option<&std::path::Path>,
    cwd: &std::path::Path,
) -> Option<String> {
    let canonical = |path: &std::path::Path| {
        std::fs::canonicalize(path)
            .unwrap_or_else(|_| path.to_path_buf())
            .display()
            .to_string()
    };
    if let Some(dir) = beads_dir {
        if !dir.trim().is_empty() {
            return Some(canonical(std::path::Path::new(dir)));
        }
    }
    if let Some(dir) = configured {
        return Some(canonical(dir));
    }
    let mut current = cwd.to_path_buf();
    loop {
        let candidate = current.join(".beads");
        if candidate.is_dir() {
            return Some(canonical(&candidate));
        }
        if !current.pop() {
            return None;
        }
    }
}

/// A stable 64-bit FNV-1a fingerprint of a row's serialized form.
///
/// **Stable on purpose.** `DefaultHasher`'s algorithm is explicitly not
/// guaranteed to stay the same across Rust releases, and a fingerprint that
/// changed when the toolchain did would make every issue look modified and
/// re-export the whole board for no reason. FNV-1a is fixed by its constants, so
/// an unchanged row fingerprints the same forever.
///
/// The whole row is hashed — including its dependency records — so an edge
/// added, re-kinded or dropped changes the fingerprint of the issue that carries
/// it, which is the one that has to be re-sent.
pub fn fingerprint(row: &BeadsRow) -> Result<u64> {
    let text = serde_json::to_string(row)
        .map_err(|e| TixError::InvalidInput(format!("could not fingerprint a beads row: {e}")))?;
    Ok(fnv1a(text.as_bytes()))
}

/// FNV-1a, 64-bit.
fn fnv1a(bytes: &[u8]) -> u64 {
    const OFFSET_BASIS: u64 = 0xcbf2_9ce4_8422_2325;
    const PRIME: u64 = 0x0000_0100_0000_01b3;
    let mut hash = OFFSET_BASIS;
    for byte in bytes {
        hash ^= u64::from(*byte);
        hash = hash.wrapping_mul(PRIME);
    }
    hash
}

/// The rows an export should send, the fingerprints to record for them, and how
/// many rows were already up to date.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct BeadsPayload {
    /// The rows to send, in plan order.
    pub rows: Vec<BeadsRow>,
    /// `(beads id, fingerprint)` for each row in [`Self::rows`], to be recorded
    /// once the push succeeds.
    pub fingerprints: Vec<(String, u64)>,
    /// How many rows matched the watermark and were left out.
    pub unchanged: usize,
}

/// Choose the rows to send, given the export watermark.
///
/// Pure, and separate from the push so the incremental rule can be tested without
/// a store or a `bd`. Rows are taken by value so the payload moves rather than
/// clones them — on a first run this is the whole board.
pub fn select_payload(
    rows: Vec<BeadsRow>,
    exported: &BTreeMap<String, u64>,
    all: bool,
) -> Result<BeadsPayload> {
    let mut payload = BeadsPayload::default();
    for row in rows {
        let fingerprint = fingerprint(&row)?;
        if all || exported.get(&row.id).copied() != Some(fingerprint) {
            payload.fingerprints.push((row.id.clone(), fingerprint));
            payload.rows.push(row);
        } else {
            payload.unchanged += 1;
        }
    }
    Ok(payload)
}

/// The state file's name, at the project root.
pub const STATE_FILE: &str = ".tix-beads-state.json";

/// The state file's path for a project root.
#[must_use]
pub fn state_path(root: &Path) -> PathBuf {
    root.join(STATE_FILE)
}

/// Read the state, or the empty state when there is none (and when the file is
/// unreadable or malformed, which is reported by the caller's first write rather
/// than failing every read).
#[must_use]
pub fn load_state(root: &Path) -> State {
    std::fs::read_to_string(state_path(root))
        .ok()
        .and_then(|raw| serde_json::from_str(&raw).ok())
        .unwrap_or_default()
}

/// Write the state atomically-enough for a single-writer CLI: a temp file beside
/// the target, then a rename.
pub fn save_state(root: &Path, state: &State) -> Result<()> {
    let path = state_path(root);
    let body = serde_json::to_string_pretty(state)
        .map_err(|e| TixError::InvalidInput(format!("could not encode the beads state: {e}")))?;
    let tmp = path.with_extension("json.tmp");
    std::fs::write(&tmp, body.as_bytes())?;
    std::fs::rename(&tmp, &path)?;
    Ok(())
}

/// One import pass's outcome.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ImportReport {
    pub created: usize,
    pub updated: usize,
    pub skipped: usize,
    /// How many of `created` were beads-originated issues whose id ⇄ number
    /// pairing had to be recorded, because their ids are not tix's own. Reported
    /// as a count rather than a diagnostic: nothing is unrepresentable here, it is
    /// just identity that had to be written down.
    pub foreign: usize,
    pub diagnostics: Vec<BeadsDiagnostic>,
}

/// A blank issue to fill, so the import path and the tests share one literal.
fn blank_issue() -> Issue {
    Issue {
        number: 0,
        id: None,
        title: String::new(),
        status: crate::model::Status::new(""),
        labels: Vec::new(),
        type_: None,
        priority: Priority::Medium,
        assignee: None,
        estimate: None,
        commit: None,
        gitlab_id: None,
        linear_id: None,
        mission: None,
        task: None,
        relationships: Vec::new(),
        related_issues: Vec::new(),
        files: Vec::new(),
        cycles: None,
        milestones: None,
        reviewer: None,
        start_date: None,
        due_date: None,
        created: crate::model::today(),
        updated: crate::model::today(),
        filename: String::new(),
        body: String::new(),
        history: crate::model::History::new(),
        has_retired_links_key: false,
        time_entries: Vec::new(),
        parent: None,
        series: None,
        health: None,
        projects: None,
        epics: None,
        extra_frontmatter: std::collections::BTreeMap::new(),
        archived: false,
        entered_at: None,
        source: None,
    }
}

/// Pull beads rows into the local store.
///
/// Each row is its own transaction, exactly as the GitLab/Linear import does it:
/// one remote issue's failure must not roll back the others. A row that names an
/// existing issue **overwrites it only when the bead is newer** ([`crate::sync::is_remote_newer`]),
/// which is the same conflict rule the other trackers use; an older bead is left
/// alone and counted as skipped.
///
/// The returned [`ImportReport`] names what happened, and `state` is updated in
/// memory — the caller saves it, so a read-only path never writes.
pub fn import_into(
    store: &Store,
    config: &Config,
    resolution: &CategoryResolution,
    rows: &[BeadsRow],
    state: &mut State,
) -> Result<ImportReport> {
    let by_number: BTreeMap<u32, Issue> = store
        .list()?
        .into_iter()
        .map(|issue| (issue.number, issue))
        .collect();

    let mut report = ImportReport::default();
    for row in rows {
        let mapped = map_row(row, config, resolution, &state.numbers);
        let Some(status) = mapped.status.clone() else {
            report.skipped += 1;
            report.diagnostics.push(BeadsDiagnostic {
                number: mapped.number,
                subject: format!("beads {} \"{}\"", row.id, row.title),
                field: "status".to_string(),
                message: format!(
                    "`{}` is not a column on the board; the issue was not imported",
                    row.status.as_deref().unwrap_or("open")
                ),
            });
            continue;
        };

        match mapped.number.and_then(|n| by_number.get(&n).cloned()) {
            Some(existing) => {
                let newer = match (&mapped.updated_at, existing.updated.as_str()) {
                    (Some(remote), local) => crate::sync::is_remote_newer(remote, local),
                    (None, _) => true,
                };
                if !newer {
                    report.skipped += 1;
                    continue;
                }
                let mut issue = existing;
                issue.title = mapped.title.clone();
                issue.body = mapped.body.clone();
                issue.status = crate::model::Status::new(status);
                issue.priority = mapped.priority.clone();
                issue.labels = mapped.labels.clone();
                issue.assignee = mapped.assignee.clone();
                issue.bump_updated();
                store.transaction(|| store.update(&issue.filename, &issue))?;
                report.updated += 1;
            }
            None => {
                let mut issue = blank_issue();
                issue.title = mapped.title.clone();
                issue.status = crate::model::Status::new(status);
                issue.priority = mapped.priority.clone();
                issue.labels = mapped.labels.clone();
                issue.assignee = mapped.assignee.clone();
                issue.type_ = mapped.type_.clone();
                let created = store.transaction(|| store.create(&mut issue, &mapped.body, None))?;
                let _ = created;
                report.created += 1;
                if number_from_id(&row.id, &config.beads.id_prefix).is_none() {
                    // A bead tix did not author: remember the pairing, so the next
                    // import recognises it instead of creating a second issue.
                    state.numbers.insert(row.id.clone(), issue.number);
                    report.foreign += 1;
                }
            }
        }
        if let Some(updated) = &mapped.updated_at {
            state.synced_at.insert(row.id.clone(), updated.clone());
        }
    }
    Ok(report)
}

/// Install the `beads → tix` pipe: executable `.beads/hooks/{on_create,on_update,on_close}`
/// scripts that run `tix beads pull <id>`.
///
/// beads calls a hook as `<script> <issue-id> <event>` with the issue JSON on
/// **stdin**, fire-and-forget, and silently skips a script that is not
/// executable — so the mode bit is set here and the scripts always exit 0. The
/// project root and the `tix` binary are baked in, because beads does not run
/// hooks in this directory and does not guarantee `tix` is on ITS path.
pub fn install_hooks(root: &Path, beads_dir: &Path, binary: &str) -> Result<Vec<PathBuf>> {
    let hooks_dir = beads_dir.join("hooks");
    std::fs::create_dir_all(&hooks_dir)?;
    let exe = std::env::current_exe()
        .ok()
        .filter(|p| p.exists())
        .map_or_else(|| "tix".to_string(), |p| p.display().to_string());
    let root = root.display();
    let mut written = Vec::new();
    for hook in ["on_create", "on_update", "on_close"] {
        let path = hooks_dir.join(hook);
        let body = format!(
            "#!/bin/sh\n\
             # The beads -> tix pipe, installed by `tix init --beads sync`.\n\
             # beads calls this as `{hook} <issue-id> <event>` with the issue JSON on\n\
             # stdin, fire-and-forget, and skips it silently unless it is executable.\n\
             # `tix beads pull` is idempotent and never writes to beads, so a missed\n\
             # run is repaired by the next one.\n\
             cd '{root}' 2>/dev/null || exit 0\n\
             '{exe}' beads pull --quiet >/dev/null 2>&1\n\
             exit 0\n"
        );
        crate::commands::atomic_write(&path, &body)?;
        make_executable(&path)?;
        written.push(path);
    }
    let _ = binary;
    Ok(written)
}

#[cfg(unix)]
fn make_executable(path: &Path) -> Result<()> {
    use std::os::unix::fs::PermissionsExt as _;
    let mode = std::fs::metadata(path)?.permissions().mode();
    std::fs::set_permissions(path, std::fs::Permissions::from_mode(mode | 0o111))?;
    Ok(())
}

#[cfg(not(unix))]
fn make_executable(_path: &Path) -> Result<()> {
    // beads skips a non-executable hook; on Windows the extension is what counts,
    // and a script written to `.beads/hooks/` is not runnable there anyway.
    Ok(())
}

/// One record from the beads events journal.
///
/// The journal is beads' own binlog: every committed mutation writes one ordered
/// record, and a consumer resumes from a `seq` it can checkpoint.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct EventRecord {
    #[serde(default)]
    pub seq: u64,
    #[serde(default)]
    pub ts: Option<String>,
    #[serde(default)]
    pub op: String,
    #[serde(default)]
    pub issue_id: String,
    #[serde(default)]
    pub actor: Option<String>,
    /// The resulting issue state, as the record carries it.
    #[serde(default)]
    pub issue: Option<BeadsRow>,
}

/// Read the beads events journal from `since` (inclusive), at most `limit`
/// records.
///
/// `bd events tail` writes JSON Lines in sequence order. It is only populated
/// where the journal is enabled — a workspace whose writers never turned it on
/// stays permanently empty, and an empty journal is indistinguishable from a
/// caught-up one — so a caller that got nothing is told that rather than left to
/// assume it is current.
pub fn events(
    runner: &dyn CommandRunner,
    config: &Config,
    since: u64,
    limit: u64,
) -> Result<Vec<EventRecord>> {
    let args = vec![
        "events".to_string(),
        "tail".to_string(),
        "--since".to_string(),
        since.to_string(),
        "--limit".to_string(),
        limit.to_string(),
    ];
    let outcome = runner.run(&config.beads.binary, &args, &env_for(config))?;
    if !outcome.success {
        return Err(TixError::External(format!(
            "`{} events tail` failed: {}",
            config.beads.binary,
            outcome.text().trim()
        )));
    }
    let trimmed = outcome.stdout.trim();
    if trimmed.is_empty() {
        return Ok(Vec::new());
    }
    // The CLI writes JSON Lines; the HTTP endpoint wraps them. Accept both.
    if let Ok(value) = serde_json::from_str::<serde_json::Value>(trimmed) {
        if let Some(records) = value.get("records") {
            if let Ok(parsed) = serde_json::from_value::<Vec<EventRecord>>(records.clone()) {
                return Ok(parsed);
            }
        }
    }
    Ok(trimmed
        .lines()
        .filter(|line| line.trim_start().starts_with('{'))
        .filter_map(|line| serde_json::from_str::<EventRecord>(line).ok())
        .collect())
}

/// Refuse a write when the configured mode is `mirror`.
///
/// `mirror` means read-only **by contract**: tix writes neither to beads nor to
/// an issue file, so every write path asks here first. `off` is not refused —
/// that mode means tix does nothing with beads on its own, and an explicit
/// `tix beads push` is the user's direct instruction, not an automatic one.
pub fn ensure_writes_allowed(config: &Config) -> Result<()> {
    if config.beads.mode == crate::config::BeadsMode::Mirror {
        return Err(TixError::Config(
            "beads mode is `mirror` (read-only): tix will not write to beads or to an issue \
             file. Run `tix init --beads sync` to enable writes."
                .to_string(),
        ));
    }
    Ok(())
}

/// Whether a row is blocked: one of its `blocks` dependencies names an issue that
/// is not closed.
///
/// beads derives its ready frontier this way, so reading it the same way is what
/// keeps a mirror agreeing with `bd ready` rather than guessing.
#[must_use]
pub fn is_blocked(row: &BeadsRow, status_by_id: &BTreeMap<String, String>) -> bool {
    row.dependencies
        .iter()
        .filter(|dep| dep.kind == "blocks")
        .any(|dep| {
            status_by_id
                .get(&dep.depends_on_id)
                .is_some_and(|status| status != "closed")
        })
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    #[allow(clippy::wildcard_imports)]
    use super::*;
    use crate::config::Config;
    use crate::model::Status;
    use crate::relationships::RelationKind;
    use std::cell::RefCell;

    fn issue(number: u32, title: &str) -> Issue {
        let mut issue = blank_issue();
        issue.number = number;
        issue.title = title.to_string();
        issue.status = Status::new("todo");
        issue
    }

    fn resolution_for(config: &Config) -> CategoryResolution {
        crate::workflow::resolution(config, config.default_config(), "default")
    }

    #[test]
    fn ids_are_deterministic_and_prefixed() {
        assert_eq!(beads_id("tix", 1705), "tix-1705");
        assert_eq!(beads_id("bd", 7), "bd-7");
    }

    #[test]
    fn status_follows_the_category_service_not_the_column_name() {
        let config = Config::default();
        let resolution = resolution_for(&config);
        // The default board's completion column maps to `completed`.
        let done_column =
            crate::workflow::completion_column_or_default(config.default_config(), &config);
        let mut done = issue(1, "done");
        done.status = Status::new(done_column);
        assert_eq!(status_for(&done, &resolution), "closed");
        assert_eq!(close_reason_for(&done, &resolution), None);

        // A column the board does not configure is not complete.
        let mut odd = issue(2, "odd");
        odd.status = Status::new("not-a-column");
        assert_eq!(status_for(&odd, &resolution), "open");
    }

    #[test]
    fn priority_maps_onto_the_beads_band_and_reports_a_stranger() {
        assert_eq!(priority_for(&Priority::High), (Some(1), None));
        assert_eq!(priority_for(&Priority::Medium), (Some(2), None));
        assert_eq!(priority_for(&Priority::Low), (Some(3), None));
        // `priority: ''` is unset, not a band.
        assert_eq!(priority_for(&Priority::Other(String::new())), (None, None));
        let (band, note) = priority_for(&Priority::Other("P0".into()));
        assert_eq!(band, Some(2));
        assert!(note.unwrap().contains("no beads band"));
    }

    #[test]
    fn type_falls_back_to_task_outside_the_beads_vocabulary() {
        let mut i = issue(1, "t");
        i.type_ = Some("bug".into());
        assert_eq!(issue_type_for(&i), ("bug".to_string(), None));

        i.type_ = Some("spike".into());
        let (mapped, note) = issue_type_for(&i);
        assert_eq!(mapped, "task");
        assert!(note.unwrap().contains("not a beads issue_type"));

        i.type_ = None;
        assert_eq!(issue_type_for(&i), ("task".to_string(), None));
    }

    #[test]
    fn a_blocking_edge_lands_on_the_blocked_row() {
        // 1 blocks 2  →  the record lives on 2, naming 1 as the blocker.
        let rel = Relationship::new(1, 2, RelationKind::Blocks, "two");
        let exported: HashSet<u32> = [1, 2].into_iter().collect();
        let (by_row, diags) =
            dependencies_by_row(&[rel], "tix", &crate::id::AliasIndex::default(), &exported);
        assert_eq!(diags, Vec::<BeadsDiagnostic>::new());
        let dep = &by_row[&2][0];
        assert_eq!(dep.issue_id, "tix-2");
        assert_eq!(dep.depends_on_id, "tix-1");
        assert_eq!(dep.kind, "blocks");
    }

    #[test]
    fn duplicates_and_related_use_beads_relation_names() {
        let dup = Relationship::new(3, 4, RelationKind::Duplicates, "four");
        let rel = Relationship::new(5, 6, RelationKind::Related, "six");
        let exported: HashSet<u32> = [3, 4, 5, 6].into_iter().collect();
        let (by_row, _) = dependencies_by_row(
            &[dup, rel],
            "tix",
            &crate::id::AliasIndex::default(),
            &exported,
        );
        assert_eq!(by_row[&3][0].kind, "duplicates");
        assert_eq!(by_row[&3][0].depends_on_id, "tix-4");
        // `related` is symmetric and order-normalized, so the lower number carries it.
        assert_eq!(by_row[&5][0].kind, "relates-to");
        assert_eq!(by_row[&5][0].depends_on_id, "tix-6");
    }

    #[test]
    fn an_edge_outside_the_export_set_is_reported_not_dropped() {
        let rel = Relationship::new(1, 99, RelationKind::Blocks, "gone");
        let exported: HashSet<u32> = [1].into_iter().collect();
        let (by_row, diags) =
            dependencies_by_row(&[rel], "tix", &crate::id::AliasIndex::default(), &exported);
        assert!(by_row.is_empty());
        assert_eq!(diags.len(), 1);
        assert!(diags[0].message.contains("#99"));
    }

    #[test]
    fn a_parent_becomes_a_parent_child_edge_and_local_fields_are_reported() {
        let config = Config::default();
        let resolution = resolution_for(&config);
        let mut child = issue(5, "child");
        child.parent = Some(crate::id::IssueRef::Number(4));
        child.cycles = Some(vec!["October 2026".into()]);
        child.estimate = Some("3d".into());
        let plan = plan("tix", &resolution, &[issue(4, "parent"), child], &[]);

        let child_row = plan.rows.iter().find(|r| r.id == "tix-5").unwrap();
        assert_eq!(child_row.dependencies.len(), 1);
        assert_eq!(child_row.dependencies[0].kind, "parent-child");
        assert_eq!(child_row.dependencies[0].depends_on_id, "tix-4");

        let fields: Vec<&str> = plan.diagnostics.iter().map(|d| d.field.as_str()).collect();
        assert!(fields.contains(&"cycles"), "{fields:?}");
        assert!(fields.contains(&"estimate"), "{fields:?}");
        // The parent is representable, so it is not reported.
        assert!(!fields.contains(&"parent"), "{fields:?}");
    }

    #[test]
    fn jsonl_round_trips_through_the_reader() {
        let row = BeadsRow {
            record_type: issue_record_type(),
            id: "tix-1".into(),
            title: "One".into(),
            description: Some("body".into()),
            status: Some("in_progress".into()),
            priority: Some(1),
            issue_type: Some("task".into()),
            assignee: Some("alice".into()),
            owner: None,
            labels: vec!["beads".into()],
            dependencies: vec![BeadsDependency {
                issue_id: "tix-1".into(),
                depends_on_id: "tix-2".into(),
                kind: "blocks".into(),
            }],
            close_reason: None,
            created_at: None,
            updated_at: None,
        };
        let jsonl = to_jsonl(std::slice::from_ref(&row)).unwrap();
        assert_eq!(jsonl.lines().count(), 1);
        assert_eq!(parse_rows(&jsonl).unwrap(), vec![row.clone()]);
        // An array and an envelope are accepted too.
        let array = serde_json::to_string(&vec![row.clone()]).unwrap();
        assert_eq!(parse_rows(&array).unwrap(), vec![row.clone()]);
        let envelope = format!("{{\"schema_version\":1,\"data\":{array}}}");
        assert_eq!(parse_rows(&envelope).unwrap(), vec![row.clone()]);
        // A usage banner on stdout must not fail the read.
        let noisy = format!("Thanks for using bd!\n{jsonl}");
        assert_eq!(parse_rows(&noisy).unwrap(), vec![row]);
    }

    #[test]
    fn exported_rows_omit_unset_optional_fields() {
        let config = Config::default();
        let resolution = resolution_for(&config);
        let plan = plan("tix", &resolution, &[issue(1, "One")], &[]);
        let json = serde_json::to_string(&plan.rows[0]).unwrap();
        assert!(!json.contains("close_reason"), "{json}");
        assert!(!json.contains("dependencies"), "{json}");
        assert!(!json.contains("assignee"), "{json}");
        assert!(json.contains("\"_type\":\"issue\""), "{json}");
    }

    fn beads_row(id: &str, status: &str) -> BeadsRow {
        BeadsRow {
            record_type: issue_record_type(),
            id: id.to_string(),
            title: format!("issue {id}"),
            description: None,
            status: Some(status.to_string()),
            priority: Some(2),
            issue_type: Some("task".to_string()),
            assignee: None,
            owner: None,
            labels: Vec::new(),
            dependencies: Vec::new(),
            close_reason: None,
            created_at: None,
            updated_at: None,
        }
    }

    #[test]
    fn a_fingerprint_is_stable_and_moves_with_the_row() {
        let row = beads_row("tix-1", "open");
        assert_eq!(fingerprint(&row).unwrap(), fingerprint(&row).unwrap());

        let mut changed = row.clone();
        changed.status = Some("closed".to_string());
        assert_ne!(fingerprint(&row).unwrap(), fingerprint(&changed).unwrap());

        // A dependency change must move the fingerprint of the row that carries it:
        // the edge is part of what beads is told about that issue.
        let mut edged = row.clone();
        edged.dependencies.push(BeadsDependency {
            issue_id: "tix-1".to_string(),
            depends_on_id: "tix-2".to_string(),
            kind: "blocks".to_string(),
        });
        assert_ne!(fingerprint(&row).unwrap(), fingerprint(&edged).unwrap());
    }

    #[test]
    fn only_changed_rows_are_selected_and_all_forces_every_row() {
        let first = beads_row("tix-1", "open");
        let second = beads_row("tix-2", "open");
        let mut exported = BTreeMap::new();
        exported.insert(first.id.clone(), fingerprint(&first).unwrap());

        let payload =
            select_payload(vec![first.clone(), second.clone()], &exported, false).unwrap();
        assert_eq!(payload.rows.len(), 1);
        assert_eq!(payload.rows[0].id, "tix-2");
        assert_eq!(payload.unchanged, 1);
        assert_eq!(payload.fingerprints.len(), 1);

        // Nothing changed at all: the empty payload is what skips `bd import`.
        let payload = select_payload(vec![first.clone()], &exported, false).unwrap();
        assert_eq!(payload.rows, Vec::<BeadsRow>::new());
        assert_eq!(payload.unchanged, 1);

        // `--all` ignores the watermark and sends everything.
        let payload = select_payload(vec![first, second], &exported, true).unwrap();
        assert_eq!(payload.rows.len(), 2);
        assert_eq!(payload.unchanged, 0);
        assert_eq!(payload.fingerprints.len(), 2);
    }

    #[test]
    fn the_workspace_identity_follows_bd_precedence() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("project");
        std::fs::create_dir_all(root.join(".beads")).unwrap();
        let nested = root.join("a").join("b");
        std::fs::create_dir_all(&nested).unwrap();
        let canonical =
            |path: &std::path::Path| std::fs::canonicalize(path).unwrap().display().to_string();
        let beads = root.join(".beads");
        let found = Some(canonical(&beads));

        // `BEADS_DIR` wins over everything, exactly as it does for `bd`.
        assert_eq!(
            workspace_identity_from(Some("/elsewhere/.beads"), Some(&beads), &nested),
            Some("/elsewhere/.beads".to_string())
        );
        // A blank `BEADS_DIR` is not a choice, so it falls through.
        assert_eq!(
            workspace_identity_from(Some("   "), Some(&beads), &nested),
            found.clone()
        );
        // The configured workspace beats discovery.
        assert_eq!(
            workspace_identity_from(None, Some(&beads), &nested),
            found.clone()
        );
        // Otherwise it is the nearest `.beads` walking up from the working
        // directory — here the project's own, found from two levels down.
        assert_eq!(workspace_identity_from(None, None, &nested), found);

        // And nothing at all when no workspace is above the directory.
        let bare = tmp.path().join("bare");
        std::fs::create_dir_all(&bare).unwrap();
        assert_eq!(workspace_identity_from(None, None, &bare), None);
    }

    struct FakeRunner {
        calls: RefCell<Vec<(String, Vec<String>)>>,
        ok: bool,
    }

    impl CommandRunner for FakeRunner {
        fn run(
            &self,
            program: &str,
            args: &[String],
            _env: &[(String, String)],
        ) -> Result<CommandOutcome> {
            self.calls
                .borrow_mut()
                .push((program.to_string(), args.to_vec()));
            Ok(CommandOutcome {
                success: self.ok,
                status: Some(if self.ok { 0 } else { 1 }),
                stdout: "Imported 1 issues from tix-beads.jsonl".to_string(),
                stderr: if self.ok {
                    String::new()
                } else {
                    "boom".to_string()
                },
            })
        }

        fn as_any(&self) -> &dyn std::any::Any {
            self
        }
    }

    #[test]
    fn push_shells_out_to_bd_import_with_a_staged_file() {
        let config = Config::default();
        let resolution = resolution_for(&config);
        let plan = plan("tix", &resolution, &[issue(1, "One")], &[]);
        let runner = FakeRunner {
            calls: RefCell::new(Vec::new()),
            ok: true,
        };
        let stdout = push(&runner, "bd", &plan.rows).unwrap();
        assert!(stdout.contains("Imported 1 issues"));

        let calls = runner.calls.borrow();
        assert_eq!(calls.len(), 1);
        let (program, args) = &calls[0];
        assert_eq!(program, "bd");
        assert_eq!(args[0], "import");
        assert!(args[1].ends_with(".jsonl"), "{args:?}");
    }

    #[test]
    fn a_failed_push_is_an_external_error_naming_the_tool() {
        let runner = FakeRunner {
            calls: RefCell::new(Vec::new()),
            ok: false,
        };
        let err = push(&runner, "bd", &[]).unwrap_err();
        assert!(matches!(err, TixError::External(_)), "{err:?}");
        assert!(err.to_string().contains("bd import"));
    }
}