hermes: recover interrupted provider 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-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

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

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

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

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

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"]

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(