hermes: normalize Codex tool streams

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-deduplicate-tool-stream"
+        ai.bstein.dev/config-rev: "20260812-normalize-tool-stream"
         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

@@ -273,15 +273,10 @@ 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
-    skip_event_data = False
+    pending_event_line = ""
     for line in body.decode("utf-8", errors="replace").splitlines():
-        if line == "event: response.function_call_arguments.done":
+        if line.startswith("event:"):
-            # Switchyard translates both the argument deltas and this terminal
+            pending_event_line = line
-            # snapshot into Chat Completions deltas.  That duplicates the JSON
-            # object and Hermes mistakes the result for provider truncation.
-            # The preceding deltas are complete and already validated by
-            # ``_completed_response``; suppress only the redundant snapshot.
-            skip_event_data = True
             continue
         if line.startswith("data:"):
             value = line[5:].strip()
@@ -289,20 +284,38 @@ def _normalized_stream(body: bytes, completed: dict[str, Any]) -> bytes:
                 event = json.loads(value)
             except (TypeError, ValueError, json.JSONDecodeError):
                 event = None
-            if (
+            event_type = event.get("type") if isinstance(event, dict) else ""
-                skip_event_data
+            item = event.get("item") if isinstance(event, dict) else None
-                or isinstance(event, dict)
+            if event_type == "response.function_call_arguments.done" or (
-                and event.get("type") == "response.function_call_arguments.done"
+                event_type == "response.output_item.done"
+                and isinstance(item, dict)
+                and item.get("type") == "function_call"
             ):
-                skip_event_data = False
+                # Switchyard translates each terminal function-call snapshot
+                # into another Chat Completions argument delta.  Retain the
+                # already-validated delta sequence, not its duplicate copies.
+                pending_event_line = ""
                 continue
-            if isinstance(event, dict) and event.get("type") == "response.completed":
+            if isinstance(event, dict) and event_type == "response.completed":
-                event["response"] = completed
+                stream_completed = dict(completed)
+                stream_completed["output"] = [
+                    output_item
+                    for output_item in stream_completed.get("output") or []
+                    if not isinstance(output_item, dict)
+                    or output_item.get("type") != "function_call"
+                ]
+                event["response"] = stream_completed
                 line = "data: " + json.dumps(event, separators=(",", ":"))
                 replaced_terminal = True
-        elif line:
+            if pending_event_line:
-            skip_event_data = False
+                normalized.append(pending_event_line)
+                pending_event_line = ""
+        elif pending_event_line and line:
+            normalized.append(pending_event_line)
+            pending_event_line = ""
         normalized.append(line)
+    if pending_event_line:
+        normalized.append(pending_event_line)
     if not replaced_terminal:
         raise RuntimeError("Codex stream ended without a completed event")
     # Preserve the blank event terminator required by SSE clients.  Responses

testing/tests/test_hermes_chat_quality.py

@@ -684,6 +684,12 @@ def test_codex_broker_auth_and_request_contract(tmp_path: Path, monkeypatch):
         "output"
     ] == [completed_item]
     assert normalized.endswith("\n\n")
+    streamed_function_item = {
+        "type": "function_call",
+        "name": "read_file",
+        "status": "completed",
+        "arguments": '{"path":"/tmp"}',
+    }
     streamed_function_body = (
         "event: response.function_call_arguments.delta\n"
         'data: {"type":"response.function_call_arguments.delta",'
@@ -691,16 +697,34 @@ def test_codex_broker_auth_and_request_contract(tmp_path: Path, monkeypatch):
         "event: response.function_call_arguments.done\n"
         'data: {"type":"response.function_call_arguments.done",'
         '"item_id":"call_1","arguments":"{\\"path\\":\\"/tmp\\"}"}\n\n'
+        "event: response.output_item.done\n"
+        'data: {"type":"response.output_item.done","output_index":0,'
+        '"item":{"type":"function_call","name":"read_file",'
+        '"arguments":"{\\"path\\":\\"/tmp\\"}"}}\n\n'
         "event: response.completed\n"
         "data: "
-        + json.dumps({"type": "response.completed", "response": completed})
+        + json.dumps(
+            {
+                "type": "response.completed",
+                "response": {**completed, "output": [streamed_function_item]},
+            }
+        )
         + "\n\n"
     ).encode()
     normalized_function_stream = module._normalized_stream(
-        streamed_function_body, completed
+        streamed_function_body, {**completed, "output": [streamed_function_item]}
     ).decode()
     assert "response.function_call_arguments.delta" in normalized_function_stream
     assert "response.function_call_arguments.done" not in normalized_function_stream
+    assert "response.output_item.done" not in normalized_function_stream
+    normalized_terminal = next(
+        line
+        for line in normalized_function_stream.splitlines()
+        if '"response.completed"' in line
+    )
+    assert json.loads(normalized_terminal.removeprefix("data: "))["response"][
+        "output"
+    ] == []
     with pytest.raises(RuntimeError, match="retryable incomplete response"):
         module._completed_response(
             [