scripts.canonical_conformance

Reference › scripts › canonical_conformance

A one-shot, idempotent migration that brings the live knowledge graph to the canonical schema: it purges test pollution, backfills missing keys and applies the canonical constraints. A human runs it; no module or test imports it. It uses the Bolt repository from core.graph and the factory in graph_factory. See also backfill_thread_links, which follows the same shape.

Invocation: main takes connection options (--uri, --user and a password option) and --dry-run, which classifies and reports counts without a single write. Missing connection settings are filled from the repository's dotenv file without overriding the ambient environment.

The Component id and project backfill runs first (plan_component_id_migration, run_component_id_migration), and the run aborts with exit 1 if it cannot complete, so the generic step never has to invent a Component id. plan_migration then classifies nodes: _is_test_pollution marks test leftovers for removal, and _backfill_value supplies the value for a missing canonical key. run_migration applies the plan in batches of 500 (_batched) and runs the schema DDL. The output lines read "WOULD" in a dry run and "DID" otherwise.

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/canonical_conformance.py

#!/usr/bin/env python3
"""One-shot, idempotent migration: bring the live Neo4j KG to canonical-schema
conformance (reproducibility runbook R4, the live-graph half).

The operational KG accumulated ~5.6k nodes that lack their canonical merge key
(``kg_schema.py check`` flags them). Investigation showed the overwhelming
majority are **test pollution** — nodes leaked into the production graph by
tests that wrote real ``:Session``/``:Fact``/``:Memory``/… nodes to the shared
instance over months (identifiers like ``_TEST_…``, ``test-sync-…``,
``vizint-test-…``, ``mod__test_…``, ``/tmp/pytest-…``). A small remainder is
genuinely real (architecture ``:Rule`` nodes, the ``:Architecture`` node, real
``:Volk`` nodes, historical ``:TestRun`` records, spec ``:Requirement`` /
``:UserStory`` nodes).

This migration therefore:
  1. PURGES only provably test-marked keyless nodes (high-precision markers).
  2. BACKFILLS the canonical key on every surviving keyless node, from its
     natural identifier, falling back to a deterministic ``legacy-<elementId>``
     (unique, stable) when no natural id exists or a value would collide.
  3. DROPS the plain (non-constraint-backed) indexes that collide with a
     canonical uniqueness constraint (the ``IndexAlreadyExists`` blocker).
  4. DROPS the extra/unknown constraints that key a canonical label on the
     wrong property (reported by ``check``).
  5. APPLIES the canonical DDL (``canonical_schema.get_all_ddl``).

Additionally, a dedicated step (ratified 2026-07-19,
plans/architecture-ontology-draft.md §9.2 ruling 2) re-scopes ``Component``
from its retired bare-``name`` key to the canonical project-qualified
``id = "{project}:{name}"``: it backfills ``id``/``project`` on every live
Component missing ``id`` (see :func:`plan_component_id_migration`), runs
BEFORE the generic steps above so the generic engine never has to fabricate
a Component id itself (Component is excluded from its per-node backfill scan
— see ``_CUSTOM_BACKFILL_LABELS`` — but still participates in the generic
constraint-drop/DDL-apply steps, which retire ``component_name`` and create
``component_id``). It ABORTS (non-zero exit, no writes) if any live Component
already carries an ambiguous ``project`` — a real signal this migration must
not paper over with a guessed default.

After it runs, ``python scripts/bootstrap_graphs.py --neo4j --check`` exits 0.

Idempotent: re-running is a no-op (purged nodes are gone, survivors already
carry keys, DROPs use ``IF EXISTS``, the canonical DDL uses ``IF NOT EXISTS``).

Connection defaults to ``NEO4J_URI``/``NEO4J_USERNAME``/``NEO4J_PASSWORD``;
``--uri``/``--user``/``--password`` override so a test-fleet clone can be
stamped without touching the shared default instance. ``--dry-run`` classifies
and reports counts without a single write. Take a backup first (see
``scripts`` sibling ``backup`` note in the runbook) — this deletes nodes.
"""

from __future__ import annotations

import argparse
import os
import re
import sys
from collections.abc import Mapping
from pathlib import Path

_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 GraphDatabase  # noqa: E402

