fix: harden reviewer bot guardrails

Author: PLeVasseur

.github/actions/reviewer-bot-source/action.yml

@@ -10,9 +10,32 @@ runs:
     - id: source-root
       shell: bash
       run: |
+        workspace_root="$(git -C "$GITHUB_WORKSPACE" rev-parse --show-toplevel)"
+        if [ "$workspace_root" != "$GITHUB_WORKSPACE" ]; then
+          echo "reviewer-bot source checkout root does not match GITHUB_WORKSPACE" >&2
+          exit 1
+        fi
+        checked_out_sha="$(git -C "$GITHUB_WORKSPACE" rev-parse HEAD)"
+        if [ -n "${GITHUB_SHA:-}" ] && [ "$checked_out_sha" != "$GITHUB_SHA" ]; then
+          echo "reviewer-bot source checkout does not match GITHUB_SHA" >&2
+          exit 1
+        fi
+        remote_url="$(git -C "$GITHUB_WORKSPACE" config --get remote.origin.url || true)"
+        case "$remote_url" in
+          *"$GITHUB_REPOSITORY"*) ;;
+          *)
+            echo "reviewer-bot source checkout remote does not match GITHUB_REPOSITORY" >&2
+            exit 1
+            ;;
+        esac
         if [ ! -f "$GITHUB_WORKSPACE/scripts/reviewer_bot.py" ]; then
           echo "reviewer-bot source checkout is missing scripts/reviewer_bot.py" >&2
           exit 1
         fi
+        bot_entry="$(realpath "$GITHUB_WORKSPACE/scripts/reviewer_bot.py")"
+        if [ "$bot_entry" != "$GITHUB_WORKSPACE/scripts/reviewer_bot.py" ]; then
+          echo "reviewer-bot entrypoint must be a regular file inside the trusted checkout" >&2
+          exit 1
+        fi
         printf 'bot-src-root=%s\n' "$GITHUB_WORKSPACE" >> "$GITHUB_OUTPUT"
         printf 'BOT_SRC_ROOT=%s\n' "$GITHUB_WORKSPACE" >> "$GITHUB_ENV"

scripts/reviewer_bot_lib/app.py

@@ -461,6 +461,11 @@ def execute_run(bot: AppExecutionRuntime, context: EventContext) -> ExecutionRes
             if manual_projection_policy is not None
             else True
         )
