fix: preserve projection write approval authority

Author: PLeVasseur

scripts/reviewer_bot_core/approval_policy.py

@@ -415,8 +415,6 @@ def compute_pr_approval_state_result(
         permission_evidence=permission_evidence,
         dismissal_evidence=None,
     )
-    if write_decision.response_state == "projection_failed":
-        return live_review_support.projection_failure_result(write_decision.diagnostic_reason or write_decision.write_approval_state)
     completion = {
         "completed": completion_decision.can_set_review_completed_at,
         "current_head_sha": current_head,
@@ -425,6 +423,20 @@ def compute_pr_approval_state_result(
         ),
         "authority_decision": completion_decision.to_output(),
     }
+    if write_decision.response_state == "projection_failed":
+        result = live_review_support.projection_failure_result(
+            write_decision.diagnostic_reason or write_decision.write_approval_state
+        )
+        result["completion"] = completion
+        result["write_approval"] = {
+            "has_write_approval": False,
+            "write_approvers": [],
+            "current_head_sha": current_head,
+            "response_state": write_decision.response_state,
+            "authority_decision": write_decision.to_output(),
+        }
+        result["current_head_sha"] = current_head
+        return result
     write_approval = {
         "has_write_approval": write_decision.response_state == "done",
         "write_approvers": [write_decision.approving_reviewer] if write_decision.response_state == "done" and write_decision.approving_reviewer else [],

scripts/reviewer_bot_core/reviewer_response_policy.py

@@ -475,6 +475,13 @@ def _decorate_response(
     }
 
 
+def _write_approval_authority_payload(write_approval: object) -> dict[str, object] | None:
+    if not isinstance(write_approval, dict):
+        return None
+    authority = write_approval.get("authority_decision")
+    return dict(authority) if isinstance(authority, dict) else None
+
+
 def _record_for_current_reviewer(record: dict | None | object, current_reviewer: str) -> dict | None:
     if not isinstance(record, dict):
         return None
@@ -877,14 +884,19 @@ def derive_reviewer_response_state(
         )
 
     if not isinstance(approval_result, dict) or not approval_result.get("ok"):
+        write_approval_authority = _write_approval_authority_payload(
+            approval_result.get("write_approval") if isinstance(approval_result, dict) else None
+        )
         return _decorate_response(
             state="projection_failed",
             reason="live_review_state_unknown",
             scope_fields=_current_scope_fields(review_data, current_reviewer, current_head, contributor_handoff),
+            write_approval_authority=write_approval_authority,
         )
 
     completion = approval_result["completion"]
     write_approval = approval_result["write_approval"]
+    write_approval_authority = _write_approval_authority_payload(write_approval)
     if not completion.get("completed"):
         return _decorate_response(
             state="awaiting_contributor_response",
@@ -911,6 +923,7 @@ def derive_reviewer_response_state(
             current_cycle_reviewer_handoff=reviewer_handoff,
             contributor_comment=contributor_comment,
             contributor_handoff=contributor_handoff,
+            write_approval_authority=write_approval_authority,
         )
     if authority_response_state == "projection_failed":
         authority = write_approval.get("authority_decision")
@@ -921,6 +934,7 @@ def derive_reviewer_response_state(
             state="projection_failed",
             reason=reason,
             scope_fields=_current_scope_fields(review_data, current_reviewer, current_head, contributor_handoff),
+            write_approval_authority=write_approval_authority,
         )
     if not write_approval.get("has_write_approval"):
         return _decorate_response(
@@ -934,6 +948,7 @@ def derive_reviewer_response_state(
             current_cycle_reviewer_handoff=reviewer_handoff,
             contributor_comment=contributor_comment,
             contributor_handoff=contributor_handoff,
+            write_approval_authority=write_approval_authority,
         )
     return _decorate_response(
         state="done",
@@ -946,6 +961,7 @@ def derive_reviewer_response_state(
         current_cycle_reviewer_handoff=reviewer_handoff,
         contributor_comment=contributor_comment,
         contributor_handoff=contributor_handoff,
+        write_approval_authority=write_approval_authority,
     )
 
 

tests/unit/reviewer_bot/test_reviews_live_fetch.py

@@ -411,6 +411,23 @@ def test_compute_reviewer_response_state_reports_permission_unavailable(monkeypa
 
     assert response_state["state"] == "projection_failed"
     assert response_state["reason"] == "live_review_state_unknown"
+    assert response_state["write_approval_authority"] == {
+        "issue_number": 42,
+        "head_sha": "head-1",
+        "assigned_reviewer": "alice",
+        "assigned_review_id": 10,
+        "assigned_review_state": "APPROVED",
+        "assigned_round_complete": True,
+        "write_approval_state": "blocked_unavailable_authority",
+        "write_approval_source": "github_permission_read_unavailable",
+        "approving_reviewer": "alice",
+        "approving_review_id": 10,
+        "permission_source": "github_permission_read",
+        "dismissal_supersession_status": "blocked_untrusted",
+        "response_state": "projection_failed",
+        "diagnostic_reason": "permission_unavailable",
+        "can_project_final_state": False,
+    }
 
 
 def test_trigger_mandatory_approver_escalation_sets_required_label_and_ping(monkeypatch):

tests/unit/reviewer_bot/test_reviews_projection.py

@@ -355,6 +355,51 @@ def test_status_projection_maps_reassignment_needed_and_exposes_decision_output(
     assert payload["output_keys"] == sorted(payload.keys())
 
 
+def test_status_projection_preserves_write_approval_authority_decision_output():
+    authority = {
+        "issue_number": 264,
+        "head_sha": "head-a",
+        "assigned_reviewer": "iglesias",
+        "assigned_review_id": 10,
+        "assigned_review_state": "APPROVED",
+        "assigned_round_complete": True,
+        "write_approval_state": "visibly_missing_write_approval",
+        "write_approval_source": "none_visible_after_trusted_reads",
+        "approving_reviewer": None,
+        "approving_review_id": None,
+        "permission_source": "github_permission_read",
+        "dismissal_supersession_status": "pass_not_dismissed_or_superseded",
+        "response_state": "awaiting_write_approval",
+        "diagnostic_reason": None,
+        "can_project_final_state": True,
+    }
+    decision = to_reviewer_response_decision(
+        {
+            "issue_number": 264,
+            "response_state": "awaiting_write_approval",
+            "current_scope_key": "scope-approval",
+            "current_scope_basis": "assigned_at",
+            "write_approval_authority": authority,
+        }
+    )
+
+    result = reviews_projection.derive_status_label_projection(
+        reviews_projection.StatusLabelProjectionInput(
+            issue_number=264,
+            issue_state="open",
+            actual_labels=(),
+            reviewer_response=decision,
+            reviewer_authority_outcome="tracked_reviewer_confirmed",
+            freshness_runtime_epoch="freshness_v15",
+            status_projection_epoch="status_projection_v2",
+        )
+    )
+
+    decision_output = result.projection_metadata["decision_output"]
+    assert decision_output["write_approval_authority"] == authority
+    assert result.delta.desired_status_labels == ("status: awaiting write approval",)
+
+
 @pytest.mark.parametrize("state_name", ["projection_failed", "live_read_unavailable", "unknown"])
 def test_status_projection_fail_closed_states_never_clear_existing_labels(state_name):
     decision = to_reviewer_response_decision({"response_state": state_name})