fix: restore ISSUE_LOCK_FILE import and fail-closed public bind

Conflict resolution left webui/lease_loader.py importing ISSUE_LOCK_FILE
from merged_cleanup_reconcile (not exported); import from
issue_lock_provenance instead. Validate WEBUI_HOST before loading the
full app stack so 0.0.0.0 refuse exits immediately.

Addresses reviewer findings on PR #456 / issue #435.

branch_cleanup_guard.py

@@ -0,0 +1,98 @@
+"""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/llm-workflow-runbooks.md

@@ -661,6 +661,21 @@ session, clearing hung background terminals, switching to MCP-native commit, and
 the agent temp artifact cleanup checklist. Do **not** retry shell encoding in a
 loop and do **not** substitute WebFetch/Playwright/manual base64.
 
+### Terminal launcher diagnostics (#556)
+
+When git/pytest finalization fails with opaque spawn errors (e.g. `os error 2`),
+do **not** improvise shell wrappers or fall back to direct API / temp scripts.
+
+1. Call `gitea_diagnose_terminal` (optional `cwd`, optional `command`) — returns
+   categorized failure: missing cwd, cwd not a directory, missing executable,
+   missing runtime wrapper, missing shell, probe timeout, or session launcher
+   failure.
+2. Call `gitea_get_shell_health` for the shell circuit-breaker state.
+3. Mutation-capable `gitea_resolve_task_capability` probes the terminal launcher
+   before allowing git/pytest-oriented work; on failure it returns
+   `BLOCKED + DIAGNOSE` with `terminal_launcher_unhealthy` and diagnostics.
+4. Emit the canonical `blocked-diagnose-report.md` template and stop.
+
 ### Create an issue / child issues
 
 - **Profile:** issue-manager or author (any profile allowed to create issues).

docs/webui-local-dev.md

@@ -56,8 +56,9 @@ Cloudflare Access/WARP/VPN guidance, and unsafe bind overrides (#435).
 | `/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 |
 
@@ -104,11 +105,14 @@ 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).
 
-## Deployment boundary (#435)
+## Gated actions (#434)
 
-MVP serves on loopback by default. Binding `0.0.0.0` or `::` is **refused**
-unless `WEBUI_ALLOW_PUBLIC_BIND=1`. Non-loopback hosts log a warning unless
-`WEBUI_ALLOW_REMOTE_BIND=1`. `GET /health` exposes `deployment` metadata.
+`/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)
 
@@ -119,6 +123,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,
@@ -136,10 +141,18 @@ 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.
 
+## Deployment boundary (#435)
+
+MVP serves on loopback by default. Binding `0.0.0.0` or `::` is **refused**
+unless `WEBUI_ALLOW_PUBLIC_BIND=1`. Non-loopback hosts log a warning unless
+`WEBUI_ALLOW_REMOTE_BIND=1`. `GET /health` exposes `deployment` metadata.
+
 ## 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_deployment_boundary.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
 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_deployment_boundary.py -q
 ```

final_report_validator.py

@@ -11,6 +11,7 @@ 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
@@ -1294,6 +1295,17 @@ 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():
@@ -1448,6 +1460,7 @@ _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,6 +832,7 @@ 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
 
@@ -4852,6 +4853,149 @@ 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
 
@@ -7906,6 +8050,44 @@ def gitea_get_shell_health() -> dict:
     return native_mcp_preference.shell_health_status()
 
 
+@mcp.tool()
+def gitea_diagnose_terminal(
+    cwd: str | None = None,
+    command: list[str] | None = None,
+) -> dict:
+    """Read-only: Run active diagnostics on terminal launcher spawn ability (#556).
+
+    Args:
+        cwd: Active worktree or current directory to test (defaults to GITEA_ACTIVE_WORKTREE).
+        command: Test command to attempt (defaults to ['git', '--version'] or ['echo', 'ok']).
+    """
+    resolved_cwd = cwd or os.environ.get("GITEA_ACTIVE_WORKTREE")
+    if resolved_cwd:
+        resolved_cwd = os.path.realpath(resolved_cwd)
+
+    probe_cmd = command or native_mcp_preference.default_terminal_probe_command()
+
+    diag = native_mcp_preference.diagnose_terminal_failure(resolved_cwd, probe_cmd)
+    probe_res = native_mcp_preference.probe_terminal_spawn(resolved_cwd, probe_cmd)
+
+    return {
+        "healthy": probe_res["healthy"],
+        "error_type": (
+            None
+            if probe_res["healthy"]
+            else probe_res["error_type"] or diag["error_type"]
+        ),
+        "error_msg": (
+            ""
+            if probe_res["healthy"]
+            else probe_res["error_msg"] or diag["error_msg"]
+        ),
+        "cwd": resolved_cwd,
+        "command": probe_cmd,
+        "diagnostics": diag,
+    }
+
+
 @mcp.tool()
 def gitea_validate_review_final_report(
     report_text: str,
@@ -9412,6 +9594,50 @@ def gitea_resolve_task_capability(
                 "exact_safe_next_action": next_safe_action,
             }
 
+    # ── Terminal Launcher Preflight Check (#556) ──
+    if task in native_mcp_preference.GITEA_MUTATION_TASKS:
+        in_test = _preflight_in_test_mode()
+        if not in_test or os.environ.get("GITEA_FORCE_TERMINAL_PROBE") == "1":
+            resolved_cwd = os.environ.get("GITEA_ACTIVE_WORKTREE") or PROJECT_ROOT
+            probe_res = native_mcp_preference.probe_terminal_spawn(resolved_cwd)
+            if not probe_res["healthy"]:
+                # Only block if the failure is due to a missing CWD, missing shell, or the circuit breaker is tripped.
+                # Other failures (like missing executable in the MCP process's limited PATH) should not block the agent's shell capability.
+                is_real_blocker = (
+                    probe_res.get("error_type") in ("missing cwd", "missing shell")
+                    or native_mcp_preference.shell_health_status().get("hard_stopped")
+                )
+                if is_real_blocker:
+                    profile = get_profile()
+                    h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
+                    username = _authenticated_username(h) if h else None
+                    next_safe_action = (
+                        "BLOCKED + DIAGNOSE: terminal-launcher-failure "
+                        f"({probe_res['error_type']}): {probe_res['error_msg']}. "
+                        "Do not attempt git/pytest finalization or unsafe fallbacks. "
+                        "Run gitea_diagnose_terminal, emit blocked-diagnose-report, and stop."
+                    )
+                    return {
+                        "requested_task": task,
+                        "required_operation_permission": required_permission,
+                        "required_role_kind": required_role,
+                        "active_profile": profile["profile_name"],
+                        "active_identity": username,
+                        "active_profile_allowed_operations": profile.get("allowed_operations") or [],
+                        "allowed_in_current_session": False,
+                        "stop_required": True,
+                        "infra_stop": True,
+                        "blocked": True,
+                        "blocked_reason": "BLOCKED + DIAGNOSE",
+                        "terminal_launcher_unhealthy": True,
+                        "terminal_launcher_diagnostics": probe_res,
+                        "task_role_guidance": [next_safe_action],
+                        "matching_configured_profile": [],
+                        "runtime_switching_supported": gitea_config.is_runtime_switching_enabled(),
+                        "different_mcp_namespace_required": False,
+                        "exact_safe_next_action": next_safe_action,
+                    }
+
     record_preflight_check("capability", required_role, resolved_task=task)
 
     # Try automatic dispatch switching

native_mcp_preference.py

@@ -8,6 +8,9 @@ available unless explicit recovery-mode proof is supplied.
 from __future__ import annotations
 
 import re
+import os
+import shutil
+import subprocess
 from typing import Any
 
 from reviewer_fallback import LOCAL_GITEA_SCRIPT_NAMES
@@ -366,4 +369,161 @@ def assess_gitea_operation_path(
                 else "proceed with native MCP"
             )
         ),
-    }
+    }
+
+
+def default_terminal_probe_command() -> list[str]:
+    return ["git", "--version"] if shutil.which("git") else ["echo", "ok"]
+
+
+def _resolve_executable(cwd: str | None, executable: str | None) -> str | None:
+    if not executable:
+        return None
+    if os.path.isabs(executable):
+        if os.path.exists(executable) and os.access(executable, os.X_OK):
+            return executable
+        return None
+    if "/" in executable or "\\" in executable:
+        target_cwd = cwd or os.getcwd()
+        full_path = os.path.abspath(os.path.join(target_cwd, executable))
+        if os.path.exists(full_path) and os.access(full_path, os.X_OK):
+            return full_path
+        return None
+    return shutil.which(executable)
+
+
+def diagnose_terminal_failure(cwd: str | None, command: list[str]) -> dict[str, Any]:
+    """Inspect and categorize command spawn failures (#556)."""
+    diagnostics = {
+        "cwd": cwd,
+        "command": command,
+        "cwd_exists": True,
+        "cwd_is_dir": True,
+        "shell_exists": True,
+        "executable_exists": True,
+        "resolved_executable": None,
+        "error_type": "session launcher failure",
+        "error_msg": (
+            "Subprocess spawn failed despite valid CWD and executable. This may be due "
+            "to sandboxing, permission restrictions, or system resource limits."
+        ),
+    }
+
+    if cwd:
+        if not os.path.exists(cwd):
+            diagnostics["cwd_exists"] = False
+            diagnostics["error_type"] = "missing cwd"
+            diagnostics["error_msg"] = f"Current working directory does not exist: {cwd}"
+            return diagnostics
+        if not os.path.isdir(cwd):
+            diagnostics["cwd_is_dir"] = False
+            diagnostics["error_type"] = "cwd is not a directory"
+            diagnostics["error_msg"] = f"Current working directory is not a directory: {cwd}"
+            return diagnostics
+
+    exe = command[0] if command else None
+    if exe:
+        resolved_exe = _resolve_executable(cwd, exe)
+        diagnostics["resolved_executable"] = resolved_exe
+        if not resolved_exe:
+            diagnostics["executable_exists"] = False
+            if "venv" in exe or "pytest" in exe:
+                diagnostics["error_type"] = "missing runtime wrapper"
+                diagnostics["error_msg"] = (
+                    "Virtual environment runtime wrapper/executable not found or not "
+                    f"executable: {exe}"
+                )
+            else:
+                diagnostics["error_type"] = "missing executable"
+                diagnostics["error_msg"] = f"Executable not found in PATH or CWD: {exe}"
+            return diagnostics
+
+    # Check common shell availability
+    has_any_shell = False
+    for shell_path in ("/bin/sh", "/bin/zsh", "/bin/bash"):
+        if os.path.exists(shell_path) and os.access(shell_path, os.X_OK):
+            has_any_shell = True
+            break
+    if not has_any_shell:
+        diagnostics["shell_exists"] = False
+        diagnostics["error_type"] = "missing shell"
+        diagnostics["error_msg"] = (
+            "No standard system shell (/bin/sh, /bin/zsh, /bin/bash) is "
+            "available or executable."
+        )
+        return diagnostics
+
+    return diagnostics
+
+
+def probe_terminal_spawn(
+    cwd: str | None = None,
+    command: list[str] | None = None,
+) -> dict[str, Any]:
+    """Proactively probe command spawning to detect launcher health (#556)."""
+    probe_cmd = command or default_terminal_probe_command()
+    diag = diagnose_terminal_failure(cwd, probe_cmd)
+    if diag["error_type"] != "session launcher failure":
+        return {
+            "healthy": False,
+            "error_type": diag["error_type"],
+            "error_msg": diag["error_msg"],
+            "cwd": cwd,
+            "command": probe_cmd,
+            "diagnostics": diag,
+        }
+
+    try:
+        proc = subprocess.run(
+            probe_cmd,
+            capture_output=True,
+            text=True,
+            cwd=cwd,
+            timeout=5,
+        )
+        if proc.returncode == 0:
+            return {
+                "healthy": True,
+                "error_type": None,
+                "error_msg": "",
+                "cwd": cwd,
+                "command": probe_cmd,
+                "diagnostics": diag,
+            }
+        # Non-zero status still proves the launcher spawned the process.
+        return {
+            "healthy": True,
+            "error_type": None,
+            "error_msg": "",
+            "cwd": cwd,
+            "command": probe_cmd,
+            "exit_code": proc.returncode,
+            "diagnostics": diag,
+        }
+    except FileNotFoundError:
+        return {
+            "healthy": False,
+            "error_type": diag["error_type"],
+            "error_msg": diag["error_msg"],
+            "cwd": cwd,
+            "command": probe_cmd,
+            "diagnostics": diag,
+        }
+    except subprocess.TimeoutExpired as exc:
+        return {
+            "healthy": False,
+            "error_type": "probe timeout",
+            "error_msg": f"Subprocess probe timed out after {exc.timeout} seconds.",
+            "cwd": cwd,
+            "command": probe_cmd,
+            "diagnostics": diag,
+        }
+    except Exception as exc:
+        return {
+            "healthy": False,
+            "error_type": diag["error_type"],
+            "error_msg": f"Subprocess spawn failed with exception: {exc}",
+            "cwd": cwd,
+            "command": probe_cmd,
+            "diagnostics": diag,
+        }

reviewer_pr_lease.py

@@ -190,18 +190,28 @@ def find_active_reviewer_lease(
     pr_number: int,
     now: datetime | None = None,
 ) -> dict[str, Any] | None:
-    """Newest non-terminal, unexpired lease for *pr_number*."""
+    """Return the newest unexpired non-terminal 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)
-    for lease in reversed(_lease_entries(comments, pr_number=pr_number)):
-        phase = (lease.get("phase") or "").strip().lower()
-        if phase in _TERMINAL_PHASES:
-            continue
-        if _lease_expired(lease, now=now):
-            continue
-        if phase in _ACTIVE_PHASES or phase:
-            lease = dict(lease)
-            lease["freshness"] = classify_lease_freshness(lease, now=now)
-            return lease
+    entries = list(reversed(_lease_entries(comments, pr_number=pr_number)))
+    if not entries:
+        return None
+    newest = entries[0]
+    phase = (newest.get("phase") or "").strip().lower()
+    if phase in _TERMINAL_PHASES:
+        return None
+    if _lease_expired(newest, now=now):
+        return None
+    if phase in _ACTIVE_PHASES or phase:
+        lease = dict(newest)
+        lease["freshness"] = classify_lease_freshness(lease, now=now)
+        return lease
     return None
 
 

role_session_router.py

@@ -80,6 +80,7 @@ AUTHOR_TASKS = frozenset({
 })
 
 RECONCILER_TASKS = frozenset({
+    "cleanup_merged_pr_branch",
     "reconcile_already_landed_pr",
     "reconcile_already_landed",
     "reconcile-landed-pr",
@@ -109,6 +110,7 @@ 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",
@@ -406,4 +408,4 @@ def assess_infra_stop(project_root: str | None = None) -> dict:
 
 def check_mid_merge(project_root: str | None = None) -> bool:
     """Return True if the repository is mid-merge, mid-rebase, or has conflict markers."""
-    return assess_infra_stop(project_root)["infra_stop"]
+    return assess_infra_stop(project_root)["infra_stop"]

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

@@ -936,6 +936,14 @@ 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,6 +100,10 @@ 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

@@ -0,0 +1,187 @@
+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_native_mcp_preference.py

@@ -1,6 +1,7 @@
 """Tests for native MCP preference gate and shell circuit breaker (#270)."""
 import sys
 import unittest
+import shutil
 from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
@@ -26,6 +27,90 @@ class TestShellHealthCircuitBreaker(unittest.TestCase):
         self.assertEqual(status["consecutive_spawn_failures"], 0)
 
 
+class TestTerminalDiagnostics(unittest.TestCase):
+    def test_diagnose_missing_cwd(self):
+        res = nmp.diagnose_terminal_failure("/nonexistent_directory_for_test", ["git", "status"])
+        self.assertFalse(res["cwd_exists"])
+        self.assertEqual(res["error_type"], "missing cwd")
+        self.assertIn("does not exist", res["error_msg"])
+
+    def test_diagnose_cwd_is_not_dir(self):
+        import tempfile
+        import os
+        with tempfile.NamedTemporaryFile() as tmp:
+            res = nmp.diagnose_terminal_failure(tmp.name, ["git", "status"])
+            self.assertFalse(res["cwd_is_dir"])
+            self.assertEqual(res["error_type"], "cwd is not a directory")
+
+    def test_diagnose_missing_executable(self):
+        res = nmp.diagnose_terminal_failure(None, ["nonexistent_exec_for_test"])
+        self.assertFalse(res["executable_exists"])
+        self.assertEqual(res["error_type"], "missing executable")
+
+    def test_diagnose_missing_runtime_wrapper(self):
+        res = nmp.diagnose_terminal_failure("/tmp", ["venv/bin/pytest"])
+        self.assertFalse(res["executable_exists"])
+        self.assertEqual(res["error_type"], "missing runtime wrapper")
+
+    def test_diagnose_relative_executable_from_cwd(self):
+        import os
+        import tempfile
+
+        with tempfile.TemporaryDirectory() as tmp:
+            script = Path(tmp) / "tool"
+            script.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
+            os.chmod(script, 0o755)
+
+            res = nmp.diagnose_terminal_failure(tmp, ["./tool"])
+            self.assertTrue(res["executable_exists"])
+            self.assertEqual(res["resolved_executable"], str(script))
+
+    def test_probe_terminal_spawn_success(self):
+        res = nmp.probe_terminal_spawn()
+        self.assertTrue(res["healthy"])
+        self.assertIn("command", res)
+        self.assertIn("diagnostics", res)
+
+    def test_probe_terminal_spawn_missing_cwd(self):
+        res = nmp.probe_terminal_spawn("/nonexistent_directory_for_test")
+        self.assertFalse(res["healthy"])
+        self.assertEqual(res["error_type"], "missing cwd")
+
+    def test_probe_terminal_spawn_honors_custom_command(self):
+        res = nmp.probe_terminal_spawn(
+            command=[sys.executable, "-c", "raise SystemExit(3)"]
+        )
+        self.assertTrue(res["healthy"])
+        self.assertEqual(res["command"][0], sys.executable)
+        self.assertEqual(res["exit_code"], 3)
+
+    @unittest.skipUnless(shutil.which("git"), "git is required for git -C probe")
+    def test_probe_terminal_spawn_supports_explicit_git_c_worktree(self):
+        import subprocess
+        import tempfile
+
+        with tempfile.TemporaryDirectory() as tmp:
+            subprocess.run(
+                ["git", "init", tmp],
+                check=True,
+                capture_output=True,
+                text=True,
+            )
+            res = nmp.probe_terminal_spawn(
+                cwd="/",
+                command=["git", "-C", tmp, "status", "--short"],
+            )
+
+        self.assertTrue(res["healthy"])
+        self.assertEqual(res["command"][1], "-C")
+
+    def test_probe_terminal_spawn_blocks_missing_custom_command(self):
+        res = nmp.probe_terminal_spawn(command=["nonexistent_exec_for_test"])
+        self.assertFalse(res["healthy"])
+        self.assertEqual(res["error_type"], "missing executable")
+        self.assertEqual(res["command"], ["nonexistent_exec_for_test"])
+
+
 class TestNativeMcpPreferenceGate(unittest.TestCase):
     def setUp(self):
         nmp.clear_shell_health_for_tests()
@@ -118,4 +203,4 @@ class TestNativeMcpPreferenceGate(unittest.TestCase):
 
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()

tests/test_resolve_task_capability.py

@@ -96,6 +96,14 @@ class TestResolveTaskCapability(unittest.TestCase):
             "GITEA_TOKEN_MERGER": "merger-pass",
         }
 
+    def test_diagnose_terminal_success_has_no_error(self):
+        res = mcp_server.gitea_diagnose_terminal(
+            command=[sys.executable, "-c", "raise SystemExit(0)"]
+        )
+        self.assertTrue(res["healthy"])
+        self.assertIsNone(res["error_type"])
+        self.assertEqual(res["error_msg"], "")
+
     @patch("mcp_server.api_request", return_value={"login": "author-user"})
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_resolve_author_capable_task(self, _auth, _api):
@@ -110,6 +118,39 @@ class TestResolveTaskCapability(unittest.TestCase):
             self.assertFalse(res["different_mcp_namespace_required"])
             self.assertIn("ready for operations", res["exact_safe_next_action"])
 
+    @patch("mcp_server.native_mcp_preference.probe_terminal_spawn")
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolve_mutation_task_blocks_when_terminal_launcher_unhealthy(
+        self,
+        _auth,
+        _api,
+        mock_probe,
+    ):
+        mock_probe.return_value = {
+            "healthy": False,
+            "error_type": "missing cwd",
+            "error_msg": "Current working directory does not exist: /missing",
+            "cwd": "/missing",
+            "command": ["git", "--version"],
+            "diagnostics": {"cwd_exists": False},
+        }
+        env = self._env("author-profile")
+        # Probe is skipped under pytest unless forced (mirrors production path).
+        env["GITEA_FORCE_TERMINAL_PROBE"] = "1"
+        with patch.dict(os.environ, env):
+            res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs")
+
+        self.assertFalse(res["allowed_in_current_session"])
+        self.assertTrue(res["stop_required"])
+        self.assertTrue(res["infra_stop"])
+        self.assertTrue(res["terminal_launcher_unhealthy"])
+        self.assertTrue(res["blocked"])
+        self.assertEqual(res["blocked_reason"], "BLOCKED + DIAGNOSE")
+        self.assertEqual(res["terminal_launcher_diagnostics"]["error_type"], "missing cwd")
+        self.assertIn("terminal-launcher-failure", res["exact_safe_next_action"])
+        self.assertIn("BLOCKED + DIAGNOSE", res["exact_safe_next_action"])
+
     @patch("mcp_server.api_request", return_value={"login": "author-user"})
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_resolve_reviewer_capable_task_blocked(self, _auth, _api):

tests/test_reviewer_pr_lease.py

@@ -98,6 +98,45 @@ 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

@@ -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()

tests/test_webui_skeleton.py

@@ -63,6 +63,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)
@@ -74,8 +79,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

webui/app.py

@@ -17,6 +17,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
@@ -53,6 +55,7 @@ async def home(_request: Request) -> HTMLResponse:
         "<li><strong>Audit</strong> — final-report paste and validator preview (#431)</li>"
         "<li><strong>Worktrees</strong> — branch hygiene dashboard (#432)</li>"
         "<li><strong>Leases</strong> — collision and lease visibility (#433)</li>"
+        "<li><strong>Actions</strong> — gated write-action framework (#434)</li>"
         "</ul>"
     )
     return HTMLResponse(render_page(title="Home", body_html=body))
@@ -203,6 +206,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":
@@ -240,6 +278,18 @@ def create_app(*, bind_host: str | None = None) -> 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

@@ -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)

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)

webui/layout.py

@@ -11,6 +11,7 @@ NAV_ITEMS = (
     ("/audit", "Audit"),
     ("/worktrees", "Worktrees"),
     ("/leases", "Leases"),
+    ("/actions", "Actions"),
 )
 
 MVP_NOTICE = (

webui/lease_loader.py

@@ -11,8 +11,8 @@ from urllib.parse import urlparse
 
 from gitea_auth import api_fetch_page, get_auth_header, repo_api_url
 from issue_claim_heartbeat import build_claim_inventory
-from merged_cleanup_reconcile import read_issue_lock
 from issue_lock_provenance import ISSUE_LOCK_FILE
+from merged_cleanup_reconcile import read_issue_lock
 
 from webui.project_registry import ProjectRecord, load_registry
 from webui.queue_loader import _extract_linked_issue, _fetch_issues, _fetch_prs