orchestrata_ultimate.governance.safe_mode

Reference › orchestrata_ultimate › governance › safe_mode

The system-wide safe mode, SPEC-007 stories US-PGOV-002 and FR-PGOV-008 (REQ-007-FR8-001, -005, -007, -008, REQ-007-FR7-003, REQ-007-US-002-005, -010). When governance violations come too fast, SafeModeManager switches the system into a state where only read operations are allowed. policy_engine is the importer: it checks safe mode first in evaluate, and after each violation it calls check_violation_rate, with the rate supplied by the ViolationTracker.

The state lives in a Redis key as JSON (active flag, reason, entry time and a violation count), so all processes see it. enter_safe_mode(reason) writes it, tells Prolog through the client's enter_safe_mode (a failure there is logged only) and logs a warning; a Redis failure raises SafeModeError. is_safe_mode reads the state and, once the cooldown has passed since entry, exits safe mode and returns False. exit_safe_mode clears it. check_violation_rate(rate) enters safe mode, with an explanatory reason, when the rate reaches the threshold and safe mode is not already on, and returns True if it triggered. is_action_allowed_in_safe_mode(action) is always True outside safe mode; during safe mode only read, get, list, query, search, view and check are allowed. get_safe_mode_status returns a SafeModeStatus. Threshold, window and cooldown come from constants.

At a glance

Class diagram

Dependency diagram

API

Classes

SafeModeManager

