fix(agent): supervise jobs and query process state safely

agent/runtime/cli/async_jobs.py

@@ -1,20 +1,21 @@
 from __future__ import annotations
 
 import json
-import os
-import signal
+import re
 import subprocess
 import sys
 import time
 from pathlib import Path
 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,66 +58,102 @@ 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()),
             "stderr_path": str(stderr_path.resolve()),
             "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]:

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])))

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"]