fix(cron): run scripts from job workdir

cron/scheduler.py

@@ -199,6 +199,18 @@ def _resolve_cron_enabled_toolsets(job: dict, cfg: dict) -> list[str] | None:
         )
         return None
 
+
+def _job_script_cwd(job: dict) -> str | None:
+    """Return the configured cron job workdir for script subprocesses, if valid."""
+    workdir = (job.get("workdir") or "").strip()
+    if not workdir:
+        return None
+    path = Path(workdir).expanduser()
+    if path.is_dir():
+        return str(path.resolve())
+    return None
+
+
 # Valid delivery platforms — used to validate user-supplied platform names
 # in cron delivery targets, preventing env var enumeration via crafted names.
 _KNOWN_DELIVERY_PLATFORMS = frozenset({
@@ -1781,7 +1793,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: str | None = 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
@@ -1807,6 +1819,8 @@ 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 subprocess working directory.  When omitted, scripts
+            run from their own directory for backwards compatibility.
 
     Returns:
         (success, output) — on failure *output* contains the error message so the
@@ -1863,6 +1877,8 @@ def _run_job_script(script_path: str) -> tuple[bool, str]:
     else:
         argv = [sys.executable, str(path)]
 
+    run_cwd = str(Path(cwd).expanduser().resolve()) if cwd else str(path.parent)
+
     try:
         from tools.environments.local import _sanitize_subprocess_env
 
@@ -1872,7 +1888,7 @@ def _run_job_script(script_path: str) -> tuple[bool, str]:
             capture_output=True,
             text=True,
             timeout=script_timeout,
-            cwd=str(path.parent),
+            cwd=run_cwd,
             env=_sanitize_subprocess_env(os.environ.copy()),
             **popen_kwargs,
         )
@@ -2288,23 +2304,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
         # 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).
-        _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
-
-        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_script_cwd(job))
 
         now_iso = _hermes_now().strftime("%Y-%m-%d %H:%M:%S")
 
@@ -2387,7 +2387,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
     prerun_script = None
     script_path = job.get("script")
     if script_path:
-        prerun_script = _run_job_script(script_path)
+        prerun_script = _run_job_script(script_path, cwd=_job_script_cwd(job))
         _ran_ok, _script_output = prerun_script
         if _ran_ok and not _parse_wake_gate(_script_output):
             logger.info(

tests/cron/test_cron_no_agent.py

@@ -210,6 +210,38 @@ def test_run_job_no_agent_success_returns_script_stdout(hermes_env):
     assert "RAM 92% on host" in doc
 
 
+def test_run_job_no_agent_script_uses_workdir(hermes_env, tmp_path):
+    """no_agent script jobs should run from the configured project workdir."""
+    from cron.jobs import create_job
+    from cron.scheduler import run_job
+
+    workdir = tmp_path / "project"
+    workdir.mkdir()
+    (workdir / "marker.txt").write_text("present\n")
+    script_path = hermes_env / "scripts" / "cwd_probe.py"
+    script_path.write_text(
+        "from pathlib import Path\n"
+        "import os\n"
+        "print(f'cwd={os.getcwd()}')\n"
+        "print(f'marker={Path(\"marker.txt\").read_text().strip()}')\n"
+    )
+
+    job = create_job(
+        prompt=None,
+        schedule="every 5m",
+        script="cwd_probe.py",
+        no_agent=True,
+        deliver="local",
+        workdir=str(workdir),
+    )
+    success, doc, final_response, error = run_job(job)
+    assert success is True
+    assert error is None
+    assert f"cwd={workdir}" in final_response
+    assert "marker=present" in final_response
+    assert f"cwd={workdir}" in doc
+
+
 def test_run_job_no_agent_empty_output_is_silent(hermes_env):
     """Empty stdout → SILENT_MARKER, which suppresses delivery downstream."""
     from cron.jobs import create_job

tests/cron/test_scheduler.py

@@ -2786,14 +2786,33 @@ def test_wake_true_runs_agent_with_injected_output(self):
         assert success is True
         assert err is None
 
+    def test_wake_script_uses_workdir(self, tmp_path):
+        """Agent pre-run scripts should inherit the configured job workdir."""
+        import cron.scheduler as scheduler
+
+        workdir = tmp_path / "project"
+        workdir.mkdir()
+        agent = MagicMock()
+        agent.run_conversation = MagicMock(return_value={
+            "final_response": "ok", "messages": []
+        })
+        job = self._make_job()
+        job["workdir"] = str(workdir)
+        with patch.object(scheduler, "_run_job_script",
+                          return_value=(True, "regular output")) as script_fn, \
+             patch("run_agent.AIAgent", return_value=agent):
+            scheduler.run_job(job)
+
+        script_fn.assert_called_once_with("check.py", cwd=str(workdir.resolve()))
+
     def test_script_runs_only_once_on_wake(self):
         """Wake-true path must not re-run the script inside _build_job_prompt
         (script would execute twice otherwise, wasting work and risking
         double-side-effects)."""
         import cron.scheduler as scheduler
 
         call_count = 0
-        def _script_stub(path):
+        def _script_stub(path, **kwargs):
             nonlocal call_count
             call_count += 1
             return (True, "regular output")