# NousResearch/hermes-agent · Gateway shutdown now drains in-flight API runs

## Open now

- Nothing open on GitHub.

These members address a bug where the gateway's shutdown process did not properly drain or interrupt in-flight API server runs, leaving active sessions without protection during shutdown. The work ensures both API and cron workloads are safely handled when the gateway shuts down.

## TL;DR

- 4 member(s): 0 open issue(s), 1 closed issue(s), 0 open PR(s), 2 merged PR(s), 1 PR(s) closed without merging.
- A fix was merged to main in PR #80917. Verify it addresses your case.
- **Recommended:** read PR #80917 (and any competing PRs below) before filing. Confirm the merged change covers your symptom.

## Maintainer resolution verdict

- Internal-tracker verdict over this cluster's issue members: `resolved` · deciding PR(s): `pr-63891`. This is distinct from the fix status above. Weigh it yourself.

## How this cluster was grouped

- Link evidence (`STRUCTURAL_SPANNING`): every member is connected by structural evidence: shared changed files, or a closing/superseded reference.
- Pairwise link density `0.50`: the share of member pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. `1.00` means every pair is linked.

## Closed / duplicate issues

- #63529: \[Bug\]: shutdown drain is blind to api_server in-flight runs — desk/API sessions get zero drain protection

## Prior & competing PRs

