fix: isolate cron approval context

cron/scheduler.py

@@ -1396,11 +1396,6 @@ def _run_job_impl(job: dict) -> tuple[bool, str, str, Optional[str]]:
 
     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
@@ -1430,6 +1425,7 @@ def _run_job_impl(job: dict) -> tuple[bool, str, str, Optional[str]]:
         platform="",
         chat_id="",
         chat_name="",
+        cron_session="1",
     )
     _cron_delivery_vars = (
         "HERMES_CRON_AUTO_DELIVER_PLATFORM",

gateway/session_context.py

@@ -61,6 +61,10 @@
 # private-chat topic (those lanes route only with thread id + reply anchor).
 _SESSION_MESSAGE_ID: ContextVar = ContextVar("HERMES_SESSION_MESSAGE_ID", default=_UNSET)
 
+# Cron execution marker.  Cron jobs can run inside the long-lived gateway
+# process, so this must be context-local instead of process-global.
+_CRON_SESSION: ContextVar = ContextVar("HERMES_CRON_SESSION", default=_UNSET)
+
 # Cron auto-delivery vars — set per-job in run_job() so concurrent jobs
 # don't clobber each other's delivery targets.
 _CRON_AUTO_DELIVER_PLATFORM: ContextVar = ContextVar("HERMES_CRON_AUTO_DELIVER_PLATFORM", default=_UNSET)
@@ -77,6 +81,7 @@
     "HERMES_SESSION_KEY": _SESSION_KEY,
     "HERMES_SESSION_ID": _SESSION_ID,
     "HERMES_SESSION_MESSAGE_ID": _SESSION_MESSAGE_ID,
+    "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,
@@ -107,6 +112,7 @@ def set_session_vars(
     user_name: str = "",
     session_key: str = "",
     message_id: str = "",
+    cron_session: str = "",
 ) -> list:
     """Set all session context variables and return reset tokens.
 
@@ -125,6 +131,7 @@ def set_session_vars(
         _SESSION_USER_NAME.set(user_name),
         _SESSION_KEY.set(session_key),
         _SESSION_MESSAGE_ID.set(message_id),
+        _CRON_SESSION.set(cron_session),
     ]
     return tokens
 
@@ -149,6 +156,7 @@ def clear_session_vars(tokens: list) -> None:
         _SESSION_USER_NAME,
         _SESSION_KEY,
         _SESSION_MESSAGE_ID,
+        _CRON_SESSION,
     ):
         var.set("")
 

tests/gateway/test_session_env.py

@@ -124,6 +124,19 @@ def test_get_session_env_falls_back_to_os_environ(monkeypatch):
     assert get_session_env("HERMES_SESSION_PLATFORM") == ""
 
 
+def test_live_session_masks_stale_cron_env(monkeypatch):
+    """Gateway session context should hide stale process-level cron markers."""
+    monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+
+    assert get_session_env("HERMES_CRON_SESSION") == "1"
+
+    tokens = set_session_vars(platform="discord", session_key="live-session")
+    assert get_session_env("HERMES_CRON_SESSION") == ""
+
+    clear_session_vars(tokens)
+    assert get_session_env("HERMES_CRON_SESSION") == ""
+
+
 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)

tests/tools/test_approval.py

@@ -1345,17 +1345,31 @@ def setup_method(self):
         self._saved_env = {
             k: os.environ.get(k)
             for k in ("HERMES_GATEWAY_SESSION", "HERMES_YOLO_MODE",
-                      "HERMES_SESSION_KEY", "HERMES_INTERACTIVE")
+                      "HERMES_SESSION_KEY", "HERMES_INTERACTIVE",
+                      "HERMES_CRON_SESSION")
         }
+        try:
+            from gateway.session_context import _UNSET, _VAR_MAP
+            for var in _VAR_MAP.values():
+                var.set(_UNSET)
+        except Exception:
+            pass
         os.environ.pop("HERMES_YOLO_MODE", None)
         os.environ.pop("HERMES_INTERACTIVE", None)
+        os.environ.pop("HERMES_CRON_SESSION", None)
         os.environ["HERMES_GATEWAY_SESSION"] = "1"
         os.environ["HERMES_SESSION_KEY"] = self.SESSION_KEY
 
     def teardown_method(self):
         from tools import approval as mod
         mod._gateway_queues.clear()
         mod._gateway_notify_cbs.clear()
+        try:
+            from gateway.session_context import _UNSET, _VAR_MAP
+            for var in _VAR_MAP.values():
+                var.set(_UNSET)
+        except Exception:
+            pass
         for k, v in self._saved_env.items():
             if v is None:
                 os.environ.pop(k, None)

tests/tools/test_cron_approval_mode.py

@@ -20,6 +20,9 @@ def _clear_approval_state():
     approval_module._permanent_approved.clear()
     approval_module.clear_session("default")
     approval_module.clear_session("test-session")
+    from gateway.session_context import _UNSET, _VAR_MAP
+    for var in _VAR_MAP.values():
+        var.set(_UNSET)
 
 
 # ---------------------------------------------------------------------------
@@ -200,6 +203,27 @@ def test_safe_command_allowed_in_combined_guard(self, monkeypatch):
             result = check_all_command_guards("echo hello", "local")
             assert result["approved"]
 
+    def test_contextvar_cron_session_blocks_without_process_env(self, monkeypatch):
+        """A real cron context should use cron_mode even without process-global env."""
+        monkeypatch.delenv("HERMES_CRON_SESSION", raising=False)
+        monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+        monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+        monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
+        monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
+
+        from gateway.session_context import set_session_vars, clear_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_all_command_guards("rm -rf /tmp/stuff", "local")
+                assert not result["approved"]
+                assert "BLOCKED" in result["message"]
+                assert "cron_mode" in result["message"]
+        finally:
+            clear_session_vars(tokens)
+
     def test_combined_guard_approve_mode(self, monkeypatch):
         monkeypatch.setenv("HERMES_CRON_SESSION", "1")
         monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
@@ -270,24 +294,22 @@ def test_non_cron_non_interactive_still_auto_approves(self, monkeypatch):
 class TestCronWithGatewayOrigin:
     """Cron jobs originating from a gateway platform must NOT be treated as gateway.
 
-    cron/scheduler.py binds HERMES_SESSION_PLATFORM via contextvars for
-    delivery routing (so cron output lands back in the origin chat). The
-    API-server approvals work (PR #20311) made check_dangerous_command treat
-    any contextvar-bound platform as a gateway session. That would route
-    cron-from-telegram/discord/etc. through submit_pending with no listener,
-    hanging the job instead of respecting approvals.cron_mode.
+    If a cron context ever carries platform metadata for delivery or logging,
+    the explicit cron context marker must still win over gateway approval
+    routing. Otherwise cron-from-telegram/discord/etc. could submit a pending
+    approval with no listener instead of respecting approvals.cron_mode.
     """
 
     def test_cron_with_telegram_origin_uses_cron_mode_not_gateway(self, monkeypatch):
         """Cron + contextvar platform=telegram + cron_mode=deny → BLOCKED, not pending."""
-        monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+        monkeypatch.delenv("HERMES_CRON_SESSION", raising=False)
         monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
         monkeypatch.delenv("HERMES_GATEWAY_SESSION", 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", 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"):
@@ -302,14 +324,14 @@ def test_cron_with_telegram_origin_uses_cron_mode_not_gateway(self, monkeypatch)
 
     def test_cron_with_telegram_origin_approve_mode_allows(self, monkeypatch):
         """Cron + contextvar platform=telegram + cron_mode=approve → allowed via cron path."""
-        monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+        monkeypatch.delenv("HERMES_CRON_SESSION", raising=False)
         monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
         monkeypatch.delenv("HERMES_GATEWAY_SESSION", 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="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"):
@@ -322,14 +344,14 @@ def test_cron_with_telegram_origin_approve_mode_allows(self, monkeypatch):
 
     def test_cron_with_telegram_origin_combined_guard_uses_cron_mode(self, monkeypatch):
         """check_all_command_guards must also honor cron_mode over gateway classification."""
-        monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+        monkeypatch.delenv("HERMES_CRON_SESSION", raising=False)
         monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
         monkeypatch.delenv("HERMES_GATEWAY_SESSION", 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", 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"):
@@ -339,3 +361,44 @@ 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 TestCronFlagLeakDoesNotPoisonLiveGateway:
+    """A stale process-level cron flag must not reclassify live gateway turns."""
+
+    def test_live_discord_command_uses_gateway_approval_despite_stale_cron_env(self, monkeypatch):
+        monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+        monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+        monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+        monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
+        monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
+
+        from gateway.session_context import set_session_vars, clear_session_vars
+        from unittest.mock import patch as mock_patch
+
+        session_key = "agent:main:discord:dm:live-user"
+        notified = []
+
+        def notify(data):
+            notified.append(data)
+            approval_module.resolve_gateway_approval(session_key, "once")
+
+        tokens = set_session_vars(
+            platform="discord",
+            chat_id="live-chat",
+            session_key=session_key,
+        )
+        approval_module.register_gateway_notify(session_key, notify)
+        try:
+            with (
+                mock_patch("tools.approval._get_approval_mode", return_value="manual"),
+                mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"),
+            ):
+                result = check_all_command_guards("rm -rf /tmp/stuff", "local")
+                assert result["approved"] is True
+                assert result.get("user_approved") is True
+                assert len(notified) == 1
+                assert "cron" not in (result.get("message") or "").lower()
+        finally:
+            approval_module.unregister_gateway_notify(session_key)
+            clear_session_vars(tokens)

tests/tools/test_execute_code_approval_cluster.py

@@ -25,6 +25,14 @@
 from tools.thread_context import propagate_context_to_thread
 
 
+@pytest.fixture(autouse=True)
+def _reset_session_contextvars():
+    yield
+    from gateway.session_context import _UNSET, _VAR_MAP
+    for var in _VAR_MAP.values():
+        var.set(_UNSET)
+
+
 # ---------------------------------------------------------------------------
 # 1. Context + callback propagation helper
 # ---------------------------------------------------------------------------
@@ -167,6 +175,30 @@ def test_guard_cron_deny_blocks(monkeypatch):
     assert res["outcome"] == "blocked"
 
 
+def test_guard_live_gateway_ignores_stale_cron_env(monkeypatch, gw_session):
+    """Stale process cron env must not block live gateway execute_code approval."""
+    from gateway.session_context import clear_session_vars, set_session_vars
+
+    monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+    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(
+        platform="discord",
+        chat_id="live-chat",
+        session_key=gw_session,
+    )
+    try:
+        _register_resolver(gw_session, "once")
+        res = A.check_execute_code_guard("import os; print(1)", "local")
+        assert res["approved"] is True
+        assert res.get("user_approved") is True
+        assert res.get("outcome") != "blocked"
+    finally:
+        clear_session_vars(tokens)
+
+
 def test_guard_gateway_user_approves_is_one_shot(gw_session):
     _register_resolver(gw_session, "once")
     res = A.check_execute_code_guard("import os; print(1)", "local")

tools/approval.py

@@ -99,21 +99,36 @@ def _get_session_platform() -> str:
         return os.getenv("HERMES_SESSION_PLATFORM", "") or ""
 
 
+def _is_cron_approval_context() -> bool:
+    """True when this approval decision belongs to a cron job.
+
+    Prefer the context-local marker set by ``cron.scheduler.run_job`` so a
+    scheduler tick in the long-lived gateway process cannot poison later live
+    Discord/Telegram/etc. sessions through ``os.environ``.  The env fallback is
+    kept for standalone cron/test callers that do not bind session contextvars.
+    """
+    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.
 
     Legacy gateway integrations set HERMES_GATEWAY_SESSION in process env.
     Newer concurrent gateway paths bind HERMES_SESSION_PLATFORM via
     contextvars so approval mode does not depend on process-global flags.
 
-    Cron jobs are NEVER gateway-approval contexts even when they originate
-    from a gateway platform (cron binds HERMES_SESSION_PLATFORM via
-    contextvars for delivery routing). Cron approvals are governed by
-    ``approvals.cron_mode`` config, not interactive resolve — letting cron
-    fall through to the gateway branch would submit a pending approval
-    with no listener and block the job indefinitely.
+    Cron jobs are NEVER gateway-approval contexts, even if they carry platform
+    metadata for delivery or logging. Cron approvals are governed by
+    ``approvals.cron_mode`` config, not interactive resolve — letting cron 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
@@ -968,7 +983,7 @@ def check_dangerous_command(command: str, env_type: str,
 
     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,
@@ -1205,7 +1220,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)
@@ -1491,7 +1506,7 @@ def check_execute_code_guard(code: str, env_type: str) -> dict:
     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,