feat(reviewer): wire PR inventory trust gate into live queue path (closes #196) #236

Merged
sysadmin merged 4 commits from feat/issue-196-pr-inventory-trust-gate-live-path into master 2026-07-06 10:14:23 -05:00
8 changed files with 816 additions and 33 deletions
+42 -1
View File
@@ -259,6 +259,7 @@ import issue_duplicate_gate # noqa: E402
import role_session_router # noqa: E402
import role_namespace_gate # noqa: E402
import task_capability_map # noqa: E402
import review_proofs # noqa: E402
# Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
@@ -2274,6 +2275,23 @@ def gitea_merge_pr(
return result
def _local_git_remote_url(remote_name: str) -> str | None:
"""Best-effort local ``git remote get-url`` for trust-gate corroboration."""
try:
proc = subprocess.run(
["git", "remote", "get-url", remote_name],
capture_output=True,
text=True,
cwd=PROJECT_ROOT,
)
if proc.returncode != 0:
return None
url = (proc.stdout or "").strip()
return url or None
except Exception:
return None
@mcp.tool()
def gitea_review_pr(
pr_number: int,
@@ -2341,6 +2359,8 @@ def gitea_review_pr(
prs_found_count = 0
pr_details_list = []
inventory_msg = ""
inventory_trust_gate = None
prs: list = []
h, o, r = _resolve(remote, host, org, repo)
auth = None
@@ -2425,7 +2445,28 @@ def gitea_review_pr(
if inventory_msg:
report_lines.append(inventory_msg)
else:
report_lines.append("Open PRs found: 0")
inventory_trust_gate = review_proofs.pr_inventory_trust_gate(
prs if inventory_attempted else None,
remote=remote,
org=o,
repo=r,
state="open",
authenticated_profile=profile,
local_remote_url=_local_git_remote_url(remote),
has_finality_metadata=True,
)
report_lines.extend(
review_proofs.format_pr_inventory_trust_gate_report(
inventory_trust_gate
)
)
if inventory_trust_gate.get("status") == "trusted_empty":
report_lines.append("Open PRs found: 0 (trusted_empty)")
else:
report_lines.append(
"Empty-queue claim blocked: "
"pr_inventory_trust_gate did not return trusted_empty"
)
else:
report_lines.append(inventory_msg)
+116 -1
View File
@@ -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")),
@@ -1284,6 +1312,93 @@ def pr_inventory_trust_gate(
}
def _split_repo_slug(full_repo: str) -> tuple[str | None, str | None]:
parts = (full_repo or "").split("/", 1)
if len(parts) == 2:
return parts[0].strip() or None, parts[1].strip() or None
return None, None
def assess_reviewer_queue_inventory(
repo_reports: list[dict] | None,
required_repos: list[str] | None = None,
*,
user_context: str | None = None,
) -> dict:
"""Canonical reviewer queue path: completeness plus per-repo trust gates (#196).
Any repository reporting ``open_pr_count == 0`` must pass
``pr_inventory_trust_gate`` with ``trusted_empty`` before an empty-queue
claim is allowed. A bare ``[]`` from ``gitea_list_prs`` is never sufficient.
"""
required = list(required_repos or [
"Scaled-Tech-Consulting/Gitea-Tools",
"Scaled-Tech-Consulting/mcp-control-plane",
])
completeness = assess_inventory_completeness(repo_reports, required)
trust_gates: dict[str, dict] = {}
blockers: list[str] = []
can_claim_empty = bool(completeness.get("complete"))
for report in repo_reports or []:
repo = (report.get("repo") or "").strip()
count = report.get("open_pr_count")
if not isinstance(count, int) or count != 0:
continue
org, repo_name = _split_repo_slug(repo)
list_response = report.get("list_prs_response")
if list_response is None:
list_response = []
gate = pr_inventory_trust_gate(
list_response,
remote=report.get("remote"),
org=org,
repo=repo_name,
state=report.get("state_filter"),
authenticated_profile=report.get("authenticated_profile"),
local_remote_url=report.get("local_remote_url"),
user_context=user_context or report.get("user_context"),
corroboration_open_pr_counter=report.get(
"corroboration_open_pr_counter"
),
has_finality_metadata=report.get("pagination_complete") is True,
)
trust_gates[repo] = gate
status = gate.get("status")
if status != "trusted_empty":
can_claim_empty = False
blockers.append(
f"repository '{repo}': empty PR list trust gate is "
f"'{status}'; cannot claim 'no open PRs'"
)
blockers.extend(gate.get("reasons") or [])
return {
"complete": bool(completeness.get("complete")),
"can_claim_empty_queue": can_claim_empty,
"can_claim_exhaustive": (
bool(completeness.get("can_claim_exhaustive")) and can_claim_empty
),
"inventory_reasons": list(completeness.get("reasons") or []),
"trust_gates": trust_gates,
"blockers": blockers,
"reasons": list(completeness.get("reasons") or []) + blockers,
}
def format_pr_inventory_trust_gate_report(gate: dict) -> list[str]:
"""Render trust-gate lines for MCP inventory output."""
lines = [f"pr_inventory_trust_gate.status: {gate.get('status', 'unknown')}"]
if gate.get("corroborated"):
lines.append("pr_inventory_trust_gate.corroborated: true")
for reason in gate.get("reasons") or []:
lines.append(f"pr_inventory_trust_gate.reason: {reason}")
return lines
def assess_duplicate_search_proof(report_text, matches):
"""#207: reject LLM duplicate summaries that omit known title matches."""
return issue_duplicate_gate.assess_duplicate_search_proof(
+219
View File
@@ -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": []}
@@ -17,9 +17,24 @@ Repo name disambiguation (Gitea-Tools blind review hardening):
configured repos were not checked. This is not a complete queue inventory."
- A single-repo "no open PRs" result MUST NOT be reported as global "no open PRs"
if the other configured repo was not inventoried.
- PR inventory trust gate (#196): before reporting "no open PRs" or "queue empty",
the workflow must run `pr_inventory_trust_gate` (via the live inventory path or
`review_proofs.assess_reviewer_queue_inventory`). Only `trusted_empty` allows a
clean empty-queue stop. Report `pr_inventory_trust_gate.status`, reasons, and
corroboration in the final report. A bare `[]` from `gitea_list_prs` is never
sufficient proof.
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
View File
@@ -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")
+68 -1
View File
@@ -261,12 +261,79 @@ class TestPRQueueInventory(unittest.TestCase):
for call in mock_api.call_args_list:
self.assertEqual(call.args[0], "GET")
@patch("mcp_server._local_git_remote_url")
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all):
def test_empty_inventory_runs_trust_gate_and_blocks_without_trusted_empty(
self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url
):
mock_get_profile.return_value = {
"profile_name": "gitea-author",
"allowed_operations": ["read"],
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = []
mock_api.return_value = {"login": "jcwalker3"}
mock_local_url.return_value = None
result = gitea_review_pr(
pr_number=1,
event="APPROVE",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
)
self.assertFalse(result["success"])
msg = result["message"]
self.assertIn("pr_inventory_trust_gate.status: untrusted_empty", msg)
self.assertIn("Empty-queue claim blocked", msg)
self.assertNotIn("Open PRs found: 0 (trusted_empty)", msg)
@patch("mcp_server._local_git_remote_url")
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_empty_inventory_trusted_empty_when_gate_passes(
self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url
):
mock_get_profile.return_value = {
"profile_name": "gitea-author",
"allowed_operations": ["read"],
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = []
mock_api.return_value = {"login": "jcwalker3"}
mock_local_url.return_value = (
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
)
result = gitea_review_pr(
pr_number=1,
event="APPROVE",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
)
self.assertFalse(result["success"])
msg = result["message"]
self.assertIn("pr_inventory_trust_gate.status: trusted_empty", msg)
self.assertIn("Open PRs found: 0 (trusted_empty)", msg)
@patch("mcp_server._local_git_remote_url")
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url):
"""Author profiles still cannot approve, request_changes, merge, or bypass gates even with inventory."""
mock_local_url.return_value = (
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
)
for event in ["APPROVE", "REQUEST_CHANGES"]:
mock_get_profile.return_value = {
"profile_name": "gitea-author",
+141 -17
View File
@@ -26,6 +26,7 @@ from review_proofs import ( # noqa: E402
assess_capability_proof,
assess_controller_handoff,
assess_inventory_completeness,
assess_reviewer_queue_inventory,
assess_live_state_recheck,
assess_review_mutation_final_report,
assess_role_boundary,
@@ -183,6 +184,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 +614,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 +925,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 +944,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 +995,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 +1007,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 +1039,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")
@@ -1163,6 +1212,74 @@ class TestPRInventoryTrustGate(unittest.TestCase):
self.assertTrue(res["corroborated"])
class TestAssessReviewerQueueInventory(unittest.TestCase):
"""Issue #196: trust gate wired into canonical queue inventory."""
def _repo_report(self, **overrides):
report = {
"repo": "Scaled-Tech-Consulting/Gitea-Tools",
"state_filter": "open",
"pagination_complete": True,
"open_pr_count": 0,
"list_prs_response": [],
"remote": "prgs",
"authenticated_profile": {
"profile_name": "prgs-reviewer",
"allowed_operations": ["read", "gitea.read"],
},
"local_remote_url": (
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
),
}
report.update(overrides)
return report
def test_empty_without_corroboration_blocks_empty_queue_claim(self):
result = assess_reviewer_queue_inventory([
self._repo_report(pagination_complete=False),
self._repo_report(
repo="Scaled-Tech-Consulting/mcp-control-plane",
open_pr_count=0,
pagination_complete=False,
),
])
self.assertFalse(result["can_claim_empty_queue"])
self.assertIn("untrusted_empty", str(result["trust_gates"]))
def test_trusted_empty_with_finality_allows_empty_queue_claim(self):
result = assess_reviewer_queue_inventory([
self._repo_report(),
self._repo_report(
repo="Scaled-Tech-Consulting/mcp-control-plane",
local_remote_url=(
"https://gitea.prgs.cc/Scaled-Tech-Consulting/"
"mcp-control-plane.git"
),
),
])
self.assertTrue(result["can_claim_empty_queue"])
self.assertTrue(result["can_claim_exhaustive"])
def test_user_context_indicating_open_prs_blocks_empty_claim(self):
result = assess_reviewer_queue_inventory(
[self._repo_report()],
user_context="please review open PR #195 in the queue",
)
self.assertFalse(result["can_claim_empty_queue"])
self.assertTrue(result["blockers"])
def test_nonempty_inventory_skips_empty_trust_gate_block(self):
result = assess_reviewer_queue_inventory([
self._repo_report(open_pr_count=2),
self._repo_report(
repo="Scaled-Tech-Consulting/mcp-control-plane",
open_pr_count=1,
),
])
self.assertTrue(result["complete"])
self.assertEqual(result["trust_gates"], {})
class TestCapabilityEvidence(unittest.TestCase):
"""#179 gap 1: capability claims need exact evidence."""
@@ -1306,6 +1423,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)
+166
View File
@@ -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()