hermes-agent/tests/test_scratch_dir.py

"""Scratch dir contract: TMPDIR/TMP/TEMP follow HERMES_HOME/cache/scratch unless the user set them."""

import os
import stat
import subprocess
import sys
import time
from unittest.mock import patch

import pytest

from hermes_constants import apply_scratch_tmp_env, get_scratch_dir, prune_scratch_dir


def test_scratch_env_follows_home_and_respects_user_tmpdir(tmp_path):
    """Unset temp vars → scratch of env HERMES_HOME; a Hermes-exported value re-derives for a routed
    home; a user/OS-set value (macOS ``/var/folders``, ``%TEMP%``) is never touched."""
    home_a, home_b = tmp_path / "a", tmp_path / "b"
    env = {"HERMES_HOME": str(home_a)}
    assert apply_scratch_tmp_env(env) is True
    scratch_a = str(home_a / "cache" / "scratch")
    assert env["TMPDIR"] == env["TMP"] == env["TEMP"] == env["HERMES_SCRATCH_DIR"] == scratch_a
    assert os.path.isdir(scratch_a)

    env["HERMES_HOME"] = str(home_b)  # child served under another profile's home
    assert apply_scratch_tmp_env(env) is True
    assert env["TMPDIR"] == str(home_b / "cache" / "scratch")

    user_env = {"HERMES_HOME": str(home_a), "TMPDIR": "/var/folders/zz"}
    assert apply_scratch_tmp_env(user_env) is False
    assert user_env["TMPDIR"] == "/var/folders/zz" and "HERMES_SCRATCH_DIR" not in user_env
    assert "TMP" not in user_env  # a partially user-set triple is left exactly as found


def test_bootstrap_import_exports_scratch_to_process_and_children(tmp_path):
    """``import hermes_bootstrap`` alone makes ``tempfile`` (this process AND a child) land in scratch."""
    env = {k: v for k, v in os.environ.items() if k not in ("TMPDIR", "TMP", "TEMP", "HERMES_SCRATCH_DIR")}
    env["HERMES_HOME"] = str(tmp_path)
    code = ("import tempfile, os, subprocess, sys; import hermes_bootstrap; "
            "print(tempfile.gettempdir()); "
            "print(subprocess.run([sys.executable, '-c', 'import tempfile;print(tempfile.gettempdir())'],"
            " capture_output=True, text=True).stdout.strip())")
    out = subprocess.run([sys.executable, "-c", code], env=env, capture_output=True, text=True, encoding="utf-8",
                         cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))), check=True)
    expected = str(tmp_path / "cache" / "scratch")
    assert out.stdout.split() == [expected, expected]


def test_prune_removes_idle_entries_and_keeps_trees_written_deep_inside(tmp_path):
    """Idle retention: an entry goes when nothing in its subtree was written within the window;
    a tree whose only recent write is three levels down is still in use and stays, even though
    its top-level mtime is ancient (a directory's mtime ignores writes below its children)."""
    scratch = get_scratch_dir(tmp_path, prune=False)
    idle, live, fresh = scratch / "idle", scratch / "live", scratch / "fresh.txt"
    deep = live / "lane" / "wt"
    deep.mkdir(parents=True)
    idle.mkdir()
    (idle / "f").write_text("x", encoding="utf-8")
    (deep / "log").write_text("x", encoding="utf-8")
    fresh.write_text("y", encoding="utf-8")
    ancient = time.time() - 30 * 3600
    for path in (idle, idle / "f", live, live / "lane", deep):
        os.utime(path, (ancient, ancient))
    assert prune_scratch_dir(scratch) == 1
    assert not idle.exists() and live.exists() and fresh.exists()


