hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 5 Packages Projects Releases Wiki Activity

refactor(compression): name the adopted-snapshot lease watermark in docs and strip row ids explicitly

Browse Source 
The lease watermark now means the highest row the compaction already
represents: MAX(id) at lease start, advanced to the newest row of an
adopted durable snapshot. Say so wherever it still read as the lease-start
value, and compute the adopted maximum with newest_exact_held_id() before
removing _row_id in a separate loop instead of popping inside max().

(cherry picked from commit 3e5331777a580c9ffe337b4623746fa78a966bbb)
This commit is contained in:
Eva
2026-09-29 01:20:50 +09:00
committed by kshitij
parent d7480fec0c
commit 55d82b04ae
3 changed files with 26 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
agent/conversation_compression.py
View File

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

12
hermes_state_compression.py
View File

@@ -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.

3
hermes_state_messages.py
View File

@@ -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(