fix(session): guard lossy migration by stored key

nanobot/session/manager.py

@@ -433,6 +433,24 @@ class SessionManager:
         """Legacy global session path (~/.nanobot/sessions/)."""
         return self.legacy_sessions_dir / f"{self.safe_key(key)}.jsonl"
 
+    @staticmethod
+    def _stored_key_for_path(path: Path) -> str | None:
+        """Read the stored session key from a JSONL metadata row, if present."""
+        try:
+            with open(path, encoding="utf-8") as f:
+                for line in f:
+                    line = line.strip()
+                    if not line:
+                        continue
+                    data = json.loads(line)
+                    if data.get("_type") == "metadata":
+                        stored_key = data.get("key")
+                        return stored_key if isinstance(stored_key, str) else None
+                    return None
+        except Exception:
+            return None
+        return None
+
     def get_or_create(self, key: str) -> Session:
         """
         Get an existing session or create a new one.
@@ -464,6 +482,15 @@ class SessionManager:
             for fallback_path, description in fallback_paths:
                 if not fallback_path.exists():
                     continue
+                stored_key = self._stored_key_for_path(fallback_path)
+                if stored_key and stored_key != key:
+                    logger.info(
+                        "Skipping migration for {} from {} because it belongs to {}",
+                        key,
+                        description,
+                        stored_key,
+                    )
+                    continue
                 try:
                     shutil.move(str(fallback_path), str(path))
                     logger.info("Migrated session {} from {}", key, description)

tests/agent/test_session_collision.py

@@ -93,6 +93,31 @@ def test_load_migrates_lossy_to_new_path(tmp_path: Path, monkeypatch) -> None:
     assert not lossy_path.exists()
 
 
+def test_load_does_not_migrate_lossy_path_for_different_stored_key(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    sm = _manager(tmp_path, monkeypatch)
+    first_key = "telegram:a_b"
+    second_key = "telegram:a:b"
+    lossy_path = sm._get_legacy_lossy_path(first_key)
+    assert lossy_path == sm._get_legacy_lossy_path(second_key)
+    _write_session_file(lossy_path, first_key, "belongs to first")
+
+    loaded_second = sm._load(second_key)
+
+    assert loaded_second is None
+    assert lossy_path.exists()
+    assert not sm._get_session_path(second_key).exists()
+
+    loaded_first = sm._load(first_key)
+
+    assert loaded_first is not None
+    assert loaded_first.messages[0]["content"] == "belongs to first"
+    assert sm._get_session_path(first_key).exists()
+    assert not lossy_path.exists()
+
+
 def test_safe_key_is_lossy() -> None:
     assert SessionManager.safe_key("telegram:a_b") == SessionManager.safe_key("telegram:a:b")