fix(compression): name the automatic trigger and effect in attempt telemetry

The content-free attempt record is the one place that says why a
compaction ran and what it did, but it could not answer either question:

- Every automatic caller passed no trigger, so idle, turn-start
  threshold, engine preflight, pre-API, post-tool and gateway hygiene
  attempts all logged trigger_source=auto. Each call site now passes its
  own label; the shared metric still buckets them as auto.
- tokens after, message counts and how the summary was made were
  computed and logged as prose but never reached the record. A finished
  rewrite now records messages_before/after, tokens_before/after,
  tokens_reclaimed (signed; a summary can grow a short transcript),
  token_count_method=estimate_rough, method (llm_summary,
  aux_fallback_main, deterministic_fallback), items_dropped, and the
  tool_results_pruned / reasoning_items_pruned pass counts.
- Attempts that never reached the local compressor returned silently:
  the automatic breaker gate (before and under the lease), every Codex
  app-server exit, a lease that found the session already rotated, and
  an unreadable durable cooldown. Each now logs one record with a
  commit_status (blocked, skipped, failed, aborted, committed), a route,
  and a class such as blocked:cooldown. These stay out of shared metrics,
  which never counted them.

Test doubles for _compress_context gain the trigger keyword the call
sites now pass.

Signed-off-by: mnajafian-nv <mnajafian@nvidia.com>
(cherry picked from commit 1795993d1a01d91dd76e6ec824cb606f8a03b897)

agent/context_compressor.py

@@ -5495,7 +5495,7 @@ Write only the summary body. Do not include any preamble or prefix."""
 
     def _finalize_compressed(
         self, compressed: list[dict[str, Any]], messages: list[dict[str, Any]], n_messages: int,
-        spare_pending_images: bool,
+        spare_pending_images: bool, pruned_count: int = 0,
     ) -> list[dict[str, Any]]:
         """Post-assembly cleanup: orphan pairs, media, savings, markers, replay prune, mem trim."""
         # Single-prompt cron shape: the only live instruction sits in the protected head, BEFORE the
@@ -5532,6 +5532,7 @@ Write only the summary body. Do not include any preamble or prefix."""
         _pruned_replay = _prune_stale_reasoning_replay(compressed)
         if _pruned_replay and not self.quiet_mode:
             logger.info("Pruned stale replay items from %d assistant message(s) during compaction", _pruned_replay)
+        self._record_compression_effect(n_messages, compressed, pre_estimate, pruned_count, _pruned_replay)
         self._last_compression_made_progress = True
 
         # Compaction frees the biggest allocation: hand pages back to the OS (glibc/config-gated,
