diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 2cba17fc57..4eb0f13d09 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -2634,8 +2634,9 @@ class _CompactionLifecycle: class _CompressionLease: """The per-attempt durable compression lock plus its lifecycle plumbing. - ``holder`` is None when no durable lock is owned (legacy DB, no session db); ``watermark`` is MAX(id) of - active rows at lease start (None = archive everything, no concurrent-tail preservation this cycle).""" + ``holder`` is None when no durable lock is owned (legacy DB, no session db); ``watermark`` is the highest + active row the compaction already represents: MAX(id) at lease start, advanced to the newest row of an + adopted durable snapshot (None = archive everything, no concurrent-tail preservation this cycle).""" def __init__( self, agent: Any, *, db: Any, sid: str, ttl: float, refresh_interval: Any, @@ -2747,10 +2748,11 @@ def _abort_lease( def _try_acquire_durable_lock(lease: _CompressionLease, try_acquire: Any, commit_fence: Any) -> bool: """Acquire the durable lock for ``lease.holder`` and capture the start watermark. - Watermark = MAX(id) of active rows at START: appends aren't blocked during summary; later rows are - concurrent tail that archive_and_compact re-sequences. Capture is safety-additive (fallback archives - everything), so its failure never aborts. An acquire that raises is not version skew: fail closed and - release holder-qualified best-effort (safe if never acquired).""" + Watermark = MAX(id) of active rows at START (durable-snapshot adoption may advance it to the newest adopted + row): appends aren't blocked during summary; later rows are concurrent tail that archive_and_compact + re-sequences. Capture is safety-additive (fallback archives everything), so its failure never aborts. An + acquire that raises is not version skew: fail closed and release holder-qualified best-effort (safe if never + acquired).""" try: acquired = try_acquire(lease.sid, lease.holder, ttl_seconds=lease.ttl) if acquired: @@ -2921,7 +2923,8 @@ def _adopt_grown_durable_parent(agent: Any, lease: _CompressionLease, messages: """Return the durable parent transcript when it outgrew the in-memory snapshot. Rotation only (in-place never loses rows). The snapshot predates the lease: if durable grew, a writer committed a turn — ADOPT it (aborting wedged busy sessions forever). Length check only: in-memory edits of - past turns are legal.""" + past turns are legal. Adoption advances ``lease.watermark`` to the snapshot's newest row, so publication + clones only rows the snapshot does not already carry.""" if lease.db is None or not lease.sid: return None durable_loader = getattr(type(lease.db), "get_messages_as_conversation", None) @@ -2957,8 +2960,11 @@ def _adopt_grown_durable_parent(agent: Any, lease: _CompressionLease, messages: durable_parent = durable_loader(lease.db, lease.sid, include_row_ids=True) if not (isinstance(durable_parent, list) and len(durable_parent) > len(messages)): return None - adopted_max_row_id = max((message.pop("_row_id", 0) or 0 for message in durable_parent), default=0) - if lease.watermark is not None: + from agent.conversation_compression_archive import newest_exact_held_id + adopted_max_row_id = newest_exact_held_id(durable_parent) + for message in durable_parent: + message.pop("_row_id", None) + if lease.watermark is not None and adopted_max_row_id is not None: lease.watermark = max(lease.watermark, adopted_max_row_id) logger.info( "compression: session=%s grew before lease (%d → %d msgs); adopting durable snapshot", lease.sid, len(messages), @@ -3360,7 +3366,8 @@ def _publish_rotated_compaction( if _parent_deliberately_ended(agent._session_db, old_session_id): raise RuntimeError(f"Compression parent already ended: {old_session_id}") # Foreign-tail ceiling: the flush below writes OUR rows (already in handoff); - # rows above the start watermark up to this MAX(id) are foreign appends. + # rows above the lease watermark (lease start, or the newest adopted-snapshot row) up to this MAX(id) + # are foreign appends. # No trustworthy ceiling means the clone could duplicate the handoff: skip tail preservation this rotation. _foreign_tail_ceiling = None with contextlib.suppress(Exception): diff --git a/hermes_state_compression.py b/hermes_state_compression.py index ce6679b2e8..dc53551d2f 100644 --- a/hermes_state_compression.py +++ b/hermes_state_compression.py @@ -252,11 +252,13 @@ class SessionCompressionMixin: watermark: Optional[int] = None, watermark_ceiling: Optional[int] = None) -> None: """Atomically close a parent and publish its durable compression child: closure, child row, and handoff commit in one transaction, so readers see the live parent or a complete child, never an - ended parent with a missing/empty child. *watermark* (parent's ``get_active_message_watermark`` at compression start): parent rows with ``id - > watermark`` — appends landed during the slow summary — are column-cloned into the child AFTER the - handoff. *watermark_ceiling* bounds the clone: the rotation path flushes its OWN transcript to the - parent just before publishing and those rows are already in the handoff, so only ``(watermark, - watermark_ceiling]`` is foreign tail (``None`` = unbounded). *require_lease_refresh* + + ended parent with a missing/empty child. *watermark* (the parent's highest row already represented + in the handoff: its ``get_active_message_watermark`` at compression start, or the newest row of an + adopted durable snapshot): parent rows with ``id > watermark`` — appends landed during the slow + summary — are column-cloned into the child AFTER the handoff. *watermark_ceiling* bounds the clone: + the rotation path flushes its OWN transcript to the parent just before publishing and those rows are + already in the handoff, so only ``(watermark, watermark_ceiling]`` is foreign tail (``None`` = + unbounded). *require_lease_refresh* + *compression_lock_holder* refreshes the lease on the same ``conn`` before the expiry check (no TOCTOU window), so a refresher that died on transient DB errors gets one last chance. diff --git a/hermes_state_messages.py b/hermes_state_messages.py index fc3bcd88dc..38cf3646e6 100644 --- a/hermes_state_messages.py +++ b/hermes_state_messages.py @@ -896,7 +896,8 @@ class SessionMessagesMixin: def get_active_message_watermark(self, session_id: str) -> int: """MAX(id) of the active rows (0 if none), captured at compression START: every active row above it - arrived concurrently and must survive compaction verbatim.""" + arrived concurrently and must survive compaction verbatim. Rotation may advance the lease's copy past + rows an adopted durable snapshot already carries.""" if not session_id: return 0 return int(self._read_one(