scripts.repair_project_hubs

Reference › scripts › repair_project_hubs

scripts/repair_project_hubs.py is a one-shot, idempotent repair that re-anchors mis-attributed Project hub edges in the knowledge graph: edges that point at bogus hub nodes are moved onto the canonical hub. A human runs it; it connects through orchestrata_ultimate.core.graph_factory. See also repair_orphan_hub_links.

Options: --slug (repeatable; the bogus hub slugs, default DEFAULT_BOGUS_SLUGS), --canonical (default DEFAULT_CANONICAL), --uri and --execute; without --execute the run is a dry run. repair inspects each bogus hub: _outgoing_rels lists its outgoing relationship types and counts, gated by an allow-list so unexpected types are not moved, and _move_edges MERGE-moves all edges of one type from the bogus hub onto the canonical one. main prints the mode, URI, canonical hub and slugs first and returns the result of repair.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

"""One-shot, idempotent repair: re-anchor mis-attributed Project-hub edges.

2026-07 audit (fixed at the source in ``kg_design.py`` + the analyst/compliance
volks, commit ``cd7036b0``): pipeline callers that slugified free-text task
summaries instead of resolving the canonical project slug MERGEd anonymous
``:Project`` hub nodes — one with ``slug=''`` and one with a
truncated-to-50-chars intent slug — and thousands of real ``DEFINES`` /
``HAS_SPEC`` edges (CodeClass/CodeFunction code intelligence for
orchestrata-ultimate) anchored onto them instead of the canonical
``orchestrata-ultimate`` hub.

The writer guard now makes such hubs impossible to create; this script repairs
the ones already in the live graph. For each bogus slug it MOVES every
outgoing relationship onto the canonical hub (MERGE — duplicate-safe /
idempotent) in bounded batches, then deletes the emptied bogus hub. Children
are never touched, only their Project anchoring.

Defaults to dry-run (prints the plan + counts, writes nothing). Pass
``--execute`` to apply. Connection comes from ``NEO4J_URI`` /
``NEO4J_USERNAME`` / ``NEO4J_PASSWORD`` (read at call time via
``os.environ.get`` — Key Rule), overridable per flag for worker DBs.

Usage:
    python scripts/repair_project_hubs.py                       # dry run, default slugs
    python scripts/repair_project_hubs.py --execute
    python scripts/repair_project_hubs.py --slug '' --slug other --canonical orchestrata-ultimate --execute
"""

from __future__ import annotations

import argparse
import os
import sys
from typing import Any

from neo4j import GraphDatabase

from orchestrata_ultimate.core.graph_factory import resolve_kg_database, resolve_uri

# The two production mis-writes found by the 2026-07 audit. --slug overrides.
DEFAULT_BOGUS_SLUGS = [
    "",
    "analyze-trade-offs-between-redis-streams-and-pub-s",
]
DEFAULT_CANONICAL = "orchestrata-ultimate"
BATCH_SIZE = 2000

# Relationship types are interpolated into Cypher, so they are validated
# against this fixed allow-list — never taken from user input.
_ALLOWED_RELS = ("DEFINES", "HAS_SPEC", "HAS_RTM", "HAS_COMPLIANCE")


def _single(result: Any) -> Any:
    """``.single()`` with a hard failure on an empty result (pyright-safe)."""
    rec = result.single()
    if rec is None:
        raise SystemExit("query unexpectedly returned no record")
    return rec


def _outgoing_rels(session: Any, slug: str) -> list[dict[str, Any]]:
    """Outgoing relationship types + counts, gated by the allow-list."""
    rows = session.run(
        "MATCH (b:Project {slug:$slug})-[r]->() RETURN type(r) AS t, count(*) AS c",
        slug=slug,
    ).data()
    unknown = [r["t"] for r in rows if r["t"] not in _ALLOWED_RELS]
    if unknown:
        raise SystemExit(
            f"REFUSING: hub {slug!r} has outgoing relationship types outside the "
            f"allow-list {_ALLOWED_RELS}: {unknown} — extend the allow-list "
            "deliberately after inspecting them."
        )
    return rows