@@ -5658,7 +5659,7 @@ Write only the summary body. Do not include any preamble or prefix."""
             )
         # Phase 4: Assemble compressed message list
         compressed = self._assemble_compressed(messages, compress_start, compress_end, scan, summary)
-        return self._finalize_compressed(compressed, canonical_messages, n_messages, spare_pending_images)
+        return self._finalize_compressed(compressed, canonical_messages, n_messages, spare_pending_images, pruned_count)
 
     def _assemble_compressed(
         self, messages: list[dict[str, Any]], compress_start: int, compress_end: int, scan: _HandoffScan, summary: str,

agent/context_compressor_telemetry.py

@@ -50,6 +50,10 @@ class CompressionTelemetryMixin:
             "summary_input_chars": None, "summary_input_sampled_chars": None, "summary_input_omitted_chars": None,
             "summary_input_record_count": None, "summary_input_sampled_record_count": None,
             "summary_input_elided_record_count": None,
+            # Effect of a committed rewrite (filled by _record_compression_effect; defaults describe no rewrite).
+            "method": "none", "items_dropped": 0, "messages_before": None, "messages_after": None,
+            "tokens_before": None, "tokens_after": None, "tokens_reclaimed": None, "token_count_method": None,
+            "tool_results_pruned": None, "reasoning_items_pruned": None,
         }
         self._active_compression_telemetry = self._last_compression_telemetry = telemetry
         return telemetry
@@ -92,3 +96,28 @@ class CompressionTelemetryMixin:
             accumulate = key in {"queue_wait_ms", "summary_generation_ms"} and value is not None
             telemetry[key] = (telemetry.get(key) or 0) + value if accumulate else value
 
+    def _compression_method(self) -> str:
+        """How the summary that replaced the middle window was produced."""
+        if getattr(self, "_last_summary_fallback_used", False):
+            return "deterministic_fallback"
+        if getattr(self, "_last_aux_model_failure_error", None):
+            return "aux_fallback_main"
+        return "llm_summary"
+
+    def _record_compression_effect(
+        self, messages_before: int, compressed: list[dict[str, Any]], tokens_before: int, tool_results_pruned: int,
+        reasoning_items_pruned: int,
+    ) -> None:
+        """Record what a finished rewrite did. ``tokens_before`` and the after count are rough message-only
+        estimates (system prompt and tool schemas excluded), so they compare like for like."""
+        telemetry = getattr(self, "_active_compression_telemetry", None)
+        if not isinstance(telemetry, dict):
+            return
+        tokens_after = estimate_messages_tokens_rough(compressed)
+        telemetry.update(
+            method=self._compression_method(), items_dropped=getattr(self, "_last_summary_dropped_count", 0) or 0,
+            messages_before=messages_before, messages_after=len(compressed), tokens_before=tokens_before,
+            tokens_after=tokens_after, tokens_reclaimed=tokens_before - tokens_after,
+            token_count_method="estimate_rough", tool_results_pruned=tool_results_pruned,
+            reasoning_items_pruned=reasoning_items_pruned,
+        )

agent/conversation_compression.py

@@ -31,7 +31,10 @@ from typing import Any, Callable, Dict, List, Literal, Optional, Tuple
 from agent.auxiliary_client import AuxiliaryExplicitCancellation
 from agent.context_engine import automatic_compaction_status_message, sanitize_memory_context
 from agent.conversation_compression_codex import _compress_context_via_codex_app_server
-from agent.conversation_compression_telemetry import _emit_aborted_attempt_telemetry, _emit_compression_attempt_telemetry
+from agent.conversation_compression_telemetry import (
+    _emit_aborted_attempt_telemetry, _emit_blocked_attempt_telemetry, _emit_bypassed_attempt_telemetry,
+    _emit_compression_attempt_telemetry,
+)
 from agent.memory_provider import PRE_COMPRESS_CHECKPOINT_API_VERSION
 from agent.model_metadata import estimate_messages_tokens_rough, estimate_request_tokens_rough
 from agent.session_activity import ActivityProvenance, normalize_activity_provenance
@@ -4043,10 +4046,15 @@ def _route_codex_compaction(
     """Codex owns the real thread: run its own compact under the commit fence bracket."""
     if commit_fence is not None and not commit_fence.begin_commit(getattr(agent, "_hard_interrupt_requested", None)):
         attempt.restore_compressor(agent.context_compressor)
+        _emit_bypassed_attempt_telemetry(
+            agent, attempt.started_at, commit_status="aborted", failure_class="commit_fence_cancelled",
+            approx_tokens=approx_tokens, route="codex_app_server",
+        )
         return messages, _existing_system_prompt(agent, system_message)
     try:
         return _compress_context_via_codex_app_server(
-            agent, messages, system_message, approx_tokens=approx_tokens, task_id=task_id, force=force
+            agent, messages, system_message, approx_tokens=approx_tokens, task_id=task_id, force=force,
+            started_at=attempt.started_at,
         )
     finally:
         if commit_fence is not None:
@@ -4110,8 +4118,8 @@ def compress_context(
     cooperative fence for executor callers that may time out. It prevents a late worker from mutating
     session state after its caller has moved on. verbatim_tail: The exchanges ``/compress here N`` keeps
     after ``messages``; an in-place commit stores them after the compacted head and returns head + tail.
-    trigger: Why this attempt runs (``"overflow"`` for provider-rejected requests); defaults to manual/auto
-    from ``force``. Feeds attempt telemetry only.
+    trigger: Why this attempt runs (``"overflow"``, ``"pre_api"``, ``"post_tool"``, ``"idle"``, ...); defaults
+    to manual/auto from ``force``. Feeds attempt telemetry only; shared metrics bucket it coarsely.
     snapshot_is_current: Optional host snapshot validation after durable lease admission. This protects
     edits completed before admission; it is not a substitute for the host's final publication fence.
     """
