Require and validate Controller Handoff sections in workflow final reports

- Upgrade SKILL.md §K compact format to the issue #182 canonical field set
  (Task/Repo/Role/Identity/Issue-PR/Branch-SHA/Files/Validation/Mutations/
  Current status/Blockers/Next/Safety) plus role-specific field lists for
  review/merge, author, and queue/inventory tasks.
- Point the review-pr, merge-pr, and start-issue template handoff lines at
  the exactly-titled Controller Handoff section with their role fields.
- Add review_proofs.assess_controller_handoff(): reports without the exact
  section are 'missing' (downgraded), present-but-partial are 'incomplete'
  with the absent fields listed, and role extras are enforced per role.
- Add TestControllerHandoff (8 tests) including a SKILL.md doc-contract
  test so the documented requirement cannot silently rot.

The handoff supplements the full report; full-report validation rules are
unchanged.

Closes #182

Co-Authored-By: Claude Fable 5 <[email protected]>

mcp_server.py

@@ -16,70 +16,10 @@ Configuration (mcp_config.json):
 import os
 import re
 import sys
-import json
 import functools
 import contextlib
 import subprocess
 
-
-LOCK_FILE = "/tmp/gitea_mutation_authority.lock"
-ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
-
-def record_mutation_authority(profile_name: str | None, identity: str | None, remote: str | None, task: str | None):
-    """Record the resolved capability context to fail-closed lock file."""
-    data = {
-        "initial_profile": profile_name,
-        "initial_identity": identity,
-        "current_profile": profile_name,
-        "current_identity": identity,
-        "remote": remote,
-        "task": task,
-        "role_pivot_authorized": False,
-        "role_pivot_record": None,
-    }
-    try:
-        with open(LOCK_FILE, "w", encoding="utf-8") as f:
-            json.dump(data, f)
-    except Exception:
-        pass
-
-def verify_mutation_authority(remote: str | None, host: str | None = None, required_role: str = "reviewer"):
-    """Verify that the current mutation matches the locked capability context."""
-    if not os.path.exists(LOCK_FILE):
-        raise RuntimeError("Mutation authority lock is missing (fail closed)")
-
-    try:
-        with open(LOCK_FILE, "r", encoding="utf-8") as f:
-            data = json.load(f)
-    except Exception as e:
-        raise RuntimeError(f"Could not read mutation authority lock: {e} (fail closed)")
-
-    if data.get("remote") != remote:
-        raise RuntimeError(
-            f"Mutation remote '{remote}' does not match locked remote '{data.get('remote')}' (fail closed)"
-        )
-
-    profile = get_profile()
-    active_profile = profile.get("profile_name")
-    h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
-    active_identity = _authenticated_username(h) if h else None
-
-    locked_profile = data.get("current_profile")
-    locked_identity = data.get("current_identity")
-
-    if active_profile != locked_profile or active_identity != locked_identity:
-        raise RuntimeError(
-            f"Mutation profile '{active_profile}' or identity '{active_identity}' "
-            f"does not match locked authority (profile: '{locked_profile}', identity: '{locked_identity}') (fail closed)"
-        )
-
-    # Check reviewer/author role pivot boundaries
-    if required_role == "reviewer" and "author" in str(data.get("initial_profile")).lower() and "reviewer" in str(active_profile).lower():
-        if not data.get("role_pivot_authorized"):
-            raise RuntimeError(
-                "Attempted reviewer mutation from author session without authorized role pivot (fail closed)"
-            )
-
 # Resolve the project root. MCP clients must launch this script directly with
 # the venv interpreter (venv/bin/python3) — see the config example above. We do
 # NOT os.execv() to re-point the interpreter: replacing the process after the
