Merge remote-tracking branch 'prgs/master' into feat/issue-436-ci-coverage

# Conflicts:
#	docs/webui-local-dev.md

branch_cleanup_guard.py

@@ -1,98 +0,0 @@
-"""Guards for merged-PR branch cleanup and raw git delete bypasses (#514)."""
-
-from __future__ import annotations
-
-import re
-from typing import Any
-
-PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})
-
-_RAW_BRANCH_DELETE_PATTERNS = (
-    re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+branch\s+-[dD]\b[^\n\r]*", re.I),
-    re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+push\b[^\n\r]*\s--delete\b[^\n\r]*", re.I),
-    re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+push\b[^\n\r]*\s:[^\s`]+", re.I),
-)
-
-
-def raw_branch_delete_commands(text: str | None) -> list[str]:
-    """Return raw git branch-delete commands cited in *text*."""
-    if not text:
-        return []
-    commands: list[str] = []
-    for pattern in _RAW_BRANCH_DELETE_PATTERNS:
-        commands.extend(match.group(0).strip("` ") for match in pattern.finditer(text))
-    return list(dict.fromkeys(commands))
-
-
-def assess_raw_branch_delete_report(text: str | None) -> dict[str, Any]:
-    """Fail closed when a report uses raw git branch deletion as cleanup proof."""
-    commands = raw_branch_delete_commands(text)
-    reasons = [
-        (
-            "raw git branch deletion bypasses MCP branch.delete cleanup gates: "
-            f"{command}"
-        )
-        for command in commands
-    ]
-    return {
-        "proven": not reasons,
-        "block": bool(reasons),
-        "commands": commands,
-        "reasons": reasons,
-        "safe_next_action": (
-            "use gitea_cleanup_merged_pr_branch or another approved cleanup "
-            "helper with explicit branch.delete capability"
-            if reasons
-            else "proceed"
-        ),
-    }
-
-
-def assess_merged_pr_branch_cleanup(
-    *,
-    pr_number: int,
-    head_branch: str,
-    merged: bool,
-    remote_branch_exists: bool,
-    open_pr_heads: set[str],
-    head_on_target: bool | None,
-    delete_capability_allowed: bool,
-    confirmation: str | None,
-    protected_branches: frozenset[str] | None = None,
-) -> dict[str, Any]:
-    """Assess whether a merged PR source branch can be deleted via MCP."""
-    protected = protected_branches or PROTECTED_BRANCHES
-    expected_confirmation = f"CLEANUP MERGED PR {pr_number} BRANCH {head_branch}"
-    reasons: list[str] = []
-    if not merged:
-        reasons.append("PR is not merged")
-    if not remote_branch_exists:
-        reasons.append("remote branch already absent")
-    if not head_branch:
-        reasons.append("PR head branch is missing")
-    if head_branch in protected:
-        reasons.append(f"branch '{head_branch}' is protected")
-    if head_branch in open_pr_heads:
-        reasons.append("an open PR still references this head branch")
-    if head_on_target is False:
-        reasons.append("PR head is not an ancestor of the target branch")
-    if head_on_target is None:
-        reasons.append("PR head ancestry could not be proven")
-    if not delete_capability_allowed:
-        reasons.append("gitea.branch.delete capability is not allowed")
-    if confirmation != expected_confirmation:
-        reasons.append(
-            "confirmation must equal "
-            f"'{expected_confirmation}' for branch cleanup"
-        )
-
-    safe = not reasons
-    return {
-        "pr_number": pr_number,
-        "head_branch": head_branch,
-        "expected_confirmation": expected_confirmation,
-        "remote_branch_exists": remote_branch_exists,
-        "safe_to_delete": safe,
-        "block_reasons": reasons,
-        "recommended_action": "delete_remote_branch" if safe else "keep_remote_branch",
-    }

docs/webui-local-dev.md

@@ -49,9 +49,8 @@ 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 |
-| `/actions` | Gated write-action registry — all disabled in MVP (#434) |
+| `/leases` | Stub — lease visibility (#433) |
-| `/api/actions` | JSON action registry with capability metadata |
+| `/worktrees` | Stub — hygiene dashboard (#432) |
-| `/api/actions/{id}/preview` | Mutation ledger preview (GET, read-only) |
 | `/leases` | Lease and collision visibility (#433) |
 | `/api/leases` | JSON lease/collision export |
 
@@ -98,15 +97,6 @@ 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.
@@ -117,30 +107,6 @@ 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,
-in-progress claim inventory (#268), reviewer PR lease comments when present
-(`<!-- mcp-review-lease:v1 -->`, #407), duplicate open PRs per issue (#400),
-and duplicate local branches per issue. Links to collision-history backend
-issues (#267, #268, #400, #407) are included. No lease acquire/release from UI.
-
-## Runtime health (#430)
-
-`/runtime` surfaces read-only MCP/runtime diagnostics for the default registry
-project: active profile and role kind, authenticated identity (when credentials
-are available), config model/mode, local vs remote `master` SHA sync, shell
-health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
-checkout is behind merged safety-gate changes. Restart guidance links to #420;
-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 (#436)
 
 Run the full hermetic web UI suite (all `test_webui_*.py` modules):
@@ -166,8 +132,6 @@ Or invoke unittest directly:
 
 ```bash
 python3 -m unittest discover -s tests -p 'test_webui_*.py' -q
-```
-
 ## Lease visibility (#433)
 
 `/leases` surfaces read-only lease and collision state: local issue lock file,

final_report_validator.py

@@ -11,7 +11,6 @@ import inspect
 import re
 from typing import Any, Callable
 
-import branch_cleanup_guard
 import issue_acceptance_gate
 import issue_lock_provenance
 import reviewer_handoff_consistency
@@ -1295,17 +1294,6 @@ def _rule_shared_author_reviewer_same_run(report_text: str) -> list[dict[str, st
     )
 
 
-def _rule_shared_raw_branch_delete_bypass(report_text: str) -> list[dict[str, str]]:
-    result = branch_cleanup_guard.assess_raw_branch_delete_report(report_text)
-    if not result["block"]:
-        return []
-    return _findings_from_reasons(
-        "shared.raw_branch_delete_bypass",
-        result["reasons"],
-        field="Cleanup mutations",
-        severity="block",
-        safe_next_action=result["safe_next_action"],
-    )
 def _rule_reviewer_workflow_load_boundary(report_text: str) -> list[dict[str, str]]:
     """#403: require structured workflow-load helper result, not file-view narrative."""
     if not report_text.strip():
@@ -1460,7 +1448,6 @@ _SHARED_ISSUE_LOCK_RULES = (
     _rule_shared_issue_lock_external_state,
     _rule_shared_manual_lock_pr_override,
     _rule_shared_author_reviewer_same_run,
-    _rule_shared_raw_branch_delete_bypass,
     _rule_shared_canonical_state_update,
 )
 

gitea_mcp_server.py

@@ -832,7 +832,6 @@ import pr_work_lease  # noqa: E402
 import workflow_skill  # noqa: E402
 import conflict_fix_classification  # noqa: E402
 import native_mcp_preference  # noqa: E402
-import branch_cleanup_guard  # noqa: E402
 import thread_state_ledger_validator  # noqa: E402
 import master_parity_gate  # noqa: E402
 
@@ -4853,149 +4852,6 @@ def gitea_delete_branch(
     return {"success": True, "message": f"Remote branch '{branch}' deleted."}
 
 
-@mcp.tool()
-def gitea_cleanup_merged_pr_branch(
-    pr_number: int,
-    confirmation: str,
-    branch: str | None = None,
-    remote: str = "dadeschools",
-    host: str | None = None,
-    org: str | None = None,
-    repo: str | None = None,
-    worktree_path: str | None = None,
-) -> dict:
-    """Delete a merged PR source branch through the guarded MCP path (#514)."""
-    gate_reasons = _profile_operation_gate("gitea.branch.delete")
-    if gate_reasons:
-        return {
-            "success": False,
-            "performed": False,
-            "required_permission": "gitea.branch.delete",
-            "reasons": gate_reasons,
-            "permission_report": _permission_block_report("gitea.branch.delete"),
-        }
-
-    profile = get_profile()
-    active_role = _role_kind(
-        profile.get("allowed_operations", []),
-        profile.get("forbidden_operations", []),
-    )
-    if active_role == "reviewer":
-        return {
-            "success": False,
-            "performed": False,
-            "required_permission": "gitea.branch.delete",
-            "reasons": [
-                "reviewer profile is not authorized for merged branch cleanup "
-                "(fail closed)"
-            ],
-            "permission_report": _permission_block_report("gitea.branch.delete"),
-        }
-
-    if worktree_path is None or "/branches/" not in os.path.realpath(worktree_path):
-        return {
-            "success": False,
-            "performed": False,
-            "required_permission": "gitea.branch.delete",
-            "reasons": [
-                "merged branch cleanup requires an explicit branches/ worktree "
-                "path; root checkout branch ref mutation is blocked (fail closed)"
-            ],
-        }
-
-    verify_preflight_purity(
-        remote,
-        worktree_path=worktree_path,
-        task="cleanup_merged_pr_branch",
-    )
-
-    h, o, r = _resolve(remote, host, org, repo)
-    auth = _auth(h)
-    base = repo_api_url(h, o, r)
-    pr = api_request("GET", f"{base}/pulls/{pr_number}", auth)
-    pr_head = pr.get("head") or {}
-    head_branch = branch or pr_head.get("ref") or ""
-    head_sha = pr_head.get("sha")
-    target_branch = (pr.get("base") or {}).get("ref") or "master"
-
-    if branch and branch != pr_head.get("ref"):
-        return {
-            "success": False,
-            "performed": False,
-            "pr_number": pr_number,
-            "branch": branch,
-            "reasons": [
-                f"requested branch '{branch}' does not match PR head "
-                f"'{pr_head.get('ref')}'"
-            ],
-        }
-
-    open_prs = api_get_all(f"{base}/pulls?state=open", auth)
-    open_heads = {
-        str((open_pr.get("head") or {}).get("ref"))
-        for open_pr in open_prs
-        if (open_pr.get("head") or {}).get("ref")
-    }
-    remote_exists = _remote_branch_exists(h, o, r, auth, head_branch)
-    target_ref = (
-        f"{remote}/{target_branch}"
-        if remote in REMOTES
-        else f"origin/{target_branch}"
-    )
-    head_on_target = merged_cleanup_reconcile.is_head_ancestor_of_ref(
-        PROJECT_ROOT,
-        head_sha,
-        target_ref,
-    )
-    assessment = branch_cleanup_guard.assess_merged_pr_branch_cleanup(
-        pr_number=pr_number,
-        head_branch=head_branch,
-        merged=bool(pr.get("merged") or pr.get("merged_at")),
-        remote_branch_exists=remote_exists,
-        open_pr_heads=open_heads,
-        head_on_target=head_on_target,
-        delete_capability_allowed=True,
-        confirmation=confirmation,
-    )
-    if not assessment["safe_to_delete"]:
-        return {
-            "success": False,
-            "performed": False,
-            "pr_number": pr_number,
-            "branch": head_branch,
-            "assessment": assessment,
-            "reasons": assessment["block_reasons"],
-        }
-
-    import urllib.parse
-
-    encoded_branch = urllib.parse.quote(head_branch, safe="")
-    url = f"{base}/branches/{encoded_branch}"
-    with _audited(
-        "cleanup_merged_pr_branch",
-        host=h,
-        remote=remote,
-        org=o,
-        repo=r,
-        pr_number=pr_number,
-        target_branch=head_branch,
-        request_metadata={
-            "branch": head_branch,
-            "required_permission": "gitea.branch.delete",
-            "cleanup_path": "gitea_cleanup_merged_pr_branch",
-        },
-    ):
-        api_request("DELETE", url, auth)
-    return {
-        "success": True,
-        "performed": True,
-        "pr_number": pr_number,
-        "branch": head_branch,
-        "message": f"Merged PR #{pr_number} source branch '{head_branch}' deleted.",
-        "assessment": assessment,
-    }
-
-
 def _remote_branch_exists(h: str, o: str, r: str, auth: str, branch: str) -> bool:
     import urllib.parse
 

reviewer_pr_lease.py

@@ -190,26 +190,16 @@ def find_active_reviewer_lease(
     pr_number: int,
     now: datetime | None = None,
 ) -> dict[str, Any] | None:
-    """Return the newest unexpired non-terminal lease for *pr_number*.
+    """Newest non-terminal, unexpired lease for *pr_number*."""
-
-    Append-only lease markers form a ledger: the **newest** marker is
-    authoritative. A later terminal phase (``released`` / ``done`` /
-    ``blocked``) ends the lease even when older ``claimed`` markers remain
-    on the thread (#577). Skipping only terminal markers and walking older
-    claims incorrectly re-arms a lease after a successful release.
-    """
     now = now or datetime.now(timezone.utc)
-    entries = list(reversed(_lease_entries(comments, pr_number=pr_number)))
+    for lease in reversed(_lease_entries(comments, pr_number=pr_number)):
-    if not entries:
+        phase = (lease.get("phase") or "").strip().lower()
-        return None
-    newest = entries[0]
-    phase = (newest.get("phase") or "").strip().lower()
         if phase in _TERMINAL_PHASES:
-        return None
+            continue
-    if _lease_expired(newest, now=now):
+        if _lease_expired(lease, now=now):
-        return None
+            continue
         if phase in _ACTIVE_PHASES or phase:
-        lease = dict(newest)
+            lease = dict(lease)
             lease["freshness"] = classify_lease_freshness(lease, now=now)
             return lease
     return None

role_session_router.py

@@ -80,7 +80,6 @@ AUTHOR_TASKS = frozenset({
 })
 
 RECONCILER_TASKS = frozenset({
-    "cleanup_merged_pr_branch",
     "reconcile_already_landed_pr",
     "reconcile_already_landed",
     "reconcile-landed-pr",
@@ -110,7 +109,6 @@ TASK_REQUIRED_ROLE = {
     "reconcile_already_landed_pr": "reconciler",
     "reconcile_already_landed": "reconciler",
     "reconcile-landed-pr": "reconciler",
-    "cleanup_merged_pr_branch": "reconciler",
     # #309: reconciler tasks close already-landed PRs/issues only.
     "reconcile_close_landed_pr": "reconciler",
     "reconcile_close_landed_issue": "reconciler",

skills/llm-project-workflow/workflows/review-merge-pr.md

@@ -936,14 +936,6 @@ Confirm:
 
 Clean only the session-owned `branches/` review worktree if the project workflow explicitly allows cleanup.
 
-Do not delete source branches from reviewer mode with raw git commands. Commands
-such as `git branch -d <branch>` and `git push <remote> --delete <branch>` are
-not proof of authorized cleanup and bypass MCP role/capability gates. Merged PR
-source branch cleanup must use an explicit MCP cleanup tool such as
-`gitea_cleanup_merged_pr_branch` or another approved cleanup helper with exact
-`gitea.branch.delete` authority, merged-PR proof, no open PR using the branch,
-target-branch ancestry proof, a `branches/` worktree path, and cleanup mutations
-reported separately from review mutations.
 Review, baseline, and merge-simulation worktrees created during this run are
 transient and are removed automatically at successful completion once they are
 clean, carry no open PR, and hold no active lease (#401). Use

task_capability_map.py

@@ -100,10 +100,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.branch.delete",
         "role": "author",
     },
-    "cleanup_merged_pr_branch": {
-        "permission": "gitea.branch.delete",
-        "role": "reconciler",
-    },
     "commit_files": {
         "permission": "gitea.repo.commit",
         "role": "author",

tests/test_branch_cleanup_guard.py

@@ -1,187 +0,0 @@
-import sys
-import unittest
-from unittest.mock import patch
-
-sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
-
-import branch_cleanup_guard as guard  # noqa: E402
-import mcp_server  # noqa: E402
-import task_capability_map  # noqa: E402
-from final_report_validator import assess_final_report_validator  # noqa: E402
-from mcp_server import gitea_cleanup_merged_pr_branch  # noqa: E402
-
-FAKE_AUTH = "token fake"
-
-
-class TestRawBranchDeleteGuard(unittest.TestCase):
-    def test_detects_local_and_remote_raw_git_delete_commands(self):
-        text = """
-        Cleanup mutations:
-        - git branch -d feat/issue-485-lease-comments-non-list-guard
-        - git push prgs --delete feat/issue-485-lease-comments-non-list-guard
-        """
-        commands = guard.raw_branch_delete_commands(text)
-        self.assertEqual(len(commands), 2)
-        self.assertIn("git branch -d", commands[0])
-        self.assertIn("git push prgs --delete", commands[1])
-
-    def test_final_report_blocks_raw_delete_as_cleanup_proof(self):
-        report = """
-        ## Controller Handoff
-        - Task: review_pr
-        - Cleanup mutations: git push prgs --delete feat/branch
-        """
-        result = assess_final_report_validator(report, "review_pr")
-        self.assertTrue(result["blocked"])
-        self.assertTrue(
-            any("shared.raw_branch_delete_bypass" in r for r in result["reasons"])
-        )
-
-    def test_cleanup_task_requires_branch_delete_capability(self):
-        self.assertEqual(
-            task_capability_map.required_permission("cleanup_merged_pr_branch"),
-            "gitea.branch.delete",
-        )
-        self.assertEqual(
-            task_capability_map.required_role("cleanup_merged_pr_branch"),
-            "reconciler",
-        )
-
-
-class TestMergedPrBranchCleanupTool(unittest.TestCase):
-    def setUp(self):
-        self._remotes = patch.dict(
-            mcp_server.REMOTES,
-            {
-                "prgs": {
-                    "host": "gitea.example.com",
-                    "org": "Example-Org",
-                    "repo": "Example-Repo",
-                }
-            },
-        )
-        self._remotes.start()
-        patch("gitea_audit.audit_enabled", return_value=False).start()
-        self.mock_api = patch("mcp_server.api_request").start()
-        self.mock_all = patch("mcp_server.api_get_all", return_value=[]).start()
-        patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
-        patch(
-            "mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref",
-            return_value=True,
-        ).start()
-
-    def tearDown(self):
-        patch.stopall()
-
-    def test_reviewer_without_delete_authority_fails_closed(self):
-        patch(
-            "mcp_server.get_profile",
-            return_value={
-                "profile_name": "reviewer",
-                "allowed_operations": ["gitea.read", "gitea.pr.review"],
-                "forbidden_operations": ["gitea.branch.delete"],
-            },
-        ).start()
-        res = gitea_cleanup_merged_pr_branch(
-            pr_number=487,
-            confirmation="CLEANUP MERGED PR 487 BRANCH feat/branch",
-            branch="feat/branch",
-            remote="prgs",
-            worktree_path="/tmp/repo/branches/cleanup",
-        )
-        self.assertFalse(res["performed"])
-        self.assertEqual(res["required_permission"], "gitea.branch.delete")
-        self.mock_api.assert_not_called()
-
-    def test_root_checkout_cleanup_fails_closed(self):
-        patch(
-            "mcp_server.get_profile",
-            return_value={
-                "profile_name": "branch-cleanup",
-                "allowed_operations": ["gitea.read", "gitea.branch.delete"],
-                "forbidden_operations": [],
-            },
-        ).start()
-        res = gitea_cleanup_merged_pr_branch(
-            pr_number=487,
-            confirmation="CLEANUP MERGED PR 487 BRANCH feat/branch",
-            branch="feat/branch",
-            remote="prgs",
-            worktree_path="/tmp/repo/Gitea-Tools",
-        )
-        self.assertFalse(res["performed"])
-        self.assertIn("root checkout", " ".join(res["reasons"]))
-        self.mock_api.assert_not_called()
-
-    def test_authorized_cleanup_deletes_through_api_path(self):
-        branch = "feat/issue-485-lease-comments-non-list-guard"
-        patch(
-            "mcp_server.get_profile",
-            return_value={
-                "profile_name": "branch-cleanup",
-                "allowed_operations": ["gitea.read", "gitea.branch.delete"],
-                "forbidden_operations": [],
-            },
-        ).start()
-        self.mock_api.side_effect = [
-            {
-                "number": 487,
-                "merged": True,
-                "merged_at": "2026-07-08T01:00:00Z",
-                "head": {"ref": branch, "sha": "a" * 40},
-                "base": {"ref": "master"},
-            },
-            {},
-            {},
-        ]
-        res = gitea_cleanup_merged_pr_branch(
-            pr_number=487,
-            confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}",
-            branch=branch,
-            remote="prgs",
-            worktree_path="/tmp/repo/branches/cleanup",
-        )
-        self.assertTrue(res["performed"])
-        delete_calls = [
-            call for call in self.mock_api.call_args_list if call.args[0] == "DELETE"
-        ]
-        self.assertEqual(len(delete_calls), 1)
-        self.assertIn("branches/feat%2Fissue-485", delete_calls[0].args[1])
-
-    def test_confirmation_required_before_delete(self):
-        branch = "feat/issue-485-lease-comments-non-list-guard"
-        patch(
-            "mcp_server.get_profile",
-            return_value={
-                "profile_name": "branch-cleanup",
-                "allowed_operations": ["gitea.read", "gitea.branch.delete"],
-                "forbidden_operations": [],
-            },
-        ).start()
-        self.mock_api.side_effect = [
-            {
-                "number": 487,
-                "merged": True,
-                "merged_at": "2026-07-08T01:00:00Z",
-                "head": {"ref": branch, "sha": "a" * 40},
-                "base": {"ref": "master"},
-            },
-            {},
-        ]
-        res = gitea_cleanup_merged_pr_branch(
-            pr_number=487,
-            confirmation="wrong",
-            branch=branch,
-            remote="prgs",
-            worktree_path="/tmp/repo/branches/cleanup",
-        )
-        self.assertFalse(res["performed"])
-        self.assertIn("confirmation", " ".join(res["reasons"]))
-        delete_calls = [
-            call for call in self.mock_api.call_args_list if call.args[0] == "DELETE"
-        ]
-        self.assertFalse(delete_calls)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_reviewer_pr_lease.py

@@ -98,45 +98,6 @@ class TestReviewerLeaseFreshness(unittest.TestCase):
         )
 
 
-class TestReviewerLeaseNewestWins(unittest.TestCase):
-    """#577: terminal release must supersede older claimed markers."""
-
-    def test_released_after_claimed_has_no_active_lease(self):
-        claim = _lease_comment(516, "session-a", phase="claimed")
-        claim["id"] = 1
-        release = _lease_comment(516, "session-a", phase="released")
-        release["id"] = 2
-        active = leases.find_active_reviewer_lease(
-            [claim, release], pr_number=516
-        )
-        self.assertIsNone(active)
-
-    def test_new_claim_after_release_is_active(self):
-        claim = _lease_comment(516, "session-a", phase="claimed")
-        claim["id"] = 1
-        release = _lease_comment(516, "session-a", phase="released")
-        release["id"] = 2
-        reclaim = _lease_comment(516, "session-b", phase="claimed")
-        reclaim["id"] = 3
-        active = leases.find_active_reviewer_lease(
-            [claim, release, reclaim], pr_number=516
-        )
-        self.assertIsNotNone(active)
-        self.assertEqual(active.get("session_id"), "session-b")
-        self.assertEqual(active.get("phase"), "claimed")
-        self.assertEqual(active.get("comment_id"), 3)
-
-    def test_done_after_claimed_has_no_active_lease(self):
-        claim = _lease_comment(516, "session-a", phase="claimed")
-        claim["id"] = 10
-        done = _lease_comment(516, "session-a", phase="done")
-        done["id"] = 11
-        active = leases.find_active_reviewer_lease(
-            [claim, done], pr_number=516
-        )
-        self.assertIsNone(active)
-
-
 class TestReviewerLeaseMutationGate(unittest.TestCase):
     def setUp(self):
         leases.clear_session_lease()

tests/test_webui_gated_actions.py

@@ -1,128 +0,0 @@
-"""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,11 +61,6 @@ 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)
@@ -77,8 +72,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", "/actions")
+        paths = ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases")
-        hrefs = ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases", "/actions")
+        hrefs = ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases")
         for path in paths:
             with self.subTest(path=path):
                 text = self.client.get(path).text

webui/app.py

@@ -16,8 +16,6 @@ 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
@@ -54,7 +52,6 @@ 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))
@@ -203,41 +200,6 @@ 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":
@@ -272,18 +234,6 @@ 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

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

@@ -1,201 +0,0 @@
-"""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,7 +11,6 @@ NAV_ITEMS = (
     ("/audit", "Audit"),
     ("/worktrees", "Worktrees"),
     ("/leases", "Leases"),
-    ("/actions", "Actions"),
 )
 
 MVP_NOTICE = (