fix(sessions): skip a re-handed uid only when the agent saw it durable

The salvaged #134799 guard skipped any no-snapshot, no-row-id dict whose
message_uid was ACTIVE, so SessionDB.append_messages_batch dropped a fresh
message carrying a copied uid — the exact case main's contract
test_fresh_identical_message_is_never_adopted_by_payload pins ("even a copied
message_uid is not proof").

Narrow the trigger to provenance: the agent flush records the uids it itself
saw durable in its current session (rows it flushed, history it scanned as
already persisted) and marks a later marker-stripped copy of one of them
RESTORED_FROM_STORE on its batch row. The write transaction skips only those
marked rows whose uid is still ACTIVE. A direct append with a bare copied uid
inserts as on main.

agent/message_metadata.py

@@ -19,7 +19,11 @@ from agent.conversation_compression_archive import MERGED_DURABLE_ROWS, RETIRED_
 # stored-row CAS digest and the durable row adopted onto the live dict. Never transcript payload.
 DB_ROW_SNAPSHOT = "_db_row_snapshot"
 CANONICAL_ROW = "_canonical_row"
-REPAIR_BOOKKEEPING_FIELDS = frozenset({DB_ROW_SNAPSHOT, CANONICAL_ROW})
+# Set by the agent flush on a batch row (never a live dict) whose ``message_uid`` the agent itself saw durable in
+# this session (written by an earlier flush, or loaded from the store with its marker). Only such a row, with no
+# row address or snapshot, may be skipped as a re-insert of an ACTIVE uid; a bare copied uid proves nothing.
+RESTORED_FROM_STORE = "_restored_from_store"
+REPAIR_BOOKKEEPING_FIELDS = frozenset({DB_ROW_SNAPSHOT, CANONICAL_ROW, RESTORED_FROM_STORE})
 # Unanswered text a merged user row held before the current turn was absorbed
 # (agent_runtime_helpers._merge_consecutive_users); the persist override keeps it.
 # It repeats the row's own content, so pricing it would double the estimate.

agent/session_persistence.py

@@ -24,8 +24,8 @@ from agent.memory_context import sanitize_context
 from agent.tool_dispatch_helpers import _is_multimodal_tool_result, _multimodal_text_summary
 from agent.trajectory import save_trajectory as _save_trajectory_to_file
 from agent.message_metadata import (
-    DB_ROW_SNAPSHOT, MERGED_TURN_PREFIX, REPAIR_BOOKKEEPING_FIELDS, TOOL_CALL_UID, copy_identity_fields,
-    tool_call_uid_from_history)
+    DB_ROW_SNAPSHOT, MERGED_TURN_PREFIX, REPAIR_BOOKKEEPING_FIELDS, RESTORED_FROM_STORE, TOOL_CALL_UID,
+    copy_identity_fields, message_uid_or_none, tool_call_uid_from_history)
 from agent.transcript_repair import sync_flushed_message_markers
 
 
@@ -199,6 +199,16 @@ def _db_flush_seed_ids(agent) -> set:
     return seed_ids if isinstance(seed_ids, set) else set()
 
 
+def _db_flush_store_uids(agent) -> set:
+    """``message_uid``s this agent saw durable in its CURRENT session (flushed by it, or scanned as already-durable
+    history): the provenance that lets the write path skip a marker-stripped copy of one of them (#134799)."""
+    session_id = getattr(agent, "session_id", None)
+    scope = getattr(agent, "_db_store_uid_scope", None)
+    if not (isinstance(scope, tuple) and scope[0] == session_id):
+        scope = agent._db_store_uid_scope = (session_id, set())
+    return scope[1]
+
+
 def _db_flush_scan_start(agent, messages: list[dict]) -> int:
     """Skip the identity-matched, still-marked prefix of the previous flush's snapshot."""
     scan_start = 0