@@ -400,84 +340,6 @@ def gitea_create_issue(
     return _with_optional_url({"number": data["number"]}, data.get("html_url"))
 
 
-@mcp.tool()
-def gitea_lock_issue(
-    issue_number: int,
-    branch_name: str,
-    remote: str = "dadeschools",
-    host: str | None = None,
-    org: str | None = None,
-    repo: str | None = None,
-) -> dict:
-    """Lock exactly one Gitea issue and its branch name to ensure durable tracking.
-
-    Args:
-        issue_number: The tracking issue number.
-        branch_name: The branch name (must match (fix|feat|docs|chore)/issue-<issue_number>-<desc>).
-        remote: Known instance — 'dadeschools' or 'prgs'.
-        host: Override Gitea host.
-        org: Override Org.
-        repo: Override Repo.
-    """
-    # 1. Enforce branch name includes issue number
-    expected_pattern = f"issue-{issue_number}"
-    if expected_pattern not in branch_name:
-        raise ValueError(
-            f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)"
-        )
-
-    # 2. Check if the issue already has an open PR (reuse protection)
-    h, o, r = _resolve(remote, host, org, repo)
-    auth = _auth(h)
-    url = f"{repo_api_url(h, o, r)}/pulls?state=open"
-
-    try:
-        prs = api_get_all(url, auth)
-    except Exception as e:
-        raise RuntimeError(f"Could not list open PRs to verify issue lock: {e}")
-
-    for pr in prs:
-        pr_head = pr.get("head", {}).get("ref", "")
-        pr_title = pr.get("title", "")
-        pr_body = pr.get("body", "")
-
-        if expected_pattern in pr_head:
-            raise ValueError(
-                f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}, branch '{pr_head}') (fail closed)"
-            )
-
-        patterns = [
-            f"closes #{issue_number}",
-            f"fixes #{issue_number}",
-        ]
-        text_to_check = f"{pr_title} {pr_body}".lower()
-        if any(p in text_to_check for p in patterns):
-            raise ValueError(
-                f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}) via Closes/Fixes reference (fail closed)"
-            )
-
-    data = {
-        "issue_number": issue_number,
-        "branch_name": branch_name,
-        "remote": remote,
-        "org": o,
-        "repo": r,
-    }
-
-    try:
-        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
-            json.dump(data, f)
-    except Exception as e:
-        raise RuntimeError(f"Could not write issue lock file: {e}")
-
-    return {
-        "success": True,
-        "message": f"Successfully locked issue #{issue_number} to branch '{branch_name}' (fail-closed check complete).",
-        "issue_number": issue_number,
-        "branch_name": branch_name,
-    }
-
-
 @mcp.tool()
 def gitea_create_pr(
     title: str,
@@ -505,41 +367,6 @@ def gitea_create_pr(
         dict with 'number' of the created PR ('url' only with the reveal opt-in).
     """
     h, o, r = _resolve(remote, host, org, repo)
-
-    # ── Issue Lock Validation (Issue #194 / #196) ──
-    if not os.path.exists(ISSUE_LOCK_FILE):
-        raise RuntimeError("Issue lock is missing (fail closed). Call gitea_lock_issue first.")
-
-    try:
-        with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
-            lock_data = json.load(f)
-    except Exception as e:
-        raise RuntimeError(f"Could not read issue lock file: {e} (fail closed)")
-
-    locked_issue = lock_data.get("issue_number")
-    locked_branch = lock_data.get("branch_name")
-
-    if head != locked_branch:
-        raise ValueError(
-            f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)"
-        )
-
-    # Check for forbidden terms anywhere in title/body
-    forbidden_terms = ["equivalent", "related", "same as"]
-    text_to_check = f"{title} {body}".lower()
-    for term in forbidden_terms:
-        if term in text_to_check:
-            raise ValueError(
-                f"PR title or body contains forbidden term '{term}' (fail closed)"
-            )
-
-    # Ensure Closes #<locked_issue> or Fixes #<locked_issue> is present exactly
-    closes_pattern = re.compile(rf"\b(closes|fixes)\s+#{locked_issue}\b", re.IGNORECASE)
-    if not closes_pattern.search(text_to_check):
-        raise ValueError(
-            f"PR title or body must contain 'Closes #{locked_issue}' or 'Fixes #{locked_issue}' exactly to ensure durable tracking (fail closed)"
-        )
-
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/pulls"
     payload = {"title": title, "body": body, "head": head, "base": base}
@@ -1164,12 +991,6 @@ def gitea_submit_pr_review(
     }
     reasons = result["reasons"]
 
-    try:
-        verify_mutation_authority(remote, host, required_role="reviewer")
-    except RuntimeError as e:
-        reasons.append(str(e))
-        return result
-
     # Gate 1 — valid review action (no mutation on unknown action).
     if action not in _REVIEW_ACTIONS:
         reasons.append(
@@ -1490,12 +1311,6 @@ def gitea_merge_pr(
     }
     reasons = result["reasons"]
 
-    try:
-        verify_mutation_authority(remote, host, required_role="reviewer")
-    except RuntimeError as e:
-        reasons.append(str(e))
-        return result
-
     # Gate 1 — valid merge method (no API call on a bad method).
     if do not in _MERGE_METHODS:
         reasons.append(
@@ -3172,25 +2987,6 @@ def gitea_activate_profile(
     after_profile = get_profile()["profile_name"]
     after_identity = _authenticated_username(h) if h else None
 
-    # 4.5 Record pivot in mutation authority lock
-    if os.path.exists(LOCK_FILE):
-        try:
-            with open(LOCK_FILE, "r", encoding="utf-8") as f:
-                lock_data = json.load(f)
-            lock_data["current_profile"] = after_profile
-            lock_data["current_identity"] = after_identity
-            lock_data["role_pivot_authorized"] = True
-            lock_data["role_pivot_record"] = {
-                "from_profile": before_profile,
-                "to_profile": after_profile,
-                "from_identity": before_identity,
-                "to_identity": after_identity
-            }
-            with open(LOCK_FILE, "w", encoding="utf-8") as f:
-                json.dump(lock_data, f)
-        except Exception:
-            pass
-
     # 5. Audit the switch if auditing is on
     _audit(
         "activate_profile",
@@ -3680,8 +3476,6 @@ def gitea_resolve_task_capability(
             "STOP: the active profile cannot perform the requested task; "
             "follow exact_safe_next_action instead of improvising.")
 
-    record_mutation_authority(profile["profile_name"], username, remote if remote in REMOTES else None, task)
-
     return {
         "requested_task": task,
         "required_operation_permission": required_permission,

review_pr.py

@@ -24,7 +24,7 @@ venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3")
 if os.path.exists(venv_python) and sys.executable != venv_python:
     os.execv(venv_python, [venv_python] + sys.argv)
 
-from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url, get_profile
+from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url
 
 
 def main(argv=None):
@@ -60,29 +60,6 @@ def main(argv=None):
 
     host, org, repo = resolve_remote(args)
 
-    # ── Mutation Authority context wall check (Issue #194) ──
-    LOCK_FILE = "/tmp/gitea_mutation_authority.lock"
-    
-    if os.path.exists(LOCK_FILE):
-        try:
-            with open(LOCK_FILE, "r", encoding="utf-8") as f:
-                lock_data = json.load(f)
-            
-            # Resolve current CLI profile
-            cli_profile = get_profile().get("profile_name")
-            locked_profile = lock_data.get("current_profile")
-            
-            if cli_profile != locked_profile:
-                print(
-                    f"Mismatched active profile vs mutation profile (CLI override rejected): "
-                    f"CLI profile '{cli_profile}' does not match locked active profile '{locked_profile}' (fail closed)",
-                    file=sys.stderr
-                )
-                return 3
-        except Exception as e:
-            print(f"Mutation authority check failed: {e}", file=sys.stderr)
-            return 3
-
     body = args.body
     if args.body_file:
         if args.body_file == "-":

review_proofs.py

@@ -18,54 +18,6 @@ import re
 
 _FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
 
-
-# Repo name disambiguation rules for blind PR queue review.
-# User phrases referencing the "MCP Gitea tool" (or similar) must resolve to
-# Gitea-Tools repo, not be confused with mcp-control-plane.
-# If ambiguous (e.g. just "open PRs"), check both configured repos.
-REPO_ALIASES = {
-    "gitea-tools": "Scaled-Tech-Consulting/Gitea-Tools",
-    "gitea tool": "Scaled-Tech-Consulting/Gitea-Tools",
-    "mcp gitea tool": "Scaled-Tech-Consulting/Gitea-Tools",
-    "gitea mcp tool": "Scaled-Tech-Consulting/Gitea-Tools",
-    "gitea-tools repo": "Scaled-Tech-Consulting/Gitea-Tools",
-    "mcp-control-plane": "Scaled-Tech-Consulting/mcp-control-plane",
-    "mcp control plane": "Scaled-Tech-Consulting/mcp-control-plane",
-}
-
-
-def resolve_repos_from_user_reference(
-    reference: str, configured: list[str] | None = None
-) -> list[str]:
-    """Resolve a user reference string to the list of target repos to inventory.
-
-    - Exact aliases for Gitea-Tools map only to Gitea-Tools.
-    - mcp-control-plane aliases map only to it.
-    - Empty, ambiguous, or general "open PRs" default to all configured repos
-      (both by default).
-    - Returns subset of configured; never invents new repos.
-    """
-    if configured is None:
-        configured = [
-            "Scaled-Tech-Consulting/Gitea-Tools",
-            "Scaled-Tech-Consulting/mcp-control-plane",
-        ]
-    if not reference or not reference.strip():
-        return list(configured)
-
-    ref_lower = reference.lower()
-    matched = []
-    for alias, full_name in REPO_ALIASES.items():
-        if alias in ref_lower:
-            if full_name not in matched:
-                matched.append(full_name)
-    if matched:
-        # return only the matched ones that are in configured, preserving order
-        return [r for r in configured if r in matched]
-    # no specific alias match → check all (complete inventory required)
-    return list(configured)
-
-
 SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION = (
     "evidence missing: report contamination as unknown and "
     "choose another PR or stop"
@@ -527,35 +479,6 @@ def assess_controller_handoff(report_text, role=None):
             "reasons": [f"handoff missing required field: {m}"
                         for m in missing],
         }
-
-    # Validate issue/PR references for exact number and no forbidden terms (Issue #194 / #196)
-    fields_dict = {}
-    for line in section:
-        stripped = line.strip().lstrip("-*").strip()
-        if ":" in stripped:
-            k, v = stripped.split(":", 1)
-            fields_dict[k.strip().lower()] = v.strip()
-
-    for alias in ("selected issue", "pr number opened", "pr opened", "pr number", "selected pr"):
-        val = fields_dict.get(alias)
-        if val:
-            numbers = re.findall(r"\d+", val)
-            has_forbidden = any(term in val.lower() for term in ("equivalent", "related", "same", "/"))
-            if len(numbers) != 1 or has_forbidden:
-                field_name = "Selected issue/PR"
-                for name, aliases in list(HANDOFF_BASE_FIELDS) + list(HANDOFF_ROLE_FIELDS.get(role or "", ())):
-                    if alias in aliases:
-                        field_name = name
-                        break
-                return {
-                    "verdict": "incomplete",
-                    "downgraded": True,
-                    "missing_fields": [field_name],
-                    "reasons": [
-                        f"{field_name} must specify exactly one number and no ambiguous references (got: '{val}')"
-                    ],
-                }
-
     return {
         "verdict": "complete",
         "downgraded": False,

scripts/worktree-start

@@ -40,16 +40,6 @@ start_ref="${2:-prgs/master}"
 
 # Enforce issue-linked, traceable branch names (issue → branch → worktree → PR).
 if [[ "$allow_unlinked" -eq 0 ]]; then
-  if [[ ! -f "/tmp/gitea_issue_lock.json" ]]; then
-    echo "Error: Issue lock file '/tmp/gitea_issue_lock.json' is missing. You must lock exactly one issue before branch creation (fail closed)." >&2
-    exit 2
-  fi
-  locked_branch=$(python3 -c "import json; print(json.load(open('/tmp/gitea_issue_lock.json')).get('branch_name', ''))")
-  if [[ "$branch" != "$locked_branch" ]]; then
-    echo "Error: Requested branch '$branch' does not match locked branch '$locked_branch' (fail closed)." >&2
-    exit 2
-  fi
-
   if [[ "$branch" =~ ^(fix|feat|docs|chore)/issue-[0-9]+-.+ ]] \
      || [[ "$branch" =~ ^review/pr-[0-9]+-.+ ]]; then
     :

skills/llm-project-workflow/SKILL.md

@@ -181,21 +181,11 @@ Worktree folder = branch with `/` replaced by `-`
    that the diff base is the PR base branch. If `HEAD` does not match the
    pinned head, **stop before review/merge**
    (`review_proofs.verify_pinned_head_checkout`).
-6. **Inventory proof (#173 + repo disambiguation hardening):** a blind queue
+6. **Inventory proof (#173):** a blind queue review must prove listing
-   review must prove listing completeness before claiming "only PRs found".
+   completeness before claiming "only PRs found": both configured
-   Use repo-name disambiguation:
+   repositories checked, open-PR filters stated, pagination handled or
-   - "Gitea-Tools" / "gitea tool" / "MCP Gitea tool" / "gitea MCP tool" /
+   explicitly not needed, and the total open PR count per repo reported
-     "gitea-tools repo" resolve **only** to `Scaled-Tech-Consulting/Gitea-Tools`.
+   (`review_proofs.assess_inventory_completeness`).
-   - "mcp-control-plane" resolves only to `Scaled-Tech-Consulting/mcp-control-plane`.
-   - Ambiguous ("open PRs", no explicit repo, "MCP Gitea tooling") → inventory
-     **both** configured repos.
-   Report must state exactly which repo(s) were checked. If only one checked:
-   "Only <repo> was checked. Other configured repos were not checked. This is
-   not a complete queue inventory." Never let a single-repo zero hide PRs in
-   the other.
-   Both configured repos must be reported with state filter, pagination proof,
-   and open-PR count (`review_proofs.assess_inventory_completeness` and
-   `resolve_repos_from_user_reference`).
 7. Inspect the full diff; confirm scope matches the linked issue; flag unrelated files.
 8. Run the tests. Validation reporting must include the exact command and
    exact results: pass/fail, counts of tests passed/skipped/failed, any

skills/llm-project-workflow/templates/review-pr.md

@@ -5,19 +5,6 @@ Copy, fill the `<...>` fields, and paste as the task prompt.
 ```text
 Task: review PR #<pr> for issue #<n>.
 
-Repo name disambiguation (Gitea-Tools blind review hardening):
-- "Gitea-Tools", "gitea tool", "MCP Gitea tool", "gitea MCP tool", "gitea-tools repo"
-  → MUST resolve to `Scaled-Tech-Consulting/Gitea-Tools` (never treat as mcp-control-plane).
-- "mcp-control-plane", "mcp control plane" → only `Scaled-Tech-Consulting/mcp-control-plane`.
-- If user says "open PRs", "the queue", "MCP Gitea tooling" without explicit repo,
-  or reference is ambiguous: check BOTH configured repos:
-  `Scaled-Tech-Consulting/Gitea-Tools` and `Scaled-Tech-Consulting/mcp-control-plane`.
-- In the final report, always state exactly which repo(s) were checked.
-  If only one was checked: explicitly say "Only <repo> was checked. Other
-  configured repos were not checked. This is not a complete queue inventory."
-- A single-repo "no open PRs" result MUST NOT be reported as global "no open PRs"
-  if the other configured repo was not inventoried.
-
 Rules (llm-project-workflow):
 - Review in a SEPARATE detached review worktree, never the author's folder.
 - You must NOT be the PR author. If the authenticated user == PR author, stop.

tests/test_create_issue.py

@@ -11,7 +11,6 @@ import json
 import sys
 import tempfile
 import unittest
-import contextlib
 from unittest.mock import MagicMock, patch
 
 # The module under test lives in the repo root, not a package.
@@ -39,8 +38,7 @@ class TestArgParsing(unittest.TestCase):
     @patch("create_issue.api_request", return_value={"number": 1, "html_url": "http://x/1"})
     @patch("create_issue.get_credentials", return_value=FAKE_CREDS)
     def test_minimal_args(self, _cred, _api):
-        with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
+        rc = create_issue.main(["--title", "Hello"])
-            rc = create_issue.main(["--title", "Hello"])
         self.assertEqual(rc, 0)
 
     def test_missing_title_exits(self):
@@ -52,8 +50,7 @@ class TestArgParsing(unittest.TestCase):
     @patch("create_issue.get_credentials", return_value=FAKE_CREDS)
     def test_remote_choices(self, _cred, _api):
         for remote in ("dadeschools", "prgs"):
-            with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
+            rc = create_issue.main(["--remote", remote, "--title", "X"])
-                rc = create_issue.main(["--remote", remote, "--title", "X"])
             self.assertEqual(rc, 0, f"--remote {remote} should be accepted")
 
     def test_invalid_remote_exits(self):
@@ -141,8 +138,7 @@ class TestAuthFailure(unittest.TestCase):
 
     @patch("create_issue.get_credentials", return_value=("", ""))
     def test_no_credentials_returns_1(self, _cred):
-        with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
+        rc = create_issue.main(["--title", "T"])
-            rc = create_issue.main(["--title", "T"])
         self.assertEqual(rc, 1)
 
 
@@ -156,8 +152,7 @@ class TestAPIError(unittest.TestCase):
     def test_api_error_returns_1(self, _cred):
         with patch("create_issue.api_request",
                    side_effect=RuntimeError("HTTP 422: duplicate")):
-            with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
+            rc = create_issue.main(["--title", "Dup"])
-                rc = create_issue.main(["--title", "Dup"])
             self.assertEqual(rc, 1)
 
 

tests/test_create_pr.py

@@ -7,7 +7,6 @@ import io
 import json
 import sys
 import unittest
-import contextlib
 from unittest.mock import MagicMock, patch
 
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
@@ -35,8 +34,7 @@ class TestArgParsing(unittest.TestCase):
     @patch("create_pr.urllib.request.urlopen", return_value=_mock_urlopen())
     @patch("create_pr.get_credentials", return_value=FAKE_CREDS)
     def test_minimal_required_args(self, _cred, _url):
-        with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
+        rc = create_pr.main(["--title", "PR Title", "--head", "feat/branch"])
-            rc = create_pr.main(["--title", "PR Title", "--head", "feat/branch"])
         self.assertEqual(rc, 0)
 
     def test_missing_title_exits(self):

tests/test_manage_labels.py

@@ -2,11 +2,9 @@
 
 All API calls are mocked — no real network or keychain access.
 """
-import io
 import json
 import sys
 import unittest
-import contextlib
 from unittest.mock import MagicMock, call, patch
 
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
@@ -35,8 +33,7 @@ class TestLabelCreation(unittest.TestCase):
 
         # Patch sys.argv to avoid --dry
         with patch.object(sys, "argv", ["manage_labels.py"]):
-            with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
+            manage_labels.main()
-                manage_labels.main()
 
         # The GET call happens, but no POST calls for label creation
         get_calls = [c for c in mock_api.call_args_list if c[0][0] == "GET"]
@@ -62,8 +59,7 @@ class TestLabelCreation(unittest.TestCase):
 
         mock_api.side_effect = side_effect
         with patch.object(sys, "argv", ["manage_labels.py"]):
-            with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
+            manage_labels.main()
-                manage_labels.main()
 
         post_calls = [
             c for c in mock_api.call_args_list
@@ -83,8 +79,7 @@ class TestDryRun(unittest.TestCase):
         mock_api.return_value = []  # no existing labels
 
         with patch.object(sys, "argv", ["manage_labels.py", "--dry"]):
-            with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
+            manage_labels.main()
-                manage_labels.main()
 
         # Only the GET call should be made, no POST or PUT
         for c in mock_api.call_args_list:
@@ -112,8 +107,7 @@ class TestLabelMapping(unittest.TestCase):
 
         mock_api.side_effect = side_effect
         with patch.object(sys, "argv", ["manage_labels.py"]):
-            with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
+            manage_labels.main()
-                manage_labels.main()
 
         put_calls = [c for c in mock_api.call_args_list if c[0][0] == "PUT"]
         self.assertEqual(len(put_calls), len(manage_labels.MAPPING))

tests/test_mcp_server.py

@@ -33,16 +33,10 @@ from mcp_server import (  # noqa: E402
     gitea_submit_pr_review,
     gitea_list_issue_comments,
     gitea_create_issue_comment,
-    gitea_lock_issue,
 )
 from gitea_auth import get_profile  # noqa: E402
 import gitea_config  # noqa: E402
 
-import mcp_server
-# Globally disable verification check for existing isolated unit tests
-_real_verify = mcp_server.verify_mutation_authority
-mcp_server.verify_mutation_authority = lambda *args, **kwargs: None
-
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 
 
@@ -91,13 +85,10 @@ class TestCreatePR(unittest.TestCase):
 
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    @patch("os.path.exists", return_value=True)
+    def test_creates_pr(self, _auth, mock_api):
-    @patch("builtins.open")
-    def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api):
-        mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
         with patch.dict(os.environ, {}, clear=True):
-            result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
+            result = gitea_create_pr(title="feat: X", head="feat/x", base="main")
         self.assertEqual(result["number"], 3)
         self.assertNotIn("url", result)
         payload = mock_api.call_args[0][3]
@@ -106,13 +97,10 @@ class TestCreatePR(unittest.TestCase):
 
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    @patch("os.path.exists", return_value=True)
+    def test_create_pr_reveal_opt_in_includes_url(self, _auth, mock_api):
-    @patch("builtins.open")
-    def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api):
-        mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
         with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
-            result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
+            result = gitea_create_pr(title="feat: X", head="feat/x", base="main")
         self.assertIn("pulls/3", result["url"])
 
 
@@ -2308,170 +2296,3 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase):
             "gitea.issue.comment", reviewer["allowed_operations"],
             reviewer.get("forbidden_operations", []))
         self.assertTrue(ok)
