import os, json, uuid from pathlib import Path WORKSPACE = "/workspace" HF_LOCAL_DIR = Path(WORKSPACE) / "hf" SESSIONS_DIR = HF_LOCAL_DIR / "sessions" SESSIONS_DIR.mkdir(parents=True, exist_ok=True) class SessionManager: def __init__(self): self.current_session = None # ------------------ CREATE / LOAD ------------------ def create_session(self, session_id=None): session_id = session_id or f"sess_{uuid.uuid4().hex[:8]}" session_path = SESSIONS_DIR / session_id session_path.mkdir(parents=True, exist_ok=True) self.current_session = session_id # initialize files only if not exist workspace_file = session_path / "workspace.json" history_file = session_path / "history.json" if not workspace_file.exists(): self._write_json("workspace.json", {}) if not history_file.exists(): self._write_json("history.json", []) print(f"[SESSION] Created {session_id}") return session_id def load_session(self, session_id): session_path = SESSIONS_DIR / session_id if not session_path.exists(): raise Exception(f"❌ Session '{session_id}' not found. Try create_session().") self.current_session = session_id print(f"[SESSION] Loaded {session_id}") return session_id # ------------------ PATH ------------------ def _get_path(self, filename): if not self.current_session: raise Exception("❌ No active session") return SESSIONS_DIR / self.current_session / filename # ------------------ READ / WRITE ------------------ def _write_json(self, filename, data): path = self._get_path(filename) with open(path, "w") as f: json.dump(data, f, indent=2) def _read_json(self, filename): path = self._get_path(filename) if not path.exists(): return {} if filename == "workspace.json" else [] return json.load(open(path)) # ------------------ WORKSPACE ------------------ def get_workspace(self): return self._read_json("workspace.json") def update_workspace(self, key, value): data = self.get_workspace() data[key] = value self._write_json("workspace.json", data) # ------------------ HISTORY ------------------ def append_history(self, item): hist = self._read_json("history.json") hist.append(item) self._write_json("history.json", hist) def get_history(self): return self._read_json("history.json")