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

fix(sessions): atomically verify display history before export deletion

Browse Source 
(cherry picked from commit f6db4ffd686728d7adc8391e5b1144786a388246)
This commit is contained in:
JoaoMarcos44
2026-09-23 07:13:05 -03:00
committed by kshitij
parent fe8b643db6
commit 539c6b9feb
6 changed files with 146 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
hermes_state_messages.py
View File

@@ -1002,6 +1002,44 @@ class SessionMessagesMixin:
msg.pop(key)
return msg
@staticmethod
def _display_rows_from_conn(conn, session_id: str, *, limit: Optional[int] = None,
offset: int = 0, latest: bool = False):
"""One display-history projection for normal reads and transactional verification."""
direction = "DESC" if latest else "ASC"
return conn.execute(
f"""WITH page AS (
SELECT display_order FROM messages
WHERE session_id = ? AND (active = 1 OR compacted = 1)
GROUP BY display_order ORDER BY display_order {direction}
LIMIT ? OFFSET ?
)
SELECT chosen.* FROM page
JOIN messages AS chosen ON chosen.id = (
SELECT candidate.id FROM messages AS candidate
WHERE candidate.session_id = ?
AND candidate.display_order = page.display_order
AND (candidate.active = 1 OR candidate.compacted = 1)
ORDER BY candidate.active DESC, candidate.id DESC LIMIT 1
)
ORDER BY page.display_order ASC""",
(session_id, -1 if limit is None else limit, offset, session_id),
).fetchall()
def _display_messages_from_conn(self, conn, session_id: str) -> Optional[List[Dict[str, Any]]]:
"""Exact display snapshot on an already-held transaction; None means fail closed."""
if conn.execute("SELECT 1 FROM sessions WHERE id = ? LIMIT 1", (session_id,)).fetchone() is None:
return None
if conn.execute(
"SELECT 1 FROM messages WHERE session_id = ? AND (active = 1 OR compacted = 1) "
"AND display_order IS NULL LIMIT 1", (session_id,),
).fetchone():
return None
return [
self._row_to_message_dict(row, warn_context="verified delete", summary_flag=True)
for row in self._display_rows_from_conn(conn, session_id)
]
@staticmethod
def _active_clause(include_inactive: bool, include_compacted: bool) -> str:
"""Audit: every row; display: active plus compaction-archived (never Undo/Rewind rows); default: live."""
@@ -1019,23 +1057,9 @@ class SessionMessagesMixin:
raise ValueError("after_id is incompatible with include_compacted (deduped display reads use offset paging)")
active_clause = self._active_clause(include_inactive, include_compacted)
if include_compacted and not include_inactive and self._ensure_display_order(session_id):
direction = "DESC" if latest else "ASC"
sql = f"""WITH page AS (
SELECT display_order FROM messages
WHERE session_id = ? AND (active = 1 OR compacted = 1)
GROUP BY display_order ORDER BY display_order {direction}
LIMIT ? OFFSET ?
)
SELECT chosen.* FROM page
JOIN messages AS chosen ON chosen.id = (
SELECT candidate.id FROM messages AS candidate
WHERE candidate.session_id = ?
AND candidate.display_order = page.display_order
AND (candidate.active = 1 OR candidate.compacted = 1)
ORDER BY candidate.active DESC, candidate.id DESC LIMIT 1
)
ORDER BY page.display_order ASC"""
rows = self._read_all(sql, [session_id, -1 if limit is None else limit, offset, session_id])
with self._read_ctx() as conn:
rows = self._display_rows_from_conn(
conn, session_id, limit=limit, offset=offset, latest=latest)
elif include_compacted:
# Read-only legacy stores cannot persist display identities; keep only fixed-width
# identities and representative ids while scanning, then fetch the selected payloads.