aristo-cli 0.2.0

//! Canon-verify dispatch path for `aristo verify`.
//!
//! Replaces the original `pending_full > 0 => NotImplemented` arm in
//! `commands/verify/mod.rs`. For Full-resolved entries that are
//! canon-bound (`aristos:` or `kanon:`), this module:
//!
//! 1. Builds [`VerifySessionTag`]s from the index + canon-matches cache.
//! 2. Runs the push-first precheck via [`aristo_core::git`].
//! 3. POSTs `/canon/verify/sessions` and prints `session_id` +
//!    `view_url` for the user (detach default per WORKFLOW.md §7c row 1).
//!
//! Auth is required: the §14 design predicates canon-verify on having
//! an `arta_*` token (we re-derive scopes server-side and authorize
//! against `repository_flavors`). If no token is resolved, the SDK
//! surfaces an actionable "run `aristo auth login`" hint and skips —
//! it does NOT graceful-degrade (the user explicitly asked to verify;
//! we shouldn't silently no-op).
//!
//! Non-canon-bound `verify="full"` entries fall through to the
//! existing deferred-design `NotImplemented` arm — they're a separate
//! verification mechanism (see `docs/deferred/verify-test-design.md`)
//! and out of scope for §14.

use std::collections::HashSet;
use std::path::Path;
use std::time::{Duration, Instant};

use anstyle::{Ansi256Color, AnsiColor, Color, Style};
use textwrap::core::display_width;

use aristo_core::canon::CanonMatchesFile;
use aristo_core::canon_verify::{
    AnnotationOutcomeStatus, CallFrame, DifferentialReport, FaultSpec, FieldDivergence, Finding,
    FrameRole, GetVerifySessionResponse, HttpVerifyClient, PostVerifySessionResponse,
    SessionStatus, Snapshot, TestOutcome, TestOutcomeStatus, TestStep, VerifyClient, VerifyError,
    VerifySessionRequest, VerifySessionTag,
};
use aristo_core::expectations::{Expectation, ExpectationsFile};
use aristo_core::index::{AnnotationId, IndexEntry, IndexFile, IntentEntry};

use super::card::Card;
use super::waiver;
use crate::workspace::Workspace;
use crate::{CliError, CliResult};

/// Long-poll server-side hold-time (§7c row 9). The server may ignore
/// this and return immediately in the prototype — the SDK still sends
/// the hint for forward-compat.
const LONGPOLL_WAIT_SECS: u32 = 30;

/// Default between-poll backoff when the server returns immediately.
/// Keeps the CLI rendering smooth (§7c row 9 "3s render") and bounded.
/// Overridable via `ARISTO_VERIFY_POLL_MS` for tests; production
/// callers never set it.
const POLL_INTERVAL: Duration = Duration::from_secs(3);

fn poll_interval() -> Duration {
    if let Ok(ms) = std::env::var("ARISTO_VERIFY_POLL_MS") {
        if let Ok(n) = ms.parse::<u64>() {
            return Duration::from_millis(n);
        }
    }
    POLL_INTERVAL
}

/// Heartbeat cadence (§7c row 13: "still running… (N seconds elapsed)").
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(60);

/// Identifies an entry pending canon-verify dispatch. Lightweight —
/// the actual request body assembly happens in [`build_tags`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct CanonDispatchEntry<'a> {
    pub id: &'a AnnotationId,
    pub entry: &'a IntentEntry,
}

/// From the full set of Full-resolved annotation ids (computed by the
/// existing dispatcher loop), partition them into the canon-verify
/// bucket vs. the legacy non-canon bucket.
///
/// Canon-bound = id namespace is `aristos:` or `kanon:` per
/// [`AnnotationId::is_canon_bound`].
pub(crate) fn partition_full<'a>(
    index: &'a IndexFile,
    pending_full_ids: &[&'a AnnotationId],
) -> (
    Vec<CanonDispatchEntry<'a>>,
    Vec<&'a AnnotationId>, // non-canon-bound; keep the NotImplemented hint
) {
    let mut canon: Vec<CanonDispatchEntry<'a>> = Vec::new();
    let mut other: Vec<&'a AnnotationId> = Vec::new();
    for id in pending_full_ids {
        let Some(entry) = index.entries.get(*id) else {
            continue;
        };
        let IndexEntry::Intent(intent) = entry else {
            // Assume entries don't carry `verify` (resolve to Bool(false));
            // they never reach the Full arm. Defensive.
            continue;
        };
        if id.is_canon_bound() {
            canon.push(CanonDispatchEntry { id, entry: intent });
        } else {
            other.push(id);
        }
    }
    (canon, other)
}