@@ -4137,6 +4145,7 @@ def compress_context(
     # All automatic entrypoints honor compressor cooldown/breaker state; hygiene's
     # fresh AIAgent loads the persisted streak via bind_session_state() first.
     if not force and _automatic_compression_gate_blocks(agent, bypass_cooldown):
+        _emit_blocked_attempt_telemetry(agent, attempt.started_at, approx_tokens)
         return messages, _existing_system_prompt(agent, system_message)
 
     _pre_msg_count = len(messages)
@@ -4183,6 +4192,10 @@ def compress_context(
         return messages, _existing_system_prompt(agent, system_message)
     _adopted = _adopt_if_parent_rotated(agent, lease, messages, system_message)
     if _adopted is not None:
+        _emit_bypassed_attempt_telemetry(
+            agent, attempt.started_at, commit_status="skipped", failure_class="session_ownership_lost",
+            approx_tokens=approx_tokens,
+        )
         return _adopted
 
     # Snapshot durable cooldown only once we own the lease. Runs for force=True
@@ -4194,12 +4207,17 @@ def compress_context(
         # Durable cooldown read failed under a built-in compressor: force=True could
         # clear an unknown newer row before cancellation could restore it. Abort.
         lease.release()
+        _emit_bypassed_attempt_telemetry(
+            agent, attempt.started_at, commit_status="aborted", failure_class="cooldown_state_unreadable",
+            approx_tokens=approx_tokens,
+        )
         return messages, _existing_system_prompt(agent, system_message)
 
     # Another path may have compacted this session in place since construction;
     # re-read breaker state under the lock, not the bind_session_state() snapshot.
     if not force and _automatic_compression_gate_blocks(agent, bypass_cooldown, include_cooldown=False):
         lease.release()
+        _emit_blocked_attempt_telemetry(agent, attempt.started_at, approx_tokens)
         return messages, _existing_system_prompt(agent, system_message)
 
     # Interrupts/redirects must not tear a summary in half. Use the explicit stop

agent/conversation_compression_codex.py

@@ -12,6 +12,8 @@ import contextlib
 import logging
 from typing import Any, Optional
 
+from agent.conversation_compression_telemetry import _emit_bypassed_attempt_telemetry
+
 logger = logging.getLogger("agent.conversation_compression")
 
 
@@ -48,37 +50,46 @@ def _record_codex_compaction_failure(agent: Any, error: str) -> None:
 
 def _compress_context_via_codex_app_server(
     agent: Any, messages: list, system_message: Optional[str], *, approx_tokens: Optional[int] = None,
-    task_id: str = "default", force: bool = False,
+    task_id: str = "default", force: bool = False, started_at: float,
 ) -> tuple[list, str]:
     """Route compaction to Codex app-server for Codex-owned threads.
     Rewriting the local transcript would not shrink the Codex thread, so Codex compacts its own thread and
-    Hermes' transcript is left unchanged."""
+    Hermes' transcript is left unchanged. Every returning exit logs one attempt record."""
     from agent.conversation_compression import (
         COMPACTION_STATUS, _CompressionActivityHeartbeat, _emit_compaction_done, _existing_system_prompt,
         _reset_read_dedup_caches, _swallow,
     )
+
+    def _record(commit_status: str, failure_class: Optional[str], method: str = "none") -> None:
+        _emit_bypassed_attempt_telemetry(
+            agent, started_at, commit_status=commit_status, failure_class=failure_class, approx_tokens=approx_tokens,
+            route="codex_app_server", method=method,
+        )
+
     _sid = getattr(agent, "session_id", None) or "none"
     _tokens = f"{approx_tokens:,}" if approx_tokens else "unknown"
     auto_mode = str(getattr(agent, "codex_app_server_auto_compaction", "native") or "native").lower()
     if auto_mode not in {"native", "hermes", "off"}:
         auto_mode = "native"
-    skip_reason = None
+    skip_reason = skip_status = skip_class = None
     if not force and auto_mode != "hermes":
-        skip_reason = f"mode={auto_mode} force=false"
+        skip_reason, skip_status, skip_class = f"mode={auto_mode} force=false", "skipped", f"codex_auto_{auto_mode}"
     elif not force:
         # Automatic entrypoints honor the compressor-owned cooldown: a recent compaction
         # failed, and retrying every turn is what thrashes.
         _cooldown_remaining = _codex_compaction_cooldown_remaining(agent)
         if _cooldown_remaining > 0:
             skip_reason = f"failure cooldown active for {_cooldown_remaining:.0f}s"
+            skip_status, skip_class = "blocked", "blocked:cooldown"
     codex_session = getattr(agent, "_codex_session", None)
     if skip_reason is None and codex_session is None:
-        skip_reason = "no active codex thread"
+        skip_reason, skip_status, skip_class = "no active codex thread", "skipped", "codex_no_thread"
     if skip_reason is not None:
         logger.info(
             "codex app-server compaction skipped: %s (session=%s messages=%d tokens=~%s)", skip_reason, _sid,
             len(messages), _tokens,
         )
+        _record(skip_status, skip_class)
         return messages, _existing_system_prompt(agent, system_message)
     logger.info("codex app-server compaction started: session=%s messages=%d tokens=~%s", _sid, len(messages), _tokens)
     with contextlib.suppress(Exception):
@@ -101,6 +112,7 @@ def _compress_context_via_codex_app_server(
         # The transcript is returned unchanged, so the session is still over
         # threshold. Without a brake the next turn retries immediately.
         _record_codex_compaction_failure(agent, str(getattr(result, "error", None) or "compaction interrupted"))
+        _record("failed", "codex_compaction_failed")
         return messages, _existing_system_prompt(agent, system_message)
     with _swallow('codex compaction bookkeeping failed', exc_info=True):
         from agent.codex_runtime import _record_codex_app_server_compaction, _record_codex_app_server_usage
@@ -115,6 +127,7 @@ def _compress_context_via_codex_app_server(
         getattr(result, "thread_id", None) or "", getattr(result, "turn_id", None) or "",
     )
     existing_prompt = _existing_system_prompt(agent, system_message)
+    _record("committed", None, method="provider")
     # Terminal edge only on success — failure/interrupt paths above return
     # without it, matching the main compress_context() gating.
     _emit_compaction_done(agent)

agent/conversation_compression_telemetry.py

@@ -41,6 +41,7 @@ def _emit_compression_attempt_telemetry(
             telemetry = {}
         payload = dict(telemetry)
         payload.setdefault("event", "compression_attempt")
+        payload.setdefault("route", "hermes")
         payload.setdefault("attempt_id", getattr(agent, "_compression_attempt_id", "") or uuid.uuid4().hex)
         payload.setdefault("session_id", getattr(agent, "session_id", "") or "")
         payload.update(
@@ -76,3 +77,49 @@ def _emit_aborted_attempt_telemetry(agent: Any, started_at: float, failure_class
     _emit_compression_attempt_telemetry(
         agent, started_at=started_at, commit_status="aborted", split_status="aborted", failure_class=failure_class
     )
+
+
+def _emit_bypassed_attempt_telemetry(
+    agent: Any, started_at: float, *, commit_status: str, failure_class: str | None, approx_tokens: Any,
+    route: str = "hermes", method: str = "none",
+) -> None:
+    """Log one attempt that never reached the local compressor: an automatic gate blocked it, Codex owns the
+    thread, or the session lease showed another path already owns the work. The compressor's telemetry still
+    describes an earlier attempt, so this record starts from the attempt seed. Shared metrics have never
+    counted these exits and still do not."""
+    try:
+        compressor = getattr(agent, "context_compressor", None)
+        seed = getattr(compressor, "_compression_telemetry_seed", None)
+        seed = seed if isinstance(seed, dict) else {}
+        payload = {
+            "event": "compression_attempt", "route": route, "method": method, "failure_class": failure_class,
+            "attempt_id": getattr(agent, "_compression_attempt_id", "") or seed.get("attempt_id") or uuid.uuid4().hex,
+            "session_id": getattr(agent, "session_id", "") or "", "trigger_source": seed.get("trigger_source") or "unknown",
+            "main_provider": getattr(agent, "provider", "") or "", "main_model": getattr(agent, "model", "") or "",
+            # Private caches only: the public properties can trigger a synchronous context-length probe.
+            "main_context_limit": getattr(compressor, "_resolved_context_length", None),
+            "effective_threshold": getattr(compressor, "_threshold_tokens", None),
+            "current_estimated_tokens": approx_tokens if isinstance(approx_tokens, int) else None,
+            "total_duration_ms": int((time.monotonic() - started_at) * 1000), "commit_status": commit_status,
+            "split_status": "not_applicable", "fallback_used": False,
+        }
+        logger.info(
+            "context compression attempt telemetry: %s", json.dumps(payload, sort_keys=True, separators=(",", ":"))
+        )
+    except Exception as exc:
+        logger.debug("failed to emit compression attempt telemetry: %s", exc, exc_info=True)
+
+
+def _emit_blocked_attempt_telemetry(agent: Any, started_at: float, approx_tokens: Any) -> None:
+    """Record an automatic attempt the breaker gate refused. The class keeps only the guard's name
+    (``blocked:cooldown``, ``blocked:structural_backoff``, ``blocked:ineffective``), never its seconds."""
+    reason = None
+    try:
+        reason_fn = getattr(getattr(agent, "context_compressor", None), "_compression_block_reason", None)
+        reason = reason_fn() if callable(reason_fn) else None
+    except Exception:
+        logger.debug("compression block-reason read failed", exc_info=True)
+    guard = reason.split(":", 1)[0] if isinstance(reason, str) and reason else "unknown"
+    _emit_bypassed_attempt_telemetry(
+        agent, started_at, commit_status="blocked", failure_class=f"blocked:{guard}", approx_tokens=approx_tokens
+    )

agent/turn_context_compaction.py

@@ -203,7 +203,7 @@ def _idle_compaction(
     if _idle_status:
         agent._emit_status(_idle_status)
     out.messages, out.active_system_prompt = agent._compress_context(
-        messages, system_message, approx_tokens=_idle_tokens, task_id=effective_task_id
+        messages, system_message, approx_tokens=_idle_tokens, task_id=effective_task_id, trigger="idle",
     )
     # ``_compress_context`` returns the INPUT list object when it skips; only
     # re-baseline and re-anchor after a real compaction.
@@ -381,7 +381,7 @@ def _run_preflight_passes(
         _orig_tokens = _preflight_tokens
         out.messages, out.active_system_prompt = agent._compress_context(
             _preflight_input, system_message, approx_tokens=_preflight_tokens,
-            task_id=effective_task_id,
+            task_id=effective_task_id, trigger="turn_start_threshold",
         )
         if out.messages is _preflight_input and compression_skipped_due_to_lock(agent):
             # Lock-skip: another path holds the lock, so this is a DEFER, not proof of
@@ -458,7 +458,8 @@ def _engine_preflight_maintenance(
     )
     _engine_input = out.messages
     out.messages, out.active_system_prompt = agent._compress_context(
-        _engine_input, system_message, approx_tokens=_preflight_tokens, task_id=effective_task_id
+        _engine_input, system_message, approx_tokens=_preflight_tokens, task_id=effective_task_id,
+        trigger="engine_preflight",
     )
     # ``_compress_context`` returns the INPUT list on every skip path and an engine
     # may no-op; re-baseline/re-anchor only after a REAL compaction.

agent/turn_preflight.py

@@ -146,7 +146,7 @@ def run_preflight_compression(
         _pre_api_input = v.messages
         v.messages, v.active_system_prompt = agent._compress_context(
             v.messages, system_message, approx_tokens=request_pressure_tokens,
-            task_id=effective_task_id,
+            task_id=effective_task_id, trigger="pre_api",
         )
         if context_compression_timed_out(agent):
             # Progress-aware timeout: never reached the provider — refund the
@@ -313,7 +313,7 @@ def compress_after_tool_results(
         # Pass overhead-aware _real_tokens, not last_prompt_tokens (0 in the
         # no-usage fallback), so the overflow guard sees the true size.
         messages, active_system_prompt = agent._compress_context(
-            messages, system_message, approx_tokens=_real_tokens, task_id=effective_task_id
+            messages, system_message, approx_tokens=_real_tokens, task_id=effective_task_id, trigger="post_tool",
         )
         if messages is _post_tool_input and compression_skipped_due_to_lock(agent):
             # Lock-skip no-op is a temporary defer, not evidence about compressibility:

gateway/run.py

@@ -254,7 +254,7 @@ async def run_codex_hygiene_compaction(
     # copy_context carries profile secret scope / HERMES_HOME override (executors don't propagate ContextVars).
     worker_future = asyncio.get_running_loop().run_in_executor(
         None, copy_context().run,
-        lambda: agent._compress_context(history, "", approx_tokens=approx_tokens, task_id=session_id or "default"))
+        lambda: agent._compress_context(history, "", approx_tokens=approx_tokens, task_id=session_id or "default", trigger="gateway_hygiene"))
     track_worker = getattr(gateway, "_track_deferred_agent_worker", None)
     if callable(track_worker):
         # ``wait_for`` only cancels the asyncio wrapper; keep the running executor thread visible to shutdown.

gateway/run_turn.py

@@ -1315,7 +1315,7 @@ class GatewayTurnMixin:
                 # task_id = the live turn's dedup bucket (session row id), never "" (= reset every task).
                 lambda: _hyg_agent._compress_context(
                     _hyg_msgs, "", approx_tokens=plan.approx_tokens, commit_fence=_hyg_commit_fence,
-                    task_id=session_entry.session_id or "default",
+                    task_id=session_entry.session_id or "default", trigger="gateway_hygiene",
                 ),
             )
             # Register the live worker with shutdown NOW, not only once it is deferred: the default

hermes_cli/observability/shared_metrics_fields.py

@@ -118,6 +118,10 @@ def setup_completed_fields(*, surface: Any, provider: Any) -> dict[str, str]:
 
 _COMPRESSION_TRIGGERS = {
     **dict.fromkeys(("auto", "threshold", "preflight", "auto_threshold"), "auto"),
+    # The attempt log's precise automatic triggers stay one coarse bucket here.
+    **dict.fromkeys(
+        ("idle", "turn_start_threshold", "engine_preflight", "pre_api", "post_tool", "gateway_hygiene"), "auto",
+    ),
     **dict.fromkeys(("overflow", "context_overflow", "overflow_error", "error"), "overflow"),
     **dict.fromkeys(("manual", "user", "command", "slash"), "manual"),
 }

tests/agent/test_api_content_sidecar.py

@@ -652,7 +652,7 @@ class TestPrologueMoaAndInPlaceBackfill:
             get_active_compression_failure_cooldown=lambda: None,
         )
 
-        def _compress(messages, _system, approx_tokens=None, task_id=None):
+        def _compress(messages, _system, approx_tokens=None, task_id=None, trigger=None):
             # Emulate compress_context in in_place mode: archive_and_compact
             # already inserted these rows (api_content=NULL — the stamp has
             # not happened yet), fresh copies replace the live dicts.

tests/agent/test_compression_attempt_effect.py

@@ -0,0 +1,221 @@
+"""Attempt telemetry names why a compression ran, what it changed, and how the summary was made.
+
+Covers the fields added to the content-free attempt record (trigger label, effect counts, method) and the
+records for exits that never reach the local compressor (automatic gate, Codex route).
+"""
+
+import json
+import logging
+import time
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+from agent.context_compressor import ContextCompressor
+from agent.conversation_compression import compress_context
+from agent.transports.codex_app_server_session import TurnResult
+from hermes_cli.observability import shared_metrics_events
+from hermes_cli.observability.shared_metrics_fields import compression_fields
+
+SECRET = "TOPSECRET_TRANSCRIPT_TEXT"
+
+
+class _TodoStore:
+    def format_for_injection(self):
+        return ""
+
+
+class _Agent:
+    def __init__(self, compressor):
+        self.context_compressor = compressor
+        self.session_id = "session-effect-test"
+        self.platform = "cli"
+        self.model = "test/main-model"
+        self.provider = "test-provider"
+        self.tools = []
+        self._compression_feasibility_checked = True
+        self.compression_in_place = False
+        self._memory_manager = None
+        self._session_db = None
+        self._todo_store = _TodoStore()
+        self._cached_system_prompt = None
+
+    def _emit_status(self, _message):
+        pass
+
+    def _emit_warning(self, _message):
+        pass
+
+    def _invalidate_system_prompt(self):
+        self._cached_system_prompt = None
+
+    def _build_system_prompt(self, system_message):
+        return system_message
+
+    def commit_memory_session(self, _messages):
+        pass
+
+
+def _compressor():
+    with patch("agent.context_compressor.get_model_context_length", return_value=100_000):
+        compressor = ContextCompressor(
+            model="test/main-model", provider="test-provider", threshold_percent=0.50, quiet_mode=True,
+            config_context_length=100_000,
+        )
+    compressor.tail_token_budget = 10
+    return compressor
+
+
+def _messages():
+    msgs = [{"role": "system", "content": "system prompt"}]
+    filler = " ".join(["context"] * 200)
+    for idx in range(10):
+        msgs.append({"role": "user", "content": f"user message {idx} {SECRET} {filler}"})
+        msgs.append({"role": "assistant", "content": f"assistant reply {idx} {SECRET} {filler}"})
+    return msgs
+
+
+def _attempt_records(caplog):
+    prefix = "context compression attempt telemetry: "
+    return [
+        json.loads(record.getMessage().split(prefix, 1)[1])
+        for record in caplog.records
+        if prefix in record.getMessage()
+    ]
+
+
+@pytest.fixture
+def shared_metric_rows(monkeypatch):
+    rows = []
+    monkeypatch.setattr(shared_metrics_events, "record_compression", lambda **kw: rows.append(compression_fields(**kw)))
+    return rows
+
+
+def test_committed_summary_records_trigger_effect_and_method_without_content(caplog, shared_metric_rows):
+    agent = _Agent(_compressor())
+
+    with patch.object(agent.context_compressor, "_generate_summary", return_value="SANITIZED SUMMARY"):
+        with caplog.at_level(logging.INFO, logger="agent.conversation_compression"):
+            compressed, _ = compress_context(agent, _messages(), "system prompt", approx_tokens=80_000, trigger="pre_api")
+
+    [record] = _attempt_records(caplog)
+    assert record["trigger_source"] == "pre_api"
+    assert (record["route"], record["commit_status"], record["method"]) == ("hermes", "committed", "llm_summary")
+    assert record["items_dropped"] == 0
+    assert record["messages_before"] == len(_messages())
+    assert record["messages_after"] < record["messages_before"]
+    assert record["tokens_after"] < record["tokens_before"]
+    assert record["tokens_reclaimed"] == record["tokens_before"] - record["tokens_after"]
+    assert record["token_count_method"] == "estimate_rough"
+    assert record["tool_results_pruned"] == 0 and record["reasoning_items_pruned"] == 0
+    assert SECRET not in json.dumps(record) and "SANITIZED SUMMARY" not in json.dumps(record)
+    # The precise label stays in the log; the shared metric keeps its coarse bucket.
+    assert shared_metric_rows == [
+        {"trigger": "auto", "outcome": "success", "context_fill_bucket": "75_to_90", "failure_class": "none"}
+    ]
+
+
+def test_deterministic_fallback_commit_records_its_method_and_dropped_items(caplog):
+    agent = _Agent(_compressor())
+
+    with patch.object(agent.context_compressor, "_generate_summary", return_value=None):
+        with caplog.at_level(logging.INFO, logger="agent.conversation_compression"):
+            compressed, _ = compress_context(agent, _messages(), "system prompt", approx_tokens=80_000)
+
+    [record] = _attempt_records(caplog)
+    assert record["commit_status"] == "committed"
+    assert record["method"] == "deterministic_fallback"
+    assert record["items_dropped"] > 0
+    assert record["failure_class"] == "summary_generation_failed"
+    assert SECRET not in json.dumps(record)
+
+
+def test_gate_blocked_attempt_logs_one_blocked_record_and_no_shared_metric(caplog, shared_metric_rows):
+    compressor = _compressor()
+    agent = _Agent(compressor)
+    messages = _messages()
+
+    with patch.object(type(compressor), "_compression_block_reason", return_value="cooldown:42"), \
+            patch.object(type(compressor), "_automatic_compression_blocked", return_value=True), \
+            patch.object(compressor, "_generate_summary") as summary:
+        with caplog.at_level(logging.INFO, logger="agent.conversation_compression"):
+            returned, _ = compress_context(agent, messages, "system prompt", approx_tokens=80_000, trigger="post_tool")
+
+    assert returned is messages
+    summary.assert_not_called()
+    [record] = _attempt_records(caplog)
+    assert (record["commit_status"], record["failure_class"]) == ("blocked", "blocked:cooldown")
+    assert (record["trigger_source"], record["current_estimated_tokens"]) == ("post_tool", 80_000)
+    assert record["attempt_id"] == agent._compression_attempt_id
+    assert shared_metric_rows == []
+
+
+class _CodexSession:
+    def __init__(self, result):
+        self.result = result
+
+    def compact_thread(self):
+        return self.result
+
+    def close(self):
+        pass
+
+
+def _codex_agent(result, auto_mode):
+    agent = SimpleNamespace(
+        api_mode="codex_app_server", codex_app_server_auto_compaction=auto_mode, session_id="codex-session",
+        platform="cli", model="gpt-test", provider="openai-codex", _cached_system_prompt="cached prompt",
+        _codex_session=_CodexSession(result), _compression_activity_heartbeat_interval=0.1,
+        context_compressor=SimpleNamespace(
+            compression_count=0, last_compression_rough_tokens=0, last_prompt_tokens=1, last_completion_tokens=0,
+            awaiting_real_usage_after_compression=False,
+        ),
+    )
+    agent._touch_activity = lambda *_args, **_kwargs: None
+    agent._emit_status = agent._emit_warning = lambda _message: None
+    return agent
+
+
+@pytest.mark.parametrize(
+    ("auto_mode", "force", "result", "expected"),
+    [
+        ("native", False, TurnResult(thread_id="t1"), ("skipped", "codex_auto_native", "none")),
+        ("hermes", True, TurnResult(thread_id="t1", turn_id="c1"), ("committed", None, "provider")),
+        ("hermes", True, TurnResult(thread_id="t1", error="boom"), ("failed", "codex_compaction_failed", "none")),
+    ],
+)
+def test_codex_route_logs_exactly_one_record_per_exit(caplog, shared_metric_rows, auto_mode, force, result, expected):
+    agent = _codex_agent(result, auto_mode)
+
+    with caplog.at_level(logging.INFO, logger="agent.conversation_compression"):
+        compress_context(agent, [{"role": "user", "content": SECRET}], "system", approx_tokens=90_000, force=force)
+
+    [record] = _attempt_records(caplog)
+    assert (record["commit_status"], record["failure_class"], record["method"]) == expected
+    assert record["route"] == "codex_app_server"
+    assert record["trigger_source"] == ("manual" if force else "auto")
+    assert SECRET not in json.dumps(record) and "boom" not in json.dumps(record)
+    assert shared_metric_rows == []
+
+
+@pytest.mark.parametrize(
+    "trigger", ["idle", "turn_start_threshold", "engine_preflight", "pre_api", "post_tool", "gateway_hygiene"]
+)
+def test_precise_automatic_triggers_keep_the_coarse_shared_metric_bucket(trigger):
+    fields = compression_fields(trigger=trigger, outcome="success", tokens_before=1, context_length=10)
+    assert fields["trigger"] == "auto"
+
+
+def test_blocked_record_never_carries_the_cooldown_seconds(caplog):
+    from agent.conversation_compression_telemetry import _emit_blocked_attempt_telemetry
+
+    compressor = SimpleNamespace(_compression_block_reason=lambda: "structural_backoff:917")
+    agent = SimpleNamespace(context_compressor=compressor, session_id="s", _compression_attempt_id="a")
+
+    with caplog.at_level(logging.INFO, logger="agent.conversation_compression"):
+        _emit_blocked_attempt_telemetry(agent, time.monotonic(), None)
+
+    [record] = _attempt_records(caplog)
+    assert record["failure_class"] == "blocked:structural_backoff"
+    assert "917" not in json.dumps(record)

tests/agent/test_run_agent_codex_responses.py

@@ -1735,7 +1735,7 @@ def test_run_conversation_compresses_mid_turn_before_output_budget_exhaustion(mo
 
     compress_calls = []
 
-    def _fake_compress_context(messages, system_message, *, approx_tokens=None, task_id="default", focus_topic=None):
+    def _fake_compress_context(messages, system_message, *, approx_tokens=None, task_id="default", focus_topic=None, trigger=None):
         compress_calls.append(approx_tokens)
         return [
             {"role": "user", "content": "[summary of prior tool-heavy work]"},
@@ -1799,7 +1799,7 @@ def test_mid_turn_compaction_does_not_double_persist_in_place_rows(monkeypatch,
                 {"role": "tool", "tool_call_id": call.id, "content": "x" * 80_000}
             )
 
-    def _fake_compress_context(messages, system_message, *, approx_tokens=None, task_id="default", focus_topic=None):
+    def _fake_compress_context(messages, system_message, *, approx_tokens=None, task_id="default", focus_topic=None, trigger=None):
         # Emulate the real in-place compaction DB side effect: soft-archive the
         # prior rows and insert the compacted set under the SAME session id,
         # then reset the flush identity seed — exactly as archive_and_compact +

website/docs/developer-guide/context-compression-and-caching.md

@@ -226,6 +226,37 @@ attempt anyway:
   the next failure would extend the ladder (#100661). If that attempt fails,
   the cooldown is recorded normally.
 
+#### Attempt telemetry
+
+Every `compress_context` call that returns logs one content-free INFO line,
+`context compression attempt telemetry: {...}` (logger
+`agent.conversation_compression`). It never carries message text, summary
+text, the focus topic or error messages.
+
+- `trigger_source` names why the attempt ran: `manual`, `idle`,
+  `turn_start_threshold`, `engine_preflight`, `pre_api`, `post_tool`,
+  `gateway_hygiene` or `overflow`.
+- `route` is `hermes` for the local compressor and `codex_app_server` when the
+  Codex thread owns compaction.
+- `commit_status` is `committed`, `aborted`, `failed`, `skipped` or `blocked`.
+  A `blocked` attempt was refused by an automatic guard; its `failure_class`
+  names the guard (`blocked:cooldown`, `blocked:structural_backoff`,
+  `blocked:ineffective`).
+- `method` says how the summary was produced: `llm_summary`,
+  `aux_fallback_main` (the summary model failed and the main model wrote it),
+  `deterministic_fallback` (static anchors summary; `items_dropped` counts the
+  replaced messages), `provider` (Codex) or `none`.
+- `messages_before` / `messages_after` and `tokens_before` / `tokens_after` /
+  `tokens_reclaimed` describe the compressor's rewrite.
+  `token_count_method: estimate_rough` marks them as rough message-only
+  estimates (system prompt and tool schemas excluded), so they compare like
+  for like. `tool_results_pruned` and `reasoning_items_pruned` count the
+  deterministic passes that ran before and after the summary.
+
+The opt-in `hermes.compression.count` shared metric keeps coarse triggers
+(`auto`, `manual`, `overflow`) and counts only attempts the local compressor
+ran; blocked and Codex-routed attempts appear in the log line only.
+
 
 ## Configuration