"""Tests for per-job workdir support in cron jobs. Covers: - jobs.create_job: param plumbing, validation, default-None preserved - jobs._normalize_workdir: absolute / relative / missing / file-not-dir - jobs.update_job: set, clear, re-validate - tools.cronjob_tools.cronjob: create + update JSON round-trip, schema includes workdir, _format_job exposes it when set - scheduler.tick(): partitions workdir jobs off the thread pool, restores TERMINAL_CWD in finally, honours the env override during run_job """ from __future__ import annotations import sys import pytest @pytest.fixture() def tmp_cron_dir(tmp_path, monkeypatch): """Isolate cron job storage into a temp dir so tests don't stomp on real jobs.""" monkeypatch.setattr("cron.jobs.CRON_DIR", tmp_path / "cron") monkeypatch.setattr("cron.jobs.JOBS_FILE", tmp_path / "cron" / "jobs.json") monkeypatch.setattr("cron.jobs.OUTPUT_DIR", tmp_path / "cron" / "output") return tmp_path # --------------------------------------------------------------------------- # jobs._normalize_workdir # --------------------------------------------------------------------------- class TestNormalizeWorkdir: def test_empty_string_returns_none(self): from cron.jobs import _normalize_workdir assert _normalize_workdir("") is None assert _normalize_workdir(" ") is None def test_absolute_existing_dir_returns_resolved_str(self, tmp_path): from cron.jobs import _normalize_workdir result = _normalize_workdir(str(tmp_path)) assert result == str(tmp_path.resolve()) @pytest.mark.platforms("linux") def test_tilde_expands(self, tmp_path, monkeypatch): from cron.jobs import _normalize_workdir # expanduser keys off USERPROFILE on native Windows, HOME elsewhere. home_var = "USERPROFILE" if sys.platform == "win32" else "HOME" monkeypatch.setenv(home_var, str(tmp_path)) result = _normalize_workdir("~") assert result == str(tmp_path.resolve()) def test_relative_path_rejected(self): from cron.jobs import _normalize_workdir with pytest.raises(ValueError, match="absolute path"): _normalize_workdir("some/relative/path") def test_missing_dir_rejected(self, tmp_path): from cron.jobs import _normalize_workdir missing = tmp_path / "does-not-exist" with pytest.raises(ValueError, match="does not exist"): _normalize_workdir(str(missing)) def test_file_not_dir_rejected(self, tmp_path): from cron.jobs import _normalize_workdir f = tmp_path / "file.txt" f.write_text("hi") with pytest.raises(ValueError, match="not a directory"): _normalize_workdir(str(f)) # --------------------------------------------------------------------------- # jobs.create_job and update_job # --------------------------------------------------------------------------- class TestCreateJobWorkdir: def test_workdir_stored_when_set(self, tmp_cron_dir): from cron.jobs import create_job, get_job job = create_job( prompt="hello", schedule="every 1h", workdir=str(tmp_cron_dir), ) stored = get_job(job["id"]) assert stored["workdir"] == str(tmp_cron_dir.resolve()) def test_create_rejects_invalid_workdir(self, tmp_cron_dir): from cron.jobs import create_job with pytest.raises(ValueError): create_job( prompt="hello", schedule="every 1h", workdir="not/absolute", ) class TestUpdateJobWorkdir: def test_set_workdir_via_update(self, tmp_cron_dir): from cron.jobs import create_job, get_job, update_job job = create_job(prompt="x", schedule="every 1h") update_job(job["id"], {"workdir": str(tmp_cron_dir)}) assert get_job(job["id"])["workdir"] == str(tmp_cron_dir.resolve()) def test_clear_workdir_with_none(self, tmp_cron_dir): from cron.jobs import create_job, get_job, update_job job = create_job( prompt="x", schedule="every 1h", workdir=str(tmp_cron_dir) ) update_job(job["id"], {"workdir": None}) assert get_job(job["id"])["workdir"] is None def test_update_rejects_invalid_workdir(self, tmp_cron_dir): from cron.jobs import create_job, update_job job = create_job(prompt="x", schedule="every 1h") with pytest.raises(ValueError): update_job(job["id"], {"workdir": "nope/relative"}) # --------------------------------------------------------------------------- # tools.cronjob_tools: end-to-end JSON round-trip # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # scheduler.tick(): workdir jobs use the parallel execution lane # --------------------------------------------------------------------------- class TestTickWorkdirPartition: """Workdir is per execution, so it must not force a global serial lane.""" def test_workdir_jobs_overlap_on_parallel_pool(self, tmp_path, monkeypatch): import cron.scheduler as sched import threading workdir_a = tmp_path / "a" workdir_b = tmp_path / "b" workdir_a.mkdir() workdir_b.mkdir() jobs = [ {"id": "a", "name": "A", "workdir": str(workdir_a)}, {"id": "b", "name": "B", "workdir": str(workdir_b)}, ] monkeypatch.setattr(sched, "get_due_jobs", lambda: jobs) monkeypatch.setattr(sched, "claim_job_for_fire", lambda *_a, **_kw: True) monkeypatch.setattr(sched, "_maybe_run_worktree_maintenance", lambda: None) barrier = threading.Barrier(2, timeout=5) calls: list[tuple[str, str]] = [] calls_lock = threading.Lock() def fake_run_job(job, *, defer_agent_teardown=None, **_kw): with calls_lock: calls.append((job["id"], threading.current_thread().name)) barrier.wait() return True, "output", "response", None monkeypatch.setattr(sched, "run_job", fake_run_job) monkeypatch.setattr(sched, "save_job_output", lambda _jid, _o: None) monkeypatch.setattr(sched, "mark_job_run", lambda *_a, **_kw: None) monkeypatch.setattr(sched, "_deliver_result", lambda *_a, **_kw: None) assert sched.tick(verbose=False, sync=True) == 2 assert {job_id for job_id, _thread in calls} == {"a", "b"} assert all(thread.startswith("cron-parallel") for _job, thread in calls) # --------------------------------------------------------------------------- # scheduler.run_job: per-task cwd + skip_context_files wiring # --------------------------------------------------------------------------- class TestRunJobTerminalCwd: """ run_job binds workdir to its unique task CWD without mutating ambient TERMINAL_CWD, and clears the task record in finally — even on error. AIAgent is stubbed so no real API call happens. """ @staticmethod def _install_stubs(monkeypatch, observed: dict): """Patch enough of run_job's deps that it executes without real creds.""" import os import sys import cron.scheduler as sched from cron import scheduler_delivery as sched_delivery class FakeAgent: def __init__(self, **kwargs): observed["skip_context_files"] = kwargs.get("skip_context_files") observed["load_soul_identity"] = kwargs.get("load_soul_identity") observed["terminal_cwd_during_init"] = os.environ.get( "TERMINAL_CWD", "_UNSET_" ) def run_conversation(self, *_a, task_id=None, **_kw): from tools.terminal_tool import get_session_cwd observed["task_id"] = task_id observed["task_cwd_during_run"] = get_session_cwd(task_id) observed["terminal_cwd_during_run"] = os.environ.get( "TERMINAL_CWD", "_UNSET_" ) return {"final_response": "done", "messages": [{"role": "assistant", "content": "done"}]} def get_activity_summary(self): return {"seconds_since_activity": 0.0} fake_mod = type(sys)("run_agent") fake_mod.AIAgent = FakeAgent monkeypatch.setitem(sys.modules, "run_agent", fake_mod) # Bypass the real provider resolver — it reads ~/.hermes and credentials. from hermes_cli import runtime_provider as _rtp monkeypatch.setattr( _rtp, "resolve_runtime_provider", lambda **_kw: { "provider": "test", "api_key": "k", "base_url": "http://test.local", "api_mode": "chat_completions", }, ) # Stub scheduler helpers that would otherwise hit the filesystem / config. monkeypatch.setattr(sched, "_build_job_prompt", lambda job, prerun_script=None, **kw: "hi") monkeypatch.setattr(sched_delivery, "_resolve_origin", lambda job: None) monkeypatch.setattr(sched, "_resolve_delivery_target", lambda job: None) monkeypatch.setattr(sched, "_resolve_cron_enabled_toolsets", lambda job, cfg: None) # Unlimited inactivity so the poll loop returns immediately. monkeypatch.setenv("HERMES_CRON_TIMEOUT", "0") # run_job calls load_dotenv(~/.hermes/.env, override=True), which will # happily clobber TERMINAL_CWD out from under us if the real user .env # has TERMINAL_CWD set (common on dev boxes). Stub it out. import dotenv monkeypatch.setattr(dotenv, "load_dotenv", lambda *_a, **_kw: True) def test_no_workdir_leaves_terminal_cwd_untouched(self, monkeypatch): """When workdir is absent, run_job must not touch TERMINAL_CWD at all — whatever value was present before the call should be present after. We don't assert on the *content* of TERMINAL_CWD (other tests in the same process may leave it set to something like '.'); we just check it's unchanged by run_job. """ import os import cron.scheduler as sched # Pin TERMINAL_CWD to a sentinel via monkeypatch so we control both # the before-value and the after-value regardless of cross-test state. monkeypatch.setenv("TERMINAL_CWD", "/cron-test-sentinel") before = os.environ["TERMINAL_CWD"] observed: dict = {} self._install_stubs(monkeypatch, observed) job = { "id": "xyz", "name": "no-wd-job", "workdir": None, "schedule_display": "manual", } success, *_ = sched.run_job(job) assert success is True # Feature is OFF — skip_context_files stays True. assert observed["skip_context_files"] is True # Cron still forces SOUL.md identity even when cwd context files stay off. assert observed["load_soul_identity"] is True # TERMINAL_CWD saw the same value during init as it had before. assert observed["terminal_cwd_during_init"] == before # And after run_job completes, it's still the sentinel (nothing # overwrote or cleared it). assert os.environ["TERMINAL_CWD"] == before def test_workdir_is_bound_to_unique_task_without_mutating_process_env( self, monkeypatch, tmp_path ): import os import cron.scheduler as sched from tools.terminal_tool import get_session_cwd baseline = str(tmp_path / "baseline") workdir = tmp_path / "project" (tmp_path / "baseline").mkdir() workdir.mkdir() monkeypatch.setenv("TERMINAL_CWD", baseline) observed: dict = {} self._install_stubs(monkeypatch, observed) success, *_ = sched.run_job( { "id": "cwd-bound", "name": "cwd-bound", "workdir": str(workdir), "schedule_display": "manual", } ) assert success is True assert observed["skip_context_files"] is False assert observed["task_id"].startswith("cron:cwd-bound:") assert observed["task_cwd_during_run"] == str(workdir) assert observed["terminal_cwd_during_run"] == baseline assert os.environ["TERMINAL_CWD"] == baseline assert get_session_cwd(observed["task_id"]) is None def test_agent_prerun_script_receives_configured_workdir( self, monkeypatch, tmp_path ): import cron.scheduler as sched workdir = tmp_path / "project" workdir.mkdir() observed: dict = {} self._install_stubs(monkeypatch, observed) def run_script(job, script_path, workdir=None, cancel_event=None): observed["script_workdir"] = workdir return True, '{"wakeAgent": false}' monkeypatch.setattr( sched, "_run_job_script_with_claim_heartbeat", run_script ) success, *_ = sched.run_job( { "id": "agent-script-workdir", "name": "agent-script-workdir", "prompt": "Review the project.", "script": "collect.py", "workdir": str(workdir), "schedule_display": "manual", } ) assert success is True assert observed["script_workdir"] == str(workdir) def test_build_job_prompt_inline_script_receives_configured_workdir(monkeypatch, tmp_path): """Callers that skip the wake-gate (no cached ``prerun_script``) run the script inline from ``_build_job_prompt``; that path must honour the job's workdir too.""" from cron import scheduler_prompt, scheduler_script workdir = tmp_path / "project" workdir.mkdir() observed: dict = {} def run_script(script_path, workdir=None, cancel_event=None, interpreter=None): observed["script_workdir"] = workdir return True, "collected data" monkeypatch.setattr(scheduler_script, "_run_job_script", run_script) prompt = scheduler_prompt._build_job_prompt( {"id": "inline", "name": "inline", "prompt": "Review.", "script": "collect.py", "workdir": str(workdir)}) assert observed["script_workdir"] == str(workdir) assert "collected data" in prompt