/// Build the [`VerifySessionTag`] list for a set of canon-bound
/// Full-resolved entries. Joins the index entry (for `annotation_id`
/// via `binding.linked` + source_path via `file:line`) with the
/// canon-matches cache (for `canon_id` + `version`).
///
/// Entries whose `linked` is missing (Local binding state — which
/// shouldn't happen for canon-prefixed ids but is defensively
/// handled), whose canon-matches entry is absent, or whose accepted-
/// match list is empty are silently dropped from the dispatch.
/// They'd fail the server-side eligibility check anyway and surface
/// a clearer error on the next `aristo canon refresh`.
pub(crate) fn build_tags(
    entries: &[CanonDispatchEntry<'_>],
    matches: &CanonMatchesFile,
) -> Vec<VerifySessionTag> {
    entries
        .iter()
        .filter_map(|d| build_one_tag(d, matches))
        .collect()
}

fn build_one_tag(
    dispatch: &CanonDispatchEntry<'_>,
    matches: &CanonMatchesFile,
) -> Option<VerifySessionTag> {
    let annotation_id = match &dispatch.entry.binding {
        aristo_core::index::BindingState::Local => return None,
        aristo_core::index::BindingState::Bound { linked } => linked.as_str().to_string(),
        aristo_core::index::BindingState::Certified { linked, .. } => linked.as_str().to_string(),
    };
    // Strip `aristos:` / `kanon:` prefix to get the bare canon_id.
    let canon_id = strip_canon_prefix(dispatch.id.as_str()).to_string();
    // Pull version from canon-matches cache. The cache is keyed by
    // the FULL AnnotationId (matching the source-form id used in
    // both source and index).
    let cache_entry = matches.entries.get(dispatch.id)?;
    let accepted = cache_entry.accepted_matches.first()?;
    let version = accepted.version.clone();

    let source_path = format_source_path(&dispatch.entry.file, dispatch.entry.site.as_str());

    Some(VerifySessionTag {
        annotation_id,
        canon_id,
        version,
        source_path,
    })
}

fn strip_canon_prefix(id: &str) -> &str {
    id.strip_prefix("aristos:")
        .or_else(|| id.strip_prefix("kanon:"))
        .unwrap_or(id)
}

/// Format `file:line` from the index entry's `file` + the line
/// extracted from `site`. The site format is `"fn foo (line N)"`
/// per the walker / index emit; absent a line (defensive), we emit
/// just the file path so the user still sees a clickable hint.
fn format_source_path(file: &str, site: &str) -> String {
    match extract_line(site) {
        Some(line) => format!("{file}:{line}"),
        None => file.to_string(),
    }
}

fn extract_line(site: &str) -> Option<u32> {
    let after = site.split(" (line ").nth(1)?;
    let trimmed = after.trim_end_matches(')');
    trimmed.parse().ok()
}

/// What the dispatcher does after partitioning + tag-building. Surfaced
/// as a separate fn so tests can drive it with a mock client.
#[cfg(test)]
pub(crate) fn dispatch_session<C: VerifyClient + ?Sized>(
    client: &C,
    req: &VerifySessionRequest,
) -> Result<PostVerifySessionResponse, VerifyError> {
    client.post_session(req)
}

/// End-to-end canon-verify dispatch:
///
/// 1. Build the tag list (silent skip for entries with no
///    canon-matches cache entry — they'd 4xx anyway).
/// 2. If empty, return 0 (nothing to dispatch — the existing skip-
///    counters in `emit_summary` already informed the user).
/// 3. Push-first precheck via git.
/// 4. Derive `commit_sha` + `repo_full_name`.
/// 5. Construct the HTTP client + POST.
/// 6. Print `session_id` + `view_url`.
///
/// Returns the number of tags actually dispatched (for the summary
/// emit on the caller side). Errors are CLI errors so the caller can
/// `?` them up.
pub(crate) fn run_canon_dispatch(
    workspace_root: &Path,
    matches_path: &Path,
    expectations_path: &Path,
    canon_entries: &[CanonDispatchEntry<'_>],
    tags_filter: Option<&[String]>,
    wait: bool,
) -> CliResult<usize> {
    if canon_entries.is_empty() {
        return Ok(0);
    }

    // 1. Auth — required up front. If no token, surface actionable hint.
    let creds = aristo_core::auth::resolve_full().map_err(no_auth_to_cli_error)?;

    // 2. Build tags from index + cache. Drop the no-version cases.
    let matches = CanonMatchesFile::read(matches_path).map_err(|e| CliError::Other {
        message: format!(
            "failed to read canon-matches cache at {}: {e}",
            matches_path.display()
        ),
        exit_code: 1,
    })?;
    let mut tags = build_tags(canon_entries, &matches);

    // 2b. Optional --tags filter (E4): subset to the requested ids.
    // Validates each requested id (reject `arta_*` — those are server-
    // side opaque refs that the user never sees in source).
    if let Some(requested) = tags_filter {
        let allowed = build_tag_filter_set(canon_entries, requested)?;
        tags.retain(|t| allowed.contains(&t.annotation_id));
        if tags.is_empty() {
            return Err(CliError::Other {
                message: format!(
                    "--tags filter matched zero eligible canon-bound entries. \
                     Requested ids: {}",
                    requested.join(", ")
                ),
                exit_code: 1,
            });
        }
    }
    if tags.is_empty() {
        // Every canon-bound entry was missing a cache entry. Surface
        // a helpful message rather than POSTing empty.
        println!(
            "\n→ {} canon-bound entries are pending verification, but no `accepted_matches` were found in",
            canon_entries.len()
        );
        println!(
            "  .aristo/canon-matches.toml. Run `aristo canon refresh` to repopulate the cache."
        );
        return Ok(0);
    }

    // 3. Resolve repo + commit_sha via git.
    let repo_full_name = match aristo_core::auth::derive_repo_full_name(workspace_root) {
        Ok(r) => r,
        Err(e) => {
            // ARISTO_REPO env override as a CI escape hatch.
            std::env::var("ARISTO_REPO").map_err(|_| CliError::Other {
                message: format!(
                    "could not determine repo for verify: {e}\n  \
                     Set ARISTO_REPO=<owner/repo> to override (CI use)."
                ),
                exit_code: 1,
            })?
        }
    };
    let commit_sha =
        aristo_core::git::rev_parse_head(workspace_root).map_err(|e| CliError::Other {
            message: format!("git rev-parse HEAD failed: {e}"),
            exit_code: 1,
        })?;

    // 4. Push-first precheck (WORKFLOW.md §4 + §7c).
    let pushed =
        aristo_core::git::commit_present_on_remote(workspace_root, &commit_sha).map_err(|e| {
            CliError::Other {
                message: format!("git push-first precheck failed: {e}"),
                exit_code: 1,
            }
        })?;
    if !pushed {
        return Err(CliError::Other {
            message: format!(
                "HEAD ({}) is not pushed to origin. Push your branch first; \
                 `aristo verify --watch` for local-edit sync is planned but \
                 not yet shipped.",
                short_sha(&commit_sha)
            ),
            exit_code: 1,
        });
    }

    // 5. Build the HTTP client. ARETTA_API_URL overrides for tests.
    let base_url =
        std::env::var("ARETTA_API_URL").unwrap_or_else(|_| creds.server.as_str().to_string());
    let client: Box<dyn VerifyClient> = if let Some(mock) = test_mock_client_from_env() {
        mock
    } else {
        Box::new(HttpVerifyClient::new(base_url, &creds.token))
    };

    // 6. POST.
    let req = VerifySessionRequest {
        repo_full_name,
        commit_sha,
        tags,
    };
    let resp = client.post_session(&req).map_err(verify_error_to_cli)?;
    let dispatched = req.tags.len();
    print_session_dispatched(&req, &resp);

    // 7. Optional poll-to-completion (--wait).
    if wait {
        // Phase 16 (c): the user's accepted gaps are a read-time join over
        // the results — they decide which failures are "known" (green) and
        // catch a waived property that has started passing (red ratchet).
        let expectations =
            ExpectationsFile::read(expectations_path).map_err(|e| CliError::Other {
                message: format!("failed to read {}: {e}", expectations_path.display()),
                exit_code: 1,
            })?;
        let final_snapshot = poll_until_terminal(&*client, &resp.session_id)?;
        render_final_snapshot(&final_snapshot, &expectations);
        let verdict = waiver::evaluate(&final_snapshot, &expectations);
        if verdict.is_red() {
            return Err(exit_error_for(&verdict));
        }
    }

    Ok(dispatched)
}

/// The `--wait` failure error, worded from the waiver-aware verdict.
/// `failed` counts only un-waived (genuine) property failures; a tripped
/// ratchet is surfaced explicitly so the message points at the fix.
fn exit_error_for(verdict: &waiver::WaiverVerdict) -> CliError {
    let ratchet = if verdict.ratchet_breaches > 0 {
        format!(", {} accepted-gap-now-passes", verdict.ratchet_breaches)
    } else {
        String::new()
    };
    CliError::Other {
        message: format!(
            "verify reported {} failed, {} build_failed, {} inconclusive{ratchet}",
            verdict.unwaived_failed, verdict.build_failed, verdict.inconclusive
        ),
        exit_code: 1,
    }
}

/// Re-attach to an existing session (E3 — `aristo verify --view <id>`).
/// One GET (or polling loop with --wait), render, exit-code derive.
/// Skips POST + push-first precheck entirely.
pub(crate) fn run_view_session(session_id: &str, wait: bool) -> CliResult<()> {
    let creds = aristo_core::auth::resolve_full().map_err(no_auth_to_cli_error)?;
    let base_url =
        std::env::var("ARETTA_API_URL").unwrap_or_else(|_| creds.server.as_str().to_string());
    let client: Box<dyn VerifyClient> = if let Some(mock) = test_mock_client_from_env() {
        mock
    } else {
        Box::new(HttpVerifyClient::new(base_url, &creds.token))
    };

    let snapshot = if wait {
        poll_until_terminal(&*client, session_id)?
    } else {
        client
            .get_session(session_id, None)
            .map_err(verify_error_to_cli)?
    };
    // `--view` doesn't require a workspace (it only needs an HTTP client),
    // so with no `.aristo/` above cwd we apply no waivers. When a workspace
    // IS present we read its expectations the same way the dispatch path
    // does — a *malformed* file hard-errors (loud, not silently dropped);
    // only a genuinely-absent file reads as empty.
    let expectations = match Workspace::find(None) {
        Ok(ws) => ExpectationsFile::read(&ws.expectations_path()).map_err(|e| CliError::Other {
            message: format!("failed to read {}: {e}", ws.expectations_path().display()),
            exit_code: 1,
        })?,
        Err(_) => ExpectationsFile::default(),
    };
    render_final_snapshot(&snapshot, &expectations);
    if wait {
        let verdict = waiver::evaluate(&snapshot, &expectations);
        if verdict.is_red() {
            return Err(exit_error_for(&verdict));
        }
    }
    Ok(())
}

/// Long-poll loop until the session reaches a terminal state. Renders
/// an intermediate snapshot at first non-terminal response so the user
/// sees the in-flight state; emits a heartbeat every 60s.
fn poll_until_terminal<C: VerifyClient + ?Sized>(
    client: &C,
    session_id: &str,
) -> CliResult<GetVerifySessionResponse> {
    let started = Instant::now();
    let mut last_heartbeat = started;
    let mut intermediate_rendered = false;

    loop {
        let snapshot = client
            .get_session(session_id, Some(LONGPOLL_WAIT_SECS))
            .map_err(verify_error_to_cli)?;
        if snapshot.status.is_terminal() {
            return Ok(snapshot);
        }
        if !intermediate_rendered {
            // First non-terminal poll: show what's running.
            println!();
            println!(
                "  status: {} ({} of {} annotations verified so far)",
                status_label(snapshot.status),
                snapshot.summary.verified,
                snapshot.summary.total_annotations
            );
            intermediate_rendered = true;
        }
        if last_heartbeat.elapsed() >= HEARTBEAT_INTERVAL {
            let elapsed = started.elapsed().as_secs();
            println!("  still running… ({elapsed}s elapsed)");
            last_heartbeat = Instant::now();
        }
        // Back off briefly so we don't hammer when the server returns
        // immediately (i.e., when ?wait= is ignored — current proxy
        // state).
        std::thread::sleep(poll_interval());
    }
}

fn build_tag_filter_set(
    canon_entries: &[CanonDispatchEntry<'_>],
    requested: &[String],
) -> CliResult<HashSet<String>> {
    let mut allowed: HashSet<String> = HashSet::new();
    // Build a map from canon-bound AnnotationId.as_str() → linked arta_*
    // so the user can pass either source-form ids (`aristos:foo`) or
    // canon-id suffixes (`foo`) without surprises.
    let mut by_source: std::collections::HashMap<&str, String> = std::collections::HashMap::new();
    let mut by_canon: std::collections::HashMap<&str, String> = std::collections::HashMap::new();
    for d in canon_entries {
        let linked = match &d.entry.binding {
            aristo_core::index::BindingState::Bound { linked } => linked.as_str().to_string(),
            aristo_core::index::BindingState::Certified { linked, .. } => {
                linked.as_str().to_string()
            }
            aristo_core::index::BindingState::Local => continue,
        };
        by_source.insert(d.id.as_str(), linked.clone());
        let canon = strip_canon_prefix(d.id.as_str());
        by_canon.insert(canon, linked);
    }
    for raw in requested {
        let id = raw.trim();
        if id.is_empty() {
            continue;
        }
        if id.starts_with("arta_") {
            return Err(CliError::Other {
                message: format!(
                    "--tags rejects opaque server ids (got `{id}`). Pass the source-form id \
                     instead, e.g. `--tags aristos:foo,kanon:bar`."
                ),
                exit_code: 2,
            });
        }
        if let Some(linked) = by_source.get(id) {
            allowed.insert(linked.clone());
        } else if let Some(linked) = by_canon.get(id) {
            allowed.insert(linked.clone());
        } else {
            return Err(CliError::Other {
                message: format!(
                    "--tags id `{id}` is not a canon-bound entry in this workspace's index"
                ),
                exit_code: 1,
            });
        }
    }
    Ok(allowed)
}

// ─── Rendering (WORKFLOW.md §6 — CLI form) ──────────────────────────────────

fn render_final_snapshot(snapshot: &GetVerifySessionResponse, expectations: &ExpectationsFile) {
    println!();
    println!(
        "session {} — verifying {} against canon {}",
        snapshot.session_id,
        short_sha(&snapshot.user_commit_sha),
        snapshot.canon_version
    );
    println!(
        "status: {} ({}/{} verified)",
        status_label(snapshot.status),
        snapshot.summary.verified,
        snapshot.summary.total_annotations
    );
    println!();
    println!("{:<55}  {:<14} TESTS", "ANNOTATION", "STATUS");
    for ann in &snapshot.annotations {
        // Phase 16 (c): is this annotation a user-accepted gap?
        let waiver_match: Option<(AnnotationId, &Expectation)> =
            waiver::waiver_key(&ann.tier, &ann.canon_id)
                .and_then(|id| expectations.get(&id).map(|e| (id, e)));
        // A clean accepted gap is a waived property `Failed` with no
        // operational test failure hiding in its test set — an operational
        // break falls through to the normal card so it can't be masked.
        let waived_failing = matches!(ann.status, AnnotationOutcomeStatus::Failed)
            && waiver_match.is_some()
            && !waiver::tests_operationally_broken(&ann.tests);
        let ratchet_breach =
            matches!(ann.status, AnnotationOutcomeStatus::Verified) && waiver_match.is_some();

        let (icon, word) = if waived_failing {
            ("[gap]", "accepted")
        } else {
            (
                annotation_icon(ann.status),
                annotation_status_word(ann.status),
            )
        };
        let passed = ann
            .tests
            .iter()
            .filter(|t| matches!(t.status, TestOutcomeStatus::Pass))
            .count();
        let header = format!(
            "{}{}@{} ({})",
            ann.tier, ann.canon_id, ann.version, ann.scope
        );
        let tests_summary = if ann.tests.is_empty() {
            "(no coverage)".to_string()
        } else {
            format!("{}/{} passed", passed, ann.tests.len())
        };
        println!("{:<55}  {} {:<11}  {}", header, icon, word, tests_summary);
        println!("  {}", ann.source_path);

        // A waived gap replaces the violation card with the accepted-gap
        // frame; a waived property that now passes trips the ratchet. In
        // both cases the per-test card loop is skipped.
        if let Some((id, exp)) = &waiver_match {
            if waived_failing {
                // Show the SAME violation detail, reframed as an accepted
                // known failure (orange headline) with the user's reason as a
                // footer. Fall back to the bare frame if no report is attached.
                let repro = format!("{}{}", ann.tier, ann.canon_id);
                match first_failing_report(&ann.tests) {
                    Some(report) => render_violation_card(report, &repro, CardKind::Accepted(exp)),
                    None => render_accepted_gap(exp),
                }
                continue;
            }
            if ratchet_breach {
                render_ratchet_breach(id);
                continue;
            }
        }

        for t in &ann.tests {
            if !matches!(
                t.status,
                TestOutcomeStatus::Fail
                    | TestOutcomeStatus::BuildFailed
                    | TestOutcomeStatus::CloneFailed
                    | TestOutcomeStatus::Timeout
                    | TestOutcomeStatus::Error
            ) {
                continue;
            }
            // Phase 16: a structured DifferentialReport renders a
            // violation card instead of the terse bullet. Fall back to
            // the bullet when no report is attached.
            match &t.report {
                Some(report) => render_violation_card(
                    report,
                    &format!("{}{}", ann.tier, ann.canon_id),
                    CardKind::Violated,
                ),
                None => render_test_bullet(t),
            }
        }

        // Un-waived property failure: point the user at the one-liner that
        // acknowledges it as a known gap. Future runs then report a "known
        // gap (accepted)" instead of a failure (with a strict ratchet so a
        // stale waiver can't rot). Offered only for a property `Failed` —
        // the only outcome a waiver can forgive.
        if matches!(ann.status, AnnotationOutcomeStatus::Failed) && waiver_match.is_none() {
            anstream::println!(
                "    {C_BOLD}Known limitation?{C_BOLD:#} acknowledge it so future runs report a known gap, not a failure:"
            );
            anstream::println!(
                "      {C_DIM}aristo verify --accept {}{} --because \"<why this is OK for now>\"{C_DIM:#}",
                ann.tier, ann.canon_id
            );
            anstream::println!();
        }
    }

    // Phase 16 (c): a waiver that matched no annotation this run is stale or
    // drifted (renamed upstream, removed from source, or never applicable).
    // Surface it on stderr so the silent-non-match can't rot unnoticed.
    let matched: std::collections::BTreeSet<AnnotationId> = snapshot
        .annotations
        .iter()
        .filter_map(|ann| {
            let id = waiver::waiver_key(&ann.tier, &ann.canon_id)?;
            expectations.is_waived(&id).then_some(id)
        })
        .collect();
    for id in expectations.entries.keys() {
        if !matched.contains(id) {
            eprintln!(
                "  note: waiver for `{}` in .aristo/expectations.toml matched no annotation this \
                 run — renamed, removed, or stale. `aristo list` to check; remove if no longer needed.",
                id.as_str()
            );
        }
    }

    println!();
    // No `view_url` on GetVerifySessionResponse; the dashboard link is
    // derivable from session_id when needed but we surface only the
    // session id here to keep the output tight.
}

/// The terse fall-back row for a failing test with no attached report:
/// `• <bin> <word> — see <stderr_url>`.
fn render_test_bullet(t: &TestOutcome) {
    let bin = t.test_binary.as_deref().unwrap_or("(session)");
    let word = test_status_word(t.status);
    match &t.stderr_url {
        Some(url) => println!("  • {bin} {word} — see {url}"),
        None => println!("  • {bin} {word}"),
    }
}

/// Phase 16 (c) — the bare accepted-gap frame, used only when the failing
/// test carried no structured report (so there's no violation body to
/// show). The rich path reuses [`render_violation_card`] with
/// [`CardKind::Accepted`].
fn render_accepted_gap(exp: &Expectation) {
    let mut card = Card::new();
    card.raw(
        format!("{C_WARN}⚠ KNOWN PROPERTY FAILURE{C_WARN:#}"),
        "⚠ KNOWN PROPERTY FAILURE",
    );
    card.blank();
    card.wrap_hang("accepted  ", &exp.reason, C_WARN);
    if let Some(tracking) = &exp.tracking {
        card.line(&format!("tracking  {tracking}"), C_DIM, 0);
    }
    anstream::println!();
    anstream::print!("{}", card.render());
    anstream::println!();
}

/// Phase 16 (c) — the strict ratchet. A waived property that now passes
/// is a red signal: the gap is closed, so the waiver is stale and must be
/// removed before it silently masks a future regression.
fn render_ratchet_breach(id: &AnnotationId) {
    let mut card = Card::new();
    card.raw(
        format!("{C_HEAD}✗ accepted gap now passes{C_HEAD:#}"),
        "✗ accepted gap now passes",
    );
    card.blank();
    card.wrap(
        "This property now holds, so the waiver is stale — remove it from \
         .aristo/expectations.toml:",
        Style::new(),
        0,
    );
    card.blank();
    card.line(id.as_str(), C_DIM, 2);
    anstream::println!();
    anstream::print!("{}", card.render());
    anstream::println!();
}

// ── Card styling. `anstream` auto-strips these when stdout isn't a TTY
// (piped / CI / `NO_COLOR`), so the `-`/`+` markers still carry the
// meaning with color removed.
const C_HEAD: Style = Style::new()
    .bold()
    .fg_color(Some(Color::Ansi(AnsiColor::Red)));
const C_BOLD: Style = Style::new().bold();
const C_DIM: Style = Style::new().dimmed();
const C_DEL: Style = Style::new().fg_color(Some(Color::Ansi(AnsiColor::Red)));
const C_ADD: Style = Style::new().fg_color(Some(Color::Ansi(AnsiColor::Green)));
/// Orange — a known, accepted failure: not a passing run, but not a red
/// build break either. 256-color so it reads as orange (not yellow).
const C_WARN: Style = Style::new()
    .bold()
    .fg_color(Some(Color::Ansi256(Ansi256Color(208))));

/// How a property-failure card is framed.
enum CardKind<'a> {
    /// Un-waived: red `PROPERTY VIOLATED`; the caller prints the reproduce
    /// command + accept hint below the card.
    Violated,
    /// A user-accepted gap: orange `KNOWN PROPERTY FAILURE`; the same
    /// violation body, plus the user's reason + tracker as a footer.
    Accepted(&'a Expectation),
}

/// Render a [`DifferentialReport`] as a structured card. User-framed (no
/// model/verification internals); every value is driven off the report
/// fields. The only difference between an un-waived failure and an
/// accepted gap is the headline colour/label and the footer — the
/// violation detail (statement, diff, provenance) is identical, because an
/// accepted gap is still a real, visible failure the user chose to live
/// with.
fn render_violation_card(report: &DifferentialReport, repro_id: &str, kind: CardKind<'_>) {
    let card = build_violation_card(report, &kind);

    anstream::println!();
    anstream::print!("{}", card.render());

    // Reproduce — only for an un-waived failure (an accepted gap needs no
    // action). Kept BELOW the card so the command stays one copy-pasteable
    // line.
    if matches!(kind, CardKind::Violated) {
        anstream::println!();
        anstream::println!("  {C_BOLD}Reproduce{C_BOLD:#}");
        anstream::println!("    {C_DIM}aristo verify --filter id={repro_id} --rerun{C_DIM:#}");
    }
    anstream::println!();
}

/// Build the violation [`Card`] (no printing) so tests can render it to a
/// string and assert on the framed text. The card order is: headline →
/// statement → impl → fault banner → "what turso ran" → "where your code
/// broke it" → "divergence observed" → (accepted footer).
fn build_violation_card(report: &DifferentialReport, kind: &CardKind<'_>) -> Card {
    let Finding::StateEq {
        expected,
        actual,
        divergence,
    } = &report.finding;

    let mut card = Card::new();
    // Headline — red "violated" vs. orange "known failure".
    match kind {
        CardKind::Violated => card.raw(
            format!(
                "{C_HEAD}✗ PROPERTY VIOLATED{C_HEAD:#}   {C_BOLD}{}{C_BOLD:#}",
                report.property.canon_id
            ),
            &format!("✗ PROPERTY VIOLATED   {}", report.property.canon_id),
        ),
        CardKind::Accepted(_) => card.raw(
            format!(
                "{C_WARN}⚠ KNOWN PROPERTY FAILURE{C_WARN:#}   {C_BOLD}{}{C_BOLD:#}",
                report.property.canon_id
            ),
            &format!("⚠ KNOWN PROPERTY FAILURE   {}", report.property.canon_id),
        ),
    }
    card.blank();
    card.wrap(&report.property.statement, Style::new(), 0);

    // The impl anchor — the user's own code. The spec anchor (the closed
    // verification harness) is deliberately not surfaced.
    if let Some(s) = &report.property.impl_source {
        let loc = match &s.snippet {
            Some(snip) => format!("impl   {}:{}  ·  {snip}", s.path, s.line),
            None => format!("impl   {}:{}", s.path, s.line),
        };
        card.blank();
        card.line(&loc, C_DIM, 0);
    }

    // The injected fault that led to the divergence. Additive — absent in
    // a Slice-1 report, so older reports render exactly as before.
    if let Some(fault) = &report.scenario.fault {
        render_fault_banner(&mut card, fault);
    }

    // "what turso ran" — the top-level test workload (the SQL the test
    // issued), the faulting statement flagged. Additive.
    if !report.scenario.test_shape.is_empty() {
        card.blank();
        render_test_shape(&mut card, &report.scenario.test_shape);
    }

    // "where your code broke it" — the turso call-path spine. Additive.
    if !report.scenario.call_path.is_empty() {
        card.blank();
        render_call_path(&mut card, report);
    }

    // "divergence observed" — the masked counterexample, AFTER the call
    // path so the user reads cause (the broken call) then effect (the
    // diverging field).
    card.blank();
    render_divergence(&mut card, report, expected, actual, divergence);

    // Accepted footer: the user's acknowledgement, inside the card.
    if let CardKind::Accepted(exp) = kind {
        card.blank();
        card.wrap_hang("accepted  ", &exp.reason, C_WARN);
        if let Some(tracking) = &exp.tracking {
            card.line(&format!("tracking  {tracking}"), C_DIM, 0);
        }
    }

    card
}

/// The fault banner (Slice 3): the named, parameterized injected fault,
/// e.g. `fault  fail_nth · sync #1 → EIO  ·  fired 1×`. Turso-observable
/// only — the policy identity + the injected error, never the model.
fn render_fault_banner(card: &mut Card, fault: &FaultSpec) {
    let op = fault.op.to_lowercase();
    let plain = format!(
        "fault  {} · {op} #{} → {}  ·  fired {}×",
        fault.kind, fault.nth, fault.error, fault.fired_count
    );
    let styled = format!(
        "{C_WARN}fault{C_WARN:#}  {} · {op} #{} → {C_WARN}{}{C_WARN:#}  {C_DIM}·  fired {}×{C_DIM:#}",
        fault.kind, fault.nth, fault.error, fault.fired_count
    );
    card.raw(styled, &plain);
}

/// "what turso ran" (Slice 3a): the top-level test workload — the SQL
/// statements the test issued — as plain labels. Driven off
/// `scenario.test_shape`; the op-trace timeline is no longer surfaced (it
/// stays in the contract data but is not rendered). The per-step `faulted`
/// flag stays on the wire but is not rendered as prose — the "divergence
/// observed" block carries the failure; a generic faulted indicator is a
/// deferred improvement.
fn render_test_shape(card: &mut Card, steps: &[TestStep]) {
    card.raw(
        format!("{C_BOLD}what turso ran{C_BOLD:#}"),
        "what turso ran",
    );
    for step in steps {
        card.line(&step.label, C_DIM, 2);
    }
}

/// "where your code broke it" (Slice 3a): the turso call-path spine.
///
/// The header NAMES the entry frame (`… · inside Connection::execute`);
/// the body renders `call_path[1..]` (the entry is already named, so it's
/// dropped from the list). The spine is a pre-order walk; `depth`
/// reconstructs the tree. Trunk frames (`depth <= fork_depth`) render
/// flush at the base indent; from the fork the two branches — the
/// pre-fault effect write and the fault — render with `├─`/`└─`
/// connectors + `│`/space guides so they read as siblings.
///
/// Every shown frame is turso (the spine is IP-filtered upstream); labels
/// are shortened for display. The fault frame carries a glyph-gated
/// marker derived from the report (`✗ {op} → {error}   fault injected
/// here`); the effect frame renders as a plain spine frame — its diverging
/// values live in the "divergence observed" block below, not on the frame.
/// Meaning survives with colour stripped.
fn render_call_path(card: &mut Card, report: &DifferentialReport) {
    let frames = &report.scenario.call_path;
    let Some(entry) = frames.first() else {
        return;
    };

    // Header: name the entry frame inline; render the rest below it.
    let entry_label = shorten_frame_label(&entry.label);
    card.raw(
        format!(
            "{C_BOLD}where your code broke it{C_BOLD:#}   {C_DIM}·   inside {entry_label}{C_DIM:#}"
        ),
        &format!("where your code broke it   ·   inside {entry_label}"),
    );
    card.blank();

    let body = &frames[1..];
    let fork_depth = fork_depth(frames);

    // Base indent of the spine inside the card.
    const BASE: usize = 2;

    // The fault marker, derived from the injected fault (op + error). The
    // effect frame carries no marker — its diverging values live in the
    // "divergence observed" block, not on the frame.
    let fault_marker = report.scenario.fault.as_ref().map(|f| {
        format!(
            "✗ {} → {}   fault injected here",
            f.op.to_lowercase(),
            f.error
        )
    });

    // Pass 1: build each frame's plain prefix+label and remember which
    // carry a marker, so the markers can be column-aligned.
    struct RenderedFrame {
        prefix: String,
        label: String,
        role: FrameRole,
    }
    let mut rendered: Vec<RenderedFrame> = Vec::with_capacity(body.len());
    for (i, frame) in body.iter().enumerate() {
        let rel = frame.depth.saturating_sub(fork_depth) as usize;
        let prefix = if rel == 0 {
            " ".repeat(BASE)
        } else {
            format!("{}{}", " ".repeat(BASE), tree_prefix(body, i, rel))
        };
        rendered.push(RenderedFrame {
            prefix,
            label: shorten_frame_label(&frame.label),
            role: frame.role,
        });
    }

    // The marker column: 2 past the fault frame's plain end (only the fault
    // frame carries an inline marker).
    let marker_col = rendered
        .iter()
        .filter(|r| matches!(r.role, FrameRole::Fault))
        .map(|r| display_width(&r.prefix) + display_width(&r.label))
        .max()
        .map(|w| w + 2);

    // Pass 2: emit. The prefix is plain (counts toward width directly); the
    // label is dim. Only the fault frame carries an inline marker; the
    // effect frame renders like any other spine frame.
    for r in &rendered {
        match r.role {
            FrameRole::Normal | FrameRole::Effect => {
                let plain = format!("{}{}", r.prefix, r.label);
                let styled = format!("{}{C_DIM}{}{C_DIM:#}", r.prefix, r.label);
                card.raw(styled, &plain);
            }
            FrameRole::Fault => {
                let marker = fault_marker.as_deref().unwrap_or("✗ fault injected here");
                let here = display_width(&r.prefix) + display_width(&r.label);
                let gap = " ".repeat(marker_col.unwrap_or(here + 2).saturating_sub(here));
                let plain = format!("{}{}{gap}{marker}", r.prefix, r.label);
                let styled = format!(
                    "{}{C_DIM}{}{C_DIM:#}{gap}{C_WARN}{marker}{C_WARN:#}",
                    r.prefix, r.label
                );
                card.raw(styled, &plain);
            }
        }
    }
}

/// The `├─`/`└─` connector + `│`/space guide prefix for a forked frame at
/// relative level `rel` (`= depth - fork_depth >= 1`), index `i` in the
/// subtree slice `body`. Standard pre-order tree drawing: for each
/// ancestor level a `│  ` guide if that branch is still open (a later
/// sibling exists before the branch closes), else three spaces; at the
/// frame's own level a `├─ ` if a later sibling follows, else `└─ `.
fn tree_prefix(body: &[CallFrame], i: usize, rel: usize) -> String {
    // The caller's `fork_depth` for this frame: depth[i] - rel.
    let fork = body[i].depth.saturating_sub(rel as u32);
    let rel_at = |idx: usize| (body[idx].depth.saturating_sub(fork)) as usize;

    // Scanning forward from `i`, is there another frame at relative `level`
    // before one shallower than `level` (i.e. is that branch still open)?
    let has_later_sibling = |level: usize| {
        for j in (i + 1)..body.len() {
            let rj = rel_at(j);
            if rj < level {
                return false;
            }
            if rj == level {
                return true;
            }
        }
        false
    };

    let mut out = String::new();
    for level in 1..rel {
        out.push_str(if has_later_sibling(level) {
            "│  "
        } else {
            "   "
        });
    }
    out.push_str(if has_later_sibling(rel) {
        "├─ "
    } else {
        "└─ "
    });
    out
}

/// The depth of the first branching node on the spine — the shallowest
/// `depth` at which two distinct later frames hang off the same parent.
/// For a pure (un-forked) path no such node exists and the whole spine is
/// the flush trunk, so we return `u32::MAX` (every `depth - fork_depth`
/// saturates to 0 → indent stays flush).
fn fork_depth(frames: &[CallFrame]) -> u32 {
    // In a pre-order spine, descent increments depth by one. A *drop* —
    // a frame whose depth is `<=` its predecessor's — means the walk has
    // popped back up and re-opened a branch off the shared prefix; that
    // branch's parent sits at `this.depth - 1`. The shallowest such parent
    // is the first branching node. A pure path never drops, so `fork`
    // stays `u32::MAX` and every subtree indent saturates to 0 (flush).
    let mut fork = u32::MAX;
    let mut prev: Option<u32> = None;
    for f in frames {
        if let Some(p) = prev {
            if f.depth <= p && f.depth >= 1 {
                fork = fork.min(f.depth - 1);
            }
        }
        prev = Some(f.depth);
    }
    fork
}

/// Shorten a wire symbol for display: drop the leading `turso_core::`
/// module path and keep the last 1–2 `::`-segments, so
/// `turso_core::storage::wal::WalFile::prepare_wal_finish` →
/// `WalFile::prepare_wal_finish` and `turso_core::vdbe::execute::op_halt`
/// → `op_halt`. Deterministic: a trailing `Type::method` pair is kept
/// whole; a bare free function keeps just its name.
fn shorten_frame_label(label: &str) -> String {
    let segs: Vec<&str> = label.split("::").collect();
    let n = segs.len();
    if n == 0 {
        return label.to_string();
    }
    // The last segment is always the fn/op name. If the penultimate
    // segment looks like a type (UpperCamel), keep the `Type::name` pair;
    // otherwise (a module path) keep just the name.
    let last = segs[n - 1];
    if n >= 2 {
        let penult = segs[n - 2];
        let is_type = penult
            .chars()
            .next()
            .is_some_and(|c| c.is_ascii_uppercase());
        if is_type {
            return format!("{penult}::{last}");
        }
    }
    last.to_string()
}

/// "divergence observed" (Slice 3a): the masked counterexample, rendered
/// AFTER the call path. A bold header naming the mask (compared fields +
/// how many were ignored), then the aligned `expected`/`actual` rows
/// (each tagged with its snapshot's side-label), then the provenance as a
/// hanging `why` paragraph. Driven entirely off the report.
fn render_divergence(
    card: &mut Card,
    report: &DifferentialReport,
    expected: &Snapshot,
    actual: &Snapshot,
    divergence: &[FieldDivergence],
) {
    // Header: the mask. `compared [a, b]   (N fields ignored)`.
    let compared = report.relation.compared.join(", ");
    let n_ignored = report.relation.ignored.len();
    let field_word = if n_ignored == 1 { "field" } else { "fields" };
    card.raw(
        format!(
            "{C_BOLD}divergence observed{C_BOLD:#}   {C_DIM}·   compared [{compared}]   \
             ({n_ignored} {field_word} ignored){C_DIM:#}"
        ),
        &format!(
            "divergence observed   ·   compared [{compared}]   ({n_ignored} {field_word} ignored)"
        ),
    );
    card.blank();

    // Aligned expected/actual rows. The tag column is 11 wide (so the
    // `field = value` clause starts in a common column); the value column
    // is padded so the snapshot side-label lands in a common column.
    const LABEL_W: usize = 11;
    const VALUE_W: usize = 21;
    for d in divergence {
        diff_row(
            card,
            "expected",
            &d.field,
            &d.expected,
            &expected.label,
            C_DEL,
            LABEL_W,
            VALUE_W,
        );
        diff_row(
            card,
            "actual",
            &d.field,
            &d.actual,
            &actual.label,
            C_ADD,
            LABEL_W,
            VALUE_W,
        );
    }

    // Provenance: the proxy/durability collapse story, hung off `why`.
    if let Some(why) = divergence.first().and_then(|d| d.provenance.as_deref()) {
        card.blank();
        card.wrap_hang("  why  ", why, C_DIM);
    }
}

/// One `expected`/`actual` row of the divergence block: a left-justified
/// `tag`, then `{field} = {value}` padded to `value_w`, then the snapshot
/// `side` label. The `field = value` clause is coloured by `vstyle`
/// (red/green) so the diff survives colour-stripping via the tag word.
#[allow(clippy::too_many_arguments)]
fn diff_row(
    card: &mut Card,
    tag: &str,
    field: &str,
    value: &str,
    side: &str,
    vstyle: Style,
    label_w: usize,
    value_w: usize,
) {
    let clause = format!("{field} = {value}");
    let label_pad = " ".repeat(label_w.saturating_sub(display_width(tag)).max(1));
    let value_pad = " ".repeat(value_w.saturating_sub(display_width(&clause)).max(1));
    let plain = format!("{tag}{label_pad}{clause}{value_pad}{side}");
    let styled = format!(
        "{C_DIM}{tag}{C_DIM:#}{label_pad}{vstyle}{clause}{vstyle:#}{value_pad}{C_DIM}{side}{C_DIM:#}"
    );
    card.raw_indented(styled, &plain, 2);
}

/// The report on the first failing test of an annotation, if any.
fn first_failing_report(tests: &[TestOutcome]) -> Option<&DifferentialReport> {
    tests.iter().find_map(|t| {
        matches!(
            t.status,
            TestOutcomeStatus::Fail
                | TestOutcomeStatus::BuildFailed
                | TestOutcomeStatus::CloneFailed
                | TestOutcomeStatus::Timeout
                | TestOutcomeStatus::Error
        )
        .then_some(t.report.as_ref())
        .flatten()
    })
}

fn status_label(s: SessionStatus) -> &'static str {
    match s {
        SessionStatus::Queued => "queued",
        SessionStatus::Running => "running",
        SessionStatus::Done => "done",
        SessionStatus::Failed => "failed",
        SessionStatus::TimedOut => "timed out",
        SessionStatus::Cancelled => "cancelled",
    }
}

fn annotation_icon(s: AnnotationOutcomeStatus) -> &'static str {
    match s {
        AnnotationOutcomeStatus::Verified => "[ok]",
        AnnotationOutcomeStatus::Failed => "[fail]",
        AnnotationOutcomeStatus::BuildFailed => "[warn]",
        AnnotationOutcomeStatus::Inconclusive => "[?]",
        AnnotationOutcomeStatus::NoCoverage => "[--]",
    }
}

fn annotation_status_word(s: AnnotationOutcomeStatus) -> &'static str {
    match s {
        AnnotationOutcomeStatus::Verified => "verified",
        AnnotationOutcomeStatus::Failed => "failed",
        AnnotationOutcomeStatus::BuildFailed => "build_failed",
        AnnotationOutcomeStatus::Inconclusive => "inconclusive",
        AnnotationOutcomeStatus::NoCoverage => "no_coverage",
    }
}

