feat(preflight): block workspace edits before identity/capability verification (Closes #210 )

2026-07-05 16:52:20 -04:00
8 changed files with 117 additions and 714 deletions
@@ -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
@@ -91,11 +31,20 @@ PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
 if PROJECT_ROOT not in sys.path:
    sys.path.insert(0, PROJECT_ROOT)
-PREFLIGHT_FILE = "/tmp/gitea_preflight_check.json"
+import json
 _preflight_whoami_called = False
 _preflight_capability_called = False
 _preflight_whoami_violation = False
 _preflight_capability_violation = False
 _preflight_resolved_role = None
 def record_preflight_check(type_name: str, resolved_role: str | None = None):
    """Record a pre-flight check (whoami or capability) and check for workspace edits."""
-    import time
+    global _preflight_whoami_called, _preflight_capability_called
    global _preflight_whoami_violation, _preflight_capability_violation
    global _preflight_resolved_role
    is_dirty = False
    in_test = "pytest" in sys.modules or "unittest" in sys.modules
    if in_test and not os.environ.get("GITEA_TEST_FORCE_DIRTY"):
@@ -113,76 +62,49 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
        except Exception:
            pass
    data = {}
    if os.path.exists(PREFLIGHT_FILE):
        try:
            with open(PREFLIGHT_FILE, "r", encoding="utf-8") as f:
                data = json.load(f)
        except Exception:
            pass
    if is_dirty:
-        if type_name == "whoami" and not data.get("whoami_called"):
+        if type_name == "whoami" and not _preflight_whoami_called:
-            data["whoami_preflight_violation"] = True
+            _preflight_whoami_violation = True
-        if type_name == "capability" and not data.get("capability_called"):
+        if type_name == "capability" and not _preflight_capability_called:
-            data["capability_preflight_violation"] = True
+            _preflight_capability_violation = True
    if type_name == "whoami":
-        data["whoami_called"] = True
+        _preflight_whoami_called = True
        data["whoami_timestamp"] = time.time()
    elif type_name == "capability":
-        data["capability_called"] = True
+        _preflight_capability_called = True
        data["capability_timestamp"] = time.time()
        if resolved_role:
-            data["role"] = resolved_role
+            _preflight_resolved_role = resolved_role
    try:
        with open(PREFLIGHT_FILE, "w", encoding="utf-8") as f:
            json.dump(data, f)
    except Exception:
        pass
 def verify_preflight_purity(remote: str | None = None):
    """Verify that identity and capability were verified prior to edits, and that reviewers made no edits."""
    in_test = "pytest" in sys.modules or "unittest" in sys.modules
    if in_test and not os.environ.get("GITEA_TEST_FORCE_DIRTY"):
        return
    if not os.path.exists(PREFLIGHT_FILE):
        raise RuntimeError("Pre-flight order violation: Identity and capability verification were skipped (fail closed)")
-    try:
+    if not _preflight_whoami_called:
        with open(PREFLIGHT_FILE, "r", encoding="utf-8") as f:
            data = json.load(f)
    except Exception as e:
        raise RuntimeError(f"Could not read pre-flight check record: {e} (fail closed)")
    if not data.get("whoami_called"):
        raise RuntimeError("Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)")
-    if not data.get("capability_called"):
+    if not _preflight_capability_called:
        raise RuntimeError("Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)")
-    if data.get("whoami_preflight_violation"):
+    if _preflight_whoami_violation:
        raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_whoami verification (fail closed)")
-    if data.get("capability_preflight_violation"):
+    if _preflight_capability_violation:
        raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_resolve_task_capability verification (fail closed)")
    is_dirty = False
-    if os.environ.get("GITEA_TEST_ENVIRONMENT") == "1" and not os.environ.get("GITEA_TEST_FORCE_DIRTY"):
+    try:
-        is_dirty = False
+        res = subprocess.run(
-    else:
+            ["git", "status", "--porcelain"],
-        try:
+            capture_output=True, text=True, cwd=PROJECT_ROOT
-            res = subprocess.run(
+        )
-                ["git", "status", "--porcelain"],
+        for line in res.stdout.splitlines():
-                capture_output=True, text=True, cwd=PROJECT_ROOT
+            if line and not line.startswith("??"):
-            )
+                is_dirty = True
-            for line in res.stdout.splitlines():
+                break
-                if line and not line.startswith("??"):
+    except Exception:
-                    is_dirty = True
+        pass
                    break
        except Exception:
            pass
-    if data.get("role") == "reviewer" and is_dirty:
+    if _preflight_resolved_role == "reviewer" and is_dirty:
        raise RuntimeError("Reviewer role violation: Reviewer profile is forbidden from modifying tracked workspace files (fail closed)")
 from mcp.server.fastmcp import FastMCP  # noqa: E402
@@ -478,8 +400,8 @@ def gitea_create_issue(
    Returns:
        dict with 'number' of the created issue ('url' only with the reveal opt-in).
    """
    h, o, r = _resolve(remote, host, org, repo)
    verify_preflight_purity(remote)
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    url = f"{repo_api_url(h, o, r)}/issues"
    try:
@@ -495,84 +417,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,
@@ -599,43 +443,8 @@ def gitea_create_pr(
    Returns:
        dict with 'number' of the created PR ('url' only with the reveal opt-in).
    """
    h, o, r = _resolve(remote, host, org, repo)
    verify_preflight_purity(remote)
-
+    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}
@@ -1261,12 +1070,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(
@@ -1475,8 +1278,8 @@ def gitea_commit_files(
    Returns:
        dict with success status and commit/branch information.
    """
    h, o, r = _resolve(remote, host, org, repo)
    verify_preflight_purity(remote)
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    url = f"{repo_api_url(h, o, r)}/contents"
@@ -1589,12 +1392,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(
@@ -2841,9 +2638,9 @@ def gitea_whoami(
        'email', 'server', 'remote', and 'profile' (safe runtime profile
        metadata: profile_name + allowed_operations; never the token).
    """
    record_preflight_check("whoami")
    if remote not in REMOTES:
        raise ValueError(f"Unknown remote '{remote}'. Choose from: {list(REMOTES)}")
    record_preflight_check("whoami")
    h = host or REMOTES[remote]["host"]
    auth = _auth(h)
    url = gitea_url(h, "/api/v1/user")
@@ -3272,25 +3069,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,6 +3458,7 @@ def gitea_resolve_task_capability(
    required_permission = TASK_MAP[task]["permission"]
    required_role = TASK_MAP[task]["role"]
    record_preflight_check("capability", required_role)
    profile = get_profile()
@@ -3781,8 +3560,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,
@@ -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 == "-":
@@ -593,7 +593,6 @@ def assess_controller_handoff(report_text, role=None, local_edits=False):
    The handoff supplements the full report; this helper never validates
    the full report body, only the continuation summary.
    """
    import re
    section = _handoff_section_lines(report_text)
    if section is None:
        return {
@@ -634,27 +633,6 @@ def assess_controller_handoff(report_text, role=None, local_edits=False):
                        for m in missing],
        }
    # Validate issue/PR references for exact number and no forbidden terms (Issue #194 / #196)
    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_list in required:
                    if alias in aliases_list:
                        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}')"
                    ],
                }
    if local_edits:
        workspace_mutations_val = fields_dict.get("workspace mutations", "").strip().lower()
        if not workspace_mutations_val or workspace_mutations_val == "none":
@@ -675,7 +653,6 @@ def assess_controller_handoff(report_text, role=None, local_edits=False):
    }
 # ── PR Inventory Trust Gate (Issue #194) ──────────────────────────────────────
 #
 # A reviewer agent may not convert an empty PR list response into a definitive
@@ -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
    :
@@ -5,7 +5,6 @@ the MCP protocol) with mocked API responses.
 """
 import json
 import os
 os.environ["GITEA_TEST_ENVIRONMENT"] = "1"
 import sys
 import unittest
 from unittest.mock import patch, MagicMock
@@ -34,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"
@@ -92,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]
@@ -107,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"])
@@ -2311,253 +2298,86 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase):
        self.assertTrue(ok)
-class TestVerifyMutationAuthority(unittest.TestCase):
+# ---------------------------------------------------------------------------
-    """Test verification lock logic under various configurations."""
+# Pre-flight ordering and workspace edit block (#210)
-
+# ---------------------------------------------------------------------------
    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))
 class TestPreflightVerification(unittest.TestCase):
