From 84b841c72765ee1249cc6b9e4957ce1bf51ae606 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 15:34:39 -0400 Subject: [PATCH] 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 --- docs/webui-local-dev.md | 18 ++- tests/test_webui_gated_actions.py | 128 +++++++++++++++++++ tests/test_webui_skeleton.py | 11 +- webui/app.py | 50 ++++++++ webui/gated_action_views.py | 101 +++++++++++++++ webui/gated_actions.py | 201 ++++++++++++++++++++++++++++++ webui/layout.py | 1 + 7 files changed, 505 insertions(+), 5 deletions(-) create mode 100644 tests/test_webui_gated_actions.py create mode 100644 webui/gated_action_views.py create mode 100644 webui/gated_actions.py diff --git a/docs/webui-local-dev.md b/docs/webui-local-dev.md index 0a5b935..92e10b8 100644 --- a/docs/webui-local-dev.md +++ b/docs/webui-local-dev.md @@ -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 ``` \ No newline at end of file diff --git a/tests/test_webui_gated_actions.py b/tests/test_webui_gated_actions.py new file mode 100644 index 0000000..87d4086 --- /dev/null +++ b/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() \ No newline at end of file diff --git a/tests/test_webui_skeleton.py b/tests/test_webui_skeleton.py index 5dd23a1..2960fc5 100644 --- a/tests/test_webui_skeleton.py +++ b/tests/test_webui_skeleton.py @@ -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) diff --git a/webui/app.py b/webui/app.py index a97d1ad..4557f7b 100644 --- a/webui/app.py +++ b/webui/app.py @@ -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: "
  • Audit — final-report paste and validator preview (#431)
    • " "
  • Worktrees — branch hygiene dashboard (#432)
    • " "
  • Leases — collision and lease visibility (#433)
    • " + "
  • Actions — gated write-action framework (#434)
    • " "" ) 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}, ) \ No newline at end of file diff --git a/webui/gated_action_views.py b/webui/gated_action_views.py new file mode 100644 index 0000000..0d01d61 --- /dev/null +++ b/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 ( + "
    " + "Preview mutation ledger" + f"
    {_escape(ledger_json)}
    " + f"

    Sample params: " + f"{_escape(json.dumps(sample_params))}

    " + f"

    JSON preview

    " + "
    " + ) + + +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 ( + "
    " + f"

    {_escape(action.label)}

    " + f"

    {_escape(action.description)}

    " + "

    " + f"Task {_escape(action.task_key)} · " + f"MCP {_escape(action.mcp_tool)}
    " + f"Requires {_escape(action.required_permission)} " + f"({_escape(action.required_role)} profile)

    " + f"" + f"{_ledger_block(action)}" + "
    " + ) + + +ACTION_PAGE_STYLES = """ + +""" + + +def render_actions_page(registry: ActionRegistry) -> str: + cards = "".join(_action_card(action) for action in registry.actions) + body = ( + "

    Gated actions

    " + "

    Future write actions are registered here but remain " + "disabled in the read-only MVP. Each action " + "declares the MCP capability and profile role from " + "task_capability_map; previews show the mutation " + "ledger before any execution path ships.

    " + f"

    {len(registry.actions)} registered actions · " + "execution: fail-closed

    " + f"{cards}" + "

    JSON registry

    " + f"{ACTION_PAGE_STYLES}" + ) + return render_page(title="Actions", body_html=body) \ No newline at end of file diff --git a/webui/gated_actions.py b/webui/gated_actions.py new file mode 100644 index 0000000..baefab8 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/webui/layout.py b/webui/layout.py index a506c2c..47bedc1 100644 --- a/webui/layout.py +++ b/webui/layout.py @@ -11,6 +11,7 @@ NAV_ITEMS = ( ("/audit", "Audit"), ("/worktrees", "Worktrees"), ("/leases", "Leases"), + ("/actions", "Actions"), ) MVP_NOTICE = (