orchestrata_ultimate.knowledge_graph.cxp_auditor

Reference › orchestrata_ultimate › knowledge_graph › cxp_auditor

SpikeAuditor writes an audit trail of CXP packet routing into the operational graph (orchkg on orchgraph), through a synchronous graph repository from core.graph (Spec 008, REQ-008-093 to -097). The postman module is its importer. log_packet runs one MERGE on a (:CXPPacket) keyed by message_id and sets its sender, recipient, subject, JSON payload, timestamp, priority and delivery and work status. When the packet has a thread id it also merges a Thread node under a deterministic id from cs.thread_entity_id, links the packet with IN_THREAD, and hangs the thread off the root Project (slug from _project_slug) with HAS_THREAD; without a thread id it logs a warning and writes the packet alone. log_routing adds a ROUTED_TO edge between two packets by message id. _ensure_indexes creates the thread_id lookup index. The audit is a side channel and never breaks routing: every method returns False when the graph is unavailable or the write fails, and _available_now retries the connection at a throttled interval so a transient outage at construction recovers on later use. Labels and relationship names come from canonical_schema.

At a glance

Class diagram

Dependency diagram

API

Classes

SpikeAuditor

Line 37

Logs CXP packets and routing events to the graph.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/knowledge_graph/cxp_auditor.py

# mypy: ignore-errors
"""Spike Auditor — graph audit trail for CXP packet routing."""

from __future__ import annotations

import json
import logging
import os
import time
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph import create_graph_repository
from orchestrata_ultimate.core.graph_factory import resolve_kg_dialect
from orchestrata_ultimate.knowledge_graph import canonical_schema as cs
from orchestrata_ultimate.knowledge_graph.schema import to_memgraph_ddl

if TYPE_CHECKING:
    from orchestrata_ultimate.core.graph import SyncGraphRepository
    from orchestrata_ultimate.core.packet import CXPPacket

_logger = logging.getLogger(__name__)

_DEFAULT_PROJECT_SLUG = "orchestrata-ultimate"


@traces(spec="008", story="US-004", reqs=["REQ-008-093"])
def _project_slug() -> str:
    """Resolve the root ``:Project`` slug this writer's hub edge MERGEs onto.

    Read at call time, never at import (Key Rule / RUL-003): the test suite and
    the CLI both override ``ORCHESTRATA_PROJECT_ID`` per process.
    """
    return os.environ.get("ORCHESTRATA_PROJECT_ID", _DEFAULT_PROJECT_SLUG)


