diff --git a/agent/context_compressor.py b/agent/context_compressor.py index dabe554c6d..16721abb9f 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -5488,7 +5488,7 @@ Write only the summary body. Do not include any preamble or prefix.""" ) return canonical_messages prepared = self._take_prepared_summary( - messages, compress_start, compress_end, fresh=bool(force or focus_topic or bypass_cooldown), + messages, compress_start, fresh=bool(force or focus_topic or bypass_cooldown), provider_context=bool(memory_context.strip()), ) if prepared: diff --git a/agent/context_compressor_window.py b/agent/context_compressor_window.py index 51e5b31fa3..c984124c45 100644 --- a/agent/context_compressor_window.py +++ b/agent/context_compressor_window.py @@ -44,8 +44,7 @@ class CompactionWindowMixin: return working, pruned_count, compress_start, compress_end def _take_prepared_summary( - self, messages: list[dict[str, Any]], compress_start: int, compress_end: int, *, fresh: bool, - provider_context: bool, + self, messages: list[dict[str, Any]], compress_start: int, *, fresh: bool, provider_context: bool, ) -> Optional[PreparedSummary]: """The summary prepared in the background for this window, which an automatic compaction splices at ITS boundary; every newer message stays outside the summary, in the tail. ``fresh`` compactions @@ -56,7 +55,7 @@ class CompactionWindowMixin: if provider_context: self.prepared_compaction.stop_for_provider_context() return self.prepared_compaction.take( - messages, self._session_id, compress_start, compress_end, eligible=not (fresh or provider_context), + messages, self._session_id, compress_start, eligible=not (fresh or provider_context), ) def _adopt_prepared_summary(self, prepared: PreparedSummary) -> str: diff --git a/agent/prepared_compaction.py b/agent/prepared_compaction.py index 2645e39763..8b02245141 100644 --- a/agent/prepared_compaction.py +++ b/agent/prepared_compaction.py @@ -9,8 +9,7 @@ compressor are untouched, so the prompt cache is too. At the next automatic threshold compaction ``compress`` takes the completed summary if the messages it covers are still the same messages: same session, -same head boundary, a boundary no later than the current tail cut, and the same -message structure (roles, tool call and result ids) up to that boundary, each +same head boundary, and the same message structure (roles, tool call and result ids) up to that boundary, each message vouched for by its durable ``message_uid``. The text of a persisted message is not compared: plugins rewrite text in place, and a summary written before such a rewrite still summarises the same messages. Only a message that @@ -18,7 +17,9 @@ has no uid on one side (caller-supplied history that is never persisted) is compared by text, since nothing else can tell it from a different message. On a hit the summary replaces ``messages[start:candidate_end]`` and every message after the candidate boundary, however many arrived since the pass started, stays verbatim in the -tail. On a miss compaction runs inline exactly as without the flag. +tail. Where the fresh plan would cut does not matter, earlier or later: the +summary is never redone over a moved cut, and the candidate boundary is the +tail's start. On a miss compaction runs inline exactly as without the flag. When no completed summary exists but a running pass fits, compaction waits for that pass instead of starting the same summary over: the pass's streamed @@ -176,16 +177,13 @@ def prefix_identities(messages: list[dict[str, Any]], end: int) -> tuple[_Identi return tuple(_message_identity(msg) for msg in messages[:end]) -def _mismatch( - window: _Window, messages: list[dict[str, Any]], session_id: str, start: int, end: int, -) -> Optional[_Miss]: - """Why *window* cannot splice into the planned window ``[start, end)`` of ``messages``, or None.""" +def _mismatch(window: _Window, messages: list[dict[str, Any]], session_id: str, start: int) -> Optional[_Miss]: + """Why *window* cannot splice into ``messages`` with the head ending at ``start``, or None. The fresh tail + cut is not consulted: the candidate's boundary is where the tail starts (one summary, never redone).""" if window.session_id != session_id: return _Miss("other_session", "another session") if window.compress_start != start: return _Miss("head_moved", f"head moved ({window.compress_start} -> {start})") - if window.compress_end > end: - return _Miss("inside_tail", f"prepared boundary {window.compress_end} is inside the protected tail (cut {end})") if window.compress_end > len(messages): return _Miss("transcript_shorter", "transcript shorter than the prepared window") for idx, ((structure, uid, digest), msg) in enumerate(zip(window.identities, messages)): @@ -277,10 +275,10 @@ class PreparedCompaction: _record("auto_disabled", reason="provider_context") def take( - self, messages: list[dict[str, Any]], session_id: str, start: int, end: int, *, eligible: bool, + self, messages: list[dict[str, Any]], session_id: str, start: int, *, eligible: bool, ) -> Optional[PreparedSummary]: - """The completed candidate when it fits ``compress()``'s planned window, else what a running pass - whose window fits publishes while this compaction waits for it, else None. Either way the state is + """The completed candidate when it still covers the same messages after ``compress()``'s head, else + what a running pass whose window does publishes while this compaction waits for it, else None. Either way the state is cleared and a running pass is cancelled: this compaction consumes or supersedes it.""" self._count_unspliced() with self._lock: @@ -291,11 +289,11 @@ class PreparedCompaction: "provider_context" if self._needs_provider_context else "fresh_required", "fresh summary required", ) if reason is None and entry is not None: - reason = _mismatch(entry, messages, session_id, start, end) + reason = _mismatch(entry, messages, session_id, start) waited = None try: if eligible and (entry is None or reason is not None): - if pending is not None and _mismatch(pending.window, messages, session_id, start, end) is None: + if pending is not None and _mismatch(pending.window, messages, session_id, start) is None: wait_started = time.monotonic() entry, reason = self._await(pending) waited = time.monotonic() - wait_started @@ -324,8 +322,8 @@ class PreparedCompaction: _record("discarded", reason=reason.code, wait_seconds=waited) return None logger.info( - "prepared compaction: splice hit session=%s window=%d-%d (fresh cut %d)", session_id, entry.compress_start, - entry.compress_end, end, + "prepared compaction: splice hit session=%s window=%d-%d", session_id, entry.compress_start, + entry.compress_end, ) taken = replace(entry, waited=waited) # counted as used once adopt() splices it with self._lock: diff --git a/hermes_cli/observability/schemas/hermes.shared_metrics.v4.schema.json b/hermes_cli/observability/schemas/hermes.shared_metrics.v4.schema.json index 3201f0ffe9..de9ed285cb 100644 --- a/hermes_cli/observability/schemas/hermes.shared_metrics.v4.schema.json +++ b/hermes_cli/observability/schemas/hermes.shared_metrics.v4.schema.json @@ -5357,7 +5357,6 @@ "fresh_required", "head_moved", "host_cancelled", - "inside_tail", "none", "other_session", "pass_failed_while_waiting", diff --git a/hermes_cli/observability/shared_metrics_prepared_compaction.py b/hermes_cli/observability/shared_metrics_prepared_compaction.py index 3f7c061138..497b549524 100644 --- a/hermes_cli/observability/shared_metrics_prepared_compaction.py +++ b/hermes_cli/observability/shared_metrics_prepared_compaction.py @@ -22,7 +22,7 @@ PREPARED_COMPACTION_EVENTS = frozenset({ PREPARED_COMPACTION_ORIGINS = frozenset({"none", "tool_batch", "turn_end"}) PREPARED_COMPACTION_REASONS = frozenset({ "agent_released", "aux_fallback_main", "compaction_aborted", "fresh_required", "head_moved", "host_cancelled", - "inside_tail", "none", "other_session", "pass_failed_while_waiting", "provider_context", "rewound", + "none", "other_session", "pass_failed_while_waiting", "provider_context", "rewound", "running_not_fit", "session_boundary", "shape_changed", "superseded", "text_changed", "transcript_shorter", "wait_timeout", }) diff --git a/tests/agent/test_prepared_compaction.py b/tests/agent/test_prepared_compaction.py index e071b98c71..4686fcc6d1 100644 --- a/tests/agent/test_prepared_compaction.py +++ b/tests/agent/test_prepared_compaction.py @@ -5,7 +5,7 @@ Behaviour contracts, with the summariser stubbed at its model boundary: * flag off (the default) leaves the whole compression path as it was: no state, no pass, inline summary; * a pass never changes the live transcript or compressor, and runs under the caller's profile scope; * one pass per compaction: the candidate splices at its own boundary and every message after it, however - many, stays verbatim in the tail; + many, stays verbatim in the tail, wherever the fresh plan would cut; * in-place text rewrites (plugins, the no-LLM prune, blank-echo drop) keep the candidate; a changed message structure, a rewound transcript, a moved head or another session's transcript never splices; * without a completed candidate, compaction waits for a running pass whose window fits (forwarding its @@ -265,6 +265,40 @@ def test_growth_past_the_candidate_beyond_the_tail_budget_still_splices_keeping_ assert cc.prepared_compaction._entry is None +@pytest.mark.parametrize("pass_state", ["completed", "running"]) +def test_a_fresh_cut_before_the_candidate_boundary_still_splices_and_the_boundary_starts_the_tail(pass_state): + """The fresh plan cutting earlier than the pass did (cheaper tail rows) never redoes the summary.""" + cc = _compressor() + msgs = _messages(30) + if pass_state == "running": + cc.gate = threading.Event() + running = pc.maybe_prepare(_agent(cc), msgs, origin="test") + assert running is not None + pending = cc.prepared_compaction._pending + threading.Thread(target=lambda: wait_until(lambda: pending.on_progress is not None) and cc.gate.set()).start() + else: + _prepare(cc, msgs) + window = (cc.prepared_compaction._entry or cc.prepared_compaction._pending.window) + # The tail rows got cheaper since the pass planned (the no-LLM prune), so the fresh plan keeps more rows. + later = [m if i < window.compress_end else {**m, "content": m["content"][:8]} for i, m in enumerate(_grow(msgs, 2))] + _, _, start, end = cc._plan_compaction_window(later) + assert start == window.compress_start and end < window.compress_end + + from agent.auxiliary_client import aux_progress_hook + + with aux_progress_hook(lambda: None): + compressed = cc.compress(later, current_tokens=cc.threshold_tokens + 1) + + if pass_state == "running": + running.join(5.0) + assert [c["worker"] for c in cc.calls] == [True], "one summary, written by the pass" + kept = later[window.compress_end:] + assert kept, "the pass planned its own tail" + for original, row in zip(kept, compressed[-len(kept):]): + assert original["content"] in row["content"] + assert cc.prepared_compaction._entry is None + + def test_in_place_text_rewrite_of_summarised_messages_keeps_the_candidate(): """Plugins rewrite message text in place (redaction, pre-compaction notes): still the same messages.""" cc = _compressor() @@ -370,11 +404,11 @@ def test_identity_survives_a_session_db_round_trip_and_catches_a_rewind(tmp_path db.append_message("S1", role=m["role"], content=m["content"]) replayed = [msgs[0]] + db.get_messages_as_conversation("S1") window = pc._Window("S1", 4, 15, pc.prefix_identities(replayed, 15)) - assert pc._mismatch(window, msgs, "S1", 4, 15) is None, "the live dicts before their first persist" - assert pc._mismatch(window, replayed, "S1", 4, 15) is None + assert pc._mismatch(window, msgs, "S1", 4) is None, "the live dicts before their first persist" + assert pc._mismatch(window, replayed, "S1", 4) is None rewound = copy.deepcopy(replayed) rewound[6]["message_uid"] = "a-new-row" - assert "different message" in pc._mismatch(window, rewound, "S1", 4, 15) + assert "different message" in pc._mismatch(window, rewound, "S1", 4) # --------------------------------------------------------------------------- running passes diff --git a/tests/hermes_cli/test_shared_metrics_prepared_compaction.py b/tests/hermes_cli/test_shared_metrics_prepared_compaction.py index 5dad104753..8a500f618e 100644 --- a/tests/hermes_cli/test_shared_metrics_prepared_compaction.py +++ b/tests/hermes_cli/test_shared_metrics_prepared_compaction.py @@ -190,7 +190,7 @@ def test_waste_modes_say_why(monkeypatch, marks): aborted = _compressor() _run_pass(aborted, msgs, "tool-batch") held = aborted.prepared_compaction._entry - taken = aborted.prepared_compaction.take(msgs, held.session_id, held.compress_start, held.compress_end, eligible=True) + taken = aborted.prepared_compaction.take(msgs, held.session_id, held.compress_start, eligible=True) assert taken is not None before = len(marks) aborted.on_session_end(held.session_id, msgs) diff --git a/website/docs/developer-guide/context-compression-and-caching.md b/website/docs/developer-guide/context-compression-and-caching.md index 1f144b377c..45ebaee8a6 100644 --- a/website/docs/developer-guide/context-compression-and-caching.md +++ b/website/docs/developer-guide/context-compression-and-caching.md @@ -684,8 +684,7 @@ transcript, the system prompt and the prompt cache are untouched until compactio per compaction: while a summary is held, no further pass starts, however much the prompt grows. At the next automatic threshold compaction `compress()` splices the completed summary when it still -covers the same messages: same session, unchanged head boundary, a candidate boundary no later than -the fresh tail cut, and the same message structure up to that boundary (role, tool call ids and +covers the same messages: same session, unchanged head boundary, and the same message structure up to that boundary (role, tool call ids and names, tool result ids), each message vouched for by its durable `message_uid`. The text of a persisted message, tool arguments and tool bodies are not compared: the no-LLM prune and plugins rewrite them in place, and the summary still describes the same messages. A message without a uid on @@ -693,7 +692,10 @@ one side (caller-supplied `conversation_history` that is never persisted) is com nothing else can tell it from a different message. A dropped or inserted message, or a rewind or branch into the window (new `message_uid`s), is a miss. The summary then covers `[head, candidate boundary)` and every later message, however many arrived after the pass started, is -kept as tail; tail rules such as lean-mode tool stubbing still apply to them. Summary placement and +kept as tail; tail rules such as lean-mode tool stubbing still apply to them. Where the fresh plan +would cut does not matter: a cut that moved earlier (the prune made tail rows cheaper, tool-group +alignment) or later never discards the summary, because one prepared summary per compaction is never +redone. The pass planned its own window with a full tail, so the tail is never empty. Summary placement and alternation handling are the normal Phase 4 assembly. The splice leaves the live compressor as the inline call would have: a pass that recovered on the main model after the summary model failed hands over that fallback (later compactions use the main model, and the user sees the broken-model diff --git a/website/docs/developer-guide/relay-shared-metrics.md b/website/docs/developer-guide/relay-shared-metrics.md index 046839e2f3..99d60e9b4d 100644 --- a/website/docs/developer-guide/relay-shared-metrics.md +++ b/website/docs/developer-guide/relay-shared-metrics.md @@ -575,7 +575,7 @@ appears in any dimension: the free-text splice-miss reason is reduced to a close | Metric | Dimensions | Question it answers | |---|---|---| -| `hermes.prepared_compaction.count` | event (`pass_started`, `pass_completed`, `pass_failed`, `pass_cancelled`, `used`, `discarded`, `compaction_failed`, `auto_disabled`, `skipped_busy`, `internal_error`), origin (`tool_batch`, `turn_end`: the hook that started the pass; `none` on compaction-side events), reason (`none`, `fresh_required`, `provider_context`, `other_session`, `head_moved`, `inside_tail`, `transcript_shorter`, `shape_changed`, `rewound`, `text_changed`, `running_not_fit`, `wait_timeout`, `host_cancelled`, `pass_failed_while_waiting`, `superseded`, `session_boundary`, `agent_released`, `compaction_aborted`, `aux_fallback_main`), failure class (the `hermes.compression.count` vocabulary; `none` unless the event is a failure), wait bucket (`none`, `lt_1s`, `1s_to_5s`, `5s_to_30s`, `30s_to_120s`, `gte_120s`: how long a compaction waited on a running pass) | Do prepared passes pay off, and if not, why. Pass side (from the background thread): `pass_started`, `pass_completed` (summary published), `pass_failed` (the summariser returned nothing; its class is the one an inline attempt would log, a terminal summary failure first), `pass_cancelled` (the pass saw its cancel event; reason `superseded`, `session_boundary`, `agent_released`, `wait_timeout` or `host_cancelled`, set by whoever cancelled it), `skipped_busy` (the process-wide cap of two concurrent passes refused a pass; counted once per refused window, not per tool batch) and `internal_error` (failure class `exception`). Compaction side: `used` (the summary was spliced into the compacted transcript, counted at the splice itself; reason `aux_fallback_main` when the pass fell back from the summary model to the main model), `discarded` (a completed or running summary the compaction did not use, with the splice-miss code and any wait; a session boundary dropping one reads `session_boundary`, releasing the agent (cache eviction, rewind, close, a one-shot or managed-worker turn ending) reads `agent_released`, and a summary the compaction took but then aborted or no-opped before splicing reads `compaction_aborted`), `compaction_failed` (a compaction that spliced a prepared summary but did not commit, with its attempt class) and `auto_disabled` (the first compaction needing memory-provider context turned passes off for the agent). One pass can count on both sides, so never add sides together: `discarded` + `pass_cancelled` for a running pass that did not fit or was dropped at a boundary (a boundary that wakes a waiting compaction counts `discarded` once, from the boundary). | +| `hermes.prepared_compaction.count` | event (`pass_started`, `pass_completed`, `pass_failed`, `pass_cancelled`, `used`, `discarded`, `compaction_failed`, `auto_disabled`, `skipped_busy`, `internal_error`), origin (`tool_batch`, `turn_end`: the hook that started the pass; `none` on compaction-side events), reason (`none`, `fresh_required`, `provider_context`, `other_session`, `head_moved`, `transcript_shorter`, `shape_changed`, `rewound`, `text_changed`, `running_not_fit`, `wait_timeout`, `host_cancelled`, `pass_failed_while_waiting`, `superseded`, `session_boundary`, `agent_released`, `compaction_aborted`, `aux_fallback_main`), failure class (the `hermes.compression.count` vocabulary; `none` unless the event is a failure), wait bucket (`none`, `lt_1s`, `1s_to_5s`, `5s_to_30s`, `30s_to_120s`, `gte_120s`: how long a compaction waited on a running pass) | Do prepared passes pay off, and if not, why. Pass side (from the background thread): `pass_started`, `pass_completed` (summary published), `pass_failed` (the summariser returned nothing; its class is the one an inline attempt would log, a terminal summary failure first), `pass_cancelled` (the pass saw its cancel event; reason `superseded`, `session_boundary`, `agent_released`, `wait_timeout` or `host_cancelled`, set by whoever cancelled it), `skipped_busy` (the process-wide cap of two concurrent passes refused a pass; counted once per refused window, not per tool batch) and `internal_error` (failure class `exception`). Compaction side: `used` (the summary was spliced into the compacted transcript, counted at the splice itself; reason `aux_fallback_main` when the pass fell back from the summary model to the main model), `discarded` (a completed or running summary the compaction did not use, with the splice-miss code and any wait; a session boundary dropping one reads `session_boundary`, releasing the agent (cache eviction, rewind, close, a one-shot or managed-worker turn ending) reads `agent_released`, and a summary the compaction took but then aborted or no-opped before splicing reads `compaction_aborted`), `compaction_failed` (a compaction that spliced a prepared summary but did not commit, with its attempt class) and `auto_disabled` (the first compaction needing memory-provider context turned passes off for the agent). One pass can count on both sides, so never add sides together: `discarded` + `pass_cancelled` for a running pass that did not fit or was dropped at a boundary (a boundary that wakes a waiting compaction counts `discarded` once, from the boundary). | #### Agent-harness accuracy