from orchestrata_ultimate.core.graph import BoltSyncRepository  # noqa: E402
from orchestrata_ultimate.core.graph_factory import (  # noqa: E402
    resolve_kg_dialect,
    resolve_uri,
)
from orchestrata_ultimate.knowledge_graph.canonical_schema import (  # noqa: E402
    CANONICAL_MERGE_KEYS,
    ExtraConstraint,
    apply,
    check,
    get_all_ddl,
)

# --- test-pollution detection -------------------------------------------------

# Strong markers: unambiguous test-artifact tokens.
_STRONG = re.compile(
    r"_TEST_|test-sync-|vizint-test-|mod__test_|Cls__test_|/tmp/pytest|"
    r"test_real_neo4j_|test_real_graph_|nonexistent_|totally_bogus|_does_not_exist",
    re.IGNORECASE,
)
# Weak marker: a leading ``test_``/``test-`` token.
_WEAK = re.compile(r"(?:^|[^a-z0-9])test[_-]", re.IGNORECASE)
_STRUCTURED_ID_FIELDS = (
    "id",
    "name",
    "session_id",
    "volk_id",
    "task_id",
    "key",
    "tag",
    "req_id",
    "story_id",
    "run_id",
    "slug",
    "path",
    "root_path",
    "fact_id",
)
# Markers are scanned ONLY against identifier fields (+ ``data``, the audit blob
# where Event/ToolCall/… embed their test session_id) — NEVER against free-text
# prose. Scanning prose deleted a real :Rule on the first run: its description
# "…start_test_services…" matched ``_TEST_`` case-insensitively (``_test_``).
# A marker in an *identifier* is a reliable test signal; a marker in prose is not.
_STRONG_SCAN_FIELDS = _STRUCTURED_ID_FIELDS + ("data",)

# Natural-id source properties per label, ordered by preference. Anything not
# listed (or whose sources are all absent) backfills to ``legacy-<elementId>``.
_BACKFILL_SOURCES: dict[str, tuple[str, ...]] = {
    "Requirement": ("req_id",),
    "UserStory": ("story_id",),
    "TestRun": ("run_id",),
    "Rule": ("name",),
    "Architecture": ("name",),
    "Volk": ("alias", "name", "volk_id"),
    "Project": ("slug", "id", "name"),
    "Session": ("id",),
    "Memory": ("key", "id"),
    "Fact": ("fact_id",),
}


def _is_test_pollution(props: Mapping[str, object]) -> bool:
    """True iff a keyless node carries a high-precision test marker in an
    IDENTIFIER field (or the audit ``data`` blob) — never in free-text prose."""
    for field in _STRONG_SCAN_FIELDS:
        value = props.get(field)
        if value is not None and _STRONG.search(str(value)):
            return True
    for field in _STRUCTURED_ID_FIELDS:
        value = props.get(field)
        if value is not None and _WEAK.search(str(value)):
            return True
    return False


def _backfill_value(label: str, eid: str, props: Mapping[str, object]) -> str:
    """Deterministic canonical-key value for a surviving keyless node."""
    for src in _BACKFILL_SOURCES.get(label, ()):
        value = props.get(src)
        if value not in (None, ""):
            return str(value)
    return f"legacy-{eid}"


# --- Component id/project backfill (ratified 2026-07-19, ruling 2) ------------

# Labels the generic per-node backfill loop below (plan_migration) must NOT
# touch — they have a dedicated, more-specific migration step instead (a
# generic ``legacy-<elementId>`` fill would be actively wrong for them). Only
# the missing-key SCAN is skipped; these labels still participate fully in
# the generic index/constraint-drop and canonical-DDL-apply steps.
_COMPONENT_LABEL = "Component"
_COMPONENT_DEFAULT_PROJECT = "orchestrata-ultimate"
_CUSTOM_BACKFILL_LABELS: frozenset[str] = frozenset({_COMPONENT_LABEL})


