From 41bfdc6e523ced1e834cf0ae636833c15eaac4de Mon Sep 17 00:00:00 2001 From: jenkins Date: Wed, 12 Aug 2026 10:41:47 -0300 Subject: [PATCH] hermes: recover interrupted provider streams --- services/hermes/agent-deployment.yaml | 2 +- services/hermes/chat-statefulset.yaml | 25 ++++- services/hermes/kustomization.yaml | 1 + .../hermes/scripts/patch_codex_runtime.py | 105 ++++++++++++++++++ .../hermes/scripts/patch_stream_recovery.py | 23 ++++ testing/tests/test_hermes_chat_quality.py | 18 ++- testing/tests/test_hermes_cli_lanes.py | 6 +- 7 files changed, 175 insertions(+), 5 deletions(-) create mode 100644 services/hermes/scripts/patch_stream_recovery.py diff --git a/services/hermes/agent-deployment.yaml b/services/hermes/agent-deployment.yaml index 067686af3..0e6b2687e 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-normalize-tool-stream" + ai.bstein.dev/config-rev: "20260812-stream-recovery" 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/chat-statefulset.yaml b/services/hermes/chat-statefulset.yaml index f2be6440a..63cd15396 100644 --- a/services/hermes/chat-statefulset.yaml +++ b/services/hermes/chat-statefulset.yaml @@ -29,7 +29,7 @@ spec: ai.bstein.dev/router-wire-contract: ollama-numeric-keepalive ai.bstein.dev/isolation: one Hermes process and PVC per Keycloak subject ai.bstein.dev/model-policy: uniform automatic policy with per-user overrides - ai.bstein.dev/config-rev: "20260812-routing-continuity" + ai.bstein.dev/config-rev: "20260812-stream-recovery" vault.hashicorp.com/agent-inject: "true" vault.hashicorp.com/role: hermes-chat vault.hashicorp.com/agent-inject-secret-anthropic-token: kv/data/atlas/hermes/agent-tokens @@ -177,6 +177,26 @@ spec: resources: requests: {cpu: 25m, memory: 64Mi} limits: {cpu: 100m, memory: 128Mi} + - name: patch-stream-recovery + image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 + imagePullPolicy: IfNotPresent + command: + - /opt/hermes/.venv/bin/python + - /opt/coordinator/patch_stream_recovery.py + - /opt/hermes/agent/conversation_loop.py + - /patched/conversation_loop.py + securityContext: + allowPrivilegeEscalation: false + runAsUser: 10000 + runAsGroup: 10000 + seccompProfile: + type: RuntimeDefault + volumeMounts: + - {name: coordinator, mountPath: /opt/coordinator, readOnly: true} + - {name: stream-recovery-patch, mountPath: /patched} + resources: + requests: {cpu: 25m, memory: 64Mi} + limits: {cpu: 100m, memory: 128Mi} containers: - name: hermes image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 @@ -217,6 +237,7 @@ spec: # their own PVCs; only provider credentials are shared here. - {name: provider-auth, mountPath: /shared-auth} - {name: auth-patch, mountPath: /opt/hermes/hermes_cli/auth.py, subPath: auth.py} + - {name: stream-recovery-patch, mountPath: /opt/hermes/agent/conversation_loop.py, subPath: conversation_loop.py} - {name: image-plugin, mountPath: /opt/hermes/plugins/image_gen/atlas-broker, readOnly: true} - {name: auto-router-plugin, mountPath: /opt/data/plugins/auto-router, readOnly: true} readinessProbe: @@ -319,6 +340,8 @@ spec: defaultMode: 0555 - name: auth-patch emptyDir: {} + - name: stream-recovery-patch + emptyDir: {} - name: auto-router-plugin configMap: name: hermes-auto-router-plugin diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml index ac58d26d1..d6f4ecd86 100644 --- a/services/hermes/kustomization.yaml +++ b/services/hermes/kustomization.yaml @@ -72,6 +72,7 @@ configMapGenerator: - migrate_herdr_state.py=scripts/migrate_herdr_state.py - patch_hermes_auth.py=scripts/patch_hermes_auth.py - patch_codex_runtime.py=scripts/patch_codex_runtime.py + - patch_stream_recovery.py=scripts/patch_stream_recovery.py - patch_tui_gateway.py=scripts/patch_tui_gateway.py - patch_ttyd_index.py=scripts/patch_ttyd_index.py - routing_catalog.py=scripts/routing_catalog.py diff --git a/services/hermes/scripts/patch_codex_runtime.py b/services/hermes/scripts/patch_codex_runtime.py index 527f36fb3..d377114db 100644 --- a/services/hermes/scripts/patch_codex_runtime.py +++ b/services/hermes/scripts/patch_codex_runtime.py @@ -202,6 +202,90 @@ RETRY_FALLBACK_DISPATCH_AFTER = ''' while retry_count < max_retries: # ── Nous Portal rate limit guard ────────────────────── ''' +STREAM_RECOVERY_BEFORE = ''' # If we have prior messages, roll back to last complete state + if len(messages) > 1: + agent._vprint(f"{agent.log_prefix} ⏪ Rolling back to last complete assistant turn") + rolled_back_messages = agent._get_messages_up_to_last_assistant(messages) + + agent._cleanup_task_resources(effective_task_id) + agent._persist_session(messages, conversation_history) + + return { + "final_response": "Response truncated due to output length limit", + "messages": rolled_back_messages, + "api_calls": api_call_count, + "completed": False, + "partial": True, + "error": "Response truncated due to output length limit" + } + else: + # First message was truncated - mark as failed + agent._flush_status_buffer() + agent._vprint(f"{agent.log_prefix}❌ First response truncated - cannot recover", force=True) + agent._persist_session(messages, conversation_history) + return { + "final_response": "First response truncated due to output length limit", + "messages": messages, + "api_calls": api_call_count, + "completed": False, + "failed": True, + "error": "First response truncated due to output length limit" + } +''' +STREAM_RECOVERY_AFTER = ''' # A partial-stream stub is a transport failure, not proof that + # the model exhausted its output budget. Retry the unchanged + # turn through the routed alias so Switchyard can choose a + # healthy provider. Never append the incomplete response. + _is_transport_stub = ( + getattr(response, "id", "") == PARTIAL_STREAM_STUB_ID + ) + if _is_transport_stub and truncated_tool_call_retries < 4: + truncated_tool_call_retries += 1 + agent._buffer_vprint( + f"⚠️ Provider stream ended before a complete response — " + f"rerouting ({truncated_tool_call_retries}/4)..." + ) + agent._session_messages = messages + continue + + _incomplete_error = ( + "Provider stream repeatedly ended before a complete response" + if _is_transport_stub + else "Response truncated due to output length limit" + ) + # If we have prior messages, roll back to last complete state. + if len(messages) > 1: + agent._vprint(f"{agent.log_prefix} ⏪ Rolling back to last complete assistant turn") + rolled_back_messages = agent._get_messages_up_to_last_assistant(messages) + + agent._cleanup_task_resources(effective_task_id) + agent._persist_session(messages, conversation_history) + + return { + "final_response": _incomplete_error, + "messages": rolled_back_messages, + "api_calls": api_call_count, + "completed": False, + "partial": True, + "error": _incomplete_error, + } + + agent._flush_status_buffer() + agent._vprint( + f"{agent.log_prefix}❌ First response did not complete", + force=True, + ) + agent._persist_session(messages, conversation_history) + return { + "final_response": _incomplete_error, + "messages": messages, + "api_calls": api_call_count, + "completed": False, + "failed": True, + "error": _incomplete_error, + } +''' + FALLBACK_CONTEXT_BEFORE = '''def run_codex_app_server_turn( ''' FALLBACK_CONTEXT_AFTER = '''def build_cross_provider_codex_prompt( @@ -384,10 +468,31 @@ def patch_loop(source: Path, destination: Path) -> None: RETRY_FALLBACK_DISPATCH_AFTER, "Codex retry-loop fallback dispatch", ) + content = patch_stream_recovery_text(content) destination.parent.mkdir(parents=True, exist_ok=True) destination.write_text(content, encoding="utf-8") +def patch_stream_recovery_text(content: str) -> str: + """Retry transport stubs and keep output-limit reporting truthful.""" + return _replace_once( + content, + STREAM_RECOVERY_BEFORE, + STREAM_RECOVERY_AFTER, + "partial stream recovery", + ) + + +def patch_stream_recovery(source: Path, destination: Path) -> None: + """Patch only shared stream recovery for non-Codex Hermes lanes.""" + content = source.read_text(encoding="utf-8") + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text( + patch_stream_recovery_text(content), + encoding="utf-8", + ) + + def patch_auxiliary(source: Path, destination: Path) -> None: """Let auxiliary tasks reuse the current Codex CLI access token safely.""" content = source.read_text(encoding="utf-8") diff --git a/services/hermes/scripts/patch_stream_recovery.py b/services/hermes/scripts/patch_stream_recovery.py new file mode 100644 index 000000000..8df63e839 --- /dev/null +++ b/services/hermes/scripts/patch_stream_recovery.py @@ -0,0 +1,23 @@ +#!/usr/bin/env python3 +"""Add bounded provider-stream recovery to the Hermes conversation loop.""" + +from __future__ import annotations + +import argparse +from pathlib import Path + +from patch_codex_runtime import patch_stream_recovery + + +def main() -> int: + """Patch one conversation loop without changing provider adapters.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("source", type=Path) + parser.add_argument("destination", type=Path) + args = parser.parse_args() + patch_stream_recovery(args.source, args.destination) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/testing/tests/test_hermes_chat_quality.py b/testing/tests/test_hermes_chat_quality.py index de3abd328..2ae6130d5 100644 --- a/testing/tests/test_hermes_chat_quality.py +++ b/testing/tests/test_hermes_chat_quality.py @@ -540,12 +540,26 @@ def test_chat_reasoning_uses_switchyard_without_owner_credentials(): statefulset = _documents(HERMES / "chat-statefulset.yaml")[0] assert statefulset["spec"]["template"]["metadata"]["annotations"][ "ai.bstein.dev/config-rev" - ] == "20260812-routing-continuity" + ] == "20260812-stream-recovery" + pod_spec = statefulset["spec"]["template"]["spec"] + patch_init = next( + item for item in pod_spec["initContainers"] + if item["name"] == "patch-stream-recovery" + ) + assert patch_init["command"][-2:] == [ + "/opt/hermes/agent/conversation_loop.py", + "/patched/conversation_loop.py", + ] hermes = next( item - for item in statefulset["spec"]["template"]["spec"]["containers"] + for item in pod_spec["containers"] if item["name"] == "hermes" ) + assert { + "name": "stream-recovery-patch", + "mountPath": "/opt/hermes/agent/conversation_loop.py", + "subPath": "conversation_loop.py", + } in hermes["volumeMounts"] assert not any( mount["mountPath"].endswith("/.codex") for mount in hermes["volumeMounts"] diff --git a/testing/tests/test_hermes_cli_lanes.py b/testing/tests/test_hermes_cli_lanes.py index 3683ea3d0..0d899925e 100644 --- a/testing/tests/test_hermes_cli_lanes.py +++ b/testing/tests/test_hermes_cli_lanes.py @@ -1051,7 +1051,8 @@ def test_codex_runtime_patch_uses_cli_and_forwards_route(tmp_path: Path): loop = tmp_path / "conversation_loop.py" loop.write_text( codex_runtime_patch.FALLBACK_DISPATCH_BEFORE - + codex_runtime_patch.RETRY_FALLBACK_DISPATCH_BEFORE, + + codex_runtime_patch.RETRY_FALLBACK_DISPATCH_BEFORE + + codex_runtime_patch.STREAM_RECOVERY_BEFORE, encoding="utf-8", ) loop_out = tmp_path / "patched/conversation_loop.py" @@ -1064,6 +1065,9 @@ def test_codex_runtime_patch_uses_cli_and_forwards_route(tmp_path: Path): ) api_kwargs = loop_content.find("agent._build_api_kwargs", retry_dispatch) assert api_kwargs == -1 or retry_dispatch < api_kwargs + assert "Provider stream ended before a complete response" in loop_content + assert "_is_transport_stub" in loop_content + assert "rerouting ({truncated_tool_call_retries}/4)" in loop_content auxiliary = tmp_path / "auxiliary_client.py" auxiliary.write_text(