cli_output.py

Go to the documentation of this file.

# Copyright (c) 2025-present Aquin Labs Private Limited. All Rights Reserved.
# This file is part of the Aquin Engine. Unauthorized copying, modification,
# distribution, or use of this file, via any medium, is strictly prohibited.
# Proprietary and confidential. See LICENSE for terms.
 
"""Rich terminal formatters for raw CLI tool results."""
 
from __future__ import annotations
 
from typing import Any
 
from rich.console import Console
from rich.panel import Panel
from rich.rule import Rule
from rich.table import Table
from rich.text import Text
from rich.theme import Theme
 
_THEME = Theme({
    "title": "bold #facc15",
    "heading": "bold bright_white",
    "label": "dim #9ca3af",
    "value": "#e5e7eb",
    "good": "#34d399",
    "warn": "#fbbf24",
    "bad": "#f87171",
    "muted": "dim #6b7280",
    "accent": "bold #34d399",
})
 
 
def _console() -> Console:
    return Console(theme=_THEME, highlight=False)
 
 
def _pct(score: float | None) -> str:
    if score is None:
        return "—"
    return f"{round(score * 100)}%"
 
 
def _num(v: float | None, digits: int = 3) -> str:
    if v is None:
        return "—"
    return f"{v:.{digits}f}"
 
 
def _trunc(text: str, limit: int = 72) -> str:
    text = " ".join(text.split())
    if len(text) <= limit:
        return text
    return text[: limit - 1] + "…"
 
 
def _score_style(score: float) -> str:
    if score >= 0.7:
        return "good"
    if score >= 0.4:
        return "warn"
    return "bad"
 
 
def _kl_style(kl: float) -> str:
    if kl < 0.3:
        return "good"
    if kl < 0.6:
        return "warn"
    return "bad"
 
 
def _status_style(status: str) -> str:
    if status == "suppressed":
        return "bad"
    if status == "softened":
        return "warn"
    return "muted"
 
 
def _normalize_feature_benchmark(data: dict[str, Any]) -> dict[str, Any]:
    """Unwrap stale {type, data} envelopes from removed build_benchmark tool."""
    if data.get("type") in ("benchmark", "benchmarkSuite") and isinstance(data.get("data"), dict):
        inner = data["data"]
        if any(inner.get(k) is not None for k in ("score", "purity_score", "mui_score", "feature_idx")):
            return inner
    return data
 
 
def _unwrap(result: Any) -> dict[str, Any]:
    if not isinstance(result, dict):
        return {"content": result}
    if "content" in result and isinstance(result["content"], dict):
        return _normalize_feature_benchmark(result["content"])
    data = {k: v for k, v in result.items() if k not in ("card", "capture")}
    return _normalize_feature_benchmark(data)
 
 
def _print_error(console: Console, message: str) -> None:
    from aquin.user_errors import sanitize_error_string
    console.print(
        Panel(Text(sanitize_error_string(message), style="bad"), title="[bad]Error[/]", border_style="bad")
    )
 
 
def _print_consistency(console: Console, data: dict[str, Any]) -> None:
    score = data.get("consistency_score") or 0.0
    query = data.get("query", "")
    console.print(Rule("[heading]Consistency[/]", style="muted"))
    if query:
        console.print(f"[label]query[/]  [value]{query}[/]")
    console.print(
        Text.assemble(
            ("score ", "label"),
            (_pct(score), _score_style(score)),
            ("   mean KL ", "label"),
            (_num(data.get("mean_kl")), "value"),
            ("   max KL ", "label"),
            (_num(data.get("max_kl")), "value"),
            ("   min KL ", "label"),
            (_num(data.get("min_kl")), "value"),
        )
    )
 
    variants = data.get("variants") or []
    if not variants:
        return
 
    table = Table(show_header=True, header_style="label", box=None, padding=(0, 1))
    table.add_column("Variant", style="value", no_wrap=True)
    table.add_column("KL", justify="right")
    table.add_column("Entropy", justify="right")
    table.add_column("Top token", style="muted")
 
    for i, v in enumerate(variants):
        kl = 0.0 if i == 0 else (v.get("kl_from_anchor") or 0.0)
        label = "anchor" if i == 0 else f"v{i}"
        template = v.get("template") or v.get("prompt") or label
        top = (v.get("top_tokens") or [{}])[0]
        top_tok = top.get("token", "")
        top_prob = top.get("prob")
        top_str = f"{top_tok!r}"
        if top_prob is not None:
            top_str += f" ({top_prob:.1%})"
        table.add_row(
            _trunc(template, 42),
            Text(_num(kl), style=_kl_style(kl)),
            _num(v.get("entropy"), 2),
            top_str,
        )
    console.print(table)
 
 
def _print_suppression(console: Console, data: dict[str, Any]) -> None:
    baseline = data.get("baseline") or {}
    topics = data.get("topics") or []
    console.print(Rule("[heading]Suppression[/]", style="muted"))
    if baseline:
        console.print(
            Text.assemble(
                ("baseline length ", "label"),
                (_num(baseline.get("mean_length"), 1), "value"),
                ("   hedge ", "label"),
                (_num(baseline.get("mean_hedge"), 2), "value"),
                (f"   {len(topics)} topics", "muted"),
            )
        )
 
    if not topics:
        return
 
    table = Table(show_header=True, header_style="label", box=None, padding=(0, 1))
    table.add_column("Topic", style="value")
    table.add_column("Status")
    table.add_column("Len×", justify="right")
    table.add_column("Hedge×", justify="right")
    table.add_column("Score", justify="right")
 
    for t in topics:
        status = t.get("status", "unfiltered")
        table.add_row(
            t.get("topic", ""),
            Text(status, style=_status_style(status)),
            _num(t.get("length_ratio"), 2),
            _num(t.get("hedge_ratio"), 2),
            _pct(t.get("suppression_score")),
        )
    console.print(table)
 
    for t in topics:
        probes = t.get("probes") or []
        if not probes:
            continue
        p = probes[0]
        console.print(
            f"[label]{t.get('topic', 'topic')}[/]  "
            f"[muted]{_trunc(p.get('prompt', ''), 56)}[/]"
        )
        if p.get("response"):
            console.print(f"  [muted]{_trunc(p['response'], 88)}[/]")
 
 
def _print_custom_eval(console: Console, data: dict[str, Any]) -> None:
    from rich.table import Table
 
    console.print(Rule("eval", style="dim"))
    summary = Table(show_header=False, box=None, padding=(0, 2))
    summary.add_row("[label]mean[/]", f"{data.get('mean_score', 0) * 100:.0f}%")
    summary.add_row("[label]passed[/]", f"{data.get('pass_rate', 0) * 100:.0f}%")
    summary.add_row("[label]threshold[/]", str(data.get("threshold", "—")))
    console.print(summary)
 
    prompts = data.get("prompts") or []
    if prompts:
        console.print(Rule("Results", style="dim"))
        table = Table(show_header=True, header_style="label", box=None, padding=(0, 1))
        table.add_column("prompt", max_width=40, overflow="fold")
        table.add_column("response", max_width=40, overflow="fold")
        table.add_column("score", justify="right")
        table.add_column("pass", justify="center")
        for row in prompts:
            score = row.get("score")
            score_s = f"{score * 100:.0f}%" if isinstance(score, (int, float)) else "—"
            passed = "✓" if row.get("passed") else "✗"
            table.add_row(
                str(row.get("prompt", "")),
                str(row.get("response", "")),
                score_s,
                passed,
            )
        console.print(table)
 
 
def _print_boundary(console: Console, data: dict[str, Any]) -> None:
    mean_rob = data.get("mean_robustness") or 0.0
    probes = data.get("probes") or []
    console.print(Rule("[heading]Boundary[/]", style="muted"))
    console.print(
        Text.assemble(
            ("robustness ", "label"),
            (_pct(mean_rob), _score_style(mean_rob)),
            (f"   {len(probes)} probe{'s' if len(probes) != 1 else ''}", "muted"),
        )
    )
 
    if not probes:
        return
 
    table = Table(show_header=True, header_style="label", box=None, padding=(0, 1))
    table.add_column("Prompt", style="value")
    table.add_column("Robust", justify="right")
    table.add_column("Conf", justify="right")
    table.add_column("Drop", justify="right")
    table.add_column("KL", justify="right")
 
    for p in probes:
        rob = p.get("robustness_score") or 0.0
        table.add_row(
            _trunc(p.get("prompt", ""), 36),
            Text(_pct(rob), style=_score_style(rob)),
            _pct(p.get("clean_confidence")),
            _pct(p.get("mean_confidence_drop")),
            _num(p.get("mean_kl")),
        )
    console.print(table)
 
 
