feat(compression): add opt-in idle-triggered context compaction

Long-lived sessions (e.g. a Telegram thread resumed over hours/days)
accumulate a large context that the existing size-based threshold only
trims once it crosses `threshold × context_window`. Until then every
turn re-reads the full history, which on large-context models can mean
hundreds of K of cache-read tokens per call even across long idle gaps.

Add a time-based trigger that complements (does not replace) the size
threshold: when a session resumes after `compression.idle_compact_after_seconds`
of inactivity, compact the accumulated history up front, before the first
reply. Disabled by default (0), so existing behaviour is unchanged.

The trigger reuses `_last_activity_ts` (the last time the turn loop did
work) to measure the idle gap at turn start, gates the token estimate
behind a cheap gap pre-check, and skips compaction when the context is
already at/below the post-compression target (threshold × target_ratio)
so a short idle thread never pays for a summarization that saves nothing.
It also defers to an active compression-failure cooldown.

The decision is factored into a pure predicate, `_should_idle_compact`,
which is unit-tested without a live agent.

agent/agent_init.py

@@ -1869,6 +1869,12 @@ def init_agent(
             codex_app_server_auto_compaction,
         )
         codex_app_server_auto_compaction = "native"
+    # Opt-in idle compaction: compact a session up front when it resumes after
+    # this many seconds of inactivity (0 = disabled). Time-based, so it
+    # complements the size-based threshold above. Consumed by build_turn_context().
+    compression_idle_compact_after_seconds = max(
+        0, int(_compression_cfg.get("idle_compact_after_seconds", 0))
+    )
 
     # Read optional explicit context_length override for the auxiliary
     # compression model. Custom endpoints often cannot report this via
