scripts.sweep_test_pollution

Reference › scripts › sweep_test_pollution

scripts/sweep_test_pollution.py sweeps stale test-pollution nodes from the knowledge graph, on the primary database and on the per-worker databases. Crashed or killed test runs never run fixture teardown, so their seeded _test_-marked nodes stay in the shared graph; this script finds and deletes them. start_test_services imports sweep_if_polluted and calls it with deep=True at suite start, before any worker has seeded anything; a human can also run it directly. It loads the environment through orchestrata_ultimate.core.env_bootstrap (_load_dotenv, the RUL-019 contract).

Targets are (uri, database) pairs. _all_uris lists every distinct graph URI the suite touches, _memgraph_worker_count finds the top of the orchkg_w<N> database range, and _all_kg_targets combines them; _driver builds a driver from the environment mapping. _match_clause builds the match for a pollution criterion. count_pollution counts matches and sweep_pollution deletes them (DETACH DELETE). _leased_elsewhere skips a worker database that another swarm currently holds under a slot lease (see slot_lease). Options: --apply deletes (the default only reports) and --deep also sweeps the deep-marker criterion (colon-qualified marker ids, uppercase _TEST_ ids, test_id-stamped nodes, test-only labels, probe sessions) on every target including the primary, which is safe only when nothing has been seeded yet. Legacy id-less File and Directory pollution is a worker-database criterion only. An unreachable target is reported per URI.

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

#!/usr/bin/env python3
"""Sweep stale test-pollution nodes from the Neo4j KG — primary AND worker DBs.

Root cause (backlog #124): crashed / SIGKILL'd test runs skip fixture
teardown and leak their seeded nodes into the shared KG instances — the
primary (10.53.1.11) and every per-worker DB (``NEO4J_URI_GW*`` in .env).
Every leaked node carries one of two fingerprints the test fixtures stamp:

- ``n.id CONTAINS '_test_'``          — the per-test ``test_id`` marker
- ``n.path`` / ``n.file_path`` starting with ``/tmp`` — tmp-workspace files

The id-marker clause EXCLUDES ids containing ``:`` — canonical project-
qualified ids (``orchestrata-ultimate:pkg.mod.symbol``) may legitimately
contain ``_test_`` when the source symbol name does (``QaVolk._handle_test_gen``,
``_is_test_path``, ...); the 2026-07-15 first apply deleted 8 such Functions
and 29 Methods before this guard existed (restored by re-running the AST
sync). Test-seeded marker ids never carry a project qualifier. Nodes are
deleted in batches so a large backlog cannot blow a single transaction.

Safety model: the sweep runs from ``start_test_services.py`` BEFORE workers
start, i.e. at a moment when no test of this run has seeded anything yet.
Concurrent independent suite runs against the same KG are already
unsupported (single-machine, xdist banned).

A second, structurally distinct fingerprint class covers ``:Project`` nodes
(``research-ast-json-vs-neo4j.md`` work item 7, 2026-07-18 audit): a live
count found 1513 of 1514 Project nodes are throwaway — only
``id='orchestrata-ultimate'`` (== ``slug``) is the real, AST-synced project.
See ``REAL_PROJECT_SLUG`` and the ``'Project' IN labels(n)`` branch of
``_MATCH_CLAUSE`` below for the derived patterns and their rationale.

A third, WORKER-DB-ONLY criterion (Workstream B3, plans/parallel-collision-
fix-plan.md, 2026-07-22) covers id-less ``:File``/``:Directory`` nodes —
pre-canonical-schema-ruling leftovers that can still collide on the legacy
bare-``path`` constraint or shadow a canonical, id-keyed node. Kept as a
SEPARATE clause/functions (``_LEGACY_FILE_DIR_MATCH_CLAUSE``,
``count_legacy_file_dir_pollution``/``sweep_legacy_file_dir_pollution``),
never folded into ``_MATCH_CLAUSE``, and only ever applied to worker URIs —
the primary's File/Directory nodes are live, id-keyed production sync data
that a heuristic bulk-delete must never touch.

A fourth, DEEP-MARKER criterion (2026-07-26 audit) catches ~5,500 residual
pollution nodes the three clauses above cannot see, because each escapes a
guard those clauses rely on:

- Colon-qualified marker ids (``n.id =~ '.*_test_[0-9a-f]{8,}.*'``) — the
  classic clause's ``NOT CONTAINS ':'`` guard, added to protect canonical
  project-qualified ids, also shields these genuinely-polluted colon-bearing
  marker ids (e.g. ``extra-methods-_test_<hex>:mod__test_<hex>``,
  ``intent:thread-_test_<hex>``). Verified live: the only
  ``orchestrata-ultimate:``-prefixed matches are themselves fixture-seeded
  Components stamped with ``test_id`` — no canonical false positives.
- The uppercase ``_TEST_[0-9a-f]{8,}`` variant the ``test_id`` pytest
  fixture (``tests/conftest.py``) actually stamps into ids — the classic
  clause's marker text is lowercase-only and never checks it.
- ``n.test_id IS NOT NULL`` directly — the fixture-stamped marker property
  itself, independent of whether it also leaked into ``id``.
- Test-only labels (``CanonTest_test_[0-9a-f]+``, ``DLTest``,
  ``__NqcAllowWritesTest``) and probe Session prefixes
  (``INTENTPROBE-``/``POLICYPROBE-``/``AUDITPROBE-``/``kgtest-``
  unconditionally; ``test-``/``sess-``/``conn-`` only when the Session node
  is isolated, i.e. ``NOT (n)--()``, so a live, edge-connected Session can
  never match on prefix alone).

This criterion is deliberately kept as a SEPARATE clause/function pair
(``_DEEP_MARKER_CLAUSE``, ``count_deep_marker_pollution``/
``sweep_deep_marker_pollution``) and NEVER folded into ``_MATCH_CLAUSE``:
the sweeper's own live-sweep tests (``test_sweep_if_polluted_sweeps_real_
pollution_...``, ``test_main_apply_...``) run a REAL global sweep MID-SUITE
while parallel tests have live seeded fixtures, and every seeded fixture
carries ``test_id`` / a marker-embedded id — folding these rules into the
always-on clause would let those tests destroy concurrent tests' data. It is
opt-in via the ``--deep`` CLI flag (and a ``deep=`` kwarg on
``sweep_if_polluted``) and is applied to EVERY URI including the primary
(unlike the worker-DB-only legacy File/Directory criterion above) — it is
safe there because it only ever fires from ``start_test_services.py``
BEFORE workers start, mirroring the base sweep's own safety model.

CLI: dry-run by default — reports what WOULD be deleted; ``--apply`` deletes.
"""

