scripts.repair_divergent_volk_aliases

Reference › scripts › repair_divergent_volk_aliases

scripts/repair_divergent_volk_aliases.py is a one-shot, idempotent repair that folds lowercase-alias Volk duplicate nodes in the knowledge graph onto the canonical node. A human runs it; it connects through orchestrata_ultimate.core.graph_factory (resolve_kg_database).

Options: --apply (write; the default is a dry run), --uri (repeatable; default every configured instance, found by configured_uris, which lists each distinct URI the environment sets, primary first) and --label (a test seam; production uses Volk). For each instance _repair_instance surveys every alias pair (_survey) and, with --apply, calls _repair, which re-points the duplicate's relationships to the canonical node and deletes the duplicate. It prints a header per URI marked apply or dry run. With no URI configured it exits 1; with nothing to repair it says so and exits 0.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Constants

Functions

Source

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

"""One-shot, idempotent repair: fold lowercase-alias Volk duplicates onto the
canonical Volk node.

``alias`` is the canonical Volk merge key (REQ-106-003). A Volk whose alias is
a lowercase package-directory name (``junior``, ``qa``, ``architect``) sitting
beside the real kernel-written node (``J:Junior``, ``U:Qa``, ``U:Ar``) is a
DIVERGENT IDENTITY, not a second volk — the same class of defect
``kernel/audit/writers.py`` already learned when MERGEing on ``volk_id``
created divergent nodes.

The writer that minted these is already fixed: ``GraphHistoryTracker.record``
resolves a harness ``volk_name`` to its canonical alias before the Volk MERGE,
and ``tests/harness/test_history.py::test_neo4j_record_canonicalizes_volk_alias``
holds that line. This script repairs the duplicates earlier runs already wrote
— found on the per-worker DBs ``.13``/``.19`` during the session-254 suite
hardening, carrying 154 real ``EVALUATED_BY`` edges of harness scenario
history between them.

Repair, per duplicate: MERGE every inbound ``EVALUATED_BY`` onto the canonical
node (duplicate-safe / idempotent), DELETE the old edges, then delete the
now-isolated duplicate. History is preserved; only the identity is unified.

Deliberately conservative — it declines to delete rather than guess:
  * no canonical counterpart present -> SKIP (renaming would be inventing an
    identity, and the lowercase node might be the only record);
  * any edge left after the re-point -> SKIP the delete and report the count,
    because an unhandled relationship type means this script does not yet know
    how to preserve that history.

Defaults to dry-run (prints the plan + counts, writes nothing). Pass
``--apply`` to execute.

Targets default to EVERY configured instance, read from ``KG_ROUTING_URI`` with
``KG_URI`` as fallback and deduplicated. (This paragraph used to name
``NEO4J_URI`` plus a ``NEO4J_URI_GW<N>`` per-worker fleet; those keys and the
separate per-worker instances behind them were retired on 2026-07-27 — worker
isolation is now a ``database=`` argument on one cluster, and the code below has
read the KG_* keys since. The prose was the only thing left describing the old
fleet as current.) That default is deliberate: a repair applied
only to the primary leaves the suite — which runs against the worker DBs —
still red, which is exactly how these duplicates went unnoticed. Connection
values are read from the injected ``env`` mapping at call time, never
hardcoded (Key Rule / RUL-003).

Usage:
    python scripts/repair_divergent_volk_aliases.py                    # dry run
    python scripts/repair_divergent_volk_aliases.py --apply
    python scripts/repair_divergent_volk_aliases.py --uri bolt://host:7687 --apply
"""

from __future__ import annotations

import argparse
import os
import sys
from collections.abc import Mapping
from typing import Any

from neo4j import GraphDatabase

from orchestrata_ultimate.core.graph_factory import resolve_kg_database

# Lowercase package-directory name -> canonical KG alias. Mirrors the volk
# roster in docs/LEVEL0.md; a name absent here is never touched.
LOWERCASE_TO_CANONICAL: dict[str, str] = {
    "analyst": "U:BA",
    "architect": "U:Ar",
    "codedirector": "U:Cd",
    "compliance": "U:Cp",
    "concierge": "U:Cx",
    "documentor": "U:Do",
    "junior": "J:Junior",
    "pm": "U:PM",
    "qa": "U:Qa",
    "senior_dev": "U:Sr",
    "ux": "U:Ux",
}

# Env keys holding the graph address. There is exactly ONE endpoint since the
# 2026-07-27 decommission: the per-worker clones this used to enumerate
# (NEO4J_URI_GW<N> -> .13/.14/.19) were destroyed with the primary (.11), and
# slots are isolated by DATABASE (orchkg_w<N>) on the single cluster.
_PRIMARY_URI_KEY = "KG_ROUTING_URI"
_FALLBACK_URI_KEY = "KG_URI"


def configured_uris(env: Mapping[str, str]) -> list[str]:
    """Every distinct Neo4j URI this environment configures, primary first.

    Derived from the environment rather than a literal fleet list so the
    script follows the .env of whatever environment it runs in — including the
    isolated rebuild fleet — and so no IP is ever baked into source.
    """
    uris: list[str] = []
    for key in (_PRIMARY_URI_KEY, _FALLBACK_URI_KEY):
        value = env.get(key)
        if value and value not in uris:
            uris.append(value)
    return uris


