scripts.repair_orphan_hub_links

Reference › scripts › repair_orphan_hub_links

scripts/repair_orphan_hub_links.py repairs the knowledge-graph root-connectivity invariant from the session-254 ruling: every node must lead to the root Project node. It links the orphans that can be resolved and deletes narrowly keyed test residue. Its read-only counterpart is check_kg_connectivity; see also repair_project_hubs and backfill_thread_links. A human runs it. It uses orchestrata_ultimate.core.graph, orchestrata_ultimate.core.graph_factory, orchestrata_ultimate.knowledge_graph and orchestrata_ultimate.knowledge_graph.canonical_schema.

compute_orphans precomputes the orphan set that every rule reads from. LINK rules (for example link_sessions, link_parameters and the opt-in Spec rule) attach orphans to hubs; _require_constant enforces RUL-016, so a LINK rule must cite a constant from the canonical schema. DELETE rules are run by apply_deletes, and check_delete_drift compares the actual count per DELETE rule, scoped to the precomputed orphan ids, with what was planned. plan_spec_new_projects lists project slugs that would need new Project hubs, which only --allow-new-projects permits. Work is batched (_batched). Options: --uri, --user, --password, --project (default from _project_slug), --apply (default is a dry run), --include-spec-hubs (the Spec LINK rule, off by default) and --allow-new-projects. Run directly, main loads .env without overriding the ambient environment so the graph URI resolves.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

DeleteRule (dataclass)

Decorators: dataclass(frozen=True) · Line 145

A narrowly-keyed DELETE verdict, scoped to the precomputed orphan set.

RunReport (dataclass)

Decorators: dataclass · Line 190

Accumulated plan/execution output -- returned for tests, printed for humans.

Functions

Source

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

#!/usr/bin/env python3
"""Repair the KG root-connectivity invariant (session-254 ruling): every node
must be reachable from a root ``:Project`` node.

Measured live 2026-07-27: 8447 of 64913 nodes unreachable
(``scripts/check_kg_connectivity.py``). ``plans/hub-link-plan.md`` §4
classifies every orphaned label as **LINK** (a resolvable anchor already
exists, the edge was simply never written) or **DELETE** (narrowly-keyed
test/fixture residue). This script applies both verdicts.

Design (``plans/hub-link-plan.md`` §8, BUILDER-BRIEF.md §8):

- **Dry-run by default.** ``--apply`` to write. Every line is prefixed
  ``[WOULD]`` (dry-run) or ``[DID]`` (applied).
- **Idempotent.** Every write is a ``MERGE`` (or a narrowly-scoped
  ``DETACH DELETE`` gated on an exact-count check) — re-running after a
  successful apply changes nothing.
- **The orphan set is computed ONCE, up front**, via
  ``check_kg_connectivity.unreachable_node_ids`` (the identical BFS the
  connectivity gate itself runs — never a re-derived traversal), *then*
  DELETE rules are checked/applied, *then* LINK rules. This ordering is
  load-bearing: the 22 test ``:Fact`` orphans stop being orphans the
  instant their ``:Session`` gets its hub edge, so recomputing per-rule
  would silently skip them.
- **Every LINK rule cites the canonical_schema constant it applies** —
  printed in the plan output, and a rule whose constant is empty refuses to
  run (RUL-016 enforced in code, not just prose).
- **Every DELETE rule's predicate is checked against its expected count
  first.** A drifted count means the graph changed under the plan and the
  verdict needs re-deriving by a human, not a wider predicate — the script
  aborts (no writes at all) rather than guessing.
- **Spec is the one LINK rule that can mint NEW ``:Project`` hubs** (one per
  distinct ``project_slug`` among orphan Specs). Two of the seven generated
  slugs are degenerate (``project_slug: null``, and the slugified
  instruction ``"implement-as-specified"``) -- DECISION D1 in the brief is
  unresolved, so the whole Spec rule is OFF by default
  (``--include-spec-hubs`` to enable) and additionally refuses to run if it
  would mint more than one new hub without ``--allow-new-projects``. A
  ``project_slug`` of ``null``/``''`` is NEVER minted, with or without
  either flag -- that is not a design call, it is the same invariant
  ``kg_design._require_project_slug`` already enforces at write time.
- **The Thread family (Intent/PipelineRun/Artifact/CXPPacket/Message/
  ToolCall) delegates to** ``backfill_thread_links.run_backfill`` (extended
  in this same wave to cover Intent/PipelineRun) rather than re-implementing
  Thread-minting Cypher. That function keys off ``thread_id`` presence, not
  orphan status, so it is safe to call regardless of ordering, and it keeps
  ``backfill_thread_links.check_hub``'s guard: it MATCHes (never MERGEs)
  the ``:Project`` hub for the given ``--project``, aborting rather than
  minting one.
- **UserStory/Requirement(non-husk)/GatewayAuditEvent get no direct rule.**
  They are rescued transitively (via the Spec and Session fixes
  respectively). ``--apply`` verifies this explicitly at the end instead of
  assuming it.
- **Connection is read from the environment at call time** —
  ``KG_ROUTING_URI``/``KG_URI`` (no hardcoded default, RUL-011); database
  routing is ``core.graph_factory.resolve_kg_database()`` (implicit in every
  ``BoltSyncRepository`` call made here).
- **Never run with ``--apply`` against the live production graph from this
  worktree** — the orchestrator performs the single production run.

Usage::

    python scripts/repair_orphan_hub_links.py                    # dry run
    python scripts/repair_orphan_hub_links.py --apply
    python scripts/repair_orphan_hub_links.py --apply \\
        --include-spec-hubs --allow-new-projects
    python scripts/repair_orphan_hub_links.py --uri bolt://10.53.1.13:7687
"""