@pytest.mark.platforms("posix")  # POSIX directory modes
class TestScratchDirPermissionPolicy:
    """get_scratch_dir must honor the home permission policy instead of a blanket 0700:
    an explicit HERMES_HOME_MODE and a managed/shared home win (#117347)."""

    @staticmethod
    def _native_mode_after_chmod(path, requested=0o2770):
        """Return what this host/filesystem preserves from a real chmod request.

        macOS may clear setgid when the directory group is not one of the caller's groups.
        The portable contract is to request the operator's mode and preserve every bit the
        host accepts, not to pretend all POSIX filesystems retain identical special bits.
        """
        os.chmod(path, requested)
        return stat.S_IMODE(os.stat(path).st_mode)

    def _isolate_env(self, monkeypatch, tmp_path):
        # A known, marker-free effective home: each case below plants `.managed` in the home whose
        # scratch dir it exercises (`get_scratch_dir(home)` reads the marker there), and the real
        # home's own marker must not leak into callers that still resolve the effective home.
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "home"))
        for var in ("HERMES_HOME_MODE", "HERMES_MANAGED", "HERMES_CONTAINER",
                    "HERMES_SKIP_CHMOD", "HERMES_UID", "HERMES_GID"):
            monkeypatch.delenv(var, raising=False)

    def test_default_is_owner_only(self, tmp_path, monkeypatch):
        self._isolate_env(monkeypatch, tmp_path)
        scratch = get_scratch_dir(tmp_path, prune=False)
        assert stat.S_IMODE(os.stat(scratch).st_mode) == 0o700

    def test_explicit_home_mode_uses_native_chmod_semantics(self, tmp_path, monkeypatch):
        self._isolate_env(monkeypatch, tmp_path)
        probe = tmp_path / "mode-probe"
        probe.mkdir()
        expected = self._native_mode_after_chmod(probe)
        monkeypatch.setenv("HERMES_HOME_MODE", "2770")
        scratch = get_scratch_dir(tmp_path, prune=False)
        assert stat.S_IMODE(os.stat(scratch).st_mode) == expected

    def test_managed_env_leaves_preexisting_mode_untouched(self, tmp_path, monkeypatch):
        self._isolate_env(monkeypatch, tmp_path)
        monkeypatch.setenv("HERMES_MANAGED", "nixos")
        pre = tmp_path / "cache" / "scratch"
        pre.mkdir(parents=True)
        expected = self._native_mode_after_chmod(pre)
        scratch = get_scratch_dir(tmp_path, prune=False)
        assert stat.S_IMODE(os.stat(scratch).st_mode) == expected

    def test_managed_marker_file_leaves_preexisting_mode_untouched(self, tmp_path, monkeypatch):
        home = tmp_path / "home"
        home.mkdir()
        self._isolate_env(monkeypatch, tmp_path)
        (home / ".managed").write_text("nixos", encoding="utf-8")
        pre = home / "cache" / "scratch"
        pre.mkdir(parents=True)
        expected = self._native_mode_after_chmod(pre)
        scratch = get_scratch_dir(home, prune=False)
        assert stat.S_IMODE(os.stat(scratch).st_mode) == expected

    def test_empty_managed_marker_counts_as_managed(self, tmp_path, monkeypatch):
        # Legacy NixOS module wrote an empty marker; config.get_managed_system treats it as
        # managed, so the parity twin must too (not re-enable chmod on managed installs).
        home = tmp_path / "home"
        home.mkdir()
        self._isolate_env(monkeypatch, tmp_path)
        (home / ".managed").write_text("", encoding="utf-8")
        pre = home / "cache" / "scratch"
        pre.mkdir(parents=True)
        expected = self._native_mode_after_chmod(pre)
        scratch = get_scratch_dir(home, prune=False)
        assert stat.S_IMODE(os.stat(scratch).st_mode) == expected

    def test_unreadable_managed_marker_counts_as_managed(self, tmp_path, monkeypatch):
        # A marker that exists but cannot be read (OSError -> "") still counts as managed.
        home = tmp_path / "home"
        home.mkdir()
        self._isolate_env(monkeypatch, tmp_path)
        (home / ".managed").mkdir()
        pre = home / "cache" / "scratch"
        pre.mkdir(parents=True)
        expected = self._native_mode_after_chmod(pre)
        scratch = get_scratch_dir(home, prune=False)
        assert stat.S_IMODE(os.stat(scratch).st_mode) == expected

    def test_effective_home_marker_does_not_govern_another_homes_scratch(self, tmp_path, monkeypatch):
        # The caller's home decides the policy. A boot caller (``export_scratch_tmp_env``) and
        # ``hermes doctor`` have already resolved theirs, so the marker lookup must not fall back
        # to ``get_hermes_home()``: for a sticky profile with HERMES_HOME unset that lookup warns
        # about the default profile on every command, and it is the wrong home to judge by anyway.
        selected = tmp_path / "home"
        selected.mkdir()
        self._isolate_env(monkeypatch, tmp_path)
        (selected / ".managed").write_text("nixos", encoding="utf-8")
        other = tmp_path / "other"
        pre = other / "cache" / "scratch"
        pre.mkdir(parents=True)
        os.chmod(pre, 0o750)
        assert stat.S_IMODE(os.stat(get_scratch_dir(other, prune=False)).st_mode) == 0o700

    def test_canonical_container_signal_without_env_override_keeps_operator_mode(self, tmp_path, monkeypatch):
        # Podman/containerd/K8s runtimes often don't export HERMES_CONTAINER; the canonical
        # _detect_container breadth (not the narrower legacy signal set) must skip the chmod.
        self._isolate_env(monkeypatch, tmp_path)
        monkeypatch.setattr("hermes_constants._detect_container", lambda: True)
        pre = tmp_path / "cache" / "scratch"
        pre.mkdir(parents=True)
        os.chmod(pre, 0o750)
        scratch = get_scratch_dir(tmp_path, prune=False)
        assert stat.S_IMODE(os.stat(scratch).st_mode) == 0o750

    def test_repeated_calls_do_not_strip_filesystem_supported_mode_bits(self, tmp_path, monkeypatch):
        self._isolate_env(monkeypatch, tmp_path)
        probe = tmp_path / "mode-probe"
        probe.mkdir()
        expected = self._native_mode_after_chmod(probe)
        monkeypatch.setenv("HERMES_HOME_MODE", "2770")
        first = get_scratch_dir(tmp_path, prune=False)
        second = get_scratch_dir(tmp_path, prune=False)
        assert first == second
        assert stat.S_IMODE(os.stat(second).st_mode) == expected

    def test_container_keeps_operator_mode_but_honors_explicit(self, tmp_path, monkeypatch):
        self._isolate_env(monkeypatch, tmp_path)
        monkeypatch.setenv("HERMES_CONTAINER", "1")
        pre = tmp_path / "cache" / "scratch"
        pre.mkdir(parents=True)
        os.chmod(pre, 0o750)
        assert stat.S_IMODE(os.stat(get_scratch_dir(tmp_path, prune=False)).st_mode) == 0o750
        probe = tmp_path / "mode-probe"
        probe.mkdir()
        expected = self._native_mode_after_chmod(probe)
        monkeypatch.setenv("HERMES_HOME_MODE", "2770")
        assert stat.S_IMODE(os.stat(get_scratch_dir(tmp_path, prune=False)).st_mode) == expected

    def test_hermes_uid_gid_applied_to_scratch(self, tmp_path, monkeypatch):
        self._isolate_env(monkeypatch, tmp_path)
        monkeypatch.setenv("HERMES_UID", "1000")
        monkeypatch.setenv("HERMES_GID", "911")
        with patch.object(os, "chown") as mock_chown:
            get_scratch_dir(tmp_path, prune=False)
        mock_chown.assert_called_once_with(tmp_path / "cache" / "scratch", 1000, 911)


