tambo 0.10.0

use std::{ffi::OsString, path::Path};

use anyhow::{bail, Context, Result};
use serde::Deserialize;

use crate::{
    adapter::adapter_for_plan,
    configuration,
    crew::{self, CrewEvidence},
    embedding, emission,
    model::{ModelArtifactState, Plan, ShareState, State, ToolState},
};

#[derive(Deserialize)]
struct ImportPromotion {
    eligible: bool,
    blockers: Vec<String>,
}

#[derive(Deserialize)]
struct ImportResult {
    outcome: String,
    share_id: String,
    staging_graph: String,
    promotion: ImportPromotion,
}

/// What `apply` should do about an installed tool whose version is not the one
/// this Tambo build reviewed (aegis-5ctwu3).
///
/// Split out as a pure function ON PURPOSE: the act it decides is a network
/// download, so a test that drove `apply` end to end would be measuring GitHub
/// rather than the decision. Every branch below is covered by a unit test; the
/// branch that could not be, would not have been.
#[derive(Debug, PartialEq, Eq)]
pub(crate) enum Convergence {
    /// Installed IS the reviewed release. Nothing to do.
    Current,
    /// Installed is strictly older. Converge upward, then prove by read-back.
    Converge,
    /// Installed is at or ahead of the reviewed release, or the two versions
    /// are not comparable. Do NOT install — that would be a downgrade or a
    /// replacement on a guess (aegis-48dvl3). Say so instead.
    Decline,
    /// Skewed, and the operator asked for no installs. A MISSING tool already
    /// fails under `--skip-install`; a skewed one reporting "applied" was the
    /// inconsistency this bead is about.
    RefuseSkipInstall,
}

pub(crate) fn decide_convergence(
    installed: &str,
    desired: &str,
    is_current: bool,
    skip_install: bool,
) -> Convergence {
    if is_current {
        return Convergence::Current;
    }
    if skip_install {
        return Convergence::RefuseSkipInstall;
    }
    match crate::release_update::behind_reviewed(installed, desired) {
        Some(true) => Convergence::Converge,
        // `None` is "cannot compare", NOT "stale". Treating it as stale is how
        // an unrecognised build gets replaced on a guess.
        Some(false) | None => Convergence::Decline,
    }
}