from __future__ import annotations

import argparse
import os
import re
import sys
from collections.abc import Mapping
from pathlib import Path
from typing import TYPE_CHECKING, LiteralString

# Sibling-script bare import (same pattern as scripts/_irc_send.py's
# ``_irc_conversation`` import) -- swarm-efficiency Fix 5: lets a sweep skip
# an ``orchkg_w<N>`` database another swarm currently holds a live lease on,
# without pulling the orchestrata_ultimate package into this stdlib+neo4j-only
# tool. See ``_leased_elsewhere`` below for the one call site.
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import slot_lease  # noqa: E402

from orchestrata_ultimate.core.env_bootstrap import load_env_contract

_ORCHKG_WORKER_DB_RE = re.compile(r"orchkg_w(\d+)$")


def _leased_elsewhere(database: str, env: Mapping[str, str]) -> bool:
    """``True`` when ``database`` is an ``orchkg_w<N>`` database currently
    leased (``scripts/slot_lease.py``) to an owner other than the caller
    (``SLOT_LEASE_OWNER`` in ``env``).

    Without this, one swarm's sweeper could ``DETACH DELETE`` nodes a
    DIFFERENT swarm's live test run just seeded into that slot's database --
    exactly the race ``scripts/slot_lease.py`` (swarm-efficiency Fix 5)
    exists to prevent. Non-worker databases (``orchkg`` itself, or anything
    not matching the ``orchkg_w<N>`` pattern) are never filtered here.

    The lease-file location is resolved from THIS module's own injected
    ``env`` mapping (``SLOT_LEASE_STATE``), not ``slot_lease``'s own
    ``os.environ`` default -- matching this file's existing convention of
    reading every input through the injected mapping (see the module
    docstring's "standalone stdlib+neo4j tool" note), and letting tests
    point it at a ``tmp_path`` lease file with no ``os.environ`` mutation.
    """
    m = _ORCHKG_WORKER_DB_RE.match(database)
    if not m:
        return False
    slot = int(m.group(1))
    state_override = env.get("SLOT_LEASE_STATE", "").strip()
    state_path = Path(state_override) if state_override else None
    owner = slot_lease.active_owner(slot, state_path=state_path)
    if owner is None:
        return False
    # bool(): slot_lease.active_owner returns Any, and Any != str is Any.
    return bool(owner != env.get("SLOT_LEASE_OWNER", ""))


if TYPE_CHECKING:
    from neo4j import Driver

BATCH_SIZE = 1000

# The single real, AST-synced Project — GraphSyncer.sync_all is the only
# writer that ever sets `.id` (canonical merge key is `.slug` per
# canonical_schema.py, but this project's real node happens to carry the
# same value in both). Every Project-pollution clause below excludes BOTH
# properties against this constant, twice over, so the allowlist holds even
# though `.id`/`.slug` disagree as writer keys in general (§4 of the audit
# report). Verified live 2026-07-18: `MATCH (p:Project {id:'orchestrata-
# ultimate'}) RETURN p` -> the only Project node with `module_count`/
# `total_loc`/`synced_at` populated; every other Project node in the graph
# (1513 of 1514) is throwaway. See ``tests/scripts/test_sweep_test_
# pollution.py::test_real_project_slug_is_protected_by_construction``.
REAL_PROJECT_SLUG = "orchestrata-ultimate"

