fix(cron): isolate approval context per job

cron/scheduler.py

@@ -3008,13 +3008,9 @@ 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).
+    # Use ContextVars for per-job session and delivery state so cron approval
+    # policy cannot leak into concurrent or later live gateway turns, and
+    # parallel jobs cannot clobber each other's targets.
     from gateway.session_context import set_session_vars, clear_session_vars, _VAR_MAP
 
     # Cron execution is an internal scheduler context, not a live inbound
@@ -3069,6 +3065,7 @@ def run_job(
         # See declare_stateless_channel(). Upstream: #53027, #63142.
         async_delivery=False,
         cwd=_job_workdir or "",
+        cron_session=True,
     )
     _cron_delivery_vars = (
         "HERMES_CRON_AUTO_DELIVER_PLATFORM",

gateway/session_context.py

@@ -115,8 +115,9 @@ def session_context_engaged() -> bool:
 # propagates that into this contextvar at session-bind time.
 _SESSION_ASYNC_DELIVERY: ContextVar = ContextVar("HERMES_SESSION_ASYNC_DELIVERY", default=_UNSET)
 
-# Cron auto-delivery vars — set per-job in run_job() so concurrent jobs
-# don't clobber each other's delivery targets.
+# Cron identity and auto-delivery vars are set per-job in run_job() so
+# concurrent jobs and live gateway turns cannot clobber each other's state.
+_CRON_SESSION: ContextVar = ContextVar("HERMES_CRON_SESSION", default=_UNSET)
 _CRON_AUTO_DELIVER_PLATFORM: ContextVar = ContextVar("HERMES_CRON_AUTO_DELIVER_PLATFORM", default=_UNSET)
 _CRON_AUTO_DELIVER_CHAT_ID: ContextVar = ContextVar("HERMES_CRON_AUTO_DELIVER_CHAT_ID", default=_UNSET)
 _CRON_AUTO_DELIVER_THREAD_ID: ContextVar = ContextVar("HERMES_CRON_AUTO_DELIVER_THREAD_ID", default=_UNSET)
@@ -135,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,
@@ -212,6 +214,7 @@ def set_session_vars(
     cwd: str = "",
     async_delivery: bool = True,
     ui_session_id: str = "",
+    cron_session: bool = False,
 ) -> list:
     """Set all session context variables and return reset tokens.
 
@@ -227,6 +230,9 @@ 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`` binds unattended approval policy to this task without a
+    process-global environment flag that could leak into live gateway turns.
     """
     # Mark the session-context machinery engaged for this process. The
     # subprocess-env bridge uses this to switch from "os.environ fallback" to
@@ -248,6 +254,7 @@ def set_session_vars(
         _SESSION_MESSAGE_ID.set(message_id),
         _SESSION_PROFILE.set(profile),
         _SESSION_ASYNC_DELIVERY.set(bool(async_delivery)),
+        _CRON_SESSION.set("1" if cron_session else ""),
     ]
     try:
         from agent.runtime_cwd import set_session_cwd
@@ -283,6 +290,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

hermes_cli/bang_shell.py

@@ -64,14 +64,23 @@ def bang_shell_enabled() -> bool:
     remote-execution surface with no approving human at the keyboard.
     """
     try:
-        from utils import env_var_enabled
+        from utils import env_var_enabled, is_truthy_value
     except Exception:  # pragma: no cover - utils is always importable in-tree
+        def is_truthy_value(value, default=False):  # type: ignore[misc]
+            return str(value).strip().lower() in {"1", "true", "yes", "on"}
+
         def env_var_enabled(name, default=""):  # type: ignore[misc]
-            return str(os.getenv(name, default)).strip().lower() in {"1", "true", "yes", "on"}
+            return is_truthy_value(os.getenv(name, default))
 
     if env_var_enabled("HERMES_GATEWAY_SESSION"):
         return False
-    if env_var_enabled("HERMES_CRON_SESSION"):
+    try:
+        from gateway.session_context import get_session_env
+
+        cron_session = get_session_env("HERMES_CRON_SESSION", "")
+    except Exception:
+        cron_session = os.getenv("HERMES_CRON_SESSION", "")
+    if is_truthy_value(cron_session):
         return False
     if (os.getenv("HERMES_SESSION_PLATFORM") or "").strip():
         return False

tests/cli/test_bang_shell_mode.py

@@ -85,6 +85,16 @@ def test_disabled_in_non_cli_contexts(self, monkeypatch, var, value):
         monkeypatch.setenv(var, value)
         assert bang_shell_enabled() is False
 
+    def test_disabled_in_task_local_cron_context(self, monkeypatch):
+        monkeypatch.delenv("HERMES_CRON_SESSION", raising=False)
+        from gateway.session_context import clear_session_vars, set_session_vars
+
+        tokens = set_session_vars(cron_session=True)
+        try:
+            assert bang_shell_enabled() is False
+        finally:
+            clear_session_vars(tokens)
+
 
 # ── execution ──────────────────────────────────────────────────────────────
 

tests/cron/test_cron_approval_context.py

@@ -0,0 +1,60 @@
+"""Regression coverage for task-local cron approval identity."""
+
+import os
+from unittest.mock import MagicMock, patch
+
+from cron.scheduler import run_job
+
+_RUNTIME = {
+    "api_key": "test-key",
+    "base_url": "https://example.invalid/v1",
+    "provider": "openrouter",
+    "api_mode": "chat_completions",
+}
+
+
+def test_run_job_scopes_cron_identity_without_process_leak(tmp_path, monkeypatch):
+    """Cron policy is active during the run and absent from later gateway turns."""
+    monkeypatch.delenv("HERMES_CRON_SESSION", raising=False)
+    observed = {}
+
+    def run_conversation(_prompt):
+        from gateway.session_context import get_session_env
+        from tools.approval import _is_cron_approval_context
+
+        observed["context_value"] = get_session_env("HERMES_CRON_SESSION", "")
+        observed["approval_context"] = _is_cron_approval_context()
+        observed["process_value"] = os.environ.get("HERMES_CRON_SESSION")
+        return {"final_response": "ok"}
+
+    job = {"id": "cron-context", "name": "test", "prompt": "hello"}
+    fake_db = MagicMock()
+
+    with (
+        patch("cron.scheduler._hermes_home", tmp_path),
+        patch("cron.scheduler._resolve_origin", return_value=None),
+        patch("hermes_cli.env_loader.load_hermes_dotenv"),
+        patch("hermes_cli.env_loader.reset_secret_source_cache"),
+        patch("hermes_state.SessionDB", return_value=fake_db),
+        patch(
+            "hermes_cli.runtime_provider.resolve_runtime_provider",
+            return_value=_RUNTIME,
+        ),
+        patch("tools.mcp_tool.discover_mcp_tools", return_value=[]),
+        patch("run_agent.AIAgent") as mock_agent_cls,
+    ):
+        mock_agent = MagicMock()
+        mock_agent.run_conversation.side_effect = run_conversation
+        mock_agent_cls.return_value = mock_agent
+
+        success, _output, final_response, error = run_job(job)
+
+    assert success is True
+    assert final_response == "ok"
+    assert error is None
+    assert observed == {
+        "context_value": "1",
+        "approval_context": True,
+        "process_value": None,
+    }
+    assert os.environ.get("HERMES_CRON_SESSION") is None

tests/tools/test_cron_approval_mode.py

@@ -13,10 +13,14 @@
 
 @pytest.fixture(autouse=True)
 def _clear_approval_state():
+    from gateway.session_context import reset_session_vars
+
+    reset_session_vars()
     approval_module._permanent_approved.clear()
     approval_module.clear_session("default")
     approval_module.clear_session("test-session")
     yield
+    reset_session_vars()
     approval_module._permanent_approved.clear()
     approval_module.clear_session("default")
     approval_module.clear_session("test-session")
@@ -360,27 +364,44 @@ def test_non_cron_non_interactive_still_auto_approves(self, monkeypatch):
         assert result["approved"]
 
 
+class TestCronContextIsolation:
+    """Cron identity must be task-local in a concurrent gateway process."""
+
+    def test_stale_process_cron_flag_does_not_override_gateway_context(self, monkeypatch):
+        """A completed cron run must not force later Discord approvals onto CLI."""
+        monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+        monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+
+        from gateway.session_context import set_session_vars, clear_session_vars
+        from tools.approval import _is_gateway_approval_context
+
+        tokens = set_session_vars(platform="discord", chat_id="456")
+        try:
+            assert _is_gateway_approval_context() is True
+        finally:
+            clear_session_vars(tokens)
+
+
 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.
+    """Cron jobs must not be treated as live gateway approval contexts.
+
+    Cron identity is bound independently from delivery routing. Even if a cron
+    execution carries a platform context, it must use approvals.cron_mode
+    rather than submit a pending approval with no live listener.
     """
 
     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=True
+        )
         try:
             from unittest.mock import patch as mock_patch
             with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"):
@@ -395,14 +416,16 @@ 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=True
+        )
         try:
             from unittest.mock import patch as mock_patch
             with mock_patch("tools.approval._get_cron_approval_mode", return_value="approve"):

tests/tools/test_mcp_elicitation.py

@@ -221,6 +221,40 @@ def fake_consent(*_args, **_kwargs):
             f"consent call; got {seen!r}"
         )
 
+    def test_gateway_context_ignores_stale_process_cron_identity(self, monkeypatch):
+        """A prior cron run must not route Discord elicitation to CLI."""
+        import contextvars
+        from types import SimpleNamespace
+
+        from gateway.session_context import clear_session_vars, set_session_vars
+        from tools.approval import _is_gateway_approval_context
+
+        monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+        tokens = set_session_vars(
+            platform="discord", chat_id="456", session_key="discord:456"
+        )
+        try:
+            captured = contextvars.copy_context()
+        finally:
+            clear_session_vars(tokens)
+
+        seen = []
+
+        def fake_consent(*_args, **_kwargs):
+            seen.append(_is_gateway_approval_context())
+            return "accept" if seen[-1] else "decline"
+
+        owner = SimpleNamespace(_pending_call_context=captured)
+        handler = ElicitationHandler("soul_broker", {"timeout": 5}, owner=owner)
+
+        with patch(
+            "tools.approval.request_elicitation_consent", side_effect=fake_consent
+        ):
+            result = asyncio.run(handler(context=None, params=_form_params()))
+
+        assert result.action == "accept"
+        assert seen == [True]
+
     def test_missing_captured_context_falls_back_to_direct_call(self):
         """Without an owner (or with an owner that hasn't entered a tool
         call) the handler must still invoke the consent router -- just

tests/tools/test_request_tool_approval.py

@@ -77,8 +77,7 @@ def test_cli_session_persists_session_only(self, monkeypatch):
     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 @@ def test_cron_deny_mode_blocks(self, monkeypatch):
     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

tools/approval.py

@@ -224,6 +224,16 @@ def _get_session_platform() -> str:
         return os.getenv("HERMES_SESSION_PLATFORM", "") or ""
 
 
+def _is_cron_approval_context() -> bool:
+    """Return task-local cron identity with env fallback for legacy callers."""
+    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 +248,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
@@ -2899,7 +2909,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,
@@ -3427,7 +3437,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)
@@ -3878,7 +3888,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,