diff --git a/agent/context_compressor_telemetry.py b/agent/context_compressor_telemetry.py index d150c4339c..669c953e6c 100644 --- a/agent/context_compressor_telemetry.py +++ b/agent/context_compressor_telemetry.py @@ -50,7 +50,7 @@ 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). + # Candidate effect is captured by the compressor, then replaced with the exact committed shape. "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, @@ -121,3 +121,18 @@ class CompressionTelemetryMixin: token_count_method="estimate_rough", tool_results_pruned=tool_results_pruned, reasoning_items_pruned=reasoning_items_pruned, ) + + def _record_committed_compression_effect( + self, messages_before: list[dict[str, Any]], messages_after: list[dict[str, Any]], + ) -> None: + """Replace candidate counts with the exact transcript shape that crossed the commit boundary.""" + telemetry = getattr(self, "_active_compression_telemetry", None) + if not isinstance(telemetry, dict): + return + tokens_before = estimate_messages_tokens_rough(messages_before) + tokens_after = estimate_messages_tokens_rough(messages_after) + telemetry.update( + messages_before=len(messages_before), messages_after=len(messages_after), + tokens_before=tokens_before, tokens_after=tokens_after, + tokens_reclaimed=tokens_before - tokens_after, token_count_method="estimate_rough", + ) diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 9668357099..ba0bc2b726 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -33,7 +33,7 @@ from agent.context_engine import automatic_compaction_status_message, sanitize_m from agent.conversation_compression_codex import _compress_context_via_codex_app_server from agent.conversation_compression_telemetry import ( _emit_aborted_attempt_telemetry, _emit_blocked_attempt_telemetry, _emit_bypassed_attempt_telemetry, - _emit_compression_attempt_telemetry, + _emit_compression_attempt_telemetry, _record_committed_attempt_effect, _snapshot_compression_effect_input, ) from agent.memory_provider import PRE_COMPRESS_CHECKPOINT_API_VERSION from agent.model_metadata import estimate_messages_tokens_rough, estimate_request_tokens_rough @@ -4257,10 +4257,10 @@ def compress_context( messages, compressed = phase.messages, phase.compressed messages_before_compression = phase.messages_before_compression approx_tokens, _pre_msg_count = phase.approx_tokens, phase.pre_msg_count + effect_messages_before = _snapshot_compression_effect_input(messages_before_compression, verbatim_tail) _commit_fence_entered = False try: - # Capture the verdict before rotation callbacks: lifecycle hooks may reset - # compressor fields on rebind; record only after the full boundary commits. + # Capture the verdict before rotation callbacks reset compressor fields on rebind. _compression_made_progress, _compression_used_fallback, _compression_feasibility_skip = ( bool(getattr(agent.context_compressor, name, False)) for name in ("_last_compression_made_progress", "_last_summary_fallback_used", "_last_feasibility_skip") @@ -4293,8 +4293,7 @@ def compress_context( _warn_summary_or_aux_fallback(agent) # A just-delivered reply the engine folded away must stay live or the # next render drops it from the surface (#118900). It runs FIRST: the - # todo fold rewrites the trailing user row (its follower would no longer - # match) and both later passes place themselves around the tail, so the + # todo fold rewrites the trailing user row, and both later passes place themselves around the tail, so the # reply has to be back in its chronological slot before they look. from agent.conversation_compression_reply_anchor import _ensure_compressed_keeps_last_assistant_reply @@ -4321,7 +4320,7 @@ def compress_context( # The reinserted copy keeps the original's _row_id/timestamp (production flush stamps # both); carry exactly that one row so the commit rewinds the durable original instead # of archiving it compacted=1 next to a fresh twin (display would show it twice). The - # todo fold / user-anchor rows added above are NOT carried: they keep their own class. + # todo fold / user-anchor rows above keep their own class and are not carried. carried_messages=[reinserted_reply] if reinserted_reply is not None else None, ) if commit.refused_prompt is not None: @@ -4344,6 +4343,8 @@ def compress_context( lifecycle.commit_status = ( "committed" if split_status in {"not_applicable", "in_place_committed", "rotated_committed"} else "aborted" ) + if lifecycle.commit_status == "committed": + _record_committed_attempt_effect(agent, effect_messages_before, compressed, verbatim_tail, split_status) _emit_compression_attempt_telemetry( agent, started_at=attempt.started_at, commit_status=lifecycle.commit_status, split_status=split_status, failure_class=("session_split_failed" if split_status in {"failed_not_indexed", "aborted"} else None), @@ -4351,8 +4352,7 @@ def compress_context( ) return compressed, new_system_prompt finally: - # Release the OLD session's lock only after rotation and all post-rotation - # bookkeeping; a waking contender then sees the NEW id and acquires on that. + # Release the old lock after rotation bookkeeping so a waking contender sees the new session id. try: lease.release() finally: diff --git a/agent/conversation_compression_telemetry.py b/agent/conversation_compression_telemetry.py index a4475aaf48..9da06f428b 100644 --- a/agent/conversation_compression_telemetry.py +++ b/agent/conversation_compression_telemetry.py @@ -7,6 +7,7 @@ consumers keep one source. from __future__ import annotations +import copy import json import logging import time @@ -26,6 +27,38 @@ logger = logging.getLogger("agent.conversation_compression") _GENERIC_ABORT_VERDICTS = frozenset({"no_progress", "summary_generation_aborted"}) +def _snapshot_compression_effect_input(messages: list, verbatim_tail: list | None) -> list | None: + """Content-only snapshot for final effect accounting; telemetry must never break compression.""" + try: + if not verbatim_tail: + return messages + from hermes_cli.partial_compress import rejoin_compressed_head_and_tail + + return rejoin_compressed_head_and_tail(messages, copy.deepcopy(verbatim_tail)) + except Exception: + logger.debug("failed to snapshot compression effect input", exc_info=True) + return None + + +def _record_committed_attempt_effect( + agent: Any, messages_before: list | None, messages_after: list, verbatim_tail: list | None, split_status: str, +) -> None: + """Finalize effect fields from the transcript shape that crossed the commit boundary.""" + if messages_before is None: + return + try: + if verbatim_tail and split_status == "not_applicable": + from hermes_cli.partial_compress import rejoin_compressed_head_and_tail + + # Without a SessionDB, the manual-compression caller performs this rejoin after compress_context returns. + messages_after = rejoin_compressed_head_and_tail(messages_after, verbatim_tail) + recorder = getattr(agent.context_compressor, "_record_committed_compression_effect", None) + if callable(recorder): + recorder(messages_before, messages_after) + except Exception: + logger.debug("failed to record committed compression effect", exc_info=True) + + def _attempt_seed( agent: Any, *, attempt_began: bool = True, attempt_seed: dict[str, Any] | None = None, ) -> dict[str, Any]: @@ -138,6 +171,16 @@ def _emit_bypassed_attempt_telemetry( ) except Exception as exc: logger.debug("failed to emit compression attempt telemetry: %s", exc, exc_info=True) + finally: + # compress_context opened the thread-local shared-metric lifecycle before reaching this exit. + # Bypassed attempts are log-only, so retire that lifecycle explicitly; otherwise an unrelated + # pool-saturation emit on the same thread can consume its trigger and token count. + try: + from hermes_cli.observability.shared_metrics_events import discard_compression_attempt + + discard_compression_attempt() + except Exception: + logger.debug("failed to discard bypassed compression metric", exc_info=True) def _emit_blocked_attempt_telemetry( diff --git a/hermes_cli/observability/shared_metrics_events.py b/hermes_cli/observability/shared_metrics_events.py index 5114dd492e..87af4fc9a4 100644 --- a/hermes_cli/observability/shared_metrics_events.py +++ b/hermes_cli/observability/shared_metrics_events.py @@ -76,6 +76,11 @@ def begin_compression_attempt(trigger: str, tokens_before: Any) -> None: _compression_attempt.pending = (trigger, tokens_before) +def discard_compression_attempt() -> None: + """Clear this thread's pending attempt without recording a shared metric.""" + _compression_attempt.pending = None + + def finish_compression_attempt( commit_status: str, failure_class: str | None, context_length: Any, agent: Any = None, ) -> None: diff --git a/tests/agent/test_compression_attempt_effect.py b/tests/agent/test_compression_attempt_effect.py index d5d57674c1..c918ca4a7d 100644 --- a/tests/agent/test_compression_attempt_effect.py +++ b/tests/agent/test_compression_attempt_effect.py @@ -151,6 +151,26 @@ def test_gate_blocked_attempt_logs_one_blocked_record_and_no_shared_metric(caplo assert shared_metric_rows == [] +def test_bypassed_attempt_does_not_leak_shared_metric_state_into_pool_saturation(caplog, shared_metric_rows): + from agent.conversation_compression import run_compress_context_with_progress_timeout + + compressor = _compressor() + agent = _Agent(compressor) + with patch.object(type(compressor), "_compression_block_reason", return_value="cooldown:42"), \ + patch.object(type(compressor), "_automatic_compression_blocked", return_value=True): + compress_context(agent, _messages(), "system prompt", approx_tokens=80_000, trigger="post_tool") + + with patch("agent.conversation_compression._try_admit_compression_job", return_value=False): + with caplog.at_level(logging.INFO, logger="agent.conversation_compression"): + run_compress_context_with_progress_timeout( + worker=lambda _fence: pytest.fail("a refused job must not run"), messages=_messages(), + system_prompt_fallback="system prompt", idle_timeout_seconds=5.0, total_ceiling_seconds=5.0, + telemetry_agent=agent, + ) + + assert shared_metric_rows == [] + + class _CodexSession: def __init__(self, result): self.result = result @@ -365,6 +385,7 @@ def test_session_backed_commit_keeps_the_attempts_own_record(caplog, tmp_path, m session_db=SessionDB(db_path=tmp_path / "state.db"), session_id="db-session", skip_context_files=True, skip_memory=True, ) + agent._session_db.create_session(agent.session_id, "cli", model=agent.model) agent.compression_in_place = in_place agent.context_compressor.tail_token_budget = 10 try: @@ -380,6 +401,44 @@ def test_session_backed_commit_keeps_the_attempts_own_record(caplog, tmp_path, m assert record["tokens_after"] is not None and record["middle_window_tokens"] is not None +@pytest.mark.parametrize("in_place", [True, False]) +def test_committed_effect_includes_the_retained_partial_compression_tail( + caplog, tmp_path, monkeypatch, in_place, +): + from agent.model_metadata import estimate_messages_tokens_rough + from hermes_state import SessionDB + from run_agent import AIAgent + + monkeypatch.setenv("OPENROUTER_API_KEY", "test-key") + agent = AIAgent( + api_key="test-key", base_url="https://openrouter.ai/api/v1", model="test/model", quiet_mode=True, + session_db=SessionDB(db_path=tmp_path / "state.db"), session_id="db-session", + skip_context_files=True, skip_memory=True, + ) + agent._session_db.create_session(agent.session_id, "cli", model=agent.model) + agent.compression_in_place = in_place + agent.context_compressor.tail_token_budget = 10 + messages = _messages() + head, tail = messages[:-4], messages[-4:] + head_tokens = estimate_messages_tokens_rough(head) + try: + with patch.object(agent.context_compressor, "_generate_summary", return_value="SANITIZED SUMMARY"): + with caplog.at_level(logging.INFO, logger="agent.conversation_compression"): + compressed, _ = agent._compress_context( + head, "sys", approx_tokens=80_000, force=True, verbatim_tail=tail, + ) + finally: + agent.close() + + [record] = _attempt_records(caplog) + assert record["commit_status"] == "committed" + assert record["messages_before"] == len(messages) + assert record["messages_after"] == len(compressed) + assert record["tokens_before"] > head_tokens + assert record["tokens_after"] == estimate_messages_tokens_rough(compressed) + assert record["tokens_reclaimed"] == record["tokens_before"] - record["tokens_after"] + + def test_engine_that_returns_an_empty_transcript_logs_one_aborted_record(caplog, shared_metric_rows): """An engine that hands back no messages for a non-empty transcript is refused before any split, and the attempt still logs its one record and counts once.""" diff --git a/website/docs/developer-guide/context-compression-and-caching.md b/website/docs/developer-guide/context-compression-and-caching.md index 1ca45a7c2e..1b6b7c1662 100644 --- a/website/docs/developer-guide/context-compression-and-caching.md +++ b/website/docs/developer-guide/context-compression-and-caching.md @@ -258,7 +258,8 @@ text, the focus topic or error messages. `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. + `tokens_reclaimed` describe the transcript that crossed the commit boundary, + including retained `/compress here N` tail rows and boundary anchors. `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