# Fingerprints: the per-test id marker, tmp-workspace paths (all three
# property spellings writers use, including the file-watcher's underscore
# `_file_path`), and e2e demo-run scratch workspaces (logs/e2e-runs is
# gitignored per-run scratch — its code structure does not belong in the
# operational KG). Gap history: the 2026-07-15 KG re-sync found 179 UUID
# "NewClass" nodes the original two fingerprints missed.
_MATCH_CLAUSE = (
    "MATCH (n) WHERE "
    "(n.id IS NOT NULL AND n.id CONTAINS '_test_' AND NOT n.id CONTAINS ':') "
    "OR (n.path IS NOT NULL AND n.path STARTS WITH '/tmp') "
    "OR (n.file_path IS NOT NULL AND n.file_path STARTS WITH '/tmp') "
    "OR (n._file_path IS NOT NULL AND (n._file_path STARTS WITH '/tmp' "
    "OR n._file_path CONTAINS '/logs/e2e-runs/')) "
    "OR (n.path IS NOT NULL AND n.path CONTAINS '/logs/e2e-runs/') "
    "OR (n.file_path IS NOT NULL AND n.file_path CONTAINS '/logs/e2e-runs/') "
    # Rule pollution keyed by `file_id`, not `path`/`file_path` (2026-07-19
    # audit): code-intelligence graph-sync tests (e.g.
    # test_code_intelligence_graph.py::test_sync_prolog_file) sync a
    # tmp_path .pl file, creating :Rule nodes stamped with
    # `file_id=<pytest tmp dir>/rules.pl` and a UUID `id` (no `_test_`
    # marker, no teardown) -- file_id is a husk/fingerprint the id/path/
    # file_path clauses above never checked, so 172 live Rule nodes under
    # `/tmp/pytest-of-*` leaked past every existing sweep. Same STARTS
    # WITH/CONTAINS pair as `_file_path` above.
    "OR (n.file_id IS NOT NULL AND (n.file_id STARTS WITH '/tmp' "
    "OR n.file_id CONTAINS '/logs/e2e-runs/')) "
    # Pre-canonicalisation Session leakage (session 241): tests merged
    # :Session on bare ids ('test-sync-<hex>', 'test_<hex>', 'test_<hex>_sess')
    # with no session_id — writers now converge on session_id, so a
    # test-prefixed id-only Session node is always stale.
    "OR (n:Session AND n.session_id IS NULL AND n.id IS NOT NULL "
    "AND n.id STARTS WITH 'test')"
    # Legacy AuditEntry husks (2026-07-15): before the 2026-07-12 test_id
    # scoping fix, audit tests wrote literal entry_ids with no test_id.
    # Accumulated duplicates break the canonical audit_entry_entry_id
    # uniqueness constraint (ensure_schema) suite-wide. Production entry
    # ids are UUID-shaped and never match these literals.
    "OR (n:AuditEntry AND n.test_id IS NULL "
    "AND n.entry_id IN ['e1', 'e2', 'entry-1'])"
    # Orphaned carryover WorkItems (2026-07-26 audit): test_session_neo4j_
    # store_coverage.py::test_full_lifecycle called persist_carryover twice
    # per run with literal text 'do work' and a random `w-<6 hex>` id, then
    # its finally-block DETACH DELETE listed only [sid, sid2, pid] — never
    # the WorkItem ids. cleanup_session severed the CARRIED_OVER_FROM/_TO
    # edges, leaving the WorkItem an isolated husk with no FK property
    # (340 accumulated = 170 runs; anchors unrecoverable by construction).
    # The test is fixed to delete its WorkItems; this fingerprint clears the
    # backlog. Production carryover items carry real handoff text — the
    # literal 'do work' + isolation guard cannot match them.
    "OR (n:WorkItem AND n.text = 'do work' AND n.id STARTS WITH 'w-' "
    "AND NOT (n)--())"
    # Project-node pollution (research-ast-json-vs-neo4j.md work item 7,
    # 2026-07-18 audit): three independent fingerprints, none of which the
    # id-marker clause above catches, because Project's canonical merge key
    # is `slug`, not `id` (canonical_schema.py) — `GraphSyncer` is the only
    # writer that ever sets `.id`, and it never stamps a `_test_` marker
    # into it.
    #   1. `slug` carries the SAME `_test_` marker convention as the id
    #      clause above (case-sensitive, same reason: `test_id` fixtures
    #      are stamped `_TEST_<hex>` — uppercase — precisely so raw
    #      embedding of `test_id` into an unrelated string, e.g. a survivor
    #      probe's slug, can never accidentally satisfy a lowercase-only
    #      marker check), just on the OTHER property — pytest fixtures that
    #      seed Project nodes stamp `slug`, not `id` (verified live: 1456
    #      of 1513 pollution nodes).
    #   2. `root_path` under /tmp — ephemeral tmp-workspace syncs (3 nodes).
    #   3. `id IS NULL` (never AST-synced as a real source project) AND no
    #      `DEFINES` edge — un-cleaned ad-hoc dogfood/demo pipeline runs
    #      (`kg_design.ensure_project()` MERGEs a throwaway Project per
    #      intent-derived slug, e.g. "build-an-adder-cli-<hex>", each with
    #      at most one scratch `HAS_SPEC`/`HAS_TASK` child). The `NOT
    #      DEFINES` guard deliberately EXEMPTS two live outlier id-less
    #      nodes (`slug=''` and a redis-streams-intent slug) that legitimately
    #      DEFINE thousands of real CodeClass/CodeFunction nodes — a
    #      caller bug (wrong/empty `project_slug` passed to a code-trace
    #      sync) that needs data reconciliation, not a pollution sweep;
    #      deleting them would sever real trace edges (2026-07-18 audit,
    #      not swept — see the builder report for this session).
    "OR ('Project' IN labels(n) "
    "AND coalesce(n.id, '') <> 'orchestrata-ultimate' "
    "AND coalesce(n.slug, '') <> 'orchestrata-ultimate' "
    "AND (coalesce(n.slug, '') CONTAINS '_test_' "
    "OR (n.root_path IS NOT NULL AND n.root_path STARTS WITH '/tmp') "
    "OR (n.id IS NULL AND NOT (n)-[:DEFINES]-())))"
)


def _load_dotenv() -> None:
    """Load the RUL-019 two-phase environment contract.

    Delegates to ``core.env_bootstrap.load_env_contract`` — THE canonical
    loader — instead of parsing ``.env`` here. The hand-rolled copy this
    replaced (byte-identical in four scripts) stopped after phase 1, and
    ``.env`` is the prod-BASE layer: it carries ``CXP_GRPC_HOST=localhost``
    for the host user-units, so a script that loads it and stops resolves
    PRODUCTION endpoints while believing it is on the test fleet. The
    canonical loader then applies the ``ORCHESTRATA_ENV_PROFILE`` overlay,
    which is the half that points the process at the right infrastructure,
    and honours ``declare_owned`` so it cannot reset a test worker's
    per-slot isolation.
    """
    load_env_contract()


def _driver(env: Mapping[str, str]) -> Driver:
    """Build a Neo4j driver from the given env mapping (no hardcoded infra)."""
    from neo4j import GraphDatabase

    uri = env.get("KG_ROUTING_URI") or env.get("KG_URI")
    if not uri:
        raise RuntimeError(
            "KG_ROUTING_URI or KG_URI must be set in the environment "
            "(no hardcoded infra defaults — see .env)"
        )
    # No auth: every remaining graph endpoint negotiates none. The credential
    # keys this used to require were retired with the Neo4j containers.
    return GraphDatabase.driver(uri, auth=None)


