feat(diagnostics): canonical-graph validation gate - identity, direction, hyperedge schema, determinism

graphify/cli.py

@@ -1479,7 +1479,8 @@ def dispatch_command(cmd: str) -> None:
             print(
                 "Usage: graphify diagnose multigraph "
                 "[--graph path] [--json] [--max-examples N] "
-                "[--directed] [--undirected] [--extract-path path]",
+                "[--directed] [--undirected] [--extract-path path] "
+                "[--fail-on-noncanonical]",
                 file=sys.stderr,
             )
             sys.exit(1)
@@ -1490,6 +1491,10 @@ def dispatch_command(cmd: str) -> None:
         direction_flag: str | None = None
         json_output = False
         extract_path: Path | None = None
+        # Opt-in, not the default: existing scripts/CI that already run this
+        # command must keep getting exit 0 on a non-canonical (but structurally
+        # loadable) graph unless they explicitly ask to gate on it.
+        fail_on_noncanonical = False
 
         i = 3
         while i < len(sys.argv):
@@ -1539,6 +1544,8 @@ def dispatch_command(cmd: str) -> None:
                     print("error: --extract-path requires a path", file=sys.stderr)
                     sys.exit(1)
                 extract_path = Path(sys.argv[i])
+            elif arg == "--fail-on-noncanonical":
+                fail_on_noncanonical = True
             else:
                 print(f"error: unknown diagnose option {arg}", file=sys.stderr)
                 sys.exit(1)
@@ -1567,6 +1574,9 @@ def dispatch_command(cmd: str) -> None:
         else:
             print(format_diagnostic_report(summary))
 
+        if fail_on_noncanonical and not summary.get("canonical", True):
+            sys.exit(1)
+
     elif cmd == "add":
         if len(sys.argv) < 3:
             print(

graphify/diagnostics.py

@@ -11,6 +11,7 @@
 
 import networkx as nx
 
+from graphify.validate import validate_extraction
 
 _SUPPRESSION_DECL_RE = re.compile(r"^\s*(?P<name>seen_[A-Za-z0-9_]+)\s*[:=]")
 _TYPE_TUPLE_RE = re.compile(r"set\[tuple\[(?P<inside>[^\]]+)\]\]")
@@ -153,6 +154,143 @@ def scan_producer_suppression_sites(path: str | Path) -> dict[str, Any]:
     }
 
 