def plan_component_id_migration(
    repo: BoltSyncRepository,
    *,
    label: str = _COMPONENT_LABEL,
    default_project: str = _COMPONENT_DEFAULT_PROJECT,
) -> tuple[list[dict[str, str]], str | None]:
    """Compute the Component ``id``/``project`` backfill plan.

    Every live Component node still keyed on the retired bare ``name`` gets a
    project-qualified ``id = "{default_project}:{name}"`` plus a ``project``
    property — the same ``"{project}:{natural}"`` scheme File/Directory/
    Package already use. Pure read — never mutates.

    Args:
        repo: A ``BoltSyncRepository``.
        label: The node label to scan; defaults to ``"Component"``. Tests
            pass an isolated test-only label to exercise this against a
            throwaway node without touching real Component data.
        default_project: The project prefix to backfill with; defaults to
            ``"orchestrata-ultimate"`` — the only project this repo's own
            Component nodes (Gateway/Kernel/graph/Redis/VizInt UI, written by
            ``scripts/sync_architecture_to_graph.py``) can legitimately carry.

    Returns:
        ``(rows, abort_reason)``. ``rows`` is ``[{"eid", "name", "id"}, ...]``
        for every node missing ``id``. ``abort_reason`` is non-``None`` (and
        ``rows`` is always ``[]`` in that case) when a node's existing
        ``project`` property already diverges from ``default_project``, or a
        node has no ``name`` to derive an id from at all — both are
        ambiguous-provenance signals this migration refuses to guess
        through; the caller must investigate and re-run.
    """
    rows = repo.execute_read(
        f"MATCH (n:`{label}`) WHERE n.id IS NULL "
        f"RETURN elementId(n) AS eid, properties(n) AS props"
    )
    plan: list[dict[str, str]] = []
    for row in rows:
        eid, props = row["eid"], row["props"]
        name = props.get("name")
        if not name:
            return [], (
                f"{label} node {eid} has no `name` property — cannot derive "
                "a canonical id, aborting"
            )
        existing_project = props.get("project")
        if existing_project not in (None, "", default_project):
            return [], (
                f"{label} node {eid} ({name!r}) already carries "
                f"project={existing_project!r} — ambiguous provenance for "
                f"the {default_project!r} default, aborting"
            )
        plan.append({"eid": eid, "name": str(name), "id": f"{default_project}:{name}"})
    return plan, None


def run_component_id_migration(
    repo: BoltSyncRepository,
    *,
    dry_run: bool,
    label: str = _COMPONENT_LABEL,
    default_project: str = _COMPONENT_DEFAULT_PROJECT,
) -> tuple[int, str | None]:
    """Execute (or, when ``dry_run``, only report) the Component backfill.

    Idempotent: once every live node under ``label`` carries ``id``,
    :func:`plan_component_id_migration` returns an empty plan and this is a
    no-op.

    Returns:
        ``(count, abort_reason)`` — ``count`` is the number of nodes planned
        (dry-run) or actually written (execute); ``abort_reason`` mirrors
        :func:`plan_component_id_migration`'s, and when set no write happens.
    """
    plan, abort_reason = plan_component_id_migration(
        repo, label=label, default_project=default_project
    )
    if abort_reason is not None or dry_run or not plan:
        return len(plan), abort_reason
    for row in plan:
        repo.execute_write(
            f"MATCH (n:`{label}`) WHERE elementId(n) = $eid "
            f"SET n.id = $id, n.project = $project",
            {"eid": row["eid"], "id": row["id"], "project": default_project},
        )
    return len(plan), None


