fix(reviewer-bot): harden stage2 readiness closure gates

scripts/reviewer_bot_lib/reconcile.py

@@ -593,10 +593,10 @@ def _build_result(state_changed: bool, pr_number: int) -> WorkflowRunHandlerResu
     pr_number = parsed_payload.pr_number
     if pr_number <= 0:
         raise RuntimeError("Deferred context is missing a valid PR number")
-    bot.collect_touched_item(pr_number)
-    review_data = ensure_review_entry(state, pr_number, create=True)
+    review_data = ensure_review_entry(state, pr_number)
     if review_data is None:
-        raise RuntimeError(f"No review entry available for PR #{pr_number}")
+        raise RuntimeError(f"No active review entry available for PR #{pr_number}")
+    bot.collect_touched_item(pr_number)
     try:
         handler = _workflow_run_handler_for_payload(parsed_payload)
         if handler is None:

tests/contract/reviewer_bot/test_stage2_closure_artifacts.py

@@ -0,0 +1,144 @@
+import json
+from pathlib import Path
+
+import pytest
+
+pytestmark = pytest.mark.contract
+
+
+def _load_fixture(name: str) -> dict:
+    return json.loads(Path("tests/fixtures/workflow_contracts", name).read_text(encoding="utf-8"))
+
+
+def _transition_notice_gate_ready(payload: dict, expected_ref: str) -> bool:
+    return (
+        payload["artifact_id"] == "transition-notice-fallback-closure"
+        and payload["evaluated_repo"] == "rustfoundation/safety-critical-rust-coding-guidelines"
+        and payload["evaluated_ref"] == expected_ref
+        and payload["closure_ready"] is True
+        and payload["remaining_transition_due_without_notice"] == []
+    )
+
+
+def _deferred_payload_gate_ready(payload: dict, expected_ref: str) -> bool:
+    return (
+        payload["artifact_id"] == "deferred-payload-legacy-closure"
+        and payload["evaluated_repo"] == "rustfoundation/safety-critical-rust-coding-guidelines"
+        and payload["evaluated_ref"] == expected_ref
+        and payload["closure_ready"] is True
+        and payload["retained_workflow_inventory_matches"] is True
+        and payload["blocking_workflows"] == []
+        and payload["queued_or_in_progress_runs"] == []
+        and payload["legacy_artifacts_remaining"] == []
+    )
+
+
+def _stage2_closure_cluster_gate(
+    transition_notice_payload: dict,
+    deferred_payload: dict,
+    *,
+    expected_ref: str,
+) -> bool:
+    return _transition_notice_gate_ready(transition_notice_payload, expected_ref) and _deferred_payload_gate_ready(
+        deferred_payload, expected_ref
+    )
+
+
+@pytest.mark.parametrize(
+    ("fixture_name", "expected_ready"),
+    [
+        ("stage2_transition_notice_fallback_closure_green.json", True),
+        ("stage2_transition_notice_fallback_closure_blocked.json", False),
+    ],
+)
+def test_transition_notice_closure_fixture_schema_and_gate_rule(fixture_name, expected_ready):
+    payload = _load_fixture(fixture_name)
+
+    assert set(payload) == {
+        "artifact_id",
+        "generated_at",
+        "evaluated_repo",
+        "evaluated_ref",
+        "state_issue_number",
+        "closure_ready",
+        "active_reviews_scanned",
+        "resolved_by_marker_backfill",
+        "resolved_by_legacy_prose_backfill",
+        "resolved_by_new_marker_notice",
+        "remaining_transition_due_without_notice",
+        "commands_run",
+    }
+    assert payload["closure_ready"] is expected_ready
+    assert payload["closure_ready"] == (payload["remaining_transition_due_without_notice"] == [])
+
+
+@pytest.mark.parametrize(
+    ("fixture_name", "expected_ready"),
+    [
+        ("stage2_deferred_payload_legacy_closure_green.json", True),
+        ("stage2_deferred_payload_legacy_closure_blocked.json", False),
+    ],
+)
+def test_deferred_payload_closure_fixture_schema_and_gate_rule(fixture_name, expected_ready):
+    payload = _load_fixture(fixture_name)
+
+    assert set(payload) == {
+        "artifact_id",
+        "generated_at",
+        "evaluated_repo",
+        "evaluated_ref",
+        "closure_ready",
+        "retained_workflow_inventory_matches",
+        "blocking_workflows",
+        "queued_or_in_progress_runs",
+        "legacy_artifacts_remaining",
+        "control_plane_actions_applied",
+        "commands_run",
+    }
+    assert payload["closure_ready"] is expected_ready
+    assert payload["closure_ready"] == (
+        payload["retained_workflow_inventory_matches"]
+        and payload["blocking_workflows"] == []
+        and payload["queued_or_in_progress_runs"] == []
+        and payload["legacy_artifacts_remaining"] == []
+    )
+
+
+def test_stage2_closure_artifacts_reject_stale_attempt_refs():
+    transition_notice_payload = _load_fixture("stage2_transition_notice_fallback_closure_green.json")
+    deferred_payload = _load_fixture("stage2_deferred_payload_legacy_closure_green.json")
+
+    assert _transition_notice_gate_ready(
+        transition_notice_payload,
+        "0000000000000000000000000000000000000000",
+    ) is False
+    assert _deferred_payload_gate_ready(
+        deferred_payload,
+        "0000000000000000000000000000000000000000",
+    ) is False
+
+
+def test_stage2_closure_cluster_gate_requires_both_current_attempt_green_artifacts():
+    transition_notice_payload = _load_fixture("stage2_transition_notice_fallback_closure_green.json")
+    deferred_payload = _load_fixture("stage2_deferred_payload_legacy_closure_green.json")
+    blocked_transition_notice_payload = _load_fixture(
+        "stage2_transition_notice_fallback_closure_blocked.json"
+    )
+    blocked_deferred_payload = _load_fixture("stage2_deferred_payload_legacy_closure_blocked.json")
+    expected_ref = transition_notice_payload["evaluated_ref"]
+
+    assert _stage2_closure_cluster_gate(
+        transition_notice_payload,
+        deferred_payload,
+        expected_ref=expected_ref,
+    ) is True
+    assert _stage2_closure_cluster_gate(
+        blocked_transition_notice_payload,
+        deferred_payload,
+        expected_ref=expected_ref,
+    ) is False
+    assert _stage2_closure_cluster_gate(
+        transition_notice_payload,
+        blocked_deferred_payload,
+        expected_ref=expected_ref,
+    ) is False

