MCPcopy Create free account
hub / github.com/ScaleML/AgentSPEX / get

Method get

src/sandbox_vm/tools/session_management.py:259–282  ·  view source on GitHub ↗
(self, session_id: str)

Source from the content-addressed store, hash-verified

257 atexit.register(self.shutdown)
258
259 def get(self, session_id: str) -> SessionRuntime:
260 if not session_id:
261 raise KeyError("session_id_REQUIRED")
262 with self._lock:
263 rt = self._sessions.get(session_id)
264 if rt:
265 return rt
266 # Evict LRU if needed
267 if len(self._sessions) >= MAX_SESSIONS:
268 stalest_sid, stalest_rt = None, None
269 for sid, srt in self._sessions.items():
270 if stalest_rt is None or srt.last_used < stalest_rt.last_used:
271 stalest_sid, stalest_rt = sid, srt
272 if stalest_sid:
273 self._sessions.pop(stalest_sid, None)
274 try:
275 stalest_rt.close_sync()
276 except Exception:
277 pass
278 root = (SESSIONS_DIR / session_fs_key(session_id)).resolve()
279 rt = SessionRuntime(session_id=session_id, root=root)
280 rt.ensure_dirs()
281 self._sessions[session_id] = rt
282 return rt
283
284 def touch(self, session_id: str) -> None:
285 with self._lock:

Callers 15

_replay_nextMethod · 0.80
invokeMethod · 0.80
__init__Method · 0.80
_env_boolMethod · 0.80
_render_eventMethod · 0.80
_append_lineMethod · 0.80
runMethod · 0.80
host_workspaceFunction · 0.80
workspace_persistentFunction · 0.80
init_run_envFunction · 0.80

Calls 5

ensure_dirsMethod · 0.95
session_fs_keyFunction · 0.85
SessionRuntimeClass · 0.85
close_syncMethod · 0.80
resolveMethod · 0.80