-
-
-class TestVerifyMutationAuthority(unittest.TestCase):
-    """Test verification lock logic under various configurations."""
-
-    def setUp(self):
-        self.patch_profile = patch("mcp_server.get_profile")
-        self.mock_profile = self.patch_profile.start()
-        self.patch_username = patch("mcp_server._authenticated_username")
-        self.mock_username = self.patch_username.start()
-
-        # Restore real function for these tests
-        self._old_verify = mcp_server.verify_mutation_authority
-        mcp_server.verify_mutation_authority = _real_verify
-
-    def tearDown(self):
-        self.patch_profile.stop()
-        self.patch_username.stop()
-        mcp_server.verify_mutation_authority = self._old_verify
-        # Clean up lock file
-        if os.path.exists("/tmp/gitea_mutation_authority.lock"):
-            os.remove("/tmp/gitea_mutation_authority.lock")
-
-    def test_missing_lock_fails_closed(self):
-        if os.path.exists("/tmp/gitea_mutation_authority.lock"):
-            os.remove("/tmp/gitea_mutation_authority.lock")
-        with self.assertRaises(RuntimeError) as ctx:
-            _real_verify("prgs")
-        self.assertIn("lock is missing", str(ctx.exception))
-
-    def test_mismatched_remote_fails(self):
-        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
-            json.dump({
-                "remote": "dadeschools",
-                "current_profile": "prgs-reviewer",
-                "current_identity": "sysadmin"
-            }, f)
-        with self.assertRaises(RuntimeError) as ctx:
-            _real_verify("prgs")
-        self.assertIn("does not match locked remote", str(ctx.exception))
-
-    def test_mismatched_profile_fails(self):
-        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
-            json.dump({
-                "remote": "prgs",
-                "current_profile": "prgs-author",
-                "current_identity": "jcwalker3"
-            }, f)
-        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
-        self.mock_username.return_value = "sysadmin"
-
-        with self.assertRaises(RuntimeError) as ctx:
-            _real_verify("prgs")
-        self.assertIn("does not match locked authority", str(ctx.exception))
-
-    def test_author_to_reviewer_pivot_blocked_without_authorization(self):
-        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
-            json.dump({
-                "remote": "prgs",
-                "initial_profile": "prgs-author",
-                "initial_identity": "jcwalker3",
-                "current_profile": "prgs-reviewer",
-                "current_identity": "sysadmin",
-                "role_pivot_authorized": False
-            }, f)
-        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
-        self.mock_username.return_value = "sysadmin"
-
-        with self.assertRaises(RuntimeError) as ctx:
-            _real_verify("prgs", required_role="reviewer")
-        self.assertIn("without authorized role pivot", str(ctx.exception))
-
-    def test_allowed_when_match(self):
-        with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
-            json.dump({
-                "remote": "prgs",
-                "initial_profile": "prgs-reviewer",
-                "initial_identity": "sysadmin",
-                "current_profile": "prgs-reviewer",
-                "current_identity": "sysadmin",
-                "role_pivot_authorized": False
-            }, f)
-        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
-        self.mock_username.return_value = "sysadmin"
-
-        # Should pass without exception
-        _real_verify("prgs")
-
-
-class TestIssueLocking(unittest.TestCase):
-    """Test issue locking and PR gating constraints."""
-
-    def tearDown(self):
-        if os.path.exists("/tmp/gitea_issue_lock.json"):
-            os.remove("/tmp/gitea_issue_lock.json")
-
-    @patch("mcp_server.api_get_all")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_lock_issue_success(self, _auth, mock_api):
-        mock_api.return_value = [] # no open PRs
-        res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
-        self.assertTrue(res["success"])
-        self.assertTrue(os.path.exists("/tmp/gitea_issue_lock.json"))
-
-    def test_lock_issue_mismatch_branch_fails(self):
-        with self.assertRaises(ValueError) as ctx:
-            gitea_lock_issue(issue_number=196, branch_name="feat/issue-195-mutations", remote="prgs")
-        self.assertIn("must contain locked issue pattern", str(ctx.exception))
-
-    @patch("mcp_server.api_get_all")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_lock_issue_reused_by_open_pr_branch(self, _auth, mock_api):
-        mock_api.return_value = [{
-            "number": 200,
-            "head": {"ref": "feat/issue-196-boundary"},
-            "title": "Some PR",
-            "body": "No closes ref"
-        }]
-        with self.assertRaises(ValueError) as ctx:
-            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
-        self.assertIn("already tied to an open PR", str(ctx.exception))
-
-    @patch("mcp_server.api_get_all")
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, mock_api):
-        mock_api.return_value = [{
-            "number": 200,
-            "head": {"ref": "feat/other-branch"},
-            "title": "Some PR",
-            "body": "fixes #196"
-        }]
-        with self.assertRaises(ValueError) as ctx:
-            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
-        self.assertIn("already tied to an open PR", str(ctx.exception))
-
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_pr_missing_lock_fails(self, _auth):
-        if os.path.exists("/tmp/gitea_issue_lock.json"):
-            os.remove("/tmp/gitea_issue_lock.json")
-        with self.assertRaises(RuntimeError) as ctx:
-            gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
-        self.assertIn("Issue lock is missing", str(ctx.exception))
-
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_pr_branch_mismatch_fails(self, _auth):
-        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
-            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
-        with self.assertRaises(ValueError) as ctx:
-            gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
-        self.assertIn("does not match locked branch", str(ctx.exception))
-
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_pr_forbidden_terms_fails(self, _auth):
-        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
-            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
-        for term in ("equivalent to #196", "related to #196", "same as #196"):
-            with self.assertRaises(ValueError) as ctx:
-                gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
-            self.assertIn("contains forbidden term", str(ctx.exception))
-
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_pr_missing_closes_ref_fails(self, _auth):
-        with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
-            json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
-        with self.assertRaises(ValueError) as ctx:
-            gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
-        self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))

