diff --git a/CLAUDE.md b/CLAUDE.md index c207217..d0cf8a3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -80,6 +80,13 @@ pytest/ruff config). Each folder with commands has its own `python -m`. record-format spec: the environment's public interface. - `lab/RUNTIME.md` — design (2026-10-08): controller (laptop: ledger, referee, secrets) and worker (GPU VM: jails, clients, profilers) behind a job interface; trust zones and rules; why srt cannot jail GPU code on Linux. +- `lab/build.py` — the dependency build step: a tree whose change touched the target's `deps` (pyproject: dependency + tables only, checked by the audit) gets a venv built by the target's `build` command in a no-GPU, `--network none` + container whose only way out is a CONNECT proxy to the package-index hosts; cached by the files' hash; jobs run + under `engine.using(build.python_for(tree))`. +- `lab/gpucheck.py` — `python -m lab.gpucheck` on a GPU VM (as root, like the worker): what the jails need from a real + GPU host, PASS/FAIL each: jail GPU, workbench (GPU, root, apt/pip, metadata blocked), quiet, exclusivity, ncu, + hostprof, the dependency build. - `lab/worker.py` — the job interface, built for the measured jobs: `test`, `measure`, `equiv`, `profile` take input trees + JSON args, write an out dir, return JSON; the tools call `self.worker`. `LAB_WORKER=local` (default, in process) or `ssh` (the `lab.vm` VM): inputs by content hash, re-hashed on the worker before every job; GpuBusy, @@ -246,6 +253,20 @@ Linux the jail is a container with the NVIDIA runtime (`lab/safety/container.py` check (`vm-setup.sh` prints `jail: gpu ok`) has not run yet. The agent's CLI still runs under srt, which on Ubuntu 24.04 needs `kernel.apparmor_restrict_unprivileged_userns=0` (in `vm-setup.sh`). +## First agent dry run (Nebius H200, 2026-10-10) +`LAB_WORKER=ssh`, laptop controller, `claude-fable-5-1`, one session ($2.53 as the proxy priced it, 24 turns, 465 s; +~1 h 20 min of GPU for the whole run). Validated on the real GPU: the GPU in the network-less jail, metadata blocked, +base and candidate served jailed through the port bridge, `profile`, `trace` (after mounting nsys's install root), +`test` (after caching the tests' gpt2 tokenizer), the remote workbench over `ssh -R`, the proxy's pricing. The agent +fused RMSNorm/RoPE in Triton and added a batch-1 decode graph: seen-split short-tier `single_stream` tpot p99 21.4 -> +13.1 ms; not submitted, correctness not run (equiv needs `data/equiv/ref`, absent in a worktree). A base dev-tier equiv +on this engine took > 46 min for 774 of 1,597 items at concurrency 16 (vLLM: ~10 min): the tool descriptions' "10 min +at tier dev" is wrong for this engine. Workbench processes cannot change clocks or power limits on that platform. +`python -m lab.gpucheck` then passed every item on Nebius H200s (2026-10-13, three runs, fixes between them): +jail GPU, quiet, exclusivity, hostprof; workbench (after blocking only 169.254.169.254: the /16 took DNS with it); +the dependency build (after allowing pypi.nvidia.com); ncu (after linking the real /opt/nvidia install). The jail's +user can read GPU performance counters on that image. + ## Next GPU (needs spend approval): resolve the BF16 puzzle (plain BF16 on the dev tier at concurrency 32 and 128 on one VM, ~20 min, ~$1.50); first real run of `python -m regimes run all` against the engine and against vLLM. diff --git a/lab/RUNTIME.md b/lab/RUNTIME.md index 0489eb4..fd3a748 100644 --- a/lab/RUNTIME.md +++ b/lab/RUNTIME.md @@ -4,8 +4,8 @@ Design, 2026-10-08. Decided with the user: the lab splits into a controller and measurement VM (cost); NVIDIA only for now; the jail on Linux is a container with the NVIDIA runtime (option A below). Built: the container jail (`lab/safety/container.py`), GPU exclusivity during measurements, and the measured jobs behind the worker interface (`lab/worker.py`), the API proxy (`lab/apiproxy.py`), the workbench -container with its pause during measurements, and the agent on a remote worker. Not built: the dependency build -step. Open decisions are marked **Decide**. +container with its pause during measurements, the agent on a remote worker, and the dependency build step +(`lab/build.py`). Open decisions are marked **Decide**. ## The line (the user, 2026-10-09) @@ -173,7 +173,8 @@ not yet over SSH to a VM. `agent`, `exec`, `snapshot`, `restore` and `facts` com interpreter, weights, CUDA and the CLI binary read-only, so an install in the workbench never reaches a measurement. Container root is uid 0 on the host kernel (no user-namespace remap): an escape is host root, accepted under decision 5; `userns-remap` is the hardening step. `vm-setup.sh` drops container traffic to - 169.254.0.0/16 (the cloud metadata service) and checks it. + 169.254.169.254 (the cloud metadata service; only that address, a cloud's DNS can sit elsewhere in 169.254/16) and + checks both that and the workbench's internet. - **Pause** (`container.quiet`): around every `measure`, `equiv` and `profile` job, processes in the workbench holding the GPU are killed, the container is paused, a second sweep catches any that started before the pause landed, and it waits for their memory to be released; thawed after. The killed processes are in the record and @@ -206,10 +207,31 @@ on the VM: on the VM, mounted into the workbench, where a socat serves it on the container's own 127.0.0.1:4000. It listens on no interface of the VM. - The write guard checks the VM's workspace path; the workbench is removed over SSH when the session ends. +- The workbench is frozen on the VM (`python -m lab.worker pause`) for the whole of every lab tool call, so the root + rsync of a pull or push never races a process in it swapping a file or a directory for a link (2026-10-13 review: + a swapped directory could have carried the held-out split out, or let root write anywhere on the VM). - Tested without a VM: rsync mirroring (content edits, deletions, links), the VM-side `workbench` command, and the argument quoting through bash and ssh (an `ssh` stand-in that prints the remote command). Not yet tested against a real VM. +## The dependency build step, as built (2026-10-10) + +- The target names its dependency files (`deps`) and the command that installs them (`build`, run with + `$UV_PROJECT_ENVIRONMENT` set). For inference-server that is vm-setup.sh's recipe: `uv sync --locked`, then the + CUDA torch wheel at the locked version. +- The agent may change those files; a `pyproject.toml` only in its dependency tables (`[project] dependencies`, + `optional-dependencies`, `[dependency-groups]`, `[tool.uv] sources` and `index`): any other change, its test and + lint config included, is a violation, and deleting a dependency file is one too. +- Each job asks `build.python_for(tree)`: a tree whose change left the dependency files as in its base (HEAD~1 of + the two-commit pristine tree) runs on the engine's venv; a changed one gets a venv built for it, cached by the + files' hash. The build runs in a measurement-room container without the GPU and with `--network none`; its only way + out is a host-side CONNECT proxy, over a unix socket, to pypi.org, files.pythonhosted.org and the https hosts its + files and command name. A failed build is a failed `test`/`profile` or an `error` record carrying the build's + output. The tool descriptions state it as a fact: only the change reaches the engine, installed from scratch, and + the engine then runs offline. +- Tested in Docker-in-Docker: the build room gets 200 from PyPI through the proxy, 403 for another host, and no + direct route out. The real inference-server build (torch, CUDA) is untested until the GPU run. + ## What moves where - Controller: `session`, `agent` (SDK side), the tool front ends in `tools`/`evaltools`/`proftools` (policy, diff --git a/lab/agent.py b/lab/agent.py index 836034d..8debf11 100644 --- a/lab/agent.py +++ b/lab/agent.py @@ -121,6 +121,7 @@ class ClaudeAgentSDK: srt); its model calls go through the lab's API proxy, so no credential is ever inside; lab tools run out here.""" name = "claude" + remote = True # can work on a Remote worker's host (its CLI over ssh) @staticmethod def containerized() -> bool: @@ -133,7 +134,10 @@ def _wrapper(self, spec: AgentSpec, cli: Path, base_url: str) -> Path: w = spec.remote.workbench(spec.workbench, spec.scratch.name, spec.remote_workspace, claude_agent_sdk.__version__) *ssh, target = spec.remote.prefix + # The SDK may launch this more than once, and that sshd need not unlink a socket a launch left behind; + # -n: that call must not read the SDK's stdin. line = (f'args=$(printf " %q" "$@")\ncmd={shlex.quote(shlex.join(w["argv"]))}\n' + f'{shlex.join(ssh)} -n -T {target} {shlex.quote("rm -f " + shlex.quote(w["sock"]))} || exit 1\n' f'exec {shlex.join(ssh)} -T -o ExitOnForwardFailure=yes -o StreamLocalBindUnlink=yes ' f'-R {w["sock"]}:127.0.0.1:{urlsplit(base_url).port} {target} "$cmd$args"') elif self.containerized(): @@ -148,7 +152,10 @@ def _wrapper(self, spec: AgentSpec, cli: Path, base_url: str) -> Path: cli_tmp = Path(f"{tmp_root}/claude-{os.getuid()}") # the CLI's own per-project scratch; Bash fails without it cli_tmp.mkdir(parents=True, exist_ok=True) config = jail.settings([home.resolve(), tmp.resolve(), ws, cli_tmp.resolve()], engine.venv(), - [base_url.removeprefix("http://")], readonly=_weights(), python=engine.python()) + [base_url.removeprefix("http://")], readonly=_weights(), python=engine.python(), + local_binding=True) # srt on macOS reaches 127.0.0.1 only with it (and then + # every local port: dev only; measurements that count run + # on Linux, where each jail has its own network) argv = jail.wrap(config, spec.scratch / "srt.json", [str(cli)]) keep = " ".join(f'"{k}=${k}"' for k in PASS_ENV) sets = " ".join(f"{k}={shlex.quote(v)}" for k, v in cli_env(home, tmp, ws, base_url).items()) @@ -162,10 +169,20 @@ async def _run(self, spec: AgentSpec, base_url: str) -> AgentReply: from claude_agent_sdk import (ClaudeAgentOptions, HookMatcher, ResultMessage, create_sdk_mcp_server, query, tool) + # The workbench is frozen for every lab tool call, wherever it runs: nothing in it can swap a file or a dir for a + # link while the tool snapshots, restores, copies into the workspace, or (Remote) rsyncs it as root on the VM. + if spec.remote is not None: + frozen = lambda: spec.remote.paused(spec.workbench) # noqa: E731 + else: + frozen = lambda: container.paused(spec.workbench if self.containerized() else None) # noqa: E731 + def adapt(t: ToolSpec): async def call(args: dict[str, Any]) -> dict[str, Any]: try: - text = await asyncio.to_thread(t.fn, args) # a long test run must not block the transport + def run(): + with frozen(): + return t.fn(args) + text = await asyncio.to_thread(run) # a long test run must not block the transport except Exception as e: # the agent sees the refusal, the harness keeps going text = f"{t.name} refused: {e}" return {"content": [{"type": "text", "text": text}]} diff --git a/lab/apiproxy.py b/lab/apiproxy.py index d0ee5d1..aed95a6 100644 --- a/lab/apiproxy.py +++ b/lab/apiproxy.py @@ -54,10 +54,11 @@ def credential() -> dict[str, str]: class _Usage: - """Usage read off a response as it streams by: JSON, or SSE message_start + message_delta.""" + """Usage read off a response as it streams by: JSON, or SSE message_start + message_delta. `complete` once the + final output count is known.""" def __init__(self, sse: bool): - self.sse, self.buf, self.model, self.usage = sse, b"", None, {} + self.sse, self.buf, self.model, self.usage, self.complete = sse, b"", None, {}, False def feed(self, chunk: bytes) -> None: self.buf += chunk @@ -78,6 +79,7 @@ def _event(self, data: bytes) -> None: self.usage.update(e["message"].get("usage") or {}) elif e.get("type") == "message_delta": self.usage.update(e.get("usage") or {}) + self.complete = True def done(self) -> None: if self.sse: @@ -88,14 +90,14 @@ def done(self) -> None: except ValueError: return if isinstance(body, dict) and isinstance(body.get("usage"), dict): - self.model, self.usage = body.get("model"), body["usage"] + self.model, self.usage, self.complete = body.get("model"), body["usage"], True class Proxy: """`with Proxy(cap_usd, host) as p: ... p.url`; `p.spent_usd` and `p.usage` are what the session cost.""" def __init__(self, cap_usd: float, host: str = "127.0.0.1", upstream: str = UPSTREAM): - self.cap_usd, self.spent_usd, self.requests = cap_usd, 0.0, 0 + self.cap_usd, self.spent_usd, self.requests, self.reserved_usd = cap_usd, 0.0, 0, 0.0 self.usage = {"input_tokens": 0, "output_tokens": 0, "cache_read_input_tokens": 0, "cache_creation_input_tokens": 0} self.auth = credential() @@ -142,20 +144,27 @@ def _handle(self, h, body: bytes | None) -> None: path = urlsplit(h.path).path if path not in ALLOWED and not (path.startswith("/v1/models/") and path.count("/") == 3): return self._refuse(h, 403, f"{path} is not forwarded") - model = None + model, max_tokens = None, 0 if body: try: req = json.loads(body) except ValueError: return self._refuse(h, 400, "the request body is not JSON") - model = req.get("model") + model, max_tokens = req.get("model"), int(req.get("max_tokens") or 0) if path == "/v1/messages" and (not isinstance(model, str) or price(model) is None): return self._refuse(h, 403, f"no price for model {model!r}; priced: {sorted(PRICES)}") if req.get("speed") == "fast": return self._refuse(h, 403, "fast mode is not priced here") + metered = path == "/v1/messages" + if metered and (not body or not isinstance(model, str) or price(model) is None): + return self._refuse(h, 403, "a /v1/messages request needs a JSON body naming a priced model") + # Its worst case is reserved before it goes (input: at most a token per byte of body; output: all of max_tokens), + # so concurrent requests cannot overshoot the cap. + hold = (len(body) * price(model)[0] + max_tokens * price(model)[1]) / 1e6 if metered else 0.0 with self.lock: - if path == "/v1/messages" and self.spent_usd >= self.cap_usd: - return self._refuse(h, 403, f"this session's ${self.cap_usd:.2f} is spent") + if metered and self.spent_usd + self.reserved_usd + hold > self.cap_usd: + return self._refuse(h, 403, f"this session's ${self.cap_usd:.2f} is spent (or would be by this request)") + self.reserved_usd += hold headers = {k: v for k, v in h.headers.items() if k.lower() not in HOP | {"x-api-key", "authorization"}} headers.update(self.auth, host=self.upstream.netloc, **{"accept-encoding": "identity"}) if "authorization" in self.auth: @@ -163,6 +172,7 @@ def _handle(self, h, body: bytes | None) -> None: headers["anthropic-beta"] = ",".join(dict.fromkeys([*betas, OAUTH_BETA])) conn = (http.client.HTTPSConnection if self.upstream.scheme == "https" else http.client.HTTPConnection)( self.upstream.netloc, timeout=900) + usage = None try: conn.request(h.command, h.path, body=body, headers=headers) resp = conn.getresponse() @@ -178,13 +188,22 @@ def _handle(self, h, body: bytes | None) -> None: h.wfile.write(b"%x\r\n%s\r\n" % (len(chunk), chunk)) h.wfile.flush() h.wfile.write(b"0\r\n\r\n") - usage.done() - finally: + finally: # charged however it ended: a dropped client still spent conn.close() - if path == "/v1/messages" and usage.usage: - spent = cost_usd(usage.model if usage.model and price(usage.model) else model, usage.usage) - with self.lock: - self.spent_usd += spent + if usage is not None: + usage.done() + self._charge(metered and usage is not None, usage, model, max_tokens, hold) + + def _charge(self, metered: bool, usage: "_Usage | None", model: str | None, max_tokens: int, hold: float) -> None: + """An incomplete response is charged its input as far as seen and its whole `max_tokens` of output.""" + got = dict(usage.usage) if usage is not None else {} + if metered and got and not usage.complete: + got["output_tokens"] = max(got.get("output_tokens") or 0, max_tokens) + spent = cost_usd(usage.model if usage.model and price(usage.model) else model, got) if metered and got else 0.0 + with self.lock: + self.reserved_usd -= hold + self.spent_usd += spent + if metered and got: self.requests += 1 for k in self.usage: - self.usage[k] += usage.usage.get(k) or 0 + self.usage[k] += got.get(k) or 0 diff --git a/lab/build.py b/lab/build.py new file mode 100644 index 0000000..b3e3ef3 --- /dev/null +++ b/lab/build.py @@ -0,0 +1,182 @@ +"""The dependency build step (lab/RUNTIME.md). A measurement room is rebuilt from the agent's change alone and runs +offline, so a change to the engine's dependency files (the target's `deps`) is installed from scratch for it: a +fresh venv built by the target's `build` command in a container with no GPU, no secrets and no network but a +CONNECT proxy to the package-index hosts, cached by the dependency files' hash. A tree whose change leaves the dependency +files as they were in its base keeps the engine's venv (built from that base by vm-setup.sh), so the base and an +unchanged candidate run identically. A measured tree is a two-commit repo (base, change; grader.pristine_tree); one +without that history, such as the exported base itself, has no change.""" + +from __future__ import annotations + +import hashlib +import os +import re +import select +import shutil +import socket +import socketserver +import subprocess +import tempfile +import threading +from pathlib import Path + +from lab import engine, target +from lab.safety import container + +# PyPI, plus the wheel hosts that CUDA builds of torch pull from (its nvidia-* dependencies live on NVIDIA's index); +# every https host the dependency files and the build command name is added per build. +INDEX_HOSTS = ("pypi.org", "files.pythonhosted.org", "download.pytorch.org", "download-r2.pytorch.org", + "pypi.nvidia.com") +PROXY_PORT_IN = 3128 +LOG_TAIL = 4000 +TIMEOUT_S = 3600 + + +class BuildFailed(RuntimeError): + """The tree's dependencies could not be installed; the message carries the build's output.""" + + +def deps_sha(tree: Path) -> str: + h = hashlib.sha256() + for rel in target.load().deps: + p = Path(tree) / rel + h.update(rel.encode() + b"\0" + (p.read_bytes() if p.is_file() else b"") + b"\0") + return h.hexdigest() + + +def hosts(tree: Path, build: str) -> set[str]: + """The index hosts a build may reach: the defaults and every https host its dependency files and command name.""" + text = build + "".join((Path(tree) / r).read_text(errors="replace") for r in target.load().deps + if (Path(tree) / r).is_file()) + return set(INDEX_HOSTS) | {m.lower() for m in re.findall(r"https://([A-Za-z0-9.-]+)", text)} + + +def changed(tree: Path) -> bool: + """Whether the tree's change touched a dependency file: its working files against its base commit (HEAD~1).""" + tree = Path(tree) + if not target.load().deps or not (tree / ".git").exists(): + return False + for rel in target.load().deps: + old = subprocess.run(["git", "-C", str(tree), "show", f"HEAD~1:{rel}"], capture_output=True) + if old.returncode != 0 and b"exists on disk, but not in" not in old.stderr and b"does not exist" not in old.stderr: + return False # no base commit: nothing to compare against + now = (tree / rel).read_bytes() if (tree / rel).is_file() else None + if (old.stdout if old.returncode == 0 else None) != now: + return True + return False + + +def python_for(tree: Path) -> str: + """The interpreter that runs `tree`: the engine's own, or a venv built from the tree's changed dependency files.""" + if not changed(tree): + return engine.python() + venv = cache_dir() / deps_sha(tree) + if not (venv / ".lab-built").exists(): + build(tree, venv) + return str(venv / "env" / "bin" / "python") + + +def cache_dir() -> Path: + from lab import worker + return worker.cache_dir() / "venvs" + + +class _Connect(socketserver.BaseRequestHandler): + """One CONNECT host:443 to an allowed host, then bytes both ways; anything else is refused.""" + allowed: set[str] = set() + + def handle(self): + head = b"" + while b"\r\n\r\n" not in head and len(head) < 8192: + chunk = self.request.recv(4096) + if not chunk: + return + head += chunk + line = head.split(b"\r\n", 1)[0].decode(errors="replace").split() + host, _, port = (line[1] if len(line) > 1 else "").rpartition(":") + if len(line) < 2 or line[0] != "CONNECT" or port != "443" or host.lower() not in self.allowed: + self.request.sendall(b"HTTP/1.1 403 Forbidden\r\ncontent-length: 0\r\n\r\n") + return + try: + up = socket.create_connection((host, 443), timeout=30) + except OSError: + self.request.sendall(b"HTTP/1.1 502 Bad Gateway\r\ncontent-length: 0\r\n\r\n") + return + self.request.sendall(b"HTTP/1.1 200 Connection established\r\n\r\n") + with up: + pair = (self.request, up) + while True: + ready, _, _ = select.select(pair, [], [], 300) + if not ready: + return + for s in ready: + data = s.recv(65536) + if not data: + return + (up if s is self.request else self.request).sendall(data) + + +class IndexProxy: + """`with IndexProxy(hosts, sock):` a CONNECT-only proxy on unix socket `sock` that reaches `hosts` and nothing else.""" + + def __init__(self, allowed: set[str], sock: Path): + handler = type("Handler", (_Connect,), {"allowed": {h.lower() for h in allowed}}) + self.server = socketserver.ThreadingUnixStreamServer(str(sock), handler) + self.server.daemon_threads = True + os.chmod(sock, 0o666) # the build runs as the jail's user + + def __enter__(self): + threading.Thread(target=self.server.serve_forever, daemon=True).start() + return self + + def __exit__(self, *exc): + self.server.shutdown() + self.server.server_close() + + +def build(tree: Path, venv: Path) -> None: + """The target's `build` in a container: the tree's dependency files only, the engine's interpreter and uv read-only, + uv's cache and `venv` writable (uv makes the environment at `venv/env`), the index proxy as the only way out. + A failed build leaves no venv behind.""" + t = target.load() + if not t.build: + raise BuildFailed("the dependency files changed, and the target has no `build` command to install them") + if not container.available(): + raise BuildFailed("a dependency build needs the container jail (docker and the lab-jail image)") + uv = shutil.which("uv") or str(Path.home() / ".local" / "bin" / "uv") + interpreter = Path(engine.python()).resolve() + venv.parent.mkdir(parents=True, exist_ok=True) + uv_cache = cache_dir().parent / "uv-cache" + uv_cache.mkdir(parents=True, exist_ok=True) + work = Path(tempfile.mkdtemp(prefix="build-", dir=cache_dir())) + src, sock_dir = work / "src", Path(tempfile.mkdtemp(prefix="lab-index-")) # short: a unix socket path + sock_dir.chmod(0o711) # the jail's user reaches the socket, lists nothing + src.mkdir() + for rel in t.deps: + if (Path(tree) / rel).is_file(): + (src / rel).parent.mkdir(parents=True, exist_ok=True) + shutil.copyfile(Path(tree) / rel, src / rel) + shutil.rmtree(venv, ignore_errors=True) # an interrupted earlier build + venv.mkdir() + container.give([src, venv, uv_cache]) + name = container.new_name() + env = {"UV_PROJECT_ENVIRONMENT": str(venv / "env"), "UV_CACHE_DIR": str(uv_cache), "UV_PYTHON": str(interpreter), + "UV_PYTHON_DOWNLOADS": "never", "HOME": str(src), "PATH": f"{Path(uv).parent}:/usr/bin:/bin", + "HTTPS_PROXY": f"http://127.0.0.1:{PROXY_PORT_IN}", "HTTP_PROXY": f"http://127.0.0.1:{PROXY_PORT_IN}"} + argv = container.argv(name, src, ["sh", "-c", f"socat TCP-LISTEN:{PROXY_PORT_IN},bind=127.0.0.1,fork,reuseaddr " + f"UNIX-CONNECT:{sock_dir}/index.sock & {t.build}"], + env, [src, venv, uv_cache, sock_dir], [Path(uv).parent, interpreter.parent.parent], str(interpreter), + gpu=False) + try: + with IndexProxy(hosts(tree, t.build), sock_dir / "index.sock"): + p = subprocess.run(argv, capture_output=True, text=True, timeout=TIMEOUT_S) + except subprocess.TimeoutExpired: + p = subprocess.CompletedProcess(argv, -9, "", f"timed out after {TIMEOUT_S}s") + finally: + container.remove(name) + shutil.rmtree(work, ignore_errors=True) + shutil.rmtree(sock_dir, ignore_errors=True) + if p.returncode != 0 or not (venv / "env" / "bin" / "python").exists(): + shutil.rmtree(venv, ignore_errors=True) + raise BuildFailed(f"dependency build exited {p.returncode}: {(p.stdout + p.stderr)[-LOG_TAIL:]}") + (venv / ".lab-built").write_text(deps_sha(tree)) diff --git a/lab/engine.py b/lab/engine.py index 9210cff..1445286 100644 --- a/lab/engine.py +++ b/lab/engine.py @@ -4,6 +4,7 @@ from __future__ import annotations import sys +from contextlib import contextmanager from pathlib import Path ENV_ROOT = Path(__file__).resolve().parents[1] @@ -15,9 +16,24 @@ def repo() -> Path: return target.load().engine_repo +_using: list[str] = [] + + +@contextmanager +def using(py: str): + """`python()` is `py` inside the block: the venv a tree's own dependency files were built into (lab/build.py).""" + _using.append(py) + try: + yield + finally: + _using.pop() + + def python() -> str: """The target's engine interpreter (LAB_ENGINE_PYTHON overrides); this interpreter if that file does not exist. The engine's venv needs what its test and lint commands import (e.g. `uv sync --extra dev`).""" + if _using: + return _using[-1] from lab import target py = target.load().engine_python return py if Path(py).exists() else sys.executable diff --git a/lab/gpucheck.py b/lab/gpucheck.py new file mode 100644 index 0000000..ebb7bed --- /dev/null +++ b/lab/gpucheck.py @@ -0,0 +1,188 @@ +"""What lab/RUNTIME.md's jails need from a real GPU host, checked on one: run on the GPU VM, as root the way the remote +worker runs (`sudo`), with the engine's venv first on PATH. Each check prints PASS or FAIL with its evidence; the exit +status is the number of failures. `--only NAME,...` runs a subset; `--skip-build` leaves out the dependency build. + + sudo -n env HOME="$HOME" PATH="$HOME/inference-server/.venv/bin:$PATH" LAB_ENGINE_REPO="$HOME/inference-server" \\ + LAB_TARGET=targets/inference-server.toml python -m lab.gpucheck +""" + +from __future__ import annotations + +import argparse +import json +import os +import shutil +import subprocess +import sys +import tempfile +import time +import traceback +from pathlib import Path + +from lab import engine, gpu, serve, worker +from lab.safety import container, grader + +HOLD = "import torch, time; x = torch.empty(2**30, dtype=torch.uint8, device='cuda'); print('holding', flush=True); time.sleep(600)" +MEASURE = {"names": ["single_stream"], "split": "seen", "tier": "short", "seed": 0, "jailed": True, "passive": False} +results: list[tuple[str, bool, str]] = [] + + +def check(name: str): + def wrap(fn): + fn.check = name + return fn + return wrap + + +def base_tree(root: Path) -> Path: + """The worker's engine checkout as a two-commit tree (base, empty change): the shape every job gets.""" + tree = root / "base" + shutil.copytree(engine.repo(), tree, ignore=shutil.ignore_patterns(".venv", ".git", "__pycache__", "*.pyc")) + grader._commit(tree, "base", init=True) + grader._commit(tree, "change") + return tree + + +def holder(name: str, wb: bool, tmp: Path) -> subprocess.Popen: + """A process holding 1 GiB of GPU memory: in a workbench container (`wb`) or on the host.""" + py = engine.python() + if wb: + argv = container.workbench_argv(name, tmp, [py, "-c", HOLD], {"HOME": str(tmp)}, [tmp], [engine.venv()], py) + else: + argv = [py, "-c", HOLD] + p = subprocess.Popen(argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) + deadline = time.monotonic() + 120 + while time.monotonic() < deadline and "holding" not in (p.stdout.readline() or ""): + if p.poll() is not None: + raise RuntimeError(f"the GPU holder exited {p.returncode}") + return p + + +@check("jail_gpu") +def jail_gpu(tmp: Path) -> str: + """A measurement room (no network, read-only root, uid 10001) sees the GPU.""" + argv = container.argv(container.new_name(), tmp, ["nvidia-smi", "-L"], {}, [tmp], [], engine.python(), gpu=True) + out = subprocess.run(argv, capture_output=True, text=True, timeout=120) + assert out.returncode == 0 and "GPU 0" in out.stdout, out.stdout + out.stderr + return out.stdout.strip() + + +@check("workbench") +def workbench(tmp: Path) -> str: + """The workbench: the GPU, root, apt and pip over the internet, and no cloud metadata service.""" + script = ("cat /etc/resolv.conf; set -e; nvidia-smi -L; id -u; apt-get update -qq; apt-get install -y -qq --no-install-recommends " + "python3-pip >/dev/null; pip3 install --quiet --break-system-packages six; python3 -c 'import six'; " + "echo pip-ok; python3 -c \"import socket; socket.create_connection(('169.254.169.254', 80), timeout=3)\" " + "&& echo METADATA-REACHABLE || echo metadata-blocked") + argv = container.workbench_argv(container.new_name(), tmp, ["sh", "-c", script], {"HOME": str(tmp)}, [tmp], [], + engine.python()) + out = subprocess.run(argv, capture_output=True, text=True, timeout=600, stdin=subprocess.DEVNULL) + text = out.stdout + out.stderr + assert out.returncode == 0 and "GPU 0" in text and "pip-ok" in text and "metadata-blocked" in text, text[-2000:] + assert "\n0\n" in f"\n{out.stdout}", "not root inside the workbench (run this check as root, like the worker)" + return "GPU, root, apt, pip; metadata blocked" + + +@check("quiet") +def quiet(tmp: Path) -> str: + """`container.quiet` kills the workbench's GPU holder, freezes the rest, and thaws it after.""" + name = container.new_name() + p = holder(name, True, tmp) + try: + mine = container.pids(name) + assert any(a["pid"] in mine for a in gpu.compute_apps() or []), "the holder is not on the GPU" + with container.quiet(name) as killed: + state = subprocess.run(["docker", "inspect", "-f", "{{.State.Paused}}", name], capture_output=True, + text=True).stdout.strip() + left = [a for a in gpu.compute_apps() or [] if a["pid"] in mine] + after = subprocess.run(["docker", "inspect", "-f", "{{.State.Paused}}", name], capture_output=True, + text=True).stdout.strip() + assert killed and not left and state == "true" and after == "false", (killed, left, state, after) + return f"killed {killed}; paused during, thawed after" + finally: + p.kill() + container.remove(name) + + +@check("exclusive") +def exclusive(tmp: Path) -> str: + """A clean measurement is not contaminated (only the engine's processes on the GPU); a foreign one is refused.""" + tree = base_tree(tmp) + clean = worker.Local().call("measure", {"tree": tree}, MEASURE, tmp / "clean") + p = holder(None, False, tmp) + try: + worker.Local().call("measure", {"tree": tree}, MEASURE, tmp / "busy") + raise AssertionError("measured with a foreign process on the GPU") + except serve.GpuBusy as e: + refused = e.apps + finally: + p.kill() + v = clean["results"][0] + return f"clean: {v['objective']} {v['value']:.4f} valid={v['valid']}; refused with {refused}" + + +@check("ncu") +def ncu(tmp: Path) -> str: + """Nsight Compute inside a measurement room (needs the GPU's performance counters for the jail's user).""" + r = worker.Local().call("profile", {"tree": base_tree(tmp)}, {"kind": "kernel", "args": {"launch_count": 2}}, + tmp / "out") + assert not r.get("refused") and r["returncode"] == 0 and r["head"], r.get("refused") or r["output"][-1500:] + return r["head"][:300] + + +@check("hostprof") +def hostprof(tmp: Path) -> str: + """py-spy inside a measurement room.""" + r = worker.Local().call("profile", {"tree": base_tree(tmp)}, {"kind": "hostprof", "args": {"seconds": 5}}, + tmp / "out") + assert not r.get("refused") and r["returncode"] == 0 and r["head"], r.get("refused") or r["output"][-1500:] + return r["head"][:300] + + +@check("build") +def build_step(tmp: Path) -> str: + """A dependency change is built from scratch (index-only network) and the built venv runs torch on the GPU.""" + from lab import build + tree = tmp / "base" + shutil.copytree(engine.repo(), tree, ignore=shutil.ignore_patterns(".venv", ".git", "__pycache__", "*.pyc")) + grader._commit(tree, "base", init=True) + uv = shutil.which("uv") or str(Path.home() / ".local" / "bin" / "uv") + subprocess.run([uv, "add", "--no-sync", "six"], cwd=tree, check=True, capture_output=True) + grader._commit(tree, "change") + assert build.changed(tree), "the change to pyproject.toml/uv.lock was not seen" + t0 = time.monotonic() + py = build.python_for(tree) + out = subprocess.run([py, "-c", "import six, torch; print(torch.__version__, torch.cuda.is_available())"], + capture_output=True, text=True) + assert out.returncode == 0 and out.stdout.strip().endswith("True"), out.stdout + out.stderr + return f"built in {time.monotonic() - t0:.0f}s: {py}: torch {out.stdout.strip()}" + + +CHECKS = [jail_gpu, workbench, quiet, exclusive, ncu, hostprof, build_step] + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(prog="python -m lab.gpucheck", description=__doc__.split("\n\n")[0]) + ap.add_argument("--only", help="comma-separated: " + ",".join(c.check for c in CHECKS)) + ap.add_argument("--skip-build", action="store_true") + a = ap.parse_args(argv) + want = set(a.only.split(",")) if a.only else {c.check for c in CHECKS} - ({"build"} if a.skip_build else set()) + print(f"as uid {os.getuid()}; GPU: {gpu.query()}", flush=True) + for c in CHECKS: + if c.check not in want: + continue + tmp = Path(tempfile.mkdtemp(prefix=f"gpucheck-{c.check}-")) + t0 = time.monotonic() + try: + detail, ok = c(tmp), True + except Exception as e: + detail, ok = f"{type(e).__name__}: {e}\n{traceback.format_exc()[-1500:]}", False + results.append((c.check, ok, detail)) + print(f"{'PASS' if ok else 'FAIL'} {c.check} ({time.monotonic() - t0:.0f}s): {detail}", flush=True) + shutil.rmtree(tmp, ignore_errors=True) + print(json.dumps({n: ok for n, ok, _ in results})) + return sum(not ok for _, ok, _ in results) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/lab/safety/container.py b/lab/safety/container.py index fcb6382..9fb5ec7 100644 --- a/lab/safety/container.py +++ b/lab/safety/container.py @@ -74,7 +74,7 @@ def mounts(writable: list[Path], readonly: list[Path], python: str) -> list[str] def argv(name: str, workdir: Path, command: list[str], env: dict[str, str], writable: list[Path], - readonly: list[Path], python: str, bridge_port: int | None = None) -> list[str]: + readonly: list[Path], python: str, bridge_port: int | None = None, gpu: bool | None = None) -> list[str]: """`docker run` for `command`. With `bridge_port`, the engine's 127.0.0.1:port is also served on the unix socket `/bridge.sock`, which the host side of `Bridge` forwards to the host's 127.0.0.1:port.""" uid, gid = user() @@ -82,7 +82,7 @@ def argv(name: str, workdir: Path, command: list[str], env: dict[str, str], writ "--tmpfs", "/tmp:exec", "--cap-drop", "ALL", "--security-opt", "no-new-privileges", "--pids-limit", "8192", "--shm-size", "16g", "--user", f"{uid}:{gid}", "--workdir", str(workdir.resolve()), "--label", "lab.jail=1"] - if gpus(): + if gpus() if gpu is None else gpu: out += ["--gpus", "all"] if (cpus := cpuset()): out += ["--cpuset-cpus", cpus] @@ -138,6 +138,32 @@ def _kill_gpu_holders(name: str) -> list[dict]: return hit +def pause(name: str | None) -> bool: + """Freeze a running container; True only if this call froze it (False: absent, stopped, or already paused).""" + if not name or _docker() is None or not running(name): + return False + if subprocess.run([_docker(), "inspect", "-f", "{{.State.Paused}}", name], capture_output=True, + text=True).stdout.strip() == "true": + return False + subprocess.run([_docker(), "pause", name], capture_output=True, check=True) + return True + + +def unpause(name: str) -> None: + subprocess.run([_docker(), "unpause", name], capture_output=True) + + +@contextmanager +def paused(name: str | None): + """The container frozen for the block (a no-op if it is not running); left paused if it already was.""" + did = pause(name) + try: + yield + finally: + if did: + unpause(name) + + @contextmanager def quiet(name: str | None): """The workbench paused for a measurement: processes in it holding the GPU are killed (and listed), the rest @@ -147,16 +173,13 @@ def quiet(name: str | None): return from lab import gpu killed = _kill_gpu_holders(name) - subprocess.run([_docker(), "pause", name], capture_output=True, check=True) - try: + with paused(name): killed += _kill_gpu_holders(name) # anything that took the GPU before the pause landed deadline = time.monotonic() + 15 while killed and time.monotonic() < deadline and any( a["pid"] in {k["pid"] for k in killed} for a in gpu.compute_apps() or []): time.sleep(0.2) # until the driver has released their memory yield killed - finally: - subprocess.run([_docker(), "unpause", name], capture_output=True) def new_name() -> str: diff --git a/lab/safety/grader.py b/lab/safety/grader.py index 04195ad..ac03976 100644 --- a/lab/safety/grader.py +++ b/lab/safety/grader.py @@ -16,7 +16,7 @@ from lab import engine, gpu, target from lab.safety import container, jail -from lab.safety.surfaces import ALWAYS_DENY, HIDDEN, may_write +from lab.safety.surfaces import ALWAYS_DENY, HIDDEN, deps_only, may_write IGNORED = ("*/__pycache__/*", "__pycache__/*", "*.pyc", ".pytest_cache/*", "*/.pytest_cache/*", ".ruff_cache/*", "*/.ruff_cache/*", "*.egg-info/*", ".DS_Store", "*/.DS_Store", @@ -94,6 +94,12 @@ def audit(repo: Path, base: str, workspace: Path) -> Audit: for name in dirs + files: p = Path(root) / name rel = p.relative_to(workspace).as_posix() + if not p.is_symlink() and any(fnmatch(rel, g) for g in IGNORED): + continue + if _scratch(rel, blobs) and (p.is_symlink() or not (p.is_dir() or p.is_file()) + or (p.is_file() and p.stat().st_size > MAX_FILE_BYTES)): + a.scratch.append(rel) # never reaches the pristine tree: a venv's links, big wheels + continue if p.is_symlink(): a.violations.append(f"symlink: {rel}") elif p.is_dir() or any(fnmatch(rel, g) for g in IGNORED): @@ -121,9 +127,24 @@ def audit(repo: Path, base: str, workspace: Path) -> Audit: for rel in a.modified + a.deleted: if not may_write(rel, new_file=False): a.violations.append(f"may not change: {rel}") + deps = target.load().deps + for rel in a.deleted: + if rel in deps: + a.violations.append(f"may not delete a dependency file: {rel}") + for rel in a.modified: + if rel in deps and rel.endswith(".toml"): + old = subprocess.run(["git", "-C", str(repo), "show", f"{base}:{rel}"], capture_output=True, text=True).stdout + if not deps_only(old, (workspace / rel).read_text(errors="replace")): + a.violations.append(f"may change only the dependency tables of {rel}") return a +def _scratch(rel: str, blobs: dict) -> bool: + """Outside what the agent may write and not a base file: left out of every tree, whatever it is.""" + return (rel not in blobs and not may_write(rel, new_file=True) + and not any(fnmatch(rel.lower(), p.lower()) for p in ALWAYS_DENY + HIDDEN)) + + def pristine_tree(repo: Path, base: str, workspace: Path, result: Audit, dest: Path) -> None: """A fresh export of `base` plus only the audited changes, as a two-commit repo (base, change).""" if not result.ok: diff --git a/lab/safety/surfaces.py b/lab/safety/surfaces.py index caf520c..2b10286 100644 --- a/lab/safety/surfaces.py +++ b/lab/safety/surfaces.py @@ -22,11 +22,41 @@ def write_surface() -> tuple[tuple[str, ...], tuple[str, ...]]: HIDDEN = ("corpus/*/heldout.jsonl",) +# The parts of a pyproject.toml that are dependencies; everything else in it (test and lint config) is protected. +DEP_TABLES = (("project", "dependencies"), ("project", "optional-dependencies"), ("dependency-groups",), + ("tool", "uv", "sources"), ("tool", "uv", "index")) + + def may_write(path: str, *, new_file: bool) -> bool: - """Whether the agent may touch repo-relative `path`.""" + """Whether the agent may touch repo-relative `path`. The target's dependency files may be changed (a + pyproject.toml only in its dependency tables: `deps_only`), though some are on the deny list for the rest.""" + if not new_file and path in target.load().deps: + return True if any(fnmatch(path.lower(), p.lower()) for p in ALWAYS_DENY + HIDDEN): # macOS ignores case return False write, add_only = write_surface() if any(fnmatch(path, p) for p in write): return True return new_file and any(fnmatch(path, p) for p in add_only) + + +def deps_only(old: str, new: str) -> bool: + """Whether a pyproject.toml changed nothing but its dependency tables (and still parses).""" + import tomllib + + def rest(text: str) -> dict: + d = tomllib.loads(text) + for keys in DEP_TABLES: + path = [d] + for k in keys[:-1]: + path.append(path[-1].get(k) if isinstance(path[-1], dict) else None) + if isinstance(path[-1], dict): + path[-1].pop(keys[-1], None) + for parent, k, child in reversed(list(zip(path, keys, path[1:]))): # tables emptied by it go too + if child == {}: + parent.pop(k) + return d + try: + return rest(old) == rest(new) + except tomllib.TOMLDecodeError: + return False diff --git a/lab/session.py b/lab/session.py index aa6f96b..39341a2 100644 --- a/lab/session.py +++ b/lab/session.py @@ -128,6 +128,8 @@ def run(cfg: RunConfig, provider: agent.Provider | None = None) -> dict: summary = {"run": run_id, "base": base, "sessions": 0, "stopped": None, "spent_usd": budget.spent_usd} wk = worker.load() remote = wk if isinstance(wk, worker.Remote) else None # then the agent works on that host, not here + if remote and not getattr(provider, "remote", False): # its edits here would be overwritten by the mirror + raise ValueError(f"provider {provider.name!r} works on this machine; it cannot run with LAB_WORKER=ssh") rws = remote.workspace_path(run_id) if remote else None if cfg.baseline: Toolbox(Session(run_id, f"{run_id}-baseline", run_dir, ws, budget, cfg.ledger_root, task=cfg.task, diff --git a/lab/target.py b/lab/target.py index affe662..4f1d850 100644 --- a/lab/target.py +++ b/lab/target.py @@ -51,6 +51,8 @@ class Target: add_only: tuple[str, ...] test: str lint: str + deps: tuple[str, ...] # dependency files the agent may change (pyproject: its dependency tables only) + build: str | None # installs `deps` into $UV_PROJECT_ENVIRONMENT (lab/build.py) reference: Server | None reference_dir: Path | None interactive: Limits @@ -96,7 +98,7 @@ def parse(path: Path) -> Target: path=path, model=m["name"], chat_kwargs=dict(m.get("chat_kwargs") or {}), max_model_len=m.get("max_model_len"), engine_repo=repo, engine_python=py, engine=_server(e), write=tuple(e.get("write") or ()), add_only=tuple(e.get("add_only") or ()), - test=e["test"], lint=e["lint"], + test=e["test"], lint=e["lint"], deps=tuple(e.get("deps") or ()), build=e.get("build"), reference=_server(r), reference_dir=(ENV_ROOT / r["dir"]) if r and r.get("dir") else None, interactive=Limits(**lim.get("interactive", {"ttft_s": 0.5, "tpot_s": 0.030})), conversational=Limits(**lim.get("conversational", {"ttft_s": 2.0, "tpot_s": 0.100})), diff --git a/lab/tools.py b/lab/tools.py index 8ca82ed..f475d68 100644 --- a/lab/tools.py +++ b/lab/tools.py @@ -231,16 +231,24 @@ def specs(self) -> list[ToolSpec]: paused = (" While it runs, the container your shell runs in is paused, and processes in it holding the GPU are " "killed and listed in the result." if ClaudeAgentSDK.containerized() else "") + deps = target.load().deps + offline = (f" Only your change reaches the engine here: if it changes {' or '.join(deps)}, those are installed " + "from scratch for it (package indexes only), and the engine then runs with no network, so anything it " + "needs at runtime (draft models, prebuilt kernels, files a library downloads when it is used) has to be " + "in your change." if deps else " The engine runs with no network.") + def gpu(t: str) -> str: return (f" Costs GPU time: ~{t} per call, charged at ${rate:g}/h." if known else - f" Costs GPU time: ~{t} per call; no GPU rate is configured on this host, so it is charged $0.") + paused + f" Costs GPU time: ~{t} per call; no GPU rate is configured on this host, so it is charged $0." + ) + offline + paused short, full = suite.TIERS["short"], suite.TIERS["full"] windows = (f"one engine start plus, per regime, a {short['final_s']:.0f} s measurement window at tier short or " f"{full['final_s']:.0f} s at full, plus {short['probe_s']:.0f} s / {full['probe_s']:.0f} s probes for " f"regimes that search for a load") specs = [ ToolSpec("test", "Lint and run the test suite on a pristine copy of your current workspace, inside the " - "referee's jail. Returns PASS or FAIL, lint and test status, and the tail of each output." + free, + "referee's jail. Returns PASS or FAIL, lint and test status, and the tail of each output." + offline + + free, obj, self.test), ToolSpec("profile", "Run the engine in process under the profiler, jailed, on `requests` sampled from a seen " "corpus class (`corpus_class`, default steady_interactive; `seed`) or on synthetic token ids " diff --git a/lab/vm-setup.sh b/lab/vm-setup.sh index 4b0ab31..93cf334 100755 --- a/lab/vm-setup.sh +++ b/lab/vm-setup.sh @@ -30,6 +30,11 @@ echo "== CUDA toolkit" nvcc --version 2>/dev/null | tail -1 || ls /usr/local/cuda/bin/nvcc 2>/dev/null || echo "nvcc: not on PATH" echo "== Nsight" +# CUDA images keep ncu in /usr/local/cuda/bin, on no login PATH, and that one is only a launcher for the install under +# /opt/nvidia/nsight-compute/: link the real one, so the tool's jail mounts its whole install root. +ncu_real=$(ls -d /opt/nvidia/nsight-compute/*/ncu 2>/dev/null | sort -V | tail -1) +if [ -n "$ncu_real" ]; then $SUDO ln -sf "$ncu_real" /usr/local/bin/ncu +elif ! command -v ncu >/dev/null && [ -x /usr/local/cuda/bin/ncu ]; then $SUDO ln -sf /usr/local/cuda/bin/ncu /usr/local/bin/ncu; fi if ! command -v nsys >/dev/null || ! command -v ncu >/dev/null; then $SUDO apt-get update -qq $SUDO apt-get install -y -qq nsight-systems nsight-compute 2>/dev/null \ @@ -67,9 +72,12 @@ fi $SUDO docker build -q -t lab-jail -f "$here/safety/jail.Dockerfile" "$here/safety" >/dev/null $SUDO docker run --rm --gpus all --network none lab-jail nvidia-smi -L && echo "jail: gpu ok" || echo "jail: no GPU in the container" # The agent's workbench has the internet but not the cloud's metadata service (instance credentials, user-data). -$SUDO iptables -C DOCKER-USER -d 169.254.0.0/16 -j DROP 2>/dev/null || $SUDO iptables -I DOCKER-USER -d 169.254.0.0/16 -j DROP +# Only that address: a cloud's DNS resolver can sit in 169.254.0.0/16 too, and the workbench needs names. +$SUDO iptables -C DOCKER-USER -d 169.254.169.254 -j DROP 2>/dev/null || $SUDO iptables -I DOCKER-USER -d 169.254.169.254 -j DROP $SUDO docker run --rm lab-jail python3 -c "import socket; socket.create_connection(('169.254.169.254', 80), timeout=3)" \ 2>/dev/null && echo "workbench: METADATA REACHABLE" || echo "workbench: metadata blocked" +$SUDO docker run --rm lab-jail python3 -c "import socket; socket.create_connection(('pypi.org', 443), timeout=10)" \ + 2>/dev/null && echo "workbench: internet ok (DNS and egress)" || echo "workbench: NO INTERNET (DNS or egress)" echo "== counters: can a non-admin process read GPU performance counters?" # ncu needs NVreg_RestrictProfilingToAdminUsers=0 (or root). Recorded as a fact about this venue. @@ -91,8 +99,18 @@ echo "== peaks: HBM bandwidth and bf16 GEMM, measured (validity/peaks.py; the ce ENV_DIR="${LAB_VM_ENV_DIR:-$HOME/BlameGraph}"; ENGINE_DIR="$PWD" (cd "$ENV_DIR" && "$ENGINE_DIR/.venv/bin/python" -m validity peaks --seconds 3) || echo "peaks: not measured (finding, not failure)" -echo "== model cache" +echo "== model cache: the target's model and the Hub repos its tests read (every jail is offline)" mkdir -p "$HOME/.cache/huggingface" +.venv/bin/python - "$here/../${LAB_TARGET:-targets/inference-server.toml}" <<'PY' \ + || echo "model cache: download failed (finding, not failure)" +import sys, tomllib +from huggingface_hub import snapshot_download +t = tomllib.load(open(sys.argv[1], "rb")) +print("cached", snapshot_download(t["model"]["name"])) +for repo in t["engine"].get("hub", []): # what tests read is tokenizers and configs, not weights + print("cached", snapshot_download(repo, allow_patterns=["*.json", "*.txt", "*.model", "*.tiktoken"])) +PY +chmod -R a+rX "$HOME/.cache/huggingface" # the jails' own user reads it (hub metadata is written 0600) du -sh "$HOME/.cache/huggingface" 2>/dev/null || true echo "== smoke" diff --git a/lab/vm.py b/lab/vm.py index 622cf9f..4343e14 100644 --- a/lab/vm.py +++ b/lab/vm.py @@ -190,7 +190,9 @@ def push(vm: VM, record: Record) -> None: """Both working trees, minus .git and everything .gitignore excludes (secrets, weights, venvs, ledger, runs).""" # .venv is named explicitly: a gitignore-only exclusion did not protect it from --delete on the VM. for local, remote in ((engine.repo(), vm.remote_dir), (engine.ENV_ROOT, vm.env_dir)): - _rsync(vm, "--delete", "--exclude=.git", "--exclude=.venv", "--exclude=__pycache__", "--filter=:- .gitignore", + # lab/runs is named explicitly too: it holds the worker's cache and the agent's live workspaces there. + _rsync(vm, "--delete", "--exclude=.git", "--exclude=.venv", "--exclude=__pycache__", "--exclude=/lab/runs", + "--filter=:- .gitignore", f"{local}/", f"{ssh_target(vm, record)}:{remote}/") diff --git a/lab/worker.py b/lab/worker.py index ee584d4..c07d794 100644 --- a/lab/worker.py +++ b/lab/worker.py @@ -4,6 +4,8 @@ the output dir back; the worker re-hashes every tree right before using it and refuses one that does not match, so what is measured is what the controller audited. GpuBusy, Contaminated, NotReady and ValueError cross intact. measure, equiv and profile take `workbench`, the agent's container: it is paused for the job, and processes in it holding the GPU are killed and listed under `killed`. +Every job runs the tree with the interpreter `build.python_for` gives it: the engine's, or a venv built from the +tree's own dependency files. test lint + the engine's suite on `tree` -> {lint, tests} measure serve `tree`, run regimes; passive data into out -> {results, ready_s} @@ -21,6 +23,7 @@ from __future__ import annotations import dataclasses +from contextlib import contextmanager import hashlib import io import json @@ -60,21 +63,28 @@ def tree_sha(root: Path) -> str: # -- jobs (run where the GPU is) --------------------------------------------------------------------------------- def job_test(inputs: dict, out: Path, args: dict, hooks: dict) -> dict: + from lab import build from lab.safety import grader tree = inputs["tree"] - lint = grader.run_lint(tree) - tests = grader.run_tests(tree) if lint.passed else grader.Run(False, -1, "skipped: lint failed") + try: + py = build.python_for(tree) + except build.BuildFailed as e: # a failed test run, recorded like any other + skipped = grader.Run(False, -1, str(e)) + return {"lint": dataclasses.asdict(skipped), "tests": dataclasses.asdict(skipped)} + with engine.using(py): + lint = grader.run_lint(tree) + tests = grader.run_tests(tree) if lint.passed else grader.Run(False, -1, "skipped: lint failed") return {"lint": dataclasses.asdict(lint), "tests": dataclasses.asdict(tests)} def job_measure(inputs: dict, out: Path, args: dict, hooks: dict) -> dict: """`args`: names, split, tier, seed, jailed, passive (keep device/telemetry/client rows in `out`).""" - from lab import artifacts + from lab import artifacts, build from lab.evaltools import run_regimes t = target.load() rows = [] if args["passive"] else None try: - with container.quiet(args.get("workbench")) as killed, serve.Served( + with engine.using(build.python_for(inputs["tree"])), container.quiet(args.get("workbench")) as killed, serve.Served( inputs["tree"], t.engine, log=out / ENGINE_LOG, jailed=args["jailed"], artifacts=out if args["passive"] else None) as srv: results = run_regimes(srv.url, t, args["names"], args["split"], args["tier"], args["seed"], rows) @@ -91,12 +101,13 @@ def job_equiv(inputs: dict, out: Path, args: dict, hooks: dict) -> dict: answers already in `prior`'s equiv.outputs.jsonl are reused, not asked again.""" from correctness import client, run as crun from correctness.gate import Thresholds + from lab import build t = target.load() if (inputs["prior"] / "equiv.outputs.jsonl").exists(): shutil.copyfile(inputs["prior"] / "equiv.outputs.jsonl", out / "equiv.outputs.jsonl") enc = hooks.get("encoder") or crun.HFEncoder(t.model, t.chat_kwargs) - with container.quiet(args.get("workbench")) as killed, serve.Served( - inputs["tree"], t.engine, log=out / ENGINE_LOG, jailed=args["jailed"], artifacts=out) as srv: + with engine.using(build.python_for(inputs["tree"])), container.quiet(args.get("workbench")) as killed, \ + serve.Served(inputs["tree"], t.engine, log=out / ENGINE_LOG, jailed=args["jailed"], artifacts=out) as srv: res = crun.candidate(inputs["reference"], srv.url, t.model, out / "equiv.json", enc, Thresholds(min_score_ratio=t.min_score_ratio, min_length_ratio=t.min_length_ratio), concurrency=args["concurrency"], api=client.Api(t.engine.api, t.chat_kwargs), @@ -107,7 +118,12 @@ def job_equiv(inputs: dict, out: Path, args: dict, hooks: dict) -> dict: def job_profile(inputs: dict, out: Path, args: dict, hooks: dict) -> dict: """`args`: kind (profile | trace | kernel | hostprof), the tool's args, workbench. Files go to `out/files`.""" - with container.quiet(args.get("workbench")) as killed: + from lab import build + try: + py = build.python_for(inputs["tree"]) + except build.BuildFailed as e: # a failed profile run, recorded like any other + return {"returncode": -1, "output": str(e)[-3000:], "instrument": None, "argv": [], "head": "", "killed": []} + with engine.using(py), container.quiet(args.get("workbench")) as killed: return {**_profile(inputs, out, args, hooks), "killed": killed} @@ -135,7 +151,9 @@ def _profile(inputs: dict, out: Path, args: dict, hooks: dict) -> dict: exe = str(Path(exe).resolve()) work = tree / proftools.OUT work.mkdir() - read = [Path(exe).parent] + read = [Path(exe).parent, Path(exe).parent.parent] # the install root: nsys's importer lives beside its binary dir + if Path("/opt/nvidia").is_dir(): + read.append(Path("/opt/nvidia")) # where launchers like CUDA's bin/ncu find the real install src = T.workload_flags(a, harness, proftools._int(a, "requests", *proftools.WORKLOAD["requests"])) argv = [*ins["wrap"](exe, work), engine.python(), "-m", "lab.profile", *ins["flags"], *src, *ins["extra"](exe), "--out", str(work / "bundle")] @@ -207,8 +225,11 @@ class Remote: """Jobs on another host. `prefix` is the argv that runs one shell command there (ssh ... user@host); empty runs it here through sh, which the tests use. `env_dir` is this repo on that host, `python` its interpreter.""" - def __init__(self, prefix: list[str], env_dir: str, python: str, env: dict[str, str], rsh: str | None = None): - self.prefix, self.env_dir, self.python, self.env, self.rsh = prefix, env_dir, python, env, rsh + def __init__(self, prefix: list[str], env_dir: str, python: str, env: dict[str, str], rsh: str | None = None, + sudo: bool = False): + """`sudo`: the login user is not root, so the worker's side runs as root (the workbench is root inside, the + measurement rooms drop to their own uid), keeping the login user's HOME (weights, venv, uv).""" + self.prefix, self.env_dir, self.python, self.env, self.rsh, self.sudo = prefix, env_dir, python, env, rsh, sudo self._root: str | None = None def _where(self, path: str) -> str: @@ -216,7 +237,8 @@ def _where(self, path: str) -> str: def _rsync(self, src: str, dst: str) -> None: """By content, not size and mtime: an edit within the same second must not be missed.""" - p = subprocess.run(["rsync", "-a", "--checksum", "--delete", *(["-e", self.rsh] if self.rsh else []), src, dst], + p = subprocess.run(["rsync", "-a", "--checksum", "--delete", *(["-e", self.rsh] if self.rsh else []), + *(["--rsync-path", "sudo -n rsync"] if self.sudo else []), src, dst], capture_output=True, text=True) if p.returncode != 0: raise WorkerError(f"rsync {src} -> {dst}: {p.stderr.strip()[-1000:]}") @@ -243,6 +265,16 @@ def workbench(self, name: str, session: str, workspace: str, sdk_version: str) - req = {"name": name, "session": session, "workspace": workspace, "sdk_version": sdk_version} return json.loads(self._run("workbench", json.dumps(req).encode())) + @contextmanager + def paused(self, name: str | None): + """The workbench on the worker host frozen for the block (as `container.paused` there).""" + did = bool(name) and self._run(f"pause {shlex.quote(name)}").decode().strip() == "paused" + try: + yield + finally: + if did: + self._run(f"unpause {shlex.quote(name)}") + def remove_workbench(self, name: str | None) -> None: if name: self._run(f"rm {shlex.quote(name)}") @@ -252,7 +284,8 @@ def _run(self, command: str, stdin: bytes = b"") -> bytes: if any(c in v for c in '"`\\'): raise WorkerError(f"unsafe value in the worker's env: {v!r}") exports = " ".join(f'{k}="{v}"' for k, v in self.env.items()) - line = f"cd {self.env_dir} && {exports} {self.python} -m lab.worker {command}" + root = 'sudo -n env HOME="$HOME" PATH="$PATH" ' if self.sudo else "" + line = f"cd {self.env_dir} && {root}{exports} {self.python} -m lab.worker {command}" argv = [*self.prefix, line] if self.prefix else ["sh", "-c", line] p = subprocess.run(argv, input=stdin, capture_output=True) if p.returncode != 0: @@ -290,7 +323,7 @@ def load(**hooks): home = lambda p: p.replace("~", "$HOME", 1) # noqa: E731 - expanded by the worker's shell return Remote(["ssh", *v.ssh_opts, vm.ssh_target(v, record)], v.env_dir, home(v.remote_dir) + "/.venv/bin/python", {"LAB_TARGET": spec.relative_to(engine.ENV_ROOT).as_posix(), "LAB_ENGINE_REPO": home(v.remote_dir)}, - rsh=f"ssh {shlex.join(v.ssh_opts)}") + rsh=f"ssh {shlex.join(v.ssh_opts)}", sudo=v.login != "root") # -- the worker host's side --------------------------------------------------------------------------------------- @@ -352,6 +385,8 @@ def _workbench(req: dict) -> dict: from lab import agent sock_dir = Path(f"/tmp/lab-proxy-{req['session']}") sock_dir.mkdir(mode=0o700, exist_ok=True) + if os.environ.get("SUDO_UID"): # sshd binds the tunnel's socket here as the login user + os.chown(sock_dir, int(os.environ["SUDO_UID"]), int(os.environ["SUDO_GID"])) (sock_dir / "proxy.sock").unlink(missing_ok=True) scratch = engine.ENV_ROOT / "lab" / "runs" / "remote" / "scratch" / req["session"] argv = agent.workbench_command(req["name"], Path(req["workspace"]), scratch, agent_cli(req["sdk_version"]), @@ -376,9 +411,13 @@ def main(argv: list[str] | None = None) -> int: print(json.dumps(_workbench(json.loads(sys.stdin.buffer.read())))) elif argv[:1] == ["rm"] and len(argv) == 2: container.remove(argv[1]) + elif argv[:1] == ["pause"] and len(argv) == 2: # prints "paused" only if this call froze it + print("paused" if container.pause(argv[1]) else "") + elif argv[:1] == ["unpause"] and len(argv) == 2: + container.unpause(argv[1]) else: print(f"usage: python -m lab.worker has SHA... | put SHA | run {{{','.join(JOBS)}}} | root | mkdir PATH | " - "workbench | rm NAME", file=sys.stderr) + "workbench | rm NAME | pause NAME | unpause NAME", file=sys.stderr) return 2 return 0 diff --git a/targets/inference-server.toml b/targets/inference-server.toml index de30fc0..b668fb3 100644 --- a/targets/inference-server.toml +++ b/targets/inference-server.toml @@ -18,6 +18,11 @@ write = ["src/inference_server/*"] # what the agent may add, modify or add_only = ["tests/test_*.py"] # what it may add but never change test = "python -m pytest -q -p no:cacheprovider -m 'not heavy and not needs_host'" lint = "python -m ruff check ." +hub = ["gpt2"] # Hub repos whose tokenizer the engine's tests read; cached at setup (jails are offline) +deps = ["pyproject.toml", "uv.lock"] # the agent may change these (pyproject: its dependency tables only) +# Installs a changed `deps` into $UV_PROJECT_ENVIRONMENT (lab/build.py): vm-setup.sh's recipe, the lock with the +# CUDA torch wheel at the locked version (the lock pins the CPU index on Linux). +build = '''uv sync --locked --extra dev --no-install-project && uv pip install --python "$UV_PROJECT_ENVIRONMENT/bin/python" --reinstall "torch==$(awk '/^name = "torch"$/{getline; sub(/version = "/,""); sub(/"/,""); print; exit}' uv.lock)" --index-url https://download.pytorch.org/whl/cu130''' [engine.telemetry] # optional: engine-native files serve collects after teardown env = { TELEMETRY_DIR = "{dir}", TIMELINE_DIR = "{dir}/timeline" } # request rows (lab/TRACE.md); events.jsonl apart diff --git a/tests/test_lab_agent.py b/tests/test_lab_agent.py index 4b1fa92..b96a8aa 100644 --- a/tests/test_lab_agent.py +++ b/tests/test_lab_agent.py @@ -60,10 +60,12 @@ def workbench(self, name, session, workspace, version): assert "-R /tmp/lab-proxy-scratch/proxy.sock:127.0.0.1:5555 root@203.0.113.5" in text fake = tmp_path / "bin" / "ssh" # an ssh that prints the remote command it was given fake.parent.mkdir() - fake.write_text('#!/bin/sh\nfor a; do last="$a"; done\nprintf "%s" "$last"\n') + fake.write_text('#!/bin/sh\nfor a; do last="$a"; done\nprintf "%s\\n" "$last"\n') fake.chmod(0o755) out = subprocess.run(["bash", str(script), "--output-format", "stream-json", "it's \"quoted\" $HOME"], capture_output=True, text=True, env={"PATH": f"{fake.parent}:/usr/bin:/bin"}) - remote_argv = shlex.split(out.stdout) + first, launch = out.stdout.splitlines() + assert first == "rm -f /tmp/lab-proxy-scratch/proxy.sock" # a stale socket from an earlier launch goes + remote_argv = shlex.split(launch) assert remote_argv == ["docker", "run", "-i", "--name", "lab-workbench-r1-s1", "lab-jail", "sh", "-c", 'exec "$0" "$@"', "/c/claude", "--output-format", "stream-json", "it's \"quoted\" $HOME"] diff --git a/tests/test_lab_apiproxy.py b/tests/test_lab_apiproxy.py index bc03393..5b0c036 100644 --- a/tests/test_lab_apiproxy.py +++ b/tests/test_lab_apiproxy.py @@ -127,3 +127,74 @@ def test_no_credential_means_no_proxy(monkeypatch): monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False) with pytest.raises(RuntimeError): apiproxy.Proxy(5.0) + + +def test_a_client_that_drops_mid_stream_is_charged_its_input_and_all_of_max_tokens(monkeypatch): + import time + seen = [] + + class Slow(BaseHTTPRequestHandler): + def log_message(self, *a): + pass + + def do_POST(self): + self.rfile.read(int(self.headers["content-length"])) + self.send_response(200) + self.send_header("content-type", "text/event-stream") + self.end_headers() + start = b'data: {"type":"message_start","message":{"model":"claude-opus-5-5","usage":{"input_tokens":1000,"output_tokens":1}}}\n\n' + self.wfile.write(start) + self.wfile.flush() + try: + for _ in range(50): # the rest never comes before the client hangs up + time.sleep(0.05) + self.wfile.write(b'data: {"type":"ping"}\n\n') + self.wfile.flush() + except OSError: + pass + seen.append(True) + + srv = ThreadingHTTPServer(("127.0.0.1", 0), Slow) + threading.Thread(target=srv.serve_forever, daemon=True).start() + monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-real") + with apiproxy.Proxy(50.0, upstream=f"http://127.0.0.1:{srv.server_address[1]}") as p: + c = http.client.HTTPConnection(urlsplit(p.url).netloc, timeout=10) + c.request("POST", "/v1/messages", body=json.dumps({"model": "claude-opus-5-5", "max_tokens": 4000, "stream": True}), + headers={"content-type": "application/json"}) + r = c.getresponse() + r.read1(200) + c.close() # hang up + deadline = time.time() + 10 + while not p.requests and time.time() < deadline: + time.sleep(0.1) + srv.shutdown() + p_in, p_out = apiproxy.PRICES["claude-opus-5-5"][:2] + assert p.spent_usd == pytest.approx((1000 * p_in + 4000 * p_out) / 1e6) and p.reserved_usd == pytest.approx(0) + + +def test_a_request_whose_worst_case_would_pass_the_cap_is_refused_before_it_goes(upstream, monkeypatch): + url, seen = upstream + monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-real") + with apiproxy.Proxy(0.10, upstream=url) as p: # 10k output tokens of opus is $0.20 + status, body = post(p, "/v1/messages", {"model": "claude-opus-5-5", "max_tokens": 10000}) + assert status == 403 and b"would be" in body and seen == [] + + +def test_a_bodyless_or_unpriced_messages_request_is_refused_not_crashed(upstream, monkeypatch): + url, seen = upstream + monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-real") + with apiproxy.Proxy(5.0, upstream=url) as p: + c = http.client.HTTPConnection(urlsplit(p.url).netloc, timeout=10) + c.request("GET", "/v1/messages") + r = c.getresponse() + assert r.status == 403 and b"priced model" in r.read() + assert post(p, "/v1/messages", {"max_tokens": 10})[0] == 403 + assert seen == [] + + +def test_the_input_is_reserved_too(upstream, monkeypatch): + url, seen = upstream + monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-real") + with apiproxy.Proxy(0.05, upstream=url) as p: # 20k bytes of opus input is up to $0.08 + status, body = post(p, "/v1/messages", {"model": "claude-opus-5-5", "max_tokens": 1, "pad": "x" * 20000}) + assert status == 403 and seen == [] diff --git a/tests/test_lab_build.py b/tests/test_lab_build.py new file mode 100644 index 0000000..7893977 --- /dev/null +++ b/tests/test_lab_build.py @@ -0,0 +1,148 @@ +"""lab.build: when a tree gets its own venv, the cache, the index-only proxy, and (Linux + docker + internet) a build +room that reaches the package index through the proxy and nothing else.""" + +from __future__ import annotations + +import dataclasses +import socket +import threading +from pathlib import Path + +import pytest + +from lab import build, engine, target +from lab.safety import container, jail + + +@pytest.fixture +def deps_target(tmp_path, monkeypatch): + """The committed target with its engine repo here, holding one dependency file.""" + repo = tmp_path / "engine" + repo.mkdir() + (repo / "pyproject.toml").write_text('[project]\nname = "e"\ndependencies = ["a"]\n') + t = dataclasses.replace(target.load(), engine_repo=repo, deps=("pyproject.toml",)) + monkeypatch.setattr(target, "load", lambda path=None: t) + monkeypatch.setenv("LAB_WORKER_CACHE", str(tmp_path / "cache")) + return t, repo + + +def pristine(tmp_path, repo: Path, change: dict[str, str]) -> Path: + """A two-commit tree like grader.pristine_tree's: base, then `change`.""" + from lab.safety import grader + from tests.lab_fixtures import make_repo + tree = tmp_path / "tree" + grader.export(make_repo(tmp_path), "HEAD", tree) + (tree / "pyproject.toml").write_text((repo / "pyproject.toml").read_text()) + grader._commit(tree, "base", init=True) + for rel, text in change.items(): + (tree / rel).write_text(text) + grader._commit(tree, "change") + return tree + + +def test_unchanged_dependencies_keep_the_engines_venv_and_changed_ones_build_once(deps_target, tmp_path, monkeypatch): + t, repo = deps_target + tree = pristine(tmp_path, repo, {"src/inference_server/engine.py": "def speed():\n return 2\n"}) + assert not build.changed(tree) and build.python_for(tree) == engine.python() + flat = tmp_path / "flat" # an exported base: no history, no change + flat.mkdir() + (flat / "pyproject.toml").write_text("anything") + assert not build.changed(flat) + built = [] + + def fake(tree, venv): + built.append(venv) + (venv / "env" / "bin").mkdir(parents=True) + (venv / ".lab-built").write_text("") + monkeypatch.setattr(build, "build", fake) + (tree / "pyproject.toml").write_text('[project]\nname = "e"\ndependencies = ["a", "b"]\n') + assert build.changed(tree) + py = build.python_for(tree) + assert py == str(built[0] / "env" / "bin" / "python") and build.python_for(tree) == py and len(built) == 1 + + +def test_the_allowed_hosts_are_the_index_defaults_plus_what_the_files_and_build_name(deps_target, tmp_path): + tree = tmp_path / "tree" + tree.mkdir() + (tree / "pyproject.toml").write_text('[[tool.uv.index]]\nurl = "https://download.pytorch.org/whl/cu130"\n') + got = build.hosts(tree, "uv sync --index-url https://mirror.example.org/simple") + assert got == {*build.INDEX_HOSTS, "mirror.example.org"} and "pypi.nvidia.com" in got + + +def proxy_request(sock_path: Path, line: bytes) -> bytes: + c = socket.socket(socket.AF_UNIX) + c.connect(str(sock_path)) + c.sendall(line) + out = c.recv(4096) + return c, out + + +def test_the_index_proxy_tunnels_to_allowed_hosts_only(tmp_path, monkeypatch): + echo = socket.socket() + echo.bind(("127.0.0.1", 0)) + echo.listen() + + def serve(): + conn, _ = echo.accept() + conn.sendall(conn.recv(100).upper()) + conn.close() + threading.Thread(target=serve, daemon=True).start() + real = socket.create_connection + monkeypatch.setattr(build.socket, "create_connection", + lambda addr, timeout=None: real(echo.getsockname(), timeout=timeout)) # "pypi.org:443" here + sock = Path("/tmp") / f"lab-test-{id(tmp_path)}.sock" + with build.IndexProxy({"pypi.org"}, sock): + c, head = proxy_request(sock, b"CONNECT pypi.org:443 HTTP/1.1\r\nHost: pypi.org\r\n\r\n") + assert head.startswith(b"HTTP/1.1 200") + c.sendall(b"hello") + assert c.recv(100) == b"HELLO" + for line in (b"CONNECT example.com:443 HTTP/1.1\r\n\r\n", b"CONNECT pypi.org:80 HTTP/1.1\r\n\r\n", + b"GET http://pypi.org/ HTTP/1.1\r\n\r\n"): + assert proxy_request(sock, line)[1].startswith(b"HTTP/1.1 403") + sock.unlink(missing_ok=True) + + +def _online() -> bool: + try: + socket.create_connection(("pypi.org", 443), timeout=5).close() + return True + except OSError: + return False + + +PROBE = r'''import os, socket, urllib.request, urllib.error +def get(url): + try: + return urllib.request.urlopen(url, timeout=20).status + except urllib.error.HTTPError as e: + return e.code + except Exception as e: # a refused tunnel is a URLError carrying the proxy's 403 + return "403" if "403" in str(e) else type(e).__name__ +direct = "no" +try: + socket.create_connection(("1.1.1.1", 443), timeout=5); direct = "yes" +except OSError: + pass +print("index", get("https://pypi.org/simple/pip/"), "other", get("https://example.com/"), "direct", direct) +''' + + +@pytest.mark.skipif(not (jail.backend() == "container" and container.available()), reason="needs the container jail") +@pytest.mark.skipif(not _online(), reason="needs the internet") +def test_a_build_room_reaches_the_index_through_the_proxy_and_nothing_else(deps_target, tmp_path, monkeypatch): + t, repo = deps_target + tree = tmp_path / "tree" + tree.mkdir() + (tree / "pyproject.toml").write_text('[project]\nname = "e"\ndependencies = ["b"]\n') + (tree / "probe.py").write_text(PROBE) + out = tmp_path / "probe.out" + out.touch() + cmd = (f'python3 -c "$(cat {tree}/probe.py)" > "$(dirname "$UV_PROJECT_ENVIRONMENT")/probe.out" 2>&1; ' + 'mkdir -p "$UV_PROJECT_ENVIRONMENT/bin" && ln -s /usr/bin/python3 "$UV_PROJECT_ENVIRONMENT/bin/python"') + monkeypatch.setattr(target, "load", lambda path=None: dataclasses.replace(t, build=cmd)) + real = container.argv + monkeypatch.setattr(container, "argv", lambda *a, **k: real(*a[:5], [*a[5], tree], *a[6:], **k)) # probe readable + venv = build.cache_dir() / "probe" + build.build(tree, venv) + said = (venv / "probe.out").read_text() + assert said.split() == ["index", "200", "other", "403", "direct", "no"], said diff --git a/tests/test_lab_container.py b/tests/test_lab_container.py index d65078c..fee4b4e 100644 --- a/tests/test_lab_container.py +++ b/tests/test_lab_container.py @@ -162,11 +162,25 @@ def test_quiet_kills_the_workbenchs_gpu_holders_then_pauses_and_thaws(monkeypatc apps = [[{"pid": 101, "used_mib": 9000, "name": "python"}, {"pid": 999, "used_mib": 1, "name": "other"}]] monkeypatch.setattr(gpu, "compute_apps", lambda: apps[0]) monkeypatch.setattr(container.os, "kill", lambda pid, sig: killed.append(pid) or apps.__setitem__(0, [apps[0][1]])) - monkeypatch.setattr(container.subprocess, "run", lambda argv, **k: calls.append(argv[1:]) or subprocess.CompletedProcess(argv, 0)) + monkeypatch.setattr(container.subprocess, "run", lambda argv, **k: calls.append(argv[1:]) or + subprocess.CompletedProcess(argv, 0, "false", "")) + states = lambda: [c for c in calls if c[0] in ("pause", "unpause")] # noqa: E731 with container.quiet("wb") as got: - assert calls == [["pause", "wb"]] + assert states() == [["pause", "wb"]] assert killed == [101] and [a["pid"] for a in got] == [101] # 999 is not the workbench's - assert calls == [["pause", "wb"], ["unpause", "wb"]] + assert states() == [["pause", "wb"], ["unpause", "wb"]] + + +def test_an_already_paused_workbench_is_left_paused(monkeypatch): + calls = [] + monkeypatch.setattr(container, "_docker", lambda: "docker") + monkeypatch.setattr(container, "running", lambda n: True) + monkeypatch.setattr(container.subprocess, "run", lambda argv, **k: calls.append(argv[1]) or + subprocess.CompletedProcess(argv, 0, "true", "")) + with container.paused("wb"): + with container.paused("wb"): + pass + assert "pause" not in calls and "unpause" not in calls def test_quiet_is_a_no_op_without_a_running_workbench(monkeypatch): diff --git a/tests/test_lab_runtime.py b/tests/test_lab_runtime.py index 97ec719..ec54c11 100644 --- a/tests/test_lab_runtime.py +++ b/tests/test_lab_runtime.py @@ -255,3 +255,17 @@ def script(n, ws: Path): assert t.model in system and t.engine_repo.name in system and "`stop`" in system and "`note`" in system assert "{" not in system.replace(t.engine.launch, "").replace(json.dumps(t.engine.env), "") # all filled assert not ADVICE.search(system), ADVICE.search(system) + + +def test_a_provider_that_works_locally_is_refused_with_a_remote_worker(cfg, monkeypatch): + from lab import worker + + class Far(worker.Remote): + def __init__(self): + pass + + def workspace_path(self, run_id): + return "/far" + monkeypatch.setattr(worker, "load", lambda **k: Far()) + with pytest.raises(ValueError, match="cannot run with LAB_WORKER=ssh"): + session.run(cfg, ScriptedProvider(lambda n, ws: iter(()))) diff --git a/tests/test_lab_safety.py b/tests/test_lab_safety.py index db6f432..ad6c1c2 100644 --- a/tests/test_lab_safety.py +++ b/tests/test_lab_safety.py @@ -19,7 +19,9 @@ ("tests/test_engine.py", False, False), # existing tests are evidence ("tests/conftest.py", False, False), ("src/inference_server/sitecustomize.py", True, False), - ("pyproject.toml", False, False), + ("pyproject.toml", False, True), # a dependency file: its dependency tables only (the audit) + ("uv.lock", False, True), + ("sub/pyproject.toml", False, False), ("lab/ledger.py", False, False), ("corpus/a/heldout.jsonl", True, False), ("CORPUS/A/HELDOUT.JSONL", True, False), @@ -82,7 +84,7 @@ def test_write_hook_lets_the_agent_edit_its_own_new_test(tmp_path): (ws / "tests/test_new.py").write_text("x") assert allowed(ws, base, "tests/test_new.py") # still "new": not in the base tree assert not allowed(ws, base, "tests/test_engine.py") - assert not allowed(ws, base, "../outside.py") and not allowed(ws, base, "pyproject.toml") + assert not allowed(ws, base, "../outside.py") and not allowed(ws, base, "tests/conftest.py") def test_jailed_timeout_is_a_failed_run(tmp_path, monkeypatch): @@ -135,3 +137,49 @@ def test_pristine_refuses_a_failed_audit(tmp_path): a = grader.audit(repo, "HEAD", ws) with pytest.raises(ValueError): grader.pristine_tree(repo, "HEAD", ws, a, tmp_path / "p") + + +PYPROJECT = """[project] +name = "x" +dependencies = ["a"] + +[tool.pytest.ini_options] +addopts = "-q" +""" + + +def test_a_pyproject_may_change_its_dependencies_and_nothing_else(tmp_path): + from lab.safety.surfaces import deps_only + assert deps_only(PYPROJECT, PYPROJECT.replace('["a"]', '["a", "b>=2"]')) + assert deps_only(PYPROJECT, PYPROJECT + '\n[tool.uv.sources]\nb = { index = "x" }\n') + assert not deps_only(PYPROJECT, PYPROJECT.replace('"-q"', '"-q -k nothing"')) # test config is the evaluator's + assert not deps_only(PYPROJECT, PYPROJECT + "[broken") + + +def test_the_audit_holds_a_pyproject_to_its_dependency_tables(tmp_path): + repo = make_repo(tmp_path) + ws = tmp_path / "ws" + grader.export(repo, "HEAD", ws) + text = (ws / "pyproject.toml").read_text() + (ws / "pyproject.toml").write_text(text.replace("[tool.ruff]", '[project]\nname = "t"\ndependencies = ["b"]\n\n[tool.ruff]')) + assert grader.audit(repo, "HEAD", ws).ok is False # [project].name is not a dependency + (ws / "pyproject.toml").write_text(text.replace("line-length = 100", "line-length = 300")) + assert "may change only the dependency tables of pyproject.toml" in grader.audit(repo, "HEAD", ws).violations + (ws / "pyproject.toml").write_text(text + '\n[dependency-groups]\nextra = ["c"]\n') + assert grader.audit(repo, "HEAD", ws).ok + (ws / "pyproject.toml").unlink() + assert "may not delete a dependency file: pyproject.toml" in grader.audit(repo, "HEAD", ws).violations + + +def test_a_venv_in_scratch_is_left_out_but_a_link_in_the_surface_is_a_violation(tmp_path): + repo = make_repo(tmp_path) + ws = tmp_path / "ws" + grader.export(repo, "HEAD", ws) + venv = ws / ".venv" / "bin" + venv.mkdir(parents=True) + (venv / "python").symlink_to("/usr/bin/python3") + (ws / ".venv" / "big.whl").write_bytes(b"x" * (grader.MAX_FILE_BYTES + 1)) + a = grader.audit(repo, "HEAD", ws) + assert a.ok and {".venv/bin/python", ".venv/big.whl"} <= set(a.scratch) + (ws / "src/inference_server/sneaky.py").symlink_to("/etc/passwd") + assert "symlink: src/inference_server/sneaky.py" in grader.audit(repo, "HEAD", ws).violations diff --git a/tests/test_lab_worker.py b/tests/test_lab_worker.py index 89fa359..b007197 100644 --- a/tests/test_lab_worker.py +++ b/tests/test_lab_worker.py @@ -135,3 +135,13 @@ def test_the_worker_builds_its_own_workbench_with_the_tunnelled_proxy(remote, tm assert argv[1:3] == ["run", "-i"] and "lab-workbench-r1-s1" in argv and str(cli) == argv[-1] assert w["sock"] == "/tmp/lab-proxy-r1-s1/proxy.sock" and "UNIX-CONNECT:/tmp/lab-proxy-r1-s1/proxy.sock" in argv[-2] assert "ANTHROPIC_BASE_URL=http://127.0.0.1:4000" in argv and not any("sk-" in a for a in argv) + + +@pytest.mark.parametrize("said,unpaused", [(b"paused\n", True), (b"\n", False)]) +def test_remote_pause_thaws_only_what_it_froze(said, unpaused): + r = worker.Remote(["ssh", "host"], "~/BlameGraph", "python", {}) + calls = [] + r._run = lambda command, stdin=b"": calls.append(command) or (said if command.startswith("pause") else b"") + with r.paused("lab-workbench-r1-s1"): + assert calls == ["pause lab-workbench-r1-s1"] + assert (calls[-1] == "unpause lab-workbench-r1-s1") is unpaused