From a420b9755c837927a1f770da6c968686a8f5e590 Mon Sep 17 00:00:00 2001
From: finn763 <165816600+finn763@users.noreply.github.com>
Date: Sun, 27 Sep 2026 16:53:15 +0800
Subject: [PATCH 1/6] fix(plugins): fire on_session_delete on every
 session-deletion path

---
 gateway/platforms/api_server.py               |   2 +
 hermes_cli/hooks.py                           |   1 +
 hermes_cli/lifecycle.py                       |  19 +++
 hermes_cli/plugins.py                         |   2 +-
 hermes_state_common.py                        |  30 +++++
 hermes_state_maintenance.py                   |   6 +-
 hermes_state_profile_repair.py                |  11 +-
 hermes_state_sessions.py                      |   9 +-
 .../hermes_state/test_session_delete_hook.py  | 121 ++++++++++++++++++
 website/docs/developer-guide/plugins/index.md |   1 +
 website/docs/user-guide/features/hooks.md     |  26 ++++
 .../current/developer-guide/plugins/index.md  |   1 +
 .../current/user-guide/features/hooks.md      |   1 +
 13 files changed, 225 insertions(+), 5 deletions(-)
 create mode 100644 tests/hermes_state/test_session_delete_hook.py

diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py
index 9e8096c3233b2..f50c17616db81 100644
--- a/gateway/platforms/api_server.py
+++ b/gateway/platforms/api_server.py
@@ -3122,6 +3122,8 @@ def _atomic(conn):
                     try:
                         db._resolve_title_conflict(conn, session_id, clean_title)
                     except ValueError as exc:  # the DB's uniqueness rule; undo the INSERT
+                        # No ``on_session_delete`` here: this rolls back the row inserted in the
+                        # same transaction, so the session never existed — firing would double-count it.
                         conn.execute("DELETE FROM sessions WHERE id = ?", (session_id,))
                         return None, f"title:{exc}"
                 conn.execute("UPDATE sessions SET title = ? WHERE id = ?", (clean_title, session_id))
diff --git a/hermes_cli/hooks.py b/hermes_cli/hooks.py
index 32a1532634097..1f2df15c24f20 100644
--- a/hermes_cli/hooks.py
+++ b/hermes_cli/hooks.py
@@ -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"},
     "pre_api_request": {
         "session_id": "test-session", "task_id": "test-task", "platform": "cli",
         "model": "claude-sonnet-4-6", "provider": "anthropic",
diff --git a/hermes_cli/lifecycle.py b/hermes_cli/lifecycle.py
index 6ff300d729b99..6bee3b1541236 100644
--- a/hermes_cli/lifecycle.py
+++ b/hermes_cli/lifecycle.py
@@ -95,3 +95,22 @@ def finalize_session(**kwargs: Any) -> list[Any]:
             logger.warning("Core Relay session finalization failed", exc_info=True)
 
     return _plugin_hooks("on_session_finalize", **kwargs)
+
+
+def notify_session_deleted(session_id: str, **kwargs: Any) -> list[Any]:
+    """Fire the ``on_session_delete`` post-commit observer (#124511).
+
+    Call only after the delete transaction commits and only when a row was
+    actually removed. Observer-only: returns ignored, failures fail open
+    (logged, never raised) so a slow/broken plugin cannot break deletion.
+    Unbounded/caller-thread like ``on_session_finalize`` — cleanup work
+    ("close a handle, unlink a directory") must not be abandoned mid-way.
+    # ponytail: single process-local fan-out; per-surface reason/platform
+    # enrichment stays with callers, add a queue here only if a callback
+    # measurably blocks a delete path.
+    """
+    try:
+        return invoke_hook("on_session_delete", session_id=session_id, **kwargs)
+    except Exception:
+        logger.warning("on_session_delete dispatch failed", exc_info=True)
+        return []
diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py
index c7fad5e387d71..52cfd2d5175d6 100644
--- a/hermes_cli/plugins.py
+++ b/hermes_cli/plugins.py
@@ -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
diff --git a/hermes_state_common.py b/hermes_state_common.py
index c3b0a08cc7bdf..1546cba068697 100644
--- a/hermes_state_common.py
+++ b/hermes_state_common.py
@@ -392,6 +392,36 @@ def _id_chunks(ids, size: int = _SQL_IN_CHUNK):
         yield ids[start:start + size]
 
 
+def emit_session_deleted(deleted_ids=None, *, reason=None, sessions_dir=None):
+    """Fire the ``on_session_delete`` post-commit observer for committed deletes (#124511).
+
+    Late-bound seam: the state layer must not import ``hermes_cli`` at module
+    top level, so the dispatch resolves inside the call and fails open — a
+    missing/broken plugin stack never breaks deletion. Call only after the
+    write transaction commits, with the ids actually removed. One invocation
+    per delete operation; ``deleted_ids`` rides along only when the count
+    differs from 1.
+    """
+    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 notify_session_deleted
+    except Exception:
+        return
+    payload: dict = {"session_id": ids[0]}
+    if len(ids) > 1:
+        payload["deleted_ids"] = ids
+    if reason is not None:
+        payload["reason"] = reason
+    if sessions_dir is not None:
+        payload["sessions_dir"] = str(sessions_dir)
+    try:
+        notify_session_deleted(**payload)
+    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")
 
diff --git a/hermes_state_maintenance.py b/hermes_state_maintenance.py
index d37556a004cec..6fd0cc5f5f858 100644
--- a/hermes_state_maintenance.py
+++ b/hermes_state_maintenance.py
@@ -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
 
@@ -136,6 +136,8 @@ def _do(conn):
         collect_retired_media(self)
         for sid in removed_ids if sessions_dir else ():
             self._remove_session_files(sessions_dir, sid)
+        if removed_ids:
+            emit_session_deleted(removed_ids, reason="ghost_prune", sessions_dir=sessions_dir)
         return len(removed_ids)
 
     def _guarded_ids(self, conn, ids: Iterable[str]) -> set:
@@ -373,6 +375,8 @@ def _do(conn):
             self._remove_session_files(sessions_dir, sid)
         if deleted_ids is not None:
             deleted_ids.extend(removed_ids)
+        if count:
+            emit_session_deleted(removed_ids, reason="prune", sessions_dir=sessions_dir)
         return count
 
     def _page_pragmas(self, names: tuple[str, ...], fail_msg: str) -> Optional[list]:
diff --git a/hermes_state_profile_repair.py b/hermes_state_profile_repair.py
index 716f1a1494601..0ebb6e2b4632f 100644
--- a/hermes_state_profile_repair.py
+++ b/hermes_state_profile_repair.py
@@ -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__)
 
@@ -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([session_id], reason="profile_repair")
+        return bool(deleted)
 
     # ── routing index ─────────────────────────────────────────────────────────
 
diff --git a/hermes_state_sessions.py b/hermes_state_sessions.py
index 28b7cf601218d..68eb98df26e9d 100644
--- a/hermes_state_sessions.py
+++ b/hermes_state_sessions.py
@@ -22,7 +22,7 @@
     _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
 
@@ -1757,6 +1757,8 @@ def _do(conn):
         collect_retired_media(self)
         for sid in removed_ids:
             self._remove_session_files(sessions_dir, sid)
+        if deleted:
+            emit_session_deleted(removed_ids, sessions_dir=sessions_dir)
         return bool(deleted)
 
     def discard_unadmitted_session(self, session_id: str) -> bool:
@@ -1817,6 +1819,7 @@ def _do(conn):
         collect_retired_media(self)
         if deleted:
             self._remove_session_files(sessions_dir, session_id)