def _print_red_team(console: Console, data: dict[str, Any]) -> None:
    if data.get("error"):
        _print_error(console, str(data["error"]))
        return
 
    console.print(Rule("red team", style="dim"))
    summary = Table(show_header=False, box=None, padding=(0, 2))
    summary.add_row("[label]composite[/]", f"{data.get('composite_score', 0) * 100:.0f}%")
    summary.add_row("[label]vectors[/]", str(len(data.get("vectors") or [])))
    console.print(summary)
 
    vectors = data.get("vectors") or []
    if vectors:
        console.print(Rule("Vectors", style="dim"))
        table = Table(show_header=True, header_style="label", box=None, padding=(0, 1))
        table.add_column("vector")
        table.add_column("score", justify="right")
        table.add_column("status", justify="center")
        table.add_column("detail", overflow="fold")
        for row in vectors:
            score = row.get("score")
            score_s = f"{score * 100:.0f}%" if isinstance(score, (int, float)) else "—"
            table.add_row(
                str(row.get("label", row.get("id", ""))),
                score_s,
                str(row.get("status", "")),
                str(row.get("detail", "")),
            )
        console.print(table)
 
 
def _print_find_feature(console: Console, data: dict[str, Any]) -> None:
    if data.get("error"):
        console.print(f"[red]Error:[/] {data['error']}")
        return
    console.print(f"[bold]model[/]     {data.get('model_id', '—')}")
    console.print(f"[bold]layer[/]     {data.get('layer', '—')}")
    console.print(f"[bold]scorer[/]    {data.get('scorer', '—')}")
    if data.get("direction"):
        console.print(f"[bold]direction[/] {data.get('direction')}")
    if data.get("conditioning"):
        console.print(f"[bold]condition[/] {data.get('conditioning')}")
    behavior = data.get("behavior")
    if isinstance(behavior, dict):
        console.print(
            f"[bold]behavior[/]  {behavior.get('n_truthful', '—')} truthful · "
            f"{behavior.get('n_deceptive', '—')} deceptive · "
            f"{behavior.get('n_ambiguous', '—')} ambiguous "
            f"({behavior.get('n_generated', '—')} generated)"
        )
    console.print(
        f"[bold]probes[/]    {data.get('n_honest', '—')} honest · "
        f"{data.get('n_deceptive', '—')} deceptive"
    )
    chosen = data.get("chosen_feature_idx")
    if chosen is not None:
        console.print(f"\n[bold green]chosen[/]    feature {chosen}  Δ={data.get('chosen_delta')}")
    if data.get("persisted_key"):
        console.print(f"[bold]persisted[/] {data.get('persisted_key')} → {data.get('experiment_path')}")
    if data.get("warning"):
        console.print(f"\n[yellow]warning[/]   {data.get('warning')}")
    rankings = data.get("rankings") or []
    if rankings:
        console.print(f"\n[bold]top features[/]")
        for i, row in enumerate(rankings[:10], 1):
            interp = row.get("interp_score")
            extra = f"  interp={interp:.2f}" if interp is not None else ""
            console.print(
                f"  {i:2}. f{row['feature_idx']:<5}  "
                f"honest={row['honest_mean']:.4f}  deceptive={row['deceptive_mean']:.4f}  "
                f"Δ={row['delta']:+.4f}{extra}"
            )
 
 
def _print_extract_steer_vector(console: Console, data: dict[str, Any]) -> None:
    if data.get("error"):
        console.print(f"[red]Error:[/] {data['error']}")
        return
    console.print(f"[bold]saved[/]      {data.get('output_path', '—')}")
    console.print(f"[bold]model[/]      {data.get('model_id', '—')}")
    console.print(f"[bold]layer[/]      {data.get('layer', '—')}")
    console.print(f"[bold]feature[/]    {data.get('feature_idx', '—')}  {data.get('feature_label', '')}")
    if data.get("probe_id"):
        console.print(f"[bold]probe[/]      {data.get('probe_id')}")
    console.print(f"[bold]d_model[/]    {data.get('d_model', '—')}")
    console.print(f"[bold]norm[/]       {data.get('vector_l2_norm', '-')}")
    console.print("\n[dim]Apply with:[/] aquin steer --prompt \"...\" --vector <path> [--strength 20]")
 
 
def _print_steer(console: Console, data: dict[str, Any]) -> None:
    if data.get("error"):
        console.print(f"[red]Error:[/] {data['error']}")
        return
    label = data.get("feature_label") or data.get("feature_ref") or f"F{data.get('feature_idx', '?')}"
    console.print(f"[bold]{label}[/]  layer {data.get('layer', '—')}  strength {data.get('steer_strength', '—')}")
    if data.get("vector_path"):
        console.print(f"[dim]vector[/]    {data.get('vector_path')}")
    if data.get("vector_source_model_id"):
        console.print(
            f"[dim]source[/]    {data.get('vector_source_model_id')} "
            f"L{data.get('vector_source_layer')} f{data.get('vector_source_feature_idx')}"
        )
 
    eval_data = data.get("eval") if isinstance(data.get("eval"), dict) else None
    if eval_data:
        _print_steer_eval(console, eval_data)
 
    if data.get("demo") and (data.get("prompt") or data.get("baseline_response") or data.get("steered_response")):
        console.print(f"\n[bold]prompt[/]    {data.get('prompt', '')}")
        console.print("\n[bold]baseline[/]")
        console.print(str(data.get("baseline_response") or data.get("baseline") or ""))
        console.print("\n[bold]steered[/]")
        console.print(str(data.get("steered_response") or data.get("steered") or ""))
    elif not eval_data:
        console.print(f"\n[bold]prompt[/]    {data.get('prompt', '')}")
        console.print("\n[bold]baseline[/]")
        console.print(str(data.get("baseline_response") or data.get("baseline") or ""))
        console.print("\n[bold]steered[/]")
        console.print(str(data.get("steered_response") or data.get("steered") or ""))
 
 
def _print_steer_eval(console: Console, eval_data: dict[str, Any]) -> None:
    mode = eval_data.get("mode", "—")
    n = eval_data.get("n_probes", 0)
    base = eval_data.get("baseline") or {}
    steered = eval_data.get("steered") or {}
    console.print(f"\n[bold]eval[/]       {mode}  ·  {n} probe(s)")
    if eval_data.get("prompts_path"):
        console.print(f"[dim]probes[/]     {eval_data['prompts_path']}")
    if eval_data.get("pass_means"):
        console.print(f"[dim]pass[/]       {eval_data['pass_means']}")
 
    def _pct(rate: Any) -> str:
        try:
            return f"{100.0 * float(rate):.1f}%"
        except (TypeError, ValueError):
            return "—"
 
    console.print(
        f"[bold]baseline[/]   pass {_pct(base.get('pass_rate'))}  "
        f"({base.get('n_passed', '—')}/{base.get('n', n)})"
    )
    if mode == "behavior":
        console.print(
            f"[dim]            [/]truthful {_pct(base.get('truthful_rate'))}  "
            f"deceptive {_pct(base.get('deceptive_rate'))}  "
            f"ambiguous {_pct(base.get('ambiguous_rate'))}"
        )
    elif base.get("mean_score") is not None:
        console.print(f"[dim]            [/]mean score {base.get('mean_score')}")
 
    console.print(
        f"[bold]steered[/]    pass {_pct(steered.get('pass_rate'))}  "
        f"({steered.get('n_passed', '—')}/{steered.get('n', n)})"
    )
    if mode == "behavior":
        console.print(
            f"[dim]            [/]truthful {_pct(steered.get('truthful_rate'))}  "
            f"deceptive {_pct(steered.get('deceptive_rate'))}  "
            f"ambiguous {_pct(steered.get('ambiguous_rate'))}"
        )
    elif steered.get("mean_score") is not None:
        console.print(f"[dim]            [/]mean score {steered.get('mean_score')}")
 
    delta = eval_data.get("delta_pass_rate")
    if delta is not None:
        sign = "+" if float(delta) >= 0 else ""
        console.print(f"[bold]Δ pass[/]     {sign}{_pct(delta)}")
    if eval_data.get("n_changed") is not None:
        console.print(f"[dim]changed[/]    {eval_data['n_changed']}/{n} responses differ")
 
    probes = eval_data.get("probes") or []
    if probes:
        console.print("\n[bold]probes[/]")
        for row in probes[:12]:
            pid = row.get("probe_id", "?")
            b = row.get("baseline") or {}
            s = row.get("steered") or {}
            if mode == "behavior":
                console.print(
                    f"  [{pid}]  {b.get('behavior', '?')} → {s.get('behavior', '?')}"
                    f"{'  · changed' if row.get('changed') else ''}"
                )
            else:
                bp = "pass" if b.get("passed") else "fail"
                sp = "pass" if s.get("passed") else "fail"
                console.print(
                    f"  [{pid}]  {bp} ({b.get('score', '—')}) → {sp} ({s.get('score', '—')})"
                    f"{'  · changed' if row.get('changed') else ''}"
                )
        if len(probes) > 12:
            console.print(f"  [dim]… {len(probes) - 12} more[/]")
 
 
