fix(tools): bound execute_code's lifecycle probe and keep the terminal guard answerable to /stop

execute_code ran the same unbounded _is_supervised_gateway_process() probe
ahead of every cell, so the wedge #111922 bounds in terminal_tool still hung
an execute_code call (and its cron slot) forever: share the cell's deadline
and fail closed with a retryable error when the probe renders no verdict.

Moving the terminal pre-exec guard onto a deadline worker made it blind to
/stop, which keys on the tool thread's ident: record the acting-for tid in a
contextvar (copied into the worker by run_bounded_sync) so is_interrupted()
on the worker honours the tool thread's bit too.

Floor the guard's share of the deadline at 30s so a short command timeout
does not turn the guard's own cold-start cost (imports, git probes under
load) into a refusal — tests/tools/test_terminal_error_redaction.py was red
on the branch for exactly that.

tests/tools/test_execute_code_lifecycle_guard_deadline.py

@@ -0,0 +1,33 @@
+"""execute_code's gateway-lifecycle identity probe shares the cell deadline (#111922).
+
+The same ``_is_supervised_gateway_process`` probe that wedged ``terminal_tool`` runs
+ahead of every ``execute_code`` cell; a probe that renders no verdict must fail closed
+within the deadline instead of holding the tool call (and its cron slot) forever.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+
+import tools.code_execution_tool as cet
+import tools.terminal_tool as terminal_module
+
+
+def test_wedged_lifecycle_probe_returns_bounded_error_without_running(monkeypatch):
+    import tools.process_registry as process_registry
+
+    monkeypatch.setattr(cet, "SANDBOX_AVAILABLE", True)
+    monkeypatch.setattr(cet, "_load_config", lambda: {"timeout": 0.05})
+    monkeypatch.setattr(terminal_module, "_PRE_EXEC_GUARD_MIN_TIMEOUT_S", 0)
+    monkeypatch.setattr(process_registry, "_is_supervised_gateway_process", lambda: time.sleep(1))
+    ran: list[str] = []
+    monkeypatch.setattr(cet, "_get_env_config", lambda: ran.append("env") or {"env_type": "local"}, raising=False)
+
+    start = time.monotonic()
+    result = json.loads(cet.execute_code("print('hi')"))
+    elapsed = time.monotonic() - start
+
+    assert elapsed < 0.5, f"lifecycle probe wedged execute_code for {elapsed:.2f}s"
+    assert "did not finish" in result["error"]
+    assert ran == [], "a probe with no verdict must not fail open into execution"

tests/tools/test_terminal_pre_guard_deadline.py

@@ -31,6 +31,7 @@ def _plan(timeout: float = 0.05) -> SimpleNamespace:
 def stubbed_pipeline(monkeypatch):
     """Stub planning/env/approval/execution; returns the list of executions that happened."""
     calls: list[str] = []
+    monkeypatch.setattr(terminal_module, "_PRE_EXEC_GUARD_MIN_TIMEOUT_S", 0)
     monkeypatch.setattr(terminal_module, "_plan_execution", lambda *_a, **_k: _plan())
     monkeypatch.setattr(terminal_module, "_acquire_env", lambda *_a, **_k: object())
     monkeypatch.setattr(
@@ -71,3 +72,20 @@ def test_completed_pre_execution_guard_verdicts_pass_through(monkeypatch, stubbe
     monkeypatch.setattr(terminal_module, "_pre_exec_block", _rejecting_probe)
     assert terminal_module.terminal_tool("echo ok") == '{"status":"blocked"}'
     assert stubbed_pipeline == ["foreground"]
+
+
+def test_pre_execution_guard_on_the_deadline_worker_sees_the_tool_threads_interrupt(monkeypatch, stubbed_pipeline):
+    """/stop keys on the tool thread's tid; the guard chain moved onto a worker must still see it."""
+    from tools.interrupt import is_interrupted, set_interrupt
+
+    seen: list[bool] = []
+    monkeypatch.setattr(terminal_module, "_pre_exec_block", lambda *_a, **_k: seen.append(is_interrupted()))
+
+    set_interrupt(True)
+    try:
+        terminal_module.terminal_tool("echo ok")
+    finally:
+        set_interrupt(False)
+
+    assert seen == [True], "guard on the deadline worker was blind to the tool thread's interrupt bit"
+    assert is_interrupted() is False, "the tool thread's own interrupt view must not leak past the guard"

tools/code_execution_tool.py

