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 <mnajafian@nvidia.com>
(cherry picked from commit 6e6bc81fc0e074c88d1d632e8110d7490e203015)

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:

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:

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.

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):

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.<key>`. 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.<key>`. 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