Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2edb10045a | ||
|
|
3ea9167f3a | ||
|
|
f1757fe590 |
+1
-136
@@ -1052,125 +1052,6 @@ def _prs_from_list_response(list_prs_response: list | dict | None) -> list | Non
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
# ── Workspace-vs-worktree mutation consistency (Issue #313) ──────────────────
|
|
||||||
#
|
|
||||||
# A worktree reset/checkout/clean is a workspace mutation even when the
|
|
||||||
# reviewer edited no files by hand. Final reports must therefore never say
|
|
||||||
# "Workspace mutations: none" once a worktree mutation occurred, and every
|
|
||||||
# observed worktree / git ref mutation must appear under its precise
|
|
||||||
# category field.
|
|
||||||
|
|
||||||
WORKTREE_MUTATION_MARKERS = (
|
|
||||||
"reset",
|
|
||||||
"checkout",
|
|
||||||
"switch",
|
|
||||||
"restore",
|
|
||||||
"clean",
|
|
||||||
"stash",
|
|
||||||
"merge",
|
|
||||||
"rebase",
|
|
||||||
"cherry-pick",
|
|
||||||
"worktree add",
|
|
||||||
"worktree remove",
|
|
||||||
"worktree prune",
|
|
||||||
)
|
|
||||||
|
|
||||||
GIT_REF_MUTATION_MARKERS = (
|
|
||||||
"fetch",
|
|
||||||
"pull",
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def _report_labeled_fields(report_text):
|
|
||||||
"""Return {lowercase label: value} for every 'Label: value' line."""
|
|
||||||
fields = {}
|
|
||||||
for line in (report_text or "").splitlines():
|
|
||||||
stripped = line.strip().lstrip("-*").strip()
|
|
||||||
if ":" in stripped:
|
|
||||||
k, v = stripped.split(":", 1)
|
|
||||||
fields[k.strip().lower()] = v.strip()
|
|
||||||
return fields
|
|
||||||
|
|
||||||
|
|
||||||
def _field_claims_none(fields, label):
|
|
||||||
"""True when *label* is present and its value is empty or 'none...'."""
|
|
||||||
value = fields.get(label)
|
|
||||||
if value is None:
|
|
||||||
return False
|
|
||||||
value = value.strip().lower()
|
|
||||||
return not value or value.startswith("none")
|
|
||||||
|
|
||||||
|
|
||||||
def _classify_git_commands(observed_commands):
|
|
||||||
"""Split observed git commands into worktree and ref mutations."""
|
|
||||||
worktree_cmds = []
|
|
||||||
ref_cmds = []
|
|
||||||
for cmd in observed_commands or []:
|
|
||||||
lowered = " ".join((cmd or "").lower().split())
|
|
||||||
if "git" not in lowered:
|
|
||||||
continue
|
|
||||||
if any(marker in lowered for marker in WORKTREE_MUTATION_MARKERS):
|
|
||||||
worktree_cmds.append(cmd)
|
|
||||||
elif any(marker in lowered for marker in GIT_REF_MUTATION_MARKERS):
|
|
||||||
ref_cmds.append(cmd)
|
|
||||||
return worktree_cmds, ref_cmds
|
|
||||||
|
|
||||||
|
|
||||||
def assess_workspace_mutation_consistency(report_text, observed_commands=None):
|
|
||||||
"""#313: reject 'Workspace mutations: none' after worktree mutations.
|
|
||||||
|
|
||||||
*observed_commands* is the optional list of git commands the workflow
|
|
||||||
actually ran. Even without it, a report that itself lists a non-none
|
|
||||||
``Worktree mutations`` value while claiming ``Workspace mutations:
|
|
||||||
none`` is internally contradictory and fails.
|
|
||||||
|
|
||||||
Returns {'complete', 'downgraded', 'reasons'}; fails closed on any
|
|
||||||
contradiction or unreported observed mutation.
|
|
||||||
"""
|
|
||||||
fields = _report_labeled_fields(report_text)
|
|
||||||
worktree_cmds, ref_cmds = _classify_git_commands(observed_commands)
|
|
||||||
|
|
||||||
reported_worktree = fields.get("worktree mutations")
|
|
||||||
reported_worktree_nonnone = bool(
|
|
||||||
reported_worktree and not reported_worktree.strip().lower().startswith("none")
|
|
||||||
)
|
|
||||||
|
|
||||||
reasons = []
|
|
||||||
|
|
||||||
workspace_none = _field_claims_none(fields, "workspace mutations")
|
|
||||||
if workspace_none and (worktree_cmds or reported_worktree_nonnone):
|
|
||||||
detail = worktree_cmds[0] if worktree_cmds else reported_worktree
|
|
||||||
reasons.append(
|
|
||||||
"report claims 'Workspace mutations: none' but worktree "
|
|
||||||
f"mutations occurred ({detail}); use precise categories such as "
|
|
||||||
"'File edits by reviewer: none' plus 'Worktree mutations: <cmd>'"
|
|
||||||
)
|
|
||||||
|
|
||||||
if worktree_cmds and (
|
|
||||||
"worktree mutations" not in fields
|
|
||||||
or _field_claims_none(fields, "worktree mutations")
|
|
||||||
):
|
|
||||||
reasons.append(
|
|
||||||
"observed worktree mutation not reported under 'Worktree "
|
|
||||||
f"mutations': {worktree_cmds[0]}"
|
|
||||||
)
|
|
||||||
|
|
||||||
if ref_cmds and (
|
|
||||||
"git ref mutations" not in fields
|
|
||||||
or _field_claims_none(fields, "git ref mutations")
|
|
||||||
):
|
|
||||||
reasons.append(
|
|
||||||
"observed git ref mutation not reported under 'Git ref "
|
|
||||||
f"mutations': {ref_cmds[0]}"
|
|
||||||
)
|
|
||||||
|
|
||||||
return {
|
|
||||||
"complete": not reasons,
|
|
||||||
"downgraded": bool(reasons),
|
|
||||||
"reasons": reasons,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
|
def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
|
||||||
justification=None):
|
justification=None):
|
||||||
"""Assess reviewer/author role separation for blind queue workflows.
|
"""Assess reviewer/author role separation for blind queue workflows.
|
||||||
@@ -3510,17 +3391,6 @@ _PROOF_WORDING_RULES: tuple[dict, ...] = (
|
|||||||
"text_evidence": _PAGINATION_FINALITY_EVIDENCE,
|
"text_evidence": _PAGINATION_FINALITY_EVIDENCE,
|
||||||
"allow_prior_label": False,
|
"allow_prior_label": False,
|
||||||
},
|
},
|
||||||
{
|
|
||||||
"id": "same_as_master",
|
|
||||||
"pattern": re.compile(r"\bsame as master\b", re.I),
|
|
||||||
"session_keys": ("baseline_worktree_proof", "clean_baseline_proof"),
|
|
||||||
"text_evidence": re.compile(
|
|
||||||
r"baseline worktree|clean baseline|diff base.*master|"
|
|
||||||
r"base[- ]equivalent.*master|worktree proof",
|
|
||||||
re.I,
|
|
||||||
),
|
|
||||||
"allow_prior_label": False,
|
|
||||||
},
|
|
||||||
{
|
{
|
||||||
"id": "no_file_edits",
|
"id": "no_file_edits",
|
||||||
"pattern": re.compile(
|
"pattern": re.compile(
|
||||||
@@ -3586,12 +3456,7 @@ def assess_proof_wording(
|
|||||||
*,
|
*,
|
||||||
session_evidence: dict | None = None,
|
session_evidence: dict | None = None,
|
||||||
) -> dict:
|
) -> dict:
|
||||||
"""Issue #330: reject unsupported proof phrases in final reports.
|
"""Issue #330: reject unsupported proof phrases in final reports."""
|
||||||
|
|
||||||
Forbidden wording such as ``inventory complete``, ``live proof``, and
|
|
||||||
``same as master`` is allowed only when matching current-session evidence
|
|
||||||
appears in the report text or in *session_evidence*.
|
|
||||||
"""
|
|
||||||
text = report_text or ""
|
text = report_text or ""
|
||||||
evidence = dict(session_evidence or {})
|
evidence = dict(session_evidence or {})
|
||||||
violations: list[str] = []
|
violations: list[str] = []
|
||||||
|
|||||||
@@ -1,93 +0,0 @@
|
|||||||
"""Doc-contract checks for proof wording enforcement (#330)."""
|
|
||||||
import unittest
|
|
||||||
|
|
||||||
from review_proofs import assess_proof_wording, build_final_report
|
|
||||||
|
|
||||||
|
|
||||||
class TestProofWording(unittest.TestCase):
|
|
||||||
def test_rejects_inventory_complete_without_pagination_proof(self):
|
|
||||||
result = assess_proof_wording(
|
|
||||||
"Open PR inventory is complete because only 3 PRs were returned."
|
|
||||||
)
|
|
||||||
self.assertFalse(result["proven"])
|
|
||||||
self.assertTrue(result["block"])
|
|
||||||
|
|
||||||
def test_rejects_default_page_size_assumption(self):
|
|
||||||
result = assess_proof_wording(
|
|
||||||
"Inventory exhaustive: fewer than the default Gitea page size returned."
|
|
||||||
)
|
|
||||||
self.assertFalse(result["proven"])
|
|
||||||
self.assertIn("default_page_size_assumption", result["flagged_rules"])
|
|
||||||
|
|
||||||
def test_allows_inventory_complete_with_finality_metadata(self):
|
|
||||||
result = assess_proof_wording(
|
|
||||||
"Inventory complete. pagination_complete: true, is_final_page: true"
|
|
||||||
)
|
|
||||||
self.assertTrue(result["proven"])
|
|
||||||
|
|
||||||
def test_allows_inventory_complete_with_session_evidence(self):
|
|
||||||
result = assess_proof_wording(
|
|
||||||
"Queue inventory complete.",
|
|
||||||
session_evidence={"pagination_complete": True},
|
|
||||||
)
|
|
||||||
self.assertTrue(result["proven"])
|
|
||||||
|
|
||||||
def test_rejects_live_proof_without_session_evidence(self):
|
|
||||||
result = assess_proof_wording("Live proof confirms mergeable state.")
|
|
||||||
self.assertFalse(result["proven"])
|
|
||||||
self.assertIn("live_proof", result["flagged_rules"])
|
|
||||||
|
|
||||||
def test_allows_prior_blocker_reuse_wording(self):
|
|
||||||
result = assess_proof_wording(
|
|
||||||
"Prior blocker reused; head SHA unchanged since blocking review. "
|
|
||||||
"Live proof not claimed for this skip."
|
|
||||||
)
|
|
||||||
self.assertTrue(result["proven"])
|
|
||||||
|
|
||||||
def test_allows_live_proof_with_current_session_evidence(self):
|
|
||||||
result = assess_proof_wording(
|
|
||||||
"Live proof: gitea_view_pr revalidated in the current session."
|
|
||||||
)
|
|
||||||
self.assertTrue(result["proven"])
|
|
||||||
|
|
||||||
def test_rejects_same_as_master_without_baseline_proof(self):
|
|
||||||
result = assess_proof_wording("Diff is same as master.")
|
|
||||||
self.assertFalse(result["proven"])
|
|
||||||
self.assertIn("same_as_master", result["flagged_rules"])
|
|
||||||
|
|
||||||
def test_allows_same_as_master_with_baseline_wording(self):
|
|
||||||
result = assess_proof_wording(
|
|
||||||
"Clean baseline worktree proof: diff base matches master."
|
|
||||||
)
|
|
||||||
self.assertTrue(result["proven"])
|
|
||||||
|
|
||||||
def test_rejects_file_edits_none_without_ledger(self):
|
|
||||||
result = assess_proof_wording("Workspace mutations: file edits none.")
|
|
||||||
self.assertFalse(result["proven"])
|
|
||||||
self.assertIn("no_file_edits", result["flagged_rules"])
|
|
||||||
|
|
||||||
def test_allows_file_edits_none_with_mutation_ledger(self):
|
|
||||||
result = assess_proof_wording(
|
|
||||||
"Mutation ledger: tracked file edits: none",
|
|
||||||
session_evidence={"mutation_ledger_clean": True},
|
|
||||||
)
|
|
||||||
self.assertTrue(result["proven"])
|
|
||||||
|
|
||||||
def test_build_final_report_downgrades_on_proof_wording_violation(self):
|
|
||||||
report = build_final_report(
|
|
||||||
{"proven": True},
|
|
||||||
{"complete": True},
|
|
||||||
{"claimable": True, "verdict": "strong"},
|
|
||||||
{"status": "clean"},
|
|
||||||
True,
|
|
||||||
False,
|
|
||||||
True,
|
|
||||||
report_text="Inventory complete after listing 4 open PRs.",
|
|
||||||
)
|
|
||||||
self.assertEqual(report["grade"], "downgraded")
|
|
||||||
self.assertFalse(report["proof_wording_proven"])
|
|
||||||
self.assertTrue(report["proof_wording_violations"])
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
unittest.main()
|
|
||||||
@@ -1,165 +0,0 @@
|
|||||||
"""Tests for workspace-vs-worktree mutation consistency verifier (#313).
|
|
||||||
|
|
||||||
Final reports must not claim ``Workspace mutations: none`` when worktree
|
|
||||||
mutations (reset --hard, checkout, clean, worktree add/remove, ...)
|
|
||||||
occurred, and must report observed worktree / git ref mutations under the
|
|
||||||
precise category fields.
|
|
||||||
"""
|
|
||||||
import sys
|
|
||||||
import unittest
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
||||||
|
|
||||||
from review_proofs import assess_workspace_mutation_consistency # noqa: E402
|
|
||||||
|
|
||||||
|
|
||||||
class TestWorkspaceNoneContradiction(unittest.TestCase):
|
|
||||||
def test_reset_hard_with_workspace_none_fails(self):
|
|
||||||
report = (
|
|
||||||
"Controller Handoff\n"
|
|
||||||
"- Workspace mutations: none (no local file edits)\n"
|
|
||||||
)
|
|
||||||
result = assess_workspace_mutation_consistency(
|
|
||||||
report,
|
|
||||||
observed_commands=["git reset --hard FETCH_HEAD"],
|
|
||||||
)
|
|
||||||
self.assertFalse(result["complete"])
|
|
||||||
self.assertTrue(result["downgraded"])
|
|
||||||
self.assertTrue(
|
|
||||||
any("workspace mutations" in r.lower() for r in result["reasons"])
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_checkout_with_workspace_none_fails(self):
|
|
||||||
report = "- Workspace mutations: none\n"
|
|
||||||
result = assess_workspace_mutation_consistency(
|
|
||||||
report,
|
|
||||||
observed_commands=["git checkout feat/some-branch"],
|
|
||||||
)
|
|
||||||
self.assertFalse(result["complete"])
|
|
||||||
|
|
||||||
def test_worktree_add_with_workspace_none_fails(self):
|
|
||||||
report = "- Workspace mutations: none\n"
|
|
||||||
result = assess_workspace_mutation_consistency(
|
|
||||||
report,
|
|
||||||
observed_commands=["git worktree add /tmp/review-pr1 abc123"],
|
|
||||||
)
|
|
||||||
self.assertFalse(result["complete"])
|
|
||||||
|
|
||||||
def test_clean_with_workspace_none_fails(self):
|
|
||||||
report = "- Workspace mutations: none\n"
|
|
||||||
result = assess_workspace_mutation_consistency(
|
|
||||||
report,
|
|
||||||
observed_commands=["git clean -fd"],
|
|
||||||
)
|
|
||||||
self.assertFalse(result["complete"])
|
|
||||||
|
|
||||||
def test_self_reported_worktree_mutation_contradicts_workspace_none(self):
|
|
||||||
# No observed command log; the report itself carries the
|
|
||||||
# contradiction (#313 observed behavior).
|
|
||||||
report = (
|
|
||||||
"- Worktree mutations: git reset --hard FETCH_HEAD\n"
|
|
||||||
"- Workspace mutations: none (no local file edits)\n"
|
|
||||||
)
|
|
||||||
result = assess_workspace_mutation_consistency(report)
|
|
||||||
self.assertFalse(result["complete"])
|
|
||||||
self.assertTrue(
|
|
||||||
any("workspace mutations" in r.lower() for r in result["reasons"])
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class TestPreciseCategoriesPass(unittest.TestCase):
|
|
||||||
def test_file_edits_none_with_worktree_mutation_listed_passes(self):
|
|
||||||
report = (
|
|
||||||
"- File edits by reviewer: none\n"
|
|
||||||
"- Worktree mutations: git reset --hard FETCH_HEAD\n"
|
|
||||||
"- Git ref mutations: git fetch prgs\n"
|
|
||||||
)
|
|
||||||
result = assess_workspace_mutation_consistency(
|
|
||||||
report,
|
|
||||||
observed_commands=[
|
|
||||||
"git fetch prgs",
|
|
||||||
"git reset --hard FETCH_HEAD",
|
|
||||||
],
|
|
||||||
)
|
|
||||||
self.assertTrue(result["complete"])
|
|
||||||
self.assertFalse(result["downgraded"])
|
|
||||||
self.assertEqual(result["reasons"], [])
|
|
||||||
|
|
||||||
def test_noop_report_passes(self):
|
|
||||||
report = (
|
|
||||||
"- File edits by reviewer: none\n"
|
|
||||||
"- Worktree mutations: none\n"
|
|
||||||
"- Git ref mutations: none\n"
|
|
||||||
)
|
|
||||||
result = assess_workspace_mutation_consistency(
|
|
||||||
report, observed_commands=[]
|
|
||||||
)
|
|
||||||
self.assertTrue(result["complete"])
|
|
||||||
self.assertEqual(result["reasons"], [])
|
|
||||||
|
|
||||||
def test_fetch_only_with_ref_mutation_reported_passes(self):
|
|
||||||
# Fetch is a git ref mutation, not a worktree mutation; a
|
|
||||||
# workspace-none claim is not contradicted by fetch alone.
|
|
||||||
report = (
|
|
||||||
"- Workspace mutations: none\n"
|
|
||||||
"- Git ref mutations: git fetch prgs master\n"
|
|
||||||
)
|
|
||||||
result = assess_workspace_mutation_consistency(
|
|
||||||
report,
|
|
||||||
observed_commands=["git fetch prgs master"],
|
|
||||||
)
|
|
||||||
self.assertTrue(result["complete"])
|
|
||||||
|
|
||||||
|
|
||||||
class TestObservedMutationsMustBeReported(unittest.TestCase):
|
|
||||||
def test_reset_without_worktree_mutation_field_fails(self):
|
|
||||||
report = "- File edits by reviewer: none\n"
|
|
||||||
result = assess_workspace_mutation_consistency(
|
|
||||||
report,
|
|
||||||
observed_commands=["git reset --hard FETCH_HEAD"],
|
|
||||||
)
|
|
||||||
self.assertFalse(result["complete"])
|
|
||||||
self.assertTrue(
|
|
||||||
any("worktree mutation" in r.lower() for r in result["reasons"])
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_reset_with_worktree_mutations_none_fails(self):
|
|
||||||
report = (
|
|
||||||
"- File edits by reviewer: none\n"
|
|
||||||
"- Worktree mutations: none\n"
|
|
||||||
)
|
|
||||||
result = assess_workspace_mutation_consistency(
|
|
||||||
report,
|
|
||||||
observed_commands=["git reset --hard FETCH_HEAD"],
|
|
||||||
)
|
|
||||||
self.assertFalse(result["complete"])
|
|
||||||
|
|
||||||
def test_fetch_without_ref_mutation_field_fails(self):
|
|
||||||
report = (
|
|
||||||
"- File edits by reviewer: none\n"
|
|
||||||
"- Worktree mutations: none\n"
|
|
||||||
)
|
|
||||||
result = assess_workspace_mutation_consistency(
|
|
||||||
report,
|
|
||||||
observed_commands=["git fetch prgs master"],
|
|
||||||
)
|
|
||||||
self.assertFalse(result["complete"])
|
|
||||||
self.assertTrue(
|
|
||||||
any("git ref mutation" in r.lower() for r in result["reasons"])
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_fetch_with_ref_mutations_none_fails(self):
|
|
||||||
report = (
|
|
||||||
"- Worktree mutations: none\n"
|
|
||||||
"- Git ref mutations: none\n"
|
|
||||||
)
|
|
||||||
result = assess_workspace_mutation_consistency(
|
|
||||||
report,
|
|
||||||
observed_commands=["git fetch prgs master"],
|
|
||||||
)
|
|
||||||
self.assertFalse(result["complete"])
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
unittest.main()
|
|
||||||
Reference in New Issue
Block a user