tests/contract/reviewer_bot/test_stage2_deferred_payload_residue.py

@@ -0,0 +1,128 @@
+import json
+from pathlib import Path
+
+import pytest
+
+pytestmark = pytest.mark.contract
+
+
+def _load_cases() -> dict:
+    return json.loads(
+        Path("tests/fixtures/workflow_contracts/deferred_payload_residue_cases.json").read_text(
+            encoding="utf-8"
+        )
+    )
+
+
+def _simulate_residue_audit(case: dict) -> dict:
+    blocking_workflows = []
+    queued_or_in_progress_runs = []
+    legacy_artifacts_remaining = []
+    evaluated_ref = case["evaluated_ref"]
+
+    for workflow in case["workflows"]:
+        if workflow["classification"] == "required_retained" and workflow["state"] != "active":
+            blocking_workflows.append(
+                {
+                    "workflow_name": workflow["workflow_name"],
+                    "workflow_path": workflow["workflow_path"],
+                    "reason": "required_retained_workflow_disabled",
+                }
+            )
+        if workflow["classification"] == "removed_legacy" and workflow["state"] == "active":
+            blocking_workflows.append(
+                {
+                    "workflow_name": workflow["workflow_name"],
+                    "workflow_path": workflow["workflow_path"],
+                    "reason": "removed_legacy_workflow_still_active",
+                }
+            )
+
+    for run in case["runs"]:
+        if run["status"] not in {"queued", "in_progress"}:
+            continue
+        if run["classification"] == "removed_legacy":
+            reason = "removed_legacy_workflow_run"
+        elif run["head_sha"] != evaluated_ref:
+            reason = "noncurrent_head_sha"
+        else:
+            continue
+        queued_or_in_progress_runs.append(
+            {
+                "workflow_name": run["workflow_name"],
+                "workflow_path": run["workflow_path"],
+                "run_id": run["run_id"],
+                "run_attempt": run["run_attempt"],
+                "status": run["status"],
+                "head_sha": run["head_sha"],
+                "reason": reason,
+            }
+        )
+
+    for artifact in case["artifacts"]:
+        if artifact["downloadable"] and artifact["payload_schema_version"] in {1, 2}:
+            legacy_artifacts_remaining.append(
+                {
+                    "workflow_name": artifact["workflow_name"],
+                    "workflow_path": artifact["workflow_path"],
+                    "run_id": artifact["run_id"],
+                    "run_attempt": artifact["run_attempt"],
+                    "artifact_id": artifact["artifact_id"],
+                    "payload_schema_version": artifact["payload_schema_version"],
+                    "payload_kind": artifact["payload_kind"],
+                    "artifact_name": artifact["artifact_name"],
+                    "payload_filename": artifact["payload_filename"],
+                }
+            )
+
+    return {
+        "retained_workflow_inventory_matches": case["retained_workflow_inventory_matches"],
+        "blocking_workflows": blocking_workflows,
+        "queued_or_in_progress_runs": queued_or_in_progress_runs,
+        "legacy_artifacts_remaining": legacy_artifacts_remaining,
+        "closure_ready": case["retained_workflow_inventory_matches"]
+        and blocking_workflows == []
+        and queued_or_in_progress_runs == []
+        and legacy_artifacts_remaining == [],
+    }
+
+
+def _select_deferred_payload(files: list[str]) -> str | None:
+    json_files = sorted(path for path in files if path.endswith(".json"))
+    if len(json_files) > 1:
+        raise RuntimeError(f"Expected at most one deferred payload, found {len(json_files)}")
+    if len(json_files) == 1:
+        return json_files[0]
+    return None
+
+
+@pytest.mark.parametrize("case", _load_cases()["residue_cases"], ids=lambda case: case["id"])
+def test_b5e_residue_cases_are_simulated_locally(case):
+    simulated = _simulate_residue_audit(case)
+
+    assert simulated == case["expected"]
+
+
+@pytest.mark.parametrize(
+    "case",
+    _load_cases()["artifact_selection_cases"],
+    ids=lambda case: case["id"],
+)
+def test_reconcile_artifact_selection_cases_fail_closed(case):
+    if case["expected_multiple"]:
+        with pytest.raises(RuntimeError, match="Expected at most one deferred payload"):
+            _select_deferred_payload(case["files"])
+        return
+
+    assert _select_deferred_payload(case["files"]) == case["expected_selected_path"]
+
+
+def test_router_zero_artifact_success_is_not_classified_as_residue():
+    case = next(case for case in _load_cases()["residue_cases"] if case["id"] == "router_zero_artifact_success")
+
+    simulated = _simulate_residue_audit(case)
+
+    assert simulated["blocking_workflows"] == []
+    assert simulated["queued_or_in_progress_runs"] == []
+    assert simulated["legacy_artifacts_remaining"] == []
+    assert simulated["closure_ready"] is True

