fix(gateway): interrupt api_server runs on shutdown timeout

gateway/platforms/api_server.py

@@ -997,6 +997,25 @@ def active_agent_work_count(self) -> int:
         except Exception:
             return 0
 
+    def interrupt_active_runs(self, reason: str) -> int:
+        """Interrupt active API-run agents during gateway shutdown.
+
+        The gateway drain accounts for API-server work through
+        ``active_agent_work_count()``, but those agents are owned by this
+        adapter rather than ``GatewayRunner._running_agents``.  Expose the same
+        cooperative interrupt used by ``POST /v1/runs/{run_id}/stop`` so a
+        shutdown timeout can stop long-running API work before process teardown.
+        """
+        interrupted = 0
+        for run_id, agent in list(self._active_run_agents.items()):
+            try:
+                agent.interrupt(reason)
+                interrupted += 1
+                logger.debug("[api_server] interrupted active run %s during shutdown", run_id)
+            except Exception as exc:
+                logger.debug("[api_server] failed interrupting active run %s: %s", run_id, exc)
+        return interrupted
+
     @staticmethod
     def _gateway_is_draining() -> bool:
         """Whether the owning gateway currently refuses new agent turns."""

gateway/run.py

@@ -4164,6 +4164,16 @@ def _active_api_run_count(self) -> int:
         except Exception:
             return 0
 
+    def _interrupt_api_server_runs(self, reason: str) -> int:
+        """Interrupt API-server agents that are not in ``_running_agents``."""
+        try:
+            adapter = getattr(self, "adapters", {}).get(Platform.API_SERVER)
+            helper = getattr(adapter, "interrupt_active_runs", None)
+            return max(0, int(helper(reason))) if callable(helper) else 0
+        except Exception as exc:
+            logger.debug("Failed interrupting api_server runs during shutdown: %s", exc)
+            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
@@ -5786,6 +5796,9 @@ def _interrupt_running_agents(self, reason: str) -> None:
                 logger.debug("Interrupted running agent for session %s during shutdown", session_key)
             except Exception as e:
                 logger.debug("Failed interrupting agent during shutdown: %s", e)
+        interrupted_api = self._interrupt_api_server_runs(reason)
+        if interrupted_api:
+            logger.debug("Interrupted %d api_server run(s) during shutdown", interrupted_api)
 
     async def _notify_active_sessions_of_shutdown(self) -> None:
         """Send shutdown/restart notifications to active chats and home channels.
@@ -8307,7 +8320,10 @@ def _phase_elapsed() -> float:
                     _INTERRUPT_REASON_GATEWAY_RESTART if self._restart_requested else _INTERRUPT_REASON_GATEWAY_SHUTDOWN
                 )
                 interrupt_deadline = asyncio.get_running_loop().time() + 5.0
-                while self._running_agents and asyncio.get_running_loop().time() < interrupt_deadline:
+                while (
+                    self._running_agents
+                    or self._active_api_run_count()
+                ) and asyncio.get_running_loop().time() < interrupt_deadline:
                     self._update_runtime_status("draining")
                     await asyncio.sleep(0.1)
 

tests/gateway/test_api_server_active_work_drain.py

@@ -144,6 +144,15 @@ def test_does_not_double_count_started_run_agent(self):
 
         assert adapter.active_agent_work_count() == 1
 
+    def test_interrupt_active_runs_interrupts_adapter_owned_agents(self):
+        adapter = APIServerAdapter(PlatformConfig(enabled=True))
+        agent = MagicMock()
+        adapter._active_run_agents = {"run-1": agent}
+
+        assert adapter.interrupt_active_runs("gateway shutdown") == 1
+
+        agent.interrupt.assert_called_once_with("gateway shutdown")
+
 
 class TestDrainWaitsForApiWork:
     @pytest.mark.asyncio
@@ -209,6 +218,17 @@ async def test_drain_times_out_if_api_run_outlives_the_window(self):
 
         assert timed_out is True
 
+    def test_shutdown_interrupt_reaches_api_server_runs(self):
+        runner, _adapter = make_restart_runner()
+        api = APIServerAdapter(PlatformConfig(enabled=True))
+        agent = MagicMock()
+        api._active_run_agents = {"run-1": agent}
+        runner.adapters = {Platform.API_SERVER: api}
+
+        runner._interrupt_running_agents("gateway shutdown")
+
+        agent.interrupt.assert_called_once_with("gateway shutdown")
+
     @pytest.mark.asyncio
     async def test_drain_still_waits_for_chat_cron_and_api_work(self):
         import cron.scheduler as sched