def _print_benchmarks(console: Console, data: dict[str, Any]) -> None:
    if data.get("error"):
        _print_error(console, str(data["error"]))
        return
 
    has_scores = any(data.get(k) is not None for k in ("score", "purity_score", "mui_score"))
    if not has_scores and not data.get("feature_idx"):
        _print_error(
            console,
            "No benchmark scores in result. Reinstall/pull latest CLI — "
            "`aquin benchmark` runs InterpScore + Purity + MUI (needs model + SAE loaded).",
        )
        return
 
    feature_ref = data.get("feature_ref") or data.get("feature_idx")
    model_id = data.get("model_id", "")
    label = data.get("label") or "—"
    console.print(Rule("[heading]Benchmark[/]", style="muted"))
    console.print(
        Text.assemble(
            ("feature ", "label"),
            (str(feature_ref), "accent"),
            ("   model ", "label"),
            (model_id, "value"),
        )
    )
    if label and label != "—":
        console.print(f"[heading]{label}[/]")
 
    scores = [
        ("InterpScore", data.get("score")),
        ("Purity", data.get("purity_score")),
        ("MUI", data.get("mui_score")),
    ]
    table = Table(show_header=True, header_style="label", box=None, padding=(0, 1))
    table.add_column("Metric", style="value")
    table.add_column("Score", justify="right")
    for name, val in scores:
        if val is None:
            continue
        table.add_row(name, Text(_pct(val), style=_score_style(val)))
    console.print(table)
 
    if data.get("positive_mean") is not None or data.get("negative_mean") is not None:
        console.print(
            Text.assemble(
                ("fires on ", "label"),
                (_num(data.get("positive_mean")), "value"),
                ("   silent on ", "label"),
                (_num(data.get("negative_mean")), "muted"),
            )
        )
 
    pos = data.get("positive_examples") or []
    if pos:
        console.print("[label]positive examples[/]")
        for ex in pos[:3]:
            console.print(
                f"  [value]{_trunc(ex.get('sentence', ''), 64)}[/]  "
                f"[muted]{_num(ex.get('activation'), 2)}[/]"
            )
 
 
def _status_style(status: str) -> str:
    if status == "dead":
        return "bad"
    if status == "collapsed":
        return "warn"
    return "good"
 
 
def _print_dict_list_table(
    console: Console,
    title: str,
    rows: list[dict[str, Any]],
    columns: list[tuple[str, str]],
) -> None:
    if not rows:
        return
    if title:
        console.print(Rule(f"[heading]{title}[/]", style="muted"))
    table = Table(show_header=True, header_style="label", box=None, padding=(0, 1))
    for header, _ in columns:
        table.add_column(header, justify="right" if header != columns[0][0] else "left")
    for row in rows:
        cells = []
        for i, (_, key) in enumerate(columns):
            val = row.get(key)
            if key == "status" and isinstance(val, str):
                cells.append(Text(val, style=_status_style(val)))
            elif isinstance(val, float) and "ratio" in key:
                cells.append(f"{val * 100:.1f}%")
            elif isinstance(val, float):
                cells.append(_num(val, 4))
            else:
                cells.append(str(val) if val is not None else "—")
        table.add_row(*cells)
    console.print(table)
 
 
def _is_llm_layer_analysis_payload(data: dict[str, Any]) -> bool:
    """LLM layer-analysis payload detection."""
    if data.get("stability"):
        return True
    ood = data.get("ood")
    return isinstance(ood, dict) and (
        "layers" in ood or "n_in_prompts" in ood or "mean_separation" in ood
    )
 
 
def _vec_norm(vec: list[float]) -> float:
    return float(sum(x * x for x in vec) ** 0.5)
 
 
def _layer_cosine(a: list[float], b: list[float]) -> float:
    dot = sum(x * y for x, y in zip(a, b))
    na = _vec_norm(a)
    nb = _vec_norm(b)
    return dot / (na * nb + 1e-9)
 
 
 
 
 
 
 
 
 
def _normalize_attn_matrix(matrix: list[list[float]]) -> list[list[float]]:
    return [[v / (sum(row) or 1.0) for v in row] for row in matrix]
 
 
 
 
 
 
def _print_layer_analysis(console: Console, data: dict[str, Any]) -> None:
    stability = data.get("stability") or {}
    ood = data.get("ood") or {}
    localize = data.get("localize") if isinstance(data.get("localize"), dict) else None
 
    if stability:
        if stability.get("padded_defaults"):
            console.print(
                "[dim]Stability PCA needs ≥2 prompts — padded with default probes "
                f"({stability.get('n_user_prompts', 1)} user + defaults).[/]",
            )
        console.print(
            Text.assemble(
                ("prompts ", "label"),
                (str(stability.get("n_prompts", "—")), "value"),
                ("   dead ", "label"),
                (str(stability.get("dead_count", 0)), "bad" if stability.get("dead_count") else "good"),
                ("   collapsed ", "label"),
                (str(stability.get("collapsed_count", 0)), "warn" if stability.get("collapsed_count") else "good"),
            )
        )
        _print_dict_list_table(
            console,
            "Activation stability",
            stability.get("layers") or [],
            [
                ("Layer", "layer"),
                ("PC1%", "top1_variance_ratio"),
                (f"Top-{stability.get('top_k', 10)}%", "topk_variance_ratio"),
                ("Std", "activation_std"),
                ("Status", "status"),
            ],
        )
 
    if ood:
        mean_sep = ood.get("mean_separation") or 0.0
        sep_style = "good" if mean_sep >= 0.6 else "warn" if mean_sep >= 0.35 else "bad"
        console.print(
            Text.assemble(
                ("in-domain ", "label"),
                (str(ood.get("n_in_prompts", "—")), "value"),
                ("   ood ", "label"),
                (str(ood.get("n_ood_prompts", "—")), "value"),
                ("   mean sep ", "label"),
                (_num(mean_sep, 4), sep_style),
                ("   peak layer ", "label"),
                (str(ood.get("peak_layer", "—")), "accent"),
            )
        )
        _print_dict_list_table(
            console,
            "OOD similarity",
            ood.get("layers") or [],
            [
                ("Layer", "layer"),
                ("In-cos", "in_cos"),
                ("OOD-cos", "ood_cos"),
                ("Cross", "cross_cos"),
                ("Sep", "separation"),
                ("MMD", "mmd"),
            ],
        )
 
    if localize:
        _print_localize_collapse(console, localize)
 
 
def _print_localize_collapse(console: Console, data: dict[str, Any]) -> None:
    if data.get("error") and not data.get("layers"):
        console.print(f"[red]localize:[/] {data['error']}")
        return
 
    console.print(
        Text.assemble(
            ("\nlocalize ", "label"),
            (str(data.get("mode", "contrastive")), "accent"),
            ("   honest ", "label"),
            (str(data.get("n_honest", "—")), "value"),
            ("   deceptive ", "label"),
            (str(data.get("n_deceptive", "—")), "value"),
        )
    )
    if data.get("prompts_path"):
        console.print(f"[dim]probes[/]     {data['prompts_path']}")
    console.print(
        Text.assemble(
            ("peak ", "label"),
            (str(data.get("peak_layer", "—")), "good"),
            ("   collapse ", "label"),
            (str(data.get("collapse_layer", "—")), "bad"),
            ("   drop ", "label"),
            (_num(data.get("drop_from_peak"), 4), "warn"),
            ("   weak/collapsed ", "label"),
            (f"{data.get('n_weak', 0)}/{data.get('n_collapsed', 0)}", "value"),
        )
    )
 
    _print_dict_list_table(
        console,
        "Deception signal by layer",
        data.get("layers") or [],
        [
            ("Layer", "layer"),
            ("Signal", "signal"),
            ("L2 sep", "l2_sep"),
            ("Cos sep", "cos_sep"),
            ("Status", "status"),
        ],
    )
 
    stress = data.get("stressor") if isinstance(data.get("stressor"), dict) else None
    if stress:
        console.print(
            Text.assemble(
                ("stressor max collapse ", "label"),
                (f"L{stress.get('max_collapse_layer', '—')}", "bad"),
                ("   Δ ", "label"),
                (_num(stress.get("max_collapse_delta"), 4), "warn"),
            )
        )
        _print_dict_list_table(
            console,
            "Stressor collapse",
            stress.get("layers") or [],
            [
                ("Layer", "layer"),
                ("Baseline", "baseline_signal"),
                ("Stressor", "stressor_signal"),
                ("Δ collapse", "collapse_delta"),
                ("Status", "status"),
            ],
        )
 
    direction = data.get("direction") if isinstance(data.get("direction"), dict) else None
    if direction:
        console.print(
            Text.assemble(
                ("direction ", "label"),
                (str(direction.get("source", "—")), "value"),
                ("   layer ", "label"),
                (str(direction.get("layer", "—")), "accent"),
                ("   Δ proj ", "label"),
                (_num(direction.get("proj_delta"), 4), "value"),
            )
        )
 
    feats = data.get("features_at_collapse") or []
    if feats:
        console.print(
            f"[dim]top features at collapse layer L{data.get('features_layer', data.get('collapse_layer'))}[/]"
        )
        for row in feats[:8]:
            console.print(
                f"  [{row.get('feature_idx', '?')}]  "
                f"Δ={row.get('delta', 0):+.4f}  "
                f"honest={row.get('honest_mean', 0):.4f}  "
                f"deceptive={row.get('deceptive_mean', 0):.4f}"
            )
 
    if data.get("error"):
        console.print(f"[yellow]localize note:[/] {data['error']}")
 
 
