Merge pull request 'fix: scope pre-flight workspace checks to session-owned deltas (closes #252)' (#254) from feat/issue-252-preflight-foreign-edits into master
This commit was merged in pull request #254.
This commit is contained in:
+174
-51
@@ -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:
|
||||
|
||||
+69
-12
@@ -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])
|
||||
|
||||
Reference in New Issue
Block a user