@pytest.mark.platforms("posix")  # POSIX file modes
def test_secure_file_skips_chmod_on_canonical_container_signal(tmp_path, monkeypatch):
    """_secure_file skips on the same canonical container signal that apply_secure_dir_policy /
    get_scratch_dir already honor (one policy implementation in hermes_constants)."""
    from hermes_cli import config

    monkeypatch.setenv("HERMES_HOME", str(tmp_path / "home"))
    for var in ("HERMES_MANAGED", "HERMES_CONTAINER", "HERMES_SKIP_CHMOD"):
        monkeypatch.delenv(var, raising=False)
    monkeypatch.setattr("hermes_constants._detect_container", lambda: True)
    f = tmp_path / "config.yaml"
    f.write_text("", encoding="utf-8")
    os.chmod(f, 0o640)
    config._secure_file(f)
    assert stat.S_IMODE(os.stat(f).st_mode) == 0o640


def test_prune_reaps_process_living_in_idle_entry_and_spares_live_tree(tmp_path):
    """A process whose cwd is inside an idle entry is gone by the time the entry is
    (a lane's headless browsers survived for days with a deleted cwd); one living in a
    tree that is still being written is not touched."""
    import subprocess

    scratch = get_scratch_dir(tmp_path, prune=False)
    idle, live = scratch / "idle-lane" / "wt", scratch / "live-lane" / "wt"
    idle.mkdir(parents=True)
    live.mkdir(parents=True)
    ancient = time.time() - 30 * 3600
    for path in (idle.parent, idle, live.parent, live):
        os.utime(path, (ancient, ancient))
    (live / "log").write_text("still writing", encoding="utf-8")
    sleeper = [sys.executable, "-c", "import time; time.sleep(60)"]
    doomed = subprocess.Popen(sleeper, cwd=str(idle), stdin=subprocess.DEVNULL)
    spared = subprocess.Popen(sleeper, cwd=str(live), stdin=subprocess.DEVNULL)
    try:
        assert prune_scratch_dir(scratch) == 1
        assert doomed.wait(timeout=10) is not None
        assert spared.poll() is None
        assert not idle.parent.exists() and live.exists()
    finally:
        for proc in (doomed, spared):
            if proc.poll() is None:
                proc.kill()
                proc.wait(timeout=10)


