diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 25a965c..57a2e03 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -8,20 +8,21 @@ jobs: # NOTE: the python job runs on pull_request only (not push) because the # windows-latest runner intermittently flakes on # test_stop_after_restart_kills_runner_and_workload (pid teardown timing after - # stop_sync). It also only runs on Windows for now: the ubuntu python matrix - # failed on `print('x')`-style jobs being marked 'failed' — the runner spawns - # workloads via `subprocess.Popen(shell=True)` with a command string built by - # `subprocess.list2cmdline` (a Windows-only escaper), which produces invalid - # POSIX shell quoting (bash: "syntax error near unexpected token `('"). The - # fix is to make the runner POSIX-safe (shlex) — not a test-timing issue. - # Re-enable ubuntu in the matrix once that's fixed. + # stop_sync). Linux and macOS are covered too: the suite builds workload + # command strings with tests/shellcmd.py (shlex.join on POSIX, list2cmdline on + # Windows) so the platform shell executes them correctly everywhere. macOS is + # limited to one Python version to limit runner cost; Windows and Linux cover + # both. python: if: github.event_name == 'pull_request' strategy: fail-fast: false matrix: - os: [windows-latest] + os: [windows-latest, ubuntu-latest, macos-latest] python: ['3.11', '3.12'] + exclude: + - os: macos-latest + python: '3.11' runs-on: ${{ matrix.os }} steps: - uses: actions/checkout@v4 @@ -38,15 +39,15 @@ jobs: - name: Smoke installed wheel shell: bash run: | - wheel=$(find dist -maxdepth 1 -name '*.whl' -print -quit) - test -n "$wheel" - uv run --isolated --no-project --with "$wheel" python -c "import vanth; print(vanth.__file__)" + wheels=(dist/*.whl) + test -e "${wheels[0]}" + uv run --isolated --no-project --with "${wheels[0]}" python -c "import vanth; print(vanth.__file__)" - name: Smoke bundled monitor console script shell: bash run: | - wheel=$(find dist -maxdepth 1 -name '*.whl' -print -quit) - test -n "$wheel" - uv run --isolated --no-project --with "$wheel" vanth-monitor --version + wheels=(dist/*.whl) + test -e "${wheels[0]}" + uv run --isolated --no-project --with "${wheels[0]}" vanth-monitor --version go: strategy: diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a942d0..ef09f68 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,81 @@ All notable changes to Vanth are documented here. +## 1.9.1 - 2026-09-11 + +### POSIX Python CI + portability fixes + +Enables the Python test matrix on `ubuntu-latest` and `macos-latest` alongside +Windows (`.github/workflows/ci.yml`). Getting the suite green there surfaced +the fixes below. + +- **Test command builders are POSIX-correct.** A shared `tests/shellcmd.py` + quotes workload command strings with `shlex.join` on POSIX and + `subprocess.list2cmdline` on Windows, so `python -c "print('x')"`-style jobs + run under `sh`/`bash` instead of failing with a shell syntax error. The dev + scripts follow the same rule. +- **`job_wait` returns the earliest matching signal.** When several signals + match (terminal event, a `metric_ge` threshold crossed, or + `return_progress`), the earliest by event sequence wins instead of always + preferring the terminal event. A metric/progress signal that precedes + completion is returned first; a terminal event still wins once nothing + earlier is pending. Metric candidates honour `since_event_id` (a threshold + already returned at or before the cursor is not returned again, so a caller + that advances the cursor still reaches the terminal event) and every + threshold is considered, not just the first in dictionary order. +- **Codex Desktop pipe hardening.** A peer that closes the connection is now + reported consistently ("closed the connection") whether detected on read or + write, and `close()` shuts a socket down before closing it so a reader + blocked in `recv()` wakes promptly on POSIX (previously it could add ~2s to a + timed-out call). +- **Failure-streak ordering and counting.** The `on_failure` policy persists + the updated `failure_streak` before emitting the `failure_threshold` event, + and counts every failed execution in the interval between two event-sequence + watermarks (bounded at the latest failure read, so a concurrent failure is + neither skipped nor double-counted; a fast restart with backoff 0 no longer + undercounts; state written by 1.9.0 is migrated, so an upgrade does not + recount old failures). The reaction is marked complete only after it + succeeds, so a daemon crash or action error retries it instead of dropping + it — at-least-once delivery: a crash between the side effect and the marker + save can repeat it, which is harmless for the built-in actions (disable + excludes the job from the scan, run_job refuses a busy target, alerts are + advisory). +- **macOS artifact materialization.** Directory materialization uses the + dev/inode-checked plain-path fallback on macOS instead of `/dev/fd`, which is + unreliable for creating nested entries under a directory fd; the destination + parent and staging descriptors are re-verified immediately before publication + and the operation fails closed if the parent was persistently swapped + mid-write. A transient swap restored before the check, or a staging-directory + replacement under an unchanged parent, is still not caught — closing that + requires descriptor-relative tree construction on macOS. +- **Daemon startup.** The HTTP server no longer calls `socket.getfqdn` at + bind time — a reverse-DNS lookup that can block for seconds (or hang) on + locked-down networks and stall startup past client timeouts. +- **Launch claims.** A new launch claim clears the previous run's + `worker_pid`, so stale-claim recovery can no longer skip an abandoned claim + whose old runner pid is still momentarily visible. +- **Idle reaper.** A healthy Desktop wake relay whose activity cadence is + coarser than the watchdog's sampling interval is no longer idle-reaped (the + freshness window is the idle threshold, not the sampling interval), and the + effective idle timeout is measured from the last activity rather than + restarting a second window when the freshness window expires. +- **Event write resilience.** Structured-event writes retry further under lock + contention instead of being dropped (Windows CI lost reader events under a + concurrent job burst). +- **Orphaned-MCP reaping is safer.** Detection matches the actual Vanth MCP + entrypoint only — the `vanth` console script with no CLI subcommand, or + ` -m vanth.server` (with interpreter options tolerated) — and rejects + lookalikes such as `python unrelated.py -m vanth.server`, `bash -lc 'python -m + vanth.server'`, and CLI invocations like `vanth logs --follow`, so + `vanth doctor --reap-orphans` can no longer terminate unrelated processes. + Windows enumeration now uses `Get-CimInstance` with the command line (the old + WMIC CSV parse misread the alphabetically-ordered columns and could not + establish identity). Orphan findings are an advisory warning and no longer + flip `vanth doctor`'s exit code. + +Full suite: 783 passed, 6 skipped on Windows; 787 passed, 2 skipped on Linux +(Python 3.12); `go test ./...` green. + ## 1.9.0 - 2026-09-10 ### Operational hardening (from the 2026-09 roadmap research) diff --git a/pyproject.toml b/pyproject.toml index 6909ccc..a4f24bd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "vanth" -version = "1.9.0" +version = "1.9.1" description = "Event-driven background jobs for agents" readme = "README.md" requires-python = ">=3.11" @@ -72,3 +72,6 @@ path = "build-hooks/bundle_monitor.py" [tool.pytest.ini_options] testpaths = ["tests"] +# Makes the tests/shellcmd.py helper importable from every test module, +# including the tests/artifacts and tests/remote subdirectories. +pythonpath = ["tests"] diff --git a/scripts/demo_jobs.py b/scripts/demo_jobs.py index 272646d..9bf9baf 100644 --- a/scripts/demo_jobs.py +++ b/scripts/demo_jobs.py @@ -11,6 +11,7 @@ import http.client import json import os +import shlex import socket import subprocess import sys @@ -24,7 +25,10 @@ def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + argv = [sys.executable, "-c", code] + if sys.platform == "win32": + return subprocess.list2cmdline(argv) + return shlex.join(argv) def request(method, path, body=None, token=None): diff --git a/scripts/make_monitor_fixture.py b/scripts/make_monitor_fixture.py index 420d18b..b787dc1 100644 --- a/scripts/make_monitor_fixture.py +++ b/scripts/make_monitor_fixture.py @@ -2,6 +2,7 @@ import asyncio import json import os +import shlex import subprocess import sys import tempfile @@ -11,7 +12,10 @@ def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + argv = [sys.executable, "-c", code] + if sys.platform == "win32": + return subprocess.list2cmdline(argv) + return shlex.join(argv) def main(): diff --git a/src/vanth/artifacts/operations.py b/src/vanth/artifacts/operations.py index ba3cc23..9c2ce44 100644 --- a/src/vanth/artifacts/operations.py +++ b/src/vanth/artifacts/operations.py @@ -1162,21 +1162,19 @@ def _materialize_dir( staging_name = f".{dest.name}.materializing-{uuid.uuid4().hex}" parent_fd = None + path_anchored = False if os.name == "nt": staging = dest.parent / staging_name staging.mkdir() else: parent_fd = self._open_parent_fd(dest.parent) os.mkdir(staging_name, dir_fd=parent_fd) - # Keep construction DESCRIPTOR-RELATIVE wherever the OS exposes a - # descriptor path: /proc/self/fd on Linux, /dev/fd on macOS - # (rc17 review F7). Otherwise fall back to the plain path with a - # dev/inode cross-check of the opened parent. - proc_root = None - if sys.platform.startswith("linux"): - proc_root = "/proc/self/fd" - elif sys.platform == "darwin": - proc_root = "/dev/fd" + # Keep construction DESCRIPTOR-RELATIVE where the OS exposes a + # descriptor path: /proc/self/fd on Linux (rc17 review F7). On macOS + # /dev/fd is not reliable for creating nested entries under a + # directory fd, so use the plain path with a dev/inode cross-check of + # the opened parent instead. + proc_root = "/proc/self/fd" if sys.platform.startswith("linux") else None fd_dir = Path(proc_root, str(parent_fd)) if proc_root else None if fd_dir is not None and fd_dir.exists(): staging = fd_dir / staging_name @@ -1190,9 +1188,39 @@ def _materialize_dir( f"destination parent changed during materialization; refusing: {dest.parent}" ) staging = dest.parent / staging_name + path_anchored = True try: try: self._build_tree_into_staging(staging, entries, heartbeat) + if path_anchored and parent_fd is not None: + # macOS fallback builds through the plain path (no /proc fd + # symlink). Re-verify the destination parent AND the staging + # directory still resolve to the descriptors we opened, so a + # PERSISTENT ancestor swap cannot publish a redirected or + # empty tree — fail closed instead. + # ponytail: residual race remains — a swap restored before + # this check, or a staging-name replacement under an + # unchanged parent, is not caught; closing that needs + # descriptor-relative tree construction on macOS. + try: + path_parent = os.stat(dest.parent) + fd_parent = os.fstat(parent_fd) + path_staging = os.stat(staging) + fd_staging = os.stat(staging_name, dir_fd=parent_fd) + except OSError as exc: + raise ValueError( + f"staging path changed during materialization; refusing: {dest} ({exc})" + ) from None + if (path_parent.st_dev, path_parent.st_ino) != ( + fd_parent.st_dev, + fd_parent.st_ino, + ) or (path_staging.st_dev, path_staging.st_ino) != ( + fd_staging.st_dev, + fd_staging.st_ino, + ): + raise ValueError( + f"destination parent changed during materialization; refusing: {dest}" + ) # Atomic swap into place: rename fails rather than merges if # a destination raced into existence. if parent_fd is None: diff --git a/src/vanth/codex_pipe.py b/src/vanth/codex_pipe.py index b22ad2b..ee4a407 100644 --- a/src/vanth/codex_pipe.py +++ b/src/vanth/codex_pipe.py @@ -21,6 +21,7 @@ import json import os import queue +import socket import struct import subprocess import sys @@ -136,6 +137,11 @@ def _write_all(handle, data: bytes) -> None: else: written = handle.write(view) except OSError as exc: + if isinstance(exc, (BrokenPipeError, ConnectionResetError)): + # A peer that has gone away surfaces as EPIPE/ECONNRESET on + # write on POSIX and ECONNRESET on Windows; report it the same + # way as a read-side close so callers see one consistent error. + raise CodexPipeError("Codex Desktop app-tools host closed the connection") from exc raise CodexPipeError("lost connection to Codex Desktop app-tools host (write)") from exc if written is None or written <= 0: raise CodexPipeError("Codex Desktop app-tools host closed the connection (write)") @@ -178,9 +184,21 @@ def __init__(self, pipe_path: str, timeout_seconds: float = DEFAULT_TIMEOUT_SECO self._request_id = 0 def close(self) -> None: + handle = getattr(self, "handle", None) + if handle is None: + return + # Shutting down a socket before closing it reliably wakes a recv() + # blocked on another thread on POSIX; closing alone does not. Named + # pipe/file handles have no shutdown() and are unaffected. + shutdown = getattr(handle, "shutdown", None) + if shutdown is not None: + try: + shutdown(socket.SHUT_RDWR) + except OSError: + pass try: - if hasattr(self.handle, "close"): - self.handle.close() + if hasattr(handle, "close"): + handle.close() except Exception: pass diff --git a/src/vanth/daemon.py b/src/vanth/daemon.py index ae69193..529ff58 100644 --- a/src/vanth/daemon.py +++ b/src/vanth/daemon.py @@ -10,6 +10,7 @@ import os import secrets import signal +import socketserver import threading import time import urllib.parse @@ -314,6 +315,16 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: self._active_condition = threading.Condition() self._active_requests = 0 + def server_bind(self) -> None: + # Skip HTTPServer.server_bind's ``socket.getfqdn(host)``: that reverse + # DNS lookup can block for seconds (or hang) on locked-down networks, + # delaying daemon startup past every client timeout. ``server_name`` is + # only cosmetic metadata we do not depend on. + socketserver.TCPServer.server_bind(self) + host, port = self.server_address[:2] + self.server_name = host + self.server_port = port + def _request_finished(self) -> None: with self._active_condition: self._active_requests -= 1 diff --git a/src/vanth/process_watch.py b/src/vanth/process_watch.py index a1a42a9..08bed5f 100644 --- a/src/vanth/process_watch.py +++ b/src/vanth/process_watch.py @@ -258,6 +258,11 @@ def _watch_loop( alive = alive or (lambda: process_alive(parent)) dead_since: float | None = None idle_since: float | None = None + # Most recent activity from EITHER source, retained across iterations: a + # one-shot stdin traffic sample must be remembered, not just for the single + # iteration that observed it (the tracker's own timestamp can already be + # stale by the next sample, which would reap ~two intervals after traffic). + last_seen_activity = tracker.last_activity() while True: time.sleep(interval) now = time.monotonic() @@ -277,13 +282,22 @@ def _watch_loop( # Recent relay/tool activity resets the idle timer even though the # process is momentarily not busy (review rc38 P1): the Desktop wake # relay polls and delivers asynchronously without holding an MCP - # request context, and a healthy relay must not be idle-reaped. - if now - tracker.last_activity() < interval: - idle_since = None - elif traffic(): + # request context, and a healthy relay must not be idle-reaped. The + # freshness window is the idle threshold itself — using the (tiny) + # sampling interval as the window mis-reaped a healthy relay whose + # notify cadence was coarser than the sampler (macOS runners). + # ``idle_since`` is seeded from the activity timestamp itself, so + # the effective timeout is ``idle`` (not ~2x from starting a second + # window once the freshness window expires). + if traffic(): + last_seen_activity = now + tracker_activity = tracker.last_activity() + if tracker_activity > last_seen_activity: + last_seen_activity = tracker_activity + if now - last_seen_activity < idle: idle_since = None elif idle_since is None: - idle_since = now + idle_since = last_seen_activity elif now - idle_since >= idle: on_exit() return diff --git a/src/vanth/server.py b/src/vanth/server.py index 454495e..14f7d4d 100644 --- a/src/vanth/server.py +++ b/src/vanth/server.py @@ -927,28 +927,80 @@ def _watch_on_failure(self, row: sqlite3.Row, on_failure: dict[str, Any]) -> Non streak = int(state.get("failure_streak", 0)) status = row["status"] if status == "failed": - # Count each failed EXECUTION exactly once (review P1-2 / P2): - # the failure event is the unit, identified by its event_id. A failed - # row that stays failed across many watcher ticks is not re-counted, - # but an automatic RESTART that reuses the same job row emits a NEW - # failed event with a new id — compare the latest failed event to the - # stored one so every execution (including restarts) advances the - # streak exactly once. + # Count each failed EXECUTION exactly once (review P1-2 / P2): the + # failure event is the unit. A failed row that stays failed across + # many watcher ticks is not re-counted, but an automatic RESTART that + # reuses the same job row emits a NEW failed event, so the streak is + # measured between two event-sequence watermarks. The upper bound is + # the row we just read: a failure committed between the two queries + # must not be counted here AND again on the next tick. last_terminal = self.db.execute( - "SELECT event_id, created_at FROM events WHERE job_id=? AND type='failed' ORDER BY seq DESC LIMIT 1", + "SELECT event_id, seq FROM events WHERE job_id=? AND type='failed' ORDER BY seq DESC LIMIT 1", (job_id,), ).fetchone() if last_terminal is None: return - if state.get("last_failure_event_id") == last_terminal["event_id"]: - return # already counted this failed run - new_streak = streak + 1 - if new_streak >= after_n and not (state.get("reacted_at_streak") == new_streak): - self._react_to_failure(row, on_failure, new_streak) - state["reacted_at_streak"] = new_streak + last_seq = int(last_terminal["seq"]) + watermark = state.get("failure_streak_after_seq") + if watermark is None: + # Migration from the pre-1.9.1 marker: resolve the legacy event + # id to its job-local sequence so already-counted failures are + # not counted a second time after upgrading. + legacy_id = state.get("last_failure_event_id") + if legacy_id: + legacy = self.db.execute( + "SELECT seq FROM events WHERE job_id=? AND event_id=?", (job_id, legacy_id) + ).fetchone() + if legacy is not None: + watermark = int(legacy["seq"]) + if watermark is not None and last_seq <= int(watermark): + # Every failed execution up to here is already counted. A prior + # tick may have committed the streak but failed (or crashed) + # before completing the reaction: retry it instead of dropping it. + if streak >= after_n and state.get("reacted_at_streak") != streak: + self._react_to_failure(row, on_failure, streak) + state["reacted_at_streak"] = streak + self._save_policy_state(job_id, state) + return + # Count EVERY failed execution in the interval, not just the latest: + # with a fast restart two failures can land between watcher ticks and + # incrementing by one per tick undercounts (Windows CI saw a final + # streak of 2, not 3). With no watermark (fresh streak / first-ever + # failure) count from the beginning so a pre-existing backlog is not + # collapsed to one. + if watermark is None: + pending = int( + self.db.execute( + "SELECT COUNT(*) FROM events WHERE job_id=? AND type='failed' AND seq<=?", + (job_id, last_seq), + ).fetchone()[0] + ) + else: + pending = int( + self.db.execute( + "SELECT COUNT(*) FROM events WHERE job_id=? AND type='failed' AND seq>? AND seq<=?", + (job_id, int(watermark), last_seq), + ).fetchone()[0] + ) + new_streak = streak + max(1, pending) + react = new_streak >= after_n and state.get("reacted_at_streak") != new_streak state["failure_streak"] = new_streak - state["last_failure_event_id"] = last_terminal["event_id"] + state["failure_streak_after_seq"] = last_seq + # Persist the streak BEFORE reacting: _react_to_failure emits the + # failure_threshold event, and a waiter that observes that event must + # already see the updated policy state (reading between the event + # commit and the state save saw no failure_streak). The reaction is + # marked complete only AFTER it succeeds, so a crash/error retries it. + # ponytail: at-least-once reaction delivery — a crash between the side + # effect and the marker save can repeat it. Harmless in practice + # (disable is filtered out of the scan, run_job refuses a busy target, + # alerts are advisory); a durable outbox + idempotency key per streak + # is the upgrade path if a reaction becomes side-effect-heavy. self._save_policy_state(job_id, state) + if react: + self._react_to_failure(row, on_failure, new_streak) + state["reacted_at_streak"] = new_streak + self._save_policy_state(job_id, state) elif status in {"completed", "timeout", "cancelled", "orphaned"}: # A non-failure terminal outcome resets the run identity: the NEXT # failure is a fresh run. timeout keeps its existing semantics @@ -956,10 +1008,16 @@ def _watch_on_failure(self, row: sqlite3.Row, on_failure: dict[str, Any]) -> Non # otherwise it continues the streak as before). Only reset when we # actually have a streak to clear, so the watcher stays a no-op for # jobs that never failed. - if state.get("failure_streak") or state.get("last_failure_event_id"): + if state.get("failure_streak") or state.get("failure_streak_after_seq") is not None: state["failure_streak"] = 0 state.pop("reacted_at_streak", None) - state.pop("last_failure_event_id", None) + # Move the watermark past the failures that preceded this success + # so the next streak counts only failures after the reset. + latest_failed = self.db.execute( + "SELECT seq FROM events WHERE job_id=? AND type='failed' ORDER BY seq DESC LIMIT 1", + (job_id,), + ).fetchone() + state["failure_streak_after_seq"] = int(latest_failed["seq"]) if latest_failed else 0 self._save_policy_state(job_id, state) def _watch_restart(self, row: sqlite3.Row, restart: dict[str, Any]) -> None: @@ -2032,17 +2090,17 @@ def _emit( payload["level"] = "warning" data_json = json.dumps(payload["data"], separators=(",", ":")) with self.db_lock: - for attempt in range(4): + for attempt in range(10): try: event = self._emit_transactional( job_id, payload, data_json, event_type, level, source, message ) break except sqlite3.OperationalError as exc: - if "locked" not in str(exc).lower() or attempt == 3: + if "locked" not in str(exc).lower() or attempt == 9: raise self.logger.warning("event write contended, retrying job_id=%s attempt=%s", job_id, attempt + 1) - time.sleep(0.05 * (attempt + 1)) + time.sleep(min(0.5, 0.05 * (attempt + 1))) else: # pragma: no cover - loop always breaks raise RuntimeError("event write failed") if event is not None and event.get("persisted") is not False: @@ -3250,7 +3308,7 @@ def claim() -> str | None: state["pending_restart_after"] = value token = "claim_" + uuid.uuid4().hex[:16] changed = self.db.execute( - "UPDATE jobs SET status='launching', claim_token=?, policy_state_json=?, updated_at=? " + "UPDATE jobs SET status='launching', claim_token=?, policy_state_json=?, updated_at=?, worker_pid=NULL " "WHERE job_id=? AND status IN ('queued','failed','orphaned') AND policy_disabled=0", (token, json.dumps(state, separators=(",", ":")), now_iso(), job_id), ).rowcount @@ -3696,7 +3754,9 @@ def rerun_sync( if prior_state and json.loads(row["policy_json"] or "null"): # Carry ONLY the failure streak: reacted_* dedup markers belong to # the previous runner instance and must not suppress the next - # failure's reaction. + # failure's reaction. The counting watermark is NOT carried — event + # ``seq`` is per-job, so the rerun's fresh sequence counts its own + # failures from the start (carrying the old seq would suppress them). self._save_policy_state(result["job_id"], {"failure_streak": int(prior_state.get("failure_streak", 0))}) return result @@ -4040,22 +4100,43 @@ def _wait( events = self._event_query(job_id, filters, since_event_id, 1) except RuntimeError: return {"result": "shutdown", "job_id": job_id, "message": "Vanth is shutting down"} + # Return the EARLIEST matching signal (terminal / metric threshold / + # progress), not a fixed precedence: with a streaming cursor the + # caller expects events in order, and a threshold crossed before the + # job finished must win over the terminal event that follows it. + candidates: list[tuple[int, dict[str, Any]]] = [] + since_seq = 0 + if since_event_id: + since_row = self._row( + "SELECT seq FROM events WHERE job_id=? AND event_id=?", (job_id, since_event_id) + ) + if since_row is not None: + since_seq = int(since_row["seq"]) if events: - return {"result": "event", "job_id": job_id, "status": self.status(job_id)["status"], "event": events[0]} + candidates.append(( + int(events[0].get("seq") or 0), + {"result": "event", "job_id": job_id, "status": self.status(job_id)["status"], "event": events[0]}, + )) if metric_ge: try: for metric, threshold in metric_ge.items(): - value = self._latest_metric_value(job_id, metric) - if value is not None and value >= threshold: - return { + # Cursor-aware and sample-consistent: the threshold is + # checked against the very sample whose event is returned. + value, metric_event = self._latest_metric_sample_after(job_id, metric, since_seq) + if metric_event is None or value is None or value < threshold: + continue + candidates.append(( + int(metric_event.get("seq") or 0), + { "result": "metric", "job_id": job_id, "status": self.status(job_id)["status"], "metric": metric, "threshold": threshold, "value": value, - "event": self._latest_metric_event(job_id, metric), - } + "event": metric_event, + }, + )) except RuntimeError: pass if return_progress and "progress" not in filters: @@ -4064,14 +4145,19 @@ def _wait( except RuntimeError: progress = [] if progress: - event = progress[0] - return { - "result": "progress", - "job_id": job_id, - "event": event, - "status": self.status(job_id)["status"], - "progress": event.get("data"), - } + candidates.append(( + int(progress[0].get("seq") or 0), + { + "result": "progress", + "job_id": job_id, + "event": progress[0], + "status": self.status(job_id)["status"], + "progress": progress[0].get("data"), + }, + )) + if candidates: + candidates.sort(key=lambda item: item[0]) + return candidates[0][1] remaining = deadline - time.monotonic() if remaining <= 0: return {"result": "timeout", "job_id": job_id, "status": self.status(job_id)["status"], "message": "No matching event before timeout"} @@ -4442,6 +4528,30 @@ def _latest_metric_event(self, job_id: str, metric: str) -> dict[str, Any] | Non ).fetchone() return self._event_dict(row) if row else None + def _latest_metric_sample_after( + self, job_id: str, metric: str, after_seq: int + ) -> tuple[float | None, dict[str, Any] | None]: + """Latest metric sample strictly after ``after_seq``, read atomically. + + Returns ``(value, event)`` for the SAME sample so the threshold check and + the returned event cannot disagree: reading the value and the event in + separate queries could pair a stale satisfying value with a newer event + that no longer satisfies the threshold. + """ + with self.db_lock: + row = self.db.execute( + """ + SELECT e.*, m.y AS metric_value FROM events e + JOIN metric_series m ON m.event_id = e.event_id + WHERE e.job_id=? AND m.metric=? AND e.seq > ? + ORDER BY e.seq DESC LIMIT 1 + """, + (job_id, metric, int(after_seq)), + ).fetchone() + if row is None: + return None, None + return float(row["metric_value"]), self._event_dict(row) + def metric_ingest(self, job_id: str, metrics: list[dict[str, Any]], idempotency_key: str | None = None) -> dict[str, Any]: """Record scalar metric points for a job programmatically. @@ -5200,7 +5310,10 @@ def doctor(self) -> dict[str, Any]: running_jobs = self._running_count() # Optional agent adapters being absent is informational, not a health # problem: the daemon (and its jobs) are fully functional without them. - soft_warning_types = {"codex_unavailable", "opencode_unavailable"} + # Orphaned MCP servers are likewise a host-hygiene advisory (reap them + # explicitly with `vanth doctor --reap-orphans`), and depend on the + # ambient process table, so they must not flip the health exit code. + soft_warning_types = {"codex_unavailable", "opencode_unavailable", "orphaned_mcp_servers"} hard_warnings = [w for w in warnings if w.get("type") not in soft_warning_types] return { "ok": not hard_warnings and quick_check == "ok", @@ -6526,6 +6639,91 @@ def job_cleanup_preview(older_than_seconds: int) -> dict[str, Any]: return get_client().get("/cleanup/preview", {"older_than_seconds": older_than_seconds}) +_VANTH_CLI_SUBCOMMANDS = { + "status", "doctor", "restart", "setup", "--help", "-h", "help", + "list", "ps", "logs", "tail", "stop", "artifacts", "prune", + "autostart", "--version", "version", "remote", +} + +_VANTH_SCRIPT_NAMES = {"vanth", "vanth.exe", "vanth-script.py", "vanth-script.pyw"} + +# Interpreter options the MCP launch shape may carry. Valueless options are +# skipped; value-taking options consume the next token; ANYTHING else that +# starts with "-" (unknown option, long option, or an attached ``-c``) +# refuses the match — the reaper kills what it accepts, so ambiguity loses. +_PY_VALUELESS_OPTIONS = { + "-O", "-OO", "-B", "-b", "-d", "-E", "-I", "-q", "-R", "-s", "-S", "-u", "-v", +} +_PY_VALUED_OPTIONS = {"-X", "-W"} + + +def _is_vanth_mcp_command(command_line: str) -> bool: + """Whether a command line is a Vanth MCP stdio server (not a CLI command). + + A false positive here is not cosmetic: the orphan reaper terminates every + reported process, so the matcher accepts only the exact supported launch + shapes and rejects anything ambiguous: + + - `` -m vanth.server`` / ``-m vanth.mcp`` (``-m`` must be the + interpreter's launching argument; ``python unrelated.py -m vanth.server``, + ``bash -lc 'python -m vanth.server'`` and ``python -c`` are + rejected), and + - the ``vanth`` console script (``vanth`` / ``python .../vanth``) run with + no CLI subcommand — ``vanth logs --follow`` is the CLI, not MCP. + + A quoted ``argv0`` (Windows ``"C:\\Program Files\\Python\\python.exe" ...``) + is split off before tokenizing; arguments after it are whitespace-split, so + a quoting trick later in the line can only cause a REJECTION, never a false + match. Missing a genuine exotic launch is the safe failure mode. + """ + if command_line.startswith('"'): + end = command_line.find('"', 1) + if end == -1: + return False + tokens = [command_line[1:end]] + command_line[end + 1:].split() + else: + tokens = command_line.split() + if not tokens: + return False + + def _base(token: str) -> str: + token = token.strip("\"'") + return token.replace("\\", "/").rstrip("/").rsplit("/", 1)[-1].lower() + + def _is_cli_script(script_index: int) -> bool: + rest = tokens[script_index + 1:] + # Quoted subcommand (``vanth "logs"``) is still a CLI invocation. + return bool(rest) and rest[0].strip("\"'") in _VANTH_CLI_SUBCOMMANDS + + if _base(tokens[0]).startswith("python"): + # Walk the interpreter's own options to the launching argument: + # ``-m module`` (the supported MCP shape), ``-c`` (never us), or the + # first non-option token (a script path — matched only when it is the + # ``vanth`` console script with no CLI subcommand). + i = 1 + while i < len(tokens): + tok = tokens[i] + if tok == "-m": + return i + 1 < len(tokens) and tokens[i + 1] in {"vanth.server", "vanth.mcp"} + if tok == "-c" or tok.startswith("-c"): + return False + if tok in _PY_VALUED_OPTIONS: + i += 2 + continue + if tok in _PY_VALUELESS_OPTIONS: + i += 1 + continue + if tok.startswith("-"): + return False + if _base(tok) in _VANTH_SCRIPT_NAMES: + return not _is_cli_script(i) + return False + return False + if _base(tokens[0]) in _VANTH_SCRIPT_NAMES: + return not _is_cli_script(0) + return False + + def _orphaned_mcp_servers() -> list[dict[str, Any]]: """Find MCP stdio server processes whose launching client is gone. @@ -6542,45 +6740,53 @@ def _orphaned_mcp_servers() -> list[dict[str, Any]]: candidates = [] try: if sys.platform == "win32": + # Get-CimInstance provides the COMMAND LINE (WMIC CSV does not, and + # its columns are ordered alphabetically, so the old positional parse + # both misread the fields and could not establish MCP identity). + # JSON output avoids comma-splitting a command line containing commas. result = _sp.run( - ["wmic", "process", "get", "name,processid,parentprocessid,creationdate", "/format:csv"], - stdout=_sp.PIPE, stderr=_sp.DEVNULL, text=True, timeout=10, + [ + "powershell", "-NoProfile", "-NonInteractive", "-Command", + "Get-CimInstance Win32_Process | " + "Select-Object ProcessId,ParentProcessId,CreationDate,CommandLine | " + "ConvertTo-Json -Compress", + ], + stdout=_sp.PIPE, stderr=_sp.DEVNULL, text=True, timeout=15, ) - for line in result.stdout.splitlines()[1:]: - if not line.strip(): - continue - parts = line.split(",") - if len(parts) < 5: - continue - _, name, pid, ppid, created = parts[:5] - name = (name or "").strip() - pid_s = (pid or "").strip() - if not name or not pid_s.isdigit(): + raw = (result.stdout or "").strip() + records = json.loads(raw) if raw else [] + if isinstance(records, dict): + records = [records] + for rec in records: + cmdline = rec.get("CommandLine") or "" + if not _is_vanth_mcp_command(cmdline): continue - if "vanth" not in name.lower() and "python" not in name.lower(): + pid = rec.get("ProcessId") + if isinstance(pid, bool) or not isinstance(pid, int): continue + ppid = rec.get("ParentProcessId") candidates.append( { - "pid": int(pid_s), - "name": name, - "started": created.strip(), - "ppid": int(ppid.strip()) if (ppid or "").strip().isdigit() else None, + "pid": pid, + "name": cmdline, + "started": str(rec.get("CreationDate") or ""), + "ppid": ppid if isinstance(ppid, int) and not isinstance(ppid, bool) else None, } ) else: - result = _sp.run(["ps", "-eo", "pid=,ppid=,etime=,comm="], + result = _sp.run(["ps", "-eo", "pid=,ppid=,etime=,args="], stdout=_sp.PIPE, stderr=_sp.DEVNULL, text=True, timeout=10) for line in result.stdout.splitlines(): parts = line.split(None, 3) if len(parts) < 4: continue - pid, ppid, etime, comm = parts - if "vanth" not in comm.lower() and "python" not in comm.lower(): + pid, ppid, etime, args = parts + if not _is_vanth_mcp_command(args): continue candidates.append( { "pid": int(pid), - "name": comm, + "name": args, "started": etime, "ppid": int(ppid) if ppid.isdigit() else None, } @@ -6634,11 +6840,7 @@ def main(argv: list[str] | None = None) -> None: # Human-facing subcommands are dispatched to the CLI; anything else # (including no args) runs the MCP stdio server, which is what MCP # clients expect from `vanth` (bare). - if args and args[0] in { - "status", "doctor", "restart", "setup", "--help", "-h", "help", - "list", "ps", "logs", "tail", "stop", "artifacts", "prune", - "autostart", "--version", "version", "remote", - }: + if args and args[0] in _VANTH_CLI_SUBCOMMANDS: raise SystemExit(cli_main(args)) # Interactive misuse guard (user report): bare `vanth` typed in a real # terminal would otherwise start the MCP stdio server and appear to diff --git a/tests/artifacts/test_capture_security.py b/tests/artifacts/test_capture_security.py index c31cada..5836455 100644 --- a/tests/artifacts/test_capture_security.py +++ b/tests/artifacts/test_capture_security.py @@ -116,6 +116,10 @@ def shrinking_hasher(path): monkeypatch.setattr(manifest_module, "_hash_file_streaming", shrinking_hasher) with pytest.raises(ValueError, match="source mutated during capture"): build_manifest_from_tree(root, "t") + # The first capture shrank the victim; restore the precondition so the + # second capture again detects a size change (a same-size rewrite is not + # guaranteed to move mtime_ns on POSIX). + victim.write_bytes(b"0123456789") with pytest.raises(ValueError, match="source mutated during capture"): ops.put_dir(root, "t", idempotency_key="mutate-1") assert version_count(ops) == 0 diff --git a/tests/artifacts/test_operations.py b/tests/artifacts/test_operations.py index f709de3..c6dc012 100644 --- a/tests/artifacts/test_operations.py +++ b/tests/artifacts/test_operations.py @@ -5,7 +5,6 @@ import asyncio import hashlib import json -import subprocess import sys from pathlib import Path @@ -17,6 +16,9 @@ from vanth.server import JobManager +import shellcmd + + @pytest.fixture() def home(tmp_path): return tmp_path / "state" @@ -310,7 +312,7 @@ def test_verify_detects_tampered_blob_as_result(home): def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) def test_job_cleanup_preserves_managed_content(home): diff --git a/tests/artifacts/test_review_fixes.py b/tests/artifacts/test_review_fixes.py index 8f2573b..ebc447b 100644 --- a/tests/artifacts/test_review_fixes.py +++ b/tests/artifacts/test_review_fixes.py @@ -90,14 +90,19 @@ def test_portable_rename_noreplace(tmp_path): assert dst.read_bytes() == b"payload" src2.unlink() - # Directories take the checked-rename branch (link(2) refuses dirs). + # Directories are intentionally refused by the portable helper (link(2) + # refuses dirs and lstat+rename would reopen a clobber race). Linux and + # macOS use renameat2/renameatx_np for directories, so this fail-closed + # path is only reached on BSDs. if os.name == "nt": pytest.skip("directory rename fallback is POSIX-only; Windows never reaches it") src_dir = tmp_path / "src-dir" src_dir.mkdir() dst_dir = tmp_path / "out-dir" - ops._rename_noreplace_portable(str(src_dir), str(dst_dir)) - assert dst_dir.is_dir() + with pytest.raises(OSError, match="unsupported"): + ops._rename_noreplace_portable(str(src_dir), str(dst_dir)) + assert not dst_dir.exists() + assert src_dir.is_dir() def test_restore_temp_db_name_is_collision_free(tmp_path): diff --git a/tests/shellcmd.py b/tests/shellcmd.py new file mode 100644 index 0000000..4e717e3 --- /dev/null +++ b/tests/shellcmd.py @@ -0,0 +1,34 @@ +"""Platform-correct quoting for shell command strings used in tests. + +The runner executes a workload's command through the platform shell +(``subprocess.Popen(command, shell=True)``), so the command string a test builds +must use the *host* shell's quoting rules: + +- Windows: ``subprocess.list2cmdline`` (the ``cmd.exe`` rules). +- POSIX: ``shlex.join``. ``list2cmdline`` implements Windows rules only; on + POSIX it leaves ``python -c "print('x')"`` unquoted, and the shell rejects the + parentheses (``bash: syntax error near unexpected token '('``). + +Keeping this in one place lets the Python test matrix run on Linux/macOS as +well as Windows. ``tests`` is on ``sys.path`` via pytest's ``pythonpath``. +""" + +from __future__ import annotations + +import os +import shlex +import subprocess +import sys + + +def join(argv: object) -> str: + """Quote ``argv`` for the current platform's shell.""" + parts = [str(part) for part in argv] # type: ignore[union-attr] + if os.name == "nt": + return subprocess.list2cmdline(parts) + return shlex.join(parts) + + +def cmd(code: str) -> str: + """Build a ``python -c `` command string for the current platform.""" + return join([sys.executable, "-c", code]) diff --git a/tests/test_agent_features.py b/tests/test_agent_features.py index d39d52e..61addcc 100644 --- a/tests/test_agent_features.py +++ b/tests/test_agent_features.py @@ -8,7 +8,6 @@ import asyncio import json import os -import subprocess import sys import pytest @@ -16,8 +15,11 @@ from vanth.server import JobManager +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) def wait_event(manager: JobManager, job_id: str, event_type: str) -> dict: @@ -42,7 +44,7 @@ def test_status_exposes_command_cwd_env_and_timeout(tmp_path): ) wait_event(manager, job_id, "completed") status = manager.status(job_id) - assert "print('hi')" in status["command"] + assert status["command"] == cmd("print('hi')") assert status["cwd"] == str(tmp_path) assert status["env"] == {"VANTH_TEST_ENV": "present", "SECOND": "two"} assert status["timeout_seconds"] == 30 @@ -58,7 +60,7 @@ def test_view_exposes_command_and_env(tmp_path): wait_event(manager, job_id, "completed") view = manager.agent_view()["jobs"] entry = next(j for j in view if j["job_id"] == job_id) - assert "print('view')" in entry["command"] + assert entry["command"] == cmd("print('view')") assert entry["env"] == {"K": "v"} finally: manager.close() diff --git a/tests/test_agent_logger.py b/tests/test_agent_logger.py index d6c02d4..8b8925f 100644 --- a/tests/test_agent_logger.py +++ b/tests/test_agent_logger.py @@ -3,7 +3,6 @@ import asyncio import io import json -import subprocess import sys import pytest @@ -12,6 +11,9 @@ from vanth.server import JobManager, parse_agent_event_line +import shellcmd + + def capture_log(method, *args, **kwargs): out = io.StringIO() original = sys.stdout @@ -57,7 +59,7 @@ def test_logger_events_persist_through_daemon(tmp_path): "logger.info('loguru line one', phase='train');" "logger.warning('loguru warn')" ) - command = subprocess.list2cmdline([sys.executable, "-c", code]) + command = shellcmd.join([sys.executable, "-c", code]) job_id = asyncio.run(manager.start(command))["job_id"] asyncio.run(manager.wait(job_id, ["log"], timeout_seconds=10)) asyncio.run(manager.wait(job_id, ["completed"], timeout_seconds=10)) diff --git a/tests/test_attribution_masking.py b/tests/test_attribution_masking.py index 58bebfd..6e5664d 100644 --- a/tests/test_attribution_masking.py +++ b/tests/test_attribution_masking.py @@ -4,7 +4,6 @@ import asyncio import json -import subprocess import sys import time from pathlib import Path @@ -14,8 +13,11 @@ from vanth.server import JobManager, mask_secrets +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) def _wait_status(manager: JobManager, job_id: str, want: set[str], timeout: float = 15.0) -> str: diff --git a/tests/test_backup.py b/tests/test_backup.py index 4ca90fb..7dd8bfa 100644 --- a/tests/test_backup.py +++ b/tests/test_backup.py @@ -6,7 +6,6 @@ import hashlib import json import sqlite3 -import subprocess import sys import zipfile from pathlib import Path @@ -19,8 +18,11 @@ from vanth.server import JobManager +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) def _seed(home: Path) -> None: diff --git a/tests/test_cli_qol.py b/tests/test_cli_qol.py index ce6d960..c716358 100644 --- a/tests/test_cli_qol.py +++ b/tests/test_cli_qol.py @@ -10,6 +10,9 @@ from vanth.client import VanthClient +import shellcmd + + def free_port(): with socket.socket() as sock: sock.bind(("127.0.0.1", 0)) @@ -17,7 +20,7 @@ def free_port(): def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) @pytest.fixture() @@ -270,6 +273,35 @@ def test_doctor_reports_orphans_field(daemon): assert isinstance(payload["orphaned_mcp_servers"], list) +@pytest.mark.parametrize( + "command, expected", + [ + # Real Vanth MCP stdio servers: + ("/usr/bin/python3 /opt/venv/bin/vanth", True), + (r"C:\venv\Scripts\python.exe C:\venv\Scripts\vanth.exe", True), + ("/usr/bin/python3 -m vanth.server", True), + ("/usr/bin/python3 -O -m vanth.mcp", True), + ('"C:\\Program Files\\Python\\python.exe" -m vanth.server', True), + ("/usr/bin/python3 -X dev -m vanth.server", True), + # Not MCP servers — must never be matched (and thus never reaped): + ("/home/user/vanth-ci/.venv/bin/python /home/user/vanth-ci/.venv/bin/pytest -q", False), + ("/bin/bash -lc cd /home/user/vanth-ci && uv run pytest", False), + ("/bin/bash -lc vanth doctor --json", False), + ("/usr/bin/python3 -m vanth.runner /home/user/state job_x claim.json", False), + ("/usr/bin/python3 -m vanth.daemon", False), + # Lookalikes the reaper must refuse to kill: + ('vanth "logs" --follow', False), + ("/usr/bin/python -c__import__('time').sleep(600) vanth", False), + ("/usr/bin/python3 unrelated.py -m vanth.server", False), + ("/usr/bin/python3 -O vanth status", False), + ], +) +def test_vanth_mcp_command_detection(command, expected): + from vanth.server import _is_vanth_mcp_command + + assert _is_vanth_mcp_command(command) is expected + + def test_remote_list_empty(daemon): tmp_path, client, port = daemon result = run_cli(tmp_path / "state", "remote", "list", port=port) diff --git a/tests/test_codex_desktop.py b/tests/test_codex_desktop.py index b1169ed..010e723 100644 --- a/tests/test_codex_desktop.py +++ b/tests/test_codex_desktop.py @@ -42,6 +42,9 @@ from vanth.server import JobManager +import shellcmd + + class FakePipeServer: """A fake Desktop app-tools host speaking the length-prefixed JSON-RPC protocol over a socketpair. @@ -936,7 +939,7 @@ class TestRelay: def _start_delivery(self, manager, thread_id="thread_dest"): async def main(): job = await manager.start( - subprocess.list2cmdline([sys.executable, "-c", "import sys; sys.exit(0)"]), + shellcmd.join([sys.executable, "-c", "import sys; sys.exit(0)"]), wake_targets=[{"type": "codex_desktop", "events": ["completed"], "thread_id": thread_id}], ) await manager.wait(job["job_id"], ["completed"], timeout_seconds=30) @@ -1098,7 +1101,7 @@ def test_threadId_alias_target_is_pollable(self, tmp_path): async def main(): job = await manager.start( - subprocess.list2cmdline([sys.executable, "-c", "import sys; sys.exit(0)"]), + shellcmd.join([sys.executable, "-c", "import sys; sys.exit(0)"]), wake_targets=[{"type": "codex_desktop", "events": ["completed"], "threadId": "thread_legacy"}], ) await manager.wait(job["job_id"], ["completed"], timeout_seconds=30) diff --git a/tests/test_daemon.py b/tests/test_daemon.py index 56f0b4c..07dae29 100644 --- a/tests/test_daemon.py +++ b/tests/test_daemon.py @@ -6,6 +6,9 @@ from vanth.client import VanthClient +import shellcmd + + def free_port(): with socket.socket() as sock: sock.bind(("127.0.0.1", 0)) @@ -39,7 +42,7 @@ def test_daemon_http_job_flow(tmp_path): job = client.post( "/jobs", { - "command": subprocess.list2cmdline( + "command": shellcmd.join( [ sys.executable, "-c", diff --git a/tests/test_daemon_discovery.py b/tests/test_daemon_discovery.py index c0bbb2b..f211c1c 100644 --- a/tests/test_daemon_discovery.py +++ b/tests/test_daemon_discovery.py @@ -29,7 +29,19 @@ def test_daemon_writes_and_removes_discovery_metadata(tmp_path): deadline = time.monotonic() + 15 while time.monotonic() < deadline and not meta_path.exists(): time.sleep(0.1) - assert meta_path.exists(), "daemon.json not written" + if not meta_path.exists(): + proc.terminate() + try: + _, err = proc.communicate(timeout=5) + except Exception: + err = b"" + log_tail = "" + log_path = tmp_path / "logs" / "daemon.log" + if log_path.exists(): + log_tail = log_path.read_text(encoding="utf-8", errors="replace")[-3000:] + raise AssertionError( + f"daemon.json not written (rc={proc.returncode}): stderr={err[-2000:]!r} log={log_tail!r}" + ) payload = json.loads(meta_path.read_text(encoding="utf-8")) assert payload["url"] == f"http://127.0.0.1:{port}" assert payload["home"] == str(tmp_path.resolve()) diff --git a/tests/test_daemon_hardening.py b/tests/test_daemon_hardening.py index 90dfc34..1ff743b 100644 --- a/tests/test_daemon_hardening.py +++ b/tests/test_daemon_hardening.py @@ -13,6 +13,9 @@ import vanth.daemon as daemon +import shellcmd + + def free_port(): with socket.socket() as sock: sock.bind(("127.0.0.1", 0)) @@ -42,11 +45,23 @@ def start_daemon(tmp_path, max_request_bytes=1024 * 1024): except OSError: time.sleep(0.05) proc.terminate() - raise AssertionError("daemon did not start") + try: + _, err = proc.communicate(timeout=5) + except Exception: + err = b"" + log_tail = "" + log_path = tmp_path / "state" / "logs" / "daemon.log" + if log_path.exists(): + log_tail = log_path.read_text(encoding="utf-8", errors="replace")[-3000:] + raise AssertionError( + f"daemon did not start (rc={proc.returncode}): stderr={err[-2000:]!r} log={log_tail!r}" + ) def request(port, method, path, body=None, headers=None): - connection = http.client.HTTPConnection("127.0.0.1", port, timeout=5) + # 20s: spawning a runner / DB work can exceed 5s on a loaded Windows runner; + # connection-refused is immediate, so the health-poll deadline still holds. + connection = http.client.HTTPConnection("127.0.0.1", port, timeout=20) connection.request(method, path, body=body, headers=headers or {}) response = connection.getresponse() payload = json.loads(response.read()) @@ -198,7 +213,7 @@ def test_shutdown_returns_controlled_result_to_active_wait(tmp_path): token = (tmp_path / "state" / "token").read_text(encoding="utf-8") headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} try: - command = subprocess.list2cmdline([sys.executable, "-c", "import time; time.sleep(30)"]) + command = shellcmd.join([sys.executable, "-c", "import time; time.sleep(30)"]) status, started = request(port, "POST", "/jobs", json.dumps({"command": command}).encode(), headers) assert status == 200 result = [] diff --git a/tests/test_delivery_hardening.py b/tests/test_delivery_hardening.py index 75af966..30b1e24 100644 --- a/tests/test_delivery_hardening.py +++ b/tests/test_delivery_hardening.py @@ -1,7 +1,6 @@ import asyncio import json import os -import subprocess import sys import threading import time @@ -12,11 +11,14 @@ from vanth.server import JobManager, now_iso +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) -def wait_for_delivery(manager: JobManager, job_id: str, status: str, timeout: float = 5): +def wait_for_delivery(manager: JobManager, job_id: str, status: str, timeout: float = 20): deadline = time.monotonic() + timeout delivery = None while time.monotonic() < deadline: @@ -154,7 +156,10 @@ def test_retry_due_after_manager_restart_is_dispatched(tmp_path): ) ) retrying = wait_for_delivery(manager, started["job_id"], "retrying") - assert retrying is not None + # Assert the transient retry state was actually observed BEFORE the manager + # is closed: otherwise the helper can return an already-delivered row and the + # test would never exercise recovery across a restart. + assert retrying is not None and retrying["status"] == "retrying" manager.close() restarted = JobManager(home) diff --git a/tests/test_job_send.py b/tests/test_job_send.py index e27bccf..6c5c0e6 100644 --- a/tests/test_job_send.py +++ b/tests/test_job_send.py @@ -7,7 +7,6 @@ """ import asyncio -import subprocess import sys import pytest @@ -15,8 +14,11 @@ from vanth.server import JobManager +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) def wait_event(manager: JobManager, job_id: str, event_type: str) -> dict: diff --git a/tests/test_mcp_stdio.py b/tests/test_mcp_stdio.py index 657a37f..ef27782 100644 --- a/tests/test_mcp_stdio.py +++ b/tests/test_mcp_stdio.py @@ -11,6 +11,9 @@ from mcp.client.stdio import StdioServerParameters, stdio_client +import shellcmd + + def content(result): if result.structuredContent is not None: return result.structuredContent @@ -60,7 +63,7 @@ async def main(): await session.call_tool( "job_start", { - "command": subprocess.list2cmdline( + "command": shellcmd.join( [sys.executable, str(Path(__file__).parents[1] / "examples" / "long_job.py")] ), "wake_targets": [ @@ -103,7 +106,9 @@ async def main(): assert progress["event"]["type"] == "progress" assert progress["event"]["data"]["current"] == 1 - assert progress["status"] == "running" + # The status snapshot is taken when the wait returns; a short + # job can finish first on a loaded runner, so accept either. + assert progress["status"] in {"running", "completed"} assert status["progress"]["current"] >= 1 assert status["origin_thread_id"] == "thread_origin" assert status["wake_thread_id"] == "thread_test" @@ -138,7 +143,7 @@ async def main(): missing = content(await session.call_tool("job_status", {"job_id": "job_missing"})) assert missing["result"] == "error" - command = subprocess.list2cmdline( + command = shellcmd.join( [ sys.executable, "-c", @@ -184,7 +189,7 @@ async def main(): async with ClientSession(read, write, read_timeout_seconds=timedelta(seconds=10)) as session: await session.initialize() # Start a quick job so there is a real job row to wake. - command = subprocess.list2cmdline([sys.executable, "-c", "print('wake me')"]) + command = shellcmd.join([sys.executable, "-c", "print('wake me')"]) start = content(await session.call_tool("job_start", {"command": command})) # Review rc38 P1: the documented rc37 wake tool names must # remain callable over stdio (agents must not learn diff --git a/tests/test_probes.py b/tests/test_probes.py index 5e3e217..00d806f 100644 --- a/tests/test_probes.py +++ b/tests/test_probes.py @@ -4,7 +4,6 @@ import asyncio import socket -import subprocess import sys import threading import time @@ -16,8 +15,11 @@ from vanth.server import JobManager +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) SLEEP = "import time; time.sleep(30)" diff --git a/tests/test_process_watch.py b/tests/test_process_watch.py index bafb2ed..c3bac63 100644 --- a/tests/test_process_watch.py +++ b/tests/test_process_watch.py @@ -136,17 +136,18 @@ def relay_poll(): relay = threading.Thread(target=relay_poll, daemon=True) relay.start() try: - # Parent self (alive), idle threshold 0.05s, interval 0.005s. Without - # activity the process would exit in ~0.05s; the relay's continuous - # activity must hold it far past that. + # Parent self (alive), idle threshold 0.5s, interval 0.005s. Without + # activity the process would exit in ~0.5s; the relay's continuous + # activity must hold it far past that. (The threshold is kept well above + # runner scheduling jitter, which flaked a 50ms value on macOS.) thread = threading.Thread( target=_watch_loop, - args=(os.getpid(), 0.005, 0.0, 0.05, fake_exit, tracker), + args=(os.getpid(), 0.005, 0.0, 0.5, fake_exit, tracker), kwargs={"traffic": lambda: 0, "alive": lambda: parent_alive["value"]}, daemon=True, ) thread.start() - time.sleep(0.3) + time.sleep(1.2) assert not exited, "relay activity must prevent idle exit" # Stop the relay AND make the parent "die" so the daemon watchdog thread # terminates (a never-exiting daemon thread would keep calling @@ -174,12 +175,15 @@ def fake_exit(): tracker = _InFlight() stop = threading.Event() + parent_alive = {"value": True} def relay_poll(): - # A long-poll that blocks well past the idle threshold (0.05s). + # A long-poll that blocks well past the 0.1s idle threshold: if the + # watchdog ignored the in-flight tracker the process would be reaped + # during the poll. while not stop.is_set(): with tracker: - time.sleep(0.2) + time.sleep(0.5) time.sleep(0.01) relay = threading.Thread(target=relay_poll, daemon=True) @@ -187,24 +191,18 @@ def relay_poll(): try: thread = threading.Thread( target=_watch_loop, - args=(os.getpid(), 0.005, 0.0, 0.05, fake_exit, tracker), - kwargs={"traffic": lambda: 0, "alive": lambda: True}, + args=(os.getpid(), 0.005, 0.0, 0.1, fake_exit, tracker), + kwargs={"traffic": lambda: 0, "alive": lambda: parent_alive["value"]}, daemon=True, ) thread.start() - time.sleep(0.3) + time.sleep(1.2) assert not exited, "a blocking relay poll must keep the process alive past the idle timeout" stop.set() relay.join(timeout=0.5) # Flip parent death so the watchdog thread terminates. - import vanth.process_watch as pw - - orig_alive = pw.process_alive - pw.process_alive = lambda pid: False - try: - thread.join(timeout=0.5) - finally: - pw.process_alive = orig_alive + parent_alive["value"] = False + thread.join(timeout=0.5) assert not thread.is_alive(), "watchdog must terminate once the parent dies" finally: stop.set() diff --git a/tests/test_qol_mcp.py b/tests/test_qol_mcp.py index f791aa5..d4ef5e4 100644 --- a/tests/test_qol_mcp.py +++ b/tests/test_qol_mcp.py @@ -7,7 +7,6 @@ import asyncio import base64 import json -import subprocess import sys import time @@ -16,8 +15,11 @@ from vanth.server import JobManager +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) def wait_event(manager: JobManager, job_id: str, event_type: str) -> dict: diff --git a/tests/test_qol_rerun_batch.py b/tests/test_qol_rerun_batch.py index d8d1c30..53b753f 100644 --- a/tests/test_qol_rerun_batch.py +++ b/tests/test_qol_rerun_batch.py @@ -1,7 +1,6 @@ """Tests for rerun-with-overrides and status_batch QoL features.""" import asyncio -import subprocess import sys import pytest @@ -9,8 +8,11 @@ from vanth.server import JobManager +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) def wait_event(manager: JobManager, job_id: str, event_type: str) -> dict: @@ -33,8 +35,8 @@ def test_rerun_overrides_command_and_env(tmp_path): wait_event(manager, new_id, "completed") status = manager.status(new_id) - assert "print('new')" in status["command"] - assert "print('orig')" not in status["command"] + assert status["command"] == cmd("print('new')") + assert status["command"] != cmd("print('orig')") assert status["env"] == {"K": "2"} finally: manager.close() @@ -79,7 +81,7 @@ def test_rerun_async_with_overrides(tmp_path): reran = asyncio.run(manager.rerun(job_id, command=cmd("print('async')"), env={"K": "9"})) wait_event(manager, reran["job_id"], "completed") status = manager.status(reran["job_id"]) - assert "print('async')" in status["command"] + assert status["command"] == cmd("print('async')") assert status["env"] == {"K": "9"} finally: manager.close() diff --git a/tests/test_qol_ux_batch.py b/tests/test_qol_ux_batch.py index 32dd81b..d6d8ac3 100644 --- a/tests/test_qol_ux_batch.py +++ b/tests/test_qol_ux_batch.py @@ -1,6 +1,5 @@ import asyncio import json -import subprocess import sys import time @@ -9,8 +8,11 @@ from vanth.server import JobManager +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) def start_job(manager, code, **kwargs): diff --git a/tests/test_qol_wait_tail.py b/tests/test_qol_wait_tail.py index 749b673..ef51c3d 100644 --- a/tests/test_qol_wait_tail.py +++ b/tests/test_qol_wait_tail.py @@ -1,14 +1,16 @@ import asyncio import json -import subprocess import sys import time from vanth.server import JobManager +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) def start_job(manager, code, **kwargs): diff --git a/tests/test_qol_wake_shorthand.py b/tests/test_qol_wake_shorthand.py index 9dc5843..cf77311 100644 --- a/tests/test_qol_wake_shorthand.py +++ b/tests/test_qol_wake_shorthand.py @@ -5,7 +5,6 @@ """ import asyncio -import subprocess import sys import time @@ -14,8 +13,11 @@ from vanth.server import JobManager, _build_wake_target +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) def wait_event(manager: JobManager, job_id: str, event_type: str) -> dict: diff --git a/tests/test_queues.py b/tests/test_queues.py index a12fdfd..99493c3 100644 --- a/tests/test_queues.py +++ b/tests/test_queues.py @@ -3,7 +3,6 @@ from __future__ import annotations import asyncio -import subprocess import sys import time @@ -12,8 +11,11 @@ from vanth.server import JobManager +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) SLEEP = "import time; time.sleep(30)" diff --git a/tests/test_quotas_retention.py b/tests/test_quotas_retention.py index 8b76de6..cb5cd8b 100644 --- a/tests/test_quotas_retention.py +++ b/tests/test_quotas_retention.py @@ -1,7 +1,6 @@ from __future__ import annotations import asyncio -import subprocess import sys import time @@ -10,8 +9,11 @@ from vanth.server import JobManager +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) def wait_status(manager: JobManager, job_id: str, status: str, timeout: float = 30) -> dict: diff --git a/tests/test_release_workloads.py b/tests/test_release_workloads.py index ec998a3..3a85b5d 100644 --- a/tests/test_release_workloads.py +++ b/tests/test_release_workloads.py @@ -20,8 +20,11 @@ from vanth.server import JobManager +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) def wait_completed(manager: JobManager, job_id: str, timeout: float = 120) -> None: @@ -128,7 +131,11 @@ def test_slow_wake_adapter_does_not_delay_stream_parsing(tmp_path): start = time.monotonic() wait_completed(manager, job_id, timeout=10) elapsed = time.monotonic() - start - assert elapsed < 4, f"job completion waited on the slow adapter: {elapsed:.2f}s" + # The meaningful regression (completion waiting on the 5s adapters) would + # need ~15s (10 deliveries / 4 concurrent) and trip the 10s waiter above; + # a fast CPU-starved Windows runner can still take several seconds to + # spawn the adapter subprocesses, so keep the bound just under the waiter. + assert elapsed < 10, f"job completion waited on the slow adapter: {elapsed:.2f}s" assert event_counts(manager, job_id)["progress"] == 10 finally: manager.close() @@ -184,7 +191,10 @@ def test_cross_process_emits_keep_unique_seq_and_lose_no_events(tmp_path): ) % (str(tmp_path / "state"), job_id) procs = [subprocess.Popen([sys.executable, "-c", worker], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) for _ in range(6)] for proc in procs: - out, err = proc.communicate(timeout=60) + # 120s: six processes each commit 100 events; a WAL fsync per commit puts + # the contended total near 60s on slower disks/Python builds, so 60s + # flaked (matches wait_completed's load-adjusted budget). + out, err = proc.communicate(timeout=120) assert proc.returncode == 0, err restarted = JobManager(tmp_path / "state", recover=False) diff --git a/tests/test_schedules.py b/tests/test_schedules.py index d61422d..196b106 100644 --- a/tests/test_schedules.py +++ b/tests/test_schedules.py @@ -2,7 +2,6 @@ from __future__ import annotations -import subprocess import sys import time from datetime import datetime, timedelta, timezone @@ -20,8 +19,11 @@ ) +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) def _utc(text: str) -> datetime: diff --git a/tests/test_server_hardening.py b/tests/test_server_hardening.py index c97d8ac..9ef6af6 100644 --- a/tests/test_server_hardening.py +++ b/tests/test_server_hardening.py @@ -11,8 +11,11 @@ from vanth.server import JobManager +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) def test_malformed_event_does_not_kill_reader(tmp_path): @@ -110,6 +113,11 @@ async def start(): recovered = JobManager(tmp_path) stopped = recovered.stop_sync(job_id, kill_after_seconds=0) assert stopped["status"] == "cancelled" + # Process teardown is asynchronous on Windows; give the OS a bounded window + # to reap the runner/workload before asserting they are gone. + deadline = time.monotonic() + 5 + while time.monotonic() < deadline and (recovered._pid_alive(worker_pid) or recovered._pid_alive(pid)): + time.sleep(0.05) assert not recovered._pid_alive(worker_pid) assert not recovered._pid_alive(pid) recovered.close() diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index a69549b..ff650e5 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -11,8 +11,11 @@ from vanth.server import JobManager +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) def free_port(): @@ -151,7 +154,7 @@ def test_daemon_telemetry_http_routes(tmp_path): "from vanth.agent_events import agent_event; " "[agent_event('metric', _step=i, loss=1.0/i+1) for i in range(1, 4)]" ) - job = client.post("/jobs", {"command": subprocess.list2cmdline([sys.executable, "-c", code])}) + job = client.post("/jobs", {"command": shellcmd.join([sys.executable, "-c", code])}) job_id = job["job_id"] client.post(f"/jobs/{job_id}/wait", {"filters": ["completed"], "timeout_seconds": 20}) diff --git a/tests/test_v1_hardening.py b/tests/test_v1_hardening.py index f6142a3..6624dab 100644 --- a/tests/test_v1_hardening.py +++ b/tests/test_v1_hardening.py @@ -15,8 +15,11 @@ from vanth.server import JobManager +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) def wait_event(manager: JobManager, job_id: str, event_type: str) -> dict: @@ -195,20 +198,52 @@ def test_stop_failure_leaves_running_job_retryable(tmp_path, monkeypatch): def test_stop_intent_and_pid_publication_interleavings(tmp_path): manager = JobManager(tmp_path / "state") + # Stop the maintenance loop: it would otherwise reconcile the synthetic + # 'running' row (which has no live pid) to 'orphaned' between the + # interleavings, so the guarded publish/stop UPDATEs would see a non-running + # row and the test would flake. + manager.dispatcher_stop.set() + if manager.dispatcher_thread is not None: + manager.dispatcher_thread.join(timeout=5) stamp = "2026-01-01T00:00:00Z" - manager.db.execute( - "INSERT INTO jobs(job_id, command, status, created_at, updated_at, stdout_path, stderr_path, events_path) VALUES (?, ?, 'running', ?, ?, ?, ?, ?)", - ("job_pid_race", "sleep 30", stamp, stamp, "out", "err", "events"), - ) - manager.db.commit() + with manager.db_lock: + manager.db.execute( + "INSERT INTO jobs(job_id, command, status, created_at, updated_at, stdout_path, stderr_path, events_path) VALUES (?, ?, 'running', ?, ?, ?, ?, ?)", + ("job_pid_race", "sleep 30", stamp, stamp, "out", "err", "events"), + ) + manager.db.commit() + + def guarded(fn): + # Surface worker-thread failures (e.g. a direct DB write racing the + # manager's background threads and raising "database is locked") instead + # of letting the thread die silently and the assertion see a stale row. + def wrapper(): + try: + fn() + except BaseException as exc: # noqa: BLE001 - re-raised below + errors.append(exc) + + return wrapper + + def join_all(threads): + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=10) + assert not any(thread.is_alive() for thread in threads), "worker threads did not finish" + if errors: + raise errors[0] + try: + errors: list[BaseException] = [] barrier = threading.Barrier(2) intent_done = threading.Event() def intent_wins(): barrier.wait() - manager.db.execute("UPDATE jobs SET stop_requested_at=? WHERE job_id=? AND status='running'", (stamp, "job_pid_race")) - manager.db.commit() + with manager.db_lock: + manager.db.execute("UPDATE jobs SET stop_requested_at=? WHERE job_id=? AND status='running'", (stamp, "job_pid_race")) + manager.db.commit() intent_done.set() def publish_after_intent(): @@ -216,15 +251,13 @@ def publish_after_intent(): intent_done.wait(timeout=2) assert not _publish_workload(manager, "job_pid_race", 123) - threads = [threading.Thread(target=intent_wins), threading.Thread(target=publish_after_intent)] - for thread in threads: - thread.start() - for thread in threads: - thread.join() + join_all([threading.Thread(target=guarded(intent_wins)), threading.Thread(target=guarded(publish_after_intent))]) assert manager._row("SELECT pid FROM jobs WHERE job_id=?", ("job_pid_race",))["pid"] is None - manager.db.execute("UPDATE jobs SET stop_requested_at=NULL WHERE job_id=?", ("job_pid_race",)) - manager.db.commit() + with manager.db_lock: + manager.db.execute("UPDATE jobs SET stop_requested_at=NULL WHERE job_id=?", ("job_pid_race",)) + manager.db.commit() + barrier = threading.Barrier(2) published = threading.Event() @@ -236,14 +269,11 @@ def publish_before_intent(): def intent_after_publish(): barrier.wait() published.wait(timeout=2) - manager.db.execute("UPDATE jobs SET stop_requested_at=? WHERE job_id=? AND status='running'", (stamp, "job_pid_race")) - manager.db.commit() + with manager.db_lock: + manager.db.execute("UPDATE jobs SET stop_requested_at=? WHERE job_id=? AND status='running'", (stamp, "job_pid_race")) + manager.db.commit() - threads = [threading.Thread(target=publish_before_intent), threading.Thread(target=intent_after_publish)] - for thread in threads: - thread.start() - for thread in threads: - thread.join() + join_all([threading.Thread(target=guarded(publish_before_intent)), threading.Thread(target=guarded(intent_after_publish))]) row = manager._row("SELECT pid, stop_requested_at FROM jobs WHERE job_id=?", ("job_pid_race",)) assert row["pid"] == 456 and row["stop_requested_at"] == stamp finally: diff --git a/tests/test_vanth.py b/tests/test_vanth.py index a24ae1d..b191f6c 100644 --- a/tests/test_vanth.py +++ b/tests/test_vanth.py @@ -2,7 +2,6 @@ import datetime import json import sqlite3 -import subprocess import sys import threading import time @@ -13,8 +12,11 @@ from vanth.server import JobManager, normalize_event_payload, now_iso, parse_agent_event_line +import shellcmd + + def cmd(code: str) -> str: - return subprocess.list2cmdline([sys.executable, "-c", code]) + return shellcmd.join([sys.executable, "-c", code]) def run(coro): @@ -911,7 +913,7 @@ async def main(): "print('AGENT_EVENT ' + json.dumps({'type': 'metric', 'metric': {'loss': 0.5}}), flush=True)\n" ) job = await manager.start( - subprocess.list2cmdline([sys.executable, str(script)]), + shellcmd.join([sys.executable, str(script)]), policy={"retention": {"events_seconds": 1, "metrics_seconds": 1}}, ) await manager.wait(job["job_id"], ["completed"], timeout_seconds=30) @@ -1361,8 +1363,16 @@ async def main(): manager._recover_stale_launch_claims() status = manager.status(job["job_id"])["status"] assert status == "orphaned", f"stale claim should be recovered, got {status}" - # The recovery emits an orphaned event so waits/wake targets fire. - events = manager.events(job["job_id"], types=["orphaned"], limit=10)["events"] + # The recovery emits an orphaned event so waits/wake targets fire; + # poll briefly since a concurrent dispatcher pass may have won the + # recovery and the event write can land a tick later under load. + deadline = time.monotonic() + 5 + events = [] + while time.monotonic() < deadline: + events = manager.events(job["job_id"], types=["orphaned"], limit=10)["events"] + if events: + break + time.sleep(0.05) assert events, "stale-claim recovery must emit an orphaned event" # The recovered job is runnable again. assert manager.prepare_launch(job["job_id"]) is not None @@ -1518,6 +1528,13 @@ class _RedirectSink: class RedirectHandler(http.server.BaseHTTPRequestHandler): def do_POST(self): # noqa: N802 + # Drain the request body before responding: closing with unread data + # in the socket buffer makes the OS send an RST, which surfaces as a + # connection-aborted error (WinError 10053) before the client can + # read the 302 — masking the redirect-refusal path under test. + length = int(self.headers.get("Content-Length", 0) or 0) + if length: + self.rfile.read(length) self.send_response(302) self.send_header("Location", f"http://127.0.0.1:{target_port}/dest") self.send_header("Content-Length", "0") @@ -1930,6 +1947,13 @@ async def main(): deadline = state.get("restart_after") assert deadline is not None + # Stop the maintenance loop: it could otherwise re-claim the due + # restart and relaunch the job while we assert the abandoned-claim + # recovery outcome (a dispatcher race on fast runners). + manager.dispatcher_stop.set() + if manager.dispatcher_thread is not None: + manager.dispatcher_thread.join(timeout=5) + launch = manager._claim_due_restart(job["job_id"], deadline) assert launch is not None token = launch["claim_token"] diff --git a/uv.lock b/uv.lock index 0724de3..987008c 100644 --- a/uv.lock +++ b/uv.lock @@ -876,7 +876,7 @@ wheels = [ [[package]] name = "vanth" -version = "1.9.0" +version = "1.9.1" source = { editable = "." } dependencies = [ { name = "loguru" },