From fd35b7310a89420e66afc0e3a73603009b6586e3 Mon Sep 17 00:00:00 2001 From: abhim-dv Date: Fri, 11 Sep 2026 17:28:09 -0700 Subject: [PATCH 01/14] Enable Linux/macOS Python CI with POSIX-safe test commands The Python matrix ran on windows-latest only because workload commands were built with subprocess.list2cmdline (Windows quoting), which the POSIX shell rejects (syntax error near unexpected token). Fixing that and running the suite on Linux surfaced several portability issues, all addressed here. - tests/shellcmd.py: shlex.join on POSIX, list2cmdline on Windows; every test and dev-script command builder routes through it. pytest gains pythonpath=["tests"] so the helper imports from all test dirs. - .github/workflows/ci.yml: python matrix now windows-latest/ubuntu-latest/ macos-latest (macOS limited to one Python version for runner cost); the wheel smoke steps are POSIX-portable (bash arrays instead of GNU find -maxdepth). - job_wait returns the EARLIEST matching signal (terminal event, metric_ge threshold, or return_progress) by event sequence instead of always preferring the terminal event. - codex_pipe: a peer close is reported consistently (closed the connection) on read or write, and close() shuts a socket down before closing so a reader blocked in recv() wakes promptly on POSIX. - _orphaned_mcp_servers: POSIX detection matches only the Vanth MCP entrypoint (the vanth console script or python -m vanth.server), so --reap-orphans can no longer terminate unrelated processes (e.g. pytest inside the checkout). Orphan findings are advisory and no longer flip doctor's exit code. - Test hardening for POSIX: locked direct DB writes in the pid/publication race, stale portable-rename directory expectation, same-size capture mutation, and load-adjusted budgets for the cross-process burst and slow wake adapter. Version 1.9.1. Windows: 783 passed, 6 skipped. Linux (3.12): 787 passed, 2 skipped. go test ./... green. --- .github/workflows/ci.yml | 29 +++---- CHANGELOG.md | 34 +++++++++ pyproject.toml | 5 +- scripts/demo_jobs.py | 6 +- scripts/make_monitor_fixture.py | 6 +- src/vanth/codex_pipe.py | 22 +++++- src/vanth/server.py | 97 ++++++++++++++++++------ tests/artifacts/test_capture_security.py | 4 + tests/artifacts/test_operations.py | 6 +- tests/artifacts/test_review_fixes.py | 11 ++- tests/shellcmd.py | 34 +++++++++ tests/test_agent_features.py | 10 ++- tests/test_agent_logger.py | 6 +- tests/test_attribution_masking.py | 6 +- tests/test_backup.py | 6 +- tests/test_cli_qol.py | 27 ++++++- tests/test_codex_desktop.py | 7 +- tests/test_daemon.py | 5 +- tests/test_daemon_hardening.py | 5 +- tests/test_delivery_hardening.py | 6 +- tests/test_job_send.py | 6 +- tests/test_mcp_stdio.py | 9 ++- tests/test_probes.py | 6 +- tests/test_qol_mcp.py | 6 +- tests/test_qol_rerun_batch.py | 12 +-- tests/test_qol_ux_batch.py | 6 +- tests/test_qol_wait_tail.py | 6 +- tests/test_qol_wake_shorthand.py | 6 +- tests/test_queues.py | 6 +- tests/test_quotas_retention.py | 6 +- tests/test_release_workloads.py | 10 ++- tests/test_schedules.py | 6 +- tests/test_server_hardening.py | 5 +- tests/test_telemetry.py | 7 +- tests/test_v1_hardening.py | 67 ++++++++++------ tests/test_vanth.py | 8 +- uv.lock | 2 +- 37 files changed, 385 insertions(+), 121 deletions(-) create mode 100644 tests/shellcmd.py 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..85afa38 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,40 @@ 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. +- **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). +- **Orphaned-MCP reaping is safer.** POSIX detection now matches the actual + Vanth MCP entrypoint (`vanth` console script or `python -m vanth.server`) + rather than any process whose command line merely mentions a Vanth path, so + `vanth doctor --reap-orphans` can no longer terminate unrelated processes + (such as a `pytest` run inside the checkout). Orphan findings are reported as + 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/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/server.py b/src/vanth/server.py index 454495e..0419172 100644 --- a/src/vanth/server.py +++ b/src/vanth/server.py @@ -4040,22 +4040,35 @@ 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]]] = [] 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 { - "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), - } + metric_event = self._latest_metric_event(job_id, metric) + candidates.append(( + int((metric_event or {}).get("seq") or 0), + { + "result": "metric", + "job_id": job_id, + "status": self.status(job_id)["status"], + "metric": metric, + "threshold": threshold, + "value": value, + "event": metric_event, + }, + )) + break except RuntimeError: pass if return_progress and "progress" not in filters: @@ -4064,14 +4077,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"} @@ -5200,7 +5218,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 +6547,36 @@ def job_cleanup_preview(older_than_seconds: int) -> dict[str, Any]: return get_client().get("/cleanup/preview", {"older_than_seconds": older_than_seconds}) +def _is_vanth_mcp_command(command_line: str) -> bool: + """Whether a POSIX command line is a Vanth MCP stdio server. + + Matches only the supported launch shapes — the ``vanth`` console script + (executable basename ``vanth``) or ``python -m vanth.server`` — so an + unrelated process that merely mentions a Vanth path (for example another + ``pytest`` running inside the checkout, whose command line contains + ``.../vanth-ci/...``) is never mistaken for an MCP server and reaped. + """ + tokens = command_line.split() + if not tokens: + return False + + def _base(token: str) -> str: + return token.replace("\\", "/").rstrip("/").rsplit("/", 1)[-1].lower() + + for i in range(len(tokens) - 1): + if tokens[i] == "-m" and tokens[i + 1] in {"vanth.server", "vanth.mcp"}: + return True + names = {"vanth", "vanth.exe", "vanth-script.py"} + if _base(tokens[0]) in names: + return True + # Console script run through the interpreter: ``python .../vanth``. Require + # the interpreter as argv0 so a shell running ``vanth status`` (whose argv0 + # is the shell) is not mistaken for the stdio server. + if len(tokens) >= 2 and _base(tokens[0]).startswith("python") and _base(tokens[1]) in names: + return True + return False + + def _orphaned_mcp_servers() -> list[dict[str, Any]]: """Find MCP stdio server processes whose launching client is gone. @@ -6568,19 +6619,19 @@ def _orphaned_mcp_servers() -> list[dict[str, Any]]: } ) 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, } 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..5c004e5 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,28 @@ 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), + # 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), + ], +) +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_hardening.py b/tests/test_daemon_hardening.py index 90dfc34..23204b7 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)) @@ -198,7 +201,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..8d86efe 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,8 +11,11 @@ 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): 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..1e5ed5d 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": [ @@ -138,7 +141,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 +187,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_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..b8c39b1 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: @@ -184,7 +187,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..e100f78 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): 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..436bee5 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: @@ -196,19 +199,44 @@ 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") 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 +244,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 +262,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..d6bfc68 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) 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" }, From 80e027ef8dea739f868eeb142e02d7f988d7c2a2 Mon Sep 17 00:00:00 2001 From: abhim-dv Date: Fri, 11 Sep 2026 18:04:10 -0700 Subject: [PATCH 02/14] Fix cross-platform CI findings (1.9.1 follow-up) The first matrix run (ubuntu/macos/windows) passed on Linux and exposed: - Windows: `on_failure` emitted the failure_threshold event before persisting failure_streak, so a waiter could read the policy state without the key. Persist the streak before reacting. - macOS: directory materialization used /dev/fd, which is unreliable for creating nested entries under a directory fd; use the dev/inode-checked plain-path fallback there. - Windows: the stop-after-restart test now waits (bounded) for OS process teardown instead of asserting immediately. - Daemon-start test helpers now surface the daemon's stderr and log tail on failure, to diagnose the remaining macOS daemon-start failures. --- CHANGELOG.md | 6 ++++++ src/vanth/artifacts/operations.py | 15 ++++++--------- src/vanth/server.py | 12 +++++++++--- tests/test_daemon_discovery.py | 14 +++++++++++++- tests/test_daemon_hardening.py | 12 +++++++++++- tests/test_server_hardening.py | 5 +++++ 6 files changed, 50 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 85afa38..9d03789 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,12 @@ the fixes below. 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.** The `on_failure` policy now persists the updated + `failure_streak` before emitting the `failure_threshold` event, so a waiter + that observes the event always sees the updated policy state. +- **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. - **Orphaned-MCP reaping is safer.** POSIX detection now matches the actual Vanth MCP entrypoint (`vanth` console script or `python -m vanth.server`) rather than any process whose command line merely mentions a Vanth path, so diff --git a/src/vanth/artifacts/operations.py b/src/vanth/artifacts/operations.py index ba3cc23..8050b3f 100644 --- a/src/vanth/artifacts/operations.py +++ b/src/vanth/artifacts/operations.py @@ -1168,15 +1168,12 @@ def _materialize_dir( 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 diff --git a/src/vanth/server.py b/src/vanth/server.py index 0419172..0d8e58f 100644 --- a/src/vanth/server.py +++ b/src/vanth/server.py @@ -943,12 +943,18 @@ def _watch_on_failure(self, row: sqlite3.Row, on_failure: dict[str, Any]) -> Non 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 + 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"] + if react: + state["reacted_at_streak"] = new_streak + # 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). self._save_policy_state(job_id, state) + if react: + self._react_to_failure(row, on_failure, new_streak) 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 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 23204b7..5c94551 100644 --- a/tests/test_daemon_hardening.py +++ b/tests/test_daemon_hardening.py @@ -45,7 +45,17 @@ 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): diff --git a/tests/test_server_hardening.py b/tests/test_server_hardening.py index e100f78..9ef6af6 100644 --- a/tests/test_server_hardening.py +++ b/tests/test_server_hardening.py @@ -113,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() From 1b3f38737c05a9b472d4570a8d1810f8005b00e6 Mon Sep 17 00:00:00 2001 From: abhim-dv Date: Fri, 11 Sep 2026 18:38:51 -0700 Subject: [PATCH 03/14] Fix remaining CI findings: daemon startup, launch claims, relay timing - daemon: override server_bind to skip http.server's socket.getfqdn reverse DNS lookup, which blocked daemon startup on the macOS runner (daemon.json never written, /health never served, MCP clients timed out). - server: _claim_launch now clears worker_pid, so stale-claim recovery cannot skip an abandoned launch whose previous runner pid is still visible. - tests: relax the relay-activity watchdog idle threshold from 50ms (below macOS scheduling jitter) and make the blocking-relay test tear down via parent liveness instead of relying on the idle timer. --- CHANGELOG.md | 6 ++++++ src/vanth/daemon.py | 11 +++++++++++ src/vanth/server.py | 2 +- tests/test_process_watch.py | 28 ++++++++++++---------------- 4 files changed, 30 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9d03789..981b11d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,12 @@ the fixes below. - **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. +- **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. - **Orphaned-MCP reaping is safer.** POSIX detection now matches the actual Vanth MCP entrypoint (`vanth` console script or `python -m vanth.server`) rather than any process whose command line merely mentions a Vanth path, so 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/server.py b/src/vanth/server.py index 0d8e58f..12dbacb 100644 --- a/src/vanth/server.py +++ b/src/vanth/server.py @@ -3256,7 +3256,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 diff --git a/tests/test_process_watch.py b/tests/test_process_watch.py index bafb2ed..566fbd4 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,6 +175,7 @@ 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). @@ -187,24 +189,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.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, "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() From 6a2420b02f5ebd17ad428918384c2893de9c532d Mon Sep 17 00:00:00 2001 From: abhim-dv Date: Fri, 11 Sep 2026 18:53:46 -0700 Subject: [PATCH 04/14] Harden idle reaper freshness window and restart-claim test race - process_watch: reset the idle timer when activity occurred within the idle threshold, not within the (tiny) sampling interval. A relay whose notify cadence is coarser than the sampler was mis-reaped on macOS runners. - tests: stop the maintenance loop before asserting the abandoned-claim recovery outcome, so the restart policy cannot re-claim and relaunch the job mid-assertion (a fast-runner race). --- CHANGELOG.md | 3 +++ src/vanth/process_watch.py | 7 +++++-- tests/test_vanth.py | 7 +++++++ 3 files changed, 15 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 981b11d..9fae63f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,9 @@ the fixes below. - **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). - **Orphaned-MCP reaping is safer.** POSIX detection now matches the actual Vanth MCP entrypoint (`vanth` console script or `python -m vanth.server`) rather than any process whose command line merely mentions a Vanth path, so diff --git a/src/vanth/process_watch.py b/src/vanth/process_watch.py index a1a42a9..230308d 100644 --- a/src/vanth/process_watch.py +++ b/src/vanth/process_watch.py @@ -277,8 +277,11 @@ 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: + # 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). + if now - tracker.last_activity() < idle: idle_since = None elif traffic(): idle_since = None diff --git a/tests/test_vanth.py b/tests/test_vanth.py index d6bfc68..eee817f 100644 --- a/tests/test_vanth.py +++ b/tests/test_vanth.py @@ -1932,6 +1932,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"] From 36573d1db5a295d237ef57f01274a69551b5b615 Mon Sep 17 00:00:00 2001 From: abhim-dv Date: Fri, 11 Sep 2026 19:10:47 -0700 Subject: [PATCH 05/14] Retry contended event writes and de-flake the MCP status assertion - server: extend _emit's lock-contention retry budget (10 attempts, capped backoff) so reader-parsed AGENT_EVENTs are not dropped under a concurrent job burst on Windows. - tests: accept running-or-completed for the status snapshot taken when a job_wait returns (a short job can finish first on a loaded runner). --- CHANGELOG.md | 3 +++ src/vanth/server.py | 6 +++--- tests/test_mcp_stdio.py | 4 +++- 3 files changed, 9 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9fae63f..a449deb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -41,6 +41,9 @@ the fixes below. - **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). +- **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.** POSIX detection now matches the actual Vanth MCP entrypoint (`vanth` console script or `python -m vanth.server`) rather than any process whose command line merely mentions a Vanth path, so diff --git a/src/vanth/server.py b/src/vanth/server.py index 12dbacb..b945a13 100644 --- a/src/vanth/server.py +++ b/src/vanth/server.py @@ -2038,17 +2038,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: diff --git a/tests/test_mcp_stdio.py b/tests/test_mcp_stdio.py index 1e5ed5d..ef27782 100644 --- a/tests/test_mcp_stdio.py +++ b/tests/test_mcp_stdio.py @@ -106,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" From 406487ed78f2ecd074f52edcb7977c66dc5378c6 Mon Sep 17 00:00:00 2001 From: abhim-dv Date: Fri, 11 Sep 2026 19:27:49 -0700 Subject: [PATCH 06/14] Raise the delivery-wait budget for Windows CI load The quick-job retry delivery test waited 5s for a stop/retry/deliver sequence (retry_delay_seconds=1, 0.2s dispatcher poll). Under the Windows runner's full suite load that occasionally exceeded 5s; use a 20s ceiling (it is a max wait, so passing runs are unaffected). --- tests/test_delivery_hardening.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_delivery_hardening.py b/tests/test_delivery_hardening.py index 8d86efe..f867311 100644 --- a/tests/test_delivery_hardening.py +++ b/tests/test_delivery_hardening.py @@ -18,7 +18,7 @@ def cmd(code: str) -> str: 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: From cf67e308d49759838d5e4987e9be4ea81a2293ca Mon Sep 17 00:00:00 2001 From: abhim-dv Date: Fri, 11 Sep 2026 19:47:47 -0700 Subject: [PATCH 07/14] Count every pending failed execution in the failure streak The on_failure watcher compared only the latest 'failed' event to the stored one and incremented the streak by one. With a fast restart (backoff 0) two failures can land between watcher ticks, undercounting the streak (Windows CI saw a final streak of 2 instead of 3). Count all failed events after the last counted one instead. --- CHANGELOG.md | 7 ++++--- src/vanth/server.py | 26 +++++++++++++++++++++++--- 2 files changed, 27 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a449deb..6183e0a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,9 +26,10 @@ the fixes below. 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.** The `on_failure` policy now persists the updated - `failure_streak` before emitting the `failure_threshold` event, so a waiter - that observes the event always sees the updated policy state. +- **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 since the last watcher tick (fast restarts + with backoff 0 no longer undercount the streak). - **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. diff --git a/src/vanth/server.py b/src/vanth/server.py index b945a13..937b94b 100644 --- a/src/vanth/server.py +++ b/src/vanth/server.py @@ -935,14 +935,34 @@ def _watch_on_failure(self, row: sqlite3.Row, on_failure: dict[str, Any]) -> Non # stored one so every execution (including restarts) advances the # streak exactly once. 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"]: + stored_failure_id = state.get("last_failure_event_id") + if stored_failure_id == last_terminal["event_id"]: return # already counted this failed run - new_streak = streak + 1 + # Count EVERY failed execution since the last counted one, not just + # the latest: with a fast restart (backoff 0) two failures can land + # between watcher ticks, and incrementing by one per tick undercounts + # the streak (Windows CI observed a final streak of 2, not 3). + pending = 1 + if stored_failure_id: + stored = self.db.execute( + "SELECT seq FROM events WHERE job_id=? AND event_id=?", (job_id, stored_failure_id) + ).fetchone() + if stored is not None: + pending = max( + 1, + int( + self.db.execute( + "SELECT COUNT(*) FROM events WHERE job_id=? AND type='failed' AND seq > ?", + (job_id, int(stored["seq"])), + ).fetchone()[0] + ), + ) + new_streak = streak + 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"] From f1bebf678857195c00a561c472b18c65cdf93d78 Mon Sep 17 00:00:00 2001 From: abhim-dv Date: Fri, 11 Sep 2026 20:04:57 -0700 Subject: [PATCH 08/14] Drain the POST body in the webhook-redirect test server The redirect handler replied 302 without reading the request body; closing the socket with unread data makes the OS send an RST, which surfaced as WinError 10053 before the client could read the 302, masking the redirect-refusal path on Windows. Read Content-Length bytes first. --- tests/test_vanth.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/test_vanth.py b/tests/test_vanth.py index eee817f..4f4231a 100644 --- a/tests/test_vanth.py +++ b/tests/test_vanth.py @@ -1520,6 +1520,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") From e754cb7c361988b2ace6e000fec9fa11c734e4bb Mon Sep 17 00:00:00 2001 From: abhim-dv Date: Fri, 11 Sep 2026 20:22:41 -0700 Subject: [PATCH 09/14] Stop the dispatcher in the pid/stop-intent interleaving test The test inserts a synthetic 'running' row with no live pid; the background reconciler could orphan it between the guarded publish/stop UPDATEs, so a guarded write saw a non-running row and returned False (Windows CI flake). Stop the maintenance loop before the interleavings. --- tests/test_v1_hardening.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/test_v1_hardening.py b/tests/test_v1_hardening.py index 436bee5..6624dab 100644 --- a/tests/test_v1_hardening.py +++ b/tests/test_v1_hardening.py @@ -198,6 +198,13 @@ 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" with manager.db_lock: manager.db.execute( From 60623548d95ac6f10632d1ba8e5fbeb15777fef1 Mon Sep 17 00:00:00 2001 From: abhim-dv Date: Fri, 11 Sep 2026 20:37:11 -0700 Subject: [PATCH 10/14] Poll for the orphaned event after stale-claim recovery The recovery can be won by a concurrent dispatcher pass and the orphaned event write can land a tick later under load; assert on a bounded poll instead of an immediate read (Windows CI saw the status recovered but the event not yet visible). --- tests/test_vanth.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/tests/test_vanth.py b/tests/test_vanth.py index 4f4231a..b191f6c 100644 --- a/tests/test_vanth.py +++ b/tests/test_vanth.py @@ -1363,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 From 87492f2ed94463cc6fddf3fdbda21dbdf5204009 Mon Sep 17 00:00:00 2001 From: abhim-dv Date: Fri, 11 Sep 2026 20:55:25 -0700 Subject: [PATCH 11/14] Raise the daemon-hardening HTTP client timeout A POST /jobs that spawns a runner can exceed 5s on a loaded Windows runner; use 20s (connection-refused is immediate, so the health-poll deadline still holds). --- tests/test_daemon_hardening.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/test_daemon_hardening.py b/tests/test_daemon_hardening.py index 5c94551..1ff743b 100644 --- a/tests/test_daemon_hardening.py +++ b/tests/test_daemon_hardening.py @@ -59,7 +59,9 @@ def start_daemon(tmp_path, max_request_bytes=1024 * 1024): 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()) From 6eeda4f859b4bec906ba115eae4ecd5dfc91f687 Mon Sep 17 00:00:00 2001 From: abhim-dv Date: Fri, 11 Sep 2026 21:09:23 -0700 Subject: [PATCH 12/14] Relax the slow-wake-adapter elapsed bound for slow runners Spawning 10 adapter subprocesses CPU-starves a 2-core Windows runner, so job completion took >4s even without waiting on the adapters. The real regression (completion waiting on the 5s adapters) needs ~15s and trips the 10s waiter, so bound just under it. --- tests/test_release_workloads.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/tests/test_release_workloads.py b/tests/test_release_workloads.py index b8c39b1..3a85b5d 100644 --- a/tests/test_release_workloads.py +++ b/tests/test_release_workloads.py @@ -131,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() From 7f06959adc440e6ed6fcf5e1e1571de8d4839665 Mon Sep 17 00:00:00 2001 From: abhim-dv Date: Mon, 14 Sep 2026 10:59:33 -0700 Subject: [PATCH 13/14] Fix review findings: wait cursor, MCP reaper safety, failure accounting Independent review of the POSIX-CI/1.9.1 changes surfaced several issues: - job_wait: metric candidates now honour since_event_id (a satisfied threshold at/before the cursor no longer starves every later signal) and every threshold is considered instead of breaking on dictionary order. - Orphaned-MCP detection: reject false positives (python unrelated.py -m vanth.server, bash -lc ..., vanth CLI subcommands) since the reaper kills each match; Windows uses Get-CimInstance command lines (the WMIC CSV parse misread alphabetic columns and could not establish identity). - on_failure: count failed executions over a bounded seq interval (no under/over-count), and mark the reaction complete only after it succeeds so a crash/error retries it. - macOS materialization: re-verify parent + staging descriptors before publication and fail closed on an ancestor swap. - Idle reaper: measure from last activity so the timeout is idle, not ~2x. - Tests: the blocking-relay test now blocks longer than the idle threshold; the delivery-retry test asserts the retrying state was actually observed. --- CHANGELOG.md | 36 ++-- src/vanth/artifacts/operations.py | 27 +++ src/vanth/process_watch.py | 12 +- src/vanth/server.py | 266 ++++++++++++++++++++---------- tests/test_delivery_hardening.py | 5 +- tests/test_process_watch.py | 8 +- 6 files changed, 244 insertions(+), 110 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6183e0a..7a0c2af 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,7 +20,10 @@ the fixes below. `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. + 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 @@ -28,11 +31,16 @@ the fixes below. 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 since the last watcher tick (fast restarts - with backoff 0 no longer undercount the streak). + 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). The reaction is marked complete only after it succeeds, so a + daemon crash or action error retries it instead of dropping it. - **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. + unreliable for creating nested entries under a directory fd; the parent and + staging descriptors are re-verified immediately before publication and the + operation fails closed if an ancestor was swapped mid-write. - **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. @@ -41,16 +49,22 @@ the fixes below. 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). + 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.** POSIX detection now matches the actual - Vanth MCP entrypoint (`vanth` console script or `python -m vanth.server`) - rather than any process whose command line merely mentions a Vanth path, so - `vanth doctor --reap-orphans` can no longer terminate unrelated processes - (such as a `pytest` run inside the checkout). Orphan findings are reported as - an advisory warning and no longer flip `vanth doctor`'s exit code. +- **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. diff --git a/src/vanth/artifacts/operations.py b/src/vanth/artifacts/operations.py index 8050b3f..ee04c88 100644 --- a/src/vanth/artifacts/operations.py +++ b/src/vanth/artifacts/operations.py @@ -1162,6 +1162,7 @@ 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() @@ -1187,9 +1188,35 @@ 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 + # racing ancestor swap cannot publish a redirected or empty + # tree — fail closed instead. + 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/process_watch.py b/src/vanth/process_watch.py index 230308d..547cd0b 100644 --- a/src/vanth/process_watch.py +++ b/src/vanth/process_watch.py @@ -281,12 +281,16 @@ def _watch_loop( # 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). - if now - tracker.last_activity() < idle: - idle_since = None - elif traffic(): + # Measure from the last observed activity and seed ``idle_since`` + # with it, so the effective timeout is ``idle`` (not ~2x from + # starting a second window once the freshness window expires). + last_activity = tracker.last_activity() + if traffic(): + last_activity = now + if now - last_activity < idle: idle_since = None elif idle_since is None: - idle_since = now + idle_since = last_activity elif now - idle_since >= idle: on_exit() return diff --git a/src/vanth/server.py b/src/vanth/server.py index 937b94b..e50c200 100644 --- a/src/vanth/server.py +++ b/src/vanth/server.py @@ -927,54 +927,64 @@ 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, seq FROM events WHERE job_id=? AND type='failed' ORDER BY seq DESC LIMIT 1", (job_id,), ).fetchone() if last_terminal is None: return - stored_failure_id = state.get("last_failure_event_id") - if stored_failure_id == last_terminal["event_id"]: - return # already counted this failed run - # Count EVERY failed execution since the last counted one, not just - # the latest: with a fast restart (backoff 0) two failures can land - # between watcher ticks, and incrementing by one per tick undercounts - # the streak (Windows CI observed a final streak of 2, not 3). - pending = 1 - if stored_failure_id: - stored = self.db.execute( - "SELECT seq FROM events WHERE job_id=? AND event_id=?", (job_id, stored_failure_id) - ).fetchone() - if stored is not None: - pending = max( - 1, - int( - self.db.execute( - "SELECT COUNT(*) FROM events WHERE job_id=? AND type='failed' AND seq > ?", - (job_id, int(stored["seq"])), - ).fetchone()[0] - ), - ) - new_streak = streak + pending + last_seq = int(last_terminal["seq"]) + watermark = state.get("failure_streak_after_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"] - if react: - state["reacted_at_streak"] = new_streak + 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). + # commit and the state save saw no failure_streak). The reaction is + # marked complete only AFTER it succeeds, so a crash/error retries it. 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 @@ -982,10 +992,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: @@ -3722,7 +3738,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 @@ -4071,6 +4089,13 @@ def _wait( # 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: candidates.append(( int(events[0].get("seq") or 0), @@ -4080,21 +4105,29 @@ def _wait( try: for metric, threshold in metric_ge.items(): value = self._latest_metric_value(job_id, metric) - if value is not None and value >= threshold: - metric_event = self._latest_metric_event(job_id, metric) - candidates.append(( - int((metric_event or {}).get("seq") or 0), - { - "result": "metric", - "job_id": job_id, - "status": self.status(job_id)["status"], - "metric": metric, - "threshold": threshold, - "value": value, - "event": metric_event, - }, - )) - break + if value is None or value < threshold: + continue + # Cursor-aware: a threshold satisfied only by an event at + # or before since_event_id must not be returned again, or + # a caller that advances the cursor would be handed the + # same metric forever and never reach the terminal event. + # No early break: every satisfied threshold is a candidate + # and the earliest event wins (dict order must not decide). + metric_event = self._latest_metric_event_after(job_id, metric, since_seq) + if metric_event is None: + 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": metric_event, + }, + )) except RuntimeError: pass if return_progress and "progress" not in filters: @@ -4486,6 +4519,22 @@ 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_event_after( + self, job_id: str, metric: str, after_seq: int + ) -> dict[str, Any] | None: + """Latest metric event strictly after ``after_seq`` (events.seq).""" + with self.db_lock: + row = self.db.execute( + """ + SELECT e.* 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() + return self._event_dict(row) if row else None + 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. @@ -6573,33 +6622,64 @@ 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"} + + def _is_vanth_mcp_command(command_line: str) -> bool: - """Whether a POSIX command line is a Vanth MCP stdio server. + """Whether a command line is a Vanth MCP stdio server (not a CLI command). - Matches only the supported launch shapes — the ``vanth`` console script - (executable basename ``vanth``) or ``python -m vanth.server`` — so an - unrelated process that merely mentions a Vanth path (for example another - ``pytest`` running inside the checkout, whose command line contains - ``.../vanth-ci/...``) is never mistaken for an MCP server and reaped. + 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`` (the ``-m`` must be the + interpreter's first argument; ``python unrelated.py -m vanth.server`` and + ``bash -lc 'python -m vanth.server'`` are rejected), and + - the ``vanth`` console script (``vanth`` / ``python .../vanth``) run with + no CLI subcommand — ``vanth logs --follow`` is the CLI, not MCP. """ 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() - for i in range(len(tokens) - 1): - if tokens[i] == "-m" and tokens[i + 1] in {"vanth.server", "vanth.mcp"}: - return True - names = {"vanth", "vanth.exe", "vanth-script.py"} - if _base(tokens[0]) in names: - return True - # Console script run through the interpreter: ``python .../vanth``. Require - # the interpreter as argv0 so a shell running ``vanth status`` (whose argv0 - # is the shell) is not mistaken for the stdio server. - if len(tokens) >= 2 and _base(tokens[0]).startswith("python") and _base(tokens[1]) in names: - return True + def _is_cli_script(script_index: int) -> bool: + rest = tokens[script_index + 1:] + return bool(rest) and rest[0] in _VANTH_CLI_SUBCOMMANDS + + if _base(tokens[0]).startswith("python"): + # Walk the interpreter's own options to where the launching argument is: + # ``-m module`` (the supported MCP shape), ``-c payload`` (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": + return False + if tok in ("-X", "-W"): # options that consume a following value + i += 2 + continue + if tok.startswith("-"): + i += 1 + continue + 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 @@ -6619,29 +6699,37 @@ 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: @@ -6711,11 +6799,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/test_delivery_hardening.py b/tests/test_delivery_hardening.py index f867311..30b1e24 100644 --- a/tests/test_delivery_hardening.py +++ b/tests/test_delivery_hardening.py @@ -156,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_process_watch.py b/tests/test_process_watch.py index 566fbd4..c3bac63 100644 --- a/tests/test_process_watch.py +++ b/tests/test_process_watch.py @@ -178,10 +178,12 @@ def fake_exit(): 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) @@ -189,7 +191,7 @@ def relay_poll(): try: thread = threading.Thread( target=_watch_loop, - args=(os.getpid(), 0.005, 0.0, 0.5, fake_exit, tracker), + args=(os.getpid(), 0.005, 0.0, 0.1, fake_exit, tracker), kwargs={"traffic": lambda: 0, "alive": lambda: parent_alive["value"]}, daemon=True, ) From 3c54dd8900eed7ae7a525f058ecdcfe1b04bf738 Mon Sep 17 00:00:00 2001 From: abhim-dv Date: Mon, 14 Sep 2026 13:31:09 -0700 Subject: [PATCH 14/14] Fix second-review findings: legacy streak migration, matcher refusals, stdin activity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up review of the review fixes surfaced three regressions and two gaps: - on_failure: policy state written by 1.9.0 (last_failure_event_id, no seq watermark) was recounted from scratch on upgrade, double-counting failures. The legacy marker is now resolved to its job-local sequence before counting. - Idle reaper: a one-shot stdin traffic sample was only remembered for the iteration that observed it, so a process could be reaped ~two sampling intervals after traffic instead of one idle timeout later. The stdin timestamp is now retained across iterations. - MCP matcher: refuses ambiguous forms instead of guessing — quoted subcommands (vanth "logs" --follow), attached -c payloads, unknown interpreter options, and long options all reject; a quoted argv0 (Windows "C:\Program Files\...python.exe") is split off so genuine launches with spaces in the interpreter path still match. Missing an exotic launch is the safe failure mode for a reaper that kills what it accepts. - job_wait metric_ge: the threshold value and the returned event are read in one joined query, so a stale satisfying value can no longer be paired with a newer event that no longer satisfies the threshold. - macOS materialization: comment/CHANGELOG now state the residual race (transient swap restored before the check, staging-name replacement under an unchanged parent) instead of claiming full fail-closed coverage. - Matcher tests extended with the lookalike cases; reaction retry documented as at-least-once with the upgrade path noted. --- CHANGELOG.md | 18 ++++-- src/vanth/artifacts/operations.py | 8 ++- src/vanth/process_watch.py | 21 ++++--- src/vanth/server.py | 95 ++++++++++++++++++++++--------- tests/test_cli_qol.py | 7 +++ 5 files changed, 108 insertions(+), 41 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a0c2af..ef09f68 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,13 +34,21 @@ the fixes below. 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). The reaction is marked complete only after it succeeds, so a - daemon crash or action error retries it instead of dropping it. + 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 parent and - staging descriptors are re-verified immediately before publication and the - operation fails closed if an ancestor was swapped mid-write. + 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. diff --git a/src/vanth/artifacts/operations.py b/src/vanth/artifacts/operations.py index ee04c88..9c2ce44 100644 --- a/src/vanth/artifacts/operations.py +++ b/src/vanth/artifacts/operations.py @@ -1196,8 +1196,12 @@ def _materialize_dir( # 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 - # racing ancestor swap cannot publish a redirected or empty - # tree — fail closed instead. + # 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) diff --git a/src/vanth/process_watch.py b/src/vanth/process_watch.py index 547cd0b..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() @@ -281,16 +286,18 @@ def _watch_loop( # 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). - # Measure from the last observed activity and seed ``idle_since`` - # with it, so the effective timeout is ``idle`` (not ~2x from - # starting a second window once the freshness window expires). - last_activity = tracker.last_activity() + # ``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_activity = now - if now - last_activity < idle: + 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 = last_activity + 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 e50c200..14f7d4d 100644 --- a/src/vanth/server.py +++ b/src/vanth/server.py @@ -942,6 +942,17 @@ def _watch_on_failure(self, row: sqlite3.Row, on_failure: dict[str, Any]) -> Non return 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) @@ -980,6 +991,11 @@ def _watch_on_failure(self, row: sqlite3.Row, on_failure: dict[str, Any]) -> Non # 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) @@ -4104,17 +4120,10 @@ def _wait( if metric_ge: try: for metric, threshold in metric_ge.items(): - value = self._latest_metric_value(job_id, metric) - if value is None or value < threshold: - continue - # Cursor-aware: a threshold satisfied only by an event at - # or before since_event_id must not be returned again, or - # a caller that advances the cursor would be handed the - # same metric forever and never reach the terminal event. - # No early break: every satisfied threshold is a candidate - # and the earliest event wins (dict order must not decide). - metric_event = self._latest_metric_event_after(job_id, metric, since_seq) - if metric_event is None: + # 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), @@ -4519,21 +4528,29 @@ 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_event_after( + def _latest_metric_sample_after( self, job_id: str, metric: str, after_seq: int - ) -> dict[str, Any] | None: - """Latest metric event strictly after ``after_seq`` (events.seq).""" + ) -> 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.* FROM events e + 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() - return self._event_dict(row) if row else None + 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. @@ -6630,6 +6647,15 @@ def job_cleanup_preview(older_than_seconds: int) -> dict[str, Any]: _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). @@ -6638,13 +6664,25 @@ def _is_vanth_mcp_command(command_line: str) -> bool: reported process, so the matcher accepts only the exact supported launch shapes and rejects anything ambiguous: - - `` -m vanth.server`` / ``-m vanth.mcp`` (the ``-m`` must be the - interpreter's first argument; ``python unrelated.py -m vanth.server`` and - ``bash -lc 'python -m vanth.server'`` are rejected), and + - `` -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. """ - tokens = command_line.split() + 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 @@ -6654,26 +6692,29 @@ def _base(token: str) -> str: def _is_cli_script(script_index: int) -> bool: rest = tokens[script_index + 1:] - return bool(rest) and rest[0] in _VANTH_CLI_SUBCOMMANDS + # 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 where the launching argument is: - # ``-m module`` (the supported MCP shape), ``-c payload`` (never us), or - # the first non-option token (a script path — matched only when it is the + # 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": + if tok == "-c" or tok.startswith("-c"): return False - if tok in ("-X", "-W"): # options that consume a following value + if tok in _PY_VALUED_OPTIONS: i += 2 continue - if tok.startswith("-"): + 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 diff --git a/tests/test_cli_qol.py b/tests/test_cli_qol.py index 5c004e5..c716358 100644 --- a/tests/test_cli_qol.py +++ b/tests/test_cli_qol.py @@ -281,12 +281,19 @@ def test_doctor_reports_orphans_field(daemon): (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):