tests/test_merge_pr.py

@@ -45,17 +45,13 @@ class TestMergeDisabled(unittest.TestCase):
         mock_api.assert_not_called()
 
     def test_message_points_to_gated_workflow(self):
-        from _pytest.monkeypatch import MonkeyPatch
         import io
+        import contextlib
         with patch("merge_pr.get_auth_header", return_value=FAKE_CREDS), \
                 patch("merge_pr.api_request") as mock_api:
             buf = io.StringIO()
-            monkeypatch = MonkeyPatch()
+            with contextlib.redirect_stderr(buf):
-            monkeypatch.setattr(sys, "stderr", buf)
-            try:
                 rc = merge_pr.main(["--pr-number", "81"])
-            finally:
-                monkeypatch.undo()
         self.assertEqual(rc, 2)
         mock_api.assert_not_called()
         msg = buf.getvalue().lower()

tests/test_migrate_profiles.py

@@ -9,8 +9,6 @@ import shutil
 from unittest.mock import patch
 from io import StringIO
 
-from _pytest.monkeypatch import MonkeyPatch
-
 # Add project root to sys.path
 PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 if PROJECT_ROOT not in sys.path:
@@ -129,29 +127,24 @@ class TestMigrateProfiles(unittest.TestCase):
         v2_data = migrate_profiles.migrate_v1_to_v2(self.v1_content)
         self.assertTrue(migrate_profiles.validate_v2_data(v2_data))
 
