fix(kanban): keep task cwd off sys.path of module-form workers

cron/scheduler_worker_env.py

@@ -26,6 +26,8 @@
 
 logger = logging.getLogger(__name__)
 
+SAFE_PATH_PIN_MARKER = "_HERMES_PINNED_SAFEPATH"
+
 
 def _installed_purelib() -> Path | None:
     try:
@@ -78,6 +80,17 @@ def pin_hermes_tree_on_pythonpath(worker_env: dict, repo_root: Path) -> dict:
     root = str(repo_root)
     if _installed_purelib() == Path(root).resolve():
         return worker_env
+    # The child's cwd (task workspace) is first on sys.path for ``-m``; a checkout there
+    # would shadow the pinned tree with its own dependency-less copy.
+    # Process-scoped: the worker latches ``sys.flags.safe_path`` at interpreter start, then
+    # the worker's boot (``hermes_cli.main`` / ``cron.worker_bootstrap``) restores the user's own value so the agent's own children (terminal /
+    # execute_code) keep the normal script-dir/cwd sys.path entry. 
+    # The marker records the user's own value ("-" absent, "+<value>" set) so boot can
+    # restore it; setdefault keeps the first record if the helper runs twice on one env.
+    if SAFE_PATH_PIN_MARKER not in worker_env:
+        user = worker_env.get("PYTHONSAFEPATH")
+        worker_env[SAFE_PATH_PIN_MARKER] = "-" if user is None else "+" + user
+    worker_env["PYTHONSAFEPATH"] = "1"
     existing = [e for e in worker_env.get("PYTHONPATH", "").split(os.pathsep) if e]
     dependency = _committed_dependency_site_packages(Path(root))
     pinned = [root, *([str(dependency)] if dependency is not None else [])]

cron/worker_bootstrap.py

@@ -48,6 +48,12 @@
 
 def worker_bootstrap() -> None:
     """Run PM's dependency boot in the marked external worker (again after a relaunch)."""
+    # sys.flags.safe_path is already latched; keep PYTHONSAFEPATH out of our children.
+    # Restore the user's own value ("-" absent, "+<value>" set) recorded by the pin helper.
+    if pinned := os.environ.pop("_HERMES_PINNED_SAFEPATH", None):
+        os.environ.pop("PYTHONSAFEPATH", None)
+        if pinned[0] == "+":
+            os.environ["PYTHONSAFEPATH"] = pinned[1:]
     if not os.environ.get(WORKER_MARKER):
         return
     from pm.environments import activate_dependencies

hermes_cli/main.py

@@ -39,6 +39,14 @@
 import re
 import sys
 
+# A worker pinned with PYTHONSAFEPATH (cron.scheduler_worker_env) latched it at interpreter
+# start; drop it so this agent's own children keep their normal script-dir/cwd sys.path.
+# Restore the user's own value ("-" absent, "+<value>" set) recorded by the pin helper.
+if _pinned := os.environ.pop("_HERMES_PINNED_SAFEPATH", None):
+    os.environ.pop("PYTHONSAFEPATH", None)
+    if _pinned[0] == "+":
+        os.environ["PYTHONSAFEPATH"] = _pinned[1:]
+
 # Inline path math so ``python hermes_cli/main.py`` (script mode: sys.path[0]
 # is hermes_cli/, not the repo root) can import hermes_cli._startup_fast.
 _bootstrap_root = os.path.realpath(os.path.join(os.path.dirname(__file__), os.pardir))

tests/cron/test_restart_safe_worker.py

@@ -739,10 +739,74 @@ def test_pin_restores_the_committed_generation_site_packages(tmp_path):
     # A runner that owns its dependencies has no committed generation: tree only.
     pm.environments.runtime_facts_path(repo_root).unlink()
     assert worker_env_mod.pin_hermes_tree_on_pythonpath({}, repo_root) == {
-        "PYTHONPATH": str(repo_root)
+        "PYTHONPATH": str(repo_root), "PYTHONSAFEPATH": "1",
+        "_HERMES_PINNED_SAFEPATH": "-",
     }
 
 
+_BOOT_PATHS = {
+    "hermes_cli.main": "import hermes_cli.main",
+    "cron.worker_bootstrap": "import cron.worker_bootstrap as w; w.worker_bootstrap()",
+}
+
+
+@pytest.mark.parametrize("boot", sorted(_BOOT_PATHS))
+@pytest.mark.parametrize("user", [None, "", "1", "x"])
+def test_pinned_safe_path_restores_user_value_for_children(tmp_path, boot, user):
+    """Helper-fed env -> real worker boot -> real child. The worker latches
+    sys.flags.safe_path; the child sees exactly the user's own PYTHONSAFEPATH (absent,
+    empty or set), no Hermes marker, and imports its own sibling unless the user set it."""
+    import subprocess
+    import cron.scheduler_worker_env as worker_env_mod
+
+    repo_root = Path(worker_env_mod.__file__).resolve().parent.parent
+    base = {k: v for k, v in os.environ.items() if k != "PYTHONSAFEPATH"}
+    if user is not None:
+        base["PYTHONSAFEPATH"] = user
+    env = worker_env_mod.pin_hermes_tree_on_pythonpath(base, repo_root)
+    (tmp_path / "sib_mod.py").write_text("X = 1")
+    (tmp_path / "child.py").write_text(
+        "import os\n"
+        "try:\n    import sib_mod; ok = True\nexcept ImportError:\n    ok = False\n"
+        "print(repr(os.environ.get('PYTHONSAFEPATH')), '_HERMES_PINNED_SAFEPATH' in os.environ, ok)"
+    )
+    code = (
+        _BOOT_PATHS[boot] + "\nimport subprocess, sys\n"
+        "print(sys.flags.safe_path)\n"
+        f"print(subprocess.run([sys.executable, {str(tmp_path / 'child.py')!r}], "
+        "capture_output=True, text=True, timeout=120).stdout.strip())"
+    )
+    run = subprocess.run([sys.executable, "-c", code], env=env, cwd=tmp_path,
+                         capture_output=True, text=True, timeout=240)
+    assert run.returncode == 0, run.stderr
+    latched, child = run.stdout.strip().splitlines()[-2:]
+    assert latched == "True"  # cwd-shadow protection stays on in the worker
+    # Windows cannot hold an empty env var in os.environ: "" reads back as absent.
+    want_var = "None" if user is None or (user == "" and os.name == "nt") else repr(user)
+    # PYTHONSAFEPATH is honoured by CPython only when non-empty.
+    assert child == f"{want_var} False {not user}"
+
+
+def test_pinned_worker_ignores_decoy_hermes_cli_in_cwd(tmp_path):
+    """cwd holding its own ``hermes_cli`` must not shadow the pinned install."""
+    import subprocess
+
+    import hermes_cli
+    from hermes_cli.kanban_db_dispatch import _module_hermes_argv, _propagate_module_import_root
+
+    decoy = tmp_path / "hermes_cli"
+    decoy.mkdir()
+    (decoy / "__init__.py").write_text("raise SystemExit('DECOY IMPORTED')")
+    env = dict(os.environ)
+    _propagate_module_import_root(_module_hermes_argv(), env)  # kanban dispatcher path
+    out = subprocess.run(
+        [sys.executable, "-c", "import hermes_cli;print(hermes_cli.__file__)"],
+        cwd=tmp_path, env=env, capture_output=True, text=True,
+    )
+    assert out.returncode == 0, out.stderr
+    assert Path(out.stdout.strip()).resolve() == Path(hermes_cli.__file__).resolve()
+
+
 _BOOT_ORDER_PROBE = """
 import json, os, sys
 import pm.environments