orchestrata_ultimate.kernel.pipeline_audit

Reference › orchestrata_ultimate › kernel › pipeline_audit

The audit mixin of PipelineRunner, imported only by pipeline. It gives the runner an AuditLogger and a way to write audit records without ever blocking or failing a routing step. It serves SPEC-001 story US-008 (REQ-001-US-008-020).

_init_audit_logger, called from boot, builds the logger through build_default_audit_logger from audit_bootstrap, reading an environment mapping that defaults to the process environment. _start_audit_background starts the logger's flush and worker loops; both production boot paths (the monolithic deployment in cli_main and the kernel packet handler in the agents package) call boot inside a running event loop, so the start succeeds there, while a synchronous caller leaves the logger constructed but not started and its records only buffer. _stop_audit_background stops it. _audit_fire hands a coroutine to the fire-and-forget helper. The pipeline_escalation and pipeline_session mixins use it to record routing decisions, escalations and session lifecycle events. The logger itself is described in the audit package.

At a glance

Class diagram

Dependency diagram

API

Classes

PipelineAuditMixin

Line 32

SPEC-001 US-008 audit-logger lifecycle helpers for PipelineRunner.

Source

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

# mypy: ignore-errors
"""PipelineRunner audit mixin.

Extracted from pipeline.py to keep PipelineRunner under the 300-line class
limit (matches pipeline_session.py / pipeline_escalation.py / pipeline_kg.py).

Wires the SPEC-001 US-008 (REQ-008-*) audit subsystem (``kernel/audit/``,
previously built with zero production callers) into the real, live kernel
lifecycle backing the out-of-process agent-runtime CXP gRPC kernel
(``agents/runtime_main.py`` / ``agents/kernel_handler.py``): message routing
(REQ-008-001), policy queries (REQ-008-002), and escalation (REQ-008-003)
seams live in ``pipeline_escalation.py``; session lifecycle (REQ-008-004)
lives in ``pipeline_session.py``; both call back into this mixin's
``self.audit`` + ``self._audit_fire`` helper. Disabled by default in tests
(``ORCHESTRATA_KERNEL_AUDIT_DISABLED=1`` in ``tests/conftest.py``, mirroring
``ORCHESTRATA_KERNEL_SESSION_DISABLED``); enabled by default in a real boot.
"""

import logging
from collections.abc import Coroutine, Mapping
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.kernel.audit_bootstrap import (
    build_default_audit_logger,
    fire_and_forget_audit,
)

_logger = logging.getLogger(__name__)


class PipelineAuditMixin:
    """SPEC-001 US-008 audit-logger lifecycle helpers for PipelineRunner.

    Expects nothing from the host class beyond attribute storage — unlike
    the session/escalation mixins this one owns its own state
    (``self.audit``, set by ``PipelineRunner.__init__``/``_init_audit_logger``)
    rather than reading state set up by a sibling mixin. Deliberately no
    class-level ``audit:`` annotation here: ``_EscalationHost`` (the
    TYPE_CHECKING-only Protocol in ``pipeline_escalation.py``) already
    declares it for the composed ``PipelineRunner``, and pyright treats two
    independent ``audit`` declarations across sibling mixin bases as an
    incompatible override.
    """

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-020"])
    def _init_audit_logger(self, env: Mapping[str, str] | None = None) -> None:
        """Construct (but do not start) the kernel's :class:`AuditLogger`.

        Called from ``boot()``. ``env`` is the injection seam (defaults to
        ``os.environ`` inside :func:`build_default_audit_logger`) — tests
        pass an explicit mapping to enable audit despite the suite-wide
        ``ORCHESTRATA_KERNEL_AUDIT_DISABLED=1`` default, without
        monkeypatching the environment.
        """
        self.audit = build_default_audit_logger(env)

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-020"])
    def _start_audit_background(self) -> None:
        """Best-effort start the audit logger's background flush/worker
        loops. Both real production boot paths
        (``kernel/cli_main.py::_boot_agent_runtime`` and
        ``agents/kernel_handler.py::build_kernel_packet_handler``'s
        self-boot fallback) call ``boot()`` from within a running asyncio
        event loop, so this reliably starts in production; a synchronous
        test/CLI call site with no running loop leaves the logger
        constructed-but-unstarted (log() calls still buffer safely, see
        AuditLogger.log — never raises, never blocks boot()).
        """
        if self.audit is None:
            return
        self._audit_fire(self.audit.start())

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-020"])
    def _stop_audit_background(self) -> None:
        """Best-effort stop + flush the audit logger on shutdown."""
        if self.audit is None:
            return
        self._audit_fire(self.audit.stop())

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-020"])
    def _audit_fire(self, coro: Coroutine[Any, Any, Any]) -> None:
        """Fire-and-forget an audit coroutine — never blocks, never raises."""
        fire_and_forget_audit(coro)