lab: the agent on a remote worker (CLI over SSH, workspace mirror, proxy tunnel)

CLAUDE.md

@@ -83,7 +83,9 @@ pytest/ruff config). Each folder with commands has its own `python -m`.
 - `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.
+  Contaminated, NotReady, ValueError cross intact. Remote is tested through a local shell, not yet over SSH. With
+  `ssh` the agent works on the VM too: its workspace mirrored by rsync (pulled before every audit), its workbench
+  built there (`workbench`), its CLI's stdio over `ssh -R`, the API proxy tunnelled to a unix socket in the workbench.
 - `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` + `<stem>.meta.json`, with schema version and drop counts;

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. Not built: the agent on a *remote* worker (its CLI over SSH, the
-workspace jobs, the proxy through a tunnel) and the dependency build step. Open decisions are marked **Decide**.
+container with its pause during measurements, and the agent on a remote worker. Not built: the dependency build
+step. Open decisions are marked **Decide**.
 
 ## The line (the user, 2026-10-09)
 
@@ -189,6 +189,27 @@ not yet over SSH to a VM. `agent`, `exec`, `snapshot`, `restore` and `facts` com
   the macOS srt path (srt is not installed on the dev box). Tested in Docker-in-Docker: workbench argv, root and
   writable system dirs, reaching the gateway, pause and thaw; the proxy against a fake API.
 
+## The agent on a remote worker, as built (2026-10-10)
+
+With `LAB_WORKER=ssh` the session loop, ledger, referee, budget and proxy stay on the controller and the agent works
+on the VM:
+
+- **Workspace**: the live copy is on the VM (`<repo>/lab/runs/remote/<run>/workspace`); the controller's is its
+  mirror. rsync by checksum (`--delete`): pushed before each session and after `restore` or a profile bundle copied
+  in, pulled before every lab tool's snapshot and audit and after the session. Pulled files are untrusted: links stay
+  links, and the audit refuses them as before.
+- **Workbench**: `python -m lab.worker workbench` on the VM builds that host's `docker run` (its GPU, paths, cores)
+  with the Linux CLI of the controller's Agent SDK version (installed once with uv into the worker cache). The
+  controller's `cli.sh` is `ssh -T -R <sock>:127.0.0.1:<proxy> vm '<docker run ...>' <SDK args>`: the CLI's stdio
+  is the SDK's over SSH, so the lab tools stay an in-process MCP server on the controller.
+- **Proxy**: on the controller, bound to its loopback. SSH forwards it to a unix socket in `/tmp/lab-proxy-<session>/`
+  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.
+- 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.
+
 ## What moves where
 
 - Controller: `session`, `agent` (SDK side), the tool front ends in `tools`/`evaltools`/`proftools` (policy,

lab/agent.py

@@ -9,6 +9,7 @@
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, Callable, Protocol
+from urllib.parse import urlsplit
 
 from lab import apiproxy, engine
 from lab.safety import container, jail
@@ -47,6 +48,8 @@ class AgentSpec:
     timeout_s: float = 3600.0
     model: str = field(default_factory=lambda: os.environ.get("LAB_MODEL", "claude-fable-5-1"))
     workbench: str | None = None            # the container name of the agent's workbench (Linux)
+    remote: Any = None                      # a lab.worker.Remote: the workbench runs on that host
+    remote_workspace: str | None = None     # the workspace's path there
 
 
 @dataclass
@@ -75,6 +78,44 @@ def load(name: str | None = None) -> Provider:
     raise ValueError(f"unknown agent provider {name!r}; LAB_AGENT_PROVIDER is claude or openai")
 
 
+PROXY_PORT_IN = 4000              # where a remote workbench serves the tunnelled proxy, inside its own network
+
+
+def _scratch(scratch: Path) -> tuple[Path, Path]:
+    home, tmp = scratch / "home", scratch / "tmp"
+    home.mkdir(parents=True, exist_ok=True)
+    tmp.mkdir(exist_ok=True)
+    return home, tmp
+
+
+def _weights() -> list[Path]:
+    return [HF_HUB] if HF_HUB.exists() else []      # model weights read-only; the token beside them stays hidden
+
+
+def cli_env(home: Path, tmp: Path, ws: Path, base_url: str) -> dict[str, str]:
+    """The CLI's whole environment inside its jail: no credential, only the proxy's URL and its placeholder."""
+    return {"HOME": str(home), "CLAUDE_CONFIG_DIR": str(home), "TMPDIR": str(tmp), "PYTHONPATH": str(ws / "src"),
+            "PYTHONDONTWRITEBYTECODE": "1", "HF_HUB_CACHE": str(HF_HUB), "HF_HUB_OFFLINE": "1",
+            "HF_HOME": str(tmp / "hf-home"), "ANTHROPIC_BASE_URL": base_url,
+            "ANTHROPIC_API_KEY": apiproxy.PLACEHOLDER, "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1"}
+
+
+def workbench_command(name: str, ws: Path, scratch: Path, cli: Path, base_url: str | None = None,
+                      sock_dir: Path | None = None) -> list[str]:
+    """`docker run` for the CLI's workbench on this host; the SDK's arguments go after it. With `sock_dir`, the proxy
+    is the unix socket `proxy.sock` there (tunnelled from the controller), served inside on 127.0.0.1:PROXY_PORT_IN."""
+    home, tmp = _scratch(scratch)
+    ws = Path(ws).resolve()
+    env = cli_env(home, tmp, ws, base_url or f"http://127.0.0.1:{PROXY_PORT_IN}")
+    env["PATH"] = f"{Path(engine.python()).parent}:{SYSTEM_PATH}"      # the agent's `python` is the engine's
+    command = [str(cli)] if sock_dir is None else [
+        "sh", "-c", f'socat TCP-LISTEN:{PROXY_PORT_IN},bind=127.0.0.1,fork,reuseaddr UNIX-CONNECT:{sock_dir}/proxy.sock '
+        '& exec "$0" "$@"', str(cli)]
+    writable = [home.resolve(), tmp.resolve(), ws] + ([Path(sock_dir)] if sock_dir else [])
+    return container.workbench_argv(name, ws, command, env, writable, [engine.venv(), *_weights(), cli.parent],
+                                    engine.python())
+
+
 class ClaudeAgentSDK:
     """Claude through the Agent SDK CLI, jailed with a wiped env (on Linux in the workbench container, elsewhere under
     srt); its model calls go through the lab's API proxy, so no credential is ever inside; lab tools run out here."""
@@ -86,35 +127,33 @@ def containerized() -> bool:
         return jail.backend() == "container" and container.available()
 
     def _wrapper(self, spec: AgentSpec, cli: Path, base_url: str) -> Path:
-        home, tmp = spec.scratch / "home", spec.scratch / "tmp"
-        home.mkdir(parents=True, exist_ok=True)
-        tmp.mkdir(exist_ok=True)
-        ws = spec.workspace.resolve()
-        weights = [HF_HUB] if HF_HUB.exists() else []      # model weights read-only; the token beside them stays hidden
-        env = {"HOME": str(home), "CLAUDE_CONFIG_DIR": str(home), "TMPDIR": str(tmp), "PYTHONPATH": str(ws / "src"),
-               "PYTHONDONTWRITEBYTECODE": "1", "HF_HUB_CACHE": str(HF_HUB), "HF_HUB_OFFLINE": "1",
-               "HF_HOME": str(tmp / "hf-home"), "ANTHROPIC_BASE_URL": base_url,
-               "ANTHROPIC_API_KEY": apiproxy.PLACEHOLDER, "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1"}
-        engine_bin = Path(engine.python()).parent                 # the agent's `python` is the engine's
-        if self.containerized():
-            env["PATH"] = f"{engine_bin}:{SYSTEM_PATH}"
-            argv = container.workbench_argv(spec.workbench, ws, [str(cli)], env, [home.resolve(), tmp.resolve(), ws],
-                                            [engine.venv(), *weights, cli.parent], engine.python())
+        script = spec.scratch / "cli.sh"
+        if spec.remote is not None:            # the worker host builds its own workbench; the proxy goes down the tunnel
+            import claude_agent_sdk
+            w = spec.remote.workbench(spec.workbench, spec.scratch.name, spec.remote_workspace,
+                                      claude_agent_sdk.__version__)
+            *ssh, target = spec.remote.prefix
+            line = (f'args=$(printf " %q" "$@")\ncmd={shlex.quote(shlex.join(w["argv"]))}\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():
+            argv = workbench_command(spec.workbench, spec.workspace, spec.scratch, cli, base_url=base_url)
             at = argv.index(container.IMAGE)
             argv[at:at] = [w for k in PASS_ENV for w in ("--env", k)]     # the SDK's values, passed through
-            line = f"exec {shlex.join(argv)} \"$@\""
+            line = f'exec {shlex.join(argv)} "$@"'
         else:
+            home, tmp = _scratch(spec.scratch)
+            ws = spec.workspace.resolve()
             tmp_root = "/private/tmp" if sys.platform == "darwin" else "/tmp"
             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())
             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 env.items())
-            line = f'exec env -i {keep} "PATH={engine_bin}:$PATH" {sets} {shlex.join(argv)} "$@"'
-        script = spec.scratch / "cli.sh"
-        script.write_text(f"#!/bin/sh\n{line}\n")
+            sets = " ".join(f"{k}={shlex.quote(v)}" for k, v in cli_env(home, tmp, ws, base_url).items())
+            line = f'exec env -i {keep} "PATH={Path(engine.python()).parent}:$PATH" {sets} {shlex.join(argv)} "$@"'
+        script.write_text(f"#!/bin/bash\n{line}\n")
         script.chmod(0o700)
         return script
 
@@ -142,7 +181,8 @@ async def call(args: dict[str, Any]) -> dict[str, Any]:
             output_format={"type": "json_schema", "schema": OUTPUT_SCHEMA},
             cli_path=str(self._wrapper(spec, cli, base_url)),
             hooks={"PreToolUse": [HookMatcher(matcher=WRITE_TOOLS,
-                                              hooks=[write_guard(spec.workspace, spec.base_files)])]},
+                                              hooks=[write_guard(Path(spec.remote_workspace or spec.workspace),
+                                                                 spec.base_files)])]},
         )
         reply = AgentReply(None, 0.0, 0, "no result message")
         async for msg in query(prompt=spec.prompt, options=options):
@@ -156,7 +196,7 @@ def run(self, spec: AgentSpec) -> AgentReply:
         """Never raises. The spend is what the API proxy priced, whatever the CLI reports or however it ended."""
         spec.scratch.mkdir(parents=True, exist_ok=True)
         try:
-            host = container.bridge_gateway() if self.containerized() else "127.0.0.1"
+            host = container.bridge_gateway() if spec.remote is None and self.containerized() else "127.0.0.1"
             proxy = apiproxy.Proxy(spec.max_budget_usd, host)
         except Exception as e:                      # nothing was sent: nothing spent
             return AgentReply(None, 0.0, 0, f"provider failed: {type(e).__name__}: {e}")
@@ -167,8 +207,10 @@ def run(self, spec: AgentSpec) -> AgentReply:
                 reply = AgentReply(None, 0.0, 0, f"timed out after {spec.timeout_s:.0f}s")
             except Exception as e:
                 reply = AgentReply(None, 0.0, 0, f"provider failed: {type(e).__name__}: {e}")
-            finally:
-                if spec.workbench and self.containerized():
-                    container.remove(spec.workbench)        # a CLI killed mid-session leaves its container behind
+            finally:                                    # a CLI killed mid-session leaves its container behind
+                if spec.remote is not None:
+                    spec.remote.remove_workbench(spec.workbench)
+                elif spec.workbench and self.containerized():
+                    container.remove(spec.workbench)
         reply.cost_usd, reply.cost_estimated, reply.usage = proxy.spent_usd, False, dict(proxy.usage)
         return reply

lab/session.py

@@ -11,7 +11,7 @@
 from dataclasses import dataclass, field
 from pathlib import Path
 
-from lab import agent, engine, experiments, ledger, target
+from lab import agent, engine, experiments, ledger, target, worker
 from lab.budget import Budget, BudgetExceeded
 from lab.safety import grader
 from lab.task import Task
@@ -34,6 +34,8 @@ class Session:
     profile_runner: object = None          # None: lab.tools.default_profile_runner
     task: Task | None = None
     workbench: str | None = None           # the agent's container: paused while the tools measure
+    worker: object = None                  # None: lab.worker.load() per toolbox
+    remote_workspace: str | None = None    # the workspace's mirror on a Remote worker, where the agent works
 
 
 @dataclass
@@ -124,27 +126,36 @@ def run(cfg: RunConfig, provider: agent.Provider | None = None) -> dict:
     done = sum(1 for _ in ledger.records(cfg.ledger_root, run=run_id, kind="session"))
     base_files = grader.base_files(cfg.repo, base)
     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
+    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)).baseline(base)
+        Toolbox(Session(run_id, f"{run_id}-baseline", run_dir, ws, budget, cfg.ledger_root, task=cfg.task,
+                        worker=remote)).baseline(base)
 
     for n in range(done + 1, done + cfg.max_sessions + 1):
         if budget.remaining_usd < MIN_SESSION_USD:
             summary["stopped"] = "budget"
             break
         s = Session(run_id, f"{run_id}-s{n}", run_dir, ws, budget, cfg.ledger_root, task=cfg.task,
-                    workbench=f"lab-workbench-{run_id}-s{n}")
+                    workbench=f"lab-workbench-{run_id}-s{n}", worker=remote, remote_workspace=rws)
         tools = Toolbox(s)
         snap = ws.snapshot()
         spec = agent.AgentSpec(system="", prompt=brief(cfg, s, snap.id, n), workspace=ws.path,
                                scratch=run_dir / "scratch" / s.session_id, tools=tools.specs(),
                                base_files=base_files, max_turns=cfg.max_turns,
-                               max_budget_usd=min(cfg.session_usd, budget.remaining_usd), workbench=s.workbench)
+                               max_budget_usd=min(cfg.session_usd, budget.remaining_usd), workbench=s.workbench,
+                               remote=remote, remote_workspace=rws)
         spec.system = system_prompt(target.load(), spec)
         t0 = time.monotonic()
         try:
+            if remote:
+                remote.push(ws.path, rws)
             reply = provider.run(spec)
         except Exception as e:                  # a provider that raises still gets charged and recorded
             reply = agent.AgentReply(None, spec.max_budget_usd, 0, f"provider raised: {e}", True)
+        if remote:
+            remote.pull(rws, ws.path)           # what the agent left on the worker is what is snapshotted
         clean_pristine(run_dir)
         try:
             budget.charge(reply.cost_usd, s.session_id)

lab/tools.py

@@ -88,13 +88,21 @@ def _audited(self, tool: str, args: dict):
         self._t0 = time.monotonic()
         if tool in GPU_TOOLS and self.s.budget.remaining_usd <= 0 and gpu_rate()[0] > 0:
             raise BudgetExceeded(f"{tool}: no budget left for GPU time")
+        self._sync("down")
         snap = self.s.workspace.snapshot()
         if snap.violations:
             self.violation = "; ".join(snap.violations)
             self._record("note", tool, args, {"violation": snap.violations}, snap)
             raise PermissionError(f"workspace violates the write surface: {self.violation}")
         return snap
 
+    def _sync(self, way: str) -> None:
+        """With the agent on a Remote worker: its workspace there is the live one, this one its mirror."""
+        rws = getattr(self.s, "remote_workspace", None)
+        if rws:
+            (self.worker.pull(rws, self.s.workspace.path) if way == "down" else
+             self.worker.push(self.s.workspace.path, rws))
+
     def _pristine(self) -> Path:
         a = self.s.workspace.audit()
         dest = self.s.run_dir / "pristine"
@@ -110,6 +118,7 @@ def _keep(self, src: Path) -> tuple[str, str]:
         dest = self.s.workspace.path / visible
         shutil.rmtree(dest, ignore_errors=True)
         shutil.copytree(src, dest)
+        self._sync("up")
         return blob, str(visible)
 
     # -- tools ---------------------------------------------------------------------------
@@ -167,6 +176,7 @@ def restore(self, args: dict) -> str:
         sid = args["snapshot"]
         self._audited("restore", args)
         self.s.workspace.restore(sid)
+        self._sync("up")
         snap = self._audited("restore", args)
         self._record("note", "restore", args, {"restored": sid, "now": snap.id}, snap)
         return f"workspace restored to {sid}"