tests/contract/reviewer_bot/test_workflow_artifact_contracts.py

@@ -3,6 +3,7 @@
 from types import SimpleNamespace
 
 import pytest
+import yaml
 
 pytestmark = pytest.mark.contract
 
@@ -14,6 +15,12 @@ def _load_fixture_payload(relative_path: str) -> dict:
     return data["payload"]
 
 
+def _load_workflow_job(relative_path: str) -> dict:
+    workflow = yaml.safe_load(Path(relative_path).read_text(encoding="utf-8"))
+    job_name = "route-pr-comment" if relative_path.endswith("reviewer-bot-pr-comment-router.yml") else "observer"
+    return workflow["jobs"][job_name]
+
+
 @pytest.mark.parametrize(
     ("workflow_path",),
     [
@@ -77,6 +84,49 @@ def test_deferred_comment_payload_parses_without_artifact_name_field():
     assert parsed.identity.source_event_name == "issue_comment"
 
 
+@pytest.mark.parametrize(
+    ("fixture_path", "workflow_path"),
+    [
+        (
+            "tests/fixtures/observer_payloads/workflow_pr_comment_deferred.json",
+            ".github/workflows/reviewer-bot-pr-comment-router.yml",
+        ),
+        (
+            "tests/fixtures/observer_payloads/workflow_pr_review_submitted_deferred.json",
+            ".github/workflows/reviewer-bot-pr-review-submitted-observer.yml",
+        ),
+        (
+            "tests/fixtures/observer_payloads/workflow_pr_review_dismissed_deferred.json",
+            ".github/workflows/reviewer-bot-pr-review-dismissed-observer.yml",
+        ),
+        (
+            "tests/fixtures/observer_payloads/workflow_pr_review_comment_deferred.json",
+            ".github/workflows/reviewer-bot-pr-review-comment-observer.yml",
+        ),
+    ],
+)
+def test_deferred_payload_fixtures_match_upload_name_and_payload_name_helpers(
+    fixture_path, workflow_path
+):
+    payload = _load_fixture_payload(fixture_path)
+    job = _load_workflow_job(workflow_path)
+    build_step = job["steps"][0]
+    upload_step = job["steps"][1]
+    rendered_upload_name = (
+        upload_step["with"]["name"]
+        .replace("${{ github.run_id }}", str(payload["source_run_id"]))
+        .replace("${{ github.run_attempt }}", str(payload["source_run_attempt"]))
+    )
+
+    assert rendered_upload_name == reconcile_payloads.artifact_expected_name(payload)
+    assert build_step["env"]["PAYLOAD_PATH"].endswith(
+        reconcile_payloads.artifact_expected_payload_name(payload)
+    )
+    assert upload_step["with"]["path"].endswith(
+        reconcile_payloads.artifact_expected_payload_name(payload)
+    )
+
+
 def test_validate_workflow_run_artifact_identity_rejects_run_attempt_mismatch(monkeypatch):
     monkeypatch.setenv("WORKFLOW_RUN_TRIGGERING_ID", "1")
     monkeypatch.setenv("WORKFLOW_RUN_TRIGGERING_ATTEMPT", "2")

tests/contract/reviewer_bot/test_workflow_files.py

@@ -138,6 +138,16 @@ def test_workflow_summaries_and_runbook_references_exist():
     assert "docs/reviewer-bot-review-freshness-operator-runbook.md" in reconcile
 
 
+def test_reconcile_workflow_selects_at_most_one_recursive_json_payload():
+    workflow_text = Path(".github/workflows/reviewer-bot-reconcile.yml").read_text(encoding="utf-8")
+
+    assert "files = sorted(Path(os.environ['RUNNER_TEMP']).joinpath('observer-artifact').rglob('*.json'))" in workflow_text
+    assert "if len(files) > 1:" in workflow_text
+    assert "Expected at most one deferred payload" in workflow_text
+    assert "if len(files) == 1:" in workflow_text
+    assert "DEFERRED_CONTEXT_PATH=" in workflow_text
+
+
 @pytest.mark.parametrize(
     ("fixture_path", "workflow_file", "payload_kind", "expected_event_name", "expected_event_action"),
     [