def _all_uris(env: Mapping[str, str]) -> list[str]:
    """Every distinct graph URI the suite touches.

    Once the per-worker Neo4j fleet (``NEO4J_URI_GW<N>``) was destroyed on
    2026-07-27 there is exactly one endpoint: the orchgraph cluster. Worker
    isolation moved from "one instance per slot" to "one DATABASE per slot"
    (``orchkg_w<N>``), which ``_all_kg_targets`` enumerates — so this returns
    the single routing URI rather than a fleet.
    """
    uri = env.get("KG_ROUTING_URI") or env.get("KG_URI")
    return [uri] if uri else []


def _session_database(env: Mapping[str, str]) -> str:
    """Resolve the target database for a sweep session from ``env``.

    Mirrors ``core.graph_factory.resolve_kg_database`` but reads the
    injected ``env`` mapping directly (this script stays a standalone
    stdlib+neo4j tool — no ``orchestrata_ultimate`` import — matching its
    existing style).

    Falls back to ``"neo4j"`` ONLY on the Neo4j engine, where that is the
    Community edition's single database. On ``KG_ENGINE=memgraph`` there is
    no database of that name and, worse, an unset ``KG_DATABASE`` would send
    a DELETE sweep at the cluster's default database — the shared semantic
    graph — so a missing value fails loudly instead.
    """
    database = env.get("KG_DATABASE")
    if database:
        return database
    if env.get("KG_ENGINE", "neo4j") == "memgraph":
        raise RuntimeError(
            "KG_DATABASE is not set and KG_ENGINE=memgraph — refusing to "
            "sweep the orchgraph cluster's default database (the shared "
            "semantic graph); pass KG_DATABASE in the env mapping"
        )
    return "neo4j"


def _memgraph_worker_count(env: Mapping[str, str]) -> int:
    """Highest worker index in the ``orchkg_w<N>`` database range.

    Read from ``KG_WORKER_COUNT``. This used to be derived from the number of
    ``NEO4J_URI_GW<N>`` keys in .env — the per-worker Neo4j fleet, destroyed
    2026-07-27. That derivation was load-bearing by accident: once those keys
    were removed it would have returned 0, and this sweeper would have cleaned
    ``orchkg`` and ``orchkg_w0`` while silently leaving w1..w7 to rot — the
    exact failure ``_all_uris`` documents (186 AuditEntry husks found on .19
    on 2026-07-15 after a primary-only sweep).

    Raises rather than defaulting (RUL-003): a wrong count here does not fail,
    it under-sweeps, which is invisible until the leaked nodes cause a
    slowdown weeks later.
    """
    raw = env.get("KG_WORKER_COUNT", "").strip()
    if not raw.isdigit():
        raise RuntimeError(
            "KG_WORKER_COUNT is not set to an integer — the sweeper cannot "
            "guess how many orchkg_w<N> worker databases exist, and guessing "
            "low silently leaves worker pollution behind (see .env)"
        )
    return int(raw)


def _all_kg_targets(env: Mapping[str, str]) -> list[tuple[str, str]]:
    """Every distinct ``(uri, database)`` pair the suite touches.

    Neo4j engine (default, ``KG_ENGINE`` unset or ``"neo4j"``): one
    database name (``_session_database``) per physical URI from
    ``_all_uris`` — Community edition has no multi-tenancy, so every
    instance uses the same database name.

    Memgraph engine (``KG_ENGINE=memgraph``): ONE cluster routing URI,
    MANY databases — the production ``orchkg`` plus every worker's
    ``orchkg_w<N>`` (Neo4j -> orchgraph migration Deliverable 4,
    plans/neo4j-to-orchgraph-migration-plan.md Phase 2 Step 2.3), mirroring
    the per-worker mapping tests/conftest.py sets up for test runs.

    ``KG_DATABASES`` (comma-separated) pins the Memgraph database set
    explicitly. It exists because the cutover moved target selection off
    the URI axis and onto the database axis: before it, an env mapping
    could restrict a sweep to one target by naming one URI, and afterwards
    every mapping fanned out across the whole cluster regardless. Callers
    that need a bounded sweep — the tests that prove the primary-vs-worker
    gate, above all — set it. Unset in production, where the full fan-out
    is the point. Pinned targets are trusted as-is (never lease-filtered) --
    only the AUTO-derived ``orchkg_w<N>`` fan-out below is (swarm-efficiency
    Fix 5, ``_leased_elsewhere``): a caller that names a database explicitly
    has already made the call this filter exists to make automatically.
    """
    if env.get("KG_ENGINE", "neo4j") == "memgraph":
        uri = env.get("KG_ROUTING_URI") or env.get("MEMGRAPH_URI")
        if not uri:
            # Pre-existing gap surfaced by pyright while touching this
            # function for Fix 5 (was `str | None` silently returned in the
            # target tuples) -- fail loud here rather than one layer down
            # in `_driver_from_env`'s generic "KG_ROUTING_URI or KG_URI"
            # message, which would otherwise mask that MEMGRAPH_URI was the
            # missing key for this engine.
            raise RuntimeError(
                "KG_ROUTING_URI or MEMGRAPH_URI must be set for KG_ENGINE=memgraph"
            )
        pinned = [
            d.strip() for d in env.get("KG_DATABASES", "").split(",") if d.strip()
        ]
        databases = pinned or (
            ["orchkg"]
            + [
                f"orchkg_w{i}"
                for i in range(_memgraph_worker_count(env) + 1)
                if not _leased_elsewhere(f"orchkg_w{i}", env)
            ]
        )
        return [(uri, database) for database in databases]
    database = _session_database(env)
    return [(uri, database) for uri in _all_uris(env)]


