DataOps-platform/app/core/llm/agent_governance.py

"""Govern internal Agents without turning the platform into an Agent builder."""

from __future__ import annotations

import base64
import copy
import hashlib
import hmac
import json
import re
import uuid
from collections.abc import Callable
from datetime import datetime, timedelta
from typing import Any

from app.core.common.identifiers import new_governance_uid
from app.core.common.timezone_utils import now_china

AUTONOMY_LEVELS = frozenset(
    {"read_only", "suggestion", "approval_execution", "low_risk_automatic"}
)
AGENT_STATUSES = frozenset({"draft", "active", "suspended", "retired"})
INTERFACE_TYPES = frozenset({"api", "mcp"})
TOOL_ACTIONS = frozenset({"read", "suggest", "execute"})
RISK_LEVELS = frozenset({"low", "medium", "high", "critical"})
ENVIRONMENTS = frozenset({"development", "test", "production"})
REQUEST_DECISIONS = frozenset(
    {
        "authorized",
        "denied",
        "pending_approval",
        "approved_for_manual_execution",
        "executed",
    }
)
CODE_PATTERN = re.compile(r"^[A-Z][A-Z0-9_]{2,119}$")
TOOL_PATTERN = re.compile(r"^[A-Za-z][A-Za-z0-9_.:/-]{1,199}$")
PROHIBITED_TOOLS = frozenset(
    {
        "drop_database",
        "disable_audit",
        "export_secrets",
        "grant_permission",
        "execute_shell",
        "modify_security_policy",
    }
)
INJECTION_PATTERNS = (
    ("ignore_instructions", re.compile(r"(?i)ignore\s+(all\s+)?(previous|prior|system)\s+instructions?")),
    ("reveal_secrets", re.compile(r"(?i)(reveal|show|leak|print).{0,30}(secret|api[-_ ]?key|token|password)")),
    ("override_tools", re.compile(r"(?i)(bypass|override|disable).{0,30}(permission|policy|guard|audit)")),
    ("system_prompt", re.compile(r"(?i)(system\s+prompt|developer\s+message|begin\s+system)")),
    ("ignore_instructions_zh", re.compile(r"忽略.{0,12}(指令|规则|系统)")),
    ("reveal_secrets_zh", re.compile(r"(泄露|显示|输出).{0,16}(密钥|口令|令牌|密码|api\s*key)", re.I)),
    ("bypass_policy_zh", re.compile(r"(绕过|关闭|禁用).{0,16}(权限|策略|审计|防护)")),
)


def _closed(value: Any, allowed: set[str], label: str) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise ValueError(f"{label} must be an object")
    unknown = sorted(set(value) - allowed)
    if unknown:
        raise ValueError(f"{label} contains unsupported fields: {', '.join(unknown)}")
    return copy.deepcopy(value)


def _string(value: Any, label: str, maximum: int = 1000) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ValueError(f"{label} is required")
    result = value.strip()
    if len(result) > maximum:
        raise ValueError(f"{label} exceeds {maximum} characters")
    return result


def _uid(value: Any, label: str) -> str:
    try:
        return str(uuid.UUID(str(value)))
    except (TypeError, ValueError, AttributeError) as error:
        raise ValueError(f"{label} must be a UUID") from error


def _list(value: Any, label: str, minimum: int = 0) -> list[Any]:
    if not isinstance(value, list) or len(value) < minimum:
        raise ValueError(f"{label} must contain at least {minimum} items")
    return copy.deepcopy(value)


def _canonical(value: Any) -> bytes:
    return json.dumps(
        value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
    ).encode("utf-8")


def _hash(value: Any) -> str:
    return hashlib.sha256(_canonical(value)).hexdigest()


def _urlsafe_encode(value: bytes) -> str:
    return base64.urlsafe_b64encode(value).decode("ascii").rstrip("=")


def _urlsafe_decode(value: str) -> bytes:
    return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))


