fix(#488): the consumer switch — replay skips are occurrence-bound via the emission projection

compaction.py

@@ -19,6 +19,7 @@
 import json
 import logging
 import time
+from collections import Counter
 from typing import Any, Dict, List, Optional
 
 from .dag import SummaryNode
@@ -33,6 +34,7 @@
     _finalize_emission_descriptors,
     _has_lossy_redacted_identity,
     _project_emitted_occurrences,
+    _proof_user_identity,
 )
 from .sanitize import _contains_sensitive_redaction
 from .sqlite_util import _is_sqlite_locked_error
@@ -342,8 +344,10 @@ def _has_ignored_backlog_outside_fresh_tail(self, messages: List[Dict[str, Any]]
         if fresh_tail_start <= leading_anchor_count:
             return False
         previous_store_id_map = self._current_compress_store_ids_by_message_id
+        occurrences, v4 = self._replay_occurrences(messages)  # the complete list, sliced (A2, F5)
         self._current_compress_store_ids_by_message_id = self._get_store_id_map_for_messages(
-            messages[leading_anchor_count:fresh_tail_start]
+            messages[leading_anchor_count:fresh_tail_start],
+            occurrences[leading_anchor_count:fresh_tail_start] if v4 else None,
         )
         try:
             return any(
@@ -429,7 +433,10 @@ def _leaf_compaction_candidate_status_once(
         generated_placeholder_hashes = self._load_generated_ignored_placeholder_hashes()
         if self._compiled_ignore_message_patterns or generated_placeholder_hashes:
             previous_store_id_map = self._current_compress_store_ids_by_message_id
-            self._current_compress_store_ids_by_message_id = self._get_store_id_map_for_messages(candidate_raw)
+            occurrences, v4 = self._replay_occurrences(messages)  # the complete list, sliced (A2, F5)
+            self._current_compress_store_ids_by_message_id = self._get_store_id_map_for_messages(
+                candidate_raw, occurrences[leading_anchor_count:fresh_tail_start] if v4 else None
+            )
             try:
                 filtered_candidate_raw: list[Dict[str, Any]] = []
                 for msg in candidate_raw:
@@ -530,13 +537,15 @@ def compress(self, messages: List[Dict[str, Any]],
         """Run compaction and leave a terminal public status on every failure."""
         try:
             self._pending_emission_candidates = []
+            self._compress_occurrences = None
             with self._fresh_tail_pressure_yield_invocation():
                 result = self._compress_impl(
                     messages,
                     current_tokens=current_tokens,
                     focus_topic=focus_topic,
                     force=force,
                 )
+            self._compress_occurrences = None
             if (
                 isinstance(result, list)
                 and result is not messages
@@ -559,6 +568,7 @@ def compress(self, messages: List[Dict[str, Any]],
             self._rekey_host_rewrite_watch(messages, result)
             return result
         except BaseException:
+            self._compress_occurrences = None
             self._last_compression_status = "error"
             self._last_compression_noop_reason = ""
             raise
@@ -588,9 +598,6 @@ def _record_compress_commit_proof(self, messages, result) -> None:
                 or self._ingest_cursor != len(result)
             ):
                 return
-            carried_rows = self._store.get_batch(
-                sorted(set(self._get_store_ids_for_messages(result)))
-            )
             state = (
                 self._lifecycle.get_by_conversation(self._conversation_id)
                 if self._conversation_id
@@ -632,13 +639,18 @@ def _record_compress_commit_proof(self, messages, result) -> None:
                     ):
                         emissions.extend(carried)
             emissions.sort(key=lambda item: item["output_occurrence"]["index"])
+            # The output's rows map through THIS proof's emissions (A3), never a DAG-shaped strip (F1).
+            occurrences, _v4 = self._replay_occurrences(result, {"version": 4, **emission_binding, "emissions": emissions})
+            store_ids = self._get_store_id_map_for_messages(result, occurrences)
+            carried_rows = self._store.get_batch(sorted({store_ids[id(m)] for m in result if id(m) in store_ids}))
             proof = {
                 "version": _COMPACTION_COMMIT_PROOF_VERSION,
                 **emission_binding,
                 "session_id": self._session_id,
                 "conversation_id": self._conversation_id,
-                "input": [self._proof_replay_identity(m) for m in messages],
-                "output": [self._proof_replay_identity(m) for m in result],
+                # Full identities: the multiplicity witness (F4); output_effective is the projection.
+                "input": [self._proof_replay_identity(m, strip_carrier=False) for m in messages],
+                "output": [self._proof_replay_identity(m, strip_carrier=False) for m in result],
                 "end_consumed": False,
                 "carry_ranges": self._coalesce_compression_carry_ranges(
                     (str(row["session_id"]), store_id - 1, store_id)
@@ -653,8 +665,12 @@ def _record_compress_commit_proof(self, messages, result) -> None:
                 # No-progress compress: Hermes has nothing to commit, and an
                 # end call with this list must stay a real session end.
                 return
-            effective_rows = [m for m in result if not self._is_replayed_context_scaffold_message(m)]
-            proof["output_effective"] = [self._proof_replay_identity(m) for m in effective_rows]
+            _projection, output_identities = self._occurrence_replay_identities(
+                result, {**proof, **emission_binding}
+            )
+            proof["output_effective"] = [
+                _proof_user_identity(identity) for identity in output_identities if identity is not None
+            ]
             proof["native"] = self._last_compression_status == "host_native"
             if proof["native"]:
                 matched_tool_ids = _matched_tool_call_ids(result)
@@ -670,10 +686,8 @@ def _record_compress_commit_proof(self, messages, result) -> None:
                     )
                     proof["skip_landing"].append(not identity[2] and not identity[3])
                 summary_index = getattr(self, "_last_native_summary_index", None)
-                rows_before_summary = result[:summary_index] if summary_index is not None else ()
                 proof["native_summary_index"] = sum(
-                    not self._is_replayed_context_scaffold_message(row)
-                    for row in rows_before_summary
+                    identity is not None for identity in output_identities[:summary_index]
                 ) if summary_index is not None else None
             proof["published"] = self._last_compression_status == "compacted"
             self._last_emission_descriptors = {
@@ -1133,6 +1147,14 @@ def _compress_impl(self, messages: List[Dict[str, Any]],
         # replay-safe view so quarantined assistant loops do not enter summaries
         # or provider context after the durable row has been written.
         working_messages = self._ingest_messages(messages)
+        # #488: project the complete admitted list ONCE; subset consumers read a row's occurrence
+        # here. Only a row at exactly one position is registered (F6): a copy or an aliased row
+        # is unproven (full identity).
+        occurrences, v4 = self._replay_occurrences(working_messages)
+        positions = Counter(id(m) for m in working_messages)
+        self._compress_occurrences = {
+            id(m): occurrence for m, occurrence in zip(working_messages, occurrences) if positions[id(m)] == 1
+        } if v4 else None
         self._prepare_retained_user_anchor(working_messages)
         native_cleanup_only = bool(
             self._config.native_recovery
@@ -1290,9 +1312,10 @@ def _compress_impl(self, messages: List[Dict[str, Any]],
                 break
 
             candidate_start = leading_anchor_count
-            while (
-                candidate_start < fresh_tail_start
-                and self._is_replayed_context_scaffold_message(working_messages[candidate_start])
+            while candidate_start < fresh_tail_start and (
+                self._is_replayed_context_scaffold_message(working_messages[candidate_start])
+                if self._compress_occurrences is None  # no v4 proof in force: the rc4 contract
+                else self._compress_occurrences.get(id(working_messages[candidate_start]), (None, ()))[1] is None
             ):
                 candidate_start += 1
             if candidate_start > leading_anchor_count:

engine.py

@@ -134,7 +134,7 @@
 from .placeholder_ledger import PlaceholderLedgerMixin
 from .reconcile import _COMPACTION_COMMIT_PROOF_METADATA_PREFIX, ReconcileMixin, _PRESERVED_OBJECTIVE_CONTEXT_PREFIX
 from .reconcile import _emission_identity
-from .reconcile import _has_lossy_redacted_identity
+from .reconcile import _has_lossy_redacted_identity, _proof_user_identity
 from .compaction import CompactionMixin
 from .reset_state import ResetStateMixin
 from .bypass import BypassMixin
@@ -2647,7 +2647,7 @@ def _prepare_retained_user_anchor(
             return None
         registered_row = self._load_retained_user_anchor_row()
         # A host may have trimmed the retained prompt in place (#498), even before an override.
-        live_identity = self._message_replay_identity(messages[1])
+        live_identity = self._message_replay_identity(messages[1], strip_carrier=False)
         if registered_row is not None and self._anchor_row_admits(live_identity, registered_row):
             later_real_users = self._durable_real_user_messages(
                 stop_after=1,
@@ -3940,7 +3940,7 @@ def _on_session_end_unlocked(
             and not proof.get("end_consumed")
             and proof.get("input") is not None
             and len(messages) == len(proof["input"])
-            and [self._proof_replay_identity(m) for m in messages] == proof["input"]
+            and [self._proof_replay_identity(m, strip_carrier=False) for m in messages] == proof["input"]
         ):
             # Compaction commit (#483): every input row is already durable and the
             # cursor indexes compress()'s output, so skip the re-ingest. Still
@@ -4717,8 +4717,8 @@ def _stored_row_externalized_text_for_pattern_matching(self, msg: Dict[str, Any]
     def _is_cached_active_replay_message_at_index(self, idx: int, msg: Dict[str, Any]) -> bool:
         if idx < 0 or idx >= len(self._last_active_replay_messages):
             return False
-        return self._message_replay_identity(msg) == self._message_replay_identity(
-            self._last_active_replay_messages[idx]
+        return self._message_replay_identity(msg, strip_carrier=False) == self._message_replay_identity(
+            self._last_active_replay_messages[idx], strip_carrier=False
         )
 
     def _matches_ignore_message_patterns(self, msg: Dict[str, Any], *, stored_row: bool = False) -> bool:
@@ -4824,8 +4824,8 @@ def _remember_active_replay_messages(
         original_messages: List[Dict[str, Any]],
         active_replay_messages: List[Dict[str, Any]],
     ) -> List[Dict[str, Any]]:
-        self._last_active_replay_source_identities = [
-            self._message_replay_identity(message) for message in original_messages
+        self._last_active_replay_source_identities = [  # full identity (#488): no carrier aliasing
+            self._message_replay_identity(message, strip_carrier=False) for message in original_messages
         ]
         self._last_active_replay_messages = self._copy_active_replay_messages_preserving_generated_ids(
             active_replay_messages
@@ -4847,7 +4847,7 @@ def _cached_active_replay_messages(
         self,
         original_messages: List[Dict[str, Any]],
     ) -> Optional[List[Dict[str, Any]]]:
-        identities = [self._message_replay_identity(message) for message in original_messages]
+        identities = [self._message_replay_identity(message, strip_carrier=False) for message in original_messages]
         if identities == getattr(self, "_last_active_replay_source_identities", None):
             cached = getattr(self, "_last_active_replay_messages", None)
             if cached is not None:
@@ -4871,12 +4871,12 @@ def _remap_cursor_through_host_merge(self, messages, proof) -> Optional[int]:
         if not target:
             return None
         effective: list = []
-        for index, message in enumerate(messages):
+        for index, identity in enumerate(self._occurrence_replay_identities(messages, proof)[1]):
             if len(effective) == len(target):
                 return index if effective == target else None
-            if self._is_verified_replay_scaffold_message(message):
+            if identity is None:
                 continue
-            effective.append(self._proof_replay_identity(message))
+            effective.append(_proof_user_identity(identity))
             if effective != target[: len(effective)]:
                 return None
         return len(messages) if effective == target else None
@@ -4898,10 +4898,10 @@ def _remap_cursor_through_native_host_repair(self, messages, proof) -> Optional[
             return None
         skip_metadata_valid = len(skip_landing) == len(target)
         matched = 0
-        for index, message in enumerate(messages):
-            if self._is_verified_replay_scaffold_message(message):
+        for index, identity in enumerate(self._occurrence_replay_identities(messages, proof)[1]):
+            if identity is None:
                 continue
-            identity = self._proof_replay_identity(message)
+            identity = _proof_user_identity(identity)
             if _has_lossy_redacted_identity(identity):
                 return None
             try:
@@ -5272,11 +5272,11 @@ def _ingest_messages(
             if native_lossy:
                 self._ingest_cursor_needs_reconcile = True
             elif n >= self._ingest_cursor and [
-                self._proof_replay_identity(m) for m in messages[: self._ingest_cursor]
+                self._proof_replay_identity(m, strip_carrier=False) for m in messages[: self._ingest_cursor]
             ] == proof["output"]:
                 self._compress_commit_proof = None
             elif (proof.get("end_consumed") or proof.get("native")) and host_input is not None and n >= len(host_input) and [
-                self._proof_replay_identity(m) for m in messages[: len(host_input)]
+                self._proof_replay_identity(m, strip_carrier=False) for m in messages[: len(host_input)]
             ] == host_input:
                 self._ingest_cursor = len(host_input)
                 cursor = self._ingest_cursor
@@ -5391,7 +5391,7 @@ def _ingest_messages(
                 and len(cached_active_replay_messages) >= cursor
             ):
                 current_prefix_identities = [
-                    self._message_replay_identity(message) for message in messages[:cursor]
+                    self._message_replay_identity(message, strip_carrier=False) for message in messages[:cursor]
                 ]
                 if current_prefix_identities == cached_source_identities[:cursor]:
                     replay_messages = (