orchestrata_ultimate.governance.audit

Reference › orchestrata_ultimate › governance › audit

The governance audit trail, SPEC-007 (REQ-007-FR10-056, REQ-007-US-001-006, -011, REQ-007-US-002-004). GovernanceAudit records each PolicyDecision and Violation (types in types) so that a decision can be reviewed later. policy_engine is the importer and writes through it from _record. It is separate from the kernel's general audit package.

The constructor takes an optional graph audit trail (orchestrator.graph_audit), an optional Redis state manager and a key prefix, from constants by default. log_decision builds an entry with a new id, the decision type, action, actor, allowed flag, reason, articles checked, cache-hit flag, duration and timestamp. It writes it to the graph as a graph-side policy decision (outcome allowed or denied, with the Prolog query and result from the decision metadata when present) and pushes the JSON to a Redis list <prefix>:decisions capped at 10000 entries. Failures in either sink are logged and do not raise. log_violation records a violation the same way, query_decisions(limit, actor) returns recent decisions, and get_audit_trail(action_id) returns the entries for one action.

At a glance

Class diagram

Dependency diagram

API

Classes

GovernanceAudit

Decorators: traces(spec='007', story='US-PGOV-001', reqs=['REQ-007-US-001-011', 'REQ-007-US-002-004']) · Line 35

Governance-specific audit trail.

Source

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

# mypy: ignore-errors
"""Governance audit trail for Orchestrata.

Provides governance-specific audit logging via graph and Redis.
"""

# Spec: 007a - Prolog Governance

from __future__ import annotations

import json
import logging
import uuid
from typing import TYPE_CHECKING, Any, cast

if TYPE_CHECKING:
    from orchestrata_ultimate.infra.redis_state import RedisStateManager
    from orchestrata_ultimate.orchestrator.graph_audit import GraphAuditTrail

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.governance.constants import (
    AUDIT_REDIS_PREFIX,
    DEFAULT_AUDIT_QUERY_LIMIT,
)
from orchestrata_ultimate.governance.types import PolicyDecision, Violation

_logger = logging.getLogger(__name__)


