diff --git a/cron/scheduler.py b/cron/scheduler.py index 9e645d3728f3d..54d6fff1d1436 100644 --- a/cron/scheduler.py +++ b/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 = [] diff --git a/tests/cron/test_cron_workdir.py b/tests/cron/test_cron_workdir.py index 253bde4769bb3..33fa2a8c0159e 100644 --- a/tests/cron/test_cron_workdir.py +++ b/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 diff --git a/tests/cron/test_parallel_pool.py b/tests/cron/test_parallel_pool.py index 4c4d3f4887e83..23599a3dd00ee 100644 --- a/tests/cron/test_parallel_pool.py +++ b/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 diff --git a/tests/cron/test_run_one_job.py b/tests/cron/test_run_one_job.py index decb6c4e35ffc..0b6a927977f10 100644 --- a/tests/cron/test_run_one_job.py +++ b/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.""" diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py index 8cb81cb6177a8..1a46ec1f28b26 100644 --- a/tests/cron/test_scheduler.py +++ b/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):