orchestrata_ultimate.kernel.audit.writers

Reference › orchestrata_ultimate › kernel › audit › writers

The storage backends of the audit trail, SPEC-001 story US-008 (REQ-001-US-008-011 to -018). AuditLogWriter is the abstract base with write, write_batch and close; the logger holds a list of writers and sends each entry (see model) to all of them. The curated kernel page names the graph writer Neo4jAuditWriter; in this source the class is GraphAuditWriter.

• FileAuditWriter appends one JSON line per entry to a dated file in the configured log directory and flushes after each line. _should_rotate triggers when the file reaches the size limit or the date-based path changes; _rotate_log compresses the old file in a worker and _cleanup_old_logs removes archives past the retention period. File errors are logged and make write return False.
• GraphAuditWriter creates an :AuditEntry node with the entry's fields and links it to the matching :Volk node (by alias) and to the :Session node when they exist. _get_repo lazily creates and checks an async graph repository from the configuration; if the driver is missing or the connection fails, it logs and write returns False, so a missing graph never blocks the file writer.

Each writer holds its own lock, and the logger treats a writer failure as a logged error, not an exception.

At a glance

Class diagram

Dependency diagram

API

Classes

AuditLogWriter

Decorators: traces(spec='001', story='US-008', reqs=['REQ-001-US-008-018']) · Line 36

Base class for audit log writers (REQ-008-020).

FileAuditWriter