@@ -308,15 +318,20 @@ def _db_flush_collect(agent, messages: list[dict], conversation_history: Optiona
     batch_rows: list[dict[str, Any]] = []
     batch_msgs: list[dict] = []
     tool_uid_owners: dict = {}  # tool_call_uid_from_history memo; the scanned dicts outlive this loop
-    # Durable-uid re-insert guard lives in resolve_and_repair_transcript_batch (the write transaction):
-    # there it shares the repair path's BEGIN IMMEDIATE, sees only the flushed suffix (not the whole
-    # transcript), and defers to the CAS rewrite for snapshot-carrying dicts — a live edit on a restored
-    # dict still reaches its row. Nothing to probe here.
+    # Durable-uid re-insert guard: a row whose uid this agent already saw durable is marked RESTORED_FROM_STORE;
+    # resolve_and_repair_transcript_batch skips it only if that uid is still ACTIVE, inside the write
+    # transaction, and never for a snapshot-carrying dict (the CAS rewrite owns its live edit).
+    store_uids = _db_flush_store_uids(agent)
     for msg_idx in range(_db_flush_scan_start(agent, messages), len(messages)):
         msg = messages[msg_idx]
         # Append-only flush: a mid-turn persist of scaffolding would commit a synthetic turn the end-of-turn
         # drop cannot un-write. Skip regardless of position.
-        if not isinstance(msg, dict) or _is_ephemeral_scaffolding(msg) or msg.get(_DB_PERSISTED_MARKER):
+        if not isinstance(msg, dict) or _is_ephemeral_scaffolding(msg):
+            continue
+        uid = message_uid_or_none(msg)
+        if msg.get(_DB_PERSISTED_MARKER):
+            if uid is not None:
+                store_uids.add(uid)
             continue
         # Already durable (history copy or caller-seeded): stamp so future flushes skip it.
         is_history = id(msg) in history_ids
@@ -324,6 +339,8 @@ def _db_flush_collect(agent, messages: list[dict], conversation_history: Optiona
             (is_history and not replay_history) or id(msg) in seed_ids
         ) and not msg.get(_PERSIST_AFTER_ADMISSION_INTERRUPT):
             msg[_DB_PERSISTED_MARKER] = True
+            if uid is not None:
+                store_uids.add(uid)
             continue
         if getattr(agent, "_mute_notification_reply", False) and not is_history:
             # Only new rows, never the cached history prefix. Keep evidence/model
@@ -335,7 +352,10 @@ def _db_flush_collect(agent, messages: list[dict], conversation_history: Optiona
             # minted (same-batch pairing happens inside the insert).
             if (tool_uid := tool_call_uid_from_history(messages, msg_idx, tool_uid_owners)) is not None:
                 msg[TOOL_CALL_UID] = tool_uid
-        batch_rows.append(_db_flush_row(agent, msg, ov_idx == msg_idx or msg is pending_cli_message))
+        row = _db_flush_row(agent, msg, ov_idx == msg_idx or msg is pending_cli_message)
+        if uid is not None and uid in store_uids:
+            row[RESTORED_FROM_STORE] = True
+        batch_rows.append(row)
         batch_msgs.append(msg)
     return batch_rows, batch_msgs
 
@@ -351,6 +371,7 @@ def _db_flush_write(agent, batch_rows: list[dict[str, Any]], batch_msgs: list[di
         turn_lease_ttl_seconds=getattr(agent, "_active_session_turn_lease_ttl_seconds", 300.0) or 300.0,
     )
     sync_flushed_message_markers(batch_msgs, batch_rows)
+    _db_flush_store_uids(agent).update(uid for msg in batch_msgs if (uid := message_uid_or_none(msg)) is not None)
     if _newest_checkpoint_carrier(batch_msgs, "codex_reasoning_items") >= 0:
         # The insert already rewrote the older rows (SessionDB._drop_shadowed_checkpoint_rows); mirror it on
         # the live transcript so forks/compaction built from memory carry one checkpoint too. Markers stay:

agent/transcript_repair.py

@@ -10,7 +10,7 @@ from typing import Any, Callable, Dict, List, Mapping
 
 from agent.context_compressor import _DB_PERSISTED_MARKER
 from agent.message_metadata import (
-    CANONICAL_ROW, DB_ROW_SNAPSHOT, MESSAGE_UID, copy_identity_fields, message_uid_or_none)
+    CANONICAL_ROW, DB_ROW_SNAPSHOT, MESSAGE_UID, RESTORED_FROM_STORE, copy_identity_fields, message_uid_or_none)
 from hermes_state_common import _id_chunks, _placeholders
 from hermes_state_identity import _fill_missing_tool_call_uids, _restore_row_identity
 from hermes_state_messages import _MESSAGE_WRITE_COLUMNS
@@ -110,10 +110,12 @@ def _active_logical_message_row(
 def _active_uid_dedupe(
     conn: sqlite3.Connection, session_id: str, messages: list[dict[str, Any]]
 ) -> set[str]:
-    """ACTIVE durable uids among *messages*' no-snapshot, no-``_row_id`` dicts — the re-insert guard.
+    """ACTIVE durable uids among *messages*' restored, no-snapshot, no-``_row_id`` rows — the re-insert guard.
 
-    A dict rebuilt from durable rows (crash/failed-turn history restore, session resume, a repaired
-    sequence, any wire-crossing handoff) carries its durable ``message_uid`` — identity is restored on
+    Only a row the agent flush marked ``RESTORED_FROM_STORE`` (it saw that uid durable in this session) is a
+    candidate: a copied uid alone is never proof, so a fresh identical message always inserts. A dict rebuilt
+    from durable rows (crash/failed-turn history restore, session resume, a repaired sequence, any
+    wire-crossing handoff) carries its durable ``message_uid`` — identity is restored on
     EVERY projection — but neither the persistence marker (underscore-prefixed, stripped before the
     wire) nor ``_row_id``/``DB_ROW_SNAPSHOT`` (same strip). Dicts that still carry a snapshot are
     excluded: the CAS path below owns them — it rewrites the active row in place, so their live edits
@@ -125,6 +127,7 @@ def _active_uid_dedupe(
         uid
         for msg in messages
         if isinstance(msg, dict)
+        and msg.get(RESTORED_FROM_STORE) is True
         and not isinstance(msg.get("_row_id"), int)
         and not isinstance(msg.get(DB_ROW_SNAPSHOT), str)
         and (uid := message_uid_or_none(msg))