fix(api_server): run the cron-fire token verifier off the event loop

_handle_cron_fire verified the NAS-minted fire JWT by calling the
fire-verifier inline on the event loop. That verifier resolves the NAS
signing key from a JWKS URL — a synchronous HTTP GET on a cache miss (a
cold PyJWKClient, or a rotated kid the cached client doesn't know) — so a
slow or rate-limited portal stalls the whole event loop and starves every
other adapter sharing it. #64641 already documented this exact symptom
(relay 504s on high-job-count instances) and cut the fetch frequency by
caching the client per URL, but the residual cache-miss fetch still ran
inline on the loop.

Dispatch the verifier the same way the platform HTTP event verifier was
hardened: await a coroutine verifier directly, run a sync one via
asyncio.to_thread so its blocking I/O stays off the loop, and fail closed
(reject with 401, never admit the fire) if the verifier raises — this is
the only inbound that can trigger remote job execution. The verifier's
JWK-client cache is already thread-safe (threading.Lock), so moving the
call to a worker thread is safe.

Adds regression tests: a sync verifier runs on a worker thread rather
than the loop thread, a crashing verifier yields 401 with no fire, and a
coroutine verifier is awaited.

gateway/platforms/api_server.py

@@ -5615,12 +5615,29 @@ class APIServerAdapter(BasePlatformAdapter):
         token = auth[7:].strip() if auth.startswith("Bearer ") else ""
 
         cfg = load_config()
-        claims = get_fire_verifier()(
+        verifier = get_fire_verifier()
+        verify_kwargs = dict(
             token=token,
             expected_audience=cfg_get(cfg, "cron", "chronos", "expected_audience", default=""),
             jwks_or_key=cfg_get(cfg, "cron", "chronos", "nas_jwks_url", default="") or None,
             issuer=cfg_get(cfg, "cron", "chronos", "portal_url", default="") or None,
         )
+        try:
+            if asyncio.iscoroutinefunction(verifier):
+                claims = await verifier(**verify_kwargs)
+            else:
+                # The verifier resolves the NAS signing key from a JWKS URL,
+                # which is a synchronous HTTP GET on a cache miss (cold client
+                # or a rotated kid) — keep that blocking I/O off the event loop
+                # so a slow or rate-limited portal can't stall every other
+                # adapter sharing this loop. Same hardening the platform HTTP
+                # event verifier already got.
+                claims = await asyncio.to_thread(verifier, **verify_kwargs)
+        except Exception:
+            # Fail closed: a crashing verifier must never admit a fire — this
+            # is the only inbound that can trigger remote job execution.
+            logger.exception("cron fire: verifier crashed; rejecting token")
+            claims = None
         if claims is None:
             logger.warning(
                 "cron fire: rejected invalid token: %s",

tests/gateway/test_cron_fire_webhook.py

@@ -123,3 +123,118 @@ async def test_missing_job_id_400(adapter, monkeypatch):
     assert spy.fired == []
 
 
+@pytest.mark.asyncio
+async def test_fire_does_not_require_api_server_key(adapter, monkeypatch):
+    """The fire endpoint must NOT gate on API_SERVER_KEY — auth is the NAS-JWT.
+    A request with NO API key header but a valid fire token still succeeds."""
+    spy = _SpyProvider()
+    monkeypatch.setattr("cron.scheduler_provider.resolve_cron_scheduler", lambda: spy)
+    monkeypatch.setattr(
+        "plugins.cron_providers.chronos.verify.get_fire_verifier",
+        lambda: (lambda **kw: {"purpose": "cron_fire"}),
+    )
+
+    app = _create_app(adapter)
+    async with TestClient(TestServer(app)) as cli:
+        # Bearer is the FIRE token, not the API_SERVER_KEY "sk-secret".
+        resp = await cli.post("/api/cron/fire",
+                              headers={"Authorization": "Bearer nas-jwt"},
+                              json={"job_id": "j9"})
+        assert resp.status == 202
+    for _ in range(50):
+        if spy.fired:
+            break
+        await asyncio.sleep(0.01)
+    assert spy.fired == ["j9"]
+
+
+@pytest.mark.asyncio
+async def test_sync_verifier_runs_off_the_event_loop(adapter, monkeypatch):
+    """The verifier resolves the signing key from a JWKS URL — a synchronous
+    HTTP GET on a cache miss. It must run via asyncio.to_thread, NOT inline on
+    the event loop, or a slow/rate-limited portal stalls every other adapter
+    sharing the loop. Proof: the sync verifier executes on a worker thread, not
+    the loop thread.
+    """
+    loop_thread_id = threading.get_ident()
+    seen = {}
+
+    def blocking_verifier(**kw):
+        seen["thread_id"] = threading.get_ident()
+        return {"purpose": "cron_fire"}
+
+    spy = _SpyProvider()
+    monkeypatch.setattr("cron.scheduler_provider.resolve_cron_scheduler", lambda: spy)
+    monkeypatch.setattr(
+        "plugins.cron_providers.chronos.verify.get_fire_verifier",
+        lambda: blocking_verifier,
+    )
+
+    app = _create_app(adapter)
+    async with TestClient(TestServer(app)) as cli:
+        resp = await cli.post("/api/cron/fire",
+                              headers={"Authorization": "Bearer good"},
+                              json={"job_id": "off-loop"})
+        assert resp.status == 202
+
+    # If the verifier had run inline on the loop, its thread id would equal the
+    # loop thread's; to_thread puts it on a distinct worker thread.
+    assert seen["thread_id"] != loop_thread_id
+
+
+@pytest.mark.asyncio
+async def test_crashing_verifier_fails_closed_401(adapter, monkeypatch):
+    """A verifier that raises must be treated as a rejection (401), never admit
+    the fire, and never surface as a 500 — this is the only inbound that can
+    trigger remote job execution, so it fails closed.
+    """
+    spy = _SpyProvider()
+    monkeypatch.setattr("cron.scheduler_provider.resolve_cron_scheduler", lambda: spy)
+
+    def exploding_verifier(**kw):
+        raise RuntimeError("JWKS endpoint unreachable")
+
+    monkeypatch.setattr(
+        "plugins.cron_providers.chronos.verify.get_fire_verifier",
+        lambda: exploding_verifier,
+    )
+
+    app = _create_app(adapter)
+    async with TestClient(TestServer(app)) as cli:
+        resp = await cli.post("/api/cron/fire",
+                              headers={"Authorization": "Bearer boom"},
+                              json={"job_id": "abc123"})
+        assert resp.status == 401
+
+    await asyncio.sleep(0.05)
+    assert spy.fired == []
+
+
+@pytest.mark.asyncio
+async def test_async_verifier_is_awaited(adapter, monkeypatch):
+    """A coroutine verifier (a future async escape-hatch) is awaited directly
+    rather than dispatched to a thread — a valid async verify still fires.
+    """
+    spy = _SpyProvider()
+    monkeypatch.setattr("cron.scheduler_provider.resolve_cron_scheduler", lambda: spy)
+
+    async def async_verifier(**kw):
+        return {"purpose": "cron_fire", "aud": "agent:x"}
+
+    monkeypatch.setattr(
+        "plugins.cron_providers.chronos.verify.get_fire_verifier",
+        lambda: async_verifier,
+    )
+
+    app = _create_app(adapter)
+    async with TestClient(TestServer(app)) as cli:
+        resp = await cli.post("/api/cron/fire",
+                              headers={"Authorization": "Bearer good"},
+                              json={"job_id": "async-ok"})
+        assert resp.status == 202
+
+    for _ in range(50):
+        if spy.fired:
+            break
+        await asyncio.sleep(0.01)
+    assert spy.fired == ["async-ok"]