from __future__ import annotations

import argparse
import importlib.util
import os
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, cast

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

from neo4j import Driver, 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 import canonical_schema as cs  # noqa: E402

# ---------------------------------------------------------------------------
# Load the sibling scripts by file path (same technique this repo's own test
# suite already uses for these two modules -- see tests/scripts/
# test_check_kg_connectivity.py / test_backfill_thread_links.py). This keeps
# the import correct under BOTH direct CLI invocation and pytest, rather
# than relying on ambient sys.path[0] behaving differently under each.
# ---------------------------------------------------------------------------


def _load_sibling_script(name: str) -> Any:
    mod_path = Path(__file__).resolve().parent / f"{name}.py"
    spec = importlib.util.spec_from_file_location(name, mod_path)
    if spec is None or spec.loader is None:
        raise ImportError(f"cannot load sibling script {name!r} from {mod_path}")
    module = importlib.util.module_from_spec(spec)
    # Register before exec: both sibling modules declare frozen dataclasses
    # under `from __future__ import annotations` -- dataclasses' own
    # string-annotation resolution looks the module up via
    # `sys.modules[cls.__module__]`, which is None (and crashes) unless the
    # module is registered under its own name before exec_module runs.
    sys.modules[name] = module
    spec.loader.exec_module(module)
    return module


check_kg_connectivity = _load_sibling_script("check_kg_connectivity")
backfill_thread_links = _load_sibling_script("backfill_thread_links")

DEFAULT_PROJECT_SLUG = "orchestrata-ultimate"
_BATCH_SIZE = 1000

# Labels the DELETE + (non-Thread-family) LINK rules operate on. The orphan
# set is computed ONCE for exactly these labels, up front, before any write.
_ORPHAN_LABELS = (
    "Parameter",
    "Session",
    "Spec",
    "Fact",
    "Action",
    "Timeout",
    "Resolution",
    "Requirement",
)

# Labels rescued transitively -- no direct rule, verified post-apply.
_TRANSITIVE_LABELS = ("UserStory", "Requirement", "GatewayAuditEvent")

# DECISION D1 (b), ratified via the 2026-07-30 fix-all-defects goal: these
# generated "project" slugs are slugified INSTRUCTIONS, not project names --
# their whole orphan tree (Spec + attached Requirement/UserStory) is deleted,
# never minted as a :Project hub.
_DEGENERATE_SPEC_SLUGS: tuple[str, ...] = ("implement-as-specified",)


