hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases Wiki Activity

fix(relay): stop waiting on Relay after one stalled compaction mark

Browse Source 
A wedged Relay pipeline made every compaction block for SCOPE_OP_TIMEOUT
(10s) while it still held the compression lease, with no breaker. The
first stalled mark now trips a process-wide flag and later marks return
immediately; marks are observability and must never tax compaction.

Also: make relay_runtime.SCOPE_OP_TIMEOUT public (relay_compaction reads
it), fold the commit-time telemetry hold into _hold_attempt_telemetry
with one _ATTEMPT_TELEMETRY_FIELDS tuple, and drop the mapping-table
change-detector tests (the ATOF freshness and failure invariants stay).
This commit is contained in:
Teknium
2026-10-10 04:01:04 -07:00
parent a32f77e53f
commit f925b01791
8 changed files with 34 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
agent/conversation_compression.py
View File

@@ -4024,7 +4024,7 @@ def _begin_compression_attempt(
# commit-time hold keeps the trio for THIS attempt's own emit, but the next attempt (and
# every emit before it re-seeds) must start clean. AFTER the snapshot so a late-unwind
# restore still round-trips the full pre-attempt state.
for _name in ("_last_compression_telemetry", "_active_compression_telemetry", "_compression_telemetry_seed"):
for _name in _ATTEMPT_TELEMETRY_FIELDS:
with contextlib.suppress(Exception):
setattr(agent.context_compressor, _name, None)
if defer_notification and callable(getattr(agent, _PENDING_CONTEXT_ENGINE_NOTIFICATION, None)):

19
agent/relay_compaction.py
View File

@@ -20,6 +20,9 @@ from agent import relay_runtime
logger = logging.getLogger(__name__)
COMPACTION_DATA_SCHEMA = {"name": "hermes.compaction", "version": "1"}
# Marks are observability: after one stalled publish, stop waiting on Relay for the rest of the process
# so a wedged pipeline costs one bounded timeout, not one per compaction under the compression lease.
_stalled = False
def _turn_target(session_id: str) -> tuple[relay_runtime.RelayRuntime, relay_runtime.RelaySession, Any] | None:
@@ -74,9 +77,17 @@ def emit_compaction_mark(
if target is None:
logger.debug("no live Relay session for %s; %s mark not recorded", session_id or "none", name)
return False
global _stalled
if _stalled:
return False
host, session, handle = target
host.run_in_session(
session, host.relay.scope.event, name, handle=handle, data=data, data_schema=dict(COMPACTION_DATA_SCHEMA),
metadata=relay_runtime.runtime_metadata(host.runtime_id), timeout=relay_runtime._SCOPE_OP_TIMEOUT,
)
try:
host.run_in_session(
session, host.relay.scope.event, name, handle=handle, data=data,
data_schema=dict(COMPACTION_DATA_SCHEMA), metadata=relay_runtime.runtime_metadata(host.runtime_id),
timeout=relay_runtime.SCOPE_OP_TIMEOUT,
)
except TimeoutError:
_stalled = True
raise
return True

14
agent/relay_runtime.py
View File

@@ -36,7 +36,7 @@ _PROFILE_KEY_CACHE: dict[str, str] = {}
# Bound for native scope ops gating turn/session completion: a wedged pipeline costs one
# lost span, never a blocked agent.
_SCOPE_OP_TIMEOUT = 10.0
SCOPE_OP_TIMEOUT = 10.0
@@ -462,7 +462,7 @@ class RelayRuntime:
self, session: RelaySession, scope_metadata: dict[str, Any], *, resolve_parent: bool,
exit_fallback: bool = False, **push_kwargs: Any,
) -> None:
"""Push a fresh SESSION_SCOPE for ``session`` (bounded by ``_SCOPE_OP_TIMEOUT``); record handle + context.
"""Push a fresh SESSION_SCOPE for ``session`` (bounded by ``SCOPE_OP_TIMEOUT``); record handle + context.
Subagents parent under their spawning turn/session handle (``resolve_parent`` creates the parent when
unknown). ``exit_fallback``: at interpreter shutdown the executor refuses futures; push synchronously."""
parent_handle = None
@@ -486,7 +486,7 @@ class RelayRuntime:
else:
# future.result() re-raises push's own RuntimeError; keeping it outside the
# except prevents a real push failure from retrying via the exit fallback.
session.handle = future.result(timeout=_SCOPE_OP_TIMEOUT)
session.handle = future.result(timeout=SCOPE_OP_TIMEOUT)
session.context = context
def ensure_session(
@@ -535,7 +535,7 @@ class RelayRuntime:
try:
self.run_in_session(
session, self.relay.scope.pop, old_handle, output={"hermes.session.segment_reason": reason},
metadata=runtime_metadata(self.runtime_id), timeout=_SCOPE_OP_TIMEOUT,
metadata=runtime_metadata(self.runtime_id), timeout=SCOPE_OP_TIMEOUT,
)
except Exception:
logger.warning(
@@ -620,7 +620,7 @@ class RelayRuntime:
) -> Any:
"""Run a Relay operation against a session's isolated scope stack.
``timeout`` bounds the native call on the daemon executor (``TimeoutError`` on breach); ``None``
runs synchronously. Lifecycle ops pass ``_SCOPE_OP_TIMEOUT``: a wedged pipeline costs one span."""
runs synchronously. Lifecycle ops pass ``SCOPE_OP_TIMEOUT``: a wedged pipeline costs one span."""
with self._operation():
return self._run_in_session_untracked(
session, callback, *args, allow_closing=allow_closing, timeout=timeout, **kwargs
@@ -758,7 +758,7 @@ class RelayRuntime:
failure = run_in_session(
session, self._pop_with_drain, handle, output=output or {},
metadata=runtime_metadata(self.runtime_id), session_root=session.handle,
drain_limit=drain_limit, allow_closing=allow_closing, timeout=_SCOPE_OP_TIMEOUT,
drain_limit=drain_limit, allow_closing=allow_closing, timeout=SCOPE_OP_TIMEOUT,
)
except Exception as exc:
failure = exc
@@ -1082,7 +1082,7 @@ class RelaySessionCoordinator:
TURN_SCOPE, host.relay.ScopeType.Function, handle=lease.session.handle,
input=_scope_input(lease.turn_cwd),
metadata=turn_metadata,
timeout=_SCOPE_OP_TIMEOUT,
timeout=SCOPE_OP_TIMEOUT,
)
turn._previous_turn = _CURRENT_TURN.get()
_CURRENT_TURN.set(turn)

49
tests/agent/test_413_compression.py
View File

@@ -444,55 +444,6 @@ class TestHTTP413Compression:
def _context_length_400():
err = Exception(
"Error code: 400 - {'error': {'message': "
"\"This endpoint's maximum context length is 204800 tokens. "
"However, you requested about 270460 tokens.\", 'code': 400}}"
)
err.status_code = 400
return err
def _long_context_tier_429():
err = Exception("Extra usage is required for long context requests.")
err.status_code = 429
return err
@pytest.mark.parametrize(
("make_error", "reason"),
[
pytest.param(_make_413_error, "payload_too_large", id="413"),
pytest.param(_context_length_400, "context_overflow", id="context-length-400"),
pytest.param(_long_context_tier_429, "long_context_tier", id="long-context-tier-429"),
],
)
def test_overflow_recovery_passes_the_classified_reason_to_compression(agent, make_error, reason):
"""Both overflow callers already hold the classifier's reason; the attempt record gets it as
``overflow_reason`` next to ``trigger="overflow"``."""
agent.client.chat.completions.create.side_effect = [
make_error(), _mock_response(content="Recovered after compression", finish_reason="stop"),
]
prefill = [
{"role": "user", "content": "previous question"},
{"role": "assistant", "content": "previous answer"},
]
with (
patch.object(agent, "_compress_context") as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
mock_compress.return_value = ([{"role": "user", "content": "hello"}], "compressed prompt")
agent.run_conversation("hello", conversation_history=prefill)
mock_compress.assert_called_once()
kwargs = mock_compress.call_args.kwargs
assert (kwargs["trigger"], kwargs.get("overflow_reason")) == ("overflow", reason)
class TestPreflightCompression:
"""Preflight compression should compress history before the first API call."""

89
tests/agent/test_compaction_events.py
View File

@@ -141,7 +141,7 @@ def test_failed_summary_is_an_attempt_mark_not_a_compaction(published):
agent.context_compressor.abort_on_summary_failure = True
with patch.object(agent.context_compressor, "_generate_summary", return_value=None):
returned, _ = compress_context(agent, _messages(), "system prompt", approx_tokens=80_000, trigger="post_tool")
compress_context(agent, _messages(), "system prompt", approx_tokens=80_000, trigger="post_tool")
[(_sid, name, payload)] = published
assert name == "compaction.attempt"
@@ -205,21 +205,6 @@ def test_overflow_attempt_records_its_classified_overflow_reason(published):
)
@pytest.mark.parametrize(
("record", "expected"),
[
({"trigger_source": "overflow", "overflow_reason": "payload_too_large"}, "payload_too_large"),
({"trigger_source": "overflow", "overflow_reason": "long_context_tier"}, "long_context_tier"),
({"trigger_source": "overflow", "overflow_reason": "rate_limit"}, "other"),
({"trigger_source": "overflow"}, None),
({"trigger_source": "pre_api"}, None),
],
)
def test_overflow_reason_stays_in_its_closed_set(record, expected):
payload = compaction_events.attempt_payload({"commit_status": "committed", **record})
assert payload["overflow_reason"] == expected
def test_gate_blocked_attempt_publishes_exactly_one_blocked_attempt_mark(published):
compressor = _compressor()
agent = _Agent(compressor)
@@ -460,75 +445,3 @@ def test_a_split_that_fails_after_rewriting_history_is_still_a_compaction(publis
"committed", "session_split_failed", "failed_not_indexed",
)
assert (payload["in_place"], payload["session_rotated"], payload["cache_break"]) == (None, None, True)
@pytest.mark.parametrize(
("record", "name", "outcome"),
[
# Rotation published the compacted child, then indexing failed: the history was rewritten.
({"commit_status": "aborted", "split_status": "failed_not_indexed", "history_rewritten": True},
"compaction", "committed"),
# The split rolled back to the original transcript: nothing the model sees changed.
({"commit_status": "aborted", "split_status": "aborted", "history_rewritten": False},
"compaction.attempt", "failed"),
],
)
def test_the_mark_name_follows_whether_history_was_rewritten(record, name, outcome):
payload = compaction_events.attempt_payload({**record, "failure_class": "session_split_failed"})
assert (compaction_events.event_name(payload), payload["outcome"]) == (name, outcome)
assert payload["failure_class"] == "session_split_failed"
@pytest.mark.parametrize(
("split_status", "expected"),
[
("in_place_committed", (True, False)), ("rotated_committed", (False, True)),
# A session without a database has no commit mode to report.
("not_applicable", (None, None)), ("failed_not_indexed", (None, None)),
],
)
def test_commit_mode_is_null_unless_the_split_committed(split_status, expected):
payload = compaction_events.attempt_payload({"commit_status": "committed", "split_status": split_status})
assert (payload["in_place"], payload["session_rotated"]) == expected
@pytest.mark.parametrize(
("record", "expected"),
[
({"commit_status": "aborted", "failure_class": "lock_contended"}, ("skipped", "lock_contended")),
({"commit_status": "aborted", "failure_class": "would_grow"}, ("aborted", "would_grow")),
({"commit_status": "aborted", "failure_class": "exception:AcmeSecretError"}, ("failed", "exception")),
({"commit_status": "aborted", "failure_class": "acme_engine_reason"}, ("failed", "other")),
({"commit_status": "mystery", "failure_class": None}, ("other", "none")),
],
)
def test_outcome_and_failure_class_stay_in_their_closed_sets(record, expected):
payload = compaction_events.attempt_payload(record)
assert (payload["outcome"], payload["failure_class"]) == expected
assert "Acme" not in json.dumps(payload)
@pytest.mark.parametrize(
("trigger_source", "trigger", "trigger_class"),
[
("manual", "manual", "manual"), ("overflow", "overflow", "overflow"), ("gateway_hygiene", "gateway_hygiene", "auto"),
# An unlabelled automatic caller: the reason is unknown, but the attempt is known to be automatic.
("auto", "unknown", "auto"),
# No trigger, or one outside the closed set: the record cannot say what started it.
("unknown", "unknown", "unknown"), (None, "unknown", "unknown"), ("plugin_said_so", "other", "unknown"),
],
)
def test_trigger_maps_onto_the_closed_set(trigger_source, trigger, trigger_class):
payload = compaction_events.attempt_payload({"trigger_source": trigger_source, "method": "made_up"})
assert (payload["trigger"], payload["trigger_class"], payload["method"]) == (trigger, trigger_class, "other")
@pytest.mark.parametrize(
("outcome", "scope", "name"),
[
("committed", "history", "compaction"), ("committed", "provider", "compaction"),
("committed", "other", "compaction.attempt"), ("skipped", "history", "compaction.attempt"),
],
)
def test_only_a_committed_history_or_provider_rewrite_is_named_compaction(outcome, scope, name):
assert compaction_events.event_name({"outcome": outcome, "scope": scope}) == name

10
tests/agent/test_relay_compaction.py
View File

@@ -8,12 +8,13 @@ hygiene runs outside any turn). Relay resets LLM-history freshness only for the
from __future__ import annotations
import threading
import time
from typing import Any
from unittest.mock import patch
import pytest
from agent import compaction_events, relay_runtime
from agent import compaction_events, relay_compaction, relay_runtime
from agent.relay_compaction import COMPACTION_DATA_SCHEMA, emit_compaction_mark
from agent.relay_runtime import RelayRuntime, RelaySessionCoordinator
@@ -138,6 +139,7 @@ def test_no_relay_host_skips_quietly(monkeypatch):
def test_stalled_relay_event_does_not_hold_the_caller(monkeypatch, relay):
fake, runtime, coordinator = relay
monkeypatch.setattr(relay_compaction, "_stalled", False)
coordinator.acquire_conversation(profile_key=runtime.profile_key, session_id="s1", platform="cli")
release, finished = threading.Event(), threading.Event()
@@ -145,7 +147,7 @@ def test_stalled_relay_event_does_not_hold_the_caller(monkeypatch, relay):
release.wait()
monkeypatch.setattr(fake.scope, "event", stalled_event)
monkeypatch.setattr(relay_runtime, "_SCOPE_OP_TIMEOUT", 2.0)
monkeypatch.setattr(relay_runtime, "SCOPE_OP_TIMEOUT", 2.0)
result = []
def publish():
@@ -163,6 +165,10 @@ def test_stalled_relay_event_does_not_hold_the_caller(monkeypatch, relay):
assert len(result) == 1
assert isinstance(result[0], TimeoutError)
assert "Relay scope operation exceeded" in str(result[0])
# One stall trips the breaker: the next compaction returns at once instead of waiting again.
started = time.monotonic()
assert emit_compaction_mark("s1", "compaction", _committed("s1")) is False
assert time.monotonic() - started < 0.5
finally:
release.set()
publisher.join(timeout=2)

2
tests/agent/test_relay_runtime_bounded_scope_ops.py
View File

@@ -143,7 +143,7 @@ def _fast_scope_timeout(monkeypatch):
The production constant is generous (healthy ops are microseconds);
tests only need 'bounded', not the specific bound.
"""
monkeypatch.setattr(relay_runtime, "_SCOPE_OP_TIMEOUT", 1.0)
monkeypatch.setattr(relay_runtime, "SCOPE_OP_TIMEOUT", 1.0)
def _acquire(coordinator, runtime, session_id="sess-1", monkeypatch=None):
"""Acquire a conversation lease against the fake runtime."""

2
tests/agent/test_relay_session_segments.py
View File

@@ -132,7 +132,7 @@ def _teardown_runtimes():
@pytest.fixture(autouse=True)
def _fast_scope_timeout(monkeypatch):
monkeypatch.setattr(relay_runtime, "_SCOPE_OP_TIMEOUT", 1.0)
monkeypatch.setattr(relay_runtime, "SCOPE_OP_TIMEOUT", 1.0)
@pytest.fixture(autouse=True)