-    def test_dry_run_default(self):
+    @patch("sys.stdout", new_callable=StringIO)
+    def test_dry_run_default(self, mock_stdout):
         """Verify that running without -w prints generated config without modifying files."""
-        monkeypatch = MonkeyPatch()
+        output_file = os.path.join(self.temp_dir, "migrated_dry.json")
-        mock_stdout = StringIO()
+        test_args = [
-        monkeypatch.setattr(sys, "stdout", mock_stdout)
+            "migrate_profiles.py",
-        try:
+            "-i", self.input_file,
-            output_file = os.path.join(self.temp_dir, "migrated_dry.json")
+            "-o", output_file
-            test_args = [
+        ]
-                "migrate_profiles.py",
+        with patch.object(sys, "argv", test_args):
-                "-i", self.input_file,
+            with self.assertRaises(SystemExit) as cm:
-                "-o", output_file
+                migrate_profiles.main()
-            ]
+            self.assertEqual(cm.exception.code, 0)
-            with patch.object(sys, "argv", test_args):
-                with self.assertRaises(SystemExit) as cm:
-                    migrate_profiles.main()
-                self.assertEqual(cm.exception.code, 0)
 
-            self.assertFalse(os.path.exists(output_file))
+        self.assertFalse(os.path.exists(output_file))
-            self.assertFalse(os.path.exists(f"{self.input_file}.bak"))
+        self.assertFalse(os.path.exists(f"{self.input_file}.bak"))
 
