fix(cron): parallelize script jobs with workdirs

cron/scheduler.py

@@ -20,6 +20,7 @@
 import subprocess
 import sys
 import threading
+from datetime import datetime
 
 # fcntl is Unix-only; on Windows use msvcrt for file locking
 try:
@@ -2010,7 +2011,9 @@ 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 | Path | 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
@@ -2036,6 +2039,9 @@ 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.  This never changes the
+            scheduler process cwd, so independent script jobs can run safely
+            in parallel.
 
     Returns:
         (success, output) — on failure *output* contains the error message so the
@@ -2066,6 +2072,10 @@ def _run_job_script(script_path: str) -> tuple[bool, str]:
     if not path.is_file():
         return False, f"Script path is not a file: {path}"
 
+    subprocess_cwd = Path(cwd).expanduser().resolve() if cwd else path.parent
+    if not subprocess_cwd.is_dir():
+        return False, f"Script working directory is not a directory: {subprocess_cwd}"
+
     script_timeout = _get_script_timeout()
 
     # Pick an interpreter by extension.  Bash for .sh/.bash, Python for
@@ -2101,7 +2111,7 @@ def _run_job_script(script_path: str) -> tuple[bool, str]:
             capture_output=True,
             text=True,
             timeout=script_timeout,
-            cwd=str(path.parent),
+            cwd=str(subprocess_cwd),
             env=_sanitize_subprocess_env(os.environ.copy()),
             **popen_kwargs,
         )
@@ -2527,26 +2537,17 @@ def run_job(
             logger.error("Job '%s': %s", job_id, err)
             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).
+        # A no-agent workdir is subprocess-local.  Never call os.chdir() here:
+        # process-global cwd mutation would force unrelated script jobs through
+        # the sequential pool and can starve high-frequency watchdogs.
         _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
+        if _job_workdir and not Path(_job_workdir).is_dir():
+            logger.warning(
+                "Cron job '%s' workdir no longer exists; using the script directory",
+                job_name,
+            )
+            _job_workdir = None
+        ok, output = _run_job_script(script_path, cwd=_job_workdir)
 
         now_iso = _hermes_now().strftime("%Y-%m-%d %H:%M:%S")
 
@@ -3346,6 +3347,37 @@ def _teardown_cron_agent(agent, job_id: str) -> None:
         logger.debug("Job '%s': failed to reap stale auxiliary clients: %s", job_id, e)
 
 
+def _dispatch_metadata(job: dict) -> dict[str, object]:
+    started = _hermes_now()
+    scheduled_at = job.get("scheduled_at") or job.get("next_run_at")
+    lag_seconds: float | None = None
+    if scheduled_at:
+        try:
+            scheduled_dt = datetime.fromisoformat(str(scheduled_at).replace("Z", "+00:00"))
+            if scheduled_dt.tzinfo is None:
+                scheduled_dt = scheduled_dt.replace(tzinfo=started.tzinfo)
+            lag_seconds = max(0.0, (started - scheduled_dt.astimezone(started.tzinfo)).total_seconds())
+        except (TypeError, ValueError):
+            lag_seconds = None
+    return {
+        "scheduled_at": scheduled_at,
+        "dispatch_started_at": started.isoformat(),
+        "dispatch_lag_seconds": lag_seconds,
+    }
+
+
+def _annotate_dispatch_metadata(output: str, metadata: dict[str, object]) -> str:
+    lines = [
+        f"**Scheduled At:** {metadata.get('scheduled_at') or 'unknown'}",
+        f"**Dispatch Started At:** {metadata['dispatch_started_at']}",
+        f"**Dispatch Lag Seconds:** {metadata.get('dispatch_lag_seconds') if metadata.get('dispatch_lag_seconds') is not None else 'unknown'}",
+    ]
+    if output.startswith("# Cron Job:"):
+        first, separator, rest = output.partition("\n")
+        return f"{first}\n\n" + "\n".join(lines) + (f"\n{rest}" if separator else "\n")
+    return "\n".join(lines) + "\n\n" + output
+
+
 def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) -> bool:
     """Run ONE due job end-to-end: execute → save output → deliver → mark.
 
@@ -3376,6 +3408,15 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) -
             )
             return True  # not an error — already handled/removed
 
+        dispatch_metadata = _dispatch_metadata(job)
+        logger.info(
+            "Job '%s' dispatch scheduled_at=%s dispatch_started_at=%s dispatch_lag_seconds=%s",
+            job.get("name", job["id"]),
+            dispatch_metadata["scheduled_at"],
+            dispatch_metadata["dispatch_started_at"],
+            dispatch_metadata["dispatch_lag_seconds"],
+        )
+
         # Run the job under the profile's secret scope. get_secret() fails
         # closed outside a scope once profile isolation is in play (multiple
         # gateway profiles / room→profile multiplexing), and cron fires from
@@ -3424,6 +3465,7 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) -
         # swallow the error and leak the agent's subprocesses/clients (#10200).
         delivery_error = None
         try:
+            output = _annotate_dispatch_metadata(output, dispatch_metadata)
             output_file = save_job_output(job["id"], output)
             if verbose:
                 logger.info("Output saved to: %s", output_file)
@@ -3557,6 +3599,7 @@ def tick(verbose: bool = True, adapters=None, loop=None, sync: bool = True) -> i
         # bumping next_run_at forward so the grace window never expires.
         # mark_job_run() overwrites next_run_at on completion.
         for job in due_jobs:
+            job.setdefault("scheduled_at", job.get("next_run_at"))
             advance_next_run(job["id"])
 
         # Resolve max parallel workers: env var > config.yaml > unbounded.
@@ -3593,14 +3636,23 @@ def _process_job(job: dict) -> bool:
             body."""
             return run_one_job(job, adapters=adapters, loop=loop, verbose=verbose)
 