pub fn apply(plan: &Plan, state_path: &Path, skip_install: bool) -> Result<State> {
    plan.validate()?;
    let mut state = State::read(state_path)?;
    let mut failures = Vec::new();
    for &name in &plan.tools {
        if name == crate::model::ToolName::Quipu && plan.external_quipu {
            // A failed external path/version probe cannot retain an earlier proof.
            state.tools.remove(name.as_str());
            state.write(state_path)?;
        }
        let adapter = adapter_for_plan(name, plan)?;
        let desired = adapter.desired_version();
        // aegis-5ctwu3: PRESENCE IS NOT CONVERGENCE. Until this match looked at
        // `is_current`, a binary that merely EXISTED was reported "applied" and
        // left at whatever version it was — bobbin sat at 0.1.0 while this build
        // reviewed 0.17.0, and `tambo install` then failed its own verify with
        // `error: unexpected argument '--source' found`. The installer knew the
        // newer contract, verified against it, and had never installed it; the
        // operator got a clap error instead of "your bobbin is 16 minors old".
        // `st doctor` had it right the whole time ("bobbin 0.1.0 installed —
        // 0.17.0 available (STALE)"), which is the tell that this was never a
        // detection problem.
        let version = (|| -> Result<String> {
            Ok(match adapter.version() {
                Ok(installed) if adapter.is_current(&installed) => installed,

                // Known-wrong under --skip-install. The flag means "do not install",
                // not "call it applied anyway", and a MISSING tool already fails
                // here — a SKEWED one reporting success was the inconsistency.
                Ok(installed) => {
                    match decide_convergence(&installed, &desired, false, skip_install) {
                        Convergence::Current => installed,
                        Convergence::RefuseSkipInstall => bail!(
                        "{} is installed at {installed} but this Tambo build reviewed {desired}; \
                         --skip-install refuses to converge it. Install {desired}, or drop \
                         --skip-install to let apply converge it.",
                        name.as_str()
                    ),
                        Convergence::Decline => {
                            println!(
                                "{}: NOT converged — installed {installed}, reviewed {desired}. \
                             Apply does not downgrade or replace an unrecognised version; \
                             use `tambo update-release` deliberately if that is what you want.",
                                name.as_str()
                            );
                            installed
                        }
                        Convergence::Converge => {
                            eprintln!(
                                "{}: stale ({installed}); converging to reviewed {desired}",
                                name.as_str()
                            );
                            adapter.install().with_context(|| {
                                format!("{} convergence to reviewed {desired}", name.as_str())
                            })?;
                            let after = adapter.version().with_context(|| {
                                format!("{} version read-back after convergence", name.as_str())
                            })?;
                            // The read-back is the proof, not the install's exit code.
                            if !adapter.is_current(&after) {
                                bail!(
                                    "{} did not reach reviewed release {desired} (got {after}); \
                                 refusing to record it as applied",
                                    name.as_str()
                                );
                            }
                            println!("{}: converged {installed} -> {after}", name.as_str());
                            after
                        }
                    }
                }

                Err(error)
                    if !skip_install
                        && !(name == crate::model::ToolName::Quipu && plan.external_quipu) =>
                {
                    eprintln!("{}: not installed ({error:#}); installing", name.as_str());
                    adapter
                    .install()
                    .with_context(|| {
                        format!(
                            "{} install step (run `tambo doctor` to list every blocker on this host)",
                            name.as_str()
                        )
                    })?;
                    adapter.version().with_context(|| {
                        format!("{} version read-back after install", name.as_str())
                    })?
                }
                Err(error) => {
                    return Err(error).context(format!("{} version read-back", name.as_str()))
                }
            })
        })();
        // aegis-z1u9s0: the version above is the MANAGED copy's for a member, so
        // a stale copy earlier on PATH left apply saying "applied" while the
        // shell ran the older release. Apply is the step that says done; it
        // checks what PATH runs, as verify does.
        let version = version.and_then(|version| {
            if name == crate::model::ToolName::Quipu && !plan.external_quipu {
                // A current pair needs only its missing alias, never a downgrade
                // to the older reviewed release just to add a command name.
                if skip_install {
                    crate::adapter::verify_existing_quipu_alias()?;
                } else {
                    crate::adapter::ensure_existing_quipu_alias()?;
                }
            }
            if !(name == crate::model::ToolName::Quipu && plan.external_quipu) {
                check_path_resolution(name)?;
            }
            Ok(version)
        });
        let version = match version {
            Ok(version) => version,
            Err(error) => {
                // An earlier successful run is no longer evidence for this tool.
                // Persist the invalidation before continuing with independent tools.
                state.tools.remove(name.as_str());
                state.write(state_path)?;
                eprintln!("{}: FAILED: {error:#}", name.as_str());
                failures.push(format!("{}: {error:#}", name.as_str()));
                continue;
            }
        };
        let remains_verified = state
            .tools
            .get(name.as_str())
            .is_some_and(|previous| previous.version == version && previous.verified);
        state.tools.insert(
            name.as_str().to_owned(),
            ToolState {
                version,
                applied: true,
                verified: remains_verified,
            },
        );
        state.write(state_path)?;
        emission::queue_transition(
            state_path,
            name.as_str(),
            "applied",
            &state.tools[name.as_str()].version,
        )?;
        println!("{}: applied", name.as_str());
    }
    if !failures.is_empty() {
        bail!(
            "{} tool(s) failed to apply:\n{}\nSuccessful tools were saved. Fix the failures and rerun the same command. \
             Stack configuration, model provisioning, crew setup, share imports and verification \
             have not run; run `tambo doctor` for preflight guidance.",
            failures.len(),
            failures.join("\n")
        );
    }
    if let Some(config) = &plan.stack_config {
        configuration::apply(config)?;
        println!("stack configuration: applied");
    }
    if let Some(model) = &plan.embedding_model {
        let provisioned = embedding::provision(model, crate::adapter::download_https)
            .context("embedding-model provisioning step")?;
        for artifact in provisioned {
            // Like tools, a re-fetch of identical pinned bytes keeps its
            // verified mark; anything else must earn it again.
            let remains_verified = state
                .models
                .get(&artifact.name)
                .is_some_and(|previous| previous.sha256 == artifact.sha256 && previous.verified);
            state.models.insert(
                artifact.name.clone(),
                ModelArtifactState {
                    path: artifact.path.clone(),
                    sha256: artifact.sha256.clone(),
                    provisioned: true,
                    verified: remains_verified,
                },
            );
            state.write(state_path)?;
            emission::queue_transition(
                state_path,
                &format!("embedding-model/{}", artifact.name),
                "applied",
                &artifact.sha256,
            )?;
            let outcome = match artifact.outcome {
                embedding::ArtifactOutcome::Fetched => "provisioned",
                embedding::ArtifactOutcome::Current => "current",
            };
            println!("embedding-model {}: {outcome}", artifact.name);
        }
    }
    #[cfg(unix)]
    if let Some(mcp) = &plan.quipu_mcp {
        let outcome = crate::quipu_mcp::provision(&mcp.url, mcp.token_file.as_deref())
            .context("quipu MCP registration step (headersHelper)")?;
        println!("quipu mcp: {outcome}");
    }
    if let Some(selection) = &plan.crew {
        crew::apply(selection, &mut state, skip_install)?;
        state.write(state_path)?;
        for (name, runtime) in &state.crew {
            emission::queue_transition(state_path, name, "applied", &runtime.version)?;
        }
    }
    // After the tools and crew (which may have just installed st): register the
    // stack's hook bundles for the tools this plan installs (aegis-u1ybxo).
    // Each tool's OWN bundle when the installed tool prints one (aegis-5s32or.5).
    let resolved = crate::tool_bundles::resolve(&crate::hook_bundles::selected(
        plan.tools.iter().map(|t| t.as_str()),
    ));
    for (name, _, source) in &resolved {
        println!("hook bundle {name}: source {}", source.label());
    }
    for line in crate::hook_bundles::register(&crate::tool_bundles::as_pairs(&resolved), "st")? {
        println!("hook bundle {line}");
    }
    // seeds + Desire Path: redirect habitual `bd` to `sd`, never overwriting a
    // host's existing `bd` rule (aegis-fbaso4).
    if crate::dp_alias::wanted(plan.tools.iter().map(|t| t.as_str())) {
        let outcome = crate::dp_alias::apply(crate::dp_alias::locate().as_deref(), None)
            .context("Desire Path bd -> sd alias step")?;
        println!("dp alias {}", outcome.line());
    }
    load_vocabulary(plan)?;
    consume_shares(plan, &mut state, state_path)?;
    Ok(state)
}