def _match_clause(marker: str | None) -> LiteralString:
    """The node-match clause; a marker narrows to ids containing it.

    ``marker`` exists so the sweeper's own tests can scope every query to
    the nodes THEY seeded (their test_id) — a test must never sweep other
    runs' pollution as a side effect.
    """
    if marker is not None:
        return "MATCH (n) WHERE n.id IS NOT NULL AND n.id CONTAINS $marker"
    return _MATCH_CLAUSE


def count_pollution(
    env: Mapping[str, str] | None = None, *, marker: str | None = None
) -> dict[str, int]:
    """Count stale test-pollution nodes per label. Read-only."""
    env = env if env is not None else os.environ
    driver = _driver(env)
    try:
        with driver.session(database=_session_database(env)) as session:
            result = session.run(
                _match_clause(marker)
                + " RETURN coalesce(labels(n)[0], '<no-label>') AS label, "
                + "count(n) AS c ORDER BY c DESC",
                marker=marker,
            )
            return {record["label"]: record["c"] for record in result}
    finally:
        driver.close()


def sweep_pollution(
    env: Mapping[str, str] | None = None, *, marker: str | None = None
) -> int:
    """DETACH DELETE stale test-pollution nodes in batches. Returns total deleted."""
    env = env if env is not None else os.environ
    driver = _driver(env)
    total = 0
    try:
        with driver.session(database=_session_database(env)) as session:
            while True:
                record = session.run(
                    _match_clause(marker)
                    + " WITH n LIMIT $batch "
                    + "DETACH DELETE n RETURN count(*) AS deleted",
                    batch=BATCH_SIZE,
                    marker=marker,
                ).single()
                deleted = record["deleted"] if record else 0
                total += deleted
                if deleted < BATCH_SIZE:
                    break
        return total
    finally:
        driver.close()


# ---------------------------------------------------------------------------
# Legacy File/Directory pollution (Workstream B3, plans/parallel-collision-
# fix-plan.md, 2026-07-22): File/Directory's canonical merge key is `id`
# (canonical_schema.py CANONICAL_MERGE_KEYS) — every current writer
# (sync.py, kg_context.py, session/kg_sync's _stage_folders.py) already
# MERGEs on `id`. A node missing `id` entirely is a pre-ruling leftover:
# created before the writers converged on `id`, it can still collide on the
# legacy bare-`path` constraint or shadow a canonical, id-keyed node once
# the migration lands — exactly the ConstraintValidationFailed incident
# class this plan fixes. Deliberately narrower than the marker/tmp/husk
# fingerprints above: "missing the canonical key" is not itself proof of
# test-run leakage, so this is a SEPARATE opt-in sweep, never folded into
# `_MATCH_CLAUSE` — see the worker-only gating on its two call sites in
# `sweep_if_polluted`/`main` below. Live production sync data on the
# PRIMARY is id-keyed and must survive a heuristic bulk-delete; only
# worker DBs (whose File/Directory nodes are exclusively test-fixture
# writes, never real production syncs) are ever swept by this criterion.
# :TestRun joined this clause on 2026-08-07 for the identical reason.
# gateway/test_runner.py::_store_result CREATEd (not MERGEd) TestRun nodes
# with no `id` — TestRun's canonical merge key — so every gateway test-run
# write left a keyless node in whichever worker DB it ran against. The writer
# is fixed, but the leftovers still fail the canonical-schema conformance
# check for whichever worker slot later lands on that database (measured: gate
# 7 left 2 in orchkg_w2, and tests/scripts/test_kg_schema.py then failed there
# while passing on every other slot). They are also invisible to the marker /
# `test_id` fingerprints above -- carrying no test_id is precisely why they
# are NOT exempt under RUL-025.
_LEGACY_FILE_DIR_MATCH_CLAUSE = (
    "MATCH (n) WHERE (n:File OR n:Directory OR n:TestRun) AND n.id IS NULL"
)


def _legacy_file_dir_clause(marker: str | None) -> LiteralString:
    """Match clause for id-less File/Directory nodes; ``marker`` narrows to
    ``path`` containing it (mirrors ``_match_clause``'s test-scoping seam —
    File/Directory nodes carry no id here to scope by, so ``path`` is the
    marker-bearing property instead)."""
    if marker is not None:
        return (
            "MATCH (n) WHERE (n:File OR n:Directory OR n:TestRun) "
            "AND n.id IS NULL "
            "AND n.path IS NOT NULL AND n.path CONTAINS $marker"
        )
    return _LEGACY_FILE_DIR_MATCH_CLAUSE


def count_legacy_file_dir_pollution(
    env: Mapping[str, str] | None = None, *, marker: str | None = None
) -> dict[str, int]:
    """Count id-less File/Directory nodes (worker-DB-only criterion; see the
    module comment above ``_LEGACY_FILE_DIR_MATCH_CLAUSE``). Read-only."""
    env = env if env is not None else os.environ
    driver = _driver(env)
    try:
        with driver.session(database=_session_database(env)) as session:
            result = session.run(
                _legacy_file_dir_clause(marker)
                + " RETURN coalesce(labels(n)[0], '<no-label>') AS label, "
                + "count(n) AS c ORDER BY c DESC",
                marker=marker,
            )
            return {record["label"]: record["c"] for record in result}
    finally:
        driver.close()


def sweep_legacy_file_dir_pollution(
    env: Mapping[str, str] | None = None, *, marker: str | None = None
) -> int:
    """DETACH DELETE id-less File/Directory nodes in batches (worker-DB-only
    criterion). Returns total deleted."""
    env = env if env is not None else os.environ
    driver = _driver(env)
    total = 0
    try:
        with driver.session(database=_session_database(env)) as session:
            while True:
                record = session.run(
                    _legacy_file_dir_clause(marker)
                    + " WITH n LIMIT $batch "
                    + "DETACH DELETE n RETURN count(*) AS deleted",
                    batch=BATCH_SIZE,
                    marker=marker,
                ).single()
                deleted = record["deleted"] if record else 0
                total += deleted
                if deleted < BATCH_SIZE:
                    break
        return total
    finally:
        driver.close()


