Fix cron session context isolation

acp_adapter/server.py

@@ -1478,7 +1478,7 @@ def _run_agent() -> dict:
                     clear_session_vars,
                     set_session_vars,
                 )
-                session_tokens = set_session_vars(session_key=session_id)
+                session_tokens = set_session_vars(session_key=session_id, cron_session="")
             except Exception:
                 session_tokens = None
                 clear_session_vars = None  # type: ignore[assignment]

cron/scheduler.py

@@ -2914,11 +2914,6 @@ def run_job(
 
     agent = None
 
-    # Mark this as a cron session so the approval system can apply cron_mode.
-    # This env var is process-wide and persists for the lifetime of the
-    # scheduler process — every job this process runs is a cron job.
-    os.environ["HERMES_CRON_SESSION"] = "1"
-
     # Use ContextVars for per-job session/delivery state so parallel jobs
     # don't clobber each other's targets (os.environ is process-global).
     from gateway.session_context import set_session_vars, clear_session_vars, _VAR_MAP
@@ -3014,7 +3009,14 @@ def run_job(
     # statement raises.  A leaked writer would deadlock the whole scheduler
     # (every future job blocks on acquire_*); a leaked reader blocks all
     # future writers.  Acquire itself can't leak (it either blocks or returns).
+    _cron_session_var = _VAR_MAP["HERMES_CRON_SESSION"]
+    _cron_session_token = None
     try:
+        # Scope cron approval policy to this job. Keep the token so the finally
+        # restores the pre-job state instead of pinning an explicit empty value,
+        # which would suppress the legacy os.environ fallback used by standalone
+        # cron entrypoints and tests.
+        _cron_session_token = _cron_session_var.set("1")
         if _job_workdir:
             os.environ["TERMINAL_CWD"] = _job_workdir
             logger.info("Job '%s': using workdir %s", job_id, _job_workdir)
@@ -3624,6 +3626,8 @@ def _heartbeat_run_claim_if_due():
             _terminal_cwd_lock.release_read()
         # Clean up ContextVar session/delivery state for this job.
         clear_session_vars(_ctx_tokens)
+        if _cron_session_token is not None:
+            _cron_session_var.reset(_cron_session_token)
         for _var_name in _cron_delivery_vars:
             _VAR_MAP[_var_name].set("")
         if _session_db:

gateway/platforms/api_server.py

@@ -4617,6 +4617,7 @@ def _bind_api_server_session(
             session_key=session_key,
             session_id=session_id,
             async_delivery=False,
+            cron_session="",
         )
 
     async def _run_agent(

gateway/run.py

@@ -16383,6 +16383,7 @@ def _set_session_env(self, context: SessionContext) -> list:
             message_id=str(context.source.message_id) if context.source.message_id else "",
             profile=getattr(context.source, "profile", "") or "",
             async_delivery=_async_delivery,
+            cron_session="",
         )
 
     def _clear_session_env(self, tokens: list) -> None:

gateway/session_context.py

@@ -93,6 +93,12 @@ def session_context_engaged() -> bool:
 
 _SESSION_PROFILE: ContextVar = ContextVar("HERMES_SESSION_PROFILE", default=_UNSET)
 
+# Per-session cron marker. Unlike the process-global legacy env var, this is
+# scoped to one cron job / inbound session. _UNSET preserves the legacy env
+# fallback for CLI/tests; "1" marks cron; "" explicitly marks non-cron and
+# masks any leaked process env value.
+_CRON_SESSION: ContextVar = ContextVar("HERMES_CRON_SESSION", default=_UNSET)
+
 # Whether the current session's delivery channel can route an ASYNC completion
 # back to the agent AFTER the current turn ends (i.e. wake a fresh turn).
 #
@@ -133,6 +139,7 @@ def session_context_engaged() -> bool:
     "HERMES_UI_SESSION_ID": _SESSION_UI_SESSION_ID,
     "HERMES_SESSION_MESSAGE_ID": _SESSION_MESSAGE_ID,
     "HERMES_SESSION_PROFILE": _SESSION_PROFILE,
+    "HERMES_CRON_SESSION": _CRON_SESSION,
     "HERMES_CRON_AUTO_DELIVER_PLATFORM": _CRON_AUTO_DELIVER_PLATFORM,
     "HERMES_CRON_AUTO_DELIVER_CHAT_ID": _CRON_AUTO_DELIVER_CHAT_ID,
     "HERMES_CRON_AUTO_DELIVER_THREAD_ID": _CRON_AUTO_DELIVER_THREAD_ID,
@@ -169,6 +176,7 @@ def set_session_vars(
     cwd: str = "",
     async_delivery: bool = True,
     ui_session_id: str = "",
+    cron_session: Any = _UNSET,
 ) -> list:
     """Set all session context variables and return reset tokens.
 
@@ -184,6 +192,10 @@ def set_session_vars(
     background completion back to the agent after the turn ends (see
     ``_SESSION_ASYNC_DELIVERY`` / ``async_delivery_supported``). Stateless
     request/response adapters (the API server) pass ``False``.
+
+    ``cron_session`` is tri-state: ``_UNSET`` preserves legacy
+    ``os.environ["HERMES_CRON_SESSION"]`` fallback, ``"1"`` marks a cron job,
+    and ``""`` explicitly marks a non-cron session while masking leaked env.
     """
     # Mark the session-context machinery engaged for this process. The
     # subprocess-env bridge uses this to switch from "os.environ fallback" to
@@ -203,6 +215,7 @@ def set_session_vars(
         _SESSION_UI_SESSION_ID.set(ui_session_id),
         _SESSION_MESSAGE_ID.set(message_id),
         _SESSION_PROFILE.set(profile),
+        _CRON_SESSION.set(cron_session),
         _SESSION_ASYNC_DELIVERY.set(bool(async_delivery)),
     ]
     try:
@@ -238,6 +251,7 @@ def clear_session_vars(tokens: list) -> None:
         _SESSION_UI_SESSION_ID,
         _SESSION_MESSAGE_ID,
         _SESSION_PROFILE,
+        _CRON_SESSION,
     ):
         var.set("")
     # Reset async-delivery capability to the "never set" sentinel rather than a

tests/cron/conftest.py

@@ -19,3 +19,23 @@ def _default_cron_test_model(monkeypatch):
     """Pin a default HERMES_MODEL so cron run_job tests have a resolvable model."""
     monkeypatch.setenv("HERMES_MODEL", "test-cron-default-model")
     yield
+
+
+@pytest.fixture(autouse=True)
+def _reset_session_context_vars():
+    """Restore session ContextVars around cron tests that call run_job directly.
+
+    Production confines each cron run to a copied context, but direct unit tests
+    share the pytest context. ``run_job`` intentionally clears ordinary session
+    variables to explicit empty values, which would otherwise shadow legacy env
+    fallbacks used by later approval tests in the same process.
+    """
+    from gateway.session_context import _UNSET, _VAR_MAP
+
+    def _reset_all():
+        for var in _VAR_MAP.values():
+            var.set(_UNSET)
+
+    _reset_all()
+    yield
+    _reset_all()

tests/cron/test_scheduler_cron_session_isolation.py

@@ -0,0 +1,161 @@
+"""Regression test for cron-session approval isolation.
+
+A cron job must use ``approvals.cron_mode`` for its own ``execute_code`` call,
+without leaving process-global state that changes a later interactive gateway
+turn handled by the same Python process.
+"""
+
+from __future__ import annotations
+
+import os
+
+import pytest
+
+import cron.scheduler as cron_scheduler
+from gateway.session_context import (
+    clear_session_vars,
+    get_session_env,
+    reset_session_vars,
+    set_session_vars,
+)
+from tools import approval as approval_module
+
+
+class _DummySessionDB:
+    def set_session_title(self, *args, **kwargs):
+        pass
+
+    def end_session(self, *args, **kwargs):
+        pass
+
+    def close(self):
+        pass
+
+
+class _FakeCronAgent:
+    def __init__(self, *args, **kwargs):
+        self.kwargs = kwargs
+
+    def run_conversation(self, prompt):
+        result = approval_module.check_execute_code_guard(
+            "import os; print(1)", "local"
+        )
+        assert result["approved"] is False
+        assert result["outcome"] == "blocked"
+        assert get_session_env("HERMES_CRON_SESSION") == "1"
+        return {
+            "completed": True,
+            "failed": False,
+            "final_response": "cron execute_code blocked",
+            "turn_exit_reason": "",
+        }
+
+    def close(self):
+        pass
+
+
+@pytest.fixture(autouse=True)
+def _clear_approval_state(monkeypatch):
+    reset_session_vars()
+    monkeypatch.delenv("HERMES_CRON_SESSION", raising=False)
+    monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+    monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+    monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
+    approval_module._permanent_approved.clear()
+    approval_module.clear_session("default")
+    approval_module.clear_session("cron-isolation-session")
+    yield
+    approval_module._permanent_approved.clear()
+    approval_module.clear_session("default")
+    approval_module.clear_session("cron-isolation-session")
+    reset_session_vars()
+
+
+def _register_gateway_auto_approve(session_key: str) -> None:
+    def _notify(_approval_data):
+        with approval_module._lock:
+            entries = approval_module._gateway_queues.get(session_key, [])
+            if entries:
+                entry = entries[-1]
+                entry.result = "once"
+                entry.event.set()
+
+    with approval_module._lock:
+        approval_module._gateway_notify_cbs[session_key] = _notify
+
+
+def test_run_job_cron_execute_code_deny_does_not_pollute_later_gateway_execute_code(
+    monkeypatch, tmp_path
+):
+    """Cron deny stays scoped; a later gateway approval still reaches its user."""
+    monkeypatch.setenv("HERMES_MODEL", "test-model")
+    monkeypatch.setattr(approval_module, "_YOLO_MODE_FROZEN", False)
+    monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "manual")
+    monkeypatch.setattr(approval_module, "_get_cron_approval_mode", lambda: "deny")
+    monkeypatch.setattr("hermes_state.SessionDB", _DummySessionDB)
+    monkeypatch.setattr("run_agent.AIAgent", _FakeCronAgent)
+    monkeypatch.setattr(
+        "hermes_constants.resolve_reasoning_config", lambda *_args, **_kwargs: None
+    )
+    monkeypatch.setattr(
+        "hermes_cli.runtime_provider.resolve_runtime_provider",
+        lambda **_kwargs: {
+            "api_key": "test-key",
+            "base_url": None,
+            "provider": "test-provider",
+            "api_mode": None,
+            "command": None,
+            "args": None,
+        },
+    )
+    monkeypatch.setattr("tools.mcp_tool.discover_mcp_tools", lambda: [])
+    monkeypatch.setattr(cron_scheduler, "_get_hermes_home", lambda: tmp_path)
+    monkeypatch.setattr(cron_scheduler, "get_fallback_chain", lambda _cfg: [])
+    monkeypatch.setattr(
+        cron_scheduler, "_guard_job_credential_exfil", lambda _job: None
+    )
+
+    success, _output, final_response, error = cron_scheduler.run_job(
+        {
+            "id": "ctx-isolation",
+            "name": "Context Isolation",
+            "prompt": "Run safely",
+            "schedule_display": "manual",
+        }
+    )
+
+    assert success is True
+    assert error is None
+    assert final_response == "cron execute_code blocked"
+    assert os.environ.get("HERMES_CRON_SESSION") is None
+    assert get_session_env("HERMES_CRON_SESSION") == ""
+
+    # A completed in-process job must restore the truly-unset ContextVar state,
+    # not leave an explicit empty value that shadows the standalone cron env
+    # fallback in this reused context.
+    monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+    assert get_session_env("HERMES_CRON_SESSION") == "1"
+    monkeypatch.delenv("HERMES_CRON_SESSION")
+
+    session_key = "cron-isolation-session"
+    key_token = approval_module.set_current_session_key(session_key)
+    session_tokens = set_session_vars(
+        platform="discord",
+        chat_id="123",
+        session_key=session_key,
+        cron_session="",
+    )
+    monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1")
+    try:
+        _register_gateway_auto_approve(session_key)
+        result = approval_module.check_execute_code_guard(
+            "import os; print(2)", "local"
+        )
+        assert result["approved"] is True
+        assert result.get("user_approved") is True
+    finally:
+        clear_session_vars(session_tokens)
+        approval_module.reset_current_session_key(key_token)
+        with approval_module._lock:
+            approval_module._gateway_queues.pop(session_key, None)
+            approval_module._gateway_notify_cbs.pop(session_key, None)

