scripts.kg_retention

Reference › scripts › kg_retention

scripts/kg_retention.py is the recurring retention job for the knowledge graph: it ages out gateway audit exhaust and, optionally, CXP exhaust. A scheduled job or a human runs it; tests/scripts/test_kg_retention.py covers it. It connects through create_graph, resolve_uri and resolve_kg_database from orchestrata_ultimate.core.graph_factory.

Options: --retention-days N (default DEFAULT_RETENTION_DAYS) purges GatewayAuditEvent nodes older than N days together with the connection Session nodes they isolate; --cxp-exhaust also adds the clauses for CXP router-storm and marked-fixture packets; --apply deletes, and without it the run is a dry run that only counts. _retention_clauses builds the per-label age-out conditions, count counts the expired nodes per label and purge deletes them. The output lists the expired node count per label and a total, saying "Dry run" when nothing was removed. It writes to the graph only with --apply.

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

#!/usr/bin/env python3
"""Recurring KG retention: age out gateway audit exhaust + CXP exhaust.

Resurrects the RECURRING retention policy that commit e73df607 (P6, retire
dead migration one-shots) deleted along with ``purge_legacy_orphans.py``.
That script bundled two different things: migration-era one-shot purges
(correctly retired — Neo4j destroyed 2026-07-27) and the weekly retention
policy behind ``--gateway-audit-retention-days`` / ``--cxp-exhaust`` that
the ``orchestrata-kg-retention.timer`` systemd user unit runs. This script
is ONLY the recurring half, on the orchgraph seam.

What it removes (both user-ratified 2026-07-26):

1. Gateway audit exhaust older than the retention window: the gateway audit
   writer mints one ``GatewayAuditEvent`` per event plus a UUID ``Session``
   per WebSocket connection (docs/gateway-audit-rollup-2026-07-26.md). Events
   age out by timestamp; the Session clause then reaps the connection
   sessions the event purge isolated. Order matters — events first. Session-
   CLI work sessions and any Session with real edges or a non-UUID id
   survive by construction. NOTE: the dry-run Session count is pre-event-
   purge and understates; ``--apply`` reports true deletions.
2. CXP packet exhaust (with ``--cxp-exhaust``): router test storms
   (chain/loop/back, empty payloads) and unambiguously-marked fixture
   packets (docs/cxp-communication-analysis-2026-07-26.md). Every clause
   requires ``work_status`` pending — no packet that ever progressed
   through a real lifecycle can match.

Dry-run by default; ``--apply`` deletes in batches and verifies a second
pass finds nothing (exit 1 on residue). Idempotent — a clean graph reports
0 everywhere. Operational KG database only (``resolve_kg_database``).
"""

from __future__ import annotations

import argparse
import sys
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING, cast

from orchestrata_ultimate.core.graph_factory import (
    NO_AUTH,
    create_graph,
    resolve_kg_database,
    resolve_uri,
)

if TYPE_CHECKING:
    from typing import LiteralString  # noqa: F401  (cast target)

BATCH_SIZE = 1000

DEFAULT_RETENTION_DAYS = 30

_GATEWAY_UUID_RE = "'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}'"

_CXP_EXHAUST_CLAUSES: dict[str, str] = {
    "CXPPacket": (
        "(n.subject IN ['chain', 'loop', 'back', 'greeting', 'ack', ''] "
        "OR n.to_agent = 'U:Target' "
        "OR n.to_agent STARTS WITH 'TEST:' "
        "OR n.subject STARTS WITH 'ACK: ') "
        "AND coalesce(n.work_status, 'pending') = 'pending'"
    ),
}


def _retention_clauses(days: int) -> dict[str, str]:
    """Age-out clauses for gateway audit exhaust older than ``days``.

    ISO-8601 strings compare lexicographically; GatewayAuditEvent timestamps
    are written ISO+00:00, so a string comparison against the cutoff is
    exact. Events age out first; the Session clause then reaps the
    connection sessions the event purge isolated.
    """
    cutoff = (datetime.now(UTC) - timedelta(days=days)).isoformat()
    return {
        "GatewayAuditEvent": f"n.timestamp < '{cutoff}'",
        "Session": "NOT (n)--() AND n.session_id =~ " + _GATEWAY_UUID_RE,
    }


def count(session, label: str, clauses: dict[str, str]) -> int:
    q = cast(
        "LiteralString",
        f"MATCH (n:{label}) WHERE {clauses[label]} RETURN count(n) AS c",
    )
    return session.run(q).single()["c"]


def purge(session, label: str, clauses: dict[str, str]) -> int:
    deleted = 0
    while True:
        q = cast(
            "LiteralString",
            f"MATCH (n:{label}) WHERE {clauses[label]} "
            f"WITH n LIMIT {BATCH_SIZE} DETACH DELETE n RETURN count(*) AS c",
        )
        c = session.run(q).single()["c"]
        deleted += c
        if c < BATCH_SIZE:
            return deleted


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=(__doc__ or "").splitlines()[0])
    parser.add_argument(
        "--retention-days",
        type=int,
        default=DEFAULT_RETENTION_DAYS,
        metavar="N",
        help="purge GatewayAuditEvent nodes older than N days, plus the "
        f"connection Sessions they isolate (default {DEFAULT_RETENTION_DAYS})",
    )
    parser.add_argument(
        "--cxp-exhaust",
        action="store_true",
        help="also purge CXP router-storm + marked-fixture packets "
        "(docs/cxp-communication-analysis-2026-07-26.md)",
    )
    parser.add_argument(
        "--apply",
        action="store_true",
        help="delete matches (default: dry-run counts only)",
    )
    args = parser.parse_args(argv)

    clauses = _retention_clauses(args.retention_days)
    if args.cxp_exhaust:
        clauses |= _CXP_EXHAUST_CLAUSES

    driver = create_graph(uri=resolve_uri(), auth=NO_AUTH)
    rc = 0
    with driver.session(database=resolve_kg_database()) as s:
        total = 0
        for label in clauses:
            n = count(s, label, clauses)
            total += n
            print(f"{label:>17}: {n} expired node(s)")

        if not args.apply:
            print(
                f"\nDry run — {total} node(s) WOULD be removed "
                "(Session count understates pre-event-purge). "
                "Pass --apply to delete."
            )
            driver.close()
            return rc

        for label in clauses:
            deleted = purge(s, label, clauses)
            if deleted:
                print(f"{label:>17}: deleted {deleted}")

        # verify — a second pass must find nothing
        residue = sum(count(s, label, clauses) for label in clauses)
        if residue:
            print(f"FAIL: residue after apply: {residue} node(s)", file=sys.stderr)
            rc = 1
        else:
            print("verified: 0 residue")
    driver.close()
    return rc


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