feat: require pagination proof in reconciliation PR inventory (Closes #308) #376

Merged
sysadmin merged 9 commits from feat/issue-308-reconcile-inventory-pagination into master 2026-07-07 09:26:03 -05:00
6 changed files with 650 additions and 0 deletions
Showing only changes of commit d0f52cbf19 - Show all commits
+114
View File
@@ -3666,6 +3666,113 @@ def assess_reviewer_baseline_validation_proof(
}
# ---------------------------------------------------------------------------
# Full-suite failure approval gate (#323)
# ---------------------------------------------------------------------------
def assess_full_suite_failure_approval_gate(
*,
review_decision: str,
full_suite_passed: bool | None,
baseline_validation: dict | None = None,
baseline_proof: dict | None = None,
report_text: str | None = None,
project_root: str | None = None,
) -> dict:
"""#323: approving a PR whose full suite fails requires baseline proof.
Default action on a full-suite failure is REQUEST_CHANGES. Approval is
allowed only when the #325 baseline proof passes in full (branches/
worktree, pinned SHAs, matching failure signatures) and the extra #323
fields prove the failures are pre-existing: ``pr_suite_command``,
``baseline_suite_command``, ``new_tests_passed``, and a non-empty
``unrelated_explanation``.
*review_decision* is the decision the workflow intends to submit; only
``approve`` can be blocked. ``block`` is True when an approval was
requested that the proofs do not allow.
"""
decision = (review_decision or "").strip().lower().replace("-", "_")
wants_approval = decision == "approve"
reasons: list[str] = []
if full_suite_passed is True:
return {
"approve_allowed": True,
"block": False,
"default_action": "proceed",
"reasons": [],
}
default_action = "request_changes"
if full_suite_passed is None:
reasons.append(
"full-suite result not proven; approval requires a recorded "
"full-suite command and result (#323)"
)
else:
if baseline_validation is None:
baseline_validation = assess_reviewer_baseline_validation_proof(
baseline_proof=baseline_proof,
report_text=report_text,
project_root=project_root,
)
proof = baseline_proof or {}
if not proof:
reasons.append(
"full suite failed but no baseline proof was provided; "
"default action is REQUEST_CHANGES (#323)"
)
if not baseline_validation.get("proven"):
reasons.append(
"baseline validation proof missing or failed (#323/#325)"
)
reasons.extend(baseline_validation.get("reasons", []))
if proof:
worktree = (proof.get("worktree_path") or "").strip()
if not _path_under_branches(worktree, project_root):
reasons.append(
"baseline comparison must run in a clean baseline "
"worktree under branches/, not the main checkout (#323)"
)
if not (proof.get("pr_suite_command") or "").strip():
reasons.append(
"PR full-suite command/result missing from baseline "
"proof (#323)"
)
if not (proof.get("baseline_suite_command") or "").strip():
reasons.append(
"baseline full-suite command/result missing from "
"baseline proof (#323)"
)
if proof.get("new_tests_passed") is not True:
reasons.append(
"proof that new/changed tests passed is missing (#323)"
)
if not (proof.get("unrelated_explanation") or "").strip():
reasons.append(
"explanation why failures are unrelated to the PR is "
"missing (#323)"
)
if proof.get("failure_signatures_match") is not True:
reasons.append(
"PR and baseline failure signatures do not match; "
"request changes (#323)"
)
approve_allowed = not reasons
return {
"approve_allowed": approve_allowed,
"block": wants_approval and not approve_allowed,
"default_action": "proceed" if approve_allowed else default_action,
"reasons": reasons,
}
# ---------------------------------------------------------------------------
# Identity disclosure (#305)
# ---------------------------------------------------------------------------
@@ -4169,3 +4276,10 @@ def assess_reconcile_inventory_report(report_text, **kwargs):
from reviewer_reconcile_inventory import assess_reconcile_inventory_report as _assess
return _assess(report_text, **kwargs)
def assess_worktree_ownership_report(report_text, **kwargs):
"""#312: prove session-owned or safe-reuse worktree before reset/validation."""
from reviewer_worktree_ownership import assess_worktree_ownership_report as _assess
return _assess(report_text, **kwargs)
+207
View File
@@ -0,0 +1,207 @@
"""Reviewer worktree ownership and safe-reuse proof verifier (#312)."""
from __future__ import annotations
import re
from typing import Any
_SESSION_OWNED_RE = re.compile(
r"branches/(?:review-pr\d+[\w/-]*|review-[\w-]+)",
re.IGNORECASE,
)
_WORKTREE_PATH_RE = re.compile(
r"(?:review worktree path|worktree path|session-owned worktree)\s*:\s*(\S+)",
re.IGNORECASE,
)
_BRANCHES_PATH_RE = re.compile(r"/branches/|\bbranches/", re.IGNORECASE)
_MAIN_CHECKOUT_RE = re.compile(
r"main checkout|not (?:the )?main checkout|outside branches/",
re.IGNORECASE,
)
_SAFE_REUSE_RE = re.compile(r"safe[- ]reuse proof", re.IGNORECASE)
_REUSE_POLICY_RE = re.compile(
r"(?:project policy|policy) (?:allows|allowing) (?:reuse|reset)",
re.IGNORECASE,
)
_CLEAN_TRACKED_RE = re.compile(
r"(?:clean tracked state|tracked state\s*:\s*clean|no uncommitted tracked)",
re.IGNORECASE,
)
_CLEAN_UNTRACKED_RE = re.compile(
r"(?:clean untracked state|untracked state\s*:\s*clean|no untracked files)",
re.IGNORECASE,
)
_NOT_OTHER_SESSION_RE = re.compile(
r"not owned by (?:another|other) (?:active )?(?:task|session)",
re.IGNORECASE,
)
_BRANCH_BEFORE_RESET_RE = re.compile(
r"(?:branch/head before reset|head before reset|branch before reset)\s*:\s*(\S+)",
re.IGNORECASE,
)
_RESET_TARGET_RE = re.compile(
r"(?:reset target sha|reset target)\s*:\s*([0-9a-f]{7,40})",
re.IGNORECASE,
)
_DESTRUCTIVE_CMD_RE = re.compile(
r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?"
r"(?:reset\s+--hard|clean(?:\s+-[A-Za-z]+)*|checkout\s+(?!HEAD\s+--)|switch\s+)",
re.IGNORECASE,
)
_WORKSPACE_NONE_RE = re.compile(r"workspace mutations\s*:\s*none", re.IGNORECASE)
_WORKTREE_MUTATION_RE = re.compile(
r"(?:worktree(?:/index)? mutations|destructive reset)\s*:\s*(?!none\b).+",
re.IGNORECASE,
)
_RESET_IN_REPORT_RE = re.compile(r"reset\s+--hard", re.IGNORECASE)
def _command_text(entry) -> str:
if isinstance(entry, dict):
return str(entry.get("command") or "").strip()
return str(entry or "").strip()
def _destructive_commands(command_log: list | None) -> list[str]:
return [
cmd
for cmd in (_command_text(entry) for entry in (command_log or []))
if cmd and _DESTRUCTIVE_CMD_RE.search(cmd)
]
def _extract_worktree_path(text: str, session: dict) -> str:
path = (session.get("worktree_path") or "").strip()
if path:
return path
match = _WORKTREE_PATH_RE.search(text or "")
return match.group(1).strip() if match else ""
def _is_session_owned_path(path: str) -> bool:
normalized = (path or "").replace("\\", "/")
return bool(_SESSION_OWNED_RE.search(normalized))
def _safe_reuse_fields_present(text: str) -> list[str]:
missing: list[str] = []
if not _WORKTREE_PATH_RE.search(text):
missing.append("exact worktree path")
if not _BRANCHES_PATH_RE.search(text):
missing.append("worktree inside branches/")
if not _MAIN_CHECKOUT_RE.search(text):
missing.append("worktree is not the main checkout")
if not _NOT_OTHER_SESSION_RE.search(text):
missing.append("worktree not owned by another active session")
if not _CLEAN_TRACKED_RE.search(text):
missing.append("clean tracked state")
if not _CLEAN_UNTRACKED_RE.search(text):
missing.append("clean untracked state")
if not _BRANCH_BEFORE_RESET_RE.search(text):
missing.append("branch/head before reset")
if not _RESET_TARGET_RE.search(text):
missing.append("reset target SHA")
if not _REUSE_POLICY_RE.search(text):
missing.append("explicit project policy allowing reuse/reset")
return missing
def assess_worktree_ownership_report(
report_text: str,
*,
ownership_session: dict | None = None,
command_log: list | None = None,
) -> dict[str, Any]:
"""Prove reviewer worktree ownership before reset or validation (#312)."""
text = report_text or ""
session = dict(ownership_session or {})
reasons: list[str] = []
worktree_path = _extract_worktree_path(text, session)
destructive = _destructive_commands(command_log or session.get("command_log"))
if not destructive and session.get("destructive_reset"):
destructive = ["git reset --hard (session)"]
session_owned = bool(
session.get("session_owned")
or (worktree_path and _is_session_owned_path(worktree_path))
)
safe_reuse = bool(session.get("safe_reuse") or _SAFE_REUSE_RE.search(text))
main_checkout = bool(session.get("main_checkout"))
dirty_tracked = session.get("dirty_tracked")
dirty_untracked = session.get("dirty_untracked")
other_session_owned = bool(session.get("other_session_owned"))
if worktree_path or destructive or session.get("validation_ran"):
if not worktree_path:
reasons.append("report missing exact review worktree path")
elif main_checkout or "branches/" not in worktree_path.replace("\\", "/"):
reasons.append("review worktree must be inside branches/, not the main checkout")
elif not _BRANCHES_PATH_RE.search(worktree_path.replace("\\", "/")):
reasons.append("review worktree path must be under branches/")
if other_session_owned and not safe_reuse:
reasons.append(
"reused worktree appears owned by another active task/session; "
"safe-reuse proof required"
)
if dirty_tracked:
reasons.append(
"worktree has uncommitted tracked changes before reset or validation"
)
if dirty_untracked and safe_reuse:
reasons.append("safe-reuse proof requires clean untracked state")
if safe_reuse and not session_owned:
reasons.extend(
f"safe-reuse proof missing {field}"
for field in _safe_reuse_fields_present(text)
)
if destructive:
if not session_owned and not safe_reuse:
reasons.append(
"destructive worktree commands forbidden without session-owned "
"worktree or safe-reuse proof"
)
if _WORKSPACE_NONE_RE.search(text) and (
_RESET_IN_REPORT_RE.search(text) or any("reset" in c.lower() for c in destructive)
):
reasons.append(
"final report must not claim 'Workspace mutations: none' when "
"git reset --hard occurred"
)
if not _WORKTREE_MUTATION_RE.search(text) and not _RESET_IN_REPORT_RE.search(text):
reasons.append(
"destructive reset must be reported under Worktree/index mutations "
"or destructive reset operations"
)
if (
worktree_path
and not session_owned
and not safe_reuse
and (destructive or session.get("validation_ran"))
and not _is_session_owned_path(worktree_path)
):
reasons.append(
"reused branch-named worktree requires safe-reuse proof before reset or validation"
)
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"worktree_path": worktree_path or None,
"session_owned": session_owned,
"safe_reuse": safe_reuse,
"destructive_commands": destructive,
"safe_next_action": (
"use a fresh session-owned review worktree under branches/review-pr<N>-* "
"or document full safe-reuse proof before destructive reset"
if reasons
else "proceed"
),
}
+20
View File
@@ -62,7 +62,24 @@ class TestPreflightWarnings(unittest.TestCase):
self.assertIn("_encode_commit_payload.py", status["preflight_warnings"][0])
# Issue-write tools are profile-gated (#69); gitea_lock_issue requires
# gitea.issue.comment (see task_capability_map), so the gate must be
# seeded exactly like tests/test_mcp_server.py::TestIssueLocking (#359).
ISSUE_WRITE_ENV = {
"GITEA_ALLOWED_OPERATIONS": (
"gitea.issue.create,gitea.issue.close,gitea.issue.comment"
),
}
class TestIssueLockArtifactWarning(unittest.TestCase):
def setUp(self):
self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
self._env_patcher.start()
def tearDown(self):
self._env_patcher.stop()
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server._auth", return_value="token x")
@patch("mcp_server._resolve", return_value=("h", "o", "r"))
@@ -72,6 +89,9 @@ class TestIssueLockArtifactWarning(unittest.TestCase):
mock_state.return_value = {
"current_branch": "master",
"porcelain_status": "?? _emit_payload.py\n",
"base_equivalent": True,
"inspected_git_root": "/scratch/wt",
"base_branch": "origin/master",
}
result = mcp_server.gitea_lock_issue(
issue_number=261,
+6
View File
@@ -152,3 +152,9 @@ def test_reconcile_inventory_verifier_exported():
from review_proofs import assess_reconcile_inventory_report
assert callable(assess_reconcile_inventory_report)
def test_worktree_ownership_verifier_exported():
from review_proofs import assess_worktree_ownership_report
assert callable(assess_worktree_ownership_report)
+166
View File
@@ -24,6 +24,7 @@ from review_proofs import ( # noqa: E402
ISSUE_SELECTION_CONTINUATION_EXPLICIT,
ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
assess_author_pr_report,
assess_full_suite_failure_approval_gate,
assess_queue_ordering_proof,
assess_reviewer_baseline_validation_proof,
assess_capability_evidence,
@@ -2539,5 +2540,170 @@ class TestReviewerBaselineValidationProof(unittest.TestCase):
self.assertFalse(report["baseline_validation_proven"])
class TestFullSuiteFailureApprovalGate(unittest.TestCase):
"""Issue #323: full-suite failure blocks approval without baseline proof."""
ROOT = "/repo/Gitea-Tools"
def _good_proof(self, **overrides):
proof = {
"worktree_path": f"{self.ROOT}/branches/baseline-master-pr280",
"baseline_target_sha": PINNED,
"pr_head_sha": OTHER,
"baseline_failures": ["tests/test_foo.py::test_bar"],
"pr_failures": ["tests/test_foo.py::test_bar"],
"failure_signatures_match": True,
"clean_before": True,
"clean_after": True,
"pr_suite_command": "venv/bin/pytest tests/",
"baseline_suite_command": "venv/bin/pytest tests/",
"new_tests_passed": True,
"unrelated_explanation": (
"failures touch agent_temp_artifacts only; PR changes "
"review_proofs handoff fields"
),
}
proof.update(overrides)
return proof
def _good_report(self):
return (
"Full suite failed on PR worktree and on baseline.\n"
"- Validation command: venv/bin/pytest tests/\n"
f"- Working directory: {self.ROOT}/branches/review-pr-280\n"
"Failures are pre-existing on master.\n"
)
def test_full_suite_pass_allows_approval_without_baseline(self):
result = assess_full_suite_failure_approval_gate(
review_decision="approve",
full_suite_passed=True,
project_root=self.ROOT,
)
self.assertTrue(result["approve_allowed"])
self.assertFalse(result["block"])
self.assertEqual(result["default_action"], "proceed")
def test_full_suite_failure_defaults_to_request_changes(self):
result = assess_full_suite_failure_approval_gate(
review_decision="request_changes",
full_suite_passed=False,
project_root=self.ROOT,
)
self.assertFalse(result["block"])
self.assertEqual(result["default_action"], "request_changes")
def test_approval_with_failure_and_no_proof_blocks(self):
result = assess_full_suite_failure_approval_gate(
review_decision="approve",
full_suite_passed=False,
project_root=self.ROOT,
)
self.assertFalse(result["approve_allowed"])
self.assertTrue(result["block"])
self.assertEqual(result["default_action"], "request_changes")
def test_approval_with_matching_baseline_proof_allowed(self):
result = assess_full_suite_failure_approval_gate(
review_decision="approve",
full_suite_passed=False,
baseline_proof=self._good_proof(),
report_text=self._good_report(),
project_root=self.ROOT,
)
self.assertTrue(result["approve_allowed"])
self.assertFalse(result["block"])
def test_mismatched_failure_signatures_block_approval(self):
result = assess_full_suite_failure_approval_gate(
review_decision="approve",
full_suite_passed=False,
baseline_proof=self._good_proof(
failure_signatures_match=False,
pr_failures=["tests/test_other.py::test_other"],
),
report_text=self._good_report(),
project_root=self.ROOT,
)
self.assertFalse(result["approve_allowed"])
self.assertTrue(result["block"])
def test_main_checkout_baseline_blocks_approval(self):
result = assess_full_suite_failure_approval_gate(
review_decision="approve",
full_suite_passed=False,
baseline_proof=self._good_proof(worktree_path=self.ROOT),
report_text=self._good_report(),
project_root=self.ROOT,
)
self.assertFalse(result["approve_allowed"])
self.assertTrue(result["block"])
def test_missing_new_tests_proof_blocks_approval(self):
result = assess_full_suite_failure_approval_gate(
review_decision="approve",
full_suite_passed=False,
baseline_proof=self._good_proof(new_tests_passed=None),
report_text=self._good_report(),
project_root=self.ROOT,
)
self.assertFalse(result["approve_allowed"])
self.assertTrue(any(
"new" in reason.lower() and "test" in reason.lower()
for reason in result["reasons"]
))
def test_missing_unrelated_explanation_blocks_approval(self):
result = assess_full_suite_failure_approval_gate(
review_decision="approve",
full_suite_passed=False,
baseline_proof=self._good_proof(unrelated_explanation=""),
report_text=self._good_report(),
project_root=self.ROOT,
)
self.assertFalse(result["approve_allowed"])
def test_missing_suite_commands_block_approval(self):
result = assess_full_suite_failure_approval_gate(
review_decision="approve",
full_suite_passed=False,
baseline_proof=self._good_proof(
pr_suite_command="", baseline_suite_command=""),
report_text=self._good_report(),
project_root=self.ROOT,
)
self.assertFalse(result["approve_allowed"])
self.assertTrue(any(
"command" in reason.lower() for reason in result["reasons"]
))
def test_unknown_suite_result_blocks_approval(self):
result = assess_full_suite_failure_approval_gate(
review_decision="approve",
full_suite_passed=None,
project_root=self.ROOT,
)
self.assertFalse(result["approve_allowed"])
self.assertTrue(result["block"])
def test_vague_same_as_master_claim_without_proof_blocks(self):
result = assess_full_suite_failure_approval_gate(
review_decision="approve",
full_suite_passed=False,
report_text="Validation: failures same as master; approving.",
project_root=self.ROOT,
)
self.assertFalse(result["approve_allowed"])
self.assertTrue(result["block"])
def test_non_approve_decision_never_blocks(self):
result = assess_full_suite_failure_approval_gate(
review_decision="comment",
full_suite_passed=None,
project_root=self.ROOT,
)
self.assertFalse(result["block"])
if __name__ == "__main__":
unittest.main()
+137
View File
@@ -0,0 +1,137 @@
"""Tests for reviewer worktree ownership verifier (#312)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from reviewer_worktree_ownership import assess_worktree_ownership_report # noqa: E402
def _fresh_review_report() -> str:
return "\n".join([
"Review worktree path: branches/review-pr280-feat-issue-275-claim",
"Worktree is inside branches/ and not the main checkout.",
"Tracked state: clean; untracked state: clean.",
"Official PR-head validation on session-owned worktree.",
"Worktree mutations: none",
])
def _safe_reuse_report() -> str:
return "\n".join([
"Safe-reuse proof for existing review worktree.",
"Review worktree path: branches/docs-issue-260-forbid-commit-fallbacks",
"Worktree inside branches/; not the main checkout.",
"Not owned by another active task/session.",
"Clean tracked state; clean untracked state.",
"Branch/head before reset: feat/issue-260-example",
"Reset target SHA: abc123def4567890abcdef1234567890abcdef12",
"Project policy allows reuse/reset for this clean review worktree.",
"Worktree mutations: git reset --hard FETCH_HEAD",
"Destructive reset operations: git reset --hard FETCH_HEAD",
])
class TestWorktreeOwnership(unittest.TestCase):
def test_fresh_session_owned_passes(self):
result = assess_worktree_ownership_report(
_fresh_review_report(),
ownership_session={
"validation_ran": True,
"worktree_path": "branches/review-pr280-feat-issue-275-claim",
},
)
self.assertTrue(result["proven"], result["reasons"])
self.assertTrue(result["session_owned"])
def test_safe_reuse_with_reset_passes(self):
result = assess_worktree_ownership_report(
_safe_reuse_report(),
ownership_session={
"safe_reuse": True,
"worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks",
},
command_log=["git reset --hard FETCH_HEAD"],
)
self.assertTrue(result["proven"], result["reasons"])
def test_destructive_reset_without_proof_blocks(self):
report = "\n".join([
"Review worktree path: branches/docs-issue-260-forbid-commit-fallbacks",
"Worktree mutations: git reset --hard FETCH_HEAD",
])
result = assess_worktree_ownership_report(
report,
ownership_session={
"validation_ran": True,
"worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks",
},
command_log=["git reset --hard FETCH_HEAD"],
)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
def test_reused_dirty_tracked_blocks(self):
result = assess_worktree_ownership_report(
_fresh_review_report(),
ownership_session={
"validation_ran": True,
"worktree_path": "branches/review-pr280-feat-issue-275-claim",
"dirty_tracked": True,
},
)
self.assertFalse(result["proven"])
self.assertTrue(any("tracked" in r.lower() for r in result["reasons"]))
def test_safe_reuse_dirty_untracked_blocks(self):
result = assess_worktree_ownership_report(
_safe_reuse_report(),
ownership_session={
"safe_reuse": True,
"dirty_untracked": True,
"worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks",
},
command_log=["git reset --hard FETCH_HEAD"],
)
self.assertFalse(result["proven"])
self.assertTrue(any("untracked" in r.lower() for r in result["reasons"]))
def test_workspace_none_with_reset_blocks(self):
report = "\n".join([
"Review worktree path: branches/review-pr280-feat-issue-275-claim",
"Workspace mutations: none",
"Worktree mutations: git reset --hard FETCH_HEAD",
])
result = assess_worktree_ownership_report(
report,
ownership_session={
"session_owned": True,
"worktree_path": "branches/review-pr280-feat-issue-275-claim",
},
command_log=["git reset --hard FETCH_HEAD"],
)
self.assertFalse(result["proven"])
self.assertTrue(any("workspace mutations" in r.lower() for r in result["reasons"]))
def test_main_checkout_blocks(self):
result = assess_worktree_ownership_report(
"Review worktree path: /Users/dev/Gitea-Tools",
ownership_session={
"validation_ran": True,
"main_checkout": True,
"worktree_path": "/Users/dev/Gitea-Tools",
},
)
self.assertFalse(result["proven"])
class TestExport(unittest.TestCase):
def test_review_proofs_reexport(self):
from review_proofs import assess_worktree_ownership_report as exported
self.assertTrue(callable(exported))
if __name__ == "__main__":
unittest.main()