def _move_edges(session: Any, slug: str, canonical: str, rel: str) -> int:
    """MERGE-move all ``rel`` edges from the bogus hub onto the canonical hub."""
    moved_total = 0
    while True:
        rec = session.run(
            f"MATCH (b:Project {{slug:$slug}})-[d:{rel}]->(n) "
            f"WITH b, d, n LIMIT $batch "
            f"MATCH (c:Project {{slug:$canonical}}) "
            f"MERGE (c)-[:{rel}]->(n) "
            f"DELETE d RETURN count(*) AS moved",
            slug=slug,
            canonical=canonical,
            batch=BATCH_SIZE,
        ).single()
        moved = rec["moved"] if rec else 0
        moved_total += moved
        if moved == 0:
            return moved_total


def repair(
    uri: str,
    user: str,
    password: str,
    bogus_slugs: list[str],
    canonical: str,
    execute: bool,
) -> int:
    drv = GraphDatabase.driver(uri, auth=(user, password))
    try:
        with drv.session(database=resolve_kg_database()) as s:
            canon_count = _single(
                s.run(
                    "MATCH (c:Project {slug:$slug}) RETURN count(c) AS c",
                    slug=canonical,
                )
            )["c"]
            if canon_count != 1:
                print(
                    f"ABORT: canonical hub {canonical!r} count={canon_count} "
                    "(need exactly 1)."
                )
                return 1
            for slug in bogus_slugs:
                rows = _outgoing_rels(s, slug)
                if not rows:
                    exists = _single(
                        s.run(
                            "MATCH (b:Project {slug:$slug}) RETURN count(b) AS c",
                            slug=slug,
                        )
                    )["c"]
                    print(f"hub {slug!r}: {exists} node(s), 0 outgoing edges")
                    if execute and exists:
                        s.run(
                            "MATCH (b:Project {slug:$slug}) DETACH DELETE b", slug=slug
                        )
                        print(f"[DID] hub {slug!r}: deleted")
                    continue
                for row in rows:
                    if execute:
                        moved = _move_edges(s, slug, canonical, row["t"])
                        print(
                            f"[DID] hub {slug!r}: moved {moved} {row['t']} -> {canonical!r}"
                        )
                    else:
                        print(
                            f"[WOULD] hub {slug!r}: move {row['c']} {row['t']} -> {canonical!r}"
                        )
                if execute:
                    left = _single(
                        s.run(
                            "MATCH (b:Project {slug:$slug})-[e]-() RETURN count(e) AS c",
                            slug=slug,
                        )
                    )["c"]
                    print(f"hub {slug!r}: {left} edges remain post-move")
                    s.run("MATCH (b:Project {slug:$slug}) DETACH DELETE b", slug=slug)
                    print(f"[DID] hub {slug!r}: deleted")
                else:
                    print(f"[WOULD] hub {slug!r}: delete after move")
            if execute:
                canon_edges = _single(
                    s.run(
                        "MATCH (c:Project {slug:$slug})-[e]->() RETURN count(e) AS c",
                        slug=canonical,
                    )
                )["c"]
                remaining = _single(
                    s.run(
                        "MATCH (p:Project) WHERE p.slug IN $slugs RETURN count(p) AS c",
                        slugs=bogus_slugs,
                    )
                )["c"]
                print(
                    f"VERIFY: canonical {canonical!r} outgoing edges = {canon_edges}; "
                    f"bogus hubs remaining = {remaining}"
                )
                if remaining:
                    return 1
        return 0
    finally:
        drv.close()


def main(argv: list[str] | None = None, env: dict[str, str] | None = None) -> int:
    """CLI entry point; ``env`` is the injection seam for tests."""
    env = env if env is not None else dict(os.environ)
    parser = argparse.ArgumentParser(
        description="Re-anchor mis-attributed Project-hub edges (dry-run by default)."
    )
    parser.add_argument("--slug", action="append", help="Bogus hub slug (repeatable).")
    parser.add_argument("--canonical", default=DEFAULT_CANONICAL)
    parser.add_argument("--uri", default=None)
    parser.add_argument(
        "--execute", action="store_true", help="Apply (default: dry run)."
    )
    args = parser.parse_args(argv)
    uri = args.uri or resolve_uri(env=env)
    user = ""
    password = ""
    slugs = args.slug if args.slug is not None else DEFAULT_BOGUS_SLUGS
    mode = "EXECUTE" if args.execute else "DRY-RUN"
    print(
        f"repair_project_hubs [{mode}] uri={uri} canonical={args.canonical!r} "
        f"slugs={slugs!r}"
    )
    return repair(uri, user, password, slugs, args.canonical, args.execute)


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