fix(compression): report committed effects without leaking metric state

Finalize effect counts from the transcript that crosses the commit boundary, including retained partial-compression tails and boundary anchors. Explicitly discard bypassed shared-metric state so a later refusal cannot consume stale attempt metadata.

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

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",
+        )

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:

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(

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:

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."""

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