diff --git a/plugins/providers/cape/README.md b/plugins/providers/cape/README.md index 46dceb9..68dbed8 100644 --- a/plugins/providers/cape/README.md +++ b/plugins/providers/cape/README.md @@ -10,55 +10,107 @@ nodes with strong per-request isolation. Requests sharing a CACHED_IDLE — a *serial* model, one command at a time per session). This backend maps one Agentix sandbox onto exactly one long-lived CAPE request inside a per-sandbox session: the request bind-mounts the -bundle's `/nix` tree read-only, prints an `AGENTIX_ENDPOINT -` marker line to stdout, and execs `/nix/runtime/bootstrap.sh`. -The provider discovers the endpoint by polling `cape status` and -`cape logs` for that marker (it never submits a second command into the -session, so it does not depend on same-session concurrency or on -cross-request workspace persistence), then health-checks `GET /health` -on the discovered endpoint. - -## Contract status — ASSUMED CLI, not verified - -**The CAPE CLI verb surface implemented here -(`cape run/status/logs/cancel`, their flags, the request-id stdout -shape, and the `cape status` JSON schema) is an *assumed* contract.** -It was reverse-documented from a sibling project's adapter of the -same CLI, whose authors record that the official CAPE CLI was never -obtained: the contract has only ever been exercised -against fakes and local emulators. Treat every invocation in -`agentix/provider/cape.py` (`_CapeCli` is the single class that knows -the CLI) as a hypothesis to confirm. - -Verification checklist for when the real CLI arrives: - -- [ ] Run `cape --help` (and per-verb `--help`) and diff the verb set - against `run` / `status` / `logs` / `cancel`. -- [ ] Compare the `cape run` argv table (`--controller-url`, `--token`, - `--pool`, `--user`, `--workspace`, `--image`, `--gpus`, - `--cpu-cores`, `--memory-gb`, `--gpu-mode`, `--isolation-policy`, - `--runtime-adapter`, `--max-duration-seconds`, `--cwd`, - `--session-key`, `--bind`, `--env`, `-- `) flag by - flag, including the "stdout is exactly one `req-...` line" parse - rule. -- [ ] Compare the `cape status` JSON schema: the `state` field, the - terminal-state set (COMPLETED / FAILED / CANCELLED / EXPIRED / - LOST / INFEASIBLE), and the `request_id` echo this provider - cross-checks. -- [ ] Confirm `cape logs` can return the stdout of a still-RUNNING - request — endpoint discovery depends on it. This assumption is - *new in this provider* (the reference adapter only called logs on - terminal requests). -- [ ] Confirm the session model is serial (RUNNING_COMMAND ↔ - CACHED_IDLE). This provider deliberately submits only one request - per session, so it works either way — but tooling built on top - must not assume same-session concurrency. -- [ ] Confirm how the token is passed — `--token` on argv is visible in - process listings; prefer an env var or token-file option if the - real CLI supports one. -- [ ] Confirm how a workload's node/port can be discovered — the - stdout-marker dance here exists only because the assumed contract - has no native endpoint query. +bundle's `/nix` tree read-only and a per-sandbox meta directory +read-write at `/agentix-meta`, creates its workspace, writes an +`AGENTIX_ENDPOINT ` marker file into the meta dir, and +execs `/nix/runtime/bootstrap.sh`. The provider discovers the endpoint +by polling the *host side* of that marker file, interleaved with +`cape status` (it never submits a second command into the session, so +it does not depend on same-session concurrency or on cross-request +workspace persistence), then health-checks `GET /health` on the +discovered endpoint. + +## Contract status — verified against the real CLI + +**The CLI surface implemented here (`cape run/status/logs/cancel`) has +been verified item by item against the CAPE source.** The original +checklist items now all have source-verified answers; `_CapeCli` in +`agentix/provider/cape.py` remains the single class that knows the CLI. + +What matched the previously assumed contract: + +* Verb set: `run` / `status` / `logs` / `cancel` all exist with + `--controller-url` and `--token` on each. (`cape wait` also exists + but is deliberately unused: its process exit code is 0 both on + timeout and on null-exit-code terminal states, so only the status + JSON `state` field can be trusted.) +* `cape run` flags: `--image`, `--gpus`, `--cpu-cores`, `--memory-gb`, + `--gpu-mode whole`, `--isolation-policy`, `--runtime-adapter`, + `--max-duration-seconds`, `--session-key`, + `--bind src:dst[:ro|rw]`, `--env K=V`, and the `-- ` + remainder are all real, with matching semantics. +* Without `--wait`, `cape run` prints exactly one request id on + stdout. Real ids are `req-%06d`; the provider matches + `req-\d{6,}`. +* `cape status` prints one JSON object on stdout (the human-readable + state line goes to stderr); the terminal-state set is exactly + {COMPLETED, FAILED, CANCELLED, EXPIRED, LOST, INFEASIBLE}; the + `request_id` echo the provider cross-checks is always present. The + status JSON has **no stdout/stderr fields** — output only travels + through `cape logs`. +* `cape cancel` takes `--reason`, is an idempotent no-op on already + terminal requests, and rc=0 confirms the cancel. HTTP errors are a + single stderr JSON line (`{"status": , "error": ...}`) with + rc=1; argv errors exit 2. +* The token can **only** be passed as `--token` on argv — the real CLI + reads no env var and no token file, and there is no mint endpoint. + This provider's token-file/env indirection is client-side + convenience feeding that flag; the value is redacted from all error + text, but `ps` visibility cannot be avoided from the CLI side + (direct HTTP with the `X-CAPE-Token` header is the only + alternative). +* The session model is serial (RUNNING_COMMAND ↔ CACHED_IDLE), as + assumed; one sandbox = one request = one unique session key remains + the right mapping. + +What the verification changed: + +* **`--task-id` is required** by the real parser (argparse exits 2 + without it). The provider now passes the sandbox id. +* **`--pool` / `--user` are required.** `CapeProviderConfig.pool` and + `.user` are checked at first use and fail fast with a clear error + instead of an argparse usage dump. +* **`--workspace` does not exist** — the real parser rejects it with + `unrecognized arguments`, rc=2. The flag is gone; the boot script + now creates the workspace itself (`mkdir -p && cd `). For + the same reason `--cwd` is no longer passed: no component creates + the directory, and every CAPE runtime adapter fails at process + start on a missing cwd. +* **Endpoint discovery was redesigned.** `cape logs` succeeds for + RUNNING requests but returns *empty* output until the command exits: + the node agent reports stdout/stderr once, after reaping the + process — CAPE has no streaming/incremental log channel and no + native endpoint query (the status `ports[]` field is a request-spec + echo that never reaches the node). The former stdout-marker + + logs-polling design could therefore never see the marker of a + never-exiting runtime server. Discovery now uses a **marker file + through a rw bind**: each sandbox gets `/` + (created before submit, removed on delete), bound at + `/agentix-meta`; the boot script writes the + `AGENTIX_ENDPOINT ` line to `/agentix-meta/endpoint`; + the provider polls the host side of that file, interleaved with + `cape status` terminal-state checks. Logs are fetched only on + terminal states — where they *are* populated — to enrich error + messages. +* **Cancel treats a controller 404 as already-gone.** After a + journal-less controller restart the request id is unknown; the old + behavior kept bookkeeping (and its port) forever. A 404 in the + CLI's stderr JSON now confirms deletion. + +### Networking: pin a host-network runtime adapter + +The client-assigned-port scheme (`AGENTIX_BIND_PORT` plus a direct +probe of `:`) requires the workload to share the host's +network stack. CAPE's apptainer and bubblewrap adapters do not unshare +the network namespace, so they work; **CAPE's podman adapter runs +without host networking**, so an endpoint bound inside it is +unreachable and this provider cannot work on podman-only pools. The +default `runtime_adapter` stays `None` (pool default), but deployments +must pin `apptainer` or `bubblewrap` (or ensure the pool default is a +host-network adapter). `local-process` shares everything and suits +CPU-only local validation; with it, bind specs are no-ops, and the +boot script falls back to the host-side meta dir and bundle paths +(valid because that adapter shares the host filesystem). ## Install @@ -81,7 +133,10 @@ from agentix.provider.cape import CapeProvider, CapeProviderConfig provider = CapeProvider( CapeProviderConfig( controller_url="https://cape-controller.example:8443", - pool="coding-agent-gpu", + pool="coding-agent-gpu", # required (`cape run --pool`) + user="alice", # required (`cape run --user`) + meta_root="/mnt/shared/agentix-meta", # required (endpoint discovery) + runtime_adapter="apptainer", # pin a host-network adapter token_file="~/.config/cape/token", ) ) @@ -103,6 +158,11 @@ Backend-specific notes: shared filesystem, prior upload, a CAPE-side template — is an **open question**; this iteration deliberately ships no `BundleDeployer` / `agentix deploy cape` until bundle transport is decided. +* **`meta_root` must be visible to both the submitter and the pool + nodes** — the same shared-filesystem assumption `bundle` already + makes, so it adds no new deployment requirement. Each sandbox gets + its own `/` directory (rw-bound at + `/agentix-meta`), which `delete()` removes after a confirmed cancel. * **`url_template`** controls how the runtime URL is built from the discovered `host`/`port`. The default `http://{host}:{port}` assumes the submitter can route to pool nodes directly; behind an SSH tunnel, @@ -128,10 +188,12 @@ Backend-specific notes: after `create()`, a fresh provider cannot see or delete the old sandbox (`delete()` of an unknown id is a silent no-op). Recovery is server-side cleanup by the `agentix-cape-...` session-key convention, - or waiting out `--max-duration-seconds`. The assumed CLI has no + or waiting out `--max-duration-seconds`. The real CLI has no list/query-by-session verb to rebuild the map from. * **Delete retries instead of leaking.** `delete()` drops bookkeeping - only after the controller confirms the cancel (`cape cancel` rc=0); a + only after the controller confirms the cancel (`cape cancel` rc=0) or + reports the request as unknown (HTTP 404 — e.g. after a journal-less + controller restart, when the request is gone anyway); any other failed cancel logs a warning and keeps the record so calling `delete()` again retries it. * **Runtime ports.** Each live sandbox of one provider instance gets a @@ -140,8 +202,7 @@ Backend-specific notes: sandboxes of the same provider can never answer each other's health probes (same node or same SSH tunnel). Port collisions with *other* submitters or users on the same node cannot be reserved from this - side — that is a known limitation of the assumed contract (no - controller-side port brokering). + side — the real contract has no controller-side port brokering. ## License diff --git a/plugins/providers/cape/agentix/provider/cape.py b/plugins/providers/cape/agentix/provider/cape.py index f3e7740..4238e09 100644 --- a/plugins/providers/cape/agentix/provider/cape.py +++ b/plugins/providers/cape/agentix/provider/cape.py @@ -14,31 +14,64 @@ does not depend on same-session concurrency or on cross-request workspace persistence. - - `create()` submits a runtime request whose workload prints an - `AGENTIX_ENDPOINT ` marker line to stdout (first IP of - `hostname -i`, plus the assigned bind port) and then execs the - bundle's `/nix/runtime/bootstrap.sh`. The provider polls - `cape status` (to detect early death) and `cape logs` (to find the - marker), then health-checks `GET /health` on the discovered - endpoint with a raw TCP probe (never an env-proxy-aware HTTP - client, which would hang behind a corp proxy or SSH tunnel). - - `delete()` cancels the runtime request; bookkeeping is dropped only - after the controller confirms the cancel, so a failed cancel can be - retried with another `delete()`. + - `create()` submits a runtime request whose workload creates the + per-sandbox workspace, writes an `AGENTIX_ENDPOINT ` + marker file into the per-sandbox meta directory (rw-bound at + `/agentix-meta`), and execs the bundle's + `/nix/runtime/bootstrap.sh`. The provider discovers the endpoint by + polling the *host side* of that marker file, interleaved with + `cape status` (to fail fast when the workload dies), then + health-checks `GET /health` on the discovered endpoint with a raw + TCP probe (never an env-proxy-aware HTTP client, which would hang + behind a corp proxy or SSH tunnel). + - `delete()` cancels the runtime request and removes the meta dir; + bookkeeping is dropped only after the controller confirms the + cancel (or reports the request as unknown — HTTP 404 — after e.g. a + journal-less controller restart), so a failed cancel can be retried + with another `delete()`. - `config.bundle` for this backend is an OPAQUE node-visible path to an already-extracted bundle tree; it is bind-mounted read-only at `/nix`. How the bundle tree gets onto CAPE nodes (shared FS, prior upload) is deliberately out of scope — there is no `BundleDeployer` in this iteration. - -ASSUMED CLI CONTRACT — the `cape run/status/logs/cancel` verb surface -implemented in `_CapeCli` was reverse-documented from a sibling -project's adapter of the same CLI and has only ever been -exercised against fakes and emulators; the official CAPE CLI has never -been obtained. Endpoint discovery additionally assumes `cape logs` can -return the stdout of a still-RUNNING request. Verify every verb, flag, -and the `cape status` JSON schema against the real CLI before -production use (see the provider README's "Contract status" checklist). + - `CapeProviderConfig.meta_root` must be a directory visible to both + the submitter and the pool nodes — the same shared-filesystem + assumption `bundle` already makes. Each sandbox uses + `/`, created before submit and removed on + delete. + +CLI CONTRACT — verified item by item against the CAPE source. The key +verified facts this provider is built on: + + * `cape run` requires `--pool`, `--user`, and `--task-id` (argparse + exits 2 without them) and has NO `--workspace` flag. Without + `--wait`, stdout is exactly one line: the request id (`req-%06d`, + matched here with `req-\\d{6,}`). + * `cape status` prints a single JSON object on stdout (the + human-readable state line goes to stderr). Terminal states are + COMPLETED / FAILED / CANCELLED / EXPIRED / LOST / INFEASIBLE, and + the JSON carries no stdout/stderr fields. + * `cape logs` succeeds for RUNNING requests but returns EMPTY output + until the command exits — the node agent reports stdout/stderr + once, after reaping the process; CAPE has no streaming/incremental + log channel. Endpoint discovery therefore CANNOT go through logs + (the runtime server never exits); it uses the marker file above. + Logs are fetched only for terminal requests, as error diagnostics. + * The token can only be passed as `--token` on argv (the CLI reads no + env var or file). This provider's token-file/env indirection is + client-side convenience that feeds that flag; the literal value is + redacted from error text, but `ps` visibility cannot be avoided + from the CLI side. + +NETWORKING — the client-assigned-port scheme (`AGENTIX_BIND_PORT` plus +a direct probe of `:`) requires the workload to share the +host's network stack. Pin `CapeProviderConfig.runtime_adapter` to a +host-network adapter: apptainer and bubblewrap do not unshare the +network namespace, but CAPE's podman adapter runs without host +networking, so an endpoint bound inside it would be unreachable. +(`local-process` shares everything and suits CPU-only local +validation.) The default stays None — the pool default applies — but +deployments must ensure a host-network adapter serves Agentix requests. """ from __future__ import annotations @@ -50,6 +83,8 @@ import math import os import re +import shlex +import shutil from collections.abc import Mapping, Sequence from dataclasses import dataclass from pathlib import Path @@ -71,17 +106,30 @@ logger = logging.getLogger("agentix.provider.cape") -_REQUEST_ID_RE = re.compile(r"req-[A-Za-z0-9][A-Za-z0-9_.-]*") -"""Shape of the request id `cape run` prints on stdout (assumed contract).""" +_REQUEST_ID_RE = re.compile(r"req-\d{6,}") +"""Shape of the request id `cape run` prints on stdout (`req-%06d` in +the CAPE source; more than six digits once the sequence outgrows the +minimum width).""" _TERMINAL_STATES = frozenset({"COMPLETED", "FAILED", "CANCELLED", "EXPIRED", "LOST", "INFEASIBLE"}) -"""Terminal request states in the `cape status` JSON (assumed contract).""" +"""Terminal request states in the `cape status` JSON (verified against +the CAPE source; the non-terminal states are SUBMITTED / QUEUED / +LEASED / PREPARING_IMAGE / STARTING_SESSION / RUNNING / RECONCILING).""" _REASON_UNSAFE_RE = re.compile(r"[^A-Za-z0-9_.-]") """Characters stripped from `cape cancel --reason` strings.""" _ENDPOINT_MARKER = "AGENTIX_ENDPOINT" -"""Prefix of the stdout marker line the boot script prints before exec.""" +"""Prefix of the marker line the boot script writes to the meta dir.""" + +_META_MOUNT = "/agentix-meta" +"""In-sandbox mount point of the per-sandbox meta directory (rw bind).""" + +_ENDPOINT_FILE_NAME = "endpoint" +"""Marker file name inside the per-sandbox meta directory.""" + +_BUNDLE_BOOT_RELATIVE = BUNDLE_RUNTIME_ENTRYPOINT.removeprefix(BUNDLE_NIX_ROOT + "/") +"""Bootstrap path relative to the bundle root (`runtime/bootstrap.sh`).""" _MEMORY_RE = re.compile(r"(\d+)\s*([kmgt])?i?b?", re.IGNORECASE) _MEMORY_UNIT_BYTES = {"k": 1 << 10, "m": 1 << 20, "g": 1 << 30, "t": 1 << 40} @@ -95,8 +143,8 @@ class CapeProviderConfig(BaseModel): Every field is optional or defaulted so `CapeProvider()` constructs with zero arguments — the plugin registry instantiates providers - with `cls()`. Anything unset falls back to environment variables at - first use. + with `cls()`. Fields the real CLI cannot work without (`pool`, + `user`, `meta_root`) are checked at first use, not construction. """ binary: str | None = Field( @@ -120,12 +168,31 @@ class CapeProviderConfig(BaseModel): default="CAPE_TOKEN", description="Env var consulted for the token when no token file is configured.", ) - pool: str | None = Field(default=None, description="Optional CAPE pool name.") - user: str | None = Field(default=None, description="Optional CAPE user identity.") + pool: str | None = Field( + default=None, + description="CAPE pool name (`--pool`). The real CLI requires it; " + "checked at first use.", + ) + user: str | None = Field( + default=None, + description="CAPE user identity (`--user`). The real CLI requires it; " + "checked at first use.", + ) + meta_root: str | None = Field( + default=None, + description="Directory for per-sandbox meta dirs (`/`), " + "used for endpoint discovery: the boot script writes the endpoint marker " + "file through a rw bind of the meta dir, and the provider polls the host " + "side. Must be visible to both the submitter and the pool nodes — the " + "same shared-filesystem assumption `bundle` makes. Checked at first use.", + ) workspace_root: str = Field( default="/workspace", description="Node-side base directory; each sandbox uses " - "`/` as its session workspace and cwd.", + "`/` as its workspace. The boot script " + "creates it (`mkdir -p`) and cd's into it before exec'ing the runtime — " + "nothing else creates it, which is also why `--cwd` is never passed " + "(every CAPE runtime adapter fails at process start on a missing cwd).", ) cpu_cores: int | None = Field( default=None, @@ -142,12 +209,16 @@ class CapeProviderConfig(BaseModel): isolation_policy: str = Field(default="default", description="`--isolation-policy` value.") runtime_adapter: str | None = Field( default=None, - description="Optional `--runtime-adapter` (e.g. `apptainer`).", + description="Optional `--runtime-adapter`. Production deployments should pin " + "a host-network adapter (`apptainer` or `bubblewrap`): the client-assigned-" + "port scheme needs the workload on the host network stack, and CAPE's " + "podman adapter has no host networking. `local-process` suits CPU-only " + "local validation.", ) extra_binds: list[str] = Field( default_factory=list, description="Raw `src:dst[:ro|rw]` bind specs passed through in addition to " - "the bundle's `/nix` bind.", + "the bundle's `/nix` bind and the meta dir's `/agentix-meta` bind.", ) runtime_port_base: int = Field( default=8710, @@ -184,9 +255,9 @@ class CapeProviderConfig(BaseModel): ) transient_failure_limit: int = Field( default=5, - description="Consecutive `cape status`/`cape logs` failures tolerated during " - "endpoint discovery before the create is failed (a single controller " - "blip must not cancel a lease that already queued onto a GPU).", + description="Consecutive `cape status` failures tolerated during endpoint " + "discovery before the create is failed (a single controller blip must " + "not cancel a lease that already queued onto a GPU).", ) @@ -218,6 +289,31 @@ def _resolve_controller_url(config: CapeProviderConfig) -> str: return url +def _required_settings(config: CapeProviderConfig) -> tuple[str, str, str]: + """Fail fast on the settings the real CLI / discovery cannot work without. + + `cape run` rejects submissions without `--pool`/`--user` (argparse + exits 2 with a usage error that would otherwise surface as an + unreadable RuntimeError), and endpoint discovery needs `meta_root`. + Returns the narrowed `(pool, user, meta_root)` triple. + """ + pool, user, meta_root = config.pool, config.user, config.meta_root + if not pool or not user or not meta_root: + missing = [ + name + for name, value in (("pool", pool), ("user", user), ("meta_root", meta_root)) + if not value + ] + raise RuntimeError( + "CapeProviderConfig." + ", CapeProviderConfig.".join(missing) + " must be set " + "before creating sandboxes: the real `cape run` requires --pool and --user, " + "and endpoint discovery needs meta_root — a directory visible to both the " + "submitter and the pool nodes (the same shared-filesystem assumption " + "`bundle` makes)" + ) + return pool, user, meta_root + + def _read_token(config: CapeProviderConfig) -> str: """Read the CAPE token, fresh on every operation (supports rotation). @@ -295,29 +391,50 @@ def _memory_gb(resource: SandboxResource | None, config: CapeProviderConfig) -> return max(1, math.ceil(num_bytes / (1 << 30))) -def _boot_script() -> str: +def _boot_script(*, workspace: str, meta_dir: str, bundle: str) -> str: """Workload for the long-lived runtime request. - Prints the `AGENTIX_ENDPOINT ` marker to stdout (first - IP of `hostname -i`, plus the assigned bind port) and execs the - bundle's bootstrap entry point. The marker is read back host-side - via `cape logs` during endpoint discovery — nothing is written to - the workspace, so discovery does not depend on cross-request - workspace persistence or on a second same-session command. + Creates and enters the per-sandbox workspace (nothing else creates + it — `--cwd` is deliberately not passed, because every CAPE runtime + adapter fails at process start when the cwd does not exist), writes + the `AGENTIX_ENDPOINT ` marker file into the meta dir, + and execs the bundle's bootstrap entry point. + + The marker goes to `/agentix-meta/endpoint` — the rw bind of the + host-side per-sandbox meta dir — and the provider polls the host + side of that file during endpoint discovery. `cape logs` cannot be + used for discovery: CAPE has no streaming logs (stdout/stderr are + reported once, after the command exits), so a runtime server that + never exits never publishes anything through logs. + + Adapters that ignore bind specs but share the host filesystem + (local-process) fall back to the host-side meta dir path and the + host-side bundle bootstrap path, so the same script stays honest + there. A marker-write failure exits non-zero on purpose: the + request goes terminal and `create()` fails fast with diagnostics + instead of burning the whole discovery budget. """ + quoted_ws = shlex.quote(workspace) + host_meta = shlex.quote(meta_dir) + host_boot = shlex.quote(f"{bundle.rstrip('/')}/{_BUNDLE_BOOT_RELATIVE}") return ( - f'printf "{_ENDPOINT_MARKER} %s %s\\n" "$(hostname -i | cut -d" " -f1)" ' - f'"${{{BIND_PORT_ENV}}}" && ' - f"exec {BUNDLE_RUNTIME_ENTRYPOINT}" + f"mkdir -p {quoted_ws} && cd {quoted_ws} || exit 1; " + f"if [ -d {_META_MOUNT} ]; then m={_META_MOUNT}; else m={host_meta}; fi; " + f'h=$(hostname -i 2>/dev/null | cut -d" " -f1); [ -n "$h" ] || h=$(hostname); ' + f'printf "{_ENDPOINT_MARKER} %s %s\\n" "$h" "${{{BIND_PORT_ENV}}}" ' + f'> "$m/{_ENDPOINT_FILE_NAME}" || exit 1; ' + f'b={BUNDLE_RUNTIME_ENTRYPOINT}; [ -e "$b" ] || b={host_boot}; ' + f'exec "$b"' ) def _parse_endpoint(text: str) -> tuple[str, int] | None: - """Find the `AGENTIX_ENDPOINT ` marker line in workload stdout. + """Find the `AGENTIX_ENDPOINT ` marker line in `text`. - The explicit marker prefix keeps unrelated two-token output (banner - lines like `GPU 0`) from being misread as an endpoint. - """ + The explicit marker prefix keeps unrelated two-token content from + being misread as an endpoint, and a partially written marker file + (seen mid-`printf` on a shared FS) simply fails to parse until the + next poll.""" for line in text.splitlines(): parts = line.split() if len(parts) == 3 and parts[0] == _ENDPOINT_MARKER and parts[2].isdigit(): @@ -325,15 +442,52 @@ def _parse_endpoint(text: str) -> tuple[str, int] | None: return None +def _read_endpoint_file(path: Path) -> str: + """Best-effort read of the host-side marker file ('' when absent). + + Synchronous file IO — call via `asyncio.to_thread` (meta_root + typically lives on a shared FS that must not block the loop).""" + try: + return path.read_text(encoding="utf-8", errors="replace") + except OSError: + return "" + + +def _stderr_reports_unknown_request(stderr: str) -> bool: + """True when `cape` stderr carries the CLI's single-line HTTP-error + JSON (`{"status": , "error": ...}`) with status 404 — the + controller does not know the request id (e.g. it restarted without + a journal), so a cancel can be treated as already done.""" + for line in stderr.splitlines(): + line = line.strip() + if not line.startswith("{"): + continue + try: + payload = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(payload, dict) and payload.get("status") == 404: + return True + return False + + +async def _remove_meta_dir(meta_dir: Path) -> None: + """Best-effort removal of a per-sandbox meta dir (never raises).""" + await asyncio.to_thread(shutil.rmtree, meta_dir, ignore_errors=True) + + class _CapeCli: - """The ONE place that knows the assumed `cape` CLI verb surface. + """The ONE place that knows the `cape` CLI verb surface. Every method's contract (verbs, flags, stdout/JSON shapes, terminal - states) is an assumption reverse-documented from a sibling - project's adapter — it has never been checked against a real - `cape` binary. - Keep all CLI knowledge in this class so a contract correction after - real-CLI verification is a single-class change. + states) has been verified item by item against the CAPE source. + The CLI also offers `cape wait`, which this provider deliberately + does not use: its process exit code is 0 both on timeout (printing + a non-terminal status) and on null-exit-code terminal states, so it + must never be trusted — polling `cape status` and parsing the JSON + `state` field is the only sound approach. Keep all CLI knowledge in + this class so any future contract correction is a single-class + change. """ def __init__(self, config: CapeProviderConfig) -> None: @@ -391,8 +545,7 @@ async def run( self, *, session_key: str, - workspace: str, - cwd: str, + task_id: str, image: str, gpus: int, workload: Sequence[str], @@ -401,28 +554,35 @@ async def run( cpu_cores: int | None = None, memory_gb: int | None = None, ) -> str: - """ASSUMED CLI contract — verify against the real `cape` CLI before production use. + """Submit one workload request and return its request id. - Submits one workload request and returns its request id: + Verified against the real parser: - cape run --controller-url U --token T [--pool P] [--user USR] - --workspace WS --image IMG --gpus N [--cpu-cores N] + cape run --controller-url U --token T --pool P --user USR + --task-id ID --image IMG --gpus N [--cpu-cores N] [--memory-gb N] --gpu-mode whole --isolation-policy X - [--runtime-adapter Y] --max-duration-seconds T --cwd CWD - --session-key KEY [--bind src:dst[:ro|rw]]... [--env K=V]... - -- - - stdout, after stripping blank lines, must be exactly one line - matching `req-[A-Za-z0-9][A-Za-z0-9_.-]*`. + [--runtime-adapter Y] --max-duration-seconds T + --session-key KEY [--bind src:dst[:ro|rw]]... + [--env K=V]... -- + + `--pool` / `--user` / `--task-id` are required (argparse exits 2 + without them); there is NO `--workspace` flag; `--cwd` is + deliberately not passed because no component creates the + directory — the boot script `mkdir -p && cd`'s instead. Without + `--wait`, stdout (after stripping blank lines) is exactly one + line matching `req-\\d{6,}`. """ - token = await asyncio.to_thread(_read_token, self._config) cfg = self._config + pool, user = cfg.pool, cfg.user + if not pool or not user: + raise RuntimeError( + "`cape run` requires --pool and --user; set CapeProviderConfig.pool " + "and CapeProviderConfig.user" + ) + token = await asyncio.to_thread(_read_token, cfg) argv: list[str] = ["run", *self._common(token)] - if cfg.pool: - argv += ["--pool", cfg.pool] - if cfg.user: - argv += ["--user", cfg.user] - argv += ["--workspace", workspace, "--image", image, "--gpus", str(int(gpus))] + argv += ["--pool", pool, "--user", user, "--task-id", task_id] + argv += ["--image", image, "--gpus", str(int(gpus))] if cpu_cores is not None: argv += ["--cpu-cores", str(int(cpu_cores))] if memory_gb is not None: @@ -431,7 +591,7 @@ async def run( if cfg.runtime_adapter: argv += ["--runtime-adapter", cfg.runtime_adapter] argv += ["--max-duration-seconds", str(int(cfg.max_duration_seconds))] - argv += ["--cwd", cwd, "--session-key", session_key] + argv += ["--session-key", session_key] for bind in binds: argv += ["--bind", bind] for key, value in (env or {}).items(): @@ -449,14 +609,16 @@ async def run( return lines[0] async def status(self, request_id: str) -> dict[str, object]: - """ASSUMED CLI contract — verify against the real `cape` CLI before production use. - - `cape status --controller-url U --token T` prints one JSON - object with at least a `state` field; terminal states are - COMPLETED / FAILED / CANCELLED / EXPIRED / LOST / INFEASIBLE. A - `request_id` field, when present and non-null, must echo the - queried id — a mismatch means the CLI answered for a different - request and is treated as an infrastructure error, not trusted. + """`cape status --controller-url U --token T`. + + Verified: prints one JSON object on stdout (the human-readable + state line goes to stderr, so it never pollutes the parse) with + at least `state` and a non-empty `request_id` echo; terminal + states are COMPLETED / FAILED / CANCELLED / EXPIRED / LOST / + INFEASIBLE; the JSON carries NO stdout/stderr fields (logs only + travel through `cape logs`). A `request_id` echo mismatch means + the CLI answered for a different request and is treated as an + infrastructure error, not trusted. """ token = await asyncio.to_thread(_read_token, self._config) rc, stdout, stderr = await self._exec( @@ -484,11 +646,15 @@ async def status(self, request_id: str) -> dict[str, object]: return payload async def cancel(self, request_id: str, reason: str) -> bool: - """ASSUMED CLI contract — verify against the real `cape` CLI before production use. - - `cape cancel --controller-url U --token T --reason R`. - Returns True only when the CLI confirmed the cancel (rc=0); - ordinary failures are swallowed (best-effort) after a redacted + """`cape cancel --controller-url U --token T --reason R`. + + Verified: rc=0 confirms the cancel (idempotent no-op on already + terminal requests, printing the post-cancel status JSON). + Returns True when the CLI confirmed the cancel OR when the + controller reported the request as unknown (HTTP 404 in the + CLI's stderr JSON) — after a journal-less controller restart the + request is gone and bookkeeping must not stick forever. + Ordinary failures are swallowed (best-effort) after a redacted warning. External cancellation is NOT swallowed: the in-flight cancel RPC gets a bounded shielded window to reach the controller, then CancelledError is re-raised so callers such as @@ -523,6 +689,13 @@ async def _cancel_once(self, request_id: str, reason: str) -> bool: verb="cancel", ) if rc != 0: + if _stderr_reports_unknown_request(stderr): + logger.info( + "cape cancel %s: controller no longer knows the request (HTTP 404); " + "treating it as already gone", + request_id, + ) + return True logger.warning( "cape cancel %s (reason=%s) exited rc=%d: %s", request_id, @@ -533,21 +706,15 @@ async def _cancel_once(self, request_id: str, reason: str) -> bool: return False return True - async def logs(self, request_id: str, *, check: bool = False) -> tuple[str, str]: - """ASSUMED CLI contract — verify against the real `cape` CLI before production use. - - `cape logs --controller-url U --token T` prints the - workload's stdout/stderr. ADDITIONAL ASSUMPTION introduced by - this provider: `cape logs` can return the stdout of a - still-RUNNING request — endpoint discovery depends on it, and - the reference adapter only ever called logs on terminal - requests. Verify this explicitly against the real CLI. - - With `check=False` (default) this is best-effort — failures - collapse to `("", "")` (error-message enrichment only). With - `check=True` a failed invocation raises RuntimeError so endpoint - discovery can tell a broken verb apart from a marker that simply - has not been printed yet. + async def logs(self, request_id: str) -> tuple[str, str]: + """`cape logs --controller-url U --token T`. + + Verified: prints the workload's stdout/stderr, BUT both are + empty until the command exits — the node agent reports output + once, after reaping the process; there is no streaming channel. + This method is therefore only called on terminal requests (whose + logs are populated) to enrich error messages. Best-effort: + failures collapse to `("", "")`. """ try: token = await asyncio.to_thread(_read_token, self._config) @@ -555,12 +722,8 @@ async def logs(self, request_id: str, *, check: bool = False) -> tuple[str, str] ["logs", request_id, *self._common(token)], token=token, verb="logs" ) except Exception: - if check: - raise return "", "" if rc != 0: - if check: - raise RuntimeError(f"cape logs {request_id} failed (rc={rc}): {stderr}") return "", "" return stdout, stderr @@ -572,12 +735,13 @@ class _CapeSandboxRecord: request_id: str session_key: str workspace: str + meta_dir: str runtime_url: str runtime_port: int class CapeProvider(SandboxProvider): - """Sandbox CRUD via the `cape` CLI (assumed contract; see module docstring).""" + """Sandbox CRUD via the `cape` CLI (see module docstring).""" def __init__(self, config: CapeProviderConfig | None = None) -> None: self.config = config or CapeProviderConfig() @@ -601,15 +765,27 @@ def _allocate_port(self) -> int: ) async def create(self, config: SandboxConfig) -> Sandbox: + _pool, _user, meta_root = _required_settings(self.config) sandbox_id = SandboxId(f"cape-{uuid4().hex[:12]}") session_key = f"agentix-{sandbox_id}" workspace = f"{self.config.workspace_root.rstrip('/')}/{sandbox_id}" + meta_dir = Path(meta_root).expanduser() / str(sandbox_id) port = self._allocate_port() + try: + await asyncio.to_thread(meta_dir.mkdir, parents=True, exist_ok=True) + except OSError as exc: + self._inflight_ports.discard(port) + raise RuntimeError(f"cannot create the sandbox meta dir {meta_dir}: {exc}") from exc - binds = [f"{config.bundle}:{BUNDLE_NIX_ROOT}:ro", *self.config.extra_binds] + binds = [ + f"{config.bundle}:{BUNDLE_NIX_ROOT}:ro", + f"{meta_dir}:{_META_MOUNT}:rw", + *self.config.extra_binds, + ] env = {BIND_PORT_ENV: str(port), **(config.env or {})} resource = config.resource gpus = resource.gpu if resource is not None and resource.gpu is not None else 0 + boot = _boot_script(workspace=workspace, meta_dir=str(meta_dir), bundle=config.bundle) # The submit runs shielded: if we are cancelled mid-`cape run`, the # CLI still finishes inside its own timeout, the request id is @@ -621,11 +797,10 @@ async def create(self, config: SandboxConfig) -> Sandbox: run_task = asyncio.ensure_future( self._cli.run( session_key=session_key, - workspace=workspace, - cwd=workspace, + task_id=str(sandbox_id), image=config.image, gpus=gpus, - workload=["sh", "-c", _boot_script()], + workload=["sh", "-c", boot], binds=binds, env=env, cpu_cores=_cpu_cores(resource, self.config), @@ -645,9 +820,11 @@ async def create(self, config: SandboxConfig) -> Sandbox: ) if harvested is not None: await self._cli.cancel(harvested, "agentix_create_cancelled") + await _remove_meta_dir(meta_dir) raise except BaseException: self._inflight_ports.discard(port) + await _remove_meta_dir(meta_dir) raise logger.info("CAPE runtime request %s submitted for sandbox %s", server_req, sandbox_id) @@ -657,19 +834,23 @@ async def create(self, config: SandboxConfig) -> Sandbox: try: loop = asyncio.get_running_loop() deadline = loop.time() + self.config.create_timeout_seconds - host, marker_port = await self._discover_endpoint(server_req, deadline=deadline) + host, marker_port = await self._discover_endpoint( + server_req, endpoint_file=meta_dir / _ENDPOINT_FILE_NAME, deadline=deadline + ) runtime_url = self.config.url_template.format(host=host, port=marker_port) await self._wait_healthy(server_req, runtime_url, deadline) except BaseException: self._sandboxes.pop(sandbox_id, None) self._inflight_ports.discard(port) await self._cli.cancel(server_req, "agentix_create_failed") + await _remove_meta_dir(meta_dir) raise self._sandboxes[sandbox_id] = _CapeSandboxRecord( request_id=server_req, session_key=session_key, workspace=workspace, + meta_dir=str(meta_dir), runtime_url=runtime_url, runtime_port=port, ) @@ -678,23 +859,34 @@ async def create(self, config: SandboxConfig) -> Sandbox: # `call_deadline` on the returned handle post-create. return Sandbox(sandbox_id=sandbox_id, runtime_url=runtime_url, status="running") - async def _discover_endpoint(self, server_req: str, *, deadline: float) -> tuple[str, int]: - """Find the node/port the runtime bound, from the workload's stdout. - - Polls `cape status` (a terminal state means the runtime died — - fail fast with its logs) and `cape logs` (looking for the - `AGENTIX_ENDPOINT` marker). Transient status/logs failures are - tolerated up to `transient_failure_limit` consecutive times so a - single controller blip cannot kill a create whose lease already - queued onto a GPU. No second request is ever submitted into the - session — the assumed session model is serial (RUNNING_COMMAND ↔ - CACHED_IDLE), so a same-session probe could queue forever behind - the never-ending runtime command. + async def _discover_endpoint( + self, server_req: str, *, endpoint_file: Path, deadline: float + ) -> tuple[str, int]: + """Wait for the workload to write the endpoint marker file. + + CAPE has no streaming logs — `cape logs` returns empty output + until the command exits, and the runtime server never exits — + so discovery polls the *host side* of the per-sandbox marker + file the boot script writes through the rw-bound meta dir, + interleaved with `cape status` so a workload that died before + publishing fails fast (its logs ARE populated once the request + is terminal, and are fetched then for diagnostics). Transient + status failures are tolerated up to `transient_failure_limit` + consecutive times so a single controller blip cannot kill a + create whose lease already queued onto a GPU. No second request + is ever submitted into the session — the session model is + serial (RUNNING_COMMAND ↔ CACHED_IDLE), so a same-session probe + would queue forever behind the never-ending runtime command. """ loop = asyncio.get_running_loop() limit = self.config.transient_failure_limit failures = 0 while True: + marker = await asyncio.to_thread(_read_endpoint_file, endpoint_file) + if marker: + endpoint = _parse_endpoint(marker) + if endpoint is not None: + return endpoint try: status = await self._cli.status(server_req) except RuntimeError as exc: @@ -704,42 +896,17 @@ async def _discover_endpoint(self, server_req: str, *, deadline: float) -> tuple f"cape status {server_req} failed {failures} consecutive times " f"during endpoint discovery: {exc}" ) from exc - if loop.time() >= deadline: - raise TimeoutError( - f"CAPE runtime request {server_req} did not publish its endpoint " - f"within {self.config.create_timeout_seconds}s" - ) from exc - await asyncio.sleep(self.config.poll_interval_seconds) - continue - state = str(status.get("state", "")) - if state in _TERMINAL_STATES: - stdout, stderr = await self._cli.logs(server_req) - raise RuntimeError( - f"CAPE runtime request {server_req} reached terminal state {state} " - f"before publishing its endpoint.\n" - f"--- workload stdout ---\n{stdout}\n" - f"--- workload stderr ---\n{stderr}" - ) - try: - stdout, _stderr = await self._cli.logs(server_req, check=True) - except (RuntimeError, OSError) as exc: - failures += 1 - if failures >= limit: + else: + failures = 0 + state = str(status.get("state", "")) + if state in _TERMINAL_STATES: + stdout, stderr = await self._cli.logs(server_req) raise RuntimeError( - f"cape logs {server_req} failed {failures} consecutive times " - f"during endpoint discovery: {exc}" - ) from exc - if loop.time() >= deadline: - raise TimeoutError( - f"CAPE runtime request {server_req} did not publish its endpoint " - f"within {self.config.create_timeout_seconds}s" - ) from exc - await asyncio.sleep(self.config.poll_interval_seconds) - continue - failures = 0 - endpoint = _parse_endpoint(stdout) - if endpoint is not None: - return endpoint + f"CAPE runtime request {server_req} reached terminal state {state} " + f"before publishing its endpoint marker.\n" + f"--- workload stdout ---\n{stdout}\n" + f"--- workload stderr ---\n{stderr}" + ) if loop.time() >= deadline: raise TimeoutError( f"CAPE runtime request {server_req} did not publish its endpoint " @@ -754,8 +921,9 @@ async def _wait_healthy(self, server_req: str, runtime_url: str, deadline: float client library: proxy env vars (`http_proxy`, ...) would leak into loopback/tunnel probes on corp-proxy hosts and hang them. Every fifth round the server request's state is re-checked so a - runtime that printed its marker and then crashed fails fast with - its logs instead of probing a dead endpoint for the whole budget. + runtime that wrote its marker and then crashed fails fast with + its logs (populated once terminal) instead of probing a dead + endpoint for the whole budget. """ parts = urlsplit(runtime_url) host = parts.hostname or "127.0.0.1" @@ -818,14 +986,16 @@ async def get(self, sandbox_id: SandboxId) -> SandboxInfo: ) async def delete(self, sandbox_id: SandboxId) -> None: - """Cancel the sandbox's runtime request. + """Cancel the sandbox's runtime request and remove its meta dir. Unknown ids are a silent no-op (`session()` calls this on the user's exception path, so a raise here would mask the original error). Bookkeeping is dropped only after the controller - confirms the cancel — a rejected/failed cancel keeps the record - (with a warning) so a later `delete()` can retry instead of - silently leaking the GPU lease. + confirms the cancel — or reports the request as unknown (HTTP + 404, e.g. after a journal-less controller restart), which is + treated as already-gone. A rejected/failed cancel keeps the + record (with a warning) so a later `delete()` can retry instead + of silently leaking the GPU lease. """ record = self._sandboxes.get(sandbox_id) if record is None: @@ -841,6 +1011,7 @@ async def delete(self, sandbox_id: SandboxId) -> None: return self._sandboxes.pop(sandbox_id, None) self._inflight_ports.discard(record.runtime_port) + await _remove_meta_dir(Path(record.meta_dir)) logger.info("Deleted sandbox %s (request %s)", sandbox_id, record.request_id) diff --git a/plugins/providers/cape/pyproject.toml b/plugins/providers/cape/pyproject.toml index 9d91f34..45c6f68 100644 --- a/plugins/providers/cape/pyproject.toml +++ b/plugins/providers/cape/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "hatchling.build" [project] name = "agentix-provider-cape" version = "0.1.0" -description = "CAPE lease-based GPU pool provider backend for Agentix (registers the `cape` backend; assumed CLI contract)" +description = "CAPE lease-based GPU pool provider backend for Agentix (registers the `cape` backend; CLI contract verified against the CAPE source)" requires-python = ">=3.11" dependencies = [ # Protocol + dataclasses (`SandboxProvider`, `Sandbox`, `SandboxConfig`, diff --git a/plugins/providers/cape/tests/test_cape_provider.py b/plugins/providers/cape/tests/test_cape_provider.py index 5271b96..7642216 100644 --- a/plugins/providers/cape/tests/test_cape_provider.py +++ b/plugins/providers/cape/tests/test_cape_provider.py @@ -3,26 +3,40 @@ Follows the apptainer provider test pattern: a fake executable staged in `tmp_path` records every invocation as one JSON line, and a real local HTTP server answers `GET /health` so the provider's raw-TCP -probe succeeds. The point is to lock in the *assumed* CAPE CLI surface -the provider emits (verbs, flags, workload shape) — no network beyond -127.0.0.1, no real `cape`. - -The fake models the assumed contract with a small file state machine: - - * `run` classifies each request by its workload — a workload that - execs the bundle bootstrap is a *server* request (stays RUNNING - until cancelled; overridable via `FAKE_CAPE_SERVER_STATE`); any - other workload completes immediately. Multiple sandboxes therefore - work in one test. - * `logs` prints noisy banner lines plus the `AGENTIX_ENDPOINT` marker - (port from `FAKE_CAPE_PORT`). `FAKE_CAPE_LOGS_EMPTY=N` withholds - the marker for the first N calls (exercising the discovery retry - loop); `FAKE_CAPE_LOGS_FAILURES=N` / `FAKE_CAPE_STATUS_FAILURES=N` - make the first N invocations of that verb exit non-zero - (exercising the transient-failure tolerance). - * `cancel` records a per-request cancelled marker; - `FAKE_CAPE_CANCEL_RC` forces a failing exit code (exercising the - delete-retry path). +probe succeeds. The fake models the REAL CAPE CLI contract (verified +against the CAPE source) — no network beyond 127.0.0.1, no real `cape`. + +Real-contract facts the fake reproduces: + + * `run` parses argv with the real flag table: `--pool` / `--user` / + `--task-id` are required and there is NO `--workspace` flag — + violations exit 2 with an argparse-style usage error, exactly like + the real parser. + * Request ids look like the real ones (`req-%06d`). + * `logs` returns EMPTY stdout/stderr while the request is running + (CAPE has no streaming logs); output appears only once the request + is terminal. Endpoint discovery must therefore never depend on + `cape logs` — it reads the marker file the boot script writes. + * `run` actually EXECUTES a server workload (the provider's boot + script) under `sh -c` with a stubbed node environment: + `AGENTIX_BIND_PORT` is overridden with `FAKE_CAPE_PORT` (where the + test's health server listens) and `FAKE_CAPE_STUB_PATH` is + prepended to PATH (stub `hostname` resolving to loopback). The + boot script then really writes the endpoint marker into the tmp + meta dir the provider polls. `FAKE_CAPE_EXEC_WORKLOAD=0` skips + execution so the marker never appears. + * `cancel` prints the post-cancel status JSON; `FAKE_CAPE_CANCEL_RC` + forces a failing exit code (delete-retry path) and + `FAKE_CAPE_CANCEL_404` emits the CLI's single-line HTTP-error JSON + (`{"status": 404, ...}`) on stderr with rc=1 (already-gone path). + * `FAKE_CAPE_STATUS_FAILURES=N` makes the first N `status` calls + exit non-zero (transient-failure tolerance); + `FAKE_CAPE_SERVER_STATE` overrides the server request's state. + +The fake keeps the status JSON minimal ({state, request_id, +exit_code}) — the provider only ever reads `state` and the +`request_id` echo. `cape wait` is deliberately not modeled: the +provider must never call it (its exit code is untrustworthy). """ from __future__ import annotations @@ -50,13 +64,16 @@ from agentix.provider.base import SandboxConfig, SandboxId, SandboxProvider, SandboxResource _FAKE_CAPE_BODY = ''' -"""Recording fake `cape` CLI driven by FAKE_CAPE_* env vars.""" +"""Recording fake `cape` CLI modeling the REAL contract (see test module).""" +import argparse import json import os +import subprocess import sys LOG = os.environ["FAKE_CAPE_LOG"] STATE = os.environ["FAKE_CAPE_STATE"] +TERMINAL = {"COMPLETED", "FAILED", "CANCELLED", "EXPIRED", "LOST", "INFEASIBLE"} def log(argv): @@ -77,67 +94,147 @@ def bump(name): return n -def req_kind(req): - path = os.path.join(STATE, "kind-" + req) - if os.path.exists(path): - with open(path, encoding="utf-8") as f: - return f.read().strip() - return "control" +def read_state(name, default=""): + path = os.path.join(STATE, name) + if not os.path.exists(path): + return default + with open(path, encoding="utf-8") as f: + return f.read() + + +def write_state(name, text): + with open(os.path.join(STATE, name), "w", encoding="utf-8") as f: + f.write(text) + + +def build_parser(): + parser = argparse.ArgumentParser(prog="cape") + sub = parser.add_subparsers(dest="verb", required=True) + run = sub.add_parser("run") + # Mirrors the real run parser flag for flag: --pool/--user/--task-id + # required, NO --workspace, REMAINDER workload. Sharing argparse with + # the real CLI makes rejection behavior (rc=2, usage on stderr) + # match by construction. + run.add_argument("--controller-url") + run.add_argument("--token") + run.add_argument("--pool", required=True) + run.add_argument("--user", required=True) + run.add_argument("--task-id", required=True) + run.add_argument("--image") + run.add_argument("--profile") + run.add_argument("--gpus", type=int, default=0) + run.add_argument("--cpu-cores", type=int, default=1) + run.add_argument("--memory-gb", type=int, default=1) + run.add_argument("--max-duration-seconds", type=float) + run.add_argument("--runtime-adapter") + run.add_argument("--hard-gpu-enforcement", action="store_true") + run.add_argument("--port", type=int, action="append", default=[]) + run.add_argument("--locality") + run.add_argument("--gpu-mode", default="whole") + run.add_argument("--isolation-policy", default="default") + run.add_argument("--cwd") + run.add_argument("--env", action="append", default=[]) + run.add_argument("--bind", action="append", default=[]) + run.add_argument("--log-path") + run.add_argument("--artifact-path", action="append", default=[]) + run.add_argument("--session-key") + run.add_argument("--wait", action="store_true") + run.add_argument("--timeout", type=float, default=30.0) + run.add_argument("workload_command", nargs=argparse.REMAINDER) + for name in ("status", "logs", "cancel"): + p = sub.add_parser(name) + p.add_argument("request_id") + p.add_argument("--controller-url") + p.add_argument("--token") + if name == "cancel": + p.add_argument("--reason", default="cancelled") + return parser + + +def request_state(req): + if read_state("cancelled-" + req): + return "CANCELLED" + if read_state("kind-" + req).strip() == "server": + return os.environ.get("FAKE_CAPE_SERVER_STATE", "RUNNING") + return "COMPLETED" + + +def do_run(args): + forced_stderr = os.environ.get("FAKE_CAPE_RUN_STDERR") + if forced_stderr: + sys.stderr.write(forced_stderr + "\\n") + sys.exit(1) + req = "req-%06d" % bump("run") + workload = list(args.workload_command) + if workload and workload[0] == "--": + workload = workload[1:] + if not workload: + sys.stderr.write("fake cape: empty workload command\\n") + sys.exit(1) + kind = "server" if "bootstrap.sh" in " ".join(workload) else "control" + write_state("kind-" + req, kind) + if kind == "server" and os.environ.get("FAKE_CAPE_EXEC_WORKLOAD", "1") != "0": + # Model the node agent: run the workload with the request's env + # applied, in a stubbed node environment (health-server port, + # loopback `hostname`). The boot script really writes the + # endpoint marker into the meta dir the provider polls. + env = dict(os.environ) + for item in args.env: + key, _, value = item.partition("=") + env[key] = value + if os.environ.get("FAKE_CAPE_PORT"): + env["AGENTIX_BIND_PORT"] = os.environ["FAKE_CAPE_PORT"] + if os.environ.get("FAKE_CAPE_STUB_PATH"): + env["PATH"] = os.environ["FAKE_CAPE_STUB_PATH"] + os.pathsep + env.get("PATH", "") + proc = subprocess.run(workload, env=env, capture_output=True, text=True, timeout=60) + write_state("out-" + req, proc.stdout) + write_state("err-" + req, proc.stderr) + print(req) + + +def do_status(args): + fails = int(os.environ.get("FAKE_CAPE_STATUS_FAILURES", "0")) + if bump("status") <= fails: + sys.stderr.write("fake cape: transient controller error\\n") + sys.exit(1) + req = args.request_id + state = request_state(req) + record = {"state": state, "request_id": req} + if state == "CANCELLED": + record["exit_code"] = 124 + elif state in TERMINAL: + record["exit_code"] = 0 + print(json.dumps(record)) + + +def do_logs(args): + req = args.request_id + # Real contract: stdout/stderr stay EMPTY until the command exits; + # the node agent reports output once, on terminal states only. + if request_state(req) in TERMINAL: + sys.stdout.write(read_state("out-" + req, "fake-terminal-stdout\\n")) + sys.stderr.write(read_state("err-" + req, "fake-terminal-stderr\\n")) + + +def do_cancel(args): + req = args.request_id + if os.environ.get("FAKE_CAPE_CANCEL_404"): + payload = {"status": 404, "error": {"detail": "unknown request: " + req}} + sys.stderr.write(json.dumps(payload) + "\\n") + sys.exit(1) + rc = int(os.environ.get("FAKE_CAPE_CANCEL_RC", "0")) + if rc: + sys.stderr.write("fake cape: cancel rejected\\n") + sys.exit(rc) + write_state("cancelled-" + req, "1") + print(json.dumps({"state": "CANCELLED", "request_id": req, "exit_code": None})) def main(): argv = sys.argv[1:] log(argv) - verb = argv[0] if argv else "" - if verb == "run": - stderr = os.environ.get("FAKE_CAPE_RUN_STDERR") - if stderr: - sys.stderr.write(stderr + "\\n") - sys.exit(1) - n = bump("run") - req = "req-fake-%d" % n - sep = argv.index("--") - workload = " ".join(argv[sep + 1 :]) - kind = "server" if "bootstrap.sh" in workload else "control" - with open(os.path.join(STATE, "kind-" + req), "w", encoding="utf-8") as f: - f.write(kind) - print(req) - elif verb == "status": - req = argv[1] - fails = int(os.environ.get("FAKE_CAPE_STATUS_FAILURES", "0")) - if bump("status") <= fails: - sys.stderr.write("fake cape: transient controller error\\n") - sys.exit(1) - if os.path.exists(os.path.join(STATE, "cancelled-" + req)): - print(json.dumps({"state": "CANCELLED", "exit_code": 124, "request_id": req})) - elif req_kind(req) == "server": - state = os.environ.get("FAKE_CAPE_SERVER_STATE", "RUNNING") - print(json.dumps({"state": state, "request_id": req})) - else: - print(json.dumps({"state": "COMPLETED", "exit_code": 0, "request_id": req})) - elif verb == "logs": - req = argv[1] - n = bump("logs") - lf = int(os.environ.get("FAKE_CAPE_LOGS_FAILURES", "0")) - if n <= lf: - sys.stderr.write("fake cape: transient log fetch error\\n") - sys.exit(1) - print("booting runtime...") - print("GPU 0") - le = int(os.environ.get("FAKE_CAPE_LOGS_EMPTY", "0")) - if n - lf > le: - print("AGENTIX_ENDPOINT 127.0.0.1 %s" % os.environ["FAKE_CAPE_PORT"]) - elif verb == "cancel": - req = argv[1] - rc = int(os.environ.get("FAKE_CAPE_CANCEL_RC", "0")) - if rc: - sys.stderr.write("fake cape: cancel rejected\\n") - sys.exit(rc) - with open(os.path.join(STATE, "cancelled-" + req), "w", encoding="utf-8"): - pass - else: - sys.stderr.write("fake cape: unsupported verb %r\\n" % verb) - sys.exit(2) + args = build_parser().parse_args(argv) + {"run": do_run, "status": do_status, "logs": do_logs, "cancel": do_cancel}[args.verb](args) if __name__ == "__main__": @@ -180,14 +277,25 @@ def health_port(): def cape_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, health_port: int) -> dict[str, Any]: state = tmp_path / "state" state.mkdir() + meta_root = tmp_path / "meta" + meta_root.mkdir() + workspace_root = tmp_path / "ws" log = tmp_path / "cape.log.jsonl" fake = tmp_path / "fake-bin" / "cape" fake.parent.mkdir() fake.write_text(f"#!{sys.executable}\n{_FAKE_CAPE_BODY}") fake.chmod(0o755) + # Stub `hostname` so the boot script's endpoint marker resolves to + # loopback regardless of the test host's real hostname/-i support. + stub = tmp_path / "stub-bin" + stub.mkdir() + hostname = stub / "hostname" + hostname.write_text('#!/bin/sh\necho "127.0.0.1 10.0.0.5"\n') + hostname.chmod(0o755) monkeypatch.setenv("FAKE_CAPE_LOG", str(log)) monkeypatch.setenv("FAKE_CAPE_STATE", str(state)) monkeypatch.setenv("FAKE_CAPE_PORT", str(health_port)) + monkeypatch.setenv("FAKE_CAPE_STUB_PATH", str(stub)) monkeypatch.setenv("CAPE_TOKEN", "unit-test-token") for var in ( "CAPE_TOKEN_FILE", @@ -196,18 +304,29 @@ def cape_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, health_port: int) "FAKE_CAPE_SERVER_STATE", "FAKE_CAPE_RUN_STDERR", "FAKE_CAPE_CANCEL_RC", - "FAKE_CAPE_LOGS_EMPTY", - "FAKE_CAPE_LOGS_FAILURES", + "FAKE_CAPE_CANCEL_404", + "FAKE_CAPE_EXEC_WORKLOAD", "FAKE_CAPE_STATUS_FAILURES", ): monkeypatch.delenv(var, raising=False) - return {"binary": fake, "log": log, "state": state, "port": health_port} + return { + "binary": fake, + "log": log, + "state": state, + "port": health_port, + "meta_root": meta_root, + "workspace_root": workspace_root, + } def _provider(cape_env: dict[str, Any], **overrides: Any) -> CapeProvider: settings: dict[str, Any] = { "binary": str(cape_env["binary"]), "controller_url": "http://cape-controller.test:9000", + "pool": "unit-pool", + "user": "unit-user", + "meta_root": str(cape_env["meta_root"]), + "workspace_root": str(cape_env["workspace_root"]), "poll_interval_seconds": 0.05, "create_timeout_seconds": 30.0, "client_timeout_seconds": 30.0, @@ -243,75 +362,92 @@ def _flags(argv: list[str], name: str) -> list[str]: return [argv[i + 1] for i, tok in enumerate(argv) if tok == name] +def _meta_dir(cape_env: dict[str, Any], sandbox_id: str) -> Path: + return Path(cape_env["meta_root"]) / sandbox_id + + # ── create / run face ───────────────────────────────────────────────────── -async def test_create_returns_sandbox_and_emits_assumed_run_face(cape_env: dict[str, Any]) -> None: +async def test_create_returns_sandbox_and_emits_real_run_face(cape_env: dict[str, Any]) -> None: provider = _provider(cape_env) config = _sandbox_config(env={"HF_HOME": "/tmp/hf"}, resource=SandboxResource(gpu=2)) sandbox = await provider.create(config) + meta_dir = _meta_dir(cape_env, str(sandbox.sandbox_id)) try: assert sandbox.status == "running" assert sandbox.runtime_url == f"http://127.0.0.1:{cape_env['port']}" runs = _log_entries(cape_env, verb="run") # One sandbox = one request; discovery must not submit a second - # same-session command (the assumed session model is serial). + # same-session command (the session model is serial). assert len(runs) == 1 argv = runs[0]["argv"] assert _flag(argv, "--controller-url") == "http://cape-controller.test:9000" assert _flag(argv, "--token") == "unit-test-token" + assert _flag(argv, "--pool") == "unit-pool" + assert _flag(argv, "--user") == "unit-user" + # --task-id is required by the real parser; the provider passes + # the sandbox id. + assert _flag(argv, "--task-id") == str(sandbox.sandbox_id) assert _flag(argv, "--session-key") == f"agentix-{sandbox.sandbox_id}" - assert _flag(argv, "--workspace") == f"/workspace/{sandbox.sandbox_id}" - assert _flag(argv, "--cwd") == f"/workspace/{sandbox.sandbox_id}" + # The real CLI has no --workspace flag, and --cwd would point at + # a directory nothing creates — the boot script mkdir+cd's. + assert "--workspace" not in argv + assert "--cwd" not in argv assert _flag(argv, "--image") == "docker://task-image:1" assert _flag(argv, "--gpus") == "2" assert _flag(argv, "--gpu-mode") == "whole" assert _flag(argv, "--isolation-policy") == "default" assert _flag(argv, "--max-duration-seconds") == "14400" - assert _flags(argv, "--bind") == ["/mnt/shared/bundles/sha256-abc:/nix:ro"] + assert _flags(argv, "--bind") == [ + "/mnt/shared/bundles/sha256-abc:/nix:ro", + f"{meta_dir}:/agentix-meta:rw", + ] env_args = _flags(argv, "--env") assert "AGENTIX_BIND_PORT=8710" in env_args assert "HF_HOME=/tmp/hf" in env_args # Optional flags absent from a default config. - for absent in ("--pool", "--user", "--cpu-cores", "--memory-gb", "--runtime-adapter"): + for absent in ("--cpu-cores", "--memory-gb", "--runtime-adapter"): assert absent not in argv - # Workload sits after the `--` separator: print the endpoint - # marker to stdout, then exec the bundle entry point. + # Workload sits after the `--` separator: create the workspace, + # write the endpoint marker file, then exec the bundle entry point. workload = argv[argv.index("--") + 1 :] assert workload[:2] == ["sh", "-c"] - assert "AGENTIX_ENDPOINT" in workload[2] - assert workload[2].endswith("exec /nix/runtime/bootstrap.sh") - assert "mkdir" not in workload[2] - assert ".agentix-endpoint" not in workload[2] - # Discovery consumed status + logs of the server request only. - assert _log_entries(cape_env, verb="logs") + script = workload[2] + assert f"mkdir -p {cape_env['workspace_root']}/{sandbox.sandbox_id}" in script + assert "AGENTIX_ENDPOINT" in script + assert "/agentix-meta" in script + assert "/nix/runtime/bootstrap.sh" in script + # The boot script really ran and wrote the marker the provider + # discovered; `cape logs` was never needed (real CAPE returns + # empty logs while the request runs). + assert (meta_dir / "endpoint").is_file() + assert _log_entries(cape_env, verb="logs") == [] finally: await provider.delete(sandbox.sandbox_id) + assert not meta_dir.exists() # delete removes the per-sandbox meta dir async def test_optional_flags_emitted_when_configured(cape_env: dict[str, Any]) -> None: provider = _provider( cape_env, - pool="pool-a", - user="alice", runtime_adapter="apptainer", extra_binds=["/data:/data:rw"], - isolation_policy="strict", + isolation_policy="host-stateless", ) config = _sandbox_config(resource=SandboxResource(cpu=2.5, memory="16g", gpu=1)) sandbox = await provider.create(config) try: argv = _log_entries(cape_env, verb="run")[0]["argv"] - assert _flag(argv, "--pool") == "pool-a" - assert _flag(argv, "--user") == "alice" assert _flag(argv, "--cpu-cores") == "3" # ceil(2.5) assert _flag(argv, "--memory-gb") == "16" assert _flag(argv, "--gpus") == "1" assert _flag(argv, "--runtime-adapter") == "apptainer" - assert _flag(argv, "--isolation-policy") == "strict" + assert _flag(argv, "--isolation-policy") == "host-stateless" assert _flags(argv, "--bind") == [ "/mnt/shared/bundles/sha256-abc:/nix:ro", + f"{_meta_dir(cape_env, str(sandbox.sandbox_id))}:/agentix-meta:rw", "/data:/data:rw", ] finally: @@ -330,6 +466,63 @@ async def test_config_resource_defaults_used_when_resource_unset(cape_env: dict[ await provider.delete(sandbox.sandbox_id) +async def test_missing_pool_user_meta_root_fail_fast(cape_env: dict[str, Any]) -> None: + # The real `cape run` exits 2 (argparse) without --pool/--user, and + # discovery cannot work without meta_root: fail fast, before any CLI + # call or meta dir creation. + for missing in ("pool", "user", "meta_root"): + provider = _provider(cape_env, **{missing: None}) + with pytest.raises(RuntimeError, match=missing): + await provider.create(_sandbox_config()) + assert provider._inflight_ports == set() + assert _log_entries(cape_env) == [] + assert list(Path(cape_env["meta_root"]).iterdir()) == [] + + +# ── fake-CLI contract (documents the real parser behavior) ─────────────── + + +def test_fake_cape_rejects_workspace_flag_and_requires_pool_user_task_id( + cape_env: dict[str, Any], +) -> None: + binary = str(cape_env["binary"]) + # The legacy (pre-verification) run face carried --workspace: the + # real parser rejects it with an argparse usage error, rc=2. + legacy = [ + binary, "run", "--pool", "p", "--user", "u", "--task-id", "t", + "--workspace", "/ws", "--image", "img", "--", "sh", "-c", "true", + ] # fmt: skip + proc = subprocess.run(legacy, capture_output=True, text=True) + assert proc.returncode == 2 + assert "--workspace" in proc.stderr + # --pool/--user/--task-id are required=True in the real parser. + proc = subprocess.run( + [binary, "run", "--image", "img", "--", "sh", "-c", "true"], + capture_output=True, + text=True, + ) + assert proc.returncode == 2 + for flag in ("--pool", "--user", "--task-id"): + assert flag in proc.stderr + + +async def test_fake_cape_logs_empty_while_running_populated_when_terminal( + cape_env: dict[str, Any], +) -> None: + provider = _provider(cape_env) + sandbox = await provider.create(_sandbox_config()) + binary = str(cape_env["binary"]) + logs_argv = [binary, "logs", "req-000001", "--controller-url", "u", "--token", "t"] + # Real contract: rc=0 but EMPTY output while the request runs. + proc = subprocess.run(logs_argv, capture_output=True, text=True) + assert (proc.returncode, proc.stdout, proc.stderr) == (0, "", "") + await provider.delete(sandbox.sandbox_id) + # Terminal (cancelled) → the workload's recorded output is available. + proc = subprocess.run(logs_argv, capture_output=True, text=True) + assert proc.returncode == 0 + assert proc.stderr # the boot script's exec failure landed on stderr + + # ── multi-sandbox ───────────────────────────────────────────────────────── @@ -346,15 +539,20 @@ async def test_two_sandboxes_get_distinct_sessions_ports_and_cancels( assert len(runs) == 2 keys = {_flag(r["argv"], "--session-key") for r in runs} assert keys == {f"agentix-{sb1.sandbox_id}", f"agentix-{sb2.sandbox_id}"} + task_ids = {_flag(r["argv"], "--task-id") for r in runs} + assert task_ids == {str(sb1.sandbox_id), str(sb2.sandbox_id)} ports = { e for r in runs for e in _flags(r["argv"], "--env") if e.startswith("AGENTIX_BIND_PORT=") } assert ports == {"AGENTIX_BIND_PORT=8710", "AGENTIX_BIND_PORT=8711"} - # Deleting sandbox 1 cancels only its own request. + # Deleting sandbox 1 cancels only its own request and removes only + # its own meta dir. await provider.delete(sb1.sandbox_id) cancels = _log_entries(cape_env, verb="cancel") - assert [c["argv"][1] for c in cancels] == ["req-fake-1"] + assert [c["argv"][1] for c in cancels] == ["req-000001"] + assert not _meta_dir(cape_env, str(sb1.sandbox_id)).exists() + assert _meta_dir(cape_env, str(sb2.sandbox_id)).is_dir() info = await provider.get(sb2.sandbox_id) assert info.status == "running" @@ -373,9 +571,10 @@ async def test_delete_cancels_server_request_and_is_idempotent(cape_env: dict[st cancels = _log_entries(cape_env, verb="cancel") assert cancels, "no cape cancel recorded" - assert cancels[-1]["argv"][1] == "req-fake-1" + assert cancels[-1]["argv"][1] == "req-000001" assert _flag(cancels[-1]["argv"], "--reason") == "agentix_delete" assert provider._inflight_ports == set() + assert not _meta_dir(cape_env, str(sandbox.sandbox_id)).exists() # Second delete of the same (now unknown) id is a no-op: no raise, # no extra cancel. @@ -390,12 +589,14 @@ async def test_delete_keeps_bookkeeping_when_cancel_fails_then_retries( ) -> None: provider = _provider(cape_env) sandbox = await provider.create(_sandbox_config()) + meta_dir = _meta_dir(cape_env, str(sandbox.sandbox_id)) monkeypatch.setenv("FAKE_CAPE_CANCEL_RC", "1") await provider.delete(sandbox.sandbox_id) # must not raise - # Cancel was not confirmed: the record (and its port) stay so a - # later delete() can retry instead of silently leaking the lease. + # Cancel was not confirmed: the record (and its port and meta dir) + # stay so a later delete() can retry instead of leaking the lease. assert sandbox.sandbox_id in provider._sandboxes + assert meta_dir.is_dir() info = await provider.get(sandbox.sandbox_id) assert info.status == "running" @@ -403,38 +604,73 @@ async def test_delete_keeps_bookkeeping_when_cancel_fails_then_retries( await provider.delete(sandbox.sandbox_id) assert provider._sandboxes == {} assert provider._inflight_ports == set() + assert not meta_dir.exists() + + +async def test_delete_treats_cancel_404_as_already_gone( + cape_env: dict[str, Any], monkeypatch: pytest.MonkeyPatch +) -> None: + # A journal-less controller restart forgets the request: cancel then + # fails with the CLI's stderr JSON `{"status": 404, ...}` and rc=1. + # That request is gone — bookkeeping must not stick forever. + provider = _provider(cape_env) + sandbox = await provider.create(_sandbox_config()) + meta_dir = _meta_dir(cape_env, str(sandbox.sandbox_id)) + + monkeypatch.setenv("FAKE_CAPE_CANCEL_404", "1") + await provider.delete(sandbox.sandbox_id) + assert provider._sandboxes == {} + assert provider._inflight_ports == set() + assert not meta_dir.exists() # ── create failure / cancellation paths ─────────────────────────────────── -async def test_create_failure_cancels_request_and_clears_bookkeeping( +async def test_terminal_state_fails_fast_with_terminal_logs( cape_env: dict[str, Any], monkeypatch: pytest.MonkeyPatch ) -> None: + # The workload dies before writing the marker: discovery must fail + # fast on the terminal state and enrich the error with `cape logs` + # output (populated once the request is terminal). + monkeypatch.setenv("FAKE_CAPE_EXEC_WORKLOAD", "0") monkeypatch.setenv("FAKE_CAPE_SERVER_STATE", "FAILED") provider = _provider(cape_env) - with pytest.raises(RuntimeError, match="FAILED"): + with pytest.raises(RuntimeError, match="terminal state FAILED") as excinfo: await provider.create(_sandbox_config()) + text = str(excinfo.value) + assert "fake-terminal-stdout" in text + assert "fake-terminal-stderr" in text cancels = _log_entries(cape_env, verb="cancel") assert cancels, "failed create must attempt a cancel" - assert cancels[-1]["argv"][1] == "req-fake-1" + assert cancels[-1]["argv"][1] == "req-000001" assert _flag(cancels[-1]["argv"], "--reason") == "agentix_create_failed" assert provider._sandboxes == {} assert provider._inflight_ports == set() + assert list(Path(cape_env["meta_root"]).iterdir()) == [] -async def test_discovery_retries_until_marker_appears( +async def test_discovery_tolerates_transient_status_failures( cape_env: dict[str, Any], monkeypatch: pytest.MonkeyPatch ) -> None: - # Marker withheld for the first 2 logs calls, plus 2 transient status - # failures: create must retry through both and still succeed. - monkeypatch.setenv("FAKE_CAPE_LOGS_EMPTY", "2") + # Marker withheld (workload not executed) plus 2 transient status + # failures: discovery must retry through the blips and still succeed + # once the marker file appears. + monkeypatch.setenv("FAKE_CAPE_EXEC_WORKLOAD", "0") monkeypatch.setenv("FAKE_CAPE_STATUS_FAILURES", "2") provider = _provider(cape_env) - sandbox = await provider.create(_sandbox_config()) + task = asyncio.ensure_future(provider.create(_sandbox_config())) + while len(_log_entries(cape_env, verb="status")) < 2: + assert not task.done() + await asyncio.sleep(0.02) + # Write the marker the way the workload's boot script would. + [meta_dir] = [p for p in Path(cape_env["meta_root"]).iterdir() if p.is_dir()] + (meta_dir / "endpoint").write_text(f"AGENTIX_ENDPOINT 127.0.0.1 {cape_env['port']}\n") + sandbox = await task try: - assert len(_log_entries(cape_env, verb="logs")) >= 3 + assert sandbox.status == "running" + assert len(_log_entries(cape_env, verb="status")) >= 2 finally: await provider.delete(sandbox.sandbox_id) @@ -442,6 +678,7 @@ async def test_discovery_retries_until_marker_appears( async def test_discovery_fails_after_consecutive_transient_failures( cape_env: dict[str, Any], monkeypatch: pytest.MonkeyPatch ) -> None: + monkeypatch.setenv("FAKE_CAPE_EXEC_WORKLOAD", "0") monkeypatch.setenv("FAKE_CAPE_STATUS_FAILURES", "100000") provider = _provider(cape_env, transient_failure_limit=3) with pytest.raises(RuntimeError, match="3 consecutive"): @@ -449,19 +686,24 @@ async def test_discovery_fails_after_consecutive_transient_failures( cancels = _log_entries(cape_env, verb="cancel") assert cancels and _flag(cancels[-1]["argv"], "--reason") == "agentix_create_failed" assert provider._sandboxes == {} + assert list(Path(cape_env["meta_root"]).iterdir()) == [] async def test_discovery_times_out_when_marker_never_appears( cape_env: dict[str, Any], monkeypatch: pytest.MonkeyPatch ) -> None: - monkeypatch.setenv("FAKE_CAPE_LOGS_EMPTY", "100000") + monkeypatch.setenv("FAKE_CAPE_EXEC_WORKLOAD", "0") provider = _provider(cape_env, create_timeout_seconds=1.0) with pytest.raises(TimeoutError, match="did not publish its endpoint"): await provider.create(_sandbox_config()) cancels = _log_entries(cape_env, verb="cancel") assert cancels and _flag(cancels[-1]["argv"], "--reason") == "agentix_create_failed" + # Logs are useless on a still-running request (empty by contract) — + # the timeout path must not have fetched them. + assert _log_entries(cape_env, verb="logs") == [] assert provider._sandboxes == {} assert provider._inflight_ports == set() + assert list(Path(cape_env["meta_root"]).iterdir()) == [] async def test_health_timeout_fails_create_and_cancels( @@ -482,16 +724,39 @@ async def test_health_timeout_fails_create_and_cancels( assert cancels and _flag(cancels[-1]["argv"], "--reason") == "agentix_create_failed" assert provider._sandboxes == {} assert provider._inflight_ports == set() + assert list(Path(cape_env["meta_root"]).iterdir()) == [] + + +async def test_marker_written_but_request_terminal_fails_with_logs( + cape_env: dict[str, Any], monkeypatch: pytest.MonkeyPatch +) -> None: + # The workload wrote its marker and then crashed (dead endpoint, + # terminal request): the health wait must fail fast with the + # request's logs instead of probing for the whole budget. + import socket + + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("127.0.0.1", 0)) + dead_port = s.getsockname()[1] + monkeypatch.setenv("FAKE_CAPE_PORT", str(dead_port)) + monkeypatch.setenv("FAKE_CAPE_SERVER_STATE", "FAILED") + provider = _provider(cape_env, create_timeout_seconds=30.0) + with pytest.raises(RuntimeError, match="terminal state FAILED") as excinfo: + await provider.create(_sandbox_config()) + assert "workload stderr" in str(excinfo.value) + assert provider._sandboxes == {} + assert list(Path(cape_env["meta_root"]).iterdir()) == [] async def test_cancelled_create_cancels_submitted_request( cape_env: dict[str, Any], monkeypatch: pytest.MonkeyPatch ) -> None: - # Keep discovery spinning so the cancellation lands mid-create. - monkeypatch.setenv("FAKE_CAPE_LOGS_EMPTY", "100000") + # Keep discovery spinning (marker never appears) so the cancellation + # lands mid-create. + monkeypatch.setenv("FAKE_CAPE_EXEC_WORKLOAD", "0") provider = _provider(cape_env) task = asyncio.ensure_future(provider.create(_sandbox_config())) - while not _log_entries(cape_env, verb="logs"): + while not _log_entries(cape_env, verb="status"): await asyncio.sleep(0.05) task.cancel() with pytest.raises(asyncio.CancelledError): @@ -504,6 +769,7 @@ async def test_cancelled_create_cancels_submitted_request( } assert provider._sandboxes == {} assert provider._inflight_ports == set() + assert list(Path(cape_env["meta_root"]).iterdir()) == [] # ── get ─────────────────────────────────────────────────────────────────── @@ -629,7 +895,7 @@ def test_memory_gb_mapping() -> None: def test_parse_endpoint_requires_marker_prefix() -> None: - marker = "booting runtime...\nGPU 0\nAGENTIX_ENDPOINT 10.0.0.5 8710\n" + marker = "AGENTIX_ENDPOINT 10.0.0.5 8710\n" assert _parse_endpoint(marker) == ("10.0.0.5", 8710) # Two-token noise ("GPU 0") must never be misread as an endpoint. assert _parse_endpoint("GPU 0\n") is None @@ -638,23 +904,34 @@ def test_parse_endpoint_requires_marker_prefix() -> None: assert _parse_endpoint("AGENTIX_ENDPOINT 10.0.0.5\n") is None -def test_boot_script_marker_roundtrip(tmp_path: Path) -> None: - """Run the generated boot script under a real `sh` (with a stubbed - multi-IP `hostname`) and feed its stdout to `_parse_endpoint` — - the producer/consumer pair must agree.""" +def test_boot_script_writes_marker_and_creates_workspace(tmp_path: Path) -> None: + """Run the generated boot script under a real `sh -c` (with a stubbed + multi-IP `hostname` and a tmp meta dir) and feed the marker file it + writes to `_parse_endpoint` — the producer/consumer pair must agree. + """ stub = tmp_path / "stub-bin" stub.mkdir() hostname = stub / "hostname" hostname.write_text("#!/bin/sh\necho '10.0.0.5 172.17.0.1'\n") hostname.chmod(0o755) + workspace = tmp_path / "ws" / "cape-x" + meta_dir = tmp_path / "meta" / "cape-x" + meta_dir.mkdir(parents=True) + bundle = tmp_path / "bundle" # deliberately has no runtime/bootstrap.sh + script = _boot_script(workspace=str(workspace), meta_dir=str(meta_dir), bundle=str(bundle)) env = { "PATH": f"{stub}:{os.environ.get('PATH', '/usr/bin:/bin')}", "AGENTIX_BIND_PORT": "9999", } - # `exec /nix/runtime/bootstrap.sh` fails in the test environment — - # the marker must already be on stdout by then. - proc = subprocess.run(["sh", "-c", _boot_script()], env=env, capture_output=True, text=True) - assert _parse_endpoint(proc.stdout) == ("10.0.0.5", 9999) + proc = subprocess.run(["sh", "-c", script], env=env, capture_output=True, text=True) + # The final exec of the (absent) bundle bootstrap fails — but only + # after the workspace was created and the marker was written. + assert proc.returncode != 0 + assert workspace.is_dir() + marker_file = meta_dir / "endpoint" + assert marker_file.is_file() + assert _parse_endpoint(marker_file.read_text()) == ("10.0.0.5", 9999) + assert proc.stdout == "" # the marker goes to the meta dir, not stdout # ── registry contract ───────────────────────────────────────────────────── @@ -662,9 +939,12 @@ def test_boot_script_marker_roundtrip(tmp_path: Path) -> None: def test_zero_arg_constructor_for_plugin_registry() -> None: # The plugin registry instantiates providers with `cls()` — every - # config field must be optional or defaulted. + # config field must be optional or defaulted. pool/user/meta_root + # are then checked at first use. provider = CapeProvider() assert isinstance(provider, SandboxProvider) assert provider.config.workspace_root == "/workspace" assert provider.config.url_template == "http://{host}:{port}" assert provider.config.runtime_port_base == 8710 + assert provider.config.pool is None + assert provider.config.meta_root is None