Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions contributors/emails/pieralukasz@gmail.com
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
pieralukasz
# PR #132639 fold-in
1 change: 1 addition & 0 deletions hermes_cli/hooks.py
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,7 @@ def _mtime_drift(shell_hooks, spec, entry) -> tuple[bool | None, str, str]:
},
"on_session_finalize": {"session_id": "test-session"},
"on_session_reset": {"session_id": "test-session"},
"on_session_delete": {"session_id": "test-session", "reason": "delete", "profile": "default"},
"pre_api_request": {
"session_id": "test-session", "task_id": "test-task", "platform": "cli",
"model": "claude-sonnet-4-6", "provider": "anthropic",
Expand Down
2 changes: 1 addition & 1 deletion hermes_cli/plugins.py
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,7 @@ def _install_plugin_debug_handler(force: bool = False) -> None:
# Run-all-then-pick-first (see get_plugin_error_classification). Privacy: error_message/
# error_body may be unredacted.
"transform_api_error_classification", "on_session_start", "on_session_end",
"on_session_finalize", "on_session_reset",
"on_session_finalize", "on_session_reset", "on_session_delete",
# on_skill_lifecycle: successful skill lifecycle facts (local skill name visible to plugins).
"on_skill_lifecycle", "subagent_start", "subagent_stop",
# pre_gateway_dispatch: once per incoming MessageEvent, after the internal-event guard, BEFORE
Expand Down
37 changes: 37 additions & 0 deletions hermes_state_common.py
Original file line number Diff line number Diff line change
Expand Up @@ -392,6 +392,43 @@ def _id_chunks(ids, size: int = _SQL_IN_CHUNK):
yield ids[start:start + size]


def emit_session_deleted(db, deleted_ids, *, reason, primary=None, sessions_dir=None):
"""Fire ``on_session_delete`` after a committed delete or deliberate archive (#124511).

Call right after the write transaction commits (before follow-up cleanup that can raise), with
the ids it removed or, for ``reason="archive"``, hid. *primary* is the session the caller asked
for; without it the first id stands in (bulk sweeps). ``deleted_ids`` rides along when more
than one row went. Dispatch runs under the store's own profile home, so a process touching
another profile's store (``sessions repair-profiles``) reaches THAT profile's plugins.
Late-bound and fail-open: never breaks the delete.
"""
ids = sorted({sid for sid in (deleted_ids or ()) if isinstance(sid, str) and sid})
if not ids:
return
try:
from hermes_cli.lifecycle import invoke_hook
from hermes_constants import (
get_hermes_home, profile_name_for_home, reset_hermes_home_override, set_hermes_home_override,
)

home = os.path.dirname(os.path.abspath(os.fspath(db.db_path)))
profile = profile_name_for_home(home)
payload: dict = {"session_id": primary if primary in ids else ids[0], "reason": reason, "profile": profile}
if len(ids) > 1:
payload["deleted_ids"] = ids
if sessions_dir is not None:
payload["sessions_dir"] = str(sessions_dir)
rebind = profile is not None and os.path.realpath(home) != os.path.realpath(get_hermes_home())
token = set_hermes_home_override(home) if rebind else None
try:
invoke_hook("on_session_delete", **payload)
finally:
if token is not None:
reset_hermes_home_override(token)
except Exception:
logger.warning("on_session_delete dispatch failed", exc_info=True)


_FTS_TRIGGERS = ("messages_fts_insert", "messages_fts_delete", "messages_fts_update",
"messages_fts_trigram_insert", "messages_fts_trigram_delete", "messages_fts_trigram_update")

Expand Down
36 changes: 35 additions & 1 deletion hermes_state_maintenance.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@

from hermes_state_common import (
AUTO_VACUUM_MIN_FREELIST_RATIO, _id_chunks, _non_continuation_child_sql, _placeholders, _sql_session_last_active,
escape_like as _escape_like
escape_like as _escape_like, emit_session_deleted,
)
from hermes_startup_watchdog import report_startup_progress

Expand Down Expand Up @@ -132,6 +132,7 @@ def _do(conn):
self._delete_unreferenced_system_prompts(conn)
return ids
removed_ids = self._execute_write(_do) or []
emit_session_deleted(self, removed_ids, reason="ghost_prune", sessions_dir=sessions_dir)
from hermes_state_media import collect_retired_media
collect_retired_media(self)
for sid in removed_ids if sessions_dir else ():
Expand Down Expand Up @@ -329,6 +330,37 @@ def archive_stale_sessions(self, idle_days: float, *, exclude_pinned: bool = Tru
self._auto_archive_lineage(row[0])
return len(rows)

def maybe_auto_archive(
self, idle_days: float = 3, min_interval_hours: int = 24, exclude_pinned: bool = True,
) -> dict[str, Any]:
"""Idempotent, non-destructive auto-archive of sessions idle for ``idle_days``; state_meta
``last_auto_archive`` gates runs within ``min_interval_hours``. Never raises."""
result: dict[str, Any] = {"skipped": False, "archived": 0}
try:
now = time.time()
try:
last = float(self.get_meta("last_auto_archive") or 0.0)
except (TypeError, ValueError):
last = 0.0 # corrupt meta; treat as no prior run
if last and now - last < min_interval_hours * 3600:
result["skipped"] = True
return result
# Startup-watchdog lease: the archive sweep is I/O-bound (near-zero CPU),
# which the watchdog's CPU fallback misreads as a parked deadlock.
# No-op when the watchdog is not armed; never raises.
report_startup_progress(900.0, phase="state_db_auto_archive")
archived = result["archived"] = self.archive_stale_sessions(idle_days, exclude_pinned=exclude_pinned)
# Record even a zero-archive run so we don't re-sweep every call.
self.set_meta("last_auto_archive", str(now))
if archived > 0:
logger.info(
"state.db auto-archive: archived %d session(s) idle >= %s days", archived, idle_days,
)
except Exception as exc:
logger.warning("state.db auto-archive failed: %s", exc)
result["error"] = str(exc)
return result

def prune_sessions(self, older_than_days: Optional[float] = 90, source: str | None = None,
sessions_dir: Optional[Path] = None, exclude_active_write_guards: bool = False,
deleted_ids: Optional[list] = None, **filters) -> int:
Expand Down Expand Up @@ -367,6 +399,8 @@ def _do(conn):
self._delete_unreferenced_system_prompts(conn)
return len(session_ids)
count = self._execute_write(_do)
if count:
emit_session_deleted(self, removed_ids, reason="prune", sessions_dir=sessions_dir)
from hermes_state_media import collect_retired_media
collect_retired_media(self)
for sid in removed_ids:
Expand Down
3 changes: 2 additions & 1 deletion hermes_state_portability.py
Original file line number Diff line number Diff line change
Expand Up @@ -471,7 +471,8 @@ def _retire_donor_segment(self, donor_db: Any, seg_id: str) -> bool:
# stamped even on ended segments.
donor_db.reopen_session(seg_id)
donor_db.end_session(seg_id, "adopted_by_profile")
donor_db.set_session_archived(seg_id, True)
# The session lives on in the adopting profile: not a delete for plugins.
donor_db.set_session_archived(seg_id, True, notify=False)
return True
except Exception:
logger.warning("failed to retire donor segment %s after adoption", seg_id, exc_info=True)
Expand Down
11 changes: 9 additions & 2 deletions hermes_state_profile_repair.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,11 @@
import logging
from typing import Any, Dict, Iterable, List, Optional, Set, Tuple

from hermes_state_common import _id_chunks, _placeholders as _session_ids_placeholders
from hermes_state_common import (
_id_chunks,
_placeholders as _session_ids_placeholders,
emit_session_deleted,
)

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -284,7 +288,10 @@ def _do(conn) -> bool:
conn.execute("DELETE FROM sessions WHERE id = ?", (session_id,))
self._delete_unreferenced_system_prompts(conn)
return True
return bool(self._execute_write(_do))
deleted = self._execute_write(_do)
if deleted:
emit_session_deleted(self, [session_id], reason="profile_repair")
return bool(deleted)

# ── routing index ─────────────────────────────────────────────────────────

Expand Down
27 changes: 22 additions & 5 deletions hermes_state_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
"""
import json
import uuid
from typing import NamedTuple, Optional

from agent.conversation_compression_archive import ABSORBED_ROW_IDS
from agent.message_metadata import (
Expand All @@ -13,6 +14,12 @@
from hermes_state_keys import admission_fingerprint


class SessionGoneEvent(NamedTuple):
"""The ``on_session_delete`` event a committed runtime mutation owes its plugins."""
session_ids: tuple
reason: str


class RuntimeStoreError(ValueError):
"""A refusal by the state authority. ``reason`` is the bounded wire code (callers compare it as a
string); ``field`` and ``detail`` name the check that fired and why, so a client and the gateway
Expand Down Expand Up @@ -449,11 +456,12 @@ def mutate_runtime_session(db, *, epoch: int, principal_id: str, session_id: str
key, digest = mutation_receipt_key(principal_id, session_id, request_id, operation=operation, payload=payload,
expected_revision=expected_revision, expected_generation=expected_generation)

def write(conn):
def write(conn) -> tuple[dict, Optional[SessionGoneEvent]]:
"""The receipt result plus the on_session_delete event this commit owes (None on replay)."""
_epoch(conn, epoch)
replay = replay_mutation_receipt(conn, key, digest)
if replay is not None:
return replay
return replay, None
session = conn.execute('SELECT * FROM sessions WHERE id=?', (session_id,)).fetchone()
if session is None and operation != 'import':
raise RuntimeStoreError('not_found')
Expand All @@ -475,8 +483,13 @@ def write(conn):
# The branch commit's own idle rule (queued input too), before the owner starts servers.
from hermes_state_mutation_guards import require_idle
require_idle(db, conn, list({session_id, snapshot['target']}))
return {'snapshot': snapshot}
return {'snapshot': snapshot}, None
affected, projection = apply_action(db, conn, session_id, operation, payload, prepared=_prepared)
event = None
if operation == 'delete':
event = SessionGoneEvent(tuple(projection.get('deleted_ids') or (session_id,)), 'delete')
elif payload.get('archived') is True and session is not None and not session['archived']:
event = SessionGoneEvent(tuple(affected) or (session_id,), 'archive')
conn.executemany('UPDATE sessions SET runtime_revision=runtime_revision+1 WHERE id=?',
[(target,) for target in affected])
updated = conn.execute('SELECT * FROM sessions WHERE id=?', (session_id,)).fetchone()
Expand All @@ -485,8 +498,12 @@ def write(conn):
'operation': operation, **projection}
conn.execute('INSERT INTO state_meta(key,value) VALUES(?,?)',
(key, _json({'digest': digest, 'result': result})))
return result
return db._execute_write(write)
return result, event
result, event = db._execute_write(write)
if event is not None:
from hermes_state_common import emit_session_deleted
emit_session_deleted(db, event.session_ids, reason=event.reason, primary=session_id)
return result


_IMPORT_KEY = 'gateway.prompt_admissions_import.v1'
Expand Down
60 changes: 23 additions & 37 deletions hermes_state_sessions.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,15 +14,14 @@
from agent.session_activity import (
ActivityProvenance, bound_activity_description, normalize_activity_provenance,
)
from hermes_startup_watchdog import report_startup_progress
from hermes_state_errors import SessionActiveWriteGuardError
from hermes_state_common import (
_LISTABLE_CHILD_SQL, _RECOVERABLE_END_REASONS,
_RECOVERABLE_END_REASONS_SQL, _RESET_CHILD_SQL, _RESET_END_REASONS, _legacy_reset_child_sql, _non_continuation_child_sql,
_shape_preview, _sql_preview_raw, QUEUED_PROMPT_METADATA_KEY,
_sql_in_window, _sql_json_extract, _sql_session_last_active, _sql_session_last_active_by_id,
escape_like as _escape_like, _SQL_IN_CHUNK, _id_chunks, _placeholders as _session_ids_placeholders,
_source_match_clause,
_source_match_clause, emit_session_deleted,
)
from hermes_state_compression import _CHAIN_CAP

Expand Down Expand Up @@ -935,12 +934,21 @@ def _set_lineage_column_in_transaction(self, conn, column: str, session_id: str,
(session_id, session_id, value),
).fetchall()]

def set_session_archived(self, session_id: str, archived: bool) -> bool:
def set_session_archived(self, session_id: str, archived: bool, *, notify: bool = True) -> bool:
"""Soft-hide (or unhide) a session and its compression lineage; messages are kept.
This is the DELIBERATE archive (user, CLI, API): it clears the ``auto_archived``
provenance, so re-activation never un-hides it on the user's behalf."""
return self._set_lineage_column(
"archived", session_id, int(archived), extra_set_sql=", auto_archived = 0")
provenance, so re-activation never un-hides it on the user's behalf. A real
unarchived -> archived flip fires ``on_session_delete`` (``reason="archive"``) after commit;
``notify=False`` is for housekeeping that hides a row whose session lives on elsewhere."""
def _do(conn):
row = conn.execute("SELECT archived FROM sessions WHERE id = ?", (session_id,)).fetchone()
ids = self._set_lineage_column_in_transaction(
conn, "archived", session_id, int(archived), extra_set_sql=", auto_archived = 0")
return ids, bool(archived and ids and row is not None and not row[0])
ids, newly_archived = self._execute_write(_do)
if newly_archived and notify:
emit_session_deleted(self, ids, reason="archive", primary=session_id)
return bool(ids)

def _auto_archive_lineage(self, session_id: str) -> bool:
"""The idle sweep's archive: like :meth:`set_session_archived` but stamps
Expand Down Expand Up @@ -1753,6 +1761,8 @@ def _do(conn):
removed_ids.extend(scope)
return True
deleted = self._execute_write(_do)
if deleted:
emit_session_deleted(self, removed_ids, reason="delete", primary=session_id, sessions_dir=sessions_dir)
from hermes_state_media import collect_retired_media
collect_retired_media(self)
for sid in removed_ids:
Expand Down Expand Up @@ -1813,6 +1823,8 @@ def _do(conn):
self._delete_unreferenced_system_prompts(conn)
return True
deleted = self._execute_write(_do)
if deleted:
emit_session_deleted(self, [session_id], reason="empty_sweep", sessions_dir=sessions_dir)
from hermes_state_media import collect_retired_media
collect_retired_media(self)
if deleted:
Expand Down Expand Up @@ -1900,6 +1912,9 @@ def expand(ids):
# 3 rows, however many physical links those rows had.
return len(existing)
count = self._execute_write(_do)
if count:
emit_session_deleted(self, removed_ids, reason="delete", primary=unique_ids[0] if len(unique_ids) == 1 else None,
sessions_dir=sessions_dir)
from hermes_state_media import collect_retired_media
collect_retired_media(self)
for sid in removed_ids:
Expand Down Expand Up @@ -1942,6 +1957,8 @@ def _do(conn):
self._delete_unreferenced_system_prompts(conn)
return len(session_ids)
count = self._execute_write(_do)
if count:
emit_session_deleted(self, removed_ids, reason="empty_sweep", sessions_dir=sessions_dir)
from hermes_state_media import collect_retired_media
collect_retired_media(self)
for sid in removed_ids:
Expand All @@ -1961,34 +1978,3 @@ def archive_sessions(
for row in rows:
self.set_session_archived(row["id"], True)
return len(rows)

def maybe_auto_archive(
self, idle_days: float = 3, min_interval_hours: int = 24, exclude_pinned: bool = True,
) -> dict[str, Any]:
"""Idempotent, non-destructive auto-archive of sessions idle for ``idle_days``; state_meta
``last_auto_archive`` gates runs within ``min_interval_hours``. Never raises."""
result: dict[str, Any] = {"skipped": False, "archived": 0}
try:
now = time.time()
try:
last = float(self.get_meta("last_auto_archive") or 0.0)
except (TypeError, ValueError):
last = 0.0 # corrupt meta; treat as no prior run
if last and now - last < min_interval_hours * 3600:
result["skipped"] = True
return result
# Startup-watchdog lease: the archive sweep is I/O-bound (near-zero CPU),
# which the watchdog's CPU fallback misreads as a parked deadlock.
# No-op when the watchdog is not armed; never raises.
report_startup_progress(900.0, phase="state_db_auto_archive")
archived = result["archived"] = self.archive_stale_sessions(idle_days, exclude_pinned=exclude_pinned)
# Record even a zero-archive run so we don't re-sweep every call.
self.set_meta("last_auto_archive", str(now))
if archived > 0:
logger.info(
"state.db auto-archive: archived %d session(s) idle >= %s days", archived, idle_days,
)
except Exception as exc:
logger.warning("state.db auto-archive failed: %s", exc)
result["error"] = str(exc)
return result
Loading
Loading