fix(gateway): adopt stranded bot sessions from the default store on profile resume

Pre-#93296, the desktop routed session RPCs by the focused tile, so a
profile bot's turns executed on the default backend and its canonical
session accumulated in the DEFAULT profile's state.db. Post-fix, the
profile backend correctly receives the resume — but its store has never
seen the session, so the same chat 4001s forever (unreachable instead
of misrouted). Live repro: Teknium's Developer bot, session c93770.

- hermes_state_portability: SessionDB.adopt_session_lineage_from() —
  composes the existing export_session_lineage()/import_sessions()
  primitives; donor rows are archived (never deleted) with
  end_reason=adopted_by_profile, which is deliberately NOT in
  RECOVERABLE_END_REASONS so canonical-lookup resurrection cannot undo
  an adoption. Idempotent (already-present ids skip).
- tui_gateway/methods_session: profile-scoped session.resume falls back
  to adoption from the default store right before the 4007; ids unknown
  to BOTH stores still 4007 exactly as before, and launch-profile
  resumes never consult the fallback.
- tests: 10 new (7 unit on the primitive incl. compression-lineage
  unit adoption + non-resurrectable archive; 3 handler-level through
  server.handle_request incl. the live repro shape); db-ownership
  leak test taught that the shared launch handle probe is by design.

Follow-up to #93296/#93311; part of #93091.

hermes_state_portability.py

