"""Tests for the auxiliary forward-progress streaming layer. Slow summary models must not be punished like hung ones (#see PR): when a forward-progress hook is installed (context compression), the primary auxiliary call streams and ticks the hook only for substantive payloads, so outer watchdogs (gateway session hygiene) can extend their deadline on liveness. Without a hook, behavior is byte-for-byte the old non-streaming call. """ import threading import time from types import SimpleNamespace from unittest.mock import MagicMock, patch import pytest from agent.auxiliary_client import ( _AnthropicCompletionsAdapter, _ChatStreamAccumulator, _CodexCompletionsAdapter, _acreate_with_stream, _aggregate_chat_stream, _aggregate_chat_stream_async, _anthropic_event_has_content, _aux_stream_total_ceiling, _codex_event_has_content, _create_with_progress, _notify_aux_progress, _provider_requires_stream, aux_progress_hook, ) from agent.conversation_compression import CompressionCommitFence # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _chunk(content=None, reasoning=None, reasoning_details=None, finish_reason=None, usage=None, tool_calls=None, model="m1", chunk_id="c1"): delta = SimpleNamespace( content=content, reasoning=reasoning, reasoning_content=None, reasoning_details=reasoning_details, tool_calls=tool_calls, ) choice = SimpleNamespace(delta=delta, finish_reason=finish_reason) return SimpleNamespace( id=chunk_id, model=model, choices=[choice], usage=usage, ) class _FakeClient: """OpenAI-shaped client whose create() returns a canned value or stream.""" def __init__(self, response=None, stream_chunks=None, stream_error=None): self.calls = [] self._response = response self._stream_chunks = stream_chunks self._stream_error = stream_error completions = SimpleNamespace(create=self._create) self.chat = SimpleNamespace(completions=completions) def _create(self, **kwargs): self.calls.append(kwargs) if kwargs.get("stream"): if self._stream_error is not None: raise self._stream_error return iter(self._stream_chunks or []) return self._response _COMPLETE = SimpleNamespace( id="r1", model="m1", object="chat.completion", choices=[SimpleNamespace( index=0, message=SimpleNamespace(role="assistant", content="non-streamed"), finish_reason="stop", )], usage=None, ) # --------------------------------------------------------------------------- # aux_progress_hook plumbing # --------------------------------------------------------------------------- class TestAuxProgressHook: def test_hook_installed_and_restored(self): ticks = [] with aux_progress_hook(lambda: ticks.append(1)): _notify_aux_progress() _notify_aux_progress() # outside — must not tick assert ticks == [1] def test_hook_is_thread_local(self): ticks = [] seen_in_thread = [] def _other_thread(): # No hook installed on this thread. _notify_aux_progress() seen_in_thread.append(len(ticks)) with aux_progress_hook(lambda: ticks.append(1)): t = threading.Thread(target=_other_thread) t.start() t.join() assert seen_in_thread == [0] # --------------------------------------------------------------------------- # _create_with_progress # --------------------------------------------------------------------------- class TestCreateWithProgress: def test_hook_upgrades_to_streaming_and_ticks_only_for_payload(self): empty_tool_call = SimpleNamespace( index=0, id=None, function=SimpleNamespace(name=None, arguments=""), ) chunks = [ SimpleNamespace(id=None, model=None, choices=[], usage=None), _chunk(content="", reasoning="", tool_calls=[empty_tool_call]), _chunk(reasoning="thinking..."), _chunk(content="Hello "), _chunk(content="world", finish_reason="stop", usage=SimpleNamespace(prompt_tokens=5, completion_tokens=2, total_tokens=7)), ] client = _FakeClient(stream_chunks=chunks) ticks = [] with aux_progress_hook(lambda: ticks.append(1)): result = _create_with_progress( client, {"model": "m1", "messages": [], "timeout": 30}, ) assert client.calls[0]["stream"] is True assert result.choices[0].message.content == "Hello world" assert result.choices[0].message.reasoning == "thinking..." assert result.choices[0].finish_reason == "stop" assert result.usage.total_tokens == 7 # 1 dispatch tick (preserved for the watchdog's historical liveness # signal — see _create_with_progress) + 1 per substantive chunk. assert ticks == [1, 1, 1, 1] def test_completed_response_ticks_only_terminal_signals(self): calls = [] def _create(**kwargs): calls.append(kwargs) return _COMPLETE client = SimpleNamespace( chat=SimpleNamespace(completions=SimpleNamespace(create=_create)) ) ticks = [] with aux_progress_hook(lambda: ticks.append(1)): result = _create_with_progress(client, {"model": "m1", "messages": []}) assert calls[0]["stream"] is True assert result is _COMPLETE # A completed response object carries the full summary payload, and # the dispatch tick is the watchdog's historical liveness signal: # both are one-shot terminal ticks, not per-frame keepalives, so # neither can defeat an inactivity timeout. assert ticks == [1, 1] def test_streaming_rejected_falls_back_to_plain_call(self): client = _FakeClient( response=_COMPLETE, stream_error=RuntimeError("stream is not supported by this model"), ) with aux_progress_hook(lambda: None): result = _create_with_progress( client, {"model": "m1", "messages": []}, ) assert result is _COMPLETE # streamed attempt + non-streaming fallback assert len(client.calls) == 2 assert client.calls[0].get("stream") is True assert "stream" not in client.calls[1] # --------------------------------------------------------------------------- # _aggregate_chat_stream # --------------------------------------------------------------------------- class TestAggregateChatStream: def test_tool_call_deltas_are_reassembled(self): tc0 = SimpleNamespace( index=0, id="call_1", function=SimpleNamespace(name="do_thing", arguments='{"a"'), ) tc1 = SimpleNamespace( index=0, id=None, function=SimpleNamespace(name=None, arguments=': 1}'), ) chunks = [ _chunk(tool_calls=[tc0]), _chunk(tool_calls=[tc1], finish_reason="tool_calls"), ] result = _aggregate_chat_stream(iter(chunks)) tool_calls = result.choices[0].message.tool_calls assert len(tool_calls) == 1 assert tool_calls[0].id == "call_1" assert tool_calls[0].function.name == "do_thing" assert tool_calls[0].function.arguments == '{"a": 1}' assert result.choices[0].finish_reason == "tool_calls" def test_stream_close_is_called(self): closed = [] class _Stream: def __iter__(self): return iter([_chunk(content="ok", finish_reason="stop")]) def close(self): closed.append(True) result = _aggregate_chat_stream(_Stream()) assert result.choices[0].message.content == "ok" assert closed == [True] # --------------------------------------------------------------------------- # Content-bearing progress classification # --------------------------------------------------------------------------- class TestContentBearingProgress: @pytest.mark.parametrize( "event", [ SimpleNamespace(type="response.created"), SimpleNamespace(type="response.output_item.added", item={"type": "message"}), SimpleNamespace(type="response.content_part.added", part={"type": "output_text"}), SimpleNamespace(type="response.output_text.delta", delta=""), SimpleNamespace(type="response.reasoning_summary_text.delta", delta=""), SimpleNamespace(type="response.function_call_arguments.delta", delta=""), SimpleNamespace( type="response.output_item.added", item=SimpleNamespace(type="function_call"), ), {"type": "response.output_text.delta", "delta": ""}, ], ) def test_codex_empty_and_structural_events_are_not_progress(self, event): assert _codex_event_has_content(event) is False @pytest.mark.parametrize( "event", [ SimpleNamespace(type="response.output_text.delta", delta="token"), SimpleNamespace(type="response.reasoning_summary_text.delta", delta="thought"), SimpleNamespace(type="response.function_call_arguments.delta", delta='{"x"'), SimpleNamespace( type="response.output_item.added", item=SimpleNamespace( type="function_call", id="item_1", call_id="call_1", name="lookup", ), ), {"type": "response.output_text.delta", "delta": "token"}, ], ) def test_codex_nonempty_deltas_are_progress(self, event): assert _codex_event_has_content(event) is True @pytest.mark.parametrize( ("delta", "expected"), [ (SimpleNamespace(type="text_delta", text="", thinking=None), False), (SimpleNamespace(type="text_delta", text="token", thinking=None), True), (SimpleNamespace(type="thinking_delta", text=None, thinking="thought"), True), (SimpleNamespace(type="input_json_delta", partial_json=""), False), (SimpleNamespace(type="input_json_delta", partial_json='{"x"'), True), # Signed-thinking / citation payloads the transport emits # (relay_llm.py signature_delta + citations_delta). (SimpleNamespace(type="signature_delta", signature="sig-1"), True), (SimpleNamespace(type="signature_delta", signature=""), False), (SimpleNamespace(type="citations_delta", citation={"cited": 1}), True), ], ) def test_anthropic_requires_nonempty_delta_payload(self, delta, expected): event = SimpleNamespace(type="content_block_delta", delta=delta) assert _anthropic_event_has_content(event) is expected @pytest.mark.parametrize( ("block", "expected"), [ (SimpleNamespace(type="text"), False), (SimpleNamespace(type="tool_use", id=None, name=None), False), (SimpleNamespace(type="tool_use", id="toolu_1", name="lookup"), True), ], ) def test_anthropic_tool_start_requires_identity(self, block, expected): event = SimpleNamespace(type="content_block_start", content_block=block) assert _anthropic_event_has_content(event) is expected def test_chat_stream_empty_tool_scaffolding_is_not_progress(self): ticks = [] empty_tool_call = SimpleNamespace( index=0, id=None, function=SimpleNamespace(name=None, arguments=""), ) accumulator = _ChatStreamAccumulator() with aux_progress_hook(lambda: ticks.append(1)): accumulator.feed(_chunk(tool_calls=[empty_tool_call])) assert ticks == [] @pytest.mark.parametrize( "tool_call", [ SimpleNamespace(index=0, id="call_1", function=None), SimpleNamespace( index=0, id=None, function=SimpleNamespace(name="lookup", arguments=""), ), SimpleNamespace( index=0, id=None, function=SimpleNamespace(name=None, arguments='{"q"'), ), ], ) def test_chat_stream_substantive_tool_fragments_are_progress(self, tool_call): ticks = [] accumulator = _ChatStreamAccumulator() with aux_progress_hook(lambda: ticks.append(1)): accumulator.feed(_chunk(tool_calls=[tool_call])) assert ticks == [1] def test_openrouter_reasoning_details_keep_compression_alive(self): """Reasoning-only streams must refresh the compression idle fence.""" ticks = [] accumulator = _ChatStreamAccumulator() detail = {"type": "reasoning.summary", "summary": "working..."} with aux_progress_hook(lambda: ticks.append(1)): accumulator.feed(_chunk(reasoning_details=[detail])) result = accumulator.finish() assert ticks == [1] assert result.choices[0].message.reasoning_details == [detail] def test_structural_reasoning_details_are_not_progress(self): ticks = [] accumulator = _ChatStreamAccumulator() with aux_progress_hook(lambda: ticks.append(1)): accumulator.feed( _chunk( reasoning_details=[ {"type": "reasoning.encrypted", "signature": "sig"} ] ) ) assert ticks == [] def test_codex_adapter_updates_fence_only_for_substantive_events(self): events = [ SimpleNamespace(type="response.created"), SimpleNamespace(type="response.output_text.delta", delta=""), SimpleNamespace(type="response.output_text.delta", delta="token"), SimpleNamespace( type="response.output_item.added", item=SimpleNamespace( type="function_call", id="item_1", call_id="call_1", name="lookup" ), ), ] real_client = SimpleNamespace( base_url="https://chatgpt.com/backend-api/codex", responses=SimpleNamespace(create=lambda **_kwargs: iter(events)), ) adapter = _CodexCompletionsAdapter(real_client, "gpt-5.6-sol") fence = CompressionCommitFence() touches = [] def _touch(): touches.append(1) fence.touch_progress() def _consume(stream, *, model, on_event): del model for event in stream: on_event(event) return SimpleNamespace(output=[], usage=None) with ( patch("agent.codex_runtime._consume_codex_event_stream", _consume), aux_progress_hook(_touch), ): adapter.create(messages=[{"role": "user", "content": "summarize"}]) assert touches == [1, 1] def test_anthropic_adapter_updates_fence_only_for_substantive_events(self): events = [ SimpleNamespace(type="ping"), SimpleNamespace( type="content_block_delta", delta=SimpleNamespace(type="text_delta", text=""), ), SimpleNamespace( type="content_block_delta", delta=SimpleNamespace(type="input_json_delta", partial_json='{"q"'), ), SimpleNamespace( type="content_block_start", content_block=SimpleNamespace( type="tool_use", id="toolu_1", name="lookup" ), ), ] adapter = _AnthropicCompletionsAdapter( MagicMock(), "claude-sonnet-4-6", is_oauth=False ) fence = CompressionCommitFence() touches = [] def _touch(): touches.append(1) fence.touch_progress() def _create_message(*_args, **kwargs): for event in events: kwargs["on_stream_event"](event) raise RuntimeError("stop after callback verification") with ( patch( "agent.anthropic_adapter.build_anthropic_kwargs", return_value={"model": "claude-sonnet-4-6", "messages": []}, ), patch( "agent.anthropic_adapter.create_anthropic_message", side_effect=_create_message, ), aux_progress_hook(_touch), pytest.raises(RuntimeError, match="stop after callback verification"), ): adapter.create(messages=[{"role": "user", "content": "summarize"}]) assert touches == [1, 1] def test_keepalive_chunks_do_not_reset_the_compression_fence(self): """End-to-end bug pin (#96707): content-free frames must not refresh CompressionCommitFence._last_progress. The waiter in conversation_compression charges its idle budget from seconds_since_progress(); before the fix, every keepalive chunk fed through _ChatStreamAccumulator ticked the fence, so a stalled summary stream never hit the inactivity timeout.""" fence = CompressionCommitFence() accumulator = _ChatStreamAccumulator() keepalive = SimpleNamespace(id=None, model=None, choices=[], usage=None) empty_role_chunk = _chunk(content="", reasoning="") with aux_progress_hook(fence.touch_progress): for _ in range(5): accumulator.feed(keepalive) accumulator.feed(empty_role_chunk) # No substantive payload arrived: the fence must have stayed stale. assert fence.seconds_since_progress() > 0.0 with aux_progress_hook(fence.touch_progress): accumulator.feed(_chunk(content="token")) assert fence.seconds_since_progress() < 0.05 def test_content_free_frames_still_record_ttfp_timing(self): """The fast-lane telemetry contract (#96945/#96963) survives the gating: time_to_first_progress_ms must record on the FIRST frame of any kind (transport liveness), not only on the first token.""" from agent.auxiliary_client import ( _aux_provider_response, _aux_timing_hook, _notify_aux_timing_response, ) timings: dict = {} def _timed_response() -> None: timings.setdefault("time_to_first_progress_ms", 42) keepalive = SimpleNamespace(id=None, model=None, choices=[], usage=None) accumulator = _ChatStreamAccumulator() with ( _aux_timing_hook(_aux_provider_response, _timed_response), aux_progress_hook(lambda: None), ): accumulator.feed(keepalive) assert timings["time_to_first_progress_ms"] == 42 # --------------------------------------------------------------------------- # Ceiling arithmetic # --------------------------------------------------------------------------- class TestStreamCeiling: def test_floor_applies_to_small_timeouts(self): assert _aux_stream_total_ceiling(30) == 600.0 def test_none_timeout_gets_floor(self): assert _aux_stream_total_ceiling(None) == 600.0 # --------------------------------------------------------------------------- # CompressionCommitFence progress surface # --------------------------------------------------------------------------- class TestFenceProgress: def test_touch_progress_resets_idle_clock(self): fence = CompressionCommitFence() time.sleep(0.05) assert fence.seconds_since_progress() >= 0.04 fence.touch_progress() assert fence.seconds_since_progress() < 0.05 def test_fence_hook_wiring_matches_compressor_usage(self): # conversation_compression installs fence.touch_progress as the hook; # verify the pair works end-to-end through _notify_aux_progress. fence = CompressionCommitFence() time.sleep(0.05) with aux_progress_hook(fence.touch_progress): _notify_aux_progress() assert fence.seconds_since_progress() < 0.05 # --------------------------------------------------------------------------- # Stream-only providers (credit @kudi88, PR #60686) # --------------------------------------------------------------------------- class TestProviderRequiresStream: def test_normal_endpoints_are_not(self): assert _provider_requires_stream( "openrouter", "https://openrouter.ai/api/v1" ) is False assert _provider_requires_stream("auto", None) is False assert _provider_requires_stream("auto", "") is False def test_config_marker_matches_custom_endpoint(self): with patch( "hermes_cli.config.load_config", return_value={"auxiliary": {"stream_only_base_urls": ["my-proxy.example.com"]}}, ): assert _provider_requires_stream( "custom", "https://my-proxy.example.com/v1" ) is True assert _provider_requires_stream( "custom", "https://other.example.com/v1" ) is False class TestForceStream: def test_force_stream_streams_without_a_hook(self): chunks = [_chunk(content="hi", finish_reason="stop")] client = _FakeClient(stream_chunks=chunks) # NO aux_progress_hook installed — force_stream alone must stream. result = _create_with_progress( client, {"model": "m1", "messages": []}, force_stream=True, ) assert client.calls[0]["stream"] is True assert result.choices[0].message.content == "hi" def test_force_stream_does_not_retry_nonstreaming_on_failure(self): client = _FakeClient( response=_COMPLETE, stream_error=RuntimeError("HTTP 400 bad request"), ) with pytest.raises(RuntimeError, match="bad request"): _create_with_progress( client, {"model": "m1", "messages": []}, force_stream=True, ) # No silent non-streaming retry — the provider rejects those anyway. assert len(client.calls) == 1 class TestAsyncStreamAggregation: @pytest.mark.asyncio async def test_async_stream_is_consumed_with_async_for(self): # The sweeper review of PR #60686 flagged that awaiting create() and # then iterating synchronously raises — the async contract is # ``async for``. Verify the async aggregator consumes a real async # iterator and preserves tool-call deltas. tc0 = SimpleNamespace( index=0, id="call_9", function=SimpleNamespace(name="lookup", arguments='{"q":'), ) tc1 = SimpleNamespace( index=0, id=None, function=SimpleNamespace(name=None, arguments='"x"}'), ) raw_chunks = [ _chunk(content="part1 "), _chunk(tool_calls=[tc0]), _chunk(tool_calls=[tc1], content="part2", finish_reason="tool_calls"), ] class _AsyncStream: def __init__(self, items): self._items = list(items) self.closed = False def __aiter__(self): return self async def __anext__(self): if not self._items: raise StopAsyncIteration return self._items.pop(0) async def close(self): self.closed = True stream = _AsyncStream(raw_chunks) result = await _aggregate_chat_stream_async(stream) msg = result.choices[0].message assert msg.content == "part1 part2" assert msg.tool_calls[0].function.name == "lookup" assert msg.tool_calls[0].function.arguments == '{"q":"x"}' assert result.choices[0].finish_reason == "tool_calls" assert stream.closed is True @pytest.mark.asyncio async def test_acreate_with_stream_passes_stream_kwargs(self): calls = [] class _AsyncStream: def __init__(self, items): self._items = list(items) def __aiter__(self): return self async def __anext__(self): if not self._items: raise StopAsyncIteration return self._items.pop(0) class _AsyncClient: def __init__(self): completions = SimpleNamespace(create=self._create) self.chat = SimpleNamespace(completions=completions) async def _create(self, **kwargs): calls.append(kwargs) return _AsyncStream([_chunk(content="ok", finish_reason="stop")]) result = await _acreate_with_stream( _AsyncClient(), {"model": "m1", "messages": [], "timeout": 30}, ) assert calls[0]["stream"] is True assert result.choices[0].message.content == "ok"