diff --git a/services/hermes/agent-deployment.yaml b/services/hermes/agent-deployment.yaml index 578fde802..e770bb3b1 100644 --- a/services/hermes/agent-deployment.yaml +++ b/services/hermes/agent-deployment.yaml @@ -25,7 +25,7 @@ spec: ai.bstein.dev/execution: Hermes Kanban with durable direct Codex and Claude Code CLI workers ai.bstein.dev/model-policy: Jetson-assisted AUTO routing, low through xhigh, cross-provider fallback ai.bstein.dev/placement: rpi5 preferred; Jetson deferred until state storage is available - ai.bstein.dev/config-rev: "20260812-codex-incomplete-failover" + ai.bstein.dev/config-rev: "20260812-codex-stream-terminal" vault.hashicorp.com/agent-inject: "true" vault.hashicorp.com/role: hermes-agent vault.hashicorp.com/agent-inject-secret-anthropic-token: kv/data/atlas/hermes/agent-tokens diff --git a/services/hermes/scripts/codex_broker.py b/services/hermes/scripts/codex_broker.py index 970f3caf9..15e2834b8 100644 --- a/services/hermes/scripts/codex_broker.py +++ b/services/hermes/scripts/codex_broker.py @@ -216,6 +216,30 @@ def _completed_response(lines: Iterable[str]) -> dict[str, Any]: raise RuntimeError(upstream_error or "Codex stream ended without a terminal response") +def _normalized_stream(body: bytes, completed: dict[str, Any]) -> bytes: + """Return Responses SSE with the reconstructed terminal response attached.""" + normalized: list[str] = [] + replaced_terminal = False + for line in body.decode("utf-8", errors="replace").splitlines(): + if line.startswith("data:"): + value = line[5:].strip() + try: + event = json.loads(value) + except (TypeError, ValueError, json.JSONDecodeError): + event = None + if isinstance(event, dict) and event.get("type") == "response.completed": + event["response"] = completed + line = "data: " + json.dumps(event, separators=(",", ":")) + replaced_terminal = True + normalized.append(line) + if not replaced_terminal: + raise RuntimeError("Codex stream ended without a completed event") + # Preserve the blank event terminator required by SSE clients. Responses + # streams end at response.completed; they do not require a Chat + # Completions-style [DONE] sentinel. + return ("\n".join(normalized).rstrip("\n") + "\n\n").encode("utf-8") + + class Handler(BaseHTTPRequestHandler): """Authenticated streaming proxy; request bodies and tokens are never logged.""" @@ -319,6 +343,8 @@ class Handler(BaseHTTPRequestHandler): self._json(200, completed) return + body = _normalized_stream(body, completed) + self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.send_header("Content-Length", str(len(body))) diff --git a/testing/tests/test_hermes_chat_quality.py b/testing/tests/test_hermes_chat_quality.py index 0acdcf495..7e965695f 100644 --- a/testing/tests/test_hermes_chat_quality.py +++ b/testing/tests/test_hermes_chat_quality.py @@ -660,6 +660,30 @@ def test_codex_broker_auth_and_request_contract(tmp_path: Path, monkeypatch): "data: [DONE]", ] )["output"] == [completed_item] + raw_stream = ( + "event: response.output_item.done\n" + "data: " + + json.dumps( + { + "type": "response.output_item.done", + "output_index": 0, + "item": completed_item, + } + ) + + "\n\nevent: response.completed\ndata: " + + json.dumps({"type": "response.completed", "response": completed}) + + "\n\n" + ).encode() + normalized = module._normalized_stream( + raw_stream, {**completed, "output": [completed_item]} + ).decode() + terminal_data = next( + line for line in normalized.splitlines() if '"response.completed"' in line + ) + assert json.loads(terminal_data.removeprefix("data: "))["response"][ + "output" + ] == [completed_item] + assert normalized.endswith("\n\n") with pytest.raises(RuntimeError, match="retryable incomplete response"): module._completed_response( [