class SpikeAuditor:
    """Logs CXP packets and routing events to the graph.

    Non-fatal: all Neo4j operations are wrapped in try/except.
    If the graph is unavailable, the spike continues without audit.
    """

    def __init__(self, repo: SyncGraphRepository | None = None) -> None:
        """Initialize the auditor.

        Ownership (backlog #163 phase 4): when ``repo`` is None, the auditor
        builds and owns its own repository via ``core.graph.create_graph_repository``
        — ``close()`` on it really closes the underlying driver. When a
        ``repo`` is injected, ownership belongs to whoever built *that* repo
        (shared repos no-op on ``close()``); this class no longer tracks its
        own ``_owns_driver`` flag.
        """
        self._repo: SyncGraphRepository | None = repo
        self._available = False
        self._injected = repo is not None
        self._redial_interval = 30.0
        if repo is not None:
            try:
                repo.verify_connectivity()
                self._available = True
                self._ensure_indexes()
                print("  [Auditor] Using shared graph driver")
            except Exception as e:
                self._available = False
                print(f"  [Auditor] Shared driver unavailable ({e}) — no audit trail")
        else:
            self._connect()
        self._last_dial = time.monotonic()

    @traces(spec="008", story="US-004", reqs=["REQ-008-096"])
    def _available_now(self) -> bool:
        """Audit availability, with a throttled lazy re-dial so a transient graph
        blip at construction is recovered on next use (backlog #219)."""
        if self._available:
            return True
        now = time.monotonic()
        if now - self._last_dial < self._redial_interval:
            return False
        self._last_dial = now
        if self._injected and self._repo is not None:
            try:
                self._repo.verify_connectivity()
                self._available = True
                self._ensure_indexes()
            except Exception as e:
                self._available = False
                print(f"  [Auditor] re-dial failed ({e})")
        else:
            self._connect()
        return self._available

    @traces(spec="008", story="US-004", reqs=["REQ-008-096"])
    def _connect(self):
        try:
            repo = create_graph_repository()
            self._repo = repo
            repo.verify_connectivity()
            self._available = True
            self._ensure_indexes()
            print("  [Auditor] graph connected")
        except Exception as e:
            self._available = False
            print(f"  [Auditor] graph unavailable ({e}) — no audit trail")

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-037"])
    def _ensure_indexes(self):
        """Create the CXPPacket thread_id lookup index.

        The UNIQUE constraint on message_id used to be hand-rolled here too,
        but CXPPacket graduated into ``canonical_schema.CANONICAL_MERGE_KEYS``
        2026-08-06 (REUSING this exact constraint name, ``cxp_message_id``) —
        it is now generated by ``canonical_schema.get_all_ddl()`` and applied
        wherever ``schema.get_unified_ddl()`` runs against this graph. Two
        independent ``CREATE CONSTRAINT`` statements for the same (label,
        prop) is exactly the split-brain ``canonical_schema.py`` exists to
        eliminate, so the hand-rolled copy is retired here; the thread_id
        index below has no canonical counterpart (non-identity lookup) and
        stays hand-rolled.

        Each statement is rendered for the live dialect and run through
        ``execute_autocommit``, NOT a shared ``unit_of_work``: orchgraph/
        Memgraph refuses schema DDL inside a managed/explicit transaction
        ("Index manipulation is not allowed in multicommand transactions" —
        measured live 2026-07-26), which silently left every CXP audit write
        unindexed and un-deduplicated on the new engine. Neo4j accepts DDL
        either way, so this is the portable form.
        """
        repo = self._repo
        if not self._available or repo is None:
            return
        statements = [
            "CREATE INDEX IF NOT EXISTS FOR (p:CXPPacket) ON (p.thread_id)",
        ]
        memgraph = resolve_kg_dialect() == "memgraph"
        for stmt in statements:
            if memgraph:
                translated = to_memgraph_ddl(stmt)
                if translated is None:
                    continue
                stmt = translated
            # gate-ok: a missing index degrades query SPEED, not correctness, and the statement that failed is logged; re-applying is idempotent on the next connect
            try:
                repo.execute_autocommit(stmt)
            except Exception as exc:
                if "already exists" in str(exc).lower():
                    continue
                # Name the statement, not just the exception: with several
                # index DDLs in this loop the old message could not say which
                # one failed. ("Neo4j" was also stale — the Neo4j containers
                # were retired 2026-07-27; this talks to orchgraph/Memgraph.)
                _logger.warning("graph index creation failed for %r: %s", stmt, exc)

    @traces(
        spec="008",
        story="US-004",
        reqs=[
            "REQ-008-093",
            "REQ-008-094",
            "REQ-008-095",
            "REQ-008-096",
            "REQ-008-097",
        ],
    )
    def log_packet(self, packet: CXPPacket) -> bool:
        """Create a (:CXPPacket) node from a routed packet.

        Returns:
            True if the write landed, False if the auditor is unavailable
            or the write itself failed. Non-fatal either way — the audit
            trail is a side-channel, never a reason to break the caller.
        """
        if not self._available_now() or self._repo is None:
            return False
        repo = self._repo
        try:
            ts = packet.timestamp
            if hasattr(ts, "isoformat"):
                ts = ts.isoformat()
            else:
                ts = str(ts)

            payload_str = (
                json.dumps(packet.payload, default=str) if packet.payload else "{}"
            )

            project = _project_slug()
            thread_id = packet.thread_id
            params: dict[str, Any] = {
                "message_id": packet.message_id,
                "thread_id": thread_id,
                "from_agent": packet.from_agent,
                "to_agent": packet.to_agent,
                "subject": packet.subject,
                "payload": payload_str,
                "timestamp": ts,
                "priority": packet.priority,
                "delivery_status": packet.delivery_status.value,
                "work_status": packet.work_status.value,
                "project": project,
            }

            if thread_id:
                thread_qid = cs.thread_entity_id(project, thread_id)
                params["thread_qid"] = thread_qid
                repo.execute_write(
                    "MERGE (p:CXPPacket {message_id: $message_id}) "
                    "SET p.thread_id = $thread_id,"
                    "    p.from_agent = $from_agent,"
                    "    p.to_agent = $to_agent,"
                    "    p.subject = $subject,"
                    "    p.payload = $payload,"
                    "    p.timestamp = $timestamp,"
                    "    p.priority = $priority,"
                    "    p.delivery_status = $delivery_status,"
                    "    p.work_status = $work_status "
                    "WITH p "
                    f"MERGE (t:{cs.LABEL_THREAD} {{id: $thread_qid}}) "
                    "  ON CREATE SET t.thread_id = $thread_id, t.project = $project "
                    f"MERGE (p)-[:{cs.REL_IN_THREAD}]->(t) "
                    "WITH t "
                    f"MERGE (proj:{cs.LABEL_PROJECT} {{slug: $project}}) "
                    f"MERGE (proj)-[:{cs.REL_HAS_THREAD}]->(t)",
                    params,
                )
            else:
                _logger.warning(
                    "CXPPacket %s has no thread_id; writing without thread anchor",
                    packet.message_id,
                )
                repo.execute_write(
                    "MERGE (p:CXPPacket {message_id: $message_id}) "
                    "SET p.thread_id = $thread_id,"
                    "    p.from_agent = $from_agent,"
                    "    p.to_agent = $to_agent,"
                    "    p.subject = $subject,"
                    "    p.payload = $payload,"
                    "    p.timestamp = $timestamp,"
                    "    p.priority = $priority,"
                    "    p.delivery_status = $delivery_status,"
                    "    p.work_status = $work_status",
                    params,
                )
            return True
        except Exception as e:
            _logger.error("log_packet error: %s", e)
            return False

    @traces(spec="008", story="US-004", reqs=["REQ-008-097"])
    def log_routing(
        self, from_msg_id: str, to_msg_id: str, thread_id: str, subject: str
    ) -> bool:
        """Create a [:ROUTED_TO] relationship between specific packets by message_id.

        Returns:
            True if the write landed, False if the auditor is unavailable
            or the write itself failed. Non-fatal either way.
        """
        if not self._available_now() or self._repo is None:
            return False
        repo = self._repo
        try:
            repo.execute_write(
                "MATCH (a:CXPPacket {message_id: $from_id}) "
                "MATCH (b:CXPPacket {message_id: $to_id}) "
                "MERGE (a)-[:ROUTED_TO {subject: $subj, thread_id: $tid}]->(b)",
                {
                    "from_id": from_msg_id,
                    "to_id": to_msg_id,
                    "tid": thread_id,
                    "subj": subject,
                },
            )
            return True
        except Exception as e:
            _logger.error("log_routing error: %s", e)
            return False

    def close(self):
        """Close the repository.

        A no-op when ``self._repo`` wraps a driver it does not own (shared
        semantics live on the repository itself — see ``__init__``).
        """
        if self._repo is not None:
            self._repo.close()
        self._repo = None


# Backward compatibility alias
CXPAuditor = SpikeAuditor