feat: log compression attempt telemetry

agent/context_compressor.py

@@ -22,6 +22,7 @@ import logging
 import sqlite3
 import re
 import time
+import uuid
 from typing import Any, Dict, List, Optional
 
 from agent.auxiliary_client import call_llm, _is_connection_error, aux_interrupt_protection
@@ -40,6 +41,14 @@ from tools.todo_tool import TODO_INJECTION_HEADER
 logger = logging.getLogger(__name__)
 
 
+def _safe_int(value: Any) -> int | None:
+    """Best-effort integer coercion for telemetry fields."""
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return None
+
+
 _SUMMARY_PERMANENT_QUOTA_MARKERS: tuple[str, ...] = (
     "insufficient_quota",
     "quota exceeded",
@@ -962,6 +971,102 @@ class ContextCompressor(ContextEngine):
         self.last_compression_rough_tokens = 0
         self.last_rough_tokens_when_real_prompt_fit = 0
         self.awaiting_real_usage_after_compression = False
+        self._last_compression_telemetry = None
+        self._active_compression_telemetry = None
+        self._compression_telemetry_seed = None
+
+    def _begin_compression_telemetry(
+        self,
+        *,
+        current_tokens: int | None,
+        attempt_id: str | None = None,
+        session_id: str | None = None,
+        trigger_source: str | None = None,
+    ) -> Dict[str, Any]:
+        """Initialize content-free per-attempt compression telemetry."""
+        seed = getattr(self, "_compression_telemetry_seed", None)
+        if isinstance(seed, dict):
+            attempt_id = attempt_id or seed.get("attempt_id")
+            session_id = session_id or seed.get("session_id")
+            trigger_source = trigger_source or seed.get("trigger_source")
+        telemetry: Dict[str, Any] = {
+            "event": "compression_attempt",
+            "attempt_id": attempt_id or uuid.uuid4().hex,
+            "session_id": session_id or "",
+            "trigger_source": trigger_source or "unknown",
+            "main_provider": self.provider or "",
+            "main_model": self.model or "",
+            "main_context_limit": _safe_int(self.context_length),
+            "current_estimated_tokens": _safe_int(current_tokens),
+            "effective_threshold": _safe_int(self.threshold_tokens),
+            "protected_head_tokens": None,
+            "protected_tail_tokens": None,
+            "middle_window_tokens": None,
+            "aux_prompt_tokens": None,
+            "aux_output_reservation": None,
+            "aux_provider": "",
+            "aux_model": "",
+            "effective_aux_context": None,
+            "fit_margin": None,
+            "chunking": False,
+            "chunk_count": 0,
+            "total_duration_ms": None,
+            "aux_call_duration_ms": None,
+            "fallback_used": False,
+            "commit_status": "unknown",
+            "split_status": "unknown",
+            "failure_class": None,
+        }
+        self._active_compression_telemetry = telemetry
+        self._last_compression_telemetry = telemetry
+        return telemetry
+
+    def _record_compression_regions(
+        self,
+        *,
+        head_messages: List[Dict[str, Any]],
+        middle_messages: List[Dict[str, Any]],
+        tail_messages: List[Dict[str, Any]],
+    ) -> None:
+        telemetry = getattr(self, "_active_compression_telemetry", None)
+        if not isinstance(telemetry, dict):
+            return
+        telemetry["protected_head_tokens"] = estimate_messages_tokens_rough(head_messages)
+        telemetry["middle_window_tokens"] = estimate_messages_tokens_rough(middle_messages)
+        telemetry["protected_tail_tokens"] = estimate_messages_tokens_rough(tail_messages)
+
+    def _record_aux_compression_call(
+        self,
+        *,
+        prompt_messages: List[Dict[str, Any]],
+        max_tokens: int,
+        duration_ms: int,
+        aux_provider: str | None = None,
+        aux_model: str | None = None,
+        effective_aux_context: int | None = None,
+    ) -> None:
+        telemetry = getattr(self, "_active_compression_telemetry", None)
+        if not isinstance(telemetry, dict):
+            return
+        telemetry["aux_prompt_tokens"] = estimate_messages_tokens_rough(prompt_messages)
+        telemetry["aux_output_reservation"] = _safe_int(max_tokens)
+        if aux_provider:
+            telemetry["aux_provider"] = aux_provider
+        if aux_model:
+            telemetry["aux_model"] = aux_model
+        if effective_aux_context is not None:
+            telemetry["effective_aux_context"] = _safe_int(effective_aux_context)
+        if (
+            telemetry["effective_aux_context"] is not None
+            and telemetry["aux_prompt_tokens"] is not None
+        ):
+            telemetry["fit_margin"] = (
+                telemetry["effective_aux_context"]
+                - telemetry["aux_prompt_tokens"]
+                - (telemetry["aux_output_reservation"] or 0)
+            )
+        previous = telemetry.get("aux_call_duration_ms") or 0
+        telemetry["aux_call_duration_ms"] = previous + max(0, int(duration_ms))
 
     def on_session_end(self, session_id: str, messages: List[Dict[str, Any]]) -> None:
         """Clear all per-session compaction state at a real session boundary.
@@ -1004,6 +1109,9 @@ class ContextCompressor(ContextEngine):
         self.last_compression_rough_tokens = 0
         self.last_rough_tokens_when_real_prompt_fit = 0
         self.awaiting_real_usage_after_compression = False
+        self._last_compression_telemetry = None
+        self._active_compression_telemetry = None
+        self._compression_telemetry_seed = None
 
     def bind_session_state(self, session_db: Any = None, session_id: str = "") -> None:
         """Bind the current session row so durable cooldowns can round-trip."""
@@ -1590,6 +1698,9 @@ class ContextCompressor(ContextEngine):
         # succeeded.  Silent recovery would hide the broken config.
         self._last_aux_model_failure_error: Optional[str] = None
         self._last_aux_model_failure_model: Optional[str] = None
+        self._last_compression_telemetry: Optional[Dict[str, Any]] = None
+        self._active_compression_telemetry: Optional[Dict[str, Any]] = None
+        self._compression_telemetry_seed: Optional[Dict[str, Any]] = None
 
     def update_from_response(self, usage: Dict[str, Any]):
         """Update tracked token usage from API response."""
@@ -2277,6 +2388,10 @@ Summary generation was unavailable, so this is a best-effort deterministic fallb
             _err_text = _err_text[:217].rstrip() + "..."
         self._last_aux_model_failure_error = _err_text
         self._last_aux_model_failure_model = self.summary_model
+        telemetry = getattr(self, "_active_compression_telemetry", None)
+        if isinstance(telemetry, dict):
+            telemetry["fallback_used"] = True
+            telemetry["failure_class"] = telemetry.get("failure_class") or "aux_model_fallback"
         self.summary_model = ""  # empty = use main model
         self._clear_compression_failure_cooldown()  # no cooldown — retry immediately
 
@@ -2577,13 +2692,40 @@ This compaction should PRIORITISE preserving all information related to the focu
             }
             if self.summary_model:
                 call_kwargs["model"] = self.summary_model
+            _aux_provider = ""
+            _aux_model = self.summary_model or ""
+            _aux_context = None
+            try:
+                from agent.auxiliary_client import _resolve_task_provider_model
+
+                _resolved_provider, _resolved_model, _, _, _ = _resolve_task_provider_model(
+                    "compression",
+                    model=(self.summary_model or ""),
+                )
+                _aux_provider = _resolved_provider or ""
+                _aux_model = _resolved_model or _aux_model or self.model or ""
+                if _aux_model == self.model:
+                    _aux_context = self.context_length
+            except Exception:
+                pass
             # Compression is atomic: protect the in-flight summary call from a
             # mid-turn gateway interrupt. Without this, an incoming user message
             # aborts the summary and compression falls back to a degraded static
             # marker, losing the real handoff (#23975). Re-entrant: a main-model
             # retry (_generate_summary recursion) re-enters harmlessly.
-            with aux_interrupt_protection():
-                response = call_llm(**call_kwargs)
+            _aux_call_start = time.monotonic()
+            try:
+                with aux_interrupt_protection():
+                    response = call_llm(**call_kwargs)
+            finally:
+                self._record_aux_compression_call(
+                    prompt_messages=call_kwargs["messages"],
+                    max_tokens=call_kwargs["max_tokens"],
+                    duration_ms=int((time.monotonic() - _aux_call_start) * 1000),
+                    aux_provider=_aux_provider,
+                    aux_model=_aux_model,
+                    effective_aux_context=_aux_context,
+                )
             # ``_validate_llm_response`` only guarantees ``choices[0].message``
             # exists, not that it's an object with ``.content``. Some
             # OpenAI-compatible proxies / local backends return a dict- or
@@ -3814,6 +3956,8 @@ This compaction should PRIORITISE preserving all information related to the focu
         # static-fallback — the exact data-loss #29559 describes.  Letting them
         # persist across compress() calls is safe because a successful summary
         # always clears both.
+        telemetry = self._begin_compression_telemetry(current_tokens=current_tokens)
+        telemetry["chunk_count"] = 0
 
         # Manual /compress (force=True) bypasses the failure cooldown so the
         # user can retry immediately after an auto-compress abort.  Without
@@ -3833,6 +3977,7 @@ This compaction should PRIORITISE preserving all information related to the focu
             # returns here unchanged, and the CLI appears frozen.
             self._ineffective_compression_count += 1
             self._last_compression_savings_pct = 0.0
+            telemetry["failure_class"] = "insufficient_messages"
             if not self.quiet_mode:
                 logger.warning(
                     "Cannot compress: only %d messages (need > %d). "
@@ -3887,6 +4032,12 @@ This compaction should PRIORITISE preserving all information related to the focu
                 compress_end = bridge_idx
 
         if compress_start >= compress_end:
+            self._record_compression_regions(
+                head_messages=messages[:compress_start],
+                middle_messages=[],
+                tail_messages=messages[compress_end:],
+            )
+            telemetry["failure_class"] = "no_compressible_window"
             # No compressable window — the entire transcript fits within
             # the tail budget (soft_ceiling).  Without recording this as
             # an ineffective compression the anti-thrashing guard in
@@ -3948,6 +4099,13 @@ This compaction should PRIORITISE preserving all information related to the focu
         else:
             self._summary_has_user_turn = real_user_present
 
+        self._record_compression_regions(
+            head_messages=messages[:compress_start],
+            middle_messages=turns_to_summarize,
+            tail_messages=messages[compress_end:],
+        )
+        telemetry["chunk_count"] = 1 if turns_to_summarize else 0
+
         if not self.quiet_mode:
             logger.info(
                 "Context compression triggered (%d tokens >= %d threshold)",
@@ -4007,6 +4165,12 @@ This compaction should PRIORITISE preserving all information related to the focu
             self._last_summary_dropped_count = 0  # nothing actually dropped
             self._last_summary_fallback_used = False
             self._last_compress_aborted = True
+            if self._last_summary_auth_failure:
+                telemetry["failure_class"] = "summary_auth_failure"
+            elif self._last_summary_network_failure:
+                telemetry["failure_class"] = "summary_network_failure"
+            else:
+                telemetry["failure_class"] = "summary_generation_aborted"
             if not self.quiet_mode:
                 if self._last_summary_auth_failure:
                     logger.warning(
@@ -4071,6 +4235,8 @@ This compaction should PRIORITISE preserving all information related to the focu
             n_dropped = compress_end - compress_start
             self._last_summary_dropped_count = n_dropped
             self._last_summary_fallback_used = True
+            telemetry["fallback_used"] = True
+            telemetry["failure_class"] = telemetry.get("failure_class") or "summary_generation_failed"
             summary = self._build_static_fallback_summary(
                 turns_to_summarize,
                 reason=self._last_summary_error,

agent/conversation_compression.py

@@ -30,10 +30,12 @@ from __future__ import annotations
 
 import copy
 import inspect
+import json
 import logging
 import math
 import os
 import tempfile
+import time
 import uuid
 import threading
 from datetime import datetime
@@ -173,6 +175,43 @@ def _session_was_rotated_by_compression(session_db: Any, session_id: str) -> boo
     )
 
 
+def _emit_compression_attempt_telemetry(
+    agent: Any,
+    *,
+    started_at: float,
+    commit_status: str,
+    split_status: str,
+    failure_class: str | None = None,
+) -> None:
+    """Emit one content-free JSON log line for a compression attempt."""
+    try:
+        telemetry = getattr(agent.context_compressor, "_last_compression_telemetry", None)
+        if not isinstance(telemetry, dict):
+            telemetry = {}
+        payload = dict(telemetry)
+        payload.setdefault("event", "compression_attempt")
+        payload.setdefault("attempt_id", getattr(agent, "_compression_attempt_id", "") or uuid.uuid4().hex)
+        payload.setdefault("session_id", getattr(agent, "session_id", "") or "")
+        payload["total_duration_ms"] = int((time.monotonic() - started_at) * 1000)
+        payload["commit_status"] = commit_status
+        payload["split_status"] = split_status
+        if failure_class:
+            payload["failure_class"] = failure_class
+        payload.setdefault("chunking", False)
+        payload.setdefault("chunk_count", 0)
+        payload["fallback_used"] = bool(
+            payload.get("fallback_used")
+            or getattr(agent.context_compressor, "_last_summary_fallback_used", False)
+            or getattr(agent.context_compressor, "_last_aux_model_failure_model", None)
+        )
+        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)
+
+
 def _compression_lock_holder(agent: Any) -> str:
     """Build a unique holder id for the lock: pid:tid:agent-instance:uuid.
 
@@ -904,6 +943,19 @@ def compress_context(
     ):
         raise RuntimeError("a compression notification is already pending")
 
+    _attempt_started_at = time.monotonic()
+    _attempt_id = uuid.uuid4().hex
+    _trigger_source = "manual" if force else "auto"
+    try:
+        agent._compression_attempt_id = _attempt_id
+        setattr(agent.context_compressor, "_compression_telemetry_seed", {
+            "attempt_id": _attempt_id,
+            "session_id": agent.session_id or "",
+            "trigger_source": _trigger_source,
+        })
+    except Exception:
+        pass
+
     # Codex app-server sessions: the codex agent owns the real thread context;
     # Hermes' summarizer would only rewrite a local mirror without shrinking
     # the actual thread (#36801). Route compaction to the app server's own
@@ -1110,6 +1162,18 @@ def compress_context(
             _existing_sp = getattr(agent, "_cached_system_prompt", None)
             if not _existing_sp:
                 _existing_sp = agent._build_system_prompt(system_message)
+            try:
+                if hasattr(agent.context_compressor, "_begin_compression_telemetry"):
+                    agent.context_compressor._begin_compression_telemetry(current_tokens=approx_tokens)
+            except Exception:
+                pass
+            _emit_compression_attempt_telemetry(
+                agent,
+                started_at=_attempt_started_at,
+                commit_status="aborted",
+                split_status="aborted",
+                failure_class="lock_contended",
+            )
             return messages, _existing_sp
     _lock_released = False
 
@@ -1235,7 +1299,7 @@ def compress_context(
         messages_before_compression = copy.deepcopy(messages)
         _activity_heartbeat = _CompressionActivityHeartbeat(agent).start()
         compressed = compress_fn(messages, **compress_kwargs)
-    except BaseException:
+    except BaseException as _compress_exc:
         # ANY exception after lock acquisition — memory hook, capability
         # inspection, engine lookup, or compress() — must release the lock so
         # the session isn't permanently blocked from future compression.
@@ -1243,6 +1307,13 @@ def compress_context(
             _activity_heartbeat.stop("context compression failed")
             _activity_heartbeat = None
         _release_lock()
+        _emit_compression_attempt_telemetry(
+            agent,
+            started_at=_attempt_started_at,
+            commit_status="aborted",
+            split_status="aborted",
+            failure_class=f"exception:{type(_compress_exc).__name__}",
+        )
         raise
     finally:
         if _activity_heartbeat is not None:
@@ -1278,6 +1349,16 @@ def compress_context(
                 _existing_sp = getattr(agent, "_cached_system_prompt", None)
                 if not _existing_sp:
                     _existing_sp = agent._build_system_prompt(system_message)
+                _emit_compression_attempt_telemetry(
+                    agent,
+                    started_at=_attempt_started_at,
+                    commit_status="aborted",
+                    split_status="aborted",
+                    failure_class=(
+                        getattr(agent.context_compressor, "_last_summary_error", None)
+                        and "summary_generation_aborted"
+                    ),
+                )
                 return messages, _existing_sp
             finally:
                 _release_lock()
@@ -1296,6 +1377,13 @@ def compress_context(
             _existing_sp = getattr(agent, "_cached_system_prompt", None)
             if not _existing_sp:
                 _existing_sp = agent._build_system_prompt(system_message)
+            _emit_compression_attempt_telemetry(
+                agent,
+                started_at=_attempt_started_at,
+                commit_status="aborted",
+                split_status="aborted",
+                failure_class="no_progress",
+            )
             _release_lock()
             return messages, _existing_sp
 
@@ -1377,7 +1465,9 @@ def compress_context(
             agent._cached_system_prompt = new_system_prompt
 
         _session_commit_succeeded = False
+        split_status = "not_applicable"
         if agent._session_db:
+            split_status = "pending"
             try:
                 # Trigger memory extraction on the current session before the
                 # transcript is rewritten (runs in BOTH modes — the logical
@@ -1405,6 +1495,7 @@ def compress_context(
                     # WITHOUT destroying history, unlike a hard replace_messages).
                     # See #38763.
                     agent._session_db.archive_and_compact(agent.session_id, compressed)
+                    split_status = "in_place_committed"
                     # Reset the flush identity set so the next turn's appends are
                     # diffed against the COMPACTED transcript: the compacted dicts
                     # are passed as conversation_history next turn and skipped by
@@ -1521,6 +1612,7 @@ def compress_context(
                         agent._session_db_created = True
                         raise
                     agent._session_db_created = True
+                    split_status = "rotated_committed"
                     # Carry a persistent /goal onto the continuation session.
                     # Compression mints a fresh child id; load_goal does a flat
                     # per-session lookup with no parent walk, so without this an
@@ -1559,6 +1651,7 @@ def compress_context(
                     }
                 _session_commit_succeeded = True
             except Exception as e:
+                split_status = "aborted" if locals().get("old_session_id") is None and not in_place else "failed_not_indexed"
                 # If the rotation rolled back to the parent (orphan-avoidance
                 # above), agent.session_id is the still-indexed parent and
                 # old_session_id was cleared — so this is recovery, not an
@@ -1701,6 +1794,18 @@ def compress_context(
             agent.session_id or "none", _pre_msg_count, len(compressed),
             f"{_compressed_est:,}",
         )
+        _commit_status = "committed" if split_status in {"not_applicable", "in_place_committed", "rotated_committed"} else "aborted"
+        _emit_compression_attempt_telemetry(
+            agent,
+            started_at=_attempt_started_at,
+            commit_status=_commit_status,
+            split_status=split_status,
+            failure_class=(
+                "session_split_failed"
+                if split_status in {"failed_not_indexed", "aborted"}
+                else None
+            ),
+        )
         return compressed, new_system_prompt
     finally:
         # Release the lock on the OLD session_id only AFTER rotation completed

tests/agent/test_compression_attempt_telemetry.py

@@ -0,0 +1,148 @@
+import json
+import logging
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from agent.conversation_compression import compress_context
+from agent.context_compressor import ContextCompressor
+
+
+class _TodoStore:
+    def format_for_injection(self):
+        return ""
+
+
+class _Agent:
+    def __init__(self, compressor):
+        self.context_compressor = compressor
+        self.session_id = "session-telemetry-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 _messages(secret_text="TOPSECRET_TRANSCRIPT_TEXT"):
+    msgs = [{"role": "system", "content": "system prompt"}]
+    for idx in range(10):
+        msgs.append({"role": "user", "content": f"user message {idx} {secret_text}"})
+        msgs.append({"role": "assistant", "content": f"assistant reply {idx} {secret_text}"})
+    return msgs
+
+
+def _extract_telemetry(caplog):
+    records = [
+        record.getMessage()
+        for record in caplog.records
+        if "context compression attempt telemetry:" in record.getMessage()
+    ]
+    assert len(records) == 1
+    return json.loads(records[0].split("context compression attempt telemetry: ", 1)[1])
+
+
+def test_compression_attempt_telemetry_is_metadata_only(caplog):
+    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
+    agent = _Agent(compressor)
+
+    with patch.object(compressor, "_generate_summary", return_value="SANITIZED SUMMARY"):
+        with caplog.at_level(logging.INFO, logger="agent.conversation_compression"):
+            compressed, system_prompt = compress_context(
+                agent,
+                _messages(),
+                "system prompt",
+                approx_tokens=75_000,
+                force=True,
+            )
+
+    assert system_prompt == "system prompt"
+    assert compressed is not None
+    payload = _extract_telemetry(caplog)
+
+    assert payload["event"] == "compression_attempt"
+    assert payload["attempt_id"]
+    assert payload["session_id"] == "session-telemetry-test"
+    assert payload["trigger_source"] == "manual"
+    assert payload["main_model"] == "test/main-model"
+    assert payload["main_context_limit"] == 100_000
+    assert payload["current_estimated_tokens"] == 75_000
+    assert payload["effective_threshold"] == compressor.threshold_tokens
+    assert payload["protected_head_tokens"] is not None
+    assert payload["protected_tail_tokens"] is not None
+    assert payload["middle_window_tokens"] is not None
+    assert payload["chunking"] is False
+    assert payload["chunk_count"] in {0, 1}
+    assert payload["commit_status"] == "committed"
+    assert payload["split_status"] == "not_applicable"
+    assert payload["fallback_used"] is False
+    assert isinstance(payload["total_duration_ms"], int)
+
+    raw_log = json.dumps(payload)
+    assert "TOPSECRET_TRANSCRIPT_TEXT" not in raw_log
+    assert "SANITIZED SUMMARY" not in raw_log
+    assert "user message" not in raw_log
+    assert "assistant reply" not in raw_log
+
+
+def test_aux_call_telemetry_records_durations_without_content(caplog):
+    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
+    agent = _Agent(compressor)
+    response = SimpleNamespace(
+        choices=[SimpleNamespace(message=SimpleNamespace(content="SANITIZED SUMMARY"))]
+    )
+
+    with patch("agent.context_compressor.call_llm", return_value=response):
+        with caplog.at_level(logging.INFO, logger="agent.conversation_compression"):
+            compress_context(
+                agent,
+                _messages(),
+                "system prompt",
+                approx_tokens=75_000,
+            )
+
+    payload = _extract_telemetry(caplog)
+    assert payload["aux_prompt_tokens"] is not None
+    assert payload["aux_output_reservation"] is not None
+    assert isinstance(payload["aux_call_duration_ms"], int)
+    assert payload["aux_provider"]
+    assert payload["aux_model"]
+
+    raw_log = json.dumps(payload)
+    assert "TOPSECRET_TRANSCRIPT_TEXT" not in raw_log
+    assert "SANITIZED SUMMARY" not in raw_log