fix(cron): honor workdir for no_agent job scripts

cron/scheduler.py

@@ -1545,7 +1545,7 @@ def _get_script_timeout() -> int:
     return _DEFAULT_SCRIPT_TIMEOUT
 
 
-def _run_job_script(script_path: str) -> tuple[bool, str]:
+def _run_job_script(script_path: str, cwd: Optional[str] = None) -> tuple[bool, str]:
     """Execute a cron job's data-collection script and capture its output.
 
     Scripts must reside within HERMES_HOME/scripts/.  Both relative and
@@ -1571,6 +1571,14 @@ def _run_job_script(script_path: str) -> tuple[bool, str]:
         script_path: Path to the script.  Relative paths are resolved
             against HERMES_HOME/scripts/.  Absolute and ~-prefixed paths
             are also validated to ensure they stay within the scripts dir.
+        cwd: Optional working directory for the script subprocess (a job's
+            configured ``workdir``).  When set and it exists, the script runs
+            with this as its cwd so relative paths resolve against the job's
+            project dir; otherwise the script runs from its own directory
+            (``HERMES_HOME/scripts/``), the historical default.  Passing it here
+            is required because ``subprocess.run(cwd=...)`` sets the child's cwd
+            absolutely — a parent-process ``os.chdir()`` would be overridden and
+            silently ignored.
 
     Returns:
         (success, output) — on failure *output* contains the error message so the
@@ -1631,12 +1639,13 @@ def _run_job_script(script_path: str) -> tuple[bool, str]:
         from tools.environments.local import _sanitize_subprocess_env
 
         popen_kwargs = {"creationflags": windows_hide_flags()} if sys.platform == "win32" else {}
+        run_cwd = cwd if (cwd and Path(cwd).is_dir()) else str(path.parent)
         result = subprocess.run(
             argv,
             capture_output=True,
             text=True,
             timeout=script_timeout,
-            cwd=str(path.parent),
+            cwd=run_cwd,
             env=_sanitize_subprocess_env(os.environ.copy()),
             **popen_kwargs,
         )
@@ -2004,25 +2013,21 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
             return False, "", "", err
 
         # Apply workdir if configured — lets scripts use predictable relative
-        # paths. For no_agent jobs this is just the subprocess cwd (not an
-        # agent TERMINAL_CWD bridge).
+        # paths. For no_agent jobs this is the script's subprocess cwd (not an
+        # agent TERMINAL_CWD bridge). Pass it straight to _run_job_script: a
+        # parent-process os.chdir() would be overridden by the explicit cwd= in
+        # subprocess.run (so the workdir was silently ignored), and chdir is
+        # process-global — unsafe while other jobs run in parallel.
         _job_workdir = (job.get("workdir") or "").strip() or None
-        _prior_cwd = None
-        if _job_workdir and Path(_job_workdir).is_dir():
-            _prior_cwd = os.getcwd()
-            try:
-                os.chdir(_job_workdir)
-            except OSError:
-                _prior_cwd = None
+        if _job_workdir and not Path(_job_workdir).is_dir():
+            logger.warning(
+                "Job '%s': configured workdir %r no longer exists — "
+                "running script without it",
+                job_id, _job_workdir,
+            )
+            _job_workdir = None
 
-        try:
-            ok, output = _run_job_script(script_path)
-        finally:
-            if _prior_cwd is not None:
-                try:
-                    os.chdir(_prior_cwd)
-                except OSError:
-                    pass
+        ok, output = _run_job_script(script_path, cwd=_job_workdir)
 
         now_iso = _hermes_now().strftime("%Y-%m-%d %H:%M:%S")
 

tests/cron/test_cron_workdir.py

@@ -390,3 +390,114 @@ def test_no_workdir_leaves_terminal_cwd_untouched(self, monkeypatch):
         # And after run_job completes, it's still the sentinel (nothing
         # overwrote or cleared it).
         assert os.environ["TERMINAL_CWD"] == before
+
+
+# ---------------------------------------------------------------------------
+# scheduler._run_job_script: subprocess cwd (job workdir) + no_agent path
+# ---------------------------------------------------------------------------
+
+def _seed_probe_script(sched, tmp_path, monkeypatch,
+                       body="import os\nprint(os.getcwd())\n"):
+    """Point HERMES_HOME at a temp dir and drop a script that prints its cwd.
+
+    Returns the resolved scripts dir (the historical default cwd).
+    """
+    from pathlib import Path
+
+    home = tmp_path / "home"
+    scripts = home / "scripts"
+    scripts.mkdir(parents=True, exist_ok=True)
+    (scripts / "probe.py").write_text(body, encoding="utf-8")
+    monkeypatch.setattr(sched, "_hermes_home", home)
+    return Path(scripts).resolve()
+
+
+class TestRunJobScriptCwd:
+    """_run_job_script must run the subprocess in the given cwd when provided.
+
+    Regression: the no_agent path used os.chdir(workdir) to set the script's
+    working directory, but subprocess.run() is invoked with an explicit cwd=
+    argument that overrides the parent process's cwd — so the configured
+    workdir was silently ignored and scripts always ran from HERMES_HOME/
+    scripts.  These tests fail before the fix (they'd land in the scripts dir)
+    and pass after it.
+    """
+
+    def test_honors_cwd_when_provided(self, tmp_path, monkeypatch):
+        from pathlib import Path
+        import cron.scheduler as sched
+        _seed_probe_script(sched, tmp_path, monkeypatch)
+        workdir = tmp_path / "project"
+        workdir.mkdir()
+
+        ok, output = sched._run_job_script("probe.py", cwd=str(workdir))
+        assert ok is True, output
+        assert Path(output.strip()).resolve() == workdir.resolve()
+
+    def test_defaults_to_scripts_dir_without_cwd(self, tmp_path, monkeypatch):
+        from pathlib import Path
+        import cron.scheduler as sched
+        scripts = _seed_probe_script(sched, tmp_path, monkeypatch)
+
+        ok, output = sched._run_job_script("probe.py")
+        assert ok is True, output
+        assert Path(output.strip()).resolve() == scripts
+
+    def test_falls_back_when_cwd_missing(self, tmp_path, monkeypatch):
+        from pathlib import Path
+        import cron.scheduler as sched
+        scripts = _seed_probe_script(sched, tmp_path, monkeypatch)
+        missing = tmp_path / "never-created"
+
+        ok, output = sched._run_job_script("probe.py", cwd=str(missing))
+        assert ok is True, output
+        # A non-existent cwd is ignored; the script still runs (scripts dir).
+        assert Path(output.strip()).resolve() == scripts
+
+
+class TestNoAgentScriptWorkdir:
+    """End-to-end: a no_agent job's script executes in the configured workdir."""
+
+    def test_no_agent_script_runs_in_workdir(self, tmp_path, monkeypatch):
+        from pathlib import Path
+        import cron.scheduler as sched
+        scripts = _seed_probe_script(sched, tmp_path, monkeypatch)
+        workdir = tmp_path / "project"
+        workdir.mkdir()
+
+        job = {
+            "id": "j1", "name": "wd", "no_agent": True,
+            "script": "probe.py", "workdir": str(workdir),
+        }
+        success, _doc, response, error = sched.run_job(job)
+        assert success is True, f"error={error!r}"
+        assert Path(response.strip()).resolve() == workdir.resolve()
+        assert Path(response.strip()).resolve() != scripts
+
+    def test_no_agent_without_workdir_uses_scripts_dir(self, tmp_path, monkeypatch):
+        from pathlib import Path
+        import cron.scheduler as sched
+        scripts = _seed_probe_script(sched, tmp_path, monkeypatch)
+
+        job = {
+            "id": "j2", "name": "no-wd", "no_agent": True,
+            "script": "probe.py",
+        }
+        success, _doc, response, error = sched.run_job(job)
+        assert success is True, f"error={error!r}"
+        assert Path(response.strip()).resolve() == scripts
+
+    def test_no_agent_vanished_workdir_falls_back(self, tmp_path, monkeypatch):
+        """A workdir removed after job creation is skipped, not fatal."""
+        from pathlib import Path
+        import cron.scheduler as sched
+        scripts = _seed_probe_script(sched, tmp_path, monkeypatch)
+        gone = tmp_path / "removed"  # never created
+
+        job = {
+            "id": "j3", "name": "gone-wd", "no_agent": True,
+            "script": "probe.py", "workdir": str(gone),
+        }
+        success, _doc, response, error = sched.run_job(job)
+        assert success is True, f"error={error!r}"
+        assert Path(response.strip()).resolve() == scripts