fix(cron): scope cron approval context per session

Replace the process-global HERMES_CRON_SESSION env var with a per-session
ContextVar so a cron tick in the gateway process cannot leak into unrelated
live gateway/API/TUI turns. The cron scheduler now sets the ContextVar
inside the job's try/finally scope and resets it on cleanup. Gateway, API
server, ACP adapter, and TUI gateway all pass cron_session='' to explicitly
mark their sessions as non-cron, masking any stale process env.

Co-authored-by: hinablue <hinablue@gmail.com>
Closes #37968

acp_adapter/server.py

@@ -1868,8 +1868,11 @@ class HermesACPAgent(acp.Agent):
                 # while the tools are rooted at the client's project, so the
                 # model emits absolute paths under ~/.hermes/workspace and the
                 # edit silently lands outside the editor's workspace.
+                # cron_session="" explicitly marks this as a non-cron context,
+                # masking any leaked process-global HERMES_CRON_SESSION (#37968).
                 session_tokens = set_session_vars(
                     session_key=session_id, session_id=session_id, cwd=state.cwd,
+                    cron_session="",
                 )
             except Exception:
                 session_tokens = None

cron/scheduler.py

@@ -3009,11 +3009,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
@@ -3116,7 +3111,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)
@@ -3763,6 +3765,8 @@ def run_job(
         # clear_session_vars also clears _SESSION_CWD internally, so no
         # separate clear_session_cwd() call is needed.
         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

@@ -5907,6 +5907,7 @@ class APIServerAdapter(BasePlatformAdapter):
             session_key=session_key,
             session_id=session_id,
             async_delivery=False,
+            cron_session="",
         )
 
     async def _run_agent(

gateway/run.py

@@ -20582,6 +20582,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             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

@@ -95,6 +95,12 @@ _SESSION_MESSAGE_ID: ContextVar = ContextVar("HERMES_SESSION_MESSAGE_ID", defaul
 
 _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).
 #
@@ -135,6 +141,7 @@ _VAR_MAP = {
     "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,
@@ -212,6 +219,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.
 
@@ -227,6 +235,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
@@ -247,6 +259,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:
@@ -283,6 +296,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 @@ from gateway.session_context import (
     get_session_env,
     set_session_vars,
     clear_session_vars,
+    reset_session_vars,
     _VAR_MAP,
     _UNSET,
 )
@@ -237,3 +238,38 @@ 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"
+

tests/tools/test_cron_approval_mode.py

@@ -3,6 +3,7 @@
 import pytest
 
 import tools.approval as approval_module
+from gateway.session_context import clear_session_vars, reset_session_vars, set_session_vars
 from tools.approval import (
     _get_cron_approval_mode,
     check_all_command_guards,
@@ -16,10 +17,12 @@ def _clear_approval_state():
     approval_module._permanent_approved.clear()
     approval_module.clear_session("default")
     approval_module.clear_session("test-session")
+    reset_session_vars()
     yield
     approval_module._permanent_approved.clear()
     approval_module.clear_session("default")
     approval_module.clear_session("test-session")
+    reset_session_vars()
 
 
 # ---------------------------------------------------------------------------
@@ -83,6 +86,65 @@ class TestCronApprovalModeParsing:
             assert _get_cron_approval_mode() == "deny"
 
 
+# ---------------------------------------------------------------------------
+# ContextVar cron detection
+# ---------------------------------------------------------------------------
+
+class TestCronContextVarDetection:
+    def test_legacy_env_fallback_still_marks_cron(self, monkeypatch):
+        monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+        assert approval_module._is_cron_approval_context() is True
+
+    def test_explicit_blank_masks_leaked_cron_env_for_gateway_classification(self, monkeypatch):
+        monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+        monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1")
+        tokens = set_session_vars(platform="api_server", cron_session="")
+        try:
+            assert approval_module._is_cron_approval_context() is False
+            assert approval_module._is_gateway_approval_context() is True
+        finally:
+            clear_session_vars(tokens)
+
+    def test_scoped_cron_deny_for_dangerous_all_and_execute_code(self, monkeypatch):
+        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)
+        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")
+
+        tokens = set_session_vars(cron_session="1")
+        try:
+            dangerous = check_dangerous_command("rm -rf /tmp/stuff", "local")
+            combined = check_all_command_guards("rm -rf /tmp/stuff", "local")
+            code = approval_module.check_execute_code_guard("import os", "local")
+        finally:
+            clear_session_vars(tokens)
+
+        assert dangerous["approved"] is False
+        assert combined["approved"] is False
+        assert code["approved"] is False
+        assert code["outcome"] == "blocked"
+
+    def test_non_cron_blank_context_keeps_headless_execute_code_legacy_approved(self, monkeypatch):
+        monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+        monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+        monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+        monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
+        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")
+
+        tokens = set_session_vars(cron_session="")
+        try:
+            result = approval_module.check_execute_code_guard("import os", "local")
+        finally:
+            clear_session_vars(tokens)
+
+        assert result["approved"] is True
+
+
 # ---------------------------------------------------------------------------
 # check_dangerous_command() with cron session
 # ---------------------------------------------------------------------------

tests/tools/test_execute_code_approval_cluster.py

@@ -24,6 +24,7 @@ import pytest
 
 from tools import approval as A
 from tools.thread_context import propagate_context_to_thread
+from gateway.session_context import clear_session_vars, reset_session_vars, set_session_vars
 
 
 # ---------------------------------------------------------------------------
@@ -122,6 +123,8 @@ def gw_session(monkeypatch):
     with A._lock:
         A._gateway_queues.pop(session_key, None)
         A._gateway_notify_cbs.pop(session_key, None)
+        A._permanent_approved.discard("execute_code")
+        A._session_approved.get(session_key, set()).discard("execute_code")
     try:
         yield session_key
     finally:
@@ -178,6 +181,39 @@ def test_guard_headless_local_approved(monkeypatch):
 
 
 def test_guard_cron_deny_blocks(monkeypatch):
+    monkeypatch.setattr(A, "_YOLO_MODE_FROZEN", False)
+    monkeypatch.delenv("HERMES_CRON_SESSION", raising=False)
+    monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+    monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual")
+    monkeypatch.setattr(A, "_get_cron_approval_mode", lambda: "deny")
+    tokens = set_session_vars(cron_session="1")
+    try:
+        res = A.check_execute_code_guard("import os", "local")
+    finally:
+        clear_session_vars(tokens)
+    assert res["approved"] is False
+    assert res["outcome"] == "blocked"
+
+
+def test_guard_explicit_non_cron_masks_leaked_env(monkeypatch):
+    monkeypatch.setattr(A, "_YOLO_MODE_FROZEN", False)
+    monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+    monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+    monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+    monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
+    monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual")
+    monkeypatch.setattr(A, "_get_cron_approval_mode", lambda: "deny")
+    tokens = set_session_vars(cron_session="")
+    try:
+        res = A.check_execute_code_guard("import os", "local")
+    finally:
+        clear_session_vars(tokens)
+        reset_session_vars()
+    assert res["approved"] is True
+
+
+def test_guard_legacy_env_cron_still_blocks(monkeypatch):
+    reset_session_vars()
     monkeypatch.setattr(A, "_YOLO_MODE_FROZEN", False)
     monkeypatch.setenv("HERMES_CRON_SESSION", "1")
     monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)

tests/tools/test_request_tool_approval.py

@@ -77,8 +77,7 @@ class TestRequestToolApproval:
     def test_cron_deny_mode_blocks(self, monkeypatch):
         monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False)
         monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
