Merge remote-tracking branch 'prgs/master' into feat/issue-435-auth-deployment

# 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/llm-workflow-runbooks.md

@@ -661,21 +661,6 @@ 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,9 +56,8 @@ 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 |
-| `/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` | Stub — lease visibility (#433) |
+| `/worktrees` | Stub — hygiene dashboard (#432) |
 | `/leases` | Lease and collision visibility (#433) |
 | `/api/leases` | JSON lease/collision export |
 
@@ -105,14 +104,11 @@ 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)
+## Deployment boundary (#435)
 
-`/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.
+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.
 
 ## Worktree hygiene (#432)
 
@@ -123,7 +119,6 @@ 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,
@@ -141,18 +136,10 @@ 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_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_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_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,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
 
@@ -8050,44 +7906,6 @@ 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,
@@ -9594,50 +9412,6 @@ 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,9 +8,6 @@ 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
@@ -369,161 +366,4 @@ 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,28 +190,18 @@ 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*.
-
-    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.
-    """
+    """Newest non-terminal, unexpired lease for *pr_number*."""
     now = now or datetime.now(timezone.utc)
-    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
+    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
     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",
@@ -408,4 +406,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,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_native_mcp_preference.py

@@ -1,7 +1,6 @@
 """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))
@@ -27,90 +26,6 @@ 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()
@@ -203,4 +118,4 @@ class TestNativeMcpPreferenceGate(unittest.TestCase):
 
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()

tests/test_resolve_task_capability.py

@@ -96,14 +96,6 @@ 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):
@@ -118,39 +110,6 @@ 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,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

@@ -63,11 +63,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)
@@ -79,8 +74,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")
-        hrefs = ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases", "/actions")
+        paths = ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases")
+        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

@@ -17,8 +17,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
@@ -55,7 +53,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))
@@ -206,41 +203,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":
@@ -278,18 +240,6 @@ 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

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

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 issue_lock_provenance import ISSUE_LOCK_FILE
 from merged_cleanup_reconcile import read_issue_lock
+from issue_lock_provenance import ISSUE_LOCK_FILE
 
 from webui.project_registry import ProjectRecord, load_registry
 from webui.queue_loader import _extract_linked_issue, _fetch_issues, _fetch_prs