/// Load the pinned Quechua vocabulary (aegis-1i5h1j.3) into the plan's Quipu
/// store. It is part of Quipu: a plan without Quipu, or without a store to load
/// into, loads nothing, and verify still proves the release round trip in a
/// throwaway store.
fn load_vocabulary(plan: &Plan) -> Result<()> {
    if !plan.tools.contains(&crate::model::ToolName::Quipu) {
        return Ok(());
    }
    let cached = crate::vocabulary::ensure_cached()?;
    match plan.quipu_db.as_deref() {
        Some(db) => {
            crate::vocabulary::load(db)?;
            println!(
                "vocabulary: quechua v{} loaded into {}",
                crate::vocabulary::QUECHUA_VERSION,
                db.display()
            );
        }
        None => println!(
            "vocabulary: quechua v{} cached at {} (no plan quipu_db to load into)",
            crate::vocabulary::QUECHUA_VERSION,
            cached.display()
        ),
    }
    Ok(())
}

fn consume_shares(plan: &Plan, state: &mut State, state_path: &Path) -> Result<()> {
    let Some(db) = plan.quipu_db.as_deref() else {
        return Ok(());
    };
    for share in &plan.shares {
        let result = crate::adapter::checked(
            "quipu",
            [
                OsString::from("import"),
                share.as_os_str().to_owned(),
                OsString::from("--db"),
                db.as_os_str().to_owned(),
            ],
            None,
        )
        .with_context(|| format!("import canonical Quipu share {}", share.display()))?;
        let imported: ImportResult = serde_json::from_slice(&result.stdout)
            .with_context(|| format!("parse Quipu import result for {}", share.display()))?;
        if !matches!(
            imported.outcome.as_str(),
            "staged" | "quarantined" | "unchanged"
        ) {
            bail!(
                "Quipu returned unknown share import outcome {:?} for {}",
                imported.outcome,
                share.display()
            );
        }
        state.shares.insert(
            imported.share_id.clone(),
            ShareState {
                path: share.clone(),
                staging_graph: imported.staging_graph,
                outcome: imported.outcome.clone(),
                promotion_eligible: imported.promotion.eligible,
                blockers: imported.promotion.blockers,
            },
        );
        state.write(state_path)?;
        println!(
            "share {}: {} (promotion eligible: {})",
            imported.share_id, imported.outcome, imported.promotion.eligible
        );
    }
    Ok(())
}