@traces(
    spec="007",
    story="US-PGOV-001",
    reqs=["REQ-007-US-001-011", "REQ-007-US-002-004"],
)
class GovernanceAudit:
    """Governance-specific audit trail.

    Wraps GraphAuditTrail with governance domain events.
    Falls back to Redis-only logging when graph is unavailable.
    """

    def __init__(
        self,
        graph: GraphAuditTrail | None = None,
        redis: RedisStateManager | None = None,
        key_prefix: str | None = None,
    ) -> None:
        self._graph = graph
        self._redis = redis
        self._prefix = key_prefix or AUDIT_REDIS_PREFIX

    @traces(
        spec="007",
        story="US-PGOV-001",
        reqs=["REQ-007-FR10-056", "REQ-007-US-001-006", "REQ-007-US-001-011"],
    )
    async def log_decision(self, decision: PolicyDecision) -> None:
        """Log a governance policy decision."""
        entry = {
            "id": str(uuid.uuid4()),
            "type": "policy_decision",
            "decision_type": decision.decision_type.value,
            "action": decision.action,
            "actor": decision.actor,
            "allowed": decision.allowed,
            "reason": decision.reason,
            "articles_checked": decision.articles_checked,
            "cache_hit": decision.cache_hit,
            "duration_ms": decision.duration_ms,
            "timestamp": decision.timestamp.isoformat(),
        }

        # Graph persistent storage
        if self._graph is not None:
            try:
                from orchestrata_ultimate.orchestrator.graph_audit import (
                    DecisionOutcome,
                )
                from orchestrata_ultimate.orchestrator.graph_audit import (
                    PolicyDecision as GraphPolicyDecision,
                )

                await self._graph.log_policy_decision(
                    GraphPolicyDecision(
                        decision_id=str(entry["id"]),
                        session_id=decision.metadata.get("session_id", "unknown"),
                        volk_id=decision.actor,
                        intent=decision.action,
                        tool=decision.metadata.get("tool", "unknown"),
                        outcome=(
                            DecisionOutcome.ALLOWED
                            if decision.allowed
                            else DecisionOutcome.DENIED
                        ),
                        reason=decision.reason,
                        timestamp=decision.timestamp,
                        prolog_query=decision.metadata.get("prolog_query"),
                        prolog_result=decision.metadata.get("prolog_result"),
                        cached=decision.cache_hit,
                    )
                )
            except Exception as exc:
                _logger.warning("Graph decision log failed: %s", exc)

        # Redis fallback/supplementary log
        if self._redis is not None:
            try:
                key = f"{self._prefix}:decisions"
                self._redis.client.lpush(key, json.dumps(entry))
                self._redis.client.ltrim(key, 0, 9999)
            except Exception as exc:
                _logger.debug("Redis decision log failed: %s", exc)

    @traces(
        spec="007",
        story="US-PGOV-002",
        reqs=[
            "REQ-007-FR10-058",
            "REQ-007-FR7-008",
            "REQ-007-US-002-004",
        ],
    )
    async def log_violation(self, violation: Violation) -> None:
        """Log a governance violation."""
        entry = {
            "id": violation.id,
            "type": "violation",
            "violation_type": violation.type.value,
            "action": violation.action,
            "actor": violation.actor,
            "article": violation.article,
            "severity": violation.severity,
            "timestamp": violation.timestamp.isoformat(),
            "details": violation.details,
        }

        if self._graph is not None:
            try:
                await self._graph.log_violation(entry)
            except Exception as exc:
                _logger.warning("Graph violation log failed: %s", exc)

        if self._redis is not None:
            try:
                key = f"{self._prefix}:violations"
                self._redis.client.lpush(key, json.dumps(entry))
                self._redis.client.ltrim(key, 0, 9999)
            except Exception as exc:
                _logger.debug("Redis violation log failed: %s", exc)

    async def query_decisions(
        self,
        limit: int = DEFAULT_AUDIT_QUERY_LIMIT,
        actor: str | None = None,
    ) -> list[dict[str, Any]]:
        """Query recent governance decisions."""
        # Try graph first
        if self._graph is not None:
            try:
                filters: dict[str, Any] = {"event_type": "PolicyDecision"}
                if actor:
                    filters["volk_id"] = actor
                results = await self._graph.query_audit_trail(filters)
                if actor:
                    # GraphAuditTrail.query_audit_trail honours only
                    # event_type/session_id/limit — volk_id is silently
                    # dropped from the Cypher — so apply the actor filter
                    # here to honour the same contract the Redis fallback
                    # below enforces.
                    results = [r for r in results if r.get("volk_id") == actor]
                return results[:limit]
            except Exception as exc:
                _logger.debug("Graph query failed, falling back to Redis: %s", exc)

        # Fallback to Redis
        if self._redis is not None:
            try:
                key = f"{self._prefix}:decisions"
                # lrange() is typed via redis-py's shared sync/async
                # command mixin (ResponseT); RedisStateManager.client is
                # always the sync client, which always returns a plain
                # list at runtime.
                raw_entries = cast(
                    "list[bytes | str]", self._redis.client.lrange(key, 0, limit - 1)
                )
                entries = []
                for raw in raw_entries:
                    if isinstance(raw, bytes):
                        raw = raw.decode()
                    data = json.loads(raw)
                    if actor and data.get("actor") != actor:
                        continue
                    entries.append(data)
                return entries
            except Exception as exc:
                _logger.warning("Redis query failed: %s", exc)

        return []

    @traces(
        spec="007",
        story="US-PGOV-006",
        reqs=["REQ-007-FR10-127"],
    )
    async def get_audit_trail(self, action_id: str) -> list[dict[str, Any]]:
        """Get full audit trail for a specific action."""
        if self._graph is not None:
            try:
                filters = {"correlation_id": action_id}
                return await self._graph.query_audit_trail(filters)
            except Exception as exc:
                _logger.warning(
                    "Graph audit trail query failed: %s", exc, exc_info=True
                )

        # Fallback: search Redis
        if self._redis is not None:
            try:
                entries = []
                for list_key in ["decisions", "violations"]:
                    key = f"{self._prefix}:{list_key}"
                    # See the cast() note in list_decisions() above.
                    raw_entries = cast(
                        "list[bytes | str]", self._redis.client.lrange(key, 0, -1)
                    )
                    for raw in raw_entries:
                        if isinstance(raw, bytes):
                            raw = raw.decode()
                        data = json.loads(raw)
                        if data.get("id") == action_id:
                            entries.append(data)
                return entries
            except Exception as exc:
                _logger.warning(
                    "Redis audit trail search failed: %s", exc, exc_info=True
                )

        return []