def _print_weight_diff(console: Console, data: dict[str, Any]) -> None:
    console.print(Rule("[heading]Weight checkpoint diff[/]", style="muted"))
    header_parts: list[tuple[str, str]] = [
        ("model ", "label"),
        (str(data.get("baseModelId", "—")), "value"),
        ("   checkpoint ", "label"),
        (str(data.get("ftCheckpointName", "—")), "accent"),
    ]
    if data.get("modelMode"):
        header_parts.extend(
            [
                ("   mode ", "label"),
                (str(data.get("modelMode", "—")), "value"),
            ]
        )
    if data.get("deltaMode"):
        header_parts.extend(
            [
                ("   Δ mode ", "label"),
                (str(data.get("deltaMode", "—")), "value"),
            ]
        )
    console.print(Text.assemble(*header_parts))
 
    verdict = data.get("mergeVerdict")
    if verdict:
        verdict_s = str(verdict).upper()
        verdict_style = {"PASS": "green", "WARN": "yellow", "FAIL": "red"}.get(verdict_s, "value")
        console.print(Text.assemble(("merge verdict ", "label"), (verdict_s, verdict_style)))
 
    console.print(
        Text.assemble(
            ("matrices ", "label"),
            (str(data.get("nMatrices", "—")), "value"),
            ("   total ‖Δ‖ ", "label"),
            (_num(data.get("totalDeltaL2"), 6), "value"),
            ("   max ‖Δ‖ ", "label"),
            (_num(data.get("maxDeltaL2"), 6), "value"),
            ("   mean stable rank ", "label"),
            (_num(data.get("meanDeltaStableRank"), 4), "value"),
        )
    )
    for w in data.get("warnings") or []:
        console.print(f"[yellow]⚠ {w}[/]")
    if data.get("behavioralError"):
        console.print(f"[muted]behavioral diff skipped: {data['behavioralError']}[/]")
    behavioral = data.get("behavioralDiff") or {}
    if behavioral:
        console.print(
            Text.assemble(
                ("behavioral ", "label"),
                ("consistency ", "label"),
                (_num(behavioral.get("consistencyScore"), 4), "value"),
                ("   robustness ", "label"),
                (_num(behavioral.get("robustnessScore"), 4), "value"),
            )
        )
    if data.get("saved_to"):
        console.print(f"[muted]saved → {data['saved_to']}[/]")
 
    _print_dict_list_table(
        console,
        "Layer profile",
        data.get("layerProfile") or [],
        [("Layer", "layer"), ("‖Δ‖", "delta_l2"), ("N", "n_matrices")],
    )
    collapse = data.get("collapseSignals") or []
    if collapse:
        _print_dict_list_table(
            console,
            "Rank / collapse signals",
            collapse,
            [
                ("Layer", "layer"),
                ("Matrix", "matrix"),
                ("‖Δ‖", "deltaL2"),
                ("Stable rank", "deltaStableRank"),
            ],
        )
    _print_dict_list_table(
        console,
        "Top changed matrices",
        data.get("topChanged") or data.get("matrices") or [],
        [
            ("Layer", "layer"),
            ("Matrix", "matrix"),
            ("‖Δ‖", "delta_l2"),
            ("Rel", "delta_l2_relative"),
            ("Stable rank", "delta_stable_rank"),
        ],
    )
 
 
def _print_merge_analysis(console: Console, data: dict[str, Any]) -> None:
    _print_weight_diff(console, data)
 
 
def _print_trajectory_analysis(console: Console, data: dict[str, Any]) -> None:
    console.print(Rule("[heading]Checkpoint trajectory[/]", style="muted"))
    console.print(
        Text.assemble(
            ("model ", "label"),
            (str(data.get("baseModelId", "—")), "value"),
            ("   checkpoints ", "label"),
            (str(data.get("nCheckpoints", "—")), "value"),
            ("   analyzed ", "label"),
            (str(data.get("nAnalyzed", "—")), "value"),
        )
    )
    if data.get("peakStep") is not None:
        console.print(
            Text.assemble(
                ("peak step ", "label"),
                (str(data.get("peakStep")), "value"),
                ("   peak total ‖Δ‖ ", "label"),
                (_num(data.get("peakTotalDeltaL2"), 6), "value"),
            )
        )
    if data.get("saved_to"):
        console.print(f"[muted]saved → {data['saved_to']}[/]")
    rows = []
    for step in data.get("steps") or []:
        if step.get("error"):
            rows.append(
                {
                    "step": step.get("step", "—"),
                    "name": step.get("name", "—"),
                    "totalDeltaL2": f"error: {step.get('error')}",
                    "deltaFromPrevious": "—",
                }
            )
        else:
            rows.append(
                {
                    "step": step.get("step", "—"),
                    "name": step.get("name", "—"),
                    "totalDeltaL2": step.get("totalDeltaL2"),
                    "deltaFromPrevious": step.get("deltaFromPrevious"),
                }
            )
    _print_dict_list_table(
        console,
        "Steps",
        rows,
        [
            ("Step", "step"),
            ("Name", "name"),
            ("Total ‖Δ‖", "totalDeltaL2"),
            ("Δ from prev", "deltaFromPrevious"),
        ],
    )
 
 
def _print_residual_drift(console: Console, data: dict[str, Any]) -> None:
    console.print(Rule("[heading]Residual drift (base vs FT)[/]", style="muted"))
    console.print(
        Text.assemble(
            ("model ", "label"),
            (str(data.get("baseModelId", "—")), "value"),
            ("   checkpoint ", "label"),
            (str(data.get("ftCheckpointName", "—")), "accent"),
            ("   mode ", "label"),
            (str(data.get("modelMode", "—")), "value"),
            ("   activation ", "label"),
            (str(data.get("activationMode", "—")), "value"),
        )
    )
    console.print(
        Text.assemble(
            ("probes ", "label"),
            (str(data.get("nProbes", "—")), "value"),
            ("   layers ", "label"),
            (str(data.get("nLayers", "—")), "value"),
            ("   mean drift ", "label"),
            (_num(data.get("meanDrift"), 6), "value"),
            ("   max drift ", "label"),
            (_num(data.get("maxDrift"), 6), "value"),
            ("   peak layer ", "label"),
            (str(data.get("peakLayer", "—")), "accent"),
        )
    )
    if data.get("saved_to"):
        console.print(f"[muted]saved → {data['saved_to']}[/]")
 
    _print_dict_list_table(
        console,
        "Layer profile",
        data.get("layerProfile") or [],
        [
            ("Layer", "layer"),
            ("Mean dist", "mean_cosine_distance"),
            ("Max dist", "max_cosine_distance"),
            ("Mean sim", "mean_cosine_sim"),
        ],
    )
    _print_dict_list_table(
        console,
        "Top drift layers",
        data.get("topLayers") or [],
        [("Layer", "layer"), ("Mean dist", "mean_cosine_distance")],
    )
    _print_dict_list_table(
        console,
        "Per-probe drift",
        data.get("perProbe") or [],
        [
            ("#", "probe_index"),
            ("Preview", "probe_preview"),
            ("Mean", "mean_drift"),
            ("Max", "max_drift"),
            ("Peak L", "peak_layer"),
        ],
    )
 
 
