diff --git a/arborist/cli.py b/arborist/cli.py
index 15208af..2d50bbd 100644
--- a/arborist/cli.py
+++ b/arborist/cli.py
@@ -2196,6 +2196,129 @@ def _cmd_providence_show_preflight(
     return 0
 
 
+def _cmd_controller_events(args: argparse.Namespace) -> int:
+    """List ``controller_events`` rows from one shard or every shard.
+
+    Read-only inspector for the Phase 2 advisory writes
+    (``arborist.qa.runner._emit_qa_controller_advisory``). Surfaces
+    decision / difficulty / budget_allocation rows so operators can
+    measure the Retrigger 1 signal (#000045 §3) without raw SQL.
+    """
+    import sqlite3 as _sqlite3
+    import time as _time
+
+    shards_dir = args.global_shards_dir
+    db_path = args.db
+    if shards_dir:
+        shard_paths = sorted(Path(shards_dir).glob("*.db"))
+    elif db_path:
+        shard_paths = [Path(db_path)]
+    else:
+        shard_paths = [Path(DEFAULT_DB_PATH)]
+
+    where: list[str] = []
+    params: list = []
+    if args.kind:
+        where.append("event_kind = ?")
+        params.append(args.kind)
+    if args.organism_prefix:
+        where.append("organism_root LIKE ?")
+        params.append(args.organism_prefix + "%")
+    if args.since_seconds is not None:
+        where.append("recorded_at >= ?")
+        params.append(int(_time.time()) - args.since_seconds)
+    where_sql = (" WHERE " + " AND ".join(where)) if where else ""
+    sql = (
+        "SELECT event_id, organism_root, branch_id, event_kind, label,"
+        " entropy, difficulty, allocation, body_blob, recorded_at"
+        " FROM controller_events"
+        + where_sql
+        + " ORDER BY recorded_at DESC, event_id DESC LIMIT ?"
+    )
+
+    out: list[dict] = []
+    summary: dict[str, int] = {}
+    remaining = args.limit
+    for sp in shard_paths:
+        if remaining <= 0:
+            break
+        if sp.suffix != ".db" or "-shm" in sp.name or "-wal" in sp.name:
+            continue
+        try:
+            conn = _sqlite3.connect(f"file:{sp}?mode=ro", uri=True)
+        except _sqlite3.OperationalError:
+            continue
+        conn.row_factory = _sqlite3.Row
+        has_table = conn.execute(
+            "SELECT name FROM sqlite_master "
+            "WHERE type='table' AND name='controller_events'"
+        ).fetchone()
+        if not has_table:
+            conn.close()
+            continue
+        try:
+            rows = conn.execute(sql, [*params, remaining]).fetchall()
+        except _sqlite3.OperationalError:
+            conn.close()
+            continue
+        conn.close()
+        for r in rows:
+            kind = r["event_kind"]
+            summary[kind] = summary.get(kind, 0) + 1
+            entry = {
+                "shard": sp.name,
+                "event_id": r["event_id"],
+                "kind": kind,
+                "organism_root": r["organism_root"],
+                "branch_id": r["branch_id"],
+                "label": r["label"],
+                "entropy": r["entropy"],
+                "difficulty": r["difficulty"],
+                "allocation": r["allocation"],
+                "recorded_at": r["recorded_at"],
+                "recorded_at_iso": _time.strftime(
+                    "%Y-%m-%dT%H:%M:%SZ", _time.gmtime(r["recorded_at"])
+                ),
+            }
+            if args.body:
+                try:
+                    entry["body"] = json.loads(r["body_blob"])
+                except (TypeError, ValueError):
+                    entry["body"] = r["body_blob"]
+            out.append(entry)
+        remaining = args.limit - len(out)
+
+    if args.json:
+        print(json.dumps(
+            {"summary": summary, "rows": out},
+            indent=2, ensure_ascii=False,
+        ))
+        return 0
+
+    if not out:
+        print("(no controller_events rows matched)")
+        return 0
+    print(f"# controller_events — {len(out)} row(s) across "
+          f"{len({e['shard'] for e in out})} shard(s)")
+    for kind, n in sorted(summary.items(), key=lambda kv: -kv[1]):
+        print(f"#   {kind}: {n}")
+    print()
+    print(
+        f"{'shard':<10} {'kind':<29} {'label':<10} "
+        f"{'diff':>6} {'alloc':>6} {'recorded_at_iso'}  organism"
+    )
+    for e in out:
+        org = (e["organism_root"] or "")[:48]
+        diff = "-" if e["difficulty"] is None else f"{e['difficulty']:.2f}"
+        alloc = "-" if e["allocation"] is None else f"{e['allocation']:.2f}"
+        print(
+            f"{e['shard']:<10} {e['kind']:<29} "
+            f"{(e['label'] or '-'):<10} {diff:>6} {alloc:>6} "
+            f"{e['recorded_at_iso']}  {org}"
+        )
+    return 0
+
+
 def _load_record_context(row, shards_dir, qa_db):
     """Reassemble context for a providence record. Returns text or None
     if any source doc has no hot chunks (cold)."""
@@ -4889,6 +5012,49 @@ def build_parser() -> argparse.ArgumentParser:
     )
     prov_cmd.set_defaults(func=_cmd_providence)
 