Bases: AuditLogWriter · Decorators: traces(spec='001', story='US-008', reqs=['REQ-001-US-008-012', 'REQ-001-US-008-013', 'REQ-001-US-008-014', 'REQ-001-US-… · Line 71

File-based audit log writer (REQ-001-US-008-012).

GraphAuditWriter

Bases: AuditLogWriter · Decorators: traces(spec='001', story='US-008', reqs=['REQ-001-US-008-011', 'REQ-001-US-010-013']) · Line 237

Graph-based audit log writer (REQ-001-US-008-011).

Source

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

# mypy: ignore-errors
"""Audit log writers: base, file-based, and graph-based destinations."""

import asyncio
import gzip
import json
import shutil
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph import (
    AsyncGraphRepository,
    GraphRepositoryConfig,
    create_async_graph_repository,
)
from orchestrata_ultimate.kernel.audit._common import (
    GRAPH_DRIVER_AVAILABLE,
    _logger,
    _utc_now,
)
from orchestrata_ultimate.kernel.audit.model import AuditEntry, AuditLogConfig
from orchestrata_ultimate.knowledge_graph.canonical_schema import (
    LABEL_SESSION,
    LABEL_VOLK,
    REL_BY_VOLK,
)


@traces(
    spec="001",
    story="US-008",
    reqs=["REQ-001-US-008-018"],
)
class AuditLogWriter:
    """Base class for audit log writers (REQ-008-020)."""

    @traces(
        spec="001",
        story="US-008",
        reqs=["REQ-001-US-008-018"],
    )
    async def write(self, entry: AuditEntry) -> bool:
        """Write an audit entry. Returns True on success."""
        raise NotImplementedError

    async def write_batch(self, entries: list[AuditEntry]) -> int:
        """Write multiple entries. Returns count of successful writes."""
        count = 0
        for entry in entries:
            if await self.write(entry):
                count += 1
        return count

    async def close(self) -> None:
        """Close the writer and release resources."""


@traces(
    spec="001",
    story="US-008",
    reqs=[
        "REQ-001-US-008-012",
        "REQ-001-US-008-013",
        "REQ-001-US-008-014",
        "REQ-001-US-008-015",
        "REQ-001-US-008-016",
    ],
)
class FileAuditWriter(AuditLogWriter):
    """File-based audit log writer (REQ-001-US-008-012).

    Handles log rotation (REQ-001-US-008-013), compression (REQ-001-US-008-014),
    and retention policies (REQ-001-US-008-015).
    """

    @traces(spec="001", story="US-001-001", reqs=["REQ-001-US-008-012"])
    def __init__(self, config: AuditLogConfig) -> None:
        self.config = config
        self._lock = asyncio.Lock()
        self._current_file: Path | None = None
        self._current_size = 0
        self._file_handle = None

        # Ensure log directory exists
        self.config.log_dir.mkdir(parents=True, exist_ok=True)

    @traces(spec="001", story="US-001-001", reqs=["REQ-001-US-008-012"])
    def _get_current_log_path(self) -> Path:
        """Get path for current log file."""
        date_str = _utc_now().strftime("%Y-%m-%d")
        return self.config.log_dir / f"audit_{date_str}.jsonl"

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-013"])
    def _should_rotate(self) -> bool:
        """Check if log rotation is needed."""
        if self._current_file is None:
            return False

        # Check file size
        if self._current_size >= self.config.max_file_size_mb * 1024 * 1024:
            return True

        # Check if date changed
        current_path = self._get_current_log_path()
        if current_path != self._current_file:
            return True

        return False

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-013"])
    async def _rotate_log(self) -> None:
        """Rotate the current log file."""
        if self._file_handle:
            self._file_handle.close()
            self._file_handle = None

        if self._current_file and self._current_file.exists():
            # Rename with timestamp
            timestamp = _utc_now().strftime("%Y%m%d_%H%M%S")
            archive_name = f"{self._current_file.stem}_{timestamp}.jsonl"
            archive_path = self._current_file.parent / archive_name

            shutil.move(str(self._current_file), str(archive_path))

            # Compress if configured (REQ-001-US-008-014)
            if self.config.compress_archives:
                await self._compress_file(archive_path)

        self._current_file = None
        self._current_size = 0

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-014"])
    async def _compress_file(self, path: Path) -> None:
        """Compress a log file with gzip."""
        compressed_path = path.with_suffix(".jsonl.gz")

        @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-014"])
        def compress() -> None:
            with open(path, "rb") as f_in:
                with gzip.open(compressed_path, "wb") as f_out:
                    shutil.copyfileobj(f_in, f_out)
            path.unlink()

        loop = asyncio.get_event_loop()
        await loop.run_in_executor(None, compress)

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-015"])
    async def _cleanup_old_logs(self) -> None:
        """Remove logs older than retention period (REQ-001-US-008-015)."""
        cutoff = _utc_now() - timedelta(days=self.config.retention_days)

        for log_file in self.config.log_dir.glob("audit_*.jsonl*"):
            try:
                stat = log_file.stat()
                file_time = datetime.fromtimestamp(stat.st_mtime, tz=UTC)
                if file_time < cutoff:
                    log_file.unlink()
                    _logger.info(f"Deleted old audit log: {log_file}")
            except OSError as e:
                _logger.error(f"Error cleaning up {log_file}: {e}")

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-016"])
    def _mask_sensitive_data(self, data: dict[str, Any]) -> dict[str, Any]:
        """Mask sensitive fields in data (REQ-001-US-008-016)."""
        sensitive_keys = {"password", "token", "secret", "key", "credential"}
        masked = {}

        for key, value in data.items():
            if any(s in key.lower() for s in sensitive_keys):
                masked[key] = "***REDACTED***"
            elif isinstance(value, dict):
                masked[key] = self._mask_sensitive_data(value)
            else:
                masked[key] = value

        return masked

    @traces(
        spec="001",
        story="US-008",
        reqs=["REQ-001-US-008-012", "REQ-001-US-008-018"],
    )
    async def write(self, entry: AuditEntry) -> bool:
        """Write an audit entry to file."""
        async with self._lock:
            try:
                # Check rotation
                if self._should_rotate():
                    await self._rotate_log()

                # Get current log path
                log_path = self._get_current_log_path()

                # Open file if needed
                if self._current_file != log_path:
                    if self._file_handle:
                        self._file_handle.close()  # pragma: no cover
                    self._current_file = log_path
                    self._file_handle = open(log_path, "a", encoding="utf-8")
                    self._current_size = (
                        log_path.stat().st_size if log_path.exists() else 0
                    )

                # Prepare entry
                entry_dict = entry.to_dict()
                if self.config.encrypt_sensitive:
                    entry_dict["data"] = self._mask_sensitive_data(
                        entry_dict.get("data", {})
                    )

                # Write entry
                line = json.dumps(entry_dict) + "\n"
                # Invariant: the branch above always opens a handle before
                # the first write (self._current_file starts as None and
                # log_path is never None), so it is open by this point.
                assert self._file_handle is not None
                self._file_handle.write(line)
                self._file_handle.flush()
                self._current_size += len(line.encode("utf-8"))

                return True

            except OSError as e:
                _logger.error(f"Failed to write audit entry to file: {e}")
                return False

    async def close(self) -> None:
        """Close file handles."""
        if self._file_handle:
            self._file_handle.close()
            self._file_handle = None


@traces(spec="001", story="US-008", reqs=["REQ-001-US-008-011", "REQ-001-US-010-013"])
class GraphAuditWriter(AuditLogWriter):
    """Graph-based audit log writer (REQ-001-US-008-011)."""

    @traces(spec="001", story="US-001-001", reqs=["REQ-001-US-008-011"])
    def __init__(self, config: AuditLogConfig) -> None:
        self.config = config
        self._repo: AsyncGraphRepository | None = None
        self._lock = asyncio.Lock()

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-013"])
    async def _get_repo(self) -> AsyncGraphRepository | None:
        """Get or create the owned async graph repository."""
        if self._repo is None:
            if not GRAPH_DRIVER_AVAILABLE:
                _logger.warning("graph driver not available")
                return None

            try:
                # No credential fallback: NEO4J_PASSWORD was retired with the
                # Neo4j containers on 2026-07-27, and the literal "password"
                # it defaulted to was never the credential of any live
                # instance — it could only turn a config gap into an auth
                # failure that reads like an outage.
                _password = self.config.graph_password or ""
                repo = create_async_graph_repository(
                    GraphRepositoryConfig(
                        uri=self.config.graph_uri,
                        auth=(self.config.graph_user, _password),
                        driver_kwargs={
                            "connection_timeout": self.config.connection_timeout,
                        },
                    )
                )
                # Verify connectivity
                await repo.verify_connectivity()
                self._repo = repo
                _logger.info(f"Connected to graph at {self.config.graph_uri}")
            except Exception as e:
                _logger.error(f"Failed to connect to graph: {e}")
                self._repo = None
                return None
        return self._repo

    @traces(
        spec="001",
        story="US-008",
        reqs=["REQ-001-US-008-011", "REQ-001-US-008-018"],
    )
    async def write(self, entry: AuditEntry) -> bool:
        """Write an audit entry to graph asynchronously."""
        repo = await self._get_repo()
        if repo is None:
            return False

        async with self._lock:
            try:
                # Volk's canonical merge key is `alias` (canonical_schema
                # CANONICAL_MERGE_KEYS) -- `entry.volk_id` is supposed to
                # carry the volk's alias string (e.g. "senior_dev", "qa"; see
                # pipeline_kg.py's `MERGE (v:Volk {alias: row.alias})` and
                # gateway/handlers/projects.py's `Volk {alias: ...}` MATCH),
                # but callers have historically passed non-agent strings
                # (e.g. a protocol/schema version like "v1") through this
                # field, unconditionally MERGEing a junk `:Volk` node per bad
                # value (session-254 connectivity audit: 202 live
                # `(:AuditEntry)-[:BY_VOLK]->(:Volk {alias:'v1'})` edges).
                # This writer therefore never MINTS a Volk node itself --
                # it only OPTIONAL MATCHes an existing one (by alias or the
                # legacy `volk_id` property) and attaches BY_VOLK when found.
                #
                # Session was previously handled differently -- `MERGE
                # (s:Session {session_id: ...})` unconditionally MINTED a new
                # orphan `:Session` node whenever `entry.session_id` did not
                # match an existing one (the kernel-runtime SessionManager
                # never attaches `(:Project)-[:HAS_SESSION]->(:Session)`, so a
                # minted Session was itself unreachable from the root
                # `:Project` -- exactly the floating-subgraph failure mode
                # session-254's connectivity gate exists to catch). Session is
                # now OPTIONAL MATCHed exactly like Volk: IN_SESSION is
                # attached only when a real, already-hub-linked Session
                # exists. An AuditEntry whose volk_id AND session_id both fail
                # to resolve is still written, just with neither edge -- a
                # residual floating-node risk this writer cannot close without
                # a new Project-hub relationship for AuditEntry, which is not
                # declared in canonical_schema.py (see kernel/audit/writers.py
                # module docstring / the wiring report that shipped this
                # comment for why that vocabulary addition was deliberately
                # left to the orchestrator rather than invented here).
                # entry.session_id/volk_id are also stored as plain
                # properties on the node itself (previously ONLY used as
                # MERGE-key inputs for the edges below, never persisted) --
                # a genuine gap for any graph-side query_by_session/volk_id
                # equivalent of the file-writer's query methods, found while
                # wiring this writer into production.
                query = f"""
                CREATE (e:AuditEntry {{
                    entry_id: $entry_id,
                    event_type: $event_type,
                    timestamp: datetime($timestamp),
                    session_id: $session_id,
                    volk_id: $volk_id,
                    correlation_id: $correlation_id,
                    outcome: $outcome,
                    data: $data,
                    reason: $reason,
                    duration_ms: $duration_ms
                }})
                WITH e
                OPTIONAL MATCH (v:{LABEL_VOLK})
                WHERE v.alias = $volk_id OR v.volk_id = $volk_id
                FOREACH (_ IN CASE WHEN v IS NOT NULL THEN [1] ELSE [] END |
                    CREATE (e)-[:{REL_BY_VOLK}]->(v)
                )
                WITH e
                OPTIONAL MATCH (s:{LABEL_SESSION} {{session_id: $session_id}})
                FOREACH (_ IN CASE WHEN s IS NOT NULL THEN [1] ELSE [] END |
                    CREATE (e)-[:IN_SESSION]->(s)
                )
                """
                params = {
                    "entry_id": entry.entry_id,
                    "event_type": entry.event_type.value,
                    "timestamp": entry.timestamp.isoformat(),
                    "session_id": entry.session_id,
                    "correlation_id": entry.correlation_id,
                    "volk_id": entry.volk_id,
                    "outcome": entry.outcome.value if entry.outcome else None,
                    "data": json.dumps(entry.data),
                    "reason": entry.reason,
                    "duration_ms": entry.duration_ms,
                }
                await repo.execute_write(
                    query, params, database=self.config.graph_database
                )
                return True

            except Exception as e:
                _logger.error(f"Failed to write audit entry to graph: {e}")
                return False

    async def close(self) -> None:
        """Close graph driver asynchronously."""
        if self._repo:
            await self._repo.close()
            self._repo = None