+            emit_session_deleted([session_id], sessions_dir=sessions_dir)
         return deleted
 
     def delete_sessions(
@@ -1904,6 +1907,8 @@ def expand(ids):
         collect_retired_media(self)
         for sid in removed_ids:
             self._remove_session_files(sessions_dir, sid)
+        if count:
+            emit_session_deleted(removed_ids, sessions_dir=sessions_dir)
         return count
 
     # Shared by count_empty_sessions / delete_empty_sessions so badge and sweep agree. message_count
@@ -1946,6 +1951,8 @@ def _do(conn):
         collect_retired_media(self)
         for sid in removed_ids:
             self._remove_session_files(sessions_dir, sid)
+        if count:
+            emit_session_deleted(removed_ids, reason="empty_sweep", sessions_dir=sessions_dir)
         return count
 
     def archive_sessions(
diff --git a/tests/hermes_state/test_session_delete_hook.py b/tests/hermes_state/test_session_delete_hook.py
new file mode 100644
index 0000000000000..639e9624eed6e
--- /dev/null
+++ b/tests/hermes_state/test_session_delete_hook.py
@@ -0,0 +1,121 @@
+"""on_session_delete post-commit observer (#124511).
+
+Delete paths must notify plugins after the commit so per-session on-disk
+state can be cleaned up without polling the store. Failing observers fail
+open — deletion itself never breaks.
+"""
+
+import pytest
+
+from hermes_cli import lifecycle
+from hermes_cli.plugins import VALID_HOOKS
+from hermes_state import SessionDB
+
+
+@pytest.fixture()
+def db(tmp_path):
+    session_db = SessionDB(db_path=tmp_path / "test_state.db")
+    yield session_db
+    session_db.close()
+
+
+@pytest.fixture(autouse=True)
+def _no_fts_rebuild_throttle(monkeypatch):
+    monkeypatch.setattr(SessionDB, "_FTS_REBUILD_MIN_PAUSE", 0.0)
+    monkeypatch.setattr(SessionDB, "_FTS_REBUILD_DUTY_FACTOR", 0.0)
+
+
+@pytest.fixture()
+def fired(monkeypatch):
+    calls = []
+
+    def _spy(**kwargs):
+        calls.append(kwargs)
+        return []
+
+    monkeypatch.setattr(lifecycle, "notify_session_deleted", _spy)
+    return calls
+
+
+def test_hook_is_registered():
+    assert "on_session_delete" in VALID_HOOKS
+
+
+def test_delete_session_fires_once(db, fired):
+    db.create_session(session_id="s1", source="cli")
+    db.append_message("s1", role="user", content="hi")
+
+    assert db.delete_session("s1") is True
+
+    assert len(fired) == 1
+    assert fired[0]["session_id"] == "s1"
+    assert "deleted_ids" not in fired[0]
+
+
+def test_delete_session_unknown_id_does_not_fire(db, fired):
+    assert db.delete_session("missing") is False
+    assert fired == []
+
+
+def test_delete_sessions_bulk_fires_once_with_ids(db, fired):
+    db.create_session(session_id="s1", source="cli")
+    db.create_session(session_id="s2", source="cli")
+    db.create_session(session_id="keep", source="cli")
+
+    assert db.delete_sessions(["s1", "s2", "missing"]) == 2
+
+    assert len(fired) == 1
+    assert fired[0]["session_id"] == "s1"
+    assert fired[0]["deleted_ids"] == ["s1", "s2"]
+
+
+def test_delete_empty_sessions_carries_reason(db, fired):
+    db.create_session(session_id="empty", source="cli")
+    db.end_session("empty", "done")
+
+    assert db.delete_empty_sessions() == 1
+
+    assert len(fired) == 1
+    assert fired[0]["session_id"] == "empty"
+    assert fired[0]["reason"] == "empty_sweep"
+
+
+def test_delete_moved_session_fires_with_reason(db, fired):
+    db.create_session(session_id="moved1", source="cli")
+
+    assert db.delete_moved_session("moved1") is True
+
+    assert len(fired) == 1
+    assert fired[0]["session_id"] == "moved1"
+    assert fired[0]["reason"] == "profile_repair"
+
+
+def test_delete_moved_session_unknown_id_does_not_fire(db, fired):
+    assert db.delete_moved_session("missing") is False
+    assert fired == []
+
+
+def test_failing_observer_does_not_break_delete(db, monkeypatch):
+    def _boom(**kwargs):
+        raise RuntimeError("plugin down")
+
+    monkeypatch.setattr(lifecycle, "notify_session_deleted", _boom)
+    db.create_session(session_id="s1", source="cli")
+
+    assert db.delete_session("s1") is True
+    assert db.get_session("s1") is None
+
+
+def test_notify_dispatches_hook_and_fails_open(monkeypatch):
+    seen = []
+    monkeypatch.setattr(
+        lifecycle, "invoke_hook", lambda name, **kw: seen.append((name, kw)) or []
+    )
+    lifecycle.notify_session_deleted("s1", reason="prune")
+    assert seen == [("on_session_delete", {"session_id": "s1", "reason": "prune"})]
+
+    def _boom(name, **kw):
+        raise RuntimeError("dispatch down")
+
+    monkeypatch.setattr(lifecycle, "invoke_hook", _boom)
+    assert lifecycle.notify_session_deleted("s1") == []
diff --git a/website/docs/developer-guide/plugins/index.md b/website/docs/developer-guide/plugins/index.md
index 7e2fc1ccd63dc..4d5ab02f1f052 100644
--- a/website/docs/developer-guide/plugins/index.md
+++ b/website/docs/developer-guide/plugins/index.md
@@ -1352,6 +1352,7 @@ Each hook is documented in full on the **[Event Hooks reference](../../user-guid
 | [`on_session_end`](../../user-guide/features/hooks.md#on_session_end) | End of every `run_conversation` call + CLI exit | `session_id: str, completed: bool, interrupted: bool, model: str, platform: str` | ignored |
 | [`on_session_finalize`](../../user-guide/features/hooks.md#on_session_finalize) | CLI/gateway tears down an active session | `session_id: str \| None, platform: str` | ignored |
 | [`on_session_reset`](../../user-guide/features/hooks.md#on_session_reset) | Gateway swaps in a new session key (`/new`, `/reset`) | `session_id: str, platform: str` | ignored |
+| [`on_session_delete`](../../user-guide/features/hooks.md#on_session_delete) | A session row delete commits (single, bulk, sweep, prune) | `session_id: str`, optional `deleted_ids: list[str]`, `reason: str`, `sessions_dir: str` | ignored |
 | [`gateway_platform_event`](../../user-guide/features/hooks.md#gateway_platform_event) | An authorized platform-native event is normalized at the gateway boundary (Telegram reactions currently) | `platform: str, event_type: str, payload: dict` | ignored |
 | `kanban_task_claimed` | A kanban task is claimed (dispatcher process, before the worker spawns) | `task_id: str, board: str \| None, assignee: str \| None, run_id: int \| None, profile_name: str` | ignored |
 | `kanban_task_completed` | A kanban task completes (worker process) | `task_id, board, assignee, run_id, profile_name, summary: str \| None` | ignored |
diff --git a/website/docs/user-guide/features/hooks.md b/website/docs/user-guide/features/hooks.md
index 0c42653be2d6b..725ac2888a44b 100644
--- a/website/docs/user-guide/features/hooks.md
+++ b/website/docs/user-guide/features/hooks.md
@@ -471,6 +471,7 @@ Payload fields below are the exact event-specific fields supplied by each call s
 | `on_session_end` | Observer | Canonically at each turn finalization; CLI/TUI exits have additional reduced legacy shapes. Return ignored. | Canonical: `session_id`, `task_id`, `turn_id`, `completed`, `failed`, `interrupted`, `turn_exit_reason`, `model`, `platform`; exit paths may add `reason`/`api_request_id` and omit fields. | IDs, model/platform, and outcome; canonical payload has no message body. |
 | `on_session_finalize` | Observer | CLI/TUI/gateway teardown through `finalize_session`; gateway shutdown may finalize without a reset. Return ignored. | Surface-dependent `session_id`, `platform`, optionally `reason`, `old_session_id`, `new_session_id` | Session and routing identifiers. |
 | `on_session_reset` | Observer | CLI/TUI session boundary and gateway after the replacement session exists; return ignored. | CLI: `session_id`, `platform`, `reason`; TUI: `session_id`, `platform`; gateway: those plus `reason`, `old_session_id`, `new_session_id` | Session and routing identifiers. |
+| `on_session_delete` | Observer | After a session row delete commits (single, bulk, empty-sweep, prune, ghost prune); return ignored. | `session_id`, optional `deleted_ids` (full set when the count differs from 1), `reason`, `sessions_dir` | Session identifiers only; no message content. |
 | `agent_loop_stopped` | Observer | Immediately after a real running agent is interrupted — gateway `_interrupt_and_clear_session` or TUI/desktop `session.interrupt`; return ignored. | `session_key`, `platform`, `reason`, `invalidation_reason` | Session/routing identifiers and interruption reasons; no message body. |
 | `on_skill_lifecycle` | Observer | After an authoritative skill-usage state change; return ignored. | `action`, `skill_name`, `provenance`, `task_id`, `session_id`, `use_count`, `reused`, `reuse_after_patch` | Exposes the local skill name and provenance. |
 | `subagent_start` | Observer | Child constructed and about to run; return ignored. | `parent_session_id`, `parent_turn_id`, `parent_subagent_id`, `child_session_id`, `child_subagent_id`, `child_role`, `child_goal` | Child goal may contain user/project content. |
@@ -1081,6 +1082,31 @@ def my_callback(session_id: str, platform: str, **kwargs):
 
 ---
 
+### `on_session_delete`
+
+Fires **once per delete operation after the transaction commits**, when a session row was actually removed — covering single deletes, bulk deletes, the empty-session sweep, retention prunes, and ghost-prune maintenance. This lets plugins that keep per-session on-disk state clean up without polling the session store.
+
+**Callback signature:**
+
+```python
+def my_callback(session_id: str, **kwargs):
+```
+
+| Parameter | Type | Description |
+|-----------|------|-------------|
+| `session_id` | `str` | The session that was deleted. Always present. |
+| `deleted_ids` | `list[str]`, optional | Full set removed by the operation, including cascade-deleted delegate children and every row in a bulk delete. Present when the count differs from 1. |
+| `reason` | `str`, optional | Why the delete happened (`"prune"`, `"ghost_prune"`, `"empty_sweep"`, …). |
+| `sessions_dir` | `str`, optional | Absolute path of the profile's `sessions/` dir, so a plugin does not have to re-derive the per-profile home. |
+
+**Fires:** Post-commit only, and only when the delete actually happened (unknown ids in a bulk delete are skipped silently). Return ignored; a failing callback is logged and skipped like every other observer. The payload carries no message content — only ids, the reason, and the directory.
+
+**Return value:** Ignored.
+
+**Use cases:** Delete a per-session sidecar database or cache directory, emit "session deleted" analytics, release per-session external resources.
+
+---
+
 See the **[Build a Plugin guide](../../developer-guide/plugins/index.md)** for the full walkthrough including tool schemas, handlers, and advanced hook patterns.
 
 ---
diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/developer-guide/plugins/index.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/developer-guide/plugins/index.md
index 486ec4e8c9117..2182a9ca4b62f 100644
--- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/developer-guide/plugins/index.md
+++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/developer-guide/plugins/index.md
@@ -552,6 +552,7 @@ def register(ctx):
 | [`on_session_end`](../../user-guide/features/hooks.md#on_session_end) | 每次 `run_conversation` 调用结束 + CLI 退出 | `session_id: str, completed: bool, interrupted: bool, model: str, platform: str` | 忽略 |
 | [`on_session_finalize`](../../user-guide/features/hooks.md#on_session_finalize) | CLI/网关销毁活跃会话 | `session_id: str \| None, platform: str` | 忽略 |
 | [`on_session_reset`](../../user-guide/features/hooks.md#on_session_reset) | 网关切换新会话键（`/new`、`/reset`） | `session_id: str, platform: str` | 忽略 |
+| [`on_session_delete`](../../user-guide/features/hooks.md#on_session_delete) | 会话行删除提交后（单删、批量、sweep、prune） | `session_id: str`，可选 `deleted_ids: list[str]`、`reason: str`、`sessions_dir: str` | 忽略 |
 
 大多数钩子是即发即忘的观察者——其返回值被忽略。例外是 `pre_llm_call`，它可以向对话中注入上下文。
 
diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/hooks.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/hooks.md
index dc590b393814f..515221f21d397 100644
--- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/hooks.md
+++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/hooks.md
@@ -393,6 +393,7 @@ def register(ctx):
 | `on_session_end` | 观察者 | Canonical 路径在每轮 finalize；CLI/TUI 退出还有精简 legacy shape。 | Canonical：`session_id`, `task_id`, `turn_id`, `completed`, `failed`, `interrupted`, `turn_exit_reason`, `model`, `platform`；退出路径可能增加 `reason`/`api_request_id` 并省略字段。 | ID、model/platform 和结果；canonical payload 无消息正文。 |
 | `on_session_finalize` | 观察者 | CLI/TUI/gateway 通过 `finalize_session` teardown；gateway 关闭时可只 finalize 而不 reset。忽略返回值。 | 按 surface：`session_id`, `platform`，可选 `reason`, `old_session_id`, `new_session_id` | Session 和 routing 标识。 |
 | `on_session_reset` | 观察者 | CLI/TUI session boundary，或 gateway 创建替代 session 后；忽略返回值。 | CLI：`session_id`, `platform`, `reason`；TUI：`session_id`, `platform`；gateway：另有 `reason`, `old_session_id`, `new_session_id` | Session 和 routing 标识。 |
+| `on_session_delete` | 观察者 | 会话行删除提交后（单删、批量、empty-sweep、prune、ghost prune）；忽略返回值。 | `session_id`，可选 `deleted_ids`（删除数不为 1 时为全集）、`reason`、`sessions_dir` | 仅 session 标识；无消息内容。 |
 | `on_skill_lifecycle` | 观察者 | 权威 skill 使用状态变更后；忽略返回值。 | `action`, `skill_name`, `provenance`, `task_id`, `session_id`, `use_count`, `reused`, `reuse_after_patch` | 暴露本地 skill 名和 provenance。 |
 | `subagent_start` | 观察者 | 子 agent 已构造、即将运行；忽略返回值。 | `parent_session_id`, `parent_turn_id`, `parent_subagent_id`, `child_session_id`, `child_subagent_id`, `child_role`, `child_goal` | Child goal 可能含用户/项目内容。 |
 | `subagent_stop` | 观察者 | 子 agent 退出；忽略返回值。 | `parent_session_id`, `parent_turn_id`, `child_session_id`, `child_role`, `child_summary`, `child_status`, `tool_call_history`, `duration_ms` | Summary 和已脱敏 tool-history metadata 仍可能暴露项目结构。 |

From 3cc62b3c37a2601206796b4cc6918c8a731bf93d Mon Sep 17 00:00:00 2001
From: finn763 <165816600+finn763@users.noreply.github.com>
Date: Fri, 9 Oct 2026 14:29:15 +0800
Subject: [PATCH 2/6] docs(hooks): note on_session_delete delivery is
 process-local, not cross-process (#124988)

---
 website/docs/user-guide/features/hooks.md | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/website/docs/user-guide/features/hooks.md b/website/docs/user-guide/features/hooks.md
index 725ac2888a44b..9f58bd7e4538a 100644
--- a/website/docs/user-guide/features/hooks.md
+++ b/website/docs/user-guide/features/hooks.md
@@ -1101,6 +1101,8 @@ def my_callback(session_id: str, **kwargs):
 
 **Fires:** Post-commit only, and only when the delete actually happened (unknown ids in a bulk delete are skipped silently). Return ignored; a failing callback is logged and skipped like every other observer. The payload carries no message content — only ids, the reason, and the directory.
 
+**Delivery scope:** Process-local — the callback runs only in the process that committed the delete. It is not a durable cross-process delivery channel, and it carries no exactly-once guarantee: external side effects driven from the callback should tolerate a missed or duplicated invocation.
+
 **Return value:** Ignored.
 
 **Use cases:** Delete a per-session sidecar database or cache directory, emit "session deleted" analytics, release per-session external resources.

From aa0014460331c7ce22fff1de065652c6530cc8d2 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Sun, 18 Oct 2026 00:38:09 -0700
Subject: [PATCH 3/6] feat(plugins): on_session_delete covers Desktop deletes
 and deliberate archives

Finn's state-layer hook missed the paths Desktop actually uses: single
delete and archive go through the canonical mutation receipt
(mutate_runtime_session), not SessionDB.delete_session, so a plugin never
heard about either. Archive (the "this is done" signal #132639 asked for)
fired nowhere.

- mutate_runtime_session fires after its write commits for operation
  'delete' and for a real unarchived -> archived sidebar flip; an exact
  retry replays the receipt and stays silent.
- set_session_archived fires reason="archive" on a real flip, so TUI
  session.archive, `hermes sessions archive` and deferred TUI archives are
  covered from one place. Repeats, un-archives, the idle auto-archive
  sweep and profile adoption (notify=False) stay silent.
- Payload is now session_id + reason + profile on every path (profile
  from the store's home), deleted_ids when more than one row went;
  primary id is the one the caller asked for, not the alphabetically
  first.
- One fail-open seam (emit_session_deleted) calls lifecycle.invoke_hook
  directly; the extra notify_session_deleted wrapper is gone.
- Tests trimmed to two invariants against the real hook registry.

Co-authored-by: Lucas Piera <pieralukasz@gmail.com>
---
 hermes_cli/hooks.py                           |   2 +-
 hermes_cli/lifecycle.py                       |  19 --
 hermes_state_common.py                        |  42 ++---
 hermes_state_maintenance.py                   |   4 +-
 hermes_state_portability.py                   |   3 +-
 hermes_state_profile_repair.py                |   2 +-
 hermes_state_runtime.py                       |  14 +-
 hermes_state_sessions.py                      |  26 ++-
 .../hermes_state/test_session_delete_hook.py  | 176 ++++++++----------
 .../docs/developer-guide/observer-hooks.md    |   1 +
 website/docs/developer-guide/plugins/index.md |   2 +-
 website/docs/user-guide/features/hooks.md     |  21 ++-
 website/docs/user-guide/features/plugins.md   |   2 +-
 .../current/developer-guide/plugins/index.md  |   2 +-
 .../current/user-guide/features/hooks.md      |   2 +-
 15 files changed, 156 insertions(+), 162 deletions(-)

diff --git a/hermes_cli/hooks.py b/hermes_cli/hooks.py
index 1f2df15c24f20..011fb9a3db1dc 100644
--- a/hermes_cli/hooks.py
+++ b/hermes_cli/hooks.py
@@ -127,7 +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"},
+    "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",
diff --git a/hermes_cli/lifecycle.py b/hermes_cli/lifecycle.py
index 6bee3b1541236..6ff300d729b99 100644
--- a/hermes_cli/lifecycle.py
+++ b/hermes_cli/lifecycle.py
@@ -95,22 +95,3 @@ def finalize_session(**kwargs: Any) -> list[Any]:
             logger.warning("Core Relay session finalization failed", exc_info=True)
 
     return _plugin_hooks("on_session_finalize", **kwargs)
-
-
-def notify_session_deleted(session_id: str, **kwargs: Any) -> list[Any]:
-    """Fire the ``on_session_delete`` post-commit observer (#124511).
-
-    Call only after the delete transaction commits and only when a row was
-    actually removed. Observer-only: returns ignored, failures fail open
-    (logged, never raised) so a slow/broken plugin cannot break deletion.
-    Unbounded/caller-thread like ``on_session_finalize`` — cleanup work
-    ("close a handle, unlink a directory") must not be abandoned mid-way.
-    # ponytail: single process-local fan-out; per-surface reason/platform
-    # enrichment stays with callers, add a queue here only if a callback
-    # measurably blocks a delete path.
-    """
-    try:
-        return invoke_hook("on_session_delete", session_id=session_id, **kwargs)
-    except Exception:
-        logger.warning("on_session_delete dispatch failed", exc_info=True)
-        return []
diff --git a/hermes_state_common.py b/hermes_state_common.py
index 1546cba068697..a45358749dd79 100644
--- a/hermes_state_common.py
+++ b/hermes_state_common.py
@@ -392,32 +392,32 @@ def _id_chunks(ids, size: int = _SQL_IN_CHUNK):
         yield ids[start:start + size]
 
 
-def emit_session_deleted(deleted_ids=None, *, reason=None, sessions_dir=None):
-    """Fire the ``on_session_delete`` post-commit observer for committed deletes (#124511).
-
-    Late-bound seam: the state layer must not import ``hermes_cli`` at module
-    top level, so the dispatch resolves inside the call and fails open — a
-    missing/broken plugin stack never breaks deletion. Call only after the
-    write transaction commits, with the ids actually removed. One invocation
-    per delete operation; ``deleted_ids`` rides along only when the count
-    differs from 1.
+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 only after the write transaction commits, with the ids it actually removed (or, for
+    ``reason="archive"``, hid). *primary* is the session the caller asked for; without it the
+    first removed id stands in (bulk sweeps). ``deleted_ids`` rides along only when more than
+    one row went. Late-bound and fail-open: the state layer never imports ``hermes_cli`` at module
+    top level, and a missing or raising plugin stack 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 notify_session_deleted
-    except Exception:
-        return
-    payload: dict = {"session_id": ids[0]}
-    if len(ids) > 1:
-        payload["deleted_ids"] = ids
-    if reason is not None:
-        payload["reason"] = reason
-    if sessions_dir is not None:
-        payload["sessions_dir"] = str(sessions_dir)
-    try:
-        notify_session_deleted(**payload)
+        from hermes_cli.lifecycle import invoke_hook
+        from hermes_constants import profile_name_for_home
+
+        payload: dict = {
+            "session_id": primary if primary in ids else ids[0],
+            "reason": reason,
+            "profile": profile_name_for_home(os.path.dirname(os.fspath(db.db_path))),
+        }
+        if len(ids) > 1:
+            payload["deleted_ids"] = ids
+        if sessions_dir is not None:
+            payload["sessions_dir"] = str(sessions_dir)
+        invoke_hook("on_session_delete", **payload)
     except Exception:
         logger.warning("on_session_delete dispatch failed", exc_info=True)
 
diff --git a/hermes_state_maintenance.py b/hermes_state_maintenance.py
index 6fd0cc5f5f858..345cd749e1df5 100644
--- a/hermes_state_maintenance.py
+++ b/hermes_state_maintenance.py
@@ -137,7 +137,7 @@ def _do(conn):
         for sid in removed_ids if sessions_dir else ():
             self._remove_session_files(sessions_dir, sid)
         if removed_ids:
-            emit_session_deleted(removed_ids, reason="ghost_prune", sessions_dir=sessions_dir)
+            emit_session_deleted(self, removed_ids, reason="ghost_prune", sessions_dir=sessions_dir)
         return len(removed_ids)
 
     def _guarded_ids(self, conn, ids: Iterable[str]) -> set:
@@ -376,7 +376,7 @@ def _do(conn):
         if deleted_ids is not None:
             deleted_ids.extend(removed_ids)
         if count:
-            emit_session_deleted(removed_ids, reason="prune", sessions_dir=sessions_dir)
+            emit_session_deleted(self, removed_ids, reason="prune", sessions_dir=sessions_dir)
         return count
 
     def _page_pragmas(self, names: tuple[str, ...], fail_msg: str) -> Optional[list]:
diff --git a/hermes_state_portability.py b/hermes_state_portability.py
index bdbddfb313c04..f794ca6c6825d 100644
--- a/hermes_state_portability.py
+++ b/hermes_state_portability.py
@@ -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)
diff --git a/hermes_state_profile_repair.py b/hermes_state_profile_repair.py
index 0ebb6e2b4632f..b591b6f089bf9 100644
--- a/hermes_state_profile_repair.py
+++ b/hermes_state_profile_repair.py
@@ -290,7 +290,7 @@ def _do(conn) -> bool:
             return True
         deleted = self._execute_write(_do)
         if deleted:
-            emit_session_deleted([session_id], reason="profile_repair")
+            emit_session_deleted(self, [session_id], reason="profile_repair")
         return bool(deleted)
 
     # ── routing index ─────────────────────────────────────────────────────────
diff --git a/hermes_state_runtime.py b/hermes_state_runtime.py
index 2c9500c449e3c..0ec37531785c0 100644
--- a/hermes_state_runtime.py
+++ b/hermes_state_runtime.py
@@ -449,7 +449,11 @@ 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)
 
+    # (ids, reason) for on_session_delete; reset per attempt (the write retries) and empty on replay.
+    gone: list = []
+
     def write(conn):
+        gone.clear()
         _epoch(conn, epoch)
         replay = replay_mutation_receipt(conn, key, digest)
         if replay is not None:
@@ -477,6 +481,10 @@ def write(conn):
                 require_idle(db, conn, list({session_id, snapshot['target']}))
             return {'snapshot': snapshot}
         affected, projection = apply_action(db, conn, session_id, operation, payload, prepared=_prepared)
+        if operation == 'delete':
+            gone[:] = [projection.get('deleted_ids') or [session_id], 'delete']
+        elif payload.get('archived') is True and session is not None and not session['archived']:
+            gone[:] = [[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()
@@ -486,7 +494,11 @@ def write(conn):
         conn.execute('INSERT INTO state_meta(key,value) VALUES(?,?)',
                      (key, _json({'digest': digest, 'result': result})))
         return result
-    return db._execute_write(write)
+    result = db._execute_write(write)
+    if gone:
+        from hermes_state_common import emit_session_deleted
+        emit_session_deleted(db, gone[0], reason=gone[1], primary=session_id)
+    return result
 
 
 _IMPORT_KEY = 'gateway.prompt_admissions_import.v1'
diff --git a/hermes_state_sessions.py b/hermes_state_sessions.py
index 68eb98df26e9d..7d70e8a6b6b28 100644
--- a/hermes_state_sessions.py
+++ b/hermes_state_sessions.py
@@ -935,12 +935,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, [session_id], reason="archive")
+        return bool(ids)
 
     def _auto_archive_lineage(self, session_id: str) -> bool:
         """The idle sweep's archive: like :meth:`set_session_archived` but stamps
@@ -1758,7 +1767,7 @@ def _do(conn):
         for sid in removed_ids:
             self._remove_session_files(sessions_dir, sid)
         if deleted:
-            emit_session_deleted(removed_ids, sessions_dir=sessions_dir)
+            emit_session_deleted(self, removed_ids, reason="delete", primary=session_id, sessions_dir=sessions_dir)
         return bool(deleted)
 
     def discard_unadmitted_session(self, session_id: str) -> bool:
@@ -1819,7 +1828,7 @@ def _do(conn):
         collect_retired_media(self)
         if deleted:
             self._remove_session_files(sessions_dir, session_id)
-            emit_session_deleted([session_id], sessions_dir=sessions_dir)
+            emit_session_deleted(self, [session_id], reason="empty_sweep", sessions_dir=sessions_dir)
         return deleted
 
     def delete_sessions(
@@ -1908,7 +1917,8 @@ def expand(ids):
         for sid in removed_ids:
             self._remove_session_files(sessions_dir, sid)
         if count:
-            emit_session_deleted(removed_ids, sessions_dir=sessions_dir)
+            emit_session_deleted(self, removed_ids, reason="delete", primary=unique_ids[0] if len(unique_ids) == 1 else None,
+                                 sessions_dir=sessions_dir)
         return count
 
     # Shared by count_empty_sessions / delete_empty_sessions so badge and sweep agree. message_count
@@ -1952,7 +1962,7 @@ def _do(conn):
         for sid in removed_ids:
             self._remove_session_files(sessions_dir, sid)
         if count:
-            emit_session_deleted(removed_ids, reason="empty_sweep", sessions_dir=sessions_dir)
+            emit_session_deleted(self, removed_ids, reason="empty_sweep", sessions_dir=sessions_dir)
         return count
 
     def archive_sessions(
diff --git a/tests/hermes_state/test_session_delete_hook.py b/tests/hermes_state/test_session_delete_hook.py
index 639e9624eed6e..e81582a330aeb 100644
--- a/tests/hermes_state/test_session_delete_hook.py
+++ b/tests/hermes_state/test_session_delete_hook.py
@@ -1,121 +1,95 @@
-"""on_session_delete post-commit observer (#124511).
+"""``on_session_delete`` (#124511): every committed delete or deliberate archive notifies plugins once.
 
-Delete paths must notify plugins after the commit so per-session on-disk
-state can be cleaned up without polling the store. Failing observers fail
-open — deletion itself never breaks.
+Real ``SessionDB`` + the real plugin hook registry. The callback reads the row back, so each event
+also proves dispatch ran AFTER the commit, never before.
 """
 
 import pytest
 
-from hermes_cli import lifecycle
-from hermes_cli.plugins import VALID_HOOKS
+import hermes_state_runtime as rt
+from hermes_cli.plugins import get_plugin_manager
 from hermes_state import SessionDB
 
 
 @pytest.fixture()
 def db(tmp_path):
-    session_db = SessionDB(db_path=tmp_path / "test_state.db")
+    session_db = SessionDB(db_path=tmp_path / "state.db")
     yield session_db
     session_db.close()
 
 
-@pytest.fixture(autouse=True)
-def _no_fts_rebuild_throttle(monkeypatch):
-    monkeypatch.setattr(SessionDB, "_FTS_REBUILD_MIN_PAUSE", 0.0)
-    monkeypatch.setattr(SessionDB, "_FTS_REBUILD_DUTY_FACTOR", 0.0)
-
-
 @pytest.fixture()
-def fired(monkeypatch):
-    calls = []
-
-    def _spy(**kwargs):
-        calls.append(kwargs)
-        return []
-
-    monkeypatch.setattr(lifecycle, "notify_session_deleted", _spy)
-    return calls
-
-
-def test_hook_is_registered():
-    assert "on_session_delete" in VALID_HOOKS
-
-
-def test_delete_session_fires_once(db, fired):
-    db.create_session(session_id="s1", source="cli")
-    db.append_message("s1", role="user", content="hi")
-
-    assert db.delete_session("s1") is True
-
-    assert len(fired) == 1
-    assert fired[0]["session_id"] == "s1"
-    assert "deleted_ids" not in fired[0]
-
-
-def test_delete_session_unknown_id_does_not_fire(db, fired):
-    assert db.delete_session("missing") is False
-    assert fired == []
-
-
-def test_delete_sessions_bulk_fires_once_with_ids(db, fired):
-    db.create_session(session_id="s1", source="cli")
-    db.create_session(session_id="s2", source="cli")
-    db.create_session(session_id="keep", source="cli")
-
-    assert db.delete_sessions(["s1", "s2", "missing"]) == 2
-
-    assert len(fired) == 1
-    assert fired[0]["session_id"] == "s1"
-    assert fired[0]["deleted_ids"] == ["s1", "s2"]
-
-
-def test_delete_empty_sessions_carries_reason(db, fired):
-    db.create_session(session_id="empty", source="cli")
+def hook(db):
+    """Register callbacks on the registry ``invoke_hook`` reads; restore it afterwards."""
+    manager = get_plugin_manager()
+    saved = {name: list(cbs) for name, cbs in manager._hooks.items()}
+    events: list[dict] = []
+
+    def _record(**kw):
+        row = db.get_session(kw["session_id"])
+        events.append({**kw, "row_after": None if row is None else bool(row["archived"])})
+
+    manager._hooks.setdefault("on_session_delete", []).append(_record)
+    try:
+        yield manager, events
+    finally:
+        manager._hooks = saved
+
+
+def _canonical(db, sid, operation, payload):
+    epoch = rt.begin_runtime_epoch(db, instance_id="test")
+    row = db.get_session(sid)
+    return rt.mutate_runtime_session(
+        db, epoch=epoch, principal_id="human", session_id=sid, request_id=f"{operation}-{sid}",
+        expected_revision=row["runtime_revision"], expected_generation=row["runtime_generation"],
+        operation=operation, payload=payload)
+
+
+def test_every_delete_and_archive_path_fires_once_after_commit(db, hook):
+    _, events = hook
+    for sid in ("cli", "bulk1", "bulk2", "empty", "old", "desk", "arch", "desk-arch"):
+        db.create_session(sid, source="cli")
+        if sid != "empty":
+            db.append_message(sid, role="user", content="hi")
     db.end_session("empty", "done")
-
-    assert db.delete_empty_sessions() == 1
-
-    assert len(fired) == 1
-    assert fired[0]["session_id"] == "empty"
-    assert fired[0]["reason"] == "empty_sweep"
-
-
-def test_delete_moved_session_fires_with_reason(db, fired):
-    db.create_session(session_id="moved1", source="cli")
-
-    assert db.delete_moved_session("moved1") is True
-
-    assert len(fired) == 1
-    assert fired[0]["session_id"] == "moved1"
-    assert fired[0]["reason"] == "profile_repair"
-
-
-def test_delete_moved_session_unknown_id_does_not_fire(db, fired):
-    assert db.delete_moved_session("missing") is False
-    assert fired == []
-
-
-def test_failing_observer_does_not_break_delete(db, monkeypatch):
-    def _boom(**kwargs):
+    db.end_session("old", "done")
+
+    assert db.delete_session("cli")                          # hermes sessions delete, /exit --delete
+    assert db.delete_session("missing") is False             # unknown id: silent
+    assert db.delete_sessions(["bulk1", "bulk2", "nope"]) == 2   # Desktop/dashboard bulk delete
+    assert db.delete_empty_sessions() == 1                   # "Delete empty (N)"
+    assert db.prune_sessions(older_than_days=0) == 1         # retention prune (only ended row)
+    _canonical(db, "desk", "delete", {})                     # Desktop DELETE /api/sessions/{id}
+    assert db.set_session_archived("arch", True)             # TUI session.archive, CLI archive
+    assert db.set_session_archived("arch", True)             # repeat archive: silent
+    assert db.set_session_archived("arch", False)            # un-archive: silent
+    _canonical(db, "desk-arch", "sidebar", {"archived": True})   # Desktop archive (PATCH)
+    db.archive_stale_sessions(0)                             # idle auto-archive sweep: silent
+
+    got = [(e["session_id"], e["reason"], e.get("deleted_ids"), e["row_after"]) for e in events]
+    assert got == [
+        ("cli", "delete", None, None),
+        ("bulk1", "delete", ["bulk1", "bulk2"], None),
+        ("empty", "empty_sweep", None, None),
+        ("old", "prune", None, None),
+        ("desk", "delete", None, None),
+        ("arch", "archive", None, True),
+        ("desk-arch", "archive", None, True),
+    ]
+    assert all(e["profile"] == events[0]["profile"] for e in events)
+
+
+def test_raising_callback_never_breaks_deletion(db, hook):
+    manager, events = hook
+
+    def _boom(**_):
         raise RuntimeError("plugin down")
 
-    monkeypatch.setattr(lifecycle, "notify_session_deleted", _boom)
-    db.create_session(session_id="s1", source="cli")
+    manager._hooks["on_session_delete"].insert(0, _boom)
+    db.create_session("s1", source="cli")
+    db.create_session("s2", source="cli")
 
     assert db.delete_session("s1") is True
-    assert db.get_session("s1") is None
-
-
-def test_notify_dispatches_hook_and_fails_open(monkeypatch):
-    seen = []
-    monkeypatch.setattr(
-        lifecycle, "invoke_hook", lambda name, **kw: seen.append((name, kw)) or []
-    )
-    lifecycle.notify_session_deleted("s1", reason="prune")
-    assert seen == [("on_session_delete", {"session_id": "s1", "reason": "prune"})]
-
-    def _boom(name, **kw):
-        raise RuntimeError("dispatch down")
-
-    monkeypatch.setattr(lifecycle, "invoke_hook", _boom)
-    assert lifecycle.notify_session_deleted("s1") == []
+    assert db.set_session_archived("s2", True) is True
+    assert db.get_session("s1") is None and db.get_session("s2")["archived"]
+    assert [e["session_id"] for e in events] == ["s1", "s2"]  # later observers still ran
diff --git a/website/docs/developer-guide/observer-hooks.md b/website/docs/developer-guide/observer-hooks.md
index a9cfb60b9b1b0..4eab86d065a32 100644
--- a/website/docs/developer-guide/observer-hooks.md
+++ b/website/docs/developer-guide/observer-hooks.md
@@ -99,6 +99,7 @@ Session hooks describe conversation boundaries and resets:
 | `on_session_end` | A `run_conversation` call ends, including interrupted or incomplete turns. |
 | `on_session_finalize` | CLI or gateway tears down an active session identity. |
 | `on_session_reset` | CLI or gateway moves from an old session identity to a new one. |
+| `on_session_delete` | A session delete or deliberate archive committed (`reason` says which). |
 
 Common fields include `session_id`, `completed`, `interrupted`, `reason`,
 `old_session_id`, and `new_session_id` where available.
diff --git a/website/docs/developer-guide/plugins/index.md b/website/docs/developer-guide/plugins/index.md
index 4d5ab02f1f052..8e94559319157 100644
--- a/website/docs/developer-guide/plugins/index.md
+++ b/website/docs/developer-guide/plugins/index.md
@@ -1352,7 +1352,7 @@ Each hook is documented in full on the **[Event Hooks reference](../../user-guid
 | [`on_session_end`](../../user-guide/features/hooks.md#on_session_end) | End of every `run_conversation` call + CLI exit | `session_id: str, completed: bool, interrupted: bool, model: str, platform: str` | ignored |
 | [`on_session_finalize`](../../user-guide/features/hooks.md#on_session_finalize) | CLI/gateway tears down an active session | `session_id: str \| None, platform: str` | ignored |
 | [`on_session_reset`](../../user-guide/features/hooks.md#on_session_reset) | Gateway swaps in a new session key (`/new`, `/reset`) | `session_id: str, platform: str` | ignored |
-| [`on_session_delete`](../../user-guide/features/hooks.md#on_session_delete) | A session row delete commits (single, bulk, sweep, prune) | `session_id: str`, optional `deleted_ids: list[str]`, `reason: str`, `sessions_dir: str` | ignored |
+| [`on_session_delete`](../../user-guide/features/hooks.md#on_session_delete) | A session delete or deliberate archive commits (every surface, sweeps, prune) | `session_id: str, reason: str, profile: str \| None`, optional `deleted_ids: list[str]`, `sessions_dir: str` | ignored |
 | [`gateway_platform_event`](../../user-guide/features/hooks.md#gateway_platform_event) | An authorized platform-native event is normalized at the gateway boundary (Telegram reactions currently) | `platform: str, event_type: str, payload: dict` | ignored |
 | `kanban_task_claimed` | A kanban task is claimed (dispatcher process, before the worker spawns) | `task_id: str, board: str \| None, assignee: str \| None, run_id: int \| None, profile_name: str` | ignored |
 | `kanban_task_completed` | A kanban task completes (worker process) | `task_id, board, assignee, run_id, profile_name, summary: str \| None` | ignored |
diff --git a/website/docs/user-guide/features/hooks.md b/website/docs/user-guide/features/hooks.md
index 9f58bd7e4538a..19e8f2c8b0303 100644
--- a/website/docs/user-guide/features/hooks.md
+++ b/website/docs/user-guide/features/hooks.md
@@ -471,7 +471,7 @@ Payload fields below are the exact event-specific fields supplied by each call s
 | `on_session_end` | Observer | Canonically at each turn finalization; CLI/TUI exits have additional reduced legacy shapes. Return ignored. | Canonical: `session_id`, `task_id`, `turn_id`, `completed`, `failed`, `interrupted`, `turn_exit_reason`, `model`, `platform`; exit paths may add `reason`/`api_request_id` and omit fields. | IDs, model/platform, and outcome; canonical payload has no message body. |
 | `on_session_finalize` | Observer | CLI/TUI/gateway teardown through `finalize_session`; gateway shutdown may finalize without a reset. Return ignored. | Surface-dependent `session_id`, `platform`, optionally `reason`, `old_session_id`, `new_session_id` | Session and routing identifiers. |
 | `on_session_reset` | Observer | CLI/TUI session boundary and gateway after the replacement session exists; return ignored. | CLI: `session_id`, `platform`, `reason`; TUI: `session_id`, `platform`; gateway: those plus `reason`, `old_session_id`, `new_session_id` | Session and routing identifiers. |
-| `on_session_delete` | Observer | After a session row delete commits (single, bulk, empty-sweep, prune, ghost prune); return ignored. | `session_id`, optional `deleted_ids` (full set when the count differs from 1), `reason`, `sessions_dir` | Session identifiers only; no message content. |
+| `on_session_delete` | Observer | After a session delete or deliberate archive commits (CLI, TUI, Desktop/dashboard, gateway API, sweeps, prune); return ignored. | `session_id`, `reason`, `profile`, optional `deleted_ids` (full set when more than one row went), `sessions_dir` | Session identifiers only; no message content. |
 | `agent_loop_stopped` | Observer | Immediately after a real running agent is interrupted — gateway `_interrupt_and_clear_session` or TUI/desktop `session.interrupt`; return ignored. | `session_key`, `platform`, `reason`, `invalidation_reason` | Session/routing identifiers and interruption reasons; no message body. |
 | `on_skill_lifecycle` | Observer | After an authoritative skill-usage state change; return ignored. | `action`, `skill_name`, `provenance`, `task_id`, `session_id`, `use_count`, `reused`, `reuse_after_patch` | Exposes the local skill name and provenance. |
 | `subagent_start` | Observer | Child constructed and about to run; return ignored. | `parent_session_id`, `parent_turn_id`, `parent_subagent_id`, `child_session_id`, `child_subagent_id`, `child_role`, `child_goal` | Child goal may contain user/project content. |
@@ -1084,16 +1084,31 @@ def my_callback(session_id: str, platform: str, **kwargs):
 
 ### `on_session_delete`
 
-Fires **once per delete operation after the transaction commits**, when a session row was actually removed — covering single deletes, bulk deletes, the empty-session sweep, retention prunes, and ghost-prune maintenance. This lets plugins that keep per-session on-disk state clean up without polling the session store.
+Fires **once per operation, after the transaction commits**, when a session actually went away: deleted (single, bulk, the empty-session sweep, retention prune, ghost-prune maintenance, profile repair) or deliberately archived. Every surface ends in the same `SessionDB` methods, so CLI (`hermes sessions delete|prune|archive`, `/exit --delete`), TUI, Desktop/dashboard (single and bulk delete, archive) and the gateway API server are all covered. Plugins that keep per-session state on disk can clean up without polling the session store.
 
 **Callback signature:**
 
 ```python
-def my_callback(session_id: str, **kwargs):
+def my_callback(session_id: str, reason: str, profile: str | None, **kwargs):
 ```
 
 | Parameter | Type | Description |
 |-----------|------|-------------|
+| `session_id` | `str` | The session the operation targeted. Always present. |
+| `reason` | `str` | `"delete"`, `"archive"`, `"empty_sweep"`, `"prune"`, `"ghost_prune"` or `"profile_repair"`. |
+| `profile` | `str \| None` | Profile that owns the session store (`"default"` for the root home), `None` for a store outside any profile home. |
+| `deleted_ids` | `list[str]`, optional | Every row removed, including cascade-deleted delegate children, compression-chain links and each row of a bulk delete. Present when more than one row went. |
+| `sessions_dir` | `str`, optional | The profile's `sessions/` dir when the caller passed one. |
+
+**Fires:** Post-commit only, and only when something changed: unknown ids, repeat archives and un-archives stay silent. `reason="archive"` means the user hid the session (the transcript is kept and it can be un-archived); the idle auto-archive sweep and profile adoption do not fire. A failing callback is logged and skipped like every other observer and never fails the delete. The payload carries no message content.
+
+**Delivery scope:** Process-local — the callback runs only in the process that committed the delete. It is not a durable cross-process delivery channel, and it carries no exactly-once guarantee: external side effects driven from the callback should tolerate a missed or duplicated invocation.
+
+**Return value:** Ignored.
+
+**Use cases:** Delete a per-session sidecar database or cache directory, close a linked task when the user archives its chat, emit "session deleted" analytics.
+
+-----------|------|-------------|
 | `session_id` | `str` | The session that was deleted. Always present. |
 | `deleted_ids` | `list[str]`, optional | Full set removed by the operation, including cascade-deleted delegate children and every row in a bulk delete. Present when the count differs from 1. |
 | `reason` | `str`, optional | Why the delete happened (`"prune"`, `"ghost_prune"`, `"empty_sweep"`, …). |
diff --git a/website/docs/user-guide/features/plugins.md b/website/docs/user-guide/features/plugins.md
index 734cabf237da5..3b54037423099 100644
--- a/website/docs/user-guide/features/plugins.md
+++ b/website/docs/user-guide/features/plugins.md
@@ -349,7 +349,7 @@ Plugins can register the 27 lifecycle events currently accepted by `hermes_cli.p
 |---|---|
 | **Directive/control** | `pre_tool_call`, `pre_llm_call`, `pre_verify`, `pre_gateway_dispatch` |
 | **Transform** | `transform_tool_result`, `transform_terminal_output`, `transform_llm_output`, `pre_transcription` |
-| **Observer** | `post_tool_call`, `post_llm_call`, `pre_api_request`, `post_api_request`, `api_request_error`, `pre_auxiliary_call`, `post_auxiliary_call`, `on_stream_start`, `on_stream_delta`, `on_stream_end`, `on_interim_message`, `on_session_start`, `on_session_end`, `on_session_finalize`, `on_session_reset`, `agent_loop_stopped`, `on_skill_lifecycle`, `subagent_start`, `subagent_stop`, `pre_approval_request`, `post_approval_response`, `on_human_input_request`, `on_human_input_resolved`, `pre_command`, `kanban_task_claimed`, `kanban_task_completed`, `kanban_task_blocked` |
+| **Observer** | `post_tool_call`, `post_llm_call`, `pre_api_request`, `post_api_request`, `api_request_error`, `pre_auxiliary_call`, `post_auxiliary_call`, `on_stream_start`, `on_stream_delta`, `on_stream_end`, `on_interim_message`, `on_session_start`, `on_session_end`, `on_session_finalize`, `on_session_reset`, `on_session_delete`, `agent_loop_stopped`, `on_skill_lifecycle`, `subagent_start`, `subagent_stop`, `pre_approval_request`, `post_approval_response`, `on_human_input_request`, `on_human_input_resolved`, `pre_command`, `kanban_task_claimed`, `kanban_task_completed`, `kanban_task_blocked` |
 
 These categories describe current behavior rather than defining future naming rules. Plugin middleware remains a separate registry/surface.
 ## Plugin types
diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/developer-guide/plugins/index.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/developer-guide/plugins/index.md
index 2182a9ca4b62f..7e0c6c0757f69 100644
--- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/developer-guide/plugins/index.md
+++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/developer-guide/plugins/index.md
@@ -552,7 +552,7 @@ def register(ctx):
 | [`on_session_end`](../../user-guide/features/hooks.md#on_session_end) | 每次 `run_conversation` 调用结束 + CLI 退出 | `session_id: str, completed: bool, interrupted: bool, model: str, platform: str` | 忽略 |
 | [`on_session_finalize`](../../user-guide/features/hooks.md#on_session_finalize) | CLI/网关销毁活跃会话 | `session_id: str \| None, platform: str` | 忽略 |
 | [`on_session_reset`](../../user-guide/features/hooks.md#on_session_reset) | 网关切换新会话键（`/new`、`/reset`） | `session_id: str, platform: str` | 忽略 |
-| [`on_session_delete`](../../user-guide/features/hooks.md#on_session_delete) | 会话行删除提交后（单删、批量、sweep、prune） | `session_id: str`，可选 `deleted_ids: list[str]`、`reason: str`、`sessions_dir: str` | 忽略 |
+| [`on_session_delete`](../../user-guide/features/hooks.md#on_session_delete) | 会话删除或主动归档提交后（所有入口、sweep、prune） | `session_id: str, reason: str, profile: str \| None`，可选 `deleted_ids: list[str]`、`sessions_dir: str` | 忽略 |
 
 大多数钩子是即发即忘的观察者——其返回值被忽略。例外是 `pre_llm_call`，它可以向对话中注入上下文。
 
diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/hooks.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/hooks.md
index 515221f21d397..ff4fbd8ca2ecd 100644
--- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/hooks.md
+++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/hooks.md
@@ -393,7 +393,7 @@ def register(ctx):
 | `on_session_end` | 观察者 | Canonical 路径在每轮 finalize；CLI/TUI 退出还有精简 legacy shape。 | Canonical：`session_id`, `task_id`, `turn_id`, `completed`, `failed`, `interrupted`, `turn_exit_reason`, `model`, `platform`；退出路径可能增加 `reason`/`api_request_id` 并省略字段。 | ID、model/platform 和结果；canonical payload 无消息正文。 |
 | `on_session_finalize` | 观察者 | CLI/TUI/gateway 通过 `finalize_session` teardown；gateway 关闭时可只 finalize 而不 reset。忽略返回值。 | 按 surface：`session_id`, `platform`，可选 `reason`, `old_session_id`, `new_session_id` | Session 和 routing 标识。 |
 | `on_session_reset` | 观察者 | CLI/TUI session boundary，或 gateway 创建替代 session 后；忽略返回值。 | CLI：`session_id`, `platform`, `reason`；TUI：`session_id`, `platform`；gateway：另有 `reason`, `old_session_id`, `new_session_id` | Session 和 routing 标识。 |
-| `on_session_delete` | 观察者 | 会话行删除提交后（单删、批量、empty-sweep、prune、ghost prune）；忽略返回值。 | `session_id`，可选 `deleted_ids`（删除数不为 1 时为全集）、`reason`、`sessions_dir` | 仅 session 标识；无消息内容。 |
+| `on_session_delete` | 观察者 | 会话删除或主动归档提交后（CLI、TUI、Desktop/dashboard、gateway API、sweep、prune）；忽略返回值。 | `session_id`、`reason`、`profile`，可选 `deleted_ids`（多于一行时为全集）、`sessions_dir` | 仅 session 标识；无消息内容。 |
 | `on_skill_lifecycle` | 观察者 | 权威 skill 使用状态变更后；忽略返回值。 | `action`, `skill_name`, `provenance`, `task_id`, `session_id`, `use_count`, `reused`, `reuse_after_patch` | 暴露本地 skill 名和 provenance。 |
 | `subagent_start` | 观察者 | 子 agent 已构造、即将运行；忽略返回值。 | `parent_session_id`, `parent_turn_id`, `parent_subagent_id`, `child_session_id`, `child_subagent_id`, `child_role`, `child_goal` | Child goal 可能含用户/项目内容。 |
 | `subagent_stop` | 观察者 | 子 agent 退出；忽略返回值。 | `parent_session_id`, `parent_turn_id`, `child_session_id`, `child_role`, `child_summary`, `child_status`, `tool_call_history`, `duration_ms` | Summary 和已脱敏 tool-history metadata 仍可能暴露项目结构。 |

From c412d6c1a1e1dfe1a17ef8091ed786d3e3bd816a Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Sun, 18 Oct 2026 00:39:42 -0700
Subject: [PATCH 4/6] chore: map contributor email for pieralukasz

---
 contributors/emails/pieralukasz@gmail.com | 2 ++
 1 file changed, 2 insertions(+)
 create mode 100644 contributors/emails/pieralukasz@gmail.com

diff --git a/contributors/emails/pieralukasz@gmail.com b/contributors/emails/pieralukasz@gmail.com
new file mode 100644
index 0000000000000..32189c7cea34d
--- /dev/null
+++ b/contributors/emails/pieralukasz@gmail.com
@@ -0,0 +1,2 @@
+pieralukasz
+# PR #132639 fold-in

From c566f813afa2e64a6f7b45047a007e4c6ea95b99 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Sun, 18 Oct 2026 00:46:40 -0700
Subject: [PATCH 5/6] refactor(state): move maybe_auto_archive beside
 archive_stale_sessions

hermes_state_sessions.py crossed the 2000-line target with the delete/archive
hook calls; the auto-archive wrapper belongs with the sweep it drives in
hermes_state_maintenance.py. Also drops the api_server rollback comment so
that facade does not grow.
---
 gateway/platforms/api_server.py |  2 --
 hermes_state_maintenance.py     | 31 +++++++++++++++++++++++++++++++
 hermes_state_sessions.py        | 32 --------------------------------
 3 files changed, 31 insertions(+), 34 deletions(-)

diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py
index f50c17616db81..9e8096c3233b2 100644
--- a/gateway/platforms/api_server.py
+++ b/gateway/platforms/api_server.py
@@ -3122,8 +3122,6 @@ def _atomic(conn):
                     try:
                         db._resolve_title_conflict(conn, session_id, clean_title)
                     except ValueError as exc:  # the DB's uniqueness rule; undo the INSERT
-                        # No ``on_session_delete`` here: this rolls back the row inserted in the
-                        # same transaction, so the session never existed — firing would double-count it.
                         conn.execute("DELETE FROM sessions WHERE id = ?", (session_id,))
                         return None, f"title:{exc}"
                 conn.execute("UPDATE sessions SET title = ? WHERE id = ?", (clean_title, session_id))
diff --git a/hermes_state_maintenance.py b/hermes_state_maintenance.py
index 345cd749e1df5..c5d699504c9ea 100644
--- a/hermes_state_maintenance.py
+++ b/hermes_state_maintenance.py
@@ -331,6 +331,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:
diff --git a/hermes_state_sessions.py b/hermes_state_sessions.py
index 7d70e8a6b6b28..8ba80c448f13f 100644
--- a/hermes_state_sessions.py
+++ b/hermes_state_sessions.py
@@ -14,7 +14,6 @@
 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,
@@ -1978,34 +1977,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

From 3d79bd726c53ca1be10bc3d7be1cc89ede617773 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Sun, 18 Oct 2026 01:03:55 -0700
Subject: [PATCH 6/6] fix(plugins): on_session_delete review findings

- Dispatch under the store's own profile home: `sessions repair-profiles`
  deletes from another profile's store, and the event now reaches that
  profile's plugins instead of the launch profile's.
- Archive events list every hidden chain row (deleted_ids), on both the
  SessionDB and the runtime-receipt path; the requested id stays primary.
- Fire right after the commit, before media/transcript cleanup that can
  raise, so a cleanup failure no longer swallows the event.
- mutate_runtime_session's write returns (result, SessionGoneEvent | None)
  instead of filling a shared list.
- Drop the leftover duplicate hook reference block in hooks.md.
---
 hermes_state_common.py                        | 31 ++++++++++++-------
 hermes_state_maintenance.py                   |  7 ++---
 hermes_state_runtime.py                       | 31 +++++++++++--------
 hermes_state_sessions.py                      | 19 ++++++------
 .../hermes_state/test_session_delete_hook.py  | 10 ++++--
 website/docs/user-guide/features/hooks.md     | 14 ---------
 6 files changed, 58 insertions(+), 54 deletions(-)

diff --git a/hermes_state_common.py b/hermes_state_common.py
index a45358749dd79..98b7a5a9e692f 100644
--- a/hermes_state_common.py
+++ b/hermes_state_common.py
@@ -395,29 +395,36 @@ def _id_chunks(ids, size: int = _SQL_IN_CHUNK):
 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 only after the write transaction commits, with the ids it actually removed (or, for
-    ``reason="archive"``, hid). *primary* is the session the caller asked for; without it the
-    first removed id stands in (bulk sweeps). ``deleted_ids`` rides along only when more than
-    one row went. Late-bound and fail-open: the state layer never imports ``hermes_cli`` at module
-    top level, and a missing or raising plugin stack never breaks the delete.
+    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 profile_name_for_home
+        from hermes_constants import (
+            get_hermes_home, profile_name_for_home, reset_hermes_home_override, set_hermes_home_override,
+        )
 
-        payload: dict = {
-            "session_id": primary if primary in ids else ids[0],
-            "reason": reason,
-            "profile": profile_name_for_home(os.path.dirname(os.fspath(db.db_path))),
-        }
+        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)
-        invoke_hook("on_session_delete", **payload)
+        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)
 
diff --git a/hermes_state_maintenance.py b/hermes_state_maintenance.py
index c5d699504c9ea..9ea68279ec7c3 100644
--- a/hermes_state_maintenance.py
+++ b/hermes_state_maintenance.py
@@ -132,12 +132,11 @@ 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 ():
             self._remove_session_files(sessions_dir, sid)
-        if removed_ids:
-            emit_session_deleted(self, removed_ids, reason="ghost_prune", sessions_dir=sessions_dir)
         return len(removed_ids)
 
     def _guarded_ids(self, conn, ids: Iterable[str]) -> set:
@@ -400,14 +399,14 @@ 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:
             self._remove_session_files(sessions_dir, sid)
         if deleted_ids is not None:
             deleted_ids.extend(removed_ids)
-        if count:
-            emit_session_deleted(self, removed_ids, reason="prune", sessions_dir=sessions_dir)
         return count
 
     def _page_pragmas(self, names: tuple[str, ...], fail_msg: str) -> Optional[list]:
diff --git a/hermes_state_runtime.py b/hermes_state_runtime.py
index 0ec37531785c0..ebc56f2e52606 100644
--- a/hermes_state_runtime.py
+++ b/hermes_state_runtime.py
@@ -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 (
@@ -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
@@ -449,15 +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)
 
-    # (ids, reason) for on_session_delete; reset per attempt (the write retries) and empty on replay.
-    gone: list = []
-
-    def write(conn):
-        gone.clear()
+    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')
@@ -479,12 +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':
-            gone[:] = [projection.get('deleted_ids') or [session_id], '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']:
-            gone[:] = [[session_id], 'archive']
+            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()
@@ -493,11 +498,11 @@ def write(conn):
                   'operation': operation, **projection}
         conn.execute('INSERT INTO state_meta(key,value) VALUES(?,?)',
                      (key, _json({'digest': digest, 'result': result})))
-        return result
-    result = db._execute_write(write)
-    if gone:
+        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, gone[0], reason=gone[1], primary=session_id)
+        emit_session_deleted(db, event.session_ids, reason=event.reason, primary=session_id)
     return result
 
 
diff --git a/hermes_state_sessions.py b/hermes_state_sessions.py
index 8ba80c448f13f..d3f28a7c0c3e4 100644
--- a/hermes_state_sessions.py
+++ b/hermes_state_sessions.py
@@ -947,7 +947,7 @@ def _do(conn):
             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, [session_id], reason="archive")
+            emit_session_deleted(self, ids, reason="archive", primary=session_id)
         return bool(ids)
 
     def _auto_archive_lineage(self, session_id: str) -> bool:
@@ -1761,12 +1761,12 @@ 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:
             self._remove_session_files(sessions_dir, sid)
-        if deleted:
-            emit_session_deleted(self, removed_ids, reason="delete", primary=session_id, sessions_dir=sessions_dir)
         return bool(deleted)
 
     def discard_unadmitted_session(self, session_id: str) -> bool:
@@ -1823,11 +1823,12 @@ 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:
             self._remove_session_files(sessions_dir, session_id)
-            emit_session_deleted(self, [session_id], reason="empty_sweep", sessions_dir=sessions_dir)
         return deleted
 
     def delete_sessions(
@@ -1911,13 +1912,13 @@ 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:
             self._remove_session_files(sessions_dir, sid)
-        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)
         return count
 
     # Shared by count_empty_sessions / delete_empty_sessions so badge and sweep agree. message_count
@@ -1956,12 +1957,12 @@ 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:
             self._remove_session_files(sessions_dir, sid)
-        if count:
-            emit_session_deleted(self, removed_ids, reason="empty_sweep", sessions_dir=sessions_dir)
         return count
 
     def archive_sessions(
diff --git a/tests/hermes_state/test_session_delete_hook.py b/tests/hermes_state/test_session_delete_hook.py
index e81582a330aeb..f9f805aa4cbf2 100644
--- a/tests/hermes_state/test_session_delete_hook.py
+++ b/tests/hermes_state/test_session_delete_hook.py
@@ -53,6 +53,12 @@ def test_every_delete_and_archive_path_fires_once_after_commit(db, hook):
             db.append_message(sid, role="user", content="hi")
     db.end_session("empty", "done")
     db.end_session("old", "done")
+    for tip in ("arch", "desk-arch"):  # compression chains: archiving the tip hides the root too
+        db.create_session(f"{tip}-root", source="cli")
+        db.append_message(f"{tip}-root", role="user", content="hi")
+        db.end_session(f"{tip}-root", "compression")
+        db._execute_write(lambda c, t=tip: c.execute(
+            "UPDATE sessions SET parent_session_id = ? WHERE id = ?", (f"{t}-root", t)))
 
     assert db.delete_session("cli")                          # hermes sessions delete, /exit --delete
     assert db.delete_session("missing") is False             # unknown id: silent
@@ -73,8 +79,8 @@ def test_every_delete_and_archive_path_fires_once_after_commit(db, hook):
         ("empty", "empty_sweep", None, None),
         ("old", "prune", None, None),
         ("desk", "delete", None, None),
-        ("arch", "archive", None, True),
-        ("desk-arch", "archive", None, True),
+        ("arch", "archive", ["arch", "arch-root"], True),
+        ("desk-arch", "archive", ["desk-arch", "desk-arch-root"], True),
     ]
     assert all(e["profile"] == events[0]["profile"] for e in events)
 
diff --git a/website/docs/user-guide/features/hooks.md b/website/docs/user-guide/features/hooks.md
index 19e8f2c8b0303..ed30ffec051a4 100644
--- a/website/docs/user-guide/features/hooks.md
+++ b/website/docs/user-guide/features/hooks.md
@@ -1108,20 +1108,6 @@ def my_callback(session_id: str, reason: str, profile: str | None, **kwargs):
 
 **Use cases:** Delete a per-session sidecar database or cache directory, close a linked task when the user archives its chat, emit "session deleted" analytics.
 
------------|------|-------------|
-| `session_id` | `str` | The session that was deleted. Always present. |
-| `deleted_ids` | `list[str]`, optional | Full set removed by the operation, including cascade-deleted delegate children and every row in a bulk delete. Present when the count differs from 1. |
-| `reason` | `str`, optional | Why the delete happened (`"prune"`, `"ghost_prune"`, `"empty_sweep"`, …). |
-| `sessions_dir` | `str`, optional | Absolute path of the profile's `sessions/` dir, so a plugin does not have to re-derive the per-profile home. |
-
-**Fires:** Post-commit only, and only when the delete actually happened (unknown ids in a bulk delete are skipped silently). Return ignored; a failing callback is logged and skipped like every other observer. The payload carries no message content — only ids, the reason, and the directory.
-
-**Delivery scope:** Process-local — the callback runs only in the process that committed the delete. It is not a durable cross-process delivery channel, and it carries no exactly-once guarantee: external side effects driven from the callback should tolerate a missed or duplicated invocation.
-
-**Return value:** Ignored.
-
-**Use cases:** Delete a per-session sidecar database or cache directory, emit "session deleted" analytics, release per-session external resources.
-
 ---
 
 See the **[Build a Plugin guide](../../developer-guide/plugins/index.md)** for the full walkthrough including tool schemas, handlers, and advanced hook patterns.