feat(webui): gated action framework (#434)

Register future write actions with task_capability_map metadata,
mutation-ledger previews, and fail-closed execution in the read-only MVP.
Adds /actions routes, disabled UI buttons, and tests proving ungated
attempts cannot invoke mutations.

Closes #434
This commit is contained in:
2026-07-07 15:34:39 -04:00
parent ee8e9a0247
commit 84b841c727
7 changed files with 505 additions and 5 deletions
+15 -3
View File
@@ -47,9 +47,12 @@ Optional environment variables:
| `/audit` | Stub — report audit paste (#431) |
| `/worktrees` | Stub — hygiene dashboard (#432) |
| `/leases` | Stub — lease visibility (#433) |
| `/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) |
All routes are GET-only. POST/PUT/PATCH/DELETE return `405` with
`read-only-mvp`.
All routes are GET-only except registered POST handlers, which still return
`405` with `read-only-mvp` until write paths ship.
## Project registry (#427)
@@ -82,8 +85,17 @@ 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.
## 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 -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_gated_actions.py -q
```
+128
View File
@@ -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()
+9 -2
View File
@@ -51,6 +51,11 @@ class TestWebuiSkeleton(unittest.TestCase):
with self.subTest(path=path):
self.assertEqual(self.client.get(path).status_code, 200)
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)
@@ -62,10 +67,12 @@ class TestWebuiSkeleton(unittest.TestCase):
self.assertIn("Live queue", response.text)
def test_nav_links_on_all_pages(self):
for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit"):
for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/actions"):
with self.subTest(path=path):
text = self.client.get(path).text
for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit"):
for href in (
"/queue", "/projects", "/prompts", "/runtime", "/audit", "/actions",
):
self.assertIn(f'href="{href}"', text)
+50
View File
@@ -14,6 +14,8 @@ from webui.project_registry import find_project, load_registry, registry_to_dict
from webui.project_views import render_project_detail, render_projects_list
from webui.prompt_library import find_prompt, library_to_dict
from webui.prompt_views import render_prompt_detail, render_prompts_page
from webui.gated_actions import attempt_action, load_action_registry, preview_action
from webui.gated_action_views import render_actions_page
from webui.queue_loader import load_queue_snapshot, snapshot_to_dict
from webui.queue_views import render_queue_page
@@ -41,6 +43,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))
@@ -147,6 +150,41 @@ async def leases(_request: Request) -> HTMLResponse:
)
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:
if request.method not in _READ_ONLY_METHODS:
return JSONResponse(
@@ -175,6 +213,18 @@ def create_app() -> Starlette:
Route("/audit", audit, methods=["GET"]),
Route("/worktrees", 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"],
),
],
exception_handlers={405: method_not_allowed},
)
+101
View File
@@ -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)
+201
View File
@@ -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)
+1
View File
@@ -11,6 +11,7 @@ NAV_ITEMS = (
("/audit", "Audit"),
("/worktrees", "Worktrees"),
("/leases", "Leases"),
("/actions", "Actions"),
)
MVP_NOTICE = (