orchestrata_ultimate.governance.violations

Reference › orchestrata_ultimate › governance › violations

The violation log and rate counter, SPEC-007 stories FR-PGOV-007 (REQ-007-FR7-001, -002), US-PGOV-006 and US-PGOV-002. ViolationTracker records each governance Violation (defined in types) and answers how many are happening per minute. policy_engine imports it and records a violation on each denial; the rate it reports is what trips safe_mode.

record gives the violation an id if it has none, adds it as JSON to a Redis sorted set scored by time, increments a rate counter that expires after the window (SAFE_MODE_WINDOW_SECONDS), and, if a graph audit trail is configured, logs it there too. Redis and graph failures are logged and do not stop the call. It logs a warning with the id, type, action, actor and severity and returns the id. get_violations reads recent violations filtered by type and actor, get_rate converts the windowed counter to violations per minute (zero on error or no data), should_trigger_safe_mode compares the rate with a threshold of 3 per minute by default, and clear empties the log. The Redis keys can be overridden by constructor arguments.

At a glance

Class diagram

Dependency diagram

API

Classes

ViolationTracker

Decorators: traces(spec='007', story='US-PGOV-002', reqs=['REQ-007-US-002-009', 'REQ-007-FR7-001', 'REQ-007-US-006-003']) · Line 35

Tracks governance violations.

Source

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

# mypy: ignore-errors
"""Violation tracking for Orchestrata governance.

Records violations in Redis (counters) and graph (persistent log).
"""

from __future__ import annotations

import json
import logging
import uuid
from typing import TYPE_CHECKING, 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 (
    DEFAULT_AUDIT_QUERY_LIMIT,
    REDIS_VIOLATION_PREFIX,
    REDIS_VIOLATION_RATE_KEY,
    SAFE_MODE_WINDOW_SECONDS,
)
from orchestrata_ultimate.governance.types import Violation, ViolationType

_logger = logging.getLogger(__name__)


@traces(
    spec="007",
    story="US-PGOV-002",
    reqs=["REQ-007-US-002-009", "REQ-007-FR7-001", "REQ-007-US-006-003"],
)
class ViolationTracker:
    """Tracks governance violations.

    Uses Redis for real-time counting and rate calculation.
    Uses graph for persistent violation audit trail.
    """

    def __init__(
        self,
        redis: RedisStateManager,
        graph: GraphAuditTrail | None = None,
        key_prefix: str | None = None,
        rate_key: str | None = None,
    ) -> None:
        self._redis = redis
        self._graph = graph
        self._log_key = f"{key_prefix or REDIS_VIOLATION_PREFIX}:log"
        self._rate_key = f"{rate_key or REDIS_VIOLATION_RATE_KEY}:counter"

    @traces(
        spec="007",
        story="FR-PGOV-007",
        reqs=[
            "REQ-007-FR7-001",
            "REQ-007-FR7-002",
        ],
    )
    async def record(self, violation: Violation) -> str:
        """Record a governance violation.

        Stores in Redis for rate tracking (sync client) and optionally in
        graph for the persistent audit trail (async — must be awaited).
        Returns the violation ID.
        """
        if not violation.id:
            violation.id = str(uuid.uuid4())

        # Store in Redis sorted set (score = timestamp)
        violation_data = {
            "id": violation.id,
            "type": violation.type.value,
            "action": violation.action,
            "actor": violation.actor,
            "article": violation.article,
            "severity": violation.severity,
            "timestamp": violation.timestamp.isoformat(),
            "details": violation.details,
        }
        try:
            key = self._log_key
            score = violation.timestamp.timestamp()
            self._redis.client.zadd(key, {json.dumps(violation_data): score})
            # Increment rate counter
            rate_key = self._rate_key
            pipe = self._redis.client.pipeline()
            pipe.incr(rate_key)
            pipe.expire(rate_key, SAFE_MODE_WINDOW_SECONDS)
            pipe.execute()
        except Exception as exc:
            _logger.error("Failed to record violation in Redis: %s", exc)

        # Store in graph for persistent trail
        if self._graph is not None:
            try:
                await self._graph.log_violation(violation_data)
            except Exception as exc:
                _logger.warning("Failed to log violation to graph: %s", exc)

        _logger.warning(
            "Violation recorded: id=%s type=%s action=%s actor=%s severity=%s",
            violation.id,
            violation.type.value,
            violation.action,
            violation.actor,
            violation.severity,
        )
        return violation.id

    @traces(
        spec="007",
        story="FR-PGOV-007",
        reqs=[
            "REQ-007-FR7-006",
        ],
    )
    def get_violations(
        self,
        limit: int = DEFAULT_AUDIT_QUERY_LIMIT,
        violation_type: ViolationType | None = None,
        actor: str | None = None,
    ) -> list[Violation]:
        """Retrieve recent violations from Redis."""
        try:
            key = self._log_key
            # zrevrange()/get() are typed via redis-py's shared
            # sync/async command mixin (ResponseT); RedisStateManager
            # .client is always the sync client, which always returns
            # the plain (non-awaitable) value at runtime.
            raw_entries = cast(
                "list[bytes | str]", self._redis.client.zrevrange(key, 0, limit - 1)
            )
            violations = []
            for raw in raw_entries:
                if isinstance(raw, bytes):
                    raw = raw.decode()
                data = json.loads(raw)
                v = Violation(
                    id=data["id"],
                    type=ViolationType(data["type"]),
                    action=data["action"],
                    actor=data["actor"],
                    article=data.get("article"),
                    severity=data.get("severity", "medium"),
                    details=data.get("details", {}),
                )
                if violation_type and v.type != violation_type:
                    continue
                if actor and v.actor != actor:
                    continue
                violations.append(v)
            return violations
        except Exception as exc:
            _logger.error("Failed to get violations: %s", exc)
            return []

    @traces(
        spec="007",
        story="US-PGOV-006",
        reqs=[
            "REQ-007-US-006-003",
            "REQ-007-US-006-010",
        ],
    )
    def get_rate(self, window_seconds: int = SAFE_MODE_WINDOW_SECONDS) -> float:
        """Get current violation rate (violations per minute).

        Uses Redis counter with TTL for the sliding window.
        """
        try:
            rate_key = self._rate_key
            # See the cast() note in get_violations() above.
            count = cast("bytes | str | None", self._redis.client.get(rate_key))
            if count is None:
                return 0.0
            if isinstance(count, bytes):
                count = count.decode()
            return float(count) * (60.0 / window_seconds)
        except Exception as exc:
            _logger.error("Failed to get violation rate: %s", exc)
            return 0.0

    @traces(
        spec="007",
        story="US-PGOV-002",
        reqs=["REQ-007-US-002-005"],
    )
    def should_trigger_safe_mode(self, threshold: int = 3) -> bool:
        """Check if violation rate warrants safe mode.

        Default threshold: 3 violations per minute.
        """
        rate = self.get_rate()
        return rate >= threshold

    def clear(self) -> None:
        """Clear all violation records (for testing)."""
        try:
            self._redis.client.delete(self._log_key)
            self._redis.client.delete(self._rate_key)
        except Exception as exc:
            _logger.warning("Failed to clear violations: %s", exc)