# NousResearch/hermes-agent · SSE stream termination race condition fixes ## Open now - Nothing open on GitHub. These issues and pull requests address a race condition where SSE streams fail to receive a completion signal when an agent task finishes. Multiple fix approaches involve sending a sentinel or waking the SSE writer upon task completion. ## TL;DR - 4 member(s): 0 open issue(s), 1 closed issue(s), 0 open PR(s), 1 merged PR(s), 2 PR(s) closed without merging. - A fix was merged to main in PR #24578. Verify it addresses your case. - **Recommended:** read PR #24578 (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-24578`. 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 - #24451: SSE streaming: response.completed never sent due to race condition with agent_task.done() ## Prior & competing PRs - **#24554: fix(api_server): push None sentinel via done callback to end SSE loop**. closed without merge · fix tier `suggested` - maintainer direction: category `superseded_by_pr`; implication `superseded`; refs `#24464`, `#24451` - _suggested-tier fix detail withheld; re-request with `?tier=all` to include it._ - **#24578: fix(gateway): enqueue SSE EOS sentinel on task completion**. merged to main · fix tier `verified` - maintainer direction: category `merged` - files: `gateway/platforms/api_server.py`, `tests/gateway/test_api_server.py` - fix hunks: - `gateway/platforms/api_server.py` ```diff @@ -1168,6 +1168,9 @@ def _on_tool_complete(tool_call_id, function_name, function_args, function_resul agent_ref=agent_ref, gateway_session_key=gateway_session_key, )) + # Ensure SSE drain loops can terminate without relying on polling + # agent_task.done(), which can race with queue timeout checks. + agent_task.add_done_callback(lambda _fut: _stream_q.put(None)) return await self._write_sse_chat_completion( request, completion_id, model_name, created, _stream_q, ``` ```diff @@ -2197,6 +2200,9 @@ def _on_tool_complete(tool_call_id, function_name, function_args, function_resul agent_ref=agent_ref, gateway_session_key=gateway_session_key, )) + # Ensure SSE drain loops can terminate without relying on polling + # agent_task.done(), which can race with queue timeout checks. + agent_task.add_done_callback(lambda _fut: _stream_q.put(None)) response_id = f"resp_{uuid.uuid4().hex[:28]}" model_name = body.get("model", self._model_name) ``` - `tests/gateway/test_api_server.py` ```diff @@ -681,6 +681,56 @@ async def _mock_run_agent(**kwargs): assert "[DONE]" in body assert "Hello!" in body + @pytest.mark.asyncio + async def test_stream_task_done_callback_enqueues_eos_for_chat_completions(self, adapter): + """Regression guard for #24451: completion callback must signal SSE EOS.""" + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + class _FakeTask: + def __init__(self): + self.callbacks = [] + + def add_done_callback(self, cb): + self.callbacks.append(cb) + + fake_task = _FakeTask() + + def _fake_ensure_future(coro): + # We short-circuit task scheduling in this unit test. + coro.close() + return fake_task + + with ( + patch.object( + adapter, + "_run_agent", + new=AsyncMock( + return_value=( + {"final_response": "ok", "messages": [], "api_calls": 1}, + {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}, + ) + ), + ), + patch("gateway.platforms.api_server.asyncio.ensure_future", side_effect=_fake_ensure_future), + patch.object(adapter, "_write_sse_chat_completion", new_callable=AsyncMock) as mock_write_sse, + ): + mock_write_sse.return_value = web.Response(status=200, text="ok") + resp = await cli.post( + "/v1/chat/completions", + json={ + "model": "test", + "messages": [{"role": "user", "content": "hi"}], + "stream": True, + }, + ) + assert resp.status == 200 + + assert len(fake_task.callbacks) == 1 + stream_q = mock_write_sse.call_args.args[4] + assert stream_q.empty() + fake_task.callbacks[0](fake_task) + assert stream_q.get_nowait() is None + @pytest.mark.asyncio async def test_stream_sends_keepalive_during_quiet_tool_gap(self, adapter): """Idle SSE streams should send keepalive comments while tools run silently.""" ``` ```diff @@ -1676,6 +1726,52 @@ async def _mock_run_agent(**kwargs): assert "Hello" in body assert " world" in body + @pytest.mark.asyncio + async def test_stream_task_done_callback_enqueues_eos_for_responses(self, adapter): + """Regression guard for #24451 on /v1/responses streaming path.""" + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + class _FakeTask: + def __init__(self): + self.callbacks = [] + + def add_done_callback(self, cb): + self.callbacks.append(cb) + + fake_task = _FakeTask() + + def _fake_ensure_future(coro): + # We short-circuit task scheduling in this unit test. + coro.close() + return fake_task + + with ( + patch.object( + adapter, + "_run_agent", + new=AsyncMock( + return_value=( + {"final_response": "ok", "messages": [], "api_calls": 1}, + {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}, + ) + ), + ), + patch("gateway.platforms.api_server.asyncio.ensure_future", side_effect=_fake_ensure_future), + patch.object(adapter, "_write_sse_responses", new_callable=AsyncMock) as mock_write_sse, + ): + mock_write_sse.return_value = web.Response(status=200, text="ok") + resp = await cli.post( + "/v1/responses", + json={"model": "hermes-agent", "input": "hi", "stream": True}, + ) + assert resp.status == 200 + + assert len(fake_task.callbacks) == 1 + stream_q = mock_write_sse.call_args.kwargs["stream_q"] + assert stream_q.empty() + fake_task.callbacks[0](fake_task) + assert stream_q.get_nowait() is None + @pytest.mark.asyncio async def test_stream_emits_function_call_and_output_items(self, adapter): app = _create_app(adapter) ``` ```diff @@ -3061,4 +3157,3 @@ async def test_capabilities_advertises_session_key_header(self, adapter): assert resp.status == 200 data = await resp.json() assert data["features"]["session_key_header"] == "X-Hermes-Session-Key" - ``` - **#24464: fix(api-server): wake SSE writer immediately on agent_task completion (#24451)**. closed without merge - maintainer direction: category `superseded_by_pr`; implication `superseded`; refs `#24578` ## Fix status - Merged to main in PR #24578 (cited above). No release data is tracked. Verify downstream. - Fix last verified (merged) `2026-05-12`.