From 4d91c75e667df83338b00236a7393c9937f1526e Mon Sep 17 00:00:00 2001 From: jenkins Date: Wed, 12 Aug 2026 07:55:35 -0300 Subject: [PATCH] hermes: normalize Codex tool streams --- services/hermes/agent-deployment.yaml | 2 +- services/hermes/scripts/codex_broker.py | 47 +++++++++++++++-------- testing/tests/test_hermes_chat_quality.py | 28 +++++++++++++- 3 files changed, 57 insertions(+), 20 deletions(-) diff --git a/services/hermes/agent-deployment.yaml b/services/hermes/agent-deployment.yaml index 4e94c9671..a6cd3877e 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-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 diff --git a/services/hermes/scripts/codex_broker.py b/services/hermes/scripts/codex_broker.py index 2d950547b..f998adb96 100644 --- a/services/hermes/scripts/codex_broker.py +++ b/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": - # Switchyard translates both the argument deltas and this terminal - # 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 + if line.startswith("event:"): + pending_event_line = line 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 ( - skip_event_data - or isinstance(event, dict) - and event.get("type") == "response.function_call_arguments.done" + event_type = event.get("type") if isinstance(event, dict) else "" + item = event.get("item") if isinstance(event, dict) else None + if event_type == "response.function_call_arguments.done" or ( + 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": - event["response"] = completed + if isinstance(event, dict) and event_type == "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: - skip_event_data = False + if pending_event_line: + 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 diff --git a/testing/tests/test_hermes_chat_quality.py b/testing/tests/test_hermes_chat_quality.py index c85d06866..3b8d629f1 100644 --- a/testing/tests/test_hermes_chat_quality.py +++ b/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( [