- **#63891: fix(gateway): drain API and cron work safely**. merged to main · fix tier `verified`
  - maintainer direction: category `merged`
  - files: `cron/scheduler.py`, `cron/scheduler_provider.py`, `gateway/platforms/api_server.py`, `gateway/run.py`, `tests/cron/test_scheduler.py`, `tests/cron/test_scheduler_provider.py`, `tests/gateway/restart_test_helpers.py`, `tests/gateway/test_api_server_active_work_drain.py`, `tests/gateway/test_api_server_jobs.py`, `tests/gateway/test_cron_fire_webhook.py`, `tests/gateway/test_external_drain_control.py`, `tests/gateway/test_shutdown_cache_cleanup.py`
  - fix hunks:
  - `cron/scheduler.py`

    ```diff
    @@ -3557,7 +3557,14 @@ def _notify_provider_jobs_changed() -> None:
             logger.debug("on_jobs_changed notify failed: %s", e)
     
     
    -def tick(verbose: bool = True, adapters=None, loop=None, sync: bool = True) -> int:
    +def tick(
    +    verbose: bool = True,
    +    adapters=None,
    +    loop=None,
    +    sync: bool = True,
    +    *,
    +    can_dispatch=None,
    +):
         """
         Check and run all due jobs.
         
    ```

    ```diff
    @@ -3568,7 +3575,9 @@ def tick(verbose: bool = True, adapters=None, loop=None, sync: bool = True) -> i
             verbose: Whether to print status messages
             adapters: Optional dict mapping Platform → live adapter (from gateway)
             loop: Optional asyncio event loop (from gateway) for live adapter sends
    -    
    +        can_dispatch: Optional synchronous gate; false leaves due jobs untouched
    +            for the next allowed tick
    +
         Returns:
             Number of jobs executed (0 if another tick is already running)
         """
    ```

    ```diff
    @@ -3590,6 +3599,10 @@ def tick(verbose: bool = True, adapters=None, loop=None, sync: bool = True) -> i
             return 0
     
         try:
    +        if can_dispatch is not None and not can_dispatch():
    +            logger.debug("Cron dispatch paused while gateway drains existing work")
    +            return 0
    +
             due_jobs = get_due_jobs()
     
             if verbose and not due_jobs:
    ```
  - `cron/scheduler_provider.py`

    ```diff
    @@ -156,14 +156,16 @@ class InProcessCronScheduler(CronScheduler):
     
         ``start()`` blocks in the tick loop until ``stop_event`` is set, identical
         to the pre-refactor ``_start_cron_ticker`` core loop. The caller runs it in
    -    a daemon thread.
    +    a daemon thread. ``can_dispatch`` is an optional synchronous gate supplied
    +    by GatewayRunner during external drain; skipped ticks leave due jobs intact
    +    for the next allowed tick.
         """
     
         @property
         def name(self) -> str:
             return "builtin"
     
    -    def start(self, stop_event, *, adapters=None, loop=None, interval=60):
    +    def start(self, stop_event, *, adapters=None, loop=None, interval=60, can_dispatch=None):
             import logging
             from cron.scheduler import tick as cron_tick
             from cron.jobs import record_ticker_heartbeat
    ```

    ```diff
    @@ -176,7 +178,16 @@ def start(self, stop_event, *, adapters=None, loop=None, interval=60):
             while not stop_event.is_set():
                 ok = False
                 try:
    -                cron_tick(verbose=False, adapters=adapters, loop=loop, sync=False)
    +                if can_dispatch is not None and not can_dispatch():
    +                    logger.debug("Cron dispatch paused while gateway drains existing work")
    +                else:
    +                    cron_tick(
    +                        verbose=False,
    +                        adapters=adapters,
    +                        loop=loop,
    +                        sync=False,
    +                        can_dispatch=can_dispatch,
    +                    )
                     ok = True
                 except BaseException as e:
                     # Catch BaseException (not just Exception) so a SystemExit from
    ```
  - `gateway/platforms/api_server.py`

    ```diff
    @@ -35,6 +35,9 @@
     import hashlib
     import hmac
     import json
    +from contextlib import contextmanager
    +from contextvars import ContextVar
    +from functools import wraps
     import logging
     import os
     import socket as _socket
    ```

    ```diff
    @@ -664,6 +667,66 @@ def _openai_error(message: str, err_type: str = "invalid_request_error", param:
         }
     
     
    +_api_agent_request_reservation: ContextVar[Optional[dict[str, bool]]] = ContextVar(
    +    "api_agent_request_reservation", default=None
    +)
    +
    +
    +def _admit_api_agent_request(handler):
    +    """Reserve an authenticated API turn before its handler first awaits.
    +
    +    Gateway shutdown and aiohttp requests share an event loop. Keeping the
    +    drain check and reservation in one non-awaiting block prevents a request
    +    admitted immediately before shutdown from becoming invisible while it is
    +    still parsing its body or resolving session state. The mutable reservation
    +    is intentionally shared with child tasks so agent/task bookkeeping releases
    +    this one slot exactly once.
    +    """
    +    @wraps(handler)
    +    async def _wrapped(self, request, *args, **kwargs):
    +        auth_err = self._check_auth(request)
    +        if auth_err:
    +            return auth_err
    +        draining = self._draining_response()
    +        if draining is not None:
    +            return draining
    +        reservation = {"active": True}
    +        token = [redacted:secret-assignment]
    +        self._pending_agent_requests += 1
    +        try:
    +            return await handler(self, request, *args, **kwargs)
    +        finally:
    +            if reservation["active"]:
    +                reservation["active"] = False
    +                self._pending_agent_requests = max(0, self._pending_agent_requests - 1)
    +            _api_agent_request_reservation.reset(token)
    +
    +    return _wrapped
    +
    +
    +def _release_pending_api_work(adapter, reservation: dict[str, bool]) -> None:
    +    """Release a pending-work reservation exactly once."""
    +    if reservation["active"]:
    +        reservation["active"] = False
    +        adapter._pending_agent_requests = max(0, adapter._pending_agent_requests - 1)
    +
    +
    +@contextmanager
    +def _reserve_pending_api_work(adapter):
    +    """Keep externally-triggered background work visible across awaits.
    +
    +    A handler can detach the reservation to an asyncio task; its done callback
    +    then owns release so shutdown cannot miss the handoff to background work.
    +    """
    +    reservation = {"active": True, "detached": False}
    +    adapter._pending_agent_requests += 1
    +    try:
    +        yield reservation
    +    finally:
    +        if not reservation["detached"]:
    +            _release_pending_api_work(adapter, reservation)
    +
    +
     if AIOHTTP_AVAILABLE:
         @web.middleware
         async def body_limit_middleware(request, handler):
    ```

    ```diff
    @@ -912,6 +975,64 @@ def __init__(self, config: PlatformConfig):
             # (the /v1/runs path tracks its own in-flight set via
             # _active_run_tasks).
             self._inflight_agent_runs: int = 0
    +        # Requests admitted before their handler reaches agent bookkeeping.
    +        # Shutdown counts this reservation so the request cannot slip through
    +        # the drain between its first await and _run_agent()/task registration.
    +        self._pending_agent_requests: int = 0
    +
    +    def active_agent_work_count(self) -> int:
    +        """Return all live agent work owned by this API adapter.
    +
    +        ``/v1/runs`` registers an asyncio task before it constructs and stores
    +        its agent, so ``_active_run_agents`` has a real queued-before-agent gap.
    +        Reuse the task-based accounting used by the concurrent-run limit: it
    +        covers that gap and excludes completed tasks retained until cleanup.
    +        """
    +        try:
    +            return (
    +                int(getattr(self, "_pending_agent_requests", 0))
    +                + int(self._inflight_agent_runs)
    +                + sum(not task.done() for task in self._active_run_tasks.values())
    +            )
    +        except Exception:
    +            return 0
    +
    +    @staticmethod
    +    def _gateway_is_draining() -> bool:
    +        """Whether the owning gateway currently refuses new agent turns."""
    +        try:
    +            from gateway.run import _gateway_runner_ref
    +
    +            runner = _gateway_runner_ref()
    +            return bool(
    +                runner
    +                and (
    +                    getattr(runner, "_draining", False)
    +                    or getattr(runner, "_external_drain_active", False)
    +                )
    +            )
    +        except Exception:
    +            return False
    +
    +    def _draining_response(self) -> Optional["web.Response"]:
    +        """Return a retryable response while the gateway drains existing work."""
    +        if not self._gateway_is_draining():
    +            return None
    +        return web.json_response(
    +            _openai_error(
    +                "Gateway is draining existing work; retry shortly.",
    +                code="gateway_draining",
    +            ),
    +            status=503,
    +            headers={"Retry-After": "1"},
    +        )
    +
    +    def _activate_admitted_request(self) -> None:
    +        """Transfer this request's drain reservation to agent bookkeeping."""
    +        reservation = _api_agent_request_reservation.get()
    +        if reservation and reservation["active"]:
    +            reservation["active"] = False
    +            self._pending_agent_requests = max(0, self._pending_agent_requests - 1)
     
         def _readiness_work_counts(self) -> tuple[int, int, int]:
             """Return bounded work counts from each subsystem's public state."""
    ```

    ```diff
    @@ -1919,11 +2040,9 @@ async def _handle_fork_session(self, request: "web.Request") -> "web.Response":
             fork = db.get_session(fork_id) or {"id": fork_id, "parent_session_id": source_id}
             return web.json_response({"object": "hermes.session", "session": self._session_response(fork)}, status=201)
     
    +    @_admit_api_agent_request
         async def _handle_session_chat(self, request: "web.Request") -> "web.Response":
             """POST /api/sessions/{session_id}/chat — one synchronous agent turn."""
    -        auth_err = self._check_auth(request)
    -        if auth_err:
    -            return auth_err
             gateway_session_key, key_err = self._parse_session_key_header(request)
             if key_err is not None:
                 return key_err
    ```

    ```diff
    @@ -1963,11 +2082,9 @@ async def _handle_session_chat(self, request: "web.Request") -> "web.Response":
                 headers=headers,
             )
     
    +    @_admit_api_agent_request
         async def _handle_session_chat_stream(self, request: "web.Request") -> "web.StreamResponse":
             """POST /api/sessions/{session_id}/chat/stream — SSE wrapper over _run_agent."""
    -        auth_err = self._check_auth(request)
    -        if auth_err:
    -            return auth_err
             gateway_session_key, key_err = self._parse_session_key_header(request)
             if key_err is not None:
                 return key_err
    ```

    ```diff
    @@ -2104,12 +2221,9 @@ async def _run_and_signal() -> None:
                 logger.debug("[api_server] session SSE stream error: %s", exc)
             return response
     
    +    @_admit_api_agent_request
         async def _handle_chat_completions(self, request: "web.Request") -> "web.Response":
             """POST /v1/chat/completions — OpenAI Chat Completions format."""
    -        auth_err = self._check_auth(request)
    -        if auth_err:
    -            return auth_err
    -
             # Bound total in-flight agent runs (configurable; #7483).
             limited = self._concurrency_limited_response()
             if limited is not None:
    ```

    ```diff
    @@ -2311,8 +2425,8 @@ def _on_tool_complete(tool_call_id, function_name, function_args, function_resul
                 # ``tool_progress_callback`` is intentionally not wired here:
                 # it would duplicate every emit because ``run_agent`` fires it
                 # side-by-side with ``tool_start_callback``/``tool_complete_callback``.
    -            # The structured callbacks are strictly richer (they carry the
    -            # tool_call id), so they own the chat-completions SSE channel.
    +            # The structured callbacks are strictly richer (they carry
    +            # the tool_call id), so they own the chat-completions SSE channel.
                 agent_ref = [None]
                 agent_task = asyncio.ensure_future(self._run_agent(
                     user_message=user_message,
    ```

    ```diff
    @@ -3240,12 +3354,9 @@ async def _flush_batch() -> None:
     
             return response
     
    +    @_admit_api_agent_request
         async def _handle_responses(self, request: "web.Request") -> "web.Response":
             """POST /v1/responses — OpenAI Responses API format."""
    -        auth_err = self._check_auth(request)
    -        if auth_err:
    -            return auth_err
    -
             # Bound total in-flight agent runs (configurable; #7483).
             limited = self._concurrency_limited_response()
             if limited is not None:
    ```

    ```diff
    @@ -3788,6 +3899,9 @@ async def _handle_run_job(self, request: "web.Request") -> "web.Response":
             auth_err = self._check_auth(request)
             if auth_err:
                 return auth_err
    +        draining = self._draining_response()
    +        if draining is not None:
    +            return draining
             cron_err = self._check_jobs_available()
             if cron_err:
                 return cron_err
    ```

    ```diff
    @@ -3833,31 +3947,39 @@ async def _handle_cron_fire(self, request: "web.Request") -> "web.Response":
                     self._request_audit_log_suffix(request),
                 )
                 return web.json_response({"error": "invalid fire token"}, status=401)
    +        draining = self._draining_response()
    +        if draining is not None:
    +            return draining
     
    -        try:
    -            body = await request.json()
    -        except Exception:
    -            body = {}
    -        job_id = (body or {}).get("job_id")
    -        if not job_id:
    -            return web.json_response({"error": "missing job_id"}, status=400)
    +        with _reserve_pending_api_work(self) as reservation:
    +            try:
    +                body = await request.json()
    +            except Exception:
    +                body = {}
    +            job_id = (body or {}).get("job_id")
    +            if not job_id:
    +                return web.json_response({"error": "missing job_id"}, status=400)
     
    -        from cron.scheduler_provider import resolve_cron_scheduler
    -        provider = resolve_cron_scheduler()
    +            from cron.scheduler_provider import resolve_cron_scheduler
    +            provider = resolve_cron_scheduler()
     
    -        loop = asyncio.get_running_loop()
    -        # Fire in the background (202 immediately). fire_due claims via the
    -        # store CAS, so a retry while this is in flight is de-duped.
    -        task = asyncio.create_task(
    -            asyncio.to_thread(provider.fire_due, job_id, adapters=None, loop=loop)
    -        )
    -        try:
    -            self._background_tasks.add(task)
    -            task.add_done_callback(self._background_tasks.discard)
    -        except (TypeError, AttributeError):
    -            pass
    +            loop = asyncio.get_running_loop()
    +            # Fire in the background (202 immediately). fire_due claims via the
    +            # store CAS, so a retry while this is in flight is de-duped.
    +            task = asyncio.create_task(
    +                asyncio.to_thread(provider.fire_due, job_id, adapters=None, loop=loop)
    +            )
    +            reservation["detached"] = True
    +            task.add_done_callback(
    +                lambda _task: _release_pending_api_work(self, reservation)
    +            )
    +            try:
    +                self._background_tasks.add(task)
    +                task.add_done_callback(self._background_tasks.discard)
    +            except (TypeError, AttributeError):
    +                pass
     
    -        return web.json_response({"status": "accepted", "job_id": job_id}, status=202)
    +            return web.json_response({"status": "accepted", "job_id": job_id}, status=202)
     
     
         # ------------------------------------------------------------------
    ```

    ```diff
    @@ -4011,19 +4133,22 @@ def _concurrency_limited_response(self) -> Optional["web.Response"]:
             """Return a 429 response if the concurrent-run cap is reached, else None.
     
             The cap bounds total in-flight agent activity across every
    -        agent-serving endpoint: the non-streaming chat/responses paths
    -        (tracked by ``_inflight_agent_runs``) plus the ``/v1/runs`` path
    -        (tracked by live entries in ``_active_run_tasks``).  Stream queues are
    -        transport state and may disappear while their underlying run remains
    -        active, so they must not define run concurrency. A configured value of
    -        0 disables the cap entirely.
    +        agent-serving endpoint. Reuse the same adapter-owned work count that
    +        shutdown draining uses, including an admitted request before it reaches
    +        agent/task bookkeeping. Stream queues are transport state and may
    +        disappear while their underlying run remains active, so they must not
    +        define run concurrency. A configured value of 0 disables the cap.
             """
             limit = self._max_concurrent_runs
             if limit <= 0:
                 return None
    -        inflight = self._inflight_agent_runs + sum(
    -            not task.done() for task in self._active_run_tasks.values()
    -        )
    +        inflight = self.active_agent_work_count()
    +        # The current request owns one reservation until it hands off to
    +        # _run_agent() or /v1/runs task registration. It must not consume its
    +        # own last available slot; other admitted requests remain counted.
    +        reservation = _api_agent_request_reservation.get()
    +        if reservation and reservation["active"]:
    +            inflight -= 1
             if inflight >= limit:
                 return web.json_response(
                     _openai_error(
    ```

    ```diff
    @@ -4141,6 +4266,7 @@ def _run():
                 finally:
                     clear_session_vars(tokens)
     
    +        self._activate_admitted_request()
             self._inflight_agent_runs += 1
             try:
                 return await loop.run_in_executor(None, _run)
    ```

    ```diff
    @@ -4215,12 +4341,9 @@ def _callback(event_type: str, tool_name: str = None, preview: str = None, args=
     
             return _callback
     
    +    @_admit_api_agent_request
         async def _handle_runs(self, request: "web.Request") -> "web.Response":
             """POST /v1/runs — start an agent run, return run_id immediately."""
    -        auth_err = self._check_auth(request)
    -        if auth_err:
    -            return auth_err
    -
             # Long-term memory scope header (see chat_completions for details).
             gateway_session_key, key_err = self._parse_session_key_header(request)
             if key_err is not None:
    ```

    ```diff
    @@ -4542,6 +4665,7 @@ def _run_sync():
                     self._run_approval_sessions.pop(run_id, None)
                     self._stopping_run_ids.discard(run_id)
     
    +        self._activate_admitted_request()
             task = asyncio.create_task(_run_and_close())
             self._active_run_tasks[run_id] = task
             try:
    ```
  - `gateway/run.py`

    ```diff
    @@ -4122,6 +4122,14 @@ def _request_clean_exit(self, reason: str) -> None:
         def _running_agent_count(self) -> int:
             return len(self._running_agents)
     
    +    def _active_work_count(self) -> int:
    +        """All agent work the gateway must expose and drain as one total."""
    +        return (
    +            self._running_agent_count()
    +            + self._active_cron_job_count()
    +            + self._active_api_run_count()
    +        )
    +
         def _active_cron_job_count(self) -> int:
             """Count of cron jobs currently executing, from the cron scheduler's
             own in-flight tracking (``cron.scheduler._running_job_ids``).
    ```

    ```diff
    @@ -4142,6 +4150,20 @@ def _active_cron_job_count(self) -> int:
             except Exception:
                 return 0
     
    +    def _active_api_run_count(self) -> int:
    +        """Count API-server work that is outside ``_running_agents``.
    +
    +        The primary API server owns the sole HTTP listener. Secondary multiplex
    +        profiles cannot create an ``api_server`` adapter because it binds a port,
    +        so only the primary registry is a supported source of this work.
    +        """
    +        try:
    +            adapter = getattr(self, "adapters", {}).get(Platform.API_SERVER)
    +            helper = getattr(adapter, "active_agent_work_count", None)
    +            return max(0, int(helper())) if callable(helper) else 0
    +        except Exception:
    +            return 0
    +
         # ── scale-to-zero idle detection / dormant-quiesce (Phase 0) ──────────────
         # The gateway-side BEHAVIOUR that consumes the relay scale-to-zero primitives
         # (gateway-gateway Phase 5). Pure logic lives in gateway/scale_to_zero.py; the
    ```

    ```diff
    @@ -4529,7 +4551,7 @@ def _update_runtime_status(self, gateway_state: Optional[str] = None, exit_reaso
                     gateway_state=gateway_state,
                     exit_reason=exit_reason,
                     restart_requested=self._restart_requested,
    -                active_agents=self._running_agent_count(),
    +                active_agents=self._active_work_count(),
                 )
             except Exception:
                 pass
    ```

    ```diff
    @@ -4552,7 +4574,7 @@ def _persist_active_agents(self) -> None:
             """
             try:
                 from gateway.status import write_runtime_status
    -            write_runtime_status(active_agents=self._running_agent_count())
    +            write_runtime_status(active_agents=self._active_work_count())
             except Exception:
                 pass
     
    ```

    ```diff
    @@ -4579,7 +4601,7 @@ def _enter_external_drain(self) -> None:
             logger.info(
                 "External drain ENGAGED (.drain_request.json present) — refusing "
                 "new turns; %d in-flight turn(s) will finish. Process stays up.",
    -            self._running_agent_count(),
    +            self._active_work_count(),
             )
             # Flip the persisted lifecycle state so /api/status.gateway_busy /
             # gateway_drainable track the drain. Preserve active_agents (the
    ```

    ```diff
    @@ -4630,6 +4652,10 @@ async def _drain_control_watcher(self, interval: float = 1.0) -> None:
                 try:
                     if drain_requested():
                         self._enter_external_drain()
    +                    # API and cron work live outside messaging's
    +                    # _running_agents map. Refresh the aggregate while an
    +                    # external caller polls this reversible drain state.
    +                    self._persist_active_agents()
                     else:
                         self._exit_external_drain()
                 except asyncio.CancelledError:
    ```

    ```diff
    @@ -5696,22 +5722,26 @@ async def _drain_active_agents(self, timeout: float) -> tuple[Dict[str, Any], bo
             snapshot = self._snapshot_running_agents()
             last_active_count = self._running_agent_count()
             last_cron_count = self._active_cron_job_count()
    +        last_api_count = self._active_api_run_count()
             last_status_at = 0.0
     
             def _maybe_update_status(force: bool = False) -> None:
    -            nonlocal last_active_count, last_cron_count, last_status_at
    +            nonlocal last_active_count, last_cron_count, last_api_count, last_status_at
                 now = asyncio.get_running_loop().time()
                 active_count = self._running_agent_count()
                 cron_count = self._active_cron_job_count()
    +            api_count = self._active_api_run_count()
                 if (
                     force
                     or active_count != last_active_count
                     or cron_count != last_cron_count
    +                or api_count != last_api_count
                     or (now - last_status_at) >= 1.0
                 ):
                     self._update_runtime_status("draining")
                     last_active_count = active_count
                     last_cron_count = cron_count
    +                last_api_count = api_count
                     last_status_at = now
     
             # Cron jobs run on the scheduler's own thread pool, outside
    ```

    ```diff
    @@ -5719,7 +5749,8 @@ def _maybe_update_status(force: bool = False) -> None:
             # same wait/timeout this method already applies to chat sessions,
             # or a cron job's tool work gets killed with zero warning the
             # instant it's the only active thing running (#60432).
    -        if not self._running_agents and last_cron_count == 0:
    +        # API-server / desk sessions have the same structural gap (#63529).
    +        if not self._running_agents and last_cron_count == 0 and last_api_count == 0:
                 _maybe_update_status(force=True)
                 return snapshot, False
     
    ```

    ```diff
    @@ -5729,12 +5760,20 @@ def _maybe_update_status(force: bool = False) -> None:
     
             deadline = asyncio.get_running_loop().time() + timeout
             while (
    -            (self._running_agents or self._active_cron_job_count())
    +            (
    +                self._running_agents
    +                or self._active_cron_job_count()
    +                or self._active_api_run_count()
    +            )
                 and asyncio.get_running_loop().time() < deadline
             ):
                 _maybe_update_status()
                 await asyncio.sleep(0.1)
    -        timed_out = bool(self._running_agents) or bool(self._active_cron_job_count())
    +        timed_out = (
    +            bool(self._running_agents)
    +            or bool(self._active_cron_job_count())
    +            or bool(self._active_api_run_count())
    +        )
             _maybe_update_status(force=True)
             return snapshot, timed_out
     
    ```

    ```diff
    @@ -8187,12 +8226,14 @@ def _phase_elapsed() -> float:
                         logger.debug("pre-drain mark_resume_pending failed for %s: %s", _sk, _e)
     
                 _cron_at_start = self._active_cron_job_count()
    +            _api_at_start = self._active_api_run_count()
                 _drain_started_at = time.monotonic()
                 active_agents, timed_out = await self._drain_active_agents(timeout)
                 logger.info(
                     "Shutdown phase: drain done at +%.2fs (drain took %.2fs, "
                     "timed_out=%s, active_at_start=%d, active_now=%d, "
    -                "cron_at_start=%d, cron_now=%d)",
    +                "cron_at_start=%d, cron_now=%d, "
    +                "api_at_start=%d, api_now=%d)",
                     _phase_elapsed(),
                     time.monotonic() - _drain_started_at,
                     timed_out,
    ```

    ```diff
    @@ -8200,6 +8241,8 @@ def _phase_elapsed() -> float:
                     self._running_agent_count(),
                     _cron_at_start,
                     self._active_cron_job_count(),
    +                _api_at_start,
    +                self._active_api_run_count(),
                 )
     
                 if not timed_out:
    ```

    ```diff
    @@ -8218,11 +8261,13 @@ def _phase_elapsed() -> float:
     
                 if timed_out:
                     logger.warning(
    -                    "Gateway drain timed out after %.1fs with %d active agent(s) "
    -                    "and %d in-flight cron job(s); interrupting remaining work.",
    +                    "Gateway drain timed out after %.1fs with %d active agent(s), "
    +                    "%d in-flight cron job(s), and %d api_server run(s); "
    +                    "interrupting remaining work.",
                         timeout,
                         self._running_agent_count(),
                         self._active_cron_job_count(),
    +                    self._active_api_run_count(),
                     )
                     # Mark forcibly-interrupted sessions as resume_pending BEFORE
                     # interrupting the agents.  This preserves each session's
    ```

    ```diff
    @@ -20971,13 +21016,21 @@ def restart_signal_handler():
         # historical in-process 60s ticker; an external provider (e.g. chronos)
         # may arm a schedule and return. Pass the event loop so cron delivery can
         # use live adapters (E2EE support).
    -    from cron.scheduler_provider import resolve_cron_scheduler
    +    from cron.scheduler_provider import InProcessCronScheduler, resolve_cron_scheduler
         cron_stop = threading.Event()
         cron_provider = resolve_cron_scheduler()
    +    cron_start_kwargs = {"adapters": runner.adapters, "loop": asyncio.get_running_loop()}
    +    # External cron providers own their remote scheduling contract. Only the
    +    # in-process ticker polls local due jobs, so only it receives the local
    +    # external-drain dispatch gate.
    +    if isinstance(cron_provider, InProcessCronScheduler):
    +        cron_start_kwargs["can_dispatch"] = lambda: not (
    +            runner._draining or runner._external_drain_active
    +        )
         cron_thread = threading.Thread(
             target=cron_provider.start,
             args=(cron_stop,),
    -        kwargs={"adapters": runner.adapters, "loop": asyncio.get_running_loop()},
    +        kwargs=cron_start_kwargs,
             daemon=True,
             name="cron-scheduler",
         )
    ```
  - `tests/cron/test_scheduler.py`

    ```diff
    @@ -1533,6 +1533,25 @@ def test_run_job_delivers_max_iteration_fallback_summary(self, tmp_path):
             assert "final fallback report" in output
             assert "(FAILED)" not in output
     
    +    def test_tick_skips_due_jobs_while_dispatch_is_paused(self, tmp_path):
    +        """The drain gate runs before advancing a due job's schedule."""
    +        from cron.scheduler import tick
    +
    +        job = {
    +            "id": "paused-due-job",
    +            "name": "paused due job",
    +            "schedule": {"kind": "interval", "seconds": 60},
    +            "next_run_at": "2020-01-01T00:00:00+00:00",
    +            "enabled": True,
    +        }
    +        with patch("cron.scheduler.get_due_jobs", return_value=[job]), patch(
    +            "cron.scheduler.advance_next_run"
    +        ) as advance, patch("cron.scheduler.run_one_job") as run_one:
    +            assert tick(verbose=False, sync=True, can_dispatch=lambda: False) == 0
    +
    +        advance.assert_not_called()
    +        run_one.assert_not_called()
    +
         def test_tick_marks_empty_response_as_error(self, tmp_path):
             """When run_job returns success=True but final_response is empty,
             tick() should mark the job as error so last_status != 'ok'.
    ```
  - `tests/cron/test_scheduler_provider.py`

    ```diff
    @@ -129,13 +129,11 @@ def start(self, stop_event, **kw):
     
     
     def test_abc_growth_stays_additive():
    -    """Forward-compat guard: the ABC's REQUIRED surface is exactly name+start.
    +    """The provider interface stays source-compatible with existing plugins.
     
    -    Any optional hook added later for the external provider
    -    (on_jobs_changed/fire_due/reconcile) must be NON-abstract (carry a default),
    -    so the built-in keeps satisfying the ABC without overriding them. This test
    -    fails loudly if someone makes a future hook abstract (a breaking change that
    -    would force every provider — including the built-in — to implement it).
    +    ``start`` must be the only required implementation hook: future optional
    +    behavior belongs in non-abstract default methods so custom plugins do not
    +    break on import after an upgrade.
         """
         from cron.scheduler_provider import CronScheduler
     
    ```

    ```diff
    @@ -174,6 +172,33 @@ def test_inprocess_provider_ticks_and_stops():
         assert calls[0].get("sync") is False
     
     
    +def test_inprocess_provider_skips_dispatch_while_draining():
    +    """A drain pause keeps due work pending until dispatch is re-enabled."""
    +    from cron.scheduler_provider import InProcessCronScheduler
    +
    +    calls = []
    +    stop = threading.Event()
    +    allow_dispatch = threading.Event()
    +    provider = InProcessCronScheduler()
    +
    +    with patch("cron.scheduler.tick", side_effect=lambda *a, **k: calls.append(k) or 0):
    +        thread = threading.Thread(
    +            target=provider.start,
    +            args=(stop,),
    +            kwargs={"interval": 0.01, "can_dispatch": allow_dispatch.is_set},
    +            daemon=True,
    +        )
    +        thread.start()
    +        time.sleep(0.05)
    +        assert calls == []
    +        allow_dispatch.set()
    +        assert _wait_until(lambda: len(calls) >= 1), "provider never resumed dispatch"
    +        stop.set()
    +        thread.join(timeout=5)
    +
    +    assert not thread.is_alive()
    +
    +
     def test_inprocess_provider_stop_is_noop():
         """The default stop() hook is a safe no-op (the stop_event is the real
         stop signal for the built-in)."""
    ```
  - `tests/gateway/restart_test_helpers.py`

    ```diff
    @@ -115,6 +115,18 @@ def make_restart_runner(
         runner._running_agent_count = GatewayRunner._running_agent_count.__get__(
             runner, GatewayRunner
         )
    +    runner._active_cron_job_count = GatewayRunner._active_cron_job_count.__get__(
    +        runner, GatewayRunner
    +    )
    +    runner._active_api_run_count = GatewayRunner._active_api_run_count.__get__(
    +        runner, GatewayRunner
    +    )
    +    runner._active_work_count = GatewayRunner._active_work_count.__get__(
    +        runner, GatewayRunner
    +    )
    +    runner._persist_active_agents = GatewayRunner._persist_active_agents.__get__(
    +        runner, GatewayRunner
    +    )
         runner._snapshot_running_agents = GatewayRunner._snapshot_running_agents.__get__(
             runner, GatewayRunner
         )
    ```
  - `tests/gateway/test_api_server_active_work_drain.py`

    ```diff
    @@ -0,0 +1,326 @@
    +"""Regression coverage for #63529 API-server shutdown draining.
    +
    +API-server work is adapter-owned rather than tracked by
    +``GatewayRunner._running_agents``. The shutdown drain must account for the
    +same live state as the API concurrency limiter, including a ``/v1/runs`` task
    +that exists before its agent has been constructed, and it must refuse new API
    +turns once the gateway starts draining.
    +"""
    +
    +import asyncio
    +from types import SimpleNamespace
    +from unittest.mock import AsyncMock, MagicMock, patch
    +
    +import pytest
    +from aiohttp import web
    +from aiohttp.test_utils import TestClient, TestServer
    +
    +from gateway.config import Platform, PlatformConfig
    +from gateway.platforms.api_server import APIServerAdapter
    +from tests.gateway.restart_test_helpers import make_restart_runner
    +
    +
    +class _RunTask:
    +    def __init__(self, done: bool = False):
    +        self._done = done
    +
    +    def done(self) -> bool:
    +        return self._done
    +
    +
    +def _make_api_adapter(*, inflight: int = 0, queued_ids=()):
    +    tasks = {run_id: _RunTask() for run_id in queued_ids}
    +    adapter = SimpleNamespace(
    +        platform=Platform.API_SERVER,
    +        _inflight_agent_runs=inflight,
    +        _active_run_tasks=tasks,
    +    )
    +
    +    def active_agent_work_count() -> int:
    +        return int(getattr(adapter, "_pending_agent_requests", 0)) + int(
    +            adapter._inflight_agent_runs
    +        ) + sum(not task.done() for task in adapter._active_run_tasks.values())
    +
    +    adapter.active_agent_work_count = active_agent_work_count
    +    return adapter
    +
    +
    +def _make_admission_app(adapter: APIServerAdapter) -> web.Application:
    +    app = web.Application()
    +    app.router.add_post("/api/sessions/{session_id}/chat", adapter._handle_session_chat)
    +    app.router.add_post(
    +        "/api/sessions/{session_id}/chat/stream", adapter._handle_session_chat_stream
    +    )
    +    app.router.add_post("/v1/chat/completions", adapter._handle_chat_completions)
    +    app.router.add_post("/v1/responses", adapter._handle_responses)
    +    app.router.add_post("/v1/runs", adapter._handle_runs)
    +    return app
    +
    +
    +class TestActiveApiRunCount:
    +    def test_zero_when_no_api_adapter(self):
    +        runner, _adapter = make_restart_runner()
    +        runner.adapters = {}
    +        assert runner._active_api_run_count() == 0
    +
    +    def test_delegates_to_primary_api_adapter(self):
    +        runner, _adapter = make_restart_runner()
    +        runner.adapters = {
    +            Platform.API_SERVER: _make_api_adapter(inflight=2, queued_ids=["r1"])
    +        }
    +        assert runner._active_api_run_count() == 3
    +
    +    def test_ignores_non_api_platforms(self):
    +        runner, _adapter = make_restart_runner()
    +        other = SimpleNamespace(
    +            platform=Platform.DISCORD,
    +            active_agent_work_count=lambda: 99,
    +        )
    +        runner.adapters = {Platform.DISCORD: other}
    +        assert runner._active_api_run_count() == 0
    +
    +    def test_never_raises_on_broken_adapter(self):
    +        runner, _adapter = make_restart_runner()
    +
    +        class Bad:
    +            platform = Platform.API_SERVER
    +
    +            @staticmethod
    +            def active_agent_work_count() -> int:
    +                raise RuntimeError("boom")
    +
    +        runner.adapters = {Platform.API_SERVER: Bad()}
    +        assert runner._active_api_run_count() == 0
    +
    +
    +class TestAPIServerAdapterWorkCount:
    +    def test_concurrency_limit_counts_other_pending_admissions(self):
    +        adapter = APIServerAdapter(PlatformConfig(enabled=True))
    +        adapter._max_concurrent_runs = 1
    +        adapter._pending_agent_requests = 1
    +
    +        response = adapter._concurrency_limited_response()
    +
    +        assert response is not None
    +        assert response.status == 429
    +
    +    @pytest.mark.asyncio
    +    async def test_concurrency_limit_excludes_current_pending_admission(self):
    +        adapter = APIServerAdapter(PlatformConfig(enabled=True))
    +        adapter._max_concurrent_runs = 1
    +        app = _make_admission_app(adapter)
    +
    +        async with TestClient(TestServer(app)) as client:
    +            with patch.object(adapter, "_run_agent", new=AsyncMock(return_value=({}, {}))):
    +                response = await client.post(
    +                    "/api/sessions/s/chat",
    +                    json={"message": "hello"},
    +                )
    +
    +        assert response.status == 404
    +
    +    def test_counts_pending_admission_before_agent_bookkeeping(self):
    +        adapter = APIServerAdapter(PlatformConfig(enabled=True))
    +        adapter._pending_agent_requests = 1
    +
    +        assert adapter.active_agent_work_count() == 1
    +
    +    def test_counts_live_run_task_before_agent_creation(self):
    +        adapter = APIServerAdapter(PlatformConfig(enabled=True))
    +        adapter._inflight_agent_runs = 2
    +        adapter._active_run_tasks = {
    +            "queued": _RunTask(),
    +            "finished": _RunTask(done=True),
    +        }
    +        adapter._active_run_agents = {}
    +
    +        assert adapter.active_agent_work_count() == 3
    +
    +    def test_does_not_double_count_started_run_agent(self):
    +        adapter = APIServerAdapter(PlatformConfig(enabled=True))
    +        adapter._inflight_agent_runs = 0
    +        adapter._active_run_tasks = {"run-1": _RunTask()}
    +        adapter._active_run_agents = {"run-1": object()}
    +
    +        assert adapter.active_agent_work_count() == 1
    +
    +
    +class TestDrainWaitsForApiWork:
    +    @pytest.mark.asyncio
    +    async def test_drain_returns_immediately_when_nothing_active(self):
    +        runner, _adapter = make_restart_runner()
    +        runner.adapters = {}
    +
    +        _snapshot, timed_out = await runner._drain_active_agents(5.0)
    +
    +        assert timed_out is False
    +
    +    @pytest.mark.asyncio
    +    async def test_drain_waits_for_real_queued_run_before_agent_creation(self):
    +        """A live /v1/runs task must block drain before it has an agent."""
    +        runner, _adapter = make_restart_runner()
    +        api = APIServerAdapter(PlatformConfig(enabled=True))
    +        runner.adapters = {Platform.API_SERVER: api}
    +        app = _make_admission_app(api)
    +        original_create_task = asyncio.create_task
    +        task_started = asyncio.Event()
    +        allow_task = asyncio.Event()
    +
    +        def delayed_create_task(coro):
    +            async def delayed():
    +                task_started.set()
    +                await allow_task.wait()
    +                return await coro
    +
    +            return original_create_task(delayed())
    +
    +        mock_agent = MagicMock()
    +        mock_agent.run_conversation.return_value = {"final_response": "done"}
    +        mock_agent.session_prompt_tokens = 0
    +        mock_agent.session_completion_tokens = 0
    +        mock_agent.session_total_tokens = 0
    +
    +        with patch(
    +            "gateway.platforms.api_server.asyncio.create_task",
    +            side_effect=delayed_create_task,
    +        ), patch.object(api, "_create_agent", return_value=mock_agent):
    +            async with TestClient(TestServer(app)) as client:
    +                response = await client.post("/v1/runs", json={"input": "hello"})
    +                assert response.status == 202
    +                await task_started.wait()
    +
    +                assert api._active_run_agents == {}
    +                assert runner._active_api_run_count() == 1
    +                drain_task = original_create_task(runner._drain_active_agents(2.0))
    +                await asyncio.sleep(0.1)
    +                assert not drain_task.done()
    +
    +                allow_task.set()
    +                _snapshot, timed_out = await drain_task
    +
    +        assert timed_out is False
    +
    +    @pytest.mark.asyncio
    +    async def test_drain_times_out_if_api_run_outlives_the_window(self):
    +        runner, _adapter = make_restart_runner()
    +        runner.adapters = {Platform.API_SERVER: _make_api_adapter(queued_ids=["run-1"])}
    +
    +        _snapshot, timed_out = await runner._drain_active_agents(0.1)
    +
    +        assert timed_out is True
    +
    +    @pytest.mark.asyncio
    +    async def test_drain_still_waits_for_chat_cron_and_api_work(self):
    +        import cron.scheduler as sched
    +
    +        runner, _adapter = make_restart_runner()
    +        runner._running_agents = {"session-1": MagicMock()}
    +        sched._running_job_ids.add("job-1")
    +        runner.adapters = {Platform.API_SERVER: _make_api_adapter(queued_ids=["run-1"])}
    +
    +        async def finish_all():
    +            await asyncio.sleep(0.12)
    +            runner._running_agents.clear()
    +            sched._running_job_ids.discard("job-1")
    +            runner.adapters[Platform.API_SERVER]._active_run_tasks.clear()
    +
    +        task = asyncio.create_task(finish_all())
    +        try:
    +            _snapshot, timed_out = await runner._drain_active_agents(2.0)
    +        finally:
    +            await task
    +            sched._running_job_ids.discard("job-1")
    +
    +        assert timed_out is False
    +
    +
    +class TestDrainAdmission:
    +    @pytest.mark.asyncio
    +    async def test_drain_refuses_every_agent_start_endpoint(self):
    +        adapter = APIServerAdapter(PlatformConfig(enabled=True))
    +        runner = SimpleNamespace(_draining=True, _external_drain_active=False)
    +        app = _make_admission_app(adapter)
    +        paths = (
    +            "/api/sessions/missing/chat",
    +            "/api/sessions/missing/chat/stream",
    +            "/v1/chat/completions",
    +            "/v1/responses",
    +            "/v1/runs",
    +        )
    +
    +        with patch("gateway.run._gateway_runner_ref", lambda: runner):
    +            async with TestClient(TestServer(app)) as client:
    +                for path in paths:
    +                    response = await client.post(path, json={})
    +                    payload = await response.json()
    +
    +                    assert response.status == 503
    +                    assert response.headers["Retry-After"] == "1"
    +                    assert payload["error"]["code"] == "gateway_draining"
    +
    +    @pytest.mark.asyncio
    +    async def test_external_drain_refuses_every_agent_start_endpoint(self):
    +        adapter = APIServerAdapter(PlatformConfig(enabled=True))
    +        runner = SimpleNamespace(_draining=False, _external_drain_active=True)
    +        app = _make_admission_app(adapter)
    +        paths = (
    +            "/api/sessions/missing/chat",
    +            "/api/sessions/missing/chat/stream",
    +            "/v1/chat/completions",
    +            "/v1/responses",
    +            "/v1/runs",
    +        )
    +
    +        with patch("gateway.run._gateway_runner_ref", lambda: runner):
    +            async with TestClient(TestServer(app)) as client:
    +                for path in paths:
    +                    response = await client.post(path, json={})
    +                    payload = await response.json()
    +
    +                    assert response.status == 503
    +                    assert payload["error"]["code"] == "gateway_draining"
    +
    +    @pytest.mark.asyncio
    +    async def test_admitted_request_blocks_drain_before_agent_bookkeeping(self):
    +        adapter = APIServerAdapter(PlatformConfig(enabled=True))
    +        runner, _adapter = make_restart_runner()
    +        runner.adapters = {Platform.API_SERVER: adapter}
    +        app = _make_admission_app(adapter)
    +        body_read_started = asyncio.Event()
    +        allow_body_read = asyncio.Event()
    +
    +        async def delayed_read_json(_request):
    +            body_read_started.set()
    +            await allow_body_read.wait()
    +            return {"message": "hello"}, None
    +
    +        with patch.object(
    +            adapter,
    +            "_get_existing_session_or_404",
    +            return_value=({}, None),
    +        ), patch.object(
    +            adapter,
    +            "_read_json_body",
    +            side_effect=delayed_read_json,
    +        ), patch.object(
    +            adapter,
    +            "_run_agent",
    +            new=AsyncMock(return_value=({"final_response": "done"}, {})),
    +        ):
    +            async with TestClient(TestServer(app)) as client:
    +                request_task = asyncio.create_task(
    +                    client.post("/api/sessions/missing/chat", json={})
    +                )
    +                await body_read_started.wait()
    +
    +                assert adapter._pending_agent_requests == 1
    +                drain_task = asyncio.create_task(runner._drain_active_agents(2.0))
    +                await asyncio.sleep(0.1)
    +                assert not drain_task.done()
    +
    +                allow_body_read.set()
    +                response = await request_task
    +                assert response.status == 200
    +                _snapshot, timed_out = await drain_task
    +
    +        assert timed_out is False
    ```
  - `tests/gateway/test_api_server_jobs.py`

    ```diff
    @@ -11,6 +11,7 @@
     """
     
     import logging
    +from types import SimpleNamespace
     from unittest.mock import MagicMock, patch
     
     import pytest
    ```

    ```diff
    @@ -494,6 +495,19 @@ async def test_run_job(self, adapter):
                     assert data["job"] == triggered_job
                     mock_trigger.assert_called_once_with(VALID_JOB_ID)
     
    +    @pytest.mark.asyncio
    +    async def test_run_job_refuses_during_gateway_drain(self, adapter):
    +        app = _create_app(adapter)
    +        runner = SimpleNamespace(_draining=False, _external_drain_active=True)
    +
    +        with patch("gateway.run._gateway_runner_ref", lambda: runner):
    +            async with TestClient(TestServer(app)) as cli:
    +                resp = await cli.post(f"/api/jobs/{VALID_JOB_ID}/run")
    +                payload = await resp.json()
    +
    +        assert resp.status == 503
    +        assert payload["error"]["code"] == "gateway_draining"
    +
     
     # ---------------------------------------------------------------------------
     # 17. test_auth_required
    ```
  - `tests/gateway/test_cron_fire_webhook.py`

    ```diff
    @@ -8,6 +8,9 @@
     """
     
     import asyncio
    +import threading
    +from types import SimpleNamespace
    +from unittest.mock import patch
     
     import pytest
     from aiohttp import web
    ```

    ```diff
    @@ -108,6 +111,86 @@ async def test_missing_token_401(adapter, monkeypatch):
         assert spy.fired == []
     
     
    [redacted:email]
    +async def test_valid_token_refuses_during_gateway_drain(adapter, monkeypatch):
    +    spy = _SpyProvider()
    +    runner = SimpleNamespace(_draining=False, _external_drain_active=True)
    +    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)
    +    with patch("gateway.run._gateway_runner_ref", lambda: runner):
    +        async with TestClient(TestServer(app)) as cli:
    +            response = await cli.post(
    +                "/api/cron/fire",
    +                headers={"Authorization": "Bearer good"},
    +                json={"job_id": "abc123"},
    +            )
    +            payload = await response.json()
    +
    +    assert response.status == 503
    +    assert payload["error"]["code"] == "gateway_draining"
    +    assert spy.fired == []
    +
    +
    [redacted:email]
    +async def test_valid_fire_reservation_blocks_drain_before_body_and_task(adapter, monkeypatch):
    +    runner = SimpleNamespace(_draining=False, _external_drain_active=False)
    +    body_started = asyncio.Event()
    +    release_body = asyncio.Event()
    +    fired = threading.Event()
    +    release_fire = threading.Event()
    +
    +    class BlockingProvider:
    +        def fire_due(self, job_id, *, adapters=None, loop=None):
    +            fired.set()
    +            release_fire.wait(timeout=2)
    +            return True
    +
    +    original_json = web.Request.json
    +
    +    async def delayed_json(request):
    +        body_started.set()
    +        await release_body.wait()
    +        return await original_json(request)
    +
    +    monkeypatch.setattr("cron.scheduler_provider.resolve_cron_scheduler", BlockingProvider)
    +    monkeypatch.setattr(
    +        "plugins.cron_providers.chronos.verify.get_fire_verifier",
    +        lambda: (lambda **kw: {"purpose": "cron_fire"}),
    +    )
    +    app = _create_app(adapter)
    +    with patch("gateway.run._gateway_runner_ref", lambda: runner), patch.object(
    +        web.Request, "json", delayed_json
    +    ):
    +        async with TestClient(TestServer(app)) as cli:
    +            request_task = asyncio.create_task(
    +                cli.post(
    +                    "/api/cron/fire",
    +                    headers={"Authorization": "Bearer good"},
    +                    json={"job_id": "abc123"},
    +                )
    +            )
    +            await body_started.wait()
    +            assert adapter.active_agent_work_count() == 1
    +
    +            release_body.set()
    +            response = await request_task
    +            assert response.status == 202
    +            await asyncio.to_thread(fired.wait, 2)
    +            assert adapter.active_agent_work_count() == 1
    +            release_fire.set()
    +            for _ in range(50):
    +                if adapter.active_agent_work_count() == 0:
    +                    break
    +                await asyncio.sleep(0.01)
    +
    +    assert adapter.active_agent_work_count() == 0
    +
    +
     @pytest.mark.asyncio
     async def test_missing_job_id_400(adapter, monkeypatch):
         """Valid token but no job_id → 400, no fire."""
    ```
  - `tests/gateway/test_external_drain_control.py`

    ```diff
    @@ -19,6 +19,7 @@
     
     import gateway.drain_control as dc
     from gateway.run import GatewayRunner
    +from gateway.config import Platform
     from gateway.platforms.base import MessageEvent, MessageType
     from tests.gateway.restart_test_helpers import make_restart_runner, make_restart_source
     
    ```

    ```diff
    @@ -238,6 +239,16 @@ def _drain_runner():
     
     
     class TestDrainStateMachine:
    +    def test_active_work_count_includes_api_and_cron_work(self, monkeypatch):
    +        runner, _ = _drain_runner()
    +        runner.adapters = {
    +            Platform.API_SERVER: MagicMock(active_agent_work_count=MagicMock(return_value=2))
    +        }
    +        runner._running_agents = {"session": MagicMock()}
    +        monkeypatch.setattr("cron.scheduler.get_running_job_ids", lambda: {"job-1"})
    +
    +        assert runner._active_work_count() == 4
    +
         def test_enter_sets_flag_and_flips_state(self):
             runner, _ = _drain_runner()
             runner._enter_external_drain()
    ```

    ```diff
    @@ -289,6 +300,26 @@ def test_exit_when_loop_stopped_does_not_revert(self):
     
     
     class TestDrainWatcher:
    +    @pytest.mark.asyncio
    +    async def test_watcher_persists_aggregate_work_during_external_drain(self, home, monkeypatch):
    +        runner, _ = _drain_runner()
    +        runner._drain_control_watcher = GatewayRunner._drain_control_watcher.__get__(
    +            runner, GatewayRunner
    +        )
    +        runner._persist_active_agents = MagicMock()
    +        dc.write_drain_request()
    +        task = asyncio.create_task(runner._drain_control_watcher(interval=0.01))
    +        await asyncio.sleep(0.03)
    +        runner._running = False
    +        await asyncio.sleep(0.02)
    +        task.cancel()
    +        try:
    +            await task
    +        except asyncio.CancelledError:
    +            pass
    +
    +        runner._persist_active_agents.assert_called()
    +
         @pytest.mark.asyncio
         async def test_watcher_enters_then_exits_with_marker(self, home):
             runner, _ = _drain_runner()
    ```
  - `tests/gateway/test_shutdown_cache_cleanup.py`

    ```diff
    @@ -60,6 +60,11 @@ def _active_cron_job_count(self):
             # there's never in-flight cron work to report.
             return 0
     
    +    def _active_api_run_count(self):
    +        # The shutdown log also reports adapter-owned API work (#63529).
    +        # This fake has no API server adapter, so it is always idle.
    +        return 0
    +
         def _update_runtime_status(self, *_a, **_kw):
             pass
     
    ```
- **#80917: fix(gateway): interrupt every in-flight API turn on shutdown, not just /v1/runs (#63529)**. merged to main · fix tier `verified`
  - maintainer direction: category `merged`
  - files: `gateway/platforms/api_server.py`, `gateway/run.py`, `tests/gateway/test_api_server_active_work_drain.py`
- **#63648: fix(gateway): drain in-flight api_server runs on shutdown**. closed without merge
  - maintainer direction: category `superseded_by_pr`; implication `superseded`; refs `#63891`

## Fix status

- Merged to main in PR #80917 (cited above). No release data is tracked. Verify downstream.
- Fix last verified (merged) `2026-08-07`.