From f925b01791fe052e7d55dc975750c0bd985aff91 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sat, 10 Oct 2026 04:01:04 -0700 Subject: [PATCH] fix(relay): stop waiting on Relay after one stalled compaction mark 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). --- agent/conversation_compression.py | 2 +- agent/relay_compaction.py | 19 +++- agent/relay_runtime.py | 14 +-- tests/agent/test_413_compression.py | 49 ---------- tests/agent/test_compaction_events.py | 89 +------------------ tests/agent/test_relay_compaction.py | 10 ++- .../test_relay_runtime_bounded_scope_ops.py | 2 +- tests/agent/test_relay_session_segments.py | 2 +- 8 files changed, 34 insertions(+), 153 deletions(-) diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index ff17d9facd..d87344b319 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -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)): diff --git a/agent/relay_compaction.py b/agent/relay_compaction.py index 5c7b7b224f..cd44c594db 100644 --- a/agent/relay_compaction.py +++ b/agent/relay_compaction.py @@ -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 diff --git a/agent/relay_runtime.py b/agent/relay_runtime.py index b0831f2a4d..f1a841f1c9 100644 --- a/agent/relay_runtime.py +++ b/agent/relay_runtime.py @@ -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) diff --git a/tests/agent/test_413_compression.py b/tests/agent/test_413_compression.py index c7fcf4716d..b29548cb51 100644 --- a/tests/agent/test_413_compression.py +++ b/tests/agent/test_413_compression.py @@ -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.""" diff --git a/tests/agent/test_compaction_events.py b/tests/agent/test_compaction_events.py index 2763ea8574..091951cf7f 100644 --- a/tests/agent/test_compaction_events.py +++ b/tests/agent/test_compaction_events.py @@ -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 diff --git a/tests/agent/test_relay_compaction.py b/tests/agent/test_relay_compaction.py index a49e0fc4e6..a9118f1add 100644 --- a/tests/agent/test_relay_compaction.py +++ b/tests/agent/test_relay_compaction.py @@ -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) diff --git a/tests/agent/test_relay_runtime_bounded_scope_ops.py b/tests/agent/test_relay_runtime_bounded_scope_ops.py index c940bbbfef..58d2a1b6be 100644 --- a/tests/agent/test_relay_runtime_bounded_scope_ops.py +++ b/tests/agent/test_relay_runtime_bounded_scope_ops.py @@ -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.""" diff --git a/tests/agent/test_relay_session_segments.py b/tests/agent/test_relay_session_segments.py index 37d9476cb6..9be1b07612 100644 --- a/tests/agent/test_relay_session_segments.py +++ b/tests/agent/test_relay_session_segments.py @@ -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)