-        monkeypatch.setattr(approval, "env_var_enabled",
-                            lambda v: v == "HERMES_CRON_SESSION")
+        monkeypatch.setattr(approval, "_is_cron_approval_context", lambda: True)
         monkeypatch.setattr(approval, "_get_cron_approval_mode", lambda: "deny")
         res = request_tool_approval("terminal", "smtp send")
         assert res["approved"] is False
@@ -87,8 +86,7 @@ class TestRequestToolApproval:
     def test_cron_approve_mode_allows(self, monkeypatch):
         monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False)
         monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
-        monkeypatch.setattr(approval, "env_var_enabled",
-                            lambda v: v == "HERMES_CRON_SESSION")
+        monkeypatch.setattr(approval, "_is_cron_approval_context", lambda: True)
         monkeypatch.setattr(approval, "_get_cron_approval_mode", lambda: "approve")
         res = request_tool_approval("terminal", "smtp send")
         assert res["approved"] is True
@@ -116,7 +114,7 @@ class TestRequestToolApproval:
         — a plugin-flagged action never runs ungated without a human."""
         monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False)
         monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
-        monkeypatch.setattr(approval, "env_var_enabled", lambda v: False)  # not cron
+        monkeypatch.setattr(approval, "_is_cron_approval_context", lambda: False)
         res = request_tool_approval("terminal", "smtp send")
         assert res["approved"] is False
         assert "no interactive user or gateway" in res["message"].lower()

tools/approval.py

@@ -224,6 +224,22 @@ def _get_session_platform() -> str:
         return os.getenv("HERMES_SESSION_PLATFORM", "") or ""
 
 
+def _is_cron_approval_context() -> bool:
+    """True when the current approval decision is running inside cron.
+
+    Prefer the session ContextVar so one cron job cannot taint unrelated
+    gateway/API/TUI turns in the same process. If the session context layer is
+    not engaged or unavailable, fall back to the legacy process env var for CLI
+    tests and older entrypoints.
+    """
+    try:
+        from gateway.session_context import get_session_env
+
+        return is_truthy_value(get_session_env("HERMES_CRON_SESSION", ""))
+    except Exception:
+        return env_var_enabled("HERMES_CRON_SESSION")
+
+
 def _is_gateway_approval_context() -> bool:
     """True when this call is inside a gateway/API session.
 
