mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-28 18:19:28 +00:00
32fd9d65cf
The gateway's pre-agent session-hygiene compression killed the summary
call at a fixed 30s wall-clock deadline (compression.hygiene_timeout_seconds),
regardless of whether the summary model was hung or merely slow. A reasoning
model happily streaming a large summary was cut off mid-generation, the user
got '⚠️ Context compression timed out after 30.0s', and a 300s failure
cooldown left the session oversized — a doom loop for slow-but-healthy
auxiliary models.
Timeouts are now liveness-based instead of wall-clock-based:
- agent/auxiliary_client.py: new thread-local aux_progress_hook. When
installed (only by context compression today), the primary call_llm
attempt streams (stream=True) and aggregates chunks back into a complete
response, ticking the hook per chunk. The configured timeout then acts
per stream read (idle) instead of as a total budget. Providers that
reject streaming fall back to the plain non-streaming call; auth/payment/
rate-limit/transport errors propagate unchanged into the existing
recovery chains. Codex Responses (per SSE event) and Anthropic Messages
(per stream event, via the new create_anthropic_message on_stream_event
callback) tick the same hook from inside their wire adapters.
- agent/conversation_compression.py: CompressionCommitFence gains
touch_progress()/seconds_since_progress(); compress_context() installs
fence.touch_progress as the progress hook around the compress call.
- gateway/run.py: the hygiene wait loop treats hygiene_timeout_seconds as
an INACTIVITY budget — while the fence reports fresh progress the wait
extends, bounded by the new compression.hygiene_total_ceiling_seconds
(default 600s, clamped >= the idle budget) so a degenerate trickle
stream still dies. The timeout warning now says the summary model
produced no output, which is the only case that still triggers it.
- config/docs: hygiene_total_ceiling_seconds added to DEFAULT_CONFIG and
configuration.md; hygiene_timeout_seconds documented as inactivity-based.
Tests: tests/agent/test_aux_progress_streaming.py (hook plumbing, stream
aggregation incl. tool-call deltas and reasoning deltas, rejection
fallback, ceiling kill, fence progress surface); two new gateway tests
prove a slow-but-streaming worker survives past the fixed timeout
(sabotage-verified: fails with the old fixed deadline) and a
forever-trickling worker is still cut off at the ceiling.
273 lines
10 KiB
Python
273 lines
10 KiB
Python
"""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 per chunk, 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
|
|
|
|
import pytest
|
|
|
|
from agent.auxiliary_client import (
|
|
_aggregate_chat_stream,
|
|
_aux_stream_total_ceiling,
|
|
_create_with_progress,
|
|
_notify_aux_progress,
|
|
aux_progress_hook,
|
|
)
|
|
from agent.conversation_compression import CompressionCommitFence
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _chunk(content=None, reasoning=None, finish_reason=None, usage=None,
|
|
tool_calls=None, model="m1", chunk_id="c1"):
|
|
delta = SimpleNamespace(
|
|
content=content,
|
|
reasoning=reasoning,
|
|
reasoning_content=None,
|
|
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_none_hook_is_noop_passthrough(self):
|
|
with aux_progress_hook(None):
|
|
_notify_aux_progress() # must not raise
|
|
|
|
def test_hook_exception_is_swallowed(self):
|
|
def _boom():
|
|
raise RuntimeError("hook blew up")
|
|
with aux_progress_hook(_boom):
|
|
_notify_aux_progress() # must not raise
|
|
|
|
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_no_hook_means_plain_nonstreaming_call(self):
|
|
client = _FakeClient(response=_COMPLETE)
|
|
result = _create_with_progress(client, {"model": "m1", "messages": []})
|
|
assert result is _COMPLETE
|
|
assert len(client.calls) == 1
|
|
assert "stream" not in client.calls[0]
|
|
|
|
def test_hook_upgrades_to_streaming_and_ticks_per_chunk(self):
|
|
chunks = [
|
|
_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 + 1 per chunk
|
|
assert len(ticks) >= len(chunks) + 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]
|
|
|
|
def test_auth_error_propagates_without_nonstreaming_retry(self):
|
|
class _FakeAuthError(Exception):
|
|
status_code = 401
|
|
client = _FakeClient(stream_error=_FakeAuthError("Error code: 401 - unauthorized"))
|
|
with aux_progress_hook(lambda: None):
|
|
with pytest.raises(_FakeAuthError):
|
|
_create_with_progress(client, {"model": "m1", "messages": []})
|
|
assert len(client.calls) == 1 # no silent non-streaming retry
|
|
|
|
def test_shim_returning_complete_response_passes_through(self):
|
|
# Adapters may ignore stream=True and hand back a full response.
|
|
class _ShimClient(_FakeClient):
|
|
def _create(self, **kwargs):
|
|
self.calls.append(kwargs)
|
|
return _COMPLETE
|
|
client = _ShimClient()
|
|
with aux_progress_hook(lambda: None):
|
|
result = _create_with_progress(client, {"model": "m1", "messages": []})
|
|
assert result is _COMPLETE
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _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_total_ceiling_kills_trickle_stream_as_timeout(self):
|
|
def _trickle():
|
|
while True:
|
|
time.sleep(0.01)
|
|
yield _chunk(content="x")
|
|
with pytest.raises(TimeoutError, match="timed out"):
|
|
_aggregate_chat_stream(_trickle(), total_ceiling=0.05)
|
|
|
|
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]
|
|
|
|
def test_empty_choices_chunks_are_skipped(self):
|
|
empty = SimpleNamespace(id="c", model="m", choices=[], usage=None)
|
|
chunks = [empty, _chunk(content="ok", finish_reason="stop")]
|
|
result = _aggregate_chat_stream(iter(chunks))
|
|
assert result.choices[0].message.content == "ok"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Ceiling arithmetic
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestStreamCeiling:
|
|
def test_floor_applies_to_small_timeouts(self):
|
|
assert _aux_stream_total_ceiling(30) == 600.0
|
|
|
|
def test_multiplier_wins_for_large_timeouts(self):
|
|
assert _aux_stream_total_ceiling(300) == 1200.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
|