# ---------------------------------------------------------------------------
# DEEP-MARKER pollution (2026-07-26 audit): see the module docstring's
# "fourth, DEEP-MARKER criterion" paragraph for the rationale and the exact
# safety argument for why this is a separate, opt-in (``--deep``) clause
# rather than a fold-in to ``_MATCH_CLAUSE``. Applied to every URI, including
# the primary, ONLY from the pre-suite (nothing-seeded-yet) call site.
# Shared condition body for both the global clause and the marker-scoped
# test seam below — ONE source of truth so the two can never drift.
# 2026-07-26 live-audit addendum: the marker regexes must ALSO check
# `session_id` (873 husk Sessions carry `_TEST_<hex>` there with no `id`
# at all), and the probe prefixes appear with BOTH separators in live data
# (`INTENTPROBE_f1cc...` underscore, older `INTENTPROBE-...` hyphen).
# 2026-07-26 root-connectivity census addendum: husks also carry the marker
# in `spec_id` (178 Specification '001-_test_<hex12>'), `name` (Pattern
# 'TestPattern-archtest-_test_<hex12>'), and `thread_id` (287 Intent/
# PipelineRun 'test_<hex8>' prefix) — properties the id/session_id checks
# never saw. thread_id uses the PREFIX form (no leading underscore).
#
# 2026-10-11 (ADR-0087 item 3): EVERY regex below sits behind the cheap
# ``CONTAINS`` / ``STARTS WITH`` predicate it implies. A full-string ``=~``
# match of ``.*_test_[0-9a-f]{8,}.*`` can only succeed when the value
# contains ``_test_``; ``test[-_]<hex>.*`` can only succeed when the value
# starts with ``test``; and so on. ``cheap AND regex`` therefore returns
# EXACTLY the nodes ``regex`` alone returned (proved node for node by
# tests/scripts/test_sweep_test_pollution.py::test_deep_marker_guarded_
# conditions_match_the_unguarded_oracle), while Memgraph skips the regex
# engine for every value the guard rejects. Measured read-only on the 725,636-
# node test primary: deep count 278 s -> 112 s. Merging the basic and deep
# MATCH clauses into one was measured worse (158 s) and is not done.
_DEEP_MARKER_CONDITIONS: LiteralString = (
    "(n.id IS NOT NULL AND n.id CONTAINS '_test_' "
    "AND n.id =~ '.*_test_[0-9a-f]{8,}.*') "
    "OR (n.id IS NOT NULL AND n.id CONTAINS '_TEST_' "
    "AND n.id =~ '.*_TEST_[0-9a-f]{8,}.*') "
    "OR (n.session_id IS NOT NULL AND n.session_id CONTAINS '_test_' "
    "AND n.session_id =~ '.*_test_[0-9a-f]{8,}.*') "
    "OR (n.session_id IS NOT NULL AND n.session_id CONTAINS '_TEST_' "
    "AND n.session_id =~ '.*_TEST_[0-9a-f]{8,}.*') "
    "OR (n.spec_id IS NOT NULL AND n.spec_id CONTAINS '_test_' "
    "AND n.spec_id =~ '.*_test_[0-9a-f]{8,}.*') "
    "OR (n.name IS NOT NULL AND n.name CONTAINS '_test_' "
    "AND n.name =~ '.*_test_[0-9a-f]{8,}.*') "
    "OR (n.thread_id IS NOT NULL AND n.thread_id STARTS WITH 'test' "
    "AND n.thread_id =~ 'test[-_][0-9a-f]{6,}.*') "
    "OR (n.thread_id IS NOT NULL AND n.thread_id CONTAINS '_test_' "
    "AND n.thread_id =~ '.*_test_[0-9a-f]{8,}.*') "
    "OR (n.thread_id IS NOT NULL AND n.thread_id CONTAINS '_TEST_' "
    "AND n.thread_id =~ '.*_TEST_[0-9a-f]{8,}.*') "
    "OR (n.test_id IS NOT NULL) "
    # A `_test_`-prefixed PROJECT namespace. The id patterns above require
    # `_test_` to be followed IMMEDIATELY by 8+ hex, so they miss the
    # `_test_<prefix>_<hex>` shape that test fixtures actually generate --
    # e.g. tests/scripts/test_sync_ast_to_graph.py builds project ids as
    # f"_test_satn_{uuid4().hex[:10]}". Those nodes carry NO `test_id`
    # either, so before this clause the sweep reported the primary database
    # "clean" while 30 Package/Module/Class nodes of pure test exhaust sat
    # in it, indistinguishable from production data to every gate that
    # counts code-structure nodes. No real project id starts with `_test_`,
    # so the prefix is unambiguous.
    "OR (n.project IS NOT NULL AND n.project STARTS WITH '_test_' "
    "AND n.project =~ '_test_.*') "
    # The project's own `test_id` fixture convention produces identifiers of
    # the form `test_<hex>_<suffix>` (e.g. `test_4bdb01ea_sess`,
    # `test_4bdb01ea_agent`). The infix patterns above miss them — they have
    # no leading segment before `_test_` — and the Session branch below only
    # catches them when they are degree-0, which a Memory/Agent-linked
    # Session is not. A LEADING `test[-_]<hex6+>` on an identity property is
    # unambiguous test exhaust whatever the label or degree; matched on
    # identity properties only (never `name`), since real code-intelligence
    # ids are project-qualified paths and never start with `test_`.
    "OR (n.id IS NOT NULL AND n.id STARTS WITH 'test' "
    "AND n.id =~ 'test[-_][0-9a-f]{6,}.*') "
    "OR (n.session_id IS NOT NULL AND n.session_id STARTS WITH 'test' "
    "AND n.session_id =~ 'test[-_][0-9a-f]{6,}.*') "
    "OR (n.alias IS NOT NULL AND n.alias STARTS WITH 'test' "
    "AND n.alias =~ 'test[-_][0-9a-f]{6,}.*') "
    "OR (n.task_id IS NOT NULL AND n.task_id STARTS WITH 'test' "
    "AND n.task_id =~ 'test[-_][0-9a-f]{6,}.*') "
    "OR (n:FactoryBuildResult AND NOT (n)--() "
    "AND n.task_id STARTS WITH 'T' AND n.task_id CONTAINS '-' "
    "AND n.task_id =~ 'T[0-9]+-[0-9a-f]{8,}') "
    "OR any(l IN labels(n) WHERE (l STARTS WITH 'CanonTest_test_' "
    "AND l =~ 'CanonTest_test_[0-9a-f]+') "
    "OR l IN ['DLTest', '__NqcAllowWritesTest']) "
    "OR (n:Session AND n.session_id IS NOT NULL AND ("
    "(n.session_id STARTS WITH 'INTENTPROBE' "
    "AND n.session_id =~ 'INTENTPROBE[-_].*') "
    "OR (n.session_id STARTS WITH 'POLICYPROBE' "
    "AND n.session_id =~ 'POLICYPROBE[-_].*') "
    "OR (n.session_id STARTS WITH 'AUDITPROBE' "
    "AND n.session_id =~ 'AUDITPROBE[-_].*') "
    "OR (n.session_id STARTS WITH 'kgtest' "
    "AND n.session_id =~ 'kgtest[-_].*') "
    "OR (((n.session_id STARTS WITH 'test' AND n.session_id =~ 'test[-_].*') "
    "OR n.session_id STARTS WITH 'sess-' "
    "OR n.session_id STARTS WITH 'conn-') AND NOT (n)--())))"
    # D2 (plans/hub-link-plan.md §4.12, 2026-07-30): every :Fact ever written
    # carries source='test' with {'i': N} loop-fixture data — log_fact has no
    # production caller. The self-declared source marker is the fingerprint;
    # none of the id/session_id patterns above ever matched these.
    "OR (n:Fact AND n.source = 'test')"
)