tests/gateway/test_session_env.py

@@ -10,6 +10,7 @@
     get_session_env,
     set_session_vars,
     clear_session_vars,
+    reset_session_vars,
     _VAR_MAP,
     _UNSET,
 )
@@ -327,6 +328,40 @@ async def test_run_in_executor_with_context_preserves_session_env(monkeypatch):
     }
 
 
+def test_cron_session_contextvar_preserves_legacy_env_fallback(monkeypatch):
+    """Unset cron ContextVar keeps old env-only cron callers working."""
+    monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+
+    assert get_session_env("HERMES_CRON_SESSION") == "1"
+
+
+def test_cron_session_explicit_blank_masks_leaked_env(monkeypatch):
+    """Non-cron session bindings must override a stale process cron env flag."""
+    monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+
+    tokens = set_session_vars(platform="api_server", cron_session="")
+    try:
+        assert get_session_env("HERMES_CRON_SESSION") == ""
+    finally:
+        clear_session_vars(tokens)
+
+    assert get_session_env("HERMES_CRON_SESSION") == ""
+
+
+def test_cron_session_set_clear_and_reset_tristate(monkeypatch):
+    """Cron marker supports _UNSET fallback, '1' cron, and '' explicit clear."""
+    monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+
+    tokens = set_session_vars(cron_session="1")
+    assert get_session_env("HERMES_CRON_SESSION") == "1"
+
+    clear_session_vars(tokens)
+    assert get_session_env("HERMES_CRON_SESSION") == ""
+
+    reset_session_vars()
+    assert get_session_env("HERMES_CRON_SESSION") == "1"
+
+
 @pytest.mark.asyncio
 async def test_run_in_executor_with_context_forwards_args():
     """_run_in_executor_with_context should forward *args to the callable."""