Merge pull request 'Web UI: Gated action framework (#434)' (#455) from feat/issue-434-gated-actions into master

docs/webui-local-dev.md

@@ -49,8 +49,9 @@ Optional environment variables:
 | `/api/audit` | JSON validator preview (POST `report_text`, optional `task_kind`) |
 | `/worktrees` | Worktree hygiene dashboard (#432) |
 | `/api/worktrees` | JSON worktree scan with classifications and anomalies |
-| `/leases` | Stub — lease visibility (#433) |
-| `/worktrees` | Stub — hygiene dashboard (#432) |
+| `/actions` | Gated write-action registry — all disabled in MVP (#434) |
+| `/api/actions` | JSON action registry with capability metadata |
+| `/api/actions/{id}/preview` | Mutation ledger preview (GET, read-only) |
 | `/leases` | Lease and collision visibility (#433) |
 | `/api/leases` | JSON lease/collision export |
 
@@ -97,6 +98,15 @@ credentials. The UI surfaces pagination proof (returned count, pages fetched,
 If credentials are missing or the fetch fails, the page shows an explicit error
 instead of an empty queue (fail closed).
 
+## Gated actions (#434)
+
+`/actions` registers future write actions (claim, comment, review, merge,
+delete branch, create PR/issue). Each entry declares the MCP tool, required
+permission, and profile role from `task_capability_map.py` — aligned with
+`gitea_resolve_task_capability`. Buttons are disabled; previews always render
+a mutation ledger. Direct `attempt_action` calls fail closed without invoking
+MCP tools.
+
 ## Worktree hygiene (#432)
 
 `/worktrees` scans local `branches/` directories and registered git worktrees.
@@ -106,6 +116,7 @@ referenced by the issue lock file are flagged as anomalies (#404). The page
 includes a copy/paste canonical cleanup prompt only — no deletion actions.
 
 Override scan root with `WEBUI_REPO_ROOT` (defaults to repository root).
+
 ## Lease visibility (#433)
 
 `/leases` surfaces read-only lease and collision state: local issue lock file,
@@ -126,6 +137,7 @@ no tokens or MCP restart actions are exposed.
 ## Tests
 
 ```bash
+pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_gated_actions.py -q
 pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_audit.py tests/test_webui_worktree_hygiene.py -q
 pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_lease_visibility.py tests/test_webui_runtime_health.py -q
 ```

tests/test_webui_gated_actions.py

@@ -0,0 +1,128 @@
+"""Tests for web UI gated action framework (#434)."""
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from starlette.testclient import TestClient
+
+from task_capability_map import TASK_CAPABILITY_MAP
+from webui.app import create_app
+from webui.gated_actions import (
+    attempt_action,
+    build_action_registry,
+    load_action_registry,
+    preview_action,
+)
+
+
+class TestGatedActionRegistry(unittest.TestCase):
+    def test_all_actions_disabled_by_default(self):
+        registry = build_action_registry()
+        self.assertGreater(len(registry.actions), 0)
+        for action in registry.actions:
+            with self.subTest(action=action.action_id):
+                self.assertFalse(action.enabled)
+
+    def test_actions_align_with_task_capability_map(self):
+        registry = build_action_registry()
+        for action in registry.actions:
+            with self.subTest(task=action.task_key):
+                self.assertIn(action.task_key, TASK_CAPABILITY_MAP)
+                self.assertEqual(
+                    action.required_permission,
+                    TASK_CAPABILITY_MAP[action.task_key]["permission"],
+                )
+                self.assertEqual(
+                    action.required_role,
+                    TASK_CAPABILITY_MAP[action.task_key]["role"],
+                )
+
+    def test_preview_includes_mutation_ledger(self):
+        preview = preview_action("merge_pr", pr_number=42)
+        self.assertNotIn("error", preview)
+        ledger = preview["mutation_ledger"]
+        self.assertEqual(len(ledger), 1)
+        self.assertEqual(ledger[0]["mcp_tool"], "gitea_merge_pr")
+        self.assertEqual(ledger[0]["permission"], "gitea.pr.merge")
+        self.assertIn("42", ledger[0]["target"])
+
+    def test_attempt_fails_closed_without_mcp_calls(self):
+        registry = build_action_registry()
+        with patch("webui.gated_actions._MVP_ACTIONS_ENABLED", False):
+            for action in registry.actions:
+                with self.subTest(action=action.action_id):
+                    result = attempt_action(action.action_id, issue_number=1)
+                    self.assertFalse(result["success"])
+                    self.assertEqual(result["error"], "action_disabled")
+                    self.assertIn("preview", result)
+
+    def test_attempt_module_has_no_mcp_imports(self):
+        """Ungated execution path must not import MCP server tools."""
+        import webui.gated_actions as ga
+
+        source_path = Path(ga.__file__).resolve()
+        source = source_path.read_text(encoding="utf-8")
+        self.assertNotIn("mcp_server", source)
+        result = attempt_action("merge_pr", pr_number=99)
+        self.assertFalse(result["success"])
+
+    def test_unknown_action_fails_closed(self):
+        result = attempt_action("nonexistent_action")
+        self.assertFalse(result["success"])
+        self.assertEqual(result["error"], "unknown_action")
+
+
+class TestGatedActionRoutes(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+
+    def test_actions_page_renders_disabled_buttons(self):
+        response = self.client.get("/actions")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Gated actions", response.text)
+        self.assertIn("disabled", response.text.lower())
+        self.assertIn("mutation ledger", response.text.lower())
+        self.assertIn("aria-disabled", response.text)
+        self.assertIn(" disabled", response.text)
+
+    def test_api_actions_registry(self):
+        response = self.client.get("/api/actions")
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+        self.assertFalse(data["mvp_actions_enabled"])
+        action_ids = {item["action_id"] for item in data["actions"]}
+        self.assertIn("merge_pr", action_ids)
+        self.assertIn("claim_issue", action_ids)
+        for item in data["actions"]:
+            self.assertFalse(item["enabled"])
+
+    def test_api_action_preview_json(self):
+        response = self.client.get(
+            "/api/actions/review_pr/preview?pr_number=12"
+        )
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+        self.assertEqual(data["task_key"], "review_pr")
+        self.assertEqual(data["required_role"], "reviewer")
+        self.assertEqual(len(data["mutation_ledger"]), 1)
+
+    def test_post_attempt_fails_closed(self):
+        response = self.client.post(
+            "/api/actions/merge_pr/attempt",
+            json={"pr_number": 1},
+        )
+        self.assertEqual(response.status_code, 403)
+        data = response.json()
+        self.assertFalse(data["success"])
+        self.assertEqual(data["error"], "action_disabled")
+
+    def test_nav_includes_actions_link(self):
+        text = self.client.get("/").text
+        self.assertIn('href="/actions"', text)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_webui_skeleton.py

@@ -61,6 +61,11 @@ class TestWebuiSkeleton(unittest.TestCase):
         self.assertIn("Collision warnings", response.text)
 
 
+    def test_actions_is_implemented(self):
+        response = self.client.get("/actions")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Gated actions", response.text)
+
     def test_post_is_rejected(self):
         response = self.client.post("/health")
         self.assertEqual(response.status_code, 405)
@@ -72,8 +77,8 @@ class TestWebuiSkeleton(unittest.TestCase):
         self.assertIn("Live queue", response.text)
 
     def test_nav_links_on_all_pages(self):
-        paths = ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases")
-        hrefs = ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases")
+        paths = ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases", "/actions")
+        hrefs = ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases", "/actions")
         for path in paths:
             with self.subTest(path=path):
                 text = self.client.get(path).text

webui/app.py

@@ -16,6 +16,8 @@ from webui.prompt_library import find_prompt, library_to_dict
 from webui.prompt_views import render_prompt_detail, render_prompts_page
 from final_report_validator import FINAL_REPORT_TASK_KINDS
 
+from webui.gated_actions import attempt_action, load_action_registry, preview_action
+from webui.gated_action_views import render_actions_page
 from webui.audit_validator import audit_report, audit_to_dict
 from webui.audit_views import render_audit_page
 from webui.lease_loader import load_lease_snapshot, snapshot_to_dict as lease_snapshot_to_dict
@@ -52,6 +54,7 @@ async def home(_request: Request) -> HTMLResponse:
         "<li><strong>Audit</strong> — final-report paste and validator preview (#431)</li>"
         "<li><strong>Worktrees</strong> — branch hygiene dashboard (#432)</li>"
         "<li><strong>Leases</strong> — collision and lease visibility (#433)</li>"
+        "<li><strong>Actions</strong> — gated write-action framework (#434)</li>"
         "</ul>"
     )
     return HTMLResponse(render_page(title="Home", body_html=body))
@@ -200,6 +203,41 @@ async def api_leases(_request: Request) -> JSONResponse:
     return JSONResponse(lease_snapshot_to_dict(load_lease_snapshot()))
 
 
+async def actions(_request: Request) -> HTMLResponse:
+    registry = load_action_registry()
+    return HTMLResponse(render_actions_page(registry))
+
+
+async def api_actions(_request: Request) -> JSONResponse:
+    return JSONResponse(load_action_registry().to_dict())
+
+
+async def api_action_preview(request: Request) -> JSONResponse:
+    action_id = request.path_params["action_id"]
+    params = dict(request.query_params)
+    for key in ("issue_number", "pr_number"):
+        if key in params:
+            params[key] = int(params[key])
+    result = preview_action(action_id, **params)
+    if "error" in result:
+        return JSONResponse(result, status_code=404)
+    return JSONResponse(result)
+
+
+async def api_action_attempt(request: Request) -> JSONResponse:
+    """POST is rejected by read-only guard; kept for explicit fail-closed tests."""
+    action_id = request.path_params["action_id"]
+    try:
+        body = await request.json()
+    except Exception:
+        body = {}
+    if not isinstance(body, dict):
+        body = {}
+    result = attempt_action(action_id, **body)
+    status = 403 if not result.get("success") else 200
+    return JSONResponse(result, status_code=status)
+
+
 async def method_not_allowed(request: Request, _exc: Exception) -> Response:
     path = request.url.path
     if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
@@ -234,6 +272,18 @@ def create_app() -> Starlette:
             Route("/worktrees", worktrees, methods=["GET"]),
             Route("/api/worktrees", api_worktrees, methods=["GET"]),
             Route("/leases", leases, methods=["GET"]),
+            Route("/actions", actions, methods=["GET"]),
+            Route("/api/actions", api_actions, methods=["GET"]),
+            Route(
+                "/api/actions/{action_id}/preview",
+                api_action_preview,
+                methods=["GET"],
+            ),
+            Route(
+                "/api/actions/{action_id}/attempt",
+                api_action_attempt,
+                methods=["POST"],
+            ),
             Route("/api/leases", api_leases, methods=["GET"]),
         ],
         exception_handlers={405: method_not_allowed},

webui/gated_action_views.py

@@ -0,0 +1,101 @@
+"""HTML views for gated action previews (#434)."""
+
+from __future__ import annotations
+
+import html
+import json
+
+from webui.gated_actions import ActionRegistry, GatedAction, preview_action
+from webui.layout import render_page
+
+
+def _escape(text: str) -> str:
+    return html.escape(text, quote=True)
+
+
+def _ledger_block(action: GatedAction) -> str:
+    sample_params: dict[str, object]
+    if action.action_id == "create_issue":
+        sample_params = {"title": "Example issue"}
+    elif action.action_id == "create_pr":
+        sample_params = {"head": "feat/issue-99-example", "base": "master"}
+    elif action.action_id == "delete_branch":
+        sample_params = {"branch_name": "feat/issue-99-example"}
+    elif action.action_id in {"review_pr", "merge_pr", "comment_pr", "close_pr"}:
+        sample_params = {"pr_number": 99}
+    else:
+        sample_params = {"issue_number": 99}
+
+    preview = preview_action(action.action_id, **sample_params)
+    ledger_json = json.dumps(preview.get("mutation_ledger", []), indent=2)
+    return (
+        "<details class='action-preview'>"
+        "<summary>Preview mutation ledger</summary>"
+        f"<pre class='prompt-text'>{_escape(ledger_json)}</pre>"
+        f"<p class='muted meta'>Sample params: "
+        f"<code>{_escape(json.dumps(sample_params))}</code></p>"
+        f"<p><a href=\"/api/actions/{_escape(action.action_id)}/preview?"
+        f"{_preview_query(sample_params)}\">JSON preview</a></p>"
+        "</details>"
+    )
+
+
+def _preview_query(params: dict[str, object]) -> str:
+    parts = []
+    for key, value in params.items():
+        parts.append(f"{key}={_escape(str(value))}")
+    return "&".join(parts)
+
+
+def _action_card(action: GatedAction) -> str:
+    disabled_attr = " disabled" if not action.enabled else ""
+    return (
+        "<article class='prompt-card action-card'>"
+        f"<h3>{_escape(action.label)}</h3>"
+        f"<p class='muted'>{_escape(action.description)}</p>"
+        "<p class='meta'>"
+        f"Task <code>{_escape(action.task_key)}</code> · "
+        f"MCP <code>{_escape(action.mcp_tool)}</code><br>"
+        f"Requires <code>{_escape(action.required_permission)}</code> "
+        f"({_escape(action.required_role)} profile)</p>"
+        f"<button type='button' class='copy-btn action-btn'{disabled_attr} "
+        f"aria-disabled='true' title='Disabled in read-only MVP'>"
+        f"{_escape(action.label)}</button>"
+        f"{_ledger_block(action)}"
+        "</article>"
+    )
+
+
+ACTION_PAGE_STYLES = """
+<style>
+  .action-card .action-btn[disabled] {
+    opacity: 0.45;
+    cursor: not-allowed;
+    filter: none;
+  }
+  .action-preview { margin-top: 0.75rem; }
+  .action-preview summary {
+    cursor: pointer;
+    color: var(--accent);
+    font-size: 0.9rem;
+  }
+</style>
+"""
+
+
+def render_actions_page(registry: ActionRegistry) -> str:
+    cards = "".join(_action_card(action) for action in registry.actions)
+    body = (
+        "<h2>Gated actions</h2>"
+        "<p>Future write actions are registered here but remain "
+        "<strong>disabled</strong> in the read-only MVP. Each action "
+        "declares the MCP capability and profile role from "
+        "<code>task_capability_map</code>; previews show the mutation "
+        "ledger before any execution path ships.</p>"
+        f"<p class='meta'>{len(registry.actions)} registered actions · "
+        "execution: fail-closed</p>"
+        f"{cards}"
+        "<p><a href=\"/api/actions\">JSON registry</a></p>"
+        f"{ACTION_PAGE_STYLES}"
+    )
+    return render_page(title="Actions", body_html=body)

webui/gated_actions.py

@@ -0,0 +1,201 @@
+"""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)

webui/layout.py

@@ -11,6 +11,7 @@ NAV_ITEMS = (
     ("/audit", "Audit"),
     ("/worktrees", "Worktrees"),
     ("/leases", "Leases"),
+    ("/actions", "Actions"),
 )
 
 MVP_NOTICE = (