diff --git a/Makefile b/Makefile index 5eb8af45..6bf0ef86 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,7 @@ CHECKER_SOURCES := \ src/common/__init__.py \ src/common/check_proof.py \ + src/common/active_clock.py \ src/common/verification_budget.py \ src/common/verification_toolchain.py \ src/common/cheating_detection.py \ diff --git a/docs/PAUSING.md b/docs/PAUSING.md new file mode 100644 index 00000000..b85ba76a --- /dev/null +++ b/docs/PAUSING.md @@ -0,0 +1,78 @@ +# Pausing Docker experiments + +New Docker runs on Linux hosts provide a host-owned clock for managed pauses. Use: + +```sh +tlaps-bench pause-container CONTAINER_NAME +tlaps-bench resume-container CONTAINER_NAME +``` + +A quota guard can call these commands in place of `docker pause` and `docker +unpause`, checking the exit status before recording a successful transition. +Both commands are idempotent. The container name or ID is the same one accepted +by Docker. Preflight containers are deliberately not pausable through this API. + +The runner's agent deadline and `agent_time_secs`, the grader envelope, and the +official checker's SANY/TLAPM envelope subtract recorded pauses. Ordinary idle +time, network/API waits, and solver execution still consume time. This is an +elapsed-time budget, not a CPU-time budget. + +Claude Code benchmark launches set `CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS=0` in +their child environment. This disables print mode's wall-clock ceiling for +pending background tasks, which can otherwise terminate them immediately after +a long freeze. The benchmark's configured `--timeout` remains the execution +limit; `--timeout 0` intentionally has no overall limit. The caller's environment +is unchanged. This setting does not suspend raw shell `timeout` commands or +TLAPM/solver internal timers, and does not alter already-running Claude sessions. + +Docker agent results also record `agent_wall_time_secs` and +`agent_paused_time_secs`; the difference is the measured active process runtime. +On timeout, the existing `agent_time_secs` cap can additionally exclude shutdown +or output-draining overhead. Verification +reports retain `wall_secs` and add `active_secs` and `paused_secs`. Check-session +recovery charges the persisted active budget; old sessions without these fields +conservatively charge their entire recorded wall time. Self-check and formal +grading keep independent budgets and cache namespaces. + +Each launch owns a durable ledger under `~/.tlaps-bench/runtime-clocks/`, outside +the agent's writable mounts. The directory is mounted read-only at +`/run/tlaps-bench-clock`. Its host path is recorded in the Docker label +`org.specula.tlaps-bench.active-clock`; an `active-clock.json` snapshot is copied +beside the launch's output after it exits. Retain the host ledger while retaining +or inspecting its container. The clock uses the Linux boot ID and monotonic time +and the host rejects a ledger from a previous host boot. The runner and Docker +container must share the Linux monotonic clock domain. An incompatible clock +epoch, or a host wall-clock adjustment over five seconds since clock creation, +fails closed rather than producing an untrustworthy budget measurement. + +The controller freezes the container before writing the pause start, and durably +closes the interval before thawing it. Host clock readers share the controller's +lock so they cannot observe a half-published transition. Container readers check +for concurrent publication without holding that lock across a freeze. +Freeze-command transit before the recorded start and thaw-command transit after +the close are conservatively charged. If the controller crashes between Docker and ledger +updates, rerunning the command recovers the Docker state, but any unrecorded +interval stays charged and the command reports that fact. A checker hard crash +retains the existing write-ahead reservation of at most one heartbeat tick; +restart downtime is not a new examination. As before, incomplete measurements +after a hard crash should not be treated as complete wall/CPU telemetry. + +Boundaries: + +- This requires a new run and a checker image built from the updated source. + Existing containers lack the clock mount; the control commands reject them. + It does not retroactively repair previous experiment results. +- Docker launches from non-Linux hosts keep their existing wall-clock budgets + and do not expose managed pause control. +- Plain `docker pause/unpause`, process signals, machine suspension, and an + external guard that has not adopted these commands are not tracked. In + particular, never thaw a managed pause directly: the ledger would remain + paused. The controller detects that mismatch on its next invocation and + rejects it; that run's accounting needs investigation. +- A backend that propagates an absolute deadline to its own child rejects + managed pauses until that backend supports the active clock. +- Arbitrary shell `timeout`, CLI/tool timeouts, and TLAPM/solver internal tactic + timers retain their own semantics. They can expire during a freeze even though + the official checker envelope has time left. An experiment that requires those + internal timers to be unaffected must pause between checks instead of freezing + an active check. No proof-validation rule or timeout verdict is relaxed. diff --git a/src/common/active_clock.py b/src/common/active_clock.py new file mode 100644 index 00000000..cc7ee30b --- /dev/null +++ b/src/common/active_clock.py @@ -0,0 +1,168 @@ +"""Elapsed wall time minus explicitly recorded, host-controlled container pauses.""" + +from __future__ import annotations + +import fcntl +import json +import math +import os +import subprocess +import tempfile +import time +from dataclasses import dataclass +from pathlib import Path + +CLOCK_ENV = "TLAPS_ACTIVE_CLOCK" +CLOCK_MOUNT = "/run/tlaps-bench-clock" +CLOCK_LABEL = "org.specula.tlaps-bench.active-clock" +POLL_SECS = 0.1 + + +def boot_id() -> str: + return Path("/proc/sys/kernel/random/boot_id").read_text().strip() + + +def write_state(path: Path, state: dict) -> None: + """Publish before thawing; fsync both the file and its directory.""" + with tempfile.NamedTemporaryFile(mode="w", dir=path.parent, delete=False) as stream: + temporary = Path(stream.name) + try: + json.dump(state, stream, allow_nan=False, sort_keys=True) + stream.flush() + os.fsync(stream.fileno()) + os.chmod(temporary, 0o644) + os.replace(temporary, path) + fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(fd) + finally: + os.close(fd) + finally: + temporary.unlink(missing_ok=True) + + +@dataclass(frozen=True) +class ClockReading: + wall: float + paused: float + + @property + def active(self) -> float: + return self.wall - self.paused + + +class ActiveClock: + def __init__(self, path: Path | None = None, *, verify_host_boot: bool = True): + self.path = path + self._host_reader = verify_host_boot + self._boot_id = boot_id() if path and verify_host_boot else None + + @classmethod + def from_environment(cls): + value = os.environ.get(CLOCK_ENV) + # A container's proc namespace can expose a different boot ID even + # while sharing CLOCK_MONOTONIC with the host. The host controller + # validates the host boot; readers pin the ledger's initial identity. + return cls(Path(value) if value else None, verify_host_boot=False) + + @classmethod + def create(cls, directory: Path, *, pause_supported: bool = True): + directory.mkdir(parents=True, exist_ok=False) + directory.chmod(0o755) + path = directory / "state.json" + (directory / "control.lock").touch(mode=0o644) + write_state( + path, + { + "version": 1, + "boot_id": boot_id(), + "pauses": [], + "pause_supported": pause_supported, + "monotonic_origin": time.monotonic(), + "wall_origin": time.time(), + }, + ) + return cls(path) + + def read_state(self) -> dict: + try: + state = json.loads(self.path.read_text()) + if state["version"] != 1 or not isinstance(state["boot_id"], str) or not state["boot_id"]: + raise ValueError("incompatible clock or host reboot") + if self._boot_id is None: + self._boot_id = state["boot_id"] + elif state["boot_id"] != self._boot_id: + raise ValueError("incompatible clock or host reboot") + if type(state["pause_supported"]) is not bool or type(state["pauses"]) is not list: + raise ValueError("invalid pause ledger") + for field in ("monotonic_origin", "wall_origin"): + if type(state[field]) not in (float, int) or not math.isfinite(state[field]): + raise ValueError("invalid clock origin") + previous = 0.0 + for index, interval in enumerate(state["pauses"]): + start, end = interval + if type(start) not in (float, int) or not math.isfinite(start) or start < previous: + raise ValueError("invalid pause start") + if end is None: + if index != len(state["pauses"]) - 1: + raise ValueError("unfinished pause is not last") + elif type(end) not in (float, int) or not math.isfinite(end) or end < start: + raise ValueError("invalid pause end") + previous = end if end is not None else start + return state + except (OSError, ValueError, TypeError, KeyError) as exc: + raise OSError(f"cannot read trusted active clock: {exc}") from exc + + def read(self) -> ClockReading: + if self.path is None: + return ClockReading(time.monotonic(), 0.0) + if self._host_reader: + # Host observations must not race a controller's sampled timestamp + # and its durable publication. Container readers cannot take this + # lock: freezing a shared-lock holder would deadlock the controller. + with (self.path.parent / "control.lock").open("r") as lock: + fcntl.flock(lock, fcntl.LOCK_SH) + return self._read_snapshot() + return self._read_snapshot() + + def _read_snapshot(self) -> ClockReading: + # A container may freeze between *any* two instructions, including + # between reading the ledger and sampling monotonic. Verify the same + # durable state on both sides of the timestamp before using it. + while True: + state = self.read_state() + now = time.monotonic() + wall = time.time() + if state == self.read_state(): + break + if abs((now - state["monotonic_origin"]) - (wall - state["wall_origin"])) > 5: + raise OSError("active clock has a different monotonic epoch or the host wall clock changed by over 5s") + paused = 0.0 + for start, end in state["pauses"]: + if start > now or (end is not None and end > now): + raise OSError("active-clock interval lies in the future") + paused += (now if end is None else end) - start + return ClockReading(now, paused) + + def now(self) -> float: + return self.read().active + + +def communicate(proc, timeout, *, clock: ActiveClock | None = None, input=None): + """Drain pipes while bounding active elapsed, including ordinary idle/API waits. + + A short wall-clock poll may expire while the container is frozen. Recheck + the host ledger before deciding that the actual budget has expired. + """ + clock = clock or ActiveClock.from_environment() + if clock.path is None or timeout is None: + return proc.communicate(input=input, timeout=timeout) + deadline = clock.now() + timeout + while True: + remaining = deadline - clock.now() + try: + return proc.communicate(input=input, timeout=max(0, min(POLL_SECS, remaining))) + except subprocess.TimeoutExpired as exc: + input = None # Popen retains partially written input across calls. + if clock.now() >= deadline: + raise subprocess.TimeoutExpired(exc.cmd, timeout, output=exc.output, stderr=exc.stderr) from exc diff --git a/src/common/check_proof.py b/src/common/check_proof.py index 7cbb43da..6dc99ebc 100644 --- a/src/common/check_proof.py +++ b/src/common/check_proof.py @@ -63,7 +63,6 @@ import subprocess import sys import tempfile -import time from concurrent.futures import ThreadPoolExecutor from dataclasses import asdict from pathlib import Path @@ -204,7 +203,9 @@ def run_killgroup(cmd, timeout, cwd): start_new_session=True, ) try: - out, err = proc.communicate(timeout=timeout) + from common.active_clock import communicate + + out, err = communicate(proc, timeout) return out, err, proc.returncode except BaseException: # Interrupting a check must also stop its backends before recovery. @@ -1217,7 +1218,9 @@ def _module_unit_result( "SANY did not provide a valid theorem location", ) - remaining = None if deadline is None else deadline - time.monotonic() + from common.active_clock import ActiveClock + + remaining = None if deadline is None else deadline - ActiveClock.from_environment().now() if remaining is not None and remaining <= 0: return ( { @@ -1463,7 +1466,7 @@ def _run_module_task_check( emit(f"ERROR: cannot stage module task inputs: {exc}") return 3 - remaining = None if deadline is None else deadline - time.monotonic() + remaining = session.remaining() if remaining is not None and remaining <= 0 and session.load("submitted-sany") is None: emit("SANY-STATUS: unavailable") emit("CHECK-TIMEOUT: module checker budget exhausted before SANY validation") @@ -1498,7 +1501,7 @@ def _run_module_task_check( return 1 assert sany_run.raw is not None - remaining = None if deadline is None else deadline - time.monotonic() + remaining = session.remaining() if remaining is not None and remaining <= 0 and session.load("canonical-sany") is None: emit("CHECK-TIMEOUT: module checker budget exhausted before canonical SANY validation") emit("ERROR: module checker timeout exhausted before canonical SANY validation") diff --git a/src/common/container.py b/src/common/container.py index eb554dd1..2a6355bb 100644 --- a/src/common/container.py +++ b/src/common/container.py @@ -18,6 +18,8 @@ from dataclasses import dataclass, field from pathlib import Path +from common.active_clock import CLOCK_ENV, CLOCK_LABEL, CLOCK_MOUNT, ActiveClock, communicate + IMAGE_TAG = "tlaps-bench-base" _REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) _IMAGE_BUILD_FINGERPRINT_LABEL = "org.specula.tlaps-bench.build-sha256" @@ -148,6 +150,9 @@ class ContainerConfig: # Optional stdout marker emitted after install + firewall setup and # immediately before the agent command starts. agent_start_marker: str = "" + pause_supported: bool = True + pause_accounting: bool = False + active_clock: ActiveClock | None = None @dataclass @@ -156,6 +161,7 @@ class ContainerRun: proc: subprocess.Popen container_id: str + active_clock: ActiveClock | None = None @dataclass(frozen=True) @@ -349,6 +355,11 @@ def build_docker_args(self, config: ContainerConfig) -> tuple[list[str], str]: args.extend(["-v", f"{config.result_dir}:/results:rw"]) if config.benchmark_dir: args.extend(["-v", f"{config.benchmark_dir}:/benchmark:ro"]) + if config.active_clock is not None: + directory = str(config.active_clock.path.parent) + args.extend(["-v", f"{directory}:{CLOCK_MOUNT}:ro"]) + args.extend(["--label", f"{CLOCK_LABEL}={directory}"]) + args.extend(["-e", f"{CLOCK_ENV}={CLOCK_MOUNT}/state.json"]) # tlapm is baked into the image at /opt/tlapm @@ -422,6 +433,7 @@ def build_composite_command( def run(self, config: ContainerConfig, cmd: list[str], stdin_data: str | None = None) -> ContainerRun: """Launch a container with the given command. Returns handle.""" + self._prepare_active_clock(config) docker_args, cid_file = self.build_docker_args(config) composite = self.build_composite_command( cmd, @@ -447,12 +459,24 @@ def run(self, config: ContainerConfig, cmd: list[str], stdin_data: str | None = pass container_id = self._read_cidfile(cid_file) - return ContainerRun(proc=proc, container_id=container_id) + return ContainerRun(proc=proc, container_id=container_id, active_clock=config.active_clock) + + @staticmethod + def _prepare_active_clock(config: ContainerConfig) -> None: + if not config.pause_accounting or not sys.platform.startswith("linux"): + return + if config.active_clock is None: + # Never place the trusted ledger under /workspace, /results, or an + # agent session mount. Keep it after exit for retained containers + # and interruption diagnosis; the Docker label identifies it. + directory = Path.home() / ".tlaps-bench" / "runtime-clocks" / uuid.uuid4().hex + config.active_clock = ActiveClock.create(directory, pause_supported=config.pause_supported) def run_with_output( self, config: ContainerConfig, cmd: list[str], stdin_data: str | None = None, timeout: int | None = None ) -> tuple[int, str, str]: """Run container to completion. Returns (exit_code, stdout, stderr).""" + self._prepare_active_clock(config) docker_args, cid_file = self.build_docker_args(config) composite = self.build_composite_command( cmd, @@ -463,18 +487,25 @@ def run_with_output( full_cmd = docker_args + ["bash", "-c", composite] try: - result = subprocess.run( + if config.active_clock is None: + completed = subprocess.run(full_cmd, input=stdin_data, capture_output=True, text=True, timeout=timeout) + return completed.returncode, completed.stdout, completed.stderr + proc = subprocess.Popen( full_cmd, - input=stdin_data, - capture_output=True, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True, - timeout=timeout, ) - return result.returncode, result.stdout, result.stderr - except subprocess.TimeoutExpired: + stdout, stderr = communicate(proc, timeout, clock=config.active_clock, input=stdin_data) + return proc.returncode, stdout, stderr + except BaseException: cid = self._read_cidfile(cid_file) if cid: self.kill_by_id(cid) + if "proc" in locals(): + proc.kill() + proc.communicate() raise def run_preflight( diff --git a/src/common/pause_control.py b/src/common/pause_control.py new file mode 100644 index 00000000..9e113b61 --- /dev/null +++ b/src/common/pause_control.py @@ -0,0 +1,77 @@ +"""Host-only, durable pause/resume control for benchmark containers.""" + +from __future__ import annotations + +import argparse +import fcntl +import json +import subprocess +import time +from pathlib import Path + +from common.active_clock import CLOCK_LABEL, CLOCK_MOUNT, ActiveClock, write_state + + +def inspect_container(name: str) -> dict: + completed = subprocess.run( + ["docker", "inspect", "--", name], capture_output=True, text=True, check=True, timeout=30 + ) + return json.loads(completed.stdout)[0] + + +def control(name: str, *, pause: bool) -> str: + info = inspect_container(name) + directory = info["Config"].get("Labels", {}).get(CLOCK_LABEL) + if not directory: + raise ValueError( + "container has no managed active clock; requires a new Linux-host run with a rebuilt checker image" + ) + if not any( + mount.get("Source") == directory and mount.get("Destination") == CLOCK_MOUNT and mount.get("RW") is False + for mount in info.get("Mounts", []) + ): + raise ValueError("container has no read-only active-clock mount") + clock = ActiveClock(Path(directory) / "state.json") + with (Path(directory) / "control.lock").open("a") as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + state = clock.read_state() + clock._read_snapshot() # Already holding the exclusive controller lock. + if not state["pause_supported"]: + raise ValueError("this backend propagates absolute deadlines and does not support managed pauses") + info = inspect_container(info["Id"]) + if not info["State"]["Running"]: + raise ValueError("container is not running") + frozen = info["State"]["Paused"] + recorded = bool(state["pauses"] and state["pauses"][-1][1] is None) + if recorded and not frozen: + raise ValueError("container was thawed outside the pause controller; accounting is invalid") + if pause: + if recorded: + return "already paused" + if not frozen: + subprocess.run(["docker", "pause", info["Id"]], check=True, capture_output=True, timeout=30) + # Freeze first. A controller crash before this write conservatively + # charges the gap until recovery rather than granting free runtime. + state["pauses"].append([time.monotonic(), None]) + write_state(clock.path, state) + return "paused" if not frozen else "paused; unrecorded frozen time before recovery remains charged" + if not frozen: + return "already running" + if recorded: + # Publish the complete interval BEFORE unpause: no checker can wake + # and accidentally charge the entire frozen interval against its budget. + state["pauses"][-1][1] = time.monotonic() + write_state(clock.path, state) + subprocess.run(["docker", "unpause", info["Id"]], check=True, capture_output=True, timeout=30) + return "resumed" if recorded else "resumed; unrecorded frozen time remains charged" + + +def main(*, pause: bool) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("container", help="benchmark Docker container name or ID") + args = parser.parse_args() + try: + print(control(args.container, pause=pause)) + return 0 + except (OSError, ValueError, KeyError, subprocess.SubprocessError) as exc: + parser.exit(1, f"pause control failed: {exc}\n") diff --git a/src/common/verification_budget.py b/src/common/verification_budget.py index c7a8d52e..e40c4dae 100644 --- a/src/common/verification_budget.py +++ b/src/common/verification_budget.py @@ -11,10 +11,11 @@ import resource import shutil import threading -import time from dataclasses import asdict, dataclass from pathlib import Path +from common.active_clock import ActiveClock + POLICY_VERSION = "pfs-module-budget-v1" POLICY_ENV = "TLAPS_VERIFICATION_POLICY" TOOLCHAIN_ENV = "TLAPS_VERIFICATION_TOOLCHAIN" @@ -178,6 +179,7 @@ def __init__(self, directory: Path, identity: dict, timeout: float): self._mutex = threading.RLock() self._stop = threading.Event() self._error = None + self.clock = ActiveClock.from_environment() def __enter__(self): self.directory.mkdir(parents=True, exist_ok=True) @@ -196,19 +198,30 @@ def __enter__(self): raise ValueError("invalid check-session state flags") if self.state.get("identity") != self.identity: raise ValueError("cannot resume check with different candidate, canonical inputs, tools, or policy") - for field in ("wall_secs", "cpu_secs"): + for field in ("wall_secs", "cpu_secs", "active_secs", "paused_secs"): + if field not in self.state: + continue value = self.state[field] if type(value) not in (int, float) or not math.isfinite(value) or value < 0: raise ValueError(f"invalid check-session {field}") + if {"active_secs", "paused_secs"} & self.state.keys() and ( + not {"active_secs", "paused_secs"} <= self.state.keys() + or not math.isclose( + self.state["wall_secs"], self.state["active_secs"] + self.state["paused_secs"], abs_tol=1e-6 + ) + ): + raise ValueError("inconsistent check-session active/pause accounting") if self.state.get("active"): self.state["cpu_complete"] = False else: self.state = {"identity": self.identity, "wall_secs": 0.0, "cpu_secs": 0.0, "cpu_complete": True} self.prior_wall = self.state["wall_secs"] + self.prior_active = self.state.get("active_secs", self.prior_wall) + self.prior_paused = self.state.get("paused_secs", 0.0) self.prior_cpu = self.state["cpu_secs"] - self.started = time.monotonic() + self.started = self.clock.read() self.cpu_started = cpu_seconds() - self.deadline = self.started + max(0, self.timeout - self.prior_wall) + self.deadline = self.started.active + max(0, self.timeout - self.prior_active) self._save(active=True) self._thread = threading.Thread(target=self._heartbeat, daemon=True) self._thread.start() @@ -219,9 +232,13 @@ def __enter__(self): def _save(self, *, active: bool): with self._mutex: - wall = self.prior_wall + time.monotonic() - self.started + reading = self.clock.read() + active_secs = self.prior_active + reading.active - self.started.active + reservation = min(HEARTBEAT_SECS, max(0, self.timeout - active_secs)) if active else 0 self.state.update( - wall_secs=wall + (min(HEARTBEAT_SECS, max(0, self.timeout - wall)) if active else 0), + wall_secs=self.prior_wall + reading.wall - self.started.wall + reservation, + active_secs=active_secs + reservation, + paused_secs=self.prior_paused + reading.paused - self.started.paused, cpu_secs=self.prior_cpu + max(0, cpu_seconds() - self.cpu_started), active=active, ) @@ -238,11 +255,14 @@ def _heartbeat(self): def remaining(self) -> float: if self._error: raise OSError(f"cannot persist check budget: {self._error}") - return max(0, self.deadline - time.monotonic()) + return max(0, self.deadline - self.clock.now()) def metrics(self) -> dict: + reading = self.clock.read() return { - "wall_secs": self.prior_wall + time.monotonic() - self.started, + "wall_secs": self.prior_wall + reading.wall - self.started.wall, + "active_secs": self.prior_active + reading.active - self.started.active, + "paused_secs": self.prior_paused + reading.paused - self.started.paused, "cpu_secs": self.prior_cpu + max(0, cpu_seconds() - self.cpu_started), "cpu_complete": self.state["cpu_complete"], "cpu_source": "getrusage_self_and_reaped_children", diff --git a/src/evaluator/backends/claude_code.py b/src/evaluator/backends/claude_code.py index d7740f0e..45bbd1fc 100644 --- a/src/evaluator/backends/claude_code.py +++ b/src/evaluator/backends/claude_code.py @@ -593,6 +593,12 @@ def __init__(self, model: str | None = None): self.model = model or DEFAULT_MODEL self.provider = "amazon-bedrock" if self._uses_aws_provider() else PROVIDER + def execution_environment(self, result_dir: str) -> dict[str, str]: + # Claude's print-mode background ceiling uses wall time, including + # container freezes. Keep pending checks alive until they finish or + # the benchmark's own execution deadline ends the attempt. + return {**super().execution_environment(result_dir), "CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS": "0"} + def get_credential_mounts(self) -> list[str]: if self._uses_aws_provider() and needs_aws_shared_credentials(): return ["aws"] diff --git a/src/evaluator/proof_module_result.py b/src/evaluator/proof_module_result.py index 6e51f806..dfbb6a91 100644 --- a/src/evaluator/proof_module_result.py +++ b/src/evaluator/proof_module_result.py @@ -42,13 +42,15 @@ def _string_list(value: object, *, label: str) -> list[str]: def validate_verification_metrics(metrics, expected=None): - if type(metrics) is not dict or set(metrics) != { + required = { "policy", "wall_secs", "cpu_secs", "cpu_complete", "cpu_source", - }: + } + pause_fields = {"active_secs", "paused_secs"} + if type(metrics) is not dict or set(metrics) not in (required, required | pause_fields): raise ModuleResultError("invalid module verification metrics") try: policy = VerificationPolicy.from_dict(metrics["policy"]) @@ -56,9 +58,13 @@ def validate_verification_metrics(metrics, expected=None): raise ModuleResultError(str(exc)) from exc if expected is not None and policy.proof_unit_ids != tuple(expected): raise ModuleResultError("verification policy has different proof units") - for field in ("wall_secs", "cpu_secs"): + for field in ("wall_secs", "cpu_secs", *sorted(pause_fields & metrics.keys())): if type(metrics[field]) not in (int, float) or not math.isfinite(metrics[field]) or metrics[field] < 0: raise ModuleResultError(f"invalid verification {field}") + if pause_fields <= metrics.keys() and not math.isclose( + metrics["wall_secs"], metrics["active_secs"] + metrics["paused_secs"], abs_tol=1e-6 + ): + raise ModuleResultError("verification wall time differs from active plus paused time") if type(metrics["cpu_complete"]) is not bool or metrics["cpu_source"] != "getrusage_self_and_reaped_children": raise ModuleResultError("invalid verification CPU accounting") diff --git a/src/evaluator/runner.py b/src/evaluator/runner.py index 0dcdf944..53ceeac3 100644 --- a/src/evaluator/runner.py +++ b/src/evaluator/runner.py @@ -42,6 +42,7 @@ from dataclasses import dataclass, replace from pathlib import Path +from common.active_clock import ActiveClock from common.container import ( IMAGE_TAG, ContainerConfig, @@ -167,6 +168,8 @@ "input_tokens", "output_tokens", "agent_time_secs", + "agent_wall_time_secs", + "agent_paused_time_secs", "grading_time_secs", "grader_error_time_secs", "logical_timeout_used_secs", @@ -264,6 +267,7 @@ def _write_attempt_accounting( time_secs: float | None, usage: UsageSummary, backend: Backend, + pause_metrics: dict | None = None, ) -> None: """Store non-experiment accounting beside an infra/quota attempt.""" @@ -271,6 +275,14 @@ def _write_attempt_accounting( return os.makedirs(directory, exist_ok=True) diagnostic: dict[str, object] = {"time_secs": time_secs} + if pause_metrics: + diagnostic.update( + { + key: pause_metrics[key] + for key in ("agent_wall_time_secs", "agent_paused_time_secs") + if key in pause_metrics + } + ) _attach_equivalent_cost(diagnostic, usage, backend) with open(os.path.join(directory, "accounting.json"), "w") as stream: json.dump(diagnostic, stream, indent=2) @@ -752,6 +764,8 @@ def _reset_logical_attempt_segment(result: dict[str, object], backend: Backend) result["termination_reason"] = TerminationReason.OK result["error"] = "" result["agent_time_secs"] = None if _supports_cost_time(backend) else 0.0 + result.pop("agent_wall_time_secs", None) + result.pop("agent_paused_time_secs", None) result["grading_time_secs"] = 0.0 result["logical_timeout_used_secs"] = 0.0 result["time_secs"] = None if _supports_cost_time(backend) else 0.0 @@ -808,6 +822,9 @@ def _merge_logical_attempt_accounting( result["input_tokens"] = aggregate_usage.legacy_input_tokens result["output_tokens"] = aggregate_usage.legacy_output_tokens result["agent_time_secs"] = _sum_accounting_values(previous.get("agent_time_secs"), result.get("agent_time_secs")) + for field in ("agent_wall_time_secs", "agent_paused_time_secs"): + if field in previous or field in result: + result[field] = _sum_accounting_values(previous.get(field), result.get(field)) result["grading_time_secs"] = _sum_accounting_values( previous.get("grading_time_secs"), result.get("grading_time_secs") ) @@ -1233,6 +1250,8 @@ def _run_backend_with_retries( def _run_once(workspace=workspace, canonical_dir=canonical_dir): nonlocal active_secs, attempt_secs, attempt_timeout_used_secs result["error"] = "" + result.pop("agent_wall_time_secs", None) + result.pop("agent_paused_time_secs", None) # Establish a valid empty-stream marker before process/container # startup. A launch exception is then distinguishable from a # child that flushed a malformed or truncated event. @@ -1272,6 +1291,7 @@ def _run_once(workspace=workspace, canonical_dir=canonical_dir): canonical_dir, ) wall_elapsed = time.monotonic() - t0 + wall_elapsed = max(0.0, wall_elapsed - result.pop("_container_paused_secs", 0.0)) elapsed = container_agent_secs if item.use_container and _supports_cost_time(backend) else wall_elapsed # Cooperative backends may spend a bounded grace period flushing # audit events after the logical deadline. That is not extra model @@ -1311,6 +1331,7 @@ def _prepare_quota_retry(quota_attempt: int, infra_attempt: int = attempt) -> bo backend, time_secs=attempt_secs, usage=quota_usage, + pause_metrics=result, ) return True @@ -1398,6 +1419,7 @@ def _prepare_quota_retry(quota_attempt: int, infra_attempt: int = attempt) -> bo backend, time_secs=attempt_secs, usage=attempt_usage, + pause_metrics=result, ) if fixed_workspace is None: shutil.rmtree(workspace, ignore_errors=True) @@ -1440,6 +1462,7 @@ def _prepare_quota_retry(quota_attempt: int, infra_attempt: int = attempt) -> bo time_secs=attempt_secs, usage=attempt_usage, backend=backend, + pause_metrics=result, ) attempt_usage = UsageSummary( sources=("runner.non_experiment_attempt",), @@ -1447,6 +1470,9 @@ def _prepare_quota_retry(quota_attempt: int, infra_attempt: int = attempt) -> bo warnings=("infra/quota accounting is stored separately under agent artifacts",), ) result["agent_time_secs"] = None + for field in ("agent_wall_time_secs", "agent_paused_time_secs"): + if field in result: + result[field] = None result["logical_timeout_used_secs"] = 0.0 result["time_secs"] = None result["equivalent_cost_usd"] = None @@ -2647,6 +2673,9 @@ def run_single_benchmark(item: WorkItem): with open(os.path.join(agent_dir, "transcript.txt"), "w") as f: f.write(f"Benchmark: {rel_path}\n") f.write(f"Agent time: {_format_task_time(result.get('agent_time_secs'))}\n") + if "agent_wall_time_secs" in result: + f.write(f"Agent wall time: {_format_task_time(result.get('agent_wall_time_secs'))}\n") + f.write(f"Managed pause time: {_format_task_time(result.get('agent_paused_time_secs'))}\n") if _supports_cost_time(backend): f.write(f"Equivalent cost: {_format_equivalent_cost(result.get('equivalent_cost_usd'))}\n") f.write(f"Tokens: {format_token_usage(result_token_usage(result), verbose=True)}\n") @@ -2878,16 +2907,17 @@ def _stash_failed_attempt( *, time_secs: float | None, usage: UsageSummary, + pause_metrics: dict | None = None, ) -> None: """Move a failed attempt's raw outputs to agent/attempts/attempt-N/: the retry starts clean (no stale stderr.txt) and the evidence stays debuggable.""" dest = os.path.join(agent_dir, "attempts", f"attempt-{attempt}") os.makedirs(dest, exist_ok=True) - for fname in ("output.jsonl", "stderr.txt", *backend.attempt_output_files()): + for fname in ("output.jsonl", "stderr.txt", "active-clock.json", *backend.attempt_output_files()): src = os.path.join(agent_dir, fname) if os.path.isfile(src): shutil.move(src, os.path.join(dest, fname)) - _write_attempt_accounting(dest, time_secs=time_secs, usage=usage, backend=backend) + _write_attempt_accounting(dest, time_secs=time_secs, usage=usage, backend=backend, pause_metrics=pause_metrics) def _stash_quota_attempt( @@ -2898,6 +2928,7 @@ def _stash_quota_attempt( *, time_secs: float | None, usage: UsageSummary, + pause_metrics: dict | None = None, ) -> None: """Preserve a no-model-work hard-cap launch before the next launch replaces it.""" @@ -2907,11 +2938,11 @@ def _stash_quota_attempt( f"infra-{infra_attempt}-quota-{quota_attempt}", ) os.makedirs(dest, exist_ok=True) - for fname in ("output.jsonl", "stderr.txt", *backend.attempt_output_files()): + for fname in ("output.jsonl", "stderr.txt", "active-clock.json", *backend.attempt_output_files()): src = os.path.join(agent_dir, fname) if os.path.isfile(src): shutil.move(src, os.path.join(dest, fname)) - _write_attempt_accounting(dest, time_secs=time_secs, usage=usage, backend=backend) + _write_attempt_accounting(dest, time_secs=time_secs, usage=usage, backend=backend, pause_metrics=pause_metrics) def _run_continuations( @@ -3023,6 +3054,11 @@ def _run_continuations( round_result["input_tokens"] = round_usage.legacy_input_tokens round_result["output_tokens"] = round_usage.legacy_output_tokens segment_agent_time = round_result.get("agent_time_secs") + segment_pause_metrics = { + field: round_result.get(field) + for field in ("agent_wall_time_secs", "agent_paused_time_secs") + if field in round_result + } segment_grading_time = round_result.get("grading_time_secs") segment_equivalent_cost = round_result.get("equivalent_cost_usd") segment_tool_calls = copy.deepcopy(round_result.get("tool_calls")) @@ -3067,6 +3103,8 @@ def _run_continuations( result.get("agent_time_secs"), segment_agent_time, ) + for field, value in segment_pause_metrics.items(): + result[field] = _sum_accounting_values(result.get(field), value) result["grading_time_secs"] = _sum_accounting_values( result.get("grading_time_secs"), segment_grading_time, @@ -3241,6 +3279,8 @@ def _run_backend_container( credential_mounts=backend.get_credential_mounts(), keep_container=item.keep_container, agent_start_marker=(f"__TLAPS_BENCH_AGENT_START_{uuid.uuid4().hex}__" if _supports_cost_time(backend) else ""), + pause_supported=propagated_deadline is None, + pause_accounting=True, ) name_no_ext = os.path.splitext(os.path.basename(item.benchmark_path))[0] safe = re.sub(r"[^A-Za-z0-9_.-]", "_", name_no_ext) @@ -3276,8 +3316,27 @@ def _run_backend_container( container_run = None agent_started_at: float | None = None agent_ended_at: float | None = None + active_clock = None + agent_started_reading = None + agent_ended_reading = None + + def finish_agent(): + nonlocal agent_ended_at, agent_ended_reading + if agent_ended_at is None: + agent_ended_reading = active_clock.read() if active_clock else None + agent_ended_at = agent_ended_reading.active if agent_ended_reading else time.monotonic() + + def deadline_now(): + # The host remains runnable while Docker freezes the child. Subtract + # only completed/ongoing managed pauses, never ordinary idle/API waits. + if active_clock is not None and propagated_deadline is None: + return active_clock.now() + return time.time() + try: container_run = runner.run(config, cmd, stdin_data=stdin_data) + candidate_clock = getattr(container_run, "active_clock", None) + active_clock = candidate_clock if isinstance(candidate_clock, ActiveClock) else None if item.keep_container: name = config.container_name print( @@ -3303,7 +3362,7 @@ def _run_backend_container( fcntl.fcntl(proc.stderr, fcntl.F_SETFL, flags | os.O_NONBLOCK) stderr_chunks: list[str] = [] - logical_deadline = propagated_deadline or (time.time() + timeout if timeout else None) + logical_deadline = propagated_deadline or (deadline_now() + timeout if timeout else None) grace = backend.capabilities.timeout_drain_grace if propagated_deadline is not None else 0.0 hard_deadline = logical_deadline + grace if logical_deadline is not None else None timed_out = False @@ -3311,7 +3370,7 @@ def _run_backend_container( # Stream stdout to file in real-time (and stderr separately) with open(agent_jsonl, "w") as jsonl_f: while True: - now = time.time() + now = deadline_now() if logical_deadline is not None and now >= logical_deadline: timed_out = True if timed_out and hard_deadline is not None and now >= hard_deadline: @@ -3319,8 +3378,8 @@ def _run_backend_container( runner.kill(container_run) result["agent_exit"] = -1 result["error"] = f"{backend.name} timeout after {item.timeout}s" - agent_ended_at = time.monotonic() - return agent_ended_at - agent_started_at if agent_started_at is not None else None + finish_agent() + break boundary = hard_deadline if timed_out else logical_deadline poll_timeout = min(5.0, max(boundary - now, 0.0)) if boundary is not None else 5.0 @@ -3351,7 +3410,10 @@ def _run_backend_container( for line in lines: if config.agent_start_marker and line.rstrip("\r\n") == config.agent_start_marker: if agent_started_at is None: - agent_started_at = time.monotonic() + agent_started_reading = active_clock.read() if active_clock else None + agent_started_at = ( + agent_started_reading.active if agent_started_reading else time.monotonic() + ) else: jsonl_f.write(line) jsonl_f.flush() @@ -3359,17 +3421,17 @@ def _run_backend_container( sys.stdout.write(line) sys.stdout.flush() if proc.poll() is not None and agent_ended_at is None: - agent_ended_at = time.monotonic() + finish_agent() if not chunk and proc.returncode is not None: break elif proc.poll() is not None: if agent_ended_at is None: - agent_ended_at = time.monotonic() + finish_agent() break if agent_started_at is not None and agent_ended_at is None: - agent_ended_at = time.monotonic() - if logical_deadline is not None and time.time() >= logical_deadline: + finish_agent() + if logical_deadline is not None and deadline_now() >= logical_deadline: timed_out = True result["agent_exit"] = -1 if timed_out else proc.returncode # Drain any remaining stderr @@ -3394,12 +3456,27 @@ def _run_backend_container( if container_run: runner.kill(container_run) if agent_started_at is not None: - agent_ended_at = time.monotonic() + with contextlib.suppress(OSError): + finish_agent() finally: - # A retained container keeps its credential mount sources so a - # `docker start` can still authenticate. - if not item.keep_container: - runner.cleanup_credential_tmps() + try: + if active_clock is not None: + result["_container_paused_secs"] = active_clock.read().paused + if agent_started_reading is not None and agent_ended_reading is not None: + result["agent_wall_time_secs"] = agent_ended_reading.wall - agent_started_reading.wall + result["agent_paused_time_secs"] = agent_ended_reading.paused - agent_started_reading.paused + # Agent output is no longer being consumed; preserve an audit copy. + shutil.copyfile(active_clock.path, os.path.join(agent_dir, "active-clock.json")) + except OSError as exc: + result["agent_exit"] = -2 + result["error"] = result.get("error") or f"cannot preserve active-clock accounting: {exc}" + result["agent_wall_time_secs"] = result["agent_paused_time_secs"] = None + agent_ended_at = None + finally: + # A retained container keeps its credential mount sources so a + # `docker start` can still authenticate. + if not item.keep_container: + runner.cleanup_credential_tmps() if agent_started_at is None or agent_ended_at is None: return None return agent_ended_at - agent_started_at @@ -3591,6 +3668,7 @@ def _run_grader_container( # Exactly {target}.tla + canonical deps. Replay-required modes pass a # grader-only fresh snapshot, never the one exposed to the agent. benchmark_dir=canonical_dir or os.path.dirname(item.benchmark_path), + pause_accounting=True, ) config.env["GIT_CONFIG_COUNT"] = "1" config.env["GIT_CONFIG_KEY_0"] = "safe.directory" @@ -3622,6 +3700,10 @@ def _run_grader_container( finally: runner.cleanup_credential_tmps() + if isinstance(config.active_clock, ActiveClock): + result["_grader_paused_secs"] = config.active_clock.read().paused + shutil.copyfile(config.active_clock.path, os.path.join(grading_dir, "active-clock.json")) + def _run_grader_local( item: WorkItem, @@ -3748,6 +3830,7 @@ def _grade_submission( attempt["error"] = f"cannot materialize grading inputs: {exc}" finally: elapsed = time.monotonic() - started + elapsed = max(0.0, elapsed - attempt.pop("_grader_paused_secs", 0.0)) if attempt.get("check_verdict") == "ERROR": attempt["grader_error"] = str(attempt.get("error", "")) prior = nonnegative_float(attempt.get("grader_error_time_secs")) or 0.0 diff --git a/src/tlacore/sany/dump.py b/src/tlacore/sany/dump.py index 5b5f9aa9..de1637bf 100644 --- a/src/tlacore/sany/dump.py +++ b/src/tlacore/sany/dump.py @@ -11,16 +11,20 @@ from __future__ import annotations +import contextlib import glob import json import os import re import shutil +import signal import subprocess import tempfile from dataclasses import dataclass from enum import StrEnum +from common.active_clock import ActiveClock, communicate + from ..model import Module _HERE = os.path.dirname(os.path.abspath(__file__)) @@ -98,12 +102,21 @@ def run_raw(tla_path: str, timeout: int = 180) -> SanyRun: command = (_RUN_SH, tla_path) try: - res = subprocess.run( - command, - capture_output=True, - text=True, - timeout=timeout, - ) + clock = ActiveClock.from_environment() + if clock.path is None: + res = subprocess.run(command, capture_output=True, text=True, timeout=timeout) + else: + with subprocess.Popen( + command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=True + ) as proc: + try: + stdout, stderr = communicate(proc, timeout, clock=clock) + except BaseException: + with contextlib.suppress(ProcessLookupError): + os.killpg(proc.pid, signal.SIGKILL) + proc.communicate() + raise + res = subprocess.CompletedProcess(command, proc.returncode, stdout, stderr) except subprocess.TimeoutExpired as exc: stdout = _captured_text(exc.stdout) stderr = _captured_text(exc.stderr) diff --git a/src/tlaps_bench/cli.py b/src/tlaps_bench/cli.py index 16db3239..1044e0d3 100644 --- a/src/tlaps_bench/cli.py +++ b/src/tlaps_bench/cli.py @@ -7,6 +7,8 @@ ("validate", "Batch-validate source proofs with tlapm"), ("generate", "Generate benchmarks (--mode proof-completion|proof-from-scratch; default proof-completion)"), ("score", "Score results (pass rate, per-module breakdown) from results.json"), + ("pause-container", "Pause a managed Docker container without spending its execution budget"), + ("resume-container", "Resume a container paused by pause-container"), ] PROG = "tlaps-bench" @@ -134,6 +136,14 @@ def main(argv: list[str] | None = None) -> int: if sub == "run": return _dispatch(f"{PROG} run", "evaluator.runner", "main", rest) + if sub in {"pause-container", "resume-container"}: + return _dispatch( + f"{PROG} {sub}", + "common.pause_control", + "main", + rest, + entry_kwargs={"pause": sub == "pause-container"}, + ) if sub == "check": return _dispatch( f"{PROG} check", diff --git a/tests/common/test_active_clock.py b/tests/common/test_active_clock.py new file mode 100644 index 00000000..bfe78a63 --- /dev/null +++ b/tests/common/test_active_clock.py @@ -0,0 +1,353 @@ +"""Managed pauses preserve active deadlines without exempting ordinary waits.""" + +import copy +import json +import subprocess +import sys +import threading +import time +from types import SimpleNamespace + +import pytest + +from common import active_clock, container, pause_control +from common.active_clock import CLOCK_ENV, CLOCK_LABEL, CLOCK_MOUNT, ActiveClock, communicate, write_state +from common.container import ContainerConfig, ContainerRunner +from common.verification_budget import CheckSession + + +@pytest.fixture +def controlled_clock(tmp_path, monkeypatch): + now = SimpleNamespace(value=1000.0) + monkeypatch.setattr(active_clock.time, "monotonic", lambda: now.value) + monkeypatch.setattr(active_clock.time, "time", lambda: now.value + 1700000000) + clock = ActiveClock.create(tmp_path / "clock") + monkeypatch.setenv(CLOCK_ENV, str(clock.path)) + return clock, now + + +def add_pause(clock, start, end): + state = clock.read_state() + state["pauses"].append([start, end]) + write_state(clock.path, state) + + +def test_multiple_pauses_cross_deadline_but_do_not_spend_check_budget(controlled_clock, tmp_path): + clock, now = controlled_clock + with CheckSession(tmp_path / "session", {}, 10) as session: + now.value += 2 + add_pause(clock, now.value, now.value + 100) + now.value += 103 + add_pause(clock, now.value, now.value + 200) + now.value += 200 + assert session.remaining() == 5 + metrics = session.metrics() + assert metrics["wall_secs"] == 305 + assert metrics["paused_secs"] == 300 + assert metrics["active_secs"] == 5 + with CheckSession(tmp_path / "session", {}, 10) as resumed: + assert resumed.remaining() == 5 + now.value += 5 + assert resumed.remaining() == 0 + assert resumed.metrics()["active_secs"] == 10 + + +def test_freeze_between_ledger_read_and_timestamp_does_not_charge_pause(controlled_clock, monkeypatch): + clock, now = controlled_clock + read_state = clock.read_state + calls = 0 + + def freeze_after_read(): + nonlocal calls + calls += 1 + state = read_state() + if calls == 1: + updated = copy.deepcopy(state) + updated["pauses"].append([now.value, now.value + 100]) + write_state(clock.path, updated) + now.value += 100 + return state + + monkeypatch.setattr(clock, "read_state", freeze_after_read) + assert clock.now() == 1000 + assert calls == 4 + + +def test_active_idle_counts_and_no_ledger_keeps_legacy_clock(tmp_path, monkeypatch): + monkeypatch.delenv(CLOCK_ENV, raising=False) + with CheckSession(tmp_path / "session", {}, 1) as session: + before = session.remaining() + time.sleep(0.05) + assert session.remaining() <= before - 0.04 + assert session.metrics()["paused_secs"] == 0 + + +def test_hard_restart_keeps_reserved_active_budget_after_a_pause(controlled_clock, tmp_path): + clock, now = controlled_clock + directory = tmp_path / "session" + with CheckSession(directory, {}, 10) as session: + now.value += 2 + add_pause(clock, now.value, now.value + 100) + now.value += 100 + session._save(active=True) + crash_state = copy.deepcopy(session.state) + # Simulate the last durable write surviving SIGKILL instead of graceful exit. + (directory / "state.json").write_text(json.dumps(crash_state)) + now.value += 1000 # No examination runs during restart downtime. + with CheckSession(directory, {}, 10) as resumed: + assert resumed.remaining() == 7.75 + assert resumed.metrics()["paused_secs"] == 100 + assert resumed.metrics()["cpu_complete"] is False + now.value += 1 + with CheckSession(directory, {}, 10) as resumed_again: + assert resumed_again.remaining() == 6.75 + + +def test_legacy_resume_conservatively_charges_all_recorded_wall(controlled_clock, tmp_path): + directory = tmp_path / "session" + directory.mkdir() + (directory / "state.json").write_text( + json.dumps( + { + "identity": {}, + "wall_secs": 7, + "cpu_secs": 0, + "cpu_complete": True, + "active": False, + } + ) + ) + with CheckSession(directory, {}, 10) as resumed: + assert resumed.remaining() == 3 + assert resumed.metrics()["paused_secs"] == 0 + + +def test_subprocess_wait_rechecks_active_budget_after_frozen_wall_poll(controlled_clock): + clock, now = controlled_clock + + class Child: + calls = 0 + + def communicate(self, *, input, timeout): + self.calls += 1 + if self.calls == 1: + add_pause(clock, now.value, now.value + 100) + now.value += 100 + raise subprocess.TimeoutExpired("child", timeout) + now.value += 0.5 + return "done", "" + + child = Child() + assert communicate(child, 1, clock=clock) == ("done", "") + assert child.calls == 2 + + +def test_subprocess_wait_does_expire_on_active_idle(controlled_clock): + clock, now = controlled_clock + + class Child: + def communicate(self, *, input, timeout): + now.value += timeout + raise subprocess.TimeoutExpired("child", timeout, output="partial output", stderr="partial error") + + with pytest.raises(subprocess.TimeoutExpired) as raised: + communicate(Child(), 1, clock=clock) + assert now.value == 1001 + assert raised.value.timeout == 1 + assert raised.value.output == "partial output" + assert raised.value.stderr == "partial error" + + +def test_real_idle_subprocess_timeout_is_still_enforced(tmp_path, monkeypatch): + clock = ActiveClock.create(tmp_path / "clock") + monkeypatch.setenv(CLOCK_ENV, str(clock.path)) + from common.check_proof import run_killgroup + + start = time.monotonic() + with pytest.raises(subprocess.TimeoutExpired): + run_killgroup([sys.executable, "-c", "import time; time.sleep(10)"], 0.15, str(tmp_path)) + assert time.monotonic() - start < 2 + + +def test_managed_sany_timeout_stops_children_holding_output_pipes(tmp_path, monkeypatch): + # Import the module explicitly: tlacore.sany exports a function called dump. + import importlib + + sany_dump = importlib.import_module("tlacore.sany.dump") + clock = ActiveClock.create(tmp_path / "clock") + monkeypatch.setenv(CLOCK_ENV, str(clock.path)) + script = tmp_path / "sany" + script.write_text( + f"#!{sys.executable}\nimport subprocess, sys, time\n" + "subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(3)'])\n" + "time.sleep(3)\n" + ) + script.chmod(0o755) + monkeypatch.setattr(sany_dump, "_RUN_SH", str(script)) + started = time.monotonic() + result = sany_dump.run_raw("unused.tla", timeout=0.15) + assert result.status == sany_dump.SanyStatus.UNAVAILABLE + assert "timed out" in result.detail + assert time.monotonic() - started < 2 + + +@pytest.mark.parametrize( + "mutation", + [ + lambda state: state.update(boot_id="previous-boot"), + lambda state: state.update(pauses=[[2, 1]]), + lambda state: state.update(pauses=[[1, None], [2, 3]]), + lambda state: state.update(pauses=[[float("inf"), None]]), + ], +) +def test_invalid_ledger_is_rejected(controlled_clock, mutation): + clock, _now = controlled_clock + state = clock.read_state() + mutation(state) + clock.path.write_text(json.dumps(state)) + with pytest.raises(OSError): + clock.read() + + +def test_clock_domain_mismatch_fails_closed(controlled_clock, monkeypatch): + clock, now = controlled_clock + monkeypatch.setattr(active_clock.time, "monotonic", lambda: now.value + 100) + with pytest.raises(OSError, match="different monotonic epoch"): + clock.now() + + +@pytest.fixture +def docker_control(controlled_clock, monkeypatch): + clock, now = controlled_clock + info = { + "Id": "container-id", + "Config": {"Labels": {CLOCK_LABEL: str(clock.path.parent)}}, + "State": {"Running": True, "Paused": False}, + "Mounts": [{"Source": str(clock.path.parent), "Destination": CLOCK_MOUNT, "RW": False}], + } + events = [] + + def docker(args, **_kwargs): + action = args[1] + events.append(action) + if action == "pause": + info["State"]["Paused"] = True + elif action == "unpause": + # Critically, the close must already be durable when Docker thaws. + assert clock.read_state()["pauses"][-1][1] is not None + info["State"]["Paused"] = False + return SimpleNamespace(returncode=0) + + monkeypatch.setattr(pause_control, "inspect_container", lambda _name: copy.deepcopy(info)) + monkeypatch.setattr(pause_control.subprocess, "run", docker) + return clock, now, info, events + + +def test_controller_is_idempotent_and_publishes_before_thaw(docker_control): + clock, now, _info, events = docker_control + assert pause_control.control("test", pause=True) == "paused" + assert pause_control.control("test", pause=True) == "already paused" + now.value += 100 + assert clock.now() == 1000 + assert pause_control.control("test", pause=False) == "resumed" + assert pause_control.control("test", pause=False) == "already running" + assert clock.now() == 1000 + now.value += 1 + assert clock.now() == 1001 + assert events == ["pause", "unpause"] + + +def test_host_reader_cannot_observe_unpublished_pause_timestamp(docker_control, monkeypatch): + clock, now, _info, _events = docker_control + read_done = threading.Event() + readings = [] + + def reader(): + readings.append(clock.now()) + read_done.set() + + thread = threading.Thread(target=reader) + original_write = pause_control.write_state + + def slow_publish(path, state): + now.value += 3 # fsync/publication stalls after sampling pause start. + thread.start() + assert not read_done.wait(0.05), "host clock observed a half-published transition" + original_write(path, state) + + monkeypatch.setattr(pause_control, "write_state", slow_publish) + pause_control.control("test", pause=True) + thread.join(timeout=2) + assert read_done.is_set() + assert readings == [1000] + + +def test_failed_freeze_never_grants_unrecorded_pause_time(docker_control, monkeypatch): + clock, now, _info, _events = docker_control + + def failed_freeze(command, **_kwargs): + now.value += 3 + raise subprocess.CalledProcessError(1, command) + + monkeypatch.setattr(pause_control.subprocess, "run", failed_freeze) + with pytest.raises(subprocess.CalledProcessError): + pause_control.control("test", pause=True) + assert clock.read_state()["pauses"] == [] + assert clock.now() == 1003 + + +def test_controller_recovers_pause_crash_conservatively(docker_control): + clock, now, info, events = docker_control + info["State"]["Paused"] = True # Crash after freeze, before durable pause write. + now.value += 100 + assert "remains charged" in pause_control.control("test", pause=True) + assert clock.now() == 1100 + assert events == [] + + +def test_controller_recovers_resume_crash_without_reopening_old_pause(docker_control): + clock, now, info, _events = docker_control + pause_control.control("test", pause=True) + now.value += 100 + state = clock.read_state() + state["pauses"][-1][1] = now.value + write_state(clock.path, state) # Crash after closing pause, before Docker thaw. + now.value += 20 + # Recovery does not subtract this ambiguous gap, which may include runtime. + info["State"]["Paused"] = True + assert "remains charged" in pause_control.control("test", pause=False) + assert info["State"]["Paused"] is False + assert clock.now() == 1020 + + +def test_controller_rejects_out_of_band_thaw(docker_control): + _clock, _now, info, _events = docker_control + pause_control.control("test", pause=True) + info["State"]["Paused"] = False + with pytest.raises(ValueError, match="outside"): + pause_control.control("test", pause=False) + + +def test_controller_rejects_unadapted_backend(docker_control): + clock, _now, _info, _events = docker_control + state = clock.read_state() + state["pause_supported"] = False + write_state(clock.path, state) + with pytest.raises(ValueError, match="absolute deadlines"): + pause_control.control("test", pause=True) + + +def test_container_clock_mount_is_read_only_and_separate_from_results(controlled_clock): + clock, _now = controlled_clock + config = ContainerConfig(workspace="/work", result_dir="/results", active_clock=clock) + args, _cid = ContainerRunner().build_docker_args(config) + assert f"{clock.path.parent}:{CLOCK_MOUNT}:ro" in args + assert f"{CLOCK_LABEL}={clock.path.parent}" in args + assert f"{CLOCK_ENV}={CLOCK_MOUNT}/state.json" in args + + +def test_non_linux_docker_host_keeps_legacy_wallclock(monkeypatch): + monkeypatch.setattr(container.sys, "platform", "darwin") + config = ContainerConfig(pause_accounting=True) + ContainerRunner._prepare_active_clock(config) + assert config.active_clock is None diff --git a/tests/evaluator/test_claude_code_environment.py b/tests/evaluator/test_claude_code_environment.py new file mode 100644 index 00000000..4bc46fb3 --- /dev/null +++ b/tests/evaluator/test_claude_code_environment.py @@ -0,0 +1,75 @@ +"""Claude print-mode background work follows the benchmark's execution limit.""" + +import json +import os +import sys +import time +from types import SimpleNamespace + +import pytest + +from evaluator import runner +from evaluator.backends.agentic import AgenticBackend +from evaluator.backends.claude_code import ClaudeCodeBackend + +BACKGROUND_CEILING = "CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS" + + +def test_execution_environment_preserves_parent_fields_without_mutating_caller(monkeypatch): + monkeypatch.setenv(BACKGROUND_CEILING, "600000") + inherited = {"EXISTING_BACKEND_FIELD": "keep", BACKGROUND_CEILING: "123"} + calls = [] + + def parent_environment(self, result_dir): + calls.append(result_dir) + return inherited + + monkeypatch.setattr(AgenticBackend, "execution_environment", parent_environment) + before = dict(os.environ) + result = ClaudeCodeBackend().execution_environment("/results") + + assert result == {"EXISTING_BACKEND_FIELD": "keep", BACKGROUND_CEILING: "0"} + assert calls == ["/results"] + assert inherited[BACKGROUND_CEILING] == "123" + assert dict(os.environ) == before + + +@pytest.mark.parametrize("hang", [False, True]) +def test_local_cli_child_receives_disabled_ceiling_but_runner_timeout_still_applies(tmp_path, monkeypatch, hang): + monkeypatch.setenv(BACKGROUND_CEILING, "600000") + workspace = tmp_path / "workspace" + workspace.mkdir() + agent_dir = tmp_path / "agent" + agent_dir.mkdir() + stub = tmp_path / "claude_stub.py" + stub.write_text( + "import json, os, time\n" + f"print(json.dumps({{'background_ceiling': os.environ.get({BACKGROUND_CEILING!r})}}), flush=True)\n" + + ("time.sleep(10)\n" if hang else "") + ) + backend = ClaudeCodeBackend() + monkeypatch.setattr(backend, "build_command", lambda *_args: [sys.executable, str(stub)]) + item = SimpleNamespace(timeout=2 if hang else 0, check_timeout=10, tlapm_lib="") + mode = SimpleNamespace(canonical_replay_required=False) + result = {} + output = agent_dir / "output.jsonl" + started = time.monotonic() + runner._run_backend_local( + item, + backend, + mode, + str(workspace), + str(agent_dir), + str(output), + "prompt", + result, + "/bin/true", + ) + assert json.loads(output.read_text())["background_ceiling"] == "0" + assert os.environ[BACKGROUND_CEILING] == "600000" + if hang: + assert result["agent_exit"] == -1 + assert "timeout after 2s" in result["error"] + assert time.monotonic() - started < 5 + else: + assert result["agent_exit"] == 0 diff --git a/tests/evaluator/test_infra_retry.py b/tests/evaluator/test_infra_retry.py index d26ee158..609697ff 100644 --- a/tests/evaluator/test_infra_retry.py +++ b/tests/evaluator/test_infra_retry.py @@ -581,6 +581,38 @@ def test_retry_time_keeps_only_formal_attempt_and_separates_infra_time(tmp_path, assert diagnostic["time_secs"] == 2.0 +def test_retry_pause_metrics_follow_formal_attempt_and_keep_diagnostics(tmp_path, monkeypatch): + backend = _ScriptedBackend() + calls = _install_agent(monkeypatch, backend, [STARTUP, GENUINE_FAIL]) + local_agent = runner._run_backend_local + _install_grader(monkeypatch, verdict="FAIL") + monkeypatch.setattr(runner, "_run_grader_container", runner._run_grader_local) + _no_sleep(monkeypatch) + + def container(item, backend, workspace, agent_dir, agent_jsonl, prompt, result, canonical_dir, **_kwargs): + local_agent(item, backend, item.mode, workspace, agent_dir, agent_jsonl, prompt, result, "", canonical_dir) + active, paused = ((2.0, 100.0), (5.0, 200.0))[calls["n"] - 1] + result.update( + agent_wall_time_secs=active + paused, agent_paused_time_secs=paused, _container_paused_secs=paused + ) + return active + + monkeypatch.setattr(runner, "_run_backend_container", container) + times = iter((10.0, 112.0, 120.0, 325.0, 330.0, 331.0)) + monkeypatch.setattr(runner.time, "monotonic", lambda: next(times)) + item = _work_item(tmp_path, backend, infra_retries=1) + item.use_container = True + result = runner.run_single_benchmark(item) + assert result["agent_time_secs"] == 5 + assert result["agent_wall_time_secs"] == 205 + assert result["agent_paused_time_secs"] == 200 + assert result["logical_timeout_used_secs"] == 5 + diagnostic = json.loads((tmp_path / "out/Foo/Bar/agent/attempts/attempt-0/accounting.json").read_text()) + assert diagnostic["time_secs"] == 2 + assert diagnostic["agent_wall_time_secs"] == 102 + assert diagnostic["agent_paused_time_secs"] == 100 + + def test_container_timeout_before_agent_marker_does_not_publish_setup_time(tmp_path, monkeypatch): backend = _ScriptedBackend() item = _work_item(tmp_path, backend, infra_retries=0) diff --git a/tests/evaluator/test_pause_accounting.py b/tests/evaluator/test_pause_accounting.py new file mode 100644 index 00000000..3a227ef4 --- /dev/null +++ b/tests/evaluator/test_pause_accounting.py @@ -0,0 +1,106 @@ +"""Runner deadline and error-path checks with a controlled host clock.""" + +import tempfile +from types import SimpleNamespace + +import pytest + +from common.active_clock import ActiveClock, ClockReading +from evaluator import runner +from evaluator.backends.claude_code import ClaudeCodeBackend + + +@pytest.fixture +def pipe_output(): + with tempfile.TemporaryFile(mode="w+", encoding="utf-8") as stream: + yield stream + + +@pytest.mark.parametrize("audit_error", [False, True]) +@pytest.mark.parametrize("delayed_initial_read", [False, True]) +def test_host_deadline_excludes_pause_and_cleanup_survives_audit_failure( + tmp_path, monkeypatch, audit_error, pipe_output, delayed_initial_read +): + state = SimpleNamespace(wall=1000.0, paused=0.0, reads=0, cleaned=False) + path = tmp_path / "ledger.json" + path.write_text("{}") + + class Clock(ActiveClock): + initialized = False + + def read(self): + if delayed_initial_read and not self.initialized: + self.initialized = True + state.wall += 100 + state.paused += 100 + return ClockReading(state.wall, state.paused) + + clock = Clock(path) + output = pipe_output + process = SimpleNamespace(stdout=output, stderr=None, returncode=0) + process.poll = lambda: 0 if state.reads >= 2 else None + chunks = [] + + class Container: + def run(self, config, _cmd, stdin_data=None): + assert config.pause_accounting + chunks.extend([(config.agent_start_marker + "\n").encode(), b'{"type":"result"}\n', b""]) + return SimpleNamespace(proc=process, active_clock=clock) + + def kill(self, _run): + pytest.fail("a managed pause must not exhaust the two-second active deadline") + + def cleanup_credential_tmps(self): + state.cleaned = True + + def read(_fd, _size): + value = chunks[state.reads] + state.reads += 1 + if state.reads == 2: + # The agent performed one second of work around a 100s freeze. + state.wall += 101 + state.paused += 100 + return value + + monkeypatch.setattr(runner, "ContainerRunner", Container) + monkeypatch.setattr(runner, "STREAM_AGENT_OUTPUT", False) + monkeypatch.setattr(runner.select, "select", lambda streams, *_args: (streams, [], [])) + monkeypatch.setattr(runner.os, "read", read) + monkeypatch.setattr(runner.time, "time", lambda: 1700000000 + state.wall) + if audit_error: + + def fail_copy(*_args): + raise OSError("test audit write failure") + + monkeypatch.setattr(runner.shutil, "copyfile", fail_copy) + item = SimpleNamespace( + benchmark_path=str(tmp_path / "Task.tla"), + timeout=2, + check_timeout=10, + keep_container=False, + session_dir="", + container_image="unused", + mode=SimpleNamespace(canonical_replay_required=False), + ) + result = {} + duration = runner._run_backend_container( + item, + ClaudeCodeBackend(), + str(tmp_path), + str(tmp_path), + str(tmp_path / "output.jsonl"), + "", + result, + ) + assert state.cleaned + if audit_error: + assert duration is None + assert result["agent_exit"] == -2 + assert result["agent_paused_time_secs"] is None + assert "audit write failure" in result["error"] + else: + assert result["agent_exit"] == 0 + assert duration == 1 + assert result["agent_wall_time_secs"] == 101 + assert result["agent_paused_time_secs"] == 100 + assert result["_container_paused_secs"] == (200 if delayed_initial_read else 100) diff --git a/tests/evaluator/test_proof_module_resume_regressions.py b/tests/evaluator/test_proof_module_resume_regressions.py index 33e6ac5e..114bac77 100644 --- a/tests/evaluator/test_proof_module_resume_regressions.py +++ b/tests/evaluator/test_proof_module_resume_regressions.py @@ -651,6 +651,8 @@ def test_interrupted_first_attempt_resume_keeps_accounting_and_uses_only_remaini "termination_reason": "INFRA_ERROR", "graded_after_interruption": True, "agent_time_secs": 3.0, + "agent_wall_time_secs": 103.0, + "agent_paused_time_secs": 100.0, "grading_time_secs": 1.0, "logical_timeout_used_secs": 4.0, "time_secs": 4.0, @@ -696,6 +698,8 @@ def resumed_segment(execution_item, _prompt, *_args, **_kwargs): "agent_exit": 0, "termination_reason": "OK", "agent_time_secs": 2.0, + "agent_wall_time_secs": 202.0, + "agent_paused_time_secs": 200.0, "grading_time_secs": 0.0, "logical_timeout_used_secs": 2.0, "time_secs": 2.0, @@ -728,6 +732,8 @@ def resumed_segment(execution_item, _prompt, *_args, **_kwargs): assert recovered["check_verdict"] == "PASS" assert recovered["logical_resume_count"] == 1 assert recovered["agent_time_secs"] == 5.0 + assert recovered["agent_wall_time_secs"] == 305.0 + assert recovered["agent_paused_time_secs"] == 300.0 assert recovered["logical_timeout_used_secs"] == 6.0 assert recovered["logical_timeout_remaining_secs"] == 4.0 assert recovered["grading_time_secs"] > 1.0 diff --git a/tests/integration/test_docker_pause_budget.py b/tests/integration/test_docker_pause_budget.py new file mode 100644 index 00000000..da62c4e6 --- /dev/null +++ b/tests/integration/test_docker_pause_budget.py @@ -0,0 +1,215 @@ +"""Opt-in local Docker freezer smoke; uses no model or network requests.""" + +import json +import os +import select +import subprocess +import time +import uuid +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from common.active_clock import CLOCK_ENV, CLOCK_LABEL, CLOCK_MOUNT, ActiveClock +from common.container import ContainerConfig, ContainerRunner +from common.pause_control import control +from evaluator import runner +from evaluator.backends.claude_code import ClaudeCodeBackend + + +@pytest.mark.skipif(os.environ.get("TLAPS_TEST_DOCKER_PAUSE") != "1", reason="opt-in Docker freezer smoke") +def test_real_docker_pause_preserves_running_checker_budget(tmp_path): + clock = ActiveClock.create(tmp_path / "clock") + name = "tlaps-pause-test-" + uuid.uuid4().hex[:10] + repo = Path(__file__).resolve().parents[2] + script = """ +import json, sys +from pathlib import Path +from common.verification_budget import CheckSession +from common.check_proof import run_killgroup +with CheckSession(Path('/tmp/check'), {}, 3) as session: + print('ready', flush=True) + out, err, rc = run_killgroup( + [sys.executable, '-c', + 'import time; start=time.process_time();\\nwhile time.process_time()-start < 1.5: pass'], + session.remaining(), '/tmp') + assert rc == 0, (out, err) + assert session.remaining() > 0 + print(json.dumps(session.metrics()), flush=True) +""" + command = [ + "docker", + "run", + "--rm", + "--name", + name, + "--network", + "none", + "--label", + f"{CLOCK_LABEL}={clock.path.parent}", + "-v", + f"{clock.path.parent}:{CLOCK_MOUNT}:ro", + "-v", + f"{repo / 'src'}:/code:ro", + "-e", + "PYTHONPATH=/code", + "-e", + f"{CLOCK_ENV}={CLOCK_MOUNT}/state.json", + "--entrypoint", + "python3", + os.environ.get("TLAPS_PAUSE_TEST_IMAGE", "tlaps-bench-base:latest"), + "-c", + script, + ] + process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + try: + ready, _, _ = select.select([process.stdout], [], [], 20) + assert ready, "container did not start" + first_line = process.stdout.readline().strip() + assert first_line == "ready", (first_line, process.stderr.read()) + time.sleep(0.1) + assert control(name, pause=True) == "paused" + time.sleep(4) + assert control(name, pause=False) == "resumed" + stdout, stderr = process.communicate(timeout=10) + assert process.returncode == 0, stderr + metrics = json.loads(stdout.strip()) + assert metrics["wall_secs"] > 4 + assert metrics["paused_secs"] >= 4 + assert 1 < metrics["active_secs"] < 3 + assert metrics["wall_secs"] == pytest.approx(metrics["active_secs"] + metrics["paused_secs"]) + finally: + subprocess.run(["docker", "rm", "-f", name], capture_output=True, timeout=10) + if process.poll() is None: + process.kill() + process.communicate(timeout=10) + + +@pytest.mark.skipif(os.environ.get("TLAPS_TEST_DOCKER_PAUSE") != "1", reason="opt-in Docker freezer smoke") +def test_host_container_wait_and_checker_share_the_active_clock(tmp_path): + name = "tlaps-pause-host-test-" + uuid.uuid4().hex[:10] + repo = Path(__file__).resolve().parents[2] + clock = ActiveClock.create(tmp_path / "clock") + workspace = tmp_path / "workspace" + workspace.mkdir() + ready = workspace / "ready" + script = """ +import json, sys +from pathlib import Path +from common.verification_budget import CheckSession +from common.check_proof import run_killgroup +with CheckSession(Path('/tmp/check'), {}, 3) as session: + Path('/workspace/ready').touch() + out, err, rc = run_killgroup( + [sys.executable, '-c', + 'import time; start=time.process_time();\\nwhile time.process_time()-start < 1.5: pass'], + session.remaining(), '/tmp') + assert rc == 0, (out, err) + print(json.dumps(session.metrics()), flush=True) +""" + + def quota_guard(): + deadline = time.monotonic() + 15 + while not ready.exists(): + if time.monotonic() > deadline: + raise RuntimeError("test child did not start") + time.sleep(0.02) + control(name, pause=True) + time.sleep(4) + control(name, pause=False) + + config = ContainerConfig( + image=os.environ.get("TLAPS_PAUSE_TEST_IMAGE", "tlaps-bench-base:latest"), + workspace=str(workspace), + read_only_files=[(str(repo / "src"), "/code")], + env={"PYTHONPATH": "/code"}, + keep_container=True, + container_name=name, + pause_accounting=True, + active_clock=clock, + ) + try: + with ThreadPoolExecutor(max_workers=1) as executor: + guard = executor.submit(quota_guard) + code, stdout, stderr = ContainerRunner().run_with_output( + config, + ["python3", "-c", script], + timeout=3, + ) + guard.result(timeout=10) + assert code == 0, stderr + metrics = json.loads(stdout.strip()) + assert metrics["wall_secs"] > 4 + assert metrics["paused_secs"] >= 4 + assert 1 < metrics["active_secs"] < 3 + finally: + subprocess.run(["docker", "rm", "-f", name], capture_output=True, timeout=10) + + +@pytest.mark.skipif(os.environ.get("TLAPS_TEST_DOCKER_PAUSE") != "1", reason="opt-in Docker environment smoke") +@pytest.mark.parametrize("hang", [False, True]) +def test_claude_container_environment_keeps_background_work_under_runner_deadline(tmp_path, monkeypatch, hang): + variable = "CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS" + monkeypatch.setenv(variable, "600000") + backend = ClaudeCodeBackend() + script = ( + "import json, os, time; " + f"print(json.dumps({{'background_ceiling': os.environ.get({variable!r})}}), flush=True); " + + ("time.sleep(10)" if hang else "") + ) + # Exercise the real runner -> Docker -> child environment path without + # installing a CLI, mounting credentials, or making model/network calls. + monkeypatch.setattr(backend, "build_command", lambda *_args: ["python3", "-c", script]) + monkeypatch.setattr(backend, "install_script", None) + monkeypatch.setattr(backend, "firewall_hosts", lambda: []) + monkeypatch.setattr(backend, "get_credential_mounts", lambda: []) + monkeypatch.setattr(runner, "forward_env", lambda *_args, **_kwargs: {variable: "600000"}) + monkeypatch.setattr(runner, "STREAM_AGENT_OUTPUT", False) + workspace, agent_dir = tmp_path / "workspace", tmp_path / "agent" + workspace.mkdir() + agent_dir.mkdir() + item = SimpleNamespace( + benchmark_path=str(tmp_path / "Task.tla"), + timeout=0.3 if hang else 0, + check_timeout=10, + keep_container=False, + session_dir="", + container_image=os.environ.get("TLAPS_PAUSE_TEST_IMAGE", "tlaps-bench-base:latest"), + mode=SimpleNamespace(canonical_replay_required=False), + ) + handles = [] + launch = ContainerRunner.run + + def capture(self, *args, **kwargs): + handle = launch(self, *args, **kwargs) + handles.append(handle) + return handle + + monkeypatch.setattr(ContainerRunner, "run", capture) + output = agent_dir / "output.jsonl" + result = {} + started = time.monotonic() + try: + runner._run_backend_container( + item, + backend, + str(workspace), + str(agent_dir), + str(output), + "prompt", + result, + ) + assert json.loads(output.read_text())["background_ceiling"] == "0" + assert os.environ[variable] == "600000" + if hang: + assert result["agent_exit"] == -1 + assert "timeout after 0.3s" in result["error"] + assert time.monotonic() - started < 5 + else: + assert result["agent_exit"] == 0 + finally: + for handle in handles: + if handle.container_id: + subprocess.run(["docker", "rm", "-f", handle.container_id], capture_output=True, timeout=10)