def test_prune_releases_git_worktree_registration_of_idle_entry(tmp_path):
    """Deleting a scratch entry that held a linked worktree leaves the repo with no
    dangling registration (10 sat in one repo's ``git worktree list`` after cleanup)."""
    import subprocess

    def git(*args, cwd):
        subprocess.run(["git", *args], cwd=cwd, check=True, capture_output=True,
                       stdin=subprocess.DEVNULL, env={**os.environ, "GIT_CONFIG_GLOBAL": os.devnull,
                                                     "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@x",
                                                     "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@x"})

    repo = tmp_path / "repo"
    repo.mkdir()
    git("init", "-q", cwd=repo)
    (repo / "f").write_text("x", encoding="utf-8")
    git("add", "f", cwd=repo)
    git("commit", "-q", "-m", "init", cwd=repo)
    scratch = get_scratch_dir(tmp_path, prune=False)
    tree = scratch / "lane" / "abwt"
    tree.parent.mkdir()
    git("worktree", "add", "-q", "--detach", str(tree), cwd=repo)
    ancient = time.time() - 30 * 3600
    for dirpath, dirnames, filenames in os.walk(tree.parent):
        for name in dirnames + filenames:
            os.utime(os.path.join(dirpath, name), (ancient, ancient), follow_symlinks=False)
    os.utime(tree.parent, (ancient, ancient))
    assert prune_scratch_dir(scratch) == 1
    listing = subprocess.run(["git", "worktree", "list", "--porcelain"], cwd=repo, capture_output=True,
                             text=True, stdin=subprocess.DEVNULL, check=True).stdout
    assert str(tree) not in listing and not tree.exists()