scripts.check_kg_connectivity

Reference › scripts › check_kg_connectivity

The knowledge-graph connectivity gate, a read-only check from the session-254 ruling: every node must be reachable from a root :Project node. No module or test imports it; a human or gate runs it. It uses the Bolt repository from core.graph and the factory in graph_factory. The write-side counterpart that links nodes is backfill_thread_links.

Invocation: main takes connection options (--uri, --user and a password option) and --json. Missing connection settings are filled from the repository's dotenv file without overriding the ambient environment. The exit code is 0 when the graph is connected and 1 otherwise.

compute_connectivity counts nodes per label and runs a breadth-first search outward from the project nodes (_bfs_reachable), issuing queries in batches (_batched). It returns a ConnectivityReport with the reachable and unreachable counts, which can say whether it is ok and serialize itself. _top_unreachable_by_label names the labels with the most unreachable nodes (top 20), and unreachable_node_ids lists the ids themselves, optionally restricted to labels.

At a glance

Class diagram

Dependency diagram

API

Classes

ConnectivityReport (dataclass)

Decorators: dataclass(frozen=True) · Line 72

Result of one connectivity census.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/check_kg_connectivity.py

#!/usr/bin/env python3
"""KG connectivity gate — every node must lead to the root ``:Project`` node.

Session-254 ruling: EVERY node in the operational KG must be reachable
from the root ``:Project`` node (slug ``'orchestrata-ultimate'``, though this
gate seeds from ALL ``:Project`` nodes — the KG holds 100+ projects, see
``plans/architecture-ontology-draft.md``'s multi-project invariant). A WCC
census found 6,211 nodes in 2,654 disconnected islands; this script is the
repeatable, CI/health-check-wired version of that census.

Implementation: a plain-BFS (Python-side, batched Cypher adjacency queries)
outward from every ``:Project`` node, over undirected relationships (``--``,
matching the WCC definition of "connected" used by the original census) --
deliberately **engine-portable Cypher only** (no ``CALL gds.*``, no APOC),
since this must eventually run against Memgraph too, which has neither. Uses
``id(n)`` rather than ``elementId(n)`` for the same reason (``id()`` is the
older, more widely implemented builtin; Memgraph has no ``elementId()``).

**Read-only.** No write query is ever issued.

Usage::

    python scripts/check_kg_connectivity.py                # human-readable report
    python scripts/check_kg_connectivity.py --json          # machine-readable
    python scripts/check_kg_connectivity.py --uri bolt://10.53.1.13:7687

Exit code: 1 if any node is unreachable, 0 otherwise (data repair is a
separate, later effort -- today's live graph is EXPECTED to fail this gate).
"""

from __future__ import annotations

import argparse
import json
import sys
from collections.abc import Iterable, Mapping
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

_repo_src = Path(__file__).resolve().parent.parent / "src"
if str(_repo_src) not in sys.path:
    sys.path.insert(0, str(_repo_src))

import os  # noqa: E402

from neo4j import GraphDatabase  # noqa: E402

from orchestrata_ultimate.core.graph import BoltSyncRepository  # noqa: E402
from orchestrata_ultimate.core.graph_factory import resolve_uri  # noqa: E402
from orchestrata_ultimate.knowledge_graph.canonical_schema import (  # noqa: E402
    LABEL_PROJECT,
)

_DEFAULT_BATCH_SIZE = 1000


def _label_key(labels: list[str] | None) -> str:
    """A stable, display-friendly grouping key for a node's label set."""
    if not labels:
        return "(none)"
    return ",".join(sorted(labels))


def _batched(seq: Iterable[Any], size: int) -> Iterable[list[Any]]:
    items = list(seq)
    for i in range(0, len(items), size):
        yield items[i : i + size]


@dataclass(frozen=True)
class ConnectivityReport:
    """Result of one connectivity census."""

    total_nodes: int
    reachable: int
    unreachable: int
    unreachable_by_label: dict[str, int] = field(default_factory=dict)

    @property
    def ok(self) -> bool:
        return self.unreachable == 0

    def to_dict(self) -> dict[str, Any]:
        return {
            "total_nodes": self.total_nodes,
            "reachable": self.reachable,
            "unreachable": self.unreachable,
            "unreachable_by_label": self.unreachable_by_label,
        }


def compute_connectivity(
    repo: BoltSyncRepository,
    *,
    project_label: str = LABEL_PROJECT,
    batch_size: int = _DEFAULT_BATCH_SIZE,
) -> ConnectivityReport:
    """Run the BFS census. Pure read -- issues only ``MATCH``/``RETURN``.

    ``project_label``/``batch_size`` are test seams (mirroring
    ``backfill_thread_links.plan_backfill``'s ``source_labels`` param):
    production always uses the real
    :data:`~orchestrata_ultimate.knowledge_graph.canonical_schema.LABEL_PROJECT`.
    """
    total_rows = repo.execute_read("MATCH (n) RETURN count(n) AS cnt")
    total_nodes = int(total_rows[0]["cnt"]) if total_rows else 0

    total_by_label: dict[str, int] = {}
    for row in repo.execute_read(
        "MATCH (n) RETURN labels(n) AS labels, count(n) AS cnt"
    ):
        key = _label_key(row.get("labels"))
        total_by_label[key] = total_by_label.get(key, 0) + int(row["cnt"])

    visited = _bfs_reachable(repo, project_label=project_label, batch_size=batch_size)

    reachable = len(visited)
    unreachable = total_nodes - reachable

    reachable_by_label: dict[str, int] = {}
    for key in visited.values():
        reachable_by_label[key] = reachable_by_label.get(key, 0) + 1

    return ConnectivityReport(
        total_nodes=total_nodes,
        reachable=reachable,
        unreachable=unreachable,
        unreachable_by_label=_top_unreachable_by_label(
            total_by_label, reachable_by_label
        ),
    )