@dataclass(frozen=True)
class DeleteRule:
    """A narrowly-keyed DELETE verdict, scoped to the precomputed orphan set."""

    label: str
    predicate: str  # Cypher WHERE fragment referencing node var `n`
    expected: int
    evidence: str


DELETE_RULES: tuple[DeleteRule, ...] = (
    DeleteRule(
        "Fact",
        "n.source = 'test'",
        22,
        "all 88 :Fact nodes have source='test'; data is {'i': N} loop fixtures",
    ),
    DeleteRule(
        "Action",
        "n.agent IN ['pytest', '']",
        8,
        "all 8: pytest/'edited module.py'/file_edit x4, ''/'picked option A'/decision x4; "
        "log_action has no production caller",
    ),
    DeleteRule(
        "Timeout",
        "n.data CONTAINS '_TEST_'",
        4,
        "all 4 carry _TEST_<hex>_esc escalation ids",
    ),
    DeleteRule(
        "Resolution",
        "n.conflict_id = 'c-resolve-neo4j'",
        3,
        "all 3 share that fixture conflict id and identical resolution_text",
    ),
    DeleteRule(
        "Requirement",
        "n.id = 'REQ-008-023' AND size(keys(n)) = 1",
        1,
        "zero-degree, single-property husk; no canonical twin exists",
    ),
)


@dataclass
class RunReport:
    """Accumulated plan/execution output -- returned for tests, printed for humans."""

    lines: list[str] = field(default_factory=list)
    aborted: bool = False

    def log(self, line: str) -> None:
        self.lines.append(line)
        print(line)


def _require_constant(report: RunReport, rule_name: str, constant_name: str) -> str:
    """RUL-016 enforcement: a LINK rule with no cited canonical_schema
    constant refuses to run. Returns the constant's value on success."""
    value = getattr(cs, constant_name, "")
    if not value:
        report.log(
            f"[ABORT] LINK rule {rule_name!r} cites no canonical_schema constant "
            f"({constant_name!r} is empty/undefined) -- refusing to run (RUL-016)."
        )
        raise SystemExit(1)
    return value


def _batched(seq: list[int], size: int = _BATCH_SIZE) -> list[list[int]]:
    return [seq[i : i + size] for i in range(0, len(seq), size)]


def _project_slug(env: dict[str, str] | None = None) -> str:
    env = env if env is not None else dict(os.environ)
    return env.get("ORCHESTRATA_PROJECT_ID", DEFAULT_PROJECT_SLUG)


# ---------------------------------------------------------------------------
# Orphan-set computation (once, up front)
# ---------------------------------------------------------------------------


def compute_orphans(repo: BoltSyncRepository) -> dict[str, list[int]]:
    """The precomputed orphan set every rule below reads from -- never
    recomputed mid-run (see module docstring)."""
    # cast: check_kg_connectivity is importlib-loaded (scripts/ is not a
    # package), so its attributes type as Any.
    return cast(
        "dict[str, list[int]]",
        check_kg_connectivity.unreachable_node_ids(repo, labels=set(_ORPHAN_LABELS)),
    )


# ---------------------------------------------------------------------------
# DELETE rules
# ---------------------------------------------------------------------------


def check_delete_drift(
    repo: BoltSyncRepository, orphans: dict[str, list[int]]
) -> dict[str, int]:
    """Actual count per DELETE rule, scoped to the precomputed orphan ids.
    Pure read -- never mutates."""
    actual: dict[str, int] = {}
    for rule in DELETE_RULES:
        ids = orphans.get(rule.label, [])
        if not ids:
            actual[rule.label] = 0
            continue
        rows = repo.execute_read(
            f"MATCH (n:{rule.label}) WHERE id(n) IN $ids AND {rule.predicate} "
            "RETURN count(n) AS c",
            {"ids": ids},
        )
        actual[rule.label] = int(rows[0]["c"]) if rows else 0
    return actual


