from __future__ import annotations import json import os from pathlib import Path import subprocess import sys from datetime import datetime, timedelta, timezone from recoil.pipeline.tools.autonomy import lease REPO_ROOT = Path(__file__).resolve().parents[4] def _iso(value: datetime) -> str: return value.replace(microsecond=0).isoformat().replace("+00:00", "Z") def _dead_pid() -> int: for candidate in range(999999, 999000, -1): try: os.kill(candidate, 0) except ProcessLookupError: return candidate except OSError: continue raise AssertionError("could not find an unused pid for stale lease test") def _isolate(monkeypatch, tmp_path: Path) -> Path: state_dir = tmp_path / "state" monkeypatch.setattr(lease, "STATE_DIR", state_dir) monkeypatch.setattr(lease, "LEASE_PATH", state_dir / "autonomy.lease.json") monkeypatch.setattr(lease, "LEASE_LOCK", state_dir / "autonomy.lease.lock") return state_dir def _write_existing_lease( *, run_id: str, pid: int, expires_at: datetime, tmux_session: str | None = None, write_lock: bool = True, ) -> None: now = datetime.now(timezone.utc) lease._atomic_write_json( lease.LEASE_PATH, { "schema_version": 1, "mode": "tick", "holder": f"test:{pid}:tick", "run_id": run_id, "issue_id": None, "host": "test", "pid": pid, "tmux_session": tmux_session, "worktree_path": None, "acquired_at": _iso(now - timedelta(seconds=120)), "heartbeat_at": _iso(now - timedelta(seconds=120)), "expires_at": _iso(expires_at), "night_id": "2026-06-06-evening", }, ) if write_lock: lease.LEASE_LOCK.write_text(f"{pid}\n", encoding="utf-8") def test_acquire_read_roundtrip_and_second_acquire_busy(monkeypatch, tmp_path): _isolate(monkeypatch, tmp_path) record = lease.acquire( "tick", run_id="run-a", issue_id="REC-1", ttl=60, host="host-a", pid=os.getpid(), worktree_path="/tmp/wt", ) assert record is not None assert record["schema_version"] == 1 assert record["mode"] == "tick" assert record["run_id"] == "run-a" assert record["issue_id"] == "REC-1" assert record["host"] == "host-a" assert record["pid"] == os.getpid() assert record["worktree_path"] == "/tmp/wt" assert lease.read()["run_id"] == "run-a" assert lease.is_held_fresh() assert lease.acquire("tick", run_id="run-b", ttl=60, pid=os.getpid()) is None assert lease.release("run-a") is True assert lease.read() is None assert not lease.LEASE_LOCK.exists() def test_heartbeat_extends_expiry(monkeypatch, tmp_path): _isolate(monkeypatch, tmp_path) assert lease.acquire("build", run_id="run-heartbeat", ttl=5, pid=os.getpid()) before = lease.read()["expires_at"] assert lease.heartbeat("run-heartbeat", ttl=120) is True after = lease.read()["expires_at"] assert after > before assert lease.read()["heartbeat_at"] >= lease.read()["acquired_at"] assert lease.release("run-heartbeat") is True def test_release_frees_lease(monkeypatch, tmp_path): _isolate(monkeypatch, tmp_path) assert lease.acquire("tick", run_id="run-release", ttl=60, pid=os.getpid()) assert lease.release("other-run") is False assert lease.read() is not None assert lease.release("run-release") is True assert lease.release("run-release") is True assert lease.read() is None def test_stale_dead_pid_reclaim_succeeds(monkeypatch, tmp_path): _isolate(monkeypatch, tmp_path) dead_pid = _dead_pid() _write_existing_lease( run_id="old-run", pid=dead_pid, expires_at=datetime.now(timezone.utc) - timedelta(seconds=60), write_lock=False, ) reclaimed = lease.acquire("tick", run_id="new-run", ttl=60, pid=os.getpid()) assert reclaimed is not None assert reclaimed["run_id"] == "new-run" assert lease.read()["run_id"] == "new-run" assert lease.release("new-run") is True def test_stale_live_pid_reclaim_refused(monkeypatch, tmp_path): _isolate(monkeypatch, tmp_path) live_pid = os.getpid() _write_existing_lease( run_id="old-live-run", pid=live_pid, expires_at=datetime.now(timezone.utc) - timedelta(seconds=60), write_lock=False, ) assert lease.acquire("tick", run_id="new-run", ttl=60, pid=live_pid) is None assert lease.read()["run_id"] == "old-live-run" lease.LEASE_PATH.unlink() lease.LEASE_LOCK.unlink(missing_ok=True) def test_convert_tick_to_build_preserves_run_id(monkeypatch, tmp_path): _isolate(monkeypatch, tmp_path) assert lease.acquire("tick", run_id="run-convert", ttl=60, pid=os.getpid()) assert lease.convert( "run-convert", new_mode="build", new_ttl=120, pid=12345, tmux_session="autonomy-build", ) converted = lease.read() assert converted["run_id"] == "run-convert" assert converted["mode"] == "build" assert converted["pid"] == 12345 assert converted["tmux_session"] == "autonomy-build" assert converted["holder"].endswith(":build:autonomy-build") assert lease.release("run-convert") is True def _cli_env(tmp_path: Path) -> dict[str, str]: home = tmp_path / "home" home.mkdir() env = os.environ.copy() env["HOME"] = str(home) env["PYTHONPATH"] = str(REPO_ROOT) return env def _run_cli(env: dict[str, str], *args: str) -> subprocess.CompletedProcess[str]: return subprocess.run( [sys.executable, "-m", "recoil.pipeline.tools.autonomy.lease", *args], cwd=REPO_ROOT, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, ) def _run_python(env: dict[str, str], code: str) -> subprocess.CompletedProcess[str]: return subprocess.run( [sys.executable, "-c", code], cwd=REPO_ROOT, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, ) def test_cli_heartbeat_extends_expiry_and_bad_holder_fails(tmp_path): env = _cli_env(tmp_path) result = _run_python( env, "from recoil.pipeline.tools.autonomy import lease; " "raise SystemExit(0 if lease.acquire('build', run_id='cli-run', ttl=5) else 1)", ) assert result.returncode == 0, result.stderr lease_path = Path(env["HOME"]) / ".local/state/studio-autonomy/autonomy.lease.json" before = json.loads(lease_path.read_text(encoding="utf-8"))["expires_at"] result = _run_cli(env, "heartbeat", "--run-id", "cli-run", "--ttl", "120") assert result.returncode == 0, result.stderr after = json.loads(lease_path.read_text(encoding="utf-8"))["expires_at"] assert after > before result = _run_cli(env, "heartbeat", "--run-id", "not-holder", "--ttl", "120") assert result.returncode != 0 result = _run_cli(env, "release", "--run-id", "cli-run") assert result.returncode == 0, result.stderr assert not lease_path.exists() def test_cli_release_frees_lease(tmp_path): env = _cli_env(tmp_path) result = _run_python( env, "from recoil.pipeline.tools.autonomy import lease; " "raise SystemExit(0 if lease.acquire('tick', run_id='cli-release', ttl=60) else 1)", ) assert result.returncode == 0, result.stderr result = _run_cli(env, "release", "--run-id", "cli-release") lease_path = Path(env["HOME"]) / ".local/state/studio-autonomy/autonomy.lease.json" assert result.returncode == 0, result.stderr assert not lease_path.exists()