-            stdout_output = mock_stdout.getvalue()
+        stdout_output = mock_stdout.getvalue()
-        finally:
-            monkeypatch.undo()
         self.assertIn("DRY-RUN MODE", stdout_output)
         self.assertIn("version", stdout_output)
         self.assertIn("identities", stdout_output)
@@ -172,18 +165,13 @@ class TestMigrateProfiles(unittest.TestCase):
             json.dump(sensitive, f)
 
         test_args = ["migrate_profiles.py", "-i", self.input_file]
-        monkeypatch = MonkeyPatch()
+        with patch("sys.stdout", new_callable=StringIO) as mock_stdout:
-        mock_stdout = StringIO()
-        monkeypatch.setattr(sys, "stdout", mock_stdout)
-        try:
             with patch.object(sys, "argv", test_args):
                 with self.assertRaises(SystemExit) as cm:
                     migrate_profiles.main()
                 self.assertEqual(cm.exception.code, 0)
 
-            stdout_output = mock_stdout.getvalue()
+        stdout_output = mock_stdout.getvalue()
-        finally:
-            monkeypatch.undo()
         self.assertNotIn("super-secret-token-value", stdout_output)
         self.assertNotIn("token", stdout_output.lower())
 

tests/test_prs.py

@@ -4,8 +4,6 @@ Mocks api_request and credentials.
 """
 import sys
 import unittest
-import io
-import contextlib
 from unittest.mock import patch
 
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
@@ -26,16 +24,14 @@ class TestListPRs(unittest.TestCase):
         mock_api.return_value = [
             {"number": 1, "title": "PR 1", "head": {"ref": "branch1"}, "base": {"ref": "main"}, "html_url": "http://url1", "mergeable": True}
         ]
-        with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
+        rc = list_prs.main([])
-            rc = list_prs.main([])
         self.assertEqual(rc, 0)
         mock_api.assert_called_once()
 
     @patch("list_prs.api_request", return_value=[])
     @patch("list_prs.get_auth_header", return_value=FAKE_CREDS)
     def test_list_prs_empty(self, _auth, mock_api):
-        with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
+        rc = list_prs.main(["--state", "closed"])
-            rc = list_prs.main(["--state", "closed"])
         self.assertEqual(rc, 0)
         mock_api.assert_called_once()
 

tests/test_python_cli.py

@@ -4,8 +4,6 @@ All tests mock credentials and API requests so no real network calls are made.
 """
 import sys
 import unittest
-import io
-import contextlib
 from unittest.mock import patch, MagicMock
 
 # The modules under test live in the repo root
@@ -33,8 +31,7 @@ class TestCloseIssueCLI(unittest.TestCase):
     @patch("close_issue.api_request")
     @patch("close_issue.get_auth_header", return_value=FAKE_AUTH)
     def test_successful_close(self, _auth, mock_api):
-        with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
+        rc = close_issue.main(["42"])
-            rc = close_issue.main(["42"])
         self.assertEqual(rc, 0)
         mock_api.assert_called_once()
         url = mock_api.call_args[0][1]
@@ -72,8 +69,7 @@ class TestMarkIssueCLI(unittest.TestCase):
             [{"id": 101, "name": "status:in-progress"}],
             [{"name": "status:in-progress"}],
         ]
-        with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
+        rc = mark_issue.main(["15", "start"])
-            rc = mark_issue.main(["15", "start"])
         self.assertEqual(rc, 0)
         self.assertEqual(mock_api.call_count, 2)
         

tests/test_review_pr.py

