diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index b442c0c60f9e..59deed4e7f67 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -4304,6 +4304,11 @@ class APIServerAdapter(BasePlatformAdapter): event_cb = self._make_run_event_callback(run_id, loop) + def _put_event_if_active(event: Optional[Dict]) -> None: + """Enqueue only while this run still owns live transport state.""" + if self._run_streams.get(run_id) is q: + q.put_nowait(event) + # Also wire stream_delta_callback so message.delta events flow through. def _text_cb(delta: Optional[str]) -> None: if delta is None: @@ -4311,7 +4316,7 @@ class APIServerAdapter(BasePlatformAdapter): if run_id not in self._run_streams: return try: - loop.call_soon_threadsafe(q.put_nowait, { + loop.call_soon_threadsafe(_put_event_if_active, { "event": "message.delta", "run_id": run_id, "timestamp": time.time(), @@ -4335,7 +4340,7 @@ class APIServerAdapter(BasePlatformAdapter): try: self._set_run_status(run_id, "running") if run_id in self._stopping_run_ids: - q.put_nowait({ + _put_event_if_active({ "event": "run.cancelled", "run_id": run_id, "timestamp": time.time(), @@ -4431,7 +4436,7 @@ class APIServerAdapter(BasePlatformAdapter): result, usage = await asyncio.get_running_loop().run_in_executor(None, _run_sync) if run_id in self._stopping_run_ids: - q.put_nowait({ + _put_event_if_active({ "event": "run.cancelled", "run_id": run_id, "timestamp": time.time(), @@ -4446,7 +4451,7 @@ class APIServerAdapter(BasePlatformAdapter): # block below never fires — issue #15561). elif isinstance(result, dict) and result.get("failed"): error_msg = _redact_api_error_text(result.get("error") or "agent run failed") - q.put_nowait({ + _put_event_if_active({ "event": "run.failed", "run_id": run_id, "timestamp": time.time(), @@ -4460,7 +4465,7 @@ class APIServerAdapter(BasePlatformAdapter): ) else: final_response = result.get("final_response", "") if isinstance(result, dict) else "" - q.put_nowait({ + _put_event_if_active({ "event": "run.completed", "run_id": run_id, "timestamp": time.time(), @@ -4481,7 +4486,7 @@ class APIServerAdapter(BasePlatformAdapter): last_event="run.cancelled", ) try: - q.put_nowait({ + _put_event_if_active({ "event": "run.cancelled", "run_id": run_id, "timestamp": time.time(), @@ -4498,7 +4503,7 @@ class APIServerAdapter(BasePlatformAdapter): last_event="run.failed", ) try: - q.put_nowait({ + _put_event_if_active({ "event": "run.failed", "run_id": run_id, "timestamp": time.time(), @@ -4520,7 +4525,7 @@ class APIServerAdapter(BasePlatformAdapter): pass # Sentinel: signal SSE stream to close try: - q.put_nowait(None) + _put_event_if_active(None) except Exception: pass self._active_run_agents.pop(run_id, None) diff --git a/tests/gateway/test_api_server_runs.py b/tests/gateway/test_api_server_runs.py index 957dbfac383e..9e3180acce15 100644 --- a/tests/gateway/test_api_server_runs.py +++ b/tests/gateway/test_api_server_runs.py @@ -540,7 +540,11 @@ class TestRunLifecycleSweep: adapter._sweep_orphaned_runs_once(time.time()) before = expired_queue.qsize() stream_delta("must-not-buffer") - await asyncio.sleep(0) + mock_agent.interrupt("finish test") + for _ in range(40): + if run_id not in adapter._active_run_tasks: + break + await asyncio.sleep(0.05) assert expired_queue.qsize() == before diff --git a/website/docs/user-guide/features/api-server.md b/website/docs/user-guide/features/api-server.md index 3bcb053dbe77..cbcb1f954d5b 100644 --- a/website/docs/user-guide/features/api-server.md +++ b/website/docs/user-guide/features/api-server.md @@ -276,9 +276,18 @@ Statuses are retained briefly after terminal states (`completed`, `failed`, or ` Server-Sent Events stream of the run's tool-call progress, token deltas, and lifecycle events. Designed for dashboards and thick clients that want to attach/detach without losing state. +Unconsumed event buffers expire after five minutes so a detached client cannot +grow memory indefinitely. This expires transport state only: a run that is +still executing remains visible to status polling, approval, stop control, and +concurrency accounting until its executor work actually exits. A connected SSE +subscriber continues draining normally. + ### POST /v1/runs/\{run_id\}/stop Interrupt a running agent turn. The endpoint returns immediately with `{"status": "stopping"}` while Hermes asks the active agent to stop at the next safe interruption point. +The run stays tracked as `stopping` until the executor-backed work exits, then +settles as `cancelled`; requesting stop never hides a worker that is still +running. ### POST /v1/runs/\{run_id\}/approval diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/api-server.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/api-server.md index 7e2b0fa84f7a..d2e3ef148143 100644 --- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/api-server.md +++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/api-server.md @@ -270,9 +270,12 @@ Runs 接受简单的 `input` 字符串，以及可选的 `session_id`、`instruc run 的工具调用进度、token 增量和生命周期事件的 Server-Sent Events 流。专为需要附加/分离而不丢失状态的仪表板和厚客户端设计。 +未消费的事件缓冲区会在五分钟后过期，避免已断开的客户端导致内存无限增长。这里只会过期传输状态：仍在执行的 run 会继续保留在状态轮询、审批、停止控制和并发计数中，直到其 executor 工作真正退出。已连接的 SSE 订阅者会继续正常消费事件。 + ### POST /v1/runs/\{run_id\}/stop 中断正在运行的 agent 轮次。端点立即返回 `{"status": "stopping"}`，同时 Hermes 要求活跃 agent 在下一个安全中断点停止。 +run 会保持 `stopping` 并继续被跟踪，直到 executor 支持的工作退出，然后进入 `cancelled`；停止请求不会隐藏仍在运行的 worker。 ## Jobs API（后台计划任务）