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",
-    }

controller_closure_baseline_proof.py

@@ -1,63 +0,0 @@
-"""Controller-closure baseline-proof gate (#529, criterion 6).
-
-A controller (or any session) may close a tracking issue with a closure
-report that summarizes validation. When that report calls a non-zero
-test-suite exit an "expected pre-existing failure" (or baseline / known
-failure) it must carry pre-merge proof; otherwise the closure buries an
-unproven regression as an accepted baseline and weakens controller
-confidence in the durable state.
-
-This module fails closed on exactly that case by reusing the pre-merge
-baseline proof verifier (#533). A closure report is allowed when it is
-empty (no validation claim), a clean pass, or a baseline failure proven on
-the PR pre-merge base commit (or a documented known-failure record that
-predates the PR).
-"""
-
-from __future__ import annotations
-
-from typing import Any
-
-from premerge_baseline_proof import CLEAN_PASS, assess_premerge_baseline_proof
-
-
-def assess_controller_closure_baseline_proof(
-    closure_report: str | None,
-) -> dict[str, Any]:
-    """Assess whether an issue-closure report may proceed.
-
-    Returns a dict with ``block``, ``proven``, ``label``, ``reasons``,
-    ``skipped`` and ``safe_next_action``. Only blocks when the closure
-    report claims a non-zero validation exit is an expected
-    pre-existing/baseline failure without valid pre-merge proof.
-    """
-    text = (closure_report or "").strip()
-    if not text:
-        return {
-            "block": False,
-            "proven": True,
-            "label": CLEAN_PASS,
-            "reasons": [],
-            "skipped": True,
-            "safe_next_action": "",
-        }
-
-    result = assess_premerge_baseline_proof(text)
-    block = bool(result.get("block"))
-    return {
-        "block": block,
-        "proven": not block,
-        "label": result.get("label"),
-        "reasons": list(result.get("reasons") or []),
-        "skipped": bool(result.get("skipped", False)),
-        "safe_next_action": (
-            result.get("safe_next_action")
-            or (
-                "provide pre-merge baseline proof (base commit, tested commit, "
-                "command, exit status, failure signature) before closing on an "
-                "expected pre-existing failure"
-            )
-        )
-        if block
-        else "",
-    }

docs/developer-testing-guidelines.md

@@ -54,25 +54,6 @@ Use `-q` for a compact summary and `-v` to see individual test names.
 ./venv/bin/python -m pytest tests/ -q
 ```
 
-### Web UI suite (#436)
-
-Hermetic unittest modules matching `test_webui_*.py` cover route rendering,
-read-only guards, registry/prompt/queue loaders, and optional child-issue
-modules when present on the branch.
-
-```bash
-./scripts/test-webui
-./scripts/ci-webui-check          # skip unless the diff touches web UI paths
-WEBUI_CI_FORCE=1 ./scripts/ci-webui-check
-```
-
-`scripts/test-webui` defaults `WEBUI_TEST_OFFLINE=1` so route coverage never
-needs Gitea credentials or MCP daemon credential access. Set
-`WEBUI_TEST_OFFLINE=0` only for explicit operator live-fetch checks.
-
-Wire `scripts/ci-webui-check` into Jenkins (or equivalent) for PRs that touch
-`webui/`, `tests/test_webui_*`, or `docs/webui*`.
-
 ### Run targeted tests
 
 ```bash

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).
@@ -811,32 +796,6 @@ author work. Do not switch from `prgs-reconciler` to `prgs-author` only to run
 - **Reports:** label cleanup actions as reconciler cleanup (not author mutation).
 - **Prompt:** `As prgs-reconciler, dry-run then execute gitea_reconcile_merged_cleanups for recently merged PRs without switching to prgs-author.`
 
-### Superseded PR / satisfied issue reconciliation (#525)
-
-Closing duplicate or superseded PRs after a canonical PR has merged is
-**reconciler** work. Do not switch to `prgs-author` only to close the duplicate
-PR, close the satisfied issue, or file a narrow follow-up discovered during
-reconciliation.
-
-- **Profile:** `prgs-reconciler`.
-- **Tasks:** `reconcile_close_superseded_pr`,
-  `reconcile_close_satisfied_issue`, and
-  `reconcile_create_followup_issue`.
-- **Tool:** `gitea_reconcile_superseded_by_merged_pr`.
-- **Required proof:**
-  1. Live target PR state is open, but it is not independently required and no
-     mergeable work remains.
-  2. Live superseding PR state is closed and merged.
-  3. Superseding PR head is an ancestor of freshly fetched `master`.
-  4. Merge commit SHA is recorded.
-  5. Canonical close comment cites the superseding PR and merge commit.
-  6. Linked issue closure is attempted only when the issue is open and
-     explicitly satisfied by the superseding PR.
-- **Fail closed:** missing ancestry proof, missing canonical comment,
-  mergeable target PR, same target/superseding PR, or missing
-  `gitea.pr.close` / `gitea.issue.close` capability.
-- **Prompt:** `As prgs-reconciler, reconcile PR #N as superseded by merged PR #M; post the canonical close comment, close the superseded PR, and close issue #K only if the merged PR fully satisfies it.`
-
 ### Stop on blocker
 
 - **Any profile.** If a required gate cannot be satisfied — identity
