feat(auto-agents): R3.7 post-push CI verification + per-cycle budget

Closes the local-vs-remote CI divergence the 2026-05-17 run-7
observation exposed: the implementer worker's ``ci_run_local_gate
--fast`` doesn't catch what remote CI does (test sharding,
parallel-job orchestration, remote-only timing). So the worker
could push a commit + claim ``outcome=resolved`` even when remote
CI would reject it; the dispatcher would treat the cycle as a
success and not re-dispatch.

With this gate, the dispatcher is the source of truth for "did
this PR actually pass CI."

How it works:
- Runs ONLY when the worker claims ``outcome=resolved`` AND
  ``head_sha_advanced is True`` (a real push happened).
- Polls Forgejo's combined CI status on the post-session head_sha
  every 10 s for up to 90 s.
- If CI lands in {failure, error}: rewrites parsed_json's
  ``outcome`` to ``post-push-ci-failed`` so
  ``_implementer_escalation.decide`` routes it as a failure
  (ESCALATE / EXHAUSTED). Stashes the original outcome +
  failing-context list under ``_post_push_ci_verification`` for
  telemetry.
- If CI is pending after the budget: outcome unchanged (don't
  penalise the worker for slow CI; next dispatcher cycle
  re-classifies).
- If fetch fails: outcome unchanged (Forgejo flake protection —
  "I couldn't check" must not equal "the worker lied").
- Dry-run short-circuits to no-op so ``--dry-run`` cycles don't
  burn 90 s polling.

Per-cycle polling budget:

Without a cycle-wide cap, a dispatcher cycle processing N PRs all
landing in ``outcome=resolved`` after a push could spend
``N * 90 s`` polling — at 5 PRs/cycle that's 7.5 min eating the
dispatcher cycle budget. Added a sliding-window budget:

- ``IMPLEMENTER_POST_PUSH_CI_CYCLE_BUDGET_S`` (default 300 s) caps
  total polling time across all verify calls in one dispatcher
  cycle.
- ``IMPLEMENTER_POST_PUSH_CI_CYCLE_RESET_AFTER_S`` (default 120 s)
  auto-resets the budget after that much idle — naturally fires
  on cycle boundaries without the dispatcher's outer loop having
  to call a reset hook.
- Per-call budget is also capped by remaining cycle budget so a
  near-exhausted cycle doesn't get blown through by one fat call.
- ``reset_post_push_ci_cycle_budget()`` exposed for tests and any
  future dispatcher hook that wants to reset explicitly.

Tests:
- 22 tests in ``TestPostPushCIVerify`` (the pre-existing class)
  cover happy path, rewrite-on-fail, head_sha_advanced gating,
  feature-flag short-circuit, transport flake handling, failing-
  contexts extraction, etc.
- 4 new tests in ``TestPostPushCIVerifyCycleBudget`` pin the
  budget contract: exhausted budget skips verification, idle
  auto-reset works, explicit reset zeroes state, per-call cap
  respects remaining cycle budget.

Tuning notes:
- 90 s per-call: enough to catch fast-failing checks (lint/format
  fail in 30-60 s typical) without dominating the cycle.
- 10 s poll interval: 9 polls per per-call budget. Forgejo's CI
  status endpoint is fast (< 1 s typical).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

tests/auto_agents/test_dispatch_implementer.py

@@ -2251,3 +2251,465 @@ class TestMetadataOnlyClassifier:
         item = {"head": {"sha": "abc"}}
         pf = self._prefetch(ci_state="error")
         assert driver._classify_metadata_only(item, pf) is False