fn test_status_word(s: TestOutcomeStatus) -> &'static str {
    match s {
        TestOutcomeStatus::Pass => "passed",
        TestOutcomeStatus::Fail => "failed",
        TestOutcomeStatus::BuildFailed => "build_failed",
        TestOutcomeStatus::CloneFailed => "clone_failed",
        TestOutcomeStatus::Timeout => "timeout",
        TestOutcomeStatus::Error => "error",
    }
}

fn print_session_dispatched(req: &VerifySessionRequest, resp: &PostVerifySessionResponse) {
    let annotation_word = if resp.plan_size == 1 {
        "annotation"
    } else {
        "annotations"
    };
    println!();
    println!(
        "→ verify session dispatched — verifying {} {annotation_word} against {}",
        resp.plan_size,
        short_sha(&req.commit_sha)
    );
    println!("  session: {}", resp.session_id);
    println!("  view:    {}", resp.view_url);
    println!();
    println!(
        "  Re-attach with: aristo verify --view {} --wait",
        resp.session_id
    );
}

fn short_sha(sha: &str) -> String {
    sha.chars().take(7).collect()
}

fn no_auth_to_cli_error(e: aristo_core::auth::AuthError) -> CliError {
    CliError::Other {
        message: format!(
            "verify requires authentication: {e}\n  \
             Run `aristo auth login` to sign in.\n  \
             (Without a token, the SDK skips canon-bound `verify=\"full\"` \
             entries — they'd be rejected server-side anyway.)"
        ),
        exit_code: 1,
    }
}

