diff --git a/agent/message_metadata.py b/agent/message_metadata.py index 86bcbe3c31..52c0c126dd 100644 --- a/agent/message_metadata.py +++ b/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. diff --git a/agent/session_persistence.py b/agent/session_persistence.py index b8a48402d8..76237a3407 100644 --- a/agent/session_persistence.py +++ b/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: diff --git a/agent/transcript_repair.py b/agent/transcript_repair.py index ed34938f2c..a9f9a19333 100644 --- a/agent/transcript_repair.py +++ b/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))