+# Relations where direction is citer -> citee between whole-file document/paper
+# nodes (extraction-spec.md's "references"/"cites" direction rule for docs,
+# mirroring the "calls" direction rule for code). Deliberately excludes "calls"
+# and other relation types, whose direction semantics differ.
+_CITATION_RELATIONS = frozenset({"references", "cites"})
+_WHOLE_FILE_TYPES = frozenset({"document", "paper"})
+
+
+def _find_edge_direction_suspects(
+    extraction: dict[str, Any], node_ids: set[str]
+) -> list[dict[str, Any]]:
+    """Flag references/cites edges whose own source_file matches the TARGET
+    node's file rather than the SOURCE node's - the extraction-spec.md
+    self-check for a reversed citation edge (the file that made the assertion
+    is the target, meaning the target cited the source, not the other way
+    round).
+
+    High precision by construction: a node's source_file is where that node
+    was authored; an edge's source_file is where the assertion was found. When
+    those disagree in exactly this way, the edge was very likely recorded
+    backwards - the bug class that inflated a real corpus's ARCHITECTURE.md/
+    PRODUCT.md "god node" degree with edges actually asserted by the ADRs
+    citing them, not by the docs themselves. Scoped to document/paper whole-
+    file nodes on both ends, matching the spec rule this check enforces.
+    """
+    nodes = extraction.get("nodes", [])
+    if not isinstance(nodes, list):
+        return []
+    file_by_id: dict[str, str] = {}
+    type_by_id: dict[str, str] = {}
+    ids_by_file: dict[str, set[str]] = defaultdict(set)
+    for node in nodes:
+        if not isinstance(node, dict):
+            continue
+        nid = node.get("id")
+        sf = node.get("source_file")
+        if not isinstance(nid, str) or not isinstance(sf, str) or not sf:
+            continue
+        file_by_id[nid] = sf
+        type_by_id[nid] = node.get("file_type")
+        ids_by_file[sf].add(nid)
+
+    suspects: list[dict[str, Any]] = []
+    for edge in _edge_list(extraction):
+        if not isinstance(edge, dict) or edge.get("relation") not in _CITATION_RELATIONS:
+            continue
+        source = edge.get("source", edge.get("from"))
+        target = edge.get("target", edge.get("to"))
+        edge_source_file = edge.get("source_file")
+        if not (
+            isinstance(source, str) and isinstance(target, str)
+            and source in node_ids and target in node_ids
+            and source in file_by_id  # source node's own file must be known
+            and type_by_id.get(source) in _WHOLE_FILE_TYPES
+            and type_by_id.get(target) in _WHOLE_FILE_TYPES
+            and isinstance(edge_source_file, str) and edge_source_file
+        ):
+            continue
+        if file_by_id[source] == edge_source_file:
+            continue  # matches the source node's own file - direction looks right.
+        if target in ids_by_file.get(edge_source_file, ()):
+            suspects.append(
+                {
+                    "source": source,
+                    "target": target,
+                    "relation": edge["relation"],
+                    "edge_source_file": edge_source_file,
+                    "source_node_file": file_by_id[source],
+                }
+            )
+    return suspects
+
+
+def _find_duplicate_whole_file_candidates(
+    extraction: dict[str, Any],
+) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    """Flag document/paper nodes that probably represent the same real file
+    under two different ids - a canonical-identity violation.
+
+    Returns (hard, soft):
+      - hard: id B == id A + "_document" (or "_file"/"_page") - the exact
+        whole-file-node suffix a subagent must never invent (extraction-spec.md
+        "Whole-file nodes"), almost never a coincidence.
+      - soft: labels look like the same file (one is the other's exact prefix
+        up to a " - "/" -- "/" (" separator, e.g. "ARCHITECTURE.md" vs
+        "ARCHITECTURE.md - Avenoria Technical Architecture") on two nodes with
+        different source_file - plausible but not certain, kept separate from
+        `hard` so it never blocks canonical status on its own.
+    """
+    nodes = [
+        n for n in extraction.get("nodes", [])
+        if isinstance(n, dict) and n.get("file_type") in _WHOLE_FILE_TYPES
+        and isinstance(n.get("id"), str) and isinstance(n.get("label"), str)
+    ]
+    by_id = {n["id"]: n for n in nodes}
+
+    hard: list[dict[str, Any]] = []
+    hard_pairs: set[tuple[str, str]] = set()
+    for n in nodes:
+        for suffix in ("_document", "_file", "_page"):
+            candidate = n["id"] + suffix
+            other = by_id.get(candidate)
+            if other is None:
+                continue
+            pair = tuple(sorted((n["id"], other["id"])))
+            if pair in hard_pairs:
+                continue
+            hard_pairs.add(pair)
+            hard.append(
+                {"node_a": n["id"], "node_b": other["id"], "reason": f"id suffix {suffix!r}"}
+            )
+
+    soft: list[dict[str, Any]] = []
+    seen_soft: set[tuple[str, str]] = set()
+    separators = (" - ", " -- ", " (")
+    for n in nodes:
+        for m in nodes:
+            if n["id"] >= m["id"]:
+                continue
+            pair = (n["id"], m["id"])
+            if pair in hard_pairs or n.get("source_file") == m.get("source_file"):
+                continue  # already caught by the id-suffix check, or same file
+                          # (e.g. a heading node) - not a whole-file duplicate.
+            shorter, longer = sorted((n["label"], m["label"]), key=len)
+            if not longer.startswith(shorter):
+                continue
+            rest = longer[len(shorter):]
+            if not any(rest.startswith(sep) for sep in separators):
+                continue
+            if pair in seen_soft:
+                continue
+            seen_soft.add(pair)
+            soft.append({"node_a": n["id"], "node_b": m["id"], "reason": "label prefix"})
+
+    return hard, soft
+
+
 def diagnose_extraction(
     extraction: dict[str, Any],
     *,
@@ -161,12 +299,27 @@ def diagnose_extraction(
     max_examples: int = 5,
     extract_path: str | Path | None = None,
 ) -> dict[str, Any]:
-    """Summarize same-endpoint edge-collapse risk for one JSON graph/extraction dict."""
+    """Summarize same-endpoint edge-collapse risk for one JSON graph/extraction dict.
+
+    Also runs the graph-validity gate this function's docstring didn't used to
+    cover: schema errors, reversed citation edges, and probable duplicate
+    whole-file nodes. The `canonical` field is the answer to "can architectural
+    metrics (centrality, clustering, dependency analysis) computed from this
+    graph be trusted, or are they informational only until the issues below are
+    fixed?" A non-canonical graph is not corrupt or unusable - it can still be
+    built and browsed - but its structural metrics may not reflect the real
+    architecture and should be caveated accordingly.
+    """
     from graphify.build import build_from_json
 
     node_ids = _node_ids(extraction)
     raw_edges = _edge_list(extraction)
     canonical_edges = [_canonical_edge(edge) for edge in raw_edges]
+    schema_errors = validate_extraction(extraction)
+    edge_direction_suspects = _find_edge_direction_suspects(extraction, node_ids)
+    duplicate_node_candidates, duplicate_node_candidates_soft = (
+        _find_duplicate_whole_file_candidates(extraction)
+    )
 
     # Code-typed semantic nodes the extractor could not verify against the source
     # it read (#1949): likely-inferred (or hallucinated) symbols surfaced from a
@@ -245,7 +398,36 @@ def diagnose_extraction(
         Path(extract_path) if extract_path else Path(__file__).with_name("extract.py")
     )
 
+    # What blocks `canonical`: schema errors, dangling/missing edge endpoints,
+    # reversed citation edges, and hard (id-suffix) duplicate whole-file nodes.
+    # What does NOT block it (informational only, per the report/warnings
+    # below): edge-collapse counts, self-loops, and the soft (label-prefix)
+    # duplicate candidates - each has a plausible benign explanation on its
+    # own, so they are visibility, not proof.
+    canonical_issues: list[str] = []
+    if schema_errors:
+        canonical_issues.append(f"{len(schema_errors)} schema error(s)")
+    if missing_endpoint_edges:
+        canonical_issues.append(f"{missing_endpoint_edges} edge(s) with a missing endpoint")
+    if dangling_endpoint_edges:
+        canonical_issues.append(f"{dangling_endpoint_edges} edge(s) with a dangling endpoint")
+    if edge_direction_suspects:
+        canonical_issues.append(
+            f"{len(edge_direction_suspects)} references/cites edge(s) with a likely-reversed direction"
+        )
+    if duplicate_node_candidates:
+        canonical_issues.append(
+            f"{len(duplicate_node_candidates)} probable duplicate whole-file node pair(s)"
+        )
+    canonical = not canonical_issues
+
     return {
+        "canonical": canonical,
+        "canonical_issues": canonical_issues,
+        "schema_errors": schema_errors,
+        "edge_direction_suspects": edge_direction_suspects,
+        "duplicate_node_candidates": duplicate_node_candidates,
+        "duplicate_node_candidates_soft": duplicate_node_candidates_soft,
         "node_count": len(node_ids),
         "unverified_node_count": unverified_node_count,
         "raw_edge_count": len(raw_edges),
@@ -347,8 +529,17 @@ def format_diagnostic_json(summary: dict[str, Any]) -> dict[str, Any]:
 
 def format_diagnostic_report(summary: dict[str, Any]) -> str:
     suppression = summary.get("producer_suppression", {})
+    canonical = summary.get("canonical", True)
+    verdict = "CANONICAL" if canonical else "NON-CANONICAL"
     lines = [
         "[graphify] MultiDiGraph edge-collapse diagnostic",
+        f"verdict: {verdict}"
+        + (
+            ""
+            if canonical
+            else " - treat centrality/clustering/dependency metrics as informational "
+            "only until the issues below are fixed"
+        ),
         f"input: {summary.get('input_path', '<in-memory>')}",
         "input_stage: provided JSON (normal graph.json is post-build)",
         f"effective_directed: {summary.get('effective_directed', '<direct-call>')}",
@@ -358,6 +549,10 @@ def format_diagnostic_report(summary: dict[str, Any]) -> str:
         f"valid_candidate_edges: {summary['valid_candidate_edges']}",
         f"missing_endpoint_edges: {summary['missing_endpoint_edges']}",
         f"dangling_endpoint_edges: {summary['dangling_endpoint_edges']}",
+        f"schema_errors: {len(summary.get('schema_errors', []))}",
+        f"edge_direction_suspects: {len(summary.get('edge_direction_suspects', []))}",
+        f"duplicate_node_candidates: {len(summary.get('duplicate_node_candidates', []))}"
+        f" (+{len(summary.get('duplicate_node_candidates_soft', []))} soft)",
         f"self_loop_edges: {summary['self_loop_edges']}",
         f"exact_duplicate_edges: {summary['exact_duplicate_edges']}",
         f"directed_unique_endpoint_pairs: {summary['directed_unique_endpoint_pairs']}",
@@ -379,6 +574,26 @@ def format_diagnostic_report(summary: dict[str, Any]) -> str:
         f"post_build_edges: {summary['post_build_edge_count']}",
         f"producer_suppression_sites: {suppression.get('total_sites', 0)}",
     ]
+    if summary.get("schema_errors"):
+        lines.append("schema_errors:")
+        for error in summary["schema_errors"][:8]:
+            lines.append(f"  - {error}")
+    if summary.get("edge_direction_suspects"):
+        lines.append("edge_direction_suspects (likely reversed references/cites):")
+        for s in summary["edge_direction_suspects"][:8]:
+            lines.append(
+                f"  - {s['source']} --{s['relation']}--> {s['target']} "
+                f"(edge asserted in {s['edge_source_file']!r}, which is {s['target']}'s own file, "
+                f"not {s['source']}'s {s['source_node_file']!r})"
+            )
+    if summary.get("duplicate_node_candidates"):
+        lines.append("duplicate_node_candidates (probable same-file split, id suffix):")
+        for d in summary["duplicate_node_candidates"][:8]:
+            lines.append(f"  - {d['node_a']} <-> {d['node_b']} ({d['reason']})")
+    if summary.get("duplicate_node_candidates_soft"):
+        lines.append("duplicate_node_candidates_soft (label prefix, needs a human glance):")
+        for d in summary["duplicate_node_candidates_soft"][:8]:
+            lines.append(f"  - {d['node_a']} <-> {d['node_b']} ({d['reason']})")
     if summary.get("post_build_error"):
         lines.append(f"post_build_error: {summary['post_build_error']}")
     if suppression.get("error"):

graphify/skill-agents.md

@@ -452,7 +452,7 @@ Replace INPUT_PATH with the actual path.
 
 ### Step 4.5 - Graph health check (read-only integrity gate)
 
-A non-destructive diagnostic on the extraction, before labeling. It surfaces edge collapse, dangling/missing endpoints, and self-loops — the silent-corruption modes of incremental updates and AST/LLM id mismatches. Read-only; never aborts.
+A non-destructive diagnostic on the extraction, before labeling. It surfaces edge collapse, dangling/missing endpoints, self-loops, schema errors, likely-reversed citation edges, and probable duplicate whole-file nodes (canonical-identity violations) — the silent-corruption modes of incremental updates and AST/LLM id mismatches. Read-only; never aborts.
 
 ```bash
 $(cat graphify-out/.graphify_python) -c "
@@ -471,11 +471,19 @@ flags = [f'{summary[k]} {label}' for k, label in (
     ('undirected_same_endpoint_collapsed_edges', 'collapsed (undirected) edges'),
 ) if summary.get(k, 0)]
 print('GRAPH HEALTH WARNING: ' + '; '.join(flags) + ' - graph may be incomplete/corrupt.' if flags else 'Graph health: OK (no dangling/missing/collapsed edges).')
+is_canonical = summary.get('canonical', True)
+issues_text = '; '.join(summary.get('canonical_issues', []))
+canonical_line = 'CANONICAL: ' + str(is_canonical)
+if not is_canonical:
+    canonical_line += ' - ' + issues_text
+print(canonical_line)
 "
 ```
 
 Substitute `IS_DIRECTED` and `INPUT_PATH` as in Step 4. If a `GRAPH HEALTH WARNING` prints, surface it in the final summary (do not abort — the graph is still usable, but the integrity issue must be visible, per the Honesty Rules).
 
+**If `CANONICAL: False`:** the graph has a structural defect stronger than a collapse warning — a schema error, a dangling/missing edge endpoint, a references/cites edge that is very likely drawn backwards, or two nodes that are probably the same real file under two different ids (see `duplicate_node_candidates` in the report). Say so plainly in the final summary, and while it stays non-canonical: treat god-node/centrality rankings, community/clustering results, and dependency-direction claims as **informational only**, not settled fact — a node's inflated apparent importance may be an artifact of exactly this kind of split, not real architectural coupling. Do not silently present god-node degree or "X is the most central node" as fact against a non-canonical graph; name the specific defect instead. `duplicate_node_candidates_soft` (label-prefix matches without the id-suffix pattern) does not by itself flip `canonical` to False — treat it as a lead worth a glance, not a proven defect.
+
 ### Step 5 - Label communities
 
 Read `graphify-out/.graphify_analysis.json`. For each community key, look at its node labels and write a 2-5 word plain-language name (e.g. "Attention Mechanism", "Training Pipeline", "Data Loading").