@@ -2,8 +2,6 @@
 
 Mocks api_request and credentials.
 """
-import io
-import os
 import sys
 import unittest
 from unittest.mock import patch
@@ -29,11 +27,6 @@ FAKE_PR_DATA = {
 
 class TestArgParsing(unittest.TestCase):
 
-    def setUp(self):
-        self.exists_patcher = patch("os.path.exists", return_value=False)
-        self.exists_patcher.start()
-        self.addCleanup(self.exists_patcher.stop)
-
     @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
     def test_missing_pr_number_exits(self, _auth):
         with self.assertRaises(SystemExit):
@@ -42,11 +35,6 @@ class TestArgParsing(unittest.TestCase):
 
 class TestAPIPayload(unittest.TestCase):
 
-    def setUp(self):
-        self.exists_patcher = patch("os.path.exists", return_value=False)
-        self.exists_patcher.start()
-        self.addCleanup(self.exists_patcher.stop)
-
     @patch("review_pr.api_request")
     @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
     def test_payload_fields_and_workflow(self, _auth, mock_api):
@@ -91,19 +79,15 @@ class TestAPIPayload(unittest.TestCase):
         self.assertEqual(mock_api.call_count, 0)
 
     def test_merge_flag_message_points_to_gated_workflow(self):
-        from _pytest.monkeypatch import MonkeyPatch
         import io
+        import contextlib
         with patch("review_pr.get_auth_header", return_value=FAKE_CREDS), \
                 patch("review_pr.api_request") as mock_api:
             buf = io.StringIO()
-            monkeypatch = MonkeyPatch()
+            with contextlib.redirect_stderr(buf):
-            monkeypatch.setattr(sys, "stderr", buf)
-            try:
                 rc = review_pr.main([
                     "--pr-number", "81", "--event", "APPROVE", "--merge",
                 ])
-            finally:
-                monkeypatch.undo()
         self.assertEqual(rc, 2)
         self.assertEqual(mock_api.call_count, 0)
         msg = buf.getvalue().lower()
@@ -111,70 +95,5 @@ class TestAPIPayload(unittest.TestCase):
         self.assertIn("gitea_merge_pr", msg)
 
 
-class TestMutationAuthorityLock(unittest.TestCase):
-    """Issue #194: verify that the CLI tool rejects profile overrides when mismatched with lock."""
-
-    @patch("review_pr.get_profile")
-    def test_cli_blocked_on_profile_mismatch(self, mock_get_profile):
-        mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
-
-        original_exists = os.path.exists
-        def conditional_exists(path):
-            if "gitea_mutation_authority.lock" in str(path):
-                return True
-            return original_exists(path)
-
-        original_open = open
-        def conditional_open(file, *args, **kwargs):
-            if "gitea_mutation_authority.lock" in str(file):
-                return io.StringIO('{"current_profile": "prgs-author"}')
-            return original_open(file, *args, **kwargs)
-
-        from _pytest.monkeypatch import MonkeyPatch
-        import io
-        buf = io.StringIO()
-        monkeypatch = MonkeyPatch()
-        monkeypatch.setattr(sys, "stderr", buf)
-
-        with patch("os.path.exists", side_effect=conditional_exists), \
-             patch("builtins.open", side_effect=conditional_open):
-            try:
-                rc = review_pr.main([
-                    "--pr-number", "81", "--event", "APPROVE",
-                ])
-            finally:
-                monkeypatch.undo()
-
-        self.assertEqual(rc, 3)
-        msg = buf.getvalue().lower()
-        self.assertIn("cli override rejected", msg)
-
-    @patch("review_pr.get_profile")
-    @patch("review_pr.api_request")
-    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
-    def test_cli_allowed_on_profile_match(self, _auth, mock_api, mock_get_profile):
-        mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
-        mock_api.side_effect = [FAKE_PR_DATA, {}]
-
-        original_exists = os.path.exists
-        def conditional_exists(path):
-            if "gitea_mutation_authority.lock" in str(path):
-                return True
-            return original_exists(path)
-
-        original_open = open
-        def conditional_open(file, *args, **kwargs):
-            if "gitea_mutation_authority.lock" in str(file):
-                return io.StringIO('{"current_profile": "prgs-reviewer"}')
-            return original_open(file, *args, **kwargs)
-
-        with patch("os.path.exists", side_effect=conditional_exists), \
-             patch("builtins.open", side_effect=conditional_open):
-            rc = review_pr.main([
-                "--pr-number", "81", "--event", "APPROVE",
-            ])
-        self.assertEqual(rc, 0)
-
-
 if __name__ == "__main__":
     unittest.main()

tests/test_review_proofs.py