# --- migration steps ----------------------------------------------------------


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 plan_migration(
    repo: BoltSyncRepository,
    merge_keys: Mapping[str, str] | None = None,
) -> tuple[
    list[str], dict[str, list[dict[str, str]]], list[str], list[ExtraConstraint]
]:
    """Compute (purge_eids, backfill_by_label, drop_indexes, drop_constraints).

    ``backfill_by_label`` maps a label to ``[{"eid", "val"}, ...]``; the target
    property is ``merge_keys[label]``. Pure read — never mutates. This is the
    dry-run core and the test seam: pass a ``{test_only_label: key}`` map to
    exercise the full plan (purge / backfill / dup re-key / drops) against an
    isolated label without touching a real canonical one.
    """
    keys = merge_keys if merge_keys is not None else CANONICAL_MERGE_KEYS
    purge_eids: list[str] = []
    backfill: dict[str, list[dict[str, str]]] = {}
    # Existing key values per label, so a backfill never collides with a node
    # that already conforms (or with an earlier backfill in the same run).
    used: dict[str, set[str]] = {}

    for label, key in keys.items():
        if label in _CUSTOM_BACKFILL_LABELS:
            # Handled by a dedicated pre-step (e.g. run_component_id_migration)
            # that derives a real value, not a generic legacy-<elementId>
            # fallback — see _CUSTOM_BACKFILL_LABELS.
            continue
        rows = repo.execute_read(
            f"MATCH (n:`{label}`) WHERE n.`{key}` IS NULL "
            f"RETURN elementId(n) AS eid, properties(n) AS props"
        )
        if not rows:
            continue
        used.setdefault(
            label,
            {
                str(r["v"])
                for r in repo.execute_read(
                    f"MATCH (n:`{label}`) WHERE n.`{key}` IS NOT NULL "
                    f"RETURN n.`{key}` AS v"
                )
            },
        )
        for row in rows:
            eid, props = row["eid"], row["props"]
            if _is_test_pollution(props):
                purge_eids.append(eid)
                continue
            value = _backfill_value(label, eid, props)
            if value in used[label]:
                value = f"legacy-{eid}"  # guaranteed unique
            used[label].add(value)
            backfill.setdefault(label, []).append({"eid": eid, "val": value})

    # Duplicate-key groups: nodes that ALREADY carry the canonical key but share
    # a value block CREATE CONSTRAINT ... IS UNIQUE (check() never sees these —
    # it only flags MISSING keys). Resolve WITHOUT data loss: purge the provably
    # -test members, then, if a group still has >1 survivor, keep the first and
    # re-key the rest to a unique deterministic ``legacy-<elementId>``. Re-keying
    # (not deleting) is the correct call for members we cannot PROVE are test —
    # the canonical key was a non-unique placeholder anyway, and the node's data
    # / timestamps / relationships are preserved.
    for label, key in keys.items():
        dup_groups = repo.execute_read(
            f"MATCH (n:`{label}`) WHERE n.`{key}` IS NOT NULL "
            f"WITH n.`{key}` AS v, count(*) AS c, "
            f"collect({{eid: elementId(n), props: properties(n)}}) AS members "
            f"WHERE c > 1 RETURN v AS v, members AS members"
        )
        for group in dup_groups:
            members = group["members"]
            survivors = []
            for m in members:
                if _is_test_pollution(m["props"]):
                    purge_eids.append(m["eid"])
                else:
                    survivors.append(m)
            # Keep survivors[0]'s existing value; re-key the rest uniquely.
            for m in survivors[1:]:
                backfill.setdefault(label, []).append(
                    {"eid": m["eid"], "val": f"legacy-{m['eid']}"}
                )

    # Colliding plain indexes: on NEO4J a non-constraint-backed single-prop
    # index on a (label, key) pair blocks CREATE CONSTRAINT ... IS UNIQUE.
    #
    # There is no such collision on orchgraph/Memgraph, and reconciling one
    # would be actively harmful: a uniqueness constraint there does NOT back
    # lookups, so the plain index is the thing making `MATCH (n:L {id: ...})`
    # anything other than a full scan — and the two coexist happily (verified
    # live 2026-07-26: CREATE INDEX ON :L(id) then CREATE CONSTRAINT ON (n:L)
    # ASSERT n.id IS UNIQUE both succeed). `SHOW INDEXES` is Neo4j-only
    # grammar besides, so this step is skipped outright on that engine.
    canon_pairs = set(keys.items())
    drop_indexes: list[str] = []
    if resolve_kg_dialect() != "memgraph":
        for idx in repo.execute_read("SHOW INDEXES"):
            labels = idx.get("labelsOrTypes") or []
            props = idx.get("properties") or []
            owner = idx.get("owningConstraint")
            if owner:
                continue
            if (
                len(labels) == 1
                and len(props) == 1
                and (labels[0], props[0]) in canon_pairs
            ):
                drop_indexes.append(idx["name"])

    # Extra/unknown constraints (canonical label, wrong key) — from check().
    # The ENTRY is carried, not just its name: orchgraph constraints are
    # unnamed, so the drop has to be spelled by (label, property) there.
    report = check(repo, dict(keys))
    drop_constraints = list(report.extra_constraints)

    return purge_eids, backfill, drop_indexes, drop_constraints


def _batched(seq: list[str], size: int = 500):
    for i in range(0, len(seq), size):
        yield seq[i : i + size]