/// Refuse to call a tool verified when the copy PATH runs is not the copy that
/// was verified (aegis-70qlhs). A different file with the same version is a
/// note, not a failure: it runs the same build.
fn check_path_resolution(name: crate::model::ToolName) -> Result<()> {
    let path = std::env::var_os("PATH");
    for &program in crate::adapter::programs(name) {
        match crate::adapter::path_resolution(program, path.as_deref()) {
            crate::adapter::PathResolution::Managed => {}
            crate::adapter::PathResolution::NotOnPath => println!(
                "{}: note: `{program}` is not on PATH; add the install directory to PATH to run it",
                name.as_str()
            ),
            crate::adapter::PathResolution::SameBuild { runs, managed } => println!(
                "{}: note: PATH runs {} rather than {}, but both report the same version",
                name.as_str(),
                runs.display(),
                managed.display()
            ),
            crate::adapter::PathResolution::Shadowed {
                runs,
                runs_version,
                managed,
                managed_version,
            } => anyhow::bail!(
                "{program} is SHADOWED: PATH runs {} ({}), not the verified {} ({}). \
                 Remove the stale copy or put {} earlier on PATH",
                runs.display(),
                runs_version.trim(),
                managed.display(),
                managed_version.trim(),
                managed
                    .parent()
                    .map(|d| d.display().to_string())
                    .unwrap_or_default()
            ),
        }
    }
    Ok(())
}

pub fn verify(plan: &Plan, state_path: &Path, evidence: &CrewEvidence) -> Result<State> {
    plan.validate()?;
    let mut state = State::read(state_path)?;
    // The hook bundles apply registers for the plan's tools (aegis-u1ybxo).
    // Asserted for every plan, as apply registers for every plan, and FIRST:
    // it is independent of the tools' functional checks, and a skewed tool
    // failing below must not hide whether the hooks are registered (measured
    // on a fresh Mac, where a bobbin skew left the bundles never asserted).
    for line in hook_bundles_verify(plan)? {
        println!("hook bundle {line}");
    }
    if plan.tools.contains(&crate::model::ToolName::Quipu) && plan.requires_remote_quipu() {
        if let Some(record) = state.tools.get_mut("quipu") {
            record.verified = false;
            state.write(state_path)?;
        }
        let server = std::env::var("QUIPU_SERVER")
            .ok()
            .filter(|url| !url.trim().is_empty())
            .unwrap_or_else(|| {
                plan.quipu_mcp
                    .as_ref()
                    .map(|mcp| mcp.url.clone())
                    .unwrap_or_else(|| "http://localhost:3030".to_owned())
            });
        let source = crate::quipu_credential::verify(&server)
            .context("live Quipu client credential verification")?;
        println!("quipu credential: canonical file 0400/0700 and authenticated read verified; source: {source}");
    }
    for &name in &plan.tools {
        if name == crate::model::ToolName::Quipu && plan.external_quipu {
            // A failed external path/version probe cannot retain an earlier proof.
            state.tools.remove(name.as_str());
            state.write(state_path)?;
        }
        let adapter = adapter_for_plan(name, plan)?;
        let version = adapter
            .version()
            .with_context(|| format!("{} version read-back", name.as_str()))?;
        // aegis-5ctwu3 item 3. A verification failure on a SKEWED tool used to
        // surface the tool's own complaint — `error: unexpected argument
        // '--source' found` — which points at an argument, not at the cause. We
        // verify against the contract of the release we reviewed, so when the
        // installed version is not that release, the skew IS the finding and
        // must be said first.
        adapter.verify().with_context(|| {
            let desired = adapter.desired_version();
            if adapter.is_current(&version) {
                format!("{} functional verification", name.as_str())
            } else {
                format!(
                    "{} functional verification — VERSION SKEW: {version} is installed, \
                     this Tambo build reviewed and verifies against {desired}. The error \
                     below is most likely that skew, not a broken tool. Converge with \
                     `tambo apply` (or `tambo update`), then re-verify.",
                    name.as_str()
                )
            }
        })?;
        if !(name == crate::model::ToolName::Quipu && plan.external_quipu) {
            check_path_resolution(name)?;
        }
        state.tools.insert(
            name.as_str().to_owned(),
            ToolState {
                version,
                applied: true,
                verified: true,
            },
        );
        state.write(state_path)?;
        emission::queue_transition(
            state_path,
            name.as_str(),
            "verified",
            &state.tools[name.as_str()].version,
        )?;
        println!("{}: verified", name.as_str());
    }
    // After the tools: Desire Path itself was verified above (aegis-fbaso4).
    if crate::dp_alias::wanted(plan.tools.iter().map(|t| t.as_str())) {
        let outcome = crate::dp_alias::verify(crate::dp_alias::locate().as_deref(), None)
            .context("Desire Path bd -> sd alias")?;
        println!("dp alias {}", outcome.line());
    }
    if let Some(model) = &plan.embedding_model {
        // This is deliberately a re-hash, not an embed round-trip; see
        // embedding::verify for why claiming more would be a banner.
        embedding::verify(model).context("embedding-model artifact re-hash step")?;
        for artifact in &model.artifacts {
            state.models.insert(
                artifact.name.clone(),
                ModelArtifactState {
                    path: model.destination.join(&artifact.name),
                    sha256: artifact.sha256.clone(),
                    provisioned: true,
                    verified: true,
                },
            );
            state.write(state_path)?;
            emission::queue_transition(
                state_path,
                &format!("embedding-model/{}", artifact.name),
                "verified",
                &artifact.sha256,
            )?;
            println!("embedding-model {}: digest re-checked", artifact.name);
        }
    }
    #[cfg(unix)]
    if let Some(mcp) = &plan.quipu_mcp {
        let notes = crate::quipu_mcp::verify(&mcp.url)
            .context("quipu MCP authenticated-write probe (headersHelper)")?;
        for note in notes {
            println!("quipu mcp: {note}");
        }
        println!("quipu mcp: authenticated MCP write reached Quipu's parser; unauthenticated control was refused");
    }
    if let Some(selection) = &plan.crew {
        crew::verify(selection, evidence, &mut state)?;
        state.write(state_path)?;
        for (name, runtime) in &state.crew {
            emission::queue_transition(state_path, name, "verified", &runtime.version)?;
        }
    }
    Ok(state)
}

