fix: command hardening — crash guards, empty-state disclosure, metric fidelity, perf caps

Crash fixes:
- owner/uses: route unbounded raw `IN (...)` clauses through batched_in so a
  directory with >999 files, or a bare name resolving to >999 symbols, no
  longer hits SQLITE_MAX_VARIABLE_NUMBER ("too many SQL variables").

Correctness:
- command_advice: `roam <cmd> --help` examples are no longer reported
  non-executable — `--help` is an eager flag that short-circuits before
  positional validation.
- graph modularity was always 0.0: a NotAPartition over an incomplete cluster
  partition was swallowed by a bare except. Repair the partition with
  singletons for uncovered nodes across clusters/fingerprint/simulate; real
  repos now report Q~0.8 instead of "no community structure".

Pattern-2 empty-state disclosure (no more silent clean/PASS/HEALTHY on an empty
corpus): cycles, clusters, dashboard, dead, clones, debt, alerts, complexity,
and uses now emit an explicit state via the shared resolve.empty_corpus_state
helper; clusters also discloses trivial_clustering when modularity Q<=0; verify
discloses state=no_changes on an empty diff.

Perf guards (defensive, behavior-preserving below the caps):
- partition + cut: k-sample (seeded) betweenness on large graphs.
- cut: pre-bucket cross-cluster edges in one pass (drops the O(clusters^2*|E|)
  rescan) and gate max-flow min-cut behind a node-count cap.

Tests added/updated for every change; full suite green on the CI loadgroup surface.

Author: Cranot

src/roam/command_advice.py

@@ -202,7 +202,16 @@ def _parse_status(tokens: list[str]) -> tuple[str, str | None]:
         return "not_checked", load_reason
     if command is None:
         return "failed", f"unknown roam subcommand: {tokens[idx]}"
-    return _make_context_status(command, tokens[idx], _strip_global_flags(tokens[idx + 1 :]))
+    sub_args = tokens[idx + 1 :]
+    # `--help` / `-h` is an EAGER flag: Click short-circuits and exits 0 before
+    # any positional/argument validation, so `roam <cmd> --help` is ALWAYS
+    # executable regardless of required positionals. Because `--help` lives in
+    # _GLOBAL_FLAGS_NO_VALUE it would otherwise be stripped, degrading
+    # `roam search --help` to `roam search` and raising a spurious "Missing
+    # parameter: pattern" — a false FAIL on a 100%-valid copy-paste command.
+    if "--help" in sub_args or "-h" in sub_args:
+        return "parsed", None
+    return _make_context_status(command, tokens[idx], _strip_global_flags(sub_args))
 
 
 def _is_root_level_invocation(tokens: list[str]) -> bool:

src/roam/commands/cmd_alerts.py

@@ -17,7 +17,7 @@
 
 from roam.capability import roam_capability
 from roam.commands.metrics_history import collect_metrics, get_snapshots
-from roam.commands.resolve import ensure_index
+from roam.commands.resolve import empty_corpus_state, ensure_index
 from roam.db.connection import find_project_root, open_db
 from roam.output._severity import severity_rank
 from roam.output.formatter import WarningsOut, json_envelope, to_json
@@ -1350,6 +1350,35 @@ def _run_check_cx(phase, fn, *args, default=None, **kwargs):
             return default
 
     with open_db(readonly=True) as conn:
+        # B8 (Pattern-2): on a 0-symbol corpus, "no alerts — all metrics within
+        # normal ranges" is a silent SAFE — nothing was analyzed. Live metrics
+        # are all vacuously healthy because there is no code. Disclose the empty
+        # corpus explicitly (canonical state + partial_success + a verdict that
+        # names it) instead of an all-clear that an agent reads as "repo is fine".
+        _empty = empty_corpus_state(conn)
+        if _empty is not None:
+            empty_verdict = "no alerts — no code indexed in corpus (0 symbols; run `roam index --force`)"
+            if json_mode:
+                click.echo(
+                    to_json(
+                        json_envelope(
+                            "alerts",
+                            summary={
+                                "verdict": empty_verdict,
+                                "total": 0,
+                                "critical": 0,
+                                "warning": 0,
+                                "info": 0,
+                                "snapshots_analyzed": 0,
+                                **_empty,
+                            },
+                            alerts=[],
+                        )
+                    )
+                )
+            else:
+                click.echo(f"VERDICT: {empty_verdict}")
+            return
         # W607-CX: ``get_snapshots`` substrate -- DB-row ingest.
         snaps_raw = _run_check_cx("get_snapshots", get_snapshots, conn, _trend_snapshot_limit(), default=[])
         if snaps_raw is None:

src/roam/commands/cmd_clones.py

@@ -19,7 +19,7 @@
 import click
 
 from roam.capability import roam_capability
-from roam.commands.resolve import ensure_index
+from roam.commands.resolve import empty_corpus_state, ensure_index
 from roam.db.connection import open_db
 from roam.index.file_roles import is_test as _is_test
 from roam.output.confidence import (
@@ -803,6 +803,20 @@ def _score_classify_run(_clusters):
                 default={"state": "DEGRADED", "scanned": 0},
             )
 
+            # W808 — Pattern-2 empty-corpus disclosure. When there are no
+            # clusters AND the index holds zero symbols, the "No structural
+            # clones detected" verdict would be a silent SAFE on an empty
+            # corpus (vacuously clean — nothing was scanned). Distinguish
+            # that 0-symbol case from the populated-but-no-clones case (and
+            # from the below-threshold / pre-filter gate-collapse cases,
+            # which are a DIFFERENT axis owned by W805-CCCC) via the
+            # canonical ``empty_corpus_state`` helper. Only the 0-symbol
+            # corpus yields ``state="empty_corpus"``; if symbols exist but
+            # no clones surfaced, the verdict stays the populated SAFE line.
+            _empty = empty_corpus_state(conn) if not clusters else None
+            if _empty is not None:
+                verdict_with_conf = "no structural clones — no symbols indexed in corpus (run `roam index --force`)"
+
             summary_payload = {
                 "verdict": verdict_with_conf,
                 "clusters": len(clusters),
@@ -824,6 +838,13 @@ def _score_classify_run(_clusters):
                 "run_state": _score_dict["state"],
                 **_cap_summary,
             }
+            # W808 — stamp the canonical empty-corpus disclosure
+            # (``state="empty_corpus"`` + ``partial_success=True``) onto
+            # the summary so consumers never read the empty-corpus SAFE as
+            # a clean result. Applied AFTER ``_cap_summary`` so the state
+            # field is not clobbered.
+            if _empty is not None:
+                summary_payload.update(_empty)
             # W607-BQ + DC: union the cap-hit disclosure list with the
             # substrate-CALL marker list AND the aggregation-phase
             # marker list. All three bins flush into the same

src/roam/commands/cmd_clusters.py

@@ -14,7 +14,7 @@
 import click
 
 from roam.capability import roam_capability
-from roam.commands.resolve import ensure_index
+from roam.commands.resolve import empty_corpus_state, ensure_index
 from roam.db.connection import open_db
 from roam.db.queries import ALL_CLUSTERS
 from roam.graph.builder import build_symbol_graph
