diff --git a/docs/webui-local-dev.md b/docs/webui-local-dev.md index 7aa6e85..990914e 100644 --- a/docs/webui-local-dev.md +++ b/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 ``` \ 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 8c6dd46..d9de18e 100644 --- a/tests/test_webui_skeleton.py +++ b/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 diff --git a/webui/app.py b/webui/app.py index ad39928..4ab38ed 100644 --- a/webui/app.py +++ b/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: "
  • 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)) @@ -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}, 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 = (