/// Compare the running toolchain to the releases reviewed by this Tambo
/// build. Returns true only when every selected tool is current.
pub(crate) fn reviewed_drift_status(installed: &str, desired: &str) -> &'static str {
    match crate::release_update::behind_reviewed(installed, desired) {
        Some(true) => "update available",
        Some(false) if crate::release_update::behind_reviewed(desired, installed) == Some(true) => {
            "ahead of reviewed pin; downgrade refused"
        }
        Some(false) | None => "version drift; review required",
    }
}

/// A tool whose binary a package manager owns is judged by its package receipt
/// and never installed by caboodle (aegis-6f1, option A): the package lane is
/// its only writer. Returns the line to print and whether it is current.
fn package_owned_status(name: crate::model::ToolName, desired: &str) -> Option<(String, bool)> {
    use crate::package::{apt_receipt, primary_program, standing, Standing};
    let receipt = primary_program(name)
        .and_then(|program| apt_receipt(&program))
        .or_else(|| crate::package::brew_receipt(name))?;
    let status = standing(&receipt.version, desired);
    let verdict = match status {
        Standing::Current => "current",
        Standing::Ahead => "ahead of reviewed pin; downgrade refused",
        Standing::Behind => {
            "behind reviewed pin; delivered by the package lane, not installed by caboodle"
        }
        Standing::Unknown => "version drift; review required",
    };
    Some((
        format!(
            "{}: {verdict} (package-owned: {} {} {}; reviewed: {desired})",
            name.as_str(),
            receipt.manager,
            receipt.package,
            receipt.version
        ),
        status == Standing::Current,
    ))
}

