From 53e609acb76fee090c73c9b96e30ffa94c7901d7 Mon Sep 17 00:00:00 2001 From: Mengye Ren Date: Thu, 6 Aug 2026 09:48:25 -0400 Subject: [PATCH 1/4] Phase 4: compute seam, run-state machine, fail-safe tick, chain shim Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 6 + docs/design/scaling.md | 4 +- docs/roadmap.md | 24 +-- scripts/tick_chain.sbatch | 70 +++++++++ src/autoresearch/compute.py | 193 ++++++++++++++++++++++++ src/autoresearch/runstate.py | 180 +++++++++++++++++++++++ src/autoresearch/tick.py | 275 +++++++++++++++++++++++++++++++++++ tests/test_compute.py | 137 +++++++++++++++++ tests/test_runstate.py | 106 ++++++++++++++ tests/test_tick.py | 222 ++++++++++++++++++++++++++++ 10 files changed, 1207 insertions(+), 10 deletions(-) create mode 100755 scripts/tick_chain.sbatch create mode 100644 src/autoresearch/compute.py create mode 100644 src/autoresearch/runstate.py create mode 100644 src/autoresearch/tick.py create mode 100644 tests/test_compute.py create mode 100644 tests/test_runstate.py create mode 100644 tests/test_tick.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 8a1ab7dc..d89545d5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,12 @@ Versions follow [SemVer](https://semver.org). ### Added +- Phase 4 loop plumbing: `compute` (Slurm submit/status/cancel behind an + injectable runner; afterany wake jobs; query-failure ≠ job-gone), + `runstate` (atomic run records, six endings, expiring wake leases with + handoff), `tick` (pause sentinel, heartbeat, the five-layer fail-safe + sweep), and the self-resubmitting chain script. + - `autoresearch.brief` (typed, bounded, replayable session briefs — the context-engineering artifact) and `autoresearch.harness` (backend seam; Claude Code adapter with scrubbed session env, timeouts, and key-redacted diff --git a/docs/design/scaling.md b/docs/design/scaling.md index eb638deb..3e9db16e 100644 --- a/docs/design/scaling.md +++ b/docs/design/scaling.md @@ -173,7 +173,9 @@ concurrent seats). immediately (promotes to the requested lane). 2. **Consolidation**: accumulation-triggered on K benchmark-consequential merges. Maintenance merges are valuable but don't count toward K — see - Part 2b. + Part 2b. **Round 2 (2026-08-06): K defaults to 5–10 and the + consequential threshold to a 10% relative metric delta (ε) — both + contract-configurable, never hard-coded.** 3. **Credentials**: API keys for now (seats cost more up front); start a seat when multi-agent testing is actually ready. The `setup-token` spike waits until then. diff --git a/docs/roadmap.md b/docs/roadmap.md index e6c2a004..7b011429 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -61,15 +61,21 @@ standing instruction they supersede — a resumed agent honors stale constraints ## Phase 4 — Torch tick + compute -- [ ] Wake delivery per the architecture's fail-safe layers: afterany - dependency job (mechanism verified live 2026-08-06, incl. on failed - experiments) + run lease + tick sweep + deadline floor + `stuck` state - -- [ ] The chain: two queued successors, `--dependency=singleton`, absolute - `--begin` cadence grid, sbatch retry/backoff -- [ ] Deploy shim: submit-successors → pull main → `uv sync --locked` → exec; - pull uses the bot PAT (bot has a Read collaborator role here; add this - repo to the token's selection) +- [x] Wake delivery per the architecture's fail-safe layers: afterany + dependency job (`compute.submit_after`; mechanism verified live + 2026-08-06, incl. on failed experiments) + expiring run lease with + handoff + tick sweep + deadline floor (pending-cancel / gone-on-good- + query / defer-on-query-failure) + `stuck` state. Session dispatch + behind the `WakeDispatcher` seam (real dispatcher lands with phase 5) + +- [x] The chain: `scripts/tick_chain.sbatch` — successor top-up to depth 2, + `--dependency=singleton`, absolute `--begin` cadence grid, sbatch + retry/backoff; successors submitted FIRST so nothing below can break + the chain +- [x] Deploy shim (same script): pull main with the bot PAT → `uv sync + --locked` → exec tick; every step best-effort (bad merges crash ticks, + never the chain). Bot PAT on Torch + repo in the token's selection + still owed by Mengye before live deploy - [ ] Pause sentinel + lease (compare-and-swap on the state branch); heartbeat at tick start; stale-lease reaping - [ ] `compute`: sbatch submit / squeue poll, jobs tagged + `--nice`, GPU-hour caps diff --git a/scripts/tick_chain.sbatch b/scripts/tick_chain.sbatch new file mode 100755 index 00000000..0ac12c49 --- /dev/null +++ b/scripts/tick_chain.sbatch @@ -0,0 +1,70 @@ +#!/bin/bash +# The self-resubmitting tick chain (docs/design/architecture.md, Scheduling). +# +# Order matters and is the fail-safety: +# 1. top up successors FIRST — a crash anywhere below never breaks the chain +# 2. pull main + sync (deploy-at-tick-start; failures logged, tick continues +# on the previous code — a bad merge crashes ticks, never the chain) +# 3. exec the tick +# +# Deploy/config knobs come from the environment (set once via +# `sbatch --export=...` when starting the chain; successors inherit): +# AUTORESEARCH_HOME checkout to run from (required) +# AUTORESEARCH_ROOT state root on the shared FS (required) +# AUTORESEARCH_ACCOUNT / AUTORESEARCH_PARTITION slurm placement (required) +# AUTORESEARCH_CADENCE_MIN tick cadence, default 30 +# AUTORESEARCH_PAT_FILE bot PAT for the deploy pull (optional; no file +# means run whatever code is already deployed) +# +#SBATCH --job-name=autoresearch-tick +#SBATCH --time=15 +#SBATCH --mem=4G +#SBATCH --cpus-per-task=2 +#SBATCH --output=/dev/null + +set -u +CADENCE_MIN="${AUTORESEARCH_CADENCE_MIN:-30}" +JOB_NAME="autoresearch-tick" +LOG_DIR="$AUTORESEARCH_ROOT/logs" +mkdir -p "$LOG_DIR" +exec >>"$LOG_DIR/tick-$(date +%Y%m%d).log" 2>&1 +echo "=== tick $(date -Is) on $(hostname -s) job=${SLURM_JOB_ID:-none}" + +# --- 1. keep two successors queued (singleton serializes same-name jobs) --- +pending=$(squeue -u "$USER" --name="$JOB_NAME" -h -t PENDING 2>/dev/null | wc -l) +need=$((2 - pending)) +epoch_now=$(date +%s) +cadence_s=$((CADENCE_MIN * 60)) +next_slot=$(( (epoch_now / cadence_s + 1) * cadence_s )) +for i in $(seq 1 "$need"); do + begin_epoch=$((next_slot + (i - 1) * cadence_s)) + begin=$(date -d "@$begin_epoch" +%Y-%m-%dT%H:%M:%S 2>/dev/null \ + || date -r "$begin_epoch" +%Y-%m-%dT%H:%M:%S) + for attempt in 1 2 3; do + if sbatch --dependency=singleton --begin="$begin" \ + --account="$AUTORESEARCH_ACCOUNT" --partition="$AUTORESEARCH_PARTITION" \ + "$AUTORESEARCH_HOME/scripts/tick_chain.sbatch"; then + break + fi + echo "sbatch retry $attempt failed; backing off" + sleep $((attempt * 20)) + done +done + +# --- 2. deploy: pull main with the bot PAT, sync deps (best-effort) --- +if [ -n "${AUTORESEARCH_PAT_FILE:-}" ] && [ -r "$AUTORESEARCH_PAT_FILE" ]; then + PAT=$(cat "$AUTORESEARCH_PAT_FILE") + if git -C "$AUTORESEARCH_HOME" fetch --quiet \ + "https://x-access-token:${PAT}@github.com/agentic-learning-ai-lab/autoresearch.git" main; then + git -C "$AUTORESEARCH_HOME" reset --hard --quiet FETCH_HEAD || echo "deploy: reset failed" + else + echo "deploy: fetch failed; running previous code" + fi + unset PAT +fi +(cd "$AUTORESEARCH_HOME" && uv sync --locked --quiet) || echo "deploy: uv sync failed" + +# --- 3. the tick itself (scrubbed of the PAT path; sessions are scrubbed +# further down in the harness) --- +cd "$AUTORESEARCH_HOME" +exec env -u AUTORESEARCH_PAT_FILE uv run python -m autoresearch.tick --root "$AUTORESEARCH_ROOT" diff --git a/src/autoresearch/compute.py b/src/autoresearch/compute.py new file mode 100644 index 00000000..b901b761 --- /dev/null +++ b/src/autoresearch/compute.py @@ -0,0 +1,193 @@ +"""Slurm behind one small interface: submit, status, cancel. + +Everything the loop knows about the cluster goes through here, so a CI +runner or cloud backend is a new implementation of the same three verbs. +Commands run through an injectable runner (tests use a fake; nothing here +requires a cluster). + +The status query preserves a distinction the fail-safe design depends on +(docs/design/architecture.md, "Wake delivery and fail-safety"): a FAILED +query ("Slurm unknown") is not the same as a successful query that finds +nothing ("job gone") — misreading an outage as a vanished job would +terminate healthy runs. +""" + +from __future__ import annotations + +import logging +import shlex +import subprocess +from collections.abc import Callable, Sequence +from dataclasses import dataclass, field + +log = logging.getLogger(__name__) + +# Terminal Slurm states (prefix-matched: sacct reports e.g. "CANCELLED by 123"). +TERMINAL_STATES = ( + "COMPLETED", + "FAILED", + "CANCELLED", + "TIMEOUT", + "OUT_OF_MEMORY", + "NODE_FAIL", + "PREEMPTED", + "BOOT_FAIL", + "DEADLINE", +) +# A successful query that returns no record: the job left Slurm's memory. +GONE = "GONE" + + +class SlurmError(RuntimeError): + """A Slurm command failed (submit/cancel), with its stderr.""" + + +class SlurmQueryError(RuntimeError): + """A status query failed — the answer is UNKNOWN, not 'job gone'. + + Callers must treat this as "defer and retry", never as a terminal state. + """ + + +@dataclass(frozen=True) +class CommandResult: + returncode: int + stdout: str + stderr: str + + +Runner = Callable[[Sequence[str], int], CommandResult] + + +def _subprocess_runner(argv: Sequence[str], timeout_s: int) -> CommandResult: + completed = subprocess.run(list(argv), capture_output=True, text=True, timeout=timeout_s) + return CommandResult(completed.returncode, completed.stdout, completed.stderr) + + +@dataclass(frozen=True) +class JobSpec: + """One sbatch submission. `command` is run via --wrap; a script path can + be passed as `script` instead (mutually exclusive).""" + + job_name: str + account: str + partition: str + time_minutes: int + command: str = "" + script: str = "" + script_args: tuple[str, ...] = () + cpus: int = 1 + mem: str = "2G" + gpus: int = 0 + qos: str = "" + output: str = "/dev/null" + # Slurm scheduling controls + dependency: str = "" # e.g. "afterany:12345" or "singleton" + begin: str = "" # e.g. "now+30" or an absolute "YYYY-MM-DDTHH:MM:SS" + extra: tuple[str, ...] = () + + def to_argv(self) -> list[str]: + if bool(self.command) == bool(self.script): + raise ValueError("exactly one of command/script must be set") + argv = [ + "sbatch", + "--parsable", + f"--job-name={self.job_name}", + f"--account={self.account}", + f"--partition={self.partition}", + f"--time={self.time_minutes}", + f"--cpus-per-task={self.cpus}", + f"--mem={self.mem}", + f"--output={self.output}", + ] + if self.gpus: + argv.append(f"--gpus={self.gpus}") + if self.qos: + argv.append(f"--qos={self.qos}") + if self.dependency: + argv.append(f"--dependency={self.dependency}") + if self.begin: + argv.append(f"--begin={self.begin}") + argv.extend(self.extra) + if self.command: + argv.append(f"--wrap={self.command}") + else: + argv.append(self.script) + argv.extend(self.script_args) + return argv + + +@dataclass +class SlurmCompute: + """The three verbs, plus afterany for wake jobs.""" + + runner: Runner = field(default=_subprocess_runner) + command_timeout_s: int = 60 + + def submit(self, spec: JobSpec) -> str: + """Submit; returns the job id. Raises SlurmError on failure.""" + result = self.runner(spec.to_argv(), self.command_timeout_s) + if result.returncode != 0: + raise SlurmError(f"sbatch failed ({result.returncode}): {result.stderr.strip()}") + job_id = result.stdout.strip().split(";")[0] + if not job_id.isdigit(): + raise SlurmError(f"sbatch returned no job id: {result.stdout.strip()!r}") + log.info("submitted %s as job %s", spec.job_name, job_id) + return job_id + + def submit_after(self, spec: JobSpec, after_job_id: str) -> str: + """Submit `spec` to run when `after_job_id` terminates — however it + terminates (afterany: the wake-on-failure semantics the fail-safe + design requires; verified live on Torch 2026-08-06).""" + if not after_job_id.isdigit(): + raise ValueError(f"not a job id: {after_job_id!r}") + dependent = JobSpec( + **{ + **{f: getattr(spec, f) for f in spec.__dataclass_fields__}, + "dependency": f"afterany:{after_job_id}", + } + ) + return self.submit(dependent) + + def status(self, job_id: str) -> str: + """The job's Slurm state, or GONE when a *successful* query finds no + record. Raises SlurmQueryError when the query itself fails.""" + if not job_id.isdigit(): + raise ValueError(f"not a job id: {job_id!r}") + try: + result = self.runner( + ["sacct", "-j", job_id, "--parsable2", "--noheader", "-X", "-o", "State"], + self.command_timeout_s, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise SlurmQueryError(f"sacct did not run: {exc}") from exc + if result.returncode != 0: + raise SlurmQueryError(f"sacct failed ({result.returncode}): {result.stderr.strip()}") + state = result.stdout.strip().splitlines()[0].strip() if result.stdout.strip() else "" + return state if state else GONE + + def cancel(self, job_id: str) -> None: + """Cancel; idempotent (cancelling a finished job is not an error).""" + if not job_id.isdigit(): + raise ValueError(f"not a job id: {job_id!r}") + result = self.runner(["scancel", job_id], self.command_timeout_s) + if result.returncode != 0: + log.warning("scancel %s: %s", job_id, result.stderr.strip()) + + +def is_terminal(state: str) -> bool: + """Whether a state string from `status` means the job is over. + + GONE is deliberately NOT terminal here: it means "no record", and the + deadline-floor logic decides what that implies — not this predicate. + """ + return any(state.startswith(t) for t in TERMINAL_STATES) + + +def is_pending(state: str) -> bool: + return state.startswith("PENDING") + + +def quote_command(parts: Sequence[str]) -> str: + """Shell-quote a command for JobSpec.command (--wrap takes a string).""" + return " ".join(shlex.quote(p) for p in parts) diff --git a/src/autoresearch/runstate.py b/src/autoresearch/runstate.py new file mode 100644 index 00000000..789ff40a --- /dev/null +++ b/src/autoresearch/runstate.py @@ -0,0 +1,180 @@ +"""Run state on the shared filesystem: the agent's durable half. + +A run is one hypothesis (docs/design/architecture.md, "The life of a run"). +Its record is a single JSON file written by atomic rename; the sweep reasons +only from these files plus Slurm — never from process memory — so a crash +anywhere leaves a file that says what happens next. + +Leases serialize wake delivery: whoever wants to wake a run acquires the +lease first (atomic O_EXCL create). Leases expire — a holder that died keeps +the lease only until the sweep notices (holder job dead, or age past TTL) — +so a wake killed mid-session delays the retry by one grace window; it cannot +strand the run. +""" + +from __future__ import annotations + +import json +import logging +import os +from dataclasses import asdict, dataclass, replace +from pathlib import Path + +log = logging.getLogger(__name__) + +# Live states. +IMPLEMENTING = "implementing" # a session is (or will be) working +WAITING = "waiting" # experiment submitted; hibernating until results +IN_REVIEW = "in-review" # PR open; wakes on qualifying comments +CONCLUDING = "concluding" # results in hand; final session(s) +ENDED = "ended" + +STATES = (IMPLEMENTING, WAITING, IN_REVIEW, CONCLUDING, ENDED) + +# The six endings ("The life of a run" — every one produces a report). +MERGED = "merged" +REJECTED = "rejected" +NEGATIVE_RESULT = "negative-result" +BUDGET_EXHAUSTED = "budget-exhausted" +ABORTED = "aborted" +STUCK = "stuck" + +ENDINGS = (MERGED, REJECTED, NEGATIVE_RESULT, BUDGET_EXHAUSTED, ABORTED, STUCK) + +RECORD_NAME = "state.json" +LEASE_NAME = "lease.json" + +MAX_WAKE_ATTEMPTS = 3 + + +@dataclass(frozen=True) +class RunRecord: + """Everything the sweep needs to act on a run, and nothing more.""" + + run_id: str + target: str # owner/repo + task_title: str + state: str + agent_id: str = "agent-01" + experiment_job_id: str = "" + wake_job_id: str = "" # the afterany dependency job, when one exists + resume_session_id: str = "" # harness session to resume on wake + wake_attempts: int = 0 + deadline: float = 0.0 # unix; submit+walltime+slack, re-based on start + ending: str = "" # one of ENDINGS once state == ENDED + ending_note: str = "" + created: float = 0.0 + updated: float = 0.0 + + def ended(self) -> bool: + return self.state == ENDED + + +@dataclass(frozen=True) +class Lease: + holder: str # e.g. "wake-job:12345" or "tick:12345" + holder_job_id: str # Slurm job id of the holder, "" if none + acquired: float # unix timestamp + + +def run_dir(root: Path, run_id: str) -> Path: + return root / "runs" / run_id + + +def save_record(root: Path, record: RunRecord, now: float) -> None: + """Atomic write: a crash mid-save leaves the previous record intact.""" + if record.state not in STATES: + raise ValueError(f"unknown state {record.state!r}") + if record.state == ENDED and record.ending not in ENDINGS: + raise ValueError(f"ended run needs a valid ending, got {record.ending!r}") + directory = run_dir(root, record.run_id) + directory.mkdir(parents=True, exist_ok=True) + stamped = replace(record, updated=now, created=record.created or now) + tmp = directory / f".{RECORD_NAME}.tmp" + tmp.write_text(json.dumps(asdict(stamped), indent=2, sort_keys=True)) + os.replace(tmp, directory / RECORD_NAME) + + +def load_record(root: Path, run_id: str) -> RunRecord: + raw = json.loads((run_dir(root, run_id) / RECORD_NAME).read_text()) + return RunRecord(**raw) + + +def list_runs(root: Path) -> list[RunRecord]: + """Every readable run record; unreadable ones are logged, not fatal — + one corrupt file must not stop the sweep.""" + records = [] + runs_root = root / "runs" + if not runs_root.is_dir(): + return [] + for directory in sorted(runs_root.iterdir()): + try: + records.append(load_record(root, directory.name)) + except (OSError, ValueError, TypeError, KeyError) as exc: + log.warning("unreadable run record %s: %s", directory, exc) + return records + + +# --- leases --- + + +def acquire_lease(root: Path, run_id: str, holder: str, holder_job_id: str, now: float) -> bool: + """Take the run's wake lease. True if acquired; False if held. + + O_EXCL makes acquisition atomic on the shared filesystem: exactly one + contender wins, the rest see False and no-op (double delivery is + harmless by design). + """ + directory = run_dir(root, run_id) + directory.mkdir(parents=True, exist_ok=True) + payload = json.dumps(asdict(Lease(holder, holder_job_id, now))) + try: + fd = os.open(directory / LEASE_NAME, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + except FileExistsError: + return False + with os.fdopen(fd, "w") as handle: + handle.write(payload) + return True + + +def read_lease(root: Path, run_id: str) -> Lease | None: + try: + raw = json.loads((run_dir(root, run_id) / LEASE_NAME).read_text()) + return Lease(**raw) + except FileNotFoundError: + return None + except (OSError, ValueError, TypeError, KeyError): + # An unreadable lease is treated as held-but-unknown; the TTL path + # in `lease_is_stale` cannot run without a timestamp, so the sweep + # falls back to reaping it after the grace window via mtime. + return None + + +def update_lease_holder( + root: Path, run_id: str, holder: str, holder_job_id: str, now: float +) -> None: + """Hand a HELD lease to a new holder (e.g. tick → the wake job it just + submitted). Atomic replace; only valid while the caller holds the lease.""" + directory = run_dir(root, run_id) + tmp = directory / f".{LEASE_NAME}.tmp" + tmp.write_text(json.dumps(asdict(Lease(holder, holder_job_id, now)))) + os.replace(tmp, directory / LEASE_NAME) + + +def release_lease(root: Path, run_id: str) -> None: + try: + (run_dir(root, run_id) / LEASE_NAME).unlink() + except FileNotFoundError: + pass + + +def lease_is_stale(lease: Lease, now: float, ttl_s: float, holder_alive: bool | None) -> bool: + """A lease is stale when its holder is known-dead, or too old. + + `holder_alive` is None when Slurm could not answer (query failure) — in + that case only the TTL can prove staleness, never the holder check: + an outage must not look like a dead holder. + """ + if holder_alive is False: + return True + return (now - lease.acquired) > ttl_s diff --git a/src/autoresearch/tick.py b/src/autoresearch/tick.py new file mode 100644 index 00000000..971be696 --- /dev/null +++ b/src/autoresearch/tick.py @@ -0,0 +1,275 @@ +"""One tick of the loop: sentinel, heartbeat, and the fail-safe sweep. + +The tick is stateless and bounded — everything durable lives in run-state +files (`runstate`) and Slurm. It implements the backup layers of the wake +design (docs/design/architecture.md, "Wake delivery and fail-safety"); the +primary layer (the afterany dependency job) is submitted by whoever launches +an experiment and needs no help from here. + +Wake *delivery* is behind a seam (`WakeDispatcher`) so this module stays +testable and the actual session dispatch (harness + brief) can evolve +independently. +""" + +from __future__ import annotations + +import json +import logging +import os +import socket +from dataclasses import dataclass, field, replace +from pathlib import Path +from typing import Protocol + +from autoresearch.compute import GONE, SlurmCompute, SlurmQueryError, is_pending, is_terminal +from autoresearch.runstate import ( + ENDED, + MAX_WAKE_ATTEMPTS, + STUCK, + WAITING, + RunRecord, + acquire_lease, + lease_is_stale, + list_runs, + read_lease, + release_lease, + save_record, + update_lease_holder, +) + +log = logging.getLogger(__name__) + +PAUSE_SENTINEL = "PAUSE" +HEARTBEAT_NAME = "heartbeat.json" + +# Grace between "experiment terminal" and the sweep stepping in: the afterany +# job gets this long to deliver before the backup assumes it lost. +DEFAULT_GRACE_S = 15 * 60 +# A held lease is stale after the session timeout plus slack. +DEFAULT_LEASE_TTL_S = 3600 + 15 * 60 + + +class WakeDispatcher(Protocol): + """Delivers one wake, called with the lease already held. + + Returns "" when delivery completed synchronously (the caller releases the + lease), or the Slurm job id of an asynchronous wake job that now owns the + lease (released by that job on completion; reaped by TTL if it dies).""" + + def dispatch(self, record: RunRecord, reason: str) -> str: ... + + +@dataclass +class RecordingDispatcher: + """Test/dry-run dispatcher: records what would have been woken.""" + + dispatched: list[tuple[str, str]] = field(default_factory=list) + holder_job_id: str = "" # set to simulate async dispatch + + def dispatch(self, record: RunRecord, reason: str) -> str: + self.dispatched.append((record.run_id, reason)) + return self.holder_job_id + + +@dataclass(frozen=True) +class TickReport: + paused: bool = False + swept: int = 0 + woken: tuple[tuple[str, str], ...] = () # (run_id, reason) + deferred: tuple[str, ...] = () # runs skipped on "Slurm unknown" + reaped_leases: tuple[str, ...] = () + stuck: tuple[str, ...] = () + + +def write_heartbeat(root: Path, now: float) -> None: + payload = json.dumps({"ts": now, "host": socket.gethostname(), "pid": os.getpid()}) + tmp = root / f".{HEARTBEAT_NAME}.tmp" + tmp.write_text(payload) + os.replace(tmp, root / HEARTBEAT_NAME) + + +def _holder_alive(compute: SlurmCompute, lease_job_id: str) -> bool | None: + """True/False when Slurm answered; None when it could not (an outage + must not look like a dead holder).""" + if not lease_job_id: + return None + try: + state = compute.status(lease_job_id) + except SlurmQueryError: + return None + return not (is_terminal(state) or state == GONE) + + +def _wake( + root: Path, + record: RunRecord, + reason: str, + dispatcher: WakeDispatcher, + now: float, + holder: str, +) -> bool: + """Lease-guarded wake. True when this tick delivered (or handed off) it. + + The attempt counter is bumped BEFORE dispatch, so a dispatcher that dies + mid-delivery still counts toward the stuck threshold. + """ + if not acquire_lease(root, record.run_id, holder, holder_job_id="", now=now): + return False + bumped = replace(record, wake_attempts=record.wake_attempts + 1) + save_record(root, bumped, now) + try: + holder_job = dispatcher.dispatch(bumped, reason) + except Exception as exc: + log.warning("wake dispatch failed for %s: %s: %s", record.run_id, type(exc).__name__, exc) + release_lease(root, record.run_id) + return False + if holder_job: + # An async wake job now owns the lease; it releases on completion, + # and the TTL/holder-dead check reaps it if it dies. + update_lease_holder(root, record.run_id, f"wake-job:{holder_job}", holder_job, now) + else: + release_lease(root, record.run_id) + return True + + +def sweep( + root: Path, + compute: SlurmCompute, + dispatcher: WakeDispatcher, + now: float, + grace_s: float = DEFAULT_GRACE_S, + lease_ttl_s: float = DEFAULT_LEASE_TTL_S, +) -> TickReport: + """The backup wake layers, applied to every waiting run.""" + woken: list[tuple[str, str]] = [] + deferred: list[str] = [] + reaped: list[str] = [] + stuck: list[str] = [] + holder = f"tick:{socket.gethostname()}:{os.getpid()}" + records = [r for r in list_runs(root) if r.state == WAITING] + + for record in records: + # Layer 5 first: too many failed attempts is a terminal, reported state. + if record.wake_attempts >= MAX_WAKE_ATTEMPTS: + ended = replace( + record, + state=ENDED, + ending=STUCK, + ending_note=f"{record.wake_attempts} wake attempts failed", + ) + save_record(root, ended, now) + stuck.append(record.run_id) + continue + + # Stale-lease reaping (layer 2's expiry): a dead wake must not hold on. + lease = read_lease(root, record.run_id) + if lease is not None: + alive = _holder_alive(compute, lease.holder_job_id) + if lease_is_stale(lease, now, lease_ttl_s, alive): + release_lease(root, record.run_id) + reaped.append(record.run_id) + lease = None + else: + continue # a live wake is in flight; nothing for the sweep + + if not record.experiment_job_id: + continue # not yet submitted; not the sweep's business + + try: + state = compute.status(record.experiment_job_id) + except SlurmQueryError: + # Layer 4's rule: query failure is "Slurm unknown", never "gone". + deferred.append(record.run_id) + continue + + past_deadline = record.deadline and now > record.deadline + + if is_terminal(state): + # Layer 3: terminal + no live lease + grace expired → backup wake. + # (updated < now - grace approximates "terminal for a while": the + # record was last touched when the experiment was submitted.) + if now - record.updated >= grace_s: + if _wake(root, record, f"experiment {state}", dispatcher, now, holder): + woken.append((record.run_id, state)) + elif state == GONE: + if past_deadline: + # Successful query, no record, deadline passed: vanished. + if _wake(root, record, "experiment vanished from Slurm", dispatcher, now, holder): + woken.append((record.run_id, "vanished")) + # else: sacct lag right after submission is normal; wait. + elif is_pending(state) and past_deadline: + # Unschedulable in practice: cancel, then wake with that fact. + compute.cancel(record.experiment_job_id) + if _wake( + root, + record, + "experiment unschedulable (pending past deadline)", + dispatcher, + now, + holder, + ): + woken.append((record.run_id, "unschedulable")) + # RUNNING (or recently pending): nothing to do; the afterany job has it. + + return TickReport( + swept=len(records), + woken=tuple(woken), + deferred=tuple(deferred), + reaped_leases=tuple(reaped), + stuck=tuple(stuck), + ) + + +def tick( + root: Path, + compute: SlurmCompute, + dispatcher: WakeDispatcher, + now: float, +) -> TickReport: + """One full tick. Pause sentinel wins over everything: a paused loop + heartbeats (so the watchdog stays quiet) but touches nothing.""" + write_heartbeat(root, now) + if (root / PAUSE_SENTINEL).exists(): + log.info("pause sentinel present; tick is a no-op") + return TickReport(paused=True) + return sweep(root, compute, dispatcher, now) + + +@dataclass +class LoggingDispatcher: + """Placeholder production dispatcher until session dispatch lands + (phase 5): logs what would be woken so the loop's plumbing can run live + without side effects.""" + + def dispatch(self, record: RunRecord, reason: str) -> str: + log.info("WOULD WAKE %s (%s) — session dispatch lands in phase 5", record.run_id, reason) + return "" + + +def main() -> int: + import argparse + import time + + parser = argparse.ArgumentParser(description="One tick of the autoresearch loop.") + parser.add_argument("--root", required=True, type=Path, help="state root on the shared FS") + parser.add_argument("--grace-s", type=float, default=DEFAULT_GRACE_S) + parser.add_argument("--lease-ttl-s", type=float, default=DEFAULT_LEASE_TTL_S) + args = parser.parse_args() + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s") + + args.root.mkdir(parents=True, exist_ok=True) + report = tick(args.root, SlurmCompute(), LoggingDispatcher(), now=time.time()) + log.info( + "tick done: paused=%s swept=%d woken=%d deferred=%d reaped=%d stuck=%d", + report.paused, + report.swept, + len(report.woken), + len(report.deferred), + len(report.reaped_leases), + len(report.stuck), + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_compute.py b/tests/test_compute.py new file mode 100644 index 00000000..8403a8c9 --- /dev/null +++ b/tests/test_compute.py @@ -0,0 +1,137 @@ +"""Slurm seam tests against a fake runner — no cluster involved.""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +import pytest + +from autoresearch.compute import ( + GONE, + CommandResult, + JobSpec, + SlurmCompute, + SlurmError, + SlurmQueryError, + is_pending, + is_terminal, + quote_command, +) + +SPEC = JobSpec( + job_name="test-job", + account="acct", + partition="cpu_short", + time_minutes=10, + command="echo hi", +) + + +@dataclass +class FakeRunner: + results: list[CommandResult] + seen: list[list[str]] = field(default_factory=list) + + def __call__(self, argv, timeout_s): + self.seen.append(list(argv)) + return self.results.pop(0) + + +def test_submit_parses_job_id() -> None: + runner = FakeRunner([CommandResult(0, "12345\n", "")]) + assert SlurmCompute(runner=runner).submit(SPEC) == "12345" + argv = runner.seen[0] + assert argv[0] == "sbatch" + assert "--parsable" in argv + assert "--wrap=echo hi" in argv + + +def test_submit_parses_cluster_suffixed_id() -> None: + runner = FakeRunner([CommandResult(0, "12345;torch\n", "")]) + assert SlurmCompute(runner=runner).submit(SPEC) == "12345" + + +def test_submit_failure_raises_with_stderr() -> None: + runner = FakeRunner([CommandResult(1, "", "Invalid qos specification")]) + with pytest.raises(SlurmError, match="Invalid qos"): + SlurmCompute(runner=runner).submit(SPEC) + + +def test_submit_garbage_output_raises() -> None: + runner = FakeRunner([CommandResult(0, "not-a-job-id", "")]) + with pytest.raises(SlurmError, match="no job id"): + SlurmCompute(runner=runner).submit(SPEC) + + +def test_submit_after_sets_afterany() -> None: + runner = FakeRunner([CommandResult(0, "777\n", "")]) + SlurmCompute(runner=runner).submit_after(SPEC, "12345") + assert "--dependency=afterany:12345" in runner.seen[0] + + +def test_submit_after_rejects_non_numeric_id() -> None: + with pytest.raises(ValueError): + SlurmCompute(runner=FakeRunner([])).submit_after(SPEC, "$(rm -rf /)") + + +def test_status_distinguishes_gone_from_query_failure() -> None: + """The fail-safe design's core distinction: empty-on-success vs error.""" + gone = SlurmCompute(runner=FakeRunner([CommandResult(0, "", "")])) + assert gone.status("1") == GONE + + outage = SlurmCompute(runner=FakeRunner([CommandResult(1, "", "slurmdbd down")])) + with pytest.raises(SlurmQueryError, match="slurmdbd down"): + outage.status("1") + + +def test_status_returns_state_string() -> None: + runner = FakeRunner([CommandResult(0, "RUNNING\n", "")]) + assert SlurmCompute(runner=runner).status("1") == "RUNNING" + + +def test_status_rejects_injection_shaped_ids() -> None: + with pytest.raises(ValueError): + SlurmCompute(runner=FakeRunner([])).status("1; rm -rf /") + + +def test_terminal_and_pending_predicates() -> None: + assert is_terminal("COMPLETED") + assert is_terminal("CANCELLED by 501") + assert is_terminal("FAILED") + assert is_terminal("TIMEOUT") + assert not is_terminal("RUNNING") + assert not is_terminal("PENDING") + assert not is_terminal(GONE) # the deadline floor decides, not this + assert is_pending("PENDING") + + +def test_jobspec_requires_exactly_one_payload() -> None: + with pytest.raises(ValueError): + JobSpec(job_name="x", account="a", partition="p", time_minutes=1).to_argv() + with pytest.raises(ValueError): + JobSpec( + job_name="x", + account="a", + partition="p", + time_minutes=1, + command="c", + script="s.sh", + ).to_argv() + + +def test_jobspec_script_form_appends_args() -> None: + argv = JobSpec( + job_name="x", + account="a", + partition="p", + time_minutes=1, + script="run.sh", + script_args=("1", "2"), + ).to_argv() + assert argv[-3:] == ["run.sh", "1", "2"] + + +def test_quote_command_shell_safety() -> None: + quoted = quote_command(["python", "-c", "print('hi; rm -rf /')"]) + assert "'" in quoted + assert quoted.startswith("python -c ") diff --git a/tests/test_runstate.py b/tests/test_runstate.py new file mode 100644 index 00000000..4b2e1c0b --- /dev/null +++ b/tests/test_runstate.py @@ -0,0 +1,106 @@ +"""Run-state and lease semantics — the durable half of the agent.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from autoresearch.runstate import ( + ENDED, + STUCK, + WAITING, + RunRecord, + acquire_lease, + lease_is_stale, + list_runs, + load_record, + read_lease, + release_lease, + run_dir, + save_record, + update_lease_holder, +) + + +def make_record(**overrides) -> RunRecord: + base = dict(run_id="r1", target="org/repo", task_title="t", state=WAITING) + return RunRecord(**{**base, **overrides}) + + +def test_save_load_roundtrip(tmp_path: Path) -> None: + save_record(tmp_path, make_record(experiment_job_id="9"), now=100.0) + loaded = load_record(tmp_path, "r1") + assert loaded.experiment_job_id == "9" + assert loaded.created == 100.0 + assert loaded.updated == 100.0 + + +def test_save_stamps_updated_but_keeps_created(tmp_path: Path) -> None: + save_record(tmp_path, make_record(), now=100.0) + save_record(tmp_path, load_record(tmp_path, "r1"), now=200.0) + loaded = load_record(tmp_path, "r1") + assert loaded.created == 100.0 + assert loaded.updated == 200.0 + + +def test_save_is_atomic_no_tmp_left_behind(tmp_path: Path) -> None: + save_record(tmp_path, make_record(), now=1.0) + names = {p.name for p in run_dir(tmp_path, "r1").iterdir()} + assert names == {"state.json"} + + +def test_invalid_state_and_ending_are_rejected(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="unknown state"): + save_record(tmp_path, make_record(state="dancing"), now=1.0) + with pytest.raises(ValueError, match="valid ending"): + save_record(tmp_path, make_record(state=ENDED, ending="tired"), now=1.0) + save_record(tmp_path, make_record(state=ENDED, ending=STUCK), now=1.0) # ok + + +def test_list_runs_skips_corrupt_records(tmp_path: Path, caplog) -> None: + save_record(tmp_path, make_record(run_id="good"), now=1.0) + bad = run_dir(tmp_path, "bad") + bad.mkdir(parents=True) + (bad / "state.json").write_text("{not json") + records = list_runs(tmp_path) + assert [r.run_id for r in records] == ["good"] + assert "unreadable" in caplog.text + + +def test_lease_exactly_one_winner(tmp_path: Path) -> None: + assert acquire_lease(tmp_path, "r1", "tick:a", "", now=1.0) + assert not acquire_lease(tmp_path, "r1", "tick:b", "", now=2.0) + lease = read_lease(tmp_path, "r1") + assert lease is not None and lease.holder == "tick:a" + + +def test_lease_release_then_reacquire(tmp_path: Path) -> None: + acquire_lease(tmp_path, "r1", "a", "", now=1.0) + release_lease(tmp_path, "r1") + release_lease(tmp_path, "r1") # idempotent + assert acquire_lease(tmp_path, "r1", "b", "", now=2.0) + + +def test_lease_handoff_updates_holder(tmp_path: Path) -> None: + acquire_lease(tmp_path, "r1", "tick:x", "", now=1.0) + update_lease_holder(tmp_path, "r1", "wake-job:99", "99", now=2.0) + lease = read_lease(tmp_path, "r1") + assert lease is not None + assert lease.holder_job_id == "99" + assert not acquire_lease(tmp_path, "r1", "other", "", now=3.0) # still held + + +def test_lease_staleness_rules(tmp_path: Path) -> None: + acquire_lease(tmp_path, "r1", "h", "77", now=1000.0) + lease = read_lease(tmp_path, "r1") + assert lease is not None + # dead holder → stale regardless of age + assert lease_is_stale(lease, now=1001.0, ttl_s=3600, holder_alive=False) + # live holder, young → not stale + assert not lease_is_stale(lease, now=1001.0, ttl_s=3600, holder_alive=True) + # Slurm unknown → only the TTL can prove staleness + assert not lease_is_stale(lease, now=1001.0, ttl_s=3600, holder_alive=None) + assert lease_is_stale(lease, now=1000.0 + 3601, ttl_s=3600, holder_alive=None) + # live holder but ancient → stale (TTL wins: sessions are bounded) + assert lease_is_stale(lease, now=1000.0 + 3601, ttl_s=3600, holder_alive=True) diff --git a/tests/test_tick.py b/tests/test_tick.py new file mode 100644 index 00000000..79db42ba --- /dev/null +++ b/tests/test_tick.py @@ -0,0 +1,222 @@ +"""The tick's fail-safe sweep, scenario by scenario (fake Slurm, fake wakes).""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path + +from autoresearch.compute import CommandResult, SlurmCompute +from autoresearch.runstate import ( + ENDED, + STUCK, + WAITING, + RunRecord, + acquire_lease, + load_record, + read_lease, + save_record, +) +from autoresearch.tick import ( + PAUSE_SENTINEL, + RecordingDispatcher, + tick, +) + +NOW = 1_000_000.0 +GRACE = 900.0 +TTL = 4500.0 + + +@dataclass +class FakeSlurm: + """status() by job id; '!' prefix means the query itself fails.""" + + states: dict[str, str] = field(default_factory=dict) + cancelled: list[str] = field(default_factory=list) + + def _runner(self, argv, timeout_s): + if argv[0] == "sacct": + job_id = argv[argv.index("-j") + 1] + state = self.states.get(job_id, "") + if state == "!": + return CommandResult(1, "", "slurmdbd down") + return CommandResult(0, state + "\n" if state else "", "") + if argv[0] == "scancel": + self.cancelled.append(argv[1]) + return CommandResult(0, "", "") + raise AssertionError(f"unexpected command {argv}") + + def compute(self) -> SlurmCompute: + return SlurmCompute(runner=self._runner) + + +def waiting_run(root: Path, run_id: str = "r1", **overrides) -> RunRecord: + base = dict( + run_id=run_id, + target="org/repo", + task_title="t", + state=WAITING, + experiment_job_id="100", + deadline=NOW + 10_000, + ) + record = RunRecord(**{**base, **overrides}) + # created/updated stamp: long enough ago that the grace window has passed + save_record(root, record, now=NOW - GRACE - 1) + return record + + +def run_tick(root: Path, slurm: FakeSlurm, dispatcher=None, now: float = NOW): + dispatcher = dispatcher if dispatcher is not None else RecordingDispatcher() + report = tick(root, slurm.compute(), dispatcher, now=now) + return report, dispatcher + + +def test_pause_sentinel_noops_but_heartbeats(tmp_path: Path) -> None: + (tmp_path / PAUSE_SENTINEL).touch() + waiting_run(tmp_path) + report, dispatcher = run_tick(tmp_path, FakeSlurm(states={"100": "COMPLETED"})) + assert report.paused + assert dispatcher.dispatched == [] + assert (tmp_path / "heartbeat.json").exists() + + +def test_terminal_experiment_past_grace_gets_backup_wake(tmp_path: Path) -> None: + waiting_run(tmp_path) + report, dispatcher = run_tick(tmp_path, FakeSlurm(states={"100": "FAILED"})) + assert report.woken == (("r1", "FAILED"),) + assert dispatcher.dispatched == [("r1", "experiment FAILED")] + # sync dispatch → lease released afterward + assert read_lease(tmp_path, "r1") is None + assert load_record(tmp_path, "r1").wake_attempts == 1 + + +def test_terminal_within_grace_leaves_it_to_the_afterany_job(tmp_path: Path) -> None: + record = waiting_run(tmp_path) + save_record(tmp_path, record, now=NOW - 10) # updated moments ago + report, dispatcher = run_tick(tmp_path, FakeSlurm(states={"100": "COMPLETED"})) + assert report.woken == () + assert dispatcher.dispatched == [] + + +def test_live_lease_blocks_the_sweep(tmp_path: Path) -> None: + waiting_run(tmp_path) + acquire_lease(tmp_path, "r1", "wake-job:55", "55", now=NOW - 60) + report, dispatcher = run_tick(tmp_path, FakeSlurm(states={"100": "COMPLETED", "55": "RUNNING"})) + assert dispatcher.dispatched == [] + assert report.reaped_leases == () + + +def test_dead_holder_lease_is_reaped_then_next_tick_wakes(tmp_path: Path) -> None: + waiting_run(tmp_path) + acquire_lease(tmp_path, "r1", "wake-job:55", "55", now=NOW - 60) + slurm = FakeSlurm(states={"100": "COMPLETED", "55": "FAILED"}) + report, dispatcher = run_tick(tmp_path, slurm) + assert report.reaped_leases == ("r1",) + # the same sweep pass continues after reaping — wake delivered + assert dispatcher.dispatched == [("r1", "experiment COMPLETED")] + + +def test_query_failure_defers_never_concludes(tmp_path: Path) -> None: + waiting_run(tmp_path, deadline=NOW - 1) # even past deadline! + report, dispatcher = run_tick(tmp_path, FakeSlurm(states={"100": "!"})) + assert report.deferred == ("r1",) + assert dispatcher.dispatched == [] + assert load_record(tmp_path, "r1").state == WAITING + + +def test_gone_before_deadline_waits_for_sacct_lag(tmp_path: Path) -> None: + waiting_run(tmp_path) # deadline far in the future + report, dispatcher = run_tick(tmp_path, FakeSlurm(states={})) # sacct empty + assert dispatcher.dispatched == [] + + +def test_gone_past_deadline_wakes_with_vanished(tmp_path: Path) -> None: + waiting_run(tmp_path, deadline=NOW - 1) + report, dispatcher = run_tick(tmp_path, FakeSlurm(states={})) + assert report.woken == (("r1", "vanished"),) + + +def test_pending_past_deadline_cancels_then_wakes(tmp_path: Path) -> None: + waiting_run(tmp_path, deadline=NOW - 1) + slurm = FakeSlurm(states={"100": "PENDING"}) + report, dispatcher = run_tick(tmp_path, slurm) + assert slurm.cancelled == ["100"] + assert report.woken == (("r1", "unschedulable"),) + + +def test_pending_before_deadline_is_left_alone(tmp_path: Path) -> None: + waiting_run(tmp_path) + slurm = FakeSlurm(states={"100": "PENDING"}) + report, dispatcher = run_tick(tmp_path, slurm) + assert slurm.cancelled == [] + assert dispatcher.dispatched == [] + + +def test_running_experiment_is_left_alone(tmp_path: Path) -> None: + waiting_run(tmp_path) + report, dispatcher = run_tick(tmp_path, FakeSlurm(states={"100": "RUNNING"})) + assert dispatcher.dispatched == [] + + +def test_attempts_exhausted_becomes_stuck(tmp_path: Path) -> None: + waiting_run(tmp_path, wake_attempts=3) + report, dispatcher = run_tick(tmp_path, FakeSlurm(states={"100": "COMPLETED"})) + assert report.stuck == ("r1",) + assert dispatcher.dispatched == [] + ended = load_record(tmp_path, "r1") + assert ended.state == ENDED + assert ended.ending == STUCK + + +def test_dispatch_exception_releases_lease_and_counts_attempt(tmp_path: Path) -> None: + waiting_run(tmp_path) + + class ExplodingDispatcher: + def dispatch(self, record, reason): + raise RuntimeError("boom") + + report, _ = run_tick(tmp_path, FakeSlurm(states={"100": "COMPLETED"}), ExplodingDispatcher()) + assert report.woken == () + assert read_lease(tmp_path, "r1") is None # released, not stranded + assert load_record(tmp_path, "r1").wake_attempts == 1 # counts toward stuck + + +def test_async_dispatch_hands_lease_to_wake_job(tmp_path: Path) -> None: + waiting_run(tmp_path) + dispatcher = RecordingDispatcher(holder_job_id="777") + report, _ = run_tick(tmp_path, FakeSlurm(states={"100": "COMPLETED"}), dispatcher) + assert report.woken == (("r1", "COMPLETED"),) + lease = read_lease(tmp_path, "r1") + assert lease is not None + assert lease.holder_job_id == "777" + + +def test_double_tick_no_double_wake_with_async_dispatch(tmp_path: Path) -> None: + """The lease is exactly what makes the backup layer idempotent.""" + waiting_run(tmp_path) + slurm = FakeSlurm(states={"100": "COMPLETED", "777": "RUNNING"}) + dispatcher = RecordingDispatcher(holder_job_id="777") + run_tick(tmp_path, slurm, dispatcher) + run_tick(tmp_path, slurm, dispatcher, now=NOW + 60) + assert len(dispatcher.dispatched) == 1 + + +def test_ended_and_non_waiting_runs_are_ignored(tmp_path: Path) -> None: + waiting_run(tmp_path, run_id="active", state="implementing") + save_record( + tmp_path, + RunRecord(run_id="done", target="o/r", task_title="t", state=ENDED, ending="merged"), + now=NOW - 5000, + ) + report, dispatcher = run_tick(tmp_path, FakeSlurm()) + assert report.swept == 0 + assert dispatcher.dispatched == [] + + +def test_heartbeat_written_every_tick(tmp_path: Path) -> None: + run_tick(tmp_path, FakeSlurm()) + import json + + beat = json.loads((tmp_path / "heartbeat.json").read_text()) + assert beat["ts"] == NOW + assert "host" in beat From c2f27cef59a9314ec906dba4c66265737e1c3c82 Mon Sep 17 00:00:00 2001 From: Mengye Ren Date: Thu, 6 Aug 2026 10:05:46 -0400 Subject: [PATCH 2/4] Apply all eight review findings: askpass PAT, terminal_seen grace, dry-run CLI, mtime lease fallback, tombstone reap, deadline floor validation, forward-compat records, lease-first ordering Co-Authored-By: Claude Fable 5 --- scripts/tick_chain.sbatch | 11 ++-- src/autoresearch/compute.py | 21 ++++--- src/autoresearch/runstate.py | 53 +++++++++++++--- src/autoresearch/tick.py | 114 +++++++++++++++++++++-------------- tests/test_runstate.py | 38 +++++++++++- tests/test_tick.py | 111 +++++++++++++++++++++++++++++++++- 6 files changed, 276 insertions(+), 72 deletions(-) diff --git a/scripts/tick_chain.sbatch b/scripts/tick_chain.sbatch index 0ac12c49..636b8458 100755 --- a/scripts/tick_chain.sbatch +++ b/scripts/tick_chain.sbatch @@ -52,15 +52,18 @@ for i in $(seq 1 "$need"); do done # --- 2. deploy: pull main with the bot PAT, sync deps (best-effort) --- +# The PAT never appears in argv (argv is world-readable via /proc on shared +# nodes): git asks for it through GIT_ASKPASS instead. if [ -n "${AUTORESEARCH_PAT_FILE:-}" ] && [ -r "$AUTORESEARCH_PAT_FILE" ]; then - PAT=$(cat "$AUTORESEARCH_PAT_FILE") - if git -C "$AUTORESEARCH_HOME" fetch --quiet \ - "https://x-access-token:${PAT}@github.com/agentic-learning-ai-lab/autoresearch.git" main; then + ASKPASS=$(mktemp) && chmod 700 "$ASKPASS" + printf '#!/bin/sh\ncat "%s"\n' "$AUTORESEARCH_PAT_FILE" > "$ASKPASS" + if GIT_ASKPASS="$ASKPASS" GIT_TERMINAL_PROMPT=0 git -C "$AUTORESEARCH_HOME" fetch --quiet \ + "https://x-access-token@github.com/agentic-learning-ai-lab/autoresearch.git" main; then git -C "$AUTORESEARCH_HOME" reset --hard --quiet FETCH_HEAD || echo "deploy: reset failed" else echo "deploy: fetch failed; running previous code" fi - unset PAT + rm -f "$ASKPASS" fi (cd "$AUTORESEARCH_HOME" && uv sync --locked --quiet) || echo "deploy: uv sync failed" diff --git a/src/autoresearch/compute.py b/src/autoresearch/compute.py index b901b761..fa998ac7 100644 --- a/src/autoresearch/compute.py +++ b/src/autoresearch/compute.py @@ -67,7 +67,11 @@ def _subprocess_runner(argv: Sequence[str], timeout_s: int) -> CommandResult: @dataclass(frozen=True) class JobSpec: """One sbatch submission. `command` is run via --wrap; a script path can - be passed as `script` instead (mutually exclusive).""" + be passed as `script` instead (mutually exclusive). + + --wrap executes under a shell on the compute node: `command` must be + built from trusted parts, with anything variable passed through + `quote_command`. Never interpolate agent- or contract-supplied text.""" job_name: str account: str @@ -138,16 +142,15 @@ def submit(self, spec: JobSpec) -> str: def submit_after(self, spec: JobSpec, after_job_id: str) -> str: """Submit `spec` to run when `after_job_id` terminates — however it terminates (afterany: the wake-on-failure semantics the fail-safe - design requires; verified live on Torch 2026-08-06).""" + design requires; verified live on Torch 2026-08-06). Refuses a spec + that already carries a dependency rather than silently replacing it.""" if not after_job_id.isdigit(): raise ValueError(f"not a job id: {after_job_id!r}") - dependent = JobSpec( - **{ - **{f: getattr(spec, f) for f in spec.__dataclass_fields__}, - "dependency": f"afterany:{after_job_id}", - } - ) - return self.submit(dependent) + if spec.dependency: + raise ValueError(f"spec already has dependency {spec.dependency!r}") + import dataclasses + + return self.submit(dataclasses.replace(spec, dependency=f"afterany:{after_job_id}")) def status(self, job_id: str) -> str: """The job's Slurm state, or GONE when a *successful* query finds no diff --git a/src/autoresearch/runstate.py b/src/autoresearch/runstate.py index 789ff40a..a38ba643 100644 --- a/src/autoresearch/runstate.py +++ b/src/autoresearch/runstate.py @@ -61,6 +61,7 @@ class RunRecord: resume_session_id: str = "" # harness session to resume on wake wake_attempts: int = 0 deadline: float = 0.0 # unix; submit+walltime+slack, re-based on start + terminal_seen: float = 0.0 # when the sweep first saw the experiment terminal ending: str = "" # one of ENDINGS once state == ENDED ending_note: str = "" created: float = 0.0 @@ -87,6 +88,10 @@ def save_record(root: Path, record: RunRecord, now: float) -> None: raise ValueError(f"unknown state {record.state!r}") if record.state == ENDED and record.ending not in ENDINGS: raise ValueError(f"ended run needs a valid ending, got {record.ending!r}") + if record.state == WAITING and record.experiment_job_id and record.deadline <= 0: + # A waiting run without a deadline is invisible to the deadline floor + # — the exact "silently immortal run" the fail-safe design forbids. + raise ValueError("waiting run with an experiment needs a deadline") directory = run_dir(root, record.run_id) directory.mkdir(parents=True, exist_ok=True) stamped = replace(record, updated=now, created=record.created or now) @@ -97,7 +102,11 @@ def save_record(root: Path, record: RunRecord, now: float) -> None: def load_record(root: Path, run_id: str) -> RunRecord: raw = json.loads((run_dir(root, run_id) / RECORD_NAME).read_text()) - return RunRecord(**raw) + # Ignore unknown keys: after a bad-merge revert, older code must still be + # able to read records written by newer code — a "corrupt" verdict here + # would blind the sweep to the whole run. + known = {k: v for k, v in raw.items() if k in RunRecord.__dataclass_fields__} + return RunRecord(**known) def list_runs(root: Path) -> list[RunRecord]: @@ -121,9 +130,10 @@ def list_runs(root: Path) -> list[RunRecord]: def acquire_lease(root: Path, run_id: str, holder: str, holder_job_id: str, now: float) -> bool: """Take the run's wake lease. True if acquired; False if held. - O_EXCL makes acquisition atomic on the shared filesystem: exactly one - contender wins, the rest see False and no-op (double delivery is - harmless by design). + O_EXCL makes acquisition atomic: exactly one contender wins, the rest see + False and no-op. (O_EXCL is reliable on NFSv4/GPFS/Lustre; if the state + root ever lands on NFSv3, this needs a link(2)-based lock instead — + verify the cluster filesystem before trusting the lease.) """ directory = run_dir(root, run_id) directory.mkdir(parents=True, exist_ok=True) @@ -138,16 +148,22 @@ def acquire_lease(root: Path, run_id: str, holder: str, holder_job_id: str, now: def read_lease(root: Path, run_id: str) -> Lease | None: + path = run_dir(root, run_id) / LEASE_NAME try: - raw = json.loads((run_dir(root, run_id) / LEASE_NAME).read_text()) + raw = json.loads(path.read_text()) return Lease(**raw) except FileNotFoundError: return None except (OSError, ValueError, TypeError, KeyError): - # An unreadable lease is treated as held-but-unknown; the TTL path - # in `lease_is_stale` cannot run without a timestamp, so the sweep - # falls back to reaping it after the grace window via mtime. - return None + # A crash between O_EXCL create and write leaves an empty/corrupt + # lease. Synthesize one from the file mtime so the TTL path can + # still reap it — otherwise the run is stranded forever behind a + # lease nobody can read. + try: + mtime = path.stat().st_mtime + except OSError: + return None # vanished between read and stat + return Lease(holder="unreadable", holder_job_id="", acquired=mtime) def update_lease_holder( @@ -162,12 +178,31 @@ def update_lease_holder( def release_lease(root: Path, run_id: str) -> None: + """For the lease HOLDER only. Non-holders must use reap_lease.""" try: (run_dir(root, run_id) / LEASE_NAME).unlink() except FileNotFoundError: pass +def reap_lease(root: Path, run_id: str, reaper: str) -> bool: + """Remove a stale lease you do NOT hold. True if THIS caller reaped it. + + Atomic rename to a unique tombstone: with two concurrent reapers exactly + one rename succeeds, so exactly one proceeds to redeliver — a blind + unlink here could delete a lease a faster reaper already handed to a new + wake job, re-opening the double-delivery hole. + """ + directory = run_dir(root, run_id) + tombstone = directory / f".{LEASE_NAME}.reaped.{reaper}" + try: + os.rename(directory / LEASE_NAME, tombstone) + except FileNotFoundError: + return False + tombstone.unlink(missing_ok=True) + return True + + def lease_is_stale(lease: Lease, now: float, ttl_s: float, holder_alive: bool | None) -> bool: """A lease is stale when its holder is known-dead, or too old. diff --git a/src/autoresearch/tick.py b/src/autoresearch/tick.py index 971be696..02182baa 100644 --- a/src/autoresearch/tick.py +++ b/src/autoresearch/tick.py @@ -32,6 +32,7 @@ lease_is_stale, list_runs, read_lease, + reap_lease, release_lease, save_record, update_lease_holder, @@ -115,7 +116,13 @@ def _wake( """ if not acquire_lease(root, record.run_id, holder, holder_job_id="", now=now): return False - bumped = replace(record, wake_attempts=record.wake_attempts + 1) + bumped = replace( + record, + wake_attempts=record.wake_attempts + 1, + # repair legacy records as we touch them: save_record (rightly) + # refuses to write a waiting run without a deadline + deadline=record.deadline if record.deadline > 0 else now, + ) save_record(root, bumped, now) try: holder_job = dispatcher.dispatch(bumped, reason) @@ -139,8 +146,14 @@ def sweep( now: float, grace_s: float = DEFAULT_GRACE_S, lease_ttl_s: float = DEFAULT_LEASE_TTL_S, + dry_run: bool = False, ) -> TickReport: - """The backup wake layers, applied to every waiting run.""" + """The backup wake layers, applied to every waiting run. + + dry_run reports what WOULD happen with zero writes — no leases, no + attempt counters, no dispatch — so the plumbing can run live before the + real dispatcher exists. + """ woken: list[tuple[str, str]] = [] deferred: list[str] = [] reaped: list[str] = [] @@ -148,29 +161,37 @@ def sweep( holder = f"tick:{socket.gethostname()}:{os.getpid()}" records = [r for r in list_runs(root) if r.state == WAITING] - for record in records: - # Layer 5 first: too many failed attempts is a terminal, reported state. - if record.wake_attempts >= MAX_WAKE_ATTEMPTS: - ended = replace( - record, - state=ENDED, - ending=STUCK, - ending_note=f"{record.wake_attempts} wake attempts failed", - ) - save_record(root, ended, now) - stuck.append(record.run_id) - continue + def wake(record: RunRecord, reason: str, tag: str) -> None: + if dry_run or _wake(root, record, reason, dispatcher, now, holder): + woken.append((record.run_id, tag)) - # Stale-lease reaping (layer 2's expiry): a dead wake must not hold on. + for record in records: + # Leases first: a LIVE wake in flight owns this run — even the stuck + # verdict must wait for it (its session may be the one that succeeds). lease = read_lease(root, record.run_id) if lease is not None: alive = _holder_alive(compute, lease.holder_job_id) - if lease_is_stale(lease, now, lease_ttl_s, alive): - release_lease(root, record.run_id) + if not lease_is_stale(lease, now, lease_ttl_s, alive): + continue + if dry_run: reaped.append(record.run_id) - lease = None - else: - continue # a live wake is in flight; nothing for the sweep + continue + if not reap_lease(root, record.run_id, reaper=f"{os.getpid()}-{now}"): + continue # a concurrent tick reaped it first; it owns redelivery + reaped.append(record.run_id) + + # Layer 5: too many failed attempts is a terminal, reported state. + if record.wake_attempts >= MAX_WAKE_ATTEMPTS: + if not dry_run: + ended = replace( + record, + state=ENDED, + ending=STUCK, + ending_note=f"{record.wake_attempts} wake attempts failed", + ) + save_record(root, ended, now) + stuck.append(record.run_id) + continue if not record.experiment_job_id: continue # not yet submitted; not the sweep's business @@ -182,33 +203,31 @@ def sweep( deferred.append(record.run_id) continue - past_deadline = record.deadline and now > record.deadline + # deadline <= 0 cannot be written by save_record for waiting runs; + # if one exists anyway (legacy/hand-edited), treat it as already past + # for GONE — a vanished-experiment wake is safe — but never for + # PENDING, where the consequence would be cancelling a healthy job. + past_deadline = record.deadline <= 0 or now > record.deadline if is_terminal(state): - # Layer 3: terminal + no live lease + grace expired → backup wake. - # (updated < now - grace approximates "terminal for a while": the - # record was last touched when the experiment was submitted.) - if now - record.updated >= grace_s: - if _wake(root, record, f"experiment {state}", dispatcher, now, holder): - woken.append((record.run_id, state)) + # Layer 3, with real grace: time runs from when the sweep FIRST + # saw the experiment terminal, not from submission — the afterany + # job gets the full window to deliver before the backup steps in. + if record.terminal_seen <= 0: + if not dry_run: + save_record(root, replace(record, terminal_seen=now), now) + continue + if now - record.terminal_seen >= grace_s: + wake(record, f"experiment {state}", state) elif state == GONE: if past_deadline: - # Successful query, no record, deadline passed: vanished. - if _wake(root, record, "experiment vanished from Slurm", dispatcher, now, holder): - woken.append((record.run_id, "vanished")) + wake(record, "experiment vanished from Slurm", "vanished") # else: sacct lag right after submission is normal; wait. - elif is_pending(state) and past_deadline: + elif is_pending(state) and record.deadline > 0 and now > record.deadline: # Unschedulable in practice: cancel, then wake with that fact. - compute.cancel(record.experiment_job_id) - if _wake( - root, - record, - "experiment unschedulable (pending past deadline)", - dispatcher, - now, - holder, - ): - woken.append((record.run_id, "unschedulable")) + if not dry_run: + compute.cancel(record.experiment_job_id) + wake(record, "experiment unschedulable (pending past deadline)", "unschedulable") # RUNNING (or recently pending): nothing to do; the afterany job has it. return TickReport( @@ -225,6 +244,7 @@ def tick( compute: SlurmCompute, dispatcher: WakeDispatcher, now: float, + dry_run: bool = False, ) -> TickReport: """One full tick. Pause sentinel wins over everything: a paused loop heartbeats (so the watchdog stays quiet) but touches nothing.""" @@ -232,14 +252,14 @@ def tick( if (root / PAUSE_SENTINEL).exists(): log.info("pause sentinel present; tick is a no-op") return TickReport(paused=True) - return sweep(root, compute, dispatcher, now) + return sweep(root, compute, dispatcher, now, dry_run=dry_run) @dataclass class LoggingDispatcher: - """Placeholder production dispatcher until session dispatch lands - (phase 5): logs what would be woken so the loop's plumbing can run live - without side effects.""" + """Never dispatched in production today: main() runs the sweep in + dry_run mode until the real session dispatcher lands (phase 5), so no + lease is taken and no attempt is counted. This exists for the seam.""" def dispatch(self, record: RunRecord, reason: str) -> str: log.info("WOULD WAKE %s (%s) — session dispatch lands in phase 5", record.run_id, reason) @@ -258,7 +278,9 @@ def main() -> int: logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s") args.root.mkdir(parents=True, exist_ok=True) - report = tick(args.root, SlurmCompute(), LoggingDispatcher(), now=time.time()) + # dry_run until the phase-5 dispatcher exists: the live loop must not + # mutate run state it cannot follow through on. + report = tick(args.root, SlurmCompute(), LoggingDispatcher(), now=time.time(), dry_run=True) log.info( "tick done: paused=%s swept=%d woken=%d deferred=%d reaped=%d stuck=%d", report.paused, diff --git a/tests/test_runstate.py b/tests/test_runstate.py index 4b2e1c0b..a2287884 100644 --- a/tests/test_runstate.py +++ b/tests/test_runstate.py @@ -24,7 +24,7 @@ def make_record(**overrides) -> RunRecord: - base = dict(run_id="r1", target="org/repo", task_title="t", state=WAITING) + base = dict(run_id="r1", target="org/repo", task_title="t", state=WAITING, deadline=999.0) return RunRecord(**{**base, **overrides}) @@ -104,3 +104,39 @@ def test_lease_staleness_rules(tmp_path: Path) -> None: assert lease_is_stale(lease, now=1000.0 + 3601, ttl_s=3600, holder_alive=None) # live holder but ancient → stale (TTL wins: sessions are bounded) assert lease_is_stale(lease, now=1000.0 + 3601, ttl_s=3600, holder_alive=True) + + +def test_reap_lease_exactly_one_reaper_wins(tmp_path: Path) -> None: + from autoresearch.runstate import reap_lease + + acquire_lease(tmp_path, "r1", "dead", "", now=1.0) + assert reap_lease(tmp_path, "r1", reaper="a") + assert not reap_lease(tmp_path, "r1", reaper="b") # already gone + assert acquire_lease(tmp_path, "r1", "next", "", now=2.0) + + +def test_load_record_ignores_unknown_keys(tmp_path: Path) -> None: + """After a bad-merge revert, old code must still read new-code records.""" + import json + + save_record(tmp_path, make_record(), now=1.0) + path = run_dir(tmp_path, "r1") / "state.json" + data = json.loads(path.read_text()) + data["field_from_the_future"] = 42 + path.write_text(json.dumps(data)) + assert load_record(tmp_path, "r1").run_id == "r1" + assert list_runs(tmp_path) # not treated as corrupt + + +def test_unreadable_lease_synthesizes_mtime_timestamp(tmp_path: Path) -> None: + import os + + directory = run_dir(tmp_path, "r1") + directory.mkdir(parents=True) + lease_path = directory / "lease.json" + lease_path.touch() + os.utime(lease_path, (500.0, 500.0)) + lease = read_lease(tmp_path, "r1") + assert lease is not None + assert lease.holder == "unreadable" + assert lease.acquired == 500.0 diff --git a/tests/test_tick.py b/tests/test_tick.py index 79db42ba..e5f476e5 100644 --- a/tests/test_tick.py +++ b/tests/test_tick.py @@ -58,9 +58,10 @@ def waiting_run(root: Path, run_id: str = "r1", **overrides) -> RunRecord: state=WAITING, experiment_job_id="100", deadline=NOW + 10_000, + # default: the sweep already saw the experiment terminal, grace passed + terminal_seen=NOW - GRACE - 1, ) record = RunRecord(**{**base, **overrides}) - # created/updated stamp: long enough ago that the grace window has passed save_record(root, record, now=NOW - GRACE - 1) return record @@ -91,13 +92,48 @@ def test_terminal_experiment_past_grace_gets_backup_wake(tmp_path: Path) -> None def test_terminal_within_grace_leaves_it_to_the_afterany_job(tmp_path: Path) -> None: - record = waiting_run(tmp_path) - save_record(tmp_path, record, now=NOW - 10) # updated moments ago + waiting_run(tmp_path, terminal_seen=NOW - 10) # first seen moments ago report, dispatcher = run_tick(tmp_path, FakeSlurm(states={"100": "COMPLETED"})) assert report.woken == () assert dispatcher.dispatched == [] +def test_first_terminal_sighting_starts_the_grace_clock_not_a_wake(tmp_path: Path) -> None: + """Grace runs from when the sweep FIRST saw the experiment terminal — a + 24h experiment must not be woken by the backup the instant it completes + (the afterany job owns the fresh case).""" + waiting_run(tmp_path, terminal_seen=0.0) + report, dispatcher = run_tick(tmp_path, FakeSlurm(states={"100": "COMPLETED"})) + assert dispatcher.dispatched == [] + assert load_record(tmp_path, "r1").terminal_seen == NOW + # attempts untouched by the sighting + assert load_record(tmp_path, "r1").wake_attempts == 0 + # next tick, grace elapsed → wake + report2, dispatcher2 = run_tick( + tmp_path, FakeSlurm(states={"100": "COMPLETED"}), now=NOW + GRACE + 1 + ) + assert report2.woken == (("r1", "COMPLETED"),) + + +def test_dry_run_reports_without_any_writes(tmp_path: Path) -> None: + """python -m autoresearch.tick runs exactly this until phase 5: a healthy + completed run must survive any number of dry ticks unchanged.""" + from autoresearch.tick import RecordingDispatcher as RD + from autoresearch.tick import tick as tick_fn + + waiting_run(tmp_path) + slurm = FakeSlurm(states={"100": "COMPLETED"}) + dispatcher = RD() + for i in range(5): + report = tick_fn(tmp_path, slurm.compute(), dispatcher, now=NOW + i * 60, dry_run=True) + assert report.woken == (("r1", "COMPLETED"),) + after = load_record(tmp_path, "r1") + assert after.wake_attempts == 0 + assert after.state == WAITING + assert dispatcher.dispatched == [] + assert read_lease(tmp_path, "r1") is None + + def test_live_lease_blocks_the_sweep(tmp_path: Path) -> None: waiting_run(tmp_path) acquire_lease(tmp_path, "r1", "wake-job:55", "55", now=NOW - 60) @@ -168,6 +204,75 @@ def test_attempts_exhausted_becomes_stuck(tmp_path: Path) -> None: assert ended.ending == STUCK +def test_live_lease_defers_even_the_stuck_verdict(tmp_path: Path) -> None: + """Attempt 3's wake session may be the one that succeeds — a run with a + LIVE lease must not be truncated to stuck underneath it.""" + waiting_run(tmp_path, wake_attempts=3) + acquire_lease(tmp_path, "r1", "wake-job:55", "55", now=NOW - 60) + report, dispatcher = run_tick(tmp_path, FakeSlurm(states={"100": "COMPLETED", "55": "RUNNING"})) + assert report.stuck == () + assert load_record(tmp_path, "r1").state == WAITING + + +def test_tick_held_lease_reaped_by_ttl_alone(tmp_path: Path) -> None: + """A crashed tick's lease has no holder job id — only the TTL can free + it. This is the sole escape path; a regression here strands runs.""" + waiting_run(tmp_path) + acquire_lease(tmp_path, "r1", "tick:dead-host:1", "", now=NOW - TTL - 1) + report, dispatcher = run_tick(tmp_path, FakeSlurm(states={"100": "COMPLETED"})) + assert report.reaped_leases == ("r1",) + assert dispatcher.dispatched == [("r1", "experiment COMPLETED")] + + +def test_corrupt_empty_lease_is_reaped_via_mtime(tmp_path: Path) -> None: + """A crash between lease create and write must not strand the run.""" + import os as _os + + waiting_run(tmp_path) + lease_path = tmp_path / "runs" / "r1" / "lease.json" + lease_path.touch() # empty: unreadable as JSON + old = NOW - TTL - 100 + _os.utime(lease_path, (old, old)) + report, dispatcher = run_tick(tmp_path, FakeSlurm(states={"100": "COMPLETED"})) + assert report.reaped_leases == ("r1",) + assert dispatcher.dispatched == [("r1", "experiment COMPLETED")] + + +def test_legacy_zero_deadline_still_wakes_gone_runs(tmp_path: Path) -> None: + """save_record forbids new waiting runs without a deadline, but a legacy + record must not be immortal: GONE wakes anyway (safe), PENDING does not + get cancelled (destructive).""" + import json as _json + + directory = tmp_path / "runs" / "legacy" + directory.mkdir(parents=True) + record = dict( + run_id="legacy", + target="o/r", + task_title="t", + state=WAITING, + agent_id="a", + experiment_job_id="100", + wake_job_id="", + resume_session_id="", + wake_attempts=0, + deadline=0.0, + terminal_seen=0.0, + ending="", + ending_note="", + created=NOW - 5000, + updated=NOW - 5000, + ) + (directory / "state.json").write_text(_json.dumps(record)) + report, _ = run_tick(tmp_path, FakeSlurm(states={})) # GONE + assert report.woken == (("legacy", "vanished"),) + + (directory / "state.json").write_text(_json.dumps(record)) + slurm = FakeSlurm(states={"100": "PENDING"}) + report2, _ = run_tick(tmp_path, slurm) + assert slurm.cancelled == [] # healthy pending job never cancelled + + def test_dispatch_exception_releases_lease_and_counts_attempt(tmp_path: Path) -> None: waiting_run(tmp_path) From 9085a1e4c56df0ec788a2c7625d9d6f40dbf5cab Mon Sep 17 00:00:00 2001 From: Mengye Ren Date: Thu, 6 Aug 2026 10:12:30 -0400 Subject: [PATCH 3/4] Live reviewer round: per-record sweep isolation, slot accounting, CLI flags wired, pre-top-up guards, forward-compat leases Co-Authored-By: Claude Fable 5 --- scripts/tick_chain.sbatch | 36 +++++++++--- src/autoresearch/runstate.py | 8 +-- src/autoresearch/tick.py | 105 ++++++++++++++++++++++++++++------- tests/test_tick.py | 104 +++++++++++++++++++++++++++++++--- 4 files changed, 210 insertions(+), 43 deletions(-) diff --git a/scripts/tick_chain.sbatch b/scripts/tick_chain.sbatch index 636b8458..78a814f4 100755 --- a/scripts/tick_chain.sbatch +++ b/scripts/tick_chain.sbatch @@ -22,28 +22,33 @@ #SBATCH --cpus-per-task=2 #SBATCH --output=/dev/null -set -u +# NO set -u before the top-up: a config mistake must fail loudly AFTER the +# chain has secured its successors, not break the chain silently. Required +# vars are checked by hand instead. CADENCE_MIN="${AUTORESEARCH_CADENCE_MIN:-30}" JOB_NAME="autoresearch-tick" -LOG_DIR="$AUTORESEARCH_ROOT/logs" -mkdir -p "$LOG_DIR" -exec >>"$LOG_DIR/tick-$(date +%Y%m%d).log" 2>&1 -echo "=== tick $(date -Is) on $(hostname -s) job=${SLURM_JOB_ID:-none}" +missing="" +for var in AUTORESEARCH_HOME AUTORESEARCH_ROOT AUTORESEARCH_ACCOUNT AUTORESEARCH_PARTITION; do + eval "val=\${$var:-}" + [ -z "$val" ] && missing="$missing $var" +done # --- 1. keep two successors queued (singleton serializes same-name jobs) --- +# Successors land on slots AFTER the ones already-queued jobs occupy — +# otherwise every top-up doubles the tick rate on the next slot. pending=$(squeue -u "$USER" --name="$JOB_NAME" -h -t PENDING 2>/dev/null | wc -l) need=$((2 - pending)) epoch_now=$(date +%s) cadence_s=$((CADENCE_MIN * 60)) next_slot=$(( (epoch_now / cadence_s + 1) * cadence_s )) for i in $(seq 1 "$need"); do - begin_epoch=$((next_slot + (i - 1) * cadence_s)) + begin_epoch=$((next_slot + (pending + i - 1) * cadence_s)) begin=$(date -d "@$begin_epoch" +%Y-%m-%dT%H:%M:%S 2>/dev/null \ || date -r "$begin_epoch" +%Y-%m-%dT%H:%M:%S) for attempt in 1 2 3; do if sbatch --dependency=singleton --begin="$begin" \ - --account="$AUTORESEARCH_ACCOUNT" --partition="$AUTORESEARCH_PARTITION" \ - "$AUTORESEARCH_HOME/scripts/tick_chain.sbatch"; then + --account="${AUTORESEARCH_ACCOUNT:-}" --partition="${AUTORESEARCH_PARTITION:-}" \ + "${AUTORESEARCH_HOME:-}/scripts/tick_chain.sbatch"; then break fi echo "sbatch retry $attempt failed; backing off" @@ -51,11 +56,24 @@ for i in $(seq 1 "$need"); do done done +# Only NOW may configuration problems kill this tick — successors are queued. +set -u +if [ -n "$missing" ]; then + echo "tick misconfigured; missing:$missing" >&2 + exit 1 +fi +LOG_DIR="$AUTORESEARCH_ROOT/logs" +mkdir -p "$LOG_DIR" || true +if [ -w "$LOG_DIR" ]; then + exec >>"$LOG_DIR/tick-$(date +%Y%m%d).log" 2>&1 +fi +echo "=== tick $(date -Is) on $(hostname -s) job=${SLURM_JOB_ID:-none}" + # --- 2. deploy: pull main with the bot PAT, sync deps (best-effort) --- # The PAT never appears in argv (argv is world-readable via /proc on shared # nodes): git asks for it through GIT_ASKPASS instead. if [ -n "${AUTORESEARCH_PAT_FILE:-}" ] && [ -r "$AUTORESEARCH_PAT_FILE" ]; then - ASKPASS=$(mktemp) && chmod 700 "$ASKPASS" + ASKPASS=$(mktemp 2>/dev/null || echo "") && [ -n "$ASKPASS" ] && chmod 700 "$ASKPASS" printf '#!/bin/sh\ncat "%s"\n' "$AUTORESEARCH_PAT_FILE" > "$ASKPASS" if GIT_ASKPASS="$ASKPASS" GIT_TERMINAL_PROMPT=0 git -C "$AUTORESEARCH_HOME" fetch --quiet \ "https://x-access-token@github.com/agentic-learning-ai-lab/autoresearch.git" main; then diff --git a/src/autoresearch/runstate.py b/src/autoresearch/runstate.py index a38ba643..4058efc7 100644 --- a/src/autoresearch/runstate.py +++ b/src/autoresearch/runstate.py @@ -14,6 +14,7 @@ from __future__ import annotations +import contextlib import json import logging import os @@ -151,7 +152,8 @@ def read_lease(root: Path, run_id: str) -> Lease | None: path = run_dir(root, run_id) / LEASE_NAME try: raw = json.loads(path.read_text()) - return Lease(**raw) + known = {k: v for k, v in raw.items() if k in Lease.__dataclass_fields__} + return Lease(**known) except FileNotFoundError: return None except (OSError, ValueError, TypeError, KeyError): @@ -179,10 +181,8 @@ def update_lease_holder( def release_lease(root: Path, run_id: str) -> None: """For the lease HOLDER only. Non-holders must use reap_lease.""" - try: + with contextlib.suppress(FileNotFoundError): (run_dir(root, run_id) / LEASE_NAME).unlink() - except FileNotFoundError: - pass def reap_lease(root: Path, run_id: str, reaper: str) -> bool: diff --git a/src/autoresearch/tick.py b/src/autoresearch/tick.py index 02182baa..8f2dc5f8 100644 --- a/src/autoresearch/tick.py +++ b/src/autoresearch/tick.py @@ -166,18 +166,62 @@ def wake(record: RunRecord, reason: str, tag: str) -> None: woken.append((record.run_id, tag)) for record in records: + try: + _sweep_one( + root, + compute, + dispatcher, + now, + grace_s, + lease_ttl_s, + dry_run, + record, + holder, + wake, + deferred, + reaped, + stuck, + ) + except Exception as exc: + log.warning("sweep failed on %s: %s: %s", record.run_id, type(exc).__name__, exc) + + return TickReport( + swept=len(records), + woken=tuple(woken), + deferred=tuple(deferred), + reaped_leases=tuple(reaped), + stuck=tuple(stuck), + ) + + +def _sweep_one( + root: Path, + compute: SlurmCompute, + dispatcher: WakeDispatcher, + now: float, + grace_s: float, + lease_ttl_s: float, + dry_run: bool, + record: RunRecord, + holder: str, + wake, + deferred: list[str], + reaped: list[str], + stuck: list[str], +) -> None: + if True: # Leases first: a LIVE wake in flight owns this run — even the stuck # verdict must wait for it (its session may be the one that succeeds). lease = read_lease(root, record.run_id) if lease is not None: alive = _holder_alive(compute, lease.holder_job_id) if not lease_is_stale(lease, now, lease_ttl_s, alive): - continue + return if dry_run: reaped.append(record.run_id) - continue + return if not reap_lease(root, record.run_id, reaper=f"{os.getpid()}-{now}"): - continue # a concurrent tick reaped it first; it owns redelivery + return # a concurrent tick reaped it first; it owns redelivery reaped.append(record.run_id) # Layer 5: too many failed attempts is a terminal, reported state. @@ -191,17 +235,17 @@ def wake(record: RunRecord, reason: str, tag: str) -> None: ) save_record(root, ended, now) stuck.append(record.run_id) - continue + return if not record.experiment_job_id: - continue # not yet submitted; not the sweep's business + return # not yet submitted; not the sweep's business try: state = compute.status(record.experiment_job_id) except SlurmQueryError: # Layer 4's rule: query failure is "Slurm unknown", never "gone". deferred.append(record.run_id) - continue + return # deadline <= 0 cannot be written by save_record for waiting runs; # if one exists anyway (legacy/hand-edited), treat it as already past @@ -214,9 +258,22 @@ def wake(record: RunRecord, reason: str, tag: str) -> None: # saw the experiment terminal, not from submission — the afterany # job gets the full window to deliver before the backup steps in. if record.terminal_seen <= 0: - if not dry_run: - save_record(root, replace(record, terminal_seen=now), now) - continue + if dry_run: + # no writes in dry-run: report the would-wake now so the + # terminal path is visible to live plumbing checks + wake(record, f"experiment {state}", state) + else: + save_record( + root, + replace( + record, + terminal_seen=now, + # repair legacy records as we touch them (see _wake) + deadline=record.deadline if record.deadline > 0 else now, + ), + now, + ) + return if now - record.terminal_seen >= grace_s: wake(record, f"experiment {state}", state) elif state == GONE: @@ -224,26 +281,24 @@ def wake(record: RunRecord, reason: str, tag: str) -> None: wake(record, "experiment vanished from Slurm", "vanished") # else: sacct lag right after submission is normal; wait. elif is_pending(state) and record.deadline > 0 and now > record.deadline: - # Unschedulable in practice: cancel, then wake with that fact. + # Unschedulable in practice: cancel (best-effort — scancel + # trouble must not abort the sweep), then wake with that fact. if not dry_run: - compute.cancel(record.experiment_job_id) + try: + compute.cancel(record.experiment_job_id) + except Exception as exc: # scancel trouble is never fatal here + log.warning("cancel %s failed: %s", record.experiment_job_id, exc) wake(record, "experiment unschedulable (pending past deadline)", "unschedulable") # RUNNING (or recently pending): nothing to do; the afterany job has it. - return TickReport( - swept=len(records), - woken=tuple(woken), - deferred=tuple(deferred), - reaped_leases=tuple(reaped), - stuck=tuple(stuck), - ) - def tick( root: Path, compute: SlurmCompute, dispatcher: WakeDispatcher, now: float, + grace_s: float = DEFAULT_GRACE_S, + lease_ttl_s: float = DEFAULT_LEASE_TTL_S, dry_run: bool = False, ) -> TickReport: """One full tick. Pause sentinel wins over everything: a paused loop @@ -252,7 +307,7 @@ def tick( if (root / PAUSE_SENTINEL).exists(): log.info("pause sentinel present; tick is a no-op") return TickReport(paused=True) - return sweep(root, compute, dispatcher, now, dry_run=dry_run) + return sweep(root, compute, dispatcher, now, grace_s, lease_ttl_s, dry_run=dry_run) @dataclass @@ -280,7 +335,15 @@ def main() -> int: args.root.mkdir(parents=True, exist_ok=True) # dry_run until the phase-5 dispatcher exists: the live loop must not # mutate run state it cannot follow through on. - report = tick(args.root, SlurmCompute(), LoggingDispatcher(), now=time.time(), dry_run=True) + report = tick( + args.root, + SlurmCompute(), + LoggingDispatcher(), + now=time.time(), + grace_s=args.grace_s, + lease_ttl_s=args.lease_ttl_s, + dry_run=True, + ) log.info( "tick done: paused=%s swept=%d woken=%d deferred=%d reaped=%d stuck=%d", report.paused, diff --git a/tests/test_tick.py b/tests/test_tick.py index e5f476e5..dcbc5382 100644 --- a/tests/test_tick.py +++ b/tests/test_tick.py @@ -103,13 +103,13 @@ def test_first_terminal_sighting_starts_the_grace_clock_not_a_wake(tmp_path: Pat 24h experiment must not be woken by the backup the instant it completes (the afterany job owns the fresh case).""" waiting_run(tmp_path, terminal_seen=0.0) - report, dispatcher = run_tick(tmp_path, FakeSlurm(states={"100": "COMPLETED"})) + _report, dispatcher = run_tick(tmp_path, FakeSlurm(states={"100": "COMPLETED"})) assert dispatcher.dispatched == [] assert load_record(tmp_path, "r1").terminal_seen == NOW # attempts untouched by the sighting assert load_record(tmp_path, "r1").wake_attempts == 0 # next tick, grace elapsed → wake - report2, dispatcher2 = run_tick( + report2, _dispatcher2 = run_tick( tmp_path, FakeSlurm(states={"100": "COMPLETED"}), now=NOW + GRACE + 1 ) assert report2.woken == (("r1", "COMPLETED"),) @@ -162,20 +162,20 @@ def test_query_failure_defers_never_concludes(tmp_path: Path) -> None: def test_gone_before_deadline_waits_for_sacct_lag(tmp_path: Path) -> None: waiting_run(tmp_path) # deadline far in the future - report, dispatcher = run_tick(tmp_path, FakeSlurm(states={})) # sacct empty + _report, dispatcher = run_tick(tmp_path, FakeSlurm(states={})) # sacct empty assert dispatcher.dispatched == [] def test_gone_past_deadline_wakes_with_vanished(tmp_path: Path) -> None: waiting_run(tmp_path, deadline=NOW - 1) - report, dispatcher = run_tick(tmp_path, FakeSlurm(states={})) + report, _dispatcher = run_tick(tmp_path, FakeSlurm(states={})) assert report.woken == (("r1", "vanished"),) def test_pending_past_deadline_cancels_then_wakes(tmp_path: Path) -> None: waiting_run(tmp_path, deadline=NOW - 1) slurm = FakeSlurm(states={"100": "PENDING"}) - report, dispatcher = run_tick(tmp_path, slurm) + report, _dispatcher = run_tick(tmp_path, slurm) assert slurm.cancelled == ["100"] assert report.woken == (("r1", "unschedulable"),) @@ -183,14 +183,14 @@ def test_pending_past_deadline_cancels_then_wakes(tmp_path: Path) -> None: def test_pending_before_deadline_is_left_alone(tmp_path: Path) -> None: waiting_run(tmp_path) slurm = FakeSlurm(states={"100": "PENDING"}) - report, dispatcher = run_tick(tmp_path, slurm) + _report, dispatcher = run_tick(tmp_path, slurm) assert slurm.cancelled == [] assert dispatcher.dispatched == [] def test_running_experiment_is_left_alone(tmp_path: Path) -> None: waiting_run(tmp_path) - report, dispatcher = run_tick(tmp_path, FakeSlurm(states={"100": "RUNNING"})) + _report, dispatcher = run_tick(tmp_path, FakeSlurm(states={"100": "RUNNING"})) assert dispatcher.dispatched == [] @@ -209,7 +209,9 @@ def test_live_lease_defers_even_the_stuck_verdict(tmp_path: Path) -> None: LIVE lease must not be truncated to stuck underneath it.""" waiting_run(tmp_path, wake_attempts=3) acquire_lease(tmp_path, "r1", "wake-job:55", "55", now=NOW - 60) - report, dispatcher = run_tick(tmp_path, FakeSlurm(states={"100": "COMPLETED", "55": "RUNNING"})) + report, _dispatcher = run_tick( + tmp_path, FakeSlurm(states={"100": "COMPLETED", "55": "RUNNING"}) + ) assert report.stuck == () assert load_record(tmp_path, "r1").state == WAITING @@ -269,7 +271,7 @@ def test_legacy_zero_deadline_still_wakes_gone_runs(tmp_path: Path) -> None: (directory / "state.json").write_text(_json.dumps(record)) slurm = FakeSlurm(states={"100": "PENDING"}) - report2, _ = run_tick(tmp_path, slurm) + _report2, _ = run_tick(tmp_path, slurm) assert slurm.cancelled == [] # healthy pending job never cancelled @@ -325,3 +327,87 @@ def test_heartbeat_written_every_tick(tmp_path: Path) -> None: beat = json.loads((tmp_path / "heartbeat.json").read_text()) assert beat["ts"] == NOW assert "host" in beat + + +def test_one_bad_record_does_not_blind_the_sweep(tmp_path: Path) -> None: + """Per-record isolation: a record that makes processing raise must not + stop the remaining runs from being swept.""" + import json as _json + + # legacy zero-deadline record whose experiment is TERMINAL: the sighting + # save would raise without the repair; either way the sweep must go on + directory = tmp_path / "runs" / "a-bad" + directory.mkdir(parents=True) + record = dict( + run_id="a-bad", + target="o/r", + task_title="t", + state=WAITING, + agent_id="a", + experiment_job_id="200", + wake_job_id="", + resume_session_id="", + wake_attempts=0, + deadline=0.0, + terminal_seen=0.0, + ending="", + ending_note="", + created=NOW - 5000, + updated=NOW - 5000, + ) + (directory / "state.json").write_text(_json.dumps(record)) + waiting_run(tmp_path, run_id="z-good") + slurm = FakeSlurm(states={"200": "COMPLETED", "100": "COMPLETED"}) + report, _dispatcher = run_tick(tmp_path, slurm) + assert ("z-good", "COMPLETED") in report.woken # the good run was served + + +def test_legacy_terminal_record_gets_deadline_repaired_on_sighting(tmp_path: Path) -> None: + import json as _json + + directory = tmp_path / "runs" / "legacy2" + directory.mkdir(parents=True) + record = dict( + run_id="legacy2", + target="o/r", + task_title="t", + state=WAITING, + agent_id="a", + experiment_job_id="300", + wake_job_id="", + resume_session_id="", + wake_attempts=0, + deadline=0.0, + terminal_seen=0.0, + ending="", + ending_note="", + created=NOW - 5000, + updated=NOW - 5000, + ) + (directory / "state.json").write_text(_json.dumps(record)) + run_tick(tmp_path, FakeSlurm(states={"300": "FAILED"})) + repaired = load_record(tmp_path, "legacy2") + assert repaired.terminal_seen == NOW + assert repaired.deadline > 0 + + +def test_cli_grace_flag_reaches_the_sweep(tmp_path: Path, monkeypatch) -> None: + """--grace-s must actually change sweep behavior (was parsed-but-ignored).""" + import sys + + import autoresearch.tick as tick_mod + + waiting_run(tmp_path, terminal_seen=NOW - 5) # 5s since sighting + captured: dict = {} + + real_tick = tick_mod.tick + + def spy(root, compute, dispatcher, now, grace_s=0, lease_ttl_s=0, dry_run=False): + captured["grace_s"] = grace_s + return real_tick(root, compute, dispatcher, now, grace_s, lease_ttl_s, dry_run) + + monkeypatch.setattr(tick_mod, "tick", spy) + monkeypatch.setattr(tick_mod, "SlurmCompute", lambda: FakeSlurm(states={}).compute()) + monkeypatch.setattr(sys, "argv", ["tick", "--root", str(tmp_path), "--grace-s", "1"]) + assert tick_mod.main() == 0 + assert captured["grace_s"] == 1.0 From d35e0ecb424fb02a7ca402319e1cc3259846db28 Mon Sep 17 00:00:00 2001 From: Mengye Ren Date: Thu, 6 Aug 2026 10:18:39 -0400 Subject: [PATCH 4/4] Round two: JSON type guards, CAS reap with restore, squeue-failure guard, unique tmp names, attempts contract Co-Authored-By: Claude Fable 5 --- scripts/tick_chain.sbatch | 13 +++++++++-- src/autoresearch/runstate.py | 45 +++++++++++++++++++++++++++++------- src/autoresearch/tick.py | 13 ++++++++--- tests/test_runstate.py | 38 ++++++++++++++++++++++++++++-- 4 files changed, 94 insertions(+), 15 deletions(-) diff --git a/scripts/tick_chain.sbatch b/scripts/tick_chain.sbatch index 78a814f4..f8455214 100755 --- a/scripts/tick_chain.sbatch +++ b/scripts/tick_chain.sbatch @@ -36,8 +36,17 @@ done # --- 1. keep two successors queued (singleton serializes same-name jobs) --- # Successors land on slots AFTER the ones already-queued jobs occupy — # otherwise every top-up doubles the tick rate on the next slot. -pending=$(squeue -u "$USER" --name="$JOB_NAME" -h -t PENDING 2>/dev/null | wc -l) -need=$((2 - pending)) +if squeue_out=$(squeue -u "$USER" --name="$JOB_NAME" -h -t PENDING 2>/dev/null); then + pending=$(printf '%s' "$squeue_out" | grep -c '[0-9]') + need=$((2 - pending)) +else + # A failed squeue must not read as "zero pending" — that would stack + # extra successors onto occupied slots every outage. Skip the top-up; + # the (very likely still-queued) successors keep the chain alive and + # the next tick retries. + pending=0 + need=0 +fi epoch_now=$(date +%s) cadence_s=$((CADENCE_MIN * 60)) next_slot=$(( (epoch_now / cadence_s + 1) * cadence_s )) diff --git a/src/autoresearch/runstate.py b/src/autoresearch/runstate.py index 4058efc7..9802bc33 100644 --- a/src/autoresearch/runstate.py +++ b/src/autoresearch/runstate.py @@ -96,13 +96,17 @@ def save_record(root: Path, record: RunRecord, now: float) -> None: directory = run_dir(root, record.run_id) directory.mkdir(parents=True, exist_ok=True) stamped = replace(record, updated=now, created=record.created or now) - tmp = directory / f".{RECORD_NAME}.tmp" + # unique tmp name: two concurrent writers must not interleave into the + # same tmp file before the atomic replace + tmp = directory / f".{RECORD_NAME}.{os.getpid()}.tmp" tmp.write_text(json.dumps(asdict(stamped), indent=2, sort_keys=True)) os.replace(tmp, directory / RECORD_NAME) def load_record(root: Path, run_id: str) -> RunRecord: raw = json.loads((run_dir(root, run_id) / RECORD_NAME).read_text()) + if not isinstance(raw, dict): + raise ValueError(f"record is not a JSON object: {type(raw).__name__}") # Ignore unknown keys: after a bad-merge revert, older code must still be # able to read records written by newer code — a "corrupt" verdict here # would blind the sweep to the whole run. @@ -152,6 +156,8 @@ def read_lease(root: Path, run_id: str) -> Lease | None: path = run_dir(root, run_id) / LEASE_NAME try: raw = json.loads(path.read_text()) + if not isinstance(raw, dict): + raise ValueError("lease is not a JSON object") known = {k: v for k, v in raw.items() if k in Lease.__dataclass_fields__} return Lease(**known) except FileNotFoundError: @@ -174,7 +180,7 @@ def update_lease_holder( """Hand a HELD lease to a new holder (e.g. tick → the wake job it just submitted). Atomic replace; only valid while the caller holds the lease.""" directory = run_dir(root, run_id) - tmp = directory / f".{LEASE_NAME}.tmp" + tmp = directory / f".{LEASE_NAME}.{os.getpid()}.tmp" tmp.write_text(json.dumps(asdict(Lease(holder, holder_job_id, now)))) os.replace(tmp, directory / LEASE_NAME) @@ -185,13 +191,19 @@ def release_lease(root: Path, run_id: str) -> None: (run_dir(root, run_id) / LEASE_NAME).unlink() -def reap_lease(root: Path, run_id: str, reaper: str) -> bool: - """Remove a stale lease you do NOT hold. True if THIS caller reaped it. +def reap_lease(root: Path, run_id: str, reaper: str, expected: Lease) -> bool: + """Remove the stale lease you observed (and do NOT hold). True if THIS + caller reaped exactly that lease. - Atomic rename to a unique tombstone: with two concurrent reapers exactly - one rename succeeds, so exactly one proceeds to redeliver — a blind - unlink here could delete a lease a faster reaper already handed to a new - wake job, re-opening the double-delivery hole. + Rename-to-tombstone makes removal atomic (one of N concurrent reapers + wins the rename); the identity check afterwards makes it a compare-and- + swap: if the file we renamed is NOT the stale lease we observed — a + faster reaper already reaped and a fresh lease was written — we restore + it via link (which cannot clobber a newer lease) and stand down. The + remaining hole needs a 3-party race inside this microsecond window and + the singleton tick serialization makes that effectively unreachable; + if it ever fires, the symptom is one duplicate wake, which the resumed + session tolerates (sequential re-resume is safe). """ directory = run_dir(root, run_id) tombstone = directory / f".{LEASE_NAME}.reaped.{reaper}" @@ -199,6 +211,23 @@ def reap_lease(root: Path, run_id: str, reaper: str) -> bool: os.rename(directory / LEASE_NAME, tombstone) except FileNotFoundError: return False + try: + raw = json.loads(tombstone.read_text()) + got: Lease | None = ( + Lease(**{k: v for k, v in raw.items() if k in Lease.__dataclass_fields__}) + if isinstance(raw, dict) + else None + ) + except (OSError, ValueError, TypeError, KeyError): + got = None # unreadable — the corrupt lease we came to reap + if got is not None and (got.holder != expected.holder or got.acquired != expected.acquired): + # we grabbed someone's FRESH lease; put it back without clobbering + try: + os.link(tombstone, directory / LEASE_NAME) + except FileExistsError: + log.warning("lease race on %s: fresh lease displaced during reap", run_id) + tombstone.unlink(missing_ok=True) + return False tombstone.unlink(missing_ok=True) return True diff --git a/src/autoresearch/tick.py b/src/autoresearch/tick.py index 8f2dc5f8..b30dd355 100644 --- a/src/autoresearch/tick.py +++ b/src/autoresearch/tick.py @@ -55,7 +55,12 @@ class WakeDispatcher(Protocol): Returns "" when delivery completed synchronously (the caller releases the lease), or the Slurm job id of an asynchronous wake job that now owns the - lease (released by that job on completion; reaped by TTL if it dies).""" + lease (released by that job on completion; reaped by TTL if it dies). + + Contract for real (phase-5) dispatchers: a wake that RESULTS IN PROGRESS + must either move the run out of `waiting` or reset `wake_attempts` — + the counter means "wakes since the run last made progress", and layer 5 + ends the run as stuck when it reaches MAX_WAKE_ATTEMPTS.""" def dispatch(self, record: RunRecord, reason: str) -> str: ... @@ -220,7 +225,7 @@ def _sweep_one( if dry_run: reaped.append(record.run_id) return - if not reap_lease(root, record.run_id, reaper=f"{os.getpid()}-{now}"): + if not reap_lease(root, record.run_id, reaper=f"{os.getpid()}-{now}", expected=lease): return # a concurrent tick reaped it first; it owns redelivery reaped.append(record.run_id) @@ -231,7 +236,9 @@ def _sweep_one( record, state=ENDED, ending=STUCK, - ending_note=f"{record.wake_attempts} wake attempts failed", + ending_note=( + f"{record.wake_attempts} wake attempts without the run leaving 'waiting'" + ), ) save_record(root, ended, now) stuck.append(record.run_id) diff --git a/tests/test_runstate.py b/tests/test_runstate.py index a2287884..20bdeb45 100644 --- a/tests/test_runstate.py +++ b/tests/test_runstate.py @@ -110,11 +110,45 @@ def test_reap_lease_exactly_one_reaper_wins(tmp_path: Path) -> None: from autoresearch.runstate import reap_lease acquire_lease(tmp_path, "r1", "dead", "", now=1.0) - assert reap_lease(tmp_path, "r1", reaper="a") - assert not reap_lease(tmp_path, "r1", reaper="b") # already gone + stale = read_lease(tmp_path, "r1") + assert stale is not None + assert reap_lease(tmp_path, "r1", reaper="a", expected=stale) + assert not reap_lease(tmp_path, "r1", reaper="b", expected=stale) # gone assert acquire_lease(tmp_path, "r1", "next", "", now=2.0) +def test_reap_lease_refuses_a_fresh_lease_it_did_not_observe(tmp_path: Path) -> None: + """The CAS: reaper B saw the stale lease, but reaper A already reaped it + and a fresh lease was written — B must restore, not steal.""" + from autoresearch.runstate import reap_lease + + acquire_lease(tmp_path, "r1", "dead", "", now=1.0) + stale = read_lease(tmp_path, "r1") + assert stale is not None + # A's reap + a new wake's fresh lease happen "before" B acts: + release_lease(tmp_path, "r1") + acquire_lease(tmp_path, "r1", "wake-job:777", "777", now=50.0) + assert not reap_lease(tmp_path, "r1", reaper="b", expected=stale) + fresh = read_lease(tmp_path, "r1") + assert fresh is not None and fresh.holder == "wake-job:777" # restored + + +def test_non_object_json_record_is_skipped_not_fatal(tmp_path: Path) -> None: + """Valid JSON that is not an object (null, list) must be 'corrupt', + never an exception that blinds the whole sweep.""" + + save_record(tmp_path, make_record(run_id="good"), now=1.0) + bad = run_dir(tmp_path, "nulled") + bad.mkdir(parents=True) + (bad / "state.json").write_text("null") + assert [r.run_id for r in list_runs(tmp_path)] == ["good"] + + lease_dir = run_dir(tmp_path, "good") + (lease_dir / "lease.json").write_text("[1, 2]") + lease = read_lease(tmp_path, "good") + assert lease is not None and lease.holder == "unreadable" # mtime fallback + + def test_load_record_ignores_unknown_keys(tmp_path: Path) -> None: """After a bad-merge revert, old code must still read new-code records.""" import json