def _print_confidence_analysis(console: Console, data: dict[str, Any]) -> None:
    console.print(
        Text.assemble(
            ("model ", "label"),
            (str(data.get("model_id", "—")), "value"),
            ("   mode ", "label"),
            (str(data.get("mode", "llm")), "accent"),
            ("   probes ", "label"),
            (str(data.get("n_probes", "—")), "value"),
            ("   mean conf ", "label"),
            (_num(data.get("mean_confidence"), 4), "value"),
            ("   ECE proxy ", "label"),
            (_num(data.get("aggregate_ece_proxy"), 4), "value"),
            ("   low-conf ", "label"),
            (str(data.get("low_confidence_count", "—")), "accent"),
        )
    )
    if data.get("join_sae"):
        console.print(f"[muted]SAE join layer {data.get('sae_layer')}[/]")
    if data.get("saved_to"):
        console.print(f"[muted]saved → {data['saved_to']}[/]")
 
    _print_dict_list_table(
        console,
        "Stressor summary",
        data.get("stressor_summary") or [],
        [
            ("Stressor", "stressor"),
            ("N", "n_probes"),
            ("Mean conf", "mean_confidence"),
            ("Entropy", "mean_entropy"),
            ("ECE", "ece_proxy"),
            ("Δ conf", "confidence_delta"),
            ("Low-conf", "low_confidence_count"),
        ],
    )
 
    cols = [
        ("ID", "id"),
        ("Stressor", "stressor"),
        ("Conf", "mean_confidence"),
        ("Max P", "max_prob"),
        ("Entropy", "entropy"),
        ("ECE", "ece_proxy"),
    ]
    if data.get("join_sae"):
        cols.extend([("L0", "mean_l0"), ("Top feat", "top_feature_idx")])
 
    _print_dict_list_table(
        console,
        "Per-probe metrics",
        data.get("probes") or [],
        cols,
    )
 
 
def _print_sae_stats(console: Console, data: dict[str, Any]) -> None:
    console.print(Rule("[heading]SAE layer statistics[/]", style="muted"))
    console.print(
        Text.assemble(
            ("model ", "label"),
            (str(data.get("model_id", "—")), "value"),
            ("   mode ", "label"),
            (str(data.get("mode", "—")), "accent"),
            ("   probes ", "label"),
            (str(data.get("n_probes", "—")), "value"),
            ("   layers ", "label"),
            (str(data.get("layers_requested") or "—"), "value"),
        )
    )
    if data.get("saved_to"):
        console.print(f"[muted]saved → {data['saved_to']}[/]")
 
    _print_dict_list_table(
        console,
        "Layer profile (mean L0 across probes)",
        data.get("layer_profile") or [],
        [
            ("Layer", "layer"),
            ("Mean L0", "mean_l0"),
            ("Sparsity", "sparsity"),
            ("Mean act", "mean_activation"),
        ],
    )
 
    for layer_entry in data.get("layer_stats") or []:
        if not layer_entry.get("sae_available"):
            console.print(f"[warn]Layer {layer_entry.get('layer')}: {layer_entry.get('error', 'no SAE')}[/]")
            continue
        layer = layer_entry.get("layer")
        top = layer_entry.get("top_features") or []
        if not top:
            continue
        console.print(Rule(f"[heading]Layer {layer} — top features[/]", style="muted"))
        _print_dict_list_table(
            console,
            None,
            top,
            [("Feature", "feature_idx"), ("Mean act", "mean_activation")],
        )
 
 
def _print_perturbation(console: Console, data: dict[str, Any]) -> None:
    raw = data.get("channels") or data.get("perturbation_results") or []
    rows = [
        {
            "channel": r.get("channel"),
            "kl": r.get("kl_mean", r.get("kl_divergence")),
        }
        for r in raw
    ]
    console.print(
        Text.assemble(
            ("layer ", "label"),
            (str(data.get("layer", "—")), "value"),
            ("   channels tested ", "label"),
            (str(len(rows)), "value"),
            ("   mean KL ", "label"),
            (_num(data.get("global_mean_kl"), 4), "value"),
            ("   max KL ", "label"),
            (_num(data.get("global_max_kl"), 4), "value"),
        )
    )
    _print_dict_list_table(
        console,
        "Channel sensitivity (KL)",
        rows[:20],
        [
            ("Channel", "channel"),
            ("KL", "kl"),
        ],
    )
    if len(rows) > 20:
        console.print(f"[muted]  … {len(rows) - 20} more channels  (use --check)[/]")
 
 
def _print_attention(console: Console, data: dict[str, Any]) -> None:
    heads = data.get("heads") or []
    top_sink = data.get("top_sink_heads") or []
    top_ind = data.get("top_induction_heads") or []
    console.print(
        Text.assemble(
            ("prompt ", "label"),
            (_trunc(data.get("prompt", ""), 48), "value"),
            ("   layers ", "label"),
            (str(data.get("n_layers", "—")), "value"),
            ("   heads/layer ", "label"),
            (str(data.get("n_heads", "—")), "value"),
        )
    )
    if top_sink:
        console.print(Rule("[heading]Top sink heads[/]", style="muted"))
        _print_dict_list_table(
            console,
            "",
            top_sink,
            [
                ("Layer", "layer"),
                ("Head", "head"),
                ("Sink", "sink_score"),
                ("Induction", "induction_score"),
            ],
        )
    n_ind_pairs = data.get("n_induction_pairs", 0)
    if n_ind_pairs == 0:
        console.print(
            "[muted]Induction: 0 — no repeated tokens in prompt "
            "(try e.g. \"The cat sat. The cat\")[/]"
        )
    elif top_ind:
        console.print(Rule("[heading]Top induction heads[/]", style="muted"))
        _print_dict_list_table(
            console,
            "",
            top_ind,
            [
                ("Layer", "layer"),
                ("Head", "head"),
                ("Sink", "sink_score"),
                ("Induction", "induction_score"),
            ],
        )
    if not top_sink and not top_ind and heads:
        console.print(f"[muted]{len(heads)} heads scored  (use --check)[/]")
 
 
def _print_check_weights(console: Console, data: dict[str, Any]) -> None:
    verdict = str(data.get("verdict", "—"))
    verdict_style = {
        "clean": "ok",
        "suspicious": "warn",
        "high_risk": "bad",
    }.get(verdict, "value")
    console.print(
        Text.assemble(
            ("verdict ", "label"),
            (verdict, verdict_style),
            ("   composite risk ", "label"),
            (f"{float(data.get('composite_risk', 0)):.1%}", "value"),
            ("   flagged ", "label"),
            (f"{data.get('pct_flagged', 0)}%", "value"),
        )
    )
    console.print(
        Text.assemble(
            ("layers ", "label"),
            (str(data.get("layers_analysed", "—")), "value"),
            ("   high ", "label"),
            (str(data.get("high_risk_count", 0)), "bad"),
            ("   suspicious ", "label"),
            (str(data.get("suspicious_count", 0)), "warn"),
            ("   clean ", "label"),
            (str(data.get("clean_count", 0)), "ok"),
        )
    )
    flags = data.get("all_flags") or []
    if flags:
        console.print(Rule("[heading]Flags[/]", style="muted"))
        for flag in flags[:8]:
            console.print(f"  [warn]•[/] {flag}")
        if len(flags) > 8:
            console.print(f"  [muted]… and {len(flags) - 8} more[/]")
 
    scored = data.get("scored_tensors") or []
    flagged = [t for t in scored if t.get("status") in ("high_risk", "suspicious")]
    if flagged:
        console.print(Rule("[heading]Top flagged tensors[/]", style="muted"))
        _print_dict_list_table(
            console,
            "",
            sorted(flagged, key=lambda t: t.get("risk_score", 0), reverse=True)[:12],
            [
                ("Layer", "layer_idx"),
                ("Tensor", "name"),
                ("Risk", "risk_score"),
                ("Status", "status"),
            ],
        )
    elif scored:
        console.print(f"[muted]{len(scored)} tensors scored — all clean[/]")
 
    rank = data.get("rank")
    if isinstance(rank, dict) and rank.get("n_layers") is not None:
        console.print(Rule("[heading]Weight matrix rank[/]", style="muted"))
        backend = rank.get("backend")
        if backend:
            console.print(f"[muted]backend[/] {backend}")
        if rank.get("skipped"):
            console.print(f"[warn]{rank.get('reason', 'rank scan skipped')}[/]")
        console.print(
            Text.assemble(
                ("layers ", "label"),
                (str(rank.get("n_layers", "—")), "value"),
                ("   collapsed ", "label"),
                (str(rank.get("collapsed_count", 0)), "bad" if rank.get("collapsed_count") else "ok"),
                ("   threshold ", "label"),
                (str(rank.get("collapse_threshold", 0.1)), "value"),
            )
        )
 
 
