From f2c8a8d5c1adbfda97a3fc59594bb4c9cb226db5 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 6 Jul 2026 12:34:45 -0400 Subject: [PATCH] fix: scope pre-flight workspace checks to session-owned deltas (#252) Pre-flight order tracking now compares porcelain baselines captured at process start, whoami, and capability resolution instead of treating any shared-worktree dirt as a sticky violation. Fresh gitea_whoami re-evaluates and clears violations; runtime context reports preflight_block_reasons when mutations would fail. Refs #252 Co-Authored-By: Claude Opus 4.8 (1M context) --- gitea_mcp_server.py | 225 ++++++++++++++++++++++++++++++--------- tests/test_mcp_server.py | 81 +++++++++++--- 2 files changed, 243 insertions(+), 63 deletions(-) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index f173581..d9641ac 100644 --- a/gitea_mcp_server.py +++ b/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: diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 328e749..133e515 100644 --- a/tests/test_mcp_server.py +++ b/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])