gitea_mcp_server.py

@@ -164,81 +164,194 @@ _preflight_capability_called = False
 _preflight_whoami_violation = False
 _preflight_capability_violation = False
 _preflight_resolved_role = None
+_process_start_porcelain: str | None = None
+_preflight_whoami_baseline_porcelain: str | None = None
+_preflight_capability_baseline_porcelain: str | None = None
+_preflight_whoami_violation_files: list[str] = []
+_preflight_capability_violation_files: list[str] = []
+_preflight_reviewer_violation_files: list[str] = []
+
+
+def _preflight_in_test_mode() -> bool:
+    return "pytest" in sys.modules or "unittest" in sys.modules
+
+
+def _ensure_process_start_porcelain() -> str:
+    """Capture the shared-worktree baseline once per MCP process (#252)."""
+    global _process_start_porcelain
+    if _process_start_porcelain is None:
+        _process_start_porcelain = _get_workspace_porcelain()
+    return _process_start_porcelain
+
+
+def _get_workspace_porcelain() -> str:
+    """Return tracked-workspace porcelain for pre-flight attribution."""
+    if os.environ.get("GITEA_TEST_FORCE_DIRTY"):
+        return " M __gitea_test_force_dirty__.py\n"
+    override = os.environ.get("GITEA_TEST_PORCELAIN")
+    if override is not None:
+        return override
+    try:
+        res = subprocess.run(
+            ["git", "status", "--porcelain"],
+            capture_output=True,
+            text=True,
+            cwd=PROJECT_ROOT,
+        )
+        return res.stdout or ""
+    except Exception:
+        return ""
+
+
+def _parse_porcelain_entries(porcelain: str) -> dict[str, str]:
+    """Map tracked path -> full porcelain line (untracked ``??`` ignored)."""
+    entries: dict[str, str] = {}
+    for line in (porcelain or "").splitlines():
+        if not line or len(line) < 4 or line.startswith("??"):
+            continue
+        path = line[3:].strip()
+        if " -> " in path:
+            path = path.split(" -> ", 1)[1].strip()
+        if path:
+            entries[path] = line
+    return entries
+
+
+def _new_tracked_changes_since(baseline: str, current: str) -> list[str]:
+    """Tracked paths that are new or changed since *baseline* porcelain."""
+    base = _parse_porcelain_entries(baseline)
+    cur = _parse_porcelain_entries(current)
+    changed = [
+        path for path, line in cur.items()
+        if path not in base or base[path] != line
+    ]
+    return sorted(changed)
+
+
+def _format_preflight_files(files: list[str]) -> str:
+    if not files:
+        return "(none)"
+    return ", ".join(files)
+
+
+def assess_preflight_status() -> dict:
+    """Non-throwing pre-flight readiness for runtime-context alignment (#252)."""
+    reasons: list[str] = []
+    if not _preflight_whoami_called:
+        reasons.append(
+            "Identity (gitea_whoami) has not been verified"
+        )
+    if not _preflight_capability_called:
+        reasons.append(
+            "Task capability (gitea_resolve_task_capability) has not been resolved"
+        )
+    if _preflight_whoami_violation:
+        reasons.append(
+            "Workspace file edits occurred before gitea_whoami verification "
+            f"(offending files: {_format_preflight_files(_preflight_whoami_violation_files)})"
+        )
+    if _preflight_capability_violation:
+        reasons.append(
+            "Workspace file edits occurred before gitea_resolve_task_capability verification "
+            f"(offending files: {_format_preflight_files(_preflight_capability_violation_files)})"
+        )
+    if _preflight_reviewer_violation_files:
+        reasons.append(
+            "Reviewer profile modified tracked workspace files after capability resolution "
+            f"(offending files: {_format_preflight_files(_preflight_reviewer_violation_files)})"
+        )
+    return {
+        "preflight_ready": not reasons,
+        "preflight_block_reasons": reasons,
+        "preflight_whoami_verified": _preflight_whoami_called,
+        "preflight_capability_resolved": _preflight_capability_called,
+        "preflight_whoami_violation_files": list(_preflight_whoami_violation_files),
+        "preflight_capability_violation_files": list(_preflight_capability_violation_files),
+        "preflight_reviewer_violation_files": list(_preflight_reviewer_violation_files),
+    }
+
 
 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."""
+    """Record a pre-flight check (whoami or capability) with session-scoped deltas."""
     global _preflight_whoami_called, _preflight_capability_called
     global _preflight_whoami_violation, _preflight_capability_violation
     global _preflight_resolved_role
+    global _preflight_whoami_baseline_porcelain, _preflight_capability_baseline_porcelain
+    global _preflight_whoami_violation_files, _preflight_capability_violation_files
+    global _preflight_reviewer_violation_files
 
-    in_test = "pytest" in sys.modules or "unittest" in sys.modules
-    if os.environ.get("GITEA_TEST_FORCE_DIRTY"):
-        is_dirty = True
-    elif in_test:
-        is_dirty = False
-    else:
-        is_dirty = False
-        try:
-            res = subprocess.run(
-                ["git", "status", "--porcelain"],
-                capture_output=True, text=True, cwd=PROJECT_ROOT
-            )
-            for line in res.stdout.splitlines():
-                if line and not line.startswith("??"):
-                    is_dirty = True
-                    break
-        except Exception:
-            pass
-
-    if is_dirty:
-        if type_name == "whoami" and not _preflight_whoami_called:
-            _preflight_whoami_violation = True
-        if type_name == "capability" and not _preflight_capability_called:
-            _preflight_capability_violation = True
+    current = _get_workspace_porcelain()
 
     if type_name == "whoami":
+        # Fresh whoami restarts the capability step and re-evaluates violations
+        # instead of replaying a sticky record (#252).
+        _preflight_capability_called = False
+        _preflight_capability_violation = False
+        _preflight_capability_violation_files = []
+        _preflight_capability_baseline_porcelain = None
+        _preflight_reviewer_violation_files = []
+
+        process_start = _ensure_process_start_porcelain()
+        whoami_delta = _new_tracked_changes_since(process_start, current)
+        _preflight_whoami_violation = bool(whoami_delta)
+        _preflight_whoami_violation_files = whoami_delta
+        _preflight_whoami_baseline_porcelain = current
         _preflight_whoami_called = True
     elif type_name == "capability":
+        baseline = _preflight_whoami_baseline_porcelain or ""
+        capability_delta = _new_tracked_changes_since(baseline, current)
+        _preflight_capability_violation = bool(capability_delta)
+        _preflight_capability_violation_files = capability_delta
+        _preflight_capability_baseline_porcelain = current
         _preflight_capability_called = True
         if resolved_role:
             _preflight_resolved_role = resolved_role
 
+
 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"):
+    """Verify that identity and capability were verified prior to session edits."""
+    global _preflight_reviewer_violation_files
+
+    in_test = _preflight_in_test_mode()
+    if in_test and not (
+        os.environ.get("GITEA_TEST_FORCE_DIRTY")
+        or os.environ.get("GITEA_TEST_PORCELAIN") is not None
+    ):
         return
 
     if not _preflight_whoami_called:
-        raise RuntimeError("Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)")
+        raise RuntimeError(
+            "Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)"
+        )
     if not _preflight_capability_called:
-        raise RuntimeError("Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)")
+        raise RuntimeError(
+            "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
+        )
 
     if _preflight_whoami_violation:
-        raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_whoami verification (fail closed)")
+        raise RuntimeError(
+            "Pre-flight order violation: Workspace file edits occurred before "
+            f"gitea_whoami verification (fail closed). Offending files: "
+            f"{_format_preflight_files(_preflight_whoami_violation_files)}"
+        )
     if _preflight_capability_violation:
-        raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_resolve_task_capability verification (fail closed)")
+        raise RuntimeError(
+            "Pre-flight order violation: Workspace file edits occurred before "
+            f"gitea_resolve_task_capability verification (fail closed). Offending files: "
+            f"{_format_preflight_files(_preflight_capability_violation_files)}"
+        )
 
-    if os.environ.get("GITEA_TEST_FORCE_DIRTY"):
-        is_dirty = True
-    elif in_test:
-        is_dirty = False
-    else:
-        is_dirty = False
-        try:
-            res = subprocess.run(
-                ["git", "status", "--porcelain"],
-                capture_output=True, text=True, cwd=PROJECT_ROOT
+    if _preflight_resolved_role == "reviewer":
+        current = _get_workspace_porcelain()
+        baseline = _preflight_capability_baseline_porcelain or ""
+        reviewer_delta = _new_tracked_changes_since(baseline, current)
+        _preflight_reviewer_violation_files = reviewer_delta
+        if reviewer_delta:
+            raise RuntimeError(
+                "Reviewer role violation: Reviewer profile is forbidden from modifying "
+                "tracked workspace files (fail closed). Offending files: "
+                f"{_format_preflight_files(reviewer_delta)}"
             )
-            for line in res.stdout.splitlines():
-                if line and not line.startswith("??"):
-                    is_dirty = True
-                    break
-        except Exception:
-            pass
-
-    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
 
@@ -3967,6 +4080,14 @@ def gitea_get_runtime_context(
             "or ask the operator to update GITEA_MCP_PROFILE to a reviewer profile."
         )
 
+    preflight = assess_preflight_status()
+    if not preflight["preflight_ready"]:
+        safe_next_action = (
+            "Complete pre-flight verification before mutating: call gitea_whoami, then "
+            "gitea_resolve_task_capability for the intended task. "
+            f"Blocked: {'; '.join(preflight['preflight_block_reasons'])}"
+        )
+
     result = {
         "active_profile": profile["profile_name"],
         "authenticated_username": username,
@@ -3981,6 +4102,8 @@ def gitea_get_runtime_context(
         "review_merge_blocked_reasons": blocked_reasons,
         "suggested_fix": suggested_fix,
         "safe_next_action": safe_next_action,
+        "preflight_ready": preflight["preflight_ready"],
+        "preflight_block_reasons": preflight["preflight_block_reasons"],
     }
 
     if reveal and h:

tests/test_mcp_server.py

@@ -3158,6 +3158,12 @@ class TestPreflightVerification(unittest.TestCase):
         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
+        self.orig_process_start = mcp_server._process_start_porcelain
+        self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
+        self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
+        self.orig_whoami_files = mcp_server._preflight_whoami_violation_files
+        self.orig_capability_files = mcp_server._preflight_capability_violation_files
+        self.orig_reviewer_files = mcp_server._preflight_reviewer_violation_files
 
         # Reset state for each test
         mcp_server._preflight_whoami_called = False
@@ -3165,6 +3171,13 @@ class TestPreflightVerification(unittest.TestCase):
         mcp_server._preflight_whoami_violation = False
         mcp_server._preflight_capability_violation = False
         mcp_server._preflight_resolved_role = None
+        mcp_server._process_start_porcelain = ""
+        mcp_server._preflight_whoami_baseline_porcelain = None
+        mcp_server._preflight_capability_baseline_porcelain = None
+        mcp_server._preflight_whoami_violation_files = []
+        mcp_server._preflight_capability_violation_files = []
+        mcp_server._preflight_reviewer_violation_files = []
+        os.environ["GITEA_TEST_PORCELAIN"] = ""
 
     def tearDown(self):
         # Restore real global variables
@@ -3174,30 +3187,42 @@ class TestPreflightVerification(unittest.TestCase):
         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"]
+        mcp_server._process_start_porcelain = self.orig_process_start
+        mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
+        mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
+        mcp_server._preflight_whoami_violation_files = self.orig_whoami_files
+        mcp_server._preflight_capability_violation_files = self.orig_capability_files
+        mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files
+        for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
+            if key in os.environ:
+                del os.environ[key]
 
     def test_preflight_whoami_violation(self):
         import mcp_server
         os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
-        mcp_server._preflight_capability_called = True
         mcp_server.record_preflight_check("whoami")
         self.assertTrue(mcp_server._preflight_whoami_violation)
+        mcp_server.record_preflight_check("capability", resolved_role="author")
 
         with self.assertRaises(RuntimeError) as ctx:
             mcp_server.verify_preflight_purity()
         self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception))
+        self.assertIn("Offending files:", str(ctx.exception))
 
     def test_preflight_capability_violation(self):
         import mcp_server
+        mcp_server.record_preflight_check("whoami")
         os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
-        mcp_server._preflight_whoami_called = True
         mcp_server.record_preflight_check("capability", resolved_role="author")
         self.assertTrue(mcp_server._preflight_capability_violation)
 
         with self.assertRaises(RuntimeError) as ctx:
             mcp_server.verify_preflight_purity()
-        self.assertIn("Workspace file edits occurred before gitea_resolve_task_capability verification", str(ctx.exception))
+        self.assertIn(
+            "Workspace file edits occurred before gitea_resolve_task_capability verification",
+            str(ctx.exception),
+        )
+        self.assertIn("Offending files:", str(ctx.exception))
 
     def test_preflight_not_called_fails_closed(self):
         import mcp_server
@@ -3213,16 +3238,48 @@ class TestPreflightVerification(unittest.TestCase):
 
     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"
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check("capability", resolved_role="reviewer")
 
-        # When dirty, reviewer edits are blocked
-        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        # Session-owned reviewer edits after capability are blocked.
+        os.environ["GITEA_TEST_PORCELAIN"] = " M reviewer_edit.py\n"
         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))
+        self.assertIn("reviewer_edit.py", str(ctx.exception))
 
-        # When clean, reviewer is allowed
-        del os.environ["GITEA_TEST_FORCE_DIRTY"]
+        # Foreign pre-existing dirty state does not block when unchanged.
+        os.environ["GITEA_TEST_PORCELAIN"] = ""
         mcp_server.verify_preflight_purity()
+
+    def test_foreign_workspace_edits_do_not_block_clean_reviewer(self):
+        """#252: concurrent-session dirt in the shared worktree is not attributed."""
+        import mcp_server
+        mcp_server._process_start_porcelain = " M foreign_author.py\n"
+        os.environ["GITEA_TEST_PORCELAIN"] = " M foreign_author.py\n"
+        mcp_server.record_preflight_check("whoami")
+        self.assertFalse(mcp_server._preflight_whoami_violation)
+        mcp_server.record_preflight_check("capability", resolved_role="reviewer")
+        self.assertFalse(mcp_server._preflight_capability_violation)
+        mcp_server.verify_preflight_purity()
+
+    def test_fresh_whoami_clears_sticky_violation(self):
+        """#252: re-running whoami re-evaluates instead of replaying sticky state."""
+        import mcp_server
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        mcp_server.record_preflight_check("whoami")
+        self.assertTrue(mcp_server._preflight_whoami_violation)
+
+        del os.environ["GITEA_TEST_FORCE_DIRTY"]
+        os.environ["GITEA_TEST_PORCELAIN"] = ""
+        mcp_server.record_preflight_check("whoami")
+        self.assertFalse(mcp_server._preflight_whoami_violation)
+        mcp_server.record_preflight_check("capability", resolved_role="reviewer")
+        mcp_server.verify_preflight_purity()
+
+    def test_runtime_context_matches_preflight_block(self):
+        """#252: safe_next_action must not claim ready while pre-flight blocks."""
+        import mcp_server
+        status = mcp_server.assess_preflight_status()
+        self.assertFalse(status["preflight_ready"])
+        self.assertIn("gitea_whoami", status["preflight_block_reasons"][0])