def apply_deletes(
    repo: BoltSyncRepository,
    orphans: dict[str, list[int]],
    drift: dict[str, int] | None = None,
    *,
    execute: bool,
    report: RunReport,
) -> None:
    verb = "DID" if execute else "WOULD"
    for rule in DELETE_RULES:
        ids = orphans.get(rule.label, [])
        if not ids:
            report.log(f"[{verb}] {rule.label}: delete 0 ({rule.evidence})")
            continue
        if execute:
            rows = repo.execute_write(
                f"MATCH (n:{rule.label}) WHERE id(n) IN $ids AND {rule.predicate} "
                "DETACH DELETE n RETURN count(n) AS c",
                {"ids": ids},
            )
            deleted = int(rows[0]["c"]) if rows else 0
            report.log(f"[DID] {rule.label}: deleted {deleted} ({rule.evidence})")
        else:
            # Report the MEASURED match count (0 after a prior apply), never
            # the historical expectation -- the plan must not overstate.
            would = drift.get(rule.label, 0) if drift is not None else rule.expected
            report.log(f"[WOULD] {rule.label}: delete {would} ({rule.evidence})")


# ---------------------------------------------------------------------------
# LINK rules — Parameter, Session, Spec
# ---------------------------------------------------------------------------


def link_parameters(
    repo: BoltSyncRepository,
    orphans: dict[str, list[int]],
    *,
    execute: bool,
    report: RunReport,
) -> None:
    constant = _require_constant(report, "Parameter", "REL_HAS_PARAMETER")
    ids = orphans.get("Parameter", [])
    verb = "DID" if execute else "WOULD"
    if not ids:
        report.log(f"[{verb}] Parameter: link 0 via canonical_schema.{constant!r}")
        return
    if not execute:
        report.log(
            f"[WOULD] Parameter: link {len(ids)} via canonical_schema.REL_HAS_PARAMETER "
            "((:Function|:Method)-[:HAS_PARAMETER]->(p), keyed on p.parent_id == owner.id)"
        )
        return
    linked = 0
    for batch in _batched(ids):
        rows = repo.execute_write(
            "MATCH (p:Parameter) WHERE id(p) IN $ids "
            "MATCH (o) WHERE o.id = p.parent_id AND (o:Function OR o:Method) "
            f"MERGE (o)-[:{cs.REL_HAS_PARAMETER}]->(p) "
            "RETURN count(p) AS c",
            {"ids": batch},
        )
        linked += int(rows[0]["c"]) if rows else 0
    report.log(
        f"[DID] Parameter: linked {linked}/{len(ids)} via canonical_schema.REL_HAS_PARAMETER"
    )
    if linked < len(ids):
        report.log(
            f"[WARN] Parameter: {len(ids) - linked} orphan(s) had no resolvable "
            "Function|Method owner -- left unlinked (does not match the measured "
            "evidence; re-derive before re-running)."
        )


def link_sessions(
    repo: BoltSyncRepository,
    orphans: dict[str, list[int]],
    *,
    project: str,
    execute: bool,
    report: RunReport,
) -> None:
    _require_constant(report, "Session", "REL_HAS_SESSION")
    ids = orphans.get("Session", [])
    verb = "DID" if execute else "WOULD"
    if not ids:
        report.log(f"[{verb}] Session: link 0 via canonical_schema.REL_HAS_SESSION")
        return
    if not execute:
        report.log(
            f"[WOULD] Session: link {len(ids)} via canonical_schema.REL_HAS_SESSION "
            f"((:Project {{slug: {project!r}}})-[:HAS_SESSION]->(s))"
        )
        return
    linked = 0
    for batch in _batched(ids):
        rows = repo.execute_write(
            "MATCH (s:Session) WHERE id(s) IN $ids "
            f"MERGE (p:{cs.LABEL_PROJECT} {{slug: $project}}) "
            f"MERGE (p)-[:{cs.REL_HAS_SESSION}]->(s) "
            "RETURN count(s) AS c",
            {"ids": batch, "project": project},
        )
        linked += int(rows[0]["c"]) if rows else 0
    report.log(
        f"[DID] Session: linked {linked}/{len(ids)} via canonical_schema.REL_HAS_SESSION "
        f"(hub slug={project!r})"
    )


