| """An environment Space as its files: what its repository declares, read without waking it. |
| |
| An OpenEnv environment is a Space repository as much as a running server: `openenv.yaml` (its manifest, sometimes in |
| a subfolder: the shallowest one counts), `models.py` (its Action, Observation and State), `server/app.py` (how the |
| server is made: `create_app(..., env_name=, max_concurrent_envs=, gradio_builder=)`), `server/*_environment.py` (the |
| episode and its reward), `Dockerfile`, `pyproject.toml` (the openenv-core it needs), `discovery.json`, the README. |
| The Hub lists a Space's files without waking it, so a sleeping Space (most of them) still shows what it is. |
| |
| GET /api/spaces/<org>/<name>/files the curated file tree and what the files declare, pinned to one revision |
| GET /api/spaces/<org>/<name>/file?f= one text file from that tree, at that revision |
| |
| Nothing here runs a Space's code: Python is read with `ast`, YAML with `safe_load` (and a bounded walk, so anchors |
| can't blow up), TOML with tomllib. Files are read anonymously (never with a visitor's token), pinned to the revision |
| listed, only files in the curated listing are served, and binaries aren't. Private Spaces are refused. Files that may |
| hold answers (a solution folder, task data) are listed but not served, as everywhere else in the explorer. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import ast |
| import json |
| import math |
| import posixpath |
| import re |
| import shlex |
| import time |
| import tomllib |
| from concurrent.futures import ThreadPoolExecutor |
| from typing import Any |
|
|
| import httpx |
| from fastapi import APIRouter, HTTPException |
|
|
| from . import catalog, config |
| from .spaces_live import ANSWER |
|
|
| router = APIRouter() |
|
|
| MAX_TEXT = 400 * 1024 |
| MAX_FETCH = 4 * 1024 * 1024 |
| MAX_LISTED = 5000 |
| INFO_TTL = 60 |
| PARSE_TTL = 6 * 3600 |
|
|
| |
| JUNK = re.compile( |
| r"(^|/)(__pycache__|\.git|\.venv|venv|node_modules|\.ipynb_checkpoints|\.pytest_cache|\.mypy_cache|\.ruff_cache)/" |
| r"|(^|/)[^/]+\.egg-info/" |
| r"|^build/|(^|/)build/(lib|bdist)[^/]*/" |
| r"|^src/(openenv|openenv_core|core)/" |
| r"|(^|/)\.gitattributes$|(^|/)\.DS_Store$|\.py[co]$") |
| BINARY = re.compile( |
| r"\.(png|jpe?g|gif|webp|bmp|ico|tiff?|avif|heic|pdf|zip|tar|gz|tgz|bz2|xz|zst|7z|rar|whl|egg|so|dylib|dll|exe|bin|" |
| r"pt|pth|ckpt|safetensors|onnx|gguf|pkl|pickle|joblib|npy|npz|h5|hdf5|parquet|arrow|feather|db|sqlite3?|mp3|wav|" |
| r"flac|ogg|m4a|mp4|webm|mov|avi|mkv|ttf|otf|woff2?|eot|jar|class|wasm)$", re.I) |
| CODE = re.compile(r"\.(py|pyi|sh|bash|js|mjs|ts|tsx|jsx|rb|go|rs|java|c|cc|cpp|h|hpp|md|html|css|ipynb)$|(^|/)Dockerfile[^/]*$", re.I) |
| |
| MANIFESTS = re.compile(r"(^|/)(openenv\.ya?ml|discovery\.json|package(-lock)?\.json|pyproject\.toml|tsconfig\.json)$", re.I) |
| DATA = re.compile(r"\.(jsonl|json|csv|tsv|parquet|arrow|ndjson)$", re.I) |
| |
| ANSWER_DIR = re.compile(r"(^|/)(solutions?|gold|answers?|oracle|ground_?truth)/", re.I) |
| TASK_DATA_DIR = re.compile(r"(^|/)(tasks?|splits?|evals?|test_?cases|episodes|answers?|labels?)/", re.I) |
|
|
|
|
| |
| def _info(spec: str) -> dict[str, Any]: |
| """The Space's revision and every file in it with its size (one Hub call; the Space isn't touched).""" |
| spec = catalog.check_spec(spec) |
|
|
| def fetch(): |
| sp = catalog._api().space_info(spec, files_metadata=True) |
| if getattr(sp, "private", False) or getattr(sp, "gated", False): |
| raise PermissionError("this Space is private") |
| card = (sp.card_data.to_dict() if getattr(sp, "card_data", None) else {}) or {} |
| rt = getattr(sp, "runtime", None) |
| files = [] |
| for s in getattr(sp, "siblings", None) or []: |
| name = getattr(s, "rfilename", None) |
| if isinstance(name, str) and name and len(name) < 512: |
| files.append((name, int(getattr(s, "size", None) or 0), getattr(s, "lfs", None) is not None)) |
| sub = getattr(sp, "subdomain", None) or "" |
| return {"id": spec, "sha": str(sp.sha or ""), "files": files, "card": card, |
| "subdomain": sub if re.match(r"^[a-z0-9-]{1,80}$", sub) else "", |
| "stage": str(getattr(rt, "stage", "") or "") if rt is not None else "", |
| "hardware": getattr(rt, "hardware", None) if rt is not None else None, |
| "requested_hardware": getattr(rt, "requested_hardware", None) if rt is not None else None, |
| "sdk": getattr(sp, "sdk", None), "tags": list(getattr(sp, "tags", None) or []), |
| "updated": sp.last_modified.isoformat() if getattr(sp, "last_modified", None) else None} |
|
|
| info = catalog._cached(("space-files-info", spec), INFO_TTL, fetch) |
| if not re.match(r"^[0-9a-f]{40}$", info["sha"]): |
| raise LookupError("the Hub didn't say which revision this Space is at") |
| return info |
|
|
|
|
| def junk(path: str) -> bool: |
| return bool(JUNK.search(path)) |
|
|
|
|
| def _withheld_by_name(path: str) -> bool: |
| """Listed but never served: reference solutions, and task data (it holds the answers the grader checks).""" |
| if ANSWER_DIR.search(path): |
| return True |
| if DATA.search(path) and (TASK_DATA_DIR.search(path) or catalog.ANSWER_FILE.search(path)): |
| return True |
| return not CODE.search(path) and not MANIFESTS.search(path) and bool(catalog.ANSWER_FILE.search(path)) |
|
|
|
|
| def listing(info: dict[str, Any]) -> tuple[list[dict[str, Any]], int, bool]: |
| """The files worth reading: junk dropped, binaries marked, answer files marked withheld.""" |
| out, dropped = [], 0 |
| for path, size, lfs in sorted(info["files"]): |
| if junk(path) or ".." in path.split("/") or path.startswith("/") or "\\" in path: |
| dropped += 1 |
| continue |
| row: dict[str, Any] = {"path": path, "size": size} |
| if lfs or BINARY.search(path): |
| row["binary"] = True |
| if _withheld_by_name(path): |
| row["withheld"] = True |
| if path == "uv.lock" or path.endswith("/uv.lock"): |
| row["collapsed"] = True |
| out.append(row) |
| return out[:MAX_LISTED], dropped, len(out) > MAX_LISTED |
|
|
|
|
| def _download(spec: str, sha: str, path: str, limit: int = MAX_TEXT + 1) -> bytes: |
| """The first `limit` bytes of one file of a public Space at a revision: anonymous, and cached on disk by revision |
| (a file at a commit never changes, so it is fetched once).""" |
| import huggingface_hub |
|
|
| p = huggingface_hub.hf_hub_download(spec, path, repo_type="space", revision=sha, token=False, |
| cache_dir=str(config.CACHE_DIR / "space-files")) |
| with open(p, "rb") as f: |
| return f.read(limit) |
|
|
|
|
| def _decode(raw: bytes) -> str | None: |
| """Text, or None for a binary (a NUL byte, or bytes that aren't UTF-8 before the cut).""" |
| if b"\x00" in raw[:8192]: |
| return None |
| try: |
| return raw.decode("utf-8") |
| except UnicodeDecodeError as e: |
| if e.start >= len(raw) - 4: |
| return raw[:e.start].decode("utf-8", "replace") |
| return None |
|
|
|
|
| def _fetch_text(spec: str, sha: str, path: str, size: int) -> str | None: |
| if size > MAX_TEXT: |
| return None |
| try: |
| return _decode(_download(spec, sha, path)[:MAX_TEXT]) |
| except Exception: |
| return None |
|
|
|
|
| |
| class _Budget: |
| def __init__(self, nodes: int = 6000): |
| self.left = nodes |
|
|
|
|
| def plain(v: Any, depth: int = 0, budget: _Budget | None = None, drop_answers: bool = True, gone: list | None = None) -> Any: |
| """A YAML/TOML value as bounded JSON: depth, sizes and the number of nodes visited are capped (YAML anchors |
| can make a small file a huge graph), and answer-like keys are left out.""" |
| budget = budget or _Budget() |
| budget.left -= 1 |
| if budget.left < 0 or depth > 7: |
| return None |
| if v is None or isinstance(v, (bool, int)): |
| return v |
| if isinstance(v, float): |
| return v if math.isfinite(v) else str(v) |
| if isinstance(v, str): |
| return v[:2000] |
| if isinstance(v, dict): |
| out = {} |
| for k, x in list(v.items())[:120]: |
| key = str(k)[:120] |
| if drop_answers and ANSWER.search(key): |
| if gone is not None: |
| gone.append(key) |
| continue |
| out[key] = plain(x, depth + 1, budget, drop_answers, gone) |
| return out |
| if isinstance(v, (list, tuple, set)): |
| return [plain(x, depth + 1, budget, drop_answers, gone) for x in list(v)[:200]] |
| if hasattr(v, "isoformat"): |
| return v.isoformat() |
| return str(v)[:200] |
|
|
|
|
| def _str(v: Any, n: int = 400) -> str | None: |
| if v is None or isinstance(v, (dict, list)): |
| return None |
| s = str(v).strip() |
| return s[:n] if s else None |
|
|
|
|
| def _num(v: Any) -> float | None: |
| return float(v) if isinstance(v, (int, float)) and not isinstance(v, bool) and math.isfinite(v) else None |
|
|
|
|
| MANIFEST_KEYS = ("spec_version", "name", "version", "type", "runtime", "app", "port", "description", "author", "license", |
| "display_name") |
|
|
|
|
| def _validation(v: Any) -> dict[str, Any] | None: |
| """OpenEnv's `validation:` block (openenv.validation.manifest), read as data: reward contract, judge pin, |
| resources, network, capabilities, type tags.""" |
| if not isinstance(v, dict): |
| return None |
| out: dict[str, Any] = {} |
| rw = v.get("reward") |
| if isinstance(rw, dict): |
| rng = rw.get("range") |
| out["reward"] = {"range": [_num(x) for x in rng[:2]] if isinstance(rng, (list, tuple)) and len(rng) >= 2 else None, |
| **{k: _num(rw.get(k)) for k in ("oracle_tolerance", "floor_margin", "variance_tolerance") if _num(rw.get(k)) is not None}} |
| j = v.get("judge") |
| if isinstance(j, dict): |
| out["judge"] = {"model": _str(j.get("model"), 200), "version": _str(j.get("version"), 100), |
| "params": plain(j.get("params"), drop_answers=False) if isinstance(j.get("params"), dict) else None} |
| r = v.get("resources") |
| if isinstance(r, dict): |
| out["resources"] = {k: (_num(r.get(k)) if k != "gpu_types" else [str(x)[:60] for x in (r.get(k) or [])[:8]] if isinstance(r.get(k), list) else None) |
| for k in ("cpu", "memory_mb", "disk_mb", "episode_timeout_s", "gpus", "gpu_types") if r.get(k) is not None} |
| n = v.get("network") |
| if isinstance(n, dict): |
| out["network"] = {"mode": _str(n.get("mode"), 40) or "public", |
| "allowed_hosts": [str(x)[:120] for x in (n.get("allowed_hosts") or [])[:30]] if isinstance(n.get("allowed_hosts"), list) else []} |
| c = v.get("capabilities") |
| if isinstance(c, dict): |
| ver = c.get("verifier") if isinstance(c.get("verifier"), dict) else {} |
| orc = c.get("oracle") if isinstance(c.get("oracle"), dict) else None |
| counts = c.get("declared_task_count") if isinstance(c.get("declared_task_count"), dict) else {} |
| out["capabilities"] = { |
| "verifier": {"kind": _str(ver.get("kind"), 40), "entry": _str(ver.get("entry"), 200)} if ver else None, |
| "oracle": {"form": _str(orc.get("form"), 40), "location": _str(orc.get("location"), 200)} if orc else None, |
| **{k: bool(c.get(k)) for k in ("set_state", "llm_judged", "rubric_tree", "task_api") if k in c}, |
| "declared_tools": [str(x)[:120] for x in (c.get("declared_tools") or [])[:100]] if isinstance(c.get("declared_tools"), list) else [], |
| "declared_task_count": {str(k)[:80]: int(x) for k, x in list(counts.items())[:40] if isinstance(x, int) and not isinstance(x, bool)}, |
| } |
| t = v.get("types") |
| if isinstance(t, dict) and isinstance(t.get("tags"), list): |
| out["types"] = [str(x)[:60] for x in t["tags"][:20]] |
| extra = sorted(str(k) for k in v if k not in ("reward", "judge", "resources", "network", "capabilities", "types")) |
| if extra: |
| out["other"] = extra[:20] |
| return out |
|
|
|
|
| TASK_FIELDS = ("id", "name", "title", "difficulty", "level", "description", "max_steps", "split", "grader", "score_range") |
|
|
|
|
| def _tasks(v: Any) -> list[dict[str, Any]] | None: |
| """A free-form `tasks:` key (hackathon manifests list them): each as {id, name, difficulty, description, …}.""" |
| items: list[tuple[Any, Any]] = [] |
| if isinstance(v, list): |
| items = [(None, x) for x in v[:500]] |
| elif isinstance(v, dict): |
| items = list(v.items())[:500] |
| else: |
| return None |
| out = [] |
| for key, x in items: |
| if isinstance(x, (str, int, float)) and not isinstance(x, bool): |
| out.append({"id": str(x)[:120]} if key is None else {"id": str(key)[:120], "description": str(x)[:600]}) |
| elif isinstance(x, dict): |
| row: dict[str, Any] = {"id": str(key)[:120]} if key is not None else {} |
| for k in TASK_FIELDS: |
| if k in x and not ANSWER.search(k): |
| val = x[k] |
| row[k] = (_str(val, 600) if not isinstance(val, (list, dict)) else plain(val, drop_answers=True)) |
| if "level" in row and "difficulty" not in row: |
| row["difficulty"] = row.pop("level") |
| if row: |
| out.append(row) |
| return out |
|
|
|
|
| def parse_manifest(text: str, path: str) -> dict[str, Any]: |
| """openenv.yaml: its canonical keys, its `validation:` block, and the free-form keys authors add (tasks, rubric, |
| reward, environment, tools; the legacy `action:`/`observation:` class names).""" |
| import yaml |
|
|
| out: dict[str, Any] = {"path": path} |
| try: |
| doc = yaml.safe_load(text) |
| except yaml.YAMLError as e: |
| mark = getattr(e, "problem_mark", None) |
| out["error"] = f"not valid YAML{f' (line {mark.line + 1})' if mark is not None else ''}" |
| return out |
| except Exception: |
| out["error"] = "couldn't be read" |
| return out |
| if not isinstance(doc, dict): |
| out["error"] = "not a YAML mapping" |
| return out |
| keys = {k: plain(doc[k]) if isinstance(doc[k], (int, float, bool)) else _str(doc[k], 1200) for k in MANIFEST_KEYS if k in doc} |
| out["keys"] = {k: v for k, v in keys.items() if v is not None} |
| tags = doc.get("tags") |
| if isinstance(tags, list): |
| out["tags"] = [str(t)[:60] for t in tags[:20]] |
| if "validation" in doc: |
| out["validation"] = _validation(doc.get("validation")) |
| tasks = _tasks(doc.get("tasks")) |
| if tasks is not None: |
| out["tasks"] = tasks |
| tools = doc.get("tools") |
| if isinstance(tools, list): |
| out["tools"] = [({"name": str(t.get("name"))[:120], "description": _str(t.get("description"), 300)} if isinstance(t, dict) and t.get("name") |
| else {"name": str(t)[:120]}) for t in tools[:100] if isinstance(t, (str, dict))] |
| gone: list[str] = [] |
| for k in ("rubric", "rubrics", "reward", "rewards", "reward_range", "grader", "scoring", "environment", "action_space", |
| "observation_space", "endpoints"): |
| if k in doc and doc[k] is not None: |
| out.setdefault("blocks", {})[k] = plain(doc[k], gone=gone) |
| for k in ("action", "observation"): |
| if isinstance(doc.get(k), str): |
| out.setdefault("legacy", {})[k] = doc[k][:120] |
| elif isinstance(doc.get(k), dict): |
| out.setdefault("blocks", {})[k] = plain(doc[k], gone=gone) |
| if gone: |
| out["withheld"] = sorted(set(gone)) |
| known = set(MANIFEST_KEYS) | {"tags", "validation", "tasks", "tools", "action", "observation"} | set(out.get("blocks", {})) |
| other = [str(k) for k in doc if k not in known] |
| if other: |
| out["other"] = other[:30] |
| return out |
|
|
|
|
| |
| ROLES = {"Action": "action", "Observation": "observation", "State": "state", "CallToolAction": "action", |
| "CallToolObservation": "observation", "ListToolsAction": "action"} |
| IMAGE = re.compile(r"((^|_)(image|img|frame|screenshot|pixels|png|jpe?g|picture|photo|render)s?(_?(b64|data|url|bytes))?$|base64|(^|_)b64$)", re.I) |
|
|
|
|
| def _src(node: ast.AST | None, n: int = 160) -> str | None: |
| if node is None: |
| return None |
| try: |
| s = ast.unparse(node) |
| except Exception: |
| return None |
| return s if len(s) <= n else s[: n - 1] + "…" |
|
|
|
|
| def _first_doc(node: ast.AST) -> str | None: |
| try: |
| d = ast.get_docstring(node) |
| except Exception: |
| return None |
| return d.strip().split("\n\n")[0].strip()[:500] if d else None |
|
|
|
|
| def _parse_py(text: str) -> ast.Module | None: |
| try: |
| return ast.parse(text) |
| except (SyntaxError, ValueError, RecursionError, MemoryError): |
| return None |
|
|
|
|
| def _field(stmt: ast.AnnAssign) -> dict[str, Any] | None: |
| if not isinstance(stmt.target, ast.Name): |
| return None |
| name = stmt.target.id |
| ann = _src(stmt.annotation, 200) or "" |
| if name.startswith("_") or name == "model_config" or ann.startswith(("ClassVar", "typing.ClassVar")): |
| return None |
| row: dict[str, Any] = {"name": name, "type": ann, "required": stmt.value is None} |
| v = stmt.value |
| if isinstance(v, ast.Call) and (_src(v.func) or "").split(".")[-1] == "Field": |
| kw = {k.arg: k.value for k in v.keywords if k.arg} |
| first = v.args[0] if v.args else kw.get("default") |
| if isinstance(first, ast.Constant) and first.value is Ellipsis: |
| row["required"] = True |
| elif first is not None: |
| row["default"] = _src(first, 80) |
| elif "default_factory" in kw: |
| row["default"] = f"{_src(kw['default_factory'], 60)}()" |
| else: |
| row["required"] = True |
| desc = kw.get("description") |
| if isinstance(desc, ast.Constant) and isinstance(desc.value, str): |
| row["description"] = desc.value[:400] |
| limits = [f"{k} {_src(kw[k], 40)}" for k in ("ge", "gt", "le", "lt", "min_length", "max_length", "pattern") if k in kw] |
| if limits: |
| row["limits"] = limits[:4] |
| elif v is not None: |
| row["default"] = _src(v, 80) |
| if IMAGE.search(name) or re.search(r"\b(Image|bytes)\b", ann): |
| row["image"] = True |
| return row |
|
|
|
|
| def parse_models(text: str, path: str) -> dict[str, Any]: |
| """models.py's classes: which are the Action, Observation and State (by their bases, followed within the file), |
| and each one's fields with type, default and description. Parsed, never imported.""" |
| tree = _parse_py(text) |
| if tree is None: |
| return {"path": path, "error": "couldn't be parsed as Python"} |
| defs = [n for n in tree.body if isinstance(n, ast.ClassDef)][:80] |
| bases = {c.name: [(_src(b, 80) or "").split(".")[-1].split("[")[0] for b in c.bases] for c in defs} |
|
|
| def role(name: str, seen: frozenset = frozenset()) -> str | None: |
| if name in ROLES: |
| return ROLES[name] |
| if name in seen or name not in bases: |
| return None |
| for b in bases[name]: |
| r = role(b, seen | {name}) |
| if r: |
| return r |
| return None |
|
|
| classes = [] |
| for c in defs: |
| kind = "enum" if any(b.endswith("Enum") for b in bases[c.name]) else None |
| r = role(c.name) |
| row: dict[str, Any] = {"name": c.name, "bases": bases[c.name][:6], "role": r, "doc": _first_doc(c)} |
| if kind == "enum": |
| row["kind"] = "enum" |
| row["values"] = [_src(s.value, 60) for s in c.body if isinstance(s, ast.Assign) and s.value is not None][:40] |
| else: |
| row["fields"] = [f for f in (_field(s) for s in c.body if isinstance(s, ast.AnnAssign)) if f][:80] |
| classes.append(row) |
| return {"path": path, "classes": classes} |
|
|
|
|
| def _consts(tree: ast.Module) -> dict[str, ast.AST]: |
| """Module-level `NAME = value`, to read `max_concurrent_envs=MAX_CONCURRENT` back to what it is.""" |
| out = {} |
| for n in tree.body: |
| if isinstance(n, ast.Assign) and len(n.targets) == 1 and isinstance(n.targets[0], ast.Name): |
| out[n.targets[0].id] = n.value |
| return out |
|
|
|
|
| def _literal_of(node: ast.AST | None, consts: dict[str, ast.AST]) -> Any: |
| """A literal, a module constant, or the default in `int(os.getenv("X", "4"))`: its value; else None.""" |
| for _ in range(3): |
| if isinstance(node, ast.Name) and node.id in consts: |
| node = consts[node.id] |
| if isinstance(node, ast.Constant) and isinstance(node.value, (str, int, float, bool)): |
| return node.value |
| if isinstance(node, ast.Call) and _src(node.func) in ("int", "float", "str") and node.args: |
| inner = node.args[0] |
| if isinstance(inner, ast.Call) and (_src(inner.func) or "").endswith(("getenv", "environ.get")) and len(inner.args) >= 2: |
| v = _literal_of(inner.args[1], consts) |
| try: |
| return int(v) if _src(node.func) == "int" else v |
| except (TypeError, ValueError): |
| return None |
| return _literal_of(inner, consts) |
| return None |
|
|
|
|
| def parse_app(text: str, path: str) -> dict[str, Any]: |
| """server/app.py: the `create_app(...)` call (its environment, Action and Observation classes, env_name, |
| max_concurrent_envs, a Gradio tab of its own, the mode) and routes it adds itself.""" |
| tree = _parse_py(text) |
| if tree is None: |
| return {"path": path, "error": "couldn't be parsed as Python"} |
| consts = _consts(tree) |
| out: dict[str, Any] = {"path": path} |
| for node in ast.walk(tree): |
| if isinstance(node, ast.Call) and (_src(node.func) or "").split(".")[-1] in ("create_app", "create_fastapi_app", "create_web_interface_app"): |
| args = [(_src(a, 80) or "") for a in node.args[:3]] |
| kw = {k.arg: k.value for k in node.keywords if k.arg} |
| out["factory"] = (_src(node.func) or "").split(".")[-1] |
| for i, (key, name) in enumerate((("env", "env"), ("action", "action_cls"), ("observation", "observation_cls"))): |
| val = args[i] if i < len(args) else _src(kw.get(name), 80) |
| if val: |
| out[key] = val.split(".")[-1] |
| for key in ("env_name", "max_concurrent_envs", "custom_tab_name", "title_override", "mode"): |
| if key in kw: |
| lit = _literal_of(kw[key], consts) |
| out[key] = lit if lit is not None else _src(kw[key], 80) |
| if "gradio_builder" in kw: |
| out["gradio_builder"] = _src(kw["gradio_builder"], 80) |
| if "state_cls" in kw: |
| out["state"] = (_src(kw["state_cls"]) or "").split(".")[-1] |
| break |
| routes = [] |
| for node in ast.walk(tree): |
| if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): |
| for d in node.decorator_list: |
| if isinstance(d, ast.Call) and isinstance(d.func, ast.Attribute) and d.func.attr in ("get", "post", "put", "delete", "websocket", "api_route") \ |
| and d.args and isinstance(d.args[0], ast.Constant) and isinstance(d.args[0].value, str): |
| routes.append(f"{d.func.attr.upper()} {d.args[0].value[:120]}") |
| if routes: |
| out["routes"] = routes[:30] |
| mounts = [_src(n.args[0], 60) for n in ast.walk(tree) if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) |
| and n.func.attr == "mount" and n.args] |
| if mounts: |
| out["mounts"] = [m for m in mounts if m][:10] |
| out["mcp"] = out.get("action") in ("CallToolAction", "ListToolsAction") or None |
| return out |
|
|
|
|
| LLM_MODULES = {"openai", "anthropic", "litellm", "groq", "mistralai", "cohere", "together", "google.generativeai", "google.genai", "vertexai"} |
| REWARD_FN = re.compile(r"(reward|score|grade|grader|judge|verify|verifier|penalt|bonus|rubric)", re.I) |
|
|
|
|
| def parse_environment(text: str, path: str) -> dict[str, Any]: |
| """server/*_environment.py: the environment class, the MCP tools it registers (`@mcp.tool`), its reward and |
| grading functions, and whether an LLM judges (a client library imported, OpenEnv's LLMJudge) or a Rubric scores.""" |
| tree = _parse_py(text) |
| if tree is None: |
| return {"path": path, "error": "couldn't be parsed as Python"} |
| imports: set[str] = set() |
| names: set[str] = set() |
| tools, fns, envs, local = [], [], [], [] |
| for node in ast.walk(tree): |
| if isinstance(node, ast.Import): |
| imports |= {a.name for a in node.names} |
| elif isinstance(node, ast.ImportFrom): |
| if node.module: |
| imports.add(node.module) |
| names |= {a.name for a in node.names} |
| if (node.level or REWARD_FN.search(node.module or "")) and (REWARD_FN.search(node.module or "") or any(REWARD_FN.search(a.name) and a.name[:1].islower() for a in node.names)): |
| local.append([node.level or 0, node.module or "", [a.name for a in node.names][:6]]) |
| elif isinstance(node, ast.Name): |
| names.add(node.id) |
| elif isinstance(node, ast.Attribute): |
| names.add(node.attr) |
| elif isinstance(node, ast.ClassDef): |
| b = [(_src(x, 80) or "").split(".")[-1].split("[")[0] for x in node.bases] |
| if any(x.endswith("Environment") for x in b): |
| envs.append({"name": node.name, "bases": b[:4], "doc": _first_doc(node)}) |
| if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): |
| if any(re.search(r"(^|\.)tool$", (_src(d.func if isinstance(d, ast.Call) else d) or "")) for d in node.decorator_list): |
| params = [a.arg for a in node.args.args if a.arg not in ("self", "cls", "ctx")][:12] |
| tools.append({"name": node.name, "params": params, "description": (_first_doc(node) or "").split("\n")[0][:300] or None}) |
| elif REWARD_FN.search(node.name) and not node.name.startswith("__"): |
| fns.append(node.name) |
| llm = sorted({m for m in imports if m in LLM_MODULES or m.split(".")[0] in LLM_MODULES}) |
| judge = bool(llm) or bool({"LLMJudge", "InferenceClient", "chat_completion"} & names) or ("completions" in names and "create" in names) |
| rubric = any("rubric" in m.lower() for m in imports) or bool({"Rubric", "RubricDict", "WeightedSum"} & names) |
| out: dict[str, Any] = {"path": path, "classes": envs[:6], "tools": tools[:60], "functions": sorted(set(fns))[:24], "reward_imports": local[:8], |
| "llm_judge": judge, "llm_modules": llm[:6], "rubric": rubric, |
| "mcp": any(e["bases"] and any(x.startswith("MCP") for x in e["bases"]) for e in envs) or bool(tools)} |
| return out |
|
|
|
|
| |
| def parse_dockerfile(text: str, path: str) -> dict[str, Any]: |
| """The image's base, whether OpenEnv's web interface is on (ENABLE_WEB_INTERFACE), the port, the command.""" |
| from .dockerfile import _lines, _subst |
|
|
| out: dict[str, Any] = {"path": path, "from": []} |
| env: dict[str, str] = {} |
| args: dict[str, str] = {} |
| for line in _lines(text)[:400]: |
| word, _, rest = line.partition(" ") |
| word = word.upper() |
| rest = rest.strip() |
| if word == "FROM": |
| parts = [p for p in rest.split() if not p.startswith("--")] |
| if parts: |
| out["from"].append(_subst(parts[0], args)[:160]) |
| if word == "ARG": |
| k, _, val = rest.partition("=") |
| if k.strip() and val: |
| args.setdefault(k.strip(), val.strip().strip("'\"")) |
| elif word in ("ENV", "ARG"): |
| try: |
| parts = shlex.split(rest) |
| except ValueError: |
| parts = rest.split() |
| if parts and "=" not in parts[0] and word == "ENV": |
| env[parts[0]] = " ".join(parts[1:]) |
| else: |
| for p in parts: |
| k, _, val = p.partition("=") |
| if word == "ENV" or k not in env: |
| env[k] = val |
| elif word == "EXPOSE": |
| m = re.match(r"(\d{2,5})", rest) |
| if m: |
| out["expose"] = int(m.group(1)) |
| elif word in ("CMD", "ENTRYPOINT"): |
| out["cmd"] = rest[:400] |
| web = env.get("ENABLE_WEB_INTERFACE") |
| if web is not None: |
| out["web_interface"] = web.strip().strip("'\"").lower() in ("true", "1", "yes") |
| m = re.search(r"--port[ =\"',]+(\d{2,5})", out.get("cmd", "")) |
| if m: |
| out["port"] = int(m.group(1)) |
| if env.get("OPENENV_MODE"): |
| out["mode"] = env["OPENENV_MODE"].strip("'\"")[:20] |
| out["from"] = list(dict.fromkeys(out["from"]))[:6] |
| return out |
|
|
|
|
| OPENENV_REQ = re.compile(r"^\s*(openenv-core|openenv_core|openenv)(?![\w.-])\s*(\[[^\]]*\])?\s*(.*)$", re.I) |
|
|
|
|
| def parse_pyproject(text: str, path: str) -> dict[str, Any]: |
| """The package, its version, Python, and the openenv-core it requires (a version range or a git ref).""" |
| try: |
| doc = tomllib.loads(text) |
| except (tomllib.TOMLDecodeError, ValueError, RecursionError): |
| return {"path": path, "error": "not valid TOML"} |
| proj = doc.get("project") if isinstance(doc.get("project"), dict) else {} |
| deps = [d for d in proj.get("dependencies") or [] if isinstance(d, str)] |
| out: dict[str, Any] = {"path": path, "package": _str(proj.get("name"), 120), "version": _str(proj.get("version"), 40), |
| "requires_python": _str(proj.get("requires-python"), 40)} |
| for d in deps: |
| m = OPENENV_REQ.match(d) |
| if not m: |
| continue |
| rest = (m.group(3) or "").strip() |
| git = re.search(r"git\+[^@#\s]+@v?([\w.\-]+)", rest) |
| ver = re.search(r"(==|>=|~=|<=|>|<|!=)\s*v?(\d+(?:\.\d+)*\w*)", rest) |
| out["openenv"] = d.strip()[:200] |
| out["openenv_package"] = m.group(1).lower().replace("_", "-") |
| out["openenv_version"] = git.group(1) if git else (ver.group(2) if ver.group(1) == "==" else ver.group(1) + ver.group(2)) if ver else None |
| break |
| scripts = proj.get("scripts") if isinstance(proj.get("scripts"), dict) else {} |
| if isinstance(scripts.get("server"), str): |
| out["server_script"] = scripts["server"][:160] |
| out["dependencies"] = len(deps) |
| return out |
|
|
|
|
| def parse_discovery(text: str, path: str) -> dict[str, Any]: |
| """discovery.json: the environment card's declarations (description, license, tags, capabilities, queries).""" |
| try: |
| doc = json.loads(text) |
| except ValueError: |
| return {"path": path, "error": "not valid JSON"} |
| if not isinstance(doc, dict): |
| return {"path": path, "error": "not a JSON object"} |
| out: dict[str, Any] = {"path": path} |
| for k in ("description", "license", "license_url", "display_name", "displayName"): |
| if _str(doc.get(k)): |
| out["display_name" if k == "displayName" else k] = _str(doc.get(k), 1200) |
| for k in ("tags", "capabilities", "representative_queries"): |
| if isinstance(doc.get(k), list): |
| out[k] = [str(x)[:200] for x in doc[k][:20] if isinstance(x, (str, int, float))] |
| return out |
|
|
|
|
| def collapse_lock(text: str) -> str: |
| """uv.lock, collapsed to what it pins: one `name version` line per package.""" |
| try: |
| doc = tomllib.loads(text) |
| except (tomllib.TOMLDecodeError, ValueError, RecursionError): |
| return text |
| pkgs = [p for p in doc.get("package") or [] if isinstance(p, dict) and p.get("name")] |
| lines = [f"# uv.lock, collapsed: {len(pkgs)} locked package{'s' if len(pkgs) != 1 else ''} (name, version)", ""] |
| width = max((len(str(p["name"])) for p in pkgs), default=0) |
| lines += [f"{str(p['name']):<{width}} {p.get('version') or (p.get('source') or {}).get('git', '') or ''}" for p in pkgs] |
| return "\n".join(lines) + "\n" |
|
|
|
|
| |
| def _dockerfile_copies(text: str | None) -> set[str]: |
| """Folders a Dockerfile COPYs or ADDs from its build context.""" |
| if not text: |
| return set() |
| from .dockerfile import _lines |
|
|
| out = set() |
| for line in _lines(text)[:400]: |
| word, _, rest = line.partition(" ") |
| if word.upper() not in ("COPY", "ADD"): |
| continue |
| parts = [p for p in rest.split() if not p.startswith("--")] |
| for src in parts[:-1]: |
| out.add(src.strip("'\"").removeprefix("./").rstrip("/")) |
| return out |
|
|
|
|
| def manifest_path(paths: list[str], dockerfile: str | None, spec: str) -> str | None: |
| """The environment's openenv.yaml: the shallowest one; among several as shallow, the folder the Dockerfile copies, |
| then the one named like the Space, then the first by name.""" |
| cands = [p for p in paths if (p == "openenv.yaml" or p.endswith("/openenv.yaml")) and "/templates/" not in f"/{p}" and not junk(p)] |
| if not cands: |
| return None |
| depth = min(p.count("/") for p in cands) |
| top = sorted(p for p in cands if p.count("/") == depth) |
| if len(top) == 1: |
| return top[0] |
| copied = _dockerfile_copies(dockerfile) |
| for p in top: |
| if posixpath.dirname(p) in copied: |
| return p |
| def norm(s: str) -> str: |
| return re.sub(r"[^a-z0-9]", "", s.lower()) |
|
|
| for p in top: |
| if norm(posixpath.dirname(p)) == norm(spec.split("/")[-1]): |
| return p |
| return top[0] |
|
|
|
|
| def _first(paths: set[str], *cands: str) -> str | None: |
| return next((c for c in cands if c in paths), None) |
|
|
|
|
| def key_files(paths: list[str], root: str, manifest: str | None) -> dict[str, str]: |
| """Where the environment's own files are: under the manifest's folder first, then the repository's root.""" |
| have = set(paths) |
| pre = f"{root}/" if root else "" |
| out: dict[str, str | None] = {"manifest": manifest} |
| out["models"] = _first(have, f"{pre}models.py", "models.py") or next( |
| (p for p in sorted(paths, key=lambda p: (p.count("/"), p)) if p.endswith("/models.py") and p.startswith(pre) and "test" not in p), None) |
| out["app"] = _first(have, f"{pre}server/app.py", f"{pre}app.py", "server/app.py") or next( |
| (p for p in sorted(paths, key=lambda p: (p.count("/"), p)) if p.endswith("server/app.py")), None) |
| envs = sorted((p for p in paths if p.startswith(f"{pre}server/") and re.search(r"(^|/|_)environment\.py$", p) and p.count("/") == pre.count("/") + 1), |
| key=lambda p: (len(p), p)) |
| out["environment"] = envs[0] if envs else next((p for p in sorted(paths, key=lambda p: (p.count("/"), p)) if p.endswith("_environment.py")), None) |
| out["client"] = _first(have, f"{pre}client.py", "client.py") |
| out["pyproject"] = _first(have, f"{pre}pyproject.toml", "pyproject.toml") |
| out["dockerfile"] = _first(have, "Dockerfile", f"{pre}server/Dockerfile", f"{pre}Dockerfile", "server/Dockerfile") |
| out["discovery"] = _first(have, f"{pre}discovery.json", "discovery.json") |
| out["inference"] = _first(have, f"{pre}inference.py", "inference.py") |
| out["readme"] = _first(have, "README.md", f"{pre}README.md") |
| out["lock"] = _first(have, f"{pre}uv.lock", "uv.lock") |
| return {k: v for k, v in out.items() if v} |
|
|
|
|
| def reward_files(env_path: str, imports: list, paths: set[str], root: str) -> list[str]: |
| """The repository files an environment imports its reward from (`from .scoring import compute_reward`, |
| `from ..rewards.reward_function import …`, `from my_env.grader import …`), found in the listing.""" |
| here = posixpath.dirname(env_path) |
| out: list[str] = [] |
| for level, module, _names in imports: |
| rel = module.replace(".", "/") |
| bases = [] |
| if level: |
| base = here |
| for _ in range(level - 1): |
| base = posixpath.dirname(base) |
| bases.append(base) |
| else: |
| bases += [root, "", posixpath.dirname(root)] |
| _, _, rest = rel.partition("/") |
| if rest: |
| bases += [root] |
| rel_alt = rest |
| else: |
| rel_alt = None |
| for base in bases: |
| for r in filter(None, [rel, None if level else rel_alt]): |
| for cand in (f"{base}/{r}.py" if base else f"{r}.py", f"{base}/{r}/__init__.py" if base else f"{r}/__init__.py"): |
| cand = posixpath.normpath(cand) |
| if cand in paths and cand != env_path and cand not in out: |
| out.append(cand) |
| return out[:6] |
|
|
|
|
| def _env_root(paths: list[str], manifest: str | None) -> str: |
| if manifest: |
| return posixpath.dirname(manifest) |
| cands = [posixpath.dirname(posixpath.dirname(p)) for p in paths if p.endswith("server/app.py")] + \ |
| [posixpath.dirname(p) for p in paths if p.endswith("models.py") and "test" not in p] |
| return min(cands, key=lambda d: (d.count("/") if d else -1, d)) if cands else "" |
|
|
|
|
| PARSERS = {"manifest": parse_manifest, "models": parse_models, "app": parse_app, "environment": parse_environment, |
| "dockerfile": parse_dockerfile, "pyproject": parse_pyproject, "discovery": parse_discovery} |
|
|
|
|
| def declared(spec: str) -> dict[str, Any]: |
| """The curated tree and what the environment's files declare, at the Space's current revision. Never wakes it.""" |
| info = _info(spec) |
| sha = info["sha"] |
|
|
| def build(): |
| files, dropped, truncated = listing(info) |
| sizes = {f["path"]: f["size"] for f in files if not f.get("binary")} |
| paths = [f["path"] for f in files] |
| docker = _fetch_text(spec, sha, "Dockerfile", sizes["Dockerfile"]) if "Dockerfile" in sizes else None |
| manifest = manifest_path(paths, docker, spec) |
| root = _env_root(paths, manifest) |
| keys = key_files(paths, root, manifest) |
| want = {role: p for role, p in keys.items() if role in PARSERS and p in sizes} |
| texts: dict[str, str | None] = {} |
| with ThreadPoolExecutor(6) as pool: |
| for role, text in zip(want, pool.map(lambda p: docker if p == "Dockerfile" and docker is not None else _fetch_text(spec, sha, p, sizes[p]), want.values())): |
| texts[role] = text |
| out: dict[str, Any] = {} |
| for role, p in want.items(): |
| if texts.get(role) is None: |
| out[role] = {"path": p, "error": "too large to read here" if sizes[p] > MAX_TEXT else "couldn't be read"} |
| continue |
| try: |
| out[role] = PARSERS[role](texts[role], p) |
| except Exception: |
| out[role] = {"path": p, "error": "couldn't be read"} |
| env = out.get("environment") |
| if isinstance(env, dict) and env.get("reward_imports"): |
| env["reward_files"] = reward_files(env["path"], env.pop("reward_imports"), set(paths), root) |
| more = [p for p in env["reward_files"] if p in sizes][:2] |
| for p, text in zip(more, ThreadPoolExecutor(2).map(lambda p: _fetch_text(spec, sha, p, sizes[p]), more)): |
| got = parse_environment(text, p) if text else None |
| if got and not got.get("error"): |
| env["llm_judge"] = env["llm_judge"] or got["llm_judge"] |
| env["rubric"] = env["rubric"] or got["rubric"] |
| env["llm_modules"] = sorted(set(env["llm_modules"]) | set(got["llm_modules"]))[:6] |
| env["functions"] = sorted(set(env["functions"]) | {f"{posixpath.basename(p)[:-3]}.{f}" for f in got["functions"][:12]})[:30] |
| elif isinstance(env, dict): |
| env.pop("reward_imports", None) |
| card = info["card"] |
| out["card"] = {k: (card.get(k) if isinstance(card.get(k), (int, float)) else _str(card.get(k), 400)) |
| for k in ("title", "short_description", "base_path", "app_port", "license", "sdk") if card.get(k) is not None} |
| return {"id": spec, "sha": sha, "short": sha[:7], "subdomain": info["subdomain"], "root": root, "keys": keys, |
| "files": files, "total": len(files), "dropped": dropped, "truncated": truncated, |
| "hardware": info["hardware"] or info["requested_hardware"], "stage": info["stage"], "updated": info["updated"], |
| "declared": out, "read_at": time.time(), |
| "partial": any(v.get("error") == "couldn't be read" for v in out.values() if isinstance(v, dict))} |
|
|
| key = ("space-files", spec, sha) |
| res = catalog._cached(key, PARSE_TTL, build) |
| if res.get("partial") and time.time() - res["read_at"] > 60: |
| catalog._memo.pop(key, None) |
| res = catalog._cached(key, PARSE_TTL, build) |
| return res |
|
|
|
|
| def file(spec: str, f: str) -> dict[str, Any]: |
| """One file of the curated listing at the listed revision: text (the first MAX_TEXT bytes of a longer one), |
| uv.lock collapsed to its pins, withheld files and binaries refused.""" |
| f = (f or "").strip() |
| if not f or len(f) > 512 or f.startswith("/") or "\\" in f or "\x00" in f or ".." in f.split("/") or posixpath.normpath(f) != f: |
| raise ValueError("not a file of this Space") |
| info = _info(spec) |
| files, _, _ = listing(info) |
| row = next((x for x in files if x["path"] == f), None) |
| if row is None: |
| raise LookupError("not a file of this Space") |
| out: dict[str, Any] = {"path": f, "size": row["size"], "sha": info["sha"]} |
| if row.get("withheld"): |
| return {**out, "withheld": True} |
| if row.get("binary"): |
| raise BinaryFile("a binary file: open it on the Hub") |
| if row["size"] > MAX_FETCH: |
| raise TooLarge(f"too large to show here ({row['size'] / 1048576:.1f} MB): open it on the Hub") |
|
|
| raw = _download(info["id"], info["sha"], f, MAX_FETCH + 1) |
| text = _decode(raw if row.get("collapsed") else raw[:MAX_TEXT]) |
| if text is None: |
| raise BinaryFile("a binary file: open it on the Hub") |
| if row.get("collapsed"): |
| return {**out, "text": collapse_lock(text), "collapsed": True} |
| if not MANIFESTS.search(f) and catalog._answer_data(f, text): |
| return {**out, "withheld": True} |
| return {**out, "text": text, "truncated": len(raw) > MAX_TEXT} |
|
|
|
|
| class BinaryFile(Exception): |
| pass |
|
|
|
|
| class TooLarge(Exception): |
| pass |
|
|
|
|
| |
| def _errors(fn, *args): |
| from huggingface_hub.errors import HfHubHTTPError, RepositoryNotFoundError |
|
|
| try: |
| return fn(*args) |
| except (RepositoryNotFoundError, PermissionError): |
| raise HTTPException(404, "No such Space, or it is private.") |
| except ValueError as e: |
| raise HTTPException(400, str(e)) |
| except LookupError as e: |
| raise HTTPException(404, str(e)) |
| except BinaryFile as e: |
| raise HTTPException(415, str(e)) |
| except TooLarge as e: |
| raise HTTPException(413, str(e)) |
| except HfHubHTTPError as e: |
| code = getattr(getattr(e, "response", None), "status_code", 0) |
| if code in (401, 403, 404): |
| raise HTTPException(404, "No such Space, or it is private.") |
| raise HTTPException(502, f"could not reach the Hub (HTTP {code})") |
| except (OSError, httpx.HTTPError) as e: |
| raise HTTPException(502, f"could not reach the Hub: {type(e).__name__}") |
|
|
|
|
| @router.get("/api/spaces/{org}/{name}/files") |
| def space_files(org: str, name: str): |
| """The Space's files (vendored OpenEnv, build output and caches left out) and what they declare: openenv.yaml, |
| models.py's classes, server/app.py, the environment's reward code, the Dockerfile and pyproject. Never wakes it.""" |
| return _errors(declared, f"{org}/{name}") |
|
|
|
|
| @router.get("/api/spaces/{org}/{name}/file") |
| def space_file(org: str, name: str, f: str = ""): |
| """One text file of the Space at the listed revision (only files in the listing; no binaries).""" |
| return _errors(file, f"{org}/{name}", f) |
|
|