+
+
+class TestPostPushCIVerify:
+    """R3.7 (2026-05-17): post-push CI verification.
+
+    When the worker claims ``outcome=resolved`` AND head_sha advanced
+    (real push), the dispatcher polls remote CI on the new SHA and
+    rewrites the outcome to ``post-push-ci-failed`` if CI lands in a
+    failure state. Closes the local-fast-gate vs remote-CI divergence
+    that lets a worker push commits + claim success even when remote
+    CI will reject them.
+
+    Tests cover the full decision matrix:
+      - resolved + push + CI passes      → outcome unchanged
+      - resolved + push + CI fails       → outcome rewritten
+      - resolved + push + CI still pending → outcome unchanged (don't
+        penalise for slow CI)
+      - resolved + no push               → no verification (handled
+        downstream by escalation predicate)
+      - non-resolved outcome             → no verification
+      - dry-run                          → no verification
+      - transport error during poll      → outcome unchanged (don't
+        penalise for Forgejo flakes)
+      - post-session head_sha fetch fails → outcome unchanged
+    """
+
+    @pytest.fixture(autouse=True)
+    def _no_sleep(self, monkeypatch):
+        """Replace ``time.sleep`` with a no-op so the budget-loop
+        runs instantly. Without this each test pays the real
+        ``IMPLEMENTER_POST_PUSH_CI_VERIFY_POLL_S`` (10 s default)
+        per poll iteration. Also explicitly enables the
+        ``IMPLEMENTER_POST_PUSH_CI_VERIFY`` feature flag so the test
+        intent — exercise the verifier — is independent of the
+        environment the suite runs in."""
+        import time as _time
+        monkeypatch.setattr(_time, "sleep", lambda _s: None)
+        monkeypatch.setenv("IMPLEMENTER_POST_PUSH_CI_VERIFY", "1")
+
+    def _stub_post_session_head(
+        self, api: FakeReviewAPI, sha: str = "newsha0000feed"
+    ) -> None:
+        """Stub ``GET /pulls/30`` to return ``sha`` as the post-
+        session head. Distinct from the pre-session ``deadbeefcafe``
+        in ``_pr_item()`` so the verifier reads a real value."""
+        api.stub(
+            "GET",
+            "/repos/owner/repo/pulls/30",
+            {
+                "status": 200,
+                "body": {
+                    "number": 30,
+                    "head": {"sha": sha, "ref": "feature/x"},
+                },
+            },
+        )
+
+    def test_resolved_outcome_unchanged_when_ci_passes(
+        self, driver, cfg, fake_implementer_api
+    ):
+        """Happy path: worker claims resolved, push happened, remote
+        CI is green → outcome unchanged."""
+        self._stub_post_session_head(fake_implementer_api)
+        stub_forgejo_ci_status(
+            fake_implementer_api,
+            sha="newsha0000feed",
+            state="success",
+            contexts=(("ci/lint", "success", "https://ci.example.test/lint"),),
+        )
+        parsed = {"outcome": "resolved", "files_touched": ["x.py"]}
+        result = driver._verify_post_push_ci(
+            cfg, 30, parsed, head_sha_advanced=True
+        )
+        assert result is parsed or result == parsed
+        assert result["outcome"] == "resolved"
+        assert "_post_push_ci_verification" not in result
+
+    def test_resolved_outcome_rewritten_when_ci_fails(
+        self, driver, cfg, fake_implementer_api
+    ):
+        """Core fix: worker claims resolved but remote CI failed —
+        outcome MUST be rewritten so escalation routes this as a
+        failure, not a success."""
+        self._stub_post_session_head(fake_implementer_api)
+        stub_forgejo_ci_status(
+            fake_implementer_api,
+            sha="newsha0000feed",
+            state="failure",
+            contexts=(
+                ("ci/lint", "failure", "https://ci.example.test/lint"),
+                ("ci/unit", "failure", "https://ci.example.test/unit"),
+            ),
+        )
+        parsed = {"outcome": "resolved", "files_touched": ["x.py"]}
+        result = driver._verify_post_push_ci(
+            cfg, 30, parsed, head_sha_advanced=True
+        )
+        assert result["outcome"] == "post-push-ci-failed"
+        verification = result["_post_push_ci_verification"]
+        assert verification["ci_state"] == "failure"
+        assert verification["head_sha"] == "newsha0000feed"
+        assert verification["original_outcome"] == "resolved"
+        assert "ci/lint" in verification["failing_contexts"]
+        assert "ci/unit" in verification["failing_contexts"]
+        # The other fields from parsed_json must survive the rewrite.
+        assert result["files_touched"] == ["x.py"]
+
+    def test_resolved_outcome_rewritten_on_error_state(
+        self, driver, cfg, fake_implementer_api
+    ):
+        """``error`` is also a terminal-fail state per
+        ``_POST_PUSH_CI_TERMINAL_FAIL_STATES``."""
+        self._stub_post_session_head(fake_implementer_api)
+        stub_forgejo_ci_status(
+            fake_implementer_api,
+            sha="newsha0000feed",
+            state="error",
+            contexts=(("ci/runner", "error", "https://ci.example.test/runner"),),
+        )
+        parsed = {"outcome": "resolved"}
+        result = driver._verify_post_push_ci(
+            cfg, 30, parsed, head_sha_advanced=True
+        )
+        assert result["outcome"] == "post-push-ci-failed"
+        assert result["_post_push_ci_verification"]["ci_state"] == "error"
+
+    def test_pending_after_budget_leaves_outcome_unchanged(
+        self, driver, cfg, fake_implementer_api, monkeypatch
+    ):
+        """Budget exhausted with CI still pending → outcome unchanged.
+        We don't penalise the worker for slow CI; the next dispatcher
+        cycle will re-classify when CI lands."""
+        # Shrink the budget so the test loop terminates quickly even
+        # if sleep is somehow real.
+        monkeypatch.setattr(driver, "_POST_PUSH_CI_VERIFY_BUDGET_S", 30)
+        monkeypatch.setattr(driver, "_POST_PUSH_CI_VERIFY_POLL_INTERVAL_S", 10)
+        self._stub_post_session_head(fake_implementer_api)
+        stub_forgejo_ci_status(
+            fake_implementer_api,
+            sha="newsha0000feed",
+            state="pending",
+            contexts=(("ci/lint", "pending", "https://ci.example.test/lint"),),
+        )
+        parsed = {"outcome": "resolved"}
+        result = driver._verify_post_push_ci(
+            cfg, 30, parsed, head_sha_advanced=True
+        )
+        assert result["outcome"] == "resolved"
+        assert "_post_push_ci_verification" not in result
+
+    def test_no_verification_when_head_sha_did_not_advance(
+        self, driver, cfg, fake_implementer_api
+    ):
+        """``head_sha_advanced=False`` means the worker claimed
+        resolved but didn't push. The escalation predicate handles
+        that case (resolved+no-push → ESCALATE); the verifier
+        short-circuits because there's nothing to poll CI against."""
+        parsed = {"outcome": "resolved"}
+        result = driver._verify_post_push_ci(
+            cfg, 30, parsed, head_sha_advanced=False
+        )
+        assert result is parsed
+        # No HTTP traffic — the verifier didn't even try to fetch CI.
+        assert all(
+            "/commits/" not in c["path"]
+            for c in fake_implementer_api.calls
+        )
+
+    def test_no_verification_when_head_sha_advanced_is_none(
+        self, driver, cfg, fake_implementer_api
+    ):
+        """``head_sha_advanced=None`` means the post-session fetch
+        failed; the escalation predicate routes to RETRY_POST_FETCH.
+        The verifier short-circuits — nothing to poll against."""
+        parsed = {"outcome": "resolved"}
+        result = driver._verify_post_push_ci(
+            cfg, 30, parsed, head_sha_advanced=None
+        )
+        assert result is parsed
+
+    def test_no_verification_when_outcome_not_resolved(
+        self, driver, cfg, fake_implementer_api
+    ):
+        """Worker claimed something other than ``resolved`` — the
+        downstream escalation predicate already handles failure
+        outcomes correctly; no verification needed."""
+        for outcome in ["rebase-failed", "noop", "blocked", "transport-error"]:
+            parsed = {"outcome": outcome}
+            result = driver._verify_post_push_ci(
+                cfg, 30, parsed, head_sha_advanced=True
+            )
+            assert result is parsed, f"outcome={outcome} should be untouched"
+
+    def test_no_verification_when_parsed_json_is_none(
+        self, driver, cfg, fake_implementer_api
+    ):
+        """Worker emitted no parseable JSON → escalation predicate
+        already routes to a failure-shaped synthesis. Verifier
+        short-circuits (nothing to inspect)."""
+        result = driver._verify_post_push_ci(
+            cfg, 30, None, head_sha_advanced=True
+        )
+        assert result is None
+
+    def test_no_verification_when_dry_run(
+        self, driver, dry_cfg, fake_implementer_api
+    ):
+        """Dry-run must short-circuit so ``--dry-run`` cycles don't
+        burn 90 s polling a real Forgejo endpoint."""
+        self._stub_post_session_head(fake_implementer_api)
+        stub_forgejo_ci_status(
+            fake_implementer_api, sha="newsha0000feed", state="failure",
+        )
+        parsed = {"outcome": "resolved"}
+        result = driver._verify_post_push_ci(
+            dry_cfg, 30, parsed, head_sha_advanced=True
+        )
+        # Dry-run returns parsed_json untouched even though CI is
+        # red — no HTTP fires in dry-run mode.
+        assert result is parsed
+        assert result["outcome"] == "resolved"
+
+    def test_transport_error_during_ci_fetch_returns_outcome_unchanged(
+        self, driver, cfg, fake_implementer_api, monkeypatch
+    ):
+        """``fetch_ci_status`` raised → don't penalise the worker for
+        a Forgejo flake. Outcome stays as the worker reported it."""
+        self._stub_post_session_head(fake_implementer_api)
+
+        review_fetch = load_tool_module("_review_fetch")
+
+        def boom(_cfg, _sha):
+            raise RuntimeError("connection reset")
+
+        monkeypatch.setattr(review_fetch, "fetch_ci_status", boom)
+
+        parsed = {"outcome": "resolved"}
+        result = driver._verify_post_push_ci(
+            cfg, 30, parsed, head_sha_advanced=True
+        )
+        assert result is parsed
+        assert result["outcome"] == "resolved"
+
+    def test_post_session_head_sha_fetch_failure_returns_outcome_unchanged(
+        self, driver, cfg, fake_implementer_api
+    ):
+        """If the post-session head_sha re-fetch returns empty (404,
+        500, transport error), there's no SHA to poll against. The
+        verifier returns the outcome unchanged."""
+        # Stub /pulls/30 to return a 500 so _fetch_post_session_head_sha
+        # returns "".
+        fake_implementer_api.stub(
+            "GET",
+            "/repos/owner/repo/pulls/30",
+            {"status": 500, "body": {}},
+        )
+        parsed = {"outcome": "resolved"}
+        result = driver._verify_post_push_ci(
+            cfg, 30, parsed, head_sha_advanced=True
+        )
+        assert result is parsed
+        assert result["outcome"] == "resolved"
+
+    def test_feature_flag_off_short_circuits_before_any_fetch(
+        self, driver, cfg, fake_implementer_api, monkeypatch
+    ):
+        """``IMPLEMENTER_POST_PUSH_CI_VERIFY=0`` MUST short-circuit
+        immediately — no head_sha fetch, no CI poll, no sleep. This
+        is the escape hatch tests in test_implementer_escalation_
+        integration.py rely on so the verifier doesn't burn the
+        budget against an unstubbed FakeReviewAPI."""
+        monkeypatch.setenv("IMPLEMENTER_POST_PUSH_CI_VERIFY", "0")
+        # Stub both endpoints so we can prove they were NOT hit.
+        self._stub_post_session_head(fake_implementer_api)
+        stub_forgejo_ci_status(
+            fake_implementer_api, sha="newsha0000feed", state="failure",
+        )
+        parsed = {"outcome": "resolved"}
+        result = driver._verify_post_push_ci(
+            cfg, 30, parsed, head_sha_advanced=True
+        )
+        assert result is parsed
+        # No traffic to either endpoint — the flag check fired first.
+        assert all(
+            "/commits/" not in c["path"]
+            and "/pulls/30" not in c["path"]
+            for c in fake_implementer_api.calls
+        )
+
+    def test_failing_contexts_extraction_filters_to_failure_states(
+        self, driver, cfg, fake_implementer_api
+    ):
+        """Mixed statuses: only the failure/error rows should appear
+        in ``failing_contexts``. A passing check on the same SHA
+        must NOT be reported as failing."""
+        self._stub_post_session_head(fake_implementer_api)
+        stub_forgejo_ci_status(
+            fake_implementer_api,
+            sha="newsha0000feed",
+            state="failure",
+            contexts=(
+                ("ci/lint", "failure", "https://ci.example.test/lint"),
+                ("ci/format", "success", "https://ci.example.test/format"),
+                ("ci/types", "error", "https://ci.example.test/types"),
+                ("ci/coverage", "pending", "https://ci.example.test/cov"),
+            ),
+        )
+        parsed = {"outcome": "resolved"}
+        result = driver._verify_post_push_ci(
+            cfg, 30, parsed, head_sha_advanced=True
+        )
+        assert result["outcome"] == "post-push-ci-failed"
+        failing = result["_post_push_ci_verification"]["failing_contexts"]
+        assert set(failing) == {"ci/lint", "ci/types"}
+
+
+class TestPostPushCIVerifyCycleBudget:
+    """Per-cycle polling-time budget. Without this, a dispatcher cycle
+    that processes N PRs each landing in ``outcome=resolved`` after
+    a push could spend ``N * IMPLEMENTER_POST_PUSH_CI_VERIFY_S``
+    seconds polling — at the default 90 s per call and 5 PRs/cycle,
+    that's 7.5 min of polling eating the dispatcher cycle.
+
+    The budget auto-resets after
+    ``IMPLEMENTER_POST_PUSH_CI_CYCLE_RESET_AFTER_S`` seconds of idle
+    so the dispatcher's outer loop doesn't need to call a reset
+    hook explicitly — a new cycle naturally exceeds that gap."""
+
+    @pytest.fixture(autouse=True)
+    def _no_sleep_and_clean_budget(self, monkeypatch, driver):
+        """Same as the parent class fixture, plus explicit budget
+        reset between tests so cross-test state doesn't leak."""
+        import time as _time
+        monkeypatch.setattr(_time, "sleep", lambda _s: None)
+        monkeypatch.setenv("IMPLEMENTER_POST_PUSH_CI_VERIFY", "1")
+        driver.reset_post_push_ci_cycle_budget()
+        yield
+        driver.reset_post_push_ci_cycle_budget()
+
+    def _stub_post_session_head(
+        self, api, sha: str = "newsha0000feed"
+    ) -> None:
+        api.stub(
+            "GET",
+            "/repos/owner/repo/pulls/30",
+            {
+                "status": 200,
+                "body": {
+                    "number": 30,
+                    "head": {"sha": sha, "ref": "feature/x"},
+                },
+            },
+        )
+
+    def test_cycle_budget_exhausted_skips_verification(
+        self, driver, cfg, fake_implementer_api, monkeypatch,
+    ):
+        """When the cumulative polling time has exceeded the cycle
+        budget, _verify_post_push_ci must short-circuit (return
+        parsed_json unchanged) instead of polling further. The
+        worker's claim survives by default — next dispatcher cycle
+        gets a fresh budget and re-verifies."""
+        # Set both budgets low so we hit the cap fast and
+        # deterministically.
+        monkeypatch.setattr(driver, "_POST_PUSH_CI_CYCLE_BUDGET_S", 30.0)
+        monkeypatch.setattr(driver, "_POST_PUSH_CI_VERIFY_BUDGET_S", 30)
+        monkeypatch.setattr(driver, "_POST_PUSH_CI_VERIFY_POLL_INTERVAL_S", 10)
+        # Pretend a prior call already consumed the full budget.
+        driver._post_push_cycle_seconds_spent = 30.0
+        driver._post_push_cycle_last_call_at = __import__("time").monotonic()
+
+        parsed = {"outcome": "resolved"}
+        # No need to stub fetch_ci_status — verifier short-circuits
+        # before the polling loop runs.
+        result = driver._verify_post_push_ci(
+            cfg, 30, parsed, head_sha_advanced=True
+        )
+        assert result is parsed or result == parsed
+        # No Forgejo call should have been made.
+        assert not any(
+            call.get("path", "").startswith(
+                "/repos/owner/repo/statuses/"
+            )
+            for call in fake_implementer_api.calls
+        )
+
+    def test_cycle_budget_resets_after_idle(self, driver, cfg, fake_implementer_api, monkeypatch):
+        """After ``_POST_PUSH_CI_CYCLE_RESET_AFTER_S`` of inactivity,
+        the next call gets a fresh budget — this is how a true new
+        dispatcher cycle naturally resets without an explicit hook."""
+        monkeypatch.setattr(driver, "_POST_PUSH_CI_CYCLE_BUDGET_S", 30.0)
+        monkeypatch.setattr(driver, "_POST_PUSH_CI_CYCLE_RESET_AFTER_S", 1.0)
+        monkeypatch.setattr(driver, "_POST_PUSH_CI_VERIFY_BUDGET_S", 30)
+        monkeypatch.setattr(driver, "_POST_PUSH_CI_VERIFY_POLL_INTERVAL_S", 10)
+
+        # Simulate "previous cycle exhausted the budget 2 seconds ago".
+        import time as _time
+        driver._post_push_cycle_seconds_spent = 30.0
+        driver._post_push_cycle_last_call_at = _time.monotonic() - 2.0
+
+        self._stub_post_session_head(fake_implementer_api)
+        stub_forgejo_ci_status(
+            fake_implementer_api,
+            sha="newsha0000feed",
+            state="success",
+            contexts=(("ci/lint", "success", "https://ci.example.test/lint"),),
+        )
+        parsed = {"outcome": "resolved"}
+        result = driver._verify_post_push_ci(
+            cfg, 30, parsed, head_sha_advanced=True
+        )
+        # Verifier actually ran (got the success rewrite path),
+        # AND the budget reset — confirms the auto-reset works.
+        assert result["outcome"] == "resolved"
+        assert driver._post_push_cycle_seconds_spent < 30.0, (
+            "budget should have reset to 0 + a single call's elapsed; "
+            f"got {driver._post_push_cycle_seconds_spent}"
+        )
+
+    def test_reset_post_push_ci_cycle_budget_zeroes_state(self, driver):
+        """Public reset hook for tests / dispatcher integration:
+        zeroes both the accumulated time and the last-call timestamp."""
+        driver._post_push_cycle_seconds_spent = 42.5
+        driver._post_push_cycle_last_call_at = 12345.6
+        driver.reset_post_push_ci_cycle_budget()
+        assert driver._post_push_cycle_seconds_spent == 0.0
+        assert driver._post_push_cycle_last_call_at == 0.0
+
+    def test_per_call_budget_capped_by_remaining_cycle_budget(
+        self, driver, cfg, fake_implementer_api, monkeypatch,
+    ):
+        """When the per-cycle budget is nearly exhausted, the next
+        call's per-call budget must be capped at what's left of the
+        cycle budget — so a near-empty cycle budget doesn't get
+        blown through by one fat call."""
+        monkeypatch.setattr(driver, "_POST_PUSH_CI_CYCLE_BUDGET_S", 100.0)
+        monkeypatch.setattr(driver, "_POST_PUSH_CI_VERIFY_BUDGET_S", 90)
+        monkeypatch.setattr(driver, "_POST_PUSH_CI_VERIFY_POLL_INTERVAL_S", 10)
+        # 95 of 100s already spent in this cycle → only 5s left.
+        driver._post_push_cycle_seconds_spent = 95.0
+        driver._post_push_cycle_last_call_at = __import__("time").monotonic()
+
+        self._stub_post_session_head(fake_implementer_api)
+        # CI is "pending" forever — verifier will loop until budget
+        # expires.
+        stub_forgejo_ci_status(
+            fake_implementer_api,
+            sha="newsha0000feed",
+            state="pending",
+            contexts=(),
+        )
+        parsed = {"outcome": "resolved"}
+        driver._verify_post_push_ci(
+            cfg, 30, parsed, head_sha_advanced=True
+        )
+        # Per-call budget should have been ~5s (cycle remaining),
+        # capped by the per-call interval logic. So at most one
+        # poll happens. Confirm cumulative spend went up but did
+        # NOT blow past the cycle budget.
+        assert driver._post_push_cycle_seconds_spent <= 100.5, (
+            f"per-call cap failed; cycle spent={driver._post_push_cycle_seconds_spent}"
+        )

