From 2159dc3e789be33ae9fbb57d4d4408fa924ff630 Mon Sep 17 00:00:00 2001 From: mnajafian-nv Date: Thu, 8 Oct 2026 00:56:25 -0700 Subject: [PATCH] fix(relay): align the hermes.compaction v1 contract with what Hermes emits Review of the first cut of the marks found four places where the record said something the attempt did not: - A split that failed after it had stored the compacted history (failed_not_indexed without a rollback) returns the compacted transcript, so the model sees it next, but the record read aborted/failed and was named compaction.attempt. Relay then never reset freshness and the compacted history was never recorded. The commit path now logs history_rewritten (whether the returned transcript replaced the input), and such a record is a committed compaction with failure_class session_split_failed and split_status failed_not_indexed. - in_place and session_rotated read false/false on a commit without a session database. They are now set only when split_status is in_place_committed or rotated_committed, and null otherwise. - trigger_class read auto for a record with no trigger or an unlisted one. It now reads unknown there; an unlabelled automatic caller (trigger_source auto) keeps trigger unknown with trigger_class auto. - The v1 sets listed values Hermes never emits (kind request_truncation and window_reduction, scope request, trigger recovery, outcome noop, token_count_method provider_usage, split_status pending and unknown). Consumers treated them as live. They are dropped from the v1 sets and documented as reserved for later versions. session_ownership_unreadable joins the attempt-only failure classes. The Compaction Marks docs now list the four sources exactly, what does not emit a mark yet (OpenAI Responses native compaction and its pre-checkpoint pruning, image shrinking after a provider rejection, request-only truncation), the closed sets and null policy, why a Codex app-server compaction keeps the compaction name (Relay records no LLM start for those turns, so the freshness reset only affects a later Hermes-side call) and is never marked twice, and two known gaps around rotating compactions. Signed-off-by: mnajafian-nv (cherry picked from commit 6e6bc81fc0e074c88d1d632e8110d7490e203015) --- agent/compaction_events.py | 58 ++++++---- agent/conversation_compression.py | 1 + agent/conversation_compression_telemetry.py | 8 +- tests/agent/test_compaction_events.py | 83 +++++++++++++- .../developer-guide/relay-shared-metrics.md | 101 ++++++++++++++---- 5 files changed, 204 insertions(+), 47 deletions(-) diff --git a/agent/compaction_events.py b/agent/compaction_events.py index b282845e4b..408a2450b9 100644 --- a/agent/compaction_events.py +++ b/agent/compaction_events.py @@ -27,33 +27,37 @@ SCHEMA_VERSION = "1" COMPACTION_MARK = "compaction" ATTEMPT_MARK = "compaction.attempt" -_KINDS = frozenset({ - "summarize", "micro_summarize", "prune_tool_results", "provider_native", "request_truncation", "window_reduction", -}) -_SCOPES = frozenset({"history", "provider", "request"}) +# The v1 sets list only values this module emits. Later versions may add request-scope truncation kinds. +_KINDS = frozenset({"summarize", "micro_summarize", "prune_tool_results", "provider_native"}) +_SCOPES = frozenset({"history", "provider"}) _METHODS = frozenset({ "llm_summary", "aux_fallback_main", "deterministic_fallback", "deterministic_prune", "provider", "none", }) _TRIGGERS = frozenset({ "manual", "turn_start_threshold", "pre_api", "post_tool", "idle", "engine_preflight", "overflow", - "gateway_hygiene", "between_turns", "proactive_prune", "provider", "recovery", "unknown", + "gateway_hygiene", "between_turns", "proactive_prune", "provider", "unknown", }) -_TRIGGER_CLASSES = {"manual": "manual", "overflow": "overflow", "provider": "provider"} -_OUTCOMES = frozenset({"committed", "aborted", "failed", "skipped", "blocked", "noop"}) +# Every other known trigger is automatic; an unknown or unlisted one asserts nothing. +_TRIGGER_CLASSES = { + "manual": "manual", "overflow": "overflow", "provider": "provider", "unknown": "unknown", "other": "unknown", +} +_TRIGGER_CLASS_VALUES = frozenset({"auto", "manual", "overflow", "provider", "unknown"}) +_OUTCOMES = frozenset({"committed", "aborted", "failed", "skipped", "blocked"}) _SPLIT_STATUSES = frozenset({ - "not_applicable", "in_place_committed", "rotated_committed", "failed_not_indexed", "aborted", "pending", "unknown", + "not_applicable", "in_place_committed", "rotated_committed", "failed_not_indexed", "aborted", }) -_TOKEN_COUNT_METHODS = frozenset({"estimate_rough", "provider_usage"}) +_TOKEN_COUNT_METHODS = frozenset({"estimate_rough"}) # Attempt classes beyond the shared-metrics closed set: guards, Codex route exits, lease races, and the # micro-compaction pass outcomes. _EXTRA_FAILURE_CLASSES = frozenset({ "blocked:cooldown", "blocked:structural_backoff", "blocked:ineffective", "blocked:unknown", "codex_auto_native", "codex_auto_off", "codex_no_thread", "codex_compaction_failed", - "session_ownership_lost", "cooldown_state_unreadable", "summary_model_benched", + "session_ownership_lost", "session_ownership_unreadable", "cooldown_state_unreadable", "summary_model_benched", "summarize_failed", "exchange_skipped", "defrag_failed", "stale_generation", "watermark_unavailable", }) # An aborted attempt the compressor ran but deliberately did not commit; other aborts are failures. _ABORTED_CLASSES = frozenset({"would_grow", "no_progress", "commit_fence_cancelled"}) +_COMMIT_MODES = frozenset({"in_place_committed", "rotated_committed"}) _MICRO_OUTCOMES = { "absorbed": "committed", "defrag": "committed", "stale_generation": "skipped", "watermark_unavailable": "skipped", } @@ -131,7 +135,9 @@ def _payload(**values: Any) -> dict[str, Any]: payload["scope"] = _closed(payload["scope"], _SCOPES) payload["method"] = _closed(payload["method"], _METHODS) or "none" payload["trigger"] = _closed(payload["trigger"], _TRIGGERS) or "unknown" - payload["trigger_class"] = _TRIGGER_CLASSES.get(payload["trigger"], "auto") + payload["trigger_class"] = ( + _closed(payload["trigger_class"], _TRIGGER_CLASS_VALUES, None) or _TRIGGER_CLASSES.get(payload["trigger"], "auto") + ) payload["outcome"] = _closed(payload["outcome"], _OUTCOMES) or "other" payload["token_count_method"] = _closed(payload["token_count_method"], _TOKEN_COUNT_METHODS) payload["attempt_id"] = payload["attempt_id"] or uuid.uuid4().hex @@ -141,7 +147,11 @@ def _payload(**values: Any) -> dict[str, Any]: return payload -def _attempt_outcome(commit_status: Any, failure: str) -> str: +def _attempt_outcome(commit_status: Any, failure: str, history_rewritten: Any) -> str: + # A split that failed after publishing the compacted history still rewrote what the model sees next: it is + # a committed compaction whose failure_class records the failure. + if history_rewritten is True: + return "committed" if commit_status != "aborted": return commit_status if commit_status in _OUTCOMES else "other" if failure in _skip_classes(): @@ -149,11 +159,12 @@ def _attempt_outcome(commit_status: Any, failure: str) -> str: return "aborted" if failure in _ABORTED_CLASSES else "failed" -def _attempt_trigger(trigger_source: Any) -> str: - # ``auto`` is what an unlabelled automatic caller records: the reason is unknown, not "other". - if trigger_source == "auto" or not trigger_source: - return "unknown" - return trigger_source +def _attempt_trigger(trigger_source: Any) -> tuple[Any, str | None]: + """``(trigger, trigger_class)``. ``auto`` is what an unlabelled automatic caller records: the reason is + unknown, but the attempt is still known to be automatic.""" + if trigger_source == "auto": + return "unknown", "auto" + return trigger_source or "unknown", None def attempt_payload(record: dict[str, Any], *, compression_count: Any = None) -> dict[str, Any]: @@ -161,19 +172,20 @@ def attempt_payload(record: dict[str, Any], *, compression_count: Any = None) -> codex = record.get("route") == "codex_app_server" failure = failure_class(record.get("failure_class")) split_status = _closed(record.get("split_status"), _SPLIT_STATUSES) - outcome = _attempt_outcome(record.get("commit_status"), failure) - committed = outcome == "committed" + outcome = _attempt_outcome(record.get("commit_status"), failure, record.get("history_rewritten")) + trigger, trigger_class = _attempt_trigger(record.get("trigger_source")) values: dict[str, Any] = {key: _int(record.get(src)) for src, key in _ATTEMPT_INTS.items()} values.update( kind="provider_native" if codex else "summarize", scope="provider" if codex else "history", official=True, - method=record.get("method"), trigger=_attempt_trigger(record.get("trigger_source")), - outcome=outcome, failure_class=failure, + method=record.get("method"), trigger=trigger, trigger_class=trigger_class, outcome=outcome, + failure_class=failure, attempt_id=record.get("attempt_id"), session_id=record.get("session_id"), token_count_method=record.get("token_count_method"), summarizer_provider=record.get("aux_provider") or None, summarizer_model=record.get("aux_model") or None, compression_count=_int(compression_count), split_status=split_status, - in_place=split_status == "in_place_committed" if committed else None, - session_rotated=split_status == "rotated_committed" if committed else None, + # Commit mode is known only for a fully committed split; anything else leaves both null. + in_place=split_status == "in_place_committed" if split_status in _COMMIT_MODES else None, + session_rotated=split_status == "rotated_committed" if split_status in _COMMIT_MODES else None, has_focus_topic=record.get("has_focus_topic") if isinstance(record.get("has_focus_topic"), bool) else None, ) if codex: diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index ba0bc2b726..88b404ce83 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -4349,6 +4349,7 @@ def compress_context( 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), commit_started_at=commit.commit_started_at, attempt_seed=attempt.seed, + history_rewritten=compressed is not messages, ) return compressed, new_system_prompt finally: diff --git a/agent/conversation_compression_telemetry.py b/agent/conversation_compression_telemetry.py index bfd51ee023..5b533875e4 100644 --- a/agent/conversation_compression_telemetry.py +++ b/agent/conversation_compression_telemetry.py @@ -85,13 +85,15 @@ def _attempt_seed( def _emit_compression_attempt_telemetry( agent: Any, *, started_at: float, commit_status: str, split_status: str, failure_class: str | None = None, commit_started_at: float | None = None, include_last_telemetry: bool = True, - attempt_seed: dict[str, Any] | None = None, + attempt_seed: dict[str, Any] | None = None, history_rewritten: bool | None = None, ) -> None: """Emit one content-free JSON log line for a compression attempt. ``include_last_telemetry=False`` is for emits that fire without an attempt having begun (pool-saturation refusals): they must not hydrate from the previous attempt's numbers. - ``attempt_seed`` is the emitting attempt's own seed (see ``_attempt_seed``).""" + ``attempt_seed`` is the emitting attempt's own seed (see ``_attempt_seed``). + ``history_rewritten`` (commit path only) says whether the returned transcript replaced the input, which + a split that failed after publishing the compacted history still did.""" try: compressor = agent.context_compressor telemetry = getattr(compressor, "_last_compression_telemetry", None) if include_last_telemetry else None @@ -111,6 +113,8 @@ def _emit_compression_attempt_telemetry( total_duration_ms=int((time.monotonic() - started_at) * 1000), commit_status=commit_status, split_status=split_status, ) + if history_rewritten is not None: + payload["history_rewritten"] = history_rewritten if commit_started_at is not None: telemetry["commit_ms"] = payload["commit_ms"] = max(0, int((time.monotonic() - commit_started_at) * 1000)) # Defer only to THIS attempt's class: an abort restore can put the previous attempt's telemetry back. diff --git a/tests/agent/test_compaction_events.py b/tests/agent/test_compaction_events.py index 2697f96448..0c822df036 100644 --- a/tests/agent/test_compaction_events.py +++ b/tests/agent/test_compaction_events.py @@ -304,6 +304,23 @@ def test_a_persistence_detached_fork_publishes_nothing(published): assert published == [] +def test_an_attempt_aborted_after_a_commit_marks_only_its_own_attempt(published): + agent = _Agent(_compressor()) + + with patch.object(agent.context_compressor, "_generate_summary", return_value="SANITIZED SUMMARY"): + compress_context(agent, _messages(), "system prompt", approx_tokens=80_000, trigger="pre_api") + compress_context( + agent, _messages(), "system prompt", approx_tokens=80_000, trigger="post_tool", + snapshot_is_current=lambda: False, + ) + + [(_, first, committed), (_, second, aborted)] = published + assert (first, second) == ("compaction", "compaction.attempt") + assert aborted["attempt_id"] != committed["attempt_id"] + assert (aborted["trigger"], aborted["failure_class"], aborted["method"]) == ("post_tool", "snapshot_stale", "none") + assert aborted["tokens_reclaimed"] is None and aborted["cache_break"] is False + + def test_rotation_publishes_after_the_split_with_the_attempts_own_session(published): """A rotating commit re-points the agent at a child session and its memory flush resets the compressor's telemetry; the record is still built from this attempt's telemetry and names the session it started in.""" @@ -331,6 +348,65 @@ def test_rotation_publishes_after_the_split_with_the_attempts_own_session(publis assert payload["tokens_after"] is not None and payload["method"] == "llm_summary" +def test_a_split_that_fails_after_rewriting_history_is_still_a_compaction(published): + """The in-place commit stored the compacted transcript, then a later bookkeeping write failed: the model + sees the compacted history next, so Relay must reset freshness, and the record says what failed.""" + from hermes_state import SessionDB + from run_agent import AIAgent + + with tempfile.TemporaryDirectory() as tmpdir, patch.dict(os.environ, {"OPENROUTER_API_KEY": "test-key"}): + db = SessionDB(db_path=Path(tmpdir) / "state.db") + db.create_session("in-place-session", "cli", model="test/model") + agent = AIAgent( + api_key="test-key", base_url="https://openrouter.ai/api/v1", model="test/model", quiet_mode=True, + session_db=db, session_id="in-place-session", skip_context_files=True, skip_memory=True, + ) + agent.compression_in_place = True + agent.context_compressor.tail_token_budget = 10 + with patch.object(agent.context_compressor, "_generate_summary", return_value="SANITIZED SUMMARY"), \ + patch.object(SessionDB, "update_system_prompt", side_effect=OSError("disk full")): + compressed, _ = agent._compress_context(_messages(), "sys", approx_tokens=80_000, trigger="pre_api") + agent.close() + + assert len(compressed) < len(_messages()) + [(_sid, name, payload)] = published + assert name == "compaction" + assert (payload["outcome"], payload["failure_class"], payload["split_status"]) == ( + "committed", "session_split_failed", "failed_not_indexed", + ) + assert (payload["in_place"], payload["session_rotated"], payload["cache_break"]) == (None, None, True) + + +@pytest.mark.parametrize( + ("record", "name", "outcome"), + [ + # Rotation published the compacted child, then indexing failed: the history was rewritten. + ({"commit_status": "aborted", "split_status": "failed_not_indexed", "history_rewritten": True}, + "compaction", "committed"), + # The split rolled back to the original transcript: nothing the model sees changed. + ({"commit_status": "aborted", "split_status": "aborted", "history_rewritten": False}, + "compaction.attempt", "failed"), + ], +) +def test_the_mark_name_follows_whether_history_was_rewritten(record, name, outcome): + payload = compaction_events.attempt_payload({**record, "failure_class": "session_split_failed"}) + assert (compaction_events.event_name(payload), payload["outcome"]) == (name, outcome) + assert payload["failure_class"] == "session_split_failed" + + +@pytest.mark.parametrize( + ("split_status", "expected"), + [ + ("in_place_committed", (True, False)), ("rotated_committed", (False, True)), + # A session without a database has no commit mode to report. + ("not_applicable", (None, None)), ("failed_not_indexed", (None, None)), + ], +) +def test_commit_mode_is_null_unless_the_split_committed(split_status, expected): + payload = compaction_events.attempt_payload({"commit_status": "committed", "split_status": split_status}) + assert (payload["in_place"], payload["session_rotated"]) == expected + + @pytest.mark.parametrize( ("record", "expected"), [ @@ -351,7 +427,10 @@ def test_outcome_and_failure_class_stay_in_their_closed_sets(record, expected): ("trigger_source", "trigger", "trigger_class"), [ ("manual", "manual", "manual"), ("overflow", "overflow", "overflow"), ("gateway_hygiene", "gateway_hygiene", "auto"), - ("auto", "unknown", "auto"), ("plugin_said_so", "other", "auto"), + # An unlabelled automatic caller: the reason is unknown, but the attempt is known to be automatic. + ("auto", "unknown", "auto"), + # No trigger, or one outside the closed set: the record cannot say what started it. + ("unknown", "unknown", "unknown"), (None, "unknown", "unknown"), ("plugin_said_so", "other", "unknown"), ], ) def test_trigger_maps_onto_the_closed_set(trigger_source, trigger, trigger_class): @@ -363,7 +442,7 @@ def test_trigger_maps_onto_the_closed_set(trigger_source, trigger, trigger_class ("outcome", "scope", "name"), [ ("committed", "history", "compaction"), ("committed", "provider", "compaction"), - ("committed", "request", "compaction.attempt"), ("skipped", "history", "compaction.attempt"), + ("committed", "other", "compaction.attempt"), ("skipped", "history", "compaction.attempt"), ], ) def test_only_a_committed_history_or_provider_rewrite_is_named_compaction(outcome, scope, name): diff --git a/website/docs/developer-guide/relay-shared-metrics.md b/website/docs/developer-guide/relay-shared-metrics.md index 831ed588a7..cf02018929 100644 --- a/website/docs/developer-guide/relay-shared-metrics.md +++ b/website/docs/developer-guide/relay-shared-metrics.md @@ -134,34 +134,95 @@ exporter for that process. ## Compaction Marks -Every history rewrite emits one Relay mark with -`data_schema: {"name": "hermes.compaction", "version": "1"}`: each -`compress_context` attempt (including blocked, Codex-routed and aborted ones), -each micro-compaction pass, each committed proactive tool-result prune, and -each Codex-native thread compaction Hermes observes. The mark is emitted -whenever the Relay runtime is live; it needs no setting because Relay -exporters are configured separately. +When the Relay runtime is live, Hermes emits one Relay mark with +`data_schema: {"name": "hermes.compaction", "version": "1"}` for each of these +sources: + +- each `compress_context` attempt that returns, including blocked, + Codex-routed and aborted ones; +- each micro-compaction pass; +- each committed proactive tool-result prune; +- each Codex app-server thread compaction that Codex ran on its own and Hermes + observed. + +The mark needs no setting because Relay exporters are configured separately. +Other history and request rewrites emit no mark yet: OpenAI Responses native +compaction and its pre-checkpoint item pruning, image shrinking after a +provider rejection, and request-only truncation. | Mark | When | Relay effect | |---|---|---| | `compaction` | `outcome: committed` with `scope: history` or `provider` | Resets LLM-history freshness for the session's agent scope, so the next LLM start records the full compacted history | -| `compaction.attempt` | Any other outcome (`aborted`, `failed`, `skipped`, `blocked`) | None | +| `compaction.attempt` | Any other outcome (`aborted`, `failed`, `skipped`, `blocked`, `other`) | None | + +A commit whose session split failed after it had already stored the compacted +history is still `committed` (the model sees the compacted history next), with +`failure_class: session_split_failed` and `split_status: failed_not_indexed`. The mark is parented to the live `hermes.turn` of the same session, or to the `hermes.session` scope when compaction runs outside a turn (gateway hygiene). +No mark is emitted from a turn Relay does not instrument (a second concurrent +turn on the same session) or from a persistence-detached fork such as +background review: their compaction rewrites only their own transcript. + `data` is flat so OpenTelemetry flattens every field into -`nemo_relay.mark.data.`. Categorical fields come from closed sets with an -`other` fallback: `kind` (`summarize`, `micro_summarize`, -`prune_tool_results`, `provider_native`), `scope`, `official` (true only for -`compress_context`), `method`, `trigger` / `trigger_class`, `outcome` and -`failure_class`. Numeric fields carry the effect (`tokens_before`, -`tokens_after`, `tokens_reclaimed`, `messages_before`, `messages_after`, -`items_dropped`) and timings (`duration_ms`, `summary_generation_ms`, -`aux_call_duration_ms`, `queue_wait_ms`, `commit_ms`). Token counts are rough -message-only estimates (`token_count_method: estimate_rough`). The payload never -contains message text, summary text, the focus topic (only -`has_focus_topic`), error text or file paths. `attempt_id` and `session_id` -are for correlation; do not use them as metric labels. +`nemo_relay.mark.data.`. Every v1 key is present on every mark; a value +that does not apply is `null`. Categorical fields come from these closed sets: + +| Field | v1 values | +|---|---| +| `kind` | `summarize`, `micro_summarize`, `prune_tool_results`, `provider_native` | +| `scope` | `history`, `provider` | +| `official` | `true` only for `compress_context` attempts | +| `method` | `llm_summary`, `aux_fallback_main`, `deterministic_fallback`, `deterministic_prune`, `provider`, `none` | +| `trigger` | `manual`, `turn_start_threshold`, `pre_api`, `post_tool`, `idle`, `engine_preflight`, `overflow`, `gateway_hygiene`, `between_turns`, `proactive_prune`, `provider`, `unknown` | +| `trigger_class` | `auto`, `manual`, `overflow`, `provider`, `unknown` | +| `outcome` | `committed`, `aborted`, `failed`, `skipped`, `blocked` | +| `split_status` | `not_applicable`, `in_place_committed`, `rotated_committed`, `failed_not_indexed`, `aborted`, or `null` | +| `token_count_method` | `estimate_rough`, or `null` | + +A value outside its set reads `other`. `failure_class` uses the +`hermes.compression.count` classes plus the attempt-only classes (guard exits +such as `blocked:cooldown`, Codex route exits such as `codex_auto_native`, the +lease exits `session_ownership_lost`, `session_ownership_unreadable` and +`cooldown_state_unreadable`, `summary_model_benched`, and the micro-compaction +outcomes), `none` when there is no class, and `other` for anything else. An +automatic caller that passes no trigger label reads `trigger: unknown` with +`trigger_class: auto`; a record with no trigger at all reads `unknown` for +both. `in_place` and `session_rotated` are set only when `split_status` is +`in_place_committed` or `rotated_committed`; `cache_break` is `null` for +`scope: provider`. + +Numeric fields carry the effect (`tokens_before`, `tokens_after`, +`tokens_reclaimed`, `messages_before`, `messages_after`, `items_dropped`) and +timings (`duration_ms`, `summary_generation_ms`, `aux_call_duration_ms`, +`queue_wait_ms`, `commit_ms`). Token counts are rough message-only estimates +(`token_count_method: estimate_rough`). The payload never contains message +text, summary text, the focus topic (only `has_focus_topic`), error text or +file paths. `attempt_id` and `session_id` are for correlation; do not use them +as metric labels. + +Codex app-server turns run their model calls inside Codex, so Relay records no +LLM start for them. Their `compaction` mark records the rewrite; its freshness +reset only affects a later Hermes-side LLM call in the same session scope. A +compaction Hermes forces is recorded once, by its `compress_context` attempt: +the next Codex turn drops notifications from the compaction turn, so it cannot +report the same compaction again. + +Known gaps in v1: + +- With `compression.in_place: false`, a second rotating compaction in the same + turn emits no mark: its new session has no Relay scope until the next turn, + which opens fresh. +- With `gateway.telemetry.session_segments.on_compaction: true`, a rotating + commit outside any turn closes the old Relay session before the mark is + emitted, so the mark is dropped. No current caller does this: manual + `/compress` defers that notification until after the mark, and gateway + hygiene compacts in place. +- Reserved for later versions and never emitted in v1: `kind` + `request_truncation` / `window_reduction`, `scope: request`, + `trigger: recovery`, `outcome: noop`, `token_count_method: provider_usage`, + and a filled `overflow_reason` (always `null` in v1). ## Process-Wide Plugin Policy and Profile Isolation