-    """Test workspace edits and pre-flight ordering verification."""
+    """Assert that pre-flight verification gates order correctly."""
    def setUp(self):
-        self.preflight_path = "/tmp/gitea_preflight_check.json"
+        # Save real global variables
-        if os.path.exists(self.preflight_path):
+        import mcp_server
-            os.remove(self.preflight_path)
+        self.orig_whoami_called = mcp_server._preflight_whoami_called
-        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        self.orig_capability_called = mcp_server._preflight_capability_called
        self.orig_whoami_violation = mcp_server._preflight_whoami_violation
        self.orig_capability_violation = mcp_server._preflight_capability_violation
        self.orig_resolved_role = mcp_server._preflight_resolved_role
        # Reset state for each test
        mcp_server._preflight_whoami_called = False
        mcp_server._preflight_capability_called = False
        mcp_server._preflight_whoami_violation = False
        mcp_server._preflight_capability_violation = False
        mcp_server._preflight_resolved_role = None
    def tearDown(self):
-        if os.path.exists(self.preflight_path):
+        # Restore real global variables
-            os.remove(self.preflight_path)
+        import mcp_server
-        os.environ.pop("GITEA_TEST_FORCE_DIRTY", None)
+        mcp_server._preflight_whoami_called = self.orig_whoami_called
        mcp_server._preflight_capability_called = self.orig_capability_called
        mcp_server._preflight_whoami_violation = self.orig_whoami_violation
        mcp_server._preflight_capability_violation = self.orig_capability_violation
        mcp_server._preflight_resolved_role = self.orig_resolved_role
        if "GITEA_TEST_FORCE_DIRTY" in os.environ:
            del os.environ["GITEA_TEST_FORCE_DIRTY"]
