From 8d74cb52dac6e5a181ba12936e48a183ed10ddc3 Mon Sep 17 00:00:00 2001 From: caya8205-2 Date: Mon, 31 Aug 2026 23:22:30 +0700 Subject: [PATCH] fix(gateway): give the routing index one store instead of the ambient one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second half of #66887. _entries is a single flat dict holding every profile's keys, so the index it persists to has to be a single file — but it was read and written through _db, which resolves whichever profile scope is active. A whole-index rewrite during one profile's turn copied every other profile's routing rows into that profile's store, and startup, which runs unscoped, then loaded a different copy than the last writer produced. That is why the startup recovery pass never sees a secondary profile's crash marker, which is the half this issue's title names. mark_turn_active() persists through the single-entry fast path (state.db only, no sessions.json mirror), so a marker written during a profile's turn landed in that profile's store and _recover_unclean_sessions(), running with no scope, read a store that had never heard of it. The turn was silently never promoted to resume_pending. Capture the gateway's own home at construction — the store is built at startup before any profile scope exists — and route the index through it: _ensure_loaded_locked, _reconcile_recovered_routing_locked, _persist_routing_data and _save_entry now use _routing_db. A pinned handle still wins, so suites that install a fake or disable the DB are unaffected. _prune_stale_sessions_locked is the mixed case and is split accordingly: it now asks _db_for_key(key) whether each session ended, because that is a per-session question, while the index write stays on the single store. One ambient handle previously answered it for every profile at once, which could prune a live secondary-profile route on the strength of the root store's copy of that session. Regression as requested on the issue: mark a turn active under a secondary profile's scope, then build a fresh store with no scope and run recover_interrupted_turns(). It promotes exactly one turn to resume_pending here and promotes zero against the previous behaviour. Co-Authored-By: Claude Opus 5 --- gateway/session.py | 57 +++++++++++++++++-- ...test_multiplex_session_db_profile_scope.py | 41 +++++++++++++ 2 files changed, 92 insertions(+), 6 deletions(-) diff --git a/gateway/session.py b/gateway/session.py index 5a6dc77d86..78b4efa4ff 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -1336,6 +1336,17 @@ class SessionStore: handles=self._db_handles, lock=self._db_handles_lock, ) + # The routing index is one process-wide structure keyed by + # ``agent::…``, not a per-profile one, so it needs exactly one + # home for its lifetime. The store is constructed at startup under the + # gateway's own home, before any profile scope exists, so capturing it + # here is what makes the index deterministic — see ``_routing_db``. + try: + from hermes_constants import get_hermes_home + + self._routing_home: Optional[Path] = Path(get_hermes_home()) + except Exception: + self._routing_home = None self._open_session_db_for_active_scope() def _open_session_db_for_active_scope(self, db_path: Optional[Path] = None): @@ -1405,6 +1416,33 @@ class SessionStore: def _db(self, value) -> None: self._db_pinned = value + @property + def _routing_db(self): + """The one store that owns the routing index, whatever scope is active. + + ``_entries`` is a single flat dict holding every profile's keys, so the + index it persists to has to be a single file too. Reading it through + ``_db`` made that file whichever profile happened to be scoped at the + time: a whole-index rewrite during one profile's turn copied every + other profile's routing rows into that profile's store, and startup — + which runs unscoped — then loaded a different copy than the one the + last writer produced. That is why a crash marker written while a + secondary profile was active is invisible to the startup recovery pass + (#66887). + + A pinned handle still wins, so the suites that install a fake or + disable the DB keep working unchanged. + """ + if self._db_pinned is not _DB_UNPINNED: + return self._db_pinned + home = getattr(self, "_routing_home", None) + if home is None: + return self._db + try: + return self._open_session_db_for_active_scope(db_path=home / "state.db") + except Exception: + return None + def _named_profile_for_key(self, session_key: Optional[str]) -> Optional[str]: """The non-default profile that owns *session_key*, or None. @@ -1618,7 +1656,7 @@ class SessionStore: # _prune_stale_sessions_locked). db_had_entries = False db_load_succeeded = False - _db = getattr(self, "_db", None) + _db = self._routing_db if _db: loader = getattr(_db, "load_gateway_routing_entries", None) if callable(loader): @@ -1710,14 +1748,21 @@ class SessionStore: legacy) are left alone, and a ``None`` DB handle (SQLite unavailable) is a no-op. DB errors are non-fatal — startup must never fail here. """ - db = getattr(self, "_db", None) - if not db or not self._entries: + if not self._entries: return stale_keys: list = [] recovered_keys = 0 try: for key, entry in self._entries.items(): + # Whether a session ended is a per-session question, so ask the + # store that owns the key. A single ambient handle answered it + # for every profile at once, which is how a live secondary + # profile session could be pruned on the strength of the root + # store's copy of it. + db = self._db_for_key(key) + if db is None: + continue row = db.get_session(entry.session_id) # row is None -> not in DB (legacy / pre-SQLite) — keep # end_reason is None -> session alive — keep @@ -1826,7 +1871,7 @@ class SessionStore: if getattr(self, "_routing_db_loaded", False) or baseline is None: return - db = getattr(self, "_db", None) + db = self._routing_db loader = getattr(db, "load_gateway_routing_entries", None) if db else None if not callable(loader): return @@ -1891,7 +1936,7 @@ class SessionStore: if revision > generation: data[key] = json.loads(entry_json) db_saved = False - _db = getattr(self, "_db", None) + _db = self._routing_db if _db: replacer = getattr(_db, "replace_gateway_routing_entries", None) if callable(replacer): @@ -2049,7 +2094,7 @@ class SessionStore: if captured is None: return entry_json, revision, candidate_entry = captured - _db = getattr(self, "_db", None) + _db = self._routing_db saver = getattr(_db, "save_gateway_routing_entry", None) if _db else None if callable(saver): save_lock = getattr(self, "_save_lock", None) diff --git a/tests/gateway/test_multiplex_session_db_profile_scope.py b/tests/gateway/test_multiplex_session_db_profile_scope.py index 1f60768f9e..10833311b3 100644 --- a/tests/gateway/test_multiplex_session_db_profile_scope.py +++ b/tests/gateway/test_multiplex_session_db_profile_scope.py @@ -738,3 +738,44 @@ def test_compression_child_write_stays_in_the_parents_profile_store(multiplex_ho # 4. root was never touched assert _session_ids(root / "state.db") == set() + + +def _restarted_store(root: Path) -> SessionStore: + """A store that really loads its index — a restart, not a primed fixture.""" + return SessionStore( + sessions_dir=root / "sessions", + config=GatewayConfig(multiplex_profiles=True), + ) + + +def test_crash_marker_from_a_secondary_profile_survives_restart(multiplex_homes): + """Startup recovery must see turn markers written under any profile. + + ``mark_turn_active`` persists through the routing index's single-entry + fast path (state.db only, no sessions.json mirror), and + ``recover_interrupted_turns`` reads that index at startup with no profile + scope installed. While the index followed the ambient store, a marker + written during a secondary profile's turn landed in that profile's + state.db and the unscoped startup pass never saw it, so the interrupted + turn was never promoted to ``resume_pending`` — the recovery half of + #66887, and the one this issue's title names. + """ + root, profile = multiplex_homes + store = _multiplex_store(root) + + scope = set_hermes_home_override(str(profile)) + try: + entry = store.get_or_create_session(_profile_source()) + assert store.mark_turn_active(entry.session_key) is not None + finally: + reset_hermes_home_override(scope) + + # Restart: fresh store, fresh index, no profile scope anywhere. + restarted = _restarted_store(root) + promoted = restarted.recover_interrupted_turns(max_age_seconds=3600) + + assert promoted == 1 + recovered = restarted._entries[entry.session_key] + assert recovered.resume_pending is True + assert recovered.resume_reason == "restart_interrupted" + assert recovered.active_turn_token is None