Spaces:
Running
Running
| """Workspaces: one isolated working directory per HuggingChat conversation. | |
| HuggingChat sends no conversation id to MCP servers, so the id travels inside the chat | |
| itself: `workspace_new` returns `ws-xxxxxx`, the model passes it to every tool call and | |
| HuggingChat replays tool history on later turns. New chat -> new history -> new workspace. | |
| """ | |
| import json | |
| import os | |
| import re | |
| import secrets | |
| import shutil | |
| import time | |
| from pathlib import Path | |
| class WorkspaceError(Exception): | |
| pass | |
| def _default_root() -> Path: | |
| env = os.environ.get("WORKSPACES_DIR") | |
| if env: | |
| return Path(env) | |
| # /data exists on Spaces with persistent storage; otherwise the disk is ephemeral. | |
| if os.path.isdir("/data") and os.access("/data", os.W_OK): | |
| return Path("/data/workspaces") | |
| return Path.home() / "workspaces" | |
| ROOT = _default_root() | |
| ROOT.mkdir(parents=True, exist_ok=True) | |
| _ID_RE = re.compile(r"^ws-[0-9a-f]{6}$") | |
| class Workspace: | |
| def __init__(self, ws_id: str): | |
| self.id = ws_id | |
| self.dir = ROOT / ws_id | |
| self.work = self.dir / "work" # what the model sees as its project folder | |
| self.meta_file = self.dir / "meta.json" | |
| self.cwd_file = self.dir / "cwd" # cwd persisted between Bash calls | |
| self.todos_file = self.dir / "todos.json" | |
| self.shells_dir = self.dir / "shells" | |
| def meta(self) -> dict: | |
| try: | |
| return json.loads(self.meta_file.read_text()) | |
| except Exception: | |
| return {} | |
| def touch(self) -> None: | |
| meta = self.meta() | |
| meta["last_used"] = time.time() | |
| self.meta_file.write_text(json.dumps(meta)) | |
| def cwd(self) -> Path: | |
| """Current directory: the last one a Bash call ended in, if it still exists.""" | |
| try: | |
| p = Path(self.cwd_file.read_text().strip()) | |
| if p.is_dir(): | |
| return p | |
| except Exception: | |
| pass | |
| return self.work | |
| def path(self, p: str) -> Path: | |
| """Resolve a tool path: absolute stays absolute, relative is from the current dir.""" | |
| path = Path(p).expanduser() | |
| return path if path.is_absolute() else self.cwd() / path | |
| def normalize(ws_id: str) -> str: | |
| ws_id = (ws_id or "").strip().lower() | |
| if re.fullmatch(r"[0-9a-f]{6}", ws_id): | |
| ws_id = f"ws-{ws_id}" | |
| return ws_id | |
| def get(ws_id: str) -> Workspace: | |
| norm = normalize(ws_id) | |
| if not ws_id or not _ID_RE.match(norm): | |
| raise WorkspaceError( | |
| "missing or invalid `workspace`. If this is a new chat, call workspace_new first. " | |
| "If this chat already has a workspace id (look for 'ws-xxxxxx' in earlier messages), " | |
| "pass it. workspace_list shows existing workspaces." | |
| ) | |
| ws = Workspace(norm) | |
| if not ws.work.is_dir(): | |
| raise WorkspaceError( | |
| f"unknown workspace '{norm}'. It may have been deleted or the Space restarted " | |
| "(disk is ephemeral without persistent storage). Call workspace_list or workspace_new." | |
| ) | |
| ws.touch() | |
| return ws | |
| def create(title: str = "") -> Workspace: | |
| for _ in range(20): | |
| ws_id = f"ws-{secrets.token_hex(3)}" | |
| if not (ROOT / ws_id).exists(): | |
| break | |
| else: | |
| raise WorkspaceError("could not allocate a workspace id") | |
| ws = Workspace(ws_id) | |
| ws.work.mkdir(parents=True) | |
| ws.shells_dir.mkdir() | |
| ws.meta_file.write_text( | |
| json.dumps({"title": title.strip()[:120], "created": time.time(), "last_used": time.time()}) | |
| ) | |
| return ws | |
| def list_all() -> list[Workspace]: | |
| out = [Workspace(p.name) for p in ROOT.iterdir() if _ID_RE.match(p.name) and (p / "work").is_dir()] | |
| return sorted(out, key=lambda w: w.meta().get("last_used", 0), reverse=True) | |
| def delete(ws_id: str) -> None: | |
| ws = get(ws_id) | |
| shutil.rmtree(ws.dir) | |