diff --git a/src/inferrsi/harness/parallel_round.py b/src/inferrsi/harness/parallel_round.py new file mode 100644 index 000000000..d9a01553a --- /dev/null +++ b/src/inferrsi/harness/parallel_round.py @@ -0,0 +1,469 @@ +"""Prepare, submit and measure one offline parallel round. + +Directory diffs stand in for git clones. ``close`` and incumbent advancement are +not implemented. A synthetic latency number is not a campaign verdict. +""" + +import hashlib +import json +import os +import shutil +import stat +import subprocess +import time +from datetime import datetime, timezone +from fnmatch import fnmatch +from pathlib import Path + +from inferrsi.harness.round_budget import ( + Binding, + BudgetError, + Entry, + Genesis, + LaneKey, + Ledger, + admit_attempt, + advance, + empty_ledger, + ensure_open, + seconds_remaining, +) +from inferrsi.infra.artifacts import digest, write_json +from inferrsi.infra.confinement import ConfinementUnavailable, confined_execute +from inferrsi.infra.device_lease import DeviceLease, LeaseError + +PLANES = {"harness", "model", "infra"} + + +class RoundPlanError(ValueError): + """The frozen round plan is internally inconsistent.""" + + +def _match(path: str, patterns: list[str]) -> bool: + return any(fnmatch(path, pattern) for pattern in patterns) + + +def _files(root: Path) -> list[str]: + hashes, _blocked = _snapshot(root) + return sorted(hashes) + + +def _snapshot(root: Path) -> tuple[dict[str, str], list[str]]: + """Hash regular files without following links. Symlinks are reported, not read.""" + hashes: dict[str, str] = {} + blocked: list[str] = [] + root_fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY) + try: + _scan(root_fd, "", hashes, blocked) + finally: + os.close(root_fd) + return hashes, blocked + + +def _scan(dirfd: int, prefix: str, hashes: dict[str, str], blocked: list[str]) -> None: + for name in os.listdir(dirfd): + if name in {".", "..", ".git"}: + continue + relative = f"{prefix}/{name}" if prefix else name + try: + info = os.stat(name, dir_fd=dirfd, follow_symlinks=False) + except OSError: + blocked.append(relative) + continue + if stat.S_ISLNK(info.st_mode): + blocked.append(relative) + continue + if stat.S_ISDIR(info.st_mode): + try: + child = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=dirfd) + except OSError: + blocked.append(relative) + continue + try: + _scan(child, relative, hashes, blocked) + finally: + os.close(child) + continue + if not stat.S_ISREG(info.st_mode): + blocked.append(relative) + continue + try: + file_fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=dirfd) + except OSError: + blocked.append(relative) + continue + try: + hasher = hashlib.sha256() + while True: + block = os.read(file_fd, 1024 * 1024) + if not block: + break + hasher.update(block) + hashes[relative] = hasher.hexdigest() + finally: + os.close(file_fd) + + +def _read_nofollow(root: Path, relative: str) -> bytes | None: + """Read one regular file through no-follow parent components, or return None.""" + dirfd = os.open(root, os.O_RDONLY | os.O_DIRECTORY) + try: + parts = relative.split("/") + for part in parts[:-1]: + try: + child = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=dirfd) + except OSError: + return None + os.close(dirfd) + dirfd = child + try: + file_fd = os.open(parts[-1], os.O_RDONLY | os.O_NOFOLLOW, dir_fd=dirfd) + except OSError: + return None + finally: + os.close(dirfd) + try: + chunks = [] + while True: + block = os.read(file_fd, 1024 * 1024) + if not block: + break + chunks.append(block) + return b"".join(chunks) + finally: + os.close(file_fd) + + +def _refusal(reason: str, **fields) -> dict: + body = {"submitted": False, "outside": [], "deleted": [], "rejected": [], "ticket": None} + body.update(fields) + body["reason"] = reason + return body + + +def _lane(plan: dict, lane_id: str) -> dict: + matches = [lane for lane in plan["lanes"] if lane["id"] == lane_id] + if len(matches) != 1: + raise RoundPlanError(f"unknown lane: {lane_id}") + return matches[0] + + +def prepare(spec: dict, incumbent: Path, round_dir: Path) -> dict: + """Validate lane scopes and write ``round-plan.json``. The digest covers the validated body.""" + if not isinstance(spec, dict): + raise RoundPlanError("round spec must be an object") + lanes = spec.get("lanes") + if not isinstance(lanes, list) or not lanes: + raise RoundPlanError("round spec needs at least one lane") + protected = spec.get("protected", []) + if not isinstance(protected, list): + raise RoundPlanError("protected paths must be a list of globs") + seen_entries: set[str] = set() + seen_ids: set[str] = set() + files = _files(incumbent) + normalized = [] + for lane in lanes: + lane_id = lane.get("id") + entry = lane.get("entry") + plane = lane.get("plane") + globs = lane.get("globs") + if not isinstance(lane_id, str) or not lane_id or lane_id in seen_ids: + raise RoundPlanError("each lane needs its own id") + if not isinstance(entry, str) or not entry: + raise RoundPlanError("each lane needs one search-plan entry") + if entry in seen_entries: + raise RoundPlanError(f"two lanes pursue {entry}") + if type(plane) is not str or plane not in PLANES: + raise RoundPlanError("a lane must name exactly one plane") + if not isinstance(globs, list) or not globs or any(type(item) is not str for item in globs): + raise RoundPlanError("a lane needs path globs") + if type(lane.get("cap")) is not int or type(lane.get("per_attempt_worst_case")) is not int: + raise RoundPlanError("cap and per-attempt cost must come from the plan") + if not isinstance(lane.get("deadline_utc"), str): + raise RoundPlanError("each lane needs a deadline") + if any(_match(path, globs) and _match(path, protected) for path in files): + raise RoundPlanError(f"{lane_id} reaches a protected path") + seen_ids.add(lane_id) + seen_entries.add(entry) + normalized.append(dict(lane)) + for index, lane in enumerate(normalized): + for other in normalized[index + 1 :]: + shared = set(lane["globs"]) & set(other["globs"]) + shared.update( + path + for path in files + if _match(path, lane["globs"]) and _match(path, other["globs"]) + ) + if shared: + raise RoundPlanError(f"{lane['id']} overlaps {other['id']}") + body = { + "round_id": spec["round_id"], + "device_id": spec["device_id"], + "incumbent_id": spec["incumbent_id"], + "protected": protected, + "build": list(spec.get("build", [])), + "trial": list(spec.get("trial", [])), + "lanes": normalized, + } + body["round_digest"] = digest(body) + round_dir.mkdir(parents=True, exist_ok=True) + write_json(round_dir / "round-plan.json", body) + return body + + +def submit( + round_dir: Path, lane_id: str, incumbent: Path, work: Path, *, now: datetime | None = None +) -> dict: + """Copy in-glob changes to the outbox. The same manifest always returns the same ticket.""" + moment = now or datetime.now(timezone.utc) + plan = json.loads((round_dir / "round-plan.json").read_text(encoding="utf-8")) + lane = _lane(plan, lane_id) + try: + ensure_open(lane["deadline_utc"], moment) + except BudgetError as exc: + return _refusal(str(exc)) + before, _incumbent_blocked = _snapshot(incumbent) + after, blocked = _snapshot(work) + changed = sorted( + path for path in set(before) | set(after) if before.get(path) != after.get(path) + ) + known = set(before) | set(after) | set(changed) + linked = [ + path + for path in blocked + if any(item == path or item.startswith(path + "/") for item in known) + ] + if linked: + return _refusal("symlink", rejected=sorted(linked)) + deleted = [path for path in changed if path not in after and _match(path, lane["globs"])] + if deleted: + return _refusal("deletion", deleted=deleted) + outside = [path for path in changed if not _match(path, lane["globs"])] + if outside or not changed: + return _refusal("unchanged" if not changed else "outside", outside=outside) + manifest = {path: after[path] for path in changed} + ticket = digest( + {"round_digest": plan["round_digest"], "lane_id": lane_id, "manifest": manifest} + )[:16] + destination = round_dir / "outbox" / ticket + if not destination.exists(): + destination.mkdir(parents=True) + try: + for path in changed: + payload = _read_nofollow(work, path) + if payload is None: + raise OSError(path) + target = destination / path + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(payload) + except OSError: + shutil.rmtree(destination) + return _refusal("symlink", rejected=list(changed)) + write_json( + destination / "manifest.json", + {"ticket": ticket, "lane_id": lane_id, "files": manifest, "deleted": []}, + ) + return {"submitted": True, "outside": [], "deleted": [], "rejected": [], "ticket": ticket} + + +def _ledger_path(round_dir: Path, lane_id: str) -> Path: + return round_dir / "ledgers" / f"{lane_id}.json" + + +def _load_ledger(round_dir: Path, plan: dict, lane: dict): + path = _ledger_path(round_dir, lane["id"]) + if not path.exists(): + return empty_ledger(plan["round_digest"], lane["id"], lane["cap"]) + raw = json.loads(path.read_text(encoding="utf-8")) + entries = tuple( + Entry( + Binding(item["plan_digest"], item["run_id"], item["worst_case"]), + item["phase"], + item["actual_cost"], + ) + for item in raw["entries"] + ) + return Ledger( + LaneKey(raw["round_digest"], raw["lane_id"]), + Genesis(raw["evidence"], raw["legacy_charge"]), + raw["cap"], + raw["revision"], + entries, + ) + + +def _save_ledger(round_dir: Path, ledger) -> None: + write_json( + _ledger_path(round_dir, ledger.key.lane_id), + { + "round_digest": ledger.key.round_digest, + "lane_id": ledger.key.lane_id, + "evidence": ledger.genesis.evidence, + "legacy_charge": ledger.genesis.legacy_charge, + "cap": ledger.cap, + "revision": ledger.revision, + "entries": [ + { + "plan_digest": entry.binding.plan_digest, + "run_id": entry.binding.run_id, + "worst_case": entry.binding.worst_case, + "phase": entry.phase, + "actual_cost": entry.actual_cost, + } + for entry in ledger.entries + ], + }, + ) + + +def _materialize(incumbent: Path, overlay: Path, destination: Path) -> None: + if destination.exists(): + shutil.rmtree(destination) + shutil.copytree(incumbent, destination, ignore=shutil.ignore_patterns("manifest.json")) + manifest = json.loads((overlay / "manifest.json").read_text(encoding="utf-8")) + for relative in manifest["files"]: + target = destination / relative + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(overlay / relative, target) + + +def _command(template: list[str], result: Path) -> tuple[str, ...]: + return tuple(part.replace("{result}", str(result)) for part in template) + + +def _record(round_dir: Path, ticket: str, **fields) -> dict: + body = {"ticket": ticket, "synthetic": True, **fields} + write_json(round_dir / "results" / f"{ticket}.json", body) + return body + + +def _claimed_entry(ledger, ticket: str): + suffix = f"@{ticket}" + for entry in ledger.entries: + if entry.binding.run_id.endswith(suffix) and entry.phase == "claimed": + return entry + return None + + +def _launch( + plan: dict, lane: dict, candidate: Path, incumbent: Path, trial_dir: Path, *, backend, current +): + """Run build and both trials. Each start is bounded by the same absolute deadline.""" + build_dir = candidate / "build-output" + build_dir.mkdir(parents=True, exist_ok=True) + trial_dir.mkdir(parents=True, exist_ok=True) + + def run(stage: str, argv, cwd, writable, readable) -> None: + remaining = min(seconds_remaining(lane["deadline_utc"], current()), 30) + try: + record = confined_execute( + argv, cwd, writable, readable, time.monotonic() + remaining, backend=backend + ) + except subprocess.TimeoutExpired as exc: + raise ConfinementUnavailable(f"{stage} timed out") from exc + if record.get("status") != "completed": + raise ConfinementUnavailable(f"{stage} {record.get('status')}") + + if plan["build"]: + run("build", tuple(plan["build"]), candidate, (candidate, build_dir), ()) + candidate_result = trial_dir / "candidate.json" + run( + "candidate", + _command(plan["trial"], candidate_result), + candidate, + (trial_dir,), + (candidate,), + ) + incumbent_result = trial_dir / "incumbent.json" + run( + "incumbent", + _command(plan["trial"], incumbent_result), + incumbent, + (trial_dir,), + (incumbent,), + ) + return candidate_result, incumbent_result + + +def serve_ticket( + round_dir: Path, + ticket: str, + incumbent: Path, + *, + backend, + now: datetime, + clock=None, +) -> dict: + """Measure one ticket. A busy device leaves it queued and does not reserve an attempt.""" + def current() -> datetime: + if clock is not None: + return clock() + return datetime.now(timezone.utc) + + plan = json.loads((round_dir / "round-plan.json").read_text(encoding="utf-8")) + result_path = round_dir / "results" / f"{ticket}.json" + if result_path.exists(): + return json.loads(result_path.read_text(encoding="utf-8")) + overlay = round_dir / "outbox" / ticket + manifest = json.loads((overlay / "manifest.json").read_text(encoding="utf-8")) + if manifest.get("deleted"): + return _record(round_dir, ticket, status="rejected", reason="deletion") + lane = _lane(plan, manifest["lane_id"]) + if _claimed_entry(_load_ledger(round_dir, plan, lane), ticket) is not None: + return _record( + round_dir, + ticket, + status="unsettled", + reason="claimed execution was not settled", + ) + try: + held = DeviceLease(round_dir / "leases").acquire(plan["device_id"]) + except LeaseError: + return {"ticket": ticket, "status": "queued"} + try: + try: + loaded = _load_ledger(round_dir, plan, lane) + ledger, binding = admit_attempt(plan, loaded, lane["id"], current(), token=ticket) + except BudgetError as exc: + return _record(round_dir, ticket, status="rejected", reason=str(exc)) + _save_ledger(round_dir, ledger) + candidate = round_dir / "candidates" / ticket + trial_dir = round_dir / "trials" / ticket + _materialize(incumbent, overlay, candidate) + try: + seconds_remaining(lane["deadline_utc"], current()) + except BudgetError as exc: + _save_ledger(round_dir, advance(ledger, binding, "cancelled")) + return _record(round_dir, ticket, status="rejected", reason=str(exc)) + ledger = advance(ledger, binding, "claimed") + _save_ledger(round_dir, ledger) + try: + candidate_result, incumbent_result = _launch( + plan, lane, candidate, incumbent, trial_dir, backend=backend, current=current + ) + except (ConfinementUnavailable, BudgetError) as exc: + return _record(round_dir, ticket, status="unsettled", reason=str(exc)) + candidate_body = json.loads(candidate_result.read_text(encoding="utf-8")) + incumbent_body = json.loads(incumbent_result.read_text(encoding="utf-8")) + candidate_metric = candidate_body["metrics"]["latency_ms"] + incumbent_metric = incumbent_body["metrics"]["latency_ms"] + finished = advance(ledger, binding, "finished", actual_cost=0) + _save_ledger(round_dir, finished) + return _record( + round_dir, + ticket, + status="measured", + candidate_latency_ms=candidate_metric, + incumbent_latency_ms=incumbent_metric, + gain_ms=incumbent_metric - candidate_metric, + confinement=getattr(backend, "name", "unknown"), + ) + finally: + held.release() + + +def ticket_status(round_dir: Path, ticket: str) -> dict: + path = round_dir / "results" / f"{ticket}.json" + if path.exists(): + return json.loads(path.read_text(encoding="utf-8")) + return {"ticket": ticket, "status": "queued"} diff --git a/src/inferrsi/harness/round_budget.py b/src/inferrsi/harness/round_budget.py new file mode 100644 index 000000000..ba4e921bf --- /dev/null +++ b/src/inferrsi/harness/round_budget.py @@ -0,0 +1,228 @@ +"""Lane attempt budgets for one parallel round. + +The transitions match ``tools/skill_eval_budget.py``: a reservation is a +permanent charge, and the same plan digest cannot be reserved twice. The +identity here is a lane, and each attempt mints its own plan digest from the +unchanged round digest, lane id and attempt ordinal. The cap comes from the +frozen round plan. This module does not authorize a skill-eval pull-request +budget, and it does not confine the filesystem. +""" + +import hashlib +import json +from dataclasses import dataclass, replace +from datetime import datetime +from pathlib import Path + + +class BudgetError(ValueError): + """Invalid, exhausted, expired, or unsafe lane-budget state.""" + + +def _text(value: str) -> None: + if type(value) is not str or not value.strip() or len(value) > 512: + raise BudgetError("Expected a bounded nonempty identifier") + + +def _digest(value: str) -> None: + digits = "0123456789abcdef" + if type(value) is not str or len(value) != 64 or any(char not in digits for char in value): + raise BudgetError("Expected SHA-256 digest") + + +def _amount(value: int, *, positive: bool = False) -> None: + if type(value) is not int or value < int(positive): + raise BudgetError("Expected nonnegative integer cost") + + +def attempt_plan_digest(round_digest: str, lane_id: str, attempt_ordinal: int) -> str: + """One immutable plan identity per attempt. The round digest stays unchanged.""" + _digest(round_digest) + _text(lane_id) + _amount(attempt_ordinal, positive=True) + payload = f"{round_digest}\n{lane_id}\n{attempt_ordinal}".encode() + return hashlib.sha256(payload).hexdigest() + + +@dataclass(frozen=True) +class LaneKey: + round_digest: str + lane_id: str + + def __post_init__(self) -> None: + _digest(self.round_digest) + _text(self.lane_id) + + +@dataclass(frozen=True) +class Genesis: + evidence: str + legacy_charge: int + + def __post_init__(self) -> None: + _text(self.evidence) + _amount(self.legacy_charge) + + +@dataclass(frozen=True) +class Binding: + plan_digest: str + run_id: str + worst_case: int + + def __post_init__(self) -> None: + _digest(self.plan_digest) + _text(self.run_id) + _amount(self.worst_case, positive=True) + + +@dataclass(frozen=True) +class Entry: + binding: Binding + phase: str = "reserved" + actual_cost: int | None = None + + def __post_init__(self) -> None: + if type(self.binding) is not Binding or self.phase not in { + "reserved", "claimed", "finished", "cancelled" + }: + raise BudgetError("Invalid reservation") + if self.phase == "finished": + _amount(self.actual_cost) + if self.actual_cost > self.binding.worst_case: + raise BudgetError("Actual cost exceeds reservation") + elif self.actual_cost is not None: + raise BudgetError("Only finished reservations have actual cost") + + +@dataclass(frozen=True) +class Ledger: + key: LaneKey + genesis: Genesis + cap: int + revision: int = 0 + entries: tuple[Entry, ...] = () + + def __post_init__(self) -> None: + if type(self.key) is not LaneKey or type(self.genesis) is not Genesis: + raise BudgetError("Missing ledger identity") + _amount(self.cap, positive=True) + _amount(self.revision) + entries_ok = type(self.entries) is tuple and all( + type(entry) is Entry for entry in self.entries + ) + if not entries_ok: + raise BudgetError("Invalid ledger entries") + expected = sum( + {"reserved": 1, "claimed": 2, "finished": 3, "cancelled": 2}[entry.phase] + for entry in self.entries + ) + if self.revision != expected: + raise BudgetError("Inconsistent transition revision") + plans = [entry.binding.plan_digest for entry in self.entries] + runs = [entry.binding.run_id for entry in self.entries] + if len(set(plans)) != len(plans) or len(set(runs)) != len(runs): + raise BudgetError("Duplicate plan or run binding") + if self.charged > self.cap: + raise BudgetError("lane cumulative budget exhausted") + + @property + def charged(self) -> int: + return self.genesis.legacy_charge + sum( + entry.binding.worst_case for entry in self.entries + ) + + +def reserve(ledger: Ledger, binding: Binding) -> Ledger: + """Idempotent for the exact binding only. Does not authorize execution.""" + if type(ledger) is not Ledger or type(binding) is not Binding: + raise BudgetError("Invalid ledger or binding") + for entry in ledger.entries: + if entry.binding == binding: + return ledger + same_plan = entry.binding.plan_digest == binding.plan_digest + same_run = entry.binding.run_id == binding.run_id + if same_plan or same_run: + raise BudgetError("Plan or run already bound differently") + return replace(ledger, revision=ledger.revision + 1, entries=ledger.entries + (Entry(binding),)) + + +def advance( + ledger: Ledger, binding: Binding, phase: str, *, actual_cost: int | None = None +) -> Ledger: + """Claim is not idempotent. Cancelling a reservation does not free capacity.""" + if type(ledger) is not Ledger or type(binding) is not Binding: + raise BudgetError("Invalid ledger or binding") + for index, entry in enumerate(ledger.entries): + if entry.binding != binding: + continue + updated = Entry(binding, phase, actual_cost) + if updated == entry and phase in {"finished", "cancelled"}: + return ledger + allowed = {"reserved": {"claimed", "cancelled"}, "claimed": {"finished"}} + if phase not in allowed.get(entry.phase, set()): + raise BudgetError("Unsafe or duplicate execution transition") + entries = ledger.entries[:index] + (updated,) + ledger.entries[index + 1 :] + return replace(ledger, revision=ledger.revision + 1, entries=entries) + raise BudgetError("No exact reservation for this plan/run") + + +def _deadline(deadline_utc: str, now: datetime) -> datetime: + if type(deadline_utc) is not str or now.tzinfo is None: + raise BudgetError("deadline and now must be timezone-aware") + try: + deadline = datetime.fromisoformat(deadline_utc) + except ValueError as exc: + raise BudgetError("deadline must be a timezone-aware timestamp") from exc + if deadline.tzinfo is None: + raise BudgetError("deadline and now must be timezone-aware") + return deadline + + +def ensure_open(deadline_utc: str, now: datetime) -> None: + """Refuse a new attempt after the lane deadline. Reservation does not check time.""" + if now > _deadline(deadline_utc, now): + raise BudgetError("lane deadline passed") + + +def seconds_remaining(deadline_utc: str, now: datetime) -> float: + """Seconds left until the same absolute cutoff. Zero or less is already closed.""" + remaining = (_deadline(deadline_utc, now) - now).total_seconds() + if remaining <= 0: + raise BudgetError("lane deadline passed") + return remaining + + +def empty_ledger(round_digest: str, lane_id: str, cap: int) -> Ledger: + return Ledger( + LaneKey(round_digest, lane_id), + Genesis("frozen-round-plan", 0), + cap, + ) + + +def admit_attempt( + plan: dict, ledger: Ledger, lane_id: str, now: datetime, *, token: str | None = None +) -> tuple[Ledger, Binding]: + """Read the lane contract, enforce the deadline, then reserve the next attempt.""" + lanes = [lane for lane in plan.get("lanes", []) if lane.get("id") == lane_id] + if len(lanes) != 1: + raise BudgetError("lane is missing from the round plan") + lane = lanes[0] + ensure_open(lane["deadline_utc"], now) + ordinal = len(ledger.entries) + 1 + run_id = f"{lane_id}#{ordinal}" if token is None else f"{lane_id}#{ordinal}@{token}" + binding = Binding( + attempt_plan_digest(plan["round_digest"], lane_id, ordinal), + run_id, + lane["per_attempt_worst_case"], + ) + return reserve(ledger, binding), binding + + +def admit_from_path( + plan_path: Path, ledger: Ledger, lane_id: str, now: datetime +) -> tuple[Ledger, Binding]: + """Re-read the frozen plan from disk so a restarted coordinator cannot skip expiry.""" + plan = json.loads(Path(plan_path).read_text(encoding="utf-8")) + return admit_attempt(plan, ledger, lane_id, now) diff --git a/src/inferrsi/infra/confinement.py b/src/inferrsi/infra/confinement.py new file mode 100644 index 000000000..43545eb38 --- /dev/null +++ b/src/inferrsi/infra/confinement.py @@ -0,0 +1,404 @@ +"""Filesystem confinement for candidate build and trial commands. + +``execute_command`` is unchanged. The namespace backend builds a fresh root +from the writable set, the readable set and the Python runtime. It does not +remount the host root, and it does not pass the coordinator environment +through. If that view cannot be built, execution is refused. ``FixtureBackend`` +is a test double: it enforces the same view with an import hook and a minimal +environment, which is not mount-namespace confinement. +""" + +import json +import os +import struct +import subprocess +import sys +import tempfile +import time +from pathlib import Path + +_ENV_ALLOW = frozenset({"PATH", "LANG", "LC_ALL", "LC_CTYPE", "TZ", "PYTHONIOENCODING"}) +_GUARD = """ +import builtins +import os + +_roots = [item for item in os.environ.get("INFERRSI_VISIBLE_ROOTS", "").split(os.pathsep) if item] +_builtin_open = builtins.open +_os_open = os.open + + +def _allowed(file): + if not isinstance(file, (str, bytes, os.PathLike)): + return True + text = os.fsdecode(file) + if text.startswith("<"): + return True + try: + resolved = os.path.realpath(text) + except OSError: + return False + return any(resolved == root or resolved.startswith(root + os.sep) for root in _roots) + + +def _open(file, *args, **kwargs): + if not _allowed(file): + raise PermissionError(file) + return _builtin_open(file, *args, **kwargs) + + +def _open_fd(path, flags, mode=0o777, *, dir_fd=None): + if dir_fd is None and not _allowed(path): + raise PermissionError(path) + return _os_open(path, flags, mode, dir_fd=dir_fd) + + +builtins.open = _open +os.open = _open_fd +""" + + +class ConfinementUnavailable(RuntimeError): + """The command was not executed because confinement could not be enforced.""" + + +def namespaces_available() -> bool: + try: + limit = int(Path("/proc/sys/user/max_user_namespaces").read_text(encoding="utf-8")) + except (OSError, ValueError): + return False + return limit > 0 + + +def minimal_environment(extra: dict | None = None) -> dict: + """Coordinator credentials stay out. Only a fixed allowlist is copied.""" + environment = {key: os.environ[key] for key in _ENV_ALLOW if key in os.environ} + if extra: + environment.update(extra) + return environment + + +def runtime_roots() -> tuple[Path, ...]: + """Python prefix needed to start the interpreter. Never the host root.""" + candidates = [Path(sys.prefix).resolve(), Path(sys.executable).resolve().parent] + roots: list[Path] = [] + for candidate in candidates: + if any(candidate == earlier or earlier in candidate.parents for earlier in roots): + continue + roots.append(candidate) + return tuple(roots) + + +def _elf_interpreter(executable: Path) -> str | None: + """PT_INTERP path recorded in the executable, or None for a static binary.""" + data = executable.read_bytes() + if data[:4] != b"\x7fELF": + return None + endian = "<" if data[5] == 1 else ">" + if data[4] == 2: + phoff = struct.unpack_from(endian + "Q", data, 32)[0] + phentsize, phnum = struct.unpack_from(endian + "HH", data, 54) + header = endian + "IIQQQQ" + elif data[4] == 1: + phoff = struct.unpack_from(endian + "I", data, 28)[0] + phentsize, phnum = struct.unpack_from(endian + "HH", data, 42) + header = endian + "IIIIIIII" + else: + raise ConfinementUnavailable("unsupported ELF class") + for index in range(phnum): + start = phoff + index * phentsize + fields = struct.unpack_from(header, data, start) + if fields[0] != 3: + continue + offset, size = (fields[2], fields[5]) if data[4] == 2 else (fields[1], fields[4]) + return data[offset : offset + size].split(b"\0", 1)[0].decode() + return None + + +def _shared_libraries(executable: Path) -> tuple[str, ...]: + """Absolute libraries ldd resolves. The virtual dynamic linker object is skipped.""" + completed = subprocess.run( + ["ldd", os.fspath(executable)], capture_output=True, text=True, check=False + ) + if completed.returncode != 0: + raise ConfinementUnavailable(completed.stderr.strip() or "ldd failed") + found: list[str] = [] + for line in completed.stdout.splitlines(): + if "=>" in line: + token = line.split("=>", 1)[1].strip().split()[0] + else: + token = line.strip().split()[0] + if token.startswith("/"): + found.append(token) + return tuple(found) + + +def runtime_files() -> tuple[Path, ...]: + """Loader and shared libraries that live outside the Python prefix.""" + executable = Path(sys.executable) + names: list[Path] = [] + interpreter = _elf_interpreter(executable) + if interpreter: + names.append(Path(interpreter)) + elif executable.read_bytes()[:4] == b"\x7fELF": + return () + for item in _shared_libraries(executable): + names.append(Path(item)) + cache = Path("/etc/ld.so.cache") + if cache.is_file(): + names.append(cache) + unique: list[Path] = [] + for path in names: + if path not in unique: + unique.append(path) + return tuple(unique) + + +def _covered(path: Path, covered: list[Path]) -> bool: + return any(path == root or root in path.parents for root in covered) + + +def namespace_plan( + writable: tuple[Path, ...], readable: tuple[Path, ...] +) -> dict: + """Mount list for a fresh root. The host root itself is not an entry.""" + mounts: list[dict] = [] + covered: list[Path] = [] + + def add(path: Path, mode: str, *, directory: bool) -> None: + source = str(path) + if source == "/" or ".." in path.parts: + raise ConfinementUnavailable("refusing a mount plan that exposes the host root") + if any(item["source"] == source for item in mounts) or _covered(path, covered): + return + mounts.append({"source": source, "mode": mode}) + if directory: + covered.append(path) + + for path in writable: + add(path, "rw", directory=True) + for path in readable: + add(path, "ro", directory=True) + for root in runtime_roots(): + add(root, "ro", directory=True) + for extra in runtime_files(): + add(extra, "ro", directory=False) + return {"mounts": mounts, "env": minimal_environment()} + + +def _within(path: Path, roots: tuple[Path, ...]) -> bool: + resolved = path.resolve() + return any(resolved == root or root in resolved.parents for root in roots) + + +def _outside_writable(argv: tuple[str, ...], writable: tuple[Path, ...]) -> Path | None: + for arg in argv[1:]: + if not arg.startswith("/"): + continue + path = Path(arg) + if not _within(path, writable): + return path + return None + + +def _run_local(argv: tuple[str, ...], cwd: Path, deadline: float, environment: dict) -> dict: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise ConfinementUnavailable("execution deadline exhausted before start") + try: + completed = subprocess.run( + argv, + cwd=cwd, + env=environment, + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + timeout=remaining, + check=False, + ) + except subprocess.TimeoutExpired as exc: + return { + "status": "timeout", + "returncode": None, + "stdout": exc.stdout or "", + "stderr": exc.stderr or "", + } + status = "completed" if completed.returncode == 0 else "failed" + return { + "status": status, + "returncode": completed.returncode, + "stdout": completed.stdout, + "stderr": completed.stderr, + } + + +class UnavailableBackend: + """Fail closed. Does not execute the candidate.""" + + name = "unavailable" + + def run( + self, + argv: tuple[str, ...], + cwd: Path, + writable: tuple[Path, ...], + readable: tuple[Path, ...], + deadline: float, + ) -> dict: + del argv, cwd, writable, readable, deadline + raise ConfinementUnavailable("confinement backend unavailable") + + +class FixtureBackend: + """Enforce the visible roots and minimal environment. Not a mount namespace.""" + + name = "fixture" + + def run( + self, + argv: tuple[str, ...], + cwd: Path, + writable: tuple[Path, ...], + readable: tuple[Path, ...], + deadline: float, + ) -> dict: + outside = _outside_writable(argv, writable) + if outside is not None: + raise ConfinementUnavailable(f"path outside writable set: {outside}") + visible = tuple(dict.fromkeys(writable + readable + runtime_roots())) + try: + guard_dir = tempfile.mkdtemp(prefix="inferrsi-guard-") + Path(guard_dir, "sitecustomize.py").write_text(_GUARD, encoding="utf-8") + except OSError as exc: + raise ConfinementUnavailable("confinement guard could not be installed") from exc + environment = minimal_environment( + { + "PYTHONPATH": guard_dir, + "INFERRSI_VISIBLE_ROOTS": os.pathsep.join(str(path) for path in visible), + } + ) + return _run_local(argv, cwd, deadline, environment) + + +class UnshareBackend: + """Run argv in a fresh root. Fail closed if the namespace cannot be built.""" + + name = "unshare" + + def run( + self, + argv: tuple[str, ...], + cwd: Path, + writable: tuple[Path, ...], + readable: tuple[Path, ...], + deadline: float, + ) -> dict: + if not writable: + raise ConfinementUnavailable("refusing to run with an empty writable set") + plan = namespace_plan(writable, readable) + remaining = deadline - time.monotonic() + if remaining <= 0: + raise ConfinementUnavailable("execution deadline exhausted before start") + config_path = None + completed = None + try: + handle = tempfile.NamedTemporaryFile( + prefix="inferrsi-confine-", suffix=".json", delete=False + ) + config_path = Path(handle.name) + handle.close() + config_path.write_text( + json.dumps({"argv": list(argv), "cwd": str(cwd), **plan}), + encoding="utf-8", + ) + completed = subprocess.run( + [ + "unshare", + "--user", + "--map-root-user", + "--mount", + "--propagation", + "private", + "--", + sys.executable, + "-c", + "import sys; from inferrsi.infra.confinement import enter_namespace; " + "enter_namespace(sys.argv[1])", + str(config_path), + ], + cwd=cwd, + env=minimal_environment(), + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + timeout=remaining, + check=False, + ) + except subprocess.TimeoutExpired as exc: + raise ConfinementUnavailable("namespace command timed out") from exc + finally: + if config_path is not None: + config_path.unlink(missing_ok=True) + if completed is None or completed.returncode != 0: + detail = "unshare confinement failed" + if completed is not None and completed.stderr.strip(): + detail = completed.stderr.strip() + raise ConfinementUnavailable(detail) + return { + "status": "completed", + "returncode": 0, + "stdout": completed.stdout, + "stderr": completed.stderr, + } + + +def enter_namespace(config_path: str) -> None: + """Bind only the planned paths into a tmpfs root, then chroot. Exit before exec on failure.""" + config = json.loads(Path(config_path).read_text(encoding="utf-8")) + jail = Path(tempfile.mkdtemp(prefix="inferrsi-jail-")) + + def mount(args: list[str]) -> None: + completed = subprocess.run(["mount", *args], capture_output=True, text=True, check=False) + if completed.returncode != 0: + raise SystemExit(completed.stderr.strip() or "mount failed") + + mount(["-t", "tmpfs", "tmpfs", str(jail)]) + for item in config["mounts"]: + source = item["source"] + source_path = Path(source) + if source == "/" or not source_path.is_absolute() or ".." in source_path.parts: + raise SystemExit("refusing to mount the host root") + target = jail.joinpath(*source_path.parts[1:]) + if source_path.is_dir() and not source_path.is_symlink(): + target.mkdir(parents=True, exist_ok=True) + else: + target.parent.mkdir(parents=True, exist_ok=True) + target.touch() + mount(["--bind", source, str(target)]) + if item["mode"] == "ro": + mount(["-o", "remount,bind,ro", str(target)]) + os.chroot(jail) + os.chdir(config["cwd"]) + argv = config["argv"] + os.execvpe(argv[0], argv, config["env"]) + + +def select_backend(): + if namespaces_available(): + return UnshareBackend() + return UnavailableBackend() + + +def confined_execute( + argv: tuple[str, ...], + cwd: Path, + writable_paths: tuple[Path, ...], + readable_paths: tuple[Path, ...], + deadline: float, + *, + backend=None, +) -> dict: + """Execute argv only inside the selected boundary. Never falls back to a plain subprocess.""" + chosen = select_backend() if backend is None else backend + writable = tuple(Path(path).resolve() for path in writable_paths) + readable = tuple(Path(path).resolve() for path in readable_paths) + return chosen.run(tuple(argv), Path(cwd), writable, readable, deadline) diff --git a/src/inferrsi/infra/device_lease.py b/src/inferrsi/infra/device_lease.py new file mode 100644 index 000000000..7c006f042 --- /dev/null +++ b/src/inferrsi/infra/device_lease.py @@ -0,0 +1,40 @@ +"""Host-local device leases. The key is the caller's device identity, not a CUDA probe.""" + +import fcntl +from pathlib import Path + + +class LeaseError(RuntimeError): + """The device identity is already held.""" + + +class _Held: + def __init__(self, handle) -> None: + self._handle = handle + + def release(self) -> None: + fcntl.flock(self._handle.fileno(), fcntl.LOCK_UN) + self._handle.close() + + def __enter__(self): + return self + + def __exit__(self, *exc) -> None: + self.release() + + +class DeviceLease: + def __init__(self, directory: Path) -> None: + self.directory = Path(directory) + self.directory.mkdir(parents=True, exist_ok=True) + + def acquire(self, device_id: str) -> _Held: + if not device_id or any(char in device_id for char in "/\\"): + raise LeaseError("device identity must be a single path segment") + handle = (self.directory / device_id).open("a+") + try: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + handle.close() + raise LeaseError(f"device busy: {device_id}") from None + return _Held(handle) diff --git a/tests/test_confinement.py b/tests/test_confinement.py new file mode 100644 index 000000000..ca5a52a3c --- /dev/null +++ b/tests/test_confinement.py @@ -0,0 +1,199 @@ +"""Confinement policy. The fixture backend is not an operating-system boundary.""" + +import json +import subprocess +import sys +import time +from pathlib import Path + +import pytest + +from inferrsi.infra.confinement import ( + ConfinementUnavailable, + FixtureBackend, + UnavailableBackend, + UnshareBackend, + _elf_interpreter, + confined_execute, + namespace_plan, + namespaces_available, + select_backend, +) + + +def test_build_write_outside_the_checkout_is_denied_and_does_not_run(tmp_path): + checkout = tmp_path / "candidate" + checkout.mkdir() + ledger = tmp_path / "ledger.json" + ledger.write_text("safe", encoding="utf-8") + argv = ( + sys.executable, + "-c", + "open(__import__('sys').argv[1], 'w').write('pwned')", + str(ledger), + ) + with pytest.raises(ConfinementUnavailable, match="outside writable"): + confined_execute( + argv, + checkout, + (checkout,), + (), + time.monotonic() + 5, + backend=FixtureBackend(), + ) + assert ledger.read_text(encoding="utf-8") == "safe" + + +def test_missing_backend_refuses_without_running(tmp_path): + marker = tmp_path / "ran" + argv = (sys.executable, "-c", "open('ran','w').write('yes')") + with pytest.raises(ConfinementUnavailable, match="unavailable"): + confined_execute( + argv, + tmp_path, + (tmp_path,), + (), + time.monotonic() + 5, + backend=UnavailableBackend(), + ) + assert not marker.exists() + + +def test_select_backend_fails_closed_when_namespaces_are_disabled(monkeypatch): + monkeypatch.setattr("inferrsi.infra.confinement.namespaces_available", lambda: False) + assert isinstance(select_backend(), UnavailableBackend) + + +def test_unshare_failure_does_not_fall_back(tmp_path, monkeypatch): + calls = [] + monkeypatch.setenv("SECRET_CANARY", "CANARY-ENV") + real_run = subprocess.run + + def fake_run(argv, **kwargs): + if not argv or argv[0] != "unshare": + return real_run(argv, **kwargs) + del kwargs + calls.append(json.loads(Path(argv[-1]).read_text(encoding="utf-8"))) + + class Completed: + returncode = 1 + stdout = "" + stderr = "unshare failed" + + return Completed() + + monkeypatch.setattr("inferrsi.infra.confinement.subprocess.run", fake_run) + with pytest.raises(ConfinementUnavailable, match="unshare failed"): + UnshareBackend().run( + (sys.executable, "-c", "print('should-not-run')"), + tmp_path, + (tmp_path,), + (), + time.monotonic() + 5, + ) + assert calls + assert "SECRET_CANARY" not in calls[0]["env"] + assert "/" not in {item["source"] for item in calls[0]["mounts"]} + + +def test_unshare_timeout_becomes_confinement_failure(tmp_path, monkeypatch): + real_run = subprocess.run + + def fake_run(argv, **kwargs): + if not argv or argv[0] != "unshare": + return real_run(argv, **kwargs) + raise subprocess.TimeoutExpired(argv, kwargs.get("timeout") or 1) + + monkeypatch.setattr("inferrsi.infra.confinement.subprocess.run", fake_run) + with pytest.raises(ConfinementUnavailable, match="timed out"): + UnshareBackend().run( + (sys.executable, "-c", "print('should-not-run')"), + tmp_path, + (tmp_path,), + (), + time.monotonic() + 5, + ) + + +def test_namespace_plan_includes_the_loader_but_not_a_private_file(tmp_path, monkeypatch): + monkeypatch.setenv("SECRET_CANARY", "CANARY-ENV") + private = tmp_path / "private" + private.mkdir() + (private / "canary.txt").write_text("CANARY-FILE", encoding="utf-8") + plan = namespace_plan((tmp_path / "work",), ()) + sources = {item["source"] for item in plan["mounts"]} + interpreter = _elf_interpreter(Path(sys.executable)) + assert interpreter is not None + assert interpreter in sources or any( + Path(interpreter).is_relative_to(Path(source)) for source in sources + ) + assert "/" not in sources + assert str(private) not in sources + assert "SECRET_CANARY" not in plan["env"] + + +@pytest.mark.skipif(not namespaces_available(), reason="user namespaces are unavailable") +def test_unshare_starts_python_and_hides_a_private_canary(tmp_path, monkeypatch): + monkeypatch.setenv("SECRET_CANARY", "CANARY-ENV") + private = tmp_path / "private" + private.mkdir() + (private / "canary.txt").write_text("CANARY-FILE", encoding="utf-8") + work = tmp_path / "work" + work.mkdir() + script = ( + "import os\n" + "print('STARTED')\n" + "print(os.environ.get('SECRET_CANARY', ''))\n" + "print(os.path.exists(%r))\n" % str(private / "canary.txt") + ) + record = UnshareBackend().run( + (sys.executable, "-c", script), + work, + (work,), + (), + time.monotonic() + 30, + ) + assert record["status"] == "completed" + assert "STARTED" in record["stdout"] + assert "CANARY-ENV" not in record["stdout"] + assert "True" not in record["stdout"] + + +def test_namespace_plan_keeps_host_root_and_secrets_out(tmp_path, monkeypatch): + monkeypatch.setenv("SECRET_CANARY", "CANARY-ENV") + plan = namespace_plan((tmp_path / "writable",), (tmp_path / "readable",)) + assert "/" not in {item["source"] for item in plan["mounts"]} + assert "SECRET_CANARY" not in plan["env"] + assert str(tmp_path / "private") not in {item["source"] for item in plan["mounts"]} + + +def test_build_and_trial_cannot_see_private_canaries(tmp_path, monkeypatch): + monkeypatch.setenv("SECRET_CANARY", "CANARY-ENV") + private = tmp_path / "private" + private.mkdir() + (private / "canary.txt").write_text("CANARY-FILE", encoding="utf-8") + checkout = tmp_path / "checkout" + checkout.mkdir() + trial = tmp_path / "trial" + trial.mkdir() + script = ( + "import os\n" + "print(os.environ.get('SECRET_CANARY', ''))\n" + "try:\n" + " print(open(%r).read())\n" + "except PermissionError:\n" + " print('FILE-HIDDEN')\n" % str(private / "canary.txt") + ) + argv = (sys.executable, "-c", script) + backend = FixtureBackend() + stages = ( + confined_execute(argv, checkout, (checkout,), (), time.monotonic() + 10, backend=backend), + confined_execute( + argv, checkout, (trial,), (checkout,), time.monotonic() + 10, backend=backend + ), + ) + for record in stages: + assert record["status"] == "completed" + assert "CANARY-ENV" not in record["stdout"] + assert "CANARY-FILE" not in record["stdout"] + assert "FILE-HIDDEN" in record["stdout"] diff --git a/tests/test_parallel_round.py b/tests/test_parallel_round.py new file mode 100644 index 000000000..a467a5c13 --- /dev/null +++ b/tests/test_parallel_round.py @@ -0,0 +1,381 @@ +"""Synthetic parallel round. Latency figures here are fixture inputs, not measurements.""" + +import json +import subprocess +import sys +import threading +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +from inferrsi.harness.parallel_round import ( + RoundPlanError, + prepare, + serve_ticket, + submit, + ticket_status, +) +from inferrsi.infra.confinement import ConfinementUnavailable, FixtureBackend +from inferrsi.infra.device_lease import DeviceLease, LeaseError + +WHEN = datetime(2026, 10, 30, tzinfo=timezone.utc) +BENCH = ( + "import json, sys\n" + "from pathlib import Path\n" + "value = float(Path('knobs/a.txt').read_text())\n" + "body = {'metrics': {'latency_ms': value}, 'synthetic': True}\n" + "Path(sys.argv[1]).write_text(json.dumps(body))\n" +) + + +def _tree(root: Path, knob: str) -> None: + (root / "knobs").mkdir(parents=True) + (root / "knobs" / "a.txt").write_text(knob, encoding="utf-8") + (root / "knobs" / "b.txt").write_text("10", encoding="utf-8") + (root / "bench.py").write_text(BENCH, encoding="utf-8") + + +def _spec(**overrides): + lanes = overrides.pop( + "lanes", + [ + { + "id": "L1", + "entry": "D-01", + "plane": "infra", + "globs": ["knobs/a.txt"], + "cap": 2, + "per_attempt_worst_case": 1, + "deadline_utc": "2099-01-01T00:00:00+00:00", + }, + { + "id": "L2", + "entry": "D-02", + "plane": "harness", + "globs": ["knobs/b.txt"], + "cap": 2, + "per_attempt_worst_case": 1, + "deadline_utc": "2099-01-01T00:00:00+00:00", + }, + ], + ) + spec = { + "round_id": "r1", + "device_id": "cpu-0", + "incumbent_id": "fixture-incumbent", + "protected": ["bench.py"], + "build": [sys.executable, "-c", "open('build-stamp','w').write('ok')"], + "trial": [sys.executable, "bench.py", "{result}"], + "lanes": lanes, + } + spec.update(overrides) + return spec + + +def test_prepare_rejects_overlapping_globs_protected_paths_and_shared_entries(tmp_path): + incumbent = tmp_path / "incumbent" + _tree(incumbent, "10") + overlap = _spec( + lanes=[ + { + "id": "L1", + "entry": "D-01", + "plane": "infra", + "globs": ["knobs/*.txt"], + "cap": 1, + "per_attempt_worst_case": 1, + "deadline_utc": "2099-01-01T00:00:00+00:00", + }, + { + "id": "L2", + "entry": "D-02", + "plane": "harness", + "globs": ["knobs/a.txt"], + "cap": 1, + "per_attempt_worst_case": 1, + "deadline_utc": "2099-01-01T00:00:00+00:00", + }, + ] + ) + with pytest.raises(RoundPlanError, match="overlaps"): + prepare(overlap, incumbent, tmp_path / "round-overlap") + protected = _spec( + lanes=[ + { + "id": "L1", + "entry": "D-01", + "plane": "infra", + "globs": ["bench.py"], + "cap": 1, + "per_attempt_worst_case": 1, + "deadline_utc": "2099-01-01T00:00:00+00:00", + } + ] + ) + with pytest.raises(RoundPlanError, match="protected"): + prepare(protected, incumbent, tmp_path / "round-protected") + shared = _spec() + shared["lanes"][1]["entry"] = "D-01" + with pytest.raises(RoundPlanError, match="pursue"): + prepare(shared, incumbent, tmp_path / "round-shared") + many = _spec() + many["lanes"][0]["plane"] = ["infra", "harness"] + with pytest.raises(RoundPlanError, match="exactly one plane"): + prepare(many, incumbent, tmp_path / "round-plane") + + +def test_prepare_records_plan_cap_and_deadline(tmp_path): + incumbent = tmp_path / "incumbent" + _tree(incumbent, "10") + plan = prepare(_spec(), incumbent, tmp_path / "round") + written = json.loads((tmp_path / "round" / "round-plan.json").read_text(encoding="utf-8")) + assert written["lanes"][0]["cap"] == 2 + assert written["lanes"][0]["deadline_utc"].startswith("2099-") + assert written["round_digest"] == plan["round_digest"] + assert len(written["round_digest"]) == 64 + + +def test_submit_rejects_paths_outside_the_lane_and_is_idempotent(tmp_path): + incumbent = tmp_path / "incumbent" + _tree(incumbent, "10") + round_dir = tmp_path / "round" + prepare(_spec(), incumbent, round_dir) + work = tmp_path / "work" + _tree(work, "8") + (work / "README").write_text("extra", encoding="utf-8") + rejected = submit(round_dir, "L1", incumbent, work) + assert rejected["submitted"] is False + assert rejected["outside"] == ["README"] + assert not (round_dir / "outbox").exists() + (work / "README").unlink() + first = submit(round_dir, "L1", incumbent, work) + second = submit(round_dir, "L1", incumbent, work) + assert first["submitted"] is True + assert first["ticket"] == second["ticket"] + + +def test_serve_records_a_synthetic_gain_and_a_held_device_stays_queued(tmp_path): + incumbent = tmp_path / "incumbent" + _tree(incumbent, "10") + round_dir = tmp_path / "round" + prepare(_spec(), incumbent, round_dir) + work = tmp_path / "work" + _tree(work, "8") + ticket = submit(round_dir, "L1", incumbent, work)["ticket"] + held = DeviceLease(round_dir / "leases").acquire("cpu-0") + queued = serve_ticket(round_dir, ticket, incumbent, backend=FixtureBackend(), now=WHEN) + assert queued["status"] == "queued" + assert ticket_status(round_dir, ticket)["status"] == "queued" + assert not (round_dir / "ledgers").exists() + held.release() + measured = serve_ticket(round_dir, ticket, incumbent, backend=FixtureBackend(), now=WHEN) + assert measured["synthetic"] is True + assert measured["candidate_latency_ms"] == 8 + assert measured["incumbent_latency_ms"] == 10 + assert measured["gain_ms"] == 2 + assert measured["confinement"] == "fixture" + assert (round_dir / "candidates" / ticket / "build-stamp").read_text(encoding="utf-8") == "ok" + + +def test_two_acquires_do_not_share_one_device(tmp_path): + lease = DeviceLease(tmp_path) + held = lease.acquire("cpu-0") + failed = threading.Event() + + def other() -> None: + try: + lease.acquire("cpu-0") + except LeaseError: + failed.set() + + thread = threading.Thread(target=other) + thread.start() + thread.join(2) + assert failed.is_set() + held.release() + lease.acquire("cpu-1").release() + + +def test_submit_rejects_symlink_without_reading_the_target(tmp_path): + incumbent = tmp_path / "incumbent" + _tree(incumbent, "10") + round_dir = tmp_path / "round" + prepare(_spec(), incumbent, round_dir) + canary = tmp_path / "canary.txt" + canary.write_text("CANARY-SECRET", encoding="utf-8") + work = tmp_path / "work" + _tree(work, "8") + (work / "knobs" / "a.txt").unlink() + (work / "knobs" / "a.txt").symlink_to(canary) + rejected = submit(round_dir, "L1", incumbent, work, now=WHEN) + assert rejected["submitted"] is False + assert rejected["reason"] == "symlink" + assert rejected["ticket"] is None + assert not (round_dir / "outbox").exists() + assert "CANARY-SECRET" not in json.dumps(rejected) + + +def test_in_scope_deletion_is_rejected_before_hashing_and_serve_does_not_run(tmp_path): + incumbent = tmp_path / "incumbent" + _tree(incumbent, "10") + round_dir = tmp_path / "round" + prepare(_spec(), incumbent, round_dir) + work = tmp_path / "work" + _tree(work, "8") + (work / "knobs" / "a.txt").unlink() + rejected = submit(round_dir, "L1", incumbent, work, now=WHEN) + assert rejected == { + "submitted": False, + "outside": [], + "deleted": ["knobs/a.txt"], + "rejected": [], + "ticket": None, + "reason": "deletion", + } + ticket = "crafted-ticket" + overlay = round_dir / "outbox" / ticket + overlay.mkdir(parents=True) + (overlay / "manifest.json").write_text( + json.dumps({"ticket": ticket, "lane_id": "L1", "files": {}, "deleted": ["knobs/a.txt"]}), + encoding="utf-8", + ) + served = serve_ticket(round_dir, ticket, incumbent, backend=FixtureBackend(), now=WHEN) + assert served["status"] == "rejected" + assert served["reason"] == "deletion" + assert not (round_dir / "candidates").exists() + + +def test_submit_after_deadline_is_refused_on_a_fresh_read(tmp_path): + incumbent = tmp_path / "incumbent" + _tree(incumbent, "10") + spec = _spec() + spec["lanes"][0]["deadline_utc"] = "2020-01-01T00:00:00+00:00" + round_dir = tmp_path / "round" + prepare(spec, incumbent, round_dir) + work = tmp_path / "work" + _tree(work, "8") + first = submit(round_dir, "L1", incumbent, work, now=WHEN) + second = submit(round_dir, "L1", incumbent, work, now=WHEN) + assert first["submitted"] is False and second["submitted"] is False + assert "deadline" in first["reason"] + assert not (round_dir / "outbox").exists() + + +def test_launch_after_deadline_cancels_before_claim(tmp_path): + incumbent = tmp_path / "incumbent" + _tree(incumbent, "10") + round_dir = tmp_path / "round" + prepare(_spec(), incumbent, round_dir) + work = tmp_path / "work" + _tree(work, "8") + ticket = submit(round_dir, "L1", incumbent, work, now=WHEN)["ticket"] + moments = [WHEN, datetime(2100, 1, 1, tzinfo=timezone.utc)] + + def clock(): + return moments.pop(0) if len(moments) > 1 else moments[0] + + rejected = serve_ticket( + round_dir, ticket, incumbent, backend=FixtureBackend(), now=WHEN, clock=clock + ) + assert rejected["status"] == "rejected" + assert "deadline" in rejected["reason"] + ledger = json.loads((round_dir / "ledgers" / "L1.json").read_text(encoding="utf-8")) + assert ledger["entries"][-1]["phase"] == "cancelled" + assert not (round_dir / "candidates" / ticket / "build-stamp").exists() + + +def test_deadline_expiring_between_build_and_candidate_stops_the_trial(tmp_path): + incumbent = tmp_path / "incumbent" + _tree(incumbent, "10") + spec = _spec() + deadline = (WHEN + timedelta(seconds=10)).isoformat() + for lane in spec["lanes"]: + lane["deadline_utc"] = deadline + round_dir = tmp_path / "round" + prepare(spec, incumbent, round_dir) + work = tmp_path / "work" + _tree(work, "8") + ticket = submit(round_dir, "L1", incumbent, work, now=WHEN)["ticket"] + moments = [WHEN, WHEN, WHEN + timedelta(seconds=1), WHEN + timedelta(seconds=11)] + + def clock(): + return moments.pop(0) + + class Counting: + name = "counting" + + def __init__(self) -> None: + self.calls = 0 + + def run(self, *args, **kwargs): + del args, kwargs + self.calls += 1 + return {"status": "completed", "returncode": 0, "stdout": "", "stderr": ""} + + backend = Counting() + rejected = serve_ticket(round_dir, ticket, incumbent, backend=backend, now=WHEN, clock=clock) + assert rejected["status"] == "unsettled" + assert "deadline" in rejected["reason"] + assert backend.calls == 1 + ledger = json.loads((round_dir / "ledgers" / "L1.json").read_text(encoding="utf-8")) + assert [entry["phase"] for entry in ledger["entries"]] == ["claimed"] + assert ticket_status(round_dir, ticket)["status"] == "unsettled" + + +def test_namespace_timeout_after_claim_is_an_unsettled_receipt(tmp_path): + incumbent = tmp_path / "incumbent" + _tree(incumbent, "10") + round_dir = tmp_path / "round" + prepare(_spec(), incumbent, round_dir) + work = tmp_path / "work" + _tree(work, "8") + ticket = submit(round_dir, "L1", incumbent, work, now=WHEN)["ticket"] + + class TimesOut: + name = "timeout" + + def run(self, *args, **kwargs): + del args, kwargs + raise subprocess.TimeoutExpired(["build"], 1) + + result = serve_ticket(round_dir, ticket, incumbent, backend=TimesOut(), now=WHEN) + assert result["status"] == "unsettled" + assert "timed out" in result["reason"] + assert result["reason"].startswith("build") + ledger = json.loads((round_dir / "ledgers" / "L1.json").read_text(encoding="utf-8")) + assert [entry["phase"] for entry in ledger["entries"]] == ["claimed"] + assert ticket_status(round_dir, ticket)["status"] == "unsettled" + + +def test_failure_after_claim_stays_unsettled_across_restart(tmp_path): + incumbent = tmp_path / "incumbent" + _tree(incumbent, "10") + round_dir = tmp_path / "round" + prepare(_spec(), incumbent, round_dir) + work = tmp_path / "work" + _tree(work, "8") + ticket = submit(round_dir, "L1", incumbent, work, now=WHEN)["ticket"] + + class StopAfterClaim: + name = "stop" + + def run(self, *args, **kwargs): + del args, kwargs + raise ConfinementUnavailable("stopped after claim") + + first = serve_ticket( + round_dir, ticket, incumbent, backend=StopAfterClaim(), now=WHEN + ) + assert first["status"] == "unsettled" + ledger_path = round_dir / "ledgers" / "L1.json" + ledger = json.loads(ledger_path.read_text(encoding="utf-8")) + assert [entry["phase"] for entry in ledger["entries"]] == ["claimed"] + (round_dir / "results" / f"{ticket}.json").unlink() + second = serve_ticket( + round_dir, ticket, incumbent, backend=StopAfterClaim(), now=WHEN + ) + assert second["status"] == "unsettled" + reloaded = json.loads(ledger_path.read_text(encoding="utf-8")) + assert [entry["phase"] for entry in reloaded["entries"]] == ["claimed"] + assert reloaded["entries"][0]["worst_case"] == ledger["entries"][0]["worst_case"] diff --git a/tests/test_round_budget.py b/tests/test_round_budget.py new file mode 100644 index 000000000..4f08cb32f --- /dev/null +++ b/tests/test_round_budget.py @@ -0,0 +1,68 @@ +"""Lane attempt budgets. Costs in these tests are plan fields, not campaign gates.""" + +import json +from datetime import datetime, timezone + +import pytest + +from inferrsi.harness.round_budget import ( + BudgetError, + admit_from_path, + advance, + attempt_plan_digest, + empty_ledger, +) + +ROUND = "a" * 64 +WHEN = datetime(2026, 10, 30, tzinfo=timezone.utc) + + +def _plan(path, *, cap=2, worst=1, deadline="2099-01-01T00:00:00+00:00"): + document = { + "round_digest": ROUND, + "lanes": [ + { + "id": "L1", + "cap": cap, + "per_attempt_worst_case": worst, + "deadline_utc": deadline, + } + ], + } + path.write_text(json.dumps(document), encoding="utf-8") + return document + + +def test_retry_with_budget_remaining_keeps_both_charges(tmp_path): + plan = _plan(tmp_path / "round-plan.json", cap=2, worst=1) + ledger = empty_ledger(ROUND, "L1", cap=2) + first, binding = admit_from_path(tmp_path / "round-plan.json", ledger, "L1", WHEN) + cancelled = advance(first, binding, "cancelled") + second, _ = admit_from_path(tmp_path / "round-plan.json", cancelled, "L1", WHEN) + assert second.charged == 2 + assert len(second.entries) == 2 + assert second.entries[0].phase == "cancelled" + assert second.entries[1].phase == "reserved" + assert attempt_plan_digest(ROUND, "L1", 1) != attempt_plan_digest(ROUND, "L1", 2) + assert plan["round_digest"] == ROUND + + +def test_exhausted_budget_rejects_the_next_attempt(tmp_path): + _plan(tmp_path / "round-plan.json", cap=1, worst=1) + ledger = empty_ledger(ROUND, "L1", cap=1) + reserved, binding = admit_from_path(tmp_path / "round-plan.json", ledger, "L1", WHEN) + cancelled = advance(reserved, binding, "cancelled") + with pytest.raises(BudgetError, match="exhausted"): + admit_from_path(tmp_path / "round-plan.json", cancelled, "L1", WHEN) + + +def test_past_deadline_rejects_even_when_budget_remains(tmp_path): + plan_path = tmp_path / "round-plan.json" + _plan(plan_path, cap=5, worst=1, deadline="2020-01-01T00:00:00+00:00") + ledger = empty_ledger(ROUND, "L1", cap=5) + before = ledger.charged + with pytest.raises(BudgetError, match="deadline"): + admit_from_path(plan_path, ledger, "L1", WHEN) + reloaded = json.loads(plan_path.read_text(encoding="utf-8")) + assert reloaded["lanes"][0]["deadline_utc"].startswith("2020-") + assert ledger.charged == before diff --git a/tools/parallel_round.py b/tools/parallel_round.py new file mode 100644 index 000000000..40ddaff59 --- /dev/null +++ b/tools/parallel_round.py @@ -0,0 +1,67 @@ +"""Offline coordinator for one parallel round. Synthetic output is not campaign evidence.""" + +import argparse +import json +import sys +from datetime import datetime, timezone +from pathlib import Path + +from inferrsi.harness.parallel_round import prepare, serve_ticket, submit, ticket_status +from inferrsi.infra.confinement import select_backend + + +def _load(path: str) -> dict: + return json.loads(Path(path).read_text(encoding="utf-8")) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + commands = parser.add_subparsers(dest="command", required=True) + + prepare_command = commands.add_parser("prepare") + prepare_command.add_argument("--spec", required=True) + prepare_command.add_argument("--incumbent", required=True, type=Path) + prepare_command.add_argument("--round-dir", required=True, type=Path) + + submit_command = commands.add_parser("submit") + submit_command.add_argument("--round-dir", required=True, type=Path) + submit_command.add_argument("--lane", required=True) + submit_command.add_argument("--incumbent", required=True, type=Path) + submit_command.add_argument("--work", required=True, type=Path) + + serve_command = commands.add_parser("serve") + serve_command.add_argument("--round-dir", required=True, type=Path) + serve_command.add_argument("--ticket", required=True) + serve_command.add_argument("--incumbent", required=True, type=Path) + + status_command = commands.add_parser("status") + status_command.add_argument("--round-dir", required=True, type=Path) + status_command.add_argument("--ticket", required=True) + + args = parser.parse_args(argv) + if args.command == "prepare": + result = prepare(_load(args.spec), args.incumbent, args.round_dir) + elif args.command == "submit": + result = submit( + args.round_dir, + args.lane, + args.incumbent, + args.work, + now=datetime.now(timezone.utc), + ) + elif args.command == "serve": + result = serve_ticket( + args.round_dir, + args.ticket, + args.incumbent, + backend=select_backend(), + ) + else: + result = ticket_status(args.round_dir, args.ticket) + json.dump(result, sys.stdout, indent=2) + sys.stdout.write("\n") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())