Create a new session and return it.
(
self,
repo_path: str,
output_dir: str,
components: Dict[str, Node],
leaf_nodes: List[str],
workspace: Optional[SessionWorkspace] = None,
)
| 66 | self._lock = threading.Lock() |
| 67 | |
| 68 | def create( |
| 69 | self, |
| 70 | repo_path: str, |
| 71 | output_dir: str, |
| 72 | components: Dict[str, Node], |
| 73 | leaf_nodes: List[str], |
| 74 | workspace: Optional[SessionWorkspace] = None, |
| 75 | ) -> SessionState: |
| 76 | """Create a new session and return it.""" |
| 77 | with self._lock: |
| 78 | self._purge_expired_locked() |
| 79 | # Evict oldest if at capacity |
| 80 | if len(self._sessions) >= _MAX_SESSIONS: |
| 81 | oldest_id = min( |
| 82 | self._sessions, |
| 83 | key=lambda sid: self._sessions[sid].last_accessed, |
| 84 | ) |
| 85 | evicted = self._sessions[oldest_id] |
| 86 | if evicted.workspace is not None: |
| 87 | evicted.workspace.cleanup() |
| 88 | del self._sessions[oldest_id] |
| 89 | session_id = uuid.uuid4().hex[:12] |
| 90 | # Ensure no collision |
| 91 | while session_id in self._sessions: |
| 92 | session_id = uuid.uuid4().hex[:12] |
| 93 | state = SessionState( |
| 94 | session_id=session_id, |
| 95 | repo_path=repo_path, |
| 96 | output_dir=output_dir, |
| 97 | components=components, |
| 98 | leaf_nodes=leaf_nodes, |
| 99 | workspace=workspace, |
| 100 | ) |
| 101 | self._sessions[session_id] = state |
| 102 | return state |
| 103 | |
| 104 | def get(self, session_id: str) -> Optional[SessionState]: |
| 105 | """Return the session or ``None`` if not found / expired.""" |