Gitea-Tools/webui/gated_actions.py

"""Disabled-by-default gated action registry for future UI writes (#434).

Every action declares the MCP capability and profile role from
``task_capability_map``. Previews always render a mutation ledger; execution
is fail-closed in the read-only MVP (no MCP or shell fallbacks).
"""

from __future__ import annotations

from dataclasses import asdict, dataclass, field
from typing import Any

from task_capability_map import required_permission, required_role

# MVP: no UI action may execute mutations until capability gates ship.
_MVP_ACTIONS_ENABLED = False


@dataclass(frozen=True)
class MutationLedgerEntry:
    """One planned mutation step shown before any write executes."""

    sequence: int
    mcp_tool: str
    permission: str
    target: str
    summary: str


@dataclass(frozen=True)
class GatedAction:
    """Registry entry tying a UI action to resolver task metadata."""

    action_id: str
    label: str
    task_key: str
    mcp_tool: str
    description: str
    enabled: bool = False

    @property
    def required_permission(self) -> str:
        return required_permission(self.task_key)

    @property
    def required_role(self) -> str:
        return required_role(self.task_key)

    def mutation_ledger(self, **params: Any) -> tuple[MutationLedgerEntry, ...]:
        """Build the mutation ledger preview for *params* (read-only)."""
        target = _format_target(self.action_id, params)
        return (
            MutationLedgerEntry(
                sequence=1,
                mcp_tool=self.mcp_tool,
                permission=self.required_permission,
                target=target,
                summary=self.description,
            ),
        )

    def preview(self, **params: Any) -> dict[str, Any]:
        ledger = self.mutation_ledger(**params)
        return {
            "action_id": self.action_id,
            "label": self.label,
            "task_key": self.task_key,
            "mcp_tool": self.mcp_tool,
            "required_permission": self.required_permission,
            "required_role": self.required_role,
            "enabled": self.enabled and _MVP_ACTIONS_ENABLED,
            "mvp_mode": "read-only",
            "mutation_ledger": [asdict(entry) for entry in ledger],
            "params": dict(params),
        }

    def attempt(self, **params: Any) -> dict[str, Any]:
        """Fail closed — never invokes MCP tools in MVP."""
        preview = self.preview(**params)
        if not preview["enabled"]:
            return {
                "success": False,
                "error": "action_disabled",
                "detail": (
                    "UI actions are disabled in the read-only MVP. "
                    "Use MCP tools with capability gates."
                ),
                "preview": preview,
            }
        return {
            "success": False,
            "error": "action_not_implemented",
            "detail": "Gated execution path is not wired in MVP.",
            "preview": preview,
        }


def _format_target(action_id: str, params: dict[str, Any]) -> str:
    if action_id == "claim_issue":
        return f"issue #{params.get('issue_number', '?')}"
    if action_id in {"comment_issue", "close_issue"}:
        return f"issue #{params.get('issue_number', '?')}"
    if action_id in {"review_pr", "merge_pr", "comment_pr", "close_pr"}:
        return f"PR #{params.get('pr_number', '?')}"
    if action_id == "delete_branch":
        return f"branch {params.get('branch_name', '?')!r}"
    if action_id == "create_pr":
        return (
            f"PR {params.get('head', '?')} → {params.get('base', 'master')}"
        )
    if action_id == "create_issue":
        return f"issue {params.get('title', '?')!r}"
    return "unspecified"


@dataclass
class ActionRegistry:
    """Ordered registry of gated UI actions."""

    actions: tuple[GatedAction, ...] = field(default_factory=tuple)

    def get(self, action_id: str) -> GatedAction | None:
        for action in self.actions:
            if action.action_id == action_id:
                return action
        return None

    def to_dict(self) -> dict[str, Any]:
        return {
            "mvp_actions_enabled": _MVP_ACTIONS_ENABLED,
            "actions": [
                {
                    "action_id": action.action_id,
                    "label": action.label,
                    "task_key": action.task_key,
                    "mcp_tool": action.mcp_tool,
                    "required_permission": action.required_permission,
                    "required_role": action.required_role,
                    "enabled": action.enabled,
                    "description": action.description,
                }
                for action in self.actions
            ],
        }


def build_action_registry() -> ActionRegistry:
    """Return the canonical MVP action set (all disabled)."""
    specs = (
        ("claim_issue", "Claim issue", "claim_issue", "gitea_mark_issue",
         "Apply status:in-progress via issue comment gate."),
        ("comment_issue", "Comment on issue", "comment_issue",
         "gitea_create_issue_comment", "Post an issue comment."),
        ("create_issue", "Create issue", "create_issue", "gitea_create_issue",
         "Open a new tracking issue."),
        ("create_pr", "Create pull request", "create_pr", "gitea_create_pr",
         "Open a PR from a locked feature branch."),
        ("review_pr", "Review PR", "review_pr", "gitea_review_pr",
         "Submit approve/request-changes/comment review."),
        ("merge_pr", "Merge PR", "merge_pr", "gitea_merge_pr",
         "Merge an approved pull request."),
        ("delete_branch", "Delete branch", "delete_branch",
         "gitea_delete_branch", "Remove a remote feature branch."),
        ("comment_pr", "Comment on PR", "comment_pr",
         "gitea_create_issue_comment", "Post a PR review thread comment."),
        ("close_pr", "Close PR", "close_pr", "gitea_edit_pr",
         "Close a pull request without merge."),
    )
    actions = tuple(
        GatedAction(
            action_id=action_id,
            label=label,
            task_key=task_key,
            mcp_tool=mcp_tool,
            description=description,
            enabled=False,
        )
        for action_id, label, task_key, mcp_tool, description in specs
    )
    return ActionRegistry(actions=actions)


_REGISTRY = build_action_registry()


def load_action_registry() -> ActionRegistry:
    return _REGISTRY


def preview_action(action_id: str, **params: Any) -> dict[str, Any]:
    action = _REGISTRY.get(action_id)
    if action is None:
        return {"error": "unknown_action", "action_id": action_id}
    return action.preview(**params)


def attempt_action(action_id: str, **params: Any) -> dict[str, Any]:
    action = _REGISTRY.get(action_id)
    if action is None:
        return {"success": False, "error": "unknown_action", "action_id": action_id}
    return action.attempt(**params)