_DEEP_MARKER_CLAUSE: LiteralString = "MATCH (n) WHERE " + _DEEP_MARKER_CONDITIONS


def _deep_marker_clause(marker: str | None) -> LiteralString:
    """Match clause for the DEEP-MARKER pollution fingerprints; ``marker``
    narrows to a ``probe`` property containing it (mirrors
    ``_legacy_file_dir_clause``'s test-scoping seam). Unlike ``_match_
    clause``'s marker mode — which fully REPLACES the fingerprint with a
    simple id-contains check — this ANDS the ``probe`` scope onto the real
    fingerprint conditions, so a scoped test both stays isolated from
    concurrent runs' data AND proves the real DEEP-MARKER logic, not just
    the scoping seam (a seeded node must satisfy the real fingerprint, not
    merely carry the marker, to be matched)."""
    if marker is not None:
        return (
            "MATCH (n) WHERE n.probe IS NOT NULL AND n.probe CONTAINS $marker "
            "AND (" + _DEEP_MARKER_CONDITIONS + ")"
        )
    return _DEEP_MARKER_CLAUSE


def count_deep_marker_pollution(
    env: Mapping[str, str] | None = None, *, marker: str | None = None
) -> dict[str, int]:
    """Count DEEP-MARKER pollution nodes per label (see the module docstring
    and ``_DEEP_MARKER_CLAUSE``). Read-only."""
    env = env if env is not None else os.environ
    driver = _driver(env)
    try:
        with driver.session(database=_session_database(env)) as session:
            result = session.run(
                _deep_marker_clause(marker)
                + " RETURN coalesce(labels(n)[0], '<no-label>') AS label, "
                + "count(n) AS c ORDER BY c DESC",
                marker=marker,
            )
            return {record["label"]: record["c"] for record in result}
    finally:
        driver.close()


def sweep_deep_marker_pollution(
    env: Mapping[str, str] | None = None, *, marker: str | None = None
) -> int:
    """DETACH DELETE DEEP-MARKER pollution nodes in batches. Returns total
    deleted. Opt-in (``--deep``) — see the module docstring."""
    env = env if env is not None else os.environ
    driver = _driver(env)
    total = 0
    try:
        with driver.session(database=_session_database(env)) as session:
            while True:
                record = session.run(
                    _deep_marker_clause(marker)
                    + " WITH n LIMIT $batch "
                    + "DETACH DELETE n RETURN count(*) AS deleted",
                    batch=BATCH_SIZE,
                    marker=marker,
                ).single()
                deleted = record["deleted"] if record else 0
                total += deleted
                if deleted < BATCH_SIZE:
                    break
        return total
    finally:
        driver.close()


def _is_worker_target(env: Mapping[str, str], uri: str, database: str) -> bool:
    """True when ``(uri, database)`` is a worker (non-primary) target.

    Neo4j engine: any URI other than ``env["NEO4J_URI"]`` (the primary
    instance). Memgraph engine: any database other than ``"orchkg"`` (the
    production database — every ``orchkg_w<N>`` is a worker database).
    """
    if env.get("KG_ENGINE", "neo4j") == "memgraph":
        return database != "orchkg"
    return uri != (env.get("KG_ROUTING_URI") or env.get("KG_URI"))