+        allow_status_label_sync = (
+            manual_projection_policy.allow_status_label_sync
+            if manual_projection_policy is not None
+            else True
+        )
         if (
             lock_required
             and event_name in {"schedule", "workflow_dispatch"}
@@ -560,7 +565,16 @@ def execute_run(bot: AppExecutionRuntime, context: EventContext) -> ExecutionRes
 
         maintenance.emit_pending_issue314_state_health_repair_summary(bot)
 
-        if touched_items:
+        if touched_items and not allow_status_label_sync:
+            _log(
+                bot,
+                "info",
+                "Skipping status-label sync because manual dispatch policy is read-only.",
+                manual_action=context.manual_action or "",
+                touched_items=touched_items,
+            )
+
+        if touched_items and allow_status_label_sync:
             if not lock_acquired:
                 raise RuntimeError(
                     "Status-label projection reached apply path without lease lock. "

scripts/reviewer_bot_lib/maintenance_schedule.py

@@ -11,6 +11,7 @@
 )
 from .overdue import (
     backfill_transition_notice_if_present,
+    backfill_transition_warning_if_present,
     check_overdue_reviews,
     handle_overdue_review_warning,
 )
@@ -139,6 +140,16 @@ def _run_tracked_pr_maintenance(bot, state: dict) -> tuple[bool, list[int]]:
 
 def _run_overdue_pass(bot, state: dict) -> bool:
     changed = False
+    active_reviews = state.get("active_reviews")
+    if isinstance(active_reviews, dict):
+        for issue_key, review_data in active_reviews.items():
+            if not isinstance(review_data, dict):
+                continue
+            try:
+                issue_number = int(issue_key)
+            except ValueError:
+                continue
+            changed = backfill_transition_warning_if_present(bot, state, issue_number) or changed
     overdue_reviews = check_overdue_reviews(bot, state)
     for review in overdue_reviews:
         issue_number = review["issue_number"]

scripts/reviewer_bot_lib/overdue.py

@@ -729,6 +729,70 @@ def _effective_response_with_cadence(bot, issue_number: int, review_data: dict,
     return effective_response, cadence, reminder_scan
 
 
+def backfill_transition_warning_if_present(bot, state: dict, issue_number: int) -> bool:
+    active_reviews = state.get("active_reviews") if isinstance(state, dict) else None
+    review_data = active_reviews.get(str(issue_number)) if isinstance(active_reviews, dict) else None
+    if not isinstance(review_data, dict):
+        return False
+    if not isinstance(review_data.get("current_reviewer"), str) or not review_data["current_reviewer"].strip():
+        return False
+    existing_warning = review_data.get("transition_warning_sent")
+    if isinstance(existing_warning, str) and existing_warning.strip():
+        return False
+    try:
+        issue_snapshot_result = bot.github.get_issue_or_pr_snapshot_result(issue_number)
+    except (AssertionError, AttributeError, RuntimeError):
+        return False
+    issue_snapshot = issue_snapshot_result.payload if issue_snapshot_result.ok else None
+    if not isinstance(issue_snapshot, dict):
+        return False
+    if str(issue_snapshot.get("state", "")).lower() == "closed":
+        return False
+
+    response_state = bot.adapters.review_state.compute_reviewer_response_state(
+        issue_number,
+        review_data,
+        issue_snapshot=issue_snapshot,
+    )
+    _effective_response, cadence, _reminder_scan = _effective_response_with_cadence(
+        bot,
+        issue_number,
+        review_data,
+        response_state,
+    )
+    receipt = cadence.warning_receipt
+    if receipt is None or receipt.receipt_kind != "warning":
+        return False
+    if receipt.source not in {"comment_scan", "live_comment_scan"}:
+        return False
+    if not isinstance(receipt.created_at, str) or not receipt.created_at.strip():
+        return False
+
+    changed = False
+    if review_data.get("transition_warning_sent") != receipt.created_at:
+        review_data["transition_warning_sent"] = receipt.created_at
+        changed = True
+    delivery_result = build_reminder_delivery_persistence_result(
+        issue_number=issue_number,
+        reviewer=receipt.reviewer,
+        head_sha=receipt.head_sha,
+        cycle_key=receipt.cycle_key,
+        scope_key=receipt.scope_key,
+        receipt_kind="warning",
+        comment_posted=False,
+        comment_id=receipt.comment_id,
+        comment_created_at=receipt.created_at,
+        state_save_attempted=False,
+        state_save_succeeded=False,
+        recovered_receipt=receipt,
+        diagnostic_reason="live_warning_receipt_backfill",
+    )
+    changed = _store_reminder_delivery_result(review_data, delivery_result) or changed
+    if changed:
+        bot.collect_touched_item(issue_number)
+    return changed
+
+
 def _find_existing_warning_comment(
     bot,
     issue_number: int,

scripts/reviewer_bot_lib/reconcile.py

@@ -252,7 +252,28 @@ def build_command_replay_receipt(
     clear_gap_allowed: bool,
     diagnostic_reason: str | None = None,
 ) -> CommandReplayReceipt:
-    if not replay_attempted:
+    normalized_side_effects = tuple(
+        sorted(
+            {
+                str(item).strip()
+                for item in command_side_effects_attempted
+                if str(item).strip()
+            }
+        )
+    )
+    normalized_command_name = command_name.strip() if isinstance(command_name, str) and command_name.strip() else None
+    consistency_error = None
+    if replay_attempted and normalized_command_name is None:
+        consistency_error = "missing_command_name_for_replay"
+    elif not replay_attempted and normalized_side_effects:
+        consistency_error = "side_effects_without_replay"
+    elif not replay_attempted and state_save_required:
+        consistency_error = "state_save_required_without_replay"
+
+    if consistency_error is not None:
+        result = "blocked_inconsistent_replay_receipt"
+        diagnostic_reason = diagnostic_reason or consistency_error
+    elif not replay_attempted:
         result = "pass_diagnostic_only"
     elif not mark_reconciled_allowed or not clear_gap_allowed:
         result = "blocked_authority_missing"
@@ -263,9 +284,9 @@ def build_command_replay_receipt(
     return CommandReplayReceipt(
         source_event_key=source_event_key,
         issue_number=issue_number,
-        command_name=command_name,
+        command_name=normalized_command_name,
         replay_attempted=replay_attempted,
-        command_side_effects_attempted=command_side_effects_attempted,
+        command_side_effects_attempted=normalized_side_effects,
         state_save_required=state_save_required,
         state_save_succeeded=state_save_succeeded,
         mark_reconciled_allowed=mark_reconciled_allowed,

tests/contract/reviewer_bot/test_workflow_files.py

@@ -423,6 +423,18 @@ def test_reviewer_bot_workflows_use_shared_source_action_without_raw_extraction(
             assert "uses: ./.github/actions/reviewer-bot-source" in text
             assert "BOT_SRC_ROOT: ${{ steps.bot-source.outputs.bot-src-root }}" in text
 
+
+def test_reviewer_bot_source_action_validates_checkout_provenance():
+    action = Path(".github/actions/reviewer-bot-source/action.yml").read_text(encoding="utf-8")
+
+    assert 'git -C "$GITHUB_WORKSPACE" rev-parse --show-toplevel' in action
+    assert 'git -C "$GITHUB_WORKSPACE" rev-parse HEAD' in action
+    assert '"$checked_out_sha" != "$GITHUB_SHA"' in action
+    assert 'git -C "$GITHUB_WORKSPACE" config --get remote.origin.url' in action
+    assert '*"$GITHUB_REPOSITORY"*' in action
+    assert 'realpath "$GITHUB_WORKSPACE/scripts/reviewer_bot.py"' in action
+    assert "printf 'BOT_SRC_ROOT=%s\\n'" in action
+
 def test_workflow_summaries_and_runbook_references_exist():
     runbook = Path("docs/reviewer-bot-review-freshness-operator-runbook.md")
     assert runbook.exists()

tests/integration/reviewer_bot/test_app_execution.py

@@ -445,6 +445,30 @@ def fake_save_state(state):
     assert saved_states[-1]["active_reviews"]["42"]["sidecars"]["repair_markers"]["status_label_projection"]["kind"] == "projection_failure"
 
 
+def test_execute_run_manual_read_only_policy_blocks_touched_status_label_sync(monkeypatch):
+    harness = AppHarness(monkeypatch)
+    harness.set_event(
+        EVENT_NAME="workflow_dispatch",
+        EVENT_ACTION="",
+        MANUAL_ACTION="preview-status-label-projection",
+        ISSUE_NUMBER=42,
+        VALIDATION_NONCE="nonce",
+    )
+    harness.stub_load_state(lambda *, fail_on_unavailable=False: make_state())
+
+    def fake_manual_dispatch(state):
+        harness.runtime.collect_touched_item(42)
+        return False
+
+    harness.stub_handler("handle_manual_dispatch", fake_manual_dispatch)
+    harness.stub_sync_status_labels(lambda current, issue_numbers: pytest.fail("read-only manual policy must not sync labels"))
+
+    result = harness.run_execute()
+
+    assert result.exit_code == 0
+    assert result.state_changed is False
+
+
 def test_execute_run_returns_failure_when_lock_release_fails(monkeypatch):
     harness = AppHarness(monkeypatch)
     harness.set_event(EVENT_NAME="issue_comment", EVENT_ACTION="created")