tools/dispatch_implementer.py

@@ -2646,6 +2646,263 @@ def _fetch_pr_state(cfg: Any, pr_number: int) -> str:
     )
 
 
+# R3.7 post-push CI verification (2026-05-17). When the worker
+# claims ``outcome=resolved`` AND head_sha advanced (a real push), the
+# dispatcher polls Forgejo CI on the new head_sha for up to this many
+# seconds. If CI lands in a terminal failure state within the budget,
+# the dispatcher REWRITES the worker's outcome from ``resolved`` to
+# ``post-push-ci-failed`` so the downstream escalation predicate
+# treats the cycle as a failure (not a success). This closes the
+# local-vs-remote-CI divergence the 2026-05-17 run-7 observation
+# exposed — the worker's ``ci_run_local_gate --fast`` doesn't catch
+# what remote CI does, so the worker can push a commit + claim
+# ``resolved`` even when remote CI will reject it. With this gate
+# in place, the dispatcher is the source of truth for "did this PR
+# actually pass CI."
+#
+# Tuning:
+#   - Default 90 s: long enough to catch the fast-failing checks
+#     (lint, format, push-validation typically fail in 30-60 s) but
+#     short enough not to dominate the dispatcher cycle.
+#   - 10 s poll interval: 9 polls per budget window. Forgejo's CI
+#     status endpoint is fast (< 1 s typical).
+#   - On budget exhaustion the verifier RETURNS the outcome unchanged
+#     (does NOT rewrite to a failure) — we don't penalise the worker
+#     for slow CI. The next cycle will re-classify if CI eventually
+#     fails.
+_POST_PUSH_CI_VERIFY_BUDGET_S = int(
+    os.environ.get("IMPLEMENTER_POST_PUSH_CI_VERIFY_S", "90")
+)
+_POST_PUSH_CI_VERIFY_POLL_INTERVAL_S = int(
+    os.environ.get("IMPLEMENTER_POST_PUSH_CI_VERIFY_POLL_S", "10")
+)
+_POST_PUSH_CI_TERMINAL_FAIL_STATES = frozenset({"failure", "error"})
+_POST_PUSH_CI_TERMINAL_PASS_STATES = frozenset({"success"})
+
+# Per-cycle polling-time budget (sliding window). Without this, a
+# dispatcher cycle that processes N PRs each landing in
+# ``outcome=resolved`` after a push could spend
+# ``N × _POST_PUSH_CI_VERIFY_BUDGET_S`` seconds polling — at the
+# default 90 s per call and 5 PRs/cycle, that's 7.5 min of polling
+# eating the cycle budget. The cycle budget caps the cumulative
+# polling time across all ``_verify_post_push_ci`` calls in a single
+# dispatcher cycle.
+#
+# Window resets automatically when more than
+# ``_POST_PUSH_CI_CYCLE_RESET_AFTER_S`` seconds pass between calls —
+# that's the dispatcher's cycle boundary in practice, since cycles
+# are typically several seconds apart. Avoids requiring the
+# dispatcher's outer loop to remember to call a reset hook.
+_POST_PUSH_CI_CYCLE_BUDGET_S = float(
+    os.environ.get("IMPLEMENTER_POST_PUSH_CI_CYCLE_BUDGET_S", "300")
+)
+_POST_PUSH_CI_CYCLE_RESET_AFTER_S = float(
+    os.environ.get("IMPLEMENTER_POST_PUSH_CI_CYCLE_RESET_AFTER_S", "120")
+)
+_post_push_cycle_seconds_spent: float = 0.0
+_post_push_cycle_last_call_at: float = 0.0
+
+
+def reset_post_push_ci_cycle_budget() -> None:
+    """Reset the per-cycle polling-time budget. Called automatically
+    when more than ``_POST_PUSH_CI_CYCLE_RESET_AFTER_S`` seconds
+    have passed since the last call; exposed for tests that want
+    to reset between assertions and for any future dispatcher hook
+    that wants to reset explicitly at cycle boundaries."""
+    global _post_push_cycle_seconds_spent, _post_push_cycle_last_call_at
+    _post_push_cycle_seconds_spent = 0.0
+    _post_push_cycle_last_call_at = 0.0
+
+
+def _maybe_reset_cycle_budget() -> None:
+    """Auto-reset the budget if the dispatcher has been idle long
+    enough that we're clearly in a new cycle."""
+    global _post_push_cycle_seconds_spent, _post_push_cycle_last_call_at
+    now = time.monotonic()
+    if _post_push_cycle_last_call_at == 0.0:
+        _post_push_cycle_last_call_at = now
+        return
+    if (now - _post_push_cycle_last_call_at) > _POST_PUSH_CI_CYCLE_RESET_AFTER_S:
+        _post_push_cycle_seconds_spent = 0.0
+    _post_push_cycle_last_call_at = now
+
+
+def _post_push_ci_verify_enabled() -> bool:
+    """Feature flag for the R3.7 post-push CI verifier. Default ``"1"``
+    (enabled in production). Tests that exercise the escalation path
+    without stubbing ``fetch_ci_status`` set ``"0"`` via an autouse
+    fixture so the verifier short-circuits instead of polling the
+    FakeReviewAPI default (which returns ``[]`` and would burn the
+    full 90 s budget on a real sleep)."""
+    return str(
+        os.environ.get("IMPLEMENTER_POST_PUSH_CI_VERIFY", "1")
+    ).strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _verify_post_push_ci(
+    cfg: Any,
+    pr_number: int,
+    parsed_json: dict[str, Any] | None,
+    head_sha_advanced: bool | None,
+) -> dict[str, Any] | None:
+    """Verify the worker's ``outcome=resolved`` claim against remote
+    CI. Returns the (possibly-rewritten) ``parsed_json``.
+
+    The verifier ONLY runs when:
+      - ``parsed_json`` is a dict with ``outcome=resolved``, AND
+      - ``head_sha_advanced is True`` (a real push happened — no
+        verification needed if the worker didn't push).
+
+    Polls ``fetch_ci_status(cfg, new_head_sha)`` every
+    :data:`_POST_PUSH_CI_VERIFY_POLL_INTERVAL_S` seconds for up to
+    :data:`_POST_PUSH_CI_VERIFY_BUDGET_S` seconds total. Verdict:
+
+      - CI state ∈ ``{success}``: outcome unchanged (worker's
+        ``resolved`` claim verified).
+      - CI state ∈ ``{failure, error}``: outcome REWRITTEN to
+        ``post-push-ci-failed``; the original outcome + the list of
+        failing contexts are stashed under ``_post_push_ci_verification``
+        for telemetry / debugging. The downstream
+        ``_implementer_escalation.decide`` will see a non-success
+        outcome and route to ESCALATE / EXHAUSTED as appropriate.
+      - CI still in ``{pending}`` after the budget expires: outcome
+        unchanged. We don't penalise the worker for slow CI — the
+        next dispatcher cycle will re-classify the PR when CI lands.
+      - Fetch fails (transport error, no status returned): outcome
+        unchanged. Treating "I couldn't check" as "the worker lied"
+        would create false positives on Forgejo flakes.
+
+    Dry-run short-circuits to no-op so ``--dry-run`` cycles don't
+    burn 90 s polling a real Forgejo endpoint.
+    """
+    if not _post_push_ci_verify_enabled():
+        return parsed_json
+    if getattr(cfg, "dry_run", False):
+        return parsed_json
+    if not isinstance(parsed_json, dict):
+        return parsed_json
+    if parsed_json.get("outcome") != "resolved":
+        return parsed_json
+    if head_sha_advanced is not True:
+        # head_sha_advanced is False → worker claims resolved but
+        # didn't push → escalation handler already routes this case
+        # (head_sha_advanced=False + outcome=resolved → ESCALATE per
+        # _implementer_escalation.decide).
+        # head_sha_advanced is None → fetch failed → already routes
+        # to RETRY_POST_FETCH.
+        # Both are handled downstream; nothing to verify here.
+        return parsed_json
+
+    # Per-cycle budget gate. Auto-resets after
+    # ``_POST_PUSH_CI_CYCLE_RESET_AFTER_S`` seconds of idle (which
+    # is well under any dispatcher cycle interval), so a true new
+    # cycle gets a fresh budget without the dispatcher's outer loop
+    # having to call a reset hook.
+    global _post_push_cycle_seconds_spent
+    _maybe_reset_cycle_budget()
+    if _post_push_cycle_seconds_spent >= _POST_PUSH_CI_CYCLE_BUDGET_S:
+        _logger.warning(
+            "post-push CI verify: PR #%s skipped — per-cycle polling "
+            "budget exhausted (%.1fs/%.1fs spent across earlier calls). "
+            "Leaving outcome unchanged; next cycle will re-verify.",
+            pr_number,
+            _post_push_cycle_seconds_spent,
+            _POST_PUSH_CI_CYCLE_BUDGET_S,
+        )
+        return parsed_json
+
+    # Re-fetch the post-push head_sha (we know it advanced; fetch
+    # fresh to poll against the right SHA).
+    new_head_sha = _fetch_post_session_head_sha(cfg, pr_number)
+    if not new_head_sha:
+        _logger.info(
+            "post-push CI verify: PR #%s head_sha_advanced=True but "
+            "post-session head_sha fetch failed; skipping verification "
+            "(outcome unchanged)",
+            pr_number,
+        )
+        return parsed_json
+
+    # Per-call budget is the min of the configured per-call budget
+    # and what's left in the cycle budget — so a near-exhausted
+    # cycle budget doesn't get blown through by one big call.
+    per_call_budget_s = min(
+        _POST_PUSH_CI_VERIFY_BUDGET_S,
+        max(0, int(_POST_PUSH_CI_CYCLE_BUDGET_S - _post_push_cycle_seconds_spent)),
+    )
+    elapsed = 0
+    final_state: str | None = None
+    failing_contexts: list[str] = []
+    while elapsed <= per_call_budget_s:
+        try:
+            ci = _review_fetch.fetch_ci_status(cfg, new_head_sha)
+        except Exception as exc:  # noqa: BLE001 — best-effort verify
+            _logger.warning(
+                "post-push CI verify: PR #%s fetch_ci_status raised %s; "
+                "leaving outcome unchanged (won't penalise worker for "
+                "Forgejo flake)",
+                pr_number, exc,
+            )
+            return parsed_json
+        if isinstance(ci, dict):
+            state = str(ci.get("state") or "").lower()
+            if state in _POST_PUSH_CI_TERMINAL_PASS_STATES:
+                _logger.info(
+                    "post-push CI verify: PR #%s @ %s passed after %ds — "
+                    "worker's resolved claim verified",
+                    pr_number, new_head_sha[:12], elapsed,
+                )
+                return parsed_json
+            if state in _POST_PUSH_CI_TERMINAL_FAIL_STATES:
+                final_state = state
+                failing_contexts = sorted({
+                    str(s.get("context") or "")
+                    for s in (ci.get("statuses") or [])
+                    if isinstance(s, dict)
+                    and str(s.get("state") or "").lower()
+                    in _POST_PUSH_CI_TERMINAL_FAIL_STATES
+                })
+                break
+        if elapsed + _POST_PUSH_CI_VERIFY_POLL_INTERVAL_S > per_call_budget_s:
+            break
+        time.sleep(_POST_PUSH_CI_VERIFY_POLL_INTERVAL_S)
+        elapsed += _POST_PUSH_CI_VERIFY_POLL_INTERVAL_S
+
+    # Charge actual polling time to the per-cycle budget so the
+    # next call in this dispatcher cycle sees the cumulative cost.
+    _post_push_cycle_seconds_spent += elapsed
+
+    if final_state in _POST_PUSH_CI_TERMINAL_FAIL_STATES:
+        _logger.warning(
+            "post-push CI verify: PR #%s @ %s — worker claimed "
+            "outcome=resolved but remote CI is %s (failing: %s). "
+            "Rewriting outcome to 'post-push-ci-failed' so escalation "
+            "treats this as a failure (closes the local-vs-remote-CI "
+            "divergence gap).",
+            pr_number, new_head_sha[:12], final_state, failing_contexts,
+        )
+        return {
+            **parsed_json,
+            "outcome": "post-push-ci-failed",
+            "_post_push_ci_verification": {
+                "head_sha": new_head_sha,
+                "ci_state": final_state,
+                "failing_contexts": failing_contexts,
+                "elapsed_seconds": elapsed,
+                "original_outcome": parsed_json.get("outcome"),
+            },
+        }
+
+    # Budget exhausted with CI still pending → outcome unchanged.
+    _logger.info(
+        "post-push CI verify: PR #%s @ %s CI still pending after %ds; "
+        "leaving worker's resolved claim intact (next cycle will "
+        "re-classify when CI lands)",
+        pr_number, new_head_sha[:12], elapsed,
+    )
+    return parsed_json
+
+
 def _compute_head_sha_tristate(
     cfg: Any, pr_number: int, pre_sha: str
 ) -> bool | None:
@@ -3109,6 +3366,14 @@ def _post_session_action_with_escalation(
         )
     head_sha_advanced = _compute_head_sha_tristate(cfg, pr_number, pre_sha)
     pr_state = _fetch_pr_state(cfg, pr_number)
+    # R3.7 post-push CI verification — if the worker claimed
+    # ``resolved`` AND head_sha advanced, poll remote CI and reject
+    # the claim if CI failed. parsed_json may be rewritten to
+    # ``outcome="post-push-ci-failed"`` so the escalation decision
+    # below sees the truth instead of the worker's optimistic claim.
+    parsed_json = _verify_post_push_ci(
+        cfg, pr_number, parsed_json, head_sha_advanced,
+    )
     action = _implementer_escalation.decide(
         parsed_json=parsed_json,
         terminal_state=terminal_state,
@@ -3148,6 +3413,16 @@ def _post_session_action_with_escalation(
             # tier advance. Re-decide.
             head_sha_advanced = _compute_head_sha_tristate(cfg, pr_number, pre_sha)
             pr_state = _fetch_pr_state(cfg, pr_number)
+            # R3.7 post-push CI verification — the prior decide()
+            # returned RETRY_POST_FETCH because head_sha_advanced was
+            # None (post-session fetch failed). If the re-fetch now
+            # succeeds with head_sha_advanced=True, the verifier
+            # finally has a SHA to poll CI against. Same idempotency
+            # contract as the other callsites: returns unchanged
+            # when the conditions for verification aren't met.
+            last_parsed = _verify_post_push_ci(
+                cfg, pr_number, last_parsed, head_sha_advanced,
+            )
             action = _implementer_escalation.decide(
                 parsed_json=last_parsed,
                 terminal_state=last_terminal_state,
@@ -3235,6 +3510,12 @@ def _post_session_action_with_escalation(
 
         head_sha_advanced = _compute_head_sha_tristate(cfg, pr_number, pre_sha)
         pr_state = _fetch_pr_state(cfg, pr_number)
+        # R3.7 post-push CI verification — see comment in initial-
+        # attempt site above. Each tier's worker session may push,
+        # so each tier's outcome needs the same verification.
+        last_parsed = _verify_post_push_ci(
+            cfg, pr_number, last_parsed, head_sha_advanced,
+        )
         action = _implementer_escalation.decide(
             parsed_json=last_parsed,
             terminal_state=last_terminal_state,