hermes: normalize Codex stream completion

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

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)))

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(
             [