def _bfs_reachable(
    repo: BoltSyncRepository,
    *,
    project_label: str = LABEL_PROJECT,
    batch_size: int = _DEFAULT_BATCH_SIZE,
) -> dict[int, str]:
    """The BFS itself, factored out of :func:`compute_connectivity` so
    :func:`unreachable_node_ids` (hub-link repair, plans/hub-link-plan.md §8)
    can reuse the identical traversal rather than re-deriving it. Returns
    ``{id(n): label_key}`` for every node reached from a ``project_label``
    seed over undirected relationships."""
    seed_rows = repo.execute_read(
        f"MATCH (p:`{project_label}`) RETURN id(p) AS eid, labels(p) AS labels"
    )
    visited: dict[int, str] = {
        int(row["eid"]): _label_key(row.get("labels")) for row in seed_rows
    }
    frontier: list[int] = list(visited.keys())

    while frontier:
        next_frontier: dict[int, str] = {}
        for batch in _batched(frontier, batch_size):
            rows = repo.execute_read(
                "MATCH (n)--(m) WHERE id(n) IN $ids "
                "RETURN DISTINCT id(m) AS eid, labels(m) AS labels",
                {"ids": batch},
            )
            for row in rows:
                eid = int(row["eid"])
                if eid not in visited and eid not in next_frontier:
                    next_frontier[eid] = _label_key(row.get("labels"))
        if not next_frontier:
            break
        visited.update(next_frontier)
        frontier = list(next_frontier.keys())

    return visited


def unreachable_node_ids(
    repo: BoltSyncRepository,
    *,
    project_label: str = LABEL_PROJECT,
    labels: Iterable[str] | None = None,
    batch_size: int = _DEFAULT_BATCH_SIZE,
) -> dict[str, list[int]]:
    """Per-label ``id(n)`` lists for nodes NOT reachable from any
    ``project_label`` seed (hub-link repair, plans/hub-link-plan.md §8: "the
    script computes the orphan set ONCE, up front, then applies DELETEs,
    then LINKs"). Read-only; reuses the exact same BFS
    :func:`compute_connectivity` runs rather than a re-derived traversal.

    ``labels`` optionally restricts the RESULT to a subset of label keys
    (e.g. just the LINK-rule owners) -- the traversal itself is never
    label-scoped, so a label filter can never silently miss a transitive
    rescue (e.g. GatewayAuditEvent riding in on Session's own fix).
    """
    visited = _bfs_reachable(repo, project_label=project_label, batch_size=batch_size)
    result: dict[str, list[int]] = {}
    for row in repo.execute_read("MATCH (n) RETURN id(n) AS eid, labels(n) AS labels"):
        eid = int(row["eid"])
        if eid in visited:
            continue
        label = _label_key(row.get("labels"))
        if labels is not None and label not in labels:
            continue
        result.setdefault(label, []).append(eid)
    return result


def _top_unreachable_by_label(
    total_by_label: Mapping[str, int],
    reachable_by_label: Mapping[str, int],
    *,
    top_n: int = 20,
) -> dict[str, int]:
    """``total_by_label - reachable_by_label`` per label, positive diffs
    only, sorted descending and capped to *top_n* -- the "top 20" of the
    report. A pure function (no DB access) so the aggregation/truncation
    logic is unit-testable without a live graph drowning small counts out
    of the top 20 (a real risk here: this repo's live KG has hundreds of
    thousands of unreachable CXPPacket nodes)."""
    diffs = {
        label: total_by_label.get(label, 0) - reachable_by_label.get(label, 0)
        for label in total_by_label
    }
    diffs = {k: v for k, v in diffs.items() if v > 0}
    return dict(sorted(diffs.items(), key=lambda kv: kv[1], reverse=True)[:top_n])


def _driver(args: argparse.Namespace, env: Mapping[str, str] | None = None):
    env = env if env is not None else os.environ
    uri = args.uri or resolve_uri(env=env)
    user = args.user or ""
    password = args.password or ""
    return GraphDatabase.driver(uri, auth=(user, password))


def _print_report(report: ConnectivityReport) -> None:
    print("KG Connectivity Report")
    print("=======================")
    print(f"Total nodes  : {report.total_nodes}")
    print(f"Reachable    : {report.reachable}")
    print(f"Unreachable  : {report.unreachable}")
    if report.unreachable_by_label:
        print("Unreachable by label (top 20):")
        for label, count in report.unreachable_by_label.items():
            print(f"  {label:<30} : {count}")
    else:
        print("Unreachable by label: none")


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="check_kg_connectivity",
        description=(
            "KG connectivity gate: every node must be reachable from a "
            "root :Project node (session-254 ruling). Read-only."
        ),
    )
    parser.add_argument("--uri", default=None)
    parser.add_argument("--user", default=None)
    parser.add_argument("--password", default=None)
    parser.add_argument(
        "--json", action="store_true", help="Emit a machine-readable JSON report."
    )
    args = parser.parse_args(argv)

    # Standalone usability: fill NEO4J_* from the repo .env when the
    # caller's environment lacks them (override=False -- ambient env always
    # wins), mirroring backfill_thread_links.main / canonical_conformance.main.
    from dotenv import load_dotenv

    load_dotenv(Path(__file__).resolve().parent.parent / ".env", override=False)

    driver = _driver(args)
    try:
        driver.verify_connectivity()
        repo = BoltSyncRepository(driver=driver)
        report = compute_connectivity(repo)
    finally:
        driver.close()

    if args.json:
        print(json.dumps(report.to_dict(), indent=2))
    else:
        _print_report(report)

    return 0 if report.ok else 1


if __name__ == "__main__":
    sys.exit(main())