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_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..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,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) diff --git a/gateway/run.py b/gateway/run.py index 2de09616f04d2..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, @@ -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 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.