From ac48add3a7fdfcb53d15c5ee7f97db352c692d39 Mon Sep 17 00:00:00 2001 From: Michael Jordan Date: Wed, 29 Jul 2026 16:15:35 -0400 Subject: [PATCH] feat(agent): report context occupancy, not just tokens saved MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tokens saved is the wrong headline for this feature. Micro-compaction is not an efficiency optimisation — the same summarization work happens either way. What it buys is (a) that work amortized across turns instead of one stall, and (b) a window kept low enough that a session runs much further before needing a hard compaction at all. Neither shows up in "net tokens saved". A session can save nothing on paper and still be a clear win on both counts. So the telemetry now carries occupancy: tokens_after as a share of the compaction threshold, plus the threshold and resolved window it was computed from. That is the number that says whether a session has headroom left. The report leads with it, and cross-references the batch `compression_attempt` lines already in the log so it can show how often the long pause actually fired — ideally never. Occupancy is read from the cached threshold only. The public `threshold_tokens` property resolves lazily and can issue a synchronous /models probe (#32221); telemetry must never be the thing that blocks a turn, so an unresolved window reports null. In practice a pass has already resolved it via the tail calculation, so the field is populated. A test pins the no-forcing behaviour directly against the emitter. The report is pure ASCII: `scripts/check_subprocess_stdin.py` currently dies on a cp1252 console before printing its results, and a diagnostic tool that crashes on the platform it is diagnosing is worse than no tool. Co-Authored-By: Claude Fable 5 --- agent/context_compressor.py | 15 +++ docs/micro-compaction.md | 24 +++++ scripts/micro_compaction_report.py | 153 +++++++++++++++++---------- tests/agent/test_micro_compaction.py | 66 ++++++++++++ 4 files changed, 201 insertions(+), 57 deletions(-) diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 59e97d48d13..319e5507af4 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -5446,6 +5446,15 @@ This compaction should PRIORITISE preserving all information related to the focu delta = tokens_after - tokens_before self._micro_compact_tokens_saved_total -= delta self._micro_compact_passes += 1 + # Cached reads only. The ``threshold_tokens`` / ``context_length`` + # properties resolve lazily and can fire a synchronous /models + # probe on first access (#32221) — telemetry must never be the + # thing that blocks a turn. Unresolved simply reports null. + threshold = self._threshold_tokens + context_limit = self._resolved_context_length + occupancy = None + if threshold and tokens_after is not None and threshold > 0: + occupancy = round(tokens_after / threshold * 100, 1) payload = { "event": "micro_compaction", "session_id": getattr(self, "_session_id", "") or "", @@ -5463,6 +5472,12 @@ This compaction should PRIORITISE preserving all information related to the focu "passes_total": self._micro_compact_passes, "tokens_saved_total": self._micro_compact_tokens_saved_total, "duration_ms": _safe_int(duration_ms), + # Headroom, not efficiency: how full the window is being kept. + # This is the number that says whether the session can keep + # going without a hard batch compaction. + "threshold_tokens": _safe_int(threshold), + "context_limit": _safe_int(context_limit), + "occupancy_pct": occupancy, "main_model": self.model or "", "aux_model": self.summary_model or "", } diff --git a/docs/micro-compaction.md b/docs/micro-compaction.md index 19419dab784..c302a1b5949 100644 --- a/docs/micro-compaction.md +++ b/docs/micro-compaction.md @@ -162,16 +162,40 @@ compaction. Everything else about compression is unchanged. ## Measuring it +Micro-compaction is not primarily a token-saving or time-saving optimisation, +and judging it on tokens saved will undersell it. The two things it actually +buys you are: + +1. **The long pause is amortized.** The same summarization work happens, but as + small increments after turns instead of one stall in the middle of a session. +2. **Your context lasts longer.** Because the middle is continuously reclaimed, + occupancy stays low instead of sawtoothing up to the threshold. A session + runs much further — often indefinitely — before it needs a hard compaction + at all. + +So the number that matters is **occupancy**: how full the window is being kept, +as a percentage of the compaction threshold. A session that holds steady around +40% has headroom to keep going; one climbing through 90% is about to stall. The +second number is **how many batch compactions actually fired** — ideally none. + +A session can save nothing on paper and still be a clear win on both counts. + 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, +"occupancy_pct":38.4,"threshold_tokens":34816,"context_limit":40960, "exchange_tokens":868,"rolling_summary_tokens":31,"passes_total":1, "tokens_saved_total":679,"duration_ms":14,...} ``` +`occupancy_pct` is `tokens_after` as a share of the compaction threshold -- the +headroom figure. It is null when the model's window has not been resolved yet: +the telemetry reads only the cached value, because resolving it can issue a +synchronous `/models` probe and telemetry must never be what blocks a turn. + `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 diff --git a/scripts/micro_compaction_report.py b/scripts/micro_compaction_report.py index 6d65a0c133a..30811a67178 100644 --- a/scripts/micro_compaction_report.py +++ b/scripts/micro_compaction_report.py @@ -1,9 +1,9 @@ #!/usr/bin/env python3 """Summarize micro-compaction telemetry from Hermes logs. -Reads the content-free ``micro compaction telemetry:`` JSON lines emitted by +Reads the content-free JSON lines emitted by ``ContextCompressor._emit_micro_compaction_telemetry`` and reports what the -feature actually did, per session and overall. +feature actually bought you. Usage: python scripts/micro_compaction_report.py [LOGFILE ...] @@ -11,11 +11,23 @@ Usage: 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. +What to look at +--------------- +The point of micro-compaction is not saving tokens or time. It is: + + (a) amortizing the one long batch-compaction pause across many turns, and + (b) keeping the context window low enough that a session runs much further + before it needs a hard compaction at all. + +So the headline numbers here are OCCUPANCY (how full the window is kept, as a +percentage of the compaction threshold) and BATCH COMPACTIONS (how often the +long pause actually fired). Net tokens saved is reported too, but it is the +least interesting figure -- a session can save nothing on paper and still be a +clear win because the stalls disappeared and the window never filled. + +Caveat: running the test suite writes telemetry into the same log. Test lines +cluster inside a sub-second window and carry an empty session_id (they group +as "(unknown)"). Use --per-session to spot them. """ from __future__ import annotations @@ -27,7 +39,8 @@ import sys from collections import defaultdict from pathlib import Path -MARKER = "micro compaction telemetry: " +MICRO_MARKER = "micro compaction telemetry: " +BATCH_MARKER = "context compression attempt telemetry: " def default_log() -> Path: @@ -35,8 +48,9 @@ def default_log() -> Path: return Path(home) / "logs" / "agent.log" -def load(paths: list[Path]) -> list[dict]: - events: list[dict] = [] +def load(paths: list[Path]) -> tuple[list[dict], list[dict]]: + micro: list[dict] = [] + batch: list[dict] = [] for path in paths: try: text = path.read_text(encoding="utf-8", errors="replace") @@ -44,88 +58,113 @@ def load(paths: list[Path]) -> list[dict]: 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 + for marker, sink in ((MICRO_MARKER, micro), (BATCH_MARKER, batch)): + idx = line.find(marker) + if idx == -1: + continue + try: + sink.append(json.loads(line[idx + len(marker):])) + except ValueError: + pass + break + return micro, batch -def fmt(n: int | None) -> str: +def pct(values: list[float]) -> tuple[float, float, float] | None: + if not values: + return None + ordered = sorted(values) + return ordered[0], ordered[len(ordered) // 2], ordered[-1] + + +def fmt(n) -> str: return "-" if n is None else f"{n:,}" -def report(events: list[dict], per_session: bool) -> int: - if not events: +def report(micro: list[dict], batch: list[dict], per_session: bool) -> int: + if not micro: 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.") + print("It may be disabled (compression.micro_compact), or no session") + print("has run long enough to trigger a pass yet.") return 1 by_session: dict[str, list[dict]] = defaultdict(list) - for e in events: + for e in micro: by_session[e.get("session_id") or "(unknown)"].append(e) outcomes: dict[str, int] = defaultdict(int) - for e in events: + for e in micro: 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] + occupancies = [e["occupancy_pct"] for e in micro if e.get("occupancy_pct") is not None] + saved = sum(-(e.get("tokens_delta") or 0) for e in micro) + absorbed = [e for e in micro if e.get("outcome") == "absorbed"] + durations = [e.get("duration_ms") or 0 for e in micro] if per_session: - print(f"{'session':<38} {'passes':>7} {'saved':>10} {'first':>8} {'last':>8}") - print("-" * 76) + print(f"{'session':<26} {'passes':>6} {'occupancy%':>18} {'batch':>6} {'saved':>10}") + print("-" * 72) + batch_by_session: dict[str, int] = defaultdict(int) + for b in batch: + batch_by_session[b.get("session_id") or "(unknown)"] += 1 for sid, evs in sorted(by_session.items(), key=lambda kv: -len(kv[1])): + occ = [e["occupancy_pct"] for e in evs if e.get("occupancy_pct") is not None] + spread = pct(occ) + occ_s = f"{spread[0]:.0f}-{spread[2]:.0f} (med {spread[1]:.0f})" if spread else "-" 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(f"{sid[:26]:<26} {len(evs):>6} {occ_s:>18} " + f"{batch_by_session.get(sid, 0):>6} {s:>+10,}") print() + print("-- headroom ----------------------------------") + spread = pct(occupancies) + if spread: + print(f"context occupancy min {spread[0]:.0f}% median {spread[1]:.0f}% max {spread[2]:.0f}%") + print(" (% of the batch-compaction threshold)") + else: + print("context occupancy unavailable (window not resolved when logged)") + print(f"batch compactions {len(batch):,}") + if batch: + print(f" micro passes each {len(micro) / len(batch):.1f}") + else: + print(" none fired -- the long pause never happened in this log") + + print() + print("-- activity ----------------------------------") print(f"sessions {len(by_session):,}") - print(f"passes {len(events):,}") + print(f"passes {len(micro):,}") 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") + print(f"pass duration median {ordered[len(ordered) // 2]:,} ms " + 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)") + print() + print("-- tokens (least interesting) ----------------") + print(f"net tokens saved {saved:+,}") + if absorbed: + sizes = [e.get("exchange_tokens") or 0 for e in absorbed] + print(f"exchanges absorbed {len(absorbed):,} " + f"(mean {sum(sizes) // len(absorbed):,} tokens each)") + print("note: the first pass in a session costs ~400 tokens of marker") + print("scaffolding; it pays back from the second pass on.") return 0 def main() -> int: - ap = argparse.ArgumentParser(description=__doc__, - formatter_class=argparse.RawDescriptionHelpFormatter) + 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) + for p in paths: + if not p.exists(): + print(f"warning: {p} does not exist", file=sys.stderr) + micro, batch = load([p for p in paths if p.exists()]) + return report(micro, batch, args.per_session) if __name__ == "__main__": diff --git a/tests/agent/test_micro_compaction.py b/tests/agent/test_micro_compaction.py index f940ff71df8..0178ad1ce53 100644 --- a/tests/agent/test_micro_compaction.py +++ b/tests/agent/test_micro_compaction.py @@ -18,6 +18,8 @@ The invariants that matter: times and then skipped, so a poison exchange can't stall every turn. """ +from unittest.mock import patch + import pytest from agent.context_compressor import ( @@ -218,6 +220,70 @@ class TestMicroCompaction: blob = json.dumps(payload) assert "answer 0" not in blob and "question 0" not in blob + def test_telemetry_reports_occupancy_without_forcing_resolution(self, caplog): + """Occupancy is the headline: how full the window is being kept. + + It must be read from the cached threshold only. The public + ``threshold_tokens`` property resolves lazily and can fire a + synchronous /models probe (#32221); telemetry must never be what + blocks a turn, so an unresolved window reports null instead. + """ + import json + import logging + + cc = _compressor() + cc.threshold_tokens = 10_000 # pin; also populates the cache + messages = _conversation(exchanges=8) + + with caplog.at_level(logging.INFO, logger="agent.context_compressor"): + cc._micro_compact(messages) + + line = next(r.getMessage() for r in caplog.records + if "micro compaction telemetry:" in r.getMessage()) + payload = json.loads(line.split("micro compaction telemetry: ", 1)[1]) + + assert payload["threshold_tokens"] == 10_000 + assert payload["occupancy_pct"] == pytest.approx( + payload["tokens_after"] / 10_000 * 100, abs=0.1 + ) + + def test_emitter_never_forces_window_resolution(self, caplog): + """The emitter reads the cached threshold, never the property. + + In a real pass the threshold is already resolved by the time + telemetry runs (the tail calculation needs it), so occupancy is + normally populated. This pins the safety property directly: with the + cache empty, emitting reports null rather than triggering the lazy + resolution — which can issue a synchronous /models probe (#32221). + """ + import json + import logging + + cc = _compressor() + cc._threshold_tokens = None + cc._resolved_context_length = None + + def explode(self): # pragma: no cover - must never be called + raise AssertionError("telemetry forced context-length resolution") + + with patch.object(type(cc), "threshold_tokens", + property(explode, lambda s, v: None)): + with caplog.at_level(logging.INFO, logger="agent.context_compressor"): + cc._emit_micro_compaction_telemetry( + outcome="absorbed", + messages_before=10, + messages_after=9, + tokens_before=500, + tokens_after=400, + ) + + line = next(r.getMessage() for r in caplog.records + if "micro compaction telemetry:" in r.getMessage()) + payload = json.loads(line.split("micro compaction telemetry: ", 1)[1]) + + assert payload["occupancy_pct"] is None + assert payload["threshold_tokens"] is None + def test_first_pass_costs_marker_overhead_then_pays_it_back(self): """The first pass can grow the transcript; later passes recover it.