Add role-boundary proofs for reviewer queue workflows

review_proofs.py

@@ -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":