fix(cron): isolate gateway approvals from environment pollution

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
@@ -3003,6 +2998,12 @@ def run_job(
     # acquire) and is a no-op for workdir-less jobs (they never mutate the env).
     _prior_terminal_cwd = os.environ.get("TERMINAL_CWD", "_UNSET_")
 
+    # Bind cron approval policy to this job's ContextVar scope. Keep the token
+    # so cleanup restores the true prior state (_UNSET for normal callers),
+    # preserving the legacy os.environ fallback used by standalone entrypoints.
+    _cron_session_var = _VAR_MAP["HERMES_CRON_SESSION"]
+    _cron_session_token = None
+
     _holds_cwd_write = _job_workdir is not None
     if _holds_cwd_write:
         _terminal_cwd_lock.acquire_write()
@@ -3015,6 +3016,7 @@ def run_job(
     # (every future job blocks on acquire_*); a leaked reader blocks all
     # future writers.  Acquire itself can't leak (it either blocks or returns).
     try:
+        _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/session_context.py

@@ -90,6 +90,9 @@ def session_context_engaged() -> bool:
 # so background-process notifications stay inside the originating Telegram
 # private-chat topic (those lanes route only with thread id + reply anchor).
 _SESSION_MESSAGE_ID: ContextVar = ContextVar("HERMES_SESSION_MESSAGE_ID", default=_UNSET)
+# Approval routing: cron state must be context-local so a scheduler tick in the
+# gateway process cannot leak into unrelated live turns.
+_CRON_SESSION: ContextVar = ContextVar("HERMES_CRON_SESSION", default=_UNSET)
 
 _SESSION_PROFILE: ContextVar = ContextVar("HERMES_SESSION_PROFILE", default=_UNSET)
 
@@ -133,6 +136,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 +173,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 +189,11 @@ 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 intentionally separate from ordinary session cleanup:
+    ``_UNSET`` leaves cron scope untouched, while ``"1"`` binds cron policy
+    for a nested context and returns a token that ``clear_session_vars`` resets.
+    This preserves the legacy process-environment fallback after cleanup.
     """
     # Mark the session-context machinery engaged for this process. The
     # subprocess-env bridge uses this to switch from "os.environ fallback" to
@@ -205,6 +215,8 @@ def set_session_vars(
         _SESSION_PROFILE.set(profile),
         _SESSION_ASYNC_DELIVERY.set(bool(async_delivery)),
     ]
+    if cron_session is not _UNSET:
+        tokens.append(_CRON_SESSION.set(cron_session))
     try:
         from agent.runtime_cwd import set_session_cwd
 
@@ -217,13 +229,10 @@ def set_session_vars(
 def clear_session_vars(tokens: list) -> None:
     """Mark session context variables as explicitly cleared.
 
-    Sets all variables to ``""`` so that ``get_session_env`` returns an empty
-    string instead of falling back to (potentially stale) ``os.environ``
-    values.  The *tokens* argument is accepted for API compatibility with
-    callers that saved the return value of ``set_session_vars``, but the
-    actual clearing uses ``var.set("")`` rather than ``var.reset(token)``
-    to ensure the "explicitly cleared" state is distinguishable from
-    "never set" (which holds the ``_UNSET`` sentinel).
+    Sets ordinary session variables to ``""`` so that ``get_session_env``
+    does not fall back to stale ``os.environ`` values. Dedicated nested scope
+    variables such as ``HERMES_CRON_SESSION`` instead reset their saved token,
+    preserving the pre-scope state and legacy fallback semantics.
     """
     for var in (
         _SESSION_PLATFORM,
@@ -240,6 +249,11 @@ def clear_session_vars(tokens: list) -> None:
         _SESSION_PROFILE,
     ):
         var.set("")
+    # Some legacy/mock gateway paths pass ``None`` because the old cleanup
+    # implementation ignored this argument entirely. Preserve that contract.
+    for token in tokens or ():
+        if getattr(token, "var", None) is _CRON_SESSION:
+            _CRON_SESSION.reset(token)
     # Reset async-delivery capability to the "never set" sentinel rather than a
     # falsy value: a cleared context should fall back to the default-supported
     # behavior (CLI / unaware paths), not be mistaken for an opted-out

tests/cron/test_scheduler.py

@@ -1821,6 +1821,56 @@ def run_conversation(self, *args, **kwargs):
         assert os.getenv("HERMES_CRON_AUTO_DELIVER_THREAD_ID") is None
         assert fake_db.close.call_count == 2
 
+    def test_run_job_keeps_cron_session_env_unchanged(self, tmp_path, monkeypatch):
+        job = {
+            "id": "cron-env-job",
+            "name": "cron env test",
+            "prompt": "hello",
+            "deliver": "local",
+        }
+        fake_db = MagicMock()
+        seen = {}
+
+        monkeypatch.setenv("HERMES_CRON_SESSION", "legacy")
+
+        class FakeAgent:
+            def __init__(self, *args, **kwargs):
+                pass
+
+            def run_conversation(self, *args, **kwargs):
+                from gateway.session_context import get_session_env
+
+                seen["env"] = os.getenv("HERMES_CRON_SESSION")
+                seen["ctx"] = get_session_env("HERMES_CRON_SESSION")
+                return {"final_response": "ok"}
+
+            def close(self):
+                pass
+
+        with patch("cron.scheduler._hermes_home", tmp_path), \
+             patch("hermes_state.SessionDB", return_value=fake_db), \
+             patch(
+                 "hermes_cli.runtime_provider.resolve_runtime_provider",
+                 return_value={
+                     "api_key": "***",
+                     "base_url": "https://example.invalid/v1",
+                     "provider": "openrouter",
+                     "api_mode": "chat_completions",
+                 },
+             ), \
+             patch("run_agent.AIAgent", FakeAgent):
+            success, output, final_response, error = run_job(job)
+
+        from gateway.session_context import get_session_env
+
+        assert success is True
+        assert error is None
+        assert final_response == "ok"
+        assert "ok" in output
+        assert seen == {"env": "legacy", "ctx": "1"}
+        assert os.getenv("HERMES_CRON_SESSION") == "legacy"
+        assert get_session_env("HERMES_CRON_SESSION") == "legacy"
+
 
 class TestRunJobConfigLogging:
     """Verify that config.yaml parse failures are logged, not silently swallowed."""

tests/gateway/test_session_env.py

@@ -139,6 +139,26 @@ def test_get_session_env_falls_back_to_os_environ(monkeypatch):
     assert get_session_env("HERMES_SESSION_PLATFORM") == ""
 
 
+def test_scoped_cron_session_restores_env_fallback(monkeypatch):
+    """Cron scope cleanup must restore, not permanently mask, env fallback."""
+    monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+
+    assert get_session_env("HERMES_CRON_SESSION") == "1"
+
+    tokens = set_session_vars(platform="telegram", cron_session="")
+    assert get_session_env("HERMES_CRON_SESSION") == ""
+
+    clear_session_vars(tokens)
+    assert get_session_env("HERMES_CRON_SESSION") == "1"
+
+
+def test_clear_session_vars_accepts_legacy_none_tokens():
+    """Mock/legacy callers may have no saved token list to provide."""
+    clear_session_vars(None)
+
+    assert get_session_env("HERMES_SESSION_PLATFORM") == ""
+
+
 def test_get_session_env_default_when_nothing_set(monkeypatch):
     """get_session_env returns default when neither contextvar nor env is set."""
     monkeypatch.delenv("HERMES_SESSION_PLATFORM", raising=False)
@@ -393,4 +413,3 @@ async def test_gateway_executor_refuses_resurrection_after_shutdown():
             await runner._run_in_executor_with_context(lambda: "second")
     finally:
         runner._shutdown_executor()
-

tests/tools/test_cron_approval_mode.py

@@ -7,6 +7,7 @@
     _get_cron_approval_mode,
     check_all_command_guards,
     check_dangerous_command,
+    check_execute_code_guard,
     detect_dangerous_command,
 )
 
@@ -188,6 +189,22 @@ def test_dangerous_command_blocked_in_combined_guard(self, monkeypatch):
             assert not result["approved"]
             assert "BLOCKED" in result["message"]
 
+    def test_exec_ask_does_not_bypass_cron_deny(self, monkeypatch):
+        monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+        monkeypatch.setenv("HERMES_EXEC_ASK", "1")
+        monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+        monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+        monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
+
+        from unittest.mock import patch as mock_patch
+
+        with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"):
+            result = check_all_command_guards("rm -rf /tmp/stuff", "local")
+
+        assert not result["approved"]
+        assert "BLOCKED" in result["message"]
+        assert result.get("status") != "pending_approval"
+
     def test_safe_command_allowed_in_combined_guard(self, monkeypatch):
         monkeypatch.setenv("HERMES_CRON_SESSION", "1")
         monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
@@ -380,7 +397,7 @@ def test_cron_with_telegram_origin_uses_cron_mode_not_gateway(self, monkeypatch)
         monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
 
         from gateway.session_context import set_session_vars, clear_session_vars
-        tokens = set_session_vars(platform="telegram", chat_id="123")
+        tokens = set_session_vars(platform="telegram", chat_id="123", cron_session="1")
         try:
             from unittest.mock import patch as mock_patch
             with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"):
@@ -402,7 +419,7 @@ def test_cron_with_telegram_origin_approve_mode_allows(self, monkeypatch):
         monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
 
         from gateway.session_context import set_session_vars, clear_session_vars
-        tokens = set_session_vars(platform="discord", chat_id="456")
+        tokens = set_session_vars(platform="discord", chat_id="456", cron_session="1")
         try:
             from unittest.mock import patch as mock_patch
             with mock_patch("tools.approval._get_cron_approval_mode", return_value="approve"):
@@ -422,7 +439,7 @@ def test_cron_with_telegram_origin_combined_guard_uses_cron_mode(self, monkeypat
         monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
 
         from gateway.session_context import set_session_vars, clear_session_vars
-        tokens = set_session_vars(platform="telegram", chat_id="789")
+        tokens = set_session_vars(platform="telegram", chat_id="789", cron_session="1")
         try:
             from unittest.mock import patch as mock_patch
             with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"):
@@ -432,3 +449,132 @@ def test_cron_with_telegram_origin_combined_guard_uses_cron_mode(self, monkeypat
                 assert result.get("status") != "approval_required"
         finally:
             clear_session_vars(tokens)
+
+
+class TestGatewayAfterCronSession:
+    """A cron marker left in process env must not disable live gateway approvals."""
+
+    def test_gateway_env_session_takes_precedence_over_cron_marker(self, monkeypatch):
+        monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+        monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1")
+        monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+        monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
+        monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
+
+        result = check_dangerous_command("rm -rf /tmp/stuff", "local")
+
+        assert not result["approved"]
+        assert result.get("status") == "approval_required"
+        assert "cron_mode" not in result["message"]
+
+    def test_combined_guard_gateway_env_session_takes_precedence_over_cron_marker(self, monkeypatch):
+        monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+        monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1")
+        monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+        monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
+        monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
+
+        result = check_all_command_guards("rm -rf /tmp/stuff", "local")
+
+        assert not result["approved"]
+        assert result.get("status") == "pending_approval"
+        assert result.get("approval_pending") is True
+        assert "cron_mode" not in result["message"]
+
+    def test_context_cron_session_takes_precedence_over_gateway_env_marker(self, monkeypatch):
+        monkeypatch.delenv("HERMES_CRON_SESSION", raising=False)
+        monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1")
+        monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+        monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
+        monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
+
+        from gateway.session_context import set_session_vars, clear_session_vars
+        tokens = set_session_vars(cron_session="1")
+        try:
+            from unittest.mock import patch as mock_patch
+            with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"):
+                result = check_all_command_guards("rm -rf /tmp/stuff", "local")
+        finally:
+            clear_session_vars(tokens)
+
+        assert not result["approved"]
+        assert "BLOCKED" in result["message"]
+        assert "cron_mode" in result["message"]
+        assert result.get("status") != "pending_approval"
+
+    def test_gateway_context_session_takes_precedence_over_cron_marker(self, monkeypatch):
+        monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+        monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+        monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+        monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
+        monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
+
+        from gateway.session_context import set_session_vars, clear_session_vars
+        tokens = set_session_vars(platform="telegram", session_key="ctx-session")
+        try:
+            result = check_dangerous_command("rm -rf /tmp/stuff", "local")
+        finally:
+            clear_session_vars(tokens)
+
+        assert not result["approved"]
+        assert result.get("status") == "approval_required"
+        assert "cron_mode" not in result["message"]
+
+    def test_combined_guard_gateway_context_session_takes_precedence_over_cron_marker(self, monkeypatch):
+        monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+        monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+        monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+        monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
+        monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
+
+        from gateway.session_context import set_session_vars, clear_session_vars
+        tokens = set_session_vars(platform="telegram", session_key="ctx-session")
+        try:
+            result = check_all_command_guards("rm -rf /tmp/stuff", "local")
+        finally:
+            clear_session_vars(tokens)
+
+        assert not result["approved"]
+        assert result.get("status") == "pending_approval"
+        assert result.get("approval_pending") is True
+        assert "cron_mode" not in result["message"]
+
+    def test_execute_code_gateway_context_session_takes_precedence_over_cron_marker(self, monkeypatch):
+        monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+        monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+        monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+        monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
+        monkeypatch.setenv("HERMES_EXEC_ASK", "1")
+
+        from gateway.session_context import set_session_vars, clear_session_vars
+        tokens = set_session_vars(platform="telegram", session_key="ctx-session")
+        try:
+            result = check_execute_code_guard("import os", "local")
+        finally:
+            clear_session_vars(tokens)
+
+        assert not result["approved"]
+        assert result.get("status") == "pending_approval"
+        assert result.get("approval_pending") is True
+        assert "cron profile is intentionally trusted" not in result["message"]
+
+    def test_execute_code_exec_ask_does_not_bypass_context_cron_deny(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_YOLO_MODE", raising=False)
+        monkeypatch.setenv("HERMES_EXEC_ASK", "1")
+
+        from gateway.session_context import clear_session_vars, set_session_vars
+        from unittest.mock import patch as mock_patch
+
+        tokens = set_session_vars(cron_session="1")
+        try:
+            with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"):
+                result = check_execute_code_guard("import os", "local")
+        finally:
+            clear_session_vars(tokens)
+
+        assert not result["approved"]
+        assert result.get("outcome") == "blocked"
+        assert "cron profile is intentionally trusted" in result["message"]