def _normalize_prompt_policy(value: Any) -> dict[str, Any]:
    body = _closed(
        value,
        {
            "trusted_instruction_sources",
            "untrusted_context_mode",
            "citation_required",
        },
        "prompt policy",
    )
    sources = sorted(
        {
            _string(item, "trusted instruction source", 80)
            for item in _list(
                body.get("trusted_instruction_sources"),
                "trusted_instruction_sources",
                1,
            )
        }
    )
    if set(sources) - {"platform_system", "published_policy", "owner_approved"}:
        raise ValueError("unsupported trusted instruction source")
    mode = _string(body.get("untrusted_context_mode"), "untrusted_context_mode", 30)
    if mode not in {"quote_only", "discard"}:
        raise ValueError("untrusted context must be quoted or discarded")
    if not isinstance(body.get("citation_required"), bool):
        raise ValueError("citation_required must be boolean")
    return {
        "trusted_instruction_sources": sources,
        "untrusted_context_mode": mode,
        "citation_required": body["citation_required"],
    }


def _normalize_evidence(value: Any, *, required: bool) -> list[dict[str, str]]:
    items = _list(value, "evidence_refs", 1 if required else 0)
    normalized = []
    for item in items:
        evidence = _closed(
            item,
            {"source_type", "source_uid", "version", "point_key"},
            "evidence reference",
        )
        normalized.append(
            {
                "source_type": _string(evidence.get("source_type"), "source_type", 60),
                "source_uid": _uid(evidence.get("source_uid"), "source_uid"),
                "version": _string(evidence.get("version"), "evidence version", 80),
                "point_key": _string(evidence.get("point_key"), "point_key", 200),
            }
        )
    return normalized


def inspect_prompt(value: Any) -> dict[str, Any]:
    prompt = _string(value, "prompt", 8000)
    signals = sorted(
        {name for name, pattern in INJECTION_PATTERNS if pattern.search(prompt)}
    )
    return {
        "safe": not signals,
        "signals": signals,
        "prompt_hash": hashlib.sha256(prompt.encode("utf-8")).hexdigest(),
        "raw_prompt_retained": False,
    }


def _autonomy_actions(level: str) -> frozenset[str]:
    return {
        "read_only": frozenset({"read"}),
        "suggestion": frozenset({"read", "suggest"}),
        "approval_execution": frozenset({"read", "suggest", "execute"}),
        "low_risk_automatic": frozenset({"read", "suggest", "execute"}),
    }[level]


