"""Run an agent CLI on a cluster node instead of the workstation. Host choice is by measured load average penalised by this workstation's in-flight placement claims: load average lags a freshly started run by tens of seconds, so probes alone put every concurrent dispatch on the same node. The checkout travels by `git push`, not by copying files: a commit snapshotting HEAD plus every tracked change and untracked, non-ignored file is pushed (delta-only) to a bare mirror on the node, then checked out into a real, working git repository at the node's workspace — this holds for a linked worktree exactly as for a plain checkout, since `git push` already resolves a linked worktree's gitdir locally. Results travel home the same way: the node snapshots its post-run state, pushes it to the same bare mirror, and the workstation fetches and applies it as an uncommitted change against the pushed commit — a delta, never a tree copy. If the local checkout changed while the dispatch was running, the result is left on a fetched ref rather than silently overwriting it. This removes two limitations of an earlier tree-mirror transport: a node-side commit's sha, parent, and message are no longer dropped — the underlying commit object crosses the wire and is fetched home, even though it still surfaces as an uncommitted change by design; and a linked worktree's gitdir no longer needs separate mirroring or remote pointer rewriting, since only ordinary git objects and refs move. One limitation remains: an out-of-tree `core.hooksPath` does not resolve on the node, so hooks do not run there. A checkout with no `HEAD` yet (freshly `git init`ed, nothing committed) falls back to a tree-copy transport, since there is nothing to push a delta against. A caller that needs no repo files — a pure-inference research or review prompt — can set `OD_DISPATCH_NO_WORKSPACE=1` to skip the transport entirely: the node gets an empty scratch directory as its working directory, nothing is pushed, nothing is pulled back. Host selection, placement claims, and containment recording are unchanged; only the transfer is skipped. `-C`/`--cd` is refused in this mode — there is no mirrored checkout to point at. """ from __future__ import annotations import hashlib import json import os import random import re import shlex import subprocess import sys import tempfile import time import uuid from dataclasses import dataclass from pathlib import Path REGISTRY_PATH = Path.home() / ".claude" / "buildbox-hosts.json" AGENT_SANDBOX_BIN = Path.home() / ".claude" / "bin" / "agent-sandbox" REMOTE_ROOT = "cdx-offload" REMOTE_GIT_COMMON_ROOT = f"{REMOTE_ROOT}/git-common" # sandbox-run's default workspace for an id, bind-mounted at CONTAINER_WORKSPACES_ROOT by its # unconditional $SANDBOX_ROOT:/sandbox mount, and the only root receiver.mjs will resolve. NODE_WORKSPACES_ROOT = "sandbox/workspaces" CONTAINER_WORKSPACES_ROOT = "/sandbox/workspaces" SEAT_ROLE = "agent-seat" SANDBOX_ROLE = "agent-sandbox" DOOR = "tailscale_ip" SYNC_EXCLUDES = ( ".worktrees", # overlayfs workdir: mode-000 by design, so rsync exits 23 on it, and a live # jail mutates it mid-transfer. ".tmpjail-work", "node_modules", "dist", "build", ".next", ".turbo", ".cache", ".venv", "target", ) SECRET_ALLOWLIST = frozenset({"modules/harness/seat/credentials.json"}) SECRET_PATH_PATTERNS = ( ("environment file (.env or .env.*)", re.compile(r"(^|/)\.env(?:\..*)?$")), ("private key/certificate (*.pem/*.key/*.p12/*.pfx)", re.compile(r"(?i)\.(?:pem|key|p12|pfx)$")), ("SSH private key (id_rsa*/id_ed25519*/id_ecdsa*/id_dsa*)", re.compile(r"(^|/)id_(?:rsa|ed25519|ecdsa|dsa)[^/]*$")), ("secret directory (.secrets/**)", re.compile(r"(^|/)\.secrets(?:/|$)")), ("secret file (*.secret)", re.compile(r"(?i)\.secret$")), ("AWS credentials", re.compile(r"(^|/)\.aws/(?:credentials|config)$")), ("GCP credentials", re.compile(r"(?i)(^|/)(?:application_default_credentials\.json|service[-_]?account(?:[^/]*)\.json)$")), ("SSH credential file", re.compile(r"(^|/)\.ssh/(?:id_[^/]+|[^/]+\.(?:pem|key|p12|pfx))$")), ) AUTH_CONFIG_NAMES = frozenset({".npmrc", ".pypirc"}) AUTH_CONFIG_RE = re.compile( r"(?im)(?:_auth(?:token)?|password|username|token|//[^\s=]+/:_authToken)\s*[=:]" ) PROBE_TIMEOUT_SEC = 10 SYNC_TIMEOUT_SEC = 900 CLAIM_DIR_ENV = "OD_PLACEMENT_CLAIM_DIR" NO_WORKSPACE_ENV = "OD_DISPATCH_NO_WORKSPACE" K3S_DISPATCH_ENV = "OD_DISPATCH_K3S" DEFAULT_CLAIM_DIR_RELPATH = ".local/state/overdeck/placement-claims" CLAIM_SUFFIX = ".claim" CLAIM_LOAD_PENALTY = 1.0 EXIT_NO_NODE = 80 # registry unreadable, no reachable agent-seat host, or no access door EXIT_CREDENTIAL = 81 # resolved credential file missing or unreadable EXIT_MIRROR = 82 # mkdir, rsync, or git-mirror preparation failed EXIT_UNCONTAINABLE = 84 # this invocation has no containerized path # agent-sandbox's own exit codes: the launcher stopped before any container ran. Its # RC_USAGE (2) is excluded — this module builds the argv, so a usage failure is a caller # bug, while 2 is a common agent exit code. RC_HOST = 3 RC_CREDENTIAL_STAGING = 10 RC_PREFLIGHT = 11 RC_SYSTEMD_RUN = 12 RC_RECURSION = 13 RC_GIT_MOUNT = 14 LAUNCHER_FAILURE_CODES = frozenset( {RC_HOST, RC_CREDENTIAL_STAGING, RC_PREFLIGHT, RC_SYSTEMD_RUN, RC_RECURSION, RC_GIT_MOUNT} ) @dataclass(frozen=True) class Credential: runtime: str slug: str | None path: Path class OffloadUnavailable(RuntimeError): """No cluster node can take this run.""" def __init__(self, message: str, code: int = EXIT_NO_NODE) -> None: super().__init__(message) self.code = code def in_container() -> bool: """Machine-local evidence, not env: an env marker is agent-settable. /proc/self/cgroup is deliberately not consulted — measured inside a rootless podman container it names neither libpod nor docker, so it would answer "not contained" and re-dispatch forever.""" return any(probe.exists() for probe in (Path("/run/.containerenv"), Path("/.dockerenv"))) def should_offload() -> bool: return not in_container() def no_workspace_requested() -> bool: return bool(os.environ.get(NO_WORKSPACE_ENV)) def load_registry(path: Path = REGISTRY_PATH) -> dict: try: return json.loads(Path(path).read_text(encoding="utf-8")) except (OSError, ValueError) as exc: raise OffloadUnavailable( f"buildbox registry unreadable at {path}: {exc}", code=EXIT_NO_NODE ) from exc def candidate_hosts(registry: dict, roles: tuple[str, ...] = (SEAT_ROLE, SANDBOX_ROLE)) -> list[dict]: """Reachable hosts carrying every role in `roles`. A host in any other state is never contacted, and one without the sandbox image provisioned is never a candidate: an unprovisioned node fails with command-not-found, and no failure returns the run here.""" hosts = [ host for host in registry.get("hosts", []) if host.get("state") == "reachable" and set(roles) <= set(host.get("roles") or []) ] if not hosts: raise OffloadUnavailable( f"no reachable host with roles {', '.join(roles)} in the registry", code=EXIT_NO_NODE, ) return hosts def access_for(host: dict, door: str = DOOR) -> dict: access = (host.get("access") or {}).get(door) if not access or not access.get("host"): raise OffloadUnavailable( f"{host.get('name')}: no usable {door} access door", code=EXIT_NO_NODE ) return access def _host_name(host: dict, access: dict) -> str: return host.get("name") or access["host"] def ssh_argv(access: dict, extra: tuple[str, ...] = ()) -> list[str]: """`-F /dev/null` keeps a sandboxed or agent-poisoned ssh_config out of the path.""" argv = [ "ssh", "-F", "/dev/null", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=accept-new", "-o", f"ConnectTimeout={PROBE_TIMEOUT_SEC}", ] if access.get("identity_file"): argv += ["-i", os.path.expanduser(access["identity_file"])] if access.get("port"): argv += ["-p", str(access["port"])] argv += list(extra) argv.append(f"{access.get('user', 'user')}@{access['host']}") return argv def rsync_shell(access: dict) -> str: return shlex.join(ssh_argv(access)[:-1]) def parse_loadavg(stdout: str) -> float | None: try: return float(stdout.split()[0]) except (AttributeError, IndexError, ValueError): return None def probe_load(access: dict, run=subprocess.run) -> float | None: try: done = run( [*ssh_argv(access), "cat /proc/loadavg"], capture_output=True, text=True, timeout=PROBE_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError): return None if done.returncode != 0: return None return parse_loadavg(done.stdout) def claim_dir() -> Path: override = os.environ.get(CLAIM_DIR_ENV) if override: return Path(override).expanduser() return Path.home() / DEFAULT_CLAIM_DIR_RELPATH def _claim_token(value: str) -> str: return re.sub(r"[^A-Za-z0-9._-]", "_", value) or "unnamed" def _process_start_ticks(pid: int) -> str | None: """Field 22 of /proc//stat, read after the last ')' so a spaced comm cannot shift it.""" try: stat = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8", errors="replace") except OSError: return None try: return stat[stat.rindex(")") + 1 :].split()[19] except (ValueError, IndexError): return None def _owner_is_live(pid: int, start_ticks: str) -> bool: """Liveness by process identity, never by elapsed time: a killed dispatch frees its host.""" current = _process_start_ticks(pid) if current is not None and start_ticks != "0": return current == start_ticks return Path(f"/proc/{pid}").exists() @dataclass class PlacementClaim: """One in-flight dispatch, recorded so a concurrent selection can see it before load does.""" host_name: str path: Path def release(self) -> None: try: self.path.unlink() except OSError: pass def _parse_claim(name: str) -> tuple[str, int, str] | None: if not name.endswith(CLAIM_SUFFIX): return None parts = name[: -len(CLAIM_SUFFIX)].rsplit("__", 3) if len(parts) != 4: return None host, pid, start_ticks, _unique = parts try: return host, int(pid), start_ticks except ValueError: return None def live_claim_counts(directory: Path) -> dict[str, int]: """In-flight claims per host token, unlinking any whose owning process is gone.""" counts: dict[str, int] = {} try: entries = list(directory.iterdir()) except OSError: return counts for entry in entries: parsed = _parse_claim(entry.name) if parsed is None: continue host, pid, start_ticks = parsed if _owner_is_live(pid, start_ticks): counts[host] = counts.get(host, 0) + 1 continue try: entry.unlink() except OSError: pass return counts def acquire_claim(host_name: str, directory: Path) -> PlacementClaim | None: """Atomic O_EXCL create of a uniquely named empty file. None means bookkeeping is off.""" pid = os.getpid() name = ( f"{_claim_token(host_name)}__{pid}__{_process_start_ticks(pid) or '0'}" f"__{uuid.uuid4().hex}{CLAIM_SUFFIX}" ) path = directory / name try: directory.mkdir(parents=True, exist_ok=True) os.close(os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)) except OSError: return None return PlacementClaim(host_name=host_name, path=path) def select_host( hosts: list[dict], probe=probe_load, claims: Path | None = None ) -> tuple[dict, dict, PlacementClaim | None]: """Least-loaded reachable node, penalised by this workstation's in-flight claims. Load average lags a newly started run, so probes alone stack concurrent dispatches on one node. The claim held by every live dispatch is what fans them out; equal scores are broken at random so simultaneous callers do not walk the same branch. The returned claim is live until released. """ directory = claim_dir() if claims is None else Path(claims) probed: list[tuple[dict, dict, float]] = [] failures: list[str] = [] for host in hosts: try: access = access_for(host) except OffloadUnavailable as exc: failures.append(str(exc)) continue load = probe(access) if load is None: failures.append(f"{host.get('name')}: unreachable or load probe failed") continue probed.append((host, access, load)) if not probed: raise OffloadUnavailable( "; ".join(failures) or "no candidate node answered", code=EXIT_NO_NODE ) held: PlacementClaim | None = None held_index: int | None = None best_index = 0 for _ in range(2 * len(probed) + 2): counts = live_claim_counts(directory) if held is not None: token = _claim_token(held.host_name) counts[token] = max(0, counts.get(token, 0) - 1) scored = [ ( load + CLAIM_LOAD_PENALTY * counts.get(_claim_token(_host_name(host, access)), 0), index, ) for index, (host, access, load) in enumerate(probed) ] best_score = min(score for score, _index in scored) tied = [index for score, index in scored if score <= best_score + 1e-9] best_index = random.choice(tied) if held_index is not None and held_index in tied: break if held is not None: held.release() held = None held_index = None host, access, _load = probed[best_index] claim = acquire_claim(_host_name(host, access), directory) if claim is None: return host, access, None held, held_index = claim, best_index index = best_index if held_index is None else held_index return probed[index][0], probed[index][1], held def git_toplevel(cwd: Path) -> Path | None: """The checkout containing `cwd`, or None when `cwd` is not inside one.""" try: done = subprocess.run( ["/usr/bin/git", "-C", str(cwd), "rev-parse", "--show-toplevel"], capture_output=True, text=True, timeout=PROBE_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError): return None root = done.stdout.strip() return Path(root) if done.returncode == 0 and root else None def repo_root(cwd: Path | None = None) -> Path: cwd = Path.cwd() if cwd is None else Path(cwd) return git_toplevel(cwd) or cwd # codex's working-directory flags. Only the mirrored tree exists inside the container, so a # workstation path forwarded verbatim makes codex exit with a bare "os error 2" before it # prints anything — the value has to be translated to its container path or the run refused. WORKDIR_FLAGS = ("-C", "--cd") def find_workdir(forwarded_argv: list[str]) -> tuple[int, int, str] | None: """(index, span, value) of the effective workdir flag; last occurrence wins, as in clap.""" found = None i = 0 while i < len(forwarded_argv): arg = forwarded_argv[i] if arg == "--": break if arg in WORKDIR_FLAGS: if i + 1 < len(forwarded_argv): found = (i, 2, forwarded_argv[i + 1]) i += 2 continue for flag in WORKDIR_FLAGS: if arg.startswith(f"{flag}="): found = (i, 1, arg[len(flag) + 1 :]) i += 1 return found def rewrite_workdir(forwarded_argv: list[str], found: tuple[int, int, str], value: str) -> list[str]: index, span, _ = found replacement = [WORKDIR_FLAGS[0], value] if span == 2 else [f"--cd={value}"] return [*forwarded_argv[:index], *replacement, *forwarded_argv[index + span :]] def sandbox_id(root: Path) -> str: """Stable per-source-path identity: two worktrees never share a remote mirror. sandbox-run enforces ^[a-z0-9][a-z0-9._-]{0,63}$, so the name part is folded to that alphabet (a capital letter in the checkout dir killed the whole dispatch); identity still comes from the digest of the unmodified path.""" digest = hashlib.sha256(str(root).encode("utf-8")).hexdigest()[:12] name = re.sub(r"[^a-z0-9._-]", "-", root.name.lower()).lstrip("._-") or "repo" return f"{name[:48]}-{digest}" def remote_rel_dir(root: Path) -> str: return f"{NODE_WORKSPACES_ROOT}/{sandbox_id(root)}" def scratch_sandbox_id(root: Path) -> str: """The `--id` a no-workspace dispatch gives agent-sandbox — distinct from `sandbox_id` for the same `root` so the container's workspace (which agent-sandbox derives from `--id`, not from `rel_dir`) can never collide with — and never run inside — a real mirror's workspace.""" return f"{sandbox_id(root)}-scratch" def remote_scratch_dir(root: Path) -> str: return f"{NODE_WORKSPACES_ROOT}/{scratch_sandbox_id(root)}" def extra_rel_dir(root: Path, path: Path, namespace: str = "pi-state") -> str: """A host-local mirror location for data outside the checkout.""" digest = hashlib.sha256(f"{root}:{path.resolve()}".encode("utf-8")).hexdigest()[:16] return f"{REMOTE_ROOT}/{namespace}/{digest}" def container_workspace(root: Path) -> str: return f"{CONTAINER_WORKSPACES_ROOT}/{sandbox_id(root)}" def _exclude_args(exclude_git: bool = False) -> list[str]: names = (*SYNC_EXCLUDES, "/.git") if exclude_git else SYNC_EXCLUDES # rsync's per-directory merge rule is the closest equivalent to Git's ignore # traversal. It handles nested .gitignore files and their negations, unlike # the old root-only --exclude list. secret_rules = [ "/.env", "/.env.*", ".env", ".env.*", "*.pem", "*.key", "*.p12", "*.pfx", "id_rsa*", "id_ed25519*", "id_ecdsa*", "id_dsa*", ".secrets", "*.secret", ".aws", ".ssh", "application_default_credentials.json", "*service-account*.json", "*service_account*.json", ] return [ *(arg for name in names for arg in ("--exclude", name)), *(arg for name in secret_rules for arg in ("--exclude", name)), "--filter", ":- .gitignore", ] def secret_exclusions(root: Path) -> dict[str, str]: """Secret-shaped workspace files mapped to the rule that denied dispatch. Traversal errors abort the dispatch: an unreadable tree cannot be certified clean. The one repository-owned credential manifest is data, not a live credential. """ excluded: dict[str, str] = {} try: paths = root.rglob("*") for path in paths: rel = path.relative_to(root).as_posix() if rel == ".git" or rel.startswith(".git/") or rel in SECRET_ALLOWLIST: continue if not path.is_file() and not path.is_symlink(): continue for label, pattern in SECRET_PATH_PATTERNS: if pattern.search(rel): excluded[rel] = label break else: if path.name in AUTH_CONFIG_NAMES: content = path.read_text(encoding="utf-8", errors="replace") if AUTH_CONFIG_RE.search(content): excluded[rel] = f"authenticated {path.name}" except OSError as exc: raise OffloadUnavailable( f"could not inspect {root} for secret files: {exc}", code=EXIT_MIRROR ) from exc return excluded def log_secret_exclusions(excluded: dict[str, str]) -> None: for path, rule in sorted(excluded.items()): print(f"cdx offload: excluded secret-shaped file {path!r} (matched {rule})", file=sys.stderr) def filtered_mirror_files(root: Path, selected: bytes) -> bytes: excluded = secret_exclusions(root) log_secret_exclusions(excluded) if not excluded: return selected paths = selected.rstrip(b"\0").split(b"\0") if selected else [] kept = [path for path in paths if path.decode("utf-8", "surrogateescape") not in excluded] return b"\0".join(kept) + (b"\0" if kept else b"") def git_mirror_files(root: Path, run=subprocess.run) -> bytes: """Tracked files, NUL-delimited for rsync. The regular rsync pass carries untracked, non-ignored files. This pass exists only to restore tracked paths excluded by nested ignore rules. """ try: done = run( ["/usr/bin/git", "-C", str(root), "ls-files", "--cached", "-z"], capture_output=True, timeout=PROBE_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError) as exc: raise OffloadUnavailable( f"could not determine mirror files for {root}: {exc}", code=EXIT_MIRROR ) from exc if done.returncode != 0: detail = str(done.stderr or "").strip() raise OffloadUnavailable( f"could not determine mirror files for {root}: {detail}", code=EXIT_MIRROR ) selected = done.stdout.encode() if isinstance(done.stdout, str) else done.stdout return filtered_mirror_files(root, selected) def tracked_sync_argv(source: str, destination: str) -> list[str]: """Sync Git-selected paths only; stdin is the NUL-delimited Git file list.""" return [ "rsync", "-az", "--from0", "--files-from=-", "--delete-missing-args", source, destination, ] def push_argv(access: dict, root: Path, rel_dir: str, exclude_git: bool = False) -> list[str]: excluded = secret_exclusions(root) log_secret_exclusions(excluded) return [ "rsync", "-az", "--delete", *_exclude_args(exclude_git), *(arg for path in excluded for arg in ("--exclude", f"/{path}")), "-e", rsync_shell(access), f"{root}/", f"{access.get('user', 'user')}@{access['host']}:{rel_dir}/", ] def pull_argv(access: dict, root: Path, rel_dir: str, exclude_git: bool = False) -> list[str]: excluded = secret_exclusions(root) log_secret_exclusions(excluded) return [ "rsync", "-az", "--delete", *_exclude_args(exclude_git), *(arg for path in excluded for arg in ("--exclude", f"/{path}")), "-e", rsync_shell(access), f"{access.get('user', 'user')}@{access['host']}:{rel_dir}/", f"{root}/", ] def push_tracked_argv(access: dict, root: Path, rel_dir: str) -> list[str]: args = tracked_sync_argv( f"{root}/", f"{access.get('user', 'user')}@{access['host']}:{rel_dir}/" ) return [*args[:-2], "-e", rsync_shell(access), *args[-2:]] def pull_tracked_argv(access: dict, root: Path, rel_dir: str) -> list[str]: args = tracked_sync_argv( f"{access.get('user', 'user')}@{access['host']}:{rel_dir}/", f"{root}/" ) return [*args[:-2], "-e", rsync_shell(access), *args[-2:]] def sync_extra_dir(access: dict, local_dir: Path, rel_dir: str, *, pull: bool, run=subprocess.run) -> str | None: """Round-trip an out-of-tree directory with the normal mirror transport.""" argv = pull_argv(access, local_dir, rel_dir) if pull else push_argv(access, local_dir, rel_dir) try: done = run(argv, capture_output=True, text=True, timeout=SYNC_TIMEOUT_SEC) except (OSError, subprocess.SubprocessError) as exc: return f"extra mirror {'pull' if pull else 'push'} failed: {exc}" if done.returncode: return f"extra mirror {'pull' if pull else 'push'} failed: {done.stderr.strip()}" return None @dataclass(frozen=True) class GitMirror: """Where a linked worktree's git plumbing lives locally and on the node.""" git_dir: Path common_dir: Path remote_common: str remote_git_dir: str def git_dirs(root: Path, run=subprocess.run) -> tuple[Path, Path] | None: """(gitdir, common dir) for `root`, both absolute. None when `root` is not a checkout.""" try: done = run( ["/usr/bin/git", "-C", str(root), "rev-parse", "--absolute-git-dir", "--git-common-dir"], capture_output=True, text=True, timeout=PROBE_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError): return None if done.returncode != 0: return None lines = done.stdout.splitlines() if len(lines) != 2 or not lines[0].strip() or not lines[1].strip(): return None git_dir = Path(lines[0].strip()) common = Path(lines[1].strip()) if not common.is_absolute(): common = Path(root) / common return git_dir.resolve(), common.resolve() def remote_git_common(common_dir: Path) -> str: """One mirror per repository: every worktree of a repo shares these objects and refs.""" digest = hashlib.sha256(str(common_dir).encode("utf-8")).hexdigest()[:12] label = common_dir.parent.name or common_dir.name return f"{REMOTE_GIT_COMMON_ROOT}/{label}-{digest}" def git_mirror(root: Path, run=subprocess.run) -> GitMirror | None: """None for a plain checkout, whose `.git` directory rides along in the tree rsync.""" dirs = git_dirs(root, run=run) if dirs is None: return None git_dir, common_dir = dirs if git_dir == common_dir: return None remote_common = remote_git_common(common_dir) return GitMirror( git_dir=git_dir, common_dir=common_dir, remote_common=remote_common, remote_git_dir=f"{remote_common}/worktrees/{git_dir.name}", ) def git_common_push_argv(access: dict, mirror: GitMirror) -> list[str]: """No --delete: a concurrent dispatch's remote commit must not lose its objects.""" return [ "rsync", "-az", "--exclude", "/worktrees", "-e", rsync_shell(access), f"{mirror.common_dir}/", f"{access.get('user', 'user')}@{access['host']}:{mirror.remote_common}/", ] def git_dir_push_argv(access: dict, mirror: GitMirror) -> list[str]: return [ "rsync", "-az", "--delete", "-e", rsync_shell(access), f"{mirror.git_dir}/", f"{access.get('user', 'user')}@{access['host']}:{mirror.remote_git_dir}/", ] def git_pointer_command(rel_dir: str, container_dir: str, mirror: GitMirror) -> str: """$HOME is resolved on the node: the rsync targets are relative to the remote home. The gitdir the worktree points at is mounted into the container at its node-absolute path, so that pointer is valid on both sides; the worktree itself is not, so its back-pointer names the container path, where git actually runs.""" inner = ( "set -eu\n" f'd="$HOME"/{shlex.quote(rel_dir)}\n' f'g="$HOME"/{shlex.quote(mirror.remote_git_dir)}\n' 'printf "gitdir: %s\\n" "$g" > "$d/.git"\n' f'printf "%s\\n" {shlex.quote(container_dir + "/.git")} > "$g/gitdir"\n' 'printf "%s\\n" "../.." > "$g/commondir"\n' 'git -C "$d" rev-parse --show-toplevel > /dev/null\n' ) return f"bash -lc {shlex.quote(inner)}" def prepare_remote_git( access: dict, host_name: str, rel_dir: str, container_dir: str, mirror: GitMirror, run=subprocess.run, ) -> str | None: """Make git work in the node's mirror. Returns a warning on failure, and never raises: a run with no git is still useful, a run refused over git is not.""" mkdir_command = f"mkdir -p {shlex.join([mirror.remote_common, mirror.remote_git_dir])}" steps: list[tuple[str, list[str], int]] = [ ("mirror directories", [*ssh_argv(access), mkdir_command], PROBE_TIMEOUT_SEC), ("objects and refs", git_common_push_argv(access, mirror), SYNC_TIMEOUT_SEC), ("worktree gitdir", git_dir_push_argv(access, mirror), SYNC_TIMEOUT_SEC), ( "gitdir pointers", [*ssh_argv(access), git_pointer_command(rel_dir, container_dir, mirror)], PROBE_TIMEOUT_SEC, ), ] for what, argv, timeout in steps: try: done = run(argv, capture_output=True, text=True, timeout=timeout) except (OSError, subprocess.SubprocessError) as exc: return git_warning_text(host_name, f"mirroring {what} failed: {exc}") if done.returncode != 0: return git_warning_text(host_name, f"mirroring {what} failed: {done.stderr.strip()}") return None def git_warning_text(host_name: str, detail: str) -> str: return ( f"!!! GIT IS NOT USABLE IN THE MIRROR ON {host_name}: {detail}. " "The run continues, but git status/add/commit/log and anything shelling out to git " "WILL FAIL on the node — do not expect to commit there." ) REMOTE_GIT_BARE_ROOT = f"{REMOTE_ROOT}/repos" PUSH_REF_PREFIX = "refs/cdx" def git_common_dir(root: Path, run=subprocess.run) -> Path | None: """Absolute git common dir for `root`, or None when `root` is not a checkout.""" try: done = run( ["/usr/bin/git", "-C", str(root), "rev-parse", "--git-common-dir"], capture_output=True, text=True, timeout=PROBE_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError): return None if done.returncode != 0: return None dir_str = done.stdout.strip() if not dir_str: return None common = Path(dir_str) if not common.is_absolute(): common = Path(root) / common return common.resolve() def repo_key(common_dir: Path) -> str: """One bare mirror per repository: every worktree of a repo shares it.""" name = common_dir.parent.name if common_dir.name == ".git" else common_dir.name digest = hashlib.sha256(str(common_dir).encode("utf-8")).hexdigest()[:12] return f"{name}-{digest}" def remote_bare_dir(key: str) -> str: return f"{REMOTE_GIT_BARE_ROOT}/{key}.git" def bare_ssh_url(access: dict, bare_abs: str) -> str: port = access.get("port") suffix = f":{port}" if port else "" return f"ssh://{access.get('user', 'user')}@{access['host']}{suffix}{bare_abs}" def git_ssh_env(access: dict) -> dict[str, str]: """`git push`/`fetch` need their own transport string, not the mirror's `-e` rsync shell.""" return {**os.environ, "GIT_SSH_COMMAND": rsync_shell(access)} def bare_mirror_ensure_script(bare_rel: str) -> str: inner = ( "set -eu\n" f'mkdir -p "$HOME"/{shlex.quote(REMOTE_GIT_BARE_ROOT)}\n' f'exec 9>"$HOME"/{shlex.quote(f"{REMOTE_GIT_BARE_ROOT}/.lock")}\n' "flock -w 600 9\n" f'b="$HOME"/{shlex.quote(bare_rel)}\n' '[ -d "$b/objects" ] || git init -q --bare "$b"\n' 'git -C "$b" config receive.shallowUpdate true\n' ) return f"bash -lc {shlex.quote(inner)}" def new_dispatch_id() -> str: """Unique per invocation, collision-impossible across concurrent sessions on any buildbox, and self-describing an age for `gc_dispatch_refs` without fetching objects.""" return f"{int(time.time())}-{uuid.uuid4().hex}" def dispatch_ref(sbox_id: str, dispatch_id: str) -> str: """One inbox ref per dispatch, not per sandbox: a stale or racing tip from another session can never block this push, since nothing else ever targets this exact ref.""" return f"{PUSH_REF_PREFIX}/{sbox_id}/{dispatch_id}/in" def result_ref(sbox_id: str) -> str: return f"{PUSH_REF_PREFIX}/{sbox_id}/out" DISPATCH_REF_TTL_SEC = 7 * 24 * 3600 DISPATCH_REF_GC_MAX_DELETE = 50 _DISPATCH_ID_RE = re.compile(r"^(\d+)-[0-9a-f]{32}$") def _dispatch_ref_age_sec(ref: str, now: float) -> float | None: """Age of a `dispatch_ref`'s embedded epoch, read from the ref name alone — no fetch. None for anything not shaped like `.../-/in`: an unrecognized ref (a future naming change, a hand-pushed one) is left alone rather than guessed at.""" parts = ref.rsplit("/", 2) if len(parts) != 3 or parts[-1] != "in": return None match = _DISPATCH_ID_RE.match(parts[-2]) if not match: return None return now - int(match.group(1)) def gc_dispatch_refs( root: Path, url: str, sbox_id: str, keep_ref: str, env: dict[str, str], host_name: str = "the node", run=subprocess.run, ) -> str | None: """Best-effort cleanup of this sandbox's dispatch refs older than `DISPATCH_REF_TTL_SEC`, run once after a successful push. Never raises and never fails the dispatch: a listing or delete failure just means the next dispatch's GC gets another chance. Returns a plain, owner-safe warning string (no internal names, no URLs) to surface non-fatally on failure, None when there was nothing to report.""" try: listed = run( ["/usr/bin/git", "-C", str(root), "ls-remote", url, f"{PUSH_REF_PREFIX}/{sbox_id}/*/in"], env=env, capture_output=True, text=True, timeout=PROBE_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError): return f"old dispatch records could not be cleaned up on {host_name}; the run is unaffected." if listed.returncode != 0: return f"old dispatch records could not be cleaned up on {host_name}; the run is unaffected." now = time.time() stale = [] for line in listed.stdout.splitlines(): fields = line.split(maxsplit=1) if len(fields) != 2: continue ref = fields[1].strip() if ref == keep_ref: continue age = _dispatch_ref_age_sec(ref, now) if age is not None and age >= DISPATCH_REF_TTL_SEC: stale.append(ref) if not stale: return None stale = stale[:DISPATCH_REF_GC_MAX_DELETE] try: deleted = run( ["/usr/bin/git", "-C", str(root), "push", "-q", "--no-verify", url, *(f":{ref}" for ref in stale)], env=env, capture_output=True, text=True, timeout=SYNC_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError): return f"old dispatch records could not be cleaned up on {host_name}; the run is unaffected." if deleted.returncode != 0: return f"old dispatch records could not be cleaned up on {host_name}; the run is unaffected." return None def _run_git(cmd: list[str], step: str, root: Path, run, **kwargs) -> subprocess.CompletedProcess: """Every local git probe/snapshot call goes through this: an `OSError` or timeout is a transport failure, converted to the same `OffloadUnavailable` the rest of the module raises on one — never a raw exception leaking out of `open_session`/`pull_back`.""" try: return run(cmd, capture_output=True, text=True, **kwargs) except (OSError, subprocess.SubprocessError) as exc: raise OffloadUnavailable(f"{step} on {root} failed: {exc}", code=EXIT_MIRROR) from exc def tree_of(root: Path, commit_ish: str, run=subprocess.run) -> str | None: done = _run_git( ["/usr/bin/git", "-C", str(root), "rev-parse", f"{commit_ish}^{{tree}}"], "rev-parse", root, run, timeout=PROBE_TIMEOUT_SEC, ) return done.stdout.strip() if done.returncode == 0 else None def _snapshot_tree(root: Path, run=subprocess.run) -> tuple[str, bool]: """(tree sha, is_dirty) for `root`'s current state, never touching HEAD or the real index: a throwaway `GIT_INDEX_FILE` does the read-tree/add/write-tree dance.""" dirty = _run_git( ["/usr/bin/git", "-C", str(root), "status", "--porcelain"], "status", root, run, timeout=PROBE_TIMEOUT_SEC, ) if dirty.returncode != 0: raise OffloadUnavailable( f"could not read the working tree status of {root}: {dirty.stderr.strip()}", code=EXIT_MIRROR, ) excluded = secret_exclusions(root) log_secret_exclusions(excluded) if not dirty.stdout.strip() and not excluded: head_tree = tree_of(root, "HEAD", run=run) return head_tree or "", False # Not `root / ".git" / ...`: for a linked worktree `.git` is a file, not a directory. index_path = Path(tempfile.gettempdir()) / f"cdx-offload-index-{os.getpid()}-{uuid.uuid4().hex}" env = {**os.environ, "GIT_INDEX_FILE": str(index_path)} try: for step, cmd in ( ("read-tree", ["/usr/bin/git", "-C", str(root), "read-tree", "HEAD"]), ("add", ["/usr/bin/git", "-C", str(root), "add", "-A", "--ignore-errors"]), ): done = _run_git(cmd, step, root, run, env=env, timeout=SYNC_TIMEOUT_SEC) if done.returncode != 0: raise OffloadUnavailable( f"snapshotting {root} failed at {step}: {done.stderr.strip()}", code=EXIT_MIRROR, ) if excluded: removed = _run_git( ["/usr/bin/git", "-C", str(root), "rm", "-q", "--cached", "--ignore-unmatch", "--", *excluded], "secret filter", root, run, env=env, timeout=SYNC_TIMEOUT_SEC, ) if removed.returncode != 0: raise OffloadUnavailable( f"secret filtering {root} failed: {removed.stderr.strip()}", code=EXIT_MIRROR ) tree = _run_git( ["/usr/bin/git", "-C", str(root), "write-tree"], "write-tree", root, run, env=env, timeout=PROBE_TIMEOUT_SEC, ) if tree.returncode != 0: raise OffloadUnavailable( f"snapshotting {root} failed at write-tree: {tree.stderr.strip()}", code=EXIT_MIRROR, ) return tree.stdout.strip(), True finally: try: index_path.unlink() except OSError: pass def snapshot_commit(root: Path, run=subprocess.run) -> str | None: """The commit fully representing `root`'s current state — HEAD plus every tracked modification and untracked, non-ignored file — as one object `git push` can send as a delta. None when `root` has no HEAD yet (a freshly `git init`ed checkout): the caller falls back to the tree-copy transport, since there is nothing to push against.""" head = _run_git( ["/usr/bin/git", "-C", str(root), "rev-parse", "--verify", "HEAD"], "rev-parse", root, run, timeout=PROBE_TIMEOUT_SEC, ) if head.returncode != 0: return None head_sha = head.stdout.strip() tree, dirty = _snapshot_tree(root, run=run) if not dirty: return head_sha id_env = { **os.environ, "GIT_AUTHOR_NAME": "cdx-offload", "GIT_AUTHOR_EMAIL": "cdx-offload@local", "GIT_COMMITTER_NAME": "cdx-offload", "GIT_COMMITTER_EMAIL": "cdx-offload@local", } commit = _run_git( ["/usr/bin/git", "-C", str(root), "commit-tree", tree, "-p", head_sha, "-m", "cdx-offload snapshot"], "commit-tree", root, run, env=id_env, timeout=PROBE_TIMEOUT_SEC, ) if commit.returncode != 0: raise OffloadUnavailable( f"snapshotting {root} failed at commit-tree: {commit.stderr.strip()}", code=EXIT_MIRROR, ) return commit.stdout.strip() def push_commit_argv(root: Path, url: str, sha: str, ref: str) -> list[str]: return ["/usr/bin/git", "-C", str(root), "push", "-q", "--no-verify", url, f"{sha}:{ref}"] def remote_checkout_script(rel_dir: str, bare_abs: str, ref: str, sha: str) -> str: """Materialize `sha` as a real, working checkout at `rel_dir`, sourcing objects from the bare mirror by alternates — the fetch itself moves only the one ref, never a tree copy. `clean` is scoped to `SYNC_EXCLUDES` so a reused workspace's installed deps survive the checkout; a bare `clean -ffd` would force every dispatch into a cold install. The lock serializes against a concurrent result-snapshot (`remote_result_script`) on the same workspace.""" excludes = " ".join(f"-e {shlex.quote(name)}" for name in SYNC_EXCLUDES) inner = ( "set -eu\n" f'w="$HOME"/{shlex.quote(rel_dir)}\n' f'b={shlex.quote(bare_abs)}\n' # The lock lives inside .git, never in the worktree root: a lock file there is an # untracked path `git add -A` in remote_result_script would pick up and round-trip # home as spurious untracked state in the local checkout. 'mkdir -p "$w/.git"\n' 'exec 9>"$w/.git/cdx-offload.lock"\n' "flock -w 600 9\n" '[ -e "$w/.git/HEAD" ] || git init -q "$w"\n' 'git -C "$w" remote add origin "$b" 2>/dev/null || git -C "$w" remote set-url origin "$b"\n' 'mkdir -p "$w/.git/objects/info"\n' 'printf "%s\\n" "$b/objects" > "$w/.git/objects/info/alternates"\n' f'git -C "$w" fetch -q origin {shlex.quote(f"+{ref}:refs/cdx/in")}\n' f'git -C "$w" checkout -qf {shlex.quote(sha)}\n' f'git -C "$w" clean -qffd {excludes}\n' ) return f"bash -lc {shlex.quote(inner)}" def remote_result_script(rel_dir: str, bare_abs: str, ref: str) -> str: """Snapshot the node's post-run state the same way `snapshot_commit` does locally, push it to the bare mirror under `ref`, and print its sha — the delta the workstation fetches back. Shares `remote_checkout_script`'s lock so the two never race on one workspace.""" inner = ( "set -eu\n" f'w="$HOME"/{shlex.quote(rel_dir)}\n' f'b={shlex.quote(bare_abs)}\n' 'exec 9>"$w/.git/cdx-offload.lock"\n' "flock -w 600 9\n" 'cd "$w"\n' 'idx="$w/.git/cdx-offload-result-index"\n' 'export GIT_INDEX_FILE="$idx"\n' 'git read-tree HEAD\n' 'git add -A --ignore-errors\n' 'tree=$(git write-tree)\n' 'rm -f "$idx"\n' 'base=$(git rev-parse HEAD)\n' 'if [ "$tree" = "$(git rev-parse HEAD^{tree})" ]; then\n' ' sha="$base"\n' "else\n" ' sha=$(GIT_AUTHOR_NAME=cdx-offload GIT_AUTHOR_EMAIL=cdx-offload@local ' 'GIT_COMMITTER_NAME=cdx-offload GIT_COMMITTER_EMAIL=cdx-offload@local ' 'git commit-tree "$tree" -p "$base" -m "cdx-offload result")\n' "fi\n" # Forced: this ref is a per-sandbox scratch pointer to the latest result, never a # history. A reused workspace whose base moved makes the next push a non-fast-forward, # and a rejected push strands the entire run's work on the node. f'git push -q --force "$b" "$sha:{ref}"\n' 'printf "%s\\n" "$sha"\n' ) return f"bash -lc {shlex.quote(inner)}" def launcher_argv( host_name: str, credential: Credential, exec_name: str, forwarded_argv: list[str], sandbox_id: str, git_common: str | None = None, git_dir: str | None = None, egress_hosts: tuple[str, ...] = (), ) -> list[str]: """`--git-common`/`--git-dir` each independently bind-mount their node-absolute path at the same path inside the container (agent-sandbox's own contract). A linked worktree's dispatch needs both; the git-push transport's bare mirror needs only `--git-common`.""" argv = [ str(AGENT_SANDBOX_BIN), "--host", host_name, "--id", sandbox_id, "--runtime", credential.runtime, ] if git_common: argv.extend(["--git-common", git_common]) if git_dir: argv.extend(["--git-dir", git_dir]) for allowed_host in egress_hosts: argv.extend(["--egress-host", allowed_host]) return [*argv, "--", exec_name, *forwarded_argv] @dataclass class Session: host_name: str access: dict root: Path rel_dir: str argv: list[str] node_home: str = "" workspace: str = "" mounts: tuple[str, ...] = () credential: Credential | None = None git_warning: str | None = None exclude_git: bool = False mirror_files: bytes = b"" claim: PlacementClaim | None = None sandbox_id: str | None = None no_workspace: bool = False git_push: bool = False bare_abs: str | None = None bare_url: str | None = None pre_sha: str | None = None def release(self) -> None: """Drop the placement claim so the node stops looking busy to the next dispatch.""" if self.claim is not None: self.claim.release() self.claim = None def pull_back(self, run=subprocess.run) -> str | None: """Bring the node's post-run state home. Returns an error string, never raises.""" if self.no_workspace: return None if self.git_push: return self._pull_back_git(run=run) try: tracked = run( pull_tracked_argv(self.access, self.root, self.rel_dir), input=self.mirror_files, capture_output=True, timeout=SYNC_TIMEOUT_SEC, ) if tracked.returncode != 0: return f"result sync from {self.host_name} failed: {tracked.stderr.strip()}" done = run( pull_argv(self.access, self.root, self.rel_dir, exclude_git=self.exclude_git), capture_output=True, text=True, timeout=SYNC_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError) as exc: return f"result sync from {self.host_name} failed: {exc}" if done.returncode != 0: return f"result sync from {self.host_name} failed: {done.stderr.strip()}" return None def _pull_back_git(self, run=subprocess.run) -> str | None: """Snapshot the node, push the result to the bare mirror, fetch it home (delta only), and apply it as an uncommitted change against the local `pre_sha`. Refuses — never overwrites — if the local checkout moved since the push; the result stays safe on the fetched ref either way.""" assert self.bare_abs is not None and self.bare_url is not None and self.pre_sha is not None ref = result_ref(self.sandbox_id or "") script = remote_result_script(self.rel_dir, self.bare_abs, ref) try: got = run( [*ssh_argv(self.access), script], capture_output=True, text=True, timeout=SYNC_TIMEOUT_SEC ) except (OSError, subprocess.SubprocessError) as exc: return f"result snapshot on {self.host_name} failed: {exc}" if got.returncode != 0: return f"result snapshot on {self.host_name} failed: {got.stderr.strip()}" node_lines = [line.strip() for line in got.stdout.splitlines() if line.strip()] if not node_lines: return f"result snapshot on {self.host_name} produced no commit" node_sha = node_lines[-1] return pull_back_git_delta( root=self.root, access=self.access, host_name=self.host_name, bare_url=self.bare_url, pre_sha=self.pre_sha, sandbox=self.sandbox_id or "", run=run, ) def pull_back_git_delta( *, root: Path, access: dict, host_name: str, bare_url: str, pre_sha: str, sandbox: str, run=subprocess.run, ) -> str | None: """Fetch a dispatched result and apply its delta with Session's refusal semantics.""" ref = result_ref(sandbox) fetch_ref = f"refs/cdx-fetch/{sandbox}" # A k3s dispatch has no per-box ssh door — the worker pushed the result itself and # empty access means "fetch with git's own transport", never a KeyError. try: fetched = run( ["/usr/bin/git", "-C", str(root), "fetch", "-q", bare_url, f"+{ref}:{fetch_ref}"], env=git_ssh_env(access) if access else None, capture_output=True, text=True, timeout=SYNC_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError) as exc: return f"result fetch from {host_name} failed: {exc}" if fetched.returncode != 0: return f"result fetch from {host_name} failed: {fetched.stderr.strip()}" try: current_tree, _ = _snapshot_tree(root, run=run) expected_tree = tree_of(root, pre_sha, run=run) except OffloadUnavailable as exc: return f"checking the local checkout of {root} failed: {exc}" if current_tree != expected_tree: return ( f"local checkout {root} changed during the {host_name} dispatch; " f"not overwriting it — the result is safe at {fetch_ref} in this repository" ) try: delta = run( ["/usr/bin/git", "-C", str(root), "diff", "--binary", pre_sha, fetch_ref], capture_output=True, timeout=SYNC_TIMEOUT_SEC, ) if delta.returncode != 0: return f"preparing the {host_name} result failed: {delta.stderr.strip()}" applied = run( ["/usr/bin/git", "-C", str(root), "apply", "--binary", "-"], input=delta.stdout, capture_output=True, timeout=SYNC_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError) as exc: return f"applying the {host_name} result failed: {exc}" if applied.returncode != 0: return f"applying the {host_name} result failed: {applied.stderr.strip()}" return None def _open_podman_session( exec_name: str, forwarded_argv: list[str], cwd: Path | None = None, registry_path: Path = REGISTRY_PATH, credential: Credential | None = None, run=subprocess.run, probe=probe_load, claims: Path | None = None, ) -> Session: """The incumbent podman dispatcher. Keep this body stable: flag-off is its contract.""" return _open_session_impl(exec_name, forwarded_argv, cwd, registry_path, credential, run, probe, claims) def open_session( exec_name: str, forwarded_argv: list[str], cwd: Path | None = None, registry_path: Path = REGISTRY_PATH, credential: Credential | None = None, run=subprocess.run, probe=probe_load, claims: Path | None = None, ) -> Session: if os.environ.get(K3S_DISPATCH_ENV) == "1": import k3s_dispatch return k3s_dispatch.open_session( exec_name=exec_name, forwarded_argv=forwarded_argv, cwd=cwd, registry_path=registry_path, credential=credential, run=run, fallback=lambda: _open_podman_session( exec_name, forwarded_argv, cwd, registry_path, credential, run, probe, claims ), ) return _open_podman_session(exec_name, forwarded_argv, cwd, registry_path, credential, run, probe, claims) def _open_session_impl( exec_name: str, forwarded_argv: list[str], cwd: Path | None = None, registry_path: Path = REGISTRY_PATH, credential: Credential | None = None, run=subprocess.run, probe=probe_load, claims: Path | None = None, ) -> Session: if credential is None: raise OffloadUnavailable( f"no credential resolved for {exec_name}", code=EXIT_CREDENTIAL ) if not credential.path.is_file() or not os.access(credential.path, os.R_OK): raise OffloadUnavailable( f"{credential.runtime} credential for {credential.slug or ''} " f"is not a readable file: {credential.path}", code=EXIT_CREDENTIAL, ) no_workspace = no_workspace_requested() workdir = find_workdir(forwarded_argv) if workdir is not None and no_workspace: raise OffloadUnavailable( f"{WORKDIR_FLAGS[0]} {workdir[2]} is not supported with {NO_WORKSPACE_ENV}=1: " "there is no mirrored checkout to point at", code=EXIT_MIRROR, ) if workdir is not None: base = Path.cwd() if cwd is None else Path(cwd) requested = Path(workdir[2]).expanduser() requested = (requested if requested.is_absolute() else base / requested).resolve() if not requested.is_dir(): raise OffloadUnavailable( f"{WORKDIR_FLAGS[0]} {workdir[2]} is not a directory", code=EXIT_MIRROR ) if git_toplevel(requested) is None: raise OffloadUnavailable( f"{WORKDIR_FLAGS[0]} {workdir[2]} is not inside a git checkout, so there is " "no bounded tree to mirror", code=EXIT_MIRROR, ) cwd = requested registry = load_registry(registry_path) host, access, claim = select_host(candidate_hosts(registry), probe=probe, claims=claims) mirror_hosts = tuple(sorted({ str(door["host"]) for item in registry.get("hosts", []) if isinstance((door := (item.get("access") or {}).get(DOOR)), dict) and door.get("host") })) name = _host_name(host, access) try: root = repo_root(cwd) commondir: Path | None = None pre_sha: str | None = None bare_abs: str | None = None bare_url: str | None = None git_push = False if no_workspace: mirror_files = b"" rel_dir = remote_scratch_dir(root) mirror = None remote_dirs = [rel_dir] else: if workdir is not None: try: inside = requested.relative_to(root) except ValueError: raise OffloadUnavailable( f"{WORKDIR_FLAGS[0]} {workdir[2]} resolves outside the mirrored checkout " f"{root}", code=EXIT_MIRROR, ) from None container = container_workspace(root) if inside != Path("."): container = f"{container}/{inside}" forwarded_argv = rewrite_workdir(forwarded_argv, workdir, container) rel_dir = remote_rel_dir(root) commondir = git_common_dir(root, run=run) pre_sha = None if commondir is None else snapshot_commit(root, run=run) git_push = commondir is not None and pre_sha is not None if git_push: mirror_files = b"" mirror = None remote_dirs = [rel_dir, REMOTE_GIT_BARE_ROOT] else: # Not a git checkout, or one with no HEAD yet (freshly `git init`ed): there # is nothing to push against, so this falls back to the tree-copy transport. mirror_files = git_mirror_files(root, run=run) mirror = git_mirror(root, run=run) remote_dirs = [rel_dir] if mirror is not None: remote_dirs += [mirror.remote_common, mirror.remote_git_dir] try: made = run( [ *ssh_argv(access), f"mkdir -p {shlex.join(remote_dirs)} && printf '%s\\n' \"$HOME\"", ], capture_output=True, text=True, timeout=PROBE_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError) as exc: raise OffloadUnavailable( f"{name}: could not prepare mirror: {exc}", code=EXIT_MIRROR ) from exc if made.returncode != 0: raise OffloadUnavailable( f"{name}: could not prepare mirror: {made.stderr.strip()}", code=EXIT_MIRROR, ) lines = [line.strip() for line in made.stdout.splitlines() if line.strip()] if not lines: raise OffloadUnavailable( f"{name}: node home is missing from prepare-mirror response", code=EXIT_MIRROR, ) node_home = lines[-1].rstrip("/") if not node_home.startswith("/"): raise OffloadUnavailable( f"{name}: node home is not absolute: {lines[-1]}", code=EXIT_MIRROR, ) workspace = f"{node_home}/{rel_dir}" mounts = ( ( f"{node_home}/{mirror.remote_common}", f"{node_home}/{mirror.remote_git_dir}", ) if mirror is not None else () ) if no_workspace: git_warning = None elif git_push: sbox = sandbox_id(root) dispatch_id = new_dispatch_id() key = repo_key(commondir) bare_rel = remote_bare_dir(key) bare_abs = f"{node_home}/{bare_rel}" bare_url = bare_ssh_url(access, bare_abs) # The container only gets $SANDBOX_ROOT:/sandbox — the bare mirror sits outside # that mount, so it needs the same --git-common bind agent-sandbox already # offers a linked worktree's gitdir, or the checkout's alternates point nowhere # inside the container and git is dead there. mounts = (bare_abs,) try: ensure = run( [*ssh_argv(access), bare_mirror_ensure_script(bare_rel)], capture_output=True, text=True, timeout=PROBE_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError) as exc: raise OffloadUnavailable( f"{name}: could not prepare the bare mirror: {exc}", code=EXIT_MIRROR ) from exc if ensure.returncode != 0: raise OffloadUnavailable( f"{name}: could not prepare the bare mirror: {ensure.stderr.strip()}", code=EXIT_MIRROR, ) in_ref = dispatch_ref(sbox, dispatch_id) try: pushed = run( push_commit_argv(root, bare_url, pre_sha, in_ref), env=git_ssh_env(access), capture_output=True, text=True, timeout=SYNC_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError) as exc: raise OffloadUnavailable(f"{name}: git push failed: {exc}", code=EXIT_MIRROR) from exc if pushed.returncode != 0: raise OffloadUnavailable( f"{name}: git push failed: {pushed.stderr.strip()}", code=EXIT_MIRROR ) git_warning = gc_dispatch_refs( root, bare_url, sbox, in_ref, git_ssh_env(access), host_name=name, run=run ) try: checkout = run( [ *ssh_argv(access), remote_checkout_script(rel_dir, bare_abs, in_ref, pre_sha), ], capture_output=True, text=True, timeout=SYNC_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError) as exc: raise OffloadUnavailable( f"{name}: remote checkout failed: {exc}", code=EXIT_MIRROR ) from exc if checkout.returncode != 0: raise OffloadUnavailable( f"{name}: remote checkout failed: {checkout.stderr.strip()}", code=EXIT_MIRROR ) else: try: tracked = run( push_tracked_argv(access, root, rel_dir), input=mirror_files, capture_output=True, timeout=SYNC_TIMEOUT_SEC, ) if tracked.returncode != 0: raise OffloadUnavailable( f"{name}: mirror push failed: {tracked.stderr.strip()}", code=EXIT_MIRROR ) pushed = run( push_argv(access, root, rel_dir, exclude_git=mirror is not None), capture_output=True, text=True, timeout=SYNC_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError) as exc: raise OffloadUnavailable( f"{name}: mirror push failed: {exc}", code=EXIT_MIRROR ) from exc if pushed.returncode != 0: raise OffloadUnavailable( f"{name}: mirror push failed: {pushed.stderr.strip()}", code=EXIT_MIRROR, ) git_warning = ( None if mirror is None else prepare_remote_git( access, name, rel_dir, container_workspace(root), mirror, run=run ) ) except BaseException: if claim is not None: claim.release() raise effective_sandbox_id = scratch_sandbox_id(root) if no_workspace else sandbox_id(root) git_common_mount = mounts[0] if len(mounts) >= 1 else None git_dir_mount = mounts[1] if len(mounts) >= 2 else None return Session( host_name=name, access=access, root=root, rel_dir=rel_dir, sandbox_id=effective_sandbox_id, node_home=node_home, workspace=workspace, mounts=mounts, credential=credential, argv=launcher_argv( host_name=name, credential=credential, exec_name=exec_name, forwarded_argv=forwarded_argv, sandbox_id=effective_sandbox_id, git_common=git_common_mount, git_dir=git_dir_mount, egress_hosts=mirror_hosts, ), git_warning=git_warning, exclude_git=mirror is not None, mirror_files=mirror_files, claim=claim, no_workspace=no_workspace, git_push=git_push, bare_abs=bare_abs, bare_url=bare_url, pre_sha=pre_sha, )