| """Recent, non-executing OpenEnv API evidence, separate from the immutable catalog. |
| |
| No wake, reset, step or tool invocation. Only public running Spaces are probed. |
| Per-Space atomic files work on both local storage and the mounted bucket. Search |
| reads a bounded cached inventory; it never waits for the network. |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import logging |
| import os |
| import re |
| import threading |
| import time |
| import tomllib |
| from concurrent.futures import ThreadPoolExecutor |
|
|
| import httpx |
| from packaging.requirements import InvalidRequirement, Requirement |
| from packaging.version import InvalidVersion, Version |
|
|
| from . import catalog, config |
|
|
| FRESH = 3600 |
| REFRESH = 1800 |
| PASS = "API checked" |
| UNKNOWN = "Not checked" |
| _cache = (None, 0.0, {}) |
| _lock = threading.Lock() |
| _write_lock = threading.Lock() |
| _refreshing = {} |
| _stop = threading.Event() |
| _worker = None |
| log = logging.getLogger("rlx") |
|
|
|
|
| def _path(spec): |
| return config.STORAGE_DIR / "openenv-checks" / f"{catalog._slug(catalog.check_spec(spec))}.json" |
|
|
|
|
| def save(rec): |
| global _cache |
| path = _path(rec["id"]) |
| |
| with _write_lock: |
| try: |
| old = json.loads(path.read_bytes()) |
| if old.get("checked_at", 0) > rec["checked_at"]: |
| return |
| except (OSError, ValueError): |
| pass |
| catalog.atomic_write(path, json.dumps(rec, separators=(",", ":")).encode()) |
| with _lock: |
| directory = path.parent |
| records = dict(_cache[2]) if _cache[0] == directory else {} |
| records[rec["id"]] = rec |
| _cache = (directory, _cache[1] if _cache[0] == directory else 0.0, records) |
|
|
|
|
| def _read_inventory(directory): |
| """Public evidence only; invalid or oversized files are ignored, never trusted.""" |
| records = {} |
| for path in sorted(directory.glob("*.json"))[:20000]: |
| try: |
| if path.stat().st_size > 32 * 1024: |
| continue |
| r = json.loads(path.read_bytes()) |
| spec = catalog.check_spec(r["id"]) |
| if r.get("schema") == 1 and isinstance(r.get("checked_at"), (float, int)): |
| records[spec] = r |
| except (OSError, ValueError, KeyError, TypeError): |
| continue |
| return records |
|
|
|
|
| def _refresh_inventory(directory, done): |
| global _cache |
| try: |
| records = _read_inventory(directory) |
| with _lock: |
| if _cache[0] == directory: |
| |
| |
| for spec, rec in _cache[2].items(): |
| if rec.get("checked_at", 0) > records.get(spec, {}).get("checked_at", 0): |
| records[spec] = rec |
| _cache = (directory, time.time(), records) |
| except Exception: |
| log.exception("Could not read stored OpenEnv checks") |
| finally: |
| with _lock: |
| _refreshing.pop(directory, None) |
| done.set() |
|
|
|
|
| def inventory(*, wait=False): |
| """Serve cached evidence immediately; refresh bucket files off the request path. |
| |
| Cold readers see no verified Spaces until the first read completes. Expiry |
| still uses checked_at. Only the background checker waits for storage so it |
| does not re-probe everything on boot. |
| """ |
| global _cache |
| directory = config.STORAGE_DIR / "openenv-checks" |
| with _lock: |
| if _cache[0] != directory: |
| _cache = (directory, 0.0, {}) |
| if time.time() - _cache[1] < 60: |
| return _cache[2] |
| done = _refreshing.get(directory) |
| if done is None: |
| done = _refreshing[directory] = threading.Event() |
| threading.Thread(target=_refresh_inventory, args=(directory, done), daemon=True, |
| name="read-openenv-checks").start() |
| records = _cache[2] |
| if not wait: |
| return records |
| done.wait() |
| with _lock: |
| return _cache[2] |
|
|
|
|
| def status(rec, now=None): |
| if not rec: |
| return UNKNOWN |
| age = (time.time() if now is None else now) - rec.get("checked_at", 0) |
| if age < 0 or age >= FRESH: |
| return "Check expired" |
| return rec.get("status") if rec.get("status") in (PASS, "Checks failed", "Not running", "Check unavailable") else UNKNOWN |
|
|
|
|
| def verified(rec): |
| """Only a fresh successful check of a running server establishes API support.""" |
| return bool(rec and rec.get("stage") == "RUNNING" and status(rec) == PASS) |
|
|
|
|
| def browseable(rec): |
| """A live, recognized environment API; this does not certify reward quality.""" |
| return verified(rec) or bool(rec and rec.get("stage") == "RUNNING" |
| and status(rec) == "Checks failed" |
| and rec.get("interface") in ("nemo-gym", "ors")) |
|
|
|
|
| def environment_interface(api): |
| """Other supported server contracts, kept distinct from OpenEnv evidence.""" |
| if not isinstance(api, dict) or not isinstance(api.get("info"), dict) or not api["info"].get("version"): |
| return None |
| paths = api.get("paths") |
| if not isinstance(paths, dict): |
| return None |
| for name, routes in (("nemo-gym", ("/seed_session", "/verify")), ("ors", ("/create_session", "/{env_name}/call"))): |
| if all(isinstance(paths.get(p), dict) and isinstance(paths[p].get("post"), dict) for p in routes): |
| return name |
| return None |
|
|
|
|
| def tag_version(tags): |
| for tag in tags or []: |
| match = re.fullmatch(r"openenv-(\d[^\s]{0,70})", str(tag), re.I) |
| if match: |
| try: |
| return {"value": str(Version(match[1])), "source": "Hub tag"} |
| except InvalidVersion: |
| pass |
| return {"value": "Unknown", "source": "Unknown"} |
|
|
|
|
| def requirement_version(text): |
| try: |
| req = Requirement(text) |
| except InvalidRequirement: |
| return None |
| if req.name.lower().replace("_", "-") not in ("openenv", "openenv-core"): |
| return None |
| if req.marker: |
| return {"value": "Conditional", "source": "Dependency constraint", "requirement": str(req)[:240]} |
| if req.url: |
| ref = re.search(r"@([^/#?\s]+)(?:#.*)?$", req.url) |
| return {"value": "git:" + ref[1][:64] if ref else "Unpinned source", "source": "Source reference"} |
| specs = list(req.specifier) |
| if len(specs) == 1 and specs[0].operator in ("==", "===") and "*" not in specs[0].version: |
| try: |
| return {"value": str(Version(specs[0].version)), "source": "Dependency pin"} |
| except InvalidVersion: |
| pass |
| return {"value": str(req.specifier) or "Unpinned", "source": "Dependency constraint"} |
|
|
|
|
| def parse_version(text, path): |
| """Repository evidence, never an assertion about the installed runtime.""" |
| if path.endswith(".toml") or path.endswith("uv.lock"): |
| try: |
| doc = tomllib.loads(text) |
| except (ValueError, RecursionError): |
| return None |
| if path.endswith("uv.lock"): |
| pins = [] |
| for pkg in doc.get("package", []): |
| if not isinstance(pkg, dict) or pkg.get("name", "").replace("_", "-") not in ("openenv", "openenv-core"): |
| continue |
| source = pkg.get("source") or {} |
| if source.get("git"): |
| ref = source["git"].rsplit("#", 1) |
| pins.append({"value": "git:" + ref[-1][:64] if len(ref) == 2 else "Unpinned source", "source": "Lockfile"}) |
| elif pkg.get("version"): |
| pins.append(requirement_version(f"openenv-core=={pkg['version']}")) |
| pins = [p for p in pins if p] |
| if len({p["value"] for p in pins}) == 1: |
| return {**pins[0], "source": "Lockfile"} |
| return {"value": "Multiple locked versions", "source": "Lockfile"} if pins else None |
| sources = doc.get("tool", {}).get("uv", {}).get("sources", {}) |
| source = sources.get("openenv") or sources.get("openenv-core") or sources.get("openenv_core") |
| if isinstance(source, dict) and source.get("git"): |
| ref = source.get("rev") or source.get("tag") or source.get("branch") |
| return {"value": "git:" + str(ref)[:64] if ref else "Unpinned source", "source": "Source reference"} |
| if source: |
| return {"value": "Custom source", "source": "Source reference"} |
| deps = doc.get("project", {}).get("dependencies", []) |
| else: |
| deps = [line.split(" #", 1)[0].strip() for line in text.splitlines()] |
| for dep in deps: |
| if isinstance(dep, str) and (v := requirement_version(dep)): |
| return v |
| return None |
|
|
|
|
| def declared_version(spec, tags): |
| from . import space_files as files |
| fallback = tag_version(tags) |
| try: |
| info = files._info(spec) |
| listing, _, _ = files.listing(info) |
| paths = [f["path"] for f in listing] |
| manifest = files.manifest_path(paths, "Dockerfile" if "Dockerfile" in paths else None, spec) |
| root = manifest.rsplit("/", 1)[0] if manifest and "/" in manifest else "" |
| keys = files.key_files(paths, root, manifest) |
| sizes = {f["path"]: f.get("size", 0) for f in listing} |
| reqs = [f"{root}/requirements.txt" if root else "requirements.txt", "requirements.txt"] |
| for path in dict.fromkeys([keys.get("lock"), keys.get("pyproject"), *reqs]): |
| if not path or path not in sizes or sizes[path] > 400 * 1024: |
| continue |
| text = files._fetch_text(spec, info["sha"], path, sizes[path]) |
| if text and (v := parse_version(text, path)): |
| return {**v, "path": path, "revision": info["sha"]} |
| except Exception: |
| pass |
| return fallback |
|
|
|
|
| def assess(api, health, metadata, schema, tools): |
| api = api if isinstance(api, dict) else {} |
| paths = api.get("paths") if isinstance(api.get("paths"), dict) else {} |
| def method(path, verb): |
| return isinstance(paths.get(path), dict) and isinstance(paths[path].get(verb), dict) |
| checks = { |
| "OpenAPI": isinstance(api.get("info"), dict) and isinstance(api["info"].get("version"), str), |
| "Health": isinstance(health, dict) and health.get("status") == "healthy", |
| "Metadata": isinstance(metadata, dict) and all(isinstance(metadata.get(k), str) for k in ("name", "description")), |
| "Schemas": isinstance(schema, dict) and all(isinstance(schema.get(k), dict) for k in ("action", "observation", "state")), |
| } |
| simulation = any(p in paths for p in ("/reset", "/step", "/state")) |
| mode = "Simulation" if simulation else "MCP service" |
| checks["Control routes" if simulation else "MCP tools"] = ( |
| all(method(p, v) for p, v in (("/reset", "post"), ("/step", "post"), ("/state", "get"))) |
| if simulation else method("/mcp", "post") and bool(tools)) |
| return {"status": PASS if all(checks.values()) else "Checks failed", "mode": mode if all(checks.values()) else None, |
| "failed": [k for k, ok in checks.items() if not ok], "tools": len(tools or [])} |
|
|
|
|
| def _task_splits(task_api): |
| from .seo_tasks import ranges |
|
|
| out = [] |
| for row in ranges(task_api): |
| if len(json.dumps(out + [row]).encode()) > 12000: |
| break |
| out.append(row) |
| return out |
|
|
|
|
| def check(spec): |
| from . import spaces_live as live |
| rec = {"schema": 1, "id": catalog.check_spec(spec), "checked_at": time.time(), "status": "Check unavailable"} |
| previous = inventory().get(spec, {}) |
| rec["version"] = previous.get("version") or {"value": "Unknown", "source": "Unknown"} |
| try: |
| hub = live.record(spec, fresh=True) |
| rec["stage"] = hub["stage"] |
| if hub["stage"] != "RUNNING": |
| rec["status"] = "Not running" |
| elif hub.get("host"): |
| urls = ("/openapi.json", "/health", "/metadata", "/schema") |
| def get(path): |
| try: |
| with live.SpaceClient(timeout=httpx.Timeout(6, connect=4), follow_redirects=False) as client: |
| response = client.get(live._url(hub, path)) |
| return response.json() if response.status_code == 200 and "json" in response.headers.get("content-type", "") else None |
| except (httpx.HTTPError, ValueError, live.SpaceError): |
| return None |
| with ThreadPoolExecutor(4) as pool: |
| api, health, meta, schema = list(pool.map(get, urls)) |
| |
| paths = api.get("paths", {}) if isinstance(api, dict) else {} |
| typed = isinstance(schema, dict) and all(isinstance(schema.get(k), dict) for k in ("action", "observation", "state")) |
| tools, _, _ = live._tools({**hub, "mcp_standard": not typed}) if isinstance(paths, dict) and "/mcp" in paths else (None, None, None) |
| rec.update(assess(api, health, meta, schema, tools)) |
| rec["interface"] = environment_interface(api) |
| from . import space_tasks |
| task_api = space_tasks.discover(hub, paths) |
| rec["task_catalog"] = space_tasks.summary(task_api) |
| rec["task_splits"] = _task_splits(task_api) |
| if rec["status"] == PASS: |
| rec["version"] = declared_version(spec, hub.get("tags")) |
| elif rec["version"]["value"] == "Unknown": |
| rec["version"] = tag_version(hub.get("tags")) |
| except Exception as exc: |
| rec["error"] = type(exc).__name__ |
| save(rec) |
| return rec |
|
|
|
|
| def observe(spec, info): |
| """Opening/rechecking a Space also updates discovery, including negative results.""" |
| try: |
| old = inventory().get(spec, {}) |
| rec = {"schema": 1, "id": spec, "checked_at": info.get("checked") or time.time(), "stage": info.get("stage"), |
| "version": old.get("version") or {"value": "Unknown", "source": "Unknown"}} |
| if not info.get("running"): |
| rec["status"] = "Not running" |
| else: |
| rec.update(assess({"info": {"version": info.get("openapi_version")}, "paths": info.get("endpoint_methods")}, |
| info.get("health"), info.get("metadata"), info.get("schema"), info.get("mcp"))) |
| rec["interface"] = environment_interface({"info": {"version": info.get("openapi_version")}, "paths": info.get("endpoint_methods")}) |
| from . import space_tasks |
| rec["task_catalog"] = space_tasks.summary(info.get("task_api")) |
| rec["task_splits"] = _task_splits(info.get("task_api")) |
| info["openenv_verified"] = verified(rec) |
| save(rec) |
| except Exception: |
| log.warning("Could not save OpenEnv check for %s", spec) |
|
|
|
|
| def scan(): |
| |
| |
| try: |
| import fcntl |
| except ImportError: |
| return _scan() |
| config.CACHE_DIR.mkdir(parents=True, exist_ok=True) |
| with (config.CACHE_DIR / "openenv-checks.lock").open("a") as lease: |
| try: |
| fcntl.flock(lease, fcntl.LOCK_EX | fcntl.LOCK_NB) |
| except BlockingIOError: |
| return [] |
| return _scan() |
|
|
|
|
| def _scan(): |
| from . import snapshot |
| old = inventory(wait=True) |
| with snapshot.use() as (_, conn): |
| candidates = list(conn.execute("SELECT id, stage FROM envs WHERE kind = 'space' ORDER BY trending DESC, likes DESC")) |
| due = pending(candidates, old) |
| def visit(spec): |
| if not _stop.is_set(): |
| return check(spec) |
| with ThreadPoolExecutor(4) as pool: |
| return [r for r in pool.map(visit, due) if r] |
|
|
|
|
| def pending(candidates, old, now=None): |
| """Bound the queue and reserve capacity for inactive candidates too. |
| |
| Never probing sleeping hosts, but refreshing their Hub metadata lets a |
| newly started Space enter the catalog without someone opening its page. |
| """ |
| now = time.time() if now is None else now |
| active, inactive = [], [] |
| for spec, stage in candidates: |
| rec = old.get(spec, {}) |
| running = rec.get("stage", stage) == "RUNNING" |
| at = rec.get("checked_at", 0) |
| interval = REFRESH if running else 6 * 3600 |
| if not rec or now - at >= interval or (running and status(rec, now) == PASS and "task_catalog" not in rec): |
| (active if running else inactive).append((spec, at)) |
| |
| |
| key = lambda item: (int(item[1] // REFRESH), item[0].split("/", 1)[0].lower() != "fineenvs", item[1]) |
| active.sort(key=key) |
| inactive.sort(key=key) |
| return [s for s, _ in active[:96] + inactive[:32]] |
|
|
|
|
| def start(): |
| global _worker |
| if os.environ.get("RLX_TEST_TMP") or os.environ.get("RLX_SPACE_CHECKS", "1") != "1" or (_worker and _worker.is_alive()): |
| return |
| _stop.clear() |
| def watch(): |
| while not _stop.is_set(): |
| try: |
| scan() |
| except Exception: |
| log.exception("OpenEnv API inventory refresh failed") |
| _stop.wait(60) |
| _worker = threading.Thread(target=watch, daemon=True, name="openenv-checks") |
| _worker.start() |
|
|
|
|
| def stop(): |
| _stop.set() |
|
|
|
|
| if __name__ == "__main__": |
| from collections import Counter |
| from . import runtime |
| runtime.hub_timeouts() |
| print(dict(Counter(r["status"] for r in scan())), flush=True) |
|
|