fn verify_error_to_cli(e: VerifyError) -> CliError {
    match e {
        VerifyError::Auth(inner) => CliError::Other {
            message: format!(
                "verify auth error: {inner}\n  \
                 Your token may be expired — re-run `aristo auth login`."
            ),
            exit_code: 1,
        },
        VerifyError::BadRequest {
            status: 402,
            message,
        } => CliError::Other {
            message: format!(
                "no canon coverage applies for your scopes — \
                 contact Aretta for DP onboarding to enable verification.\n  \
                 (server message: {message})"
            ),
            exit_code: 1,
        },
        VerifyError::BadRequest { status, message } => CliError::Other {
            message: format!("verify server rejected request (HTTP {status}): {message}"),
            exit_code: 1,
        },
        other => CliError::Other {
            message: format!("verify failed: {other}"),
            exit_code: 1,
        },
    }
}

// ─── Test-only mock plumbing ────────────────────────────────────────────────

/// Build a [`VerifyClient`] from `ARISTO_CANON_VERIFY_FIXTURE` if set.
///
/// The env var points at a JSON file with the canned POST response.
/// Used by CLI integration tests to exercise the dispatch path
/// without spinning up an HTTP server. Production callers must never
/// set this; the file-on-disk gating + env var make accidental
/// production use loud.
fn test_mock_client_from_env() -> Option<Box<dyn VerifyClient>> {
    let path = std::env::var("ARISTO_CANON_VERIFY_FIXTURE").ok()?;
    let raw = std::fs::read_to_string(&path).ok()?;
    let parsed: FixtureFile = serde_json::from_str(&raw).ok()?;
    let post_resp = parsed.post.map(|p| PostVerifySessionResponse {
        session_id: p.session_id,
        view_url: p.view_url,
        plan_size: p.plan_size,
    });
    // Pre-canned GET response sequence (for --view + --wait paths).
    let get_responses: Vec<GetVerifySessionResponse> = parsed.gets.unwrap_or_default();
    // Record the POST body to a sibling file so tests can inspect it.
    let record_path = format!("{path}.posted.json");
    let mock = match (post_resp, get_responses.is_empty()) {
        (Some(p), true) => aristo_core::canon_verify::MockVerifyClient::with_post_response(p),
        (Some(p), false) => {
            aristo_core::canon_verify::MockVerifyClient::with_post_and_gets(p, get_responses)
        }
        (None, false) => {
            aristo_core::canon_verify::MockVerifyClient::with_get_responses(get_responses)
        }
        (None, true) => return None,
    };
    Some(Box::new(RecordingMock {
        inner: mock,
        record_path,
    }))
}