def sweep_if_polluted(
    env: Mapping[str, str] | None = None, *, deep: bool = False
) -> tuple[int, str]:
    """Best-effort sweep for suite infra: never raises.

    Sweeps the primary AND every worker target — the graph fleet's
    ``NEO4J_URI_GW*`` instances, or (``KG_ENGINE=memgraph``) every
    ``orchkg_w<N>`` database on the shared orchgraph cluster (Neo4j ->
    orchgraph migration Deliverable 4, plans/neo4j-to-orchgraph-migration-
    plan.md Phase 2 Step 2.3). Returns ``(deleted_count, detail)`` — a
    target that cannot be reached contributes an error note instead of
    aborting the others (a dead KG must not abort the suite; the readiness
    gate reports that separately).

    The legacy id-less File/Directory criterion (``count_legacy_file_dir_
    pollution``/``sweep_legacy_file_dir_pollution``) is folded in for every
    WORKER target only (``_is_worker_target``) — see the module comment
    above ``_LEGACY_FILE_DIR_MATCH_CLAUSE``.

    ``deep=True`` additionally folds in the DEEP-MARKER criterion
    (``count_deep_marker_pollution``/``sweep_deep_marker_pollution``) for
    EVERY target including the primary — see the module docstring's
    "fourth, DEEP-MARKER criterion" paragraph. Callers must only pass
    ``deep=True`` from a moment when nothing has been seeded yet (the
    pre-suite call site in ``start_test_services.py``).
    """
    env = env if env is not None else os.environ
    total = 0
    details: list[str] = []
    for uri, database in _all_kg_targets(env):
        scoped = {**env, "KG_ROUTING_URI": uri, "KG_DATABASE": database}
        label_key = f"{uri}[{database}]"
        is_worker = _is_worker_target(env, uri, database)
        try:
            counts = count_pollution(scoped)
            if is_worker:
                for label, c in count_legacy_file_dir_pollution(scoped).items():
                    counts[label] = counts.get(label, 0) + c
            if deep:
                for label, c in count_deep_marker_pollution(scoped).items():
                    counts[label] = counts.get(label, 0) + c
            if not counts:
                continue
            deleted = sweep_pollution(scoped)
            if is_worker:
                deleted += sweep_legacy_file_dir_pollution(scoped)
            if deep:
                deleted += sweep_deep_marker_pollution(scoped)
            total += deleted
            breakdown = ", ".join(f"{label}={c}" for label, c in counts.items())
            details.append(f"{label_key}: swept {deleted} ({breakdown})")
        except Exception as exc:  # noqa: BLE001 - infra sweep is best-effort
            details.append(f"{label_key}: sweep skipped: {exc}")
    if not details:
        return 0, "clean"
    return total, "; ".join(details)


def main(argv: list[str] | None = None, env: Mapping[str, str] | None = None) -> int:
    """CLI entry point. Dry-run by default; ``--apply`` deletes.

    ``env`` mirrors the same injection seam as ``count_pollution``/
    ``sweep_pollution``/``sweep_if_polluted`` — defaults to the real
    ``os.environ`` (after ``_load_dotenv()`` populates it) so normal CLI use
    is unaffected, but tests can pass an explicit mapping to exercise the
    per-URI unreachable/clean/``--apply`` branches without mutating the real
    process environment (T2 — no bare ``os.environ`` writes in tests).
    """
    _load_dotenv()
    env = env if env is not None else os.environ
    parser = argparse.ArgumentParser(
        description="Sweep stale test-pollution nodes from the operational KG"
    )
    parser.add_argument(
        "--apply",
        action="store_true",
        help="Actually DETACH DELETE the matched nodes (default: report only)",
    )
    parser.add_argument(
        "--deep",
        action="store_true",
        help=(
            "Also sweep the DEEP-MARKER criterion (colon-qualified marker "
            "ids, uppercase _TEST_ ids, test_id-stamped nodes, test-only "
            "labels, probe Sessions) for EVERY URI including the primary — "
            "see the module docstring. Only safe when nothing has been "
            "seeded yet (e.g. the pre-suite call site)."
        ),
    )
    args = parser.parse_args(argv)

    # Legacy id-less File/Directory pollution is a worker-DB-only criterion
    # (see the module comment above _LEGACY_FILE_DIR_MATCH_CLAUSE) — never
    # folded in for the primary target. --deep applies to every target,
    # primary included (see the module docstring's DEEP-MARKER paragraph).
    # Targets are (uri, database) pairs — plural databases only matter for
    # KG_ENGINE=memgraph (_all_kg_targets); the Neo4j engine still yields
    # one target per physical URI, database held constant.
    any_pollution = False
    for uri, database in _all_kg_targets(env):
        scoped = {**env, "KG_ROUTING_URI": uri, "KG_DATABASE": database}
        label_key = f"{uri}[{database}]"
        is_worker = _is_worker_target(env, uri, database)
        try:
            counts = count_pollution(scoped)
            if is_worker:
                for label, c in count_legacy_file_dir_pollution(scoped).items():
                    counts[label] = counts.get(label, 0) + c
            if args.deep:
                for label, c in count_deep_marker_pollution(scoped).items():
                    counts[label] = counts.get(label, 0) + c
        except Exception as exc:  # noqa: BLE001 - report per-DB, keep going
            print(f"{label_key}: unreachable ({exc})")
            continue
        if not counts:
            print(f"{label_key}: clean")
            continue
        any_pollution = True
        total = sum(counts.values())
        print(f"{label_key}: {total} stale test-pollution nodes")
        for label, c in counts.items():
            print(f"  {label:<20} {c}")
        if args.apply:
            deleted = sweep_pollution(scoped)
            if is_worker:
                deleted += sweep_legacy_file_dir_pollution(scoped)
            if args.deep:
                deleted += sweep_deep_marker_pollution(scoped)
            print(f"  deleted {deleted} nodes (batched DETACH DELETE)")

    if not any_pollution:
        print("KG is clean: no test-pollution fingerprints found")
    elif not args.apply:
        print("\nDry run — pass --apply to delete.")
    return 0


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