[codex] Guard execute_code behind gateway approvals

tests/acp/test_server.py

@@ -971,17 +971,13 @@ def fake_agent(**kwargs):
             "hermes_cli.runtime_provider.resolve_runtime_provider",
             fake_resolve_runtime_provider,
         )
-        # Pin the parser so this test doesn't depend on live
-        # ``_KNOWN_PROVIDER_NAMES`` / ``_PROVIDER_ALIASES`` module state
-        # (sibling of the same hardening on
-        # ``test_model_switch_uses_requested_provider``).
+        # This test covers the ACP model-switch handoff, not model-string
+        # parsing. Patch the ACP resolver directly so live provider registry
+        # state from unrelated tests cannot shadow the provider under test.
         monkeypatch.setattr(
-            "hermes_cli.models.parse_model_input",
-            lambda raw, current: ("anthropic", "claude-sonnet-4-6"),
-        )
-        monkeypatch.setattr(
-            "hermes_cli.models.detect_provider_for_model",
-            lambda model, current: None,
+            HermesACPAgent,
+            "_resolve_model_selection",
+            staticmethod(lambda raw, current: ("anthropic", "claude-sonnet-4-6")),
         )
         manager = SessionManager(db=SessionDB(tmp_path / "state.db"))
 
@@ -1555,19 +1551,13 @@ def fake_agent(**kwargs):
             "hermes_cli.runtime_provider.resolve_runtime_provider",
             fake_resolve_runtime_provider,
         )
-        # Pin the model-string parser independently of the live
-        # ``_KNOWN_PROVIDER_NAMES`` / ``_PROVIDER_ALIASES`` module state.
-        # Otherwise any test in the same xdist worker that mutates those
-        # globals (e.g. registers a custom provider that shadows
-        # ``anthropic``) flakes this one — observed once in CI as
-        # ``'custom' == 'anthropic'``.
-        monkeypatch.setattr(
-            "hermes_cli.models.parse_model_input",
-            lambda raw, current: ("anthropic", "claude-sonnet-4-6"),
-        )
+        # This test covers the ACP model-switch handoff, not model-string
+        # parsing. Patch the ACP resolver directly so live provider registry
+        # state from unrelated tests cannot shadow the provider under test.
         monkeypatch.setattr(
-            "hermes_cli.models.detect_provider_for_model",
-            lambda model, current: None,
+            HermesACPAgent,
+            "_resolve_model_selection",
+            staticmethod(lambda raw, current: ("anthropic", "claude-sonnet-4-6")),
         )
         manager = SessionManager(db=SessionDB(tmp_path / "state.db"))
 

tests/tools/test_browser_supervisor.py

@@ -40,6 +40,15 @@ def _find_chrome() -> str:
     pytest.skip("no Chrome binary found")
 
 