def _scalar(row: Any, key: str) -> int:
    """One integer out of a single-row result.

    ``Any`` for the neo4j session/record types throughout this module matches
    ``scripts/repair_project_hubs.py``: the driver types ``Session.run``'s
    query as ``LiteralString``, which no f-string-built Cypher can satisfy.

    ``Session.run(...).single()`` is typed ``Record | None`` and genuinely
    returns None when a query yields no row, so indexing it directly is a
    latent TypeError. Every query here is an aggregate (``count``/``size``),
    which always produces exactly one row — but that is a property of the
    queries, not of the API, so the None case is handled rather than assumed.
    """
    if row is None:
        return 0
    return int(row[key])


def _survey(
    session: Any, lower: str, canonical: str, label: str = "Volk"
) -> dict[str, int] | None:
    """Counts for one alias pair, or None when there is no duplicate.

    ``label`` exists so this script's own tests can drive it against a
    per-test-unique label instead of the real ``:Volk`` nodes (the
    ``label=``-scoped discipline ``test_canonical_conformance.py`` uses).
    Production always uses the default.
    """
    row = session.run(
        f"MATCH (dup:`{label}` {{alias: $lower}}) "
        f"OPTIONAL MATCH (canon:`{label}` {{alias: $canonical}}) "
        "RETURN count(DISTINCT dup) AS dups, count(DISTINCT canon) AS canon, "
        "       size([(dup)-[r]-() | r]) AS edges",
        lower=lower,
        canonical=canonical,
    ).single()
    if row is None or row["dups"] == 0:
        return None
    return {"dups": row["dups"], "canon": row["canon"], "edges": row["edges"]}


def _repair(session: Any, lower: str, canonical: str, label: str = "Volk") -> str:
    """Re-point + delete one duplicate. Returns a human-readable outcome."""
    moved = _scalar(
        session.run(
            f"MATCH (dup:`{label}` {{alias: $lower}})<-[r:EVALUATED_BY]-(src) "
            f"MATCH (canon:`{label}` {{alias: $canonical}}) "
            f"MERGE (src)-[:EVALUATED_BY]->(canon) "
            f"DELETE r RETURN count(*) AS n",
            lower=lower,
            canonical=canonical,
        ).single(),
        "n",
    )
    remaining = _scalar(
        session.run(
            f"MATCH (dup:`{label}` {{alias: $lower}}) "
            f"RETURN size([(dup)-[r]-() | r]) AS e",
            lower=lower,
        ).single(),
        "e",
    )
    if remaining:
        return (
            f"re-pointed {moved} EVALUATED_BY; {remaining} other edge(s) remain "
            f"-> NOT deleting (unhandled relationship type)"
        )
    session.run(f"MATCH (dup:`{label}` {{alias: $lower}}) DELETE dup", lower=lower)
    return f"re-pointed {moved} EVALUATED_BY; deleted duplicate"


def _repair_instance(session: Any, *, apply: bool, label: str) -> int:
    """Survey (and, when ``apply``, repair) every alias pair on one instance.

    Returns how many duplicates were found, so the caller can tell "clean
    graph" apart from "repaired something".
    """
    found = 0
    for lower, canonical in LOWERCASE_TO_CANONICAL.items():
        survey = _survey(session, lower, canonical, label)
        if survey is None:
            continue
        found += 1
        print(
            f"  {lower!r} -> {canonical!r}: dup={survey['dups']} "
            f"canonical_exists={survey['canon']} edges={survey['edges']}"
        )
        if not apply:
            continue
        if survey["canon"] == 0:
            print("    SKIP: no canonical Volk to fold into")
            continue
        print(f"    {_repair(session, lower, canonical, label)}")
    return found


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Fold lowercase-alias Volk duplicates onto the canonical node."
    )
    parser.add_argument("--apply", action="store_true", help="write (default: dry run)")
    parser.add_argument(
        "--uri",
        action="append",
        dest="uris",
        help="bolt URI to repair (repeatable; default: every configured instance)",
    )
    parser.add_argument(
        "--label",
        default="Volk",
        help="node label to repair (test seam; production uses the default)",
    )
    return parser


def main(argv: list[str] | None = None, env: Mapping[str, str] | None = None) -> int:
    args = _build_parser().parse_args(argv)

    env = env if env is not None else os.environ
    uris = args.uris or configured_uris(env)
    found = 0

    if not uris:
        print(
            f"No graph URI configured — set {_PRIMARY_URI_KEY} or "
            f"{_FALLBACK_URI_KEY}, or pass --uri."
        )
        return 1

    for uri in uris:
        # No credentials: every remaining endpoint negotiates no auth.
        driver = GraphDatabase.driver(uri, auth=None)
        try:
            with driver.session(database=resolve_kg_database()) as session:
                print(f"===== {uri} {'(apply)' if args.apply else '(dry run)'}")
                found += _repair_instance(session, apply=args.apply, label=args.label)
        finally:
            driver.close()

    if found == 0:
        print("No divergent lowercase-alias Volk nodes found — nothing to repair.")
    return 0


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