Decorators: traces(spec='007', story='US-PGOV-002', reqs=['REQ-007-US-002-005', 'REQ-007-US-002-010', 'REQ-007-FR8-001', 'REQ-007-F… · Line 43

Manages safe mode lifecycle.

Source

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

# mypy: ignore-errors
"""Safe mode management for Orchestrata governance.

Triggers safe mode when violation rate exceeds threshold.
In safe mode, only read operations are allowed.
"""

from __future__ import annotations

import json
import logging
import time
from datetime import UTC
from typing import TYPE_CHECKING, cast

if TYPE_CHECKING:
    from orchestrata_ultimate.infra.redis_state import RedisStateManager

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.governance.constants import (
    REDIS_SAFE_MODE_KEY,
    SAFE_MODE_COOLDOWN_SECONDS,
    SAFE_MODE_VIOLATION_THRESHOLD,
    SAFE_MODE_WINDOW_SECONDS,
)
from orchestrata_ultimate.governance.exceptions import SafeModeError
from orchestrata_ultimate.governance.prolog_client import GovernancePrologClient
from orchestrata_ultimate.governance.types import SafeModeStatus

_logger = logging.getLogger(__name__)


@traces(
    spec="007",
    story="US-PGOV-002",
    reqs=[
        "REQ-007-US-002-005",
        "REQ-007-US-002-010",
        "REQ-007-FR8-001",
        "REQ-007-FR8-007",
    ],
)
class SafeModeManager:
    """Manages safe mode lifecycle.

    Safe mode is triggered when the violation rate exceeds
    SAFE_MODE_VIOLATION_THRESHOLD violations within
    SAFE_MODE_WINDOW_SECONDS seconds.

    In safe mode:
    - Only read operations are allowed
    - Write operations require human approval
    - System runs in degraded but safe state
    """

    @traces(spec="007", story="US-PGOV-002", reqs=["REQ-007-US-002-010"])
    def __init__(
        self,
        redis: RedisStateManager,
        prolog_client: GovernancePrologClient | None = None,
        violation_threshold: int = SAFE_MODE_VIOLATION_THRESHOLD,
        window_seconds: int = SAFE_MODE_WINDOW_SECONDS,
        cooldown_seconds: int = SAFE_MODE_COOLDOWN_SECONDS,
        redis_key: str | None = None,
    ) -> None:
        self._redis = redis
        self._prolog_client = prolog_client
        self._violation_threshold = violation_threshold
        self._window_seconds = window_seconds
        self._cooldown_seconds = cooldown_seconds
        self._redis_key = redis_key or REDIS_SAFE_MODE_KEY

    @traces(spec="007", story="FR-PGOV-008", reqs=["REQ-007-FR8-002"])
    def _safe_mode_data(self) -> dict | None:
        """Get current safe mode data from Redis."""
        try:
            # cast: redis-py's command mixin types `.get()` as
            # `ResponseT | Awaitable[ResponseT]` (shared sync/async surface);
            # ``client`` is RedisStateManager's SYNC client (precedent:
            # infra/redis_state.py's per-call `cast(...)`).
            raw = cast("str | bytes | None", self._redis.client.get(self._redis_key))
            if raw is not None:
                if isinstance(raw, bytes):
                    raw = raw.decode()
                return json.loads(raw)
        except Exception as exc:
            _logger.warning("Failed to read safe mode state: %s", exc)
        return None

    @traces(
        spec="007",
        story="FR-PGOV-008",
        reqs=["REQ-007-FR8-005", "REQ-007-FR8-008"],
    )
    def enter_safe_mode(self, reason: str) -> None:
        """Activate safe mode."""
        data = {
            "active": True,
            "reason": reason,
            "entered_at": time.time(),
            "violation_count": 0,
        }
        try:
            self._redis.client.set(self._redis_key, json.dumps(data))
        except Exception as exc:
            _logger.error("Failed to enter safe mode: %s", exc)
            raise SafeModeError(f"Cannot enter safe mode: {exc}") from exc

        if self._prolog_client is not None:
            try:
                self._prolog_client.enter_safe_mode(reason)
            except Exception as exc:
                _logger.warning("Prolog enter_safe_mode failed: %s", exc)

        _logger.warning("SAFE MODE ACTIVATED: %s", reason)

    @traces(
        spec="007",
        story="FR-PGOV-008",
        reqs=["REQ-007-FR8-005", "REQ-007-FR8-006"],
    )
    def exit_safe_mode(self) -> None:
        """Deactivate safe mode."""
        try:
            self._redis.client.delete(self._redis_key)
        except Exception as exc:
            _logger.error("Failed to exit safe mode: %s", exc)
            raise SafeModeError(f"Cannot exit safe mode: {exc}") from exc

        if self._prolog_client is not None:
            try:
                self._prolog_client.exit_safe_mode()
            except Exception as exc:
                _logger.warning("Prolog exit_safe_mode failed: %s", exc)

        _logger.info("Safe mode deactivated")

    @traces(spec="007", story="FR-PGOV-008", reqs=["REQ-007-FR8-007"])
    def is_safe_mode(self) -> bool:
        """Check if safe mode is currently active."""
        data = self._safe_mode_data()
        if data is None:
            return False
        if not data.get("active", False):
            return False
        # Check cooldown
        entered_at = data.get("entered_at", 0)
        elapsed = time.time() - entered_at
        if elapsed > self._cooldown_seconds:
            # Auto-exit after cooldown
            self.exit_safe_mode()
            return False
        return True

    @traces(
        spec="007",
        story="US-PGOV-002",
        reqs=["REQ-007-FR7-003", "REQ-007-US-002-005"],
    )
    def check_violation_rate(self, current_rate: float) -> bool:
        """Check if violation rate warrants entering safe mode.

        Returns True if safe mode was triggered.
        """
        if current_rate >= self._violation_threshold:
            if not self.is_safe_mode():
                self.enter_safe_mode(
                    f"Violation rate {current_rate:.1f}/min exceeds "
                    f"threshold {self._violation_threshold}/min"
                )
                return True
        return False

    @traces(spec="007", story="FR-PGOV-008", reqs=["REQ-007-FR8-007"])
    def get_safe_mode_status(self) -> SafeModeStatus:
        """Get detailed safe mode status."""
        data = self._safe_mode_data()
        if data is None or not data.get("active", False):
            return SafeModeStatus(active=False)

        from datetime import datetime

        entered_at = data.get("entered_at", 0)
        elapsed = time.time() - entered_at
        cooldown_remaining = max(0, self._cooldown_seconds - elapsed)

        # Auto-exit if cooldown expired
        if cooldown_remaining == 0:
            self.exit_safe_mode()
            return SafeModeStatus(active=False)

        return SafeModeStatus(
            active=True,
            reason=data.get("reason", ""),
            entered_at=datetime.fromtimestamp(entered_at, tz=UTC),
            violation_count=data.get("violation_count", 0),
            cooldown_remaining_seconds=cooldown_remaining,
        )

    @traces(spec="007", story="FR-PGOV-008", reqs=["REQ-007-FR8-001"])
    def is_action_allowed_in_safe_mode(self, action: str) -> bool:
        """Check if an action is allowed during safe mode.

        Only read operations are permitted.
        """
        if not self.is_safe_mode():
            return True
        read_actions = {"read", "get", "list", "query", "search", "view", "check"}
        return action.lower() in read_actions