+def _terminate_chrome(proc: subprocess.Popen) -> None:
+    proc.terminate()
+    try:
+        proc.wait(timeout=5)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        proc.wait(timeout=5)
+
+
 @pytest.fixture
 def chrome_cdp(request):
     """Start a headless Chrome with --remote-debugging-port, yield its WS URL.
@@ -89,18 +98,13 @@ def chrome_cdp(request):
         except Exception:
             time.sleep(0.25)
     if ws_url is None:
-        proc.terminate()
-        proc.wait(timeout=5)
+        _terminate_chrome(proc)
         shutil.rmtree(profile, ignore_errors=True)
         pytest.skip("Chrome didn't expose CDP in time")
 
     yield ws_url, port
 
-    proc.terminate()
-    try:
-        proc.wait(timeout=3)
-    except Exception:
-        proc.kill()
+    _terminate_chrome(proc)
     shutil.rmtree(profile, ignore_errors=True)
 
 

tests/tools/test_code_execution.py

@@ -17,6 +17,7 @@
 
 import json
 import os
+import tempfile
 
 os.environ["TERMINAL_ENV"] = "local"
 
@@ -836,6 +837,116 @@ def test_nonoverlapping_tools_fallback(self):
         self.assertEqual(result["status"], "success")
         self.assertIn("fallback ok", result["output"])
 
+    @unittest.skipIf(sys.platform == "win32", "UDS not available on Windows")
+    def test_gateway_execute_code_denial_blocks_child_process(self):
+        """Gateway approval denial must stop execute_code before spawn."""
+        from tools.approval import (
+            clear_session,
+            register_gateway_notify,
+            reset_current_session_key,
+            resolve_gateway_approval,
+            set_current_session_key,
+            unregister_gateway_notify,
+        )
+
+        session_key = "execute-code-deny"
+        notified = []
+        result_holder = []
+
+        with tempfile.TemporaryDirectory() as tmp:
+            marker = os.path.join(tmp, "marker.txt")
+            code = (
+                "from pathlib import Path\n"
+                f"Path({marker!r}).write_text('ran')\n"
+                "print('should-not-run')\n"
+            )
+
+            register_gateway_notify(session_key, lambda data: notified.append(data))
+            token = set_current_session_key(session_key)
+            os.environ["HERMES_GATEWAY_SESSION"] = "1"
+            os.environ["HERMES_EXEC_ASK"] = "1"
+            os.environ["HERMES_SESSION_KEY"] = session_key
+            try:
+                thread = threading.Thread(
+                    target=lambda: result_holder.append(json.loads(execute_code(
+                        code,
+                        task_id="test-exec-deny",
+                        enabled_tools=[],
+                    )))
+                )
+                thread.start()
+
+                deadline = time.monotonic() + 5
+                while not notified and time.monotonic() < deadline:
+                    time.sleep(0.05)
+
+                self.assertEqual(len(notified), 1)
+                self.assertIn("execute_code <<'PY'", notified[0]["command"])
+                self.assertFalse(os.path.exists(marker))
+
+                resolve_gateway_approval(session_key, "deny")
+                thread.join(timeout=10)
+
+                self.assertFalse(thread.is_alive())
+                self.assertEqual(result_holder[0]["status"], "blocked")
+                self.assertFalse(os.path.exists(marker))
+            finally:
+                os.environ.pop("HERMES_GATEWAY_SESSION", None)
+                os.environ.pop("HERMES_EXEC_ASK", None)
+                os.environ.pop("HERMES_SESSION_KEY", None)
+                unregister_gateway_notify(session_key)
+                clear_session(session_key)
+                reset_current_session_key(token)
+
+    @unittest.skipIf(sys.platform == "win32", "UDS not available on Windows")
+    def test_gateway_execute_code_runs_after_one_shot_approval(self):
+        """Approving the execute_code preflight allows the script to run."""
+        from tools.approval import (
+            clear_session,
+            register_gateway_notify,
+            reset_current_session_key,
+            resolve_gateway_approval,
+            set_current_session_key,
+            unregister_gateway_notify,
+        )
+
+        session_key = "execute-code-approve"
+        notified = []
+        result_holder = []
+        register_gateway_notify(session_key, lambda data: notified.append(data))
+        token = set_current_session_key(session_key)
+        os.environ["HERMES_GATEWAY_SESSION"] = "1"
+        os.environ["HERMES_EXEC_ASK"] = "1"
+        os.environ["HERMES_SESSION_KEY"] = session_key
+        try:
+            thread = threading.Thread(
+                target=lambda: result_holder.append(json.loads(execute_code(
+                    "print('approved-run')",
+                    task_id="test-exec-approve",
+                    enabled_tools=[],
+                )))
+            )
+            thread.start()
+
+            deadline = time.monotonic() + 5
+            while not notified and time.monotonic() < deadline:
+                time.sleep(0.05)
+
+            self.assertEqual(len(notified), 1)
+            resolve_gateway_approval(session_key, "once")
+            thread.join(timeout=10)
+
+            self.assertFalse(thread.is_alive())
+            self.assertEqual(result_holder[0]["status"], "success")
+            self.assertIn("approved-run", result_holder[0]["output"])
+        finally:
+            os.environ.pop("HERMES_GATEWAY_SESSION", None)
+            os.environ.pop("HERMES_EXEC_ASK", None)
+            os.environ.pop("HERMES_SESSION_KEY", None)
+            unregister_gateway_notify(session_key)
+            clear_session(session_key)
+            reset_current_session_key(token)
+
 
 # ---------------------------------------------------------------------------
 # _load_config