fix: mid-turn user message no longer waits behind a foreground terminal command

A message typed while the agent runs (CLI busy_input_mode=interrupt, gateway
priority redirect, ACP redirect) goes through AIAgent.redirect(). During tool
execution redirect() degrades to steer(), whose delivery rides the tool result
— so a long foreground command (a `sleep 285` CI poller, a build) parked the
user's message until it exited. The UI printed "Redirected current turn" while
nothing happened for minutes.

redirect() now also asks the tool workers to YIELD (tools/interrupt.request_yield).
The local terminal backend's wait loop honours it: the drain thread is stopped, the
still-running Popen is adopted by the process registry as a notify_on_complete
background session (ProcessRegistry.adopt_local — output so far seeds the buffer,
the registry reader continues from the pipe), and the tool returns immediately with
status "yielded_to_background" + session_id. The command is never killed; the
completion notification arrives as usual and process(poll/wait/log/kill) work on it.

Non-local backends and internal env.execute() consumers pass no yield_handler and
are unaffected; a stale yield bit is cleared with the interrupt bit per worker tid.
This commit is contained in:
Teknium
2026-09-05 13:27:22 -07:00
parent 9dd6634c56
commit 463292351f
11 changed files with 264 additions and 21 deletions
+14 -1
View File
@@ -9,6 +9,7 @@ import threading
from typing import Optional
from agent.interrupt_compat import request_hard_interrupt
from tools.interrupt import request_yield as _request_yield
from tools.interrupt import set_interrupt as _set_interrupt
# Same logger name as the origin module so log records / caplog filters are unchanged.
@@ -245,8 +246,20 @@ class InterruptControlMixin:
return False
# Never kill a tool to deliver guidance; the steer drain puts it on the final tool result.
# A foreground terminal command would park that delivery until it exits (a 5-minute
# `sleep` poller, a build), so ask the tool workers to YIELD: terminal hands the live
# process to the background registry and returns; tools that don't yield are unaffected.
if getattr(self, "_executing_tools", False):
return self.steer(cleaned)
accepted = self.steer(cleaned)
if accepted:
tracker = getattr(self, "_tool_worker_threads", None)
tracker_lock = getattr(self, "_tool_worker_threads_lock", None)
if tracker is not None and tracker_lock is not None:
with tracker_lock:
worker_tids = list(tracker)
for tid in worker_tids:
_request_yield(tid)
return accepted
_model_active = getattr(self, "_model_request_active", None)
with _ic_lock(self, "_pending_redirect_lock"):
+4 -2
View File
@@ -40,7 +40,8 @@ def test_foreground_command_uses_registered_task_cwd_for_existing_environment(mo
result = json.loads(terminal_tool.terminal_tool(command="pwd", task_id=task_id))
assert result["exit_code"] == 0
assert calls == [("pwd", {"timeout": 60, "cwd": "/workspace/acp", "bounded_capture": True})]
assert len(calls) == 1 and calls[0][0] == "pwd"
assert calls[0][1] | {"timeout": 60, "cwd": "/workspace/acp", "bounded_capture": True} == calls[0][1]
def test_explicit_workdir_still_wins_over_registered_task_cwd(monkeypatch):
@@ -73,7 +74,8 @@ def test_explicit_workdir_still_wins_over_registered_task_cwd(monkeypatch):
)
assert result["exit_code"] == 0
assert calls == [{"timeout": 60, "cwd": "/explicit/workdir", "bounded_capture": True}]
assert len(calls) == 1
assert calls[0] | {"timeout": 60, "cwd": "/explicit/workdir", "bounded_capture": True} == calls[0]
def test_explicit_workdir_does_not_persist_into_session_cwd(monkeypatch):
@@ -0,0 +1,88 @@
"""A user message sent while a foreground terminal command runs must not wait for the command.
``AIAgent.redirect()`` during tool execution degrades to ``steer()``, whose delivery rides
the tool result — so a long foreground command (a 5-minute ``sleep`` poller, a build) parked
the user's message until it exited. Now redirect() also asks the tool workers to YIELD: the
local terminal backend hands the still-running process to the process registry as a
notify-on-complete background session and returns immediately, without killing it.
"""
import json
import os
import threading
import time
import pytest
from agent.interrupt_control import InterruptControlMixin
from tools import interrupt as interrupt_mod
from tools.process_registry import process_registry
from tools.terminal_tool import terminal_tool
pytestmark = pytest.mark.linux_only
class _Agent(InterruptControlMixin):
_executing_tools = True
_interrupt_requested = False
_pending_steer = None
_pending_redirect = None
api_mode = "chat_completions"
def __init__(self):
self._pending_steer_lock = threading.Lock()
self._pending_redirect_lock = threading.Lock()
self._tool_worker_threads = set()
self._tool_worker_threads_lock = threading.Lock()
self._execution_thread_id = None
def test_redirect_mid_command_yields_it_to_background_without_killing_it(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
agent = _Agent()
res = {}
def worker():
with agent._tool_worker_threads_lock:
agent._tool_worker_threads.add(threading.current_thread().ident)
t0 = time.monotonic()
res["result"] = json.loads(terminal_tool("echo started; sleep 60; echo done", task_id="yield-test", timeout=90))
res["elapsed"] = time.monotonic() - t0
t = threading.Thread(target=worker, daemon=True)
t.start()
time.sleep(1.5)
assert agent.redirect("also, check the session ids") is True
assert agent._pending_steer == "also, check the session ids" # still delivered as a steer
t.join(timeout=15)
assert not t.is_alive(), "terminal tool still blocked on the command after redirect()"
r = res["result"]
try:
assert r["status"] == "yielded_to_background"
assert r["exit_code"] is None and "started" in r["output"]
assert r["notify_on_complete"] is True
# The process is alive and tracked: poll/wait/kill and the completion notification work.
assert os.path.exists(f"/proc/{r['pid']}")
assert process_registry.poll(r["session_id"])["status"] == "running"
assert not interrupt_mod.is_thread_yield_requested(t.ident)
finally:
killed = process_registry.kill_process(r["session_id"])
assert killed["status"] == "killed"
evt = process_registry.completion_queue.get(timeout=5)
assert evt["session_id"] == r["session_id"]
def test_yield_request_without_steer_leaves_foreground_wait_alone():
"""No yield handler (internal env.execute consumers) -> the request is ignored and the
command runs to completion; an unrelated stale yield bit must not leak into it either."""
from tools.environments.local import LocalEnvironment
env = LocalEnvironment()
try:
interrupt_mod.request_yield(threading.current_thread().ident)
result = env.execute("echo alpha; sleep 0.3; echo omega", timeout=10)
assert result["returncode"] == 0
assert "omega" in result["output"]
finally:
interrupt_mod.consume_yield(threading.current_thread().ident)
env.cleanup()
+27 -5
View File
@@ -19,7 +19,7 @@ from pathlib import Path
from typing import Callable, Iterable
from hermes_constants import get_hermes_home
from tools.interrupt import is_interrupted, is_thread_interrupted
from tools.interrupt import consume_yield, is_interrupted, is_thread_interrupted
from tools.environments.base_output import (
ProcessHandle, _finalize_wait_result, _new_output_collector, _start_drain_thread,
)
@@ -337,8 +337,13 @@ class BaseEnvironment(ABC):
# --- Process lifecycle ---
def _wait_for_process(
self, proc: ProcessHandle, timeout: int = 120, *,
bounded_capture: bool = False, watch_interrupt_tid: int | None = None) -> dict:
bounded_capture: bool = False, watch_interrupt_tid: int | None = None,
yield_handler: Callable[[ProcessHandle, str], dict] | None = None) -> dict:
"""Poll-based wait with interrupt checking and stdout draining (shared, not overridden).
``yield_handler(proc, output_so_far)``: when the tool thread is asked to yield
(``tools.interrupt.request_yield`` — a user message arrived mid-command), the drain
thread is stopped, the still-running process is handed to the handler and its dict
is returned as the result; the process is NOT killed.
``bounded_capture=True`` (foreground terminal-tool path only) retains at most
``tool_output.max_bytes`` in a head/tail window so a verbose subprocess cannot OOM the
process; the default keeps full fidelity for internal consumers. Fires the activity
@@ -354,7 +359,8 @@ class BaseEnvironment(ABC):
data. See #64435.
"""
output = _new_output_collector(proc, bounded_capture)
drain_thread = _start_drain_thread(proc, output)
drain_stop = threading.Event() if yield_handler is not None else None
drain_thread = _start_drain_thread(proc, output, drain_stop)
_now = time.monotonic()
deadline = _now + timeout
_activity_state = {"last_touch": _now, "start": _now}
@@ -375,6 +381,19 @@ class BaseEnvironment(ABC):
trace.interrupted()
_kill_and_join()
return self._finalize_wait_result(output, output.render(suffix="\n[Command interrupted]"), 130)
if yield_handler is not None and consume_yield(watch_interrupt_tid):
drain_stop.set()
drain_thread.join(timeout=1)
try:
handed = yield_handler(proc, output.render())
except Exception:
logger.warning("yield-to-background handoff failed; continuing to wait", exc_info=True)
handed = None
if handed is not None:
output.close_spill()
return handed
drain_stop.clear()
drain_thread = _start_drain_thread(proc, output, drain_stop)
if time.monotonic() > deadline:
trace.timed_out()
_kill_and_join()
@@ -462,7 +481,8 @@ class BaseEnvironment(ABC):
timeout: int | None = None,
stdin_data: str | None = None,
rewrite_compound_background: bool = True,
bounded_capture: bool = False) -> dict:
bounded_capture: bool = False,
yield_handler: Callable[[ProcessHandle, str], dict] | None = None) -> dict:
"""Execute a command, return {"output": str, "returncode": int}. ``bounded_capture=True``
caps retention at ``tool_output.max_bytes`` WHILE draining; only the foreground terminal
tool may set it — internal full-fidelity consumers (file-op ``cat`` reads feeding the
@@ -509,7 +529,9 @@ class BaseEnvironment(ABC):
spawned = self._run_bash(wrapped, login=login, timeout=effective_timeout, stdin_data=effective_stdin)
proc_holder.append(spawned)
return self._wait_for_process(
spawned, timeout=effective_timeout, bounded_capture=bounded_capture, watch_interrupt_tid=parent_tid)
spawned, timeout=effective_timeout, bounded_capture=bounded_capture,
watch_interrupt_tid=parent_tid,
**({"yield_handler": yield_handler} if yield_handler is not None else {}))
def _on_timeout() -> None:
if proc_holder:
+12 -7
View File
@@ -341,7 +341,7 @@ class _ThreadedProcessHandle:
# --- Stdout drain thread ---
def _drain_stdout(proc: ProcessHandle, output: _BoundedOutputCollector) -> None:
def _drain_stdout(proc: ProcessHandle, output: _BoundedOutputCollector, stop: "threading.Event | None" = None) -> None:
"""Drain ``proc.stdout`` into *output* until EOF or shortly after exit.
``for line in proc.stdout`` would block on ``readline()`` until EOF, and a backgrounded
grandchild (``cmd &``, ``setsid cmd & disown``) inherits the pipe's write end — so the
@@ -384,7 +384,7 @@ def _drain_stdout(proc: ProcessHandle, output: _BoundedOutputCollector) -> None:
while chunk := os.read(fd, 4096):
output.append(decoder.decode(chunk))
else:
_drain_fd_select(proc, fd, output, decoder)
_drain_fd_select(proc, fd, output, decoder, stop)
except Exception:
pass # closed fd / broken stream: keep what was captured
finally:
@@ -397,10 +397,13 @@ def _drain_stdout(proc: ProcessHandle, output: _BoundedOutputCollector) -> None:
pass
def _drain_fd_select(proc, fd: int, output: _BoundedOutputCollector, decoder) -> None:
"""POSIX drain: select() poll, stopping ~300ms after bash exits with the pipe idle."""
def _drain_fd_select(proc, fd: int, output: _BoundedOutputCollector, decoder, stop=None) -> None:
"""POSIX drain: select() poll, stopping ~300ms after bash exits with the pipe idle, or
when *stop* is set (the pipe is being handed to another reader — yield-to-background)."""
idle_after_exit = 0
while True:
if stop is not None and stop.is_set():
return
try:
ready, _, _ = select.select([fd], [], [], 0.1)
except (ValueError, OSError):
@@ -422,8 +425,10 @@ def _drain_fd_select(proc, fd: int, output: _BoundedOutputCollector, decoder) ->
return
def _start_drain_thread(proc: ProcessHandle, output: _BoundedOutputCollector) -> threading.Thread:
"""Start the daemon thread running :func:`_drain_stdout`."""
thread = threading.Thread(target=_drain_stdout, args=(proc, output), daemon=True)
def _start_drain_thread(
proc: ProcessHandle, output: _BoundedOutputCollector, stop: "threading.Event | None" = None,
) -> threading.Thread:
"""Start the daemon thread running :func:`_drain_stdout`; *stop* ends it early."""
thread = threading.Thread(target=_drain_stdout, args=(proc, output, stop), daemon=True)
thread.start()
return thread
+34 -1
View File
@@ -20,12 +20,15 @@ if _DEBUG_INTERRUPT:
# Interrupted thread idents + optional user-safe cause (never the user's message text).
_interrupted_threads: set[int] = set()
_interrupt_reasons: dict[int, str] = {}
# Threads asked to YIELD: hand a long-running foreground command to the background
# instead of killing it, so a mid-turn user message is not parked behind it.
_yield_threads: set[int] = set()
_lock = threading.Lock()
def set_interrupt(active: bool, thread_id: int | None = None, *, reason: str | None = None) -> None:
"""Set or clear the interrupt for *thread_id* (default: current thread); ``reason`` is
an optional user-safe cause."""
an optional user-safe cause. Clearing also drops a pending yield request."""
tid = thread_id if thread_id is not None else threading.current_thread().ident
with _lock:
(_interrupted_threads.add if active else _interrupted_threads.discard)(tid)
@@ -33,6 +36,8 @@ def set_interrupt(active: bool, thread_id: int | None = None, *, reason: str | N
_interrupt_reasons[tid] = reason
else:
_interrupt_reasons.pop(tid, None)
if not active:
_yield_threads.discard(tid)
_snapshot = set(_interrupted_threads) if _DEBUG_INTERRUPT else None
if _DEBUG_INTERRUPT:
logger.info(
@@ -58,6 +63,34 @@ def is_thread_interrupted(thread_id: int | None) -> bool:
return thread_id in _interrupted_threads
def request_yield(thread_id: int) -> None:
"""Ask the tool running on *thread_id* to yield: a foreground terminal command hands
its live process to the background registry and returns at once, so a user's mid-turn
message (``redirect()`` during tool execution) is delivered instead of parked behind it.
The command itself is never killed; that is what ``set_interrupt`` is for."""
with _lock:
_yield_threads.add(thread_id)
def is_thread_yield_requested(thread_id: int | None) -> bool:
"""Whether a yield is pending for *thread_id* (``None`` never is)."""
if thread_id is None:
return False
with _lock:
return thread_id in _yield_threads
def consume_yield(thread_id: int | None) -> bool:
"""Atomically take the pending yield for *thread_id*; True if one was pending."""
if thread_id is None:
return False
with _lock:
if thread_id in _yield_threads:
_yield_threads.discard(thread_id)
return True
return False
def run_if_not_interrupted(callback: Callable[[], None]) -> bool:
"""Run a state transition atomically with current-thread interruption.
+19
View File
@@ -890,6 +890,25 @@ class ProcessRegistry:
with suppress(Exception):
proc.wait(timeout=5)
def adopt_local(
self, proc: subprocess.Popen, *, command: str, cwd: Optional[str], task_id: str = "",
session_key: str = "", owner_task_id: str = "", output_so_far: str = "",
notify_on_complete: bool = True) -> ProcessSession:
"""Take over a still-running foreground Popen as a tracked background session
(yield-to-background: the user sent a message while the command was running).
The caller has stopped its own drain thread; the registry's reader continues from
the pipe's current position and ``output_so_far`` seeds the buffer so nothing
already captured is lost."""
session = self._new_session(command, task_id, owner_task_id, session_key, cwd)
session.process = proc
session.pid = proc.pid
session.host_start_time = self._safe_host_start_time(session.pid)
session.notify_on_complete = notify_on_complete
if output_so_far:
session.append_output(output_so_far)
self._track_started(session, self._reader_loop, f"proc-reader-{session.id}")
return session
def spawn_via_env(
self, env: Any, command: str, cwd: str = None, task_id: str = "", session_key: str = "",
timeout: int = 10, owner_task_id: str = "") -> ProcessSession:
+18 -2
View File
@@ -736,7 +736,7 @@ from tools.terminal_tool_guards import (
_foreground_background_guidance, _safe_command_preview, _validate_workdir,
gateway_lifecycle_block, self_repo_block,
)
from tools.terminal_tool_background import spawn_background_process
from tools.terminal_tool_background import _YIELDED_NOTE, spawn_background_process, yield_to_background_handler
from tools.terminal_tool_result import finalize_foreground_result
@@ -1009,6 +1009,12 @@ def _acquire_env(plan: _ExecPlan, task_id: Optional[str]) -> Any:
return new_env
def _yield_kwargs(command: str, **ctx) -> dict:
"""``env.execute`` kwargs enabling yield-to-background (local backend only)."""
handler = yield_to_background_handler(command=command, **ctx)
return {"yield_handler": handler} if handler is not None else {}
def _run_foreground(
command: str, env: Any, plan: _ExecPlan, *,
task_id: Optional[str], session_id: Optional[str], session_key: str,
@@ -1035,7 +1041,11 @@ def _run_foreground(
# bounded_capture: model-facing output keeps a head/tail window
# while streaming so a verbose command can't OOM the gateway;
# internal env.execute() consumers stay unbounded.
result = env.execute(command, timeout=effective_timeout, cwd=command_cwd, bounded_capture=True)
result = env.execute(
command, timeout=effective_timeout, cwd=command_cwd, bounded_capture=True,
**_yield_kwargs(command, env_type=env_type, cwd=command_cwd, effective_task_id=eff,
task_id=task_id, session_key=session_key),
)
break
except Exception as e:
if "timeout" in str(e).lower():
@@ -1051,6 +1061,12 @@ def _run_foreground(
max_retries, _safe_command_preview(command), type(e).__name__, e, eff, env_type)
return _error_json(_redact_terminal_error_text(f"Command execution failed: {type(e).__name__}: {e}"))
if result.get("yielded_session_id"):
return json.dumps({
"output": result.get("output", ""), "exit_code": None, "error": None,
"status": "yielded_to_background", "session_id": result["yielded_session_id"],
"pid": result.get("pid"), "notify_on_complete": True, "note": _YIELDED_NOTE,
}, ensure_ascii=False)
return finalize_foreground_result(
command=command, result=result, env=env, env_type=env_type, effective_task_id=eff,
task_id=task_id, session_id=session_id, session_key=session_key, workdir=workdir,
+45
View File
@@ -186,3 +186,48 @@ def spawn_background_process(
"output": "", "exit_code": -1,
"error": _redact_terminal_error_text(f"Failed to start background process: {e}"),
}, ensure_ascii=False)
_YIELDED_NOTE = (
"The user sent a message while this command was running, so it was moved to the "
"background WITHOUT being killed and is still running. You will be notified when it "
"exits (notify_on_complete). Read the user's message and respond to it now; use "
"process(action='poll'|'wait'|'log', session_id=...) to check on this command."
)
def yield_to_background_handler(
*, command: str, env_type: str, cwd: Optional[str], effective_task_id: str,
task_id: Optional[str], session_key: str,
):
"""Build the ``yield_handler`` a foreground ``env.execute`` calls when the tool thread is
asked to yield (a user message arrived mid-command). Local backend only: the live Popen
is adopted by the process registry as a notify-on-complete background session and the
partial output is returned to the model right away. Other backends return None (no
adoptable host process) and the foreground wait continues."""
if env_type != "local":
return None
def _handler(proc, output_so_far: str) -> dict:
from tools.process_registry import process_registry
session = process_registry.adopt_local(
proc, command=command, cwd=cwd, task_id=effective_task_id,
owner_task_id=task_id or effective_task_id, session_key=session_key,
output_so_far=output_so_far)
_stamp_routing_if_gateway(process_registry, session, session_key)
logger.info("foreground command yielded to background as %s (pid %s)", session.id, session.pid)
return {
"output": output_so_far, "returncode": None, "yielded_session_id": session.id, "pid": session.pid,
}
return _handler
def _stamp_routing_if_gateway(process_registry, session, session_key) -> None:
"""Route the adopted session's completion like a normal notify_on_complete spawn."""
from gateway.session_context import async_delivery_supported, get_session_env
if not async_delivery_supported():
session.notify_on_complete = False
return
_stamp_gateway_routing(session, get_session_env)
if session.watcher_platform:
_register_completion_watcher(process_registry, session, session_key)
+2 -2
View File
@@ -364,7 +364,7 @@ The `display.busy_input_mode` config key controls what happens when you press En
| Mode | Behavior |
|------|----------|
| `"interrupt"` (default) | Your message redirects the active turn. Model generation restarts with displayed reasoning and completed work preserved; running tools finish first |
| `"interrupt"` (default) | Your message redirects the active turn. Model generation restarts with displayed reasoning and completed work preserved. A running foreground terminal command is moved to the background (not killed — you get a completion notification) so your message is read immediately; other running tools finish first |
| `"queue"` | Your message is silently queued and sent as the next turn after the agent finishes |
| `"steer"` | Your message is injected into the current run via `/steer`, arriving at the agent after the next tool call — no interrupt, no new turn |
@@ -374,7 +374,7 @@ display:
busy_input_mode: "steer" # or "queue" or "interrupt" (default)
```
`"queue"` mode prepares a separate follow-up turn. `"steer"` always waits for the next tool-result boundary. The default `"interrupt"` mode responds sooner during model generation while avoiding cancellation of a running tool. Use `/stop` when you want to cancel the turn and its foreground work. Unknown values fall back to `"interrupt"`.
`"queue"` mode prepares a separate follow-up turn. `"steer"` always waits for the next tool-result boundary. The default `"interrupt"` mode responds sooner during model generation while avoiding cancellation of a running tool; a long foreground `terminal` command (a build, a poller) is handed to the background so the agent sees your message right away instead of after the command exits. Use `/stop` when you want to cancel the turn and its foreground work. Unknown values fall back to `"interrupt"`.
`"steer"` has two automatic fallbacks: if the agent hasn't started yet, or if images are attached, the message falls back to `"queue"` behavior so nothing is lost.
+1 -1
View File
@@ -405,7 +405,7 @@ Send a message while the agent is working to correct the active turn:
### Queue vs interrupt vs steer (busy-input mode)
By default, messaging a busy agent redirects its active turn. Two other modes are available:
By default, messaging a busy agent redirects its active turn (a running foreground terminal command is moved to the background rather than killed, so your message is read immediately). Two other modes are available:
- `queue` — follow-up messages wait and run as the next turn after the current task finishes.
- `steer` — follow-up messages are injected into the current run via `/steer`, arriving at the agent after the next tool call. No interrupt, no new turn. Falls back to `queue` behavior if the agent hasn't started yet.