def run_migration(
    repo: BoltSyncRepository,
    *,
    dry_run: bool,
    merge_keys: Mapping[str, str] | None = None,
    ddl: list[str] | None = None,
) -> dict[str, int]:
    """Execute (or, when ``dry_run``, only report) the migration.

    ``merge_keys``/``ddl`` default to the real canonical schema; tests pass a
    scoped ``{test_label: key}`` + its constraint DDL to exercise the
    destructive path in isolation.
    """
    keys = merge_keys if merge_keys is not None else CANONICAL_MERGE_KEYS
    statements = ddl if ddl is not None else get_all_ddl()
    purge_eids, backfill, drop_indexes, drop_constraints = plan_migration(repo, keys)

    stats = {
        "purged": len(purge_eids),
        "backfilled": sum(len(v) for v in backfill.values()),
        "dropped_indexes": len(drop_indexes),
        "dropped_constraints": len(drop_constraints),
        "applied_ddl": len(statements),
    }
    if dry_run:
        return stats

    for batch in _batched(purge_eids):
        repo.execute_write(
            "MATCH (n) WHERE elementId(n) IN $eids DETACH DELETE n", {"eids": batch}
        )
    # Backfill/re-key the canonical key directly, one UNWIND write per
    # label+batch (the key property name differs per label). The SET is
    # unconditional so it covers both keyless fills and duplicate-group re-keys;
    # values are deterministic, so re-running is idempotent.
    for label, items in backfill.items():
        key = keys[label]
        for i in range(0, len(items), 500):
            repo.execute_write(
                f"UNWIND $rows AS row MATCH (n:`{label}`) "
                f"WHERE elementId(n) = row.eid "
                f"SET n.`{key}` = row.val",
                {"rows": items[i : i + 500]},
            )

    # DDL, so: execute_autocommit (orchgraph rejects schema manipulation
    # inside a managed transaction), and dialect-rendered — orchgraph's
    # constraints are UNNAMED, so `DROP CONSTRAINT <name>` has nothing to name
    # and the drop is spelled by (label, property) instead. drop_indexes is
    # Neo4j-only by construction (see plan_migration).
    for name in drop_indexes:
        repo.execute_autocommit(f"DROP INDEX {name} IF EXISTS")
    memgraph = resolve_kg_dialect() == "memgraph"
    for entry in drop_constraints:
        if memgraph:
            stmt = (
                f"DROP CONSTRAINT ON (n:`{entry.label}`) "
                f"ASSERT n.`{entry.property}` IS UNIQUE"
            )
        else:
            stmt = f"DROP CONSTRAINT {entry.name} IF EXISTS"
        repo.execute_autocommit(stmt)

    apply(repo, statements)
    return stats


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="canonical_conformance",
        description="Purge test pollution + backfill keys + apply canonical schema.",
    )
    parser.add_argument("--uri", default=None)
    parser.add_argument("--user", default=None)
    parser.add_argument("--password", default=None)
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Classify + report counts without a single write.",
    )
    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 tests/conftest.py's load_dotenv so the one-shot tool does not
    # require a pre-sourced shell.
    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)

        verb = "WOULD" if args.dry_run else "DID"

        # Component id/project backfill runs FIRST (see module docstring):
        # by the time the generic steps below scan for missing canonical
        # keys, every live Component either already conforms or the whole
        # run has aborted — the generic engine never fabricates a Component
        # id itself.
        component_count, component_abort = run_component_id_migration(
            repo, dry_run=args.dry_run
        )
        if component_abort is not None:
            print(f"[ABORT] Component id/project backfill: {component_abort}")
            return 1
        print(f"[{verb}] backfill Component id/project : {component_count}")

        stats = run_migration(repo, dry_run=args.dry_run)
    finally:
        driver.close()

    print(f"[{verb}] purge test-pollution nodes : {stats['purged']}")
    print(f"[{verb}] backfill canonical keys    : {stats['backfilled']}")
    print(f"[{verb}] drop colliding plain indexes: {stats['dropped_indexes']}")
    print(f"[{verb}] drop wrong-key constraints  : {stats['dropped_constraints']}")
    print(f"[{verb}] apply canonical DDL stmts   : {stats['applied_ddl']}")
    return 0


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