-        # Partition due jobs: those with a per-job workdir mutate
+        # Partition due jobs: agent jobs with a per-job workdir mutate
         # os.environ["TERMINAL_CWD"] inside run_job, which is process-global, so
-        # they queue on the single-thread sequential pool to run one at a time.
+        # they queue on the single-thread sequential pool.  no_agent jobs pass
+        # workdir directly to subprocess.run(cwd=...) and are safe to parallelize.
         # That alone only keeps workdir jobs from overlapping EACH OTHER;
         # run_job's _terminal_cwd_lock is what additionally stops a concurrently
         # firing workdir-less parallel-pool job from observing the override.
-        sequential_jobs = [j for j in due_jobs if (j.get("workdir") or "").strip()]
-        parallel_jobs = [j for j in due_jobs if not (j.get("workdir") or "").strip()]
+        sequential_jobs = [
+            j
+            for j in due_jobs
+            if (j.get("workdir") or "").strip() and not j.get("no_agent")
+        ]
+        parallel_jobs = [
+            j
+            for j in due_jobs
+            if not ((j.get("workdir") or "").strip() and not j.get("no_agent"))
+        ]
 
         _results: list = []
         _all_futures: list = []

tests/cron/test_cron_workdir.py

@@ -390,3 +390,35 @@ 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
+
+    def test_no_agent_workdir_is_subprocess_cwd_without_process_chdir(
+        self, tmp_path, monkeypatch
+    ):
+        """Pure-script jobs must not mutate the scheduler process cwd."""
+        import os
+        import cron.scheduler as sched
+
+        hermes_home = tmp_path / "hermes-home"
+        scripts_dir = hermes_home / "scripts"
+        scripts_dir.mkdir(parents=True)
+        workdir = tmp_path / "project"
+        workdir.mkdir()
+        (scripts_dir / "show_cwd.py").write_text(
+            "from pathlib import Path\nprint(Path.cwd())\n"
+        )
+        monkeypatch.setattr(sched, "_get_hermes_home", lambda: hermes_home)
+
+        process_cwd = os.getcwd()
+        success, _doc, response, error = sched.run_job(
+            {
+                "id": "script-workdir",
+                "name": "script-workdir",
+                "no_agent": True,
+                "script": "show_cwd.py",
+                "workdir": str(workdir),
+            }
+        )
+
+        assert success is True, error
+        assert response == str(workdir.resolve())
+        assert os.getcwd() == process_cwd

tests/cron/test_parallel_pool.py

@@ -272,3 +272,48 @@ def test_get_sequential_pool_is_persistent(self):
 
         sched._shutdown_parallel_pool()
         assert sched._sequential_pool is None