class AgentGovernanceService:
    """Version Agents, issue scoped machine credentials and record every decision."""

    def __init__(
        self,
        repository,
        *,
        approval_gateway,
        credential_secret: str,
        uid_factory: Callable[[], str] = new_governance_uid,
        now_factory: Callable[[], datetime] = now_china,
        commit: Callable[[], None] = lambda: None,
        rollback: Callable[[], None] = lambda: None,
    ):
        if not isinstance(credential_secret, str) or len(credential_secret.encode()) < 32:
            raise ValueError("Agent credential secret must contain at least 32 bytes")
        self.repository = repository
        self.approval_gateway = approval_gateway
        self.secret = credential_secret.encode("utf-8")
        self.uid_factory = uid_factory
        self.now_factory = now_factory
        self.commit = commit
        self.rollback = rollback

    def _definition(self, payload: Any) -> dict[str, Any]:
        body = _closed(
            payload,
            {
                "code",
                "name",
                "purpose",
                "owner_uid",
                "business_domain_uids",
                "environments",
                "autonomy_level",
                "prompt_policy",
                "change_reason",
            },
            "governed Agent",
        )
        code = _string(body.get("code"), "Agent code", 120).upper()
        if not CODE_PATTERN.fullmatch(code):
            raise ValueError("Agent code is invalid")
        owner_uid = _uid(body.get("owner_uid"), "owner_uid")
        domains = sorted(
            {
                _uid(item, "business_domain_uid")
                for item in _list(
                    body.get("business_domain_uids"), "business_domain_uids", 1
                )
            }
        )
        environments = sorted(
            {
                _string(item, "environment", 30)
                for item in _list(body.get("environments"), "environments", 1)
            }
        )
        if not set(environments) <= ENVIRONMENTS:
            raise ValueError("unsupported Agent environment")
        level = _string(body.get("autonomy_level"), "autonomy_level", 30)
        if level not in AUTONOMY_LEVELS:
            raise ValueError("unsupported autonomy level")
        return {
            "code": code,
            "name": _string(body.get("name"), "Agent name", 300),
            "purpose": _string(body.get("purpose"), "Agent purpose", 2000),
            "owner_uid": owner_uid,
            "business_domain_uids": domains,
            "environments": environments,
            "autonomy_level": level,
            "prompt_policy": _normalize_prompt_policy(body.get("prompt_policy")),
            "change_reason": _string(
                body.get("change_reason", "initial registration"),
                "change_reason",
                1000,
            ),
        }

    def register_agent(self, payload: Any, *, actor_uid: str):
        definition = self._definition(payload)
        actor = _uid(actor_uid, "actor_uid")
        if self.repository.users_available({actor, definition["owner_uid"]}) != {
            actor,
            definition["owner_uid"],
        }:
            raise ValueError("Agent owner or actor is unavailable")
        now = self.now_factory().isoformat()
        agent_uid = self.uid_factory()
        snapshot = {key: value for key, value in definition.items() if key != "change_reason"}
        agent = {
            "uid": agent_uid,
            **snapshot,
            "machine_subject": f"agent:{definition['code'].lower()}:{agent_uid}",
            "status": "draft",
            "current_version": 1,
            "created_by": actor,
            "created_at": now,
            "updated_by": actor,
            "updated_at": now,
            "retired_at": None,
        }
        version = {
            "uid": self.uid_factory(),
            "agent_uid": agent_uid,
            "version": 1,
            "status": "draft",
            "definition": snapshot,
            "content_hash": _hash(snapshot),
            "change_reason": definition["change_reason"],
            "created_by": actor,
            "created_at": now,
            "published_by": None,
            "published_at": None,
        }
        try:
            result = self.repository.create_agent(agent, version)
            self.repository.add_event(agent_uid, "agent_registered", actor, 1, {})
            self.commit()
            return result
        except Exception:
            self.rollback()
            raise

    def list_agents(self, **filters):
        return self.repository.list_agents(**filters)

    def get_agent(self, uid: str):
        agent = self.repository.get_agent(_uid(uid, "agent_uid"))
        if agent is None:
            raise LookupError("governed Agent was not found")
        return agent

    def agent_detail(self, uid: str):
        result = self.repository.agent_detail(_uid(uid, "agent_uid"))
        if result is None:
            raise LookupError("governed Agent was not found")
        return result

    def revise_agent(
        self, uid: str, payload: Any, *, expected_version: int, actor_uid: str
    ):
        agent = self.get_agent(uid)
        actor = _uid(actor_uid, "actor_uid")
        if actor != agent["owner_uid"] or agent["status"] == "retired":
            raise PermissionError("only the active Agent owner can revise it")
        definition = self._definition(payload)
        if definition["owner_uid"] != agent["owner_uid"]:
            raise ValueError("Agent ownership transfer must use responsibility governance")
        next_version = int(expected_version) + 1
        now = self.now_factory().isoformat()
        snapshot = {key: value for key, value in definition.items() if key != "change_reason"}
        revised = {
            **agent,
            **snapshot,
            "status": "draft",
            "current_version": next_version,
            "updated_by": actor,
            "updated_at": now,
        }
        version = {
            "uid": self.uid_factory(),
            "agent_uid": agent["uid"],
            "version": next_version,
            "status": "draft",
            "definition": snapshot,
            "content_hash": _hash(snapshot),
            "change_reason": definition["change_reason"],
            "created_by": actor,
            "created_at": now,
            "published_by": None,
            "published_at": None,
        }
        try:
            result = self.repository.update_agent(
                revised, version, int(expected_version), "agent_revised", actor
            )
            self.repository.revoke_credentials(agent["uid"], actor, now)
            self.commit()
            return result
        except Exception:
            self.rollback()
            raise

    def transition_agent(
        self, uid: str, payload: Any, *, expected_version: int, actor_uid: str
    ):
        body = _closed(payload, {"action", "reason"}, "Agent transition")
        agent = self.get_agent(uid)
        actor = _uid(actor_uid, "actor_uid")
        if actor != agent["owner_uid"]:
            raise PermissionError("only the Agent owner can transition it")
        action = _string(body.get("action"), "action", 30)
        transitions = {
            ("draft", "activate"): ("active", "agent_activated"),
            ("suspended", "activate"): ("active", "agent_reactivated"),
            ("active", "suspend"): ("suspended", "agent_suspended"),
            ("draft", "retire"): ("retired", "agent_retired"),
            ("active", "retire"): ("retired", "agent_retired"),
            ("suspended", "retire"): ("retired", "agent_retired"),
        }
        target = transitions.get((agent["status"], action))
        if target is None:
            raise RuntimeError("Agent lifecycle transition is not allowed")
        if action == "activate" and not self.repository.active_grants(agent["uid"]):
            raise RuntimeError("at least one active tool grant is required")
        now = self.now_factory().isoformat()
        updated = {
            **agent,
            "status": target[0],
            "current_version": int(expected_version) + 1,
            "updated_by": actor,
            "updated_at": now,
            "retired_at": now if target[0] == "retired" else None,
        }
        try:
            result = self.repository.update_agent(
                updated, None, int(expected_version), target[1], actor
            )
            if target[0] in {"suspended", "retired"}:
                self.repository.revoke_credentials(agent["uid"], actor, now)
            self.commit()
            return result
        except Exception:
            self.rollback()
            raise

    def create_tool_grant(self, agent_uid: str, payload: Any, *, actor_uid: str):
        body = _closed(
            payload,
            {
                "interface_type",
                "tool_name",
                "action",
                "business_domain_uid",
                "environment",
                "risk_level",
                "requires_approval",
            },
            "Agent tool grant",
        )
        agent = self.get_agent(agent_uid)
        actor = _uid(actor_uid, "actor_uid")
        if actor != agent["owner_uid"] or agent["status"] == "retired":
            raise PermissionError("only the Agent owner can grant tools")
        interface_type = _string(body.get("interface_type"), "interface_type", 20)
        action = _string(body.get("action"), "tool action", 20)
        tool_name = _string(body.get("tool_name"), "tool_name", 200)
        domain = _uid(body.get("business_domain_uid"), "business_domain_uid")
        environment = _string(body.get("environment"), "environment", 30)
        risk = _string(body.get("risk_level"), "risk_level", 20)
        requires_approval = body.get("requires_approval")
        if interface_type not in INTERFACE_TYPES or action not in TOOL_ACTIONS:
            raise ValueError("unsupported interface or tool action")
        if not TOOL_PATTERN.fullmatch(tool_name):
            raise ValueError("tool name is invalid")
        if domain not in agent["business_domain_uids"]:
            raise ValueError("tool grant domain is outside Agent scope")
        if environment not in agent["environments"] or environment not in ENVIRONMENTS:
            raise ValueError("tool grant environment is outside Agent scope")
        if risk not in RISK_LEVELS:
            raise ValueError("unsupported risk level")
        if not isinstance(requires_approval, bool):
            raise ValueError("requires_approval must be boolean")
        if action not in _autonomy_actions(agent["autonomy_level"]):
            raise ValueError("tool action exceeds Agent autonomy level")
        if risk in {"high", "critical"} and not requires_approval:
            raise ValueError("high-risk grants require approval")
        if (
            agent["autonomy_level"] == "low_risk_automatic"
            and action == "execute"
            and risk != "low"
            and not requires_approval
        ):
            raise ValueError("automatic execution is limited to low risk")
        now = self.now_factory().isoformat()
        grant = {
            "uid": self.uid_factory(),
            "agent_uid": agent["uid"],
            "interface_type": interface_type,
            "tool_name": tool_name,
            "action": action,
            "business_domain_uid": domain,
            "environment": environment,
            "risk_level": risk,
            "requires_approval": requires_approval,
            "status": "active",
            "created_by": actor,
            "created_at": now,
            "revoked_by": None,
            "revoked_at": None,
        }
        try:
            result = self.repository.create_grant(grant)
            self.repository.add_event(
                agent["uid"], "tool_granted", actor, agent["current_version"],
                {"grant_uid": grant["uid"], "tool_name": tool_name, "action": action},
            )
            self.commit()
            return result
        except Exception:
            self.rollback()
            raise

    def revoke_tool_grant(self, agent_uid: str, grant_uid: str, *, actor_uid: str):
        agent = self.get_agent(agent_uid)
        actor = _uid(actor_uid, "actor_uid")
        if actor != agent["owner_uid"] or agent["status"] == "retired":
            raise PermissionError("only the Agent owner can revoke tools")
        try:
            result = self.repository.revoke_grant(
                agent["uid"], _uid(grant_uid, "grant_uid"), actor,
                self.now_factory().isoformat(),
            )
            self.repository.add_event(
                agent["uid"], "tool_revoked", actor, agent["current_version"],
                {"grant_uid": grant_uid},
            )
            self.commit()
            return result
        except Exception:
            self.rollback()
            raise

    def issue_credential(self, agent_uid: str, payload: Any, *, actor_uid: str):
        body = _closed(payload, {"ttl_seconds"}, "Agent credential request")
        agent = self.get_agent(agent_uid)
        actor = _uid(actor_uid, "actor_uid")
        if actor != agent["owner_uid"] or agent["status"] != "active":
            raise PermissionError("only the active Agent owner can issue credentials")
        try:
            ttl = int(body.get("ttl_seconds"))
        except (TypeError, ValueError) as error:
            raise ValueError("ttl_seconds must be an integer") from error
        if ttl < 60 or ttl > 900:
            raise ValueError("ttl_seconds must be between 60 and 900")
        now = self.now_factory()
        expires_at = now + timedelta(seconds=ttl)
        jti = self.uid_factory()
        claims = {
            "jti": jti,
            "sub": agent["machine_subject"],
            "agent_uid": agent["uid"],
            "business_domain_uids": agent["business_domain_uids"],
            "environments": agent["environments"],
            "iat": int(now.timestamp()),
            "exp": int(expires_at.timestamp()),
        }
        encoded = _urlsafe_encode(_canonical(claims))
        signature = _urlsafe_encode(hmac.new(self.secret, encoded.encode(), hashlib.sha256).digest())
        token = f"{encoded}.{signature}"
        record = {
            "uid": self.uid_factory(),
            "agent_uid": agent["uid"],
            "jti": jti,
            "token_digest": hashlib.sha256(token.encode()).hexdigest(),
            "issued_by": actor,
            "issued_at": now.isoformat(),
            "expires_at": expires_at.isoformat(),
            "status": "active",
            "revoked_by": None,
            "revoked_at": None,
        }
        try:
            self.repository.create_credential(record)
            self.repository.add_event(
                agent["uid"], "credential_issued", actor, agent["current_version"],
                {"credential_uid": record["uid"], "jti": jti, "expires_at": record["expires_at"]},
            )
            self.commit()
            return {"token": token, "jti": jti, "expires_at": record["expires_at"]}
        except Exception:
            self.rollback()
            raise

    def validate_credential(self, agent_uid: str, token: str) -> dict[str, Any]:
        if not isinstance(token, str) or token.count(".") != 1:
            raise PermissionError("Agent credential is malformed")
        encoded, supplied_signature = token.split(".", 1)
        expected_signature = _urlsafe_encode(
            hmac.new(self.secret, encoded.encode(), hashlib.sha256).digest()
        )
        if not hmac.compare_digest(supplied_signature, expected_signature):
            raise PermissionError("Agent credential signature is invalid")
        try:
            claims = json.loads(_urlsafe_decode(encoded))
        except (ValueError, json.JSONDecodeError) as error:
            raise PermissionError("Agent credential payload is invalid") from error
        uid = _uid(agent_uid, "agent_uid")
        if claims.get("agent_uid") != uid:
            raise PermissionError("Agent credential subject does not match")
        if int(claims.get("exp", 0)) <= int(self.now_factory().timestamp()):
            raise PermissionError("Agent credential has expired")
        stored = self.repository.get_credential(claims.get("jti"))
        if (
            not stored
            or stored.get("status") != "active"
            or stored.get("agent_uid") != uid
            or not hmac.compare_digest(
                stored.get("token_digest", ""), hashlib.sha256(token.encode()).hexdigest()
            )
        ):
            raise PermissionError("Agent credential is revoked or unknown")
        return claims

    def revoke_agent_credentials(self, agent_uid: str, *, actor_uid: str):
        agent = self.get_agent(agent_uid)
        actor = _uid(actor_uid, "actor_uid")
        if actor != agent["owner_uid"]:
            raise PermissionError("only the Agent owner can revoke credentials")
        now = self.now_factory().isoformat()
        try:
            count = self.repository.revoke_credentials(agent["uid"], actor, now)
            self.repository.add_event(
                agent["uid"], "credentials_revoked", actor,
                agent["current_version"], {"revoked_count": count},
            )
            self.commit()
            return {"agent_uid": agent["uid"], "revoked_count": count}
        except Exception:
            self.rollback()
            raise

    def _decision_record(
        self,
        agent: dict[str, Any],
        request_body: dict[str, Any],
        *,
        grant: dict[str, Any] | None,
        prompt_guard: dict[str, Any],
        evidence_refs: list[dict[str, str]],
        decision: str,
        reason_code: str,
        approval_task_uid: str | None = None,
        automatic_execution_allowed: bool = False,
    ) -> dict[str, Any]:
        if decision not in REQUEST_DECISIONS:
            raise ValueError("unsupported Agent decision")
        now = self.now_factory().isoformat()
        return {
            "uid": self.uid_factory(),
            "agent_uid": agent["uid"],
            "agent_version": agent["current_version"],
            "grant_uid": grant.get("uid") if grant else None,
            "correlation_id": request_body["correlation_id"],
            "interface_type": request_body["interface_type"],
            "tool_name": request_body["tool_name"],
            "action": request_body["action"],
            "business_domain_uid": request_body["business_domain_uid"],
            "environment": request_body["environment"],
            "risk_level": request_body["risk_level"],
            "input_digest": prompt_guard["prompt_hash"],
            "prompt_guard": prompt_guard,
            "evidence_refs": evidence_refs,
            "decision": decision,
            "reason_code": reason_code,
            "approval_task_uid": approval_task_uid,
            "automatic_execution_allowed": automatic_execution_allowed,
            "output_digest": None,
            "current_version": 1,
            "created_at": now,
            "updated_at": now,
        }

    def authorize_action(self, agent_uid: str, token: str, payload: Any):
        body = _closed(
            payload,
            {
                "interface_type",
                "tool_name",
                "action",
                "business_domain_uid",
                "environment",
                "risk_level",
                "prompt",
                "evidence_refs",
                "workflow_uid",
                "correlation_id",
            },
            "Agent action",
        )
        agent = self.get_agent(agent_uid)
        request_body = {
            "interface_type": _string(body.get("interface_type"), "interface_type", 20),
            "tool_name": _string(body.get("tool_name"), "tool_name", 200),
            "action": _string(body.get("action"), "action", 20),
            "business_domain_uid": _uid(body.get("business_domain_uid"), "business_domain_uid"),
            "environment": _string(body.get("environment"), "environment", 30),
            "risk_level": _string(body.get("risk_level"), "risk_level", 20),
            "correlation_id": _uid(body.get("correlation_id", self.uid_factory()), "correlation_id"),
        }
        if request_body["interface_type"] not in INTERFACE_TYPES:
            raise ValueError("unsupported interface type")
        if request_body["action"] not in TOOL_ACTIONS:
            raise ValueError("unsupported tool action")
        if request_body["risk_level"] not in RISK_LEVELS:
            raise ValueError("unsupported risk level")
        prompt_guard = inspect_prompt(body.get("prompt"))
        evidence_refs = _normalize_evidence(
            body.get("evidence_refs", []),
            required=(
                request_body["action"] == "suggest"
                and agent["prompt_policy"]["citation_required"]
                and prompt_guard["safe"]
            ),
        )

        reason = None
        claims = None
        try:
            claims = self.validate_credential(agent["uid"], token)
        except PermissionError:
            reason = "credential_invalid"
        if claims and (
            request_body["business_domain_uid"] not in claims["business_domain_uids"]
            or request_body["environment"] not in claims["environments"]
        ):
            reason = "credential_scope_denied"
        if agent["status"] != "active":
            reason = "agent_not_active"
        if request_body["tool_name"] in PROHIBITED_TOOLS:
            reason = "prohibited_action"
        elif not prompt_guard["safe"]:
            reason = "prompt_injection_detected"
        elif request_body["action"] not in _autonomy_actions(agent["autonomy_level"]):
            reason = "autonomy_level_denied"

        grant = None
        if reason is None:
            grant = self.repository.find_grant(
                agent["uid"],
                request_body["interface_type"],
                request_body["tool_name"],
                request_body["action"],
                request_body["business_domain_uid"],
                request_body["environment"],
            )
            if grant is None:
                reason = "tool_not_granted"
            elif grant["risk_level"] != request_body["risk_level"]:
                reason = "risk_classification_mismatch"

        if reason is not None:
            record = self._decision_record(
                agent,
                request_body,
                grant=grant,
                prompt_guard=prompt_guard,
                evidence_refs=evidence_refs,
                decision="denied",
                reason_code=reason,
            )
        else:
            needs_approval = (
                grant["requires_approval"]
                or request_body["risk_level"] in {"high", "critical"}
                or (
                    request_body["action"] == "execute"
                    and agent["autonomy_level"] == "approval_execution"
                )
            )
            automatic = bool(
                request_body["action"] == "execute"
                and request_body["risk_level"] == "low"
                and agent["autonomy_level"] == "low_risk_automatic"
                and not needs_approval
            )
            record = self._decision_record(
                agent,
                request_body,
                grant=grant,
                prompt_guard=prompt_guard,
                evidence_refs=evidence_refs,
                decision="pending_approval" if needs_approval else "authorized",
                reason_code="approval_required" if needs_approval else "policy_allowed",
                automatic_execution_allowed=automatic,
            )
            if needs_approval:
                workflow_uid = _uid(body.get("workflow_uid"), "workflow_uid")
                task = self.approval_gateway.create_agent_task(
                    record, workflow_uid, agent["owner_uid"]
                )
                record["approval_task_uid"] = task["uid"]
        try:
            result = self.repository.create_request(record)
            self.commit()
            return result
        except Exception:
            self.rollback()
            raise

    def reconcile_action(
        self, request_uid: str, *, expected_version: int, actor_uid: str
    ):
        request_record = self.repository.get_request(_uid(request_uid, "request_uid"))
        if request_record is None:
            raise LookupError("Agent action request was not found")
        if request_record["decision"] != "pending_approval":
            raise RuntimeError("Agent action is not waiting for approval")
        actor = _uid(actor_uid, "actor_uid")
        task = self.approval_gateway.get_task(request_record["approval_task_uid"])
        if not task or task["status"] not in {"approved", "rejected"}:
            raise RuntimeError("Agent approval has no final decision")
        if task["status"] == "rejected":
            decision, reason = "denied", "approval_rejected"
        elif request_record["risk_level"] in {"high", "critical"}:
            route = task.get("route_snapshot") or {}
            reviewers = {
                item.get("reviewer_uid")
                for item in task.get("reviews", [])
                if item.get("decision") == "approve"
            }
            if (
                route.get("approval_mode") != "dual_control"
                or int(route.get("min_approvals", 0)) < 2
                or len(reviewers) < 2
            ):
                raise RuntimeError("high-risk Agent action requires dual control")
            decision, reason = (
                "approved_for_manual_execution",
                "dual_control_approved_manual_only",
            )
        else:
            decision, reason = "authorized", "approval_granted"
        updated = {
            **request_record,
            "decision": decision,
            "reason_code": reason,
            "automatic_execution_allowed": False,
            "updated_at": self.now_factory().isoformat(),
        }
        try:
            result = self.repository.update_request(
                updated, int(expected_version), "approval_reconciled", actor
            )
            self.commit()
            return result
        except Exception:
            self.rollback()
            raise

    def complete_action(
        self,
        request_uid: str,
        payload: Any,
        *,
        expected_version: int,
        actor_uid: str,
    ):
        body = _closed(payload, {"output", "evidence_refs"}, "Agent action result")
        request_record = self.repository.get_request(_uid(request_uid, "request_uid"))
        if request_record is None:
            raise LookupError("Agent action request was not found")
        if request_record["risk_level"] in {"high", "critical"}:
            raise RuntimeError("high-risk automatic execution is disabled")
        if request_record["decision"] != "authorized":
            raise RuntimeError("Agent action is not authorized for completion")
        output = body.get("output")
        if not isinstance(output, dict):
            raise ValueError("Agent output must be an object")
        evidence_refs = _normalize_evidence(body.get("evidence_refs"), required=True)
        updated = {
            **request_record,
            "decision": "executed",
            "reason_code": "execution_evidence_recorded",
            "evidence_refs": evidence_refs,
            "output_digest": _hash(output),
            "updated_at": self.now_factory().isoformat(),
        }
        actor = _uid(actor_uid, "actor_uid")
        try:
            result = self.repository.update_request(
                updated, int(expected_version), "execution_recorded", actor
            )
            self.commit()
            return result
        except Exception:
            self.rollback()
            raise

    def replay(self, request_uid: str):
        result = self.repository.replay(_uid(request_uid, "request_uid"))
        if result is None:
            raise LookupError("Agent action replay was not found")
        return result

    def list_requests(self, **filters):
        return self.repository.list_requests(**filters)

    def dashboard(self):
        return self.repository.dashboard()