@@ -2295,6 +2301,9 @@ def init_agent(
     agent.compression_in_place = compression_in_place
     agent.codex_app_server_auto_compaction = codex_app_server_auto_compaction
     agent.max_compression_attempts = compression_max_attempts
+    agent.compression_idle_compact_after_seconds = (
+        compression_idle_compact_after_seconds
+    )
 
     # Reject models whose context window is below the minimum required
     # for reliable tool-calling workflows (64K tokens).

agent/turn_context.py

@@ -26,6 +26,7 @@ from __future__ import annotations
 
 import logging
 import threading
+import time
 import uuid
 from dataclasses import dataclass
 from typing import Any, Dict, List, Mapping, Optional
@@ -255,6 +256,40 @@ def _should_run_preflight_estimate(
     return estimate_messages_tokens_rough(messages) >= threshold_tokens
 
 
+def _should_idle_compact(
+    *,
+    enabled: bool,
+    idle_after_seconds: int,
+    idle_gap_seconds: float,
+    tokens: int,
+    floor_tokens: int,
+    cooldown_active: bool,
+) -> bool:
+    """Decide whether an idle-triggered compaction should run this turn.
+
+    Idle compaction is opt-in (``idle_after_seconds <= 0`` disables it). It
+    fires when a session resumes after a wall-clock gap of at least
+    ``idle_after_seconds`` since its last activity, so a long-lived thread
+    that is paused and later resumed compacts its accumulated history up
+    front instead of re-reading it on every subsequent turn.
+
+    It is orthogonal to the token-threshold trigger: it does NOT require the
+    context to exceed ``threshold_tokens``. It still skips work when the
+    context is at or below ``floor_tokens`` (the size compaction would reduce
+    *to*), so a small idle thread never pays for a summarisation that saves
+    nothing, and it defers to an active compression-failure cooldown.
+
+    Pure predicate so the policy is unit-testable without a live agent.
+    """
+    if not enabled or idle_after_seconds <= 0:
+        return False
+    if idle_gap_seconds < idle_after_seconds:
+        return False
+    if cooldown_active:
+        return False
+    return tokens > floor_tokens
+
+
 @dataclass
 class TurnContext:
     """Values produced by the turn prologue and consumed by the turn loop."""
@@ -579,6 +614,62 @@ def build_turn_context(
         if not isinstance(pending_cli_message, dict) or pending_cli_message.get("_db_persisted"):
             agent._pending_cli_user_message = None
 
+    # ── Idle-triggered compaction (opt-in; ``idle_compact_after_seconds``) ──
+    # When a session resumes after a long idle gap, compact the accumulated
+    # history up front so the rest of the conversation does not keep re-reading
+    # a large stale context on every turn. This fires on elapsed wall-clock time
+    # rather than size, so it complements (does not replace) the token-threshold
+    # preflight below. ``_last_activity_ts`` is the last time this turn loop did
+    # work; nothing has touched it yet this turn, so it measures the gap since
+    # the previous turn finished. The cheap gap pre-check gates the (more
+    # expensive) token estimate, mirroring ``_should_run_preflight_estimate``.
+    _idle_after = getattr(agent, "compression_idle_compact_after_seconds", 0)
+    if agent.compression_enabled and _idle_after > 0 and messages:
+        _idle_gap = time.time() - getattr(agent, "_last_activity_ts", time.time())
+        if _idle_gap >= _idle_after:
+            _compressor = agent.context_compressor
+            _idle_tokens = estimate_request_tokens_rough(
+                messages,
+                system_prompt=active_system_prompt or "",
+                tools=agent.tools or None,
+            )
+            # Post-compression target size: don't summarise a thread already
+            # below what compaction would reduce it to.
+            _idle_floor = int(
+                _compressor.threshold_tokens * _compressor.summary_target_ratio
+            )
+            _idle_cooldown = getattr(
+                _compressor, "get_active_compression_failure_cooldown", lambda: None
+            )()
+            if _should_idle_compact(
+                enabled=agent.compression_enabled,
+                idle_after_seconds=_idle_after,
+                idle_gap_seconds=_idle_gap,
+                tokens=_idle_tokens,
+                floor_tokens=_idle_floor,
+                cooldown_active=bool(_idle_cooldown),
+            ):
+                logger.info(
+                    "Idle compaction: %ss idle >= %ss, ~%s tokens > %s floor "
+                    "(session %s)",
+                    int(_idle_gap),
+                    _idle_after,
+                    f"{_idle_tokens:,}",
+                    f"{_idle_floor:,}",
+                    agent.session_id or "none",
+                )
+                agent._emit_status(
+                    f"💤 Resumed after {int(_idle_gap)}s idle — compacting "
+                    f"~{_idle_tokens:,} tokens before continuing."
+                )
+                messages, active_system_prompt = agent._compress_context(
+                    messages, system_message, approx_tokens=_idle_tokens,
+                    task_id=effective_task_id,
+                )
+                conversation_history = conversation_history_after_compression(
+                    agent, messages
+                )
+
     # ── Preflight context compression ──
     # Gate the (expensive) full token estimate behind a cheap pre-check.
     # See ``_should_run_preflight_estimate`` for the OR semantics that fix

cli-config.yaml.example

@@ -474,6 +474,16 @@ compression:
   # head messages, matching the pre-feature behaviour.
   protect_first_n: 3
 
+  # Idle compaction (default: 0 = disabled). When > 0, a session that resumes
+  # after at least this many seconds of inactivity compacts its accumulated
+  # history up front, before the first reply, so a long-lived thread you come
+  # back to later doesn't re-read its full stale context on every turn.
+  # Time-based, so it complements (does not replace) the size-based `threshold`
+  # above. It is skipped when the context is already small (at or below the
+  # post-compression target = threshold × target_ratio), so it never wastes a
+  # summarization on a short idle thread. Example: 1800 = compact after 30 min idle.
+  idle_compact_after_seconds: 0
+
   # To pin a specific model/provider for compression summaries, use the
   # auxiliary section below (auxiliary.compression.provider / model).
 

tests/agent/test_idle_compaction.py

@@ -0,0 +1,55 @@
+"""Tests for the opt-in idle-triggered compaction policy.
+
+Covers ``agent.turn_context._should_idle_compact`` — the pure predicate that
+decides whether a session resuming after an idle gap should compact up front.
+The predicate is intentionally side-effect-free so the policy can be verified
+without constructing a live agent or DB.
+"""
+
+from agent.turn_context import _should_idle_compact
+
+
+def _decide(**overrides):
+    """Call the predicate with sensible defaults (idle + large context => fire)."""
+    kwargs = dict(
+        enabled=True,
+        idle_after_seconds=1800,
+        idle_gap_seconds=3600.0,
+        tokens=100_000,
+        floor_tokens=40_000,
+        cooldown_active=False,
+    )
+    kwargs.update(overrides)
+    return _should_idle_compact(**kwargs)
+
+
+class TestShouldIdleCompact:
+    def test_fires_when_idle_long_enough_and_context_large(self):
+        assert _decide() is True
+
+    def test_disabled_when_idle_after_zero(self):
+        # 0 is the documented "off" value — must never fire regardless of gap.
+        assert _decide(idle_after_seconds=0, idle_gap_seconds=10_000.0) is False
+
+    def test_disabled_when_idle_after_negative(self):
+        assert _decide(idle_after_seconds=-1) is False
+
+    def test_disabled_when_compression_off(self):
+        assert _decide(enabled=False) is False
+
+    def test_skips_when_gap_below_threshold(self):
+        assert _decide(idle_gap_seconds=600.0) is False
+
+    def test_gap_exactly_at_threshold_fires(self):
+        assert _decide(idle_after_seconds=1800, idle_gap_seconds=1800.0) is True
+
+    def test_skips_when_context_at_or_below_floor(self):
+        # At/below the post-compression target there is nothing worth saving.
+        assert _decide(tokens=40_000, floor_tokens=40_000) is False
+        assert _decide(tokens=39_999, floor_tokens=40_000) is False
+
+    def test_fires_just_above_floor(self):
+        assert _decide(tokens=40_001, floor_tokens=40_000) is True
+
+    def test_defers_to_active_compression_cooldown(self):
+        assert _decide(cooldown_active=True) is False