hermes-agent/tests/cron/test_cron_workdir.py

"""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):
        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