diff --git a/CLAUDE.md b/CLAUDE.md index 7c4263d..a290788 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -63,8 +63,12 @@ pytest/ruff config). Each folder with commands has its own `python -m`. measures each regime's run-to-run band from repeated reference runs into `knowledge/noise/.json` (which `submit` reads). Exit 0 only if both hold. Needs a GPU for a real target; tested on fakes. `lab/LEDGER.md` is the record-format spec: the environment's public interface. -- `lab/RUNTIME.md` — design, not built (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/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/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, + Contaminated, NotReady, ValueError cross intact. Remote is tested through a local shell, not yet over SSH. - `lab/TRACE.md` + `lab/trace_contract.py` — the request-trace contract: per-request engine rows (trace_id = the client's `X-Trace-Id`, `arrival_ts` epoch s, `*_s` monotonic durations, TPOT = (E2E − TTFT)/(n − 1)) in a dir of `*.sqlite` (`requests` + `meta`) or `*.jsonl` + `.meta.json`, with schema version and drop counts; diff --git a/lab/README.md b/lab/README.md index 148dfd7..d7936dd 100644 --- a/lab/README.md +++ b/lab/README.md @@ -167,6 +167,10 @@ bridge; image from `python -m lab.safety.container build`, which `vm-setup.sh` r sandbox-runtime (`npm install -g @anthropic-ai/sandbox-runtime`); `LAB_JAIL` picks one. `LAB_NO_JAIL=1` waives the jail for tests and a box you trust. The agent's own CLI still runs under srt (`lab/RUNTIME.md`). +The GPU work behind `test`, `profile`, `trace`, `kernel`, `hostprof`, `bench`, `equiv` and `submit` runs as worker +jobs (`lab/worker.py`). `LAB_WORKER=local` (default) runs them in process; `LAB_WORKER=ssh` runs them on the VM +`lab.vm` names (start it and run `python -m lab.vm setup` first), shipping each tree by content hash. + ## Canaries python -m lab.canary slow_decode,kv_leak --port 8000 diff --git a/lab/RUNTIME.md b/lab/RUNTIME.md index 8059e13..6ccc544 100644 --- a/lab/RUNTIME.md +++ b/lab/RUNTIME.md @@ -2,8 +2,41 @@ Design, 2026-10-08. Decided with the user: the lab splits into a controller and a worker; no separate 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`). Not built: the split. Open decisions are marked -**Decide**. +below). Built: the container jail (`lab/safety/container.py`), GPU exclusivity during measurements, and the +measured jobs behind the worker interface (`lab/worker.py`). Not built: the agent on the worker (its container, the +API proxy) and the agent-side jobs. Open decisions are marked **Decide**. + +## The line (the user, 2026-10-09) + +Don't restrict the agent's search space; restrict only what would let it cheat or cost money. The agent never +touches (1) secrets and money: API keys, cloud credentials, the HF token; (2) the answer key: held-out workloads, +correctness reference outputs; (3) the scoreboard: ledger, measurement clients, grader, budget; (4) the GPU +during its own measurement. Everything else it can have. Decided with it: + +- **Internet:** the agent's workbench has it, open: the same capabilities Claude Code has (a shell with network, + web search and fetch, any package install), not gated behind tools. Hence nothing secret on the VM (rule 1). + Agent-facing tools are the user's to design; the lab builds none. +- **Installing libraries:** the agent can pip- and apt-install. Its room is a looser *workbench* container: GPU, + network, root inside the container (never on the VM host), writable system dirs. The tight container (no + network, no capabilities, read-only root) stays for the measurement rooms only (`test`, `profile`, served + engines). +- **Installed libraries reach measurement only through the change.** A measurement room is rebuilt from the + agent's code change alone, so a dependency change has to be part of it: the engine's dependency files (e.g. + `pyproject.toml`, `uv.lock`) join the target's write surface and are installed from scratch for the + measurement. Patching an installed library in the workbench changes no measured result. + The measurement room stays offline, a fairness rule rather than a safety one: an engine with network during + a measurement could forward requests to a remote GPU or API. So the install is its own build step before the + measurement (decided 2026-10-09): package indexes only, no GPU, no secrets, cached by lockfile hash. Anything the + engine needs at runtime (draft models, prebuilt kernels, files a library downloads on first use) must be in the + change, in the dependency files or saved in the workspace; the tool descriptions state that as a fact once the + build step exists (not before: today measurements still use the engine venv on the VM). +- **Fresh VM per run:** yes (rule 6). +- **No separate measurement VM** for now: container escape is not a realistic threat at this stage, and it + would double GPU cost. Revisit for a public leaderboard. +- **Refuse rather than kill** (#20) until the workbench can be frozen during a measurement (rule 4). + +Not decided here: held-out workloads are sampled from public datasets, so with internet the agent could +download them; the workload generator's owner decides (rephrased or non-public held-out prompts). ## Why @@ -27,7 +60,8 @@ cannot signal others), and everything in a jail dies when the jail's command exi |---|---|---|---| | Controller | laptop (later an always-on box) | session loop, ledger, referee, holdout guard, budget, snapshot store, every secret | no | | Worker | GPU VM, host side, outside every jail | job runner, measurement clients, profilers, GPU state | only through the engine's HTTP port | -| Jails | GPU VM, inside containers | agent's shell and workspace; each served engine | yes, it is the agent's | +| Workbench | GPU VM, the agent's container | agent's shell and workspace; GPU, internet, root inside | yes, it is the agent's | +| Measurement rooms | GPU VM, one tight container per job | a tree under `test`/`profile`, a served engine | its code is the agent's; nothing else reaches in | Rules: @@ -43,10 +77,10 @@ Rules: states what was killed (a fact). 5. **Measurement code is out of reach.** The clients run on the host, outside every container, from a tree the jails cannot read; their CPU cores are reserved (cpuset) so the engine cannot starve them. -6. **One VM per run, destroyed after** (**Decide**; recommended). Nothing from one run can touch the next one's numbers. Setup time is the +6. **One VM per run, created at run start and deleted at the end** (decided 2026-10-09). Nothing from one run can touch the next one's numbers. Setup time is the cost; a prebuilt image with the venv plus a reattached read-only weights disk cuts it. -7. **Network.** VM egress: the controller tunnel only. Agent jail: the API proxy only. Engine jail: nothing; its - port is reachable from the host alone. +7. **Network.** Workbench: the internet (decided 2026-10-09), so the VM holds nothing secret. Measurement rooms: + nothing; a served engine's port is reachable from the host alone. Left open by not having a separate measurement VM: the clients and the agent's code share a kernel, so a kernel or container escape defeats rules 4 and 5. A measurement VM in the same private network closes it later. @@ -119,6 +153,15 @@ JSON, raw files with hashes, and hardware facts (device, driver, clocks). Transport now: SSH from the controller, one worker. Transport later: a queue. The job list does not change. +Built (2026-10-09, `lab/worker.py`): `test`, `measure` (bench, submit and the base share it), `equiv` and `profile` +(all four instruments). The tools keep the policy, the ledger and the budget and call `self.worker`; `LAB_WORKER` +picks `local` (default: in process, as before) or `ssh` (the VM `lab.vm` names, started and set up beforehand). +Over SSH each input tree goes by content hash, at most once per worker (`has`, `put`), and the worker re-hashes a +fresh copy right before every job, refusing a mismatch. The job's out dir comes back as a tar; GpuBusy, +Contaminated, NotReady and ValueError are raised again on the controller. equiv ships the snapshot's earlier +answers with the job, so calling it again never re-rolls the gate. Tested through a local shell with the same CLI; +not yet over SSH to a VM. `agent`, `exec`, `snapshot`, `restore` and `facts` come with the agent container. + ## What moves where - Controller: `session`, `agent` (SDK side), the tool front ends in `tools`/`evaltools`/`proftools` (policy, @@ -148,7 +191,7 @@ database, still append-only. ## Order 1. ~~Decide the jail~~ (A) and build it. Done, minus the GPU check. -2. Without a GPU: the job interface with a local transport, the snapshot round trip and rule 3, the API proxy, - and the container jail's network, filesystem and freeze behaviour on any Linux box (Docker locally). +2. Without a GPU: ~~the job interface with a local transport, the snapshot round trip and rule 3~~ (done for the + measured jobs); next the agent container with its freeze, and the API proxy. 3. With a GPU (spend approval): `--gpus` and the profilers inside the container, then a first `lab.session` dry run with a tiny budget. diff --git a/lab/evaltools.py b/lab/evaltools.py index 19af29b..eb19f87 100644 --- a/lab/evaltools.py +++ b/lab/evaltools.py @@ -1,5 +1,6 @@ -"""The three tools that can make a change count: bench, equiv and submit. Each serves the agent's pristine tree -from outside the jail (lab/serve.py), measures it with the environment's own instruments, and writes the ledger. +"""The three tools that can make a change count: bench, equiv and submit. Each has the worker (lab/worker.py) serve +the agent's pristine tree outside the jail and measure it with the environment's own instruments; the policy and the +ledger stay here. bench the regimes on the seen split, short tier -> one headline per regime, raw equiv the correctness gate against the target's reference -> pass | fail | inconclusive, with every metric @@ -18,11 +19,12 @@ import json import os +import shutil import tempfile import time from pathlib import Path -from lab import artifacts, ledger, serve, target +from lab import artifacts, ledger, serve, target, worker from validity.holdout import HoldoutGuard, Refused HOLDOUT_BUDGET, HOLDOUT_THRESHOLD_PCT, HOLDOUT_SIGMA_PCT = 100, 2.0, 0.5 @@ -147,7 +149,6 @@ class EvalTools: # -- bench -------------------------------------------------------------------------- def bench(self, args: dict) -> str: snap = self._audited("bench", args) - t = target.load() names = _regimes(args, self.bench_defaults()) tier = args.get("tier") or "short" seed = args.get("seed", self.seed) @@ -155,18 +156,18 @@ def bench(self, args: dict) -> str: raise ValueError("seed must be an integer") tree = self._pristine() t0 = time.monotonic() - work, rows = self._work("bench"), [] + work = self._work("bench") try: - with serve.Served(tree, t.engine, log=self.s.run_dir / "serve-bench.log", - jailed=self.serve_jailed, artifacts=work) as srv: - results = run_regimes(srv.url, t, names, "seen", tier, seed, rows) - srv.exclusive() + r = self._measure(tree, work, "serve-bench.log", names=names, split="seen", tier=tier, passive=True, + seed=seed) except Exception as e: # a tool never crashes the session: the failure is the record - result = {**_failed(e), "seconds": time.monotonic() - t0, "artifacts": self._store(work, rows)} + result = {**_failed(e), "seconds": time.monotonic() - t0, + "artifacts": artifacts.store(work, self.s.ledger_root)} self._record("bench", "bench", args, result, snap, config={"split": "seen", "tier": tier, "seed": seed}) return _said("bench", result) - result = {"verdict": "ok", "metrics": headline(results), "regimes": results, "tier": tier, - "seconds": time.monotonic() - t0, "ready_s": srv.ready_s, "artifacts": self._store(work, rows)} + result = {"verdict": "ok", "metrics": headline(r["results"]), "regimes": r["results"], "tier": tier, + "seconds": time.monotonic() - t0, "ready_s": r["ready_s"], + "artifacts": artifacts.store(work, self.s.ledger_root)} self._record("bench", "bench", args, result, snap, config={"split": "seen", "tier": tier, "seed": seed}) return json.dumps({"snapshot": snap.id, "split": "seen", "tier": tier, "metrics": result["metrics"]}, indent=1) @@ -180,27 +181,25 @@ def equiv(self, args: dict) -> str: self._record("equiv", "equiv", args, {"verdict": "refused", "reason": f"no {tier}-tier reference outputs; " "run `python -m correctness reference` for this target first"}, snap) return "equiv refused: no reference outputs for this target" - from correctness import client, run as crun - from correctness.gate import Thresholds tree = self._pristine() out = self.s.run_dir / "equiv" / f"{snap.id}-{tier}.json" work = self._work("equiv") + # answers already collected for this snapshot are reused, never re-rolled: the gate is statistical + prior = Path(tempfile.mkdtemp(prefix="equiv-prior-", dir=self.s.run_dir)) + if out.with_suffix(".outputs.jsonl").exists(): + shutil.copyfile(out.with_suffix(".outputs.jsonl"), prior / "equiv.outputs.jsonl") try: - with serve.Served(tree, t.engine, log=self.s.run_dir / "serve-equiv.log", jailed=self.serve_jailed, - artifacts=work) as srv: - res = crun.candidate(ref, srv.url, t.model, out, self.encoder(t), - Thresholds(min_score_ratio=t.min_score_ratio, min_length_ratio=t.min_length_ratio), - concurrency=self.equiv_concurrency, api=client.Api(t.engine.api, t.chat_kwargs), - config={"tier": tier}) - srv.exclusive() + res = self.worker.call("equiv", {"tree": tree, "reference": ref, "prior": prior}, + {"tier": tier, "concurrency": self.equiv_concurrency, "jailed": self.serve_jailed}, + work) except Exception as e: - result = {**_failed(e), "artifacts": self._store(work, equiv_rows(out))} + result = {**_failed(e), "artifacts": self._equiv_store(work, out, prior)} self._record("equiv", "equiv", args, result, snap, config={"split": "seen", "tier": tier}) return _said("equiv", result) passed = {"pass": True, "fail": False}.get(res["verdict"]) # inconclusive is None: neither passing nor failing record = {"verdict": res["verdict"], "passed": passed, "reasons": res["reasons"], "gates": res["gates"], "metrics": res["metrics"], "thresholds": res["thresholds"], "tier": tier, - "artifacts": self._store(work, equiv_rows(out))} + "artifacts": self._equiv_store(work, out, prior)} self._record("equiv", "equiv", args, record, snap, config={"split": "seen", "tier": tier}) return json.dumps({"snapshot": snap.id, "tier": tier, "verdict": res["verdict"], "reasons": res["reasons"], "metrics": res["metrics"]}, indent=1, default=str) @@ -229,16 +228,11 @@ def submit(self, args: dict) -> str: base_seen = self.base_seen(t, names) guard = self.holdout_guard() tree = self._pristine() - # held-out passive data stays operator-only: never a blob, never in the record, never inside the jail - private, rows = self.s.run_dir / "heldout-private" / f"{snap.id}-{time.strftime('%Y%m%dT%H%M%S')}", [] - try: - with serve.Served(tree, t.engine, log=self.s.run_dir / "serve-submit.log", jailed=self.serve_jailed, - artifacts=private) as srv: - results = run_regimes(srv.url, t, names, "heldout", "full", self.seed, rows) - srv.exclusive() # before the holdout guard: a contaminated run spends no query - finally: - artifacts.write_rows(private, rows) - new = headline(results) + # held-out passive data stays operator-only: never a blob, never in the record, never inside the jail. + # A contaminated run raises here, before the holdout guard: it spends no query. + private = self.s.run_dir / "heldout-private" / f"{snap.id}-{time.strftime('%Y%m%dT%H%M%S')}" + r = self._measure(tree, private, "serve-submit.log", names=names, split="heldout", tier="full", passive=True) + new = headline(r["results"]) metrics, better = {}, {} for n in names: held_d, seen_d = delta_pct(base.get(n, {}), new[n]), delta_pct(base_seen.get(n, {}), seen[n]) @@ -273,16 +267,35 @@ def bench_defaults(self) -> tuple[str, ...]: equiv_concurrency = 16 serve_jailed = True - def encoder(self, t: target.Target): - from correctness.run import HFEncoder - return getattr(self.s, "encoder", None) or HFEncoder(t.model, t.chat_kwargs) - def _work(self, tool: str) -> Path: - """A scratch dir outside the jail for one served run's passive data; emptied into blobs by `_store`.""" + """A scratch dir outside the jail for one worker job's output; emptied into blobs or removed after.""" return Path(tempfile.mkdtemp(prefix=f"passive-{tool}-", dir=self.s.run_dir)) - def _store(self, work: Path, rows: list[dict]) -> dict: - artifacts.write_rows(work, rows) + def _measure(self, tree: Path, out: Path, log: str, **args) -> dict: + """The worker's `measure` job into `out`; the engine's log is appended to `/` either way.""" + try: + return self.worker.call("measure", {"tree": tree}, {"seed": self.seed, **args, "jailed": self.serve_jailed}, + out) + finally: + src = out / worker.ENGINE_LOG + if src.exists(): + with open(self.s.run_dir / log, "ab") as f: + f.write(src.read_bytes()) + src.unlink() + + def _equiv_store(self, work: Path, out: Path, prior: Path) -> dict: + """The gate's files to `out` (kept per snapshot for the next equiv), its rows and passive data as blobs.""" + shutil.rmtree(prior, ignore_errors=True) + out.parent.mkdir(parents=True, exist_ok=True) + for name, dest in (("equiv.json", out), ("equiv.outputs.jsonl", out.with_suffix(".outputs.jsonl"))): + if (work / name).exists(): + shutil.move(work / name, dest) + log = work / worker.ENGINE_LOG + if log.exists(): + with open(self.s.run_dir / "serve-equiv.log", "ab") as f: + f.write(log.read_bytes()) + log.unlink() + artifacts.write_rows(work, equiv_rows(out)) return artifacts.store(work, self.s.ledger_root) def holdout_guard(self) -> HoldoutGuard: @@ -299,9 +312,13 @@ def _base_measure(self, t: target.Target, names: list[str], split: str, tier: st tree = self.s.run_dir / f"base-tree-{split}" from lab.safety import grader grader.export(self.s.workspace.repo, self.s.workspace.base, tree) - with serve.Served(tree, t.engine, log=self.s.run_dir / f"serve-base-{split}.log", jailed=self.serve_jailed) as srv: - have.update(headline(run_regimes(srv.url, t, missing, split, tier, self.seed))) - srv.exclusive() # a contaminated base is never cached: every later submit reads it + work = self._work("base") + try: # a contaminated base raises: never cached, every later submit reads it + r = self._measure(tree, work, f"serve-base-{split}.log", names=missing, split=split, tier=tier, + passive=False) + finally: + shutil.rmtree(work, ignore_errors=True) + have.update(headline(r["results"])) cache.write_text(json.dumps(have, indent=1)) return have diff --git a/lab/proftools.py b/lab/proftools.py index da93092..c8c3f14 100644 --- a/lab/proftools.py +++ b/lab/proftools.py @@ -14,7 +14,6 @@ import time from pathlib import Path -from lab import engine from lab.agent import ToolSpec OUT = "_instr" # in the pristine tree: writable in the jail, rebuilt on every call @@ -68,84 +67,79 @@ def _head(path: Path | None) -> str: return path.read_text(errors="replace")[:SUMMARY_CHARS] if path and path.is_file() and not path.is_symlink() else "" -class ProfTools: - """Mixin for lab.tools.Toolbox: needs self._audited, self._pristine, self._record, self._keep.""" - - def trace(self, args: dict) -> str: - wl, flags = workload(args), _flags(args, "nsys_args", NSYS_FLAGS) - wrap = lambda exe, out: [exe, "profile", "--output", str(out / "trace"), "--force-overwrite", "true", - *flags, "--stats", "false"] - post = lambda exe, out: ([exe, "stats", "--report", NSYS_REPORTS, "--format", "csv", "--output", - str(out / "stats"), str(out / "trace.nsys-rep")], None) - return self._instrumented("trace", args, "nsys", wrap, wl + ["--profiler", "cuda-range"], post, - "stats_cuda_gpu_kern_sum.csv") - - def kernel(self, args: dict) -> str: +def instrument(kind: str, args: dict) -> dict: + """One instrument from validated args: its binary, `wrap(exe, out)` around lab.profile, lab.profile `flags`, + `post(exe, out)` -> (argv, file for its stdout) run after, the `summary` glob, and `extra(exe)` flags.""" + wl = workload(args) + if kind == "trace": + flags = _flags(args, "nsys_args", NSYS_FLAGS) + return {"binary": "nsys", "flags": wl + ["--profiler", "cuda-range"], "summary": "stats_cuda_gpu_kern_sum.csv", + "wrap": lambda exe, out: [exe, "profile", "--output", str(out / "trace"), "--force-overwrite", "true", + *flags, "--stats", "false"], + "post": lambda exe, out: ([exe, "stats", "--report", NSYS_REPORTS, "--format", "csv", "--output", + str(out / "stats"), str(out / "trace.nsys-rep")], None), + "extra": lambda exe: []} + if kind == "kernel": rx = kernel_regex(args["kernel_regex"]) if "kernel_regex" in args else None skip, count = _int(args, "launch_skip", 0, 0), _int(args, "launch_count", 8, 1) kset = args.get("set", "basic") if not isinstance(kset, str) or not kset: raise ValueError("set must be a non-empty string") - wl, flags = workload(args), _flags(args, "ncu_args") - wrap = lambda exe, out: [exe, "--export", str(out / "kernel"), "--force-overwrite", - *(["--kernel-name", f"regex:{rx}"] if rx else []), "--launch-skip", str(skip), - "--launch-count", str(count), "--set", kset, "--profile-from-start", "off", - "--target-processes", "all", *flags] - post = lambda exe, out: ([exe, "--import", str(out / "kernel.ncu-rep"), "--csv", "--page", "raw"], - out / "metrics.csv") - return self._instrumented("kernel", args, "ncu", wrap, wl + ["--profiler", "cuda-range"], post, "metrics.csv") - - def hostprof(self, args: dict) -> str: - wl = workload(args) + flags = _flags(args, "ncu_args") + return {"binary": "ncu", "flags": wl + ["--profiler", "cuda-range"], "summary": "metrics.csv", + "wrap": lambda exe, out: [exe, "--export", str(out / "kernel"), "--force-overwrite", + *(["--kernel-name", f"regex:{rx}"] if rx else []), "--launch-skip", str(skip), + "--launch-count", str(count), "--set", kset, "--profile-from-start", "off", + "--target-processes", "all", *flags], + "post": lambda exe, out: ([exe, "--import", str(out / "kernel.ncu-rep"), "--csv", "--page", "raw"], + out / "metrics.csv"), + "extra": lambda exe: []} + if kind == "hostprof": seconds, rate = _int(args, "seconds", 20, 1), _int(args, "rate", 100, 1) more = [w for f in _flags(args, "pyspy_args") for w in ("--pyspy-arg", f)] - extra = lambda exe: ["--profiler", "pyspy", "--seconds", str(seconds), "--rate", str(rate), "--pyspy-bin", exe, - *more] - return self._instrumented("hostprof", args, "py-spy", lambda exe, out: [], wl, None, - "bundle/*/pyspy-dump.txt", extra) - - def _instrumented(self, tool: str, args: dict, binary: str, wrap, flags: list[str], post, summary: str, - extra=lambda exe: []) -> str: - """`wrap(exe, out)` + lab.profile `flags`, jailed with the instrument readable; then `post`, jailed too - (the report was written by agent code). Kept as one blob; the head of `summary` is returned.""" - from lab import tools as T + return {"binary": "py-spy", "flags": wl, "summary": "bundle/*/pyspy-dump.txt", "wrap": lambda exe, out: [], + "post": None, + "extra": lambda exe: ["--profiler", "pyspy", "--seconds", str(seconds), "--rate", str(rate), + "--pyspy-bin", exe, *more]} + raise ValueError(f"unknown instrument {kind!r}") + + +class ProfTools: + """Mixin for lab.tools.Toolbox: needs self._audited, self._pristine, self._record, self._keep, self._work, + self.worker.""" + + def trace(self, args: dict) -> str: + return self._instrumented("trace", args) + + def kernel(self, args: dict) -> str: + return self._instrumented("kernel", args) + + def hostprof(self, args: dict) -> str: + return self._instrumented("hostprof", args) + + def _instrumented(self, tool: str, args: dict) -> str: + """The instrument around lab.profile's workload, then its report step, both jailed on the worker (the report + was written by agent code). Kept as one blob; the head of its summary is returned.""" + instrument(tool, args) # refuse bad arguments before anything is snapshotted or run snap = self._audited(tool, args) - exe = shutil.which(binary) - if not exe: - reason = f"{binary} is not on PATH on this host" - self._record("note", tool, args, {"verdict": "refused", "reason": reason}, snap) - return f"{tool} refused: {reason}" - exe = str(Path(exe).resolve()) tree = self._pristine() - out = tree / OUT - out.mkdir() - harness = T.stage_harness(tree) - read = [Path(exe).parent] - src = T.workload_flags(args, harness, _int(args, "requests", *WORKLOAD["requests"])) - argv = [*wrap(exe, out), engine.python(), "-m", "lab.profile", *flags, *src, *extra(exe), "--out", str(out / "bundle")] + out = self._work(tool) t0 = time.monotonic() - proc = T.default_profile_runner(tree, argv, read=read) - log = proc.stdout + proc.stderr - commands = [argv] - T._drop_links(out) - if post and proc.returncode == 0: - pargv, dest = post(exe, out) - commands.append(pargv) - p = T.default_profile_runner(tree, pargv, read=read) - if dest: - dest.write_text(p.stdout) - log += p.stderr if dest else p.stdout + p.stderr - proc = p if p.returncode else proc - T._drop_links(out) - found = sorted(out.glob(summary)) - head = _head(found[-1] if found else None) - blob, visible = self._keep(out) if any(p.is_file() for p in out.rglob("*")) else ("", "") - result = {"returncode": proc.returncode, "bundle": blob, "workspace_copy": visible, "instrument": exe, - "argv": commands, "seconds": time.monotonic() - t0, "output": log[-3000:]} + try: + r = self.worker.call("profile", {"tree": tree}, {"kind": tool, "args": args}, out) + if r.get("refused"): + self._record("note", tool, args, {"verdict": "refused", "reason": r["refused"]}, snap) + return f"{tool} refused: {r['refused']}" + files = out / "files" + blob, visible = self._keep(files) if files.is_dir() and any(p.is_file() for p in files.rglob("*")) else ("", "") + finally: + shutil.rmtree(out, ignore_errors=True) + result = {"returncode": r["returncode"], "bundle": blob, "workspace_copy": visible, "instrument": r["instrument"], + "argv": r["argv"], "seconds": time.monotonic() - t0, "output": r["output"]} self._record("profile", tool, args, result, snap) - if proc.returncode != 0 or not head: - return f"{tool} failed ({proc.returncode}):\n{result['output']}" - return f"{tool} output at {visible} (in your workspace; left out of your change)\n{head}" + if r["returncode"] != 0 or not r["head"]: + return f"{tool} failed ({r['returncode']}):\n{result['output']}" + return f"{tool} output at {visible} (in your workspace; left out of your change)\n{r['head']}" def prof_specs(self, gpu) -> list[ToolSpec]: """`gpu(time)` renders the cost fact for a description.""" diff --git a/lab/tools.py b/lab/tools.py index ee06e4b..b3a91ad 100644 --- a/lab/tools.py +++ b/lab/tools.py @@ -33,8 +33,11 @@ class Toolbox(EvalTools, ProfTools): """Bound to one session: knows the workspace, the budget, the ledger root and the run.""" def __init__(self, session: Any): + from lab import worker self.s = session self.violation: str | None = None + self.worker = getattr(session, "worker", None) or worker.load( + encoder=getattr(session, "encoder", None), profile_runner=getattr(session, "profile_runner", None)) # -- plumbing ------------------------------------------------------------------------ def _cost(self, tool: str) -> dict: @@ -97,9 +100,12 @@ def _keep(self, src: Path) -> tuple[str, str]: # -- tools --------------------------------------------------------------------------- def test(self, args: dict) -> str: snap = self._audited("test", args) - tree = self._pristine() - lint = grader.run_lint(tree) - tests = grader.run_tests(tree) if lint.passed else grader.Run(False, -1, "skipped: lint failed") + out = self._work("test") + try: + r = self.worker.call("test", {"tree": self._pristine()}, {}, out) + finally: + shutil.rmtree(out, ignore_errors=True) + lint, tests = grader.Run(**r["lint"]), grader.Run(**r["tests"]) result = {"lint": lint.passed, "tests": tests.passed, "returncode": tests.returncode, "lint_output": lint.output[-2000:], "test_output": tests.output[-4000:], "changed": snap.files + snap.deleted, "scratch_left_out": self.s.workspace.audit().scratch} @@ -110,22 +116,18 @@ def test(self, args: dict) -> str: def profile(self, args: dict) -> str: snap = self._audited("profile", args) - tree = self._pristine() - out = tree / "lab" / "runs" - harness = stage_harness(tree) - n = int(args.get("requests", 8)) - argv = [engine.python(), "-m", "lab.profile", "--requests", str(n), - "--max-tokens", str(args.get("max_tokens", 32)), "--out", str(out), - *workload_flags(args, harness, n)] + out = self._work("profile") t0 = time.monotonic() - proc = (self.s.profile_runner or default_profile_runner)(tree, argv) - bundles = sorted(out.glob("*")) if out.exists() else [] - blob, visible = self._keep(bundles[-1]) if bundles else ("", "") - result = {"returncode": proc.returncode, "bundle": blob, "workspace_copy": visible, - "seconds": time.monotonic() - t0, "output": (proc.stdout + proc.stderr)[-3000:]} + try: + r = self.worker.call("profile", {"tree": self._pristine()}, {"kind": "profile", "args": args}, out) + blob, visible = self._keep(out / "files") if (out / "files").is_dir() else ("", "") + finally: + shutil.rmtree(out, ignore_errors=True) + result = {"returncode": r["returncode"], "bundle": blob, "workspace_copy": visible, + "seconds": time.monotonic() - t0, "output": r["output"]} self._record("profile", "profile", args, result, snap) - if proc.returncode != 0 or not blob: - return f"profile failed ({proc.returncode}):\n{result['output']}" + if r["returncode"] != 0 or not blob: + return f"profile failed ({r['returncode']}):\n{result['output']}" return f"bundle at {visible} (in your workspace; left out of your change)\n" + "\n".join( f" {p.name}" for p in sorted((self.s.workspace.path / visible).iterdir())) diff --git a/lab/worker.py b/lab/worker.py new file mode 100644 index 0000000..1d895ce --- /dev/null +++ b/lab/worker.py @@ -0,0 +1,300 @@ +"""Where the GPU work runs (lab/RUNTIME.md). A job takes input trees and JSON arguments, writes files into an output +dir and returns JSON; the tools keep the policy, the ledger and the budget. `Local` runs a job in this process. +`Remote` ships each input by content hash, runs `python -m lab.worker run JOB` on the worker host over SSH and brings +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. + + test lint + the engine's suite on `tree` -> {lint, tests} + measure serve `tree`, run regimes; passive data into out -> {results, ready_s} + equiv serve `tree`, the correctness gate against `reference` -> the gate's result; outputs in out + profile lab.profile under torch, nsys, ncu or py-spy on `tree` -> {returncode, output, ...}; files in out + + python -m lab.worker has SHA... | put SHA < tar | run JOB < json > tar (the worker host's side) +""" + +from __future__ import annotations + +import dataclasses +import hashlib +import io +import json +import os +import shutil +import subprocess +import sys +import tarfile +import tempfile +from pathlib import Path + +from lab import engine, serve, target + +RESULT = "_result.json" +ENGINE_LOG = "engine.log" # the served engine's log, in every job's out dir + + +class WorkerError(RuntimeError): + """The worker could not run the job (transport, hash mismatch, or an unexpected failure there).""" + + +def tree_sha(root: Path) -> str: + """Content hash of a directory: every regular file's path, exec bit and bytes; refuses symlinks.""" + h = hashlib.sha256() + for p in sorted(Path(root).rglob("*")): + rel = p.relative_to(root).as_posix() + if p.is_symlink(): + raise WorkerError(f"symlink in a shipped tree: {rel}") + if p.is_file(): + h.update(f"{rel}\0{os.access(p, os.X_OK):d}\0{p.stat().st_size}\0".encode()) + h.update(p.read_bytes()) + return h.hexdigest() + + +# -- jobs (run where the GPU is) --------------------------------------------------------------------------------- + +def job_test(inputs: dict, out: Path, args: dict, hooks: dict) -> dict: + 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") + 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.evaltools import run_regimes + t = target.load() + rows = [] if args["passive"] else None + try: + with 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) + srv.exclusive() + return {"results": results, "ready_s": srv.ready_s} + finally: + if rows is not None: + artifacts.write_rows(out, rows) + + +def job_equiv(inputs: dict, out: Path, args: dict, hooks: dict) -> dict: + """`args`: tier, concurrency, jailed. The gate's outputs land in `out` as equiv.json and equiv.outputs.jsonl; + 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 + 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 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), + config={"tier": args["tier"]}) + srv.exclusive() + return res + + +def job_profile(inputs: dict, out: Path, args: dict, hooks: dict) -> dict: + """`args`: kind (profile | trace | kernel | hostprof) and the tool's args. Output files go to `out/files`.""" + from lab import proftools, tools as T + tree, kind, a = inputs["tree"], args["kind"], args["args"] + files = out / "files" + harness = T.stage_harness(tree) + if kind == "profile": + runs = tree / "lab" / "runs" + n = int(a.get("requests", 8)) + argv = [engine.python(), "-m", "lab.profile", "--requests", str(n), "--max-tokens", str(a.get("max_tokens", 32)), + "--out", str(runs), *T.workload_flags(a, harness, n)] + proc = (hooks.get("profile_runner") or T.default_profile_runner)(tree, argv) + bundles = sorted(runs.glob("*")) if runs.exists() else [] + if bundles: + T._drop_links(bundles[-1]) + shutil.copytree(bundles[-1], files) + return {"returncode": proc.returncode, "output": (proc.stdout + proc.stderr)[-3000:]} + ins = proftools.instrument(kind, a) + exe = shutil.which(ins["binary"]) + if not exe: + return {"refused": f"{ins['binary']} is not on PATH on this host"} + exe = str(Path(exe).resolve()) + work = tree / proftools.OUT + work.mkdir() + read = [Path(exe).parent] + 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")] + proc = T.default_profile_runner(tree, argv, read=read) + log, commands = proc.stdout + proc.stderr, [argv] + T._drop_links(work) + if ins["post"] and proc.returncode == 0: + pargv, dest = ins["post"](exe, work) + commands.append(pargv) + p = T.default_profile_runner(tree, pargv, read=read) + if dest: + dest.write_text(p.stdout) + log += p.stderr if dest else p.stdout + p.stderr + proc = p if p.returncode else proc + T._drop_links(work) + found = sorted(work.glob(ins["summary"])) + if any(p.is_file() for p in work.rglob("*")): + shutil.copytree(work, files) + return {"returncode": proc.returncode, "instrument": exe, "argv": commands, "output": log[-3000:], + "head": proftools._head(found[-1] if found else None)} + + +JOBS = {"test": job_test, "measure": job_measure, "equiv": job_equiv, "profile": job_profile} + +# Failures the tools turn into records; anything else is a WorkerError on the controller. +_ERRORS = {"GpuBusy": lambda e: serve.GpuBusy(e["apps"]), "Contaminated": lambda e: serve.Contaminated(e["apps"]), + "NotReady": lambda e: serve.NotReady(e["message"]), "ValueError": lambda e: ValueError(e["message"])} + + +def _error(e: BaseException) -> dict: + return {"type": type(e).__name__, "message": str(e), "apps": getattr(e, "apps", None)} + + +def _raise(err: dict | None) -> None: + if err: + make = _ERRORS.get(err["type"]) + raise make(err) if make else WorkerError(f"{err['type']}: {err['message']}") + + +# -- transports --------------------------------------------------------------------------------------------------- + +class Local: + """Jobs in this process, on the controller's own paths. `hooks`: in-process stand-ins (an encoder, a profile + runner) that tests and fakes provide; a remote worker has none.""" + + def __init__(self, **hooks): + self.hooks = {k: v for k, v in hooks.items() if v is not None} + + def call(self, job: str, inputs: dict[str, Path], args: dict, out: Path) -> dict: + out.mkdir(parents=True, exist_ok=True) + return JOBS[job](inputs, out, args, self.hooks) + + +def _tar(root: Path) -> bytes: + buf = io.BytesIO() + with tarfile.open(fileobj=buf, mode="w:gz") as t: + t.add(root, arcname=".") + return buf.getvalue() + + +def _untar(data: bytes, dest: Path) -> None: + dest.mkdir(parents=True, exist_ok=True) + with tarfile.open(fileobj=io.BytesIO(data), mode="r:gz") as t: + t.extractall(dest, filter="data") + + +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]): + self.prefix, self.env_dir, self.python, self.env = prefix, env_dir, python, env + + def _run(self, command: str, stdin: bytes = b"") -> bytes: + for v in self.env.values(): + 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}" + argv = [*self.prefix, line] if self.prefix else ["sh", "-c", line] + p = subprocess.run(argv, input=stdin, capture_output=True) + if p.returncode != 0: + raise WorkerError(f"worker `{command}` exited {p.returncode}: {p.stderr.decode(errors='replace')[-2000:]}") + return p.stdout + + def call(self, job: str, inputs: dict[str, Path], args: dict, out: Path) -> dict: + shas = {k: tree_sha(p) for k, p in inputs.items()} + have = set(self._run("has " + " ".join(shas.values())).decode().split()) if shas else set() + for k, p in inputs.items(): + if shas[k] not in have: + self._run(f"put {shas[k]}", _tar(p)) + _untar(self._run(f"run {job}", json.dumps({"inputs": shas, "args": args}).encode()), out) + reply = json.loads((out / RESULT).read_text()) + (out / RESULT).unlink() + _raise(reply.get("error")) + return reply["result"] + + +def load(**hooks): + """`LAB_WORKER`: local (default) or ssh, the VM `lab.vm` names (started and set up beforehand).""" + kind = os.environ.get("LAB_WORKER", "local") + if kind == "local": + return Local(**hooks) + if kind != "ssh": + raise ValueError(f"LAB_WORKER is local or ssh, not {kind!r}") + from lab import vm + v = vm.VM() + record = v.provider.get(v.name) + if record is None or record.state != "running": + raise WorkerError(f"{v.name} is not running: start it with `python -m lab.vm start` and set it up first") + spec = Path(os.environ.get("LAB_TARGET", "")).resolve() + if not spec.is_relative_to(engine.ENV_ROOT): + raise WorkerError(f"LAB_TARGET must be inside this repo to be found on the worker: {spec}") + 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)}) + + +# -- the worker host's side --------------------------------------------------------------------------------------- + +def cache_dir() -> Path: + return Path(os.environ.get("LAB_WORKER_CACHE") or engine.ENV_ROOT / "lab" / "runs" / "worker-cache") + + +def _put(sha: str, data: bytes) -> None: + dest = cache_dir() / sha + if dest.exists(): + return + part = Path(tempfile.mkdtemp(prefix=f"{sha[:12]}-", dir=cache_dir())) + _untar(data, part) + got = tree_sha(part) + if got != sha: + shutil.rmtree(part) + raise WorkerError(f"received tree hashes to {got}, not {sha}") + part.rename(dest) + + +def _run_job(job: str, request: dict) -> bytes: + """Fresh copies of the inputs (jobs write into them), each re-hashed right before use; out dir as a tar.""" + work = Path(tempfile.mkdtemp(prefix=f"job-{job}-")) + out = work / "out" + out.mkdir() + try: + try: + inputs = {} + for name, sha in request["inputs"].items(): + src = cache_dir() / sha + if not src.is_dir(): + raise WorkerError(f"input {name} ({sha}) was never put") + inputs[name] = work / name + shutil.copytree(src, inputs[name]) + if tree_sha(inputs[name]) != sha: + raise WorkerError(f"input {name} changed after it was put") + reply = {"result": JOBS[job](inputs, out, request["args"], {})} + except Exception as e: # the controller re-raises it; the tar still carries what was written + reply = {"error": _error(e)} + (out / RESULT).write_text(json.dumps(reply, default=str)) + return _tar(out) + finally: + shutil.rmtree(work, ignore_errors=True) + + +def main(argv: list[str] | None = None) -> int: + argv = sys.argv[1:] if argv is None else argv + cache_dir().mkdir(parents=True, exist_ok=True) + if argv[:1] == ["has"]: + print(" ".join(s for s in argv[1:] if (cache_dir() / s).is_dir())) + elif argv[:1] == ["put"] and len(argv) == 2: + _put(argv[1], sys.stdin.buffer.read()) + elif argv[:1] == ["run"] and len(argv) == 2 and argv[1] in JOBS: + sys.stdout.buffer.write(_run_job(argv[1], json.loads(sys.stdin.buffer.read()))) + else: + print(f"usage: python -m lab.worker has SHA... | put SHA | run {{{','.join(JOBS)}}}", file=sys.stderr) + return 2 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_lab_evaltools.py b/tests/test_lab_evaltools.py index f27741d..cec9d0f 100644 --- a/tests/test_lab_evaltools.py +++ b/tests/test_lab_evaltools.py @@ -158,6 +158,17 @@ def test_equiv_judges_against_the_reference(lab): assert all(j["engine"] for j in joined) # every gate request has its engine row (X-Trace-Id) +def test_a_second_equiv_on_the_same_snapshot_reuses_its_answers(lab): + tb, s = lab + tb.equiv({}) + answers = next((s.run_dir / "equiv").glob("*-dev.outputs.jsonl")) + marked = "".join(json.dumps({**json.loads(x), "kept": True}) + "\n" for x in answers.read_text().splitlines()) + answers.write_text(marked) + tb.equiv({}) + assert answers.read_text() == marked # nothing asked again: the gate is never re-rolled + assert not list(s.run_dir.glob("equiv-prior-*")) and not list(s.run_dir.glob("passive-*")) + + def test_submit_keeps_heldout_passive_data_out_of_the_ledger(lab): tb, s = lab tb.equiv({"tier": "full"}); tb.bench({"regimes": ["single_stream"]}) diff --git a/tests/test_lab_worker.py b/tests/test_lab_worker.py new file mode 100644 index 0000000..ebf9f27 --- /dev/null +++ b/tests/test_lab_worker.py @@ -0,0 +1,105 @@ +"""lab.worker: tree hashes, and the Remote transport end to end through a local shell (the same CLI ssh would run).""" + +from __future__ import annotations + +import os +import stat +import sys +from pathlib import Path + +import pytest + +from lab import engine, serve, target, worker +from lab.safety import grader +from tests.lab_fixtures import make_repo +from tests.test_lab_evaltools import _target + +MEASURE = {"names": ["single_stream"], "split": "seen", "tier": "short", "seed": 0, "jailed": False, "passive": False} + + +@pytest.fixture +def remote(tmp_path, monkeypatch): + """A Remote whose 'host' is this machine, with its own cache; returns (remote, a pristine tree, calls made).""" + monkeypatch.setenv("LAB_NO_JAIL", "1") + repo = make_repo(tmp_path) + spec = _target(tmp_path, repo, tmp_path / "ref") + monkeypatch.setenv("LAB_TARGET", str(spec)) + monkeypatch.setenv("LAB_ENGINE_REPO", str(repo)) + target._cache.clear() + tree = tmp_path / "tree" + grader.export(repo, "HEAD", tree) + r = worker.Remote([], str(engine.ENV_ROOT), sys.executable, {"LAB_WORKER_CACHE": str(tmp_path / "cache")}) + calls = [] + run = r._run + r._run = lambda command, stdin=b"": calls.append(command.split()[0]) or run(command, stdin) + yield r, tree, calls + target._cache.clear() + + +def test_tree_sha_covers_content_and_exec_bit_and_refuses_links(tmp_path): + (tmp_path / "a").mkdir() + (tmp_path / "a" / "f").write_text("x") + first = worker.tree_sha(tmp_path / "a") + (tmp_path / "a" / "f").chmod(0o755) + second = worker.tree_sha(tmp_path / "a") + (tmp_path / "a" / "f").write_text("y") + assert len({first, second, worker.tree_sha(tmp_path / "a")}) == 3 + (tmp_path / "a" / "l").symlink_to("/etc/passwd") + with pytest.raises(worker.WorkerError): + worker.tree_sha(tmp_path / "a") + + +def test_remote_runs_the_job_on_the_shipped_tree_and_ships_it_once(remote, tmp_path): + r, tree, calls = remote + res = r.call("test", {"tree": tree}, {}, tmp_path / "out1") + assert res["lint"]["passed"] and res["tests"]["passed"], res + assert calls == ["has", "put", "run"] + calls.clear() + r.call("test", {"tree": tree}, {}, tmp_path / "out2") + assert calls == ["has", "run"] # cached by hash: not shipped again + + +def test_remote_carries_gpu_busy_with_its_facts(remote, tmp_path, monkeypatch): + r, tree, _ = remote + bin_dir = tmp_path / "bin" + bin_dir.mkdir() + smi = bin_dir / "nvidia-smi" + smi.write_text(f"#!{sys.executable}\nimport sys\n" + "if any(a.startswith('--query-compute-apps') for a in sys.argv): print('4242, 9000, python')\n") + smi.chmod(smi.stat().st_mode | stat.S_IEXEC) + monkeypatch.setenv("PATH", f"{bin_dir}{os.pathsep}{os.environ['PATH']}") + with pytest.raises(serve.GpuBusy) as e: + r.call("measure", {"tree": tree}, MEASURE, tmp_path / "out") + assert e.value.apps == [{"pid": 4242, "used_mib": 9000, "name": "python"}] + + +def test_remote_carries_an_engine_that_did_not_start(remote, tmp_path, monkeypatch): + r, tree, _ = remote + spec = Path(os.environ["LAB_TARGET"]) + spec.write_text(spec.read_text().replace('launch = "python ', 'launch = "python -c \'import sys; sys.exit(3)\' #', 1)) + with pytest.raises(serve.NotReady): + r.call("measure", {"tree": tree}, MEASURE, tmp_path / "out") + assert (tmp_path / "out" / worker.ENGINE_LOG).exists() # what was written still comes back + + +def test_a_tree_changed_after_it_was_put_is_refused(remote, tmp_path): + r, tree, _ = remote + r.call("test", {"tree": tree}, {}, tmp_path / "out1") + cached = tmp_path / "cache" / worker.tree_sha(tree) + (cached / "src" / "inference_server" / "engine.py").write_text("def speed():\n return 100\n") + with pytest.raises(worker.WorkerError, match="changed after it was put"): + r.call("test", {"tree": tree}, {}, tmp_path / "out2") + + +def test_put_refuses_a_tree_that_does_not_hash_to_its_name(remote): + r, tree, _ = remote + with pytest.raises(worker.WorkerError, match="received tree hashes to"): + r._run(f"put {'0' * 64}", worker._tar(tree)) + + +def test_local_by_default_and_nothing_else_but_ssh(monkeypatch): + monkeypatch.delenv("LAB_WORKER", raising=False) + assert isinstance(worker.load(), worker.Local) + monkeypatch.setenv("LAB_WORKER", "gpu-box") + with pytest.raises(ValueError): + worker.load()