+
+    def test_no_agent_workdir_job_uses_parallel_pool(self, tmp_path, monkeypatch):
+        """A script-only cwd is subprocess-local and must not queue behind agents."""
+        import concurrent.futures
+        import cron.scheduler as sched
+
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes-home"))
+        sched._running_job_ids.clear()
+
+        class ImmediatePool:
+            def __init__(self):
+                self.submissions = 0
+
+            def submit(self, fn):
+                self.submissions += 1
+                future = concurrent.futures.Future()
+                try:
+                    future.set_result(fn())
+                except Exception as exc:  # pragma: no cover - assertion aid
+                    future.set_exception(exc)
+                return future
+
+        sequential = ImmediatePool()
+        parallel = ImmediatePool()
+        job = {
+            "id": "parallel-script-workdir",
+            "name": "parallel-script-workdir",
+            "schedule": "every 5m",
+            "enabled": True,
+            "next_run_at": "2020-01-01T00:00:00",
+            "deliver": "local",
+            "no_agent": True,
+            "script": "watchdog.py",
+            "workdir": str(tmp_path),
+        }
+
+        monkeypatch.setattr(sched, "get_due_jobs", lambda: [job])
+        monkeypatch.setattr(sched, "advance_next_run", lambda *_a, **_kw: None)
+        monkeypatch.setattr(sched, "_get_sequential_pool", lambda: sequential)
+        monkeypatch.setattr(sched, "_get_parallel_pool", lambda _workers: parallel)
+        monkeypatch.setattr(sched, "run_one_job", lambda *_a, **_kw: True)
+
+        assert sched.tick(verbose=False, sync=True) == 1
+        assert sequential.submissions == 0
+        assert parallel.submissions == 1

tests/cron/test_run_one_job.py

@@ -10,6 +10,8 @@
 the extraction didn't change `tick`'s behavior); the rest unit-test the
 extracted helper directly.
 """
+from datetime import datetime, timezone
+
 import cron.scheduler as s
 
 
@@ -66,6 +68,39 @@ def test_run_one_job_success_sequence(monkeypatch):
     assert calls[-1] == ("mark", "j2", True)
 
 
+def test_run_one_job_records_dispatch_lag_in_saved_output(monkeypatch):
+    """Saved artifacts expose the scheduled time, dispatch time, and queue lag."""
+    fixed_now = datetime(2026, 7, 9, 12, 5, tzinfo=timezone.utc)
+    saved = {}
+
+    monkeypatch.setattr(s, "_hermes_now", lambda: fixed_now)
+    monkeypatch.setattr(s, "claim_dispatch", lambda _job_id: True)
+    monkeypatch.setattr(
+        s,
+        "run_job",
+        lambda *_a, **_k: (True, "# Cron Job: test\n\nbody", "final", None),
+    )
+    monkeypatch.setattr(
+        s,
+        "save_job_output",
+        lambda _job_id, output: saved.setdefault("output", output) or "/tmp/out",
+    )
+    monkeypatch.setattr(s, "_deliver_result", lambda *_a, **_k: None)
+    monkeypatch.setattr(s, "mark_job_run", lambda *_a, **_k: None)
+
+    assert s.run_one_job(
+        {
+            "id": "lag-job",
+            "name": "lag-job",
+            "scheduled_at": "2026-07-09T12:00:00+00:00",
+        }
+    ) is True
+
+    assert "**Scheduled At:** 2026-07-09T12:00:00+00:00" in saved["output"]
+    assert "**Dispatch Started At:** 2026-07-09T12:05:00+00:00" in saved["output"]
+    assert "**Dispatch Lag Seconds:** 300.0" in saved["output"]
+
+
 def test_run_one_job_silent_skips_delivery(monkeypatch):
     """A [SILENT] final response saves output + marks the run but does NOT
     deliver."""

tests/cron/test_scheduler.py

@@ -2549,7 +2549,12 @@ def test_output_saved_even_when_delivery_suppressed(self):
             save_mock.return_value = "/tmp/out.md"
             from cron.scheduler import tick
             tick(verbose=False)
-        save_mock.assert_called_once_with("monitor-job", "# full output")
+        save_mock.assert_called_once()
+        saved_output = save_mock.call_args.args[1]
+        assert "**Scheduled At:** unknown" in saved_output
+        assert "**Dispatch Started At:**" in saved_output
+        assert "**Dispatch Lag Seconds:** unknown" in saved_output
+        assert saved_output.endswith("# full output")
         deliver_mock.assert_not_called()
 
     def test_whitespace_only_response_is_marked_failed_not_delivered(self):