@@ -14,7 +14,6 @@ each precondition of a blind queue review:
 
 These are the harness assertions from the issue's Required behavior 7.
 """
-import io
 import sys
 import unittest
 
@@ -26,7 +25,6 @@ from review_proofs import (  # noqa: E402
     assess_self_review_contamination,
     assess_validation_report,
     build_final_report,
-    resolve_repos_from_user_reference,
     verify_pinned_head_checkout,
 )
 
@@ -466,117 +464,6 @@ class TestFinalReport(unittest.TestCase):
         self.assertFalse(report["merge_allowed"])
 
 
-class TestStdoutIsolation(unittest.TestCase):
-    """Regression test for #178: tests must not close or corrupt stdout/stderr
-    (prevents need for junitxml workaround in full suite runs and review validation).
-    """
-
-    def test_stdout_remains_usable(self):
-        """After typical test activity (mocks, redirects, prints from mains), stdout should be usable."""
-        # Verify not closed
-        self.assertFalse(getattr(sys.stdout, "closed", False))
-        # Should be able to write (even if captured by pytest)
-        try:
-            sys.stdout.write("")
-            sys.stdout.flush()
-        except Exception as exc:
-            self.fail(f"stdout write failed after test activity: {exc}")
-
-    def test_stderr_remains_usable(self):
-        self.assertFalse(getattr(sys.stderr, "closed", False))
-        try:
-            sys.stderr.write("")
-            sys.stderr.flush()
-        except Exception as exc:
-            self.fail(f"stderr write failed: {exc}")
-
-
-class TestRepoNameDisambiguation(unittest.TestCase):
-    """Harness assertions for repo name disambiguation (new blind-review hardening).
-
-    "MCP Gitea tool" etc. must resolve to Gitea-Tools, not silently default to
-    mcp-control-plane. "open PRs" or ambiguous must check both. Single-repo
-    zero-result must not hide PRs in the other configured repo.
-    """
-
-    CONFIGURED = [
-        "Scaled-Tech-Consulting/Gitea-Tools",
-        "Scaled-Tech-Consulting/mcp-control-plane",
-    ]
-
-    def test_gitea_tools_aliases_resolve_to_gitea_tools_only(self):
-        for ref in [
-            "MCP Gitea tool",
-            "gitea tool",
-            "Gitea-Tools",
-            "gitea mcp tool",
-            "gitea-tools repo",
-        ]:
-            result = resolve_repos_from_user_reference(ref, self.CONFIGURED)
-            self.assertEqual(result, ["Scaled-Tech-Consulting/Gitea-Tools"])
-
-    def test_mcp_control_plane_alias_resolves_only_to_it(self):
-        result = resolve_repos_from_user_reference(
-            "mcp-control-plane", self.CONFIGURED
-        )
-        self.assertEqual(result, ["Scaled-Tech-Consulting/mcp-control-plane"])
-
-    def test_ambiguous_or_empty_defaults_to_both(self):
-        self.assertEqual(
-            resolve_repos_from_user_reference("open PRs", self.CONFIGURED),
-            self.CONFIGURED,
-        )
-        self.assertEqual(
-            resolve_repos_from_user_reference("", self.CONFIGURED),
-            self.CONFIGURED,
-        )
-        self.assertEqual(
-            resolve_repos_from_user_reference("review the queue", self.CONFIGURED),
-            self.CONFIGURED,
-        )
-
-    def test_single_repo_zero_result_does_not_hide_other(self):
-        # Simulate a run that only inventoried mcp because of bad alias resolution.
-        # The inventory must still require Gitea-Tools to claim "no open PRs".
-        mcp_only_reports = [
-            {
-                "repo": "Scaled-Tech-Consulting/mcp-control-plane",
-                "state_filter": "open",
-                "pagination_complete": True,
-                "open_pr_count": 0,
-            }
-        ]
-        result = assess_inventory_completeness(
-            repo_reports=mcp_only_reports,
-            required_repos=self.CONFIGURED,
-        )
-        self.assertFalse(result["complete"])
-        self.assertTrue(
-            any("Gitea-Tools" in r for r in result["reasons"])
-        )
-
-    def test_full_inventory_of_both_is_required_for_exhaustive_claim(self):
-        both_reports = [
-            {
-                "repo": "Scaled-Tech-Consulting/Gitea-Tools",
-                "state_filter": "open",
-                "pagination_complete": True,
-                "open_pr_count": 1,
-            },
-            {
-                "repo": "Scaled-Tech-Consulting/mcp-control-plane",
-                "state_filter": "open",
-                "pagination_complete": True,
-                "open_pr_count": 0,
-            },
-        ]
-        result = assess_inventory_completeness(
-            repo_reports=both_reports, required_repos=self.CONFIGURED
-        )
-        self.assertTrue(result["complete"])
-        self.assertTrue(result["can_claim_exhaustive"])
-
-
 class TestControllerHandoff(unittest.TestCase):
     """Issue #182: final reports must end with a Controller Handoff."""
 
@@ -657,40 +544,6 @@ class TestControllerHandoff(unittest.TestCase):
         self.assertEqual(result["verdict"], "incomplete")
         self.assertIn("No review/merge confirmation", result["missing_fields"])
 
-    def test_author_role_rejects_equivalent_or_multiple_issues(self):
-        # 1. equivalent reference blocked
-        incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
-            "- Selected issue: Issue #194 / #196 equivalent",
-            "- Claim/comment status: comment-claimed",
-            "- PR number opened: #999",
-            "- No review/merge: confirmed",
-        ])
-        res = assess_controller_handoff(incomplete_eq, role="author")
-        self.assertEqual(res["verdict"], "incomplete")
-        self.assertIn("Selected issue", res["missing_fields"])
-
-        # 2. multiple issues blocked
-        incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([
-            "- Selected issue: #194, #196",
-            "- Claim/comment status: comment-claimed",
-            "- PR number opened: #999",
-            "- No review/merge: confirmed",
-        ])
-        res = assess_controller_handoff(incomplete_multi, role="author")
-        self.assertEqual(res["verdict"], "incomplete")
-        self.assertIn("Selected issue", res["missing_fields"])
-
-    def test_author_role_rejects_fuzzy_pr_number(self):
-        incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([
-            "- Selected issue: #196",
-            "- Claim/comment status: comment-claimed",
-            "- PR number opened: PR #203 / #204 equivalent",
-            "- No review/merge: confirmed",
-        ])
-        res = assess_controller_handoff(incomplete_pr, role="author")
-        self.assertEqual(res["verdict"], "incomplete")
-        self.assertIn("PR number opened", res["missing_fields"])
-
     def test_inventory_role_requires_inventory_fields(self):
         complete = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Repositories checked: Gitea-Tools, mcp-control-plane",

tests/test_worktrees.py

@@ -14,39 +14,11 @@ BRANCHES = REPO / "branches"
 
 
 def run(script, *args):
-    branch = None
+    proc = subprocess.run(
-    for arg in args:
+        ["bash", str(SCRIPTS / script), *args],
-        if not arg.startswith("-"):
+        capture_output=True, text=True, cwd=str(REPO),
-            branch = arg
+    )
-            break
+    return proc.returncode, proc.stdout, proc.stderr
-
-    lock_file = Path("/tmp/gitea_issue_lock.json")
-    created_lock = False
-    if script == "worktree-start" and branch:
-        import re
-        import json
-        m = re.search(r"issue-(\d+)", branch)
-        if not m:
-            m = re.search(r"pr-(\d+)", branch)
-        issue_num = int(m.group(1)) if m else 999
-        lock_file.write_text(json.dumps({
-            "issue_number": issue_num,
-            "branch_name": branch,
-            "remote": "prgs",
-            "org": "Scaled-Tech-Consulting",
-            "repo": "Gitea-Tools"
-        }), encoding="utf-8")
-        created_lock = True
-
-    try:
-        proc = subprocess.run(
-            ["bash", str(SCRIPTS / script), *args],
-            capture_output=True, text=True, cwd=str(REPO),
-        )
-        return proc.returncode, proc.stdout, proc.stderr
-    finally:
-        if created_lock and lock_file.exists():
-            lock_file.unlink()
 
 
 class TestWorktreeStart(unittest.TestCase):