#[derive(serde::Deserialize)]
struct FixtureFile {
    post: Option<FixturePost>,
    gets: Option<Vec<GetVerifySessionResponse>>,
}

#[derive(serde::Deserialize)]
struct FixturePost {
    session_id: String,
    view_url: String,
    plan_size: u32,
}

/// Wraps [`MockVerifyClient`] to write the POSTed request body to a
/// sidecar file so the CLI integration test can assert wire-shape.
struct RecordingMock {
    inner: aristo_core::canon_verify::MockVerifyClient,
    record_path: String,
}

impl VerifyClient for RecordingMock {
    fn post_session(
        &self,
        req: &VerifySessionRequest,
    ) -> Result<PostVerifySessionResponse, VerifyError> {
        if let Ok(serialized) = serde_json::to_string_pretty(req) {
            let _ = std::fs::write(&self.record_path, serialized);
        }
        self.inner.post_session(req)
    }

    fn get_session(
        &self,
        session_id: &str,
        wait_seconds: Option<u32>,
    ) -> Result<aristo_core::canon_verify::GetVerifySessionResponse, VerifyError> {
        self.inner.get_session(session_id, wait_seconds)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use aristo_core::canon::{AcceptedMatch, CacheEntry, CacheMeta, PrefixTier};
    use aristo_core::index::{
        ArtaId, BindingState, CoveredRegion, IndexEntry, IntentEntry, Meta, Sha256, Status,
        VerifyLevel, VerifyMethod,
    };
    use std::collections::BTreeMap;

    fn empty_index() -> IndexFile {
        IndexFile {
            meta: Meta {
                schema_version: 1,
                generated_by: None,
                generated_at: None,
                source_root: None,
            },
            entries: BTreeMap::new(),
        }
    }

    fn zero_hash() -> Sha256 {
        Sha256::parse(&format!("sha256:{}", "0".repeat(64))).unwrap()
    }

    fn arta(s: &str) -> ArtaId {
        ArtaId::parse(s).unwrap()
    }

    /// The committed golden — byte-copy of the cross-repo contract
    /// fixture. The renderer is driven entirely off the deserialized
    /// report, so the test exercises the same `DifferentialReport` the
    /// server emits.
    const CR03_FULL: &str = include_str!("fixtures/cr03.full.json");

    fn cr03_report() -> DifferentialReport {
        serde_json::from_str(CR03_FULL).expect("cr03.full.json must deserialize")
    }

    /// Render the WHOLE violation card into a string (ANSI retained —
    /// anstream strips it at print time). The card-construction seam
    /// (`build_violation_card`) is shared with the printing path.
    fn render_full_card(report: &DifferentialReport) -> String {
        build_violation_card(report, &CardKind::Violated).render()
    }

    /// Strip ANSI SGR escape sequences (`ESC [ … m`) so a test can measure
    /// visible columns on the styled card.
    fn strip_sgr(s: &str) -> String {
        let mut out = String::new();
        let mut chars = s.chars();
        while let Some(c) = chars.next() {
            if c == '\u{1b}' {
                for n in chars.by_ref() {
                    if n == 'm' {
                        break;
                    }
                }
            } else {
                out.push(c);
            }
        }
        out
    }

    /// Card-level width: every rendered line of the WHOLE violation card —
    /// borders and content alike — stays inside the 88-column frame.
    #[test]
    fn violation_card_fits_88_cols() {
        let report = cr03_report();
        let rendered = render_full_card(&report);
        for line in rendered.lines() {
            assert!(
                display_width(line) <= 88,
                "every card line must fit 88 cols; got {} → {line:?}",
                display_width(line)
            );
        }
        eprintln!("\n{rendered}");
    }

    /// IP boundary: the card shows ONLY turso frames. None of the harness /
    /// model / proxy internals leak into the user-facing card.
    #[test]
    fn violation_card_is_turso_only_no_ip_leak() {
        let report = cr03_report();
        let rendered = render_full_card(&report);
        for forbidden in [
            "io_bridge",
            "harness",
            "lean",
            "Lean",
            "aretta",
            "InjectedTurso",
            "/checkouts/",
        ] {
            assert!(
                !rendered.contains(forbidden),
                "card must not leak {forbidden:?}; got:\n{rendered}"
            );
        }
    }

    /// "what turso ran": the section header, both SQL labels, and the
    /// faulted sub-line are present.
    #[test]
    fn card_shows_what_turso_ran() {
        let report = cr03_report();
        let rendered = render_full_card(&report);
        assert!(
            rendered.contains("what turso ran"),
            "missing 'what turso ran' header; got:\n{rendered}"
        );
        assert!(rendered.contains("PRAGMA journal_mode=WAL"));
        assert!(
            rendered.contains("CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY, v INTEGER)")
        );
        // The faulting statement renders like the others — no per-cr_03
        // "faulted on commit" sub-line (removed 2026-06-03: the wording was
        // cr_03-specific, and the "divergence observed" block already carries
        // the failure, so no information is lost on the card).
        assert!(
            !rendered.contains("faulted on commit"),
            "the cr_03-specific 'faulted on commit' note must be gone; got:\n{rendered}"
        );
    }

    /// The reworked call path: shortened labels, the named entry frame in
    /// the header, the visible fork, and the data-driven fault marker. The
    /// effect frame renders as a plain spine frame (no prose marker).
    #[test]
    fn card_call_path_fork_and_markers() {
        let report = cr03_report();
        let rendered = render_full_card(&report);

        // Header names the entry frame; the entry is NOT also listed below.
        assert!(
            rendered.contains("where your code broke it"),
            "missing call-path header; got:\n{rendered}"
        );
        assert!(
            rendered.contains("inside Connection::execute"),
            "header must name the entry frame; got:\n{rendered}"
        );

        // Shortened labels (module path dropped).
        for label in [
            "prepare_wal_start",
            "begin_write_wal_header",
            "WalFile::prepare_wal_finish",
        ] {
            assert!(
                rendered.contains(label),
                "call path must show {label:?}; got:\n{rendered}"
            );
        }

        // The fault marker: op (lowercased) + error + the fixed tail.
        assert!(
            rendered.contains("→ EIO   fault injected here"),
            "fault marker missing/changed; got:\n{rendered}"
        );
        // The effect frame renders as a plain spine frame — no prose marker
        // (the cr_03-specific "writes the WAL header to disk" was removed
        // 2026-06-03; the "divergence observed" block carries the effect, and
        // the literal was wrong for any other CR).
        assert!(
            !rendered.contains("writes the WAL header to disk"),
            "the cr_03-specific effect marker must be gone; got:\n{rendered}"
        );
        assert!(
            !rendered.contains("sets initialized"),
            "the effect frame must NOT carry divergence values; got:\n{rendered}"
        );

        // The fork is visible: begin_write_wal_header (effect, depth 12)
        // renders deeper than prepare_wal_finish (fault, depth 11), with
        // the box-drawing connectors. (The label is dim-styled, so a style
        // reset sits between the connector and the label in the ANSI
        // string — assert the connector on the same LINE as its label.)
        let line_with = |needle: &str| -> &str {
            rendered
                .lines()
                .find(|l| l.contains(needle))
                .unwrap_or_else(|| panic!("no line for {needle:?}; got:\n{rendered}"))
        };
        // Match the call-path frames specifically (the `WalFile::` form);
        // the bare `prepare_wal_finish` also appears on the `impl` line.
        assert!(
            line_with("WalFile::prepare_wal_start").contains("├─ "),
            "first branch must use the ├─ connector; got:\n{rendered}"
        );
        assert!(
            line_with("WalFile::prepare_wal_finish").contains("└─ "),
            "fault branch must use the └─ connector; got:\n{rendered}"
        );
        // Depth via the visible column where the label needle starts (the
        // tree connectors push deeper frames right). Measured on the
        // ANSI-stripped line so style codes don't shift the column.
        let label_col = |needle: &str| -> usize {
            let line = rendered
                .lines()
                .find(|l| l.contains(needle))
                .unwrap_or_else(|| panic!("no line for {needle:?}"));
            let plain = strip_sgr(line);
            display_width(&plain[..plain.find(needle).unwrap()])
        };
        assert!(
            label_col("begin_write_wal_header") > label_col("WalFile::prepare_wal_finish"),
            "effect branch must be deeper (further right) than the fault branch:\n{rendered}"
        );
    }

    /// The divergence is a SEPARATE block AFTER the call path — not inline
    /// on a frame. `initialized = true` appears below the spine, under the
    /// "divergence observed" header.
    #[test]
    fn card_divergence_block_after_call_path() {
        let report = cr03_report();
        let rendered = render_full_card(&report);
        let lines: Vec<&str> = rendered.lines().collect();

        let pos = |needle: &str| {
            lines
                .iter()
                .position(|l| l.contains(needle))
                .unwrap_or_else(|| panic!("no line for {needle:?} in:\n{rendered}"))
        };

        assert!(
            rendered.contains("divergence observed"),
            "missing 'divergence observed' header; got:\n{rendered}"
        );
        // The actual value appears, but only inside the divergence block —
        // below the call path, never on a call-path frame.
        assert!(
            rendered.contains("initialized = true"),
            "divergence must show the diverging value; got:\n{rendered}"
        );
        // Use the call-path frame specifically (`WalFile::` form); the bare
        // `prepare_wal_finish` also appears on the early `impl` line.
        assert!(
            pos("divergence observed") > pos("WalFile::prepare_wal_finish"),
            "divergence block must come AFTER the call path; got:\n{rendered}"
        );
        assert!(
            pos("initialized = true") > pos("WalFile::prepare_wal_finish"),
            "the diverging value must render below the spine, not inline; got:\n{rendered}"
        );

        // The mask line + the aligned expected/actual rows.
        assert!(rendered.contains("compared [initialized]"));
        assert!(rendered.contains("9 fields ignored"));
        assert!(rendered.contains("initialized = false"));
    }

    /// The old op-trace rendering is GONE: neither the "what we tested"
    /// header nor the numbered op rows appear anywhere in the card.
    #[test]
    fn card_has_no_op_trace_rendering() {
        let report = cr03_report();
        // Sanity: the fixture STILL carries the op_trace data (contract
        // unchanged) — it is just no longer rendered.
        assert!(
            !report.scenario.op_trace.is_empty(),
            "fixture should still carry op_trace data"
        );
        let rendered = render_full_card(&report);
        assert!(
            !rendered.contains("what we tested"),
            "the old op-trace header must be gone; got:\n{rendered}"
        );
        assert!(
            !rendered.contains("← injected"),
            "the old op-trace injected marker must be gone; got:\n{rendered}"
        );
    }

    fn intent(
        id: &str,
        binding: BindingState,
        file: &str,
        site_with_line: &str,
    ) -> (AnnotationId, IntentEntry) {
        (
            AnnotationId::parse(id).unwrap(),
            IntentEntry {
                text: "the property".into(),
                verify: VerifyLevel::Method(VerifyMethod::Full),
                status: Status::Unknown,
                text_hash: zero_hash(),
                body_hash: zero_hash(),
                file: file.into(),
                site: site_with_line.into(),
                covered_region: CoveredRegion::Function,
                binding,
                parent: None,
                last_critiqued_at_text_hash: None,
                last_critique_finding_count: None,
            },
        )
    }

    fn matches_with(id: &AnnotationId, canon_id: &str, version: &str) -> CanonMatchesFile {
        let mut f = CanonMatchesFile {
            meta: CacheMeta::default(),
            ..CanonMatchesFile::default()
        };
        f.entries.insert(
            id.clone(),
            CacheEntry {
                last_match_text_hash: "x".into(),
                canon_fetched_at: "2026-05-24T00:00:00Z".into(),
                pending_matches: vec![],
                accepted_matches: vec![AcceptedMatch {
                    canon_id: canon_id.into(),
                    version: version.into(),
                    canonical_text: "the property".into(),
                    canon_version: "v0.2.0".into(),
                    confidence: 0.95,
                    prefix_tier: PrefixTier::Aristos,
                    backed_by: Some("test backing".into()),
                    linked: None,
                    accepted_at: "2026-05-24T00:00:00Z".into(),
                    bound_at: "2026-05-24T00:00:00Z".into(),
                }],
                rejected_matches: vec![],
            },
        );
        f
    }

    // ─── partition_full ───────────────────────────────────────────────────

    #[test]
    fn partition_separates_canon_bound_from_local() {
        let (aristos_id, aristos_entry) = intent(
            "aristos:foo",
            BindingState::Bound {
                linked: arta("arta_op4q3z9NbV"),
            },
            "src/x.rs",
            "fn x (line 1)",
        );
        let (kanon_id, kanon_entry) = intent(
            "kanon:bar",
            BindingState::Bound {
                linked: arta("arta_xyz1234567"),
            },
            "src/y.rs",
            "fn y (line 10)",
        );
        let (local_id, local_entry) =
            intent("baz", BindingState::Local, "src/z.rs", "fn z (line 5)");

        let mut index = empty_index();
        index
            .entries
            .insert(aristos_id.clone(), IndexEntry::Intent(aristos_entry));
        index
            .entries
            .insert(kanon_id.clone(), IndexEntry::Intent(kanon_entry));
        index
            .entries
            .insert(local_id.clone(), IndexEntry::Intent(local_entry));

        let pending = vec![&aristos_id, &kanon_id, &local_id];
        let (canon, other) = partition_full(&index, &pending);

        assert_eq!(canon.len(), 2);
        assert_eq!(other.len(), 1);
        assert_eq!(other[0].as_str(), "baz");
        let canon_ids: Vec<_> = canon.iter().map(|c| c.id.as_str()).collect();
        assert!(canon_ids.contains(&"aristos:foo"));
        assert!(canon_ids.contains(&"kanon:bar"));
    }

    #[test]
    fn partition_returns_empty_for_no_input() {
        let index = empty_index();
        let (canon, other) = partition_full(&index, &[]);
        assert!(canon.is_empty());
        assert!(other.is_empty());
    }

    // ─── build_tags ───────────────────────────────────────────────────────

    #[test]
    fn build_tags_emits_arta_id_canon_id_version_and_source_path() {
        let (id, entry) = intent(
            "aristos:vacuum_preserves_logical_content",
            BindingState::Bound {
                linked: arta("arta_op4q3z9NbV"),
            },
            "crates/vacuum/src/lib.rs",
            "fn vacuum (line 42)",
        );
        let matches = matches_with(&id, "vacuum_preserves_logical_content", "v0.1.0");
        let dispatch = vec![CanonDispatchEntry {
            id: &id,
            entry: &entry,
        }];

        let tags = build_tags(&dispatch, &matches);
        assert_eq!(tags.len(), 1);
        let t = &tags[0];
        assert_eq!(t.annotation_id, "arta_op4q3z9NbV");
        assert_eq!(t.canon_id, "vacuum_preserves_logical_content");
        assert_eq!(t.version, "v0.1.0");
        assert_eq!(t.source_path, "crates/vacuum/src/lib.rs:42");
    }

    #[test]
    fn build_tags_works_for_certified_binding_state() {
        // Certified binding still carries `linked` — the SDK should
        // re-dispatch on --rerun (caller's responsibility; we just
        // need to handle the variant).
        use aristo_core::index::{CommitHash, VerifiedOutcome};
        let (id, entry) = intent(
            "kanon:foo",
            BindingState::Certified {
                linked: arta("arta_aaaa1234"),
                verified_outcome: VerifiedOutcome::parse(&format!("v1:{}", "A".repeat(86)))
                    .unwrap(),
                last_verified_at_commit: CommitHash::parse(&"a".repeat(40)).unwrap(),
            },
            "src/foo.rs",
            "fn foo (line 7)",
        );
        let matches = matches_with(&id, "foo", "v0.2.0");
        let dispatch = vec![CanonDispatchEntry {
            id: &id,
            entry: &entry,
        }];
        let tags = build_tags(&dispatch, &matches);
        assert_eq!(tags.len(), 1);
        assert_eq!(tags[0].annotation_id, "arta_aaaa1234");
        assert_eq!(tags[0].canon_id, "foo");
        assert_eq!(tags[0].version, "v0.2.0");
    }

    #[test]
    fn build_tags_drops_entries_missing_from_cache() {
        // Canon-bound entry but no cache row — the cache is the only
        // source of truth for `version`. Drop silently rather than
        // POST an empty version that'd 4xx server-side.
        let (id, entry) = intent(
            "aristos:foo",
            BindingState::Bound {
                linked: arta("arta_aaaa1234"),
            },
            "src/foo.rs",
            "fn foo (line 7)",
        );
        let matches = CanonMatchesFile::default();
        let dispatch = vec![CanonDispatchEntry {
            id: &id,
            entry: &entry,
        }];
        let tags = build_tags(&dispatch, &matches);
        assert!(tags.is_empty(), "missing cache entry must drop the tag");
    }

    #[test]
    fn build_tags_drops_local_binding() {
        // Defensive: canon-prefixed id but `BindingState::Local` —
        // shouldn't happen in practice but the partition is by id-
        // prefix only; the build must handle missing linked gracefully.
        let (id, entry) = intent(
            "aristos:foo",
            BindingState::Local,
            "src/foo.rs",
            "fn foo (line 7)",
        );
        let matches = matches_with(&id, "foo", "v0.1.0");
        let dispatch = vec![CanonDispatchEntry {
            id: &id,
            entry: &entry,
        }];
        let tags = build_tags(&dispatch, &matches);
        assert!(tags.is_empty());
    }

    #[test]
    fn build_tags_falls_back_to_file_only_when_site_lacks_line() {
        let (id, entry) = intent(
            "aristos:foo",
            BindingState::Bound {
                linked: arta("arta_aaaa1234"),
            },
            "src/foo.rs",
            "fn foo", // no "(line N)" suffix
        );
        let matches = matches_with(&id, "foo", "v0.1.0");
        let dispatch = vec![CanonDispatchEntry {
            id: &id,
            entry: &entry,
        }];
        let tags = build_tags(&dispatch, &matches);
        assert_eq!(tags.len(), 1);
        assert_eq!(tags[0].source_path, "src/foo.rs");
    }

    // ─── strip_canon_prefix ──────────────────────────────────────────────

    #[test]
    fn strip_canon_prefix_handles_both_tiers_and_no_prefix() {
        assert_eq!(strip_canon_prefix("aristos:foo_bar"), "foo_bar");
        assert_eq!(
            strip_canon_prefix("kanon:checkout_total_non_negative"),
            "checkout_total_non_negative"
        );
        // No prefix — pass through (defensive).
        assert_eq!(strip_canon_prefix("local_id"), "local_id");
    }

    // ─── extract_line ────────────────────────────────────────────────────

    #[test]
    fn extract_line_parses_walker_emitted_site_format() {
        assert_eq!(extract_line("fn foo (line 42)"), Some(42));
        assert_eq!(extract_line("fn really::long::path (line 1)"), Some(1));
        assert_eq!(extract_line("fn x"), None);
        assert_eq!(extract_line(""), None);
    }

    // ─── dispatch_session (with mock client) ─────────────────────────────

    #[test]
    fn dispatch_session_round_trips_through_mock() {
        let mock = aristo_core::canon_verify::MockVerifyClient::with_post_response(
            PostVerifySessionResponse {
                session_id: "01HMTEST".into(),
                view_url: "https://dev.aretta.ai/dashboard/jobs/01HMTEST".into(),
                plan_size: 1,
            },
        );
        let req = VerifySessionRequest {
            repo_full_name: "owner/repo".into(),
            commit_sha: "deadbeef".into(),
            tags: vec![VerifySessionTag {
                annotation_id: "arta_x".into(),
                canon_id: "foo".into(),
                version: "v0.1.0".into(),
                source_path: "src/x.rs:1".into(),
            }],
        };
        let resp = dispatch_session(&mock, &req).expect("dispatch ok");
        assert_eq!(resp.session_id, "01HMTEST");
        let posted = mock.posted_requests();
        assert_eq!(posted.len(), 1);
        assert_eq!(posted[0].tags[0].annotation_id, "arta_x");
    }

    #[test]
    fn dispatch_session_propagates_server_error() {
        let mock =
            aristo_core::canon_verify::MockVerifyClient::with_post_error(VerifyError::BadRequest {
                status: 402,
                message: "no_canon_coverage".into(),
            });
        let req = VerifySessionRequest {
            repo_full_name: "o/r".into(),
            commit_sha: "x".into(),
            tags: vec![],
        };
        let err = dispatch_session(&mock, &req).unwrap_err();
        match err {
            VerifyError::BadRequest { status: 402, .. } => {}
            other => panic!("expected BadRequest 402, got {other:?}"),
        }
    }
}