fix(compaction): keep the idle gap across gateway turn-start resets (RIG-9162)

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."""
 

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

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

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,8 +84,8 @@ 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_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
         arm_turn(self.agent)

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,
@@ -4106,7 +4107,9 @@ 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()
+            save_idle_gap(agent, now)
+            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

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

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.