@@ -694,8 +694,21 @@ def execute_code(
     # execute_code is a straight bypass — the terminal() path refuses `launchctl bootout ai.hermes.gateway`,
     # but the identical command inside `os.system(...)` / `subprocess.run([...])` here sailed through and
     # SIGTERM'd the gateway mid-task.
+    # The identity probe ends in a kernel process query that has wedged on macOS
+    # (#111922); share the cell's own deadline and fail CLOSED when it renders no verdict.
+    from agent.deadline import run_bounded_sync
     from tools.process_registry import _is_supervised_gateway_process
-    if _is_supervised_gateway_process():
+    from tools.terminal_tool import _PRE_EXEC_GUARD_MIN_TIMEOUT_S
+    _probe_timeout = max(_load_config().get("timeout", DEFAULT_TIMEOUT), _PRE_EXEC_GUARD_MIN_TIMEOUT_S)
+    _probe = run_bounded_sync(
+        _is_supervised_gateway_process, _probe_timeout, label="execute_code.lifecycle-guard",
+    )
+    if _probe.timed_out:
+        return tool_error(
+            f"execute_code lifecycle guard did not finish within {_probe_timeout}s "
+            "(process-identity probe wedged); the code was not run. Retry the call."
+        )
+    if _probe.value:
         from cron.lifecycle_guard import contains_gateway_lifecycle_command
         if contains_gateway_lifecycle_command(code):
             return tool_error(

tools/interrupt.py

@@ -3,6 +3,7 @@ agent session does not kill tools in other sessions (the gateway runs many agent
 process). The agent passes its execution thread id to set_interrupt(); tools call
 is_interrupted(), which checks the CURRENT thread."""
 
+import contextvars
 import logging
 import os
 import threading
@@ -24,6 +25,12 @@ _interrupt_reasons: dict[int, str] = {}
 # instead of killing it, so a mid-turn user message is not parked behind it.
 _yield_threads: set[int] = set()
 _lock = threading.Lock()
+# Tool-worker tid a deadline worker acts for. ``run_bounded_sync`` runs its worker under
+# ``contextvars.copy_context()``, so a guard chain moved onto that worker still honours
+# ``/stop`` aimed at the tool thread that spawned it (``is_interrupted`` checks both).
+acting_for_tid: contextvars.ContextVar[int | None] = contextvars.ContextVar(
+    "hermes_interrupt_acting_for_tid", default=None,
+)
 
 
 def set_interrupt(active: bool, thread_id: int | None = None, *, reason: str | None = None) -> None:
@@ -47,7 +54,7 @@ def set_interrupt(active: bool, thread_id: int | None = None, *, reason: str | N
 
 
 def is_interrupted() -> bool:
-    return is_thread_interrupted(threading.current_thread().ident)
+    return is_thread_interrupted(threading.current_thread().ident) or is_thread_interrupted(acting_for_tid.get())
 
 
 def is_thread_interrupted(thread_id: int | None) -> bool:

tools/terminal_tool.py

@@ -1127,6 +1127,12 @@ def _run_foreground(
     )
 
 
+# Floor for the pre-exec guard's share of the command deadline: a short command timeout
+# (1s in tests, a few seconds in practice) must not turn the guard's own cold-start cost
+# (module imports, git probes under load) into a refusal; the wedge it bounds lasted an hour.
+_PRE_EXEC_GUARD_MIN_TIMEOUT_S = 30
+
+
 def _pre_exec_block(
     command: str, *, env: Any, env_type: str, cwd: str,
     workdir: Optional[str], session_key: str,
@@ -1231,17 +1237,25 @@ def terminal_tool(
         # unconditionally (``force`` cannot bypass them), so the command is
         # refused with a retryable error instead of running unguarded.
         from agent.deadline import run_bounded_sync
+        from tools.interrupt import acting_for_tid
 
-        bounded_guard = run_bounded_sync(
-            lambda: _pre_exec_block(
-                command, env=env, env_type=env_type, cwd=cwd, workdir=workdir, session_key=session_key,
-            ),
-            plan.effective_timeout,
-            label="terminal.pre-exec-guard",
-        )
+        # The guard chain runs on the deadline worker; keep it answerable to /stop
+        # aimed at this tool thread (a remote-backend script read polls is_interrupted()).
+        guard_timeout = max(plan.effective_timeout, _PRE_EXEC_GUARD_MIN_TIMEOUT_S)
+        _acting_token = acting_for_tid.set(threading.current_thread().ident)
+        try:
+            bounded_guard = run_bounded_sync(
+                lambda: _pre_exec_block(
+                    command, env=env, env_type=env_type, cwd=cwd, workdir=workdir, session_key=session_key,
+                ),
+                guard_timeout,
+                label="terminal.pre-exec-guard",
+            )
+        finally:
+            acting_for_tid.reset(_acting_token)
         if bounded_guard.timed_out:
             raise _Rejected(_error_json(
-                f"Terminal pre-execution guard did not finish within {plan.effective_timeout}s "
+                f"Terminal pre-execution guard did not finish within {guard_timeout}s "
                 "(process-identity probe wedged); the command was not run. Retry the call.",
                 status="error",
             ))