def _print_umap(console: Console, data: dict[str, Any]) -> None:
    n_features = data.get("n_features")
    n_points = data.get("n_points") or len(data.get("points") or [])
    layer = data.get("layer")
    source = data.get("source", "computed")
    console.print(Rule("[heading]UMAP projection[/]", style="muted"))
    console.print(
        Text.assemble(
            ("features ", "label"),
            (str(n_features if n_features is not None else "—"), "value"),
            ("   plotted ", "label"),
            (str(n_points), "accent"),
            ("   layer ", "label"),
            (str(layer if layer is not None else "—"), "value"),
            ("   source ", "label"),
            (str(source), "muted"),
        )
    )
    if n_features and n_points and n_points < n_features:
        console.print(
            f"[muted]{n_points} points synced for web "
            f"({n_features} total features; use --check for full coords)[/]"
        )
 
 
 
 
def _print_simulate(console: Console, data: dict[str, Any]) -> None:
    from aquin.compute.card_mapper import simulation_full_card_data
 
    if data.get("status") == "error" or data.get("error"):
        _print_error(console, str(data.get("error", "simulation failed")))
        if data.get("run_id"):
            console.print(f"[muted]partial run saved: {data['run_id']}[/]")
        return
 
    sim = simulation_full_card_data(data)
    run_id = data.get("run_id") or sim.get("savedRunId") or "—"
 
    console.print(Rule("[heading]Summary[/]", style="muted"))
    summary = Table(show_header=False, box=None, padding=(0, 1))
    summary.add_column(style="label")
    summary.add_column(style="value")
    summary.add_row("run_id", str(run_id))
    summary.add_row("model", sim.get("modelClass") or sim.get("modelId") or "—")
    summary.add_row("method", str(sim.get("method") or "—"))
    if data.get("topic"):
        summary.add_row("topic", str(data["topic"]))
    cfg = sim.get("config") or {}
    if cfg.get("rank"):
        summary.add_row("LoRA", f"r={cfg['rank']} α={cfg.get('alpha', '?')}")
    if cfg.get("lr") is not None:
        summary.add_row("lr", f"{float(cfg['lr']):.2e}")
    console.print(summary)
 
    dq = sim.get("datasetQuality")
    if isinstance(dq, dict):
        console.print(Rule("[heading]Dataset[/]", style="muted"))
        dq_table = Table(show_header=False, box=None, padding=(0, 1))
        dq_table.add_column(style="label")
        dq_table.add_column(style="value")
        for key, label in (
            ("nSamples", "samples"),
            ("diversityScore", "diversity"),
            ("harmfulCount", "harmful"),
            ("shortInstructions", "short instructions"),
        ):
            if dq.get(key) is not None:
                val = dq[key]
                if isinstance(val, float):
                    val = f"{val:.3f}"
                dq_table.add_row(label, str(val))
        console.print(dq_table)
 
    sae = sim.get("saePrediction")
    if isinstance(sae, dict) and sae.get("topFeatures"):
        feats = sae["topFeatures"][:8]
        console.print(Rule("[heading]Top SAE feature shifts[/]", style="muted"))
        ft = Table(show_header=True, header_style="label", box=None, padding=(0, 1))
        ft.add_column("Feature")
        ft.add_column("Direction")
        ft.add_column("Score", justify="right")
        for f in feats:
            idx = f.get("feature_idx", "?")
            direction = f.get("direction", "")
            score = f.get("score")
            if isinstance(score, (int, float)) and abs(score) >= 1e-4:
                score_s = f"{float(score):+.4f}"
            elif isinstance(score, (int, float)):
                score_s = f"{float(score):+.2e}"
            else:
                score_s = "—"
            style = "good" if direction == "strengthen" else "warn"
            ft.add_row(str(idx), Text(direction, style=style), score_s)
        console.print(ft)
 
    infl = sim.get("influenceScores")
    if isinstance(infl, dict) and infl.get("topSamples"):
        console.print(Rule("[heading]High-impact samples[/]", style="muted"))
        it = Table(show_header=True, header_style="label", box=None, padding=(0, 1))
        it.add_column("Idx")
        it.add_column("Influence", justify="right")
        it.add_column("Direction")
        it.add_column("Instruction")
        for s in infl["topSamples"][:5]:
            direction = s.get("direction", "")
            style = "bad" if direction == "harmful" else "good"
            instr = str(s.get("instruction", ""))[:60]
            inf = s.get("influence")
            inf_s = f"{float(inf):+.4f}" if isinstance(inf, (int, float)) and abs(float(inf)) < 1e4 else (
                f"{float(inf):+.4e}" if isinstance(inf, (int, float)) else "—"
            )
            it.add_row(str(s.get("idx", "")), inf_s, Text(direction, style=style), instr)
        console.print(it)
 
    ls = sim.get("lossSharpness")
    if isinstance(ls, dict):
        lam = ls.get("maxEigenvalue")
        lam_s = f"{float(lam):.2e}" if isinstance(lam, (int, float)) else "?"
        console.print(
            f"[label]Loss landscape:[/] [value]{ls.get('sharpnessLabel', '?')}[/] "
            f"[muted](λ={lam_s})[/]"
        )
 
    md = sim.get("modelDiff")
    if isinstance(md, dict):
        parts = []
        for label, key in (
            ("consistency", "consistencyScore"),
            ("suppression", "suppressionScore"),
            ("robustness", "robustnessScore"),
        ):
            val = md.get(key)
            if isinstance(val, (int, float)):
                parts.append(f"{label}={val:.2f}")
        if parts:
            console.print(f"[label]Attack surface:[/] [value]{' · '.join(parts)}[/]")
 
    losses = sim.get("lossHistory") or []
    if losses:
        final_loss = losses[-1]
        console.print(
            f"[label]Gradient steps:[/] [value]{len(losses)}[/] "
            f"[muted]final loss {float(final_loss):.4f}[/]"
        )
 
    cal = sim.get("calibration")
    if isinstance(cal, dict) and cal.get("base_ece") is not None:
        console.print(
            f"[label]Calibration:[/] [value]base ECE {float(cal['base_ece']):.4f}[/] "
            f"[muted]→[/] [value]ft ECE {float(cal['ft_ece']):.4f}[/] "
            f"[muted](Δ {float(cal.get('ece_delta', 0)):+.4f})[/]"
        )
        low = cal.get("low_confidence_rows") or []
        if low:
            console.print(f"[muted]{len(low)} low-confidence row(s)[/]")
 
 
def _format_saved_at(iso: str) -> str:
    if not iso:
        return "—"
    try:
        from datetime import datetime
        dt = datetime.fromisoformat(iso.replace("Z", "+00:00"))
        return dt.strftime("%Y-%m-%d %H:%M")
    except ValueError:
        return iso[:16] if len(iso) >= 16 else iso
 
 
def _print_simulation_list(console: Console, data: dict[str, Any]) -> None:
    runs = data.get("runs") or []
    if not runs:
        console.print(
            "[muted]No saved simulation runs. Run[/] [accent]aquin simulate[/] "
            "[muted]to create one.[/]"
        )
        return
 
    console.print(Rule("[heading]Saved simulations[/]", style="muted"))
    table = Table(show_header=True, header_style="label", box=None, padding=(0, 1))
    table.add_column("Run ID", style="accent", no_wrap=True)
    table.add_column("Model", style="value")
    table.add_column("Saved", style="muted", no_wrap=True)
    table.add_column("Samples", justify="right", style="value")
    table.add_column("Topic / dataset", style="muted")
 
    for entry in runs:
        if isinstance(entry, str):
            table.add_row(entry, "—", "—", "—", "—")
            continue
        run_id = str(entry.get("run_id") or "—")
        model = _trunc(str(entry.get("model_id") or "—"), 28)
        saved = _format_saved_at(str(entry.get("saved_at") or ""))
        n_samples = entry.get("n_samples")
        samples = str(n_samples) if n_samples is not None else "—"
        topic = entry.get("topic")
        ds = entry.get("dataset_path") or ""
        if topic:
            subtitle = _trunc(str(topic), 36)
        elif ds:
            subtitle = _trunc(str(ds).replace("\\", "/").split("/")[-1], 36)
        else:
            subtitle = "—"
        method = entry.get("method")
        if method and method != "LoRA":
            model = f"{model} [{method}]"
        table.add_row(run_id, model, saved, samples, subtitle)
 
    console.print(table)
    count = data.get("count", len(runs))
    console.print()
    console.print(
        f"[muted]{count} run(s). Load one with:[/] "
        f"[accent]aquin replay simulation --run_id <id>[/]"
    )
 
 
def _print_loaded_simulation(console: Console, data: dict[str, Any]) -> None:
    if data.get("status") == "not_found":
        _print_error(console, f"Run not found: {data.get('run_id', '?')}")
        return
    run_id = data.get("run_id", "")
    payload = {k: v for k, v in data.items() if k != "run_id"}
    wrapped = {
        "run_id": run_id,
        "result": payload,
        "events": [],
        "model_id": payload.get("model_id") or (payload.get("meta") or {}).get("modelId"),
    }
    _print_simulate(console, wrapped)
 
 