@@ -253,23 +253,55 @@ def _clusters_json(conn, rows, min_size, quality, mermaid=None, detail=True, tok
     if mermaid is not None:
         extra["mermaid"] = mermaid
 
-    # Build verdict
+    # Build verdict + a machine-readable state (W805-BB Pattern-2). A bare
+    # "no clusters detected" conflated three distinct zero-cluster cases; an
+    # agent could not tell whether to re-index, lower --min-size, or accept
+    # that there is no modular structure. Disclose which case fired:
+    #   clusters_detected  — real decomposition surfaced (Newman Q > 0)
+    #   trivial_clustering — clusters exist but modularity Q<=0 (no real structure)
+    #   empty_corpus       — 0 symbols indexed (no graph to cluster)
+    #   no_clusters        — symbols exist but Louvain found no structure
+    #   below_min_size     — clusters exist but all smaller than --min-size
     n_visible = len(visible)
     largest = max(visible, key=lambda r: r["size"]) if visible else None
-    if largest:
-        verdict = f"{n_visible} clusters, largest: {largest['cluster_label']}({largest['size']} syms)"
+    mod_q = quality["modularity"]
+    summary = {
+        "clusters": n_visible,
+        "mismatches": sum(1 for m in mismatches if m["cluster_id"] in visible_ids),
+        "modularity_q": mod_q,
+        "mean_conductance": quality["mean_conductance"],
+    }
+    if largest and mod_q > 0.0:
+        summary["verdict"] = f"{n_visible} clusters, largest: {largest['cluster_label']}({largest['size']} syms)"
+        summary["state"] = "clusters_detected"
+    elif largest:
+        # Visible clusters exist but Newman modularity Q<=0 means there is no real
+        # community structure (Q>0.3 is meaningful; Newman 2004). Disclose so an
+        # agent does not read "N clusters" as a genuine architectural
+        # decomposition and proceed to refactor a phantom cluster. (Now that
+        # modularity is computed against a repaired partition, Q<=0 is an honest
+        # signal — a well-modularized repo reports Q~0.8, not 0.)
+        summary["verdict"] = f"{n_visible} trivial cluster(s), modularity Q={mod_q} — no community structure detected"
+        summary["state"] = "trivial_clustering"
+        summary["partial_success"] = True
     else:
-        verdict = "no clusters detected"
+        _empty = empty_corpus_state(conn)
+        if _empty is not None:
+            # Keep "no clusters" in the verdict (sealed contract) while
+            # disclosing the empty corpus via the canonical state + flag.
+            summary["verdict"] = "no clusters — no symbols indexed in corpus (run `roam index --force`)"
+            summary.update(_empty)  # state="empty_corpus", partial_success=True
+        elif not rows:
+            summary["verdict"] = "no clusters detected — no community structure found in the symbol graph"
+            summary["state"] = "no_clusters"
+            summary["partial_success"] = True
+        else:
+            summary["verdict"] = f"no clusters >= min-size {min_size} ({len(rows)} below threshold)"
+            summary["state"] = "below_min_size"
 
     envelope = json_envelope(
         "clusters",
-        summary={
-            "verdict": verdict,
-            "clusters": n_visible,
-            "mismatches": sum(1 for m in mismatches if m["cluster_id"] in visible_ids),
-            "modularity_q": quality["modularity"],
-            "mean_conductance": quality["mean_conductance"],
-        },
+        summary=summary,
         budget=token_budget,
         clusters=[
             {

src/roam/commands/cmd_complexity.py

@@ -617,7 +617,10 @@ def _merged_warnings() -> list[str]:
                 # W1298: use "functions" (LAW-4 concrete-noun anchor) over
                 # "total" so the auto-derived fact terminal is in the
                 # accepted set (test_w806_complexity_empty_corpus pin).
-                summary_norows: dict = {"verdict": verdict, "functions": 0}
+                # C3 (Pattern-2): explicit not_found state, matching the
+                # world-model siblings (side-effects/idempotency/causal-graph)
+                # so consumers switch on a field rather than parsing the verdict.
+                summary_norows: dict = {"verdict": verdict, "functions": 0, "state": "not_found"}
                 if _all_w:
                     summary_norows["partial_success"] = True
                 click.echo(

src/roam/commands/cmd_cut.py

@@ -240,6 +240,25 @@ def _detect_clusters():
         # composes against zero-counted cut analysis.
         def _compute_min_cuts():
             boundaries_local: list[dict] = []
+            # Pre-bucket cross-cluster edges in ONE pass over G.edges() (was an
+            # O(clusters^2 * |E|) rescan — the inner `for u,v in G.edges()` ran
+            # once per cluster pair). node_cluster maps node -> cluster id;
+            # pair_edges keys on the unordered cluster pair and preserves
+            # G.edges() order so the src/tgt seed pick below is unchanged.
+            from collections import defaultdict
+
+            node_cluster: dict = {}
+            for _cid, _nodes in cluster_nodes.items():
+                for _n in _nodes:
+                    node_cluster[_n] = _cid
+            pair_edges: dict[tuple, list[tuple]] = defaultdict(list)
+            for u, v in G.edges():
+                cu = node_cluster.get(u)
+                cv = node_cluster.get(v)
+                if cu is None or cv is None or cu == cv:
+                    continue
+                pair_edges[(cu, cv) if cu < cv else (cv, cu)].append((u, v))
+
             for i, c1 in enumerate(cluster_ids):
                 for c2 in cluster_ids[i + 1 :]:
                     if between:
@@ -252,13 +271,10 @@ def _compute_min_cuts():
                         ):
                             continue
 
-                    # Count cross-edges
-                    cross_edges: list[tuple] = []
+                    # Cross-edges from the pre-built bucket (no per-pair rescan).
                     nodes_c1 = cluster_nodes.get(c1, set())
                     nodes_c2 = cluster_nodes.get(c2, set())
-                    for u, v in G.edges():
-                        if (u in nodes_c1 and v in nodes_c2) or (u in nodes_c2 and v in nodes_c1):
-                            cross_edges.append((u, v))
+                    cross_edges: list[tuple] = pair_edges.get((c1, c2) if c1 < c2 else (c2, c1), [])
 
                     if not cross_edges:
                         continue
@@ -277,10 +293,19 @@ def _compute_min_cuts():
                     if src_node is None:
                         continue
 
-                    try:
-                        min_cut_edges = nx.minimum_edge_cut(UG, src_node, tgt_node)
-                        min_cut_size = len(min_cut_edges)
-                    except (nx.NetworkXError, nx.NetworkXUnbounded, nx.exception.NetworkXError):
+                    # Gate exact max-flow min-cut behind a subgraph-size cap —
+                    # nx.minimum_edge_cut is ~O(V*E^2). Above the cap, fall back
+                    # to the cross-edge-count heuristic (the same value the
+                    # except branch already uses) so huge cluster pairs do not
+                    # stall the command.
+                    if len(nodes_c1) + len(nodes_c2) <= 2000:
+                        try:
+                            min_cut_edges = nx.minimum_edge_cut(UG, src_node, tgt_node)
+                            min_cut_size = len(min_cut_edges)
+                        except (nx.NetworkXError, nx.NetworkXUnbounded, nx.exception.NetworkXError):
+                            min_cut_edges = set()
+                            min_cut_size = len(cross_edges)
+                    else:
                         min_cut_edges = set()
                         min_cut_size = len(cross_edges)
 
@@ -333,7 +358,15 @@ def _extract_leak_edges():
             if not (leak_edges or not between):
                 return leak_edge_list_local
             try:
-                ebc = nx.edge_betweenness_centrality(UG)
+                # Speed guard (mirrors graph/cycles.py's size gate): exact edge
+                # betweenness is O(V*E). k-sample pivots above a node-count cap;
+                # seed keeps the sampled leak-edge ranking reproducible.
+                _ug_n = UG.number_of_nodes()
+                if _ug_n <= 2000:
+                    ebc = nx.edge_betweenness_centrality(UG)
+                else:
+                    _k = min(_ug_n, max(200, int(_ug_n**0.5 * 5)))
+                    ebc = nx.edge_betweenness_centrality(UG, k=_k, seed=1)
             except Exception:
                 ebc = {}
 

src/roam/commands/cmd_cycles.py

@@ -15,7 +15,7 @@
 import click
 
 from roam.capability import roam_capability
-from roam.commands.resolve import ensure_index
+from roam.commands.resolve import empty_corpus_state, ensure_index
 from roam.db.connection import open_db
 from roam.graph.builder import build_symbol_graph
 from roam.graph.cycles import find_cycles, format_cycles, mark_actionable_cycles
@@ -58,6 +58,32 @@ def cycles(ctx, min_size, limit, actionable_only):
     token_budget = ctx.obj.get("budget", 0) if ctx.obj else 0
     ensure_index()
     with open_db(readonly=True) as conn:
+        # B3 (Pattern-2): a 0-symbol corpus must NOT report "clean dependency
+        # graph" — there is no graph to analyze. Disclose empty_corpus instead
+        # of a vacuous clean verdict.
+        _empty = empty_corpus_state(conn)
+        if _empty is not None:
+            empty_verdict = "no symbols indexed — no dependency graph to analyze (run `roam index --force`)"
+            if json_mode:
+                click.echo(
+                    to_json(
+                        json_envelope(
+                            "cycles",
+                            summary={
+                                "verdict": empty_verdict,
+                                "cycle_count": 0,
+                                "actionable_count": 0,
+                                **_empty,
+                            },
+                            cycles=[],
+                            budget=token_budget,
+                        )
+                    )
+                )
+            else:
+                click.echo(f"VERDICT: {empty_verdict}")
+            return
+
         graph = build_symbol_graph(conn)
         raw = find_cycles(graph, min_size=min_size)
         formatted = format_cycles(raw, conn) if raw else []

src/roam/commands/cmd_dashboard.py

@@ -587,6 +587,20 @@ def _assemble_sections():
                 if ai_rot_definition_top is not None:
                     _summary_block["ai_rot_definition"] = ai_rot_definition_top
 
+                # W805-PP (Pattern-2): a 0-symbol corpus must NOT read as a clean
+                # HEALTHY bill — there is nothing indexed to be healthy about
+                # (uncoded / not yet written / index broken / wrong cwd). The
+                # numeric health-band verdict ("Codebase is HEALTHY 100/100") is
+                # a silent SAFE here. Disclose the empty corpus explicitly via
+                # the canonical state + partial_success + an empty-naming verdict,
+                # matching cmd_health's guard and the shared empty_corpus_state.
+                if overview["symbols"] == 0:
+                    _summary_block["verdict"] = (
+                        "Codebase has 0 symbols indexed (empty corpus — run `roam index --force`)"
+                    )
+                    _summary_block["state"] = "empty_corpus"
+                    _summary_block["partial_success"] = True
+
                 _envelope_kwargs: dict = {
                     "overview": overview,
                     "health": {

src/roam/commands/cmd_dead.py

@@ -28,7 +28,7 @@
 # ``roam.index.test_conventions`` — no back-edge into ``cmd_dead``.
 from roam.commands.changed_files import is_test_file as _is_test_path
 from roam.commands.next_steps import format_next_steps_text, suggest_next_steps
-from roam.commands.resolve import ensure_index
+from roam.commands.resolve import empty_corpus_state, ensure_index
 from roam.db.connection import batched_in, find_project_root, open_db
 from roam.db.edge_kinds import CALL_EDGE_KINDS
 from roam.output.confidence import (
@@ -2295,6 +2295,23 @@ def _emit_empty_sarif():
                 if _combined_warnings_empty:
                     summary["warnings_out"] = list(_combined_warnings_empty)
                     summary["partial_success"] = True
+                # W802 Pattern-2: distinguish a genuinely empty corpus
+                # (0 symbols indexed) from a populated corpus that simply
+                # has no dead exports. On a 0-symbol corpus "no dead
+                # exports" is a vacuous SAFE — disclose the empty state
+                # explicitly via the canonical ``empty_corpus_state``
+                # helper so machine consumers read it without parsing the
+                # verdict. An empty corpus is a fully-resolved "nothing to
+                # flag" state, NOT a partial failure, so partial_success
+                # stays the literal boolean ``False`` (the helper's
+                # ``partial_success: True`` is intended for analysis
+                # commands whose graph logic degrades on empty input;
+                # ``dead`` produces a complete, correct answer here).
+                _empty_corpus = empty_corpus_state(conn)
+                if _empty_corpus is not None:
+                    summary["state"] = _empty_corpus["state"]
+                    summary["verdict"] = "no dead exports — no symbols indexed in corpus (run `roam index --force`)"
+                    summary.setdefault("partial_success", False)
                 envelope_kwargs: dict = {
                     "summary": summary,
                     "high_confidence": [],