-    @patch("subprocess.run")
+    def test_preflight_whoami_violation(self):
-    def test_record_preflight_detects_violation_whoami(self, mock_run):
+        import mcp_server
-        mock_run.return_value = MagicMock(stdout="M mcp_server.py\n")
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
-        from mcp_server import record_preflight_check, verify_preflight_purity
+        mcp_server._preflight_capability_called = True
-        record_preflight_check("whoami")
+        mcp_server.record_preflight_check("whoami")
-        with open(self.preflight_path, "r") as f:
+        self.assertTrue(mcp_server._preflight_whoami_violation)
            data = json.load(f)
        self.assertTrue(data.get("whoami_preflight_violation"))
        self.assertTrue(data.get("whoami_called"))
        data["capability_called"] = True
        with open(self.preflight_path, "w") as f:
            json.dump(data, f)
        with self.assertRaises(RuntimeError) as ctx:
-            verify_preflight_purity("prgs")
+            mcp_server.verify_preflight_purity()
-        self.assertIn("Workspace file edits occurred before gitea_whoami", str(ctx.exception))
+        self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception))
-    @patch("subprocess.run")
+    def test_preflight_capability_violation(self):
-    def test_record_preflight_detects_violation_capability(self, mock_run):
+        import mcp_server
-        mock_run.return_value = MagicMock(stdout="M mcp_server.py\n")
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
-        from mcp_server import record_preflight_check, verify_preflight_purity
+        mcp_server._preflight_whoami_called = True
-        record_preflight_check("capability", resolved_role="author")
+        mcp_server.record_preflight_check("capability", resolved_role="author")
-        with open(self.preflight_path, "r") as f:
+        self.assertTrue(mcp_server._preflight_capability_violation)
            data = json.load(f)
        self.assertTrue(data.get("capability_preflight_violation"))
        self.assertTrue(data.get("capability_called"))
        self.assertEqual(data.get("role"), "author")
        data["whoami_called"] = True
        with open(self.preflight_path, "w") as f:
            json.dump(data, f)
        with self.assertRaises(RuntimeError) as ctx:
-            verify_preflight_purity("prgs")
+            mcp_server.verify_preflight_purity()
-        self.assertIn("Workspace file edits occurred before gitea_resolve_task_capability", str(ctx.exception))
+        self.assertIn("Workspace file edits occurred before gitea_resolve_task_capability verification", str(ctx.exception))
-    @patch("subprocess.run")
+    def test_preflight_not_called_fails_closed(self):
-    def test_verify_preflight_reviewer_edits_blocked(self, mock_run):
+        import mcp_server
-        with open(self.preflight_path, "w") as f:
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
            json.dump({
                "whoami_called": True,
                "capability_called": True,
                "role": "reviewer"
            }, f)
        mock_run.return_value = MagicMock(stdout="M review_proofs.py\n")
        from mcp_server import verify_preflight_purity
        with self.assertRaises(RuntimeError) as ctx:
-            verify_preflight_purity("prgs")
+            mcp_server.verify_preflight_purity()
        self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))
        mcp_server._preflight_whoami_called = True
        with self.assertRaises(RuntimeError) as ctx2:
            mcp_server.verify_preflight_purity()
        self.assertIn("Task capability (gitea_resolve_task_capability) has not been resolved", str(ctx2.exception))
    def test_preflight_reviewer_mutation_violation(self):
        import mcp_server
        mcp_server._preflight_whoami_called = True
        mcp_server._preflight_capability_called = True
        mcp_server._preflight_resolved_role = "reviewer"
        # When dirty, reviewer edits are blocked
        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
        with self.assertRaises(RuntimeError) as ctx:
            mcp_server.verify_preflight_purity()
        self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception))
-    @patch("subprocess.run")
+        # When clean, reviewer is allowed
-    def test_verify_preflight_clean_reviewer_allowed(self, mock_run):
+        del os.environ["GITEA_TEST_FORCE_DIRTY"]
-        with open(self.preflight_path, "w") as f:
+        mcp_server.verify_preflight_purity()
            json.dump({
                "whoami_called": True,
                "capability_called": True,
                "role": "reviewer"
            }, f)
        mock_run.return_value = MagicMock(stdout="")
        from mcp_server import verify_preflight_purity
        verify_preflight_purity("prgs")
    def test_verify_preflight_skipped_fails(self):
        from mcp_server import verify_preflight_purity
        with self.assertRaises(RuntimeError) as ctx:
            verify_preflight_purity("prgs")
        self.assertIn("verification were skipped", str(ctx.exception))
@@ -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):
@@ -111,70 +99,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()
@@ -776,40 +776,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",
@@ -854,6 +820,7 @@ class TestControllerHandoff(unittest.TestCase):
        self.assertEqual(res2["verdict"], "complete")
 class TestPRInventoryTrustGate(unittest.TestCase):
    """Issue #194: unit tests for the PR inventory trust gate."""
@@ -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):