@@ -1049,72 +1008,6 @@ Special states:
 Final reports must not claim a comment was posted when
 `canonical_comment_validation.allowed` is false (#496 AC14).
 
-## Stale #332 review-decision lock cleanup (#594)
-
-#332 hard-stops a reviewer session after a terminal live review mutation
-(`approve` / `request_changes`). After #559 those locks are **durable** under
-`~/.cache/gitea-tools/session-state/review_decision_lock-<profile>.json`, so they
-can outlive the PR they protected.
-
-### When cleanup is **allowed**
-
-Use `gitea_cleanup_stale_review_decision_lock` from a **reviewer** profile when:
-
-1. A durable review-decision lock has a terminal live mutation, **and**
-2. Live Gitea state shows that terminal PR is already **merged** or **closed**
-   (so no same-PR merge sequence remains), **and**
-3. The active profile identity matches the lock, **and**
-4. Authenticated identity can be verified.
-
-Workflow:
-
-```text
-# 1) Assess only (default)
-gitea_cleanup_stale_review_decision_lock(apply=false, remote=prgs, ...)
-
-# 2) Apply after is_moot=true and cleanup_allowed=true
-gitea_cleanup_stale_review_decision_lock(
-  apply=true,
-  expected_terminal_pr=<merged-or-closed-pr>,
-  remote=prgs,
-  ...
-)
-```
-
-Successful apply:
-
-* clears in-memory + durable decision lock for that profile
-* returns a structured `audit` payload
-* posts an audit comment on the mooted PR when `gitea.pr.comment` is allowed
-  (`post_audit_comment=true` default)
-
-Same-profile auto-expire: if `gitea_merge_pr` succeeds on a PR that this same
-profile just approved, the decision lock is cleared after merge. Cross-profile
-locks (reviewer vs merger) still require the cleanup tool.
-
-### When cleanup is **forbidden**
-
-Do **not** clear when:
-
-* the last terminal PR is still **open** (active #332 hard-stop — continue merge
-  for that approve, or stop after request_changes)
-* live PR state cannot be fetched (fail closed)
-* profile identity does not match the lock
-* identity cannot be verified
-* `expected_terminal_pr` does not match the last terminal PR
-
-**Never** use manual deletion of session-state files as the normal workflow.
-**Never** use cleanup to skip review of an open PR or to bypass eligibility /
-mark-ready / submit gates on active work.
-
-### Related tools
-
-| Tool | Purpose |
-|------|---------|
-| `gitea_cleanup_stale_review_decision_lock` | Moot decision-lock cleanup (#594) |
-| `gitea_authorize_review_correction` | Operator-approved correction after a **mistaken** live review on still-active work (#211) |
-| `gitea_cleanup_post_merge_moot_lease` | Moot **lease** cleanup after merge (#515) — different object |
-
 ## Fail-closed behavior
 
 Before any mutating action the workflow verifies identity, active profile,

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,68 +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_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
-
-## Tests (#436)
-
-Run the full hermetic web UI suite (all `test_webui_*.py` modules):
-
-```bash
-./scripts/test-webui
-```
-
-CI / Jenkins multibranch can call the path-filtered gate (runs only when the
-diff touches `webui/`, `tests/test_webui_*`, or web UI docs/scripts):
-
-```bash
-./scripts/ci-webui-check
-WEBUI_CI_FORCE=1 ./scripts/ci-webui-check   # always run
-```
-
-`scripts/test-webui` sets `WEBUI_TEST_OFFLINE=1` by default. In that mode the
-queue, lease, and runtime routes use empty offline snapshots instead of Gitea
-credentials, so CI can run without MCP daemon credential access. Set
-`WEBUI_TEST_OFFLINE=0` only when deliberately validating live fetch behavior.
-
-Or invoke unittest directly:
-
-```bash
-python3 -m unittest discover -s tests -p 'test_webui_*.py' -q
-```
-
-## Lease visibility (#433)
-
-`/leases` surfaces read-only lease and collision state: local issue lock file,
-in-progress claim inventory (#268), reviewer PR lease comments when present
-(`<!-- mcp-review-lease:v1 -->`, #407), duplicate open PRs per issue (#400),
-and duplicate local branches per issue. Links to collision-history backend
-issues (#267, #268, #400, #407) are included. No lease acquire/release from UI.
-
-## Runtime health (#430)
-
-`/runtime` surfaces read-only MCP/runtime diagnostics for the default registry
-project: active profile and role kind, authenticated identity (when credentials
-are available), config model/mode, local vs remote `master` SHA sync, shell
-health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
-checkout is behind merged safety-gate changes. Restart guidance links to #420;
-no tokens or MCP restart actions are exposed.
-
-## Tests
-
-```bash
 pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_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
-```
+```

docs/wiki/Safety-and-Gates.md

@@ -15,7 +15,5 @@
 5. **Explicit merge confirmation** — `gitea_merge_pr` requires `confirmation="MERGE PR <n>"`.
 6. **No self-review / self-merge** — Authenticated user must differ from PR author for approve/merge.
 7. **Review decision lock** — Live review mutations require validation-phase dry-run and `gitea_mark_final_review_decision`.
-8. **Terminal review hard-stop (#332)** — After a terminal live review mutation, only same-PR merge after `approve` may continue. Durable locks (#559) must not be deleted by hand.
-9. **Stale decision-lock cleanup (#594)** — `gitea_cleanup_stale_review_decision_lock` may clear a durable #332 lock **only** when the last terminal mutation's PR is live-state **merged or closed**, identity/profile gates pass, and apply uses a reviewer-capable profile. Open/ambiguous locks stay fail-closed. Successful cleanup records a durable audit trail.
-10. **Redaction** — Tokens, passwords, and keychain material never appear in tool output.
-11. **Wiki publication (#224)** — `docs/wiki/` and the sync helper are prerequisites only. Closing a wiki issue requires live Gitea Wiki proof on the repo Wiki tab. See [Runbooks](Runbooks.md#wiki-publication-readiness-gate-224).
+8. **Redaction** — Tokens, passwords, and keychain material never appear in tool output.
+9. **Wiki publication (#224)** — `docs/wiki/` and the sync helper are prerequisites only. Closing a wiki issue requires live Gitea Wiki proof on the repo Wiki tab. See [Runbooks](Runbooks.md#wiki-publication-readiness-gate-224).

final_report_validator.py

@@ -11,10 +11,8 @@ import inspect
 import re
 from typing import Any, Callable
 
-import branch_cleanup_guard
 import issue_acceptance_gate
 import issue_lock_provenance
-import merger_lease_adoption
 import reviewer_handoff_consistency
 import thread_state_ledger_validator
 from mcp_native_cleanup_proof import assess_mcp_native_cleanup_proof
@@ -40,7 +38,6 @@ FINAL_REPORT_TASK_KINDS = frozenset({
     "issue_filing",
     "issue_selection",
     "inventory",
-    "controller_close",
 })
 
 _TASK_KIND_ALIASES = {
@@ -52,9 +49,6 @@ _TASK_KIND_ALIASES = {
     "reconcile_already_landed": "reconcile_already_landed",
     "work-issue": "work_issue",
     "create_issue": "issue_filing",
-    "close_issue": "controller_close",
-    "close-issue": "controller_close",
-    "controller-close": "controller_close",
 }
 
 _HANDOFF_ROLE_BY_TASK = {
@@ -66,7 +60,6 @@ _HANDOFF_ROLE_BY_TASK = {
     "issue_filing": "issue_filing",
     "issue_selection": None,
     "inventory": "inventory",
-    "controller_close": None,
 }
 
 _LEGACY_WORKSPACE_MUTATIONS_RE = re.compile(
@@ -855,27 +848,6 @@ def _rule_reviewer_premerge_baseline_proof(report_text: str) -> list[dict[str, s
     )
 
 
-def _rule_reviewer_post_merge_validation(report_text: str) -> list[dict[str, str]]:
-    """Block active approval on an already-merged/closed PR, and post-merge moot
-    validation claimed without merged-state + merge-commit proof (#529)."""
-    from post_merge_validation import assess_post_merge_validation
-
-    result = assess_post_merge_validation(report_text)
-    if not result.get("block"):
-        return []
-    return _findings_from_reasons(
-        "reviewer.post_merge_validation",
-        result.get("reasons") or [],
-        field="Validation status",
-        severity="block",
-        safe_next_action=result.get("safe_next_action")
-        or (
-            "record post-merge moot validation instead of an active approval; "
-            "cite PR state merged/closed and the merge commit SHA"
-        ),
-    )
-
-
 def _rule_reviewer_validation_status_vocabulary(
     report_text: str,
     *,
@@ -1290,24 +1262,6 @@ def _rule_shared_manual_lock_pr_override(report_text: str) -> list[dict[str, str
     )
 
 
-def _rule_shared_manual_lease_proof_handoff(report_text: str) -> list[dict[str, str]]:
-    """Block merge/review handoffs that claim unsafe lease seeding (#535)."""
-    result = merger_lease_adoption.assess_manual_lease_proof_handoff(report_text)
-    if result.get("proven"):
-        return []
-    return _findings_from_reasons(
-        "shared.manual_lease_proof_handoff",
-        result.get("reasons") or [],
-        field="Merge mutations",
-        severity="block",
-        safe_next_action=(
-            "use gitea_adopt_merger_pr_lease or gitea_acquire_reviewer_pr_lease; "
-            "cite lease_proof_source / adoption_comment_id; never claim manual "
-            "seeded/injected lease proof"
-        ),
-    )
-
-
 def _rule_shared_issue_acceptance_gate(report_text: str) -> list[dict[str, str]]:
     result = issue_acceptance_gate.validate_final_report_issue_acceptance(report_text)
     if not result.get("applicable") or result.get("valid"):
@@ -1340,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():
@@ -1504,9 +1447,7 @@ def _rule_shared_mcp_native_cleanup_proof(report_text: str) -> list[dict[str, st
 _SHARED_ISSUE_LOCK_RULES = (
     _rule_shared_issue_lock_external_state,
     _rule_shared_manual_lock_pr_override,
-    _rule_shared_manual_lease_proof_handoff,
     _rule_shared_author_reviewer_same_run,
-    _rule_shared_raw_branch_delete_bypass,
     _rule_shared_canonical_state_update,
 )
 
@@ -1540,7 +1481,6 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_linked_issue,
         _rule_reviewer_baseline_on_failure,
         _rule_reviewer_premerge_baseline_proof,
-        _rule_reviewer_post_merge_validation,
         _rule_reviewer_validation_status_vocabulary,
         _rule_reviewer_main_checkout_baseline,
         _rule_reviewer_main_checkout_path,
@@ -1633,13 +1573,6 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         *_SHARED_CANONICAL_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
     ],
-    # Controller issue closure (#529): a closure report must not bury an
-    # unproven non-zero suite exit as an "expected pre-existing failure".
-    # Kept intentionally narrow so a closure pre-check does not demand the
-    # full reviewer/author handoff schema.
-    "controller_close": [
-        _rule_reviewer_premerge_baseline_proof,
-    ],
 }
 
 

issue_content_gate.py

@@ -1,180 +0,0 @@
-"""Pre-create issue content gate (#582).
-
-Blocks issue creation when the title/body is a *vague reference* to
-out-of-band content the LLM cannot prove it has ("the drafted issue",
-"the prepared issue", "the issue we discussed", "the previous draft")
-with no durable source pointer.
-
-The rule from #582: an LLM must not fabricate an issue from stale chat
-memory. When the requested issue content is a vague reference and no
-durable source pointer (existing issue/PR/comment, checked-in file, URL,
-or scratchpad path) is present, the gate fails closed and the caller must
-return BLOCKED + DIAGNOSE naming the exact vague/missing fields.
-
-This gate intentionally does NOT require a non-empty body: title-only
-issue creation remains allowed for callers that explicitly want it. It
-only blocks content that is a placeholder reference to something the
-current context does not contain.
-"""
-
-from __future__ import annotations
-
-import re
-import unicodedata
-
-# Vague reference phrases that point at out-of-band draft content. Stored
-# normalized (lowercase, punctuation-stripped, whitespace-collapsed) so they
-# can be matched against normalized field text.
-VAGUE_REFERENCE_PHRASES: tuple[str, ...] = (
-    "the drafted issue",
-    "drafted issue",
-    "the prepared issue",
-    "prepared issue",
-    "the issue we discussed",
-    "the issue we talked about",
-    "the one we discussed",
-    "the previous draft",
-    "previous draft",
-    "the draft above",
-    "the issue above",
-    "as discussed",
-    "as we discussed",
-    "as previously discussed",
-    "per our discussion",
-    "per our conversation",
-    "per our chat",
-    "see above",
-    "same as before",
-    "the issue from earlier",
-    "the issue i drafted",
-    "the issue you drafted",
-)
-
-# A field that only restates a vague phrase (optionally with a leading verb
-# such as "create"/"add"/"open"/"file") is still a vague reference.
-_LEADING_VERBS = ("create", "add", "open", "file", "make", "please", "the")
-
-# Durable source pointers: if any of these appears, the content points at a
-# retrievable source of truth and is NOT treated as a fabricated reference.
-_DURABLE_SOURCE_REGEXES: tuple[re.Pattern[str], ...] = (
-    re.compile(r"#\d+"),                                   # #402
-    re.compile(r"\b(?:issue|pull|pr)\s+#?\d+", re.I),      # issue 402 / PR #17
-    re.compile(r"\bcomment\s+#?\d+", re.I),                # comment 8155
-    re.compile(r"https?://\S+", re.I),                     # URL
-    re.compile(                                            # checked-in file
-        r"[\w./-]+\.(?:py|md|json|txt|sh|ya?ml|toml|cfg|ini|rst)\b", re.I
-    ),
-    re.compile(r"\bscratchpad\b", re.I),                   # scratchpad path
-    re.compile(r"(?:^|\s)/[\w./-]+"),                      # absolute path
-)
-
-
-def normalize_text(text: str) -> str:
-    """Lowercase, punctuation-stripped, whitespace-collapsed text."""
-    norm = unicodedata.normalize("NFKC", (text or "").strip().lower())
-    norm = re.sub(r"[^\w\s]", " ", norm)
-    norm = re.sub(r"\s+", " ", norm).strip()
-    return norm
-
-
-def has_durable_source_pointer(text: str) -> bool:
-    """True when the raw text references a durable, retrievable source."""
-    raw = text or ""
-    return any(rx.search(raw) for rx in _DURABLE_SOURCE_REGEXES)
-
-
-def find_vague_reference(text: str) -> str | None:
-    """Return the vague phrase a field is dominated by, else ``None``.
-
-    A field is a vague reference only when it contains a known vague phrase,
-    carries no durable source pointer, and the phrase dominates the field
-    (the field is essentially just the phrase). Long, specific bodies that
-    merely contain "as discussed" in passing are not blocked.
-    """
-    if has_durable_source_pointer(text):
-        return None
-    norm = normalize_text(text)
-    if not norm:
-        return None
-    stripped = norm
-    for verb in _LEADING_VERBS:
-        if stripped.startswith(verb + " "):
-            stripped = stripped[len(verb) + 1:]
-    for phrase in VAGUE_REFERENCE_PHRASES:
-        if phrase in norm and len(stripped) <= len(phrase) + 12:
-            return phrase
-    return None
-
-
-def assess_issue_content(
-    title: str,
-    body: str = "",
-    *,
-    allow_incomplete: bool = False,
-) -> dict:
-    """Evaluate whether proposed issue content is durable enough to create.
-
-    Returns a dict with ``complete`` (bool), ``missing_fields``,
-    ``vague_fields``, ``reasons``, ``diagnose`` (joined reasons), and
-    ``has_durable_source_pointer``. ``allow_incomplete`` is an explicit
-    operator override that forces ``complete`` True.
-    """
-    t = (title or "").strip()
-    b = (body or "").strip()
-    missing_fields: list[str] = []
-    vague_fields: list[str] = []
-    reasons: list[str] = []
-
-    if not t:
-        missing_fields.append("title")
-        reasons.append("issue title is required")
-    else:
-        phrase = find_vague_reference(t)
-        if phrase:
-            vague_fields.append("title")
-            reasons.append(
-                f"title is a vague reference ('{phrase}') with no durable "
-                "content or source pointer"
-            )
-
-    if b:
-        phrase = find_vague_reference(b)
-        if phrase:
-            vague_fields.append("body")
-            reasons.append(
-                f"body is a vague reference ('{phrase}') with no durable "
-                "content or source pointer"
-            )
-
-    complete = allow_incomplete or (not missing_fields and not vague_fields)
-    return {
-        "complete": complete,
-        "missing_fields": missing_fields,
-        "vague_fields": vague_fields,
-        "reasons": reasons,
-        "diagnose": "; ".join(reasons),
-        "has_durable_source_pointer": has_durable_source_pointer(b)
-        or has_durable_source_pointer(t),
-        "override_applied": bool(
-            allow_incomplete and (missing_fields or vague_fields)
-        ),
-    }
-
-
-def pre_create_issue_content_gate(
-    title: str,
-    body: str = "",
-    *,
-    allow_incomplete: bool = False,
-) -> dict:
-    """Content gate wrapper mirroring the duplicate gate shape.
-
-    ``performed`` is True when the content is durable enough to create (or an
-    explicit override was supplied). When False, the caller must return
-    BLOCKED + DIAGNOSE and must not create the issue.
-    """
-    assessment = assess_issue_content(
-        title, body, allow_incomplete=allow_incomplete
-    )
-    assessment["performed"] = assessment["complete"]
-    return assessment

issue_workflow_labels.py

@@ -42,38 +42,12 @@ STATUS_LABEL_SPECS: tuple[LabelSpec, ...] = (
     LabelSpec("status:wontfix", "000000", "Issue will not be fixed"),
 )
 
-# Durable validation-outcome labels (#529): the four canonical distinctions a
-# reviewer report can carry. These are orthogonal to the single active status:*
-# label, so they use their own prefix and are not subject to the one-status
-# invariant.
-VALIDATION_LABEL_SPECS: tuple[LabelSpec, ...] = (
-    LabelSpec("validation:clean-pass", "0e8a16", "Validation was a clean pass"),
-    LabelSpec(
-        "validation:baseline-accepted",
-        "fbca04",
-        "Validation passed with a baseline-proven unrelated failure",
-    ),
-    LabelSpec(
-        "validation:blocked",
-        "b60205",
-        "Validation blocked by an unresolved failure",
-    ),
-    LabelSpec(
-        "validation:post-merge-moot",
-        "5319e7",
-        "Validation is post-merge moot (PR already merged/closed before review)",
-    ),
-)
-
 CANONICAL_LABEL_SPECS: tuple[LabelSpec, ...] = (
-    TYPE_LABEL_SPECS + STATUS_LABEL_SPECS + VALIDATION_LABEL_SPECS
+    TYPE_LABEL_SPECS + STATUS_LABEL_SPECS
 )
 
 TYPE_LABELS: frozenset[str] = frozenset(spec.name for spec in TYPE_LABEL_SPECS)
 STATUS_LABELS: frozenset[str] = frozenset(spec.name for spec in STATUS_LABEL_SPECS)
-VALIDATION_LABELS: frozenset[str] = frozenset(
-    spec.name for spec in VALIDATION_LABEL_SPECS
-)
 CANONICAL_LABELS: frozenset[str] = frozenset(
     spec.name for spec in CANONICAL_LABEL_SPECS
 )

master_parity_gate.py

@@ -24,26 +24,12 @@ from __future__ import annotations
 
 import os
 import subprocess
-import time
-
-# Live-remote head cache: the parity gate runs on every mutation and every
-# runtime-context read, so the ``git ls-remote`` result is cached briefly to
-# avoid a network round-trip per call (#610). Keyed by (root, remote, branch).
-_REMOTE_HEAD_CACHE: dict[tuple[str, str, str], tuple[float, str | None]] = {}
-_REMOTE_HEAD_TTL = 60.0
-
-
-def _clear_remote_head_cache() -> None:
-    """Reset the live-remote head cache (test isolation / forced refresh)."""
-    _REMOTE_HEAD_CACHE.clear()
 
 # Environment escape hatches (ops + tests):
 #   GITEA_MCP_DISABLE_PARITY_GATE  -> disable enforcement entirely (fail open).
 #   GITEA_TEST_CURRENT_HEAD        -> force the "current" HEAD read, for tests.
 ENV_DISABLE = "GITEA_MCP_DISABLE_PARITY_GATE"
 ENV_TEST_CURRENT_HEAD = "GITEA_TEST_CURRENT_HEAD"
-# GITEA_TEST_LIVE_REMOTE_HEAD -> force the live remote master read, for tests.
-ENV_TEST_LIVE_REMOTE_HEAD = "GITEA_TEST_LIVE_REMOTE_HEAD"
 
 
 def read_git_head(root: str) -> str | None:
@@ -72,57 +58,6 @@ def read_git_head(root: str) -> str | None:
     return (res.stdout or "").strip() or None
 
 
-def read_remote_master_head(
-    root: str,
-    remote: str = "origin",
-    branch: str = "master",
-    ttl: float = _REMOTE_HEAD_TTL,
-) -> str | None:
-    """Return the live remote ``branch`` commit SHA, or ``None`` (#610).
-
-    Resolves the *live* target commit via ``git ls-remote`` so parity can tell
-    a daemon that is behind the live remote master apart from one whose local
-    checkout simply hasn't been pulled. ``None`` means the live head could not
-    be resolved (offline, no such remote, git unavailable, error) -- callers
-    must treat unknown live state as *not mutation-safe* while never blocking
-    read-only diagnostics. A ``GITEA_TEST_LIVE_REMOTE_HEAD`` override takes
-    precedence so the wiring can be exercised deterministically and offline.
-
-    The result is cached for *ttl* seconds per (root, remote, branch) so the
-    gate does not run a network probe on every mutation/read (``ttl=0`` forces
-    a live probe). Both hits and ``None`` misses are cached to bound offline
-    latency; the env override bypasses the cache and the subprocess entirely.
-    """
-    forced = os.environ.get(ENV_TEST_LIVE_REMOTE_HEAD)
-    if forced is not None:
-        return forced.strip() or None
-    if not root:
-        return None
-    key = (root, remote, branch)
-    now = time.monotonic()
-    if ttl > 0:
-        cached = _REMOTE_HEAD_CACHE.get(key)
-        if cached is not None and (now - cached[0]) < ttl:
-            return cached[1]
-    sha: str | None = None
-    try:
-        res = subprocess.run(
-            ["git", "-C", root, "ls-remote", remote, f"refs/heads/{branch}"],
-            capture_output=True,
-            text=True,
-            check=False,
-            timeout=5,
-        )
-        if res.returncode == 0:
-            lines = (res.stdout or "").strip().splitlines()
-            if lines:
-                sha = lines[0].split("\t", 1)[0].split()[0].strip() or None
-    except Exception:
-        sha = None
-    _REMOTE_HEAD_CACHE[key] = (now, sha)
-    return sha
-
-
 def capture_startup_parity(root: str, head: str | None = None) -> dict:
     """Capture the process source-tree baseline once at server startup.
 
@@ -137,38 +72,18 @@ def _short(sha: str | None) -> str:
     return sha[:12] if sha else "unknown"
 
 
-def assess_master_parity(
-    startup: dict | None,
-    current_head: str | None,
-    live_remote_head: str | None = None,
-) -> dict:
+def assess_master_parity(startup: dict | None, current_head: str | None) -> dict:
     """Compare the startup baseline against the current on-disk ``HEAD``.
 
-    Pure: all HEADs are supplied by the caller. Returns a structured result:
+    Pure: both HEADs are supplied by the caller. Returns a structured result:
 
     - ``in_parity``    -- server code matches the on-disk master (or parity
       could not be determined, which is not treated as stale).
     - ``stale``        -- the on-disk master has definitively advanced past the
       running process.
-    - ``restart_required`` -- ``stale`` or ``live_stale``; the recovery action.
-    - ``determinable`` -- whether both local HEADs were known well enough to
-      compare.
+    - ``restart_required`` -- alias of ``stale``; the recovery action.
+    - ``determinable`` -- whether both HEADs were known well enough to compare.
     - ``startup_head`` / ``current_head`` / ``reasons``.
-
-    #610 adds live-remote awareness so a daemon that is stale relative to the
-    *live* remote master cannot report a mutation-safe result even when the
-    local checkout HEAD still matches the daemon's startup commit:
-
-    - ``daemon_start_head`` -- the commit the running process started at
-      (alias of ``startup_head``, named for clarity in reports).
-    - ``local_head`` -- the on-disk checkout HEAD (alias of ``current_head``).
-    - ``live_remote_head`` -- the live remote target commit, or ``None`` when it
-      could not be fetched.
-    - ``live_known`` -- whether the live remote target was resolved.
-    - ``live_stale`` -- the live remote master has advanced past the running
-      process (daemon is behind live master) even if local parity is green.
-    - ``mutation_safe`` -- the daemon code, local checkout, and live remote
-      target all agree; the only state in which a mutation may rely on parity.
     """
     startup_head = (startup or {}).get("startup_head")
     reasons: list[str] = []
@@ -176,56 +91,32 @@ def assess_master_parity(
     if startup_head is None:
         reasons.append(
             "startup commit was not captured; code parity cannot be enforced")
-        return _result(True, False, False, startup_head, current_head,
-                       live_remote_head, False, reasons)
+        return _result(True, False, False, startup_head, current_head, reasons)
 
     if current_head is None:
         reasons.append(
             "current workspace HEAD could not be read; code parity cannot be "
             "enforced")
-        return _result(True, False, False, startup_head, current_head,
-                       live_remote_head, False, reasons)
+        return _result(True, False, False, startup_head, current_head, reasons)
 
-    local_in_parity = startup_head == current_head
-    local_stale = not local_in_parity
-    if local_stale:
-        reasons.append(
-            f"MCP server started at commit {_short(startup_head)} but the "
-            f"workspace master is now {_short(current_head)}; restart the "
-            f"server to load the current capability gates")
+    if startup_head == current_head:
+        return _result(True, False, True, startup_head, current_head, reasons)
 
-    live_known = live_remote_head is not None
-    live_stale = live_known and live_remote_head != startup_head
-    if live_stale:
-        reasons.append(
-            f"live remote master is {_short(live_remote_head)} but the MCP "
-            f"server started at {_short(startup_head)}; the daemon is stale "
-            f"relative to live master -- restart/reconnect before mutating")
-
-    return _result(
-        local_in_parity, local_stale, True, startup_head, current_head,
-        live_remote_head, live_stale, reasons)
+    reasons.append(
+        f"MCP server started at commit {_short(startup_head)} but the workspace "
+        f"master is now {_short(current_head)}; restart the server to load the "
+        f"current capability gates")
+    return _result(False, True, True, startup_head, current_head, reasons)
 
 
-def _result(in_parity, stale, determinable, startup_head, current_head,
-            live_remote_head, live_stale, reasons):
-    live_known = live_remote_head is not None
-    mutation_safe = (
-        determinable and in_parity and live_known and not live_stale)
+def _result(in_parity, stale, determinable, startup_head, current_head, reasons):
     return {
         "in_parity": in_parity,
         "stale": stale,
-        "restart_required": stale or live_stale,
+        "restart_required": stale,
         "determinable": determinable,
         "startup_head": startup_head,
         "current_head": current_head,
-        # #610 distinguished signals:
-        "daemon_start_head": startup_head,
-        "local_head": current_head,
-        "live_remote_head": live_remote_head,
-        "live_known": live_known,
-        "live_stale": live_stale,
-        "mutation_safe": mutation_safe,
         "reasons": list(reasons),
     }
 
@@ -239,13 +130,11 @@ def parity_block_reasons(assessment: dict) -> list[str]:
     """Block reasons for a mutation gate (empty when the mutation may proceed).
 
     A disabled gate or an in-parity / non-determinable assessment yields no
-    reasons. A definitively stale server blocks, and (#610) a daemon that is
-    stale relative to the *live* remote master blocks even when the local
-    checkout HEAD still matches the daemon's startup commit.
+    reasons; only a definitively stale server blocks.
     """
     if gate_disabled():
         return []
-    if assessment.get("stale") or assessment.get("live_stale"):
+    if assessment.get("stale"):
         return list(assessment.get("reasons") or
                     ["server code is stale relative to master (fail closed)"])
     return []
@@ -258,10 +147,6 @@ def parity_report(assessment: dict) -> dict:
         "restart_required": True,
         "startup_head": assessment.get("startup_head"),
         "current_head": assessment.get("current_head"),
-        # #610: name the live remote target so the report distinguishes a
-        # local-code stale from a daemon-behind-live-master stale.
-        "live_remote_head": assessment.get("live_remote_head"),
-        "live_stale": bool(assessment.get("live_stale")),
         "reasons": list(assessment.get("reasons") or []),
         "recovery": [
             "The running MCP server is executing code older than the current "
@@ -272,45 +157,6 @@ def parity_report(assessment: dict) -> dict:
     }
 
 
-def parity_resolver_disagreement(
-    assessment: dict,
-    resolver_restart_required: bool,
-) -> dict | None:
-    """Typed blocker when the resolver requires restart but parity looks green.
-
-    The capability resolver (``gitea_resolve_task_capability``) detects stale
-    runtime authoritatively for mutation safety (#610). When it requires a
-    restart, local-only parity must never override it: this returns a typed,
-    fail-closed blocker that names the resolver as authoritative. Returns
-    ``None`` when the resolver does not require a restart.
-    """
-    if not resolver_restart_required:
-        return None
-    parity_optimistic = bool(assessment.get("in_parity")) and not (
-        assessment.get("stale") or assessment.get("live_stale"))
-    return {
-        "kind": "parity_resolver_disagreement",
-        "restart_required": True,
-        "resolver_authoritative": True,
-        "parity_optimistic": parity_optimistic,
-        "daemon_start_head": assessment.get("daemon_start_head"),
-        "local_head": assessment.get("local_head"),
-        "live_remote_head": assessment.get("live_remote_head"),
-        "reasons": [
-            "The capability resolver requires a restart/reconnect (stale "
-            "runtime) but master-parity reported local code as in-parity. "
-            "The resolver is authoritative for mutation safety; do not mutate "
-            "on local parity alone. Restart/reconnect the Gitea MCP server "
-            "and re-verify before mutating.",
-        ],
-        "recovery": [
-            "Trust the resolver: treat this session as stale.",
-            "Restart or /mcp reconnect the Gitea MCP namespace so it reloads "
-            "current master and live target state, then re-run preflight.",
-        ],
-    }
-
-
 def format_parity(assessment: dict) -> str:
     """One-line human summary for logs / runtime context."""
     if assessment.get("stale"):

merger_lease_adoption.py

@@ -7,7 +7,6 @@ after formal approval at the current PR head.
 
 from __future__ import annotations
 
-import re
 from datetime import datetime, timezone
 from typing import Any
 
@@ -134,115 +133,6 @@ def is_sanctioned_session_lease(session: dict[str, Any] | None) -> bool:
     return False
 
 
-def describe_session_lease_proof(
-    session: dict[str, Any] | None,
-) -> dict[str, Any]:
-    """Canonical merge-report lease proof fields (#535).
-
-    Surfaces how the in-session lease was established so merge handoffs can
-    distinguish sanctioned MCP acquire/adopt paths from bare manual seeding.
-    """
-    if not session:
-        return {
-            "lease_proof_source": None,
-            "lease_recovery_reason": None,
-            "lease_proof_kind": "none",
-            "lease_proof_sanctioned": False,
-            "lease_proof_comment_id": None,
-            "lease_adopted_from_session_id": None,
-        }
-
-    provenance = session.get("lease_provenance") or {}
-    if not isinstance(provenance, dict):
-        provenance = {}
-    source = (provenance.get("source") or "").strip() or None
-    sanctioned = is_sanctioned_session_lease(session)
-    reason = provenance.get("adoption_reason")
-    if isinstance(reason, str):
-        reason = reason.strip() or None
-    else:
-        reason = None
-
-    if source == SOURCE_ADOPT and sanctioned:
-        kind = "sanctioned_adoption"
-        reason = reason or DEFAULT_ADOPTION_REASON
-    elif source == SOURCE_ACQUIRE and sanctioned:
-        kind = "sanctioned_acquire"
-    elif source == SOURCE_HEARTBEAT and sanctioned:
-        kind = "sanctioned_heartbeat"
-    elif source:
-        kind = "unsanctioned"
-    else:
-        # Session present without provenance = classic manual _SESSION_LEASE seed.
-        kind = "unsanctioned_manual_seed"
-
-    comment_id = provenance.get("comment_id")
-    if comment_id is None:
-        comment_id = session.get("comment_id")
-
-    return {
-        "lease_proof_source": source,
-        "lease_recovery_reason": reason,
-        "lease_proof_kind": kind,
-        "lease_proof_sanctioned": sanctioned,
-        "lease_proof_comment_id": comment_id,
-        "lease_adopted_from_session_id": provenance.get("adopted_from_session_id"),
-    }
-
-
-# Phrases that claim non-MCP / fabricated lease proof in handoffs (#535).
-_UNSAFE_LEASE_PROOF_CLAIM_RE = re.compile(
-    r"(?is)\b("
-    r"manually\s+seeded\s+lease|"
-    r"manual(?:ly)?\s+seed(?:ed)?\s+(?:lease|proof)|"
-    r"seeded\s+lease\s+proof|"
-    r"injected\s+lease|"
-    r"lease\s+proof\s+injected|"
-    r"manual\s+_?SESSION_LEASE|"
-    r"_SESSION_LEASE\s+seed|"
-    r"unsanctioned\s+lease\s+proof"
-    r")\b"
-)
-
-# Evidence that the report used the sanctioned MCP adoption/acquire path.
-_SANCTIONED_LEASE_EVIDENCE_RE = re.compile(
-    r"(?is)\b("
-    r"gitea_adopt_merger_pr_lease|"
-    r"gitea_acquire_reviewer_pr_lease|"
-    r"lease_proof_source\s*[:=]\s*gitea_adopt_merger_pr_lease|"
-    r"lease_proof_source\s*[:=]\s*gitea_acquire_reviewer_pr_lease|"
-    r"lease_proof_kind\s*[:=]\s*sanctioned_adoption|"
-    r"lease_proof_kind\s*[:=]\s*sanctioned_acquire|"
-    r"adoption_comment_id\s*[:=]|"
-    r"sanctioned_adoption|"
-    r"mcp-review-lease-adoption"
-    r")\b"
-)
-
-
-def assess_manual_lease_proof_handoff(report_text: str) -> dict[str, Any]:
-    """Controller audit: block handoffs that claim unsafe lease seeding (#535).
-
-    Reports may discuss manual seeding as a blocked/historical anti-pattern only
-    when they also cite sanctioned MCP adoption/acquire evidence. Bare claims of
-    manual/seeded/injected lease proof without that evidence fail closed.
-    """
-    text = report_text or ""
-    if not _UNSAFE_LEASE_PROOF_CLAIM_RE.search(text):
-        return {"proven": True, "block": False, "reasons": []}
-    if _SANCTIONED_LEASE_EVIDENCE_RE.search(text):
-        return {"proven": True, "block": False, "reasons": []}
-    return {
-        "proven": False,
-        "block": True,
-        "reasons": [
-            "report claims manual/seeded/injected/unsanctioned lease proof without "
-            "sanctioned MCP adoption evidence (use gitea_adopt_merger_pr_lease / "
-            "gitea_acquire_reviewer_pr_lease and cite lease_proof_source; #535)"
-        ],
-    }
-
-
 def assess_adopt_merger_lease(
     comments: list[dict],
     *,

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,
-        }
+    }

post_merge_validation.py

@@ -1,222 +0,0 @@
-"""Canonical post-merge (moot) validation outcome and wording gate (#529).
-
-When a PR is already merged or closed *before* a reviewer submits their
-verdict, an ordinary "approve" is misleading: the review never gated the
-merge, so a normal active approval overstates controller confidence. The
-sanctioned outcome for that case is a **post-merge moot validation** — the
-reviewer records what validation found, but classifies it as moot rather than
-an active approval.
-
-This module defines the four canonical validation distinctions #529 requires
-and enforces the post-merge case:
-
-- ``CLEAN_PASS_OUTCOME`` — clean validation pass on an open, reviewable PR.
-- ``BASELINE_ACCEPTED_OUTCOME`` — validation pass with a baseline-proven
-  unrelated failure (proof handled by :mod:`premerge_baseline_proof`).
-- ``BLOCKED_OUTCOME`` — validation blocked by an unresolved failure.
-- ``POST_MERGE_MOOT_OUTCOME`` — the PR was already merged/closed before review
-  submission; validation is recorded as post-merge moot, not an active
-  approval.
-
-The gate blocks two mistakes:
-
-1. Claiming an *active approval* on a PR that was already merged/closed before
-   review submission (must be recorded as post-merge moot validation instead).
-2. Claiming post-merge moot validation without proof the PR is actually
-   merged/closed (a merged-state field plus a merge commit SHA).
-
-Each outcome maps to a durable ``validation:*`` process-state label so PRs and
-issues carry the distinction (#529 acceptance criterion).
-"""
-
-from __future__ import annotations
-
-import re
-from typing import Any
-
-CLEAN_PASS_OUTCOME = "clean validation pass"
-BASELINE_ACCEPTED_OUTCOME = "validation pass with baseline-proven unrelated failure"
-BLOCKED_OUTCOME = "validation blocked by unresolved failure"
-POST_MERGE_MOOT_OUTCOME = "post-merge moot validation"
-
-# Canonical validation status label a reviewer writes in the report body for the
-# post-merge case; kept in sync with validation_status_vocabulary.
-POST_MERGE_MOOT_STATUS = "post-merge moot validation"
-
-# Durable process-state labels (registered in issue_workflow_labels).
-LABEL_CLEAN_PASS = "validation:clean-pass"
-LABEL_BASELINE_ACCEPTED = "validation:baseline-accepted"
-LABEL_BLOCKED = "validation:blocked"
-LABEL_POST_MERGE_MOOT = "validation:post-merge-moot"
-
-_OUTCOME_TO_LABEL: dict[str, str] = {
-    CLEAN_PASS_OUTCOME: LABEL_CLEAN_PASS,
-    BASELINE_ACCEPTED_OUTCOME: LABEL_BASELINE_ACCEPTED,
-    BLOCKED_OUTCOME: LABEL_BLOCKED,
-    POST_MERGE_MOOT_OUTCOME: LABEL_POST_MERGE_MOOT,
-}
-
-# The PR was already merged/closed before the review was submitted.
-_MERGED_BEFORE_REVIEW_RE = re.compile(
-    r"(?:already[- ]merged"
-    r"|merged\s+before\s+review"
-    r"|closed\s+before\s+review"
-    r"|pr\s+(?:is|was)\s+(?:already\s+)?(?:merged|closed)"
-    r"|pr\s+state\s*[:=]\s*(?:merged|closed)"
-    r"|merged\s*/\s*closed)",
-    re.IGNORECASE,
-)
-# An active approval verdict is being claimed.
-_ACTIVE_APPROVAL_RE = re.compile(
-    r"(?:review\s+decision\s*[:=]\s*approve"
-    r"|submitted\s+['\"]?approve['\"]?"
-    r"|active\s+approval"
-    r"|approving\s+the\s+pr"
-    r"|posting\s+an?\s+approval)",
-    re.IGNORECASE,
-)
-# The sanctioned post-merge moot validation wording.
-_POST_MERGE_MOOT_RE = re.compile(
-    r"post[- ]merge\s+(?:moot\s+)?validation"
-    r"|post[- ]merge\s+moot"
-    r"|moot\s+validation",
-    re.IGNORECASE,
-)
-# Proof the PR is genuinely merged/closed.
-_MERGED_STATE_PROOF_RE = re.compile(
-    r"(?:pr\s+state\s*[:=]\s*(?:merged|closed)"
-    r"|merged_at\s*[:=]\s*\S"
-    r"|closed_at\s*[:=]\s*\S"
-    r"|state\s*[:=]\s*(?:merged|closed))",
-    re.IGNORECASE,
-)
-_MERGE_COMMIT_SHA_RE = re.compile(
-    r"merge\s+commit(?:\s+sha)?\s*[:=]\s*[0-9a-f]{7,40}",
-    re.IGNORECASE,
-)
-
-
-POST_MERGE_VALIDATION_TEMPLATE = (
-    "Validation status: post-merge moot validation\n"
-    "PR state: merged\n"
-    "Merge commit sha: <40-hex merge commit>\n"
-    "Validation finding: <what the validation run showed, for the record>\n"
-    "Note: PR merged/closed before review submission; recorded as post-merge "
-    "moot validation, not an active approval."
-)
-
-
-def process_state_label(outcome: str) -> str | None:
-    """Return the durable ``validation:*`` label for a canonical outcome."""
-    return _OUTCOME_TO_LABEL.get(outcome)
-
-
-def assess_post_merge_validation(
-    report_text: str,
-    *,
-    pr_merged_or_closed: bool | None = None,
-) -> dict[str, Any]:
-    """Assess post-merge moot validation wording and proof (#529).
-
-    Args:
-        report_text: The reviewer final report text.
-        pr_merged_or_closed: Optional structured signal that the PR is already
-            merged/closed. When ``True`` it forces the merged-before-review
-            path even if the report text omits the phrasing; proof fields are
-            still required in the report body for durability.
-
-    Returns a dict with ``proven``, ``block``, ``outcome``,
-    ``process_state_label``, ``reasons``, ``skipped`` and ``safe_next_action``.
-    Only blocks when the report either claims an active approval on a
-    merged/closed PR, or claims post-merge moot validation without proof.
-    """
-    text = report_text or ""
-
-    merged_signal = bool(_MERGED_BEFORE_REVIEW_RE.search(text)) or bool(
-        pr_merged_or_closed
-    )
-    active_approval = bool(_ACTIVE_APPROVAL_RE.search(text))
-    moot_wording = bool(_POST_MERGE_MOOT_RE.search(text))
-
-    # Nothing about merged-state or moot validation — not applicable here.
-    if not merged_signal and not moot_wording:
-        return {
-            "proven": True,
-            "block": False,
-            "outcome": None,
-            "process_state_label": None,
-            "reasons": [],
-            "skipped": True,
-            "safe_next_action": "",
-        }
-
-    # An active approval on a PR already merged/closed before review, without
-    # the sanctioned moot wording, overstates confidence.
-    if merged_signal and active_approval and not moot_wording:
-        return {
-            "proven": False,
-            "block": True,
-            "outcome": POST_MERGE_MOOT_OUTCOME,
-            "process_state_label": LABEL_POST_MERGE_MOOT,
-            "reasons": [
-                "PR was already merged/closed before review submission; an "
-                "active approval overstates confidence — record 'post-merge "
-                "moot validation' instead of a normal approval"
-            ],
-            "skipped": False,
-            "safe_next_action": (
-                "classify the outcome as post-merge moot validation (not an "
-                "active approval); cite PR state merged/closed and the merge "
-                "commit SHA. Template:\n" + POST_MERGE_VALIDATION_TEMPLATE
-            ),
-        }
-
-    # Post-merge moot validation claimed — require merged-state + commit proof.
-    if moot_wording:
-        reasons: list[str] = []
-        if not _MERGED_STATE_PROOF_RE.search(text):
-            reasons.append(
-                "post-merge moot validation claimed without merged/closed state "
-                "proof (state: merged/closed, or merged_at/closed_at)"
-            )
-        if not _MERGE_COMMIT_SHA_RE.search(text):
-            reasons.append(
-                "post-merge moot validation claimed without a merge commit SHA"
-            )
-        if reasons:
-            return {
-                "proven": False,
-                "block": True,
-                "outcome": POST_MERGE_MOOT_OUTCOME,
-                "process_state_label": LABEL_POST_MERGE_MOOT,
-                "reasons": reasons,
-                "skipped": False,
-                "safe_next_action": (
-                    "prove the PR is merged/closed: cite PR state and the merge "
-                    "commit SHA. Template:\n" + POST_MERGE_VALIDATION_TEMPLATE
-                ),
-            }
-        return {
-            "proven": True,
-            "block": False,
-            "outcome": POST_MERGE_MOOT_OUTCOME,
-            "process_state_label": LABEL_POST_MERGE_MOOT,
-            "reasons": [],
-            "skipped": False,
-            "safe_next_action": "",
-        }
-
-    # Merged signal present but no approval claim and no moot wording yet —
-    # guide toward the moot outcome without blocking.
-    return {
-        "proven": True,
-        "block": False,
-        "outcome": POST_MERGE_MOOT_OUTCOME,
-        "process_state_label": LABEL_POST_MERGE_MOOT,
-        "reasons": [],
-        "skipped": False,
-        "safe_next_action": (
-            "PR appears merged/closed; if submitting a verdict, record "
-            "post-merge moot validation rather than an active approval"
-        ),
-    }

remote_repo_guard.py

@@ -16,35 +16,11 @@ or the local remote URL is unavailable (best-effort corroboration only).
 
 from __future__ import annotations
 
-import re
-
 REMEDIATION = (
-    "Pass explicit org= and repo= matching the local git remote on tools that "
-    "accept those parameters (including mcp_get_control_plane_guide), "
-    "e.g. org=Scaled-Tech-Consulting repo=Gitea-Tools. "
-    "Do not pass org/repo to tools whose schema does not list them."
+    "Pass explicit org= and repo= matching the local git remote, "
+    "e.g. org=Scaled-Tech-Consulting repo=Gitea-Tools."
 )
 
-# https://host/org/repo.git  or  git@host:org/repo.git
-_REMOTE_URL_SLUG_RE = re.compile(
-    r"(?:[:/])(?P<org>[^/]+)/(?P<repo>[^/]+?)(?:\.git)?/*$"
-)
-
-
-def parse_org_repo_from_remote_url(remote_url: str | None) -> tuple[str, str] | None:
-    """Best-effort parse of org/repo from a git remote URL."""
-    url = (remote_url or "").strip()
-    if not url:
-        return None
-    match = _REMOTE_URL_SLUG_RE.search(url)
-    if not match:
-        return None
-    org = (match.group("org") or "").strip()
-    repo = (match.group("repo") or "").strip()
-    if not org or not repo:
-        return None
-    return org, repo
-
 
 def assess_remote_repo_match(
     *,

review_proofs.py

@@ -4842,22 +4842,6 @@ RECONCILER_CLOSE_REPORT_FIELDS = (
     "no review/merge confirmation",
 )
 
-RECONCILER_SUPERSESSION_REPORT_FIELDS = (
-    "identity/profile",
-    "supersession close capability proof",
-    "target PR live state",
-    "target PR independent-work proof",
-    "superseding PR merged state",
-    "superseding merge commit SHA",
-    "target branch SHA",
-    "superseding ancestry proof",
-    "canonical close comment",
-    "linked issue status",
-    "PR close result",
-    "issue close result",
-    "no review/merge confirmation",
-)
-
 
 def assess_reconciler_close_gate(
     *,
@@ -5004,157 +4988,6 @@ def assess_reconciler_close_gate(
     }
 
 
-def assess_reconciler_supersession_close_gate(
-    *,
-    target_pr_number: int | None,
-    target_pr_state: str | None,
-    target_pr_mergeable: bool | None,
-    target_independently_required: bool | None,
-    superseding_pr_number: int | None,
-    superseding_pr_state: str | None,
-    superseding_pr_merged: bool,
-    superseding_merge_commit_sha: str | None,
-    superseding_head_sha: str | None,
-    target_branch: str | None,
-    target_branch_sha: str | None,
-    superseding_head_is_ancestor_of_target: bool | None,
-    canonical_comment_valid: bool,
-    canonical_comment_mentions_superseding_pr: bool,
-    canonical_comment_mentions_merge_commit: bool,
-    close_pr_capability: bool,
-    close_issue_capability: bool = False,
-    target_issue_state: str | None = None,
-    issue_satisfied_by_superseding: bool = False,
-) -> dict:
-    """Gate reconciler closure of PRs/issues superseded by a merged PR.
-
-    This is stricter than already-landed cleanup: the target PR may be closed
-    only after a named superseding PR is live-verified merged, its head is on a
-    freshly fetched target branch, the target PR is proven not independently
-    required, and a canonical close comment cites the superseding PR and merge
-    commit.
-    """
-    reasons: list[str] = []
-    if not isinstance(target_pr_number, int) or target_pr_number <= 0:
-        reasons.append("target PR number missing or invalid (#525)")
-    if not isinstance(superseding_pr_number, int) or superseding_pr_number <= 0:
-        reasons.append("superseding PR number missing or invalid (#525)")
-    if target_pr_number and superseding_pr_number and (
-        target_pr_number == superseding_pr_number
-    ):
-        reasons.append("target PR and superseding PR must differ (#525)")
-    if (target_pr_state or "").strip().lower() != "open":
-        reasons.append("target PR must be live-verified open before close (#525)")
-    if target_pr_mergeable is True:
-        reasons.append(
-            "target PR is still mergeable; prove it is superseded before close (#525)"
-        )
-    if target_independently_required is not False:
-        reasons.append(
-            "target PR independent-work proof missing; close requires explicit "
-            "proof that no required work remains (#525)"
-        )
-    if (superseding_pr_state or "").strip().lower() != "closed":
-        reasons.append("superseding PR must be live-verified closed (#525)")
-    if superseding_pr_merged is not True:
-        reasons.append("superseding PR must be live-verified merged (#525)")
-    if not _FULL_SHA.match((superseding_merge_commit_sha or "").strip()):
-        reasons.append("superseding merge commit SHA missing or invalid (#525)")
-    if not _FULL_SHA.match((superseding_head_sha or "").strip()):
-        reasons.append("superseding head SHA missing or invalid (#525)")
-    if not (target_branch or "").strip():
-        reasons.append("target branch missing (#525)")
-    if not _FULL_SHA.match((target_branch_sha or "").strip()):
-        reasons.append("target branch SHA missing or invalid (#525)")
-    if superseding_head_is_ancestor_of_target is None:
-        reasons.append("superseding ancestry proof not checked (#525)")
-    if not canonical_comment_valid:
-        reasons.append("canonical close comment failed validation (#525)")
-    if not canonical_comment_mentions_superseding_pr:
-        reasons.append("canonical comment must cite superseding PR (#525)")
-    if not canonical_comment_mentions_merge_commit:
-        reasons.append("canonical comment must cite merge commit SHA (#525)")
-
-    never_allowed = {
-        "review_allowed": False,
-        "approve_allowed": False,
-        "request_changes_allowed": False,
-        "merge_allowed": False,
-    }
-
-    if reasons:
-        return {
-            "outcome": "GATE_NOT_PROVEN",
-            "pr_close_allowed": False,
-            "issue_close_allowed": False,
-            "reasons": reasons,
-            "required_report_fields": RECONCILER_SUPERSESSION_REPORT_FIELDS,
-            "safe_next_action": (
-                "prove superseding PR state, target branch ancestry, target "
-                "supersession, and canonical comment before closing"
-            ),
-            **never_allowed,
-        }
-
-    if superseding_head_is_ancestor_of_target is False:
-        return {
-            "outcome": "SUPERSEDING_PR_NOT_ON_TARGET",
-            "pr_close_allowed": False,
-            "issue_close_allowed": False,
-            "reasons": [
-                f"superseding PR #{superseding_pr_number} head is not an "
-                f"ancestor of {target_branch}; supersession close denied (#525)"
-            ],
-            "required_report_fields": RECONCILER_SUPERSESSION_REPORT_FIELDS,
-            "safe_next_action": "re-fetch target branch and re-check ancestry",
-            **never_allowed,
-        }
-
-    if close_pr_capability is not True:
-        return {
-            "outcome": "RECOVERY_HANDOFF_REQUIRED",
-            "pr_close_allowed": False,
-            "issue_close_allowed": False,
-            "reasons": [
-                "exact gitea.pr.close capability not proven; produce a "
-                "recovery handoff instead of closing (#525)"
-            ],
-            "required_report_fields": RECONCILER_SUPERSESSION_REPORT_FIELDS,
-            "safe_next_action": (
-                "launch a reconciler profile with gitea.pr.close and replay "
-                "the live-state proof"
-            ),
-            **never_allowed,
-        }
-
-    issue_close_allowed = (
-        (target_issue_state or "").strip().lower() == "open"
-        and issue_satisfied_by_superseding is True
-        and close_issue_capability is True
-    )
-    issue_reasons = []
-    if (target_issue_state or "").strip().lower() == "closed":
-        issue_reasons.append("linked issue already closed; no issue close attempted (#525)")
-    elif target_issue_state and not issue_close_allowed:
-        issue_reasons.append(
-            "issue close requires an open issue satisfied by the superseding "
-            "merged PR and exact gitea.issue.close capability (#525)"
-        )
-
-    return {
-        "outcome": "SUPERSESSION_CLOSE_ALLOWED",
-        "pr_close_allowed": True,
-        "issue_close_allowed": issue_close_allowed,
-        "reasons": issue_reasons,
-        "required_report_fields": RECONCILER_SUPERSESSION_REPORT_FIELDS,
-        "safe_next_action": (
-            "post the canonical comment, close the superseded PR, and close "
-            "the linked issue only when issue_close_allowed is true"
-        ),
-        **never_allowed,
-    }
-
-
 # ---------------------------------------------------------------------------
 # Identity disclosure (#305)
 # ---------------------------------------------------------------------------

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
 
 
@@ -522,241 +512,4 @@ def assess_lease_inventory(
         "active_review_leases": active,
         "stale_review_leases": stale,
         "reclaimable_review_leases": reclaimable,
-    }
-
-# Canonical next-action vocabulary for reviewer lease handoff (#599).
-NEXT_ACTION_ACQUIRE = "acquire"
-NEXT_ACTION_WAIT = "wait"
-NEXT_ACTION_RESUME_EXACT_OWNER_SESSION = "resume_exact_owner_session"
-NEXT_ACTION_RELEASE_EXPIRED_LEASE = "release_expired_lease"
-NEXT_ACTION_OPERATOR_AUTHORIZED_CLEANUP = "operator_authorized_cleanup"
-NEXT_ACTION_REPAIR_WORKTREE_BINDING = "repair_worktree_binding"
-
-_HANDOFF_CLASSIFICATIONS = frozenset({
-    "no_lease",
-    "own_active",
-    "own_expired",
-    "foreign_active",
-    "foreign_reclaimable",
-    "foreign_expired",
-    "instructed_lease_missing_with_replacement",
-    "worktree_binding_mismatch",
-})
-
-
-def _norm_path(value: str | None) -> str:
-    text = (value or "").strip()
-    if not text:
-        return ""
-    return os.path.normpath(text.rstrip("/"))
-
-
-def diagnose_reviewer_pr_lease_handoff(
-    comments: list[dict],
-    *,
-    pr_number: int,
-    current_session_id: str | None,
-    current_reviewer_identity: str | None,
-    proposed_worktree: str | None = None,
-    env_bound_worktree: str | None = None,
-    instructed_session_id: str | None = None,
-    instructed_comment_id: int | None = None,
-    now: datetime | None = None,
-) -> dict[str, Any]:
-    """Classify open-PR reviewer lease handoff and emit a canonical next action (#599).
-
-    Read-only diagnosis. Never steals, releases, or adopts a foreign lease.
-    Fail-closed acquisition rules for active foreign leases remain intact.
-
-    Returns a structured diagnosis with:
-    - classification
-    - next_action (one of wait / resume_exact_owner_session /
-      release_expired_lease / operator_authorized_cleanup /
-      repair_worktree_binding / acquire)
-    - active_lease identity fields when present
-    - worktree_binding match result
-    - instructed-lease mismatch flags
-    """
-    now = now or datetime.now(timezone.utc)
-    session_id = (current_session_id or "").strip()
-    identity = (current_reviewer_identity or "").strip()
-    instructed_sid = (instructed_session_id or "").strip()
-    reasons: list[str] = []
-
-    active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
-    session_lease = get_session_lease()
-
-    # Worktree binding: env-bound vs proposed vs active lease worktree.
-    env_wt = _norm_path(env_bound_worktree)
-    prop_wt = _norm_path(proposed_worktree)
-    lease_wt = _norm_path((active or {}).get("worktree") if active else None)
-    binding_mismatch = False
-    binding_details: dict[str, Any] = {
-        "env_bound_worktree": env_bound_worktree or None,
-        "proposed_worktree": proposed_worktree or None,
-        "lease_worktree": (active or {}).get("worktree") if active else None,
-        "match": True,
-    }
-    paths = [p for p in (env_wt, prop_wt, lease_wt) if p]
-    if len(paths) >= 2 and len(set(paths)) > 1:
-        binding_mismatch = True
-        binding_details["match"] = False
-        reasons.append(
-            "worktree binding mismatch: env/proposed/lease worktree paths disagree"
-        )
-
-    # Instructed lease gone while a different lease is active (PR #592-style).
-    instructed_missing_with_replacement = False
-    if instructed_sid or instructed_comment_id is not None:
-        if not active:
-            reasons.append(
-                "instructed lease is gone and no active replacement lease remains"
-            )
-        else:
-            owner = (active.get("session_id") or "").strip()
-            cid = active.get("comment_id")
-            sid_mismatch = bool(instructed_sid and owner and owner != instructed_sid)
-            cid_mismatch = (
-                instructed_comment_id is not None
-                and cid is not None
-                and int(cid) != int(instructed_comment_id)
-            )
-            if sid_mismatch or cid_mismatch:
-                instructed_missing_with_replacement = True
-                reasons.append(
-                    "instructed lease is gone; a different active lease replaced it "
-                    f"(active session_id={owner}, comment_id={cid})"
-                )
-
-    # Classification + next_action.
-    classification = "no_lease"
-    next_action = NEXT_ACTION_ACQUIRE
-
-    if active:
-        owner = (active.get("session_id") or "").strip()
-        freshness = active.get("freshness") or classify_lease_freshness(
-            active, now=now
-        )
-        owner_identity = (active.get("reviewer_identity") or "").strip()
-        is_own = bool(session_id and owner and owner == session_id)
-        # Same identity alone is NOT ownership for resume; session_id must match.
-        same_identity = bool(
-            identity and owner_identity and identity == owner_identity
-        )
-
-        if is_own and freshness in {"active", "stale_warning"}:
-            classification = "own_active"
-            next_action = NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
-        elif is_own and freshness in {"reclaimable", "expired"}:
-            classification = "own_expired"
-            next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
-            reasons.append(
-                f"own lease freshness is '{freshness}'; release via "
-                "gitea_release_reviewer_pr_lease then re-acquire"
-            )
-        elif not is_own and freshness in {"active", "stale_warning"}:
-            classification = "foreign_active"
-            next_action = NEXT_ACTION_WAIT
-            reasons.append(
-                f"foreign active reviewer lease (session_id={owner}, "
-                f"phase={active.get('phase')}, freshness={freshness}); "
-                "do not submit; do not steal"
-            )
-            if same_identity:
-                reasons.append(
-                    "lease identity matches current reviewer but session_id differs; "
-                    "resume only from the exact owner session_id or wait"
-                )
-        elif not is_own and freshness == "reclaimable":
-            classification = "foreign_reclaimable"
-            next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
-            reasons.append(
-                f"foreign reclaimable lease (session_id={owner}); clear only via "
-                "sanctioned gitea_release_reviewer_pr_lease when reclaimable"
-            )
-        elif not is_own and freshness == "expired":
-            classification = "foreign_expired"
-            next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
-            reasons.append(
-                f"foreign expired lease (session_id={owner}); use sanctioned release"
-            )
-        else:
-            classification = "foreign_active"
-            next_action = NEXT_ACTION_WAIT
-            reasons.append(
-                f"unclassified active lease state (session_id={owner}, "
-                f"freshness={freshness}); wait fail-closed"
-            )
-
-        if instructed_missing_with_replacement and classification.startswith(
-            "foreign"
-        ):
-            classification = "instructed_lease_missing_with_replacement"
-            # Foreign active still means wait; reclaimable still release.
-            if next_action == NEXT_ACTION_WAIT:
-                reasons.append(
-                    "replacement foreign lease is active — wait; "
-                    "operator_authorized_cleanup only with explicit operator authority"
-                )
-    else:
-        classification = "no_lease"
-        next_action = NEXT_ACTION_ACQUIRE
-        reasons.append("no active reviewer lease; acquire via gitea_acquire_reviewer_pr_lease")
-
-    # Binding mismatch is a first-class blocker before submit, but does not
-    # erase foreign-lease wait/release guidance. Override next_action only when
-    # the session would otherwise be free to acquire or resume (submit path).
-    if binding_mismatch:
-        if next_action in {
-            NEXT_ACTION_ACQUIRE,
-            NEXT_ACTION_RESUME_EXACT_OWNER_SESSION,
-        }:
-            classification = "worktree_binding_mismatch"
-            next_action = NEXT_ACTION_REPAIR_WORKTREE_BINDING
-        else:
-            reasons.append(
-                "also repair worktree binding before submit "
-                f"(next_action remains {next_action})"
-            )
-
-    lease_summary = None
-    if active:
-        lease_summary = {
-            "comment_id": active.get("comment_id"),
-            "session_id": active.get("session_id"),
-            "phase": active.get("phase"),
-            "candidate_head": active.get("candidate_head"),
-            "expires_at": active.get("expires_at"),
-            "last_activity": active.get("last_activity"),
-            "freshness": active.get("freshness")
-            or classify_lease_freshness(active, now=now),
-            "reviewer_identity": active.get("reviewer_identity"),
-            "profile": active.get("profile"),
-            "worktree": active.get("worktree"),
-            "blocker": active.get("blocker"),
-        }
-
-    return {
-        "pr_number": pr_number,
-        "classification": classification,
-        "next_action": next_action,
-        "active_lease": lease_summary,
-        "session_lease": session_lease,
-        "worktree_binding": binding_details,
-        "instructed_session_id": instructed_session_id,
-        "instructed_comment_id": instructed_comment_id,
-        "instructed_lease_missing_with_replacement": instructed_missing_with_replacement,
-        "mutation_allowed": (
-            next_action == NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
-            and not binding_mismatch
-            and bool(session_lease)
-        ),
-        "reasons": reasons,
-        "forbidden": [
-            "manual lock deletion",
-            "raw API bypass",
-            "mtime manipulation",
-            "direct _SESSION_LEASE seeding",
-            "silent foreign lease steal",
-        ],
-    }
+    }

role_session_router.py

@@ -55,6 +55,7 @@ def skip_python_scan_walk_root(project_root: str, walk_root: str) -> bool:
 
 REVIEWER_TASKS = frozenset({
     "review_pr",
+    "merge_pr",
     "blind_pr_queue_review",
     "pr_queue_cleanup",
     "pr-queue-cleanup",
@@ -62,10 +63,6 @@ REVIEWER_TASKS = frozenset({
     "approve_pr",
 })
 
-MERGER_TASKS = frozenset({
-    "merge_pr",
-})
-
 AUTHOR_TASKS = frozenset({
     "create_issue",
     "comment_issue",
@@ -83,7 +80,6 @@ AUTHOR_TASKS = frozenset({
 })
 
 RECONCILER_TASKS = frozenset({
-    "cleanup_merged_pr_branch",
     "reconcile_already_landed_pr",
     "reconcile_already_landed",
     "reconcile-landed-pr",
@@ -101,7 +97,7 @@ TASK_REQUIRED_ROLE = {
     "address_pr_change_requests": "author",
     "delete_branch": "author",
     "review_pr": "reviewer",
-    "merge_pr": "merger",
+    "merge_pr": "reviewer",
     "blind_pr_queue_review": "reviewer",
     "pr_queue_cleanup": "reviewer",
     "pr-queue-cleanup": "reviewer",
@@ -113,13 +109,9 @@ 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",
-    "reconcile_close_superseded_pr": "reconciler",
-    "reconcile_close_satisfied_issue": "reconciler",
-    "reconcile_create_followup_issue": "reconciler",
 }
 
 WRONG_ROLE_REVIEWER_MSG = (
@@ -131,10 +123,6 @@ WRONG_ROLE_RECONCILER_MSG = (
     "MCP namespace/profile with exact close capability."
 )
 
-WRONG_ROLE_MERGER_MSG = (
-    "Wrong role/session for merger task. Launch merger MCP namespace."
-)
-
 _session_last_route: dict | None = None
 
 
@@ -250,25 +238,6 @@ def route_task_session(
         _record_route(result)
         return result
 
-    if required_role == "merger":
-        result = {
-            "task_type": task_type,
-            "required_role": required_role,
-            "active_role": active_role_kind,
-            "active_profile": active_profile,
-            "route_result": ROUTE_WRONG_ROLE,
-            "downstream_allowed": False,
-            "reasons": [
-                WRONG_ROLE_MERGER_MSG,
-                "Merger tasks cannot run in author or reviewer sessions.",
-            ],
-            "message": WRONG_ROLE_MERGER_MSG,
-            "runtime_switching_supported": runtime_switching_supported,
-            "profile_switch_blocked": not runtime_switching_supported,
-        }
-        _record_route(result)
-        return result
-
     if required_role == "author":
         route = ROUTE_TO_AUTHOR
         message = (
@@ -437,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"]

scripts/ci-webui-check

@@ -1,53 +0,0 @@
-#!/usr/bin/env bash
-# Path-filtered CI gate for the internal web UI (#436).
-# Runs the hermetic web UI unittest suite when a change touches UI surfaces.
-set -euo pipefail
-
-script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
-repo_root="$(cd "$script_dir/.." && pwd)"
-cd "$repo_root"
-
-if [[ "${WEBUI_CI_FORCE:-}" == "1" ]]; then
-  exec "$script_dir/test-webui"
-fi
-
-base_ref="${WEBUI_CI_BASE_REF:-}"
-if [[ -z "$base_ref" ]]; then
-  if git rev-parse --verify prgs/master >/dev/null 2>&1; then
-    base_ref="$(git merge-base HEAD prgs/master 2>/dev/null || true)"
-  fi
-  if [[ -z "$base_ref" ]]; then
-    base_ref="${GITHUB_BASE_REF:-${CHANGE_TARGET:-master}}"
-    if git rev-parse --verify "origin/$base_ref" >/dev/null 2>&1; then
-      base_ref="$(git merge-base HEAD "origin/$base_ref")"
-    elif git rev-parse --verify "$base_ref" >/dev/null 2>&1; then
-      base_ref="$(git merge-base HEAD "$base_ref")"
-    else
-      base_ref="HEAD~1"
-    fi
-  fi
-fi
-
-changed="$(git diff --name-only "$base_ref" HEAD 2>/dev/null || true)"
-if [[ -z "$changed" ]]; then
-  echo "ci-webui-check: no changed files vs $base_ref; skipping web UI suite"
-  exit 0
-fi
-
-python_bin="${WEBUI_TEST_PYTHON:-python3}"
-if [[ -x "$repo_root/venv/bin/python" ]]; then
-  python_bin="$repo_root/venv/bin/python"
-fi
-
-if printf '%s\n' "$changed" | "$python_bin" -c "
-import sys
-from webui.ci_paths import should_run_webui_ci
-paths = [line.strip() for line in sys.stdin if line.strip()]
-raise SystemExit(0 if should_run_webui_ci(paths) else 1)
-"; then
-  echo "ci-webui-check: web UI surface changed; running suite"
-  exec "$script_dir/test-webui"
-fi
-
-echo "ci-webui-check: no web UI paths in diff; skipping suite"
-exit 0

scripts/test-webui

@@ -1,14 +0,0 @@
-#!/usr/bin/env bash
-set -euo pipefail
-script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
-repo_root="$(cd "$script_dir/.." && pwd)"
-cd "$repo_root"
-
-python_bin="${WEBUI_TEST_PYTHON:-python3}"
-if [[ -x "$repo_root/venv/bin/python" ]]; then
-  python_bin="$repo_root/venv/bin/python"
-fi
-
-pattern="${WEBUI_TEST_PATTERN:-test_webui_*.py}"
-export WEBUI_TEST_OFFLINE="${WEBUI_TEST_OFFLINE:-1}"
-exec "$python_bin" -m unittest discover -s tests -p "$pattern" "$@"

skills/llm-project-workflow/workflows/create-issue.md

@@ -344,41 +344,6 @@ If edits are needed, make the smallest correction necessary and report the corre
 
 Do not silently change requested meaning.
 
-## 14a. Enforced content preflight (#582)
-
-`gitea_create_issue` runs a **content preflight gate** before duplicate
-search or creation. It fails closed (returns `BLOCKED + DIAGNOSE`, no issue
-created) when the title/body is a *vague reference* to out-of-band draft
-content the session cannot prove it holds, and no durable source pointer is
-present.
-
-An LLM must never fabricate issue content from stale chat memory. If asked to
-create "the drafted issue" / "the prepared issue" / "the issue we discussed"
-/ "the previous draft" without the full content in the current context or a
-durable source pointer, stop and return `BLOCKED + DIAGNOSE` naming the exact
-vague/missing fields.
-
-A **durable source pointer** is one of: an existing issue/PR reference
-(`#582`), a comment id (`comment 8155`), a checked-in file path
-(`task_capability_map.py`), a URL, or a scratchpad path. When a pointer is
-present, retrieve the real content from it before creating.
-
-The gate does **not** require a non-empty body; explicit title-only creation
-stays allowed. It only blocks placeholder references. An operator may bypass
-it with `allow_incomplete_content=True` (report the override).
-
-**Invalid prompts (must block):**
-
-* "Create the drafted issue."
-* "File the prepared issue we discussed."
-* "Open the previous draft."
-
-**Valid prompts (proceed):**
-
-* Full title + body + acceptance criteria supplied inline.
-* "Create the issue drafted in `#582`." (durable pointer → fetch and use it)
-* "Create the issue from `scratchpad/issue-draft.md`." (durable file pointer)
-
 ## 15. Labels, assignees, and metadata
 
 Apply labels, assignees, milestones, or project fields only if:

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

@@ -826,75 +826,6 @@ The final report must identify:
 * whether same-PR merge continuation was allowed
 * whether the run stopped as required
 
-## 26A-1. Stale durable review-decision lock cleanup (#594)
-
-After #559, the review-decision lock is durable under
-`~/.cache/gitea-tools/session-state/` (for example
-`review_decision_lock-prgs-reviewer.json`). That durability can outlive the work
-it protects: a terminal `approve` / `request_changes` on a PR that is already
-**merged or closed** still hard-stops later unrelated reviews under #332.
-
-**Do not** delete session-state files by hand. Use the canonical MCP path.
-
-### When cleanup is allowed
-
-Use `gitea_cleanup_stale_review_decision_lock` when:
-
-1. `gitea_mark_final_review_decision` / live review is blocked by
-   `terminal review mutation already consumed ... (#332)`.
-2. Live Gitea state for the **last terminal mutation's PR** is **merged** or
-   **closed** (no same-PR merge sequence remains).
-3. Active profile identity matches the durable lock (reviewer profile with
-   `gitea.pr.review` for `apply=true`).
-4. Optional pin: pass `expected_terminal_pr` to the prior terminal PR number
-   (for example `586` when resuming after a merged #586 lock).
-
-Recommended sequence:
-
-```text
-gitea_whoami
-gitea_resolve_task_capability(task="cleanup_stale_review_decision_lock")
-gitea_cleanup_stale_review_decision_lock(apply=false, remote=..., org=..., repo=...)
-# inspect is_moot / cleanup_allowed / last_terminal_pr
-gitea_cleanup_stale_review_decision_lock(
-    apply=true,
-    expected_terminal_pr=<last_terminal_pr>,
-    remote=..., org=..., repo=...,
-)
-gitea_resolve_task_capability(task="review_pr")
-gitea_load_review_workflow()
-# continue formal mark + submit for the *new* PR
-```
-
-Successful `apply=true` clears memory + durable store and returns an audit
-record (and posts a durable PR comment on the terminal PR when
-`post_audit_comment` is true and comment permission allows).
-
-Same-profile auto-expire: after `gitea_merge_pr` succeeds for the PR that this
-profile just approved, the decision lock for that profile is cleared
-automatically. Cross-profile locks (for example reviewer approve, merger merge)
-still require `gitea_cleanup_stale_review_decision_lock`.
-
-### When cleanup is forbidden
-
-Refuse / fail-closed — keep #332 hard-stop — when:
-
-* the last terminal PR is still **open** (active review or merge sequence may
-  still apply)
-* live PR state cannot be fetched (ambiguous)
-* no lock or no terminal live mutation exists
-* profile identity does not match the lock
-* apply requested without authenticated identity or `gitea.pr.review`
-* `expected_terminal_pr` does not match the last terminal PR
-
-Do **not** use cleanup to:
-
-* bypass #332 on an open PR
-* replace `gitea_authorize_review_correction` for a mistaken review on a still
-  active PR (#211)
-* auto-approve or auto-merge any PR
-* justify `rm` of session-state files
-
 ## 26B. Per-PR reviewer lease (#407)
 
 Parallel reviewer sessions are allowed only when each session holds a distinct,
@@ -1005,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
@@ -1112,8 +1035,6 @@ Blocked handoffs must say to rerun the full workflow after the blocker clears.
 
 If the blocker is a terminal review mutation already consumed in the current run, the handoff must say that the next run must start from the beginning and must not reuse stale ready/approved state.
 
-If the referenced terminal PR is already **merged or closed** on live Gitea, the durable #332 decision lock is **moot**. Do **not** delete session-state files by hand. Use `gitea_cleanup_stale_review_decision_lock` (assess with `apply=false`, then `apply=true` with `expected_terminal_pr` set) from the reviewer profile after identity/profile checks (#594). Cleanup is **forbidden** while that PR is still open.
-
 ## 30. Final report must be precise
 
 Include:

stale_review_decision_lock.py

@@ -1,252 +0,0 @@
-"""Stale / moot #332 review-decision lock detection and cleanup policy (#594).
-
-#332 correctly hard-stops a reviewer session after a terminal live review
-mutation. After #559 those locks are durable on disk, so they can outlive the
-work they protect: when the referenced PR is already merged or closed, no
-same-PR merge sequence remains, yet the durable ledger still blocks unrelated
-new terminal reviews.
-
-This module is pure policy (no Gitea I/O). The MCP tool
-``gitea_cleanup_stale_review_decision_lock`` fetches live PR state, calls these
-helpers, and only then clears durable state when cleanup is allowed.
-
-Manual deletion of ``~/.cache/gitea-tools/session-state/*`` is never the
-canonical path.
-"""
-
-from __future__ import annotations
-
-from datetime import datetime, timezone
-from typing import Any
-
-# Keep in sync with gitea_mcp_server._TERMINAL_REVIEW_ACTIONS.
-TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"})
-
-
-def last_terminal_mutation(lock: dict | None) -> dict | None:
-    """Return the last terminal live mutation on *lock*, or None."""
-    if not lock:
-        return None
-    terminals = [
-        m
-        for m in (lock.get("live_mutations") or [])
-        if isinstance(m, dict) and m.get("action") in TERMINAL_REVIEW_ACTIONS
-    ]
-    return terminals[-1] if terminals else None
-
-
-def classify_pr_live_state(pr_live: dict | None) -> dict[str, Any]:
-    """Normalize a Gitea PR payload into merge/closed flags."""
-    pr = pr_live or {}
-    merged = bool(pr.get("merged") or pr.get("merged_at"))
-    state = (pr.get("state") or "").strip().lower() or None
-    closed = state == "closed"
-    return {
-        "pr_state": state,
-        "pr_merged": merged,
-        "pr_merged_or_closed": merged or closed,
-        "merge_commit_sha": pr.get("merge_commit_sha")
-        or pr.get("merged_commit_sha"),
-    }
-
-
-def lock_summary(lock: dict | None) -> dict[str, Any] | None:
-    """LLM-safe summary of a decision lock (no secrets)."""
-    if lock is None:
-        return None
-    last = last_terminal_mutation(lock)
-    mutations = list(lock.get("live_mutations") or [])
-    return {
-        "task": lock.get("task"),
-        "remote": lock.get("remote"),
-        "org": lock.get("org") or lock.get("ready_org"),
-        "repo": lock.get("repo") or lock.get("ready_repo"),
-        "profile_identity": lock.get("profile_identity")
-        or lock.get("session_profile_lock")
-        or lock.get("session_profile"),
-        "session_profile": lock.get("session_profile"),
-        "ready_pr_number": lock.get("ready_pr_number"),
-        "ready_action": lock.get("ready_action"),
-        "live_mutations_count": len(mutations),
-        "last_terminal": (
-            {
-                "pr_number": last.get("pr_number"),
-                "action": last.get("action"),
-                "review_id": last.get("review_id"),
-            }
-            if last
-            else None
-        ),
-        "correction_authorized": bool(lock.get("correction_authorized")),
-        "updated_at": lock.get("updated_at") or lock.get("recorded_at"),
-    }
-
-
-def assess_stale_review_decision_lock(
-    lock: dict | None,
-    *,
-    pr_live: dict | None = None,
-    pr_lookup_error: str | None = None,
-    active_profile_identity: str | None = None,
-) -> dict[str, Any]:
-    """Assess whether a durable review-decision lock is stale/moot (#594).
-
-    *pr_live* must be the live Gitea payload for the **last terminal
-    mutation's PR**. When no lock / no terminal mutation exists, cleanup is
-    not applicable. When the PR is still open (or lookup fails), cleanup is
-    forbidden and #332 hard-stop remains in force.
-    """
-    summary = lock_summary(lock)
-    result: dict[str, Any] = {
-        "has_lock": lock is not None,
-        "lock_summary": summary,
-        "last_terminal_pr": None,
-        "last_terminal_action": None,
-        "is_moot": False,
-        "cleanup_allowed": False,
-        "profile_match": True,
-        "pr_state": None,
-        "pr_merged": False,
-        "pr_merged_or_closed": False,
-        "merge_commit_sha": None,
-        "reasons": [],
-        "recovery_tool": "gitea_cleanup_stale_review_decision_lock",
-    }
-
-    if lock is None:
-        result["reasons"].append("no review decision lock present")
-        return result
-
-    # Profile identity gate (caller may re-check with env; pure assess still
-    # reports mismatch when active identity is supplied).
-    stored = (
-        (lock.get("profile_identity") or lock.get("session_profile_lock") or "")
-        .strip()
-    )
-    active = (active_profile_identity or "").strip()
-    if active and stored and active != stored:
-        result["profile_match"] = False
-        result["reasons"].append(
-            "review decision lock profile identity does not match active "
-            f"profile '{active}' (fail closed, #594)"
-        )
-        return result
-
-    last = last_terminal_mutation(lock)
-    if last is None:
-        result["reasons"].append(
-            "review decision lock has no terminal live mutations; nothing to clear"
-        )
-        return result
-
-    pr_number = last.get("pr_number")
-    action = last.get("action")
-    result["last_terminal_pr"] = pr_number
-    result["last_terminal_action"] = action
-
-    if pr_number is None:
-        result["reasons"].append(
-            "last terminal mutation missing pr_number; refuse cleanup (fail closed)"
-        )
-        return result
-
-    if pr_lookup_error:
-        result["reasons"].append(
-            f"could not load live state for PR #{pr_number}: {pr_lookup_error} "
-            "(fail closed, #594)"
-        )
-        return result
-
-    if pr_live is None:
-        result["reasons"].append(
-            f"live PR state required for terminal PR #{pr_number} before cleanup "
-            "(fail closed, #594)"
-        )
-        return result
-
-    live = classify_pr_live_state(pr_live)
-    result.update(
-        {
-            "pr_state": live["pr_state"],
-            "pr_merged": live["pr_merged"],
-            "pr_merged_or_closed": live["pr_merged_or_closed"],
-            "merge_commit_sha": live["merge_commit_sha"],
-        }
-    )
-
-    if not live["pr_merged_or_closed"]:
-        result["reasons"].append(
-            f"terminal review mutation on open PR #{pr_number} is still active "
-            f"({action}); #332 hard-stop remains — cleanup forbidden (#594)"
-        )
-        return result
-
-    # Merged or closed: lock is moot. Same-PR merge continuation is impossible.
-    result["is_moot"] = True
-    result["cleanup_allowed"] = True
-    result["reasons"].append(
-        f"terminal mutation ({action} on PR #{pr_number}) is moot: PR is "
-        f"{'merged' if live['pr_merged'] else 'closed'}; canonical cleanup allowed (#594)"
-    )
-    return result
-
-
-def build_cleanup_audit_record(
-    *,
-    assessment: dict[str, Any],
-    actor_username: str | None,
-    profile_name: str | None,
-    applied: bool,
-) -> dict[str, Any]:
-    """Structured durable audit payload for a cleanup attempt."""
-    last = (assessment.get("lock_summary") or {}).get("last_terminal") or {}
-    return {
-        "event": "stale_review_decision_lock_cleanup",
-        "issue_ref": "#594",
-        "applied": bool(applied),
-        "timestamp": datetime.now(timezone.utc).isoformat(),
-        "actor_username": actor_username,
-        "profile_name": profile_name,
-        "last_terminal_pr": assessment.get("last_terminal_pr") or last.get("pr_number"),
-        "last_terminal_action": assessment.get("last_terminal_action")
-        or last.get("action"),
-        "pr_state": assessment.get("pr_state"),
-        "pr_merged": assessment.get("pr_merged"),
-        "pr_merged_or_closed": assessment.get("pr_merged_or_closed"),
-        "merge_commit_sha": assessment.get("merge_commit_sha"),
-        "prior_live_mutations_count": (
-            (assessment.get("lock_summary") or {}).get("live_mutations_count")
-        ),
-        "prior_profile_identity": (
-            (assessment.get("lock_summary") or {}).get("profile_identity")
-        ),
-        "cleanup_allowed": assessment.get("cleanup_allowed"),
-        "is_moot": assessment.get("is_moot"),
-        "reasons": list(assessment.get("reasons") or []),
-    }
-
-
-def format_cleanup_audit_comment(audit: dict[str, Any]) -> str:
-    """Markdown body for a durable Gitea audit comment after cleanup."""
-    status = "APPLIED" if audit.get("applied") else "ASSESSED (not applied)"
-    lines = [
-        "## Stale #332 review-decision lock cleanup (#594)",
-        "",
-        f"Status: **{status}**",
-        "",
-        f"- actor: `{audit.get('actor_username')}`",
-        f"- profile: `{audit.get('profile_name')}`",
-        f"- timestamp: `{audit.get('timestamp')}`",
-        f"- last terminal: `{audit.get('last_terminal_action')}` on "
-        f"PR #{audit.get('last_terminal_pr')}",
-        f"- PR state: `{audit.get('pr_state')}` "
-        f"(merged={audit.get('pr_merged')})",
-        f"- merge_commit_sha: `{audit.get('merge_commit_sha')}`",
-        f"- prior live_mutations_count: "
-        f"`{audit.get('prior_live_mutations_count')}`",
-        f"- prior profile_identity: `{audit.get('prior_profile_identity')}`",
-        "",
-        "Manual deletion of session-state files is **not** the workflow.",
-        "This path only clears a lock when the referenced PR is merged/closed.",
-    ]
-    return "\n".join(lines)

task_capability_map.py

@@ -96,25 +96,10 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.pr.approve",
         "role": "reviewer",
     },
-    # #594: clear durable #332 decision lock only when last terminal PR is
-    # already merged/closed (moot). Apply path requires reviewer review
-    # permission; assessment itself uses gitea.read inside the tool.
-    "cleanup_stale_review_decision_lock": {
-        "permission": "gitea.pr.review",
-        "role": "reviewer",
-    },
-    "gitea_cleanup_stale_review_decision_lock": {
-        "permission": "gitea.pr.review",
-        "role": "reviewer",
-    },
     "delete_branch": {
         "permission": "gitea.branch.delete",
         "role": "author",
     },
-    "cleanup_merged_pr_branch": {
-        "permission": "gitea.branch.delete",
-        "role": "reconciler",
-    },
     "commit_files": {
         "permission": "gitea.repo.commit",
         "role": "author",
@@ -161,18 +146,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.issue.close",
         "role": "reconciler",
     },
-    "reconcile_close_superseded_pr": {
-        "permission": "gitea.pr.close",
-        "role": "reconciler",
-    },
-    "reconcile_close_satisfied_issue": {
-        "permission": "gitea.issue.close",
-        "role": "reconciler",
-    },
-    "reconcile_create_followup_issue": {
-        "permission": "gitea.issue.create",
-        "role": "reconciler",
-    },
     "post_heartbeat": {
         "permission": "gitea.issue.comment",
         "role": "author",

tests/conftest.py

@@ -1,5 +1,4 @@
 """Shared pytest fixtures for the Gitea-Tools test suite."""
-import os
 import sys
 
 import pytest
@@ -17,46 +16,13 @@ def _reset_mutation_authority(monkeypatch):
     the next test. It must never replace verify_mutation_authority with a
     no-op: individual tests that need a specific authority state set it up
     explicitly.
-
-    Session-state isolation (#559 / #590 / #594): many tests use
-    ``patch.dict(os.environ, ..., clear=True)``, which drops
-    ``GITEA_MCP_SESSION_STATE_DIR``. Relying only on the env var therefore
-    re-exposes the operator host cache
-    (``~/.cache/gitea-tools/session-state``) and its review-mutation ledger.
-    Pin ``default_state_dir`` / ``DEFAULT_STATE_DIR`` to a per-test temp dir
-    so durable load/save never touches host state even after env clears.
     """
     monkeypatch.delenv("GITEA_SESSION_PROFILE_LOCK", raising=False)
     # Isolate durable session-state files so tests never share host cache (#559).
     import tempfile
 
     _state_tmp = tempfile.TemporaryDirectory(prefix="gitea-session-state-")
-    state_dir = _state_tmp.name
-    monkeypatch.setenv("GITEA_MCP_SESSION_STATE_DIR", state_dir)
-    try:
-        import mcp_session_state
-    except Exception:
-        mcp_session_state = None
-    if mcp_session_state is not None:
-        # Survive patch.dict(..., clear=True) that drops the env var (#590).
-        # Still honor an explicit GITEA_MCP_SESSION_STATE_DIR when tests set
-        # their own temp dir (see tests/test_mcp_session_state.py).
-        monkeypatch.setattr(
-            mcp_session_state, "DEFAULT_STATE_DIR", state_dir, raising=False
-        )
-
-        def _isolated_default_state_dir(
-            _fallback: str = state_dir,
-            _env_key: str = mcp_session_state.STATE_DIR_ENV,
-        ) -> str:
-            raw = (os.environ.get(_env_key) or "").strip()
-            return raw or _fallback
-
-        monkeypatch.setattr(
-            mcp_session_state,
-            "default_state_dir",
-            _isolated_default_state_dir,
-        )
+    monkeypatch.setenv("GITEA_MCP_SESSION_STATE_DIR", _state_tmp.name)
     try:
         import mcp_server
     except Exception:
@@ -74,14 +40,6 @@ def _reset_mutation_authority(monkeypatch):
     monkeypatch.setattr(mcp_server, "_preflight_capability_baseline_porcelain", None)
     monkeypatch.setattr(mcp_server, "_preflight_resolved_task", None)
     monkeypatch.setattr(mcp_server, "_preflight_reviewer_violation_files", [])
-    # Unit tests must not depend on live host MCP process health. Tests that
-    # use patch.dict(..., clear=True) drop PYTEST_CURRENT_TEST, which would
-    # otherwise re-enable the stale-runtime probe against operator daemons.
-    monkeypatch.setattr(
-        mcp_server,
-        "_check_mcp_runtimes_diagnostics",
-        lambda *args, **kwargs: [],
-    )
     try:
         import review_workflow_load
         review_workflow_load._REVIEW_WORKFLOW_LOAD = None

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_controller_closure_baseline_proof.py

@@ -1,113 +0,0 @@
-"""Controller-closure baseline-proof gate tests (#529, criterion 6)."""
-
-import unittest
-
-from controller_closure_baseline_proof import (
-    assess_controller_closure_baseline_proof,
-)
-from premerge_baseline_proof import (
-    CLEAN_PASS,
-    PREMERGE_BASELINE_PROVEN_FAILURE,
-    UNRESOLVED_REGRESSION_RISK,
-)
-from final_report_validator import assess_final_report_validator
-
-# Complete pre-merge proof fields for a baseline claim (see #533).
-_PROOF_FIELDS = (
-    "Pre-merge base commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
-    "Tested commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
-    "Command: venv/bin/python -m pytest tests/test_foo.py -q\n"
-    "Exit status: 1\n"
-    "Failure signature: AssertionError: None is not true\n"
-)
-
-
-class TestControllerClosureBaselineProof(unittest.TestCase):
-    def test_empty_report_skipped_and_allowed(self):
-        result = assess_controller_closure_baseline_proof("")
-        self.assertFalse(result["block"])
-        self.assertTrue(result["skipped"])
-        self.assertEqual(result["label"], CLEAN_PASS)
-
-    def test_none_report_allowed(self):
-        result = assess_controller_closure_baseline_proof(None)
-        self.assertFalse(result["block"])
-        self.assertTrue(result["skipped"])
-
-    def test_clean_pass_closure_allowed(self):
-        result = assess_controller_closure_baseline_proof(
-            "Closing #529. Validation: full suite passed, exit status 0. Clean pass."
-        )
-        self.assertFalse(result["block"])
-        self.assertEqual(result["label"], CLEAN_PASS)
-
-    def test_expected_preexisting_failure_without_proof_blocked(self):
-        report = (
-            "Closing #529. Full suite: 1 failed. This is an expected "
-            "pre-existing failure, safe to close."
-        )
-        result = assess_controller_closure_baseline_proof(report)
-        self.assertTrue(result["block"])
-        self.assertFalse(result["proven"])
-        self.assertEqual(result["label"], UNRESOLVED_REGRESSION_RISK)
-        self.assertTrue(result["reasons"])
-        self.assertTrue(result["safe_next_action"])
-
-    def test_current_master_reproduction_only_blocked(self):
-        report = (
-            "Closing #529. Full suite: 1 failed. Pre-existing baseline failure — "
-            "reproduced on current master after merge.\n"
-            "Command: venv/bin/python -m pytest tests/test_foo.py -q\n"
-            "Exit status: 1\n"
-            "Failure signature: AssertionError: None is not true\n"
-        )
-        result = assess_controller_closure_baseline_proof(report)
-        self.assertTrue(result["block"])
-
-    def test_proven_baseline_closure_allowed(self):
-        report = (
-            "Closing #529. Full suite: 1 failed, an expected pre-existing "
-            "baseline failure proven on the PR pre-merge base commit.\n"
-            + _PROOF_FIELDS
-        )
-        result = assess_controller_closure_baseline_proof(report)
-        self.assertFalse(result["block"])
-        self.assertEqual(result["label"], PREMERGE_BASELINE_PROVEN_FAILURE)
-
-
-class TestControllerCloseTaskKind(unittest.TestCase):
-    """The composable validator must cover the controller_close task kind."""
-
-    def test_controller_close_blocks_unproven_baseline(self):
-        report = (
-            "Full suite: 1 failed. Expected pre-existing failure, closing the "
-            "tracking issue."
-        )
-        result = assess_final_report_validator(report, "controller_close")
-        self.assertTrue(result["blocked"])
-        self.assertTrue(
-            any(
-                "premerge_baseline_proof" in f["rule_id"]
-                for f in result["findings"]
-            )
-        )
-
-    def test_controller_close_alias_close_issue(self):
-        report = (
-            "Full suite: 1 failed. Expected pre-existing failure, closing the "
-            "tracking issue."
-        )
-        result = assess_final_report_validator(report, "close_issue")
-        self.assertTrue(result["blocked"])
-
-    def test_controller_close_allows_proven_baseline(self):
-        report = (
-            "Full suite: 1 failed, expected pre-existing baseline failure proven "
-            "on the PR pre-merge base commit.\n" + _PROOF_FIELDS
-        )
-        result = assess_final_report_validator(report, "controller_close")
-        self.assertFalse(result["blocked"])
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_final_report_validator.py

@@ -781,103 +781,6 @@ class TestMergePrRules(unittest.TestCase):
         self.assertEqual(result["grade"], "A")
         self.assertFalse(result["blocked"])
 
-    def test_manual_seeded_lease_claim_blocks_without_mcp_evidence(self):
-        lines = [
-            "## Controller Handoff",
-            "",
-            "- Task: merge PR #203",
-            "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
-            "- Role: merger",
-            "- Identity: sysadmin / prgs-merger",
-            "- Issue/PR: #182 / PR #203",
-            "- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
-            "- Files changed: review_proofs.py",
-            "- Validation: pytest in branches/review-203",
-            "- Mutations: merge completed",
-            "- File edits by reviewer: none",
-            "- Worktree/index mutations: none",
-            "- Git ref mutations: git fetch prgs master",
-            "- MCP/Gitea mutations: merge completed",
-            "- Review mutations: none",
-            "- Merge mutations: merged using manually seeded lease proof",
-            "- Cleanup mutations: none",
-            "- External-state mutations: none",
-            "- Read-only diagnostics: metadata check",
-            "- Current status: PR merged",
-            "- Blockers: none",
-            "- Next: none",
-            "- Safety: review and merge are separate roles",
-            "- Selected PR: #203",
-            "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
-            "- Reviewed head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
-            "- Final live head SHA before approval: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
-            "- Final live head SHA before merge: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
-            "- Push occurred during validation: no",
-            "- Active profile: prgs-merger",
-            "- Role kind: merger",
-            "- Merge capability source: profile",
-            "- Explicit operator authorization: MERGE PR 203",
-            "- Expected head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
-            "- Approval at current head: yes",
-            "- Merge result: PR merged successfully",
-            "- Cleanup status: complete",
-        ]
-        report = "\n".join(lines)
-        result = assess_final_report_validator(report, "merge_pr")
-        self.assertTrue(result["blocked"])
-        codes = {f.get("rule_id") for f in result.get("findings") or []}
-        self.assertIn("shared.manual_lease_proof_handoff", codes)
-
-    def test_manual_seeded_claim_allowed_when_adoption_cited(self):
-        lines = [
-            "## Controller Handoff",
-            "",
-            "- Task: merge PR #203",
-            "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
-            "- Role: merger",
-            "- Identity: sysadmin / prgs-merger",
-            "- Issue/PR: #182 / PR #203",
-            "- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
-            "- Files changed: review_proofs.py",
-            "- Validation: pytest in branches/review-203",
-            "- Mutations: merge completed",
-            "- File edits by reviewer: none",
-            "- Worktree/index mutations: none",
-            "- Git ref mutations: git fetch prgs master",
-            "- MCP/Gitea mutations: gitea_adopt_merger_pr_lease then merge",
-            "- Review mutations: none",
-            "- Merge mutations: historical manual seed replaced by sanctioned adoption",
-            "- Cleanup mutations: none",
-            "- External-state mutations: none",
-            "- Read-only diagnostics: metadata check",
-            "- Current status: PR merged",
-            "- Blockers: none",
-            "- Next: none",
-            "- Safety: review and merge are separate roles",
-            "- Selected PR: #203",
-            "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
-            "- Reviewed head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
-            "- Final live head SHA before approval: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
-            "- Final live head SHA before merge: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
-            "- Push occurred during validation: no",
-            "- Active profile: prgs-merger",
-            "- Role kind: merger",
-            "- Merge capability source: profile",
-            "- Explicit operator authorization: MERGE PR 203",
-            "- Expected head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
-            "- Approval at current head: yes",
-            "- lease_proof_source: gitea_adopt_merger_pr_lease",
-            "- lease_proof_kind: sanctioned_adoption",
-            "- adoption_comment_id: 8288",
-            "- Merge result: PR merged successfully",
-            "- Cleanup status: complete",
-        ]
-        report = "\n".join(lines)
-        result = assess_final_report_validator(report, "merge_pr")
-        self.assertFalse(result["blocked"])
-        codes = {f.get("rule_id") for f in result.get("findings") or []}
-        self.assertNotIn("shared.manual_lease_proof_handoff", codes)
-
     def test_missing_merger_fields_blocks(self):
         lines = [
             "## Controller Handoff",

tests/test_issue_content_gate.py

@@ -1,195 +0,0 @@
-"""Tests for the pre-create issue content gate (Issue #582)."""
-import sys
-import unittest
-from unittest.mock import patch
-
-sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
-
-from issue_content_gate import (  # noqa: E402
-    assess_issue_content,
-    find_vague_reference,
-    has_durable_source_pointer,
-    normalize_text,
-    pre_create_issue_content_gate,
-)
-from mcp_server import gitea_create_issue  # noqa: E402
-
-FAKE_AUTH = "Basic dGVzdDp0ZXN0"
-ISSUE_WRITE_ENV = {
-    "GITEA_ALLOWED_OPERATIONS": "gitea.issue.create",
-}
-
-
-class TestNormalizeAndPointers(unittest.TestCase):
-
-    def test_normalize_strips_punctuation_and_case(self):
-        self.assertEqual(normalize_text("  The Drafted-Issue!  "), "the drafted issue")
-
-    def test_issue_ref_is_durable_pointer(self):
-        self.assertTrue(has_durable_source_pointer("See #402 for the spec"))
-
-    def test_file_path_is_durable_pointer(self):
-        self.assertTrue(has_durable_source_pointer("Draft in task_capability_map.py"))
-
-    def test_scratchpad_is_durable_pointer(self):
-        self.assertTrue(has_durable_source_pointer("Full text in scratchpad/draft.md"))
-
-    def test_url_is_durable_pointer(self):
-        self.assertTrue(
-            has_durable_source_pointer("https://gitea.prgs.cc/issues/1 has it")
-        )
-
-    def test_plain_text_has_no_durable_pointer(self):
-        self.assertFalse(has_durable_source_pointer("just some words here"))
-
-
-class TestFindVagueReference(unittest.TestCase):
-
-    def test_bare_vague_phrases_are_flagged(self):
-        for text in (
-            "the drafted issue",
-            "The prepared issue",
-            "the issue we discussed",
-            "the previous draft",
-            "Create the drafted issue",
-            "please file the prepared issue",
-        ):
-            with self.subTest(text=text):
-                self.assertIsNotNone(find_vague_reference(text))
-
-    def test_durable_pointer_defeats_vague_reference(self):
-        # Same vague phrase but with a durable pointer -> not vague.
-        self.assertIsNone(find_vague_reference("the drafted issue in #402"))
-        self.assertIsNone(
-            find_vague_reference("the prepared issue: see scratchpad/draft.md")
-        )
-
-    def test_long_specific_body_with_incidental_phrase_is_not_vague(self):
-        body = (
-            "As discussed, the create_issue tool must reject vague references. "
-            "Add a preflight check that lists the exact missing fields and "
-            "returns BLOCKED + DIAGNOSE with concrete acceptance criteria."
-        )
-        self.assertIsNone(find_vague_reference(body))
-
-    def test_concrete_short_body_is_not_vague(self):
-        self.assertIsNone(find_vague_reference("Fix null deref in auth handler"))
-
-    def test_empty_is_not_vague(self):
-        self.assertIsNone(find_vague_reference(""))
-
-
-class TestAssessIssueContent(unittest.TestCase):
-
-    def test_missing_title_is_incomplete(self):
-        result = assess_issue_content("", "some body")
-        self.assertFalse(result["complete"])
-        self.assertIn("title", result["missing_fields"])
-
-    def test_vague_body_is_incomplete_and_diagnosed(self):
-        result = assess_issue_content("Add the thing", "the drafted issue")
-        self.assertFalse(result["complete"])
-        self.assertIn("body", result["vague_fields"])
-        self.assertIn("vague reference", result["diagnose"])
-
-    def test_vague_title_is_incomplete(self):
-        result = assess_issue_content("the prepared issue", "")
-        self.assertFalse(result["complete"])
-        self.assertIn("title", result["vague_fields"])
-
-    def test_complete_content_passes(self):
-        result = assess_issue_content(
-            "Block vague issue creation",
-            "Add a preflight gate that rejects placeholder references.",
-        )
-        self.assertTrue(result["complete"])
-        self.assertEqual(result["missing_fields"], [])
-        self.assertEqual(result["vague_fields"], [])
-
-    def test_title_only_is_allowed(self):
-        # Empty body must NOT block: title-only creation stays allowed.
-        result = assess_issue_content("A concrete specific title", "")
-        self.assertTrue(result["complete"])
-
-    def test_durable_pointer_body_passes(self):
-        result = assess_issue_content(
-            "Implement the drafted change",
-            "the drafted issue is fully specified in #582",
-        )
-        self.assertTrue(result["complete"])
-
-    def test_allow_incomplete_override(self):
-        result = assess_issue_content(
-            "the drafted issue", "the drafted issue", allow_incomplete=True
-        )
-        self.assertTrue(result["complete"])
-        self.assertTrue(result["override_applied"])
-
-    def test_gate_wrapper_sets_performed(self):
-        blocked = pre_create_issue_content_gate("Add x", "the drafted issue")
-        self.assertFalse(blocked["performed"])
-        ok = pre_create_issue_content_gate("Add x", "concrete actionable body")
-        self.assertTrue(ok["performed"])
-
-
-class TestCreateIssueContentMCPGate(unittest.TestCase):
-
-    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop")
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.api_get_all")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_blocks_vague_reference_without_creating(
-        self, _auth, mock_get_all, mock_api, mock_role_check
-    ):
-        mock_role_check.return_value = (True, [])
-        with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
-            result = gitea_create_issue(
-                title="Create the drafted issue",
-                body="the drafted issue",
-                remote="prgs",
-            )
-        self.assertFalse(result["performed"])
-        self.assertIn("content_gate", result)
-        self.assertIn("body", result["content_gate"]["vague_fields"])
-        self.assertTrue(any("BLOCKED + DIAGNOSE" in r for r in result["reasons"]))
-        mock_api.assert_not_called()
-        mock_get_all.assert_not_called()
-
-    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop")
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.api_get_all", return_value=[])
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_creates_when_content_is_concrete(
-        self, _auth, _get_all, mock_api, mock_role_check
-    ):
-        mock_role_check.return_value = (True, [])
-        mock_api.return_value = {"number": 7, "html_url": "https://x/issues/7"}
-        with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
-            result = gitea_create_issue(
-                title="Block vague issue creation",
-                body="Add a preflight gate rejecting placeholder references.",
-                remote="prgs",
-            )
-        self.assertEqual(result["number"], 7)
-
-    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop")
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.api_get_all", return_value=[])
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_override_creates_despite_vague_content(
-        self, _auth, _get_all, mock_api, mock_role_check
-    ):
-        mock_role_check.return_value = (True, [])
-        mock_api.return_value = {"number": 8, "html_url": "https://x/issues/8"}
-        with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
-            result = gitea_create_issue(
-                title="the drafted issue",
-                body="the drafted issue",
-                remote="prgs",
-                allow_incomplete_content=True,
-            )
-        self.assertEqual(result["number"], 8)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_master_parity_gate.py

@@ -78,95 +78,6 @@ class TestBlockReasonsAndReport(unittest.TestCase):
         self.assertTrue(report["recovery"])
 
 
-class TestLiveRemoteParity(unittest.TestCase):
-    """#610: parity must account for the live remote master, not just local.
-
-    The daemon can be stale relative to the live remote target while the local
-    checkout HEAD still matches the daemon's startup commit, so local parity
-    reports green even though a mutation would run against outdated code.
-    """
-
-    SHA_C = "c" * 40
-
-    def test_distinguishes_three_shas(self):
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
-        self.assertEqual(res["daemon_start_head"], SHA_A)
-        self.assertEqual(res["local_head"], SHA_A)
-        self.assertEqual(res["live_remote_head"], SHA_B)
-
-    def test_mutation_safe_only_when_all_three_match(self):
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_A)
-        self.assertTrue(res["mutation_safe"])
-        self.assertTrue(res["live_known"])
-        self.assertFalse(res["live_stale"])
-
-    def test_live_stale_when_remote_advanced_past_daemon(self):
-        # Local checkout still matches the daemon start (local parity green),
-        # but the live remote master has advanced -> daemon is live-stale.
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
-        self.assertTrue(res["in_parity"])  # local parity still green
-        self.assertTrue(res["live_stale"])
-        self.assertFalse(res["mutation_safe"])
-        self.assertTrue(any("live" in r.lower() for r in res["reasons"]))
-
-    def test_live_unknown_is_not_mutation_safe_but_not_stale(self):
-        # Non-goal: unfetchable live remote must not be treated as stale for
-        # read-only, but a mutation-safe claim fails closed.
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=None)
-        self.assertFalse(res["live_known"])
-        self.assertFalse(res["mutation_safe"])
-        self.assertFalse(res["live_stale"])
-        self.assertTrue(res["in_parity"])
-
-    def test_default_live_remote_preserves_legacy_shape(self):
-        # Callers that do not supply a live head keep the pre-#610 behavior:
-        # in-parity, not live-stale, no live-derived block.
-        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
-        self.assertFalse(res["live_stale"])
-        self.assertEqual(mp.parity_block_reasons(res), [])
-
-
-class TestLiveStaleBlockAndReport(unittest.TestCase):
-    """#610: live-staleness must block mutations and surface a typed blocker."""
-
-    def test_live_stale_produces_block_reasons(self):
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
-        self.assertTrue(mp.parity_block_reasons(res))
-
-    def test_disable_env_suppresses_live_stale_block(self):
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
-        with patch.dict(os.environ, {mp.ENV_DISABLE: "1"}):
-            self.assertEqual(mp.parity_block_reasons(res), [])
-
-    def test_resolver_disagreement_returns_typed_blocker(self):
-        # Parity says local-green, resolver says restart required -> disagreement
-        # is a typed, fail-closed blocker naming the resolver as authoritative.
-        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
-        blocker = mp.parity_resolver_disagreement(res, resolver_restart_required=True)
-        self.assertIsNotNone(blocker)
-        self.assertEqual(blocker["kind"], "parity_resolver_disagreement")
-        self.assertTrue(blocker["restart_required"])
-        self.assertTrue(blocker["resolver_authoritative"])
-
-    def test_no_disagreement_when_resolver_agrees(self):
-        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
-        self.assertIsNone(
-            mp.parity_resolver_disagreement(res, resolver_restart_required=False))
-
-    def test_live_stale_report_names_live_remote(self):
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
-        report = mp.parity_report(res)
-        self.assertEqual(report["live_remote_head"], SHA_B)
-        self.assertTrue(report["restart_required"])
-
-
 class TestReadGitHead(unittest.TestCase):
     def test_test_override_takes_precedence(self):
         with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
@@ -184,78 +95,6 @@ class TestReadGitHead(unittest.TestCase):
             self.assertIsNone(mp.read_git_head(""))
 
 
-class TestReadRemoteMasterHead(unittest.TestCase):
-    """#610: live remote master head reader (env-overridable, fails to None)."""
-
-    def test_test_override_takes_precedence(self):
-        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
-            self.assertEqual(mp.read_remote_master_head("/nonexistent"), SHA_B)
-
-    def test_blank_override_is_none(self):
-        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: "   "}):
-            self.assertIsNone(mp.read_remote_master_head("/nonexistent"))
-
-    def test_unfetchable_remote_is_none(self):
-        # No override; a bogus root/remote must fail closed to None, never raise.
-        env = {k: v for k, v in os.environ.items()
-               if k != mp.ENV_TEST_LIVE_REMOTE_HEAD}
-        with patch.dict(os.environ, env, clear=True):
-            self.assertIsNone(
-                mp.read_remote_master_head("/nonexistent", remote="nope"))
-
-
-class TestRemoteHeadCache(unittest.TestCase):
-    """#610: live remote reads are cached with a TTL to stay off the network.
-
-    The parity gate runs on every mutation and every runtime-context read, so an
-    unbounded ``git ls-remote`` per call would be a latency/flakiness regression.
-    """
-
-    def setUp(self):
-        mp._clear_remote_head_cache()
-        env = {k: v for k, v in os.environ.items()
-               if k != mp.ENV_TEST_LIVE_REMOTE_HEAD}
-        self._env = patch.dict(os.environ, env, clear=True)
-        self._env.start()
-        self.addCleanup(self._env.stop)
-        self.addCleanup(mp._clear_remote_head_cache)
-
-    def _fake_run(self, sha):
-        class _R:
-            returncode = 0
-            stdout = f"{sha}\trefs/heads/master\n"
-        calls = {"n": 0}
-
-        def run(*args, **kwargs):
-            calls["n"] += 1
-            return _R()
-        return run, calls
-
-    def test_second_call_within_ttl_uses_cache(self):
-        run, calls = self._fake_run(SHA_B)
-        with patch.object(mp.subprocess, "run", run):
-            a = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
-            b = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
-        self.assertEqual(a, SHA_B)
-        self.assertEqual(b, SHA_B)
-        self.assertEqual(calls["n"], 1)
-
-    def test_zero_ttl_bypasses_cache(self):
-        run, calls = self._fake_run(SHA_B)
-        with patch.object(mp.subprocess, "run", run):
-            mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
-            mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
-        self.assertEqual(calls["n"], 2)
-
-    def test_env_override_never_touches_subprocess(self):
-        run, calls = self._fake_run(SHA_B)
-        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
-            with patch.object(mp.subprocess, "run", run):
-                self.assertEqual(
-                    mp.read_remote_master_head("/repo", remote="prgs"), SHA_A)
-        self.assertEqual(calls["n"], 0)
-
-
 class TestServerWiring(unittest.TestCase):
     """Integration with the gate choke point in the server namespace."""
 
@@ -266,13 +105,6 @@ class TestServerWiring(unittest.TestCase):
         self._saved = self.srv._STARTUP_PARITY
         self.srv._STARTUP_PARITY = {"root": self.srv.PROJECT_ROOT,
                                     "startup_head": SHA_A}
-        # Keep the live-remote read hermetic (no real ls-remote network call):
-        # default the live master to the daemon start so parity is fully green
-        # unless a test overrides the live head explicitly (#610).
-        self._live_patch = patch.dict(
-            os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A})
-        self._live_patch.start()
-        self.addCleanup(self._live_patch.stop)
 
     def tearDown(self):
         self.srv._STARTUP_PARITY = self._saved
@@ -315,36 +147,6 @@ class TestServerWiring(unittest.TestCase):
             self.assertTrue(out["in_parity"])
             self.assertNotIn("report", out)
 
-    # --- #610: live-remote wiring -------------------------------------------
-
-    def test_live_stale_blocks_mutation_though_local_green(self):
-        # Local checkout matches the daemon start (local parity green) but the
-        # live remote master has advanced -> mutations must fail closed.
-        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
-                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
-            self.assertEqual(self.srv._master_parity_block("gitea.read"), [])
-            self.assertTrue(
-                self.srv._master_parity_block("gitea.pr.create"))
-
-    def test_assess_tool_exposes_three_distinct_shas(self):
-        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
-                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
-            out = self.srv.gitea_assess_master_parity(remote="prgs")
-            self.assertEqual(out["daemon_start_head"], SHA_A)
-            self.assertEqual(out["local_head"], SHA_A)
-            self.assertEqual(out["live_remote_head"], SHA_B)
-            self.assertTrue(out["live_stale"])
-            self.assertFalse(out["mutation_safe"])
-            self.assertIn("report", out)
-
-    def test_assess_tool_mutation_safe_when_all_three_match(self):
-        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
-                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
-            out = self.srv.gitea_assess_master_parity(remote="prgs")
-            self.assertTrue(out["mutation_safe"])
-            self.assertFalse(out["live_stale"])
-            self.assertNotIn("report", out)
-
 
 if __name__ == "__main__":
     unittest.main()

tests/test_mcp_server.py

@@ -774,9 +774,6 @@ class TestMergePR(unittest.TestCase):
     def setUp(self):
         import reviewer_pr_lease
 
-        # Clean ledger each merge test so residual/host durable #332 state
-        # cannot short-circuit eligibility gates (#590 / #594).
-        mcp_server._save_review_decision_lock(None)
         mcp_server.gitea_load_review_workflow()
         self._lease_patch = _install_owned_reviewer_lease(8)
         self._lease_patch.start()
@@ -1006,64 +1003,6 @@ class TestMergePR(unittest.TestCase):
             r["reasons"])
         self._assert_no_merge_call(mock_api)
 
-    @patch("mcp_server.api_request")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_env_clear_does_not_adopt_host_review_mutation_ledger(
-        self, _auth, mock_api
-    ):
-        """#590: patch.dict(clear=True) must not load host terminal mutations.
-
-        Simulates a dirty host session-state file for profile gitea-default
-        (the profile active when env is fully cleared). Isolation must keep
-        the #332 hard-stop ledger clean so the empty-profile gate fires.
-        """
-        import tempfile
-
-        import mcp_session_state
-
-        poison_dir = tempfile.mkdtemp(prefix="gitea-host-poison-")
-        mcp_session_state.save_state(
-            kind=mcp_session_state.KIND_DECISION_LOCK,
-            profile_identity="gitea-default",
-            state_dir=poison_dir,
-            payload={
-                "task": "review_pr",
-                "remote": "prgs",
-                "final_review_decision_ready": True,
-                "ready_pr_number": 8,
-                "ready_action": "request_changes",
-                "live_mutations": [
-                    {
-                        "pr_number": 8,
-                        "action": "request_changes",
-                        "review_id": 99,
-                        "review_state": "request_changes",
-                    }
-                ],
-            },
-        )
-        # If isolation only used the env var, clear=True would fall back to
-        # DEFAULT_STATE_DIR and adopt this poison ledger.
-        mcp_server._REVIEW_DECISION_LOCK = None
-        mock_api.side_effect = [{"login": "merger-bot"}, self._pr("author-bot")]
-        with patch.object(mcp_session_state, "DEFAULT_STATE_DIR", poison_dir):
-            with patch.dict(os.environ, {}, clear=True):
-                r = gitea_merge_pr(
-                    pr_number=8,
-                    confirmation=self._confirm(8),
-                    remote="prgs",
-                )
-        self.assertFalse(r["performed"])
-        self.assertIn(
-            "profile has no configured allowed operations (fail closed)",
-            r["reasons"],
-        )
-        self.assertFalse(
-            any("terminal review mutation already consumed" in x for x in r["reasons"]),
-            msg=f"host ledger leaked into reasons: {r['reasons']}",
-        )
-        self._assert_no_merge_call(mock_api)
-
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_profile_without_merge_permission_blocks(self, _auth, mock_api):
@@ -2733,56 +2672,6 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
         self.assertTrue(res["success"])
         self.assertEqual(res["cleanup_status"].get(1), "not present")
 
-    def test_close_issue_blocks_unproven_expected_preexisting_failure(self):
-        # #529: a closure report calling a non-zero suite exit an "expected
-        # pre-existing failure" without pre-merge proof must fail closed and
-        # never PATCH the issue state.
-        patched = []
-
-        def api_side_effect(method, url, auth, payload=None):
-            if method == "PATCH":
-                patched.append(url)
-            return {}
-        self.mock_api.side_effect = api_side_effect
-
-        res = gitea_close_issue(
-            issue_number=1,
-            closure_report=(
-                "Full suite: 1 failed. Expected pre-existing failure, safe to close."
-            ),
-        )
-        self.assertFalse(res["success"])
-        self.assertTrue(res.get("blocked"))
-        self.assertTrue(res.get("reasons"))
-        self.assertFalse(
-            patched, "must not PATCH issue state when the closure gate blocks"
-        )
-
-    def test_close_issue_allows_proven_baseline_closure(self):
-        def api_side_effect(method, url, auth, payload=None):
-            if method == "PATCH" and "issues/1" in url:
-                return {"state": "closed"}
-            if method == "GET" and "labels" in url and "issues" not in url:
-                return [{"name": "bug", "id": 2}]
-            if method == "GET" and "issues/1" in url:
-                return {"labels": [{"name": "bug"}]}
-            return {}
-        self.mock_api.side_effect = api_side_effect
-
-        res = gitea_close_issue(
-            issue_number=1,
-            closure_report=(
-                "Full suite: 1 failed, expected pre-existing baseline failure "
-                "proven on the PR pre-merge base commit.\n"
-                "Pre-merge base commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
-                "Tested commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
-                "Command: venv/bin/python -m pytest -q\n"
-                "Exit status: 1\n"
-                "Failure signature: AssertionError: None is not true\n"
-            ),
-        )
-        self.assertTrue(res["success"])
-
     def test_merge_pr_with_closes_removes_label(self):
         import reviewer_pr_lease
 
@@ -4399,236 +4288,3 @@ class TestIssue546Deadlock(unittest.TestCase):
             self.assertIsNone(reviewer_pr_lease.get_session_lease())
             # verify comment phase is released
             self.assertIn("phase: released", posted_comments[0]["body"])
-
-    def test_reviewer_pr_lease_gate_resolves_identity_with_host(self):
-        """Lease mutation gate must resolve identity using host, not remote alias."""
-        import mcp_server
-
-        resolved_hosts = []
-
-        def tracking_username(host):
-            resolved_hosts.append(host)
-            return None if host in {"prgs", "dadeschools"} else "reviewer-bot"
-
-        with _install_owned_reviewer_lease(
-            550,
-            session_id=_DEFAULT_LEASE_SESSION,
-            head_sha="abc123",
-        ), patch(
-            "mcp_server._resolve",
-            return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
-        ), patch(
-            "mcp_server._authenticated_username",
-            side_effect=tracking_username,
-        ):
-            reasons = mcp_server._reviewer_pr_lease_gate(
-                pr_number=550,
-                remote="prgs",
-                host=None,
-                org=None,
-                repo=None,
-                mutation="approve",
-                live_head_sha="abc123",
-                pinned_head_sha="abc123",
-            )
-
-        self.assertEqual(reasons, [])
-        self.assertIn("gitea.prgs.cc", resolved_hosts)
-        self.assertNotIn("prgs", resolved_hosts)
-
-    def test_acquire_reviewer_pr_lease_resolves_identity_with_host(self):
-        """Acquire path must not pass a remote alias into identity lookup."""
-        import mcp_server
-
-        resolved_hosts = []
-        posted = []
-
-        def tracking_username(host):
-            resolved_hosts.append(host)
-            return None if host in {"prgs", "dadeschools"} else "reviewer-bot"
-
-        def mock_api(method, url, auth, data=None):
-            if "/pulls/" in url:
-                return {
-                    "user": {"login": "author-user"},
-                    "state": "open",
-                    "head": {"sha": "abc123"},
-                    "mergeable": True,
-                }
-            if "/comments" in url:
-                if method == "POST":
-                    posted.append(data["body"])
-                    return {"id": 1003}
-                return []
-            return {}
-
-        with patch("mcp_server.api_request", side_effect=mock_api), patch(
-            "mcp_server._authenticated_username",
-            side_effect=tracking_username,
-        ), patch(
-            "mcp_server._resolve",
-            return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
-        ):
-            res = mcp_server.gitea_acquire_reviewer_pr_lease(
-                pr_number=550,
-                worktree="/workspace",
-                candidate_head="abc123",
-                remote="prgs",
-            )
-
-        self.assertTrue(res.get("success"), res)
-        self.assertTrue(res.get("acquired"), res)
-        self.assertEqual(res.get("comment_id"), 1003)
-        self.assertTrue(posted)
-        self.assertIn("gitea.prgs.cc", resolved_hosts)
-        self.assertNotIn("prgs", resolved_hosts)
-
-    def test_adopt_merger_pr_lease_resolves_identity_with_host(self):
-        """Adopt path must not pass a remote alias into identity lookup."""
-        import mcp_server
-        import reviewer_pr_lease
-
-        reviewer_pr_lease.clear_session_lease()
-        head = "a" * 40
-        resolved_hosts = []
-        posted = []
-        reviewer_comment = _reviewer_lease_comment(
-            550,
-            session_id="reviewer-session",
-            head_sha=head,
-            reviewer="reviewer-bot",
-        )
-
-        def tracking_username(host):
-            resolved_hosts.append(host)
-            return None if host in {"prgs", "dadeschools"} else "merger-bot"
-
-        def mock_api(method, url, auth, data=None):
-            if "/pulls/" in url and "/comments" not in url:
-                return {
-                    "number": 550,
-                    "user": {"login": "author-user"},
-                    "state": "open",
-                    "head": {"sha": head},
-                    "mergeable": True,
-                    "merged": False,
-                }
-            if "/comments" in url:
-                if method == "POST":
-                    posted.append(data["body"])
-                    return {"id": 1004}
-                return [reviewer_comment]
-            return {}
-
-        merger_profile = {
-            "profile_name": "prgs-merger",
-            "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"],
-            "forbidden_operations": [],
-        }
-        with patch("mcp_server.get_profile", return_value=merger_profile), patch(
-            "mcp_server.api_request",
-            side_effect=mock_api,
-        ), patch(
-            "mcp_server._authenticated_username",
-            side_effect=tracking_username,
-        ), patch(
-            "mcp_server._resolve",
-            return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
-        ), patch(
-            "mcp_server.gitea_get_pr_review_feedback",
-            return_value={
-                "approval_at_current_head": True,
-                "latest_approved_head_sha": head,
-            },
-        ):
-            res = mcp_server.gitea_adopt_merger_pr_lease(
-                pr_number=550,
-                worktree="/workspace",
-                expected_head_sha=head,
-                remote="prgs",
-            )
-
-        self.assertTrue(res.get("success"), res)
-        self.assertTrue(res.get("adopted"), res)
-        self.assertEqual(res.get("adoption_comment_id"), 1004)
-        self.assertTrue(posted)
-        self.assertIn("gitea.prgs.cc", resolved_hosts)
-        self.assertNotIn("prgs", resolved_hosts)
-
-    def test_release_reviewer_pr_lease_same_identity_cross_session(self):
-        """Same reviewer identity may release a lease claimed by another session.
-
-        Regression: release used to call ``_authenticated_username(remote)``
-        with the remote alias (e.g. ``prgs``) instead of the resolved host,
-        so identity resolved empty and same-identity release fail-closed even
-        when the authenticated user owned the lease.
-        """
-        import mcp_server
-        import reviewer_pr_lease
-
-        mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
-        mcp_server.gitea_load_review_workflow()
-        reviewer_pr_lease.clear_session_lease()
-
-        foreign_session = "46820-dd78cdc4aacc"
-        comments = [
-            _reviewer_lease_comment(
-                457,
-                session_id=foreign_session,
-                head_sha="ef287eaf5841d9e542a4e888ce0ae7d679dbd685",
-                reviewer="sysadmin",
-            )
-        ]
-        posted = []
-        resolved_hosts = []
-
-        def mock_api(method, url, auth, data=None):
-            if "/api/v1/user" in url:
-                return {"login": "sysadmin"}
-            if "/comments" in url:
-                if method == "POST":
-                    new_c = {
-                        "id": 8071,
-                        "body": data["body"],
-                        "user": {"login": "sysadmin"},
-                    }
-                    comments.append(new_c)
-                    posted.append(new_c)
-                    return {"id": 8071}
-                return comments
-            if "/pulls/" in url:
-                return {
-                    "user": {"login": "jcwalker3"},
-                    "state": "open",
-                    "head": {"sha": "ef287eaf5841d9e542a4e888ce0ae7d679dbd685"},
-                    "mergeable": False,
-                }
-            return {}
-
-        def tracking_username(host):
-            resolved_hosts.append(host)
-            # Empty identity when passed a remote alias (the pre-fix bug path).
-            if host in {"prgs", "dadeschools"}:
-                return None
-            return "sysadmin"
-
-        with patch("mcp_server.api_request", side_effect=mock_api), patch(
-            "mcp_server._authenticated_username", side_effect=tracking_username
-        ), patch(
-            "mcp_server._resolve",
-            return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
-        ), patch("mcp_server._auth", return_value={"Authorization": "token x"}), patch(
-            "mcp_server._verify_role_mutation_workspace", return_value=None
-        ):
-            res = mcp_server.gitea_release_reviewer_pr_lease(
-                pr_number=457, remote="prgs"
-            )
-
-        self.assertTrue(res.get("success"), res)
-        self.assertTrue(res.get("released"), res)
-        self.assertEqual(res.get("comment_id"), 8071)
-        self.assertTrue(posted)
-        self.assertIn("phase: released", posted[0]["body"])
-        # Must resolve identity with host, never the remote alias alone.
-        self.assertIn("gitea.prgs.cc", resolved_hosts)
-        self.assertNotIn("prgs", resolved_hosts)

tests/test_mcp_stale_runtime.py

@@ -67,66 +67,5 @@ class TestMcpStaleRuntime(unittest.TestCase):
         # But does NOT contain missing author profile error
         self.assertFalse(any("None of the matching profiles for task" in r for r in reasons_author))
 
-    @patch("subprocess.run")
-    @patch("os.path.getmtime")
-    @patch("os.path.exists")
-    @patch("os.getpid")
-    @patch.dict("os.environ", {"GITEA_FORCE_MCP_RUNTIME_CHECK": "1"})
-    def test_git_head_change_triggers_staleness(self, mock_getpid, mock_exists, mock_getmtime, mock_run):
-        # Set boot SHA to FAKE1 and current to FAKE2
-        gitea_mcp_server._process_boot_head_sha = "FAKE1"
-        mock_getpid.return_value = 12345
-        mock_exists.return_value = True
-
-        # Code time is in the past (fresh process chronologically)
-        code_time = datetime(2026, 7, 8, 11, 0, 0)
-        mock_getmtime.return_value = code_time.timestamp()
-
-        ps_output = (
-            "  PID LSTART COMMAND\n"
-            "12345 Wed Jul  8 12:00:00 2026 /path/to/python mcp_server.py\n"
-        )
-
-        mock_run_ps = MagicMock()
-        mock_run_ps.stdout = ps_output
-
-        mock_run_env = MagicMock()
-        mock_run_env.stdout = "GITEA_MCP_PROFILE=prgs-author"
-
-        mock_run_git = MagicMock()
-        mock_run_git.stdout = "FAKE2" # different SHA
-
-        def side_effect(args, **kwargs):
-            if args[0] == "ps" and "eww" in args:
-                return mock_run_env
-            elif args[0] == "ps":
-                return mock_run_ps
-            elif args[0] == "git" and "rev-parse" in args:
-                return mock_run_git
-            raise ValueError(f"Unexpected: {args}")
-
-        mock_run.side_effect = side_effect
-
-        # Stale even though process started AFTER code mtime, because git HEAD changed
-        reasons = gitea_mcp_server._check_mcp_runtimes_diagnostics("create_issue", ["prgs-author"])
-        self.assertTrue(any("stale-runtime: The active Gitea MCP server process is stale" in r for r in reasons))
-
-    @patch("threading.Thread")
-    @patch("os.utime")
-    @patch("os.path.exists")
-    @patch.dict("os.environ", {"MCP_CONFIG_PATH": "/tmp/mcp_config.json"})
-    def test_auto_restart_trigger_touches_and_spawns(self, mock_exists, mock_utime, mock_thread):
-        mock_exists.return_value = True
-        gitea_mcp_server._restart_triggered = False
-
-        # Ensure we are not skipped in test mode for testing purposes
-        with patch("gitea_mcp_server._preflight_in_test_mode", return_value=False):
-            gitea_mcp_server._trigger_mcp_auto_restart()
-
-        mock_utime.assert_called_once_with("/tmp/mcp_config.json", None)
-        mock_thread.assert_called_once()
-        self.assertTrue(gitea_mcp_server._restart_triggered)
-
-
 if __name__ == "__main__":
     unittest.main()

tests/test_merger_lease_adoption.py

@@ -42,96 +42,6 @@ def _lease_comment(
 HEAD = "f" * 40
 
 
-class TestDescribeSessionLeaseProof(unittest.TestCase):
-    def setUp(self):
-        leases.clear_session_lease()
-
-    def test_empty_session_is_none_kind(self):
-        proof = mla.describe_session_lease_proof(None)
-        self.assertIsNone(proof["lease_proof_source"])
-        self.assertEqual(proof["lease_proof_kind"], "none")
-        self.assertFalse(proof["lease_proof_sanctioned"])
-
-    def test_manual_seed_without_provenance(self):
-        leases.record_session_lease({
-            "pr_number": 536,
-            "session_id": "manual",
-            "comment_id": 1,
-        })
-        proof = mla.describe_session_lease_proof(leases.get_session_lease())
-        self.assertIsNone(proof["lease_proof_source"])
-        self.assertEqual(proof["lease_proof_kind"], "unsanctioned_manual_seed")
-        self.assertFalse(proof["lease_proof_sanctioned"])
-
-    def test_sanctioned_adopt_fields(self):
-        leases.record_session_lease(
-            {
-                "pr_number": 536,
-                "session_id": "merger-sess",
-                "comment_id": 700,
-            },
-            lease_provenance=mla.build_lease_provenance(
-                source=mla.SOURCE_ADOPT,
-                comment_id=700,
-                adopted_from_session_id="reviewer-sess",
-                adoption_reason=mla.DEFAULT_ADOPTION_REASON,
-            ),
-        )
-        proof = mla.describe_session_lease_proof(leases.get_session_lease())
-        self.assertEqual(proof["lease_proof_source"], mla.SOURCE_ADOPT)
-        self.assertEqual(proof["lease_proof_kind"], "sanctioned_adoption")
-        self.assertTrue(proof["lease_proof_sanctioned"])
-        self.assertEqual(
-            proof["lease_recovery_reason"], mla.DEFAULT_ADOPTION_REASON
-        )
-        self.assertEqual(proof["lease_proof_comment_id"], 700)
-        self.assertEqual(proof["lease_adopted_from_session_id"], "reviewer-sess")
-
-    def test_sanctioned_acquire_fields(self):
-        leases.record_session_lease(
-            {
-                "pr_number": 536,
-                "session_id": "rev",
-                "comment_id": 50,
-            },
-            lease_provenance=mla.build_lease_provenance(
-                source=mla.SOURCE_ACQUIRE,
-                comment_id=50,
-            ),
-        )
-        proof = mla.describe_session_lease_proof(leases.get_session_lease())
-        self.assertEqual(proof["lease_proof_source"], mla.SOURCE_ACQUIRE)
-        self.assertEqual(proof["lease_proof_kind"], "sanctioned_acquire")
-        self.assertTrue(proof["lease_proof_sanctioned"])
-
-
-class TestManualLeaseProofHandoffAudit(unittest.TestCase):
-    def test_clean_merge_report_passes(self):
-        text = (
-            "Merged via gitea_merge_pr after gitea_adopt_merger_pr_lease. "
-            "lease_proof_source: gitea_adopt_merger_pr_lease"
-        )
-        result = mla.assess_manual_lease_proof_handoff(text)
-        self.assertTrue(result["proven"])
-        self.assertFalse(result["block"])
-
-    def test_manual_seeded_claim_without_mcp_evidence_blocks(self):
-        text = "Merged using manually seeded lease proof from prior session."
-        result = mla.assess_manual_lease_proof_handoff(text)
-        self.assertFalse(result["proven"])
-        self.assertTrue(result["block"])
-        self.assertTrue(any("#535" in r for r in result["reasons"]))
-
-    def test_historical_manual_mention_allowed_with_adoption_evidence(self):
-        text = (
-            "Historical incident used manually seeded lease; this merge used "
-            "gitea_adopt_merger_pr_lease with adoption_comment_id: 8288."
-        )
-        result = mla.assess_manual_lease_proof_handoff(text)
-        self.assertTrue(result["proven"])
-        self.assertFalse(result["block"])
-
-
 class TestSanctionedProvenance(unittest.TestCase):
     def setUp(self):
         leases.clear_session_lease()

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_operator_guide.py

@@ -88,21 +88,6 @@ class TestControlPlaneGuide(GuideTestBase):
         blob = " ".join(g["guidance"]).lower()
         self.assertIn("forbidden", blob)
         self.assertIn("review", blob)
-        self.assertIn("resolved_repository", g)
-
-    @patch("mcp_server.api_request", return_value={"login": "author-bot"})
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_guide_explicit_org_repo_parameters(self, _auth, _api):
-        with patch.dict(os.environ, AUTHOR_ENV, clear=True):
-            g = mcp_get_control_plane_guide(
-                remote="prgs",
-                org="Scaled-Tech-Consulting",
-                repo="Gitea-Tools",
-            )
-        self.assertEqual(
-            g["resolved_repository"],
-            {"org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools"},
-        )
 
     @patch("mcp_server.api_request", return_value={"login": "reviewer-bot"})
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)

tests/test_post_merge_validation.py

@@ -1,119 +0,0 @@
-"""Post-merge moot validation wording + label tests (#529, criteria 1/4/5)."""
-
-import unittest
-
-from post_merge_validation import (
-    BASELINE_ACCEPTED_OUTCOME,
-    BLOCKED_OUTCOME,
-    CLEAN_PASS_OUTCOME,
-    LABEL_BASELINE_ACCEPTED,
-    LABEL_BLOCKED,
-    LABEL_CLEAN_PASS,
-    LABEL_POST_MERGE_MOOT,
-    POST_MERGE_MOOT_OUTCOME,
-    assess_post_merge_validation,
-    process_state_label,
-)
-from final_report_validator import assess_final_report_validator
-from issue_workflow_labels import CANONICAL_LABELS, VALIDATION_LABELS
-
-_MOOT_PROOF = (
-    "Validation status: post-merge moot validation\n"
-    "PR state: merged\n"
-    "Merge commit sha: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
-    "Validation finding: full suite passed, for the record.\n"
-)
-
-
-class TestPostMergeValidation(unittest.TestCase):
-    def test_not_applicable_when_no_merged_or_moot_signal(self):
-        result = assess_post_merge_validation(
-            "Review decision: approve. Validation: full suite passed."
-        )
-        self.assertFalse(result["block"])
-        self.assertTrue(result["skipped"])
-
-    def test_active_approval_on_merged_pr_blocked(self):
-        report = (
-            "PR is already merged. Review decision: approve. "
-            "Validation: full suite passed."
-        )
-        result = assess_post_merge_validation(report)
-        self.assertTrue(result["block"])
-        self.assertEqual(result["outcome"], POST_MERGE_MOOT_OUTCOME)
-        self.assertEqual(result["process_state_label"], LABEL_POST_MERGE_MOOT)
-
-    def test_active_approval_on_merged_pr_via_structured_signal_blocked(self):
-        report = "Review decision: approve. Validation: full suite passed."
-        result = assess_post_merge_validation(report, pr_merged_or_closed=True)
-        self.assertTrue(result["block"])
-
-    def test_moot_wording_without_proof_blocked(self):
-        report = "Recording post-merge moot validation for this PR."
-        result = assess_post_merge_validation(report)
-        self.assertTrue(result["block"])
-        self.assertTrue(result["reasons"])
-
-    def test_moot_wording_with_full_proof_allowed(self):
-        result = assess_post_merge_validation(_MOOT_PROOF)
-        self.assertFalse(result["block"])
-        self.assertEqual(result["outcome"], POST_MERGE_MOOT_OUTCOME)
-
-    def test_merged_signal_only_guides_without_blocking(self):
-        result = assess_post_merge_validation(
-            "PR was already merged before review; recording findings."
-        )
-        self.assertFalse(result["block"])
-        self.assertTrue(result["safe_next_action"])
-
-    def test_process_state_label_mapping(self):
-        self.assertEqual(process_state_label(CLEAN_PASS_OUTCOME), LABEL_CLEAN_PASS)
-        self.assertEqual(
-            process_state_label(BASELINE_ACCEPTED_OUTCOME), LABEL_BASELINE_ACCEPTED
-        )
-        self.assertEqual(process_state_label(BLOCKED_OUTCOME), LABEL_BLOCKED)
-        self.assertEqual(
-            process_state_label(POST_MERGE_MOOT_OUTCOME), LABEL_POST_MERGE_MOOT
-        )
-        self.assertIsNone(process_state_label("nonsense"))
-
-
-class TestValidationLabelsRegistered(unittest.TestCase):
-    def test_validation_labels_are_canonical(self):
-        for label in (
-            LABEL_CLEAN_PASS,
-            LABEL_BASELINE_ACCEPTED,
-            LABEL_BLOCKED,
-            LABEL_POST_MERGE_MOOT,
-        ):
-            self.assertIn(label, VALIDATION_LABELS)
-            self.assertIn(label, CANONICAL_LABELS)
-
-
-class TestPostMergeValidationWiredIntoReview(unittest.TestCase):
-    def test_review_pr_blocks_active_approval_on_merged_pr(self):
-        report = (
-            "PR is already merged. Review decision: approve. "
-            "Validation: full suite passed."
-        )
-        result = assess_final_report_validator(report, "review_pr")
-        self.assertTrue(result["blocked"])
-        self.assertTrue(
-            any(
-                f["rule_id"] == "reviewer.post_merge_validation"
-                for f in result["findings"]
-            )
-        )
-
-    def test_review_pr_allows_proven_post_merge_moot(self):
-        result = assess_final_report_validator(_MOOT_PROOF, "review_pr")
-        self.assertFalse(
-            any(
-                f["rule_id"] == "reviewer.post_merge_validation"
-                for f in result["findings"]
-            )
-        )
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_reconciler_supersession_close.py

@@ -1,245 +0,0 @@
-"""Issue #525: reconciler supersession close path."""
-
-import sys
-import unittest
-from unittest.mock import patch
-
-sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
-
-import mcp_server  # noqa: E402
-import role_session_router  # noqa: E402
-import task_capability_map  # noqa: E402
-from review_proofs import assess_reconciler_supersession_close_gate  # noqa: E402
-
-
-HEAD = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
-MERGE = "1111111111111111111111111111111111111111"
-TARGET = "2222222222222222222222222222222222222222"
-
-
-def _target_pr(mergeable=False):
-    return {
-        "number": 490,
-        "state": "open",
-        "mergeable": mergeable,
-        "head": {"sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
-    }
-
-
-def _superseding_pr():
-    return {
-        "number": 502,
-        "state": "closed",
-        "merged": True,
-        "merge_commit_sha": MERGE,
-        "head": {"sha": HEAD},
-    }
-
-
-class TestReconcilerSupersessionCapabilityMap(unittest.TestCase):
-    def test_supersession_tasks_route_to_reconciler(self):
-        self.assertEqual(
-            task_capability_map.required_permission(
-                "reconcile_close_superseded_pr"),
-            "gitea.pr.close",
-        )
-        self.assertEqual(
-            task_capability_map.required_role("reconcile_close_superseded_pr"),
-            "reconciler",
-        )
-        self.assertEqual(
-            task_capability_map.required_permission(
-                "reconcile_close_satisfied_issue"),
-            "gitea.issue.close",
-        )
-        self.assertEqual(
-            task_capability_map.required_role("reconcile_create_followup_issue"),
-            "reconciler",
-        )
-
-    def test_author_session_cannot_route_supersession_close(self):
-        result = role_session_router.route_task_session(
-            "reconcile_close_superseded_pr",
-            active_profile="prgs-author",
-            active_role_kind="author",
-            allowed_in_current_session=False,
-        )
-        self.assertEqual(result["route_result"], "wrong_role_stop")
-        self.assertFalse(result["downstream_allowed"])
-
-
-class TestReconcilerSupersessionCloseGate(unittest.TestCase):
-    def _gate(self, **overrides):
-        kwargs = {
-            "target_pr_number": 490,
-            "target_pr_state": "open",
-            "target_pr_mergeable": False,
-            "target_independently_required": False,
-            "superseding_pr_number": 502,
-            "superseding_pr_state": "closed",
-            "superseding_pr_merged": True,
-            "superseding_merge_commit_sha": MERGE,
-            "superseding_head_sha": HEAD,
-            "target_branch": "master",
-            "target_branch_sha": TARGET,
-            "superseding_head_is_ancestor_of_target": True,
-            "canonical_comment_valid": True,
-            "canonical_comment_mentions_superseding_pr": True,
-            "canonical_comment_mentions_merge_commit": True,
-            "close_pr_capability": True,
-            "close_issue_capability": True,
-            "target_issue_state": "open",
-            "issue_satisfied_by_superseding": True,
-        }
-        kwargs.update(overrides)
-        return assess_reconciler_supersession_close_gate(**kwargs)
-
-    def test_supersession_close_allowed_with_full_proof(self):
-        result = self._gate()
-        self.assertEqual(result["outcome"], "SUPERSESSION_CLOSE_ALLOWED")
-        self.assertTrue(result["pr_close_allowed"])
-        self.assertTrue(result["issue_close_allowed"])
-        self.assertFalse(result["review_allowed"])
-        self.assertFalse(result["merge_allowed"])
-
-    def test_mergeable_target_still_blocks(self):
-        result = self._gate(target_pr_mergeable=True)
-        self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
-        self.assertFalse(result["pr_close_allowed"])
-        self.assertTrue(any("mergeable" in r for r in result["reasons"]))
-
-    def test_requires_canonical_comment_with_merge_commit(self):
-        result = self._gate(canonical_comment_mentions_merge_commit=False)
-        self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
-        self.assertFalse(result["pr_close_allowed"])
-        self.assertTrue(any("merge commit" in r for r in result["reasons"]))
-
-    def test_superseding_head_must_be_on_target_branch(self):
-        result = self._gate(superseding_head_is_ancestor_of_target=False)
-        self.assertEqual(result["outcome"], "SUPERSEDING_PR_NOT_ON_TARGET")
-        self.assertFalse(result["pr_close_allowed"])
-
-    def test_close_capability_required(self):
-        result = self._gate(close_pr_capability=False)
-        self.assertEqual(result["outcome"], "RECOVERY_HANDOFF_REQUIRED")
-        self.assertFalse(result["pr_close_allowed"])
-
-
-class TestReconcilerSupersessionMcpTool(unittest.TestCase):
-    def _comment(self):
-        return f"""## Canonical PR State
-STATE: superseded
-NEXT_ACTION: Close superseded PR #490 and issue #470; PR #502 is canonical.
-NEXT_ACTOR: prgs-reconciler
-SUMMARY: PR #490 is superseded by merged PR #502 at merge commit {MERGE}.
-CANONICAL_ITEM: PR #502
-SUPERSEDED_ITEM: PR #490
-CLOSE_OR_KEEP_OPEN: close superseded PR #490 and satisfied issue #470
-"""
-
-    @patch("mcp_server.verify_preflight_purity")
-    @patch("mcp_server._canonical_comment_gate")
-    @patch("already_landed_reconcile.is_head_ancestor_of_ref", return_value=True)
-    @patch("already_landed_reconcile.fetch_target_branch")
-    @patch("mcp_server._profile_operation_gate", return_value=[])
-    @patch("mcp_server._auth", return_value="token test")
-    @patch("mcp_server.api_request")
-    def test_tool_posts_comment_and_closes_superseded_pr_issue(
-        self,
-        api,
-        _auth,
-        _profile_gate,
-        fetch_target,
-        _ancestor,
-        canonical_gate,
-        verify_preflight,
-    ):
-        fetch_target.return_value = {
-            "success": True,
-            "target_branch": "master",
-            "target_ref": "prgs/master",
-            "target_branch_sha": TARGET,
-        }
-        canonical_gate.return_value = {
-            "blocked": False,
-            "canonical_comment_validation": {"valid": True},
-        }
-        api.side_effect = [
-            _target_pr(),
-            _superseding_pr(),
-            {"number": 470, "state": "open"},
-            {"id": 123},
-            {"state": "closed"},
-            {"state": "closed"},
-        ]
-
-        result = mcp_server.gitea_reconcile_superseded_by_merged_pr(
-            target_pr_number=490,
-            superseding_pr_number=502,
-            target_issue_number=470,
-            target_independently_required=False,
-            issue_satisfied_by_superseding=True,
-            close_pr=True,
-            close_issue=True,
-            post_comment=True,
-            comment_body=self._comment(),
-            remote="prgs",
-            org="Scaled-Tech-Consulting",
-            repo="Gitea-Tools",
-        )
-
-        self.assertTrue(result["success"])
-        self.assertTrue(result["performed"])
-        self.assertTrue(result["pr_comment_posted"])
-        self.assertTrue(result["pr_closed"])
-        self.assertTrue(result["issue_closed"])
-        verify_preflight.assert_called_once_with(
-            "prgs", task="reconcile_close_superseded_pr")
-
-    @patch("mcp_server.verify_preflight_purity")
-    @patch("mcp_server._canonical_comment_gate")
-    @patch("already_landed_reconcile.is_head_ancestor_of_ref", return_value=True)
-    @patch("already_landed_reconcile.fetch_target_branch")
-    @patch("mcp_server._profile_operation_gate", return_value=[])
-    @patch("mcp_server._auth", return_value="token test")
-    @patch("mcp_server.api_request")
-    def test_tool_does_not_mutate_when_target_still_mergeable(
-        self,
-        api,
-        _auth,
-        _profile_gate,
-        fetch_target,
-        _ancestor,
-        canonical_gate,
-        verify_preflight,
-    ):
-        fetch_target.return_value = {
-            "success": True,
-            "target_branch": "master",
-            "target_ref": "prgs/master",
-            "target_branch_sha": TARGET,
-        }
-        canonical_gate.return_value = {
-            "blocked": False,
-            "canonical_comment_validation": {"valid": True},
-        }
-        api.side_effect = [_target_pr(mergeable=True), _superseding_pr()]
-
-        result = mcp_server.gitea_reconcile_superseded_by_merged_pr(
-            target_pr_number=490,
-            superseding_pr_number=502,
-            target_independently_required=False,
-            close_pr=True,
-            post_comment=True,
-            comment_body=self._comment(),
-            remote="prgs",
-            org="Scaled-Tech-Consulting",
-            repo="Gitea-Tools",
-        )
-
-        self.assertFalse(result["success"])
-        self.assertFalse(result["performed"])
-        self.assertFalse(result["pr_closed"])
-        self.assertFalse(result["pr_comment_posted"])
-        verify_preflight.assert_not_called()
-        self.assertEqual(api.call_count, 2)

tests/test_remote_repo_guard.py

@@ -111,23 +111,6 @@ class TestAssessRemoteRepoMatch(unittest.TestCase):
         self.assertIn("Gitea-Tools", message)
         self.assertIn("org=", message)
         self.assertIn("repo=", message)
-        # Recovery text must name a tool that actually accepts org/repo (#530 follow-up).
-        self.assertIn("mcp_get_control_plane_guide", message)
-        self.assertIn("schema does not list", message)
-
-    def test_parse_org_repo_from_https_url(self):
-        parsed = remote_repo_guard.parse_org_repo_from_remote_url(LOCAL_GITEA_TOOLS_URL)
-        self.assertEqual(parsed, ("Scaled-Tech-Consulting", "Gitea-Tools"))
-
-    def test_parse_org_repo_from_ssh_url(self):
-        parsed = remote_repo_guard.parse_org_repo_from_remote_url(
-            "[email protected]:Scaled-Tech-Consulting/Gitea-Tools.git"
-        )
-        self.assertEqual(parsed, ("Scaled-Tech-Consulting", "Gitea-Tools"))
-
-    def test_parse_org_repo_empty(self):
-        self.assertIsNone(remote_repo_guard.parse_org_repo_from_remote_url(None))
-        self.assertIsNone(remote_repo_guard.parse_org_repo_from_remote_url(""))
 
 
 class TestResolveServerWiring(unittest.TestCase):
@@ -190,82 +173,6 @@ class TestResolveServerWiring(unittest.TestCase):
         with self.assertRaises(RuntimeError):
             self.server.gitea_view_issue(issue_number=1, remote="prgs")
 
-    def test_control_plane_guide_accepts_explicit_gitea_tools(self):
-        """Guide schema accepts org/repo and resolves Gitea-Tools, not Timesheet."""
-        self._set_prgs_default_repo("Timesheet")
-        with mock.patch.object(
-            self.server, "api_request", return_value={"login": "guide-bot"}
-        ), mock.patch.object(
-            self.server, "get_auth_header", return_value="Basic dGVzdA=="
-        ), mock.patch.dict(
-            os.environ,
-            {
-                "GITEA_PROFILE_NAME": "author-test",
-                "GITEA_ALLOWED_OPERATIONS": "gitea.read",
-                "GITEA_FORBIDDEN_OPERATIONS": "gitea.pr.merge",
-            },
-            clear=False,
-        ):
-            guide = self.server.mcp_get_control_plane_guide(
-                remote="prgs",
-                org="Scaled-Tech-Consulting",
-                repo="Gitea-Tools",
-            )
-        self.assertTrue(guide["read_only"])
-        self.assertEqual(
-            guide["resolved_repository"],
-            {"org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools"},
-        )
-        self.assertNotEqual(guide["resolved_repository"]["repo"], "Timesheet")
-
-    def test_control_plane_guide_bare_remote_aligns_to_local_gitea_tools(self):
-        """Bare remote=prgs in a Gitea-Tools worktree must not stay on Timesheet."""
-        self._set_prgs_default_repo("Timesheet")
-        with mock.patch.object(
-            self.server, "api_request", return_value={"login": "guide-bot"}
-        ), mock.patch.object(
-            self.server, "get_auth_header", return_value="Basic dGVzdA=="
-        ), mock.patch.dict(
-            os.environ,
-            {
-                "GITEA_PROFILE_NAME": "author-test",
-                "GITEA_ALLOWED_OPERATIONS": "gitea.read",
-                "GITEA_FORBIDDEN_OPERATIONS": "gitea.pr.merge",
-            },
-            clear=False,
-        ):
-            guide = self.server.mcp_get_control_plane_guide(remote="prgs")
-        self.assertEqual(guide["resolved_repository"]["repo"], "Gitea-Tools")
-        self.assertEqual(
-            guide["resolved_repository"]["org"], "Scaled-Tech-Consulting"
-        )
-
-    def test_control_plane_guide_wrong_default_fails_closed_without_local(self):
-        """Without local remote context, mismatched defaults still fail closed."""
-        self._set_prgs_default_repo("Timesheet")
-        with mock.patch.object(
-            self.server, "_local_git_remote_url", return_value=None
-        ):
-            # No local URL → preference cannot recover; bare resolve uses Timesheet
-            # and guard is skipped only because local URL is missing (best-effort).
-            host, org, repo = self.server._resolve_control_plane_guide_target(
-                "prgs", None, None, None
-            )
-        self.assertEqual(repo, "Timesheet")
-
-    def test_control_plane_guide_schema_recovery_matches_parameters(self):
-        """Remediation must only recommend parameters the guide actually accepts."""
-        import inspect
-
-        sig = inspect.signature(self.server.mcp_get_control_plane_guide)
-        self.assertIn("org", sig.parameters)
-        self.assertIn("repo", sig.parameters)
-        self.assertIn("remote", sig.parameters)
-        remediation = remote_repo_guard.REMEDIATION
-        self.assertIn("mcp_get_control_plane_guide", remediation)
-        self.assertIn("org=", remediation)
-        self.assertIn("repo=", remediation)
-
 
 if __name__ == "__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):
@@ -287,49 +246,6 @@ class TestResolveTaskCapability(unittest.TestCase):
             self.assertEqual(res["required_operation_permission"], "gitea.pr.merge")
             self.assertEqual(res["required_role_kind"], "merger")
 
-    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
-    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
-    def test_reviewer_role_with_merge_permission_still_cannot_merge(self, _auth, _api):
-        # #539: exact permission alone is insufficient. The active profile
-        # role must also be merger for merge_pr.
-        config = json.loads(json.dumps(CONFIG_RESOLVER))
-        config["profiles"]["reviewer-with-merge"] = {
-            "enabled": True,
-            "context": "ctx",
-            "role": "reviewer",
-            "username": "reviewer-user",
-            "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
-            "allowed_operations": [
-                "gitea.read", "gitea.pr.review", "gitea.pr.approve",
-                "gitea.pr.merge",
-            ],
-            "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
-            "execution_profile": "reviewer-with-merge",
-        }
-        self._write_config(config)
-        with patch.dict(os.environ, self._env("reviewer-with-merge")):
-            res = mcp_server.gitea_resolve_task_capability(
-                task="merge_pr", remote="prgs"
-            )
-        self.assertEqual(res["required_role_kind"], "merger")
-        self.assertEqual(res["active_role_kind"], "reviewer")
-        self.assertTrue(res["active_profile_permission_allowed"])
-        self.assertFalse(res["allowed_in_current_session"])
-        self.assertTrue(res["stop_required"])
-        self.assertIn("cannot perform merger task", " ".join(res["task_role_guidance"]))
-
-    @patch("mcp_server.api_request", return_value={"login": "merger-user"})
-    @patch("mcp_server.get_auth_header", return_value="token merger-pass")
-    def test_merger_profile_can_resolve_merge_task(self, _auth, _api):
-        with patch.dict(os.environ, self._env("merger-profile")):
-            res = mcp_server.gitea_resolve_task_capability(
-                task="merge_pr", remote="prgs"
-            )
-        self.assertEqual(res["required_role_kind"], "merger")
-        self.assertEqual(res["active_role_kind"], "merger")
-        self.assertTrue(res["allowed_in_current_session"])
-        self.assertFalse(res["stop_required"])
-
     @patch("mcp_server.api_request")
     def test_self_review_blocked_structured(self, mock_api):
         # Test that self-approve (self-review gate) is blocked and returns structured reasons

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()
@@ -237,113 +198,5 @@ class TestReviewerLeaseMcpGate(unittest.TestCase):
         self.assertTrue(any("lease" in r.lower() for r in reasons))
 
 
-
-class TestReviewerLeaseHandoffDiagnose(unittest.TestCase):
-    """#599: canonical next-action diagnosis for open-PR lease handoff."""
-
-    def setUp(self):
-        leases.clear_session_lease()
-
-    def test_no_lease_next_action_acquire(self):
-        result = leases.diagnose_reviewer_pr_lease_handoff(
-            [],
-            pr_number=592,
-            current_session_id="me-1",
-            current_reviewer_identity="sysadmin",
-            proposed_worktree="branches/review-pr-592",
-        )
-        self.assertEqual(result["classification"], "no_lease")
-        self.assertEqual(result["next_action"], leases.NEXT_ACTION_ACQUIRE)
-        self.assertIsNone(result["active_lease"])
-
-    def test_foreign_active_wait_pr592_style(self):
-        # Instructed lease gone; newer foreign claim is active.
-        old = _lease_comment(592, "23139-da018dab43ba", phase="released")
-        old["id"] = 8640
-        active = _lease_comment(592, "51515-2bfb70f0685f", phase="claimed")
-        active["id"] = 8647
-        result = leases.diagnose_reviewer_pr_lease_handoff(
-            [old, active],
-            pr_number=592,
-            current_session_id="fresh-session",
-            current_reviewer_identity="sysadmin",
-            proposed_worktree="branches/review-pr-592-issue-590",
-            instructed_session_id="23139-da018dab43ba",
-            instructed_comment_id=8640,
-        )
-        self.assertEqual(
-            result["classification"],
-            "instructed_lease_missing_with_replacement",
-        )
-        self.assertEqual(result["next_action"], leases.NEXT_ACTION_WAIT)
-        self.assertTrue(result["instructed_lease_missing_with_replacement"])
-        self.assertEqual(result["active_lease"]["session_id"], "51515-2bfb70f0685f")
-        self.assertEqual(result["active_lease"]["comment_id"], 8647)
-        self.assertFalse(result["mutation_allowed"])
-
-    def test_foreign_reclaimable_release_expired(self):
-        reclaim = _lease_comment(
-            592, "foreign-old", phase="claimed", minutes_ago=65
-        )
-        reclaim["id"] = 99
-        result = leases.diagnose_reviewer_pr_lease_handoff(
-            [reclaim],
-            pr_number=592,
-            current_session_id="me-1",
-            current_reviewer_identity="sysadmin",
-            proposed_worktree="branches/review-pr-592",
-        )
-        self.assertEqual(result["classification"], "foreign_reclaimable")
-        self.assertEqual(
-            result["next_action"], leases.NEXT_ACTION_RELEASE_EXPIRED_LEASE
-        )
-
-    def test_own_active_resume(self):
-        head = "a" * 40
-        claim = _lease_comment(592, "me-1", phase="claimed", candidate_head=head)
-        claim["id"] = 10
-        leases.record_session_lease({
-            "pr_number": 592,
-            "session_id": "me-1",
-            "candidate_head": head,
-            "comment_id": 10,
-        }, lease_provenance=mla.build_lease_provenance(
-            source=mla.SOURCE_ACQUIRE,
-            comment_id=10,
-        ))
-        result = leases.diagnose_reviewer_pr_lease_handoff(
-            [claim],
-            pr_number=592,
-            current_session_id="me-1",
-            current_reviewer_identity="rev1",
-            proposed_worktree="branches/review-pr382",
-        )
-        self.assertEqual(result["classification"], "own_active")
-        self.assertEqual(
-            result["next_action"], leases.NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
-        )
-        self.assertTrue(result["mutation_allowed"])
-
-    def test_worktree_binding_mismatch(self):
-        claim = _lease_comment(592, "me-1", phase="claimed")
-        claim["id"] = 11
-        # _lease_comment uses branches/review-pr382
-        result = leases.diagnose_reviewer_pr_lease_handoff(
-            [claim],
-            pr_number=592,
-            current_session_id="me-1",
-            current_reviewer_identity="rev1",
-            proposed_worktree="branches/review-pr-592-issue-590",
-            env_bound_worktree="branches/review-pr-538",
-        )
-        self.assertEqual(result["classification"], "worktree_binding_mismatch")
-        self.assertEqual(
-            result["next_action"], leases.NEXT_ACTION_REPAIR_WORKTREE_BINDING
-        )
-        self.assertFalse(result["worktree_binding"]["match"])
-        self.assertFalse(result["mutation_allowed"])
-
-
-
 if __name__ == "__main__":
     unittest.main()

tests/test_role_session_router.py

@@ -52,20 +52,6 @@ CONFIG = {
             ],
             "execution_profile": "prgs-reviewer",
         },
-        "prgs-merger": {
-            "enabled": True,
-            "context": "ctx",
-            "role": "merger",
-            "username": "sysadmin",
-            "auth": {"type": "env", "name": "GITEA_TOKEN_MERGER"},
-            "allowed_operations": [
-                "gitea.read", "gitea.pr.merge", "gitea.issue.comment",
-            ],
-            "forbidden_operations": [
-                "gitea.pr.create", "gitea.branch.push", "gitea.pr.approve",
-            ],
-            "execution_profile": "prgs-merger",
-        },
         "prgs-reconciler": {
             "enabled": True,
             "context": "ctx",
@@ -118,7 +104,6 @@ class TestRoleSessionRouter(unittest.TestCase):
             "GITEA_MCP_PROFILE": profile,
             "GITEA_TOKEN_AUTHOR": "author-pass",
             "GITEA_TOKEN_REVIEWER": "reviewer-pass",
-            "GITEA_TOKEN_MERGER": "merger-pass",
             "GITEA_TOKEN_RECONCILER": "reconciler-pass",
         }
 
@@ -242,31 +227,6 @@ class TestRoleSessionRouter(unittest.TestCase):
         self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
         self.assertTrue(route["downstream_allowed"])
 
-    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
-    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
-    def test_merge_task_under_reviewer_profile_wrong_role_stop(self, _auth, _api):
-        # #539: a reviewer session must not pass the pre-task merge route
-        # even if its config accidentally includes gitea.pr.merge.
-        with patch.dict(os.environ, self._env("prgs-reviewer")):
-            route = mcp_server.gitea_route_task_session(
-                task_type="merge_pr", remote="prgs"
-            )
-        self.assertEqual(route["required_role"], "merger")
-        self.assertEqual(route["route_result"], role_session_router.ROUTE_WRONG_ROLE)
-        self.assertFalse(route["downstream_allowed"])
-        self.assertIn("merger", route["message"].lower())
-
-    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
-    @patch("mcp_server.get_auth_header", return_value="token merger-pass")
-    def test_merge_task_under_merger_profile_allowed(self, _auth, _api):
-        with patch.dict(os.environ, self._env("prgs-merger")):
-            route = mcp_server.gitea_route_task_session(
-                task_type="merge_pr", remote="prgs"
-            )
-        self.assertEqual(route["required_role"], "merger")
-        self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
-        self.assertTrue(route["downstream_allowed"])
-
     @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
     @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
     def test_author_task_under_reviewer_profile_routes_to_author(self, _auth, _api):
@@ -374,4 +334,4 @@ class TestCheckMidMerge(unittest.TestCase):
 
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()

tests/test_session_state_isolation.py

@@ -1,62 +0,0 @@
-"""Regression: durable session-state isolation survives env clears (#590)."""
-
-from __future__ import annotations
-
-import os
-from pathlib import Path
-from unittest.mock import patch
-
-import mcp_session_state
-
-
-def test_default_state_dir_survives_env_clear():
-    """conftest pins default_state_dir so clear=True cannot hit host cache."""
-    isolated = os.environ.get("GITEA_MCP_SESSION_STATE_DIR")
-    assert isolated, "conftest must set GITEA_MCP_SESSION_STATE_DIR"
-    host_cache = Path.home() / ".cache" / "gitea-tools" / "session-state"
-    with patch.dict(os.environ, {}, clear=True):
-        assert os.environ.get("GITEA_MCP_SESSION_STATE_DIR") is None
-        resolved = mcp_session_state.default_state_dir()
-        assert resolved == isolated
-        assert Path(resolved).resolve() != host_cache.resolve()
-
-
-def test_decision_lock_save_load_stays_in_isolated_dir():
-    """Review-mutation ledger files land only under the per-test state dir."""
-    isolated = os.environ.get("GITEA_MCP_SESSION_STATE_DIR")
-    assert isolated
-    with patch.dict(os.environ, {}, clear=True):
-        saved = mcp_session_state.save_state(
-            kind=mcp_session_state.KIND_DECISION_LOCK,
-            profile_identity="gitea-default",
-            payload={
-                "live_mutations": [
-                    {"pr_number": 1, "action": "request_changes"}
-                ],
-            },
-        )
-        assert saved is not None
-        path = mcp_session_state.state_file_path(
-            kind=mcp_session_state.KIND_DECISION_LOCK,
-            profile_identity="gitea-default",
-        )
-        assert path.startswith(isolated + os.sep) or path.startswith(
-            isolated + "/"
-        )
-        loaded = mcp_session_state.load_state(
-            kind=mcp_session_state.KIND_DECISION_LOCK,
-            profile_identity="gitea-default",
-        )
-        assert loaded is not None
-        assert loaded.get("live_mutations")
-        host_path = (
-            Path.home()
-            / ".cache"
-            / "gitea-tools"
-            / "session-state"
-            / "review_decision_lock-gitea-default.json"
-        )
-        # Host file may exist from operator use; this write must not refresh it.
-        # Prove our isolated file is distinct and present.
-        assert Path(path).is_file()
-        assert Path(path).resolve() != host_path.resolve()

tests/test_stale_review_decision_lock_cleanup.py

@@ -1,378 +0,0 @@
-"""Stale #332 review-decision lock cleanup (#594).
-
-Proves:
-  a. active terminal locks still block unrelated terminal reviews
-  b. merged/closed PR locks can be cleared canonically
-  c. cleanup cannot bypass review gates on open PRs
-  d. after moot cleanup, mark_ready for a different PR can proceed
-     (PR #592-style after PR #586-style stale approve)
-"""
-
-from __future__ import annotations
-
-import os
-import unittest
-from unittest.mock import patch
-
-import mcp_server
-import stale_review_decision_lock as srdl
-from mcp_server import (
-    gitea_cleanup_stale_review_decision_lock,
-    gitea_mark_final_review_decision,
-)
-
-HEAD = "a" * 40
-FAKE_AUTH = "Basic dGVzdDp0ZXN0"
-
-REVIEWER_ENV = {
-    "GITEA_PROFILE_NAME": "prgs-reviewer",
-    "GITEA_ALLOWED_OPERATIONS": (
-        "gitea.read,gitea.pr.review,gitea.pr.approve,"
-        "gitea.pr.request_changes,gitea.pr.comment"
-    ),
-    "GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer",
-}
-
-REVIEWER_PROFILE = {
-    "profile_name": "prgs-reviewer",
-    "allowed_operations": [
-        "gitea.read",
-        "gitea.pr.review",
-        "gitea.pr.approve",
-        "gitea.pr.request_changes",
-        "gitea.pr.comment",
-    ],
-    "forbidden_operations": [
-        "gitea.pr.merge",
-        "gitea.pr.create",
-        "gitea.branch.push",
-    ],
-}
-
-
-def _reviewer_profile_patch():
-    return patch.object(mcp_server, "get_profile", return_value=REVIEWER_PROFILE)
-
-
-def _lock(mutations=None, correction=False):
-    return {
-        "task": "review_pr",
-        "remote": "prgs",
-        "org": "Scaled-Tech-Consulting",
-        "repo": "Gitea-Tools",
-        "session_pid": os.getpid(),
-        "session_profile": "prgs-reviewer",
-        "session_profile_lock": "prgs-reviewer",
-        "profile_identity": "prgs-reviewer",
-        "final_review_decision_ready": False,
-        "ready_pr_number": None,
-        "ready_action": None,
-        "ready_expected_head_sha": None,
-        "ready_remote": None,
-        "ready_org": None,
-        "ready_repo": None,
-        "live_mutations": list(mutations or []),
-        "correction_authorized": correction,
-        "correction_reason": None,
-    }
-
-
-APPROVED_586 = {
-    "pr_number": 586,
-    "action": "approve",
-    "review_id": 395,
-    "review_state": "approve",
-}
-APPROVED_OPEN = {
-    "pr_number": 700,
-    "action": "approve",
-    "review_id": 1,
-    "review_state": "approve",
-}
-RC_OPEN = {
-    "pr_number": 701,
-    "action": "request_changes",
-    "review_id": 2,
-    "review_state": "request_changes",
-}
-
-
-def _seed(mutations=None, correction=False):
-    import review_workflow_load
-
-    review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT)
-    mcp_server._save_review_decision_lock(_lock(mutations, correction))
-    mcp_server.gitea_load_review_workflow()
-
-
-def _merged_pr(pr_number=586, sha="6913ac9" + "0" * 33):
-    return {
-        "number": pr_number,
-        "state": "closed",
-        "merged": True,
-        "merged_at": "2026-07-09T18:20:03Z",
-        "merge_commit_sha": sha,
-    }
-
-
-def _open_pr(pr_number=700):
-    return {
-        "number": pr_number,
-        "state": "open",
-        "merged": False,
-        "merged_at": None,
-        "merge_commit_sha": None,
-    }
-
-
-# --------------------------------------------------------------------------- #
-# Pure assessment
-# --------------------------------------------------------------------------- #
-class TestAssessStaleLock(unittest.TestCase):
-    def test_no_lock(self):
-        a = srdl.assess_stale_review_decision_lock(None)
-        self.assertFalse(a["cleanup_allowed"])
-        self.assertFalse(a["is_moot"])
-
-    def test_no_terminal_mutations(self):
-        a = srdl.assess_stale_review_decision_lock(_lock([]))
-        self.assertFalse(a["cleanup_allowed"])
-
-    def test_open_pr_not_cleanable(self):
-        a = srdl.assess_stale_review_decision_lock(
-            _lock([APPROVED_OPEN]), pr_live=_open_pr(700)
-        )
-        self.assertFalse(a["is_moot"])
-        self.assertFalse(a["cleanup_allowed"])
-        self.assertTrue(any("open PR" in r for r in a["reasons"]))
-
-    def test_merged_pr_is_moot_and_cleanable(self):
-        a = srdl.assess_stale_review_decision_lock(
-            _lock([APPROVED_586]), pr_live=_merged_pr(586)
-        )
-        self.assertTrue(a["is_moot"])
-        self.assertTrue(a["cleanup_allowed"])
-        self.assertEqual(a["last_terminal_pr"], 586)
-
-    def test_closed_unmerged_is_moot(self):
-        a = srdl.assess_stale_review_decision_lock(
-            _lock([RC_OPEN]),
-            pr_live={"number": 701, "state": "closed", "merged": False},
-        )
-        self.assertTrue(a["is_moot"])
-        self.assertTrue(a["cleanup_allowed"])
-
-    def test_profile_mismatch_blocks(self):
-        a = srdl.assess_stale_review_decision_lock(
-            _lock([APPROVED_586]),
-            pr_live=_merged_pr(586),
-            active_profile_identity="other-profile",
-        )
-        self.assertFalse(a["cleanup_allowed"])
-        self.assertFalse(a["profile_match"])
-
-    def test_lookup_error_fail_closed(self):
-        a = srdl.assess_stale_review_decision_lock(
-            _lock([APPROVED_586]),
-            pr_lookup_error="timeout",
-        )
-        self.assertFalse(a["cleanup_allowed"])
-        self.assertTrue(any("timeout" in r for r in a["reasons"]))
-
-
-# --------------------------------------------------------------------------- #
-# Hard-stop still blocks active locks
-# --------------------------------------------------------------------------- #
-class TestActiveHardStopPreserved(unittest.TestCase):
-    def tearDown(self):
-        mcp_server._save_review_decision_lock(None)
-        mcp_server.review_workflow_load.clear_review_workflow_load()
-
-    def test_open_approve_still_blocks_other_pr_mark_ready(self):
-        _seed([APPROVED_OPEN])
-        reasons = mcp_server.terminal_review_hard_stop_reasons(592, "mark_ready")
-        self.assertTrue(reasons)
-        self.assertIn("#332", reasons[0])
-        self.assertIn("gitea_cleanup_stale_review_decision_lock", reasons[0])
-
-    def test_request_changes_still_blocks_everything(self):
-        _seed([RC_OPEN])
-        for op in ("merge", "mark_ready", "review"):
-            self.assertTrue(
-                mcp_server.terminal_review_hard_stop_reasons(592, op),
-                msg=op,
-            )
-
-
-# --------------------------------------------------------------------------- #
-# MCP cleanup tool
-# --------------------------------------------------------------------------- #
-class TestCleanupTool(unittest.TestCase):
-    def setUp(self):
-        self._env = patch.dict(os.environ, REVIEWER_ENV, clear=False)
-        self._env.start()
-
-    def tearDown(self):
-        self._env.stop()
-        mcp_server._save_review_decision_lock(None)
-        mcp_server.review_workflow_load.clear_review_workflow_load()
-
-    def test_read_only_reports_moot_without_clearing(self):
-        _seed([APPROVED_586])
-        with patch.object(mcp_server, "api_request", return_value=_merged_pr()), \
-                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
-                patch.object(
-                    mcp_server, "_authenticated_username", return_value="sysadmin"
-                ), \
-                _reviewer_profile_patch(), \
-                patch.object(mcp_server, "_profile_operation_gate", return_value=[]):
-            r = gitea_cleanup_stale_review_decision_lock(
-                apply=False, remote="prgs",
-                org="Scaled-Tech-Consulting", repo="Gitea-Tools",
-            )
-        self.assertTrue(r["success"])
-        self.assertTrue(r["is_moot"])
-        self.assertTrue(r["cleanup_allowed"])
-        self.assertFalse(r["cleanup_performed"])
-        self.assertIsNotNone(mcp_server._load_review_decision_lock())
-
-    def test_apply_clears_moot_lock(self):
-        _seed([APPROVED_586])
-        posts = []
-
-        def _api(method, url, auth, payload=None):
-            if method == "GET" and "/pulls/586" in url:
-                return _merged_pr()
-            if method == "POST" and "/comments" in url:
-                posts.append(payload)
-                return {"id": 9999}
-            return {}
-
-        with patch.object(mcp_server, "api_request", side_effect=_api), \
-                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
-                patch.object(
-                    mcp_server, "_authenticated_username", return_value="sysadmin"
-                ), \
-                _reviewer_profile_patch(), \
-                patch.object(mcp_server, "_profile_operation_gate", return_value=[]), \
-                patch.object(
-                    mcp_server, "_audited",
-                    side_effect=lambda *a, **k: __import__(
-                        "contextlib"
-                    ).nullcontext(),
-                ):
-            r = gitea_cleanup_stale_review_decision_lock(
-                apply=True,
-                expected_terminal_pr=586,
-                remote="prgs",
-                org="Scaled-Tech-Consulting",
-                repo="Gitea-Tools",
-            )
-        self.assertTrue(r["cleanup_performed"], r)
-        self.assertTrue(r["audit"]["applied"])
-        self.assertIsNone(mcp_server._load_review_decision_lock())
-        self.assertEqual(r.get("audit_comment_id"), 9999)
-        self.assertTrue(posts)
-
-    def test_apply_refuses_open_pr(self):
-        _seed([APPROVED_OPEN])
-        with patch.object(mcp_server, "api_request", return_value=_open_pr(700)), \
-                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
-                patch.object(
-                    mcp_server, "_authenticated_username", return_value="sysadmin"
-                ), \
-                _reviewer_profile_patch(), \
-                patch.object(mcp_server, "_profile_operation_gate", return_value=[]):
-            r = gitea_cleanup_stale_review_decision_lock(
-                apply=True, remote="prgs",
-                org="Scaled-Tech-Consulting", repo="Gitea-Tools",
-            )
-        self.assertFalse(r["cleanup_performed"])
-        self.assertFalse(r["cleanup_allowed"])
-        self.assertIsNotNone(mcp_server._load_review_decision_lock())
-
-    def test_expected_terminal_pr_pin(self):
-        _seed([APPROVED_586])
-        with patch.object(mcp_server, "api_request", return_value=_merged_pr()), \
-                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
-                patch.object(
-                    mcp_server, "_authenticated_username", return_value="sysadmin"
-                ), \
-                _reviewer_profile_patch(), \
-                patch.object(mcp_server, "_profile_operation_gate", return_value=[]):
-            r = gitea_cleanup_stale_review_decision_lock(
-                apply=True,
-                expected_terminal_pr=999,
-                remote="prgs",
-                org="Scaled-Tech-Consulting",
-                repo="Gitea-Tools",
-            )
-        self.assertFalse(r["cleanup_performed"])
-        self.assertTrue(any("expected_terminal_pr" in x for x in r["reasons"]))
-
-    def test_after_moot_cleanup_other_pr_mark_ready_unblocked(self):
-        """PR #592-style: after clearing merged #586 lock, new mark_ready works."""
-        _seed([APPROVED_586])
-        with patch.object(mcp_server, "api_request", return_value=_merged_pr()), \
-                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
-                patch.object(
-                    mcp_server, "_authenticated_username", return_value="sysadmin"
-                ), \
-                _reviewer_profile_patch(), \
-                patch.object(mcp_server, "_profile_operation_gate", return_value=[]), \
-                patch.object(
-                    mcp_server, "_audited",
-                    side_effect=lambda *a, **k: __import__(
-                        "contextlib"
-                    ).nullcontext(),
-                ):
-            cleaned = gitea_cleanup_stale_review_decision_lock(
-                apply=True,
-                expected_terminal_pr=586,
-                remote="prgs",
-                org="Scaled-Tech-Consulting",
-                repo="Gitea-Tools",
-                post_audit_comment=False,
-            )
-        self.assertTrue(cleaned["cleanup_performed"], cleaned)
-
-        # Hard-stop gone; re-init clean lock like a fresh review_pr resolve.
-        mcp_server.init_review_decision_lock(remote="prgs", task="review_pr")
-        mcp_server.gitea_load_review_workflow()
-        self.assertEqual(
-            mcp_server.terminal_review_hard_stop_reasons(592, "mark_ready"),
-            [],
-        )
-
-        no_lease = {"block": False, "reasons": [], "mutation_allowed": True}
-        with patch.object(mcp_server, "_list_pr_lease_comments", return_value=[]), \
-                patch.object(
-                    mcp_server, "_pr_work_lease_reviewer_block", return_value=no_lease
-                ), \
-                patch.object(
-                    mcp_server,
-                    "gitea_check_pr_eligibility",
-                    return_value={
-                        "eligible": True,
-                        "reasons": [],
-                        "authenticated_user": "sysadmin",
-                        "pr_author": "jcwalker3",
-                        "self_author": False,
-                    },
-                ), \
-                patch.object(
-                    mcp_server, "_authenticated_username", return_value="sysadmin"
-                ):
-            marked = gitea_mark_final_review_decision(
-                pr_number=592,
-                action="approve",
-                expected_head_sha=HEAD,
-                remote="prgs",
-                org="Scaled-Tech-Consulting",
-                repo="Gitea-Tools",
-            )
-        self.assertTrue(marked.get("marked_ready"), marked)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_webui_ci_gate.py

@@ -1,90 +0,0 @@
-"""Tests for web UI CI path-filter gate (#436)."""
-import os
-import subprocess
-import sys
-import unittest
-from pathlib import Path
-from unittest import mock
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-from webui.ci_paths import should_run_webui_ci, touches_webui_surface
-from webui.lease_loader import load_lease_snapshot
-from webui.queue_loader import load_queue_snapshot
-from webui.runtime_health import load_runtime_snapshot
-
-_REPO_ROOT = Path(__file__).resolve().parent.parent
-
-
-class TestCiPathMatching(unittest.TestCase):
-    def test_positive_paths(self):
-        for path in (
-            "webui/app.py",
-            "tests/test_webui_skeleton.py",
-            "docs/webui-local-dev.md",
-            "scripts/test-webui",
-        ):
-            with self.subTest(path=path):
-                self.assertTrue(touches_webui_surface(path))
-
-    def test_negative_paths(self):
-        for path in ("mcp_server.py", "tests/test_mcp_server.py", "README.md"):
-            with self.subTest(path=path):
-                self.assertFalse(touches_webui_surface(path))
-
-    def test_should_run_aggregate(self):
-        self.assertTrue(should_run_webui_ci(["README.md", "webui/layout.py"]))
-        self.assertFalse(should_run_webui_ci(["mcp_server.py", "gitea_auth.py"]))
-
-
-class TestCiRunnerScript(unittest.TestCase):
-    def test_test_webui_smoke(self):
-        script = _REPO_ROOT / "scripts" / "test-webui"
-        result = subprocess.run(
-            ["bash", str(script), "-q"],
-            cwd=_REPO_ROOT,
-            capture_output=True,
-            text=True,
-            timeout=60,
-            env={
-                **os.environ,
-                "WEBUI_TEST_PATTERN": "test_webui_skeleton.py",
-            },
-        )
-        self.assertEqual(result.returncode, 0, result.stderr or result.stdout)
-
-
-class TestOfflineWebuiCiMode(unittest.TestCase):
-    def test_offline_mode_avoids_live_queue_auth(self):
-        with mock.patch.dict(os.environ, {"WEBUI_TEST_OFFLINE": "1"}):
-            with mock.patch(
-                "webui.queue_loader.get_auth_header",
-                side_effect=AssertionError("live auth should not run"),
-            ):
-                snapshot = load_queue_snapshot()
-        self.assertIsNone(snapshot.fetch_error)
-        self.assertEqual(snapshot.prs, ())
-        self.assertEqual(snapshot.issues, ())
-
-    def test_offline_mode_avoids_live_lease_auth(self):
-        with mock.patch.dict(os.environ, {"WEBUI_TEST_OFFLINE": "1"}):
-            with mock.patch(
-                "webui.lease_loader.get_auth_header",
-                side_effect=AssertionError("live auth should not run"),
-            ):
-                snapshot = load_lease_snapshot()
-        self.assertIsNone(snapshot.fetch_error)
-        self.assertEqual(snapshot.reviewer_leases, ())
-
-    def test_offline_mode_avoids_live_runtime_identity(self):
-        with mock.patch.dict(os.environ, {"WEBUI_TEST_OFFLINE": "1"}):
-            with mock.patch(
-                "webui.runtime_health.get_auth_header",
-                side_effect=AssertionError("live auth should not run"),
-            ):
-                snapshot = load_runtime_snapshot()
-        self.assertEqual(snapshot.identity_error, "offline web UI test mode")
-
-
-if __name__ == "__main__":
-    unittest.main()

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_mvp_suite.py

@@ -1,156 +0,0 @@
-"""Consolidated MVP coverage for the internal web UI (#436)."""
-from __future__ import annotations
-
-import importlib
-import os
-import sys
-import unittest
-from pathlib import Path
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-from starlette.testclient import TestClient
-from starlette.routing import Route
-
-from webui.app import create_app
-
-_REPO_ROOT = Path(__file__).resolve().parent.parent
-
-# HTML pages and JSON exports registered on master MVP.
-MVP_GET_ROUTES: tuple[tuple[str, str | tuple[str, ...]], ...] = (
-    ("/", "Operator console"),
-    ("/health", "mcp-control-plane-webui"),
-    ("/queue", "Live queue"),
-    ("/api/queue", ("prs", "issues")),
-    ("/projects", "Gitea-Tools"),
-    ("/api/projects", "projects"),
-    ("/prompts", "Prompt library"),
-    ("/api/prompts", "prompts"),
-    ("/runtime", "Runtime"),
-    ("/audit", "Audit"),
-    ("/worktrees", "Worktrees"),
-    ("/leases", "Leases"),
-)
-
-MUTATION_METHODS = ("POST", "PUT", "PATCH", "DELETE")
-
-
-def _import_optional(module_name: str):
-    try:
-        return importlib.import_module(module_name)
-    except ImportError:
-        return None
-
-
-class TestWebuiServerStartup(unittest.TestCase):
-    def test_create_app_imports_cleanly(self):
-        app = create_app()
-        self.assertIsNotNone(app)
-
-    def test_main_module_imports(self):
-        import webui.__main__ as main_mod
-
-        self.assertTrue(callable(main_mod.main))
-
-
-class TestWebuiRouteRendering(unittest.TestCase):
-    def setUp(self):
-        self.client = TestClient(create_app())
-
-    def test_all_mvp_get_routes_render(self):
-        for path, needle in MVP_GET_ROUTES:
-            with self.subTest(path=path):
-                response = self.client.get(path)
-                self.assertEqual(response.status_code, 200, path)
-                if path == "/health":
-                    assert isinstance(needle, str)
-                    self.assertEqual(response.json()["service"], needle)
-                elif path.startswith("/api/"):
-                    data = response.json()
-                    if isinstance(needle, tuple):
-                        for key in needle:
-                            self.assertIn(key, data)
-                    else:
-                        self.assertIn(needle, data)
-                else:
-                    assert isinstance(needle, str)
-                    self.assertIn(needle, response.text)
-
-    def test_registered_routes_include_mvp_paths(self):
-        app = create_app()
-        get_paths = {
-            route.path
-            for route in app.routes
-            if isinstance(route, Route) and "GET" in route.methods
-        }
-        for path, _ in MVP_GET_ROUTES:
-            with self.subTest(path=path):
-                self.assertIn(path, get_paths)
-
-
-class TestWebuiReadOnlyGuards(unittest.TestCase):
-    def setUp(self):
-        self.client = TestClient(create_app())
-
-    def test_mutation_methods_rejected_on_core_paths(self):
-        paths = ("/health", "/queue", "/projects", "/prompts", "/api/queue")
-        for path in paths:
-            for method in MUTATION_METHODS:
-                with self.subTest(path=path, method=method):
-                    response = self.client.request(method, path)
-                    self.assertEqual(response.status_code, 405)
-                    self.assertEqual(response.json()["error"], "read-only-mvp")
-
-
-class TestWebuiOptionalModules(unittest.TestCase):
-    """Exercise child-issue modules when present on the branch under test."""
-
-    def test_audit_validator_basics_when_present(self):
-        mod = _import_optional("webui.audit_validator")
-        if mod is None:
-            self.skipTest("webui.audit_validator not merged on this branch")
-        report = "## Controller Handoff\n- Task: review PR #1\n"
-        self.assertEqual(mod.infer_task_kind(report), "review_pr")
-        result = mod.audit_report(report)
-        self.assertIn("grade", result)
-
-    def test_gated_actions_fail_closed_when_present(self):
-        mod = _import_optional("webui.gated_actions")
-        if mod is None:
-            self.skipTest("webui.gated_actions not merged on this branch")
-        registry = mod.build_action_registry()
-        for action in registry.actions:
-            with self.subTest(action=action.action_id):
-                self.assertFalse(action.enabled)
-        result = mod.attempt_action("merge_pr", pr_number=1)
-        self.assertFalse(result["success"])
-
-    def test_deployment_boundary_when_present(self):
-        mod = _import_optional("webui.deployment_boundary")
-        if mod is None:
-            self.skipTest("webui.deployment_boundary not merged on this branch")
-        assessment = mod.assess_bind_host("0.0.0.0")
-        self.assertEqual(assessment.disposition, "refuse")
-
-
-class TestWebuiTestInventory(unittest.TestCase):
-    def test_webui_test_modules_exist(self):
-        modules = sorted(_REPO_ROOT.glob("tests/test_webui_*.py"))
-        names = [path.name for path in modules]
-        self.assertIn("test_webui_skeleton.py", names)
-        self.assertIn("test_webui_project_registry.py", names)
-        self.assertIn("test_webui_prompt_library.py", names)
-        self.assertIn("test_webui_queue_dashboard.py", names)
-        self.assertGreaterEqual(len(names), 5)
-
-
-class TestWebuiCiScripts(unittest.TestCase):
-    def test_runner_scripts_exist(self):
-        for name in ("test-webui", "ci-webui-check"):
-            script = _REPO_ROOT / "scripts" / name
-            self.assertTrue(script.is_file(), name)
-            self.assertTrue(os.access(script, os.X_OK), name)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_webui_queue_dashboard.py

@@ -206,9 +206,8 @@ class TestQueueRoutes(unittest.TestCase):
         self.assertEqual(data["pagination"]["issues"]["returned_count"], 3)
 
     def test_queue_fail_closed_without_credentials(self):
-        with mock.patch.dict("os.environ", {"WEBUI_TEST_OFFLINE": "0"}):
-            with mock.patch("webui.queue_loader.get_auth_header", return_value=None):
-                snapshot = load_queue_snapshot()
+        with mock.patch("webui.queue_loader.get_auth_header", return_value=None):
+            snapshot = load_queue_snapshot()
         self.assertIsNotNone(snapshot.fetch_error)
         self.assertEqual(len(snapshot.prs), 0)
 
@@ -286,4 +285,4 @@ class TestQueueFailClosedUx(unittest.TestCase):
 
 
 if __name__ == "__main__":
-    unittest.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/ci_paths.py

@@ -1,20 +0,0 @@
-"""Path filters for web UI CI gating (#436)."""
-
-from __future__ import annotations
-
-import re
-from collections.abc import Iterable
-
-_WEBUI_TOUCH_RE = re.compile(
-    r"^(?:webui/|tests/test_webui_|docs/webui|scripts/(?:run-webui|test-webui|ci-webui-check)$)"
-)
-
-
-def touches_webui_surface(path: str) -> bool:
-    """Return True when *path* is a web UI surface file."""
-    return bool(_WEBUI_TOUCH_RE.match(path.strip()))
-
-
-def should_run_webui_ci(changed_paths: Iterable[str]) -> bool:
-    """Return True when any changed path should trigger the web UI test suite."""
-    return any(touches_webui_surface(path) for path in changed_paths)

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
@@ -228,29 +228,10 @@ def load_lease_snapshot(
         )
 
     host = _host_from_url(project.remote_host)
-    offline_test = (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip() in {
-        "1",
-        "true",
-        "yes",
-    }
-
-    def _empty_pr_fetch(*_args, **_kwargs):
-        return [], None
-
-    def _empty_issue_fetch(*_args, **_kwargs):
-        return [], None
-
-    def _empty_comment_fetch(*_args, **_kwargs):
-        return []
-
-    pr_fetch = fetch_prs or (_empty_pr_fetch if offline_test else _fetch_prs)
-    issue_fetch = fetch_issues or (
-        _empty_issue_fetch if offline_test else _fetch_issues
-    )
-    comment_fetch = fetch_comments or (
-        _empty_comment_fetch if offline_test else _fetch_comments
-    )
-    using_live = not offline_test and (fetch_prs is None or fetch_issues is None)
+    pr_fetch = fetch_prs or _fetch_prs
+    issue_fetch = fetch_issues or _fetch_issues
+    comment_fetch = fetch_comments or _fetch_comments
+    using_live = fetch_prs is None or fetch_issues is None
     auth = get_auth_header(host) if using_live else "test-auth"
 
     if using_live and not auth:
@@ -347,4 +328,4 @@ def snapshot_to_dict(snapshot: LeaseSnapshot) -> dict[str, Any]:
         ],
         "collision_history": list(snapshot.collision_history),
         "fetch_error": snapshot.fetch_error,
-    }
+    }

webui/queue_loader.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import os
 import re
 from dataclasses import dataclass
 from datetime import datetime, timezone
@@ -269,23 +268,9 @@ def load_queue_snapshot(
         )
 
     host = _host_from_url(project.remote_host)
-    offline_test = (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip() in {
-        "1",
-        "true",
-        "yes",
-    }
-
-    def _empty_fetch(*_args, **_kwargs):
-        return [], _pagination_from_pages(
-            per_page=50,
-            pages_fetched=1,
-            returned_count=0,
-            is_final_page=True,
-        )
-
-    pr_fetch = fetch_prs or (_empty_fetch if offline_test else _fetch_prs)
-    issue_fetch = fetch_issues or (_empty_fetch if offline_test else _fetch_issues)
-    using_live_fetch = not offline_test and (fetch_prs is None or fetch_issues is None)
+    pr_fetch = fetch_prs or _fetch_prs
+    issue_fetch = fetch_issues or _fetch_issues
+    using_live_fetch = fetch_prs is None or fetch_issues is None
     auth = get_auth_header(host) if using_live_fetch else "test-auth"
     if using_live_fetch and not auth:
         return QueueSnapshot(
@@ -397,4 +382,4 @@ def snapshot_to_dict(snapshot: QueueSnapshot) -> dict[str, Any]:
             "prs": _page(snapshot.pr_pagination),
             "issues": _page(snapshot.issue_pagination),
         },
-    }
+    }

