Load session data from disk. If the file isn't found in the agent-namespaced directory, checks the flat root directory for backward compatibility and auto-migrates. Args: session_id: Session ID to load Returns: SessionData or None if not fou
(self, session_id: str)
| 87 | return path |
| 88 | |
| 89 | def load(self, session_id: str) -> SessionData | None: |
| 90 | """Load session data from disk. |
| 91 | |
| 92 | If the file isn't found in the agent-namespaced directory, checks the |
| 93 | flat root directory for backward compatibility and auto-migrates. |
| 94 | |
| 95 | Args: |
| 96 | session_id: Session ID to load |
| 97 | |
| 98 | Returns: |
| 99 | SessionData or None if not found |
| 100 | """ |
| 101 | path = self._session_path(session_id) |
| 102 | if not path.exists(): |
| 103 | # Backward-compat: check flat root directory and migrate |
| 104 | migrated = self._migrate_from_root(session_id) |
| 105 | if migrated is None: |
| 106 | logger.warning(f"Session not found: {session_id}") |
| 107 | return None |
| 108 | path = migrated |
| 109 | |
| 110 | try: |
| 111 | raw = path.read_text(encoding="utf-8") |
| 112 | data: SessionData = SessionData.model_validate_json(raw) |
| 113 | return data |
| 114 | except Exception as e: |
| 115 | logger.error(f"Failed to load session {session_id}: {e}") |
| 116 | return None |
| 117 | |
| 118 | def _migrate_from_root(self, session_id: str) -> Path | None: |
| 119 | """Check flat root dir for a legacy session file and migrate it. |