feat: unify agent isolation (--ignore-rules / --safe-mode) across TUI, messaging gateway, and API server

acp_adapter/session.py

@@ -393,6 +393,10 @@ def _make_agent(self, *, session_id: str, cwd: str, model: str | None = None,
             "enabled_toolsets": _expand_acp_enabled_toolsets(["hermes-acp"], mcp_server_names=configured_mcp_servers),
             "model": model or default_model,
         }
+        from agent.isolation import resolve_agent_isolation
+
+        isolated = resolve_agent_isolation()
+        kwargs.update(skip_context_files=isolated, skip_memory=isolated)
         try:
             runtime = resolve_runtime_provider(requested=requested_provider or config_provider)
             kwargs.update({

agent/isolation.py

@@ -0,0 +1,52 @@
+"""Shared agent isolation contract.
+
+The CLI, TUI, messaging gateway, and API server all construct ``AIAgent``
+instances for user sessions.  Isolation — skipping auto-injected rules
+(``AGENTS.md`` / ``SOUL.md`` / ``.cursorrules``) and memory — must behave
+identically across every entry point, otherwise ``--ignore-rules`` and
+``--safe-mode`` only affect a subset of surfaces.
+
+Resolution precedence:
+
+1. Explicitly passed flags (``ignore_rules`` / ``safe_mode``) when either is
+   provided (not ``None``) — an explicit ``False`` deliberately overrides a
+   process-level env var so callers can force isolation off;
+2. environment variables ``HERMES_IGNORE_RULES`` / ``HERMES_SAFE_MODE``
+   (``HERMES_SAFE_MODE`` implies ignore-rules);
+3. default ``False``.
+
+The CLI normalizes its flags into those env vars at startup
+(``hermes_cli/main.py``), so entry points that construct agents inside an
+already-started process can rely on the env path; direct callers can pass
+explicit flags instead.
+"""
+
+from __future__ import annotations
+
+from utils import env_var_enabled
+
+
+def resolve_agent_isolation(
+    *,
+    ignore_rules: bool | None = None,
+    safe_mode: bool | None = None,
+) -> bool:
+    """Return whether the shared agent-isolation contract is enabled.
+
+    Callers apply this one decision to both ``skip_context_files`` and
+    ``skip_memory`` because the isolation contract skips rules and memory
+    together.
+
+    Args:
+        ignore_rules: explicit ``--ignore-rules``-style flag; ``None`` means
+            "not provided" and falls through to the env/default chain.
+        safe_mode: explicit ``--safe-mode``-style flag; implies
+            ``ignore_rules`` when set.
+    """
+    if ignore_rules is not None or safe_mode is not None:
+        isolated = bool(ignore_rules or safe_mode)
+    else:
+        isolated = env_var_enabled("HERMES_IGNORE_RULES") or env_var_enabled(
+            "HERMES_SAFE_MODE"
+        )
+    return isolated

gateway/platforms/api_server.py

@@ -2174,6 +2174,10 @@ def _create_agent(
             "fallback_model": None if confirmed_runtime_lock else GatewayRunner._load_fallback_model(),
             "reasoning_config": request_reasoning_config,
             "gateway_session_key": gateway_session_key}
+        from agent.isolation import resolve_agent_isolation
+
+        isolated = resolve_agent_isolation()
+        agent_kwargs.update(skip_context_files=isolated, skip_memory=isolated)
         if request_service_tier is not _REQUEST_OPTION_MISSING:
             agent_kwargs["service_tier"] = request_service_tier
         agent = AIAgent(**agent_kwargs)

gateway/run.py

@@ -2811,6 +2811,16 @@ def _checkpoint_agent_kwargs(config: dict | None) -> dict:
         "checkpoint_max_file_size_mb": cp_cfg.get("max_file_size_mb", defaults["max_file_size_mb"])}
 
 
+def _resolve_gateway_isolation_skip_flags(
+    platform_skip_context: bool = False,
+) -> tuple[bool, bool]:
+    """Compose the platform context opt-out with process-wide isolation."""
+    from agent.isolation import resolve_agent_isolation
+
+    isolated = resolve_agent_isolation()
+    return bool(platform_skip_context or isolated), isolated
+
+
 def _resolve_gateway_model(config: dict | None = None) -> str:
     """Read model from config.yaml (single source of truth), else temporary AIAgents (e.g. /compress)
     use the hardcoded default, which fails under openai-codex."""

gateway/run_agent_cache.py

@@ -94,6 +94,7 @@ def _agent_config_signature(
         model: str, runtime: dict, enabled_toolsets: list, ephemeral_prompt: str,
         cache_keys: dict | None = None, user_id: str | None = None, user_id_alt: str | None = None,
         skip_context_files: bool = False,
+        skip_memory: bool = False,
     ) -> str:
         """Stable key from agent config: change → cached AIAgent rebuilt; unchanged → reused (frozen
         prompt + schemas for cache hits). ``user_id`` / ``user_id_alt`` participate because Honcho
@@ -128,6 +129,8 @@ def _agent_config_signature(
                 # skip_context_files changes the agent's frozen system prompt (context files in vs out):
                 # a toggled edit must rebuild the cached agent, not silently reuse it.
                 bool(skip_context_files),
+                # Memory-provider state is also frozen at construction.
+                bool(skip_memory),
             ],
             sort_keys=True, default=str,
         )

gateway/run_turn.py

@@ -2189,6 +2189,17 @@ async def _run_background_task_inner(
 
             platform_key = _platform_config_key(source.platform)
             enabled_toolsets, disabled_toolsets = self._resolve_turn_toolsets(user_config, source, platform_key)
+            from gateway.run import _resolve_gateway_isolation_skip_flags
+
+            platforms_cfg = (user_config.get("gateway") or {}).get("platforms") or {}
+            platform_skip_context = False
+            if isinstance(platforms_cfg, dict):
+                platform_cfg = platforms_cfg.get(platform_key) or {}
+                if isinstance(platform_cfg, dict):
+                    platform_skip_context = bool(platform_cfg.get("skip_context_files"))
+            skip_context_files, skip_memory = _resolve_gateway_isolation_skip_flags(
+                platform_skip_context
+            )
             pr = self._provider_routing
             max_iterations = _current_max_iterations()
             reasoning_config = self._resolve_session_reasoning_config(source=source, model=model)
@@ -2236,6 +2247,8 @@ def run_sync():
                     # Reload from disk — do not reuse the startup snapshot.
                     # See #60955.
                     fallback_model=self._refresh_fallback_model(),
+                    skip_context_files=skip_context_files,
+                    skip_memory=skip_memory,
                 )
                 try:
                     return agent.run_conversation(user_message=enriched_prompt, task_id=task_id)

gateway/run_turn_runner.py

@@ -1019,7 +1019,7 @@ def _release_evicted_agent(self, agent) -> None:
         )
 
     def _build_fresh_agent(self, turn_route, platform_key, combined_ephemeral, max_iterations,
-                           reasoning_config, pr, skip_context_files):
+                           reasoning_config, pr, skip_context_files, skip_memory):
         from gateway.run import _checkpoint_agent_kwargs
         ctx = self._ctx
         runner = self._runner
@@ -1044,6 +1044,7 @@ def _build_fresh_agent(self, turn_route, platform_key, combined_ephemeral, max_i
             # See #60955.
             fallback_model=self._runner._refresh_fallback_model(),
             skip_context_files=skip_context_files,
+            skip_memory=skip_memory,
             # Keep the persona even with minimal context: soul identity is one small file.
             load_soul_identity=True,
         )
@@ -1053,13 +1054,18 @@ def _resolve_turn_agent(self, turn_route, platform_key, combined_ephemeral, max_
         hits) or build a fresh one. Returns (agent, reused_cached_agent)."""
         ctx = self._ctx
         runner = self._runner
-        skip_context_files = self._skip_context_files(platform_key)
+        from gateway.run import _resolve_gateway_isolation_skip_flags
+
+        skip_context_files, skip_memory = _resolve_gateway_isolation_skip_flags(
+            self._skip_context_files(platform_key)
+        )
         sig = runner._agent_config_signature(
             turn_route["model"], turn_route["runtime"], ctx.enabled_toolsets, combined_ephemeral,
             cache_keys=runner._extract_cache_busting_config(ctx.user_config),
             user_id=getattr(ctx.source, "user_id", None),
             user_id_alt=getattr(ctx.source, "user_id_alt", None),
             skip_context_files=skip_context_files,
+            skip_memory=skip_memory,
         )
         cache_lock = getattr(runner, "_agent_cache_lock", None)
         cache = getattr(runner, "_agent_cache", None)
@@ -1076,7 +1082,8 @@ def _resolve_turn_agent(self, turn_route, platform_key, combined_ephemeral, max_
             self._release_evicted_agent(found.evicted)
         if agent is None:
             agent = self._build_fresh_agent(
-                turn_route, platform_key, combined_ephemeral, max_iterations, reasoning_config, pr, skip_context_files,
+                turn_route, platform_key, combined_ephemeral, max_iterations, reasoning_config, pr,
+                skip_context_files, skip_memory,
             )
             if cache_lock and cache is not None:
                 with cache_lock:

hermes_cli/cli_commands_mixin.py

@@ -1938,6 +1938,9 @@ def produce():
             with suppress(Exception):
                 set_secret_capture_callback(self._secret_capture_callback)
             try:
+                from agent.isolation import resolve_agent_isolation
+
+                isolated = resolve_agent_isolation()
                 bg_agent = AIAgent(
                     model=turn_route["model"], acp_command=runtime.get("command"),
                     acp_args=runtime.get("args"), max_iterations=self.max_turns,
@@ -1947,6 +1950,8 @@ def produce():
                     session_db=self._session_db, reasoning_config=self.reasoning_config,
                     service_tier=self.service_tier,
                     request_overrides=turn_route.get("request_overrides"),
+                    skip_context_files=isolated,
+                    skip_memory=isolated,
                     **{kw: getattr(self, attr) for kw, attr in _BG_PROVIDER_KWARGS.items()})
                 # Silence raw spinner; route thinking through TUI widget when no foreground agent is active.
                 bg_agent._print_fn = lambda *_a, **_kw: None

tests/acp_adapter/test_session.py

@@ -110,6 +110,75 @@ def __init__(self, **kwargs):
 
         assert state.agent.session_cwd == "/tmp/project"
 
+    def test_make_agent_forwards_isolation_env(self, monkeypatch):
+        class FakeAgent:
+            model = "fake-model"
+
+            def __init__(self, **kwargs):
+                self.kwargs = kwargs
+
+        monkeypatch.setattr("run_agent.AIAgent", FakeAgent)
+        monkeypatch.setattr(
+            "acp_adapter.session.load_config",
+            lambda: {"model": {"default": "fake-model"}, "mcp_servers": {}},
+            raising=False,
+        )
+        monkeypatch.setattr(
+            "hermes_cli.config.load_config",
+            lambda: {"model": {"default": "fake-model"}, "mcp_servers": {}},
+        )
+        monkeypatch.setattr(
+            "hermes_cli.runtime_provider.resolve_runtime_provider",
+            lambda requested=None: {
+                "provider": requested,
+                "api_mode": "codex_app_server",
+                "base_url": "https://example.invalid",
+                "api_key": "test-key",
+            },
+        )
+        monkeypatch.setattr("acp_adapter.session._register_task_cwd", lambda task_id, cwd: None)
+        monkeypatch.setenv("HERMES_SAFE_MODE", "1")
+
+        state = SessionManager(db=None).create_session(cwd="/tmp/project")
+
+        assert state.agent.kwargs["skip_context_files"] is True
+        assert state.agent.kwargs["skip_memory"] is True
+
+    def test_make_agent_defaults_isolation_off(self, monkeypatch):
+        class FakeAgent:
+            model = "fake-model"
+
+            def __init__(self, **kwargs):
+                self.kwargs = kwargs
+
+        monkeypatch.setattr("run_agent.AIAgent", FakeAgent)
+        monkeypatch.setattr(
+            "acp_adapter.session.load_config",
+            lambda: {"model": {"default": "fake-model"}, "mcp_servers": {}},
+            raising=False,
+        )
+        monkeypatch.setattr(
+            "hermes_cli.config.load_config",
+            lambda: {"model": {"default": "fake-model"}, "mcp_servers": {}},
+        )
+        monkeypatch.setattr(
+            "hermes_cli.runtime_provider.resolve_runtime_provider",
+            lambda requested=None: {
+                "provider": requested,
+                "api_mode": "codex_app_server",
+                "base_url": "https://example.invalid",
+                "api_key": "test-key",
+            },
+        )
+        monkeypatch.setattr("acp_adapter.session._register_task_cwd", lambda task_id, cwd: None)
+        monkeypatch.delenv("HERMES_IGNORE_RULES", raising=False)
+        monkeypatch.delenv("HERMES_SAFE_MODE", raising=False)
+
+        state = SessionManager(db=None).create_session(cwd="/tmp/project")
+
+        assert state.agent.kwargs["skip_context_files"] is False
+        assert state.agent.kwargs["skip_memory"] is False
+
 
 
 

tests/agent/test_isolation.py

@@ -0,0 +1,48 @@
+"""Unit tests for the shared agent isolation contract."""
+
+import pytest
+
+
+@pytest.mark.parametrize(
+    ("ignore_rules", "safe_mode", "env", "expected"),
+    [
+        # Default path: no flags, no env.
+        (None, None, {}, False),
+        # Env path: each variable alone enables isolation; safe mode implies
+        # ignore-rules at the flag level too.
+        (None, None, {"HERMES_IGNORE_RULES": "1"}, True),
+        (None, None, {"HERMES_SAFE_MODE": "1"}, True),
+        (None, None, {"HERMES_IGNORE_RULES": "true"}, True),
+        (None, None, {"HERMES_IGNORE_RULES": "0", "HERMES_SAFE_MODE": "0"}, False),
+        # Explicit flags win over the environment.
+        (True, None, {}, True),
+        (False, None, {"HERMES_IGNORE_RULES": "1"}, False),
+        (None, True, {}, True),
+        (False, True, {}, True),
+        (True, False, {}, True),
+        (False, False, {"HERMES_SAFE_MODE": "1"}, False),
+    ],
+)
+def test_resolve_agent_isolation(monkeypatch, ignore_rules, safe_mode, env, expected):
+    from agent.isolation import resolve_agent_isolation
+
+    monkeypatch.delenv("HERMES_IGNORE_RULES", raising=False)
+    monkeypatch.delenv("HERMES_SAFE_MODE", raising=False)
+    for name, value in env.items():
+        monkeypatch.setenv(name, value)
+
+    isolated = resolve_agent_isolation(
+        ignore_rules=ignore_rules,
+        safe_mode=safe_mode,
+    )
+    assert isolated is expected
+
+
+def test_resolve_agent_isolation_returns_single_decision(monkeypatch):
+    """The resolver exposes one decision for both agent skip flags."""
+    from agent.isolation import resolve_agent_isolation
+
+    monkeypatch.setenv("HERMES_IGNORE_RULES", "1")
+    assert resolve_agent_isolation() is True
+    monkeypatch.delenv("HERMES_IGNORE_RULES", raising=False)
+    assert resolve_agent_isolation() is False

tests/gateway/test_agent_cache.py

@@ -39,6 +39,24 @@ def test_model_change_different_signature(self):
         sig2 = GatewayRunner._agent_config_signature("claude-opus-4.6", runtime, ["hermes-telegram"], "")
         assert sig1 != sig2
 
+    def test_skip_context_files_change_different_signature(self):
+        """Toggling context-file skipping must rebuild the cached agent."""
+        from gateway.run import GatewayRunner
+
+        runtime = {"api_key": "k", "base_url": "u", "provider": "p"}
+        sig1 = GatewayRunner._agent_config_signature("m", runtime, [], "", skip_context_files=False)
+        sig2 = GatewayRunner._agent_config_signature("m", runtime, [], "", skip_context_files=True)
+        assert sig1 != sig2
+
+    def test_skip_memory_change_different_signature(self):
+        """Toggling memory skipping must rebuild the cached agent."""
+        from gateway.run import GatewayRunner
+
+        runtime = {"api_key": "k", "base_url": "u", "provider": "p"}
+        sig1 = GatewayRunner._agent_config_signature("m", runtime, [], "", skip_memory=False)
+        sig2 = GatewayRunner._agent_config_signature("m", runtime, [], "", skip_memory=True)
+        assert sig1 != sig2
+
     def test_same_token_prefix_different_full_token_changes_signature(self):
         """Tokens sharing a JWT-style prefix must not collide."""
         from gateway.run import GatewayRunner

tests/gateway/test_api_server.py

@@ -241,7 +241,8 @@ def test_custom_config_from_extra(self):
         assert adapter._cors_origins == ("http://localhost:3000",)
 
 
-    def test_create_agent_forwards_runtime_config(self, monkeypatch):
+    @staticmethod
+    def _stub_create_agent_deps(monkeypatch):
         captured = {}
 
         class FakeAgent:
@@ -276,6 +277,10 @@ def __init__(self, **kwargs):
         )
         monkeypatch.setattr("gateway.run.GatewayRunner._load_fallback_model", staticmethod(lambda: None))
         monkeypatch.setattr("hermes_cli.tools_config._get_platform_tools", lambda *_: set())
+        return captured, FakeAgent
+
+    def test_create_agent_forwards_runtime_config(self, monkeypatch):
+        captured, FakeAgent = self._stub_create_agent_deps(monkeypatch)
 
         adapter = APIServerAdapter(PlatformConfig(enabled=True))
         monkeypatch.setattr(adapter, "_ensure_session_db", lambda: None)
@@ -288,6 +293,20 @@ def __init__(self, **kwargs):
         assert captured["checkpoint_max_snapshots"] == 7
         assert captured["checkpoint_max_total_size_mb"] == 321
         assert captured["checkpoint_max_file_size_mb"] == 4
+        assert captured["skip_context_files"] is False
+        assert captured["skip_memory"] is False
+
+    def test_create_agent_honors_isolation_env(self, monkeypatch):
+        captured, _ = self._stub_create_agent_deps(monkeypatch)
+        monkeypatch.setenv("HERMES_SAFE_MODE", "1")
+
+        adapter = APIServerAdapter(PlatformConfig(enabled=True))
+        monkeypatch.setattr(adapter, "_ensure_session_db", lambda: None)
+
+        adapter._create_agent(session_id="api-session")
+
+        assert captured["skip_context_files"] is True
+        assert captured["skip_memory"] is True
 
 
 # ---------------------------------------------------------------------------