webui/runtime_health.py

@@ -217,11 +217,6 @@ def load_runtime_snapshot(
     repo = _repo_root()
     host = _host_from_url(project.remote_host)
     remote = _remote_name_for_host(host)
-    offline_test = (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip() in {
-        "1",
-        "true",
-        "yes",
-    }
     profile = get_profile()
     allowed = profile.get("allowed_operations") or []
     forbidden = profile.get("forbidden_operations") or []
@@ -252,20 +247,10 @@ def load_runtime_snapshot(
             fetch_error=str(exc),
         )
 
-    identity_fn = resolve_username
-    if identity_fn is None:
-        if offline_test:
-            identity_fn = lambda _host: (None, "offline web UI test mode")
-        else:
-            identity_fn = _resolve_username
+    identity_fn = resolve_username or _resolve_username
     username, identity_error = identity_fn(host)
 
-    git_fn = git_sync
-    if git_fn is None:
-        if offline_test:
-            git_fn = lambda _repo, _remote, _branch: (None, None, None)
-        else:
-            git_fn = _git_sync_status
+    git_fn = git_sync or _git_sync_status
     remote_ref = "prgs" if remote == "prgs" else "origin"
     local_sha, remote_sha, behind = git_fn(repo, remote_ref, project.default_branch)
 
@@ -332,4 +317,4 @@ def snapshot_to_dict(snapshot: RuntimeSnapshot) -> dict[str, Any]:
         "schema_hashes": [_hash_row(item) for item in snapshot.schema_hashes],
         "restart_guidance": snapshot.restart_guidance,
         "fetch_error": snapshot.fetch_error,
-    }
+    }