fix(gateway): restore gateway sessions from JSON snapshots

gateway/session.py

@@ -1297,17 +1297,41 @@ def rewrite_transcript(self, session_id: str, messages: List[Dict[str, Any]]) ->
     def load_transcript(self, session_id: str) -> List[Dict[str, Any]]:
         """Load all messages from a session's transcript.
 
-        state.db is the canonical store. The legacy JSONL fallback was removed
-        in spec 002 — pre-DB sessions on existing disks have already been
-        migrated (their DB row holds the full message history).
+        state.db is the canonical store.  The optional per-session JSON
+        snapshot is a fail-open replay source for installs where the DB row is
+        empty/unavailable but the same session's snapshot is intact.
         """
-        if not self._db:
+        if not session_id:
             return []
+        if self._db:
+            try:
+                messages = self._db.get_messages_as_conversation(session_id)
+                if messages:
+                    return messages
+            except Exception as e:
+                logger.debug("Could not load messages from DB: %s", e)
+        return self._load_transcript_snapshot(session_id)
+
+    def _load_transcript_snapshot(self, session_id: str) -> List[Dict[str, Any]]:
+        """Load a same-session JSON snapshot when SQLite has no replay rows."""
+        snapshot_path = self.sessions_dir / f"session_{session_id}.json"
         try:
-            return self._db.get_messages_as_conversation(session_id)
+            data = json.loads(snapshot_path.read_text(encoding="utf-8"))
         except Exception as e:
-            logger.debug("Could not load messages from DB: %s", e)
+            logger.debug("Could not load session snapshot %s: %s", snapshot_path, e)
+            return []
+        if isinstance(data, dict):
+            messages = data.get("messages", [])
+        elif isinstance(data, list):
+            messages = data
+        else:
+            return []
+        if not isinstance(messages, list):
             return []
+        return [
+            msg for msg in messages
+            if isinstance(msg, dict) and msg.get("role")
+        ]
 
     def rewind_session(self, session_id: str, n: int = 1) -> Optional[Dict[str, Any]]:
         """Back up ``n`` user turns via soft-delete, keeping rows for audit.

tests/gateway/test_session.py

@@ -545,18 +545,18 @@ def test_rewrite_with_empty_list(self, store):
         assert reloaded == []
 
 
-class TestLoadTranscriptDBOnly:
-    """After spec 002, load_transcript reads only from state.db."""
+class TestLoadTranscript:
+    """Transcript replay prefers state.db and can salvage same-session snapshots."""
 
-    def test_db_only_returns_empty_for_nonexistent(self, tmp_path, monkeypatch):
+    def test_returns_empty_for_nonexistent(self, tmp_path, monkeypatch):
         import hermes_state
         monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db")
         config = GatewayConfig()
         store = SessionStore(sessions_dir=tmp_path, config=config)
         result = store.load_transcript("nonexistent")
         assert result == []
 
-    def test_db_only_returns_messages(self, tmp_path, monkeypatch):
+    def test_returns_db_messages(self, tmp_path, monkeypatch):
         import hermes_state
         monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db")
         config = GatewayConfig()
@@ -571,6 +571,58 @@ def test_db_only_returns_messages(self, tmp_path, monkeypatch):
         assert result[0]["content"] == "db-q"
         assert result[1]["content"] == "db-a"
 
+    def test_db_messages_win_over_json_snapshot(self, tmp_path, monkeypatch):
+        import hermes_state
+        monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db")
+        config = GatewayConfig()
+        store = SessionStore(sessions_dir=tmp_path, config=config)
+        sid = "db_primary_session"
+        store._db.create_session(session_id=sid, source="gateway", model="m")
+        store._db.append_message(session_id=sid, role="user", content="db-q")
+        store._db.append_message(session_id=sid, role="assistant", content="db-a")
+        (tmp_path / f"session_{sid}.json").write_text(
+            json.dumps({
+                "session_id": sid,
+                "messages": [
+                    {"role": "user", "content": "snapshot-q"},
+                    {"role": "assistant", "content": "snapshot-a"},
+                ],
+            }),
+            encoding="utf-8",
+        )
+
+        result = store.load_transcript(sid)
+
+        assert [msg["content"] for msg in result] == ["db-q", "db-a"]
+
+    def test_loads_json_snapshot_when_db_history_empty(self, tmp_path, monkeypatch):
+        import hermes_state
+        monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db")
+        config = GatewayConfig()
+        store = SessionStore(sessions_dir=tmp_path, config=config)
+        sid = "snapshot_salvage_session"
+        store._db.create_session(session_id=sid, source="gateway", model="m")
+        (tmp_path / f"session_{sid}.json").write_text(
+            json.dumps({
+                "session_id": sid,
+                "messages": [
+                    {"role": "system", "content": "ignored by gateway replay later"},
+                    {"role": "user", "content": "snapshot-q"},
+                    {"role": "assistant", "content": "snapshot-a"},
+                    {"content": "missing role"},
+                ],
+            }),
+            encoding="utf-8",
+        )
+
+        result = store.load_transcript(sid)
+
+        assert [msg["content"] for msg in result] == [
+            "ignored by gateway replay later",
+            "snapshot-q",
+            "snapshot-a",
+        ]
+
 
 class TestSessionStoreSwitchSession:
     """Regression coverage for gateway /resume session switching semantics."""