diff --git a/agent/relay_llm.py b/agent/relay_llm.py index 4508fce94144..3481ca072376 100644 --- a/agent/relay_llm.py +++ b/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) diff --git a/agent/relay_runtime.py b/agent/relay_runtime.py index 4e3654cf7780..533604791a86 100644 --- a/agent/relay_runtime.py +++ b/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: diff --git a/tests/agent/test_relay_llm.py b/tests/agent/test_relay_llm.py index ae4eac15826a..2e09923f327e 100644 --- a/tests/agent/test_relay_llm.py +++ b/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, diff --git a/tests/hermes_cli/test_relay_shared_metrics_runtime.py b/tests/hermes_cli/test_relay_shared_metrics_runtime.py index 23258def3385..ccda2abc1f4c 100644 --- a/tests/hermes_cli/test_relay_shared_metrics_runtime.py +++ b/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