orchestrata_ultimate.kernel.audit_bootstrap

Reference › orchestrata_ultimate › kernel › audit_bootstrap

The shared boot-time helpers of the audit package. They build the process's AuditLogger and let synchronous code write audit records without blocking or failing. agents.runtime_main, agents.volk_registry, pipeline_audit and pipeline_session import it.

audit_disabled returns True when the audit-disable environment variable holds 1, true or yes; it reads an explicit mapping when one is given. build_default_audit_logger returns None when audit is disabled or when constructing an AuditLogger with the default AuditLogConfig fails, so audit can never stop a caller from booting; construction does not start the background loops. fire_and_forget_audit schedules a coroutine on the running event loop and only logs any exception it raises; with no running loop it closes the coroutine and skips the write. It keeps a reference to each task until completion. drain_audit_tasks waits for all those in-flight tasks, up to a timeout (30 seconds by default), and returns how many finished; it never raises and logs a warning if tasks remain. Shutdown paths such as ashutdown in pipeline_session use it so that closing records are not lost when the loop closes.

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/src/orchestrata_ultimate/kernel/audit_bootstrap.py

# mypy: ignore-errors
"""Shared boot-time construction + fire-and-forget helpers for kernel.audit.

SPEC-001 US-008 (REQ-008-*) built a complete audit-logging subsystem
(``kernel/audit/``) with zero production callers — every RTM row citing it
was false-IMPL. This module is the single, reusable construction site both
real production kernel surfaces use to build their :class:`AuditLogger`:

- :class:`~orchestrata_ultimate.agents.volk_registry.VolkRegistry` (used by
  BOTH ``kernel/cli_main.py``'s monolithic boot AND, more importantly, by
  ``orchestrator.pipeline_runner.PipelineRunner`` — the gateway's live,
  user-facing path — for every stage-volk registration).
- :class:`~orchestrata_ultimate.kernel.pipeline.PipelineRunner` (backs the
  out-of-process agent-runtime CXP gRPC kernel, ``agents/runtime_main.py`` /
  ``agents/kernel_handler.py``).

Disabled by default in the test suite (``tests/conftest.py`` sets
``ORCHESTRATA_KERNEL_AUDIT_DISABLED=1`` the same way it already disables the
session subsystem) so the ~1400-module suite gets zero new filesystem/Neo4j
side effects from this wiring; enabled by default in a real boot.
"""

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

from orchestrata_ultimate.kernel.audit.logger import AuditLogger
from orchestrata_ultimate.kernel.audit.model import AuditLogConfig

logger = logging.getLogger(__name__)

AUDIT_DISABLED_ENV_VAR = "ORCHESTRATA_KERNEL_AUDIT_DISABLED"


def audit_disabled(env: Mapping[str, str] | None = None) -> bool:
    """True when the audit subsystem is disabled for this process.

    ``env`` is the injection seam (defaults to ``os.environ`` at call time via
    the caller — this function accepts an explicit mapping so it never reads
    ``os.environ`` directly, matching the no-bare-``os.environ[...]`` house
    rule and letting tests override it without monkeypatch).
    """
    import os

    source = env if env is not None else os.environ
    return source.get(AUDIT_DISABLED_ENV_VAR, "").lower() in ("1", "true", "yes")


def build_default_audit_logger(
    env: Mapping[str, str] | None = None,
) -> AuditLogger | None:
    """Best-effort construct the process's :class:`AuditLogger`.

    Returns ``None`` (today's behavior — audit off) when disabled via
    :data:`AUDIT_DISABLED_ENV_VAR`, or when construction fails for any reason
    (this must never prevent a caller from booting). Construction alone does
    NOT start the background flush/worker loops — call
    :func:`fire_and_forget_audit` with ``logger.start()``/``logger.stop()``
    from a caller that has a running event loop.
    """
    if audit_disabled(env):
        return None
    try:
        return AuditLogger(config=AuditLogConfig())
    except Exception as exc:  # construction must never break kernel boot
        logger.warning("audit logger construction failed; audit disabled: %s", exc)
        return None


# Strong references to in-flight audit tasks.
#
# `loop.create_task` returns the ONLY strong reference to the task; the event
# loop itself keeps just a weak one. A task whose handle is dropped can be
# garbage-collected mid-execution, so the pre-2026-07-26 code
# (`task.add_done_callback(...)` and nothing else — a done-callback does not
# keep a task alive) could silently lose audit writes under memory pressure.
# In a compliance subsystem a dropped write is the one failure mode that must
# not be silent, so the handle is held until the task completes.
_pending_audit_tasks: set[asyncio.Task[Any]] = set()


def _log_audit_task_result(task: asyncio.Task[Any]) -> None:
    _pending_audit_tasks.discard(task)
    if task.cancelled():
        return
    exc = task.exception()
    if exc is not None:
        logger.warning("audit background task failed: %s", exc)


def fire_and_forget_audit(coro: Coroutine[Any, Any, Any]) -> None:
    """Schedule *coro* without ever blocking or raising into the caller.

    Mirrors the dispatcher's ``policy_audit_mode`` philosophy (audit
    failures log a warning and never affect the hot path): when a running
    event loop is available, the coroutine is scheduled via
    ``loop.create_task`` and any exception it raises is only logged (never
    propagated). When no loop is running (e.g. a synchronous test/CLI call
    site), the coroutine is closed immediately — audit is silently skipped,
    which is the same "zero behavior change" outcome as audit being disabled.

    The task handle is retained until completion (see
    :data:`_pending_audit_tasks`) so it cannot be collected mid-write, and so
    :func:`drain_audit_tasks` can wait for it.
    """
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        coro.close()
        return
    task = loop.create_task(coro)
    _pending_audit_tasks.add(task)
    task.add_done_callback(_log_audit_task_result)


async def drain_audit_tasks(timeout: float = 30.0) -> int:
    """Wait for every in-flight fire-and-forget audit task, return how many.

    Audit writes are scheduled from SYNCHRONOUS call sites (``register_volk``,
    ``_session_log``, the dispatcher's audit-mode hooks), so a caller that
    needs the write to be observable — a shutdown path flushing before the
    loop closes, or a test asserting that a seam is wired — otherwise has
    nothing to await and must poll until a scheduling budget expires. Polling
    turns CPU starvation under load into a reported defect that does not
    exist; that is what made
    ``tests/kernel/test_audit_logger_enforcement_proof.py`` flake through both
    a 10s and a 45s budget on 2026-07-26.

    Never raises: a task that fails has already been logged by
    :func:`_log_audit_task_result`, and exceeding *timeout* returns the count
    drained so far rather than turning a slow audit write into a caller error.
    """
    drained = 0
    while _pending_audit_tasks:
        batch = list(_pending_audit_tasks)
        done, _ = await asyncio.wait(batch, timeout=timeout)
        drained += len(done)
        if len(done) < len(batch):
            # Timed out with work still in flight — report, do not raise.
            logger.warning(
                "drain_audit_tasks timed out with %d task(s) still in flight",
                len(batch) - len(done),
            )
            break
    return drained