/// Member delivery through the stack's Homebrew tap (aegis-6f1). Applies only
/// where caboodle may drive brew and the tool has a stack formula. Returns the
/// verified version when brew delivered (or already holds) the tool, `None` to
/// fall back to the other lanes (the release download when the tap is older
/// than the pin). Never downgrades.
fn brew_update(
    name: crate::model::ToolName,
    adapter: &dyn crate::adapter::Adapter,
) -> Result<Option<String>> {
    use crate::package::{
        brew_channel, brew_converge, brew_formula, brew_program, standing, Standing,
    };
    if brew_formula(name).is_none() || brew_program().is_none() {
        return Ok(None);
    }
    let Some(channel) = brew_channel(name)? else {
        return Ok(None);
    };
    let desired = adapter.desired_version();
    let stable = format!("pkg {}", channel.stable);
    if channel.pinned {
        let installed = channel
            .installed
            .context("pinned brew formula has no installed receipt")?;
        println!(
            "{}: pinned Homebrew formula; update skipped ({} {}; stable: {})",
            name.as_str(),
            crate::package::STACK_TAP,
            installed,
            channel.stable
        );
        return brew_version(adapter, &installed).map(Some);
    }
    match channel.installed {
        Some(installed) => match standing(&installed, &stable) {
            Standing::Behind => {
                // A tap older than the minimum reviewed release cannot deliver
                // it; keep the existing package ownership/fallback decision.
                if standing(&channel.stable, &desired) == Standing::Behind {
                    return Ok(None);
                }
                brew_converge(name, true)?;
            }
            Standing::Current | Standing::Ahead => {
                let word = if standing(&installed, &stable) == Standing::Ahead {
                    "ahead of stable channel; downgrade refused"
                } else {
                    "current stable channel"
                };
                println!(
                    "{}: {word} (package-owned: brew {} {}; stable: {})",
                    name.as_str(),
                    crate::package::STACK_TAP,
                    installed,
                    channel.stable
                );
                return brew_version(adapter, &installed).map(Some);
            }
            Standing::Unknown => {
                anyhow::bail!("{} brew versions cannot be compared", name.as_str())
            }
        },
        None => {
            if standing(&channel.stable, &desired) == Standing::Behind {
                return Ok(None);
            }
            if let Ok(installed) = adapter.version() {
                if crate::release_update::behind_reviewed(&stable, &installed) == Some(true) {
                    println!(
                        "{}: installed {installed} is newer than {} {}; not installed by brew",
                        name.as_str(),
                        crate::package::STACK_TAP,
                        channel.stable
                    );
                    return Ok(Some(installed));
                }
            }
            brew_converge(name, false)?;
        }
    }
    let receipt = brew_channel(name)?
        .context("brew channel after delivery")?
        .installed
        .context("brew receipt after delivery")?;
    if !matches!(
        standing(&receipt, &stable),
        Standing::Current | Standing::Ahead
    ) {
        anyhow::bail!(
            "{} brew update did not reach stable {} (receipt: {receipt})",
            name.as_str(),
            channel.stable
        );
    }
    // Resolve after delivery: a fresh package install creates the PATH pair.
    let package_adapter = adapter.package_adapter()?;
    let adapter = package_adapter.as_deref().unwrap_or(adapter);
    adapter
        .verify()
        .with_context(|| format!("{} verification after brew delivery", name.as_str()))?;
    let version = brew_version(adapter, &receipt)?;
    println!(
        "{}: delivered by brew ({} {}) and verified",
        name.as_str(),
        crate::package::STACK_TAP,
        receipt
    );
    Ok(Some(version))
}

fn brew_version(adapter: &dyn crate::adapter::Adapter, receipt: &str) -> Result<String> {
    // Current/pinned reads also use the package pair, never an unrelated
    // Cargo copy. This is deliberately lazy until a read-back is needed.
    let package_adapter = adapter.package_adapter()?;
    let adapter = package_adapter.as_deref().unwrap_or(adapter);
    let version = adapter
        .version()
        .context("version read-back after brew delivery")?;
    // Quipu reports its client and server as two semicolon-separated
    // identities. Check both; a plain version line must not absorb the
    // separator, and a current client must not hide a stale server.
    let identities: Vec<_> = if adapter.name() == crate::model::ToolName::Quipu {
        version.split(';').take(2).collect()
    } else {
        vec![version.as_str()]
    };
    if identities.iter().any(|identity| {
        crate::package::standing(receipt, identity.trim()) != crate::package::Standing::Current
    }) {
        anyhow::bail!("brew receipt and executable version disagree");
    }
    Ok(version)
}

pub fn check_updates(plan: &Plan) -> Result<bool> {
    plan.validate()?;
    let mut current = true;
    for &name in &plan.tools {
        let adapter = adapter_for_plan(name, plan)?;
        let desired = adapter.desired_version();
        if name == crate::model::ToolName::Quipu && plan.external_quipu {
            println!(
                "quipu: externally owned; release convergence not assessed ({})",
                adapter.version()?
            );
            continue;
        }
        if let Some((line, is_current)) = package_owned_status(name, &desired) {
            current &= is_current;
            println!("{line}");
            continue;
        }
        match adapter.version() {
            Ok(installed) if adapter.is_current(&installed) => {
                println!("{}: current ({installed})", name.as_str());
            }
            Ok(installed) => {
                current = false;
                println!(
                    "{}: {} (installed: {installed}; reviewed: {desired})",
                    name.as_str(),
                    reviewed_drift_status(&installed, &desired)
                );
            }
            Err(error) => {
                current = false;
                println!(
                    "{}: missing or unreadable ({error:#}); reviewed: {desired}",
                    name.as_str()
                );
            }
        }
    }
    if let Some(selection) = &plan.crew {
        current &= crew::check_updates(selection);
    }
    Ok(current)
}

