a0_software_orchestrator/tools/plan_mode_guard.py

"""Tool: enforce Plan Mode by checking current mode (DB-backed, global)."""
from __future__ import annotations

import json
from typing import Any, Dict

from helpers.tool import Tool, Response
from usr.plugins.a0_software_orchestrator.helpers.db_state_store import (
    _db_error_handler, _db,
)


_DEFAULT_MODE = {
    "mode": "planning_only",
    "allowed_actions": [],
    "blocked_actions": [
        "code",
        "deploy",
        "destroy",
        "push_to_main",
    ],
}

_MODE_POLICIES = {
    "planning_only": {"allowed_actions": [], "blocked_actions": ["code", "deploy", "destroy", "push_to_main"]},
    "implementation_allowed": {"allowed_actions": ["code", "test", "commit"], "blocked_actions": ["deploy", "destroy", "push_to_main"]},
    "runtime_verification_allowed": {"allowed_actions": ["code", "test", "commit", "start_server"], "blocked_actions": ["deploy", "destroy"]},
    "deployment_preparation_allowed": {"allowed_actions": ["code", "test", "commit", "deploy_staging"], "blocked_actions": ["deploy_production", "destroy"]},
    "release_handoff_allowed": {"allowed_actions": ["deploy_production"], "blocked_actions": ["destroy"]},
    "maintenance_allowed": {"allowed_actions": ["code", "test", "commit", "deploy"], "blocked_actions": ["destroy"]},
}

# Plan-Mode ist plugin-weit (Bug 7 Fix). Der Key lebt in plugin_settings,
# einer Tabelle ohne project_id/FK – saubere Architektur statt
# project_id=0-Workaround in orch_kv (Migration 5).
_PLUGIN_MODE_KEY = "orchestrator_mode"


def _normalize_mode(mode_data: Any) -> Dict[str, Any]:
    """Stellt sicher, dass das gelesene Mode-Dict die erwarteten Keys hat."""
    if not isinstance(mode_data, dict):
        return dict(_DEFAULT_MODE)
    out = dict(_DEFAULT_MODE)
    out.update({
        "mode": mode_data.get("mode", _DEFAULT_MODE["mode"]),
        "allowed_actions": list(mode_data.get("allowed_actions") or []),
        "blocked_actions": list(mode_data.get("blocked_actions") or []),
    })
    return out


@_db_error_handler
def _get_plugin_mode() -> Dict[str, Any]:
    """Liest den globalen Plan-Mode aus plugin_settings.

    Returns _DEFAULT_MODE wenn kein Eintrag existiert.
    """
    db = _db()
    row = db.conn.execute(
        "SELECT value_json FROM plugin_settings WHERE key = ?",
        (_PLUGIN_MODE_KEY,),
    ).fetchone()
    if not row:
        return dict(_DEFAULT_MODE)
    try:
        return _normalize_mode(json.loads(row[0]))
    except (json.JSONDecodeError, TypeError):
        return dict(_DEFAULT_MODE)


@_db_error_handler
def _set_plugin_mode(mode_data: Dict[str, Any]) -> None:
    """Schreibt den globalen Plan-Mode in plugin_settings."""
    db = _db()
    db.conn.execute(
        "INSERT INTO plugin_settings (key, value_json, updated_at) "
        "VALUES (?, ?, datetime('now')) "
        "ON CONFLICT(key) DO UPDATE SET "
        "  value_json = excluded.value_json, updated_at = datetime('now')",
        (_PLUGIN_MODE_KEY, json.dumps(mode_data)),
    )
    db.conn.commit()


class PlanModeGuard(Tool):
    """Prüft den aktuellen Plan-Mode (plugin-weit) und entscheidet, ob eine
    Aktion erlaubt ist.

    KEIN resolve_project()-Aufruf mehr: Plan-Mode ist global, nicht pro Projekt.

    Args (kwargs):
        requested_action (str): Action-Name zum Testen (z.B. "code", "deploy").
        new_mode (str): Optional – neuen Mode setzen (für Mode-Transitions).
    """

    async def execute(
        self,
        requested_action: str = "",
        new_mode: str = "",
        **kwargs,
    ):
        # Mode-Transition (z.B. planning_only → implementation_allowed)
        if new_mode:
            if new_mode not in _MODE_POLICIES:
                return Response(
                    message=json.dumps({
                        "verdict": "ERROR",
                        "error": f"invalid mode: {new_mode}",
                        "allowed_modes": sorted(_MODE_POLICIES),
                    }, indent=2),
                    break_loop=False,
                )
            mode_data = {
                "mode": new_mode,
                "allowed_actions": list(_MODE_POLICIES[new_mode]["allowed_actions"]),
                "blocked_actions": list(_MODE_POLICIES[new_mode]["blocked_actions"]),
            }
            _set_plugin_mode(mode_data)
            return Response(
                message=json.dumps({
                    "verdict": "ALLOWED",
                    "mode": new_mode,
                    "allowed_actions": mode_data["allowed_actions"],
                    "blocked_actions": mode_data["blocked_actions"],
                    "transitioned": True,
                }, indent=2),
                break_loop=False,
            )

        # Ansonsten: aktuellen Mode lesen und Aktion prüfen
        mode_data = _get_plugin_mode()
        current_mode = mode_data["mode"]
        allowed = mode_data["allowed_actions"]
        blocked = mode_data["blocked_actions"]

        if requested_action and requested_action in blocked:
            return Response(
                message=json.dumps({
                    "verdict": "BLOCKED",
                    "mode": current_mode,
                    "action": requested_action,
                    "reason": f"action '{requested_action}' is not allowed in mode '{current_mode}'",
                }, indent=2),
                break_loop=False,
            )
        if not requested_action or requested_action in allowed:
            return Response(
                message=json.dumps({
                    "verdict": "ALLOWED",
                    "mode": current_mode,
                    "action": requested_action,
                }, indent=2),
                break_loop=False,
            )
        return Response(
            message=json.dumps({
                "verdict": "UNKNOWN",
                "mode": current_mode,
                "action": requested_action,
                "hint": f"action not in allowed={allowed} or blocked={blocked}",
            }, indent=2),
            break_loop=False,
        )