From fce5033a8b0ff24e306ec7163ceaee50fdfc6d1f Mon Sep 17 00:00:00 2001 From: Lingrui Mei Date: Thu, 23 Jul 2026 12:16:51 +0800 Subject: [PATCH] provider: verify CAPE contract against real CLI and fix run face + endpoint discovery The CAPE provider shipped with an assumed CLI contract. Every verb, flag, and output shape has now been verified item by item against the CAPE source, and the mismatches are fixed. What matched: the run/status/logs/cancel verb set; --controller-url/ --token on every verb; the run flag table (--image, --gpus, --cpu-cores, --memory-gb, --gpu-mode whole, --isolation-policy, --runtime-adapter, --max-duration-seconds, --session-key, --bind, --env, `-- `); "stdout is exactly one request id" without --wait (real ids are req-%06d; the regex is tightened to req-\d{6,}); the status JSON on stdout with the terminal-state set {COMPLETED, FAILED, CANCELLED, EXPIRED, LOST, INFEASIBLE}, a request_id echo, and no stdout/stderr fields; cancel --reason with idempotent terminal no-op; token passable only as --token argv (no env/file support in the CLI); the serial session model. What did not match (submit-fatal argv bugs): --task-id is required by the real parser and was never passed (now = sandbox id); --pool/--user are required but were optional (now checked at first use, fail-fast); --workspace does not exist (argparse rc=2 "unrecognized arguments"); --cwd pointed at a directory no component creates, so the command would die at process start. The architectural finding: CAPE has no streaming logs. `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), and there is no native endpoint query. The old stdout-marker + logs-polling discovery could therefore never see the marker of a never-exiting runtime server: every create() would burn the full 600s budget and fail. Discovery is redesigned around a marker file through a rw bind: a new required meta_root config names a submitter-and-node-visible directory (same shared-FS assumption as bundle); each sandbox gets / bound at /agentix-meta; the boot script now does `mkdir -p && cd `, writes `AGENTIX_ENDPOINT ` to /agentix-meta/endpoint, and execs the bundle bootstrap; the provider polls the host side of the marker file interleaved with `cape status` terminal-state checks, and fetches logs only on terminal states (where they are populated) for diagnostics. delete() removes the meta dir after a confirmed cancel, and treats a controller 404 on cancel as already-gone so bookkeeping cannot stick forever after a journal-less controller restart. Docs now state the host-network constraint: the client-assigned-port scheme needs apptainer/bubblewrap (host network stack); CAPE's podman adapter has no host networking, so podman-only pools cannot work. The fake cape in the tests now models the real contract (argparse- mirrored run parser rejecting --workspace and requiring --pool/--user/ --task-id with rc=2; req-%06d ids; logs empty while RUNNING; cancel 404 stderr JSON) and really executes the boot script under sh -c with a stubbed node environment, so discovery is tested against a marker file the actual script wrote. Validated against a real local CAPE deployment (controller + node join, local-process adapter, auth enabled): create() submitted a request the real parser accepted, discovery found the marker while real `cape logs` returned empty, GET /health answered 200 on the discovered URL, and delete() drove the request to CANCELLED, reaped the workload process group, and removed the meta dir - 11/11 e2e assertions passed. The local-process adapter ignores bind specs, so that run exercises the boot script's documented host-path fallbacks; container-adapter bind behavior still needs a Linux pool. Co-authored-by: Claude Fable 5 --- plugins/providers/cape/README.md | 169 ++++-- .../providers/cape/agentix/provider/cape.py | 499 ++++++++++------ plugins/providers/cape/pyproject.toml | 2 +- .../cape/tests/test_cape_provider.py | 536 +++++++++++++----- 4 files changed, 859 insertions(+), 347 deletions(-) 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