ac087e6ada
Hermes Console ran each command on a ThreadPoolExecutor and cancelled only the asyncio waiter. A command that forks an AIAgent (`curator run --consolidate`) kept its worker thread and its in-flight provider request alive after the prompt said "cancelled" — a llama.cpp generation kept decoding for 30+ minutes and held the inference slot (#106179). Root cause: the host owning the thread never knew about the agent created deep inside the synchronous command, so it could not call the existing cooperative `interrupt()` path that closes the request sockets. Fix: `agent/interrupt_scope.py` gives the host an `InterruptScope`; the console binds it around the worker (ContextVar), and every `AIAgent.run_conversation()` registers itself with the bound scope for the turn. On cancel, timeout and disconnect the console calls `scope.cancel()` (hard-interrupts every registered agent; an agent registering after the cancel is interrupted on entry so the cancel cannot lose the race with a turn that has not started) and awaits the worker Future with a 10s bound before reporting cancelled/timeout. Queued-but-unstarted work is dropped via `Future.cancel()` alone. Live repro (fake OpenAI-compatible provider blocking like llama.cpp, real /api/console, real curator dispatch, real AIAgent + direct request path): origin/main at the "cancelled" frame -> request_exited=false, worker_exited=false; with this change -> both true, provider saw the peer close, prompt reported cancelled 0.27s after the frame. Salvage of #106197 by @kyssta-exe (executor-future handle, cancel/timeout/ disconnect propagation) and #106320 by @Xixiartemis (deterministic lifecycle regression: terminal(cancelled) => no owned request or worker remains live; interrupt-on-late-registration). Both rebuilt slimmer: #106197 keyed its fallback on Future.cancel() returning False, but the handle it held was run_in_executor's asyncio wrapper, whose cancel() returns True while the thread keeps running, so its thread-name abort registry was never consulted; #106320's command-scoped ownership model is folded into one small module hooked at the turn facade instead of a per-caller `bind_agent`. Co-authored-by: kyssta-exe <218078013+kyssta-exe@users.noreply.github.com> Co-authored-by: Xixiartemis <182932319+Xixiartemis@users.noreply.github.com>
66 lines
2.4 KiB
Python
66 lines
2.4 KiB
Python
"""Host-owned cancellation for agents created deep inside synchronous work.
|
|
|
|
A host that runs a blocking command on a worker thread (Hermes Console) never sees
|
|
the ``AIAgent`` a CLI subcommand forks inside it, so it cannot call ``interrupt()``
|
|
when the user cancels. The host binds an :class:`InterruptScope` around the work;
|
|
every ``run_conversation()`` under that scope registers its agent, and
|
|
``scope.cancel()`` hard-interrupts them from any thread. Agents registering after
|
|
the cancel are interrupted immediately, so a cancel never loses the race with a
|
|
turn that has not started yet (#106179).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from contextlib import contextmanager, nullcontext
|
|
from contextvars import ContextVar
|
|
from typing import Any, Iterator, Optional
|
|
|
|
from agent.interrupt_compat import request_hard_interrupt
|
|
|
|
_ACTIVE_SCOPE: ContextVar[Optional["InterruptScope"]] = ContextVar("hermes_interrupt_scope", default=None)
|
|
|
|
|
|
class InterruptScope:
|
|
def __init__(self) -> None:
|
|
self._lock = threading.Lock()
|
|
self._agents: list[Any] = []
|
|
self.reason: Optional[str] = None
|
|
|
|
def cancel(self, reason: str) -> None:
|
|
"""Latch ``reason`` and hard-interrupt every agent running under this scope."""
|
|
with self._lock:
|
|
self.reason = reason
|
|
agents = list(self._agents)
|
|
for agent in agents:
|
|
request_hard_interrupt(agent, reason, tool_reason="host cancelled the command")
|
|
|
|
@contextmanager
|
|
def track(self, agent: Any) -> Iterator[None]:
|
|
with self._lock:
|
|
self._agents.append(agent)
|
|
reason = self.reason
|
|
if reason is not None:
|
|
request_hard_interrupt(agent, reason, tool_reason="host cancelled the command")
|
|
try:
|
|
yield
|
|
finally:
|
|
with self._lock:
|
|
self._agents.remove(agent)
|
|
|
|
|
|
@contextmanager
|
|
def bind_interrupt_scope(scope: Optional[InterruptScope]) -> Iterator[None]:
|
|
"""Make ``scope`` the owner of every agent turn started in this context."""
|
|
token = _ACTIVE_SCOPE.set(scope)
|
|
try:
|
|
yield
|
|
finally:
|
|
_ACTIVE_SCOPE.reset(token)
|
|
|
|
|
|
def track_in_interrupt_scope(agent: Any):
|
|
"""Register ``agent`` with the bound scope for the duration of its turn (no-op without one)."""
|
|
scope = _ACTIVE_SCOPE.get()
|
|
return nullcontext() if scope is None else scope.track(agent)
|