def plan_spec_new_projects(
    repo: BoltSyncRepository, orphans: dict[str, list[int]]
) -> list[str]:
    """Distinct, non-null/non-empty ``project_slug`` values among orphan
    Specs that do not already exist as a ``:Project`` node -- the "how many
    NEW hubs would this mint" check the ``--allow-new-projects`` guard
    reads. Pure read."""
    ids = orphans.get("Spec", [])
    if not ids:
        return []
    rows = repo.execute_read(
        "MATCH (s:Spec) WHERE id(s) IN $ids AND s.project_slug IS NOT NULL "
        "AND s.project_slug <> '' AND NOT s.project_slug IN $degenerate "
        "WITH DISTINCT s.project_slug AS slug "
        "OPTIONAL MATCH (p:Project {slug: slug}) "
        "WITH slug, p WHERE p IS NULL "
        "RETURN slug",
        {"ids": ids, "degenerate": list(_DEGENERATE_SPEC_SLUGS)},
    )
    return sorted(str(r["slug"]) for r in rows)


def backfill_hub_ids(
    repo: BoltSyncRepository,
    *,
    execute: bool,
    report: RunReport,
) -> None:
    """Stamp canonical identity (``id == slug``) on spec-bearing Project hubs.

    A hub minted with only ``slug`` reads as a property-less husk to
    ``sweep_test_pollution.py``'s Project fingerprint (``id IS NULL AND no
    DEFINES``) and gets swept — re-orphaning its whole spec tree.
    Idempotent: only touches hubs that own a HAS_SPEC edge and lack ``id``.
    """
    verb = "DID" if execute else "WOULD"
    count_rows = repo.execute_read(
        f"MATCH (p:{cs.LABEL_PROJECT})-[:{cs.REL_HAS_SPEC}]->(:Spec) "
        "WHERE p.id IS NULL AND p.slug IS NOT NULL "
        "RETURN count(DISTINCT p) AS c"
    )
    pending = int(count_rows[0]["c"]) if count_rows else 0
    if not pending:
        report.log("[OK] hub ids: all spec-bearing Project hubs carry id")
        return
    if execute:
        repo.execute_write(
            f"MATCH (p:{cs.LABEL_PROJECT})-[:{cs.REL_HAS_SPEC}]->(:Spec) "
            "WHERE p.id IS NULL AND p.slug IS NOT NULL "
            "SET p.id = p.slug"
        )
    report.log(f"[{verb}] hub ids: set id = slug on {pending} Project hub(s)")


def delete_degenerate_spec_trees(
    repo: BoltSyncRepository,
    orphans: dict[str, list[int]],
    *,
    execute: bool,
    report: RunReport,
) -> bool:
    """D1 (b): DETACH DELETE each degenerate orphan Spec tree.

    Scope safety: the per-tree Requirement/UserStory sums measured live
    (2026-07-30) equal the orphan-set totals exactly, proving the six trees
    are DISJOINT -- a 1..3-hop walk from a degenerate Spec touches only that
    tree. The count guard below still refuses if the tree grew or shrank
    since the verdict (same fail-loud posture as the DELETE rules).

    Returns False (and aborts the run) on drift.
    """
    spec_ids = orphans.get("Spec", [])
    if not spec_ids:
        report.log("[SKIP] degenerate Spec trees: 0 orphan Specs")
        return True
    verb = "DID" if execute else "WOULD"
    for slug in _DEGENERATE_SPEC_SLUGS:
        rows = repo.execute_read(
            "MATCH (s:Spec {project_slug: $slug}) WHERE id(s) IN $ids "
            "OPTIONAL MATCH (s)-[*1..3]-(n) WHERE n:Requirement OR n:UserStory "
            "RETURN count(DISTINCT s) AS specs, count(DISTINCT n) AS members",
            {"slug": slug, "ids": spec_ids},
        )
        specs = int(rows[0]["specs"]) if rows else 0
        members = int(rows[0]["members"]) if rows else 0
        if specs == 0:
            report.log(f"[OK] degenerate Spec tree {slug!r}: already deleted")
            continue
        if specs != 1:
            report.log(
                f"[ABORT] degenerate Spec tree {slug!r}: {specs} orphan Specs "
                "share the slug (expected 1) -- re-derive the verdict."
            )
            return False
        if execute:
            repo.execute_write(
                "MATCH (s:Spec {project_slug: $slug}) WHERE id(s) IN $ids "
                "OPTIONAL MATCH (s)-[*1..3]-(n) WHERE n:Requirement OR n:UserStory "
                "DETACH DELETE n, s",
                {"slug": slug, "ids": spec_ids},
            )
        report.log(
            f"[{verb}] degenerate Spec tree {slug!r}: delete{'d' if execute else ''} "
            f"1 Spec + {members} Requirement/UserStory node(s) (D1 option b)"
        )
    return True