def _fmt_influence(val: float | None) -> str:
    if val is None:
        return "—"
    v = float(val)
    if abs(v) >= 1e4 or (abs(v) > 0 and abs(v) < 1e-4):
        return f"{v:+.4e}"
    return f"{v:+.4f}"
 
 
def _short_param_name(name: str) -> str:
    import re
    m = re.search(r"layers\.(\d+)", name)
    layer = f"L{m.group(1)}" if m else ""
    parts = name.split(".")
    tail = ".".join(parts[-2:]) if len(parts) >= 2 else name
    return f"{layer}.{tail}" if layer else _trunc(name, 44)
 
 
def _print_simulation_compare(console: Console, data: dict[str, Any]) -> None:
    comp = data.get("comparison") if isinstance(data.get("comparison"), dict) else None
    if not comp:
        _print_generic(console, "compare simulation", data)
        return
 
    console.print(Rule("[heading]Simulation comparison[/]", style="muted"))
    header = Table(show_header=False, box=None, padding=(0, 1))
    header.add_column(style="label")
    header.add_column(style="value")
    header.add_row(str(comp.get("label_a") or "Before"), str(data.get("run_id_a") or "—"))
    header.add_row(str(comp.get("label_b") or "After"), str(data.get("run_id_b") or "—"))
    console.print(header)
 
    run_a = comp.get("run_a") if isinstance(comp.get("run_a"), dict) else {}
    run_b = comp.get("run_b") if isinstance(comp.get("run_b"), dict) else {}
    if run_a or run_b:
        meta = Table(show_header=True, header_style="label", box=None, padding=(0, 1))
        meta.add_column("")
        meta.add_column(str(comp.get("label_a") or "A"), style="value")
        meta.add_column(str(comp.get("label_b") or "B"), style="value")
        for label, key in (
            ("model", "model_id"),
            ("samples", "n_samples"),
            ("final loss", "final_loss"),
            ("sharpness", "sharpness"),
            ("influence", "influence_method"),
        ):
            va, vb = run_a.get(key), run_b.get(key)
            if va is not None or vb is not None:
                meta.add_row(label, str(va if va is not None else "—"), str(vb if vb is not None else "—"))
        console.print(meta)
 
    loss_delta = comp.get("lossDelta")
    if loss_delta is not None or run_a.get("n_samples") != run_b.get("n_samples"):
        parts = []
        sa, sb = run_a.get("n_samples"), run_b.get("n_samples")
        if sa is not None and sb is not None and sa != sb:
            parts.append(f"samples {sa}→{sb}")
        if loss_delta is not None:
            parts.append(f"final loss {run_a.get('final_loss')}→{run_b.get('final_loss')} (Δ{float(loss_delta):+.4f})")
        if parts:
            console.print(f"[label]Dataset shift:[/] [value]{' · '.join(parts)}[/]")
 
    console.print(
        f"[label]Flipped:[/] [value]{comp.get('nFlippedFeatures', 0)} features[/] "
        f"[muted]·[/] [value]{comp.get('nFlippedInfluence', 0)} influence[/] "
        f"[muted]· max feature Δ[/] [value]{comp.get('maxFeatureDelta', 0)}[/]"
    )
    if comp.get("similarRuns"):
        console.print(
            "[muted]Runs are nearly identical — same dataset + config reproduced the same prediction.[/]"
        )
    elif comp.get("saeOverlapIdentical") and not comp.get("similarRuns"):
        n_only_a = comp.get("nFeaturesOnlyA", 0)
        n_only_b = comp.get("nFeaturesOnlyB", 0)
        n_overlap = comp.get("nFeaturesOverlap", 0)
        msg = (
            f"[muted]{n_overlap} overlapping SAE features have identical scores"
        )
        if n_only_a or n_only_b:
            msg += f" · {n_only_a} only in Before · {n_only_b} only in After"
        msg += " — loss/sample count still differ.[/]"
        console.print(msg)
 
    scores = comp.get("modelScores") or {}
    deltas = comp.get("attackSurfaceDeltas") if isinstance(comp.get("attackSurfaceDeltas"), dict) else {}
    atk = Table(show_header=True, header_style="label", box=None, padding=(0, 1))
    atk.add_column("Metric")
    atk.add_column(str(comp.get("label_a") or "A"), justify="right")
    atk.add_column(str(comp.get("label_b") or "B"), justify="right")
    atk.add_column("Δ", justify="right", style="muted")
    has_atk = False
    for label, key in (
        ("Consistency", "consistencyScore"),
        ("Suppression", "suppressionScore"),
        ("Robustness", "robustnessScore"),
    ):
        side_a = scores.get("a") if isinstance(scores, dict) else {}
        side_b = scores.get("b") if isinstance(scores, dict) else {}
        va = side_a.get(key) if isinstance(side_a, dict) else None
        vb = side_b.get(key) if isinstance(side_b, dict) else None
        if isinstance(va, (int, float)) or isinstance(vb, (int, float)):
            has_atk = True
            delta_key = key.replace("Score", "")
            d = deltas.get(delta_key)
            delta_s = f"{float(d):+.3f}" if isinstance(d, (int, float)) else "—"
            atk.add_row(
                label,
                _pct(va) if isinstance(va, (int, float)) else "—",
                _pct(vb) if isinstance(vb, (int, float)) else "—",
                delta_s,
            )
    if has_atk:
        console.print(Rule("[heading]Attack surface[/]", style="muted"))
        console.print(atk)
 
    sharp = comp.get("sharpness") if isinstance(comp.get("sharpness"), dict) else {}
    if sharp.get("label_a") or sharp.get("label_b"):
        console.print(
            f"[label]Loss landscape:[/] [value]{sharp.get('label_a', '?')}[/] "
            f"[muted]→[/] [value]{sharp.get('label_b', '?')}[/]"
        )
 
    diffs = comp.get("featureDiffs") or []
    nonzero_feats = [
        f for f in diffs
        if f.get("score_delta") is not None and abs(float(f["score_delta"])) > 1e-9
    ]
    if nonzero_feats:
        console.print(Rule("[heading]SAE feature shifts (by |Δ|)[/]", style="muted"))
        ft = Table(show_header=True, header_style="label", box=None, padding=(0, 1))
        ft.add_column("Feature")
        ft.add_column("Score A", justify="right")
        ft.add_column("Score B", justify="right")
        ft.add_column("Δ", justify="right")
        ft.add_column("Dir")
        for f in nonzero_feats[:12]:
            sa, sb = f.get("score_a"), f.get("score_b")
            delta = f.get("score_delta")
            dir_s = f"{f.get('direction_a', '?')}→{f.get('direction_b', '?')}"
            if f.get("flipped"):
                dir_s = Text(dir_s + " ⚡", style="warn")
            ft.add_row(
                str(f.get("feature_idx", "?")),
                f"{float(sa):+.4f}" if isinstance(sa, (int, float)) else "—",
                f"{float(sb):+.4f}" if isinstance(sb, (int, float)) else "—",
                f"{float(delta):+.4e}" if isinstance(delta, (int, float)) else "—",
                dir_s,
            )
        console.print(ft)
    else:
        only_a = [f for f in diffs if f.get("only_in") == "a"]
        only_b = [f for f in diffs if f.get("only_in") == "b"]
        n_overlap = comp.get("nFeaturesOverlap", 0)
        if n_overlap and comp.get("saeOverlapIdentical"):
            console.print(
                f"[muted]{n_overlap} overlapping SAE features — identical scores[/]"
            )
        elif diffs:
            console.print(f"[muted]{len(diffs)} SAE features in union — no score deltas above noise[/]")
        for title, rows, score_key in (
            (f"Top SAE features only in {comp.get('label_a', 'Before')}", only_a[:8], "score_a"),
            (f"Top SAE features only in {comp.get('label_b', 'After')}", only_b[:8], "score_b"),
        ):
            if not rows:
                continue
            console.print(Rule(f"[heading]{title}[/]", style="muted"))
            ot = Table(show_header=True, header_style="label", box=None, padding=(0, 1))
            ot.add_column("Feature")
            ot.add_column("Score", justify="right")
            ot.add_column("Dir")
            for f in rows:
                sc = f.get(score_key)
                ot.add_row(
                    str(f.get("feature_idx", "?")),
                    f"{float(sc):+.4f}" if isinstance(sc, (int, float)) else "—",
                    str(f.get("direction_a" if score_key == "score_a" else "direction_b") or "—"),
                )
            console.print(ot)
 
    infl_avail = comp.get("influenceAvailable") if isinstance(comp.get("influenceAvailable"), dict) else {}
    infl = comp.get("influenceDiffs") or []
    paired_infl = [
        s for s in infl
        if s.get("influence_a") is not None and s.get("influence_b") is not None
    ]
    if infl_avail.get("a") and not infl_avail.get("b"):
        console.print(
            f"[warn]{comp.get('label_b', 'After')} has no saved influence scores[/] "
            f"[muted](re-run simulate on that config to compare influence)[/]"
        )
    elif infl_avail.get("b") and not infl_avail.get("a"):
        console.print(
            f"[warn]{comp.get('label_a', 'Before')} has no saved influence scores[/]"
        )
    if paired_infl:
        console.print(Rule("[heading]Influence changes[/]", style="muted"))
        it = Table(show_header=True, header_style="label", box=None, padding=(0, 1))
        it.add_column("Idx")
        it.add_column("A", justify="right")
        it.add_column("B", justify="right")
        it.add_column("Δ", justify="right")
        it.add_column("Instruction")
        for s in paired_infl[:8]:
            it.add_row(
                str(s.get("idx", "")),
                _fmt_influence(s.get("influence_a")),
                _fmt_influence(s.get("influence_b")),
                _fmt_influence(s.get("delta")),
                _trunc(str(s.get("instruction") or ""), 40),
            )
        console.print(it)
    elif infl_avail.get("a") or infl_avail.get("b"):
        console.print("[muted]No overlapping influence samples to compare[/]")
 
    lr = comp.get("lrDiffs") or []
    if lr:
        console.print(Rule("[heading]Effective LR deltas[/]", style="muted"))
        lt = Table(show_header=True, header_style="label", box=None, padding=(0, 1))
        lt.add_column("Param")
        lt.add_column("A", justify="right")
        lt.add_column("B", justify="right")
        lt.add_column("Δ", justify="right")
        for row in lr[:6]:
            lt.add_row(
                _short_param_name(str(row.get("param", ""))),
                _num(row.get("lr_a"), 6),
                _num(row.get("lr_b"), 6),
                _num(row.get("delta"), 6),
            )
        console.print(lt)
    elif comp.get("similarRuns"):
        console.print("[muted]Effective LR identical across shared parameters[/]")
 
 
