orchestrata_ultimate.kernel.audit.model

Reference › orchestrata_ultimate › kernel › audit › model

The data model of the audit trail, SPEC-001 story US-008 (REQ-001-US-008-001 to -016). The logger, queue and writers all use it, as does audit_bootstrap.

AuditEventType lists the event kinds: message routing, policy query, escalation, session start, end, pause and resume, resource allocation and deallocation, workflow start, step and end, volk register, deregister and heartbeat, error, config change and compliance check. AuditOutcome records the result: success, failure, denied, allowed, escalated, timeout or partial. AuditEntry is one record with an entry id, event type, timestamp, session id, correlation id (for tracing across hops), volk id, optional outcome, event data, reason and duration in milliseconds; to_dict and from_dict convert it to and from the JSON form used by the file writer and the Redis queue. AuditLogConfig is a dataclass of settings: log directory, graph URI, user, password and database (defaults come from the shared resolvers), file size and age limits for rotation (100 MB, 7 days), retention (90 days), archive compression, the in-memory buffer size (1000 entries) and flush interval (5 seconds), and switches that enable the file and graph writers.

At a glance

Class diagram

Dependency diagram

API

Classes

AuditEventType (enum)

Bases: Enum · Decorators: traces(spec='001', story='US-008', reqs=['REQ-001-US-008-001', 'REQ-001-US-008-002', 'REQ-001-US-008-003', 'REQ-001-US-… · Line 27

Types of audit events (REQ-001-US-008-001 to REQ-001-US-008-007).

AuditOutcome (enum)

Bases: Enum · Decorators: traces(spec='001', story='US-008', reqs=['REQ-001-US-008-002']) · Line 66

Outcome of audited operations.

AuditEntry (dataclass)

Decorators: traces(spec='001', story='US-008', reqs=['REQ-001-US-008-008', 'REQ-001-US-008-009', 'REQ-001-US-008-010']), dataclass · Line 88

Single audit log entry (REQ-001-US-008-008 to REQ-001-US-008-010).

AuditLogConfig (dataclass)

Decorators: traces(spec='001', story='US-008', reqs=['REQ-001-US-008-011', 'REQ-001-US-008-012', 'REQ-001-US-008-013', 'REQ-001-US-…, dataclass · Line 162

Configuration for audit logging.

Source

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

# mypy: ignore-errors
"""Audit event types, outcomes, entry, and configuration models."""

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph_factory import resolve_kg_database, resolve_uri


@traces(
    spec="001",
    story="US-008",
    reqs=[
        "REQ-001-US-008-001",
        "REQ-001-US-008-002",
        "REQ-001-US-008-003",
        "REQ-001-US-008-004",
        "REQ-001-US-008-005",
        "REQ-001-US-008-006",
        "REQ-001-US-008-007",
    ],
)
class AuditEventType(Enum):
    """Types of audit events (REQ-001-US-008-001 to REQ-001-US-008-007)."""

    # Message routing (REQ-001-US-008-001)
    MESSAGE_ROUTING = "message_routing"

    # Policy queries (REQ-001-US-008-002)
    POLICY_QUERY = "policy_query"

    # Escalation events (REQ-001-US-008-003)
    ESCALATION = "escalation"

    # Session lifecycle (REQ-001-US-008-004)
    SESSION_START = "session_start"
    SESSION_END = "session_end"
    SESSION_PAUSE = "session_pause"
    SESSION_RESUME = "session_resume"

    # Resource allocation (REQ-001-US-008-005)
    RESOURCE_ALLOCATION = "resource_allocation"
    RESOURCE_DEALLOCATION = "resource_deallocation"

    # Workflow execution (REQ-001-US-008-006)
    WORKFLOW_START = "workflow_start"
    WORKFLOW_STEP = "workflow_step"
    WORKFLOW_END = "workflow_end"

    # Volk lifecycle (REQ-001-US-008-007)
    VOLK_REGISTER = "volk_register"
    VOLK_DEREGISTER = "volk_deregister"
    VOLK_HEARTBEAT = "volk_heartbeat"

    # Additional events
    ERROR = "error"
    CONFIG_CHANGE = "config_change"
    COMPLIANCE_CHECK = "compliance_check"


@traces(spec="001", story="US-008", reqs=["REQ-001-US-008-002"])
class AuditOutcome(Enum):
    """Outcome of audited operations."""

    SUCCESS = "success"
    FAILURE = "failure"
    DENIED = "denied"
    ALLOWED = "allowed"
    ESCALATED = "escalated"
    TIMEOUT = "timeout"
    PARTIAL = "partial"


@traces(
    spec="001",
    story="US-008",
    reqs=[
        "REQ-001-US-008-008",
        "REQ-001-US-008-009",
        "REQ-001-US-008-010",
    ],
)
@dataclass
class AuditEntry:
    """Single audit log entry (REQ-001-US-008-008 to REQ-001-US-008-010).

    Attributes:
        entry_id: Unique identifier for this entry
        event_type: Type of event being logged
        timestamp: When the event occurred (REQ-001-US-008-008)
        session_id: Session context (REQ-001-US-008-009)
        correlation_id: For tracing across hops (REQ-001-US-008-010)
        volk_id: Agent that triggered the event
        outcome: Result of the operation
        data: Event-specific data
        reason: Human-readable explanation
        duration_ms: Operation duration in milliseconds
    """

    entry_id: str
    event_type: AuditEventType
    timestamp: datetime
    session_id: str
    correlation_id: str
    volk_id: str
    outcome: AuditOutcome | None = None
    data: dict[str, Any] = field(default_factory=dict)
    reason: str | None = None
    duration_ms: int | None = None

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-016"])
    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for serialization."""
        return {
            "entry_id": self.entry_id,
            "event_type": self.event_type.value,
            "timestamp": self.timestamp.isoformat(),
            "session_id": self.session_id,
            "correlation_id": self.correlation_id,
            "volk_id": self.volk_id,
            "outcome": self.outcome.value if self.outcome else None,
            "data": self.data,
            "reason": self.reason,
            "duration_ms": self.duration_ms,
        }

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-016"])
    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> AuditEntry:
        """Create from dictionary."""
        return cls(
            entry_id=data["entry_id"],
            event_type=AuditEventType(data["event_type"]),
            timestamp=datetime.fromisoformat(data["timestamp"]),
            session_id=data["session_id"],
            correlation_id=data["correlation_id"],
            volk_id=data["volk_id"],
            outcome=AuditOutcome(data["outcome"]) if data.get("outcome") else None,
            data=data.get("data", {}),
            reason=data.get("reason"),
            duration_ms=data.get("duration_ms"),
        )


@traces(
    spec="001",
    story="US-008",
    reqs=[
        "REQ-001-US-008-011",
        "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",
    ],
)
@dataclass
class AuditLogConfig:
    """Configuration for audit logging.

    Attributes:
        log_dir: Directory for file-based logs (REQ-001-US-008-012)
        graph_uri: Graph connection URI (REQ-001-US-008-011)
        graph_user: Graph username
        graph_password: Graph password
        graph_database: Graph database name
        max_file_size_mb: Max log file size before rotation (REQ-001-US-008-013)
        max_file_age_days: Max age before rotation (REQ-001-US-008-013)
        retention_days: How long to keep archived logs (REQ-001-US-008-015)
        compress_archives: Whether to compress old logs (REQ-001-US-008-014)
        encrypt_sensitive: Whether to encrypt sensitive data (REQ-001-US-008-016)
        encryption_key: Key for encryption (if enabled)
        buffer_size: In-memory buffer size before flush
        flush_interval_seconds: How often to flush buffer
        enable_graph: Whether to write to the graph
        enable_file: Whether to write to files
    """

    log_dir: Path = field(default_factory=lambda: Path("logs/audit"))
    graph_uri: str = field(default_factory=resolve_uri)
    # No credentials: NEO4J_USERNAME/NEO4J_PASSWORD were retired with the Neo4j
    # containers on 2026-07-27 and every remaining endpoint negotiates no auth.
    graph_user: str = ""
    graph_password: str | None = None
    # Pre-existing defect found while wiring this subsystem into production
    # (2026-07-26): this defaulted to the literal "orchestrata", a database
    # that does not exist on the live graph instance (only "neo4j" did —
    # knowledge_graph.constants.NEO4J_DATABASE was the canonical default used
    # everywhere else in the codebase). Every graph write from this config's
    # default construction therefore failed with
    # Neo.ClientError.Database.DatabaseNotFound and was silently swallowed
    # by GraphAuditWriter.write's broad except — invisible until this wave's
    # enforcement-proof test asserted against real graph. Never caught
    # before because this subsystem had zero production callers.
    #
    # 2026-07-26 (orchgraph cutover): the constant is no longer a safe
    # default either — "neo4j" is a database the new engine does not have.
    # Resolved at CALL time from KG_DATABASE, like every other KG consumer.
    graph_database: str = field(default_factory=resolve_kg_database)
    max_file_size_mb: int = 100
    max_file_age_days: int = 7
    retention_days: int = 90
    compress_archives: bool = True
    encrypt_sensitive: bool = False
    encryption_key: str | None = None
    buffer_size: int = 1000
    flush_interval_seconds: float = 5.0
    enable_graph: bool = True
    enable_file: bool = True
    connection_timeout: int = 30