@@ -9,8 +9,10 @@ module-level constants live in hermes_state_common.
 """
 
 import logging
+import contextlib
 import json
 import time
+from pathlib import Path
 from typing import Any, Dict, List, Optional
 
 from agent.skill_commands import SKILL_SCAFFOLD_SQL_LIKE
@@ -306,6 +308,74 @@ class SessionPortabilityMixin:
             results.append({**session, "messages": messages})
         return results
 
+    def adopt_session_lineage_from(
+        self,
+        donor_db: Any,  # a full SessionDB (mixin cannot import it — cycle)
+        session_id: str,
+        *,
+        retire_donor: bool = True,
+    ) -> Dict[str, Any]:
+        """Adopt *session_id*'s full compression lineage from *donor_db* into
+        this store.
+
+        The stranded-bot-session heal (#93091 follow-up to #93296): before the
+        desktop routed session RPCs by their target session, a profile bot's
+        turns executed on whichever backend held window focus — usually the
+        default one — so the bot's canonical session rows and messages
+        accumulated in the DEFAULT profile's state.db. Once routing was fixed,
+        the profile backend correctly received the RPCs but had no such
+        session, so the same chat 4001'd for the opposite reason. This method
+        moves the conversation to where routing now looks for it.
+
+        Composition of existing primitives (no new import/export machinery):
+        ``donor_db.export_session_lineage()`` -> ``self.import_sessions()``.
+        Import semantics apply unchanged: gateway routing, handoff, and live
+        activity fields are reset; already-present ids are skipped
+        (idempotent re-adoption after a partial run).
+
+        When ``retire_donor`` is True and at least one segment was imported
+        (or every segment already exists here), the donor rows are ARCHIVED —
+        never deleted — with ``end_reason='adopted_by_profile'`` so the
+        default profile's list stops advertising a conversation that now
+        lives elsewhere, while the bytes stay recoverable. The archive is
+        deliberately NOT in the recoverable set (agent_close/ws_orphan_reap):
+        canonical-lookup resurrection must not undo an adoption.
+
+        Returns the ``import_sessions`` result dict, plus ``adopted`` (bool)
+        and ``donor_retired`` (bool).
+        """
+        payload = donor_db.export_session_lineage(session_id)
+        if not payload:
+            return {
+                "ok": False,
+                "adopted": False,
+                "donor_retired": False,
+                "error": f"session {session_id!r} not found in donor store",
+            }
+
+        segments = payload.get("segments") or [payload]
+        result = self.import_sessions([dict(seg) for seg in segments])
+        imported = int(result.get("imported") or 0)
+        skipped = int(result.get("skipped") or 0)
+        adopted = result.get("ok", False) and (imported + skipped) == len(segments)
+
+        donor_retired = False
+        if adopted and retire_donor:
+            for seg in segments:
+                seg_id = seg.get("id")
+                if not seg_id:
+                    continue
+                with contextlib.suppress(Exception):
+                    # First end_reason wins in end_session(); reopen first so
+                    # the adoption boundary is stamped even on ended segments
+                    # (e.g. 'compression' parents).
+                    donor_db.reopen_session(seg_id)
+                    donor_db.end_session(seg_id, "adopted_by_profile")
+                    donor_db.set_session_archived(seg_id, True)
+            donor_retired = True
+
+        return {**result, "adopted": adopted, "donor_retired": donor_retired}
+
     @staticmethod
     def _import_text_or_none(value: Any, field: str) -> Optional[str]:
         if value is None:

tests/tui_gateway/test_session_resume_db_ownership.py

@@ -117,12 +117,21 @@ def _resume(**params):
 
 
 def test_resume_closes_profile_db_when_session_not_found(profile_dbs):
-    """The 'session not found' early return must not leak the handle."""
+    """The 'session not found' early return must not leak the handle.
+
+    The stranded-session adoption fallback (#93296 follow-up) may lazily
+    construct the SHARED launch handle via ``_get_db()`` while probing the
+    default store for a donor row; that handle carries ``db_path=None`` and
+    is never closed by design (see module docstring). Only the dedicated
+    profile-scoped open (``db_path=<profile>/state.db``) is the caller's to
+    close, so the leak assertion filters to path-scoped opens.
+    """
     resp = _resume(session_id="missing", profile="work")
 
     assert resp["error"]["code"] == 4007
-    assert len(profile_dbs) == 1
-    assert profile_dbs[0].closed == 1
+    scoped = [db for db in profile_dbs if db.db_path is not None]
+    assert len(scoped) == 1
+    assert scoped[0].closed == 1
 
 
 def test_resume_closes_profile_db_when_reopen_fails(profile_dbs, monkeypatch):

tests/tui_gateway/test_stranded_session_adoption.py

@@ -0,0 +1,264 @@
+"""Stranded bot-session adoption (#93296 follow-up).
+
+Pre-#93296 misrouting accumulated profile-bot sessions in the DEFAULT
+profile's state.db. Post-fix, profile-scoped resumes correctly target the
+profile's own store — which never saw the session — so the same chat 4001'd
+for the opposite reason. These tests pin the heal: the unit-level adoption
+primitive (SessionDB.adopt_session_lineage_from) and its invariants.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from hermes_state import SessionDB
+
+STRANDED_ID = "20260823_043331_c93770"
+
+
+@pytest.fixture()
+def stores(tmp_path):
+    default_db = SessionDB(db_path=tmp_path / "state.db")
+    profile_home = tmp_path / "profiles" / "developer"
+    profile_home.mkdir(parents=True)
+    profile_db = SessionDB(db_path=profile_home / "state.db")
+    yield default_db, profile_db
+    default_db.close()
+    profile_db.close()
+
+
+def _seed_stranded(db, session_id=STRANDED_ID, turns=3, title="Bot Chat", **kwargs):
+    db.create_session(session_id, source="tui", **kwargs)
+    db.set_session_title(session_id, title)
+    for i in range(1, turns + 1):
+        db.append_message(session_id, "user", f"question {i}")
+        db.append_message(session_id, "assistant", f"answer {i}")
+
+
+def test_adoption_moves_session_and_messages(stores):
+    default_db, profile_db = stores
+    _seed_stranded(default_db)
+
+    result = profile_db.adopt_session_lineage_from(default_db, STRANDED_ID)
+
+    assert result["adopted"] is True
+    assert result["imported"] == 1
+    row = profile_db.get_session(STRANDED_ID)
+    assert row is not None
+    msgs = profile_db.get_messages(STRANDED_ID)
+    assert len(msgs) == 6
+    assert msgs[0]["content"] == "question 1"
+    assert msgs[-1]["content"] == "answer 3"
+
+
+def test_donor_is_archived_not_deleted(stores):
+    default_db, profile_db = stores
+    _seed_stranded(default_db)
+
+    profile_db.adopt_session_lineage_from(default_db, STRANDED_ID)
+
+    donor = default_db.get_session(STRANDED_ID)
+    assert donor is not None, "donor row must survive (archived, never deleted)"
+    assert donor["archived"]
+    assert donor["end_reason"] == "adopted_by_profile"
+    # bytes stay recoverable
+    assert len(default_db.get_messages(STRANDED_ID)) == 6
+
+
+def test_adoption_archive_is_not_recoverable_resurrectable(stores):
+    """Canonical-lookup resurrection must NOT undo an adoption."""
+    default_db, profile_db = stores
+    _seed_stranded(default_db)
+    profile_db.adopt_session_lineage_from(default_db, STRANDED_ID)
+
+    assert "adopted_by_profile" not in SessionDB.RECOVERABLE_END_REASONS
+    assert default_db.unarchive_recoverable_session(STRANDED_ID) is False
+    assert default_db.get_session(STRANDED_ID)["archived"]
+
+
+def test_adoption_is_idempotent(stores):
+    default_db, profile_db = stores
+    _seed_stranded(default_db)
+
+    first = profile_db.adopt_session_lineage_from(default_db, STRANDED_ID)
+    second = profile_db.adopt_session_lineage_from(default_db, STRANDED_ID)
+
+    assert first["adopted"] and second["adopted"]
+    assert second["imported"] == 0 and second["skipped"] == 1
+    assert len(profile_db.get_messages(STRANDED_ID)) == 6
+
+
+def test_missing_donor_session_is_reported_not_raised(stores):
+    default_db, profile_db = stores
+    result = profile_db.adopt_session_lineage_from(default_db, "nope")
+    assert result["adopted"] is False
+    assert "not found" in result["error"]
+
+
+def test_compression_lineage_adopts_as_a_unit(stores):
+    """A compacted conversation is parent(end_reason=compression) -> child.
+
+    Adoption must carry BOTH segments so the profile store can follow the
+    continuation chain, and must retire both donor rows.
+    """
+    default_db, profile_db = stores
+    parent, child = "sess-parent", "sess-child"
+    _seed_stranded(default_db, session_id=parent, turns=2)
+    default_db.end_session(parent, "compression")
+    default_db.create_session(child, source="tui", parent_session_id=parent)
+    default_db.set_session_title(child, "Bot Chat")
+    default_db.append_message(child, "user", "post-compaction question")
+    default_db.append_message(child, "assistant", "post-compaction answer")
+
+    result = profile_db.adopt_session_lineage_from(default_db, parent)
+
+    assert result["adopted"] is True
+    assert result["imported"] == 2
+    assert profile_db.get_session(parent) is not None
+    assert profile_db.get_session(child) is not None
+    assert profile_db.get_session(child)["parent_session_id"] == parent
+    for sid in (parent, child):
+        donor = default_db.get_session(sid)
+        assert donor["archived"], f"{sid} must be retired in donor store"
+
+
+def test_adoption_does_not_touch_unrelated_sessions(stores):
+    default_db, profile_db = stores
+    _seed_stranded(default_db)
+    _seed_stranded(default_db, session_id="other-session", title="Other Chat")
+
+    profile_db.adopt_session_lineage_from(default_db, STRANDED_ID)
+
+    other = default_db.get_session("other-session")
+    assert not other["archived"]
+    assert profile_db.get_session("other-session") is None
+
+
+# -------------------------------------------------------------------------
+# Handler-level: the real session.resume JSON-RPC path (server.handle_request)
+# -------------------------------------------------------------------------
+#
+# Mirrors tests/tui_gateway/test_session_profile_db.py's harness: import the
+# real server (with env_loader/banner mocked at first import), wire _get_db()
+# to the default store, and drive session.resume with profile= + lazy=True so
+# the resume registers a live record WITHOUT building an agent.
+
+import importlib
+from unittest.mock import MagicMock, patch
+
+
+@pytest.fixture()
+def gateway(tmp_path, monkeypatch):
+    from pathlib import Path as _P
+
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setattr(_P, "home", lambda: tmp_path)
+    monkeypatch.setenv("HERMES_HOME", str(home))
+
+    with patch.dict(
+        "sys.modules",
+        {
+            "hermes_cli.env_loader": MagicMock(),
+            "hermes_cli.banner": MagicMock(),
+        },
+    ):
+        mod = importlib.import_module("tui_gateway.server")
+
+    methods = dict(mod._methods)
+
+    default_db = SessionDB(db_path=home / "state.db")
+    mod._db = default_db
+
+    profile_home = home / "profiles" / "developer"
+    profile_home.mkdir(parents=True)
+
+    # session.resume resolves the profile via hermes_cli.profiles
+    monkeypatch.setattr(
+        "hermes_cli.profiles.get_profile_dir", lambda name: str(profile_home)
+    )
+
+    yield mod, default_db, profile_home
+
+    mod._methods.clear()
+    mod._methods.update(methods)
+    mod._sessions.clear()
+    mod._pending.clear()
+    mod._answers.clear()
+    mod._db = None
+    default_db.close()
+
+
+def test_profile_resume_adopts_stranded_default_store_session(gateway):
+    """The live repro: resume a session by id on a profile whose store has
+    never seen it, while the id exists in the default store. Pre-heal this
+    was a hard 4007; now the lineage is adopted and the resume succeeds."""
+    mod, default_db, profile_home = gateway
+    _seed_stranded(default_db)
+
+    resp = mod.handle_request(
+        {
+            "id": "1",
+            "method": "session.resume",
+            "params": {
+                "session_id": STRANDED_ID,
+                "profile": "developer",
+                "lazy": True,
+            },
+        }
+    )
+
+    assert not resp.get("error"), f"resume failed: {resp.get('error')}"
+    result = resp["result"]
+    assert result.get("resumed") == STRANDED_ID
+    assert result.get("message_count") == 6
+
+    # Durably adopted into the profile's own store...
+    pdb = SessionDB(db_path=profile_home / "state.db")
+    try:
+        assert pdb.get_session(STRANDED_ID) is not None
+        assert len(pdb.get_messages(STRANDED_ID)) == 6
+    finally:
+        pdb.close()
+    # ...and retired (archived, never deleted) in the default store.
+    donor = default_db.get_session(STRANDED_ID)
+    assert donor["archived"]
+    assert donor["end_reason"] == "adopted_by_profile"
+
+
+def test_profile_resume_of_truly_unknown_session_still_4007s(gateway):
+    """Adoption must not weaken the not-found contract: an id in NEITHER
+    store keeps failing with 4007 exactly as before."""
+    mod, _default_db, _profile_home = gateway
+
+    resp = mod.handle_request(
+        {
+            "id": "2",
+            "method": "session.resume",
+            "params": {
+                "session_id": "definitely-not-anywhere",
+                "profile": "developer",
+                "lazy": True,
+            },
+        }
+    )
+
+    assert resp.get("error")
+    assert resp["error"]["code"] == 4007
+
+
+def test_launch_profile_resume_path_is_untouched(gateway):
+    """A resume WITHOUT profile scope (owns_db=False) never consults the
+    adoption fallback — unknown ids fail 4007 on the shared handle."""
+    mod, _default_db, _profile_home = gateway
+
+    resp = mod.handle_request(
+        {
+            "id": "3",
+            "method": "session.resume",
+            "params": {"session_id": "unknown-launch-id", "lazy": True},
+        }
+    )
+
+    assert resp.get("error")
+    assert resp["error"]["code"] == 4007

tui_gateway/methods_session.py

@@ -483,7 +483,49 @@ def _(rid, params: dict) -> dict:
                             },
                         },
                     )