def _print_generic(console: Console, verb: str, data: dict[str, Any]) -> None:
    if _is_llm_layer_analysis_payload(data):
        _print_layer_analysis(console, data)
        return
    if data.get("perturbation_results") or data.get("channels"):
        _print_perturbation(console, data)
        return
    if data.get("layer_norms"):
        _print_dict_list_table(
            console,
            "Residual norms",
            data["layer_norms"],
            [("Layer", "layer"), ("Norm", "norm")],
        )
        return
 
    console.print(Rule(f"[heading]{verb}[/]", style="muted"))
    skip_keys: set[str] = set()
    for key, val in data.items():
        if key == "error" or not isinstance(val, list) or not val:
            continue
        if all(isinstance(item, dict) for item in val):
            keys = set()
            for item in val:
                keys.update(item.keys())
            if len(keys) <= 6:
                cols = [(k, k) for k in sorted(keys)]
                _print_dict_list_table(console, key.replace("_", " ").title(), val, cols)
                skip_keys.add(key)
 
    table = Table(show_header=True, header_style="label", box=None, padding=(0, 1))
    table.add_column("Key", style="label")
    table.add_column("Value", style="value")
    has_rows = False
    for key, val in data.items():
        if key == "error" or key in skip_keys:
            continue
        if isinstance(val, dict):
            table.add_row(key, f"[muted]{len(val)} fields  (use --check)[/]")
            has_rows = True
        elif isinstance(val, list):
            table.add_row(key, f"[muted]{len(val)} items  (use --check)[/]")
            has_rows = True
        elif isinstance(val, float):
            table.add_row(key, _num(val))
            has_rows = True
        elif isinstance(val, (str, int, bool)) or val is None:
            table.add_row(key, str(val) if val is not None else "—")
            has_rows = True
 
    if has_rows:
        console.print(table)
    elif not skip_keys:
        console.print("[muted]No displayable fields. Use --check for full result.[/]")
 
 
def _print_dataset_generate(console: Console, data: dict[str, Any]) -> None:
    path = data.get("path") or "—"
    topic = data.get("topic") or "—"
    count = data.get("count", 0)
    console.print(Rule("[heading]Generated dataset[/]", style="muted"))
    console.print(f"[label]Topic[/]  [value]{topic}[/]")
    console.print(f"[label]Rows[/]   [value]{count}[/]")
    console.print(f"[label]File[/]   [accent]{path}[/]")
    rows = data.get("rows") or []
    if rows:
        table = Table(show_header=True, header_style="label", box=None, padding=(0, 1))
        table.add_column("#", style="muted", justify="right", width=3)
        table.add_column("Instruction", style="value", max_width=48)
        table.add_column("Response", style="muted", max_width=48)
        for i, row in enumerate(rows[:8], 1):
            instr = str(row.get("instruction", ""))
            resp = str(row.get("response", ""))
            if len(instr) > 48:
                instr = instr[:45] + "..."
            if len(resp) > 48:
                resp = resp[:45] + "..."
            table.add_row(str(i), instr, resp)
        console.print(table)
        if len(rows) > 8:
            console.print(f"[muted]… and {len(rows) - 8} more row(s)[/]")
 
 
def print_tool_result(verb: str, result: Any, *, tool_name: str | None = None) -> None:
    """Pretty-print a tool result using the same Rich theme as aquin chat."""
    console = _console()
    data = _unwrap(result)
 
    if isinstance(data, dict) and data.get("error"):
        _print_error(console, str(data["error"]))
        return
 
    console.print()
    title = verb.replace("-", " ").title()
    model_id = data.get("model_id") if isinstance(data, dict) else None
    header = f"[title]{title}[/]"
    if model_id:
        header += f"  [muted]{model_id}[/]"
    console.print(Panel.fit(header, border_style="dim"))
 
    if verb == "audit":
        if data.get("consistency"):
            _print_consistency(console, data["consistency"])
        if data.get("suppression"):
            _print_suppression(console, data["suppression"])
        if data.get("boundary"):
            _print_boundary(console, data["boundary"])
        console.print()
        return
 
    if verb == "consistency-eval":
        _print_consistency(console, data)
        console.print()
        return
 
    if verb == "suppression-eval":
        _print_suppression(console, data)
        console.print()
        return
 
    if verb == "boundary-eval":
        _print_boundary(console, data)
        console.print()
        return
 
    if verb == "eval":
        _print_custom_eval(console, data)
        console.print()
        return
 
    if verb in ("benchmark", "benchmarks"):
        _print_benchmarks(console, data)
        console.print()
        return
 
    if verb == "feature locate":
        _print_find_feature(console, data)
        console.print()
        return
 
    if verb == "extract-steer-vector" or tool_name == "extract_steer_vector":
        _print_extract_steer_vector(console, data)
        console.print()
        return
 
    if verb in ("steer", "multi-steer"):
        _print_steer(console, data)
        console.print()
        return
 
    if verb == "red-team":
        _print_red_team(console, data)
        console.print()
        return
 
    if verb == "layer-analysis":
        _print_layer_analysis(console, data)
        console.print()
        return
 
    if verb == "sae-stats":
        _print_sae_stats(console, data)
        console.print()
        return
 
    if verb in ("confidence-analysis", "check confidence"):
        _print_confidence_analysis(console, data)
        console.print()
        return
 
    if verb == "diff weight":
        _print_weight_diff(console, data)
        console.print()
        return
 
    if verb == "check trajectory":
        _print_trajectory_analysis(console, data)
        console.print()
        return
 
    if verb == "diff residue":
        _print_residual_drift(console, data)
        console.print()
        return
 
    if verb == "perturbation":
        _print_perturbation(console, data)
        console.print()
        return
 
    if verb == "attention":
        _print_attention(console, data)
        console.print()
        return
 
    if verb == "check-weights":
        _print_check_weights(console, data)
        console.print()
        return
 
    if verb == "umap":
        _print_umap(console, data)
        console.print()
        return
 
    if verb == "simulate":
        _print_simulate(console, data)
        console.print()
        return
 
    if verb in ("list simulation", "list simulations", "list-runs"):
        _print_simulation_list(console, data)
        console.print()
        return
 
    if verb in ("replay simulation", "load simulation", "load-run"):
        _print_loaded_simulation(console, data)
        console.print()
        return
 
    if verb in ("compare simulation", "compare-runs"):
        _print_simulation_compare(console, data)
        console.print()
        return
 
    if verb == "dataset-generate":
        _print_dataset_generate(console, data)
        console.print()
        return
 
    if isinstance(data, dict):
        _print_generic(console, verb, data)
    else:
        console.print(data)
    console.print()