def link_specs(
    repo: BoltSyncRepository,
    orphans: dict[str, list[int]],
    *,
    execute: bool,
    report: RunReport,
) -> None:
    _require_constant(report, "Spec", "REL_HAS_SPEC")
    ids = orphans.get("Spec", [])
    if not ids:
        report.log("[SKIP] Spec: 0 orphans")
        return
    null_rows = repo.execute_read(
        "MATCH (s:Spec) WHERE id(s) IN $ids AND "
        "(s.project_slug IS NULL OR s.project_slug = '') RETURN count(s) AS c",
        {"ids": ids},
    )
    skipped_null = int(null_rows[0]["c"]) if null_rows else 0
    if skipped_null:
        report.log(
            f"[SKIP] Spec: {skipped_null} orphan(s) with a null/empty project_slug -- "
            "NEVER minted a `:Project {slug: null}` hub, with or without any flag "
            "(kg_design._require_project_slug's own invariant, not a design call)."
        )
    linkable_rows = repo.execute_read(
        "MATCH (s:Spec) WHERE id(s) IN $ids AND s.project_slug IS NOT NULL "
        "AND s.project_slug <> '' AND NOT s.project_slug IN $degenerate "
        "RETURN count(s) AS c",
        {"ids": ids, "degenerate": list(_DEGENERATE_SPEC_SLUGS)},
    )
    linkable = int(linkable_rows[0]["c"]) if linkable_rows else 0
    if not execute:
        report.log(
            f"[WOULD] Spec: link {linkable} via canonical_schema.REL_HAS_SPEC "
            "((:Project {slug: s.project_slug})-[:HAS_SPEC]->(s))"
        )
        return
    rows = repo.execute_write(
        "MATCH (s:Spec) WHERE id(s) IN $ids AND s.project_slug IS NOT NULL "
        "AND s.project_slug <> '' AND NOT s.project_slug IN $degenerate "
        f"MERGE (p:{cs.LABEL_PROJECT} {{slug: s.project_slug}}) "
        # Canonical Project identity is id == slug (the orchestrata-ultimate
        # root hub's convention); a property-less hub reads as a husk to
        # sweep_test_pollution.py's Project fingerprint and gets deleted.
        "SET p.id = coalesce(p.id, s.project_slug) "
        f"MERGE (p)-[:{cs.REL_HAS_SPEC}]->(s) "
        "RETURN count(s) AS c",
        {"ids": ids, "degenerate": list(_DEGENERATE_SPEC_SLUGS)},
    )
    linked = int(rows[0]["c"]) if rows else 0
    report.log(
        f"[DID] Spec: linked {linked}/{linkable} via canonical_schema.REL_HAS_SPEC"
    )


# ---------------------------------------------------------------------------
# Thread family — delegates to backfill_thread_links.run_backfill
# ---------------------------------------------------------------------------


