Spaces:
Running
Running
File size: 3,860 Bytes
48174a2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | """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)
|