From 4fe59ed3eb12b144d4d6be65cf22304a131ee317 Mon Sep 17 00:00:00 2001 From: onion-hong <2922707977@qq.com> Date: Tue, 8 Sep 2026 21:59:56 +0800 Subject: [PATCH] fix(agent): supervise jobs and query process state safely --- agent/runtime/cli/async_jobs.py | 129 +++++++++++++++++++++----------- agent/scripts/run_async_job.py | 29 +++++++ tests/test_agent_async_jobs.py | 104 +++++++++++++++++++++++++ 3 files changed, 218 insertions(+), 44 deletions(-) create mode 100644 agent/scripts/run_async_job.py create mode 100644 tests/test_agent_async_jobs.py diff --git a/agent/runtime/cli/async_jobs.py b/agent/runtime/cli/async_jobs.py index f8ad7759..70e6fa7f 100644 --- a/agent/runtime/cli/async_jobs.py +++ b/agent/runtime/cli/async_jobs.py @@ -1,8 +1,7 @@ from __future__ import annotations import json -import os -import signal +import re import subprocess import sys import time @@ -10,11 +9,13 @@ from typing import Any from uuid import uuid4 -from runtime.cli.contract import ensure_manifest_dir, json_safe, response, write_manifest +import psutil +from runtime.cli.contract import _atomic_write_json, ensure_manifest_dir, json_safe, response, write_manifest SKILL_ROOT = Path(__file__).resolve().parents[2] DISPATCHER = SKILL_ROOT / "scripts" / "run_yolo_master_skill.py" +WORKER = SKILL_ROOT / "scripts" / "run_async_job.py" def async_requested(request: dict[str, Any]) -> bool: @@ -32,17 +33,21 @@ def __init__(self, root: Path | None = None): def _jobs_dir(self, request: dict[str, Any] | None = None) -> Path: base = self.root or SKILL_ROOT / "logs" / "async-jobs" - base.mkdir(parents=True, exist_ok=True) - return base + return base.resolve() def _job_dir(self, job_id: str, request: dict[str, Any] | None = None) -> Path: - path = self._jobs_dir(request) / job_id - path.mkdir(parents=True, exist_ok=True) + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,127}", job_id): + raise ValueError("Invalid job_id") + base = self._jobs_dir(request) + path = (base / job_id).resolve() + if path.parent != base: + raise ValueError("Job path escapes the jobs directory") return path def submit(self, skill: str, request: dict[str, Any], callback_url: str | None = None) -> dict[str, Any]: job_id = uuid4().hex[:12] job_dir = self._job_dir(job_id, request) + job_dir.mkdir(parents=True, exist_ok=False) child_request = json_safe(request) child_request["request_id"] = f"{request.get('request_id', skill.replace('.', '-'))}-{job_id}" child_request.setdefault("policy", {}) @@ -53,22 +58,26 @@ def submit(self, skill: str, request: dict[str, Any], callback_url: str | None = stderr_path = job_dir / "stderr.log" request_path.write_text(json.dumps(child_request, ensure_ascii=False, indent=2), encoding="utf-8") - stdout_handle = stdout_path.open("a", encoding="utf-8") - stderr_handle = stderr_path.open("a", encoding="utf-8") - proc = subprocess.Popen( - [sys.executable, str(DISPATCHER), "--request", str(request_path)], - cwd=Path(__file__).resolve().parents[3], - stdout=stdout_handle, - stderr=stderr_handle, - start_new_session=True, - ) - stdout_handle.close() - stderr_handle.close() + with stdout_path.open("a", encoding="utf-8") as stdout_handle, stderr_path.open( + "a", encoding="utf-8" + ) as stderr_handle: + proc = subprocess.Popen( + [sys.executable, str(WORKER), str(job_dir), str(DISPATCHER)], + cwd=SKILL_ROOT.parent, + stdout=stdout_handle, + stderr=stderr_handle, + start_new_session=True, + ) + try: + created = psutil.Process(proc.pid).create_time() + except psutil.NoSuchProcess: + created = None # A fast worker may already have persisted its result. status = { "job_id": job_id, "skill": skill, "status": "running", "pid": proc.pid, + "process_create_time": created, "submitted_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "request_path": str(request_path.resolve()), "stdout_path": str(stdout_path.resolve()), @@ -76,43 +85,75 @@ def submit(self, skill: str, request: dict[str, Any], callback_url: str | None = "callback_url": callback_url, "progress_path": str((ensure_manifest_dir(child_request) / "progress.jsonl").resolve()), } - status_path.write_text(json.dumps(status, ensure_ascii=False, indent=2), encoding="utf-8") + _atomic_write_json(status_path, status) return {**status, "status_path": str(status_path.resolve())} + @staticmethod + def _process(status: dict[str, Any]) -> psutil.Process | None: + """Find the recorded process without signaling it or accepting a reused PID.""" + pid, created = status.get("pid"), status.get("process_create_time") + if not isinstance(pid, int) or isinstance(pid, bool) or pid <= 0 or created is None: + return None + process = psutil.Process(pid) + if process.create_time() != created or not process.is_running() or process.status() == psutil.STATUS_ZOMBIE: + return None + return process + def status(self, job_id: str, request: dict[str, Any] | None = None) -> dict[str, Any]: - status_path = self._job_dir(job_id, request) / "status.json" + """Read worker results or probe process identity; reads never create files or send signals.""" + job_dir = self._job_dir(job_id, request) + status_path = job_dir / "status.json" if not status_path.exists(): return {"job_id": job_id, "status": "missing"} status = json.loads(status_path.read_text(encoding="utf-8")) - pid = status.get("pid") - running = False - if isinstance(pid, int): - try: - os.kill(pid, 0) - running = True - except OSError: - running = False - if not running and status.get("status") == "running": - status["status"] = "completed" - status["completed_at"] = time.strftime("%Y-%m-%dT%H:%M:%S%z") - status_path.write_text(json.dumps(status, ensure_ascii=False, indent=2), encoding="utf-8") - return status + for name in ("cancelled.json", "result.json"): + result_path = job_dir / name + if result_path.exists(): + return {**status, **json.loads(result_path.read_text(encoding="utf-8"))} + try: + running = self._process(status) is not None + except psutil.NoSuchProcess: + running = False + except psutil.AccessDenied: + return {**status, "status": "unknown"} + # A missing process is not evidence of a successful task (including legacy records). + return {**status, "status": "running" if running else "unknown"} def cancel(self, job_id: str, request: dict[str, Any] | None = None) -> dict[str, Any]: + """Terminate only the recorded worker and its descendants, then persist cancellation.""" status = self.status(job_id, request) - pid = status.get("pid") - if status.get("status") != "running" or not isinstance(pid, int): + if status.get("status") != "running": return {**status, "cancelled": False} try: - os.killpg(pid, signal.SIGTERM) - except Exception: - os.kill(pid, signal.SIGTERM) - status["status"] = "cancelled" - status["cancelled"] = True - status["cancelled_at"] = time.strftime("%Y-%m-%dT%H:%M:%S%z") - status_path = self._job_dir(job_id, request) / "status.json" - status_path.write_text(json.dumps(status, ensure_ascii=False, indent=2), encoding="utf-8") - return status + process = self._process(status) + if process is None: + return {**status, "status": "unknown", "cancelled": False} + children = process.children(recursive=True) + for target in reversed(children): + try: + target.terminate() + except psutil.NoSuchProcess: + pass + # Let the supervisor reap exited children before terminating the supervisor itself. + _, alive = psutil.wait_procs(children, timeout=3) + try: + process.terminate() + except psutil.NoSuchProcess: + pass + alive.append(process) + for target in alive: + try: + target.kill() + except psutil.NoSuchProcess: + pass + _, alive = psutil.wait_procs(alive, timeout=3) + if alive: + return {**status, "status": "unknown", "cancelled": False} + except (psutil.NoSuchProcess, psutil.AccessDenied): + return {**status, "status": "unknown", "cancelled": False} + result = {"status": "cancelled", "cancelled": True, "cancelled_at": time.strftime("%Y-%m-%dT%H:%M:%S%z")} + _atomic_write_json(self._job_dir(job_id, request) / "cancelled.json", result) + return {**status, **result} def submit_async_skill(request: dict[str, Any]) -> dict[str, Any]: diff --git a/agent/scripts/run_async_job.py b/agent/scripts/run_async_job.py new file mode 100644 index 00000000..9f2630a5 --- /dev/null +++ b/agent/scripts/run_async_job.py @@ -0,0 +1,29 @@ +"""Supervise one Agent dispatcher and persist its exit status independently of callers.""" + +from __future__ import annotations + +import subprocess +import sys +import time +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from runtime.cli.contract import _atomic_write_json + + +def run(job_dir: Path, dispatcher: Path) -> int: + """Keep terminal results separate from submission metadata to avoid fast-exit races.""" + try: + child = subprocess.Popen([sys.executable, str(dispatcher), "--request", str(job_dir / "request.json")]) + returncode = child.wait() + result = {"status": "completed" if returncode == 0 else "failed", "returncode": returncode} + except (OSError, ValueError) as exc: + result = {"status": "failed", "returncode": None, "error": str(exc)} + result["completed_at"] = time.strftime("%Y-%m-%dT%H:%M:%S%z") + _atomic_write_json(job_dir / "result.json", result) + return 0 if result["status"] == "completed" else 1 + + +if __name__ == "__main__": + raise SystemExit(run(Path(sys.argv[1]), Path(sys.argv[2]))) diff --git a/tests/test_agent_async_jobs.py b/tests/test_agent_async_jobs.py new file mode 100644 index 00000000..cf2dd09b --- /dev/null +++ b/tests/test_agent_async_jobs.py @@ -0,0 +1,104 @@ +"""Cross-platform lifecycle checks using disposable workers, never user processes.""" + +import importlib +import json +import os +import sys +import time +from pathlib import Path +from unittest.mock import Mock + +import psutil +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "agent")) +jobs = importlib.import_module("runtime.cli.async_jobs") + + +def wait_terminal(manager, job_id): + """Wait a bounded interval for the supervised worker's atomic result.""" + for _ in range(100): + status = manager.status(job_id) + if status["status"] in {"completed", "failed", "cancelled"}: + return status + time.sleep(0.05) + pytest.fail(f"worker did not finish: {status}") + + +@pytest.mark.parametrize("code", [0, 7]) +def test_worker_records_exit_code(tmp_path, monkeypatch, code): + """Fast success and failure must not be overwritten by submission metadata.""" + dispatcher = tmp_path / "dispatcher.py" + dispatcher.write_text(f"raise SystemExit({code})") + monkeypatch.setattr(jobs, "DISPATCHER", dispatcher) + monkeypatch.setattr(jobs, "ensure_manifest_dir", lambda request: tmp_path) + manager = jobs.AsyncJobManager(tmp_path / "jobs") + submitted = manager.submit("fixture", {"policy": {}}) + status = wait_terminal(manager, submitted["job_id"]) + assert status["returncode"] == code + assert status["status"] == ("completed" if code == 0 else "failed") + + +def test_status_is_read_only_and_cancel_stops_worker(tmp_path, monkeypatch): + """Windows queries must leave a running worker alive; cancellation is explicit.""" + dispatcher = tmp_path / "dispatcher.py" + ready = tmp_path / "ready" + dispatcher.write_text(f"import time\nfrom pathlib import Path\nPath({str(ready)!r}).touch()\ntime.sleep(60)") + monkeypatch.setattr(jobs, "DISPATCHER", dispatcher) + monkeypatch.setattr(jobs, "ensure_manifest_dir", lambda request: tmp_path) + manager = jobs.AsyncJobManager(tmp_path / "jobs") + job = manager.submit("fixture", {"policy": {}}) + try: + for _ in range(100): + if ready.exists(): + break + time.sleep(0.05) + assert ready.exists() + with monkeypatch.context() as context: + context.setattr(os, "kill", Mock(side_effect=AssertionError("status must not send signals"))) + assert manager.status(job["job_id"])["status"] == "running" + assert psutil.Process(job["pid"]).is_running() + assert manager.cancel(job["job_id"])["cancelled"] + assert manager.status(job["job_id"])["status"] == "cancelled" + finally: + manager.cancel(job["job_id"]) + + +@pytest.mark.parametrize("job_id", ["../escape", "/absolute", "C:\\escape", "a/b", "a\\b", ".", ""]) +def test_invalid_job_id_cannot_create_paths(tmp_path, job_id): + """Reject traversal before any mkdir or status read.""" + root = tmp_path / "jobs" + with pytest.raises(ValueError): + jobs.AsyncJobManager(root).status(job_id) + assert not root.exists() + + +def test_missing_status_creates_nothing(tmp_path): + """Existing safe legacy IDs remain queryable without creating directories.""" + root = tmp_path / "jobs" + assert jobs.AsyncJobManager(root).status("missing-job-contract")["status"] == "missing" + assert not root.exists() + + +@pytest.mark.parametrize("created", [None, -1]) +def test_legacy_or_reused_pid_cannot_be_cancelled(tmp_path, created): + """A PID alone is not enough to authorize termination, even when it exists.""" + manager = jobs.AsyncJobManager(tmp_path) + path = manager._job_dir("fixture") + path.mkdir() + (path / "status.json").write_text( + json.dumps({"status": "running", "pid": os.getpid(), "process_create_time": created}) + ) + assert manager.status("fixture")["status"] == "unknown" + assert not manager.cancel("fixture")["cancelled"] + + +def test_access_denied_is_unknown(tmp_path, monkeypatch): + """Permission failure is neither a successful exit nor a cancellable process.""" + manager = jobs.AsyncJobManager(tmp_path) + path = manager._job_dir("fixture") + path.mkdir() + (path / "status.json").write_text(json.dumps({"status": "running", "pid": 123, "process_create_time": 1})) + monkeypatch.setattr(jobs.psutil, "Process", Mock(side_effect=psutil.AccessDenied(123))) + assert manager.status("fixture")["status"] == "unknown" + assert not manager.cancel("fixture")["cancelled"]