def link_thread_family(
    repo: BoltSyncRepository,
    *,
    project: str,
    execute: bool,
    report: RunReport,
) -> bool:
    """Intent/PipelineRun/Artifact/CXPPacket/Message/ToolCall -- delegates
    to :func:`backfill_thread_links.run_backfill` rather than duplicating
    its Thread-minting Cypher (BUILDER-BRIEF.md §8). That function keys off
    ``thread_id`` presence, not precomputed orphan status, so it is safe to
    call unconditionally; its own ``check_hub`` guard MATCHes (never
    MERGEs) the ``:Project`` hub, aborting rather than minting one.

    Returns True on success, False if the hub guard aborted (nothing
    written).
    """
    _require_constant(report, "Thread family", "REL_HAS_THREAD")
    _require_constant(report, "Thread family", "REL_IN_THREAD")
    stats, abort_reason = backfill_thread_links.run_backfill(
        repo, project=project, dry_run=not execute
    )
    if abort_reason is not None:
        report.log(
            f"[ABORT] Thread family (Intent/PipelineRun/Artifact/...): {abort_reason}"
        )
        return False
    verb = "DID" if execute else "WOULD"
    report.log(
        f"[{verb}] Thread family: MERGE {stats.threads} Thread node(s) via "
        "canonical_schema.thread_entity_id + REL_HAS_THREAD, "
        f"{stats.in_thread_edges} IN_THREAD edge(s) via canonical_schema.REL_IN_THREAD"
    )
    for label, count in sorted(stats.in_thread_by_label.items()):
        report.log(f"           - {label}: {count}")
    return True


# ---------------------------------------------------------------------------
# Post-apply verification
# ---------------------------------------------------------------------------


def verify_transitive_rescues(repo: BoltSyncRepository, report: RunReport) -> bool:
    """UserStory/Requirement/GatewayAuditEvent get no direct LINK rule --
    they must have become reachable transitively (via Spec / Session).
    Recomputes a FRESH orphan set (never the pre-apply snapshot)."""
    remaining = check_kg_connectivity.unreachable_node_ids(
        repo, labels=set(_TRANSITIVE_LABELS)
    )
    ok = True
    for label in _TRANSITIVE_LABELS:
        count = len(remaining.get(label, []))
        status = "PASS" if count == 0 else "FAIL"
        if count != 0:
            ok = False
        report.log(
            f"[VERIFY] {label} transitive rescue: {status} ({count} still orphaned)"
        )
    return ok


