diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 48228ef2b720..300b7c588745 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -22,6 +22,7 @@ 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 @@ 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", @@ -941,6 +950,102 @@ def on_session_reset(self) -> None: 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 | None, + 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. @@ -983,6 +1088,9 @@ def on_session_end(self, session_id: str, messages: List[Dict[str, Any]]) -> Non 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.""" @@ -1522,6 +1630,9 @@ def __init__( # 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.""" @@ -2209,6 +2320,10 @@ def _fallback_to_main_for_compression(self, e: Exception, reason: str) -> None: _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 @@ -2500,13 +2615,43 @@ def _generate_summary( } 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"], + # Current main intentionally omits max_tokens from the aux + # call (summary_budget is prompt-level guidance only) — + # use .get() so the telemetry hook never breaks the call. + max_tokens=call_kwargs.get("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 @@ -3553,6 +3698,8 @@ def compress( # 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 @@ -3572,6 +3719,7 @@ def compress( # 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). " @@ -3599,6 +3747,12 @@ def compress( compress_end = self._find_tail_cut_by_tokens(messages, compress_start) 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 @@ -3660,6 +3814,13 @@ def compress( 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)", @@ -3719,6 +3880,12 @@ def compress( 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( @@ -3785,6 +3952,8 @@ def compress( 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, diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 6ca6163ac89f..4c105d4906e6 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -30,10 +30,12 @@ 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 _release_lock() -> None: 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 _release_lock() -> None: _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 _release_lock() -> None: _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 _release_lock() -> None: _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 _release_lock() -> None: 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 _release_lock() -> None: # 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 _release_lock() -> None: 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 _release_lock() -> None: } _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 _release_lock() -> None: 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 diff --git a/tests/agent/test_compression_attempt_telemetry.py b/tests/agent/test_compression_attempt_telemetry.py new file mode 100644 index 000000000000..9b2b587d290c --- /dev/null +++ b/tests/agent/test_compression_attempt_telemetry.py @@ -0,0 +1,151 @@ +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 + # Current main intentionally omits max_tokens from the aux summary call + # (the summary budget is prompt-level guidance only), so no output + # reservation is recorded. + assert payload["aux_output_reservation"] is 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