Merge pull request #73493 from afourniernv/fix/relay-provider-stream-error-preservation

fix(relay): preserve provider errors and scope cleanup

agent/relay_llm.py

@@ -438,6 +438,11 @@ class ManagedLlmStream(Iterator[Any]):
                 run_callback(self._on_chunk, _jsonable(chunk))
 
         def relay_finalizer() -> Any:
+            # Relay can invoke the finalizer while unwinding a provider-stream
+            # failure. Preserve that original callback error instead of
+            # replacing it with a secondary "missing terminal response" error.
+            if self._callback_error is not None:
+                return None
             try:
                 if self.final_response is not None:
                     return _jsonable(self.final_response)

agent/relay_runtime.py

@@ -722,7 +722,8 @@ class RelaySessionCoordinator:
         with turn.logical_llm_lock:
             logical_calls = list(turn.logical_llm_calls.items())
             turn.logical_llm_calls.clear()
-        for request_id, logical_handle in logical_calls:
+        for index in range(len(logical_calls) - 1, -1, -1):
+            request_id, logical_handle = logical_calls[index]
             try:
                 lease.host.run_in_session(
                     lease.session,
@@ -736,11 +737,21 @@ class RelaySessionCoordinator:
                 )
             except Exception:
                 with turn.logical_llm_lock:
-                    turn.logical_llm_calls.setdefault(request_id, logical_handle)
+                    # Relay scopes are stack-owned. If the newest remaining
+                    # handle cannot close, older handles cannot close safely
+                    # either, so retain the unclosed prefix for diagnostics.
+                    for pending_request_id, pending_handle in logical_calls[
+                        : index + 1
+                    ]:
+                        turn.logical_llm_calls.setdefault(
+                            pending_request_id,
+                            pending_handle,
+                        )
                 logger.warning(
                     "Hermes Relay logical LLM finalization failed",
                     exc_info=True,
                 )
+                break
 
     @staticmethod
     def _reset_turn_context(turn: RelayTurnContext) -> None:

tests/agent/test_relay_llm.py

@@ -181,6 +181,49 @@ def test_deferred_stream_preserves_provider_error_and_logical_scope_for_retry(
     assert "request-2" in turn.logical_llm_calls
 
 
+def test_stream_provider_error_is_not_replaced_by_finalizer_error(relay_turn):
+    _relay, turn = relay_turn
+
+    class ProviderError(Exception):
+        pass
+
+    provider_error = ProviderError("provider failed before first chunk")
+    finalizer_called = False
+
+    def failing_stream(_request):
+        def generate():
+            raise provider_error
+            yield  # pragma: no cover
+
+        return generate()
+
+    def failing_finalizer():
+        nonlocal finalizer_called
+        finalizer_called = True
+        raise RuntimeError("missing terminal response")
+
+    stream = relay_llm.stream(
+        {"model": "test-model", "input": "hi"},
+        failing_stream,
+        session_id="session-1",
+        name="test-provider",
+        model_name="test-model",
+        finalizer=failing_finalizer,
+        metadata={
+            "api_mode": "codex_responses",
+            "api_request_id": "request-provider-before-finalizer",
+        },
+        defer_logical_completion=True,
+    )
+
+    with pytest.raises(ProviderError) as caught:
+        list(stream)
+
+    assert caught.value is provider_error
+    assert finalizer_called is False
+    assert "request-provider-before-finalizer" in turn.logical_llm_calls
+
+
 def test_non_deferred_partial_stream_close_cancels_logical_call(
     relay_turn,
     monkeypatch,

tests/hermes_cli/test_relay_shared_metrics_runtime.py

@@ -1525,6 +1525,91 @@ def test_turn_cleanup_drains_logical_calls_after_turn_scope_start_failure(
     assert turn.logical_llm_calls == {}
 
 
+def test_turn_cleanup_drains_logical_calls_in_lifo_order(direct_runtime):
+    coordinator = relay_runtime.SESSION_COORDINATOR
+    profile_key = relay_runtime.current_profile_key()
+    lease = coordinator.acquire_conversation(
+        profile_key=profile_key,
+        session_id="session-lifo",
+        platform="cli",
+    )
+    assert lease.session is not None
+    turn = coordinator.begin_turn(
+        lease,
+        turn_id="turn-lifo",
+        task_id="task-lifo",
+    )
+    assert turn.handle is not None
+    runtime = lease.host
+
+    handles = []
+    for request_id in ("request-1", "request-2"):
+        handle = runtime.run_in_session(
+            lease.session,
+            runtime.relay.scope.push,
+            relay_runtime.LOGICAL_LLM_SCOPE,
+            runtime.relay.ScopeType.Function,
+            handle=turn.handle,
+            input={},
+        )
+        turn.logical_llm_calls[request_id] = handle
+        handles.append(handle)
+
+    coordinator.end_turn(turn, outcome="failed")
+    coordinator.release_conversation(lease)
+
+    logical_closes = [
+        event[1]
+        for event in direct_runtime.events
+        if event[0] == "scope.pop" and event[1] in handles
+    ]
+    assert logical_closes == list(reversed(handles))
+    assert turn.logical_llm_calls == {}
+
+
+def test_real_binding_drains_multiple_logical_calls_before_turn_close(
+    real_binding_runtime,
+    caplog,
+):
+    coordinator = relay_runtime.SESSION_COORDINATOR
+    profile_key = relay_runtime.current_profile_key()
+    lease = coordinator.acquire_conversation(
+        profile_key=profile_key,
+        session_id="session-native-lifo",
+        platform="cli",
+    )
+    assert lease.session is not None
+    turn = coordinator.begin_turn(
+        lease,
+        turn_id="turn-native-lifo",
+        task_id="task-native-lifo",
+    )
+    assert turn.handle is not None
+    runtime = lease.host
+
+    for request_id in ("request-1", "request-2"):
+        handle = runtime.run_in_session(
+            lease.session,
+            runtime.relay.scope.push,
+            relay_runtime.LOGICAL_LLM_SCOPE,
+            runtime.relay.ScopeType.Function,
+            handle=turn.handle,
+            input={},
+        )
+        turn.logical_llm_calls[request_id] = handle
+
+    coordinator.end_turn(turn, outcome="failed")
+    coordinator.release_conversation(lease)
+    coordinator.finalize_conversation(
+        profile_key=profile_key,
+        session_id=lease.session_id,
+    )
+
+    assert turn.logical_llm_calls == {}
+    assert "finalization failed" not in caplog.text
+    assert "closed with errors" not in caplog.text
+
+
 def test_shared_metrics_creates_one_task_under_concurrent_access(direct_runtime):
     runtime = relay_shared_metrics._get_runtime()
     assert runtime is not None