From 67757285f696a4614b76e928dfd86c9f07ec5704 Mon Sep 17 00:00:00 2001
From: teknium1 <127238744+teknium1@users.noreply.github.com>
Date: Fri, 18 Sep 2026 22:17:59 -0700
Subject: [PATCH] feat(sessions): `hermes sessions repair-profiles` settles
crossed-profile durable state
The per-profile store model (#88734), the parent-inheritance fence (#88381),
profile-stamped topic rows (#76423) and profile-prefixed voice keys (#75198)
are all forward-only: they put NEW state under the right profile and refuse to
widen existing damage, but nothing walks the stores and settles what earlier
releases left crossed. #113884 found 246 sessions stranded that way and could
only warn.
`hermes sessions repair-profiles` scans every profile's state.db plus the
gateway's voice-mode and sessions.json files and names six kinds of crossing:
1. `profile_name` disagreeing with the row's own session key -> relabel;
2. rows physically in another profile's store -> move (all message
generations, usage rows, system prompt) to the owning store, parents before
children so lineage survives, copy-then-delete so a crash leaves a duplicate
the next run settles;
3. `parent_session_id` crossing namespaces -> sever (own identity kept);
4. routing rows outside the default store under multiplexing -> move (an
existing row wins); routing rows for a profile that no longer exists -> drop;
5. Telegram topic bindings and voice-mode entries missing their bot's profile
-> relabel from the sessions that hold the chat (ambiguous chats reported);
6. sessions.json mirror entries for an unclaimed namespace -> drop (the legacy
import re-injects them into routing every boot).
Report-only by default. `--apply` refuses while a gateway owns any store, takes
a quick snapshot of every store first, and is idempotent. Two cases are
reported but never guessed: rows keyed to a profile that does not exist, and
`agent:main` rows inside a named profile's store (`--legacy-main rekey|move`
says which of the two histories they are).
Storage side lives in `hermes_state_profile_repair.py` (SessionDB mixin);
orchestration across stores in `hermes_cli/sessions_repair_profiles.py`; the
CLI face in `hermes_cli/sessions_cmd_repair_profiles.py` (pre-DB handler: it
opens every store itself).
Part of #88715 (PR-6). Closes the remediation gap #113884 only warns about.
---
hermes_cli/sessions_cmd.py | 10 +-
hermes_cli/sessions_cmd_repair_profiles.py | 109 +++++
hermes_cli/sessions_repair_profiles.py | 462 ++++++++++++++++++
hermes_cli/subcommands/sessions.py | 20 +
hermes_state.py | 3 +-
hermes_state_profile_repair.py | 338 +++++++++++++
.../test_sessions_repair_profiles.py | 236 +++++++++
website/docs/reference/cli-commands.md | 1 +
website/docs/user-guide/sessions.md | 46 ++
9 files changed, 1223 insertions(+), 2 deletions(-)
create mode 100644 hermes_cli/sessions_cmd_repair_profiles.py
create mode 100644 hermes_cli/sessions_repair_profiles.py
create mode 100644 hermes_state_profile_repair.py
create mode 100644 tests/hermes_cli/test_sessions_repair_profiles.py
diff --git a/hermes_cli/sessions_cmd.py b/hermes_cli/sessions_cmd.py
index 25e5d4f5db..ff11dcc26e 100644
--- a/hermes_cli/sessions_cmd.py
+++ b/hermes_cli/sessions_cmd.py
@@ -963,7 +963,15 @@ def _cmd_stats(db, args):
# -- dispatch -----------------------------------------------------------------
-_PRE_DB_HANDLERS = {"repair": _cmd_repair, "recover": _cmd_recover, "import": _cmd_import}
+def _cmd_repair_profiles(args):
+ from hermes_cli.sessions_cmd_repair_profiles import cmd_repair_profiles
+ return cmd_repair_profiles(args)
+
+
+_PRE_DB_HANDLERS = {
+ "repair": _cmd_repair, "recover": _cmd_recover, "import": _cmd_import,
+ "repair-profiles": _cmd_repair_profiles, # opens every profile's store itself
+}
_OBSERVATIONAL_DB_ACTIONS = frozenset({"list", "stats", "pinned"})
_DB_HANDLERS = {
"list": _cmd_list, "export": _cmd_export, "delete": _cmd_delete, "rename": _cmd_rename, "pinned": _cmd_pinned,
diff --git a/hermes_cli/sessions_cmd_repair_profiles.py b/hermes_cli/sessions_cmd_repair_profiles.py
new file mode 100644
index 0000000000..bc026dd13b
--- /dev/null
+++ b/hermes_cli/sessions_cmd_repair_profiles.py
@@ -0,0 +1,109 @@
+"""``hermes sessions repair-profiles`` — the CLI face of :mod:`hermes_cli.sessions_repair_profiles`.
+
+Runs pre-DB (``_PRE_DB_HANDLERS``): it opens every profile's store itself rather than the one
+ambient ``SessionDB()``. Report-only unless ``--apply``; ``--json`` for automation.
+"""
+from __future__ import annotations
+
+import json
+import sys
+from typing import Dict, List
+
+from hermes_cli.sessions_repair_profiles import (
+ Finding, default_snapshot, enumerate_stores, live_gateway_homes, scan_stores,
+)
+
+_KIND_LABELS = {
+ "mislabelled": "profile_name disagrees with the row's own session key",
+ "wrong_store": "rows in another profile's store",
+ "legacy_main": "legacy agent:main rows inside a named profile's store",
+ "unclaimed_namespace": "rows keyed to a profile that does not exist",
+ "crossed_parent": "parent_session_id crossing profile namespaces",
+ "routing_stray": "routing rows outside the default store",
+ "routing_unclaimed": "routing rows for a profile that does not exist",
+ "topic_profile_less": "Telegram topic bindings without their bot's profile",
+ "voice_profile_less": "voice-mode entries without their bot's profile",
+ "sessions_json_unclaimed": "sessions.json entries for a profile that does not exist",
+}
+
+
+def _group(findings: List[Finding]) -> Dict[str, List[Finding]]:
+ grouped: Dict[str, List[Finding]] = {}
+ for finding in findings:
+ grouped.setdefault(finding.kind, []).append(finding)
+ return grouped
+
+
+def _print_report(findings: List[Finding]) -> None:
+ for kind, rows in _group(findings).items():
+ print(f"\n{_KIND_LABELS.get(kind, kind)} ({len(rows)}):")
+ for finding in rows:
+ print(f" [{finding.store}] {finding.subject}\n {finding.detail}")
+ if finding.action:
+ print(f" → {finding.action}")
+ else:
+ print(f" ✗ not repaired — {finding.reason}")
+
+
+def cmd_repair_profiles(args) -> int:
+ stores = enumerate_stores()
+ apply = bool(getattr(args, "apply", False))
+ as_json = bool(getattr(args, "json", False))
+ legacy_main = getattr(args, "legacy_main", None) or "report"
+
+ if apply:
+ # Before any store is opened for write: a writable open runs schema init, and a live
+ # gateway holds the routing index in memory and would write it back over the repair.
+ live = live_gateway_homes(stores)
+ if live:
+ names = ", ".join(f"{profile} (pid {pid})" for profile, pid in live)
+ print(f"A gateway is running for: {names}. It holds the routing index in memory and would "
+ "write it back over this repair. Stop it (`hermes gateway stop`), then re-run --apply.",
+ file=sys.stderr)
+ return 1
+
+ plan, session = scan_stores(stores, legacy_main=legacy_main, read_only=not apply)
+ try:
+ findings = plan.findings
+ repairable = plan.repairable
+ if as_json and not apply:
+ print(json.dumps({"stores": [s.profile for s in stores], "findings": [f.as_dict() for f in findings],
+ "repairable": len(repairable)}, indent=2))
+ return 0
+ if not as_json:
+ print(f"Scanned {len(stores)} profile store(s): {', '.join(s.profile for s in stores)}")
+ if not findings:
+ print("✓ No crossed-profile state found.")
+ return 0
+ _print_report(findings)
+ if not apply:
+ print(f"\n{len(repairable)} of {len(findings)} finding(s) can be repaired. "
+ "Re-run with --apply to perform them.")
+ return 0
+ if not repairable:
+ if as_json:
+ print(json.dumps({"findings": [f.as_dict() for f in findings], "applied": {}}, indent=2))
+ else:
+ print("\nNothing to repair.")
+ return 0
+
+ if not getattr(args, "yes", False) and not as_json:
+ from hermes_cli.sessions_cmd import _confirm_prompt
+ if not _confirm_prompt(f"\nApply {len(repairable)} repair(s)? A snapshot of every store is taken first. [y/N] "):
+ print("Aborted — nothing was changed.")
+ return 0
+ result = plan.apply(session, snapshot=default_snapshot)
+ finally:
+ session.close()
+
+ if as_json:
+ print(json.dumps({"findings": [f.as_dict() for f in findings], "applied": result}, indent=2))
+ else:
+ for profile, snap in result["snapshots"].items():
+ print(f" snapshot [{profile}]: {snap or 'FAILED'}")
+ applied = ", ".join(f"{k}={v}" for k, v in sorted(result["totals"].items()) if v)
+ print(f"\nApplied: {applied or 'nothing'}")
+ for failure in result["failures"]:
+ print(f" ✗ {failure['kind']} {failure['subject']}: {failure['error']}")
+ print("Re-run without --apply to confirm the stores are clean.")
+ return 1 if result["failures"] else 0
diff --git a/hermes_cli/sessions_repair_profiles.py b/hermes_cli/sessions_repair_profiles.py
new file mode 100644
index 0000000000..d93df3665c
--- /dev/null
+++ b/hermes_cli/sessions_repair_profiles.py
@@ -0,0 +1,462 @@
+"""``hermes sessions repair-profiles`` — settle crossed-profile durable state across every store.
+
+The per-profile store model (#88734) and the identity fences that followed are forward-only: they
+put NEW rows in the right place and refuse to widen existing damage, but nothing walks the stores
+and settles what an earlier release left crossed. This does, for six kinds of crossing:
+
+1. ``sessions.profile_name`` disagreeing with the profile in the row's own ``session_key``;
+2. rows physically in the wrong store (a named profile's keys in the root ``state.db``, another
+ profile's keys in ``profiles//state.db``) — moved to the owning profile's store;
+3. ``parent_session_id`` crossing namespaces — severed (the child's own identity is untouched);
+4. ``gateway_routing`` rows in a named profile's store (the routing index lives in the gateway's
+ home; #66887 copied every profile's rows into whichever store was scoped) — moved to the routing
+ store when absent there, dropped otherwise — and root rows for a namespace no profile claims;
+5. profile-less Telegram topic bindings and ``gateway_voice_mode.json`` entries owned by a
+ non-default bot — relabelled from the evidence of the rows themselves;
+6. ``sessions.json`` mirror entries whose key namespace no served profile claims — dropped, since
+ the legacy import re-injects any key the routing DB lacks on every boot.
+
+Legacy ``agent:main:`` rows inside a named profile's store are reported but NOT repaired unless
+``--legacy-main`` says how: they are either a standalone (pre-multiplex) gateway's own history that
+must be rekeyed to ``agent:
:`` (the #113884 incident), or a default-profile chat that leaked in
+under a scoped write (#102157) and must move to the root store. The row carries no evidence that
+tells the two apart, and guessing wrong hands a conversation to the other bot.
+
+Report-only by default. ``--apply`` refuses while a gateway owning any touched store is live (it
+holds the routing index in memory and writes it back), takes a quick snapshot of every store it
+will mutate, and is idempotent: a second run finds nothing.
+"""
+from __future__ import annotations
+
+import contextlib
+import json
+import logging
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Callable, Dict, Iterable, List, Optional, Set, Tuple
+
+logger = logging.getLogger(__name__)
+
+_LEGACY_MAIN_CHOICES = ("report", "rekey", "move")
+_VOICE_MODE_FILE = "gateway_voice_mode.json"
+
+
+@dataclass(frozen=True)
+class Store:
+ """One profile's ``state.db``; ``routing`` marks the store that owns the multiplexer's index."""
+ profile: str
+ home: Path
+ routing: bool = False
+
+ @property
+ def db_path(self) -> Path:
+ return self.home / "state.db"
+
+
+@dataclass
+class Finding:
+ kind: str
+ store: str
+ subject: str
+ detail: str
+ action: Optional[str] = None # what --apply does; None = report only
+ reason: Optional[str] = None # why it is report only
+ _fix: Optional[Callable[["_Session"], Dict[str, int]]] = field(default=None, repr=False, compare=False)
+
+ @property
+ def repairable(self) -> bool:
+ return self._fix is not None
+
+ def as_dict(self) -> Dict[str, Any]:
+ return {"kind": self.kind, "store": self.store, "subject": self.subject, "detail": self.detail,
+ "action": self.action, "reason": self.reason}
+
+
+class _Session:
+ """Open store handles for one run; every store opened at most once, all released at the end."""
+
+ def __init__(self, *, read_only: bool) -> None:
+ self.read_only = read_only
+ self._dbs: Dict[Path, Any] = {}
+
+ def db(self, store: Store):
+ """The store's SessionDB. Read-only runs never create a file: a live profile that has no
+ ``state.db`` yet reads as empty; an apply run opens it for real when a move lands there."""
+ path = store.db_path
+ if path not in self._dbs:
+ if self.read_only:
+ from hermes_state import SessionDB
+ self._dbs[path] = SessionDB(path, read_only=True) if path.exists() else _EMPTY_STORE
+ else:
+ from hermes_state_registry import acquire
+ self._dbs[path] = acquire(path)
+ return self._dbs[path]
+
+ def close(self) -> None:
+ from hermes_state_registry import release_or_close
+ for db in self._dbs.values():
+ if db is _EMPTY_STORE:
+ continue
+ with contextlib.suppress(Exception):
+ release_or_close(db) if not self.read_only else db.close()
+ self._dbs.clear()
+
+
+class _EmptyStore:
+ """Stand-in for a live profile whose ``state.db`` does not exist yet: nothing to find."""
+ def find_crossed_profile_sessions(self, owner):
+ return {"mislabelled": [], "foreign": [], "crossed_parents": []}
+
+ def find_profile_less_telegram_topic_rows(self):
+ return []
+
+ def list_gateway_routing_rows(self):
+ return []
+
+ def key_profiles_for_chat(self, platform, chat_id):
+ return set()
+
+
+_EMPTY_STORE = _EmptyStore()
+
+
+# ── enumeration ───────────────────────────────────────────────────────────────
+
+def enumerate_stores() -> List[Store]:
+ """Default root plus every live named profile (a live profile claims its namespace whether or not
+ it has written a ``state.db`` yet). The root store owns the routing index: the multiplexer's
+ ``_routing_home`` is its launch home, the root."""
+ from hermes_cli.profiles import _get_default_hermes_home, _iter_named_profile_dirs
+ root = _get_default_hermes_home()
+ stores = [Store("default", root, routing=True)]
+ stores.extend(Store(entry.name, entry) for entry in _iter_named_profile_dirs())
+ return stores
+
+
+def _gateway_multiplexes(root: Path) -> bool:
+ """Does the default gateway serve every profile? Same reader as every other CLI surface: the live
+ record, else the explicit flag, else False — an unset flag is a verdict only the gateway reaches,
+ and guessing "yes" would uproot a standalone gateway's own routing index."""
+ from hermes_cli.gateway_multiplex_mode import default_gateway_multiplexes
+ try:
+ return default_gateway_multiplexes(root)
+ except Exception:
+ return False
+
+
+def live_gateway_homes(stores: Iterable[Store]) -> List[Tuple[str, int]]:
+ from gateway.status import live_gateway_pid_for_home
+ live = []
+ for store in stores:
+ pid = live_gateway_pid_for_home(store.home)
+ if pid is not None:
+ live.append((store.profile, pid))
+ return live
+
+
+# ── scan ──────────────────────────────────────────────────────────────────────
+
+class RepairPlan:
+ def __init__(self, stores: List[Store], *, legacy_main: str = "report") -> None:
+ if legacy_main not in _LEGACY_MAIN_CHOICES:
+ raise ValueError(f"legacy_main must be one of {_LEGACY_MAIN_CHOICES}")
+ self.stores = stores
+ self.by_profile = {s.profile: s for s in stores}
+ self.claimed: Set[str] = set(self.by_profile)
+ self.routing_store = next((s for s in stores if s.routing), None)
+ self.legacy_main = legacy_main
+ self.multiplexes = _gateway_multiplexes(self.routing_store.home) if self.routing_store else False
+ self.findings: List[Finding] = []
+ self._moves: Dict[Tuple[Path, Path], "_MoveBatch"] = {}
+
+ # -- helpers --
+ def _add(self, finding: Finding) -> None:
+ self.findings.append(finding)
+
+ @property
+ def repairable(self) -> List[Finding]:
+ return [f for f in self.findings if f.repairable]
+
+ # -- scan --
+ def scan(self, session: _Session) -> "RepairPlan":
+ for store in self.stores:
+ self._scan_sessions(session, store)
+ self._scan_topics(session, store)
+ for store in self.stores:
+ self._scan_routing(session, store)
+ if self.routing_store is not None:
+ self._scan_voice_modes(session, self.routing_store)
+ self._scan_sessions_json(self.routing_store)
+ return self
+
+ def _scan_sessions(self, session: _Session, store: Store) -> None:
+ db = session.db(store)
+ crossed = db.find_crossed_profile_sessions(store.profile)
+ moving: Set[str] = set()
+ for row in crossed["foreign"]:
+ self._plan_foreign_row(store, row, moving)
+ for row in crossed["mislabelled"]:
+ if row["id"] in moving:
+ continue # the move stamps the right label
+ self._add(Finding(
+ "mislabelled", store.profile, row["id"],
+ f"profile_name={row['profile_name']!r} but key {row['session_key']!r} belongs to "
+ f"{row['key_profile']!r}", action=f"relabel to {row['key_profile']!r}",
+ _fix=lambda s, st=store, sid=row["id"]: {
+ "relabelled": s.db(st).relabel_sessions_to_key_profile([sid])}))
+ for row in crossed["crossed_parents"]:
+ if row["id"] in moving:
+ continue # the import drops a parent that is not in the target store
+ self._add(Finding(
+ "crossed_parent", store.profile, row["id"],
+ f"{row['key_profile']!r} row inherits from parent {row['parent_session_id']} keyed under "
+ f"{row['parent_profile']!r}", action="sever parent_session_id",
+ _fix=lambda s, st=store, sid=row["id"]: {"severed": s.db(st).sever_crossed_parents([sid])}))
+
+ def _plan_foreign_row(self, store: Store, row: Dict[str, Any], moving: Set[str]) -> None:
+ key_profile, sid = row["key_profile"], row["id"]
+ detail = f"key {row['session_key']!r} ({row['message_count']} messages) sits in the {store.profile} store"
+ if key_profile == "default" and store.profile != "default":
+ self._plan_legacy_main_row(store, row, moving)
+ return
+ if key_profile not in self.claimed:
+ self._add(Finding(
+ "unclaimed_namespace", store.profile, sid, detail,
+ reason=f"no live profile named {key_profile!r}; create it, then rerun, or "
+ f"`hermes profile migrate-identity {key_profile} `"))
+ return
+ moving.add(sid)
+ self._add(Finding(
+ "wrong_store", store.profile, sid, detail, action=f"move to the {key_profile} store",
+ _fix=self._move_batch(store, self.by_profile[key_profile]).fix_for(sid)))
+
+ def _plan_legacy_main_row(self, store: Store, row: Dict[str, Any], moving: Set[str]) -> None:
+ sid = row["id"]
+ detail = (f"legacy default-namespace key {row['session_key']!r} ({row['message_count']} messages) "
+ f"in the {store.profile} store")
+ if self.legacy_main == "rekey":
+ self._add(Finding(
+ "legacy_main", store.profile, sid, detail, action=f"rekey to agent:{store.profile}:",
+ _fix=lambda s, st=store, i=sid: {"rekeyed": s.db(st).rekey_legacy_main_sessions([i], st.profile)}))
+ elif self.legacy_main == "move" and self.routing_store is not None:
+ moving.add(sid)
+ self._add(Finding(
+ "legacy_main", store.profile, sid, detail, action="move to the default store",
+ _fix=self._move_batch(store, self.routing_store).fix_for(sid)))
+ else:
+ self._add(Finding(
+ "legacy_main", store.profile, sid, detail,
+ reason="either a standalone gateway's own history (--legacy-main rekey) or a default "
+ "chat that leaked in (--legacy-main move); the row cannot say which"))
+
+ def _move_batch(self, src: Store, dst: Store) -> "_MoveBatch":
+ return self._moves.setdefault((src.db_path, dst.db_path), _MoveBatch(src, dst))
+
+ def _scan_topics(self, session: _Session, store: Store) -> None:
+ rows = session.db(store).find_profile_less_telegram_topic_rows()
+ for row in rows:
+ self._add(Finding(
+ "topic_profile_less", store.profile, f"chat {row['chat_id']} thread {row['thread_id']}",
+ f"telegram topic binding labelled 'default' but keyed {row['session_key']!r}",
+ action=f"relabel to {row['key_profile']!r}",
+ _fix=lambda s, st=store, r=row: s.db(st).relabel_telegram_topic_rows([r])))
+
+ def _scan_routing(self, session: _Session, store: Store) -> None:
+ from hermes_state_profile_repair import session_key_profile
+ rows = session.db(store).list_gateway_routing_rows()
+ for row in rows:
+ key_profile = session_key_profile(row["session_key"])
+ if key_profile is None:
+ continue
+ subject = f"{row['session_key']} (scope {row['scope']!r})"
+ if store.routing:
+ if key_profile not in self.claimed:
+ self._add(Finding(
+ "routing_unclaimed", store.profile, subject,
+ f"routing row for a namespace no live profile claims ({key_profile!r}); every "
+ "inbound event on it logs a missing-profile error", action="delete",
+ _fix=lambda s, st=store, r=row: {
+ "routing_deleted": s.db(st).delete_gateway_routing_rows([(r["scope"], r["session_key"])])}))
+ continue
+ if key_profile == store.profile and not self.multiplexes:
+ continue # a standalone gateway's own index lives in its own store
+ routing_store = self.routing_store
+ if routing_store is None:
+ continue
+ self._add(Finding(
+ "routing_stray", store.profile, subject,
+ f"routing row in the {store.profile} store; the index lives in the default store",
+ action="move to the default store (existing row there wins)",
+ _fix=lambda s, st=store, r=row, rs=routing_store: self._move_routing_row(s, st, rs, r)))
+
+ def _move_routing_row(self, session: _Session, store: Store, routing_store: Store,
+ row: Dict[str, Any]) -> Dict[str, int]:
+ adopted = session.db(routing_store).insert_gateway_routing_rows_if_absent(
+ [(row["scope"], row["session_key"], row["entry_json"], row["updated_at"])])
+ deleted = session.db(store).delete_gateway_routing_rows([(row["scope"], row["session_key"])])
+ return {"routing_adopted": adopted, "routing_deleted": deleted}
+
+ def _scan_voice_modes(self, session: _Session, store: Store) -> None:
+ path = store.home / _VOICE_MODE_FILE
+ try:
+ data = json.loads(path.read_text(encoding="utf-8"))
+ except (FileNotFoundError, json.JSONDecodeError, OSError):
+ return
+ if not isinstance(data, dict):
+ return
+ for key in sorted(k for k in data if isinstance(k, str) and k.count(":") == 1):
+ platform, chat_id = key.split(":", 1)
+ owners: Set[str] = set()
+ for st in self.stores:
+ owners |= session.db(st).key_profiles_for_chat(platform, chat_id)
+ if not owners or "default" in owners:
+ continue # the default bot speaks there: the unprefixed key is its own
+ if len(owners) > 1:
+ self._add(Finding(
+ "voice_profile_less", store.profile, key,
+ f"voice mode entry without a profile; chat is held by {sorted(owners)}",
+ reason="more than one non-default bot speaks in this chat"))
+ continue
+ owner = next(iter(owners))
+ self._add(Finding(
+ "voice_profile_less", store.profile, key,
+ f"voice mode {data[key]!r} without a profile; chat is held only by {owner!r}",
+ action=f"rekey to {owner}:{key}",
+ _fix=lambda s, p=path, k=key, o=owner: _rekey_voice_mode_entry(p, k, o)))
+
+ def _scan_sessions_json(self, store: Store) -> None:
+ from hermes_state_profile_repair import session_key_profile
+ path = store.home / "sessions" / "sessions.json"
+ try:
+ data = json.loads(path.read_text(encoding="utf-8"))
+ except (FileNotFoundError, json.JSONDecodeError, OSError):
+ return
+ if not isinstance(data, dict):
+ return
+ for key in sorted(k for k in data if isinstance(k, str) and not k.startswith("_")):
+ key_profile = session_key_profile(key)
+ if key_profile is None or key_profile in self.claimed:
+ continue
+ self._add(Finding(
+ "sessions_json_unclaimed", store.profile, key,
+ f"sessions.json mirror entry for a namespace no live profile claims ({key_profile!r}); "
+ "the legacy import re-injects it into routing on every boot", action="delete entry",
+ _fix=lambda s, p=path, k=key: _drop_sessions_json_entry(p, k)))
+
+ # -- apply --
+ def apply(self, session: _Session, *, snapshot: Callable[[Store], Optional[str]]) -> Dict[str, Any]:
+ """Run every repairable fix; snapshots every store first. Fixes are independent and each
+ re-checks its precondition, so a partial run leaves a state the next run completes."""
+ snapshots = {s.profile: snapshot(s) for s in self.stores if s.db_path.exists()}
+ totals: Dict[str, int] = {}
+ failures: List[Dict[str, str]] = []
+ for finding in self.repairable:
+ fix = finding._fix
+ assert fix is not None
+ try:
+ for key, count in (fix(session) or {}).items():
+ totals[key] = totals.get(key, 0) + int(count)
+ except Exception as exc: # one bad row must not abandon the rest of the plan
+ logger.warning("repair-profiles: %s on %s failed", finding.kind, finding.subject, exc_info=True)
+ failures.append({"kind": finding.kind, "subject": finding.subject,
+ "error": f"{type(exc).__name__}: {exc}"})
+ return {"snapshots": snapshots, "totals": totals, "failures": failures}
+
+
+class _MoveBatch:
+ """Every session moving from one store to another, moved together: export all, import all
+ (parents before children, so a moved child keeps its moved parent), then delete from the source.
+ Row by row, deleting a parent would detach the children still waiting in the source. Copy
+ precedes delete, so a crash between the two leaves a duplicate the next run settles
+ (``present`` on import, then the delete)."""
+
+ def __init__(self, src: Store, dst: Store) -> None:
+ self.src, self.dst = src, dst
+ self.ids: List[str] = []
+ self._result: Optional[Dict[str, Dict[str, int]]] = None
+
+ def fix_for(self, sid: str) -> Callable[[_Session], Dict[str, int]]:
+ self.ids.append(sid)
+ return lambda session: self.run(session).get(sid, {"missing": 1})
+
+ def run(self, session: _Session) -> Dict[str, Dict[str, int]]:
+ if self._result is not None:
+ return self._result
+ src_db, dst_db = session.db(self.src), session.db(self.dst)
+ payloads = {sid: src_db.export_session_for_move(sid) for sid in self.ids}
+ ordered = _parents_first([p for p in payloads.values() if p is not None])
+ result: Dict[str, Dict[str, int]] = {sid: {"missing": 1} for sid, p in payloads.items() if p is None}
+ for payload in ordered:
+ sid = payload["session"]["id"]
+ outcome = dst_db.import_moved_session(payload, profile_name=self.dst.profile)
+ if dst_db.count_messages_all(sid) < len(payload["messages"]):
+ logger.warning("repair-profiles: %s copied into %s with fewer messages than the source; "
+ "source row kept", sid, self.dst.profile)
+ result[sid] = {"copied_incomplete": 1}
+ continue
+ result[sid] = {outcome: 1}
+ for payload in ordered:
+ sid = payload["session"]["id"]
+ if "copied_incomplete" not in result[sid] and src_db.delete_moved_session(sid):
+ result[sid]["moved"] = 1
+ self._result = result
+ return result
+
+
+def _parents_first(payloads: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+ """Order moved sessions so every parent in the batch precedes its children (a cycle, impossible
+ for a well-formed store, falls back to insertion order)."""
+ pending = {p["session"]["id"]: p for p in payloads}
+ ordered: List[Dict[str, Any]] = []
+ while pending:
+ ready = [sid for sid, p in pending.items() if p["session"].get("parent_session_id") not in pending]
+ if not ready:
+ ready = list(pending)
+ for sid in ready:
+ ordered.append(pending.pop(sid))
+ return ordered
+
+
+def _rekey_voice_mode_entry(path: Path, key: str, owner: str) -> Dict[str, int]:
+ from utils import atomic_json_write
+ data = json.loads(path.read_text(encoding="utf-8"))
+ if key not in data:
+ return {}
+ data.setdefault(f"{owner}:{key}", data.pop(key))
+ data.pop(key, None)
+ atomic_json_write(path, data)
+ return {"voice_rekeyed": 1}
+
+
+def _drop_sessions_json_entry(path: Path, key: str) -> Dict[str, int]:
+ from utils import atomic_json_write
+ data = json.loads(path.read_text(encoding="utf-8"))
+ if key not in data:
+ return {}
+ del data[key]
+ atomic_json_write(path, data, mode=0o600)
+ return {"sessions_json_dropped": 1}
+
+
+# ── entry points ──────────────────────────────────────────────────────────────
+
+def scan_stores(stores: Optional[List[Store]] = None, *, legacy_main: str = "report",
+ read_only: bool = True) -> Tuple[RepairPlan, _Session]:
+ """Open every store and build the plan. The caller owns the returned session (close it)."""
+ stores = enumerate_stores() if stores is None else stores
+ session = _Session(read_only=read_only)
+ try:
+ plan = RepairPlan(stores, legacy_main=legacy_main).scan(session)
+ except Exception:
+ session.close()
+ raise
+ return plan, session
+
+
+def default_snapshot(store: Store) -> Optional[str]:
+ from hermes_cli.backup import create_quick_snapshot
+ try:
+ return create_quick_snapshot(label="repair-profiles", hermes_home=store.home)
+ except Exception as exc:
+ logger.warning("repair-profiles: snapshot of %s failed: %s", store.home, exc)
+ return None
diff --git a/hermes_cli/subcommands/sessions.py b/hermes_cli/subcommands/sessions.py
index e235c20a9e..9e8fc885c1 100644
--- a/hermes_cli/subcommands/sessions.py
+++ b/hermes_cli/subcommands/sessions.py
@@ -183,6 +183,26 @@ def build_sessions_parser(subparsers, *, cmd_sessions: Callable) -> None:
"orphan's start for them to count as the same conversation "
"(default: 900)")
+ sessions_repair_profiles = sessions_subparsers.add_parser(
+ "repair-profiles", help="Settle session/routing state that landed under the wrong profile",
+ description="Scan every profile's state.db (and the gateway's voice-mode / sessions.json "
+ "files) for durable state crossed between profiles: rows whose profile_name "
+ "disagrees with their session key, rows sitting in another profile's store, "
+ "parent links crossing namespaces, routing rows outside the default store or for a "
+ "profile that no longer exists, and Telegram topic / voice-mode entries missing "
+ "their bot's profile. Reports without touching anything unless --apply is given; "
+ "--apply refuses while a gateway is running, snapshots every store first, and is "
+ "safe to re-run.")
+ _flag(sessions_repair_profiles, "--apply", help="Perform the repairs (default: report only)")
+ _flag(sessions_repair_profiles, "--json", help="Machine-readable report")
+ _flag(sessions_repair_profiles, "--yes", "-y", default=False, help="Skip the confirmation prompt")
+ sessions_repair_profiles.add_argument(
+ "--legacy-main", choices=("report", "rekey", "move"), default="report",
+ help="What to do with agent:main rows found inside a named profile's store: 'rekey' them "
+ "to that profile (a standalone gateway's own history, e.g. after multiplexing was "
+ "switched on), 'move' them to the default store (a default chat that leaked in), or "
+ "'report' (default) — the rows themselves cannot tell the two cases apart")
+
sessions_recover = sessions_subparsers.add_parser(
"recover", help="Rebuild canonical session data into a separate clean database",
description="Offline, non-destructive recovery for a damaged state.db. The "
diff --git a/hermes_state.py b/hermes_state.py
index df1da67519..71b91c19f7 100644
--- a/hermes_state.py
+++ b/hermes_state.py
@@ -48,6 +48,7 @@ from hermes_state_sessions import SessionSessionsMixin
from hermes_state_fts import SessionFtsSetupMixin, load_fts5_cjk_extension
from hermes_state_portability import SessionPortabilityMixin
from hermes_state_telegram import SessionTelegramTopicsMixin
+from hermes_state_profile_repair import SessionProfileRepairMixin
from hermes_state_schema import SessionSchemaMixin
import hermes_state_holders as _state_holders
import hermes_state_lockguard as _lockguard
@@ -442,7 +443,7 @@ class SessionDB(
SessionSessionsMixin, SessionFtsSetupMixin, SessionSearchMixin, SessionSchemaMixin,
SessionPortabilityMixin, SessionTelegramTopicsMixin, SessionCompressionMixin,
SessionGatewayMixin, SessionMaintenanceMixin, SessionUsageMixin, SessionTitlesMixin,
- SessionMessagesMixin, SessionRewindMixin,
+ SessionMessagesMixin, SessionRewindMixin, SessionProfileRepairMixin,
):
"""SQLite-backed session storage with FTS5 search; many reader threads, one writer (WAL)."""
diff --git a/hermes_state_profile_repair.py b/hermes_state_profile_repair.py
new file mode 100644
index 0000000000..34d88cd7be
--- /dev/null
+++ b/hermes_state_profile_repair.py
@@ -0,0 +1,338 @@
+"""Crossed-profile durable state in ONE ``state.db`` — finders and fixers for
+``hermes sessions repair-profiles`` (#88715 PR-6).
+
+Every ``state.db`` belongs to exactly one profile (``/state.db`` → ``default``,
+``/profiles//state.db`` → ``name``) and every gateway session key encodes the profile
+that owns the conversation (``agent:main:…`` for the default, ``agent::…`` for a named one).
+The per-profile store model (#88734) is forward-only: it routes NEW writes to the right store but
+never touched rows that had already landed in the wrong one, and the identity fences
+(``_INHERIT_PARENT_META_SQL``, ``_recovered_row_allowed_for_active_profile``) only refuse to
+*widen* damage that already exists. This module is the backward-looking half: it names each crossed
+row and can settle it.
+
+Store-level only. Which profiles exist, which store owns which home, whether a live gateway holds the
+routing index in memory, and the JSON files outside ``state.db`` are the orchestrator's concern
+(``hermes_cli/sessions_repair_profiles.py``).
+
+Moving a session between two stores is two single-store transactions — copy into the target, then
+delete from the source — because a crash between them leaves a *duplicate*, which the next run
+settles idempotently (:meth:`import_moved_session` reports ``present``; the delete only proceeds when
+the target holds at least as many messages). The reverse order would lose the row.
+"""
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, Iterable, List, Optional, Set, Tuple
+
+from hermes_state_common import _id_chunks, _placeholders as _session_ids_placeholders
+
+logger = logging.getLogger(__name__)
+
+
+def session_key_profile(session_key: Any) -> Optional[str]:
+ """Profile encoded in an ``agent::…`` gateway key (``default`` for ``main``), or None for a
+ keyless row (CLI/subagent lineage) or a key another producer minted (hosted rooms, tests)."""
+ if not isinstance(session_key, str):
+ return None
+ parts = session_key.split(":")
+ if len(parts) < 3 or parts[0] != "agent" or not parts[1]:
+ return None
+ from gateway.session import profile_from_session_key_namespace
+ return profile_from_session_key_namespace(parts[1])
+
+
+def _stored_profile(value: Any) -> Optional[str]:
+ """``sessions.profile_name`` as a comparable label; NULL/blank is "unowned", not a crossing."""
+ name = str(value or "").strip()
+ return name or None
+
+
+def _table_columns(conn, table: str) -> List[str]:
+ return [row[1] for row in conn.execute(f"PRAGMA table_info('{table}')")]
+
+
+# Columns a moved message must NOT carry over verbatim: ``id`` is reassigned by the target's
+# AUTOINCREMENT and ``display_order`` points at a message id of the SOURCE store — left NULL, the
+# ``messages_display_order_insert`` trigger recomputes it from ``display_identity`` (a content hash,
+# store-independent) or the new id.
+_MESSAGE_MOVE_SKIP = frozenset({"id", "display_order"})
+
+
+class SessionProfileRepairMixin:
+ """Per-store crossed-profile identity: find, relabel, sever, move, and settle routing/topic rows."""
+
+ # ── finders ────────────────────────────────────────────────────────────────
+
+ def find_crossed_profile_sessions(self, owner: str) -> Dict[str, List[Dict[str, Any]]]:
+ """Keyed session rows whose identity disagrees with itself or with this store.
+
+ ``mislabelled``: ``profile_name`` names a profile other than the one in the row's own key
+ (the key is what routing, the agent cache and ``_db_for_key`` consult; the label is stamped
+ after the fact). ``foreign``: the key's profile is not *owner* — the row sits in a store that
+ the routed profile never reads. ``crossed_parents``: child and parent are both keyed and name
+ different profiles, so the NULL-fill inheritance fence would have been the only thing standing
+ between them. A row can appear in more than one list.
+ """
+ rows = self._read_all(
+ "SELECT s.id, s.session_key, s.profile_name, s.parent_session_id, s.message_count, "
+ " p.session_key AS parent_session_key "
+ "FROM sessions s LEFT JOIN sessions p ON p.id = s.parent_session_id "
+ "WHERE s.session_key LIKE 'agent:%' ORDER BY s.started_at, s.id")
+ found: Dict[str, List[Dict[str, Any]]] = {"mislabelled": [], "foreign": [], "crossed_parents": []}
+ for row in rows:
+ key_profile = session_key_profile(row["session_key"])
+ if key_profile is None:
+ continue
+ label = _stored_profile(row["profile_name"])
+ if label is not None and label != key_profile:
+ found["mislabelled"].append({
+ "id": row["id"], "session_key": row["session_key"],
+ "profile_name": label, "key_profile": key_profile})
+ if key_profile != owner:
+ found["foreign"].append({
+ "id": row["id"], "session_key": row["session_key"], "key_profile": key_profile,
+ "message_count": int(row["message_count"] or 0)})
+ parent_profile = session_key_profile(row["parent_session_key"])
+ if parent_profile is not None and parent_profile != key_profile:
+ found["crossed_parents"].append({
+ "id": row["id"], "key_profile": key_profile,
+ "parent_session_id": row["parent_session_id"], "parent_profile": parent_profile})
+ return found
+
+ def list_gateway_routing_rows(self) -> List[Dict[str, Any]]:
+ return [dict(row) for row in self._read_all(
+ "SELECT scope, session_key, entry_json, updated_at FROM gateway_routing "
+ "ORDER BY scope, session_key")]
+
+ def find_profile_less_telegram_topic_rows(self) -> List[Dict[str, Any]]:
+ """``telegram_dm_topic_bindings`` rows labelled ``default`` whose ``session_key`` names a named
+ profile: written by a multiplexer that had not yet learned to stamp the routed profile
+ (#76423). Stores whose topic tables predate ``profile_name`` have nothing to relabel."""
+ def _read(conn):
+ existing = {row[0] for row in conn.execute(
+ "SELECT name FROM sqlite_master WHERE type='table' AND name = 'telegram_dm_topic_bindings'")}
+ if not existing or "profile_name" not in _table_columns(conn, "telegram_dm_topic_bindings"):
+ return []
+ return [dict(row) for row in conn.execute(
+ "SELECT chat_id, thread_id, session_key FROM telegram_dm_topic_bindings "
+ "WHERE profile_name = 'default' ORDER BY chat_id, thread_id")]
+ out = []
+ for row in self._read_retrying_ioerr(_read):
+ key_profile = session_key_profile(row["session_key"])
+ if key_profile not in (None, "default"):
+ out.append({**row, "key_profile": key_profile})
+ return out
+
+ def count_messages_all(self, session_id: str) -> int:
+ row = self._read_one("SELECT COUNT(*) FROM messages WHERE session_id = ?", (session_id,))
+ return int(row[0]) if row else 0
+
+ # ── in-store fixers ───────────────────────────────────────────────────────
+
+ def relabel_sessions_to_key_profile(self, ids: Iterable[str]) -> int:
+ """Set ``profile_name`` to the profile in each row's own key. Re-derives the target inside the
+ transaction (never trusts a stale report)."""
+ wanted = list(ids)
+ if not wanted:
+ return 0
+
+ def _do(conn) -> int:
+ changed = 0
+ for chunk in _id_chunks(wanted):
+ rows = conn.execute(
+ f"SELECT id, session_key FROM sessions WHERE id IN ({_session_ids_placeholders(chunk)})",
+ chunk).fetchall()
+ for session_id, session_key in rows:
+ target = session_key_profile(session_key)
+ if target is None:
+ continue
+ changed += conn.execute(
+ "UPDATE sessions SET profile_name = ? WHERE id = ? AND profile_name IS NOT ?",
+ (target, session_id, target)).rowcount
+ return changed
+ return self._execute_write(_do)
+
+ def sever_crossed_parents(self, ids: Iterable[str]) -> int:
+ """Detach children from a parent keyed under another profile. Only ``parent_session_id`` is
+ cleared — ``profile_name`` stays the row's own (relabelled separately when it is wrong) — and
+ only while the crossing still holds."""
+ wanted = list(ids)
+ if not wanted:
+ return 0
+
+ def _do(conn) -> int:
+ severed = 0
+ for chunk in _id_chunks(wanted):
+ rows = conn.execute(
+ "SELECT s.id, s.session_key, p.session_key FROM sessions s "
+ "JOIN sessions p ON p.id = s.parent_session_id "
+ f"WHERE s.id IN ({_session_ids_placeholders(chunk)})", chunk).fetchall()
+ for session_id, key, parent_key in rows:
+ mine, theirs = session_key_profile(key), session_key_profile(parent_key)
+ if mine is not None and theirs is not None and mine != theirs:
+ severed += conn.execute(
+ "UPDATE sessions SET parent_session_id = NULL WHERE id = ?", (session_id,)).rowcount
+ return severed
+ return self._execute_write(_do)
+
+ # ── cross-store move ──────────────────────────────────────────────────────
+
+ def export_session_for_move(self, session_id: str) -> Optional[Dict[str, Any]]:
+ """Everything the target store needs to hold *session_id* as its own: the row (every column),
+ the resolved system prompt, EVERY message row (inactive and compacted generations included —
+ a move is not an export) and its usage rows."""
+ def _read(conn):
+ session = conn.execute("SELECT * FROM sessions WHERE id = ?", (session_id,)).fetchone()
+ if session is None:
+ return None
+ prompt = None
+ if session["system_prompt_hash"]:
+ row = conn.execute(
+ "SELECT prompt FROM system_prompts WHERE hash = ?", (session["system_prompt_hash"],)).fetchone()
+ prompt = row[0] if row else None
+ messages = [dict(r) for r in conn.execute(
+ "SELECT * FROM messages WHERE session_id = ? ORDER BY id", (session_id,))]
+ usage = [dict(r) for r in conn.execute(
+ "SELECT * FROM session_model_usage WHERE session_id = ?", (session_id,))]
+ return {"session": dict(session), "system_prompt": prompt, "messages": messages, "usage": usage}
+ return self._read_retrying_ioerr(_read)
+
+ def import_moved_session(self, payload: Dict[str, Any], *, profile_name: str) -> str:
+ """Insert a moved session into THIS store as *profile_name*'s. ``present`` when the id already
+ exists (an earlier run copied but did not delete), else ``imported``. The parent link survives
+ only when the parent is already here — a moved row must never point across stores or at a
+ row of another profile. Columns the target schema lacks are dropped, never invented."""
+ session = dict(payload["session"])
+ session_id = session["id"]
+
+ def _do(conn) -> str:
+ if conn.execute("SELECT 1 FROM sessions WHERE id = ?", (session_id,)).fetchone():
+ return "present"
+ session["profile_name"] = profile_name
+ parent_id = session.get("parent_session_id")
+ if parent_id and conn.execute("SELECT 1 FROM sessions WHERE id = ?", (parent_id,)).fetchone() is None:
+ session["parent_session_id"] = None
+ session["system_prompt_hash"] = self._store_system_prompt(conn, payload.get("system_prompt"))
+ self._insert_row(conn, "sessions", session, skip=frozenset())
+ for message in payload.get("messages") or []:
+ self._insert_row(conn, "messages", {**message, "session_id": session_id}, skip=_MESSAGE_MOVE_SKIP)
+ for usage in payload.get("usage") or []:
+ self._insert_row(conn, "session_model_usage", {**usage, "session_id": session_id}, skip=frozenset())
+ return "imported"
+ return self._execute_write(_do)
+
+ @staticmethod
+ def _insert_row(conn, table: str, values: Dict[str, Any], *, skip: frozenset) -> None:
+ columns = [c for c in _table_columns(conn, table) if c in values and c not in skip]
+ conn.execute(
+ f"INSERT INTO {table} ({', '.join(columns)}) VALUES ({', '.join('?' for _ in columns)})",
+ [values[c] for c in columns])
+
+ def delete_moved_session(self, session_id: str) -> bool:
+ """Remove a session this store no longer owns after the target confirmed it. Children left
+ behind are detached (same FK rule as :meth:`delete_session`); topic bindings on the row
+ cascade with it."""
+ def _do(conn) -> bool:
+ if conn.execute("SELECT 1 FROM sessions WHERE id = ?", (session_id,)).fetchone() is None:
+ return False
+ conn.execute("UPDATE sessions SET parent_session_id = NULL WHERE parent_session_id = ?", (session_id,))
+ conn.execute("DELETE FROM messages WHERE session_id = ?", (session_id,))
+ conn.execute("DELETE FROM session_model_usage WHERE session_id = ?", (session_id,))
+ conn.execute("DELETE FROM sessions WHERE id = ?", (session_id,))
+ self._delete_unreferenced_system_prompts(conn)
+ return True
+ return bool(self._execute_write(_do))
+
+ # ── routing index ─────────────────────────────────────────────────────────
+
+ def rekey_legacy_main_sessions(self, ids: Iterable[str], profile: str) -> int:
+ """``agent:main:…`` → ``agent::…`` (+ ``profile_name``) for rows a standalone
+ gateway wrote before this store's profile was multiplexed (#113884). Only rows still under
+ the legacy namespace change; a ``main``-named profile is written in its marked form."""
+ from gateway.session import _session_key_namespace
+ wanted = list(ids)
+ if not wanted:
+ return 0
+ new_ns = _session_key_namespace(profile) + ":"
+
+ def _do(conn) -> int:
+ changed = 0
+ for chunk in _id_chunks(wanted):
+ changed += conn.execute(
+ "UPDATE sessions SET session_key = ? || substr(session_key, 12), profile_name = ? "
+ f"WHERE id IN ({_session_ids_placeholders(chunk)}) AND substr(session_key, 1, 11) = 'agent:main:'",
+ (new_ns, profile, *chunk)).rowcount
+ return changed
+ return self._execute_write(_do)
+
+ def delete_gateway_routing_rows(self, rows: Iterable[Tuple[str, str]]) -> int:
+ wanted = list(rows)
+ if not wanted:
+ return 0
+
+ def _do(conn) -> int:
+ return sum(conn.execute(
+ "DELETE FROM gateway_routing WHERE scope = ? AND session_key = ?", (scope, key)).rowcount
+ for scope, key in wanted)
+ return self._execute_write(_do)
+
+ def insert_gateway_routing_rows_if_absent(self, rows: Iterable[Tuple[str, str, str, float]]) -> int:
+ """Adopt ``(scope, session_key, entry_json, updated_at)`` rows another store held for this
+ one's routing index. An existing key wins — the row the gateway actually loads stays."""
+ wanted = list(rows)
+ if not wanted:
+ return 0
+
+ def _do(conn) -> int:
+ return sum(conn.execute(
+ "INSERT OR IGNORE INTO gateway_routing (scope, session_key, entry_json, updated_at) "
+ "VALUES (?, ?, ?, ?)", row).rowcount for row in wanted)
+ return self._execute_write(_do)
+
+ # ── telegram topic tables ─────────────────────────────────────────────────
+
+ def relabel_telegram_topic_rows(self, rows: Iterable[Dict[str, Any]]) -> Dict[str, int]:
+ """Stamp the key's profile onto ``default``-labelled bindings (and the chat's mode row when
+ only the default one exists). A binding that would collide with one the named profile already
+ wrote is the stale duplicate and is removed."""
+ wanted = list(rows)
+ counts = {"bindings_relabelled": 0, "bindings_duplicates_removed": 0, "mode_rows_relabelled": 0}
+ if not wanted:
+ return counts
+
+ def _do(conn) -> Dict[str, int]:
+ mode_has_profile = "profile_name" in _table_columns(conn, "telegram_dm_topic_mode")
+ for row in wanted:
+ target = session_key_profile(row["session_key"])
+ if target in (None, "default"):
+ continue
+ chat_id, thread_id = row["chat_id"], row["thread_id"]
+ collides = conn.execute(
+ "SELECT 1 FROM telegram_dm_topic_bindings WHERE profile_name = ? AND chat_id = ? "
+ "AND thread_id = ?", (target, chat_id, thread_id)).fetchone()
+ if collides:
+ counts["bindings_duplicates_removed"] += conn.execute(
+ "DELETE FROM telegram_dm_topic_bindings WHERE profile_name = 'default' "
+ "AND chat_id = ? AND thread_id = ?", (chat_id, thread_id)).rowcount
+ else:
+ counts["bindings_relabelled"] += conn.execute(
+ "UPDATE telegram_dm_topic_bindings SET profile_name = ? WHERE profile_name = 'default' "
+ "AND chat_id = ? AND thread_id = ?", (target, chat_id, thread_id)).rowcount
+ if mode_has_profile and conn.execute(
+ "SELECT 1 FROM telegram_dm_topic_mode WHERE profile_name = ? AND chat_id = ?",
+ (target, chat_id)).fetchone() is None:
+ counts["mode_rows_relabelled"] += conn.execute(
+ "UPDATE telegram_dm_topic_mode SET profile_name = ? WHERE profile_name = 'default' "
+ "AND chat_id = ?", (target, chat_id)).rowcount
+ return counts
+ return self._execute_write(_do)
+
+ # ── evidence for JSON-file repairs ────────────────────────────────────────
+
+ def key_profiles_for_chat(self, platform: str, chat_id: str) -> Set[str]:
+ """Profiles whose keyed sessions hold *(platform, chat_id)* in this store — the evidence for
+ who owns a profile-less ``gateway_voice_mode.json`` entry."""
+ rows = self._read_all(
+ "SELECT DISTINCT session_key FROM sessions WHERE source = ? AND chat_id = ? "
+ "AND session_key LIKE 'agent:%'", (platform, str(chat_id)))
+ return {p for p in (session_key_profile(r["session_key"]) for r in rows) if p is not None}
diff --git a/tests/hermes_cli/test_sessions_repair_profiles.py b/tests/hermes_cli/test_sessions_repair_profiles.py
new file mode 100644
index 0000000000..c26d7917eb
--- /dev/null
+++ b/tests/hermes_cli/test_sessions_repair_profiles.py
@@ -0,0 +1,236 @@
+"""``hermes sessions repair-profiles``: the backward-looking half of the per-profile store model.
+
+The forward-only fixes (#88734 store routing, the #88381 inheritance fence, #76423 topic labels,
+#75198 voice keys) put NEW state in the right place; nothing settled what earlier releases left
+crossed. One fixture carries all six kinds of crossing; the contracts are that a dry run mutates
+nothing, that apply settles every repairable finding, and that a second apply finds nothing.
+Part of #88715.
+"""
+from __future__ import annotations
+
+import json
+import sqlite3
+from argparse import Namespace
+from pathlib import Path
+
+import pytest
+
+from hermes_state import SessionDB
+
+ROOT_KEY = "agent:main:telegram:dm:100"
+ACME_KEY = "agent:acme:telegram:dm:200"
+BETA_KEY = "agent:beta:telegram:dm:300"
+GHOST_KEY = "agent:ghost:telegram:dm:400"
+
+
+@pytest.fixture
+def homes(tmp_path, monkeypatch):
+ monkeypatch.setattr(Path, "home", lambda: tmp_path)
+ root = tmp_path / ".hermes"
+ root.mkdir()
+ monkeypatch.setenv("HERMES_HOME", str(root))
+ from hermes_cli.profiles import create_profile
+ for name in ("acme", "beta"):
+ create_profile(name, no_alias=True, no_skills=True)
+ return {"default": root, "acme": root / "profiles" / "acme", "beta": root / "profiles" / "beta"}
+
+
+def _session(db: SessionDB, sid: str, key: str, *, profile: str, parent=None, messages=2, chat_id=None,
+ source="telegram"):
+ db.create_session(sid, source, session_key=key, chat_id=chat_id or key.rsplit(":", 1)[-1],
+ chat_type="dm", profile_name=profile, parent_session_id=parent, system_prompt="sys")
+ for i in range(messages):
+ db.append_message(sid, "user" if i % 2 == 0 else "assistant", f"{sid} m{i}")
+
+
+def _seed_crossings(homes):
+ """Every defect the command knows, spread over the three stores."""
+ root = SessionDB(homes["default"] / "state.db")
+ acme = SessionDB(homes["acme"] / "state.db")
+ beta = SessionDB(homes["beta"] / "state.db")
+ scope = str((homes["default"] / "sessions").resolve())
+
+ # healthy controls — must survive untouched
+ _session(root, "ok-root", ROOT_KEY, profile="default")
+ _session(acme, "ok-acme", ACME_KEY, profile="acme")
+ root.save_gateway_routing_entry(ACME_KEY, json.dumps({"session_key": ACME_KEY, "session_id": "ok-acme"}), scope=scope)
+
+ # 1. label ≠ key namespace (row in the right store)
+ _session(acme, "mislabelled", "agent:acme:telegram:dm:201", profile="default")
+ # 2. wrong store: acme's key in the root store, with a compressed parent it points at (moves whole)
+ _session(root, "stray-parent", "agent:acme:telegram:dm:202", profile="acme", messages=3)
+ _session(root, "stray-child", "agent:acme:telegram:dm:202", profile="acme", parent="stray-parent")
+ # …and a legacy agent:main row inside acme's store (reported, not repaired by default)
+ _session(acme, "legacy-main", "agent:main:telegram:dm:203", profile="default")
+ # …and a row keyed to a profile that does not exist
+ _session(root, "ghost", GHOST_KEY, profile="ghost")
+ # 3. parent crossing namespaces inside beta's store: beta's own child forked from an acme row
+ # (which is itself a wrong-store row and moves out — the child must not follow a pointer
+ # into another profile's store)
+ _session(beta, "beta-parent", BETA_KEY, profile="beta")
+ _session(beta, "acme-in-beta", "agent:acme:telegram:dm:204", profile="acme")
+ _session(beta, "beta-child", "agent:beta:telegram:dm:301", profile="beta", parent="acme-in-beta")
+ # 4. routing rows: beta's index row copied into beta's store (#66887) + a ghost row in the root
+ beta.save_gateway_routing_entry(BETA_KEY, json.dumps({"session_key": BETA_KEY, "session_id": "beta-parent"}), scope=scope)
+ root.save_gateway_routing_entry(GHOST_KEY, json.dumps({"session_key": GHOST_KEY, "session_id": "ghost"}), scope=scope)
+ # 5. profile-less topic binding + voice-mode entry for a chat only acme's bot holds
+ acme.bind_telegram_topic(chat_id="200", thread_id="7", user_id="200", session_key=ACME_KEY,
+ session_id="ok-acme", profile_name="default")
+ (homes["default"] / "gateway_voice_mode.json").write_text(json.dumps({
+ "telegram:200": "all", # acme's chat, unprefixed
+ "telegram:100": "voice_only", # the default bot's chat — legitimately unprefixed
+ }))
+ # 6. sessions.json mirror entry for the ghost namespace
+ (homes["default"] / "sessions").mkdir(exist_ok=True)
+ (homes["default"] / "sessions" / "sessions.json").write_text(json.dumps({
+ "_README": "x", GHOST_KEY: {"session_key": GHOST_KEY, "session_id": "ghost"},
+ ROOT_KEY: {"session_key": ROOT_KEY, "session_id": "ok-root"},
+ }))
+ for db in (root, acme, beta):
+ db.close()
+
+
+def _dump(path: Path) -> dict:
+ """Every user table as sorted row tuples — the whole-file mutation oracle for the dry run."""
+ conn = sqlite3.connect(path)
+ try:
+ tables = [r[0] for r in conn.execute(
+ "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' "
+ "AND name NOT LIKE '%_fts%' ORDER BY name")]
+ return {t: sorted(map(repr, conn.execute(f"SELECT * FROM {t}").fetchall())) for t in tables}
+ finally:
+ conn.close()
+
+
+def _run(**kw) -> int:
+ from hermes_cli.sessions_cmd import cmd_sessions
+ args = Namespace(sessions_action="repair-profiles", apply=False, json=False, yes=True, legacy_main="report")
+ for k, v in kw.items():
+ setattr(args, k, v)
+ return cmd_sessions(args) or 0
+
+
+def _report() -> dict:
+ import io
+ import contextlib
+ buf = io.StringIO()
+ with contextlib.redirect_stdout(buf):
+ assert _run(json=True) == 0
+ return json.loads(buf.getvalue())
+
+
+def test_dry_run_names_every_crossing_and_changes_nothing(homes, capsys):
+ _seed_crossings(homes)
+ before = {name: _dump(home / "state.db") for name, home in homes.items()}
+ files_before = {p.name: p.read_text() for p in (homes["default"] / "gateway_voice_mode.json",
+ homes["default"] / "sessions" / "sessions.json")}
+
+ report = _report()
+
+ kinds = {(f["kind"], f["subject"]) for f in report["findings"]}
+ assert {
+ ("mislabelled", "mislabelled"),
+ ("wrong_store", "stray-parent"), ("wrong_store", "stray-child"),
+ ("legacy_main", "legacy-main"),
+ ("unclaimed_namespace", "ghost"),
+ ("wrong_store", "acme-in-beta"), ("crossed_parent", "beta-child"),
+ ("routing_unclaimed", f"{GHOST_KEY} (scope {str((homes['default'] / 'sessions').resolve())!r})"),
+ ("topic_profile_less", "chat 200 thread 7"),
+ ("voice_profile_less", "telegram:200"),
+ ("sessions_json_unclaimed", GHOST_KEY),
+ } <= kinds
+ # beta's own routing row: a standalone gateway's index lives in its own store; nothing recorded a
+ # multiplexing verdict here, so it is NOT reported as stray.
+ assert not any(f["kind"] == "routing_stray" for f in report["findings"])
+ # the healthy controls and the default bot's own voice key are not findings
+ assert not any(f["subject"] in {"ok-root", "ok-acme", "telegram:100", ROOT_KEY} for f in report["findings"])
+ # the two report-only kinds say why
+ by_kind = {f["kind"]: f for f in report["findings"]}
+ assert by_kind["legacy_main"]["action"] is None and "--legacy-main" in by_kind["legacy_main"]["reason"]
+ assert by_kind["unclaimed_namespace"]["action"] is None and "ghost" in by_kind["unclaimed_namespace"]["reason"]
+
+ assert {name: _dump(home / "state.db") for name, home in homes.items()} == before
+ assert {p.name: p.read_text() for p in (homes["default"] / "gateway_voice_mode.json",
+ homes["default"] / "sessions" / "sessions.json")} == files_before
+
+
+def test_apply_settles_every_repairable_crossing_and_is_idempotent(homes, monkeypatch, capsys):
+ _seed_crossings(homes)
+ snapshots = []
+ monkeypatch.setattr("hermes_cli.sessions_cmd_repair_profiles.default_snapshot",
+ lambda store: snapshots.append(store.profile) or f"snap-{store.profile}")
+
+ assert _run(apply=True) == 0
+ assert set(snapshots) == {"default", "acme", "beta"}
+
+ root = SessionDB(homes["default"] / "state.db")
+ acme = SessionDB(homes["acme"] / "state.db")
+ beta = SessionDB(homes["beta"] / "state.db")
+ try:
+ # 1. relabelled from the key
+ assert acme.get_session("mislabelled")["profile_name"] == "acme"
+ # 2. moved whole: parent link and every message intact in the target, gone from the source
+ assert root.get_session("stray-parent") is None and root.get_session("stray-child") is None
+ moved_child = acme.get_session("stray-child")
+ assert moved_child["profile_name"] == "acme" and moved_child["parent_session_id"] == "stray-parent"
+ assert len(acme.get_messages("stray-parent")) == 3 and len(acme.get_messages("stray-child")) == 2
+ assert acme.get_session("stray-parent")["system_prompt"] == "sys"
+ # legacy agent:main row and the ghost row are left exactly as they were
+ assert acme.get_session("legacy-main")["session_key"] == "agent:main:telegram:dm:203"
+ assert root.get_session("ghost")["session_key"] == GHOST_KEY
+ # 3. severed, own identity kept; the acme row it pointed at moved to acme's store
+ child = beta.get_session("beta-child")
+ assert child["parent_session_id"] is None and child["profile_name"] == "beta"
+ assert beta.get_session("acme-in-beta") is None and acme.get_session("acme-in-beta") is not None
+ # 4. ghost routing row dropped, acme's healthy row kept, beta's own row kept (standalone)
+ scope = str((homes["default"] / "sessions").resolve())
+ assert set(root.load_gateway_routing_entries(scope=scope)) == {ACME_KEY}
+ assert set(beta.load_gateway_routing_entries(scope=scope)) == {BETA_KEY}
+ # 5. topic binding carries acme; voice key prefixed only for acme's chat
+ assert acme.get_telegram_topic_binding(chat_id="200", thread_id="7", profile_name="acme") is not None
+ assert acme.get_telegram_topic_binding(chat_id="200", thread_id="7", profile_name="default") is None
+ voice = json.loads((homes["default"] / "gateway_voice_mode.json").read_text())
+ assert voice == {"acme:telegram:200": "all", "telegram:100": "voice_only"}
+ # 6. ghost mirror entry dropped, the default's kept
+ mirror = json.loads((homes["default"] / "sessions" / "sessions.json").read_text())
+ assert set(mirror) == {"_README", ROOT_KEY}
+ # controls untouched
+ assert root.get_session("ok-root")["profile_name"] == "default"
+ assert acme.get_session("ok-acme")["profile_name"] == "acme"
+ finally:
+ for db in (root, acme, beta):
+ db.close()
+
+ # idempotent: only the two report-only findings remain, and nothing is repairable
+ report = _report()
+ assert {f["kind"] for f in report["findings"]} == {"legacy_main", "unclaimed_namespace"}
+ assert report["repairable"] == 0
+
+
+def test_apply_refuses_while_a_gateway_owns_a_store(homes, monkeypatch, capsys):
+ _seed_crossings(homes)
+ monkeypatch.setattr("hermes_cli.sessions_cmd_repair_profiles.live_gateway_homes",
+ lambda stores: [("default", 4242)])
+ before = _dump(homes["default"] / "state.db")
+
+ assert _run(apply=True) == 1
+
+ assert "pid 4242" in capsys.readouterr().err
+ assert _dump(homes["default"] / "state.db") == before
+
+
+def test_legacy_main_rekey_adopts_a_standalone_gateways_history(homes, monkeypatch):
+ """The #113884 incident: multiplexing switched on, every existing key of the named profile's own
+ gateway stopped resolving. ``--legacy-main rekey`` gives them the namespace the multiplexer reads."""
+ _seed_crossings(homes)
+ monkeypatch.setattr("hermes_cli.sessions_cmd_repair_profiles.default_snapshot", lambda store: "snap")
+
+ assert _run(apply=True, legacy_main="rekey") == 0
+
+ acme = SessionDB(homes["acme"] / "state.db")
+ try:
+ row = acme.get_session("legacy-main")
+ assert row["session_key"] == "agent:acme:telegram:dm:203" and row["profile_name"] == "acme"
+ finally:
+ acme.close()
+ assert not any(f["kind"] == "legacy_main" for f in _report()["findings"])
diff --git a/website/docs/reference/cli-commands.md b/website/docs/reference/cli-commands.md
index f9a9089cf8..b4f012322c 100644
--- a/website/docs/reference/cli-commands.md
+++ b/website/docs/reference/cli-commands.md
@@ -1684,6 +1684,7 @@ Subcommands:
| `optimize-storage` | Migrate the full-text search index to the compact v23 external-content layout; on large databases this reclaims a large fraction of `state.db`. |
| `repair` | Repair a malformed `state.db` schema (e.g. `table messages_fts already exists`) so hidden sessions reappear; a backup is made first. |
| `repair-routing` | Re-attach gateway conversations stranded in session rows that lost their routing identity (a chat "jumping back in time" after a restart). Dry-run by default; `--apply` performs the adoptions (stop the gateway first); `--max-gap-seconds N` tunes the contiguity window. Only unambiguous cases are repaired. See [Sessions → Repair Stranded Gateway Sessions](../user-guide/sessions.md#repair-stranded-gateway-sessions). |
+| `repair-profiles` | Settle session, routing, Telegram-topic and voice-mode state that landed under the wrong profile (rows in another profile's store, labels disagreeing with the session key, parent links crossing profiles, index rows for deleted profiles). Dry-run by default; `--apply` performs the repairs after snapshotting every store (stop the gateway first); `--legacy-main rekey\|move` decides what `agent:main` rows inside a named profile's store are; `--json` for automation. See [Sessions → Repair State Crossed Between Profiles](../user-guide/sessions.md#repair-state-crossed-between-profiles). |
| `recover` | Offline, non-destructive recovery of a damaged `state.db` into a separate clean database. |
| `retitle-skills` | Regenerate titles for sessions opened with a `/skill`, using what the user actually typed; lists changes unless `--apply` is passed. |
diff --git a/website/docs/user-guide/sessions.md b/website/docs/user-guide/sessions.md
index d9f89fd0dc..d199aaf7a2 100644
--- a/website/docs/user-guide/sessions.md
+++ b/website/docs/user-guide/sessions.md
@@ -650,6 +650,52 @@ conversation stays readable via `/resume` and session search either way —
routing is the only thing the repair changes. Back up first
(`cp ~/.hermes/state.db ~/.hermes/state.db.bak`).
+### Repair State Crossed Between Profiles
+
+Every profile owns one `state.db`, and every gateway session key names the
+profile that owns the conversation (`agent:main:…` for the default profile,
+`agent::…` for a named one). Older releases could leave the two
+disagreeing — a named profile's rows written into the default store, a child
+session inheriting from another profile's row, a routing row copied into the
+wrong store, a Telegram topic or `/voice` setting saved without the bot's
+profile. Current versions put new state in the right place; `hermes sessions
+repair-profiles` settles what is already crossed.
+
+```bash
+# Report only — every store is scanned, nothing is written
+hermes sessions repair-profiles
+
+# Perform the repairs (stop the gateway first; a snapshot of every store is taken)
+hermes sessions repair-profiles --apply
+
+# Machine-readable report
+hermes sessions repair-profiles --json
+```
+
+What it finds and does:
+
+| Finding | Repair |
+|---|---|
+| `profile_name` disagrees with the row's own session key | relabel from the key |
+| rows sitting in another profile's store | move (with all messages) to the owning profile's store |
+| `parent_session_id` pointing at another profile's row | sever the link; the row's own identity is kept |
+| routing rows outside the default store (under multiplexing) | move to the default store; an existing row there wins |
+| routing rows / `sessions.json` entries for a profile that no longer exists | delete |
+| Telegram topic bindings and voice-mode entries missing their bot's profile | relabel from the sessions that hold the chat |
+
+Two cases are reported but never repaired without being told what they are:
+rows keyed to a profile that does not exist (create the profile, or
+`hermes profile migrate-identity `), and `agent:main:…` rows inside
+a named profile's store. The latter are either the history of a gateway that
+used to run standalone for that profile (`--legacy-main rekey` gives them the
+profile's namespace) or default-profile chats that leaked in under a scoped
+write (`--legacy-main move` sends them to the default store) — the rows
+themselves cannot tell the two apart.
+
+`--apply` refuses while a gateway owns any of the stores (it holds the routing
+index in memory and would write it back), and is safe to re-run: a second run
+finds nothing.
+
## Importing Sessions from Claude Code and Codex CLI