Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7966e70db6 | ||
|
|
4f466550ca | ||
|
|
6ac6b9528c | ||
|
|
4dd32bb9f7 | ||
|
|
d5d3331498 |
@@ -31,7 +31,8 @@ Wiki-related issues cannot be closed until the live Wiki is verified — see
|
||||
|
||||
| Repository | `docs/wiki/` source | Gitea Wiki published | Proof |
|
||||
|---|---|---|---|
|
||||
| `Scaled-Tech-Consulting/Gitea-Tools` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/wiki/Home); 10 pages; wiki git log head `11549ee` |
|
||||
| `Scaled-Tech-Consulting/mcp-control-plane` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane/wiki/Home) |
|
||||
| `Scaled-Tech-Consulting/Gitea-Tools` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/wiki/Home); 10 pages; wiki git log head `d1f0693` |
|
||||
| `Scaled-Tech-Consulting/mcp-control-plane` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane/wiki/Home); 10 pages (History, Home, Identity-and-Profiles, MCP-Tools, Open-Decisions, Operator-Guide, Repositories, Runbooks, Safety-and-Gates, Workflow); wiki git log head `ef3dec2` |
|
||||
|
||||
|
||||
Update this table whenever a wiki is published, re-synced, or found stale.
|
||||
+29
-1
@@ -17,6 +17,7 @@ here weakens or replaces them.
|
||||
import re
|
||||
|
||||
import issue_duplicate_gate
|
||||
from reviewer_worktree import assess_reviewer_worktree_proof
|
||||
|
||||
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
|
||||
|
||||
@@ -637,7 +638,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
role_boundary=None, review_mutation=None,
|
||||
report_text=None, review_decision_lock=None,
|
||||
controller_handoff=None, capability_proof=None,
|
||||
sweep_proof=None):
|
||||
sweep_proof=None, worktree_proof=None):
|
||||
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
|
||||
|
||||
Combines the individual proof verdicts into the final-report fields the
|
||||
@@ -707,6 +708,14 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
"downgraded": True,
|
||||
"reasons": ["review mutation proof not provided (#211)"],
|
||||
}
|
||||
if worktree_proof is not None:
|
||||
worktree = assess_reviewer_worktree_proof(worktree_proof)
|
||||
else:
|
||||
worktree = {
|
||||
"proven": False,
|
||||
"block": True,
|
||||
"reasons": ["reviewer worktree proof not provided (#233)"],
|
||||
}
|
||||
|
||||
capability_proven = bool(capability_evidence.get("proven"))
|
||||
sweep_proven = bool(sweep.get("proven"))
|
||||
@@ -719,6 +728,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
"reasons": ["review mutation proof missing"],
|
||||
}
|
||||
review_mutation_complete = bool(review_mutation.get("complete"))
|
||||
worktree_proven = bool(worktree.get("proven"))
|
||||
|
||||
downgrade_reasons = []
|
||||
if not identity_eligible:
|
||||
@@ -772,6 +782,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
if not review_mutation_complete:
|
||||
downgrade_reasons.append("review mutation proof missing or incomplete (#211)")
|
||||
downgrade_reasons.extend(review_mutation.get("reasons", []))
|
||||
if not worktree_proven:
|
||||
downgrade_reasons.append(
|
||||
"reviewer worktree safety proof missing or failed (#233)"
|
||||
)
|
||||
downgrade_reasons.extend(worktree.get("reasons", []))
|
||||
|
||||
merge_allowed = (
|
||||
identity_eligible
|
||||
@@ -782,6 +797,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
and validation.get("verdict") != "invalid"
|
||||
# #179: no merge without a proven final live-state recheck.
|
||||
and live_state_proven
|
||||
and worktree_proven
|
||||
)
|
||||
|
||||
violations = []
|
||||
@@ -825,6 +841,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
"live_state_recheck_proven": live_state_proven,
|
||||
"role_boundary_clean": role_boundary_clean,
|
||||
"review_mutation_complete": review_mutation_complete,
|
||||
"worktree_proof_proven": worktree_proven,
|
||||
"worktree_scratch_used": bool(worktree.get("scratch_used")),
|
||||
"unrelated_mutations_avoided": bool(
|
||||
worktree.get("unrelated_mutations_avoided")
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
@@ -971,6 +992,12 @@ HANDOFF_ROLE_FIELDS = {
|
||||
("Selected PR", ("selected pr",)),
|
||||
("Reviewer eligibility", ("reviewer eligibility", "eligibility")),
|
||||
("Pinned reviewed head", ("pinned reviewed head", "pinned head")),
|
||||
("Worktree path", ("worktree path", "starting worktree path")),
|
||||
("Worktree dirty", ("worktree dirty", "whether worktree was dirty")),
|
||||
("Scratch worktree used", ("scratch worktree used", "scratch clone used",
|
||||
"scratch worktree")),
|
||||
("Unrelated local mutations", ("unrelated local mutations",
|
||||
"unrelated files modified")),
|
||||
("Review decision", ("review decision", "decision")),
|
||||
("Merge result", ("merge result",)),
|
||||
("Linked issue status", ("linked issue status", "linked issue")),
|
||||
@@ -978,6 +1005,7 @@ HANDOFF_ROLE_FIELDS = {
|
||||
),
|
||||
"author": (
|
||||
("Selected issue", ("selected issue",)),
|
||||
("Issue lock proof", ("issue lock proof", "lock before diff")),
|
||||
("Claim/comment status", ("claim/comment status", "claim status",
|
||||
"claim")),
|
||||
("PR number opened", ("pr number opened", "pr opened", "pr number")),
|
||||
|
||||
@@ -0,0 +1,219 @@
|
||||
"""Fail-closed reviewer worktree and local-git safety proofs (#233).
|
||||
|
||||
Reviewer sessions must never stash, reset, or otherwise manipulate unrelated
|
||||
local changes from another session. When the active worktree has dirty tracked
|
||||
files outside the PR scope, the workflow must stop or switch to a disposable
|
||||
scratch worktree (``scripts/worktree-review``).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import shlex
|
||||
|
||||
# Subcommands that mutate unrelated local state — forbidden for reviewers.
|
||||
_FORBIDDEN_REVIEWER_GIT = re.compile(
|
||||
r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?"
|
||||
r"(?:stash(?:\s+(?:push|pop|drop|apply|clear|list))?|"
|
||||
r"checkout\s+--|"
|
||||
r"restore\s+|"
|
||||
r"reset(?:\s+(?:--hard|--soft|--mixed|--merge))?|"
|
||||
r"clean(?:\s+(?:-f|-fd|-fdx|-x|-d|-n))*|"
|
||||
r"cherry-pick|"
|
||||
r"rebase|"
|
||||
r"merge|"
|
||||
r"commit(?:\s+(?:--amend|-a|-am))?|"
|
||||
r"push(?:\s+(?:--force|--force-with-lease))?|"
|
||||
r"branch\s+-[dD]|"
|
||||
r"worktree\s+remove)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
# Read-only git operations reviewers may use for validation.
|
||||
_READONLY_REVIEWER_GIT = re.compile(
|
||||
r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?"
|
||||
r"(?:fetch|status|diff|log|show|rev-parse|branch(?:\s+--show-current)?|"
|
||||
r"worktree\s+list|worktree\s+add)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
def parse_dirty_tracked_files(porcelain: str) -> list[str]:
|
||||
"""Return tracked paths with local modifications from ``git status --porcelain``.
|
||||
|
||||
Untracked entries (``??``) are ignored — they do not block reviewer work
|
||||
when a scratch worktree is used, and authors may have unrelated untracked
|
||||
files without implying reviewer interference.
|
||||
"""
|
||||
paths: list[str] = []
|
||||
for line in (porcelain or "").splitlines():
|
||||
if not line or len(line) < 4:
|
||||
continue
|
||||
if line.startswith("??"):
|
||||
continue
|
||||
path = line[3:].strip()
|
||||
if " -> " in path:
|
||||
path = path.split(" -> ", 1)[1].strip()
|
||||
if path:
|
||||
paths.append(path)
|
||||
return paths
|
||||
|
||||
|
||||
def files_outside_pr_scope(
|
||||
dirty_files: list[str] | None,
|
||||
pr_scope_files: list[str] | None,
|
||||
) -> list[str]:
|
||||
"""Dirty tracked files not explained by the PR diff file set."""
|
||||
dirty = [p for p in (dirty_files or []) if p]
|
||||
scope = {p for p in (pr_scope_files or []) if p}
|
||||
if not dirty:
|
||||
return []
|
||||
if not scope:
|
||||
return list(dirty)
|
||||
return [path for path in dirty if path not in scope]
|
||||
|
||||
|
||||
def is_forbidden_reviewer_git_command(command: str) -> bool:
|
||||
"""True when a shell command would mutate unrelated local/remote git state."""
|
||||
text = (command or "").strip()
|
||||
if not text:
|
||||
return False
|
||||
return bool(_FORBIDDEN_REVIEWER_GIT.search(text))
|
||||
|
||||
|
||||
def is_readonly_reviewer_git_command(command: str) -> bool:
|
||||
"""True when the command is an explicitly allowed read-only git operation."""
|
||||
text = (command or "").strip()
|
||||
if not text:
|
||||
return False
|
||||
if is_forbidden_reviewer_git_command(text):
|
||||
return False
|
||||
return bool(_READONLY_REVIEWER_GIT.search(text))
|
||||
|
||||
|
||||
def assess_reviewer_git_command_log(commands: list[str] | None) -> dict:
|
||||
"""Fail closed when reviewer shell history includes forbidden git mutations."""
|
||||
forbidden = [
|
||||
cmd for cmd in (commands or []) if is_forbidden_reviewer_git_command(cmd)
|
||||
]
|
||||
if forbidden:
|
||||
return {
|
||||
"proven": False,
|
||||
"block": True,
|
||||
"forbidden_commands": forbidden,
|
||||
"reasons": [
|
||||
"reviewer workflow executed forbidden local git mutation: "
|
||||
f"{cmd!r}"
|
||||
for cmd in forbidden
|
||||
],
|
||||
"safe_next_action": (
|
||||
"stop; report worktree interference; do not stash/reset/checkout "
|
||||
"unrelated files — use scripts/worktree-review instead"
|
||||
),
|
||||
}
|
||||
return {
|
||||
"proven": True,
|
||||
"block": False,
|
||||
"forbidden_commands": [],
|
||||
"reasons": [],
|
||||
"safe_next_action": "proceed",
|
||||
}
|
||||
|
||||
|
||||
def assess_reviewer_worktree_proof(proof: dict | None) -> dict:
|
||||
"""Evaluate reviewer worktree safety before checkout/diff/validation/review.
|
||||
|
||||
*proof* keys:
|
||||
- ``worktree_path`` (required)
|
||||
- ``porcelain_status`` or ``dirty_files``
|
||||
- ``pr_scope_files`` (paths in the PR diff)
|
||||
- ``scratch_used`` (bool)
|
||||
- ``scratch_path`` (when scratch_used)
|
||||
- ``git_commands`` (shell commands executed this session)
|
||||
- ``unrelated_mutations_claimed`` (bool) — stash/reset/drop reported
|
||||
"""
|
||||
proof = dict(proof or {})
|
||||
reasons: list[str] = []
|
||||
worktree_path = (proof.get("worktree_path") or "").strip()
|
||||
if not worktree_path:
|
||||
reasons.append("reviewer worktree path not reported; fail closed")
|
||||
|
||||
if proof.get("dirty_files") is not None:
|
||||
dirty_files = list(proof.get("dirty_files") or [])
|
||||
else:
|
||||
dirty_files = parse_dirty_tracked_files(proof.get("porcelain_status") or "")
|
||||
|
||||
pr_scope = list(proof.get("pr_scope_files") or [])
|
||||
unrelated = files_outside_pr_scope(dirty_files, pr_scope)
|
||||
scratch_used = bool(proof.get("scratch_used"))
|
||||
scratch_path = (proof.get("scratch_path") or "").strip()
|
||||
|
||||
is_dirty = bool(dirty_files)
|
||||
unrelated_dirty = bool(unrelated)
|
||||
|
||||
if unrelated_dirty and not scratch_used:
|
||||
reasons.append(
|
||||
"worktree has dirty tracked files outside PR scope "
|
||||
f"({', '.join(unrelated)}); stop or use a scratch worktree"
|
||||
)
|
||||
if scratch_used and not scratch_path:
|
||||
reasons.append(
|
||||
"scratch worktree was used but scratch_path was not reported"
|
||||
)
|
||||
if proof.get("unrelated_mutations_claimed"):
|
||||
reasons.append(
|
||||
"reviewer reported stash/reset/checkout cleanup of unrelated "
|
||||
"local changes; this is forbidden"
|
||||
)
|
||||
|
||||
command_assessment = assess_reviewer_git_command_log(
|
||||
list(proof.get("git_commands") or [])
|
||||
)
|
||||
if command_assessment["block"]:
|
||||
reasons.extend(command_assessment["reasons"])
|
||||
|
||||
proven = not reasons
|
||||
return {
|
||||
"proven": proven,
|
||||
"block": not proven,
|
||||
"reasons": reasons,
|
||||
"worktree_path": worktree_path or None,
|
||||
"is_dirty": is_dirty,
|
||||
"dirty_files": dirty_files,
|
||||
"unrelated_dirty_files": unrelated,
|
||||
"scratch_used": scratch_used,
|
||||
"scratch_path": scratch_path or None,
|
||||
"unrelated_mutations_avoided": not bool(
|
||||
proof.get("unrelated_mutations_claimed")
|
||||
or command_assessment.get("forbidden_commands")
|
||||
),
|
||||
"safe_next_action": (
|
||||
"proceed"
|
||||
if proven
|
||||
else command_assessment.get("safe_next_action")
|
||||
or "stop; use scripts/worktree-review or report dirty worktree"
|
||||
),
|
||||
"forbidden_commands": command_assessment.get("forbidden_commands", []),
|
||||
}
|
||||
|
||||
|
||||
def assess_author_worktree_continuity(proof: dict | None) -> dict:
|
||||
"""Authors may keep dirty feature worktrees; reviewers may not manipulate them.
|
||||
|
||||
This helper only proves the task role is author when dirty unrelated files
|
||||
exist — it does not grant reviewers an exception.
|
||||
"""
|
||||
proof = dict(proof or {})
|
||||
role = (proof.get("task_role") or "").strip().lower()
|
||||
dirty_files = list(proof.get("dirty_files") or [])
|
||||
if role == "author" and dirty_files:
|
||||
return {
|
||||
"allowed": True,
|
||||
"reasons": [
|
||||
"author task may continue with dirty tracked files in its own "
|
||||
"worktree; reviewer interference rules do not apply"
|
||||
],
|
||||
}
|
||||
if role == "reviewer" and dirty_files:
|
||||
return assess_reviewer_worktree_proof(proof)
|
||||
return {"allowed": True, "reasons": []}
|
||||
@@ -20,6 +20,15 @@ Repo name disambiguation (Gitea-Tools blind review hardening):
|
||||
|
||||
Rules (llm-project-workflow):
|
||||
- Review in a SEPARATE detached review worktree, never the author's folder.
|
||||
- Worktree safety (#233): before checkout, diff, validation, review, or merge,
|
||||
report the starting worktree path and whether it was dirty. If unrelated
|
||||
tracked files exist outside the PR scope, STOP or run
|
||||
`scripts/worktree-review <pr-head-branch>` and validate in the scratch path.
|
||||
NEVER run `git stash`, `git stash pop/drop`, `git checkout --`, `git reset`,
|
||||
or `git clean` to manage another session's dirty files.
|
||||
- Final report must state: Worktree path, Worktree dirty (yes/no),
|
||||
Scratch worktree used (yes/no + path if yes), and confirm no unrelated local
|
||||
files were modified, stashed, reset, or dropped.
|
||||
- You must NOT be the PR author. If the authenticated user == PR author, stop.
|
||||
A different LLM-Agent-SHA does NOT make you a different actor — only a
|
||||
different authenticated Gitea user does (docs/llm-agent-sha.md).
|
||||
|
||||
+49
-13
@@ -65,6 +65,20 @@ CREATE_PR_ENV = {
|
||||
),
|
||||
}
|
||||
|
||||
ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
|
||||
|
||||
|
||||
def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
|
||||
record = {
|
||||
"issue_number": issue_number,
|
||||
"branch_name": branch_name,
|
||||
"remote": "dadeschools",
|
||||
"org": "Scaled-Tech-Consulting",
|
||||
"repo": "Gitea-Tools",
|
||||
}
|
||||
record.update(overrides)
|
||||
return record
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Create Issue
|
||||
@@ -127,15 +141,19 @@ class TestCreatePR(unittest.TestCase):
|
||||
@patch("os.path.exists", return_value=True)
|
||||
@patch("builtins.open")
|
||||
def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role):
|
||||
mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
|
||||
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
|
||||
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
|
||||
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
|
||||
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
||||
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
|
||||
self.assertEqual(result["number"], 3)
|
||||
self.assertNotIn("url", result)
|
||||
mock_exists.assert_called_with(ISSUE_LOCK_FILE)
|
||||
mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
|
||||
payload = mock_api.call_args[0][3]
|
||||
self.assertEqual(payload["head"], "feat/x")
|
||||
self.assertEqual(payload["base"], "main")
|
||||
self.assertIn("Closes #123", payload["title"])
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@@ -144,13 +162,28 @@ class TestCreatePR(unittest.TestCase):
|
||||
@patch("os.path.exists", return_value=True)
|
||||
@patch("builtins.open")
|
||||
def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role):
|
||||
mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
|
||||
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
|
||||
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
|
||||
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
|
||||
env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
|
||||
self.assertIn("pulls/3", result["url"])
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
@patch("os.path.exists", return_value=True)
|
||||
@patch("builtins.open")
|
||||
def test_create_pr_locked_issue_mismatch_fails(self, mock_open, mock_exists, _auth, _role):
|
||||
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
|
||||
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
|
||||
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
gitea_create_pr(title="feat: X Closes #999", head="feat/x", base="main")
|
||||
self.assertIn("Closes #123", str(ctx.exception))
|
||||
mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Close Issue
|
||||
@@ -2770,8 +2803,8 @@ 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")
|
||||
if os.path.exists(ISSUE_LOCK_FILE):
|
||||
os.remove(ISSUE_LOCK_FILE)
|
||||
|
||||
@patch("mcp_server.api_get_all")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
@@ -2779,7 +2812,7 @@ class TestIssueLocking(unittest.TestCase):
|
||||
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"))
|
||||
self.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
|
||||
|
||||
def test_lock_issue_mismatch_branch_fails(self):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
@@ -2816,8 +2849,8 @@ class TestIssueLocking(unittest.TestCase):
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_create_pr_missing_lock_fails(self, _auth, _role):
|
||||
if os.path.exists("/tmp/gitea_issue_lock.json"):
|
||||
os.remove("/tmp/gitea_issue_lock.json")
|
||||
if os.path.exists(ISSUE_LOCK_FILE):
|
||||
os.remove(ISSUE_LOCK_FILE)
|
||||
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
|
||||
@@ -2827,8 +2860,9 @@ class TestIssueLocking(unittest.TestCase):
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_create_pr_branch_mismatch_fails(self, _auth, _role):
|
||||
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 open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
|
||||
json.dump(_sample_issue_lock(
|
||||
issue_number=196, branch_name="feat/issue-196-mutations"), f)
|
||||
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
|
||||
@@ -2838,8 +2872,9 @@ class TestIssueLocking(unittest.TestCase):
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_create_pr_forbidden_terms_fails(self, _auth, _role):
|
||||
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 open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
|
||||
json.dump(_sample_issue_lock(
|
||||
issue_number=196, branch_name="feat/issue-196-mutations"), f)
|
||||
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
||||
for term in ("equivalent to #196", "related to #196", "same as #196"):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
@@ -2850,8 +2885,9 @@ class TestIssueLocking(unittest.TestCase):
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_create_pr_missing_closes_ref_fails(self, _auth, _role):
|
||||
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 open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
|
||||
json.dump(_sample_issue_lock(
|
||||
issue_number=196, branch_name="feat/issue-196-mutations"), f)
|
||||
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
|
||||
|
||||
+72
-17
@@ -183,6 +183,24 @@ def _good_review_mutation():
|
||||
return assess_review_mutation_final_report(report, lock)
|
||||
|
||||
|
||||
def _good_worktree(**overrides):
|
||||
proof = {
|
||||
"worktree_path": "/repo/branches/review-feat-issue-224",
|
||||
"porcelain_status": "",
|
||||
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
||||
"scratch_used": True,
|
||||
"scratch_path": "/repo/branches/review-feat-issue-224",
|
||||
"git_commands": [
|
||||
"git fetch prgs master feat/issue-224-wiki-proof-refresh",
|
||||
"git diff prgs/master...prgs/feat/issue-224-wiki-proof-refresh",
|
||||
],
|
||||
}
|
||||
proof.update(overrides)
|
||||
from reviewer_worktree import assess_reviewer_worktree_proof # noqa: E402
|
||||
|
||||
return assess_reviewer_worktree_proof(proof)
|
||||
|
||||
|
||||
def _good_role_boundary_179(**overrides):
|
||||
kwargs = {
|
||||
"task_role": "reviewer",
|
||||
@@ -595,6 +613,13 @@ class TestFinalReport(unittest.TestCase):
|
||||
"controller_handoff": _good_handoff(),
|
||||
"capability_proof": _good_capability_proof(),
|
||||
"sweep_proof": _good_secret_sweep(),
|
||||
"worktree_proof": {
|
||||
"worktree_path": "/repo/branches/review-feat-issue-224",
|
||||
"porcelain_status": "",
|
||||
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
||||
"scratch_used": True,
|
||||
"scratch_path": "/repo/branches/review-feat-issue-224",
|
||||
},
|
||||
}
|
||||
kwargs.update(overrides)
|
||||
return build_final_report(**kwargs)
|
||||
@@ -899,12 +924,17 @@ class TestControllerHandoff(unittest.TestCase):
|
||||
result = assess_controller_handoff(self.BASE_HANDOFF, role="review")
|
||||
self.assertEqual(result["verdict"], "incomplete")
|
||||
self.assertIn("Pinned reviewed head", result["missing_fields"])
|
||||
self.assertIn("Worktree path", result["missing_fields"])
|
||||
self.assertIn("Merge result", result["missing_fields"])
|
||||
|
||||
complete = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||
"- Selected PR: #999",
|
||||
"- Reviewer eligibility: passed",
|
||||
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
||||
"- Worktree path: /repo/branches/review-pr-999",
|
||||
"- Worktree dirty: no",
|
||||
"- Scratch worktree used: yes (/repo/branches/review-pr-999)",
|
||||
"- Unrelated local mutations: none",
|
||||
"- Review decision: approve",
|
||||
"- Merge result: merged",
|
||||
"- Linked issue status: closed",
|
||||
@@ -913,24 +943,46 @@ class TestControllerHandoff(unittest.TestCase):
|
||||
result = assess_controller_handoff(complete, role="review")
|
||||
self.assertEqual(result["verdict"], "complete")
|
||||
|
||||
def test_author_role_requires_author_fields(self):
|
||||
complete = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||
"- Selected issue: #182",
|
||||
def _author_role_fields(self, issue_number=182, pr_number=999):
|
||||
return [
|
||||
f"- Selected issue: #{issue_number}",
|
||||
"- Issue lock proof: lock before diff on feat/x @ master",
|
||||
"- Claim/comment status: comment-claimed",
|
||||
"- PR number opened: #999",
|
||||
f"- PR number opened: #{pr_number}",
|
||||
"- No review/merge: confirmed",
|
||||
])
|
||||
]
|
||||
|
||||
def test_handoff_role_fields_author_includes_issue_lock_proof(self):
|
||||
from review_proofs import HANDOFF_ROLE_FIELDS
|
||||
names = [name for name, _ in HANDOFF_ROLE_FIELDS["author"]]
|
||||
self.assertIn("Issue lock proof", names)
|
||||
|
||||
def test_author_role_requires_author_fields(self):
|
||||
complete = self.BASE_HANDOFF + "\n" + "\n".join(self._author_role_fields())
|
||||
result = assess_controller_handoff(complete, role="author")
|
||||
self.assertEqual(result["verdict"], "complete")
|
||||
|
||||
result = assess_controller_handoff(self.BASE_HANDOFF, role="author")
|
||||
self.assertEqual(result["verdict"], "incomplete")
|
||||
self.assertIn("Issue lock proof", result["missing_fields"])
|
||||
self.assertIn("No review/merge confirmation", result["missing_fields"])
|
||||
|
||||
def test_author_role_requires_issue_lock_proof(self):
|
||||
without_lock = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||
"- Selected issue: #182",
|
||||
"- Claim/comment status: comment-claimed",
|
||||
"- PR number opened: #999",
|
||||
"- No review/merge: confirmed",
|
||||
])
|
||||
result = assess_controller_handoff(without_lock, role="author")
|
||||
self.assertEqual(result["verdict"], "incomplete")
|
||||
self.assertIn("Issue lock proof", 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",
|
||||
"- Issue lock proof: lock before diff on feat/x @ master",
|
||||
"- Claim/comment status: comment-claimed",
|
||||
"- PR number opened: #999",
|
||||
"- No review/merge: confirmed",
|
||||
@@ -942,6 +994,7 @@ class TestControllerHandoff(unittest.TestCase):
|
||||
# 2. multiple issues blocked
|
||||
incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||
"- Selected issue: #194, #196",
|
||||
"- Issue lock proof: lock before diff on feat/x @ master",
|
||||
"- Claim/comment status: comment-claimed",
|
||||
"- PR number opened: #999",
|
||||
"- No review/merge: confirmed",
|
||||
@@ -953,6 +1006,7 @@ class TestControllerHandoff(unittest.TestCase):
|
||||
def test_author_role_rejects_fuzzy_pr_number(self):
|
||||
incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||
"- Selected issue: #196",
|
||||
"- Issue lock proof: lock before diff on feat/issue-196 @ master",
|
||||
"- Claim/comment status: comment-claimed",
|
||||
"- PR number opened: PR #203 / #204 equivalent",
|
||||
"- No review/merge: confirmed",
|
||||
@@ -984,23 +1038,17 @@ class TestControllerHandoff(unittest.TestCase):
|
||||
|
||||
def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
|
||||
# 1. Workspace mutations: none is rejected when local_edits is True
|
||||
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||
"- Selected issue: #196",
|
||||
"- Claim/comment status: comment-claimed",
|
||||
"- PR number opened: #203",
|
||||
"- No review/merge: confirmed",
|
||||
])
|
||||
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join(
|
||||
self._author_role_fields(issue_number=196, pr_number=203))
|
||||
res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
|
||||
self.assertEqual(res["verdict"], "incomplete")
|
||||
self.assertIn("Workspace mutations", res["missing_fields"])
|
||||
|
||||
# 2. Workspace mutations: edited files is allowed when local_edits is True
|
||||
complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([
|
||||
"- Selected issue: #196",
|
||||
"- Claim/comment status: comment-claimed",
|
||||
"- PR number opened: #203",
|
||||
"- No review/merge: confirmed",
|
||||
])
|
||||
complete_eq = self.BASE_HANDOFF.replace(
|
||||
"- Workspace mutations: none",
|
||||
"- Workspace mutations: edited review_proofs.py",
|
||||
) + "\n" + "\n".join(self._author_role_fields(issue_number=196, pr_number=203))
|
||||
res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
|
||||
self.assertEqual(res2["verdict"], "complete")
|
||||
|
||||
@@ -1306,6 +1354,13 @@ class TestFinalReport179Bar(unittest.TestCase):
|
||||
"controller_handoff": _good_handoff(),
|
||||
"capability_proof": _good_capability_proof(),
|
||||
"sweep_proof": _good_secret_sweep(),
|
||||
"worktree_proof": {
|
||||
"worktree_path": "/repo/branches/review-feat-issue-224",
|
||||
"porcelain_status": "",
|
||||
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
||||
"scratch_used": True,
|
||||
"scratch_path": "/repo/branches/review-feat-issue-224",
|
||||
},
|
||||
}
|
||||
kwargs.update(overrides)
|
||||
return build_final_report(**kwargs)
|
||||
|
||||
@@ -0,0 +1,166 @@
|
||||
"""Tests for reviewer worktree safety proofs (Issue #233)."""
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
from reviewer_worktree import ( # noqa: E402
|
||||
assess_author_worktree_continuity,
|
||||
assess_reviewer_git_command_log,
|
||||
assess_reviewer_worktree_proof,
|
||||
files_outside_pr_scope,
|
||||
is_forbidden_reviewer_git_command,
|
||||
is_readonly_reviewer_git_command,
|
||||
parse_dirty_tracked_files,
|
||||
)
|
||||
|
||||
|
||||
class TestParseDirtyTrackedFiles(unittest.TestCase):
|
||||
def test_ignores_untracked_files(self):
|
||||
porcelain = "?? untracked.txt\n M tracked.py\n"
|
||||
self.assertEqual(parse_dirty_tracked_files(porcelain), ["tracked.py"])
|
||||
|
||||
def test_parses_renamed_paths(self):
|
||||
porcelain = "R old.py -> new.py\n"
|
||||
self.assertEqual(parse_dirty_tracked_files(porcelain), ["new.py"])
|
||||
|
||||
|
||||
class TestFilesOutsidePrScope(unittest.TestCase):
|
||||
def test_all_dirty_in_scope_is_clean(self):
|
||||
self.assertEqual(
|
||||
files_outside_pr_scope(
|
||||
["docs/wiki/Repositories.md"],
|
||||
["docs/wiki/Repositories.md"],
|
||||
),
|
||||
[],
|
||||
)
|
||||
|
||||
def test_unrelated_dirty_files_detected(self):
|
||||
self.assertEqual(
|
||||
files_outside_pr_scope(
|
||||
["review_proofs.py", "docs/wiki/Repositories.md"],
|
||||
["docs/wiki/Repositories.md"],
|
||||
),
|
||||
["review_proofs.py"],
|
||||
)
|
||||
|
||||
|
||||
class TestForbiddenGitCommands(unittest.TestCase):
|
||||
def test_blocks_stash_operations(self):
|
||||
self.assertTrue(is_forbidden_reviewer_git_command("git stash"))
|
||||
self.assertTrue(
|
||||
is_forbidden_reviewer_git_command(
|
||||
'git stash push -m "reviewer-temp-stash" -- tests/test_mcp_server.py'
|
||||
)
|
||||
)
|
||||
self.assertTrue(is_forbidden_reviewer_git_command("git stash pop"))
|
||||
self.assertTrue(is_forbidden_reviewer_git_command("git stash drop"))
|
||||
|
||||
def test_blocks_checkout_reset_and_clean(self):
|
||||
self.assertTrue(
|
||||
is_forbidden_reviewer_git_command(
|
||||
"git checkout -- review_proofs.py tests/test_mcp_server.py"
|
||||
)
|
||||
)
|
||||
self.assertTrue(is_forbidden_reviewer_git_command("git reset --hard"))
|
||||
self.assertTrue(is_forbidden_reviewer_git_command("git clean -fd"))
|
||||
|
||||
def test_allows_readonly_commands(self):
|
||||
for cmd in (
|
||||
"git fetch prgs master",
|
||||
"git status --porcelain",
|
||||
"git diff prgs/master...HEAD",
|
||||
"git rev-parse HEAD",
|
||||
"git -C /repo log -1",
|
||||
):
|
||||
with self.subTest(cmd=cmd):
|
||||
self.assertFalse(is_forbidden_reviewer_git_command(cmd))
|
||||
self.assertTrue(is_readonly_reviewer_git_command(cmd))
|
||||
|
||||
|
||||
class TestAssessReviewerWorktreeProof(unittest.TestCase):
|
||||
def test_clean_worktree_proceeds(self):
|
||||
result = assess_reviewer_worktree_proof({
|
||||
"worktree_path": "/repo/branches/review-pr-231",
|
||||
"porcelain_status": "",
|
||||
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
||||
"scratch_used": False,
|
||||
"git_commands": ["git fetch prgs master", "git diff prgs/master...HEAD"],
|
||||
})
|
||||
self.assertTrue(result["proven"])
|
||||
self.assertFalse(result["block"])
|
||||
|
||||
def test_dirty_unrelated_without_scratch_blocks(self):
|
||||
result = assess_reviewer_worktree_proof({
|
||||
"worktree_path": "/repo",
|
||||
"dirty_files": ["review_proofs.py", "tests/test_review_proofs.py"],
|
||||
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
||||
"scratch_used": False,
|
||||
})
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["block"])
|
||||
self.assertIn("review_proofs.py", result["unrelated_dirty_files"][0])
|
||||
|
||||
def test_scratch_worktree_allows_dirty_main_repo(self):
|
||||
result = assess_reviewer_worktree_proof({
|
||||
"worktree_path": "/repo",
|
||||
"dirty_files": ["review_proofs.py"],
|
||||
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
||||
"scratch_used": True,
|
||||
"scratch_path": "/repo/branches/review-feat-issue-224",
|
||||
})
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_forbidden_command_history_blocks(self):
|
||||
result = assess_reviewer_worktree_proof({
|
||||
"worktree_path": "/repo/branches/review-pr-231",
|
||||
"porcelain_status": "",
|
||||
"git_commands": ["git stash push -m temp -- review_proofs.py"],
|
||||
})
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["forbidden_commands"])
|
||||
|
||||
def test_unrelated_mutations_claimed_blocks(self):
|
||||
result = assess_reviewer_worktree_proof({
|
||||
"worktree_path": "/repo",
|
||||
"porcelain_status": "",
|
||||
"unrelated_mutations_claimed": True,
|
||||
})
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
|
||||
class TestAuthorContinuity(unittest.TestCase):
|
||||
def test_author_may_keep_dirty_worktree(self):
|
||||
result = assess_author_worktree_continuity({
|
||||
"task_role": "author",
|
||||
"dirty_files": ["feat.py"],
|
||||
})
|
||||
self.assertTrue(result["allowed"])
|
||||
|
||||
def test_reviewer_dirty_worktree_uses_reviewer_gate(self):
|
||||
result = assess_author_worktree_continuity({
|
||||
"task_role": "reviewer",
|
||||
"worktree_path": "/repo",
|
||||
"dirty_files": ["other.py"],
|
||||
"pr_scope_files": ["docs/a.md"],
|
||||
"scratch_used": False,
|
||||
})
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
|
||||
class TestAssessReviewerGitCommandLog(unittest.TestCase):
|
||||
def test_empty_log_is_clean(self):
|
||||
result = assess_reviewer_git_command_log([])
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_mixed_log_blocks_on_forbidden(self):
|
||||
result = assess_reviewer_git_command_log([
|
||||
"git fetch prgs",
|
||||
"git stash",
|
||||
])
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertEqual(len(result["forbidden_commands"]), 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user