fix(gateway): a local turn reloads the profile's .env, as a messaging turn does

A messaging turn reloads .env before it runs (inside _current_max_iterations); a local session's turn took its saved turn budget instead and skipped that reload with it, so a key rotated in .env reached it only after a gateway restart. Side tasks now read config.yaml, whose ${VAR} references expand against the environment, so a local turn runs the same reload. Its turn budget stays the session's own.

gateway/run_turn_runner.py

@@ -1245,7 +1245,10 @@ class TurnRunner(GatewayTurnProgressMixin, GatewaySessionAgentMixin):
         every rebind. session_key propagates via contextvars (_set_session_env / set_current_session_key)
         — never os.environ["HERMES_SESSION_KEY"], which would misroute approvals across sessions.
         """
-        from gateway.run import _current_max_iterations, _normalize_empty_agent_response, _sanitize_gateway_final_response
+        from gateway.run import (
+            _current_max_iterations, _normalize_empty_agent_response, _reload_runtime_env_preserving_config_authority,
+            _sanitize_gateway_final_response,
+        )
         ctx = self._ctx
         runner = self._runner
         # Platform.LOCAL ("local") maps to the "cli" hint key the agent understands.
@@ -1268,7 +1271,14 @@ class TurnRunner(GatewayTurnProgressMixin, GatewaySessionAgentMixin):
         combined_ephemeral = self._combined_ephemeral_prompt()
         if policy and policy.skills_prompt:
             combined_ephemeral = "\n\n".join(p for p in (combined_ephemeral, policy.skills_prompt) if p)
-        max_iterations = policy.max_turns if policy else _current_max_iterations()
+        if policy:
+            # Rotated .env keys reach this turn's config reads (side tasks read config.yaml), as on the
+            # messaging path, where _current_max_iterations() does the same reload; the turn budget
+            # stays the session's own.
+            _reload_runtime_env_preserving_config_authority()
+            max_iterations = policy.max_turns
+        else:
+            max_iterations = _current_max_iterations()
         from gateway.hosted_room_execution_policy import current_room_execution_policy
         room = current_room_execution_policy()
         if room is not None:

tests/gateway/test_session_config_loaders.py

@@ -48,6 +48,43 @@ def test_in_process_turn_routes_with_the_sessions_own_provider_routing(tmp_path,
     assert caught.value.args[0] == {'only': ['Fresh'], 'sort': 'price'}
 
 
+def test_a_local_turn_reloads_the_profiles_env_file_and_keeps_its_own_turn_budget(tmp_path, monkeypatch):
+    """A local session's turn sees a key rotated in ``.env`` (side tasks read config.yaml, whose ``${VAR}``
+    references expand against the environment), as a messaging turn does; its turn budget stays the
+    session's own, never the profile's ``agent.max_turns``."""
+    import os
+    from gateway import run as gateway_run, session_policy
+    from gateway.config import Platform
+    from gateway.run_turn_runner import TurnRunner
+    from gateway.session_policy import build_policy
+    home = tmp_path / 'home'
+    home.mkdir()
+    (home / 'config.yaml').write_text(json.dumps({'agent': {'max_turns': 90}}))
+    (home / '.env').write_text('SIDE_TASK_ROTATED_KEY=rotated-in-env-file\n')
+    monkeypatch.setattr(gateway_run, '_hermes_home', home)
+    monkeypatch.setenv('SIDE_TASK_ROTATED_KEY', 'value-the-gateway-started-with')
+    policy = build_policy({'cwd': str(tmp_path), **LAUNCH, 'max_turns': 7}, {})
+    monkeypatch.setattr(session_policy, 'policy_for_source', lambda runner, source: policy)
+    runner = SimpleNamespace(
+        _provider_routing={}, _peek_session_state=lambda key: None,
+        session_authority=SimpleNamespace(profile_id=str(home)),
+        _resolve_session_agent_runtime=lambda **kw: ('m', {'provider': 'custom'}),
+        _resolve_turn_agent_config=lambda message, model, runtime: {'model': model, 'runtime': runtime})
+    turn = TurnRunner.__new__(TurnRunner)
+    turn._runner = runner
+    turn._ctx = SimpleNamespace(source=SimpleNamespace(platform=Platform.LOCAL, chat_id='c'), session_key='k',
+                                user_config={}, message='hi', inbound_message_id=None, enabled_toolsets=[])
+    turn._combined_ephemeral_prompt = lambda: None
+    turn._setup_stream_consumer = lambda platform_key: (None, None, None, False)
+
+    def built(route, platform_key, ephemeral, max_iterations, reasoning, pr):
+        raise _Built(max_iterations, os.environ.get('SIDE_TASK_ROTATED_KEY'))
+    turn._resolve_turn_agent = built
+    with pytest.raises(_Built) as caught:
+        turn._run_sync_scoped()
+    assert caught.value.args == (7, 'rotated-in-env-file')
+
+
 def test_each_new_session_reads_its_profiles_prefill_once_and_workers_match(tmp_path, monkeypatch):
     """The prefill path and file contents a session gets are its own profile's when it was created; a
     later edit reaches the next new session and never an existing one (main: one read per CLI