/// Converge drifted tools to the reviewed releases and functionally verify
/// each changed tool before recording it as current.
pub fn update(plan: &Plan, state_path: &Path, evidence: &CrewEvidence) -> Result<State> {
    plan.validate()?;
    let mut state = State::read(state_path)?;
    if crate::package::brew_program().is_some()
        && plan.tools.iter().any(|name| {
            crate::package::brew_formula(*name).is_some()
                && !(*name == crate::model::ToolName::Quipu && plan.external_quipu)
        })
    {
        crate::package::brew_refresh()?;
    }
    for &name in &plan.tools {
        let adapter = adapter_for_plan(name, plan)?;
        if name == crate::model::ToolName::Quipu && plan.external_quipu {
            println!(
                "quipu: externally owned; update skipped ({})",
                adapter.version()?
            );
            continue;
        }
        if let Some(version) = brew_update(name, adapter.as_ref())? {
            state.tools.insert(
                name.as_str().to_owned(),
                ToolState {
                    version: version.clone(),
                    applied: true,
                    verified: true,
                },
            );
            state.write(state_path)?;
            emission::queue_transition(state_path, name.as_str(), "updated", &version)?;
            continue;
        }
        if let Some((line, _)) = package_owned_status(name, &adapter.desired_version()) {
            println!("{line}");
            continue;
        }
        let before = adapter.version().ok();
        if before
            .as_deref()
            .is_some_and(|installed| adapter.is_current(installed))
        {
            if name == crate::model::ToolName::Quipu {
                crate::adapter::ensure_existing_quipu_alias()?;
            }
            println!("{}: current", name.as_str());
            continue;
        }
        #[cfg(unix)]
        crate::release_update::guard_reviewed_update(name, &adapter.desired_version())?;
        eprintln!(
            "{}: converging {:?} -> {}",
            name.as_str(),
            before,
            adapter.desired_version()
        );
        adapter
            .install()
            .with_context(|| format!("{} reviewed update install", name.as_str()))?;
        let version = adapter
            .version()
            .with_context(|| format!("{} version read-back after update", name.as_str()))?;
        if !adapter.is_current(&version) {
            bail!(
                "{} update did not reach reviewed release {} (got {})",
                name.as_str(),
                adapter.desired_version(),
                version
            );
        }
        adapter
            .verify()
            .with_context(|| format!("{} verification after update", name.as_str()))?;
        state.tools.insert(
            name.as_str().to_owned(),
            ToolState {
                version: version.clone(),
                applied: true,
                verified: true,
            },
        );
        state.write(state_path)?;
        emission::queue_transition(state_path, name.as_str(), "updated", &version)?;
        println!("{}: updated and verified", name.as_str());
    }
    if let Some(selection) = &plan.crew {
        if !crew::check_updates(selection) {
            crew::apply(selection, &mut state, false)?;
            crew::verify(selection, evidence, &mut state)?;
            for line in hook_bundles_verify(plan)? {
                println!("hook bundle {line}");
            }
            state.write(state_path)?;
            for (name, runtime) in &state.crew {
                emission::queue_transition(state_path, name, "updated", &runtime.version)?;
            }
        }
    }
    Ok(state)
}

fn hook_bundles_verify(plan: &Plan) -> Result<Vec<String>> {
    let resolved = crate::tool_bundles::resolve(&crate::hook_bundles::selected(
        plan.tools.iter().map(|t| t.as_str()),
    ));
    crate::hook_bundles::verify(&crate::tool_bundles::as_pairs(&resolved))
}

