Add role-boundary proofs for reviewer queue workflows (Issue #175) #192

Merged
sysadmin merged 1 commits from feat/issue-175-role-pivot-wall into master 2026-07-05 15:24:45 -05:00
4 changed files with 241 additions and 8 deletions
+101 -1
View File
@@ -330,9 +330,97 @@ def assess_self_review_contamination(reviewer_identity, pr_author,
}
def assess_role_boundary(proof):
"""Assess reviewer/author role separation for blind queue workflows.
Issue #175 blocks a reviewer queue task from silently becoming author
implementation work. The workflow may use both namespaces only when that
mixed use is explicit, justified, and non-mutating; author mutations after
a reviewer queue task require an explicit operator authorization.
*proof* keys:
``task_role`` ('reviewer' or 'author'), ``task_kind`` (for example
'blind_pr_queue_review'), ``reviewer_namespace_used``,
``author_namespace_used``, ``author_mutations`` (list), ``review_mutations``
(list), ``operator_authorized_author_work``, ``mixed_namespace_justification``,
``scratch_evidence_claimed``, and ``scratch_evidence_durable``.
"""
proof = proof or {}
task_role = (proof.get("task_role") or "").strip().lower()
task_kind = (proof.get("task_kind") or "").strip().lower()
author_mutations = list(proof.get("author_mutations") or [])
review_mutations = list(proof.get("review_mutations") or [])
reviewer_used = bool(proof.get("reviewer_namespace_used"))
author_used = bool(proof.get("author_namespace_used"))
authorized = bool(proof.get("operator_authorized_author_work"))
mixed_justification = (
proof.get("mixed_namespace_justification") or ""
).strip()
scratch_claimed = bool(proof.get("scratch_evidence_claimed"))
scratch_durable = bool(proof.get("scratch_evidence_durable"))
reasons = []
violations = []
if task_role not in {"reviewer", "author"}:
reasons.append("task role missing or unknown; role boundary unproven")
if task_role == "reviewer":
if author_mutations and not authorized:
violations.append(
"reviewer task performed author mutations without explicit "
"operator authorization"
)
if author_used and not mixed_justification:
reasons.append(
"reviewer task used author namespace without an explicit "
"justification"
)
if task_kind == "blind_pr_queue_review" and author_mutations:
if not authorized:
violations.append(
"blind PR queue review silently pivoted into author "
"implementation"
)
elif task_role == "author":
if review_mutations:
violations.append(
"author task performed reviewer-only mutations"
)
if reviewer_used and author_used and not mixed_justification:
reasons.append(
"mixed reviewer+author namespace use was not reported as a "
"role-boundary event"
)
if scratch_claimed and not scratch_durable:
reasons.append(
"scratch-only notes were claimed as durable evidence"
)
if violations:
status = "violation"
safe_next_action = "stop; report role-boundary violation"
elif reasons:
status = "warning"
safe_next_action = "downgrade final report; do not claim A-level proof"
else:
status = "clean"
safe_next_action = "proceed"
return {
"status": status,
"clean": status == "clean",
"reasons": reasons,
"violations": violations,
"safe_next_action": safe_next_action,
}
def build_final_report(checkout_proof, inventory, validation, contamination,
identity_eligible, merge_performed,
issue_status_verified):
issue_status_verified, role_boundary=None):
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
Combines the individual proof verdicts into the final-report fields the
@@ -348,6 +436,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
checkout_proven = bool(checkout_proof.get("proven"))
validation_claimable = bool(validation.get("claimable"))
validation_strong = validation.get("verdict") == "strong"
role_boundary = role_boundary or {
"status": "warning",
"reasons": ["role-boundary proof missing"],
"violations": [],
}
role_status = role_boundary.get("status", "warning")
downgrade_reasons = []
if not identity_eligible:
@@ -367,6 +461,9 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
downgrade_reasons.append(
f"session contamination status is '{contamination_status}'"
)
if role_status != "clean":
downgrade_reasons.append(f"role boundary status is '{role_status}'")
downgrade_reasons.extend(role_boundary.get("reasons", []))
if not issue_status_verified:
downgrade_reasons.append("linked issue status not verified")
@@ -374,6 +471,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
identity_eligible
and checkout_proven
and contamination_status == "clean"
and role_status == "clean"
and validation_claimable
and validation.get("verdict") != "invalid"
)
@@ -384,6 +482,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"merge was performed/claimed although the proofs did not allow "
"one; this run is blocked, not graded"
)
violations.extend(role_boundary.get("violations", []))
if violations:
grade = "blocked"
@@ -400,6 +499,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"pr_author_distinct_from_reviewer":
contamination_status in ("clean",),
"session_contamination": contamination_status,
"role_boundary": role_status,
"inventory_complete": bool(inventory.get("complete")),
"validated_on_pinned_head": checkout_proven and validation_claimable,
"validation_passed":
+15 -7
View File
@@ -196,19 +196,27 @@ Worktree folder = branch with `/` replaced by `-`
Both configured repos must be reported with state filter, pagination proof,
and open-PR count (`review_proofs.assess_inventory_completeness` and
`resolve_repos_from_user_reference`).
7. Inspect the full diff; confirm scope matches the linked issue; flag unrelated files.
8. Run the tests. Validation reporting must include the exact command and
7. **Role-boundary proof (#175):** a reviewer queue task must not silently
become author implementation. If no eligible PR exists, stop with the
queue report. Do not claim issues, create branches, commit, push, or open
PRs unless the operator explicitly retasks the run as author work. Mixed
reviewer+author namespace use must be reported with a justification, and
scratch-only notes are not durable evidence unless posted or committed
intentionally (`review_proofs.assess_role_boundary`).
8. Inspect the full diff; confirm scope matches the linked issue; flag unrelated files.
9. Run the tests. Validation reporting must include the exact command and
exact results: pass/fail, counts of tests passed/skipped/failed, any
ignored paths and why they are safe to ignore, and whether the command
differs from the repository's canonical validation command. Only claim a
validation result after the command has completed and its output has
been read (`review_proofs.assess_validation_report`).
9. **Do not merge if checks fail. Do not merge if the reviewer is the author.**
10. The final report must distinguish (`review_proofs.build_final_report`):
10. **Do not merge if checks fail. Do not merge if the reviewer is the author.**
11. The final report must distinguish (`review_proofs.build_final_report`):
identity eligible; PR author different from reviewer; session
contamination absent (with evidence); validation performed on the pinned
head; merge performed; issue status verified. If any proof is missing,
stop or downgrade the result instead of merging confidently.
contamination absent (with evidence); role boundary clean; validation
performed on the pinned head; merge performed; issue status verified. If
any proof is missing, stop or downgrade the result instead of merging
confidently.
## G. Merge / cleanup workflow
@@ -23,6 +23,10 @@ Rules (llm-project-workflow):
- 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).
- Do not pivot from a reviewer queue task into author implementation unless
the operator explicitly retasks the run. If author namespace was used, the
final report must justify why; author mutations after reviewer queue work
without explicit authorization are a role-boundary violation.
- Do not merge if any check fails.
Steps:
@@ -39,6 +43,10 @@ Steps:
cannot evidence whether this session authored/touched the PR branch,
report contamination as UNKNOWN (not contaminated, not clean) and choose
another PR or stop.
Role-boundary claims must also be evidence-backed (#175): report whether
reviewer namespace, author namespace, author mutations, or review mutations
occurred. Use `review_proofs.assess_role_boundary`; if it is not clean,
downgrade or stop instead of claiming an A-level run.
5. scripts/worktree-review <pr-head-branch> # detached, branches/review-*
cd branches/review-<pr-head-branch-slug>
6. Checkout proof (#173) — prove and state, before any diff review or
+117
View File
@@ -22,6 +22,7 @@ sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.par
from review_proofs import ( # noqa: E402
assess_inventory_completeness,
assess_role_boundary,
assess_self_review_contamination,
assess_validation_report,
build_final_report,
@@ -98,6 +99,21 @@ def _good_contamination():
)
def _good_role_boundary():
return assess_role_boundary(
{
"task_role": "reviewer",
"task_kind": "blind_pr_queue_review",
"reviewer_namespace_used": True,
"author_namespace_used": False,
"author_mutations": [],
"review_mutations": [],
"operator_authorized_author_work": False,
"scratch_evidence_claimed": False,
}
)
class TestCheckoutProof(unittest.TestCase):
"""Required behavior 1 + 2: prove HEAD == pinned PR head or stop."""
@@ -367,6 +383,74 @@ class TestSelfReviewContamination(unittest.TestCase):
self.assertEqual(result["status"], "unknown")
class TestRoleBoundary(unittest.TestCase):
"""Issue #175: reviewer queue tasks must not pivot into author work."""
def test_reviewer_queue_without_author_mutations_is_clean(self):
result = _good_role_boundary()
self.assertEqual(result["status"], "clean")
self.assertEqual(result["violations"], [])
def test_reviewer_queue_author_mutation_without_authorization_violates(self):
result = assess_role_boundary(
{
"task_role": "reviewer",
"task_kind": "blind_pr_queue_review",
"reviewer_namespace_used": True,
"author_namespace_used": True,
"author_mutations": ["claim issue #171", "push branch"],
"operator_authorized_author_work": False,
"mixed_namespace_justification": (
"author namespace was used for implementation"
),
}
)
self.assertEqual(result["status"], "violation")
self.assertTrue(any("pivot" in r for r in result["violations"]))
def test_mixed_namespace_use_without_justification_is_warning(self):
result = assess_role_boundary(
{
"task_role": "reviewer",
"task_kind": "blind_pr_queue_review",
"reviewer_namespace_used": True,
"author_namespace_used": True,
"author_mutations": [],
}
)
self.assertEqual(result["status"], "warning")
self.assertTrue(
any("mixed" in r.lower() for r in result["reasons"])
)
def test_author_task_cannot_perform_review_mutations(self):
result = assess_role_boundary(
{
"task_role": "author",
"reviewer_namespace_used": False,
"author_namespace_used": True,
"review_mutations": ["approve PR"],
}
)
self.assertEqual(result["status"], "violation")
self.assertTrue(
any("reviewer-only" in r for r in result["violations"])
)
def test_scratch_only_notes_are_not_durable_evidence(self):
result = assess_role_boundary(
{
"task_role": "reviewer",
"task_kind": "blind_pr_queue_review",
"reviewer_namespace_used": True,
"scratch_evidence_claimed": True,
"scratch_evidence_durable": False,
}
)
self.assertEqual(result["status"], "warning")
self.assertTrue(any("scratch-only" in r for r in result["reasons"]))
class TestFinalReport(unittest.TestCase):
"""Required behavior 6 + acceptance criteria: the report must
distinguish each proof, and only a fully proven run earns an "A"."""
@@ -380,6 +464,7 @@ class TestFinalReport(unittest.TestCase):
"identity_eligible": True,
"merge_performed": False,
"issue_status_verified": True,
"role_boundary": _good_role_boundary(),
}
kwargs.update(overrides)
return build_final_report(**kwargs)
@@ -392,6 +477,7 @@ class TestFinalReport(unittest.TestCase):
self.assertTrue(report["identity_eligible"])
self.assertTrue(report["pr_author_distinct_from_reviewer"])
self.assertEqual(report["session_contamination"], "clean")
self.assertEqual(report["role_boundary"], "clean")
self.assertTrue(report["validated_on_pinned_head"])
self.assertFalse(report["merge_performed"])
self.assertTrue(report["issue_status_verified"])
@@ -464,6 +550,37 @@ class TestFinalReport(unittest.TestCase):
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["merge_allowed"])
def test_missing_role_boundary_downgrades_and_blocks_merge(self):
kwargs = {
"checkout_proof": _good_checkout(),
"inventory": _good_inventory(),
"validation": _good_validation(),
"contamination": _good_contamination(),
"identity_eligible": True,
"merge_performed": False,
"issue_status_verified": True,
}
report = build_final_report(**kwargs)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["merge_allowed"])
self.assertEqual(report["role_boundary"], "warning")
def test_role_boundary_violation_blocks_report(self):
boundary = assess_role_boundary(
{
"task_role": "reviewer",
"task_kind": "blind_pr_queue_review",
"reviewer_namespace_used": True,
"author_namespace_used": True,
"author_mutations": ["create PR"],
"operator_authorized_author_work": False,
"mixed_namespace_justification": "implementation pivot",
}
)
report = self._report(role_boundary=boundary)
self.assertEqual(report["grade"], "blocked")
self.assertFalse(report["merge_allowed"])
class TestStdoutIsolation(unittest.TestCase):
"""Regression test for #178: tests must not close or corrupt stdout/stderr