@@ -238,7 +254,7 @@ def _is_gateway_approval_context() -> bool:
     fall through to the gateway branch would submit a pending approval
     with no listener and block the job indefinitely.
     """
-    if env_var_enabled("HERMES_CRON_SESSION"):
+    if _is_cron_approval_context():
         return False
     if env_var_enabled("HERMES_GATEWAY_SESSION"):
         return True
@@ -2903,7 +2919,7 @@ def _run_approval_gate(
 
     if not is_cli and not is_gateway:
         # Cron sessions: respect cron_mode config
-        if env_var_enabled("HERMES_CRON_SESSION"):
+        if _is_cron_approval_context():
             if _get_cron_approval_mode() == "deny":
                 return {
                     "approved": False,
@@ -3431,7 +3447,7 @@ def check_all_command_guards(command: str, env_type: str,
     # flows, we do not block on approvals and we skip external guard work.
     if not is_cli and not is_gateway and not is_ask:
         # Cron sessions: respect cron_mode config
-        if env_var_enabled("HERMES_CRON_SESSION"):
+        if _is_cron_approval_context():
             if _get_cron_approval_mode() == "deny":
                 # Run detection to get a description for the block message
                 is_dangerous, _pk, description = detect_dangerous_command(command)
@@ -3882,7 +3898,7 @@ def check_execute_code_guard(code: str, env_type: str,
     is_ask = env_var_enabled("HERMES_EXEC_ASK")
 
     # Cron: no user is present to approve arbitrary code.
-    if env_var_enabled("HERMES_CRON_SESSION"):
+    if _is_cron_approval_context():
         if _get_cron_approval_mode() == "deny":
             return {
                 "approved": False,

tools/environments/base.py

@@ -409,7 +409,7 @@ def _cwd_marker(session_id: str) -> str:
 # as the Python-side contract for the exclusion set; the dump path unsets by
 # name/prefix instead of grepping declare lines (see below / issue #71296).
 _SNAPSHOT_EXCLUDED_ENV_REGEX = (
-    "^declare -x (HERMES_SESSION_|HERMES_UI_SESSION_ID|HERMES_CRON_AUTO_DELIVER_)"
+    "^declare -x (HERMES_SESSION_|HERMES_UI_SESSION_ID|HERMES_CRON_AUTO_DELIVER_|HERMES_CRON_SESSION)"
 )
 _SHELL_ENV_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
 

tui_gateway/server.py

@@ -2944,6 +2944,7 @@ def _set_session_context(
             source=source,
             cwd=resolved,
             ui_session_id=ui_session_id,
+            cron_session="",
         )
     except Exception:
         return []