pub fn verify_questions(plan: &Plan, db: Option<&Path>) -> Result<()> {
    plan.validate()?;
    let intent = plan.intent.as_ref().context(
        "plan has no intended-use/question contract; regenerate it through the Phase 2 interview",
    )?;
    for (index, contract) in intent.anticipated_questions.iter().enumerate() {
        // The self-test question is about the stack, not the user's graph, so it
        // always runs against a scratch store seeded with its own fact.
        let scratch = if contract.is_self_test() {
            Some(seed_self_test_store()?)
        } else {
            None
        };
        let db = scratch.as_ref().map(|(_, db)| db.as_path()).or(db);
        let mut args = vec![OsString::from("read"), OsString::from(&contract.sparql)];
        if let Some(db) = db {
            args.push(OsString::from("--db"));
            args.push(db.as_os_str().to_owned());
        }
        let result = crate::adapter::checked("quipu", args, None)
            .with_context(|| format!("anticipated question {} query", index + 1))?;
        let answer = String::from_utf8_lossy(&result.stdout);
        if !answer.contains(&contract.expected) {
            anyhow::bail!(
                "anticipated question {} was executable but its answer did not contain {:?}: {}",
                index + 1,
                contract.expected,
                contract.question
            );
        }
        println!("question {}: verified — {}", index + 1, contract.question);
    }
    Ok(())
}

/// A throwaway Quipu store holding exactly the self-test fact. The tempdir is
/// returned so it lives as long as the query that reads it.
fn seed_self_test_store() -> Result<(tempfile::TempDir, std::path::PathBuf)> {
    use crate::model::SELF_TEST_MARKER;
    let dir = tempfile::tempdir().context("create self-test question store")?;
    let db = dir.path().join("self-test.db");
    let episode = dir.path().join("episode.json");
    std::fs::write(
        &episode,
        format!(
            r#"{{"name":"caboodle self-test question","episode_body":"self-test seed","source":"caboodle","group_id":"caboodle-verification","nodes":[{{"name":"{SELF_TEST_MARKER}","type":"Verification","description":"caboodle self-test question seed"}}],"edges":[]}}"#
        ),
    )
    .context("write self-test episode")?;
    crate::adapter::checked(
        "quipu",
        [
            OsString::from("episode"),
            episode.into_os_string(),
            OsString::from("--db"),
            db.clone().into_os_string(),
        ],
        None,
    )
    .context("seed the self-test question store")?;
    Ok((dir, db))
}

/// Provision a pre-issued client credential; never mint or rotate an identity.
pub fn provision_quipu_credential(source: &Path, server: &str) -> Result<()> {
    let home = std::env::var_os("HOME").context("HOME required for canonical Quipu credential")?;
    crate::quipu_credential::provision(source, server, Path::new(&home))?;
    println!(
        "quipu credential: issued file accepted and canonical installation verified (0400/0700)"
    );
    Ok(())
}

#[cfg(test)]
mod convergence_tests {
    use super::{decide_convergence, reviewed_drift_status, Convergence};

    const REVIEWED: &str = "bobbin 0.16.2";

    #[test]
    fn reviewed_drift_never_proposes_a_downgrade_or_unknown_replacement() {
        assert_eq!(
            reviewed_drift_status("sd 0.1.4 (seeds)", "seeds 0.0.4"),
            "ahead of reviewed pin; downgrade refused"
        );
        assert_eq!(
            reviewed_drift_status("sd 0.0.3", "seeds 0.0.4"),
            "update available"
        );
        for installed in ["sd dev", "sd 0.0.4", "sd 0.0.4-rc1"] {
            assert_eq!(
                reviewed_drift_status(installed, "seeds 0.0.4"),
                "version drift; review required"
            );
        }
    }

    #[test]
    fn current_is_left_alone_even_under_skip_install() {
        // Unreachable from apply today (is_current is matched first), which is why
        // this branch is covered here rather than deleted as apparently dead.
        assert_eq!(
            decide_convergence(REVIEWED, REVIEWED, true, true),
            Convergence::Current
        );
        assert_eq!(
            decide_convergence(REVIEWED, REVIEWED, true, false),
            Convergence::Current
        );
    }

    #[test]
    fn stale_converges_and_skip_install_refuses_instead_of_claiming_applied() {
        assert_eq!(
            decide_convergence("bobbin 0.1.0", REVIEWED, false, false),
            Convergence::Converge
        );
        assert_eq!(
            decide_convergence("bobbin 0.1.0", REVIEWED, false, true),
            Convergence::RefuseSkipInstall
        );
    }

    #[test]
    fn ahead_or_incomparable_is_declined_not_replaced() {
        assert_eq!(
            decide_convergence("bobbin 0.99.0", REVIEWED, false, false),
            Convergence::Decline
        );
        assert_eq!(
            decide_convergence("bobbin dev", REVIEWED, false, false),
            Convergence::Decline
        );
    }
}