Spaces:
Sleeping
Sleeping
File size: 4,132 Bytes
924a755 | 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 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """Runtime isolation and quota protection for the public Hugging Face Space.
Each Streamlit browser session receives its own SQLite database and trace tree.
An atomic lock limits expensive negotiation pipelines to one at a time while
allowing completed pipelines to wait independently at their human gates.
"""
from __future__ import annotations
import json
import os
import shutil
import time
import uuid
from dataclasses import dataclass
from pathlib import Path
from core.config import PROJECT_ROOT, settings
def is_hosted() -> bool:
return os.environ.get("AUTOSOURCE_HOSTED", "").lower() in {"1", "true", "yes"}
@dataclass(frozen=True)
class SessionWorkspace:
session_id: str
root: Path
db: Path
traces: Path
def child_env(self) -> dict[str, str]:
env = os.environ.copy()
env.update({
"AUTOSOURCE_DB_PATH": str(self.db),
"AUTOSOURCE_TRACES_DIR": str(self.traces),
"PYTHONUNBUFFERED": "1",
"PYTHONIOENCODING": "utf-8",
})
return env
def session_base() -> Path:
value = os.environ.get("AUTOSOURCE_SESSION_ROOT")
root = Path(value) if value else PROJECT_ROOT / ".runtime" / "sessions"
root.mkdir(parents=True, exist_ok=True)
return root.resolve()
def get_workspace(session_id: str) -> SessionWorkspace:
safe_id = "".join(c for c in session_id if c.isalnum() or c in "-_")[:64]
if not safe_id:
raise ValueError("invalid session id")
root = session_base() / safe_id
traces = root / "traces"
traces.mkdir(parents=True, exist_ok=True)
os.utime(root, None)
return SessionWorkspace(safe_id, root, root / "warehouse.db", traces)
def new_session_id() -> str:
return uuid.uuid4().hex
def cleanup_expired_sessions(max_age_s: int | None = None) -> int:
"""Remove abandoned ephemeral workspaces; never touches the bundled replay."""
max_age_s = max_age_s or int(settings().get("hosted", {}).get(
"session_ttl_s", 21600))
cutoff = time.time() - max_age_s
removed = 0
for path in session_base().iterdir():
try:
if path.is_dir() and path.stat().st_mtime < cutoff:
shutil.rmtree(path)
removed += 1
except OSError:
continue
return removed
def lock_path() -> Path:
value = os.environ.get("AUTOSOURCE_LIVE_LOCK")
return Path(value) if value else session_base().parent / "live-pipeline.lock"
def _read_lock() -> dict | None:
path = lock_path()
try:
return json.loads(path.read_text(encoding="utf-8"))
except (FileNotFoundError, json.JSONDecodeError, OSError):
return None
def live_slot_status() -> dict | None:
record = _read_lock()
if not record:
return None
max_age = int(settings().get("hosted", {}).get("pipeline_lock_ttl_s", 900))
if time.time() - float(record.get("created_at", 0)) > max_age:
try:
lock_path().unlink()
except OSError:
pass
return None
return record
def acquire_live_slot(session_id: str, run_id: str) -> str | None:
"""Atomically reserve the shared provider slot. Returns an ownership token."""
live_slot_status() # clears a stale lock before the atomic create
token = uuid.uuid4().hex
record = {
"token": token,
"session_id": session_id,
"run_id": run_id,
"created_at": time.time(),
}
path = lock_path()
path.parent.mkdir(parents=True, exist_ok=True)
try:
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
except FileExistsError:
return None
with os.fdopen(fd, "w", encoding="utf-8") as stream:
json.dump(record, stream)
return token
def release_live_slot(token: str) -> bool:
"""Release only when the caller owns the lock; prevents cross-session unlocks."""
record = _read_lock()
if not record or record.get("token") != token:
return False
try:
lock_path().unlink()
return True
except FileNotFoundError:
return True
except OSError:
return False
|