+    ce_cmd = sub.add_parser(
+        "controller-events",
+        help=(
+            "list controller_events advisory rows from #000037 Phase 2 "
+            "(QA-runner → controller decision/difficulty/budget rows)"
+        ),
+    )
+    ce_cmd.add_argument("--limit", type=int, default=20)
+    ce_cmd.add_argument(
+        "--kind",
+        choices=[
+            "controller_decision",
+            "controller_difficulty",
+            "controller_budget_allocation",
+        ],
+        default=None,
+        help="filter to one event_kind",
+    )
+    ce_cmd.add_argument(
+        "--organism-prefix",
+        dest="organism_prefix",
+        default=None,
+        help='match organism_root LIKE prefix (e.g. "qa:" for QA-runner advisories)',
+    )
+    ce_cmd.add_argument(
+        "--since-seconds",
+        dest="since_seconds",
+        type=int,
+        default=None,
+        help="only rows recorded within the last N seconds",
+    )
+    ce_cmd.add_argument(
+        "--body",
+        action="store_true",
+        help="include the JSON body_blob in JSON output (off by default — bodies are bulky)",
+    )
+    ce_cmd.add_argument(
+        "--json",
+        action="store_true",
+        help="emit JSON instead of a terminal table",
+    )
+    ce_cmd.set_defaults(func=_cmd_controller_events)
+
     burn_cmd = sub.add_parser(
         "burn",
         help=(
diff --git a/tests/test_cli_smoke.py b/tests/test_cli_smoke.py
index 1b8a8a6..bdf8a2e 100644
--- a/tests/test_cli_smoke.py
+++ b/tests/test_cli_smoke.py
@@ -75,7 +75,8 @@ def test_arborist_no_subcommand_returns_error():
 # contains a verb not in this list).
 _TOP_LEVEL_SUBCOMMANDS = [
     "ingest", "search", "verify", "distill", "ask", "query",
-    "inspect", "losses", "providence", "burn", "burn-kindergarten",
+    "inspect", "losses", "providence", "controller-events",
+    "burn", "burn-kindergarten",
     "reclassify", "emergent", "evict", "rehydrate", "activity",
     "stats", "canon", "analyze", "snapshot", "substrate", "memory",
     "capital", "selfmodel", "warrant-status", "warrant-resolve",
diff --git a/tests/test_prometheus_audit.py b/tests/test_prometheus_audit.py
index 39e9915..717d039 100644
--- a/tests/test_prometheus_audit.py
+++ b/tests/test_prometheus_audit.py
@@ -475,3 +475,112 @@ def test_qa_runner_advisory_idempotent_on_same_verdict(shard):
             "SELECT COUNT(*) FROM controller_events"
         ).fetchone()[0]
     assert first_count == second_count
+
+
+# ---------------------------------------------------------------------
+# `arborist controller-events` CLI inspector (#000037 follow-through)
+# ---------------------------------------------------------------------
+
+
+def _run_inspector(shard_dir, **kwargs):
+    """Invoke ``_cmd_controller_events`` against a tmp shards dir."""
+    import argparse
+    from arborist.cli import _cmd_controller_events
+
+    args = argparse.Namespace(
+        global_shards_dir=str(shard_dir),
+        db=None,
+        limit=kwargs.get("limit", 20),
+        kind=kwargs.get("kind", None),
+        organism_prefix=kwargs.get("organism_prefix", None),
+        since_seconds=kwargs.get("since_seconds", None),
+        body=kwargs.get("body", False),
+        json=kwargs.get("json", False),
+    )
+    return _cmd_controller_events(args)
+
+
+def test_inspector_lists_rows_from_a_shard_dir(shard, capsys):
+    """Inspector reads every shard with a controller_events table.
+    Empty/missing-table shards are silently skipped."""
+    conn, db_path = shard
+    with transaction(conn):
+        emit_controller_events(
+            conn, _stub_decision(), organism_root="qa:abc123"
+        )
+    rc = _run_inspector(db_path.parent, limit=20)
+    out = capsys.readouterr().out
+    assert rc == 0
+    assert "controller_events" in out
+    assert "qa:abc123" in out
+    assert "controller_decision" in out
+
+
+def test_inspector_filters_by_kind(shard, capsys):
+    """``--kind controller_difficulty`` returns only difficulty rows."""
+    conn, db_path = shard
+    with transaction(conn):
+        emit_controller_events(
+            conn, _stub_decision(), organism_root="qa:k1"
+        )
+    rc = _run_inspector(
+        db_path.parent, kind="controller_difficulty", limit=20
+    )
+    out = capsys.readouterr().out
+    assert rc == 0
+    assert "controller_difficulty" in out
+    assert "controller_decision" not in out
+    assert "controller_budget_allocation" not in out
+
+
+def test_inspector_filters_by_organism_prefix(shard, capsys):
+    """``--organism-prefix`` matches LIKE prefix."""
+    conn, db_path = shard
+    with transaction(conn):
+        emit_controller_events(
+            conn, _stub_decision(), organism_root="qa:keep"
+        )
+        emit_controller_events(
+            conn,
+            _stub_decision(selected_branch_id="X", label="REJECT"),
+            organism_root="sweep:drop",
+        )
+    rc = _run_inspector(
+        db_path.parent, organism_prefix="qa:", limit=20
+    )
+    out = capsys.readouterr().out
+    assert rc == 0
+    assert "qa:keep" in out
+    assert "sweep:drop" not in out
+
+
+def test_inspector_json_emits_summary_and_rows(shard, capsys):
+    """``--json`` returns ``{"summary": {...}, "rows": [...]}``."""
+    conn, db_path = shard
+    with transaction(conn):
+        emit_controller_events(
+            conn, _stub_decision(), organism_root="qa:json-test"
+        )
+    rc = _run_inspector(db_path.parent, json=True, limit=20)
+    payload = json.loads(capsys.readouterr().out)
+    assert rc == 0
+    assert payload["summary"]["controller_decision"] >= 1
+    assert payload["summary"]["controller_difficulty"] >= 1
+    assert any(r["organism_root"] == "qa:json-test" for r in payload["rows"])
+
+
+def test_inspector_skips_shards_without_table(tmp_path, capsys):
+    """A shards-dir with a non-arborist sqlite file is skipped, not
+    crashed."""
+    import sqlite3 as _sqlite3
+
+    bogus = tmp_path / "not-arborist.db"
+    c = _sqlite3.connect(bogus)
+    c.execute("CREATE TABLE foo (x INTEGER)")
+    c.commit()
+    c.close()
+
+    rc = _run_inspector(tmp_path, limit=5)
+    out = capsys.readouterr().out
+    assert rc == 0
+    assert "no controller_events rows matched" in out