-                return _err(rid, 4007, "session not found")
+
+                # Stranded-session adoption (#93296 follow-up): before session
+                # RPCs routed by their TARGET session, a profile bot's turns
+                # executed on the focused tile's backend — usually default —
+                # so its canonical session accumulated in the DEFAULT
+                # profile's state.db. Now that routing is correct, this
+                # profile-scoped resume is the first place the fix and the
+                # stranded data collide: the id exists in the default store
+                # but not here, and without adoption the same chat 4001s
+                # forever (the fix made it unreachable instead of misrouted).
+                # Adopt the full lineage from the default store into this
+                # profile's db, then retry the lookup. Only profile-scoped
+                # resumes reach here (owns_db); unknown ids in the default
+                # store still 4007 exactly as before.
+                if owns_db:
+                    try:
+                        default_db = _get_db()
+                        donor_row = (
+                            default_db.get_session(target)
+                            or default_db.get_session_by_title(target)
+                        ) if default_db is not None else None
+                        if donor_row:
+                            adoption = db.adopt_session_lineage_from(
+                                default_db, donor_row["id"]
+                            )
+                            if adoption.get("adopted"):
+                                logger.info(
+                                    "adopted stranded session %s (lineage of %s "
+                                    "segment(s)) from default store into profile %s",
+                                    donor_row["id"],
+                                    len(adoption.get("imported_ids") or [])
+                                    + len(adoption.get("skipped_ids") or []),
+                                    profile or "?",
+                                )
+                                found = db.get_session(donor_row["id"])
+                                if found:
+                                    target = found["id"]
+                    except Exception:
+                        logger.exception(
+                            "stranded-session adoption failed for %s", target
+                        )
+                if not found:
+                    return _err(rid, 4007, "session not found")
 
         # Follow the compression-continuation chain to the live tip so a resume on
         # a rotated-out parent id binds to the descendant that actually holds the