diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 7cbc3879b5e..59e97d48d13 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -1300,6 +1300,8 @@ class ContextCompressor(ContextEngine): self._micro_compact_rolling_summary = "" self._micro_compact_consecutive_failures = 0 self._micro_compact_last_failure_cursor = -1 + self._micro_compact_passes = 0 + self._micro_compact_tokens_saved_total = 0 def _begin_compression_telemetry( self, @@ -2155,6 +2157,8 @@ class ContextCompressor(ContextEngine): self._micro_compact_consecutive_failures: int = 0 self._micro_compact_last_failure_cursor: int = -1 self._micro_compact_defrag_threshold_tokens: int = 2000 + self._micro_compact_passes: int = 0 + self._micro_compact_tokens_saved_total: int = 0 # Defer context-length resolution to first access (#32221): # get_model_context_length() can issue a synchronous /models HTTP @@ -5332,6 +5336,15 @@ This compaction should PRIORITISE preserving all information related to the focu exchange_start, exchange_end = exchange + # Baseline for telemetry. Taken only once an exchange is in hand, so + # turns that no-op early don't pay for the scan. + _started_at = time.monotonic() + _tokens_before = estimate_messages_tokens_rough(messages) + _messages_before = n_messages + + def _elapsed_ms() -> int: + return int((time.monotonic() - _started_at) * 1000) + # Check for defrag trigger if self._needs_defrag(): self._defrag_rolling_summary(messages, exchange_start, compress_end) @@ -5339,10 +5352,19 @@ This compaction should PRIORITISE preserving all information related to the focu self._sync_micro_compact_to_db(result) self._micro_compact_consecutive_failures = 0 self._micro_compact_last_failure_cursor = -1 + self._emit_micro_compaction_telemetry( + outcome="defrag", + messages_before=_messages_before, + messages_after=len(result), + tokens_before=_tokens_before, + tokens_after=estimate_messages_tokens_rough(result), + duration_ms=_elapsed_ms(), + ) return result # Micro-summarize one exchange exchange_text = self._serialize_one_exchange(messages, exchange_start, exchange_end) + _exchange_tokens = estimate_tokens_rough(exchange_text) updated_summary = self._micro_summarize_one(exchange_text) if updated_summary is None: # Track consecutive failures on the same cursor position so we @@ -5366,6 +5388,18 @@ This compaction should PRIORITISE preserving all information related to the focu self._micro_compact_cursor = exchange_end self._micro_compact_consecutive_failures = 0 self._micro_compact_last_failure_cursor = -1 + _outcome = "exchange_skipped" + else: + _outcome = "summarize_failed" + self._emit_micro_compaction_telemetry( + outcome=_outcome, + messages_before=_messages_before, + messages_after=len(messages), + tokens_before=_tokens_before, + tokens_after=_tokens_before, + exchange_tokens=_exchange_tokens, + duration_ms=_elapsed_ms(), + ) return messages self._micro_compact_rolling_summary = updated_summary @@ -5375,8 +5409,70 @@ This compaction should PRIORITISE preserving all information related to the focu result = self._splice_micro_compact_result(messages, exchange_start, exchange_end) self._sync_micro_compact_to_db(result) + self._emit_micro_compaction_telemetry( + outcome="absorbed", + messages_before=_messages_before, + messages_after=len(result), + tokens_before=_tokens_before, + tokens_after=estimate_messages_tokens_rough(result), + exchange_tokens=_exchange_tokens, + duration_ms=_elapsed_ms(), + ) return result + def _emit_micro_compaction_telemetry( + self, + *, + outcome: str, + messages_before: int, + messages_after: int, + tokens_before: int | None, + tokens_after: int | None, + exchange_tokens: int | None = None, + duration_ms: int | None = None, + ) -> None: + """Emit one content-free JSON log line describing a micro-compaction pass. + + Mirrors ``_emit_compression_attempt_telemetry`` for the batch path. + Message counts move by one or two even when the saving is large, so the + token fields are the ones that actually answer "is this helping?". + ``tokens_delta`` is negative when the pass shrank the transcript, and + the ``*_total`` fields accumulate across the session so a whole run can + be summarised from the last line alone. + """ + try: + delta = None + if tokens_before is not None and tokens_after is not None: + delta = tokens_after - tokens_before + self._micro_compact_tokens_saved_total -= delta + self._micro_compact_passes += 1 + payload = { + "event": "micro_compaction", + "session_id": getattr(self, "_session_id", "") or "", + "outcome": outcome, + "messages_before": messages_before, + "messages_after": messages_after, + "tokens_before": _safe_int(tokens_before), + "tokens_after": _safe_int(tokens_after), + "tokens_delta": _safe_int(delta), + "exchange_tokens": _safe_int(exchange_tokens), + "rolling_summary_tokens": estimate_tokens_rough( + self._micro_compact_rolling_summary + ), + "cursor": _safe_int(self._micro_compact_cursor), + "passes_total": self._micro_compact_passes, + "tokens_saved_total": self._micro_compact_tokens_saved_total, + "duration_ms": _safe_int(duration_ms), + "main_model": self.model or "", + "aux_model": self.summary_model or "", + } + logger.info( + "micro compaction telemetry: %s", + json.dumps(payload, sort_keys=True, separators=(",", ":")), + ) + except Exception as exc: + logger.debug("failed to emit micro-compaction telemetry: %s", exc) + def _sync_micro_compact_to_db( self, compacted_messages: List[Dict[str, Any]], diff --git a/docs/micro-compaction.md b/docs/micro-compaction.md index 3fa9aa6049a..19419dab784 100644 --- a/docs/micro-compaction.md +++ b/docs/micro-compaction.md @@ -160,7 +160,47 @@ compression: Set it to `false` to disable micro-compaction and return to batch-only compaction. Everything else about compression is unchanged. -## What you'll see in the logs +## Measuring it + +Every pass emits one content-free JSON line, in the same style as the batch +compaction telemetry: + +``` +micro compaction telemetry: {"event":"micro_compaction","outcome":"absorbed", +"tokens_before":12739,"tokens_after":12060,"tokens_delta":-679, +"exchange_tokens":868,"rolling_summary_tokens":31,"passes_total":1, +"tokens_saved_total":679,"duration_ms":14,...} +``` + +`tokens_delta` is negative when the pass shrank the transcript. +`tokens_saved_total` and `passes_total` accumulate across the session, so a whole +run can be summarised from its last line. No transcript content appears in the +payload — only counts. + +To turn a log into an answer: + +``` +python scripts/micro_compaction_report.py [--per-session] [LOGFILE ...] +``` + +Defaults to `$HERMES_HOME/logs/agent.log`. It reports passes, outcome mix, net +tokens saved, mean absorbed-exchange size and pass durations. + +### Reading the numbers honestly + +**The first pass in a session usually costs tokens rather than saving them.** +Inserting the summary marker carries a fixed ~400 tokens of scaffolding — the +compaction preamble, the historical heading, the end marker — and on pass one +that is paid against a single absorbed exchange. A first pass showing +`tokens_delta: +330` is not a malfunction. + +From the second pass on, the marker is *replaced* rather than added, so the +scaffolding is already paid for and each absorbed exchange is close to pure +saving. The break-even is normally the second or third pass. This is why the +per-session view matters more than any single line: judge the feature on a +session's trajectory, not on one turn. + +The plainer human-readable lines are still there too: ``` Micro-compaction: 37 -> 36 messages @@ -168,7 +208,7 @@ Micro-compaction defrag: rolling summary re-summarized (1843 chars) Micro-compaction: skipping exchange at cursor 12 after 3 consecutive failures ``` -Message counts move by small amounts — that's the point. The token count is where +Message counts move by small amounts — that's expected. The token count is where the effect shows: absorbing one tool-heavy exchange can drop hundreds of tokens while changing the message count by one or two. diff --git a/scripts/micro_compaction_report.py b/scripts/micro_compaction_report.py new file mode 100644 index 00000000000..6d65a0c133a --- /dev/null +++ b/scripts/micro_compaction_report.py @@ -0,0 +1,132 @@ +#!/usr/bin/env python3 +"""Summarize micro-compaction telemetry from Hermes logs. + +Reads the content-free ``micro compaction telemetry:`` JSON lines emitted by +``ContextCompressor._emit_micro_compaction_telemetry`` and reports what the +feature actually did, per session and overall. + +Usage: + python scripts/micro_compaction_report.py [LOGFILE ...] + python scripts/micro_compaction_report.py --per-session + +With no LOGFILE, reads ``$HERMES_HOME/logs/agent.log`` (default ~/.hermes). + +Note on reading the numbers: the first pass in a session inserts the summary +marker, which costs a fixed ~400 tokens of scaffolding. That overhead is paid +once; from the second pass on the marker is replaced rather than added, so +each absorbed exchange is pure saving. A session with a single pass can +therefore show a net loss and still be working correctly. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from collections import defaultdict +from pathlib import Path + +MARKER = "micro compaction telemetry: " + + +def default_log() -> Path: + home = os.environ.get("HERMES_HOME") or str(Path.home() / ".hermes") + return Path(home) / "logs" / "agent.log" + + +def load(paths: list[Path]) -> list[dict]: + events: list[dict] = [] + for path in paths: + try: + text = path.read_text(encoding="utf-8", errors="replace") + except OSError as exc: + print(f"warning: cannot read {path}: {exc}", file=sys.stderr) + continue + for line in text.splitlines(): + idx = line.find(MARKER) + if idx == -1: + continue + try: + events.append(json.loads(line[idx + len(MARKER):])) + except ValueError: + continue + return events + + +def fmt(n: int | None) -> str: + return "-" if n is None else f"{n:,}" + + +def report(events: list[dict], per_session: bool) -> int: + if not events: + print("No micro-compaction telemetry found.") + print("Micro-compaction may be disabled (compression.micro_compact),") + print("or no session has run long enough to trigger a pass yet.") + return 1 + + by_session: dict[str, list[dict]] = defaultdict(list) + for e in events: + by_session[e.get("session_id") or "(unknown)"].append(e) + + outcomes: dict[str, int] = defaultdict(int) + for e in events: + outcomes[e.get("outcome", "?")] += 1 + + saved = sum(-(e.get("tokens_delta") or 0) for e in events) + absorbed = [e for e in events if e.get("outcome") == "absorbed"] + exchange_tokens = [e.get("exchange_tokens") or 0 for e in absorbed] + durations = [e.get("duration_ms") or 0 for e in events] + + if per_session: + print(f"{'session':<38} {'passes':>7} {'saved':>10} {'first':>8} {'last':>8}") + print("-" * 76) + for sid, evs in sorted(by_session.items(), key=lambda kv: -len(kv[1])): + s = sum(-(e.get("tokens_delta") or 0) for e in evs) + first = evs[0].get("tokens_before") + last = evs[-1].get("tokens_after") + print(f"{sid[:38]:<38} {len(evs):>7} {s:>+10,} {fmt(first):>8} {fmt(last):>8}") + print() + + print(f"sessions {len(by_session):,}") + print(f"passes {len(events):,}") + for name, count in sorted(outcomes.items(), key=lambda kv: -kv[1]): + print(f" {name:<20} {count:,}") + print(f"net tokens saved {saved:+,}") + if absorbed: + print(f"exchanges absorbed {len(absorbed):,}") + print(f" mean exchange size {sum(exchange_tokens) // len(absorbed):,} tokens") + print(f" mean saving/pass {saved // len(events):+,} tokens") + if durations: + ordered = sorted(durations) + print(f"pass duration median {ordered[len(ordered) // 2]:,} ms") + print(f" max {ordered[-1]:,} ms") + + multi = {s: e for s, e in by_session.items() if len(e) > 1} + if multi: + net = sum( + (evs[-1].get("tokens_after") or 0) - (evs[0].get("tokens_before") or 0) + for evs in multi.values() + ) + print() + print(f"sessions with >1 pass {len(multi):,}") + print(f" net context change {net:+,} tokens (negative is shrinkage)") + return 0 + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("logs", nargs="*", type=Path, help="log files (default: agent.log)") + ap.add_argument("--per-session", action="store_true", help="break down by session") + args = ap.parse_args() + + paths = args.logs or [default_log()] + missing = [p for p in paths if not p.exists()] + for p in missing: + print(f"warning: {p} does not exist", file=sys.stderr) + return report(load([p for p in paths if p.exists()]), args.per_session) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/agent/test_micro_compaction.py b/tests/agent/test_micro_compaction.py index f6f897de088..f940ff71df8 100644 --- a/tests/agent/test_micro_compaction.py +++ b/tests/agent/test_micro_compaction.py @@ -186,6 +186,75 @@ class TestMicroCompaction: assert len(_summary_markers(messages)) == 1 assert after < before, f"context grew: {before} -> {after}" + def test_emits_content_free_token_telemetry(self, caplog): + """Each pass logs one JSON line with the token accounting. + + Message counts barely move even when the saving is large, so the token + fields are what make the effect measurable in a real session. + """ + import json + import logging + + cc = _compressor() + messages = _conversation(exchanges=8) + + with caplog.at_level(logging.INFO, logger="agent.context_compressor"): + result = cc._micro_compact(messages) + + lines = [ + r.getMessage() for r in caplog.records + if "micro compaction telemetry:" in r.getMessage() + ] + assert len(lines) == 1 + payload = json.loads(lines[0].split("micro compaction telemetry: ", 1)[1]) + + assert payload["event"] == "micro_compaction" + assert payload["outcome"] == "absorbed" + assert payload["tokens_saved_total"] == -payload["tokens_delta"] + assert payload["passes_total"] == 1 + assert payload["messages_after"] == len(result) + assert payload["exchange_tokens"] > 0 + # Content-free: no transcript text may ride along in the payload. + blob = json.dumps(payload) + assert "answer 0" not in blob and "question 0" not in blob + + def test_first_pass_costs_marker_overhead_then_pays_it_back(self): + """The first pass can grow the transcript; later passes recover it. + + Inserting the summary marker costs a fixed ~400 tokens of scaffolding + (the compaction preamble, the historical heading and the end marker). + On pass one that overhead is paid against a single absorbed exchange, + so the net can be positive. From pass two on the marker is replaced + rather than added, so the scaffolding is already paid for and each + absorbed exchange is pure saving. Anyone reading a single turn's + telemetry needs to know this before concluding it made things worse. + """ + from agent.model_metadata import estimate_messages_tokens_rough + + cc = _compressor() + messages = _conversation(exchanges=10) + start = estimate_messages_tokens_rough(messages) + + messages = cc._micro_compact(messages) + after_first = estimate_messages_tokens_rough(messages) + + for _ in range(5): + messages = cc._micro_compact(messages) + after_many = estimate_messages_tokens_rough(messages) + + assert after_first > start, "expected one-time marker overhead" + assert after_many < after_first, "later passes must recover it" + + def test_cumulative_savings_accumulate_across_passes(self): + cc = _compressor() + messages = _conversation(exchanges=10) + + for _ in range(4): + messages = cc._micro_compact(messages) + + assert cc._micro_compact_passes == 4 + assert cc._micro_compact_tokens_saved_total > 0 + def test_defrag_triggers_once_the_rolling_summary_grows(self): cc = _compressor(summary="FRESH DEFRAGGED SUMMARY") cc._micro_compact_rolling_summary = "x" * 40_000 # far over the threshold