From 0744809c9c36848a21b4dc927906d221554fd5c1 Mon Sep 17 00:00:00 2001 From: mintaka Date: Tue, 10 Nov 2026 13:15:13 -0500 Subject: [PATCH 1/2] fix(compaction): keep the idle gap across gateway turn-start resets (RIG-9162) Co-authored-by: Matt Wilkinson --- agent/turn_context_compaction.py | 6 +++++- agent/turn_facade_lease.py | 11 +++++++++-- gateway/run.py | 11 ++++++++++- .../agent/test_compression_trigger_labels.py | 19 +++++++++++++++++++ 4 files changed, 43 insertions(+), 4 deletions(-) diff --git a/agent/turn_context_compaction.py b/agent/turn_context_compaction.py index c7ec2d65b88ce..930eab7b509cd 100644 --- a/agent/turn_context_compaction.py +++ b/agent/turn_context_compaction.py @@ -150,10 +150,14 @@ def _idle_compaction( from agent import turn_context as _tc messages = out.messages + _idle_resume_gap = getattr(agent, "_idle_resume_gap_s", None) + agent._idle_resume_gap_s = None _idle_after = getattr(agent, "compression_idle_compact_after_seconds", 0) if not (agent.compression_enabled and _idle_after > 0 and messages): return - _idle_gap = time.time() - getattr(agent, "_last_activity_ts", time.time()) + _idle_gap = _idle_resume_gap + if _idle_gap is None: + _idle_gap = time.time() - getattr(agent, "_last_activity_ts", time.time()) if _idle_gap < _idle_after: return _compressor = agent.context_compressor diff --git a/agent/turn_facade_lease.py b/agent/turn_facade_lease.py index e60d4ca45bbba..d58473e0d2f46 100644 --- a/agent/turn_facade_lease.py +++ b/agent/turn_facade_lease.py @@ -83,8 +83,15 @@ def build_threads(self) -> None: def start(self) -> None: with self._lock: self.turn_active = True - # Stamp the activity clock at turn entry: `_last_activity_ts` persists across turns, so - # without this the watchdog would measure idle from the PREVIOUS turn and abort a fresh one. + # Save the pre-turn idle gap for idle compaction before the activity clock resets. + previous_activity_ts = getattr(self.agent, "_last_activity_ts", None) + if previous_activity_ts is not None: + prior_gap = getattr(self.agent, "_idle_resume_gap_s", 0) + self.agent._idle_resume_gap_s = max( + prior_gap if isinstance(prior_gap, (int, float)) else 0, + time.time() - previous_activity_ts, + ) + # `_last_activity_ts` persists across turns; without this reset the watchdog would abort a fresh turn. self.agent._touch_activity("starting new turn") from hermes_cli.observability.shared_metrics_process import arm_turn arm_turn(self.agent) diff --git a/gateway/run.py b/gateway/run.py index 2de09616f04d2..f4ff5394b06d8 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -4106,7 +4106,16 @@ def _init_cached_agent_for_turn(agent: Any, interrupt_depth: int) -> None: See #15654, #9051. """ if interrupt_depth == 0: - agent._last_activity_ts = time.time() + now = time.time() + previous_activity_ts = getattr(agent, "_last_activity_ts", None) + if previous_activity_ts is not None: + # Preserve idle time for compaction while resetting the watchdog clock. + prior_gap = getattr(agent, "_idle_resume_gap_s", 0) + agent._idle_resume_gap_s = max( + prior_gap if isinstance(prior_gap, (int, float)) else 0, + now - previous_activity_ts, + ) + agent._last_activity_ts = now agent._last_activity_desc = "starting new turn (cached)" agent._last_activity_provenance = ActivityProvenance.UNKNOWN # Reset the SessionDB flush cursor so the new turn's messages are fully persisted — a stale diff --git a/tests/agent/test_compression_trigger_labels.py b/tests/agent/test_compression_trigger_labels.py index d7f200bd79e85..0a4fc5d77dad1 100644 --- a/tests/agent/test_compression_trigger_labels.py +++ b/tests/agent/test_compression_trigger_labels.py @@ -137,3 +137,22 @@ def test_preflight_reads_the_cooldown_after_a_moved_route_clears_it(monkeypatch) with pytest.raises(_CompressionReached): _drive_pre_api(agent) assert labels == ["pre_api"] + + +def test_idle_compaction_uses_gap_saved_before_cached_turn_reset(): + from gateway.run import GatewayRunner + + labels: list = [] + agent = _agent(labels) + agent.compression_idle_compact_after_seconds = 60 + agent._last_activity_ts = time.time() - 3_600 + agent._last_activity_desc = "previous turn activity" + agent._last_activity_provenance = None + + GatewayRunner._init_cached_agent_for_turn(agent, interrupt_depth=0) + + with pytest.raises(_CompressionReached): + _idle_compaction(agent, _outcome(_messages()), "sys", "current ask", "task") + + assert labels == ["idle"] + assert agent._idle_resume_gap_s is None From 71e3a1f174535cd4b2c67437937b11fc2ebca5ff Mon Sep 17 00:00:00 2001 From: mintaka Date: Tue, 10 Nov 2026 13:54:02 -0500 Subject: [PATCH 2/2] fix(compaction): share the idle-gap save and cover the lease path (RIG-9162) Co-authored-by: Matt Wilkinson --- agent/activity_tracking.py | 15 +++ agent/agent_init.py | 1 + agent/turn_facade_lease.py | 10 +- gateway/run.py | 10 +- .../agent/test_compression_trigger_labels.py | 19 --- tests/agent/test_idle_resume_gap.py | 115 ++++++++++++++++++ .../context-compression-and-caching.md | 3 +- 7 files changed, 137 insertions(+), 36 deletions(-) create mode 100644 tests/agent/test_idle_resume_gap.py diff --git a/agent/activity_tracking.py b/agent/activity_tracking.py index 2b293da1dd351..414613944d1fe 100644 --- a/agent/activity_tracking.py +++ b/agent/activity_tracking.py @@ -25,6 +25,21 @@ def _activity_lock(obj) -> threading.Lock: return _lock +def save_idle_gap(agent, now: Optional[float] = None) -> None: + """Preserve an unconsumed idle gap across turn-start activity-clock resets.""" + # This is time since the last real activity, not yet consumed; it carries across a turn ending before compaction + # because no provider call ran in between. + with _activity_lock(agent): + previous_activity_ts = getattr(agent, "_last_activity_ts", None) + if previous_activity_ts is None: + return + prior_gap = getattr(agent, "_idle_resume_gap_s", 0) + agent._idle_resume_gap_s = max( + prior_gap if isinstance(prior_gap, (int, float)) else 0, + (time.time() if now is None else now) - previous_activity_ts, + ) + + class ActivityTrackingMixin: """Liveness timestamps/labels and rate-limited session activity persistence.""" diff --git a/agent/agent_init.py b/agent/agent_init.py index 50cb0f1b3a49a..4e3ab4ef5e01d 100644 --- a/agent/agent_init.py +++ b/agent/agent_init.py @@ -467,6 +467,7 @@ def _set_defaults(agent, table: dict[str, Any]) -> None: # Activity tracking (API call / tool / stream chunk) for the gateway timeout handler and # "still working" notifications. Named provenances are stamped only by compression writers. "_last_activity_ts": lambda: time.time(), + "_idle_resume_gap_s": None, "_last_activity_desc": "initializing", "_last_activity_provenance": ActivityProvenance.UNKNOWN, "_session_activity_last_persist_mono": 0.0, # rate-limits durable SessionDB stamps diff --git a/agent/turn_facade_lease.py b/agent/turn_facade_lease.py index d58473e0d2f46..b8e21e403354f 100644 --- a/agent/turn_facade_lease.py +++ b/agent/turn_facade_lease.py @@ -16,6 +16,7 @@ from typing import Any, Dict, List, Optional from hermes_state_pidns import holder_namespace_token +from agent.activity_tracking import save_idle_gap # Same logger name as the origin module so log records / caplog filters are unchanged. logger = logging.getLogger("run_agent") @@ -83,14 +84,7 @@ def build_threads(self) -> None: def start(self) -> None: with self._lock: self.turn_active = True - # Save the pre-turn idle gap for idle compaction before the activity clock resets. - previous_activity_ts = getattr(self.agent, "_last_activity_ts", None) - if previous_activity_ts is not None: - prior_gap = getattr(self.agent, "_idle_resume_gap_s", 0) - self.agent._idle_resume_gap_s = max( - prior_gap if isinstance(prior_gap, (int, float)) else 0, - time.time() - previous_activity_ts, - ) + save_idle_gap(self.agent) # `_last_activity_ts` persists across turns; without this reset the watchdog would abort a fresh turn. self.agent._touch_activity("starting new turn") from hermes_cli.observability.shared_metrics_process import arm_turn diff --git a/gateway/run.py b/gateway/run.py index f4ff5394b06d8..438dba47a517e 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -31,6 +31,7 @@ from typing import Callable, Dict, Optional, Any, List, Tuple, cast from agent.async_utils import safe_schedule_threadsafe +from agent.activity_tracking import save_idle_gap from agent.i18n import t from agent.conversation_compression import ( COMPACTION_DONE_STATUS, COMPACTION_HEARTBEAT_STATUS, COMPACTION_STATUS, COMPRESSION_RETRY_CONTEXT_REDUCED_STATUS_TEMPLATE, @@ -4107,14 +4108,7 @@ def _init_cached_agent_for_turn(agent: Any, interrupt_depth: int) -> None: """ if interrupt_depth == 0: now = time.time() - previous_activity_ts = getattr(agent, "_last_activity_ts", None) - if previous_activity_ts is not None: - # Preserve idle time for compaction while resetting the watchdog clock. - prior_gap = getattr(agent, "_idle_resume_gap_s", 0) - agent._idle_resume_gap_s = max( - prior_gap if isinstance(prior_gap, (int, float)) else 0, - now - previous_activity_ts, - ) + save_idle_gap(agent, now) agent._last_activity_ts = now agent._last_activity_desc = "starting new turn (cached)" agent._last_activity_provenance = ActivityProvenance.UNKNOWN diff --git a/tests/agent/test_compression_trigger_labels.py b/tests/agent/test_compression_trigger_labels.py index 0a4fc5d77dad1..d7f200bd79e85 100644 --- a/tests/agent/test_compression_trigger_labels.py +++ b/tests/agent/test_compression_trigger_labels.py @@ -137,22 +137,3 @@ def test_preflight_reads_the_cooldown_after_a_moved_route_clears_it(monkeypatch) with pytest.raises(_CompressionReached): _drive_pre_api(agent) assert labels == ["pre_api"] - - -def test_idle_compaction_uses_gap_saved_before_cached_turn_reset(): - from gateway.run import GatewayRunner - - labels: list = [] - agent = _agent(labels) - agent.compression_idle_compact_after_seconds = 60 - agent._last_activity_ts = time.time() - 3_600 - agent._last_activity_desc = "previous turn activity" - agent._last_activity_provenance = None - - GatewayRunner._init_cached_agent_for_turn(agent, interrupt_depth=0) - - with pytest.raises(_CompressionReached): - _idle_compaction(agent, _outcome(_messages()), "sys", "current ask", "task") - - assert labels == ["idle"] - assert agent._idle_resume_gap_s is None diff --git a/tests/agent/test_idle_resume_gap.py b/tests/agent/test_idle_resume_gap.py new file mode 100644 index 0000000000000..43924521bde13 --- /dev/null +++ b/tests/agent/test_idle_resume_gap.py @@ -0,0 +1,115 @@ +"""Regression coverage for preserving idle time across a turn reset.""" + +from __future__ import annotations + +import time +from types import MethodType, SimpleNamespace + +import pytest + +from agent.activity_tracking import ActivityTrackingMixin +from agent.turn_context_compaction import CompactionOutcome, _idle_compaction +from agent.turn_facade_lease import DurableTurnLease + + +class _CompressionReached(Exception): + """Raised by the recording double after idle compaction selects its trigger.""" + + +def _agent(labels: list[str]) -> SimpleNamespace: + def record_and_stop(_messages, _system, **kwargs): + labels.append(kwargs.get("trigger")) + raise _CompressionReached + + agent = SimpleNamespace( + context_compressor=SimpleNamespace( + threshold_tokens=1_000, summary_target_ratio=0.2, + last_compression_rough_tokens=0, awaiting_real_usage_after_compression=False, + ), + compression_enabled=True, compression_idle_compact_after_seconds=60, + model="m", session_id="s1", _session_db=None, + _emit_status=lambda _message: None, _compress_context=record_and_stop, + _persist_session_activity_if_due=lambda: None, + ) + agent._touch_activity = MethodType(ActivityTrackingMixin._touch_activity, agent) + return agent + + +def _outcome() -> CompactionOutcome: + messages = [ + {"role": "user", "content": "x" * 8_000}, + {"role": "assistant", "content": "y" * 8_000}, + {"role": "user", "content": "current ask"}, + ] + return CompactionOutcome( + messages=messages, active_system_prompt="sys", conversation_history=None, + current_turn_user_idx=len(messages) - 1, + ) + + +def _patch_idle_estimate(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("agent.turn_context._preflight_request_tokens", lambda *_args: 5_000) + + +def test_idle_compaction_uses_gap_saved_before_cached_turn_reset(monkeypatch): + from gateway.run import GatewayRunner + + _patch_idle_estimate(monkeypatch) + labels: list[str] = [] + agent = _agent(labels) + agent._last_activity_ts = time.time() - 3_600 + agent._last_activity_desc = "previous turn activity" + agent._last_activity_provenance = None + + GatewayRunner._init_cached_agent_for_turn(agent, interrupt_depth=0) + + with pytest.raises(_CompressionReached): + _idle_compaction(agent, _outcome(), "sys", "current ask", "task") + + assert labels == ["idle"] + assert agent._idle_resume_gap_s is None + + +def test_lease_start_saves_idle_gap_before_clock_reset(monkeypatch): + monkeypatch.setattr("agent.periodic_scheduler.schedule", lambda *_args: None) + _patch_idle_estimate(monkeypatch) + labels: list[str] = [] + agent = _agent(labels) + agent._last_activity_ts = time.time() - 3_600 + old_activity_ts = agent._last_activity_ts + lease = DurableTurnLease(agent, object(), "s1", "holder") + + lease.start() + + assert agent._last_activity_ts > old_activity_ts + assert agent._last_activity_ts >= time.time() - 5 + with pytest.raises(_CompressionReached): + _idle_compaction(agent, _outcome(), "sys", "current ask", "task") + + assert labels == ["idle"] + + +def test_gateway_gap_survives_later_lease_start(monkeypatch): + from gateway.run import GatewayRunner + + monkeypatch.setattr("agent.periodic_scheduler.schedule", lambda *_args: None) + labels: list[str] = [] + agent = _agent(labels) + agent._last_activity_ts = time.time() - 3_600 + + GatewayRunner._init_cached_agent_for_turn(agent, interrupt_depth=0) + gateway_gap = agent._idle_resume_gap_s + DurableTurnLease(agent, object(), "s1", "holder").start() + + assert gateway_gap >= 3_590 + assert agent._idle_resume_gap_s >= gateway_gap + + +def test_idle_compaction_clears_gap_when_compression_disabled(): + agent = _agent([]) + agent.compression_enabled = False + agent._idle_resume_gap_s = 3_600 + + _idle_compaction(agent, _outcome(), "sys", "current ask", "task") + + assert agent._idle_resume_gap_s is None diff --git a/website/docs/developer-guide/context-compression-and-caching.md b/website/docs/developer-guide/context-compression-and-caching.md index 57893de91f1c2..39a02bdaf9bce 100644 --- a/website/docs/developer-guide/context-compression-and-caching.md +++ b/website/docs/developer-guide/context-compression-and-caching.md @@ -139,7 +139,8 @@ never-estimate acceptance. The following policies remain unchanged: on an estimated delta. Boundary fingerprint matching does not fingerprint the whole prefix, model, tools, or system prompt. - Opt-in idle compaction uses its own floor/cooldown and can act on unanchored - pressure; it does not share the threshold gate's one-request wait. + pressure; it does not share the threshold gate's one-request wait. A pending + idle gap is carried across turn resets until compaction consumes it. - Pre-agent gateway hygiene retains its rough-history fallback and hard-message safety valve. The replay harness's `gateway` shape reloads transcript dictionaries; it does **not** exercise that separate hygiene policy.