def final_connectivity_check(repo: BoltSyncRepository, report: RunReport) -> int:
    """Re-run the full BFS; print UNREACHABLE <n>."""
    conn_report = check_kg_connectivity.compute_connectivity(repo)
    report.log(f"UNREACHABLE {conn_report.unreachable}")
    return int(conn_report.unreachable)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def _driver(args: argparse.Namespace, env: dict[str, str] | None = None) -> Driver:
    env = env if env is not None else dict(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 run(
    repo: BoltSyncRepository,
    *,
    project: str,
    execute: bool,
    include_spec_hubs: bool,
    allow_new_projects: bool,
    orphans: dict[str, list[int]] | None = None,
) -> tuple[RunReport, int]:
    """The whole repair, against an already-connected repo. Returns
    ``(report, exit_code)``. Never raises on a refused rule -- refusal is a
    normal, reported outcome (exit code 1).

    ``orphans`` is a test seam (mirroring ``compute_connectivity``'s
    ``project_label`` / ``plan_backfill``'s ``source_labels``): production
    always computes it fresh via :func:`compute_orphans`. Tests inject a
    precomputed dict so the DELETE-drift/abort behaviour is deterministic
    and independent of whatever else lives in a shared per-worker test
    database.
    """
    report = RunReport()
    mode = "APPLY" if execute else "DRY-RUN"
    report.log(f"repair_orphan_hub_links [{mode}] project={project!r}")

    if orphans is None:
        orphans = compute_orphans(repo)
    report.log(
        "Orphan set (computed once, up front): "
        + ", ".join(
            f"{label}={len(orphans.get(label, []))}" for label in _ORPHAN_LABELS
        )
    )

    # -- Refuse-early checks (before ANY write) --------------------------
    drift = check_delete_drift(repo, orphans)
    # A rule whose predicate now matches 0 has already been applied by a
    # prior run (the one-shot/idempotent contract in the module header);
    # any OTHER count than {0, expected} is real drift and refuses early.
    mismatched = [
        (rule, drift[rule.label])
        for rule in DELETE_RULES
        if drift.get(rule.label, 0) not in (0, rule.expected)
    ]
    if mismatched:
        for rule, actual in mismatched:
            report.log(
                f"[ABORT] {rule.label} DELETE predicate matched {actual}, expected "
                f"{rule.expected} -- the graph drifted; re-derive the verdict, do not "
                "widen the predicate."
            )
        return report, 1
    for rule in DELETE_RULES:
        if drift.get(rule.label, 0) == 0 and rule.expected:
            report.log(
                f"[OK] {rule.label}: DELETE rule already applied "
                f"(predicate matches 0 of {rule.expected})"
            )

    new_spec_slugs: list[str] = []
    if include_spec_hubs:
        new_spec_slugs = plan_spec_new_projects(repo, orphans)
        if new_spec_slugs:
            report.log(
                f"Spec rule would mint {len(new_spec_slugs)} new :Project hub(s): "
                f"{new_spec_slugs!r}"
            )
        if len(new_spec_slugs) > 1 and not allow_new_projects:
            report.log(
                f"[ABORT] Spec rule would mint {len(new_spec_slugs)} new :Project hubs "
                "-- refusing without --allow-new-projects (minting hubs is a real "
                "ontology act, not a side effect)."
            )
            return report, 1
    else:
        report.log(
            "[SKIP] Spec rule: DECISION D1 unresolved (Mohamed's call -- which of the "
            "generated slugs are real vs degenerate). Pass --include-spec-hubs to enable."
        )

    # -- DELETE rules -----------------------------------------------------
    apply_deletes(repo, orphans, drift, execute=execute, report=report)

    # -- LINK rules ---------------------------------------------------------
    link_parameters(repo, orphans, execute=execute, report=report)
    link_sessions(repo, orphans, project=project, execute=execute, report=report)
    if include_spec_hubs:
        if not delete_degenerate_spec_trees(
            repo, orphans, execute=execute, report=report
        ):
            return report, 1
        link_specs(repo, orphans, execute=execute, report=report)
        backfill_hub_ids(repo, execute=execute, report=report)
    link_thread_family(repo, project=project, execute=execute, report=report)

    if not execute:
        report.log("Dry run complete -- pass --apply to write.")
        return report, 0

    # -- Post-apply verification -------------------------------------------
    transitive_ok = verify_transitive_rescues(repo, report)
    unreachable = final_connectivity_check(repo, report)
    if unreachable != 0 or not transitive_ok:
        return report, 1
    return report, 0


def main(argv: list[str] | None = None, env: dict[str, str] | None = None) -> int:
    """CLI entry point; ``env`` is the injection seam for tests.

    When ``env`` is not injected (real CLI usage), ``.env`` is loaded into
    the process environment FIRST (standalone usability, mirroring
    ``check_kg_connectivity.py``/``backfill_thread_links.py`` --
    ``override=False``, ambient env always wins) and THEN snapshotted, so
    ``KG_ROUTING_URI``/``KG_URI`` resolve even when only present in
    ``.env``. Test-injected ``env`` dicts are used as-is, untouched.
    """
    if env is None:
        from dotenv import load_dotenv

        load_dotenv(Path(__file__).resolve().parent.parent / ".env", override=False)
        env = dict(os.environ)
    parser = argparse.ArgumentParser(
        prog="repair_orphan_hub_links",
        description=(
            "Repair the KG root-connectivity invariant: LINK resolvable "
            "orphans, DELETE narrowly-keyed test residue (dry-run by default)."
        ),
    )
    parser.add_argument("--uri", default=None)
    parser.add_argument("--user", default=None)
    parser.add_argument("--password", default=None)
    parser.add_argument("--project", default=_project_slug(env))
    parser.add_argument(
        "--apply", action="store_true", help="Apply (default: dry run)."
    )
    parser.add_argument(
        "--include-spec-hubs",
        action="store_true",
        help="Enable the Spec LINK rule (DECISION D1 -- off by default).",
    )
    parser.add_argument(
        "--allow-new-projects",
        action="store_true",
        help="Allow the Spec rule to mint more than one new :Project hub.",
    )
    args = parser.parse_args(argv)

    driver = _driver(args, env)
    try:
        driver.verify_connectivity()
        repo = BoltSyncRepository(driver=driver)
        _, exit_code = run(
            repo,
            project=args.project,
            execute=args.apply,
            include_spec_hubs=args.include_spec_hubs,
            allow_new_projects=args.allow_new_projects,
        )
    finally:
        driver.close()
    return exit_code


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