diff --git a/gateway/control_socket.py b/gateway/control_socket.py index 1845936afc..9a14f8c2e2 100644 --- a/gateway/control_socket.py +++ b/gateway/control_socket.py @@ -368,3 +368,13 @@ def migrate_gateway_profile_identity(home: Path, old_name: str, new_name: str, * and a restart reconciles the in-memory copy.""" return query_gateway_control(home, "migrate-profile-identity", params={"old": old_name, "new": new_name}, timeout=timeout) + + +def purge_gateway_profile_identity(home: Path, name: str, *, + timeout: float = 8.0) -> Optional[dict[str, Any]]: + """Ask the multiplexer serving ``home`` to drop a deleted profile's routing identity now — the + in-memory index AND the durable rows, neither of which a CLI-side delete can settle: this process + writes its in-memory copy back, so it re-creates what the CLI removed. Returns its + ``{"ok": True, "dropped": N, ...}`` answer, or None when no gateway answers / the gateway predates + the verb.""" + return query_gateway_control(home, "purge-profile-identity", params={"name": name}, timeout=timeout) diff --git a/gateway/run.py b/gateway/run.py index 067f084131..4182ae9cdd 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -5079,7 +5079,7 @@ async def _start_gateway_start_control_socket(runner): # failure only means consumers fall back to the process-scan/state-file layer, exactly as before # this feature. See #92091. from gateway.control_socket import GatewayControlServer - from gateway.run_profile_reconcile import migrate_profile_identity_verb + from gateway.run_profile_reconcile import migrate_profile_identity_verb, purge_profile_identity_verb # pause-for-update: the updater asks us to drain + exit (freeing venv handles) vs. a tree-kill # (same path as SIGUSR1). Handler runs on the socket executor thread, so marshal onto the loop. # pause-for-update (#92091 step 2): the updater asks this gateway to drain in-flight turns and exit @@ -5128,7 +5128,8 @@ async def _start_gateway_start_control_socket(runner): _control_server = GatewayControlServer( verb_handlers={"pause-for-update": _pause_for_update_handler, "rescan-profiles": _rescan_profiles_handler, - "migrate-profile-identity": migrate_profile_identity_verb(runner)}) + "migrate-profile-identity": migrate_profile_identity_verb(runner), + "purge-profile-identity": purge_profile_identity_verb(runner)}) if not await _control_server.start(): _control_server = None else: diff --git a/gateway/run_profile_reconcile.py b/gateway/run_profile_reconcile.py index c127770f2b..181fbfec06 100644 --- a/gateway/run_profile_reconcile.py +++ b/gateway/run_profile_reconcile.py @@ -304,3 +304,36 @@ def migrate_profile_identity_verb(runner): logger.debug("Failed to release renamed profile state DB", exc_info=True) return _handler + + +def purge_profile_identity_verb(runner): + """Build the ``purge-profile-identity`` control-verb handler for ``hermes profile delete`` + (#111926, delete side). The live multiplexer owns the routing index in memory and writes it back + periodically, so a CLI-side DELETE of ``agent::*`` rows would be undone by its next save; + the CLI therefore asks this process to drop the durable rows AND ``SessionStore._entries``. + + Deliberately NOT part of ``_unserve_profile()``: that path also unserves names that are still + alive elsewhere in the identity story — a rename's old name leaves the served set exactly like a + delete does (its directory is gone either way) — and purging there would race the rekey it is + supposed to leave intact. Only the delete path invokes this verb. Runs on the control-socket + executor thread; ``purge_profile_routing`` takes the store lock.""" + + def _handler(params: dict) -> dict: + name = str(params.get("name") or "").strip() + if not name: + return {"ok": False, "error": "name required"} + store = getattr(runner, "session_store", None) + if store is None: + return {"ok": False, "error": "live gateway has no session store"} + try: + db_counts: Dict[str, Dict[str, int]] = {} + routing_db = getattr(store, "_routing_db", None) + if routing_db is not None and hasattr(routing_db, "purge_profile_state"): + db_counts["routing"] = routing_db.purge_profile_state(name) + dropped = store.purge_profile_routing(name) + return {"ok": True, "dropped": dropped, "db": db_counts} + except Exception as exc: + logger.warning("Profile identity purge failed for %r: %s", name, exc) + return {"ok": False, "error": f"{type(exc).__name__}: {exc}"} + + return _handler diff --git a/gateway/session.py b/gateway/session.py index 59cda21e99..af2e373d39 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -1148,6 +1148,26 @@ class SessionStore( self._save() return len(moving) + def purge_profile_routing(self, profile: str) -> int: + """Drop a deleted profile's live routing entries and persist the drop (#111926, delete side). + + The mirror of :meth:`rekey_profile_routing`, and it has to happen here for the same reason: + this index is written back by the owning process, so a durable DB delete made elsewhere is + undone by the next save of this in-memory copy — which is how a deleted profile kept + resolving. Idempotent; returns the number of entries dropped. + """ + name = (profile or "").strip() + if not name: + return 0 + ns = f"agent:{name}:" + with self._lock: + dropped = [key for key in self._entries if key.startswith(ns)] + for key in dropped: + self._entries.pop(key, None) + if dropped: + self._save() + return len(dropped) + # Compression repoint is store bookkeeping, not user activity — leave ``updated_at`` alone so a # background compression on an idle session cannot make it look fresh to the # restart-resume freshness gate (#85709). diff --git a/hermes_cli/profile_cmd.py b/hermes_cli/profile_cmd.py index f20165a35f..6e6058031e 100644 --- a/hermes_cli/profile_cmd.py +++ b/hermes_cli/profile_cmd.py @@ -286,7 +286,7 @@ def _profile_delete(args): from hermes_cli.profiles import delete_profile try: delete_profile(args.profile_name, yes=getattr(args, "yes", False)) - except (ValueError, FileNotFoundError) as e: + except (ValueError, FileNotFoundError, RuntimeError) as e: _die(f"Error: {e}") @@ -451,6 +451,21 @@ def _profile_migrate_identity(args): print(f"✓ Session/routing identity migrated: {args.old_name} → {args.new_name}") +def _profile_purge_identity(args): + """Retry the identity purge of a delete that already completed. Exits non-zero when a live + gateway would not purge (it still owns the routing index in memory), or when a database rejected + the delete (lock, partial failure).""" + from hermes_cli.profile_identity import purge_profile_identity + try: + purged = purge_profile_identity(args.profile_name) + except ValueError as e: + _die(f"Error: {e}") + if not purged: + _die(f"Error: session identity was not purged. Restart or stop the gateway, then run:\n" + f" hermes profile purge-identity {args.profile_name}", err=True) + print(f"✓ Session/routing identity purged: {args.profile_name}") + + def _profile_export(args): from hermes_cli.profiles import export_profile, get_profile_export_path name = args.profile_name @@ -590,6 +605,7 @@ PROFILE_ACTIONS = { 'show': _profile_show, 'alias': _profile_alias, 'rename': _profile_rename, + 'purge-identity': _profile_purge_identity, 'migrate-identity': _profile_migrate_identity, 'export': _profile_export, 'import': _profile_import, diff --git a/hermes_cli/profile_identity.py b/hermes_cli/profile_identity.py index b70e81ee17..5b4c536785 100644 --- a/hermes_cli/profile_identity.py +++ b/hermes_cli/profile_identity.py @@ -1,4 +1,4 @@ -"""Rekey a renamed profile's session/routing identity (#111926). +"""Rekey a renamed profile's session/routing identity (#111926), and purge a deleted profile's. ``rename_profile`` moves ``profiles//`` to ``profiles//`` so row DATA travels with the directory, but the profile name is also baked into keys and values the move never touches: @@ -7,9 +7,15 @@ directory, but the profile name is also baked into keys and values the move neve every inbound event on a chat keyed to the old name logs ``Profile 'old' does not exist`` and falls back to the global home, and renamed sessions drop out of the Desktop sidebar. +The same identity has to be *purged* when a profile is deleted — the mirror image of the rekey, +and ``delete_profile`` calls :func:`purge_profile_identity` for it. + Ownership decides who rewrites: a live multiplexer holds the routing index in memory (``SessionStore._entries``) and writes it back periodically, so a CLI-side DB rewrite would be -clobbered on its next save — the CLI delegates to the ``migrate-profile-identity`` control verb. +clobbered on its next save — the CLI delegates the rename to the ``migrate-profile-identity`` +control verb and the delete to the delete-only ``purge-profile-identity`` verb. Neither runs inside +``_unserve_profile()``: that hook fires for every name leaving the served set, and a rename's old +name leaves it exactly like a deleted one, so identity must survive it for the rekey that follows. With no live multiplexer nothing else holds the store and the durable rewrite is safe here. """ from __future__ import annotations @@ -43,6 +49,93 @@ def migrate_profile_identity(old_name: str, new_name: str) -> bool: return _migrate_profile_identity(old_canon, new_canon, _live_default_multiplexer()) +def purge_profile_identity(profile: str) -> bool: + """Purge a deleted profile's session/routing identity from the durable stores (#111926). + + The mirror of :func:`migrate_profile_identity`: a rename must rekey a profile's identity, a + delete must purge it. ``agent::*`` routing keys, ``gateway_heartbeats.profile`` and + ``delivery_obligations`` rows are bookkeeping for a profile that no longer exists; left behind, + every inbound event on a chat keyed to the deleted name enters the routing index, resolves a + profile whose directory is gone, and logs ``Profile '' does not exist`` on each event. + + A live multiplexer holds that index in memory and writes it back periodically, so it owns the + purge exactly as it owns the rename migration: the delete path asks it through the + ``purge-profile-identity`` control verb and treats the answer as the settlement — a gateway that + times out, predates the verb, or fails the purge is reported with a retry command instead of + being silently assumed. With no live multiplexer nothing else holds the store and the durable + delete happens here. Session history rows are not part of this purge — it settles identity only. + Refuses a name that is a live profile again: the purge keys off the name alone, so a same-name + profile created after the delete (the very case a failed settlement leaves behind) would + otherwise have the new incarnation's routing/heartbeat identity deleted from under it. The + delete path tombstones the directory before calling this, so the guard never blocks it. + Idempotent: purging an already-purged profile succeeds. Returns False only when the identity was + not settled — by this process or by the live gateway. + """ + from hermes_cli.profiles import _canon_valid, _live_default_multiplexer, profile_exists + canon = _canon_valid(profile) + if canon == "default": + raise ValueError("Identity purge applies to named profiles only.") + if profile_exists(canon): + raise ValueError( + f"Profile '{canon}' exists; purge-identity only settles the identity of a delete that " + "has already completed.") + return _purge_profile_identity(canon, _live_default_multiplexer()) + + +def _purge_profile_identity(canon: str, live_mux: bool) -> bool: + """Purge deleted-profile identity without racing a live gateway's in-memory routing index. + + Returns True when the identity was purged — by the gateway's control verb, or by this process's + durable delete when no gateway holds the store — and False when a live gateway did not accept + it. Never fatal to the delete, which has already happened by this point. + """ + if live_mux: + from hermes_constants import get_default_hermes_root + root = get_default_hermes_root() + try: + from gateway.control_socket import purge_gateway_profile_identity + answer = purge_gateway_profile_identity(root, canon) + except Exception as exc: + reason = f"{type(exc).__name__}: {exc}" + else: + if isinstance(answer, dict) and answer.get("ok") is True: + return True + reason = _control_answer_failure(answer) + if answer is None and _gateway_accepts_profile_identity_verb(root): + reason += (" — the gateway is running but does not implement " + "'purge-profile-identity' (an older process than this CLI)") + print( + "⚠ Profile was deleted, but the live gateway could not purge its session identity" + f" ({reason}). Restart the gateway, then run:\n" + f" hermes profile purge-identity {canon}", + file=sys.stderr) + return False + + from hermes_cli.profiles import get_profile_dir + from hermes_constants import get_default_hermes_root + from hermes_state_registry import acquire, release_or_close + root = get_default_hermes_root() + purged = True + for db_path in (root / "state.db", get_profile_dir(canon) / "state.db"): + if not db_path.exists(): + continue + db = None + try: + db = acquire(db_path) + db.purge_profile_state(canon) + except Exception as exc: + purged = False + print( + f"⚠ Profile was deleted, but identity purge failed for {db_path}: " + f"{type(exc).__name__}: {exc}", + file=sys.stderr) + finally: + if db is not None: + with contextlib.suppress(Exception): + release_or_close(db) + return purged + + def _control_answer_failure(answer) -> str: """Why a control-socket answer is not a success. Keeps the raw answer when the payload carries no reason field, so a malformed or old-gateway response stays diagnosable instead of diff --git a/hermes_cli/profiles.py b/hermes_cli/profiles.py index a3ce6412c4..2361888988 100644 --- a/hermes_cli/profiles.py +++ b/hermes_cli/profiles.py @@ -1026,6 +1026,15 @@ def _notify_multiplexer(canon: str) -> None: notify_multiplexer_profiles_changed(canon) +def _purge_identity(canon: str) -> bool: + """Settle a deleted profile's durable identity (``profile_identity.purge_profile_identity``). + + False means the filesystem delete happened but the identity settlement did not — the caller + reports that as a pending settlement rather than a clean success.""" + from hermes_cli.profile_identity import purge_profile_identity + return purge_profile_identity(canon) + + def _live_default_multiplexer() -> bool: """True when a live default gateway has recorded a served-profile set: every dir under profiles/ is then served by it, so a profile-identity change must be unrouted first.""" @@ -1319,8 +1328,11 @@ def delete_profile(name: str, yes: bool = False) -> Path: # Tombstone before rmtree so a stale serve/logging mkdir cannot relist this name live. mark_named_profile_deleted(profile_dir) # The multiplexer sees the tombstone, stops this profile's adapters and releases its handles - # into the directory before we remove it. + # into the directory before we remove it. Identity settlement is a separate delete-only + # operation below: an ordinary unserve must preserve identity, because a rename's old name + # leaves the served set exactly like a deleted one does (#111926, delete side). _notify_multiplexer(canon) + identity_settled = _purge_identity(canon) # The main serve process survives this deletion. Stop only this profile's MCP # transports and release cached stderr handles, including completed probes. @@ -1362,6 +1374,13 @@ def delete_profile(name: str, yes: bool = False) -> Path: if remove_error is not None: raise RuntimeError(f"Could not remove profile directory {profile_dir}: {remove_error}") from remove_error print(f"\nProfile '{canon}' deleted.") + if not identity_settled: + # Filesystem work and runtime teardown are done; the durable identity is not. Report the + # partial settlement as a failure (same shape as the rmtree error above) instead of a clean + # success, and name the retry. + raise RuntimeError( + f"Profile '{canon}' was deleted, but its session/routing identity settlement is still " + f"pending — run: hermes profile purge-identity {canon}") return profile_dir diff --git a/hermes_cli/subcommands/profile.py b/hermes_cli/subcommands/profile.py index e07405a0e2..a67300f912 100644 --- a/hermes_cli/subcommands/profile.py +++ b/hermes_cli/subcommands/profile.py @@ -90,6 +90,16 @@ def build_profile_parser(subparsers, *, cmd_profile: Callable) -> None: "new_name", help="New profile name (for 'default': a display name — the canonical id stays 'default')") + profile_purge = profile_subparsers.add_parser( + "purge-identity", + help="Retry a deleted profile's session/routing identity purge", + description="Re-run the session/routing identity purge that `hermes profile delete` performs " + "automatically. The profile directory is already gone when this is needed: state still " + "keyed by the deleted profile name (routing keys, heartbeats, routing/delivery rows) is " + "deleted. Run it after restarting the gateway (which reloads the routing index from the " + "DB) or after stopping it. Idempotent.") + profile_purge.add_argument("profile_name", help="Deleted profile name") + profile_migrate = profile_subparsers.add_parser( "migrate-identity", help="Retry a renamed profile's session/routing identity migration", diff --git a/hermes_state_gateway.py b/hermes_state_gateway.py index e12600b52f..40f75bedf4 100644 --- a/hermes_state_gateway.py +++ b/hermes_state_gateway.py @@ -653,6 +653,69 @@ class SessionGatewayMixin: self._execute_write(_do) return counts + def purge_profile_state(self, profile: str) -> Dict[str, int]: + """Delete exact profile identity from this state database (#111926, delete side). + + The mirror of :meth:`rekey_profile_state`: a rename must rekey a profile's identity, a + delete must purge it. ``agent::*`` routing keys, ``gateway_heartbeats.profile``, + ``delivery_obligations`` and the telegram topic tables are bookkeeping for a profile that + no longer exists — left behind, every inbound event on a chat keyed to the deleted name + enters the routing index, resolves a profile whose directory is gone, and logs + ``Profile '' does not exist`` on each event for the life of the store. + + What each store gets, and why: + + * Routing keys and heartbeat rows are hard-deleted — pure bookkeeping for a dead name. + * ``delivery_obligations`` rows are **terminalized** (``state='abandoned'``), not deleted: + a pending obligation is delivery state that should not vanish silently, and the ledger's + own retention prunes abandoned rows. Delivered history is left as it was. + * ``sessions`` rows are not deleted here: this helper settles identity, not history, and it + does not decide what a delete leaves of a profile's conversation record — ``delete_profile`` + removes the profile's own home, ``state.db`` included, with the directory. Rows in a shared + store keep whatever ownership they had; re-binding or archiving them belongs to the flow + that recreates the name, not to this purge. + + Idempotent. + """ + name = (profile or "").strip() + counts: Dict[str, int] = {} + if not name: + return counts + ns, ns_len = f"agent:{name}:", len(f"agent:{name}:") + + def _do(conn): + existing = {row[0] for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'").fetchall()} + if "gateway_routing" in existing: + counts["gateway_routing"] = conn.execute( + "DELETE FROM gateway_routing WHERE substr(session_key, 1, ?) = ?", + (ns_len, ns)).rowcount + if "gateway_heartbeats" in existing: + counts["gateway_heartbeats"] = conn.execute( + "DELETE FROM gateway_heartbeats WHERE profile = ?", (name,)).rowcount + if "delivery_obligations" in existing: + # Terminalize, never hard-delete: a pending obligation is delivery state someone may + # still care about, and the ledger's own retention prunes abandoned rows. Only + # non-terminal rows are touched — delivered history is left exactly as it was. + counts["delivery_obligations"] = conn.execute( + "UPDATE delivery_obligations SET state='abandoned', updated_at=? " + "WHERE (adapter_profile = ? OR substr(session_key, 1, ?) = ?) " + "AND state NOT IN ('delivered', 'abandoned')", + (time.time(), name, ns_len, ns)).rowcount + if "telegram_dm_topic_mode" in existing: + counts["telegram_dm_topic_mode"] = conn.execute( + "DELETE FROM telegram_dm_topic_mode WHERE profile_name = ?", (name,)).rowcount + if "telegram_dm_topic_bindings" in existing: + # A rename rewrites a binding's session_key namespace as well as its profile_name + # (:meth:`rekey_profile_state`), so matching on one alone leaves the other behind. + counts["telegram_dm_topic_bindings"] = conn.execute( + "DELETE FROM telegram_dm_topic_bindings " + "WHERE profile_name = ? OR substr(session_key, 1, ?) = ?", + (name, ns_len, ns)).rowcount + + self._execute_write(_do) + return counts + @staticmethod def session_gateway_runtime(session_meta: Optional[Dict[str, Any]]) -> Dict[str, Any]: """Read the persisted runtime route off a session row dict (``model_config`` as diff --git a/tests/gateway/test_profile_identity_purge.py b/tests/gateway/test_profile_identity_purge.py new file mode 100644 index 0000000000..5acb6c9955 --- /dev/null +++ b/tests/gateway/test_profile_identity_purge.py @@ -0,0 +1,131 @@ +"""Profile identity purge for `hermes profile delete`: the delete-only path, and what an unserve +must leave alone. + +`_unserve_profile()` runs for every name that leaves the served set — a rename's old name leaves it +exactly like a deleted one (its directory is gone either way) and the rename's rekey still needs that +identity — so the purge lives behind the delete-only ``purge-profile-identity`` control verb and +never in the unserve path (#111926, delete side). +""" + +from __future__ import annotations + +import json +import time +from pathlib import Path +from types import SimpleNamespace + +import pytest + + +def _make_store(tmp_path): + from gateway.config import GatewayConfig + from gateway.session import SessionStore + sessions_dir = tmp_path / "sessions" + sessions_dir.mkdir(exist_ok=True) + store = SessionStore( + sessions_dir, + GatewayConfig(sessions_dir=sessions_dir, write_sessions_json=False, + multiplex_profiles=True), + ) + store._ensure_loaded() + return store + + +def _entry(session_key, chat_id, profile): + from gateway.session import SessionEntry, SessionSource, Platform + from gateway.session_lifecycle import _now + now = _now() + return SessionEntry( + session_key=session_key, session_id=f"sid-{chat_id}", + platform=Platform.FEISHU, chat_type="dm", created_at=now, updated_at=now, + origin=SessionSource(platform=Platform.FEISHU, chat_id=chat_id, profile=profile), + ) + + +def _state_db(): + from hermes_constants import get_hermes_home + from hermes_state import SessionDB + return SessionDB(Path(get_hermes_home()) / "state.db") + + +def _runner_stub(store): + return SimpleNamespace( + session_store=store, + _profile_failed_platforms={}, + _profile_adapters={}, + pairing_stores={}, + _busy_text_modes_by_profile={}, + _busy_input_modes_by_profile={}, + _served_profile_homes={}, + _served_profile_signatures={}, + _agent_cache={}, + _evict_cached_agent=lambda key: None, + ) + + +@pytest.mark.asyncio +async def test_unserve_profile_keeps_identity_for_a_rename_to_rekey(tmp_path): + """Unserving is not deleting: the purge must not run here. + + A rename's old name leaves the served set exactly like a deleted one, and + ``migrate-profile-identity`` still has to find that identity to rekey it. A purge inside + ``_unserve_profile()`` would delete it first and break the rename it runs beside. + """ + from gateway.run_profile_reconcile import GatewayProfileReconcileMixin + store = _make_store(tmp_path) + with store._lock: + store._entries["agent:oldname:feishu:dm:chatA"] = _entry( + "agent:oldname:feishu:dm:chatA", "chatA", "oldname") + db = _state_db() + db.register_backend_heartbeat( + backend_id="be1", pid=1, started_at=time.time(), profile="oldname", host="h") + db.close() + + home = tmp_path / "home" + home.mkdir() + await GatewayProfileReconcileMixin._unserve_profile(_runner_stub(store), "oldname", home) + + assert "agent:oldname:feishu:dm:chatA" in store._entries + db = _state_db() + try: + assert db._read_one( + "SELECT COUNT(*) AS n FROM gateway_heartbeats WHERE profile = ?", + ("oldname",))["n"] == 1 + finally: + db.close() + + +def test_purge_verb_drops_routing_identity_and_reports_ok(tmp_path): + """The delete-only verb settles identity in the process that owns the routing index.""" + from gateway.run_profile_reconcile import purge_profile_identity_verb + store = _make_store(tmp_path) + with store._lock: + store._entries["agent:gone:feishu:dm:chatA"] = _entry( + "agent:gone:feishu:dm:chatA", "chatA", "gone") + store._entries["agent:keepme:feishu:dm:chatB"] = _entry( + "agent:keepme:feishu:dm:chatB", "chatB", "keepme") + db = _state_db() + db.register_backend_heartbeat( + backend_id="be1", pid=1, started_at=time.time(), profile="gone", host="h") + db.close() + + answer = purge_profile_identity_verb(_runner_stub(store))({"name": "gone"}) + + assert answer["ok"] is True + assert answer["dropped"] == 1 + assert "agent:gone:feishu:dm:chatA" not in store._entries + assert "agent:keepme:feishu:dm:chatB" in store._entries + db = _state_db() + try: + assert db._read_one( + "SELECT COUNT(*) AS n FROM gateway_heartbeats WHERE profile = ?", ("gone",))["n"] == 0 + finally: + db.close() + + +def test_purge_verb_refuses_without_a_name_or_store(): + """A malformed request or a gateway with no store answers failure — never a silent success.""" + from gateway.run_profile_reconcile import purge_profile_identity_verb + no_store = purge_profile_identity_verb(SimpleNamespace(session_store=None)) + assert no_store({})["ok"] is False + assert no_store({"name": "gone"})["ok"] is False diff --git a/tests/gateway/test_purge_profile_routing.py b/tests/gateway/test_purge_profile_routing.py new file mode 100644 index 0000000000..35106cc22f --- /dev/null +++ b/tests/gateway/test_purge_profile_routing.py @@ -0,0 +1,69 @@ +"""In-memory routing purge for `hermes profile delete`. + +The routing index lives in ``SessionStore._entries`` and is written back periodically, so a durable +DB delete made anywhere else is undone by this process's next save — the store has to drop its own +copy, and the drop has to be persisted, or a deleted profile's chats keep resolving to it. This is +the delete-side sibling of the rename rekey in #111926. +""" + +from __future__ import annotations + + +def _make_store(tmp_path): + from gateway.config import GatewayConfig + from gateway.session import SessionStore + sessions_dir = tmp_path / "sessions" + sessions_dir.mkdir(exist_ok=True) + store = SessionStore( + sessions_dir, + GatewayConfig(sessions_dir=sessions_dir, write_sessions_json=False, + multiplex_profiles=True), + ) + store._ensure_loaded() + return store + + +def _entry(session_key, chat_id, profile): + from gateway.session import SessionEntry, SessionSource, Platform + from gateway.session_lifecycle import _now + now = _now() + return SessionEntry( + session_key=session_key, session_id=f"sid-{chat_id}", + platform=Platform.FEISHU, chat_type="dm", created_at=now, updated_at=now, + origin=SessionSource(platform=Platform.FEISHU, chat_id=chat_id, profile=profile), + ) + + +def test_purges_deleted_namespace_and_persists_the_drop(tmp_path): + store = _make_store(tmp_path) + with store._lock: + store._entries["agent:gone:feishu:dm:chatA"] = _entry( + "agent:gone:feishu:dm:chatA", "chatA", "gone") + store._entries["agent:keepme:feishu:dm:chatB"] = _entry( + "agent:keepme:feishu:dm:chatB", "chatB", "keepme") + store._save() + + dropped = store.purge_profile_routing("gone") + + assert dropped == 1 + assert "agent:gone:feishu:dm:chatA" not in store._entries + assert "agent:keepme:feishu:dm:chatB" in store._entries + # The drop is durable: a store reloading the same home must not resurrect the deleted profile. + reloaded = _make_store(tmp_path) + assert "agent:gone:feishu:dm:chatA" not in reloaded._entries + assert "agent:keepme:feishu:dm:chatB" in reloaded._entries + + +def test_namespace_scoped_and_idempotent(tmp_path): + store = _make_store(tmp_path) + with store._lock: + store._entries["agent:foo_bar:feishu:dm:chatA"] = _entry( + "agent:foo_bar:feishu:dm:chatA", "chatA", "foo_bar") + store._entries["agent:fooXbar:feishu:dm:chatB"] = _entry( + "agent:fooXbar:feishu:dm:chatB", "chatB", "fooXbar") + + assert store.purge_profile_routing("foo_bar") == 1 + assert store.purge_profile_routing("foo_bar") == 0 + + assert "agent:foo_bar:feishu:dm:chatA" not in store._entries + assert "agent:fooXbar:feishu:dm:chatB" in store._entries diff --git a/tests/hermes_cli/test_profile_identity_purge_cmd.py b/tests/hermes_cli/test_profile_identity_purge_cmd.py new file mode 100644 index 0000000000..ff7a3791e5 --- /dev/null +++ b/tests/hermes_cli/test_profile_identity_purge_cmd.py @@ -0,0 +1,94 @@ +"""The `hermes profile purge-identity` retry path: dispatched, and honest about failure. + +`hermes profile delete` reports a pending identity settlement when it cannot purge and names this +command as the retry — so the command must actually reach a handler, and must fail loudly when the +settlement still cannot be made. Regression for the delete side of #111926. +""" +import argparse +from argparse import Namespace +from pathlib import Path + +import pytest + +from hermes_cli import profile_cmd, profiles + + +@pytest.fixture() +def profile_env(tmp_path, monkeypatch): + monkeypatch.setattr(Path, "home", lambda: tmp_path) + default_home = tmp_path / ".hermes" + default_home.mkdir(exist_ok=True) + monkeypatch.setenv("HERMES_HOME", str(default_home)) + return tmp_path + + +def test_every_profile_subcommand_has_a_dispatch_entry(): + """A subcommand that parses but has no entry in the dispatch table silently does nothing.""" + from hermes_cli.subcommands.profile import build_profile_parser + top = argparse.ArgumentParser() + subparsers = top.add_subparsers(dest="command") + build_profile_parser(subparsers, cmd_profile=lambda args: None) + profile_parser = subparsers.choices["profile"] + groups = [a for a in profile_parser._actions if isinstance(a, argparse._SubParsersAction)] + assert groups, "the profile parser exposes subcommands" + assert set(groups[0].choices) == set(profile_cmd.PROFILE_ACTIONS) - {None} + + +def test_purge_identity_reports_success(profile_env, monkeypatch, capsys): + monkeypatch.setattr("hermes_cli.profile_identity.purge_profile_identity", lambda name: True) + + profile_cmd.cmd_profile(Namespace(profile_action="purge-identity", profile_name="gone")) + + assert "identity purged: gone" in capsys.readouterr().out + + +def test_purge_identity_exits_nonzero_when_settlement_stays_pending( + profile_env, monkeypatch, capsys): + monkeypatch.setattr("hermes_cli.profile_identity.purge_profile_identity", lambda name: False) + + with pytest.raises(SystemExit) as exc: + profile_cmd.cmd_profile(Namespace(profile_action="purge-identity", profile_name="gone")) + + assert exc.value.code not in (0, None) + assert "hermes profile purge-identity gone" in capsys.readouterr().err + + +def test_purge_identity_rejects_the_default_profile(profile_env, capsys): + with pytest.raises(SystemExit): + profile_cmd.cmd_profile(Namespace(profile_action="purge-identity", profile_name="default")) + assert "named profiles only" in capsys.readouterr().out + + +def test_purge_identity_refuses_a_same_name_profile_created_after_the_delete(profile_env, capsys): + """The retry must not purge identity out from under a profile that exists again. + + The purge keys off the name alone, so the recovery flow — ``delete foo`` (settlement pending), + ``create foo``, ``purge-identity foo`` — would delete the NEW incarnation's routing/heartbeat + identity. The delete path tombstones the directory before it purges, so this refusal cannot + block the delete it belongs to (``TestDeleteProfile`` covers that path). + """ + import json + + from hermes_cli.profiles import create_profile + from hermes_state import SessionDB + + create_profile("gone", no_alias=True) + scope = str(profile_env / ".hermes" / "sessions") + db = SessionDB(profile_env / ".hermes" / "state.db") + db.save_gateway_routing_entry( + "agent:gone:feishu:dm:chatA", + json.dumps({"session_key": "agent:gone:feishu:dm:chatA"}), scope=scope) + db.close() + + with pytest.raises(SystemExit) as exc: + profile_cmd.cmd_profile(Namespace(profile_action="purge-identity", profile_name="gone")) + + assert exc.value.code not in (0, None) + assert "purge-identity only settles the identity of a delete" in capsys.readouterr().out + check = SessionDB(profile_env / ".hermes" / "state.db") + try: + # The recreated profile still owns its routing key. + assert set(check.load_gateway_routing_entries(scope=scope)) == { + "agent:gone:feishu:dm:chatA"} + finally: + check.close() diff --git a/tests/hermes_cli/test_profiles.py b/tests/hermes_cli/test_profiles.py index b2ba3f30f3..664f916149 100644 --- a/tests/hermes_cli/test_profiles.py +++ b/tests/hermes_cli/test_profiles.py @@ -388,7 +388,90 @@ class TestDeleteProfile: assert profile_dir.is_dir() assert get_active_profile() == "default" + def test_delete_purges_profile_keyed_identity(self, profile_env): + """A deleted profile must not keep routing/heartbeat/delivery identity (#111926, delete side). + The name is baked into ``agent::*`` routing keys, ``gateway_heartbeats.profile`` and + ``delivery_obligations``. Left behind, a later event on a chat keyed to the deleted name + enters the routing index, resolves a profile whose directory is gone, and logs + ``Profile '' does not exist`` on every subsequent event. + """ + from hermes_state import SessionDB + import time + + tmp_path = profile_env + create_profile("gone", no_alias=True) + create_profile("keepme", no_alias=True) + scope = str(tmp_path / ".hermes" / "sessions") + db = SessionDB(tmp_path / ".hermes" / "state.db") + db.save_gateway_routing_entry( + "agent:gone:feishu:dm:chatA", + json.dumps({"session_key": "agent:gone:feishu:dm:chatA", + "origin": {"platform": "feishu", "chat_id": "chatA", + "profile": "gone"}}), + scope=scope) + db.save_gateway_routing_entry( + "agent:keepme:feishu:dm:chatB", + json.dumps({"session_key": "agent:keepme:feishu:dm:chatB", + "origin": {"platform": "feishu", "chat_id": "chatB", + "profile": "keepme"}}), + scope=scope) + db.register_backend_heartbeat( + backend_id="be-gone", pid=1, started_at=time.time(), profile="gone", host="h") + db.register_backend_heartbeat( + backend_id="be-keep", pid=2, started_at=time.time(), profile="keepme", host="h") + db.close() + + # No live multiplexer: nothing else owns the store, so this process purges the durable rows. + with patch("hermes_cli.profiles._cleanup_gateway_service"), \ + patch("hermes_cli.profiles._live_default_multiplexer", return_value=False): + delete_profile("gone", yes=True) + + check = SessionDB(tmp_path / ".hermes" / "state.db") + try: + assert set(check.load_gateway_routing_entries(scope=scope)) == { + "agent:keepme:feishu:dm:chatB"} + assert check._read_one( + "SELECT COUNT(*) AS n FROM gateway_heartbeats WHERE profile = ?", + ("gone",))["n"] == 0 + assert check._read_one( + "SELECT COUNT(*) AS n FROM gateway_heartbeats WHERE profile = ?", + ("keepme",))["n"] == 1 + finally: + check.close() + + def test_delete_reports_pending_settlement_for_a_live_multiplexer(self, profile_env, capsys): + """With a live multiplexer the owner process purges, so the CLI must not race it (#111926). + + That process holds the routing index in memory and writes it back, so a CLI-side DELETE + would be undone by its next save. When it cannot be reached the delete is NOT a clean + success: the identity settlement is reported as pending, with the retry named. + """ + from hermes_state import SessionDB + + tmp_path = profile_env + create_profile("gone", no_alias=True) + scope = str(tmp_path / ".hermes" / "sessions") + db = SessionDB(tmp_path / ".hermes" / "state.db") + db.save_gateway_routing_entry( + "agent:gone:feishu:dm:chatA", + json.dumps({"session_key": "agent:gone:feishu:dm:chatA"}), + scope=scope) + db.close() + + with patch("hermes_cli.profiles._cleanup_gateway_service"), \ + patch("hermes_cli.profiles._live_default_multiplexer", return_value=True): + with pytest.raises(RuntimeError, match="identity settlement is still pending"): + delete_profile("gone", yes=True) + + assert "hermes profile purge-identity gone" in capsys.readouterr().err + check = SessionDB(tmp_path / ".hermes" / "state.db") + try: + # The CLI left the identity alone rather than racing the live owner. + assert set(check.load_gateway_routing_entries(scope=scope)) == { + "agent:gone:feishu:dm:chatA"} + finally: + check.close() def test_backend_scan_only_matches_this_profile(self, profile_env, monkeypatch): """The backend PID scan binds by --profile selector and skips self.""" diff --git a/tests/hermes_state/test_purge_profile_state.py b/tests/hermes_state/test_purge_profile_state.py new file mode 100644 index 0000000000..2c4b1c2d57 --- /dev/null +++ b/tests/hermes_state/test_purge_profile_state.py @@ -0,0 +1,159 @@ +"""Regression tests for profile-keyed state purge on `hermes profile delete`. + +`delete_profile` removes the profile directory and the multiplexer tears its runtime down, but the +profile name is also baked into session keys (``agent::*``), ``gateway_heartbeats.profile``, +``delivery_obligations`` and the ``gateway_routing`` index. Left stale, an inbound event on a chat +keyed to the deleted name resolves to a profile that no longer exists and floods errors.log — the +delete-side sibling of the rename rekey in #111926. +""" +import json +import time + +import pytest + +from hermes_state import SessionDB + + +@pytest.fixture +def db(tmp_path): + database = SessionDB(tmp_path / "state.db") + yield database + database.close() + + +def _write_obligation(db, monkeypatch, obligation_id, session_key, chat_id, profile, + state="pending"): + """delivery_obligations is created lazily by the delivery ledger against the same state.db.""" + monkeypatch.setenv("HERMES_HOME", str(db.db_path.parent)) + from gateway import delivery_ledger + monkeypatch.setattr(delivery_ledger, "_db_path", lambda: db.db_path) + with delivery_ledger._connect() as conn: + now = time.time() + conn.execute( + "INSERT INTO delivery_obligations (obligation_id, session_key, platform, chat_id, " + "content, state, created_at, updated_at, adapter_profile) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + (obligation_id, session_key, "feishu", chat_id, "hi", state, now, now, profile), + ) + conn.commit() + + +class TestPurgeProfileState: + def test_purges_routing_and_heartbeats_but_keeps_bystanders(self, db): + db.save_gateway_routing_entry( + "agent:gone:feishu:dm:chatA", json.dumps({"session_key": "agent:gone:feishu:dm:chatA"}), + scope="/root/sessions") + db.save_gateway_routing_entry( + "agent:keepme:feishu:dm:chatB", + json.dumps({"session_key": "agent:keepme:feishu:dm:chatB"}), scope="/root/sessions") + db.register_backend_heartbeat( + backend_id="be-gone", pid=1, started_at=time.time(), profile="gone", host="h") + db.register_backend_heartbeat( + backend_id="be-keep", pid=2, started_at=time.time(), profile="keepme", host="h") + + counts = db.purge_profile_state("gone") + + assert counts["gateway_routing"] == 1 + assert counts["gateway_heartbeats"] == 1 + routing = db.load_gateway_routing_entries(scope="/root/sessions") + assert "agent:gone:feishu:dm:chatA" not in routing + assert "agent:keepme:feishu:dm:chatB" in routing + assert db._read_one( + "SELECT profile FROM gateway_heartbeats WHERE backend_id = ?", + ("be-keep",))["profile"] == "keepme" + assert db._read_one( + "SELECT COUNT(*) AS n FROM gateway_heartbeats WHERE profile = ?", ("gone",))["n"] == 0 + + def test_abandons_pending_deliveries_instead_of_deleting_them(self, db, monkeypatch): + """A pending obligation is delivery state, not identity: terminalize it, never drop it.""" + _write_obligation(db, monkeypatch, "ob-gone", "agent:gone:feishu:dm:chatA", "chatA", "gone") + _write_obligation( + db, monkeypatch, "ob-gone-done", "agent:gone:feishu:dm:chatB", "chatB", "gone", + state="delivered") + _write_obligation( + db, monkeypatch, "ob-keep", "agent:keepme:feishu:dm:chatC", "chatC", "keepme") + + counts = db.purge_profile_state("gone") + + assert counts["delivery_obligations"] == 1 + pending = db._read_one( + "SELECT state FROM delivery_obligations WHERE obligation_id = ?", ("ob-gone",)) + assert pending["state"] == "abandoned" + # A delivered row for the same profile is history, not pending state: left exactly as it was. + delivered = db._read_one( + "SELECT state FROM delivery_obligations WHERE obligation_id = ?", ("ob-gone-done",)) + assert delivered["state"] == "delivered" + bystander = db._read_one( + "SELECT state FROM delivery_obligations WHERE obligation_id = ?", ("ob-keep",)) + assert bystander["state"] == "pending" + + def test_purge_does_not_delete_session_rows(self, db): + """Pins this helper's scope only: the purge settles identity, it never deletes rows itself. + + This is NOT a claim that a deleted profile's conversation record outlives the delete — the + delete path removes the profile's own ``state.db`` together with ``profiles//``. It + says the identity purge adds no deletion of its own, so it cannot be the step that loses + history, and nothing here pins the row's stale ownership either. + """ + db.create_session( + "sess_gone", "feishu", session_key="agent:gone:feishu:dm:chatA", + profile_name="gone", chat_id="chatA", chat_type="dm", + ) + + db.purge_profile_state("gone") + + assert db._read_one( + "SELECT id FROM sessions WHERE id = ?", ("sess_gone",)) is not None + + def test_purges_topic_bindings_matched_by_session_key_namespace(self, db): + """The rekey rewrites a binding's ``profile_name`` AND its ``session_key`` namespace, so the + purge has to match both — a binding whose ``profile_name`` names somebody else still belongs + to the deleted profile by key, and matching on ``profile_name`` alone leaves it routing + events for a profile that no longer exists.""" + db.create_session( + "sess_gone", "telegram", session_key="agent:gone:telegram:dm:chatA", + profile_name="gone", chat_id="chatA", chat_type="dm", + ) + db.create_session( + "sess_keep", "telegram", session_key="agent:keepme:telegram:dm:chatB", + profile_name="keepme", chat_id="chatB", chat_type="dm", + ) + db.bind_telegram_topic( + chat_id="chatA", thread_id="threadA", user_id="userA", + session_key="agent:gone:telegram:dm:chatA", session_id="sess_gone", + profile_name="otherholder") + db.bind_telegram_topic( + chat_id="chatB", thread_id="threadB", user_id="userB", + session_key="agent:keepme:telegram:dm:chatB", session_id="sess_keep", + profile_name="keepme") + + counts = db.purge_profile_state("gone") + + assert db._read_one( + "SELECT COUNT(*) AS n FROM telegram_dm_topic_bindings WHERE chat_id = ?", + ("chatA",))["n"] == 0 + assert db._read_one( + "SELECT COUNT(*) AS n FROM telegram_dm_topic_bindings WHERE chat_id = ?", + ("chatB",))["n"] == 1 + assert counts["telegram_dm_topic_bindings"] == 1 + + def test_underscore_in_profile_name_is_not_a_like_wildcard(self, db): + db.save_gateway_routing_entry( + "agent:foo_bar:feishu:dm:chatA", + json.dumps({"session_key": "agent:foo_bar:feishu:dm:chatA"}), scope="/root/sessions") + db.save_gateway_routing_entry( + "agent:fooXbar:feishu:dm:chatB", + json.dumps({"session_key": "agent:fooXbar:feishu:dm:chatB"}), scope="/root/sessions") + + db.purge_profile_state("foo_bar") + + routing = db.load_gateway_routing_entries(scope="/root/sessions") + assert "agent:foo_bar:feishu:dm:chatA" not in routing + assert "agent:fooXbar:feishu:dm:chatB" in routing + + def test_purge_is_idempotent(self, db): + db.register_backend_heartbeat( + backend_id="be1", pid=1, started_at=time.time(), profile="gone", host="h") + + assert db.purge_profile_state("gone")["gateway_heartbeats"] == 1 + assert db.purge_profile_state("gone")["gateway_heartbeats"] == 0 diff --git a/website/docs/reference/profile-commands.md b/website/docs/reference/profile-commands.md index 390af56449..df8bef7274 100644 --- a/website/docs/reference/profile-commands.md +++ b/website/docs/reference/profile-commands.md @@ -276,6 +276,40 @@ hermes profile migrate-identity mybot assistant # ✓ Session/routing identity migrated: mybot → assistant ``` +## `hermes profile purge-identity` + +```bash +hermes profile purge-identity +``` + +Retries the identity purge of a delete that already completed. Run it if `hermes profile delete` +reported that its session/routing identity settlement is still pending: restart the gateway (it +reloads the routing index from the database, so the purge lands), or stop it — with no gateway +holding the store the command performs the durable delete itself. + +The purge keys off `` alone, so the profile directory does not have to exist — but a profile +that is live again under that name is refused: identity is settled by name, so purging it would take +the new profile's routing with it. Routing keys (`agent::*`), heartbeat rows and the profile's +Telegram topic bindings/mode rows are deleted; `delivery_obligations` rows are marked `abandoned` +rather than dropped, so pending delivery state is not lost silently. Session rows are not deleted by +the purge itself — it settles identity, not history; whether a conversation record outlives a delete +is decided by `hermes profile delete`, which removes the profile's own `profiles//`, its +`state.db` included. Idempotent — re-running a completed purge succeeds with nothing left to purge. +Exits non-zero when the name is a live profile again, when a live gateway refuses the purge, or when +a database rejects the delete (a lock, or a partial failure). + +**Example:** + +```bash +hermes profile delete mybot +# ⚠ Profile was deleted, but the live gateway could not purge its session identity (…). +# Restart the gateway, then run: +# hermes profile purge-identity mybot + +hermes profile purge-identity mybot +# ✓ Session/routing identity purged: mybot +``` + ## `hermes profile export` ```bash diff --git a/website/docs/user-guide/profiles.md b/website/docs/user-guide/profiles.md index 3e0580c628..f3b9a28960 100644 --- a/website/docs/user-guide/profiles.md +++ b/website/docs/user-guide/profiles.md @@ -309,6 +309,7 @@ hermes profile list # show all profiles with status hermes profile show coder # detailed info for one profile hermes profile rename coder dev-bot # rename (updates alias + service) hermes profile migrate-identity coder dev-bot # retry a rename's identity migration +hermes profile purge-identity dev-bot # retry a delete's identity purge hermes profile export coder # pack into coder.tar.gz (shareable; keys stripped) hermes profile import coder.tar.gz # install an archive as a new profile ```