Web UI: Gated action framework (#434) #455

Merged
sysadmin merged 5 commits from feat/issue-434-gated-actions into master 2026-07-09 10:39:29 -05:00
7 changed files with 502 additions and 4 deletions
+14 -2
View File
@@ -49,8 +49,9 @@ Optional environment variables:
| `/api/audit` | JSON validator preview (POST `report_text`, optional `task_kind`) | | `/api/audit` | JSON validator preview (POST `report_text`, optional `task_kind`) |
| `/worktrees` | Worktree hygiene dashboard (#432) | | `/worktrees` | Worktree hygiene dashboard (#432) |
| `/api/worktrees` | JSON worktree scan with classifications and anomalies | | `/api/worktrees` | JSON worktree scan with classifications and anomalies |
| `/leases` | Stub — lease visibility (#433) | | `/actions` | Gated write-action registry — all disabled in MVP (#434) |
| `/worktrees` | Stub — hygiene dashboard (#432) | | `/api/actions` | JSON action registry with capability metadata |
| `/api/actions/{id}/preview` | Mutation ledger preview (GET, read-only) |
| `/leases` | Lease and collision visibility (#433) | | `/leases` | Lease and collision visibility (#433) |
| `/api/leases` | JSON lease/collision export | | `/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 If credentials are missing or the fetch fails, the page shows an explicit error
instead of an empty queue (fail closed). 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) ## Worktree hygiene (#432)
`/worktrees` scans local `branches/` directories and registered git worktrees. `/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. includes a copy/paste canonical cleanup prompt only — no deletion actions.
Override scan root with `WEBUI_REPO_ROOT` (defaults to repository root). Override scan root with `WEBUI_REPO_ROOT` (defaults to repository root).
## Lease visibility (#433) ## Lease visibility (#433)
`/leases` surfaces read-only lease and collision state: local issue lock file, `/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 ## Tests
```bash ```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_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 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
``` ```
+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()
+7 -2
View File
@@ -61,6 +61,11 @@ class TestWebuiSkeleton(unittest.TestCase):
self.assertIn("Collision warnings", response.text) 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): def test_post_is_rejected(self):
response = self.client.post("/health") response = self.client.post("/health")
self.assertEqual(response.status_code, 405) self.assertEqual(response.status_code, 405)
@@ -72,8 +77,8 @@ class TestWebuiSkeleton(unittest.TestCase):
self.assertIn("Live queue", response.text) self.assertIn("Live queue", response.text)
def test_nav_links_on_all_pages(self): def test_nav_links_on_all_pages(self):
paths = ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases") paths = ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases", "/actions")
hrefs = ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases") hrefs = ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases", "/actions")
for path in paths: for path in paths:
with self.subTest(path=path): with self.subTest(path=path):
text = self.client.get(path).text text = self.client.get(path).text
+50
View File
@@ -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 webui.prompt_views import render_prompt_detail, render_prompts_page
from final_report_validator import FINAL_REPORT_TASK_KINDS 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_validator import audit_report, audit_to_dict
from webui.audit_views import render_audit_page from webui.audit_views import render_audit_page
from webui.lease_loader import load_lease_snapshot, snapshot_to_dict as lease_snapshot_to_dict 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>Audit</strong> — final-report paste and validator preview (#431)</li>"
"<li><strong>Worktrees</strong> — branch hygiene dashboard (#432)</li>" "<li><strong>Worktrees</strong> — branch hygiene dashboard (#432)</li>"
"<li><strong>Leases</strong> — collision and lease visibility (#433)</li>" "<li><strong>Leases</strong> — collision and lease visibility (#433)</li>"
"<li><strong>Actions</strong> — gated write-action framework (#434)</li>"
"</ul>" "</ul>"
) )
return HTMLResponse(render_page(title="Home", body_html=body)) 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())) 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: async def method_not_allowed(request: Request, _exc: Exception) -> Response:
path = request.url.path path = request.url.path
if path in _AUDIT_MUTATION_PATHS and request.method == "POST": if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
@@ -234,6 +272,18 @@ def create_app() -> Starlette:
Route("/worktrees", worktrees, methods=["GET"]), Route("/worktrees", worktrees, methods=["GET"]),
Route("/api/worktrees", api_worktrees, methods=["GET"]), Route("/api/worktrees", api_worktrees, methods=["GET"]),
Route("/leases", leases, 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"]), Route("/api/leases", api_leases, methods=["GET"]),
], ],
exception_handlers={405: method_not_allowed}, 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"), ("/audit", "Audit"),
("/worktrees", "Worktrees"), ("/worktrees", "Worktrees"),
("/leases", "Leases"), ("/leases", "Leases"),
("/actions", "Actions"),
) )
MVP_NOTICE = ( MVP_NOTICE = (