mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-29 18:46:59 +00:00
fix(delegate): isolate async batches from parent interrupts
Detached background delegation batches (_batch_runner) no longer honor the foreground parent's interrupt flag — a busy-submit interrupt in the TUI/desktop previously fabricated 'interrupted' results for background children that should outlive the turn. Explicit cancellation still works via _batch_interrupt. Rebased onto current main from PR #65040; both interrupt-suppression regression tests aligned with the current _session test helper. Original work by @AtakanGs in #65040.
This commit is contained in:
atakan g
Teknium
parent
5423e26a7d
commit
3d30232eba
2 changed files with 222 additions and 3 deletions
|
|
@ -13,6 +13,7 @@ import threading
|
|||
import time
|
||||
import types
|
||||
|
||||
import tools.async_delegation as ad
|
||||
from tui_gateway import server
|
||||
|
||||
|
||||
|
|
@ -130,6 +131,62 @@ def test_queued_flag_overrides_mode_never_touches_live_turn(monkeypatch):
|
|||
assert session["queued_prompt"]["text"] == "next turn text"
|
||||
|
||||
|
||||
def test_busy_interrupt_mode_ignores_completed_background_delegation(monkeypatch):
|
||||
"""A terminal delegation must not suppress normal busy-turn interruption."""
|
||||
monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt")
|
||||
calls = {"interrupt": 0}
|
||||
agent = types.SimpleNamespace(
|
||||
interrupt=lambda *a, **k: calls.__setitem__("interrupt", calls["interrupt"] + 1)
|
||||
)
|
||||
session = _session(agent=agent, running=True)
|
||||
|
||||
with ad._records_lock:
|
||||
ad._records["deleg_completed"] = {
|
||||
"delegation_id": "deleg_completed",
|
||||
"status": "completed",
|
||||
"session_key": "session-key",
|
||||
"origin_ui_session_id": "sid",
|
||||
}
|
||||
|
||||
try:
|
||||
resp = server._handle_busy_submit("r1", "sid", session, "continue", "ws-1")
|
||||
finally:
|
||||
with ad._records_lock:
|
||||
ad._records.clear()
|
||||
|
||||
assert resp["result"]["status"] == "queued"
|
||||
assert calls["interrupt"] == 1
|
||||
assert session["queued_prompt"]["text"] == "continue"
|
||||
|
||||
|
||||
def test_busy_interrupt_mode_ignores_foreign_background_delegation(monkeypatch):
|
||||
"""Another tab's background work must not suppress this tab's interrupt."""
|
||||
monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt")
|
||||
calls = {"interrupt": 0}
|
||||
agent = types.SimpleNamespace(
|
||||
interrupt=lambda *a, **k: calls.__setitem__("interrupt", calls["interrupt"] + 1)
|
||||
)
|
||||
session = _session(agent=agent, running=True)
|
||||
|
||||
with ad._records_lock:
|
||||
ad._records["deleg_foreign"] = {
|
||||
"delegation_id": "deleg_foreign",
|
||||
"status": "running",
|
||||
"session_key": "foreign-key",
|
||||
"origin_ui_session_id": "foreign-sid",
|
||||
}
|
||||
|
||||
try:
|
||||
resp = server._handle_busy_submit("r1", "sid", session, "interrupt me", "ws-1")
|
||||
finally:
|
||||
with ad._records_lock:
|
||||
ad._records.clear()
|
||||
|
||||
assert resp["result"]["status"] == "queued"
|
||||
assert calls["interrupt"] == 1
|
||||
assert session["queued_prompt"]["text"] == "interrupt me"
|
||||
|
||||
|
||||
def test_busy_steer_mode_injects_when_accepted(monkeypatch):
|
||||
monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "steer")
|
||||
agent = types.SimpleNamespace(steer=lambda text: True, interrupt=lambda *a, **k: None)
|
||||
|
|
@ -289,3 +346,160 @@ def test_drain_releases_running_on_dispatch_failure(monkeypatch):
|
|||
assert server._drain_queued_prompt("r1", "sid", session) is True
|
||||
# Failure must not leave the session wedged as running.
|
||||
assert session["running"] is False
|
||||
|
||||
|
||||
def test_busy_interrupt_mode_preserves_real_background_batch_completion(
|
||||
monkeypatch, tmp_path
|
||||
):
|
||||
"""Foreground interruption must not cancel its detached async batch."""
|
||||
import json
|
||||
import queue
|
||||
import time
|
||||
|
||||
import tools.delegate_tool as dt
|
||||
from gateway.session_context import clear_session_vars, set_session_vars
|
||||
from tools.process_registry import process_registry
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
monkeypatch.setattr(server, "_load_busy_input_mode", lambda: "interrupt")
|
||||
|
||||
isolated_queue = queue.Queue()
|
||||
monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
|
||||
ad._reset_for_tests()
|
||||
|
||||
calls = {"interrupt": 0}
|
||||
|
||||
class _Parent:
|
||||
def __init__(self):
|
||||
self._delegate_depth = 0
|
||||
self.session_id = "session-key"
|
||||
self._interrupt_requested = False
|
||||
self._active_children = []
|
||||
self._active_children_lock = None
|
||||
|
||||
def interrupt(self, *_args, **_kwargs):
|
||||
calls["interrupt"] += 1
|
||||
self._interrupt_requested = True
|
||||
|
||||
parent = _Parent()
|
||||
session = _session(agent=parent, running=True)
|
||||
|
||||
release_children = threading.Event()
|
||||
all_children_started = threading.Event()
|
||||
started_lock = threading.Lock()
|
||||
started = {"count": 0}
|
||||
child_ids = iter(("child-1", "child-2", "child-3"))
|
||||
|
||||
def _build_child(**_kwargs):
|
||||
return types.SimpleNamespace(
|
||||
_delegate_role="leaf",
|
||||
_subagent_id=next(child_ids),
|
||||
)
|
||||
|
||||
def _blocking_child(task_index, goal, child=None, parent_agent=None, **_kwargs):
|
||||
with started_lock:
|
||||
started["count"] += 1
|
||||
if started["count"] == 3:
|
||||
all_children_started.set()
|
||||
|
||||
release_children.wait(timeout=10)
|
||||
return {
|
||||
"task_index": task_index,
|
||||
"status": "completed",
|
||||
"summary": f"done: {goal}",
|
||||
"api_calls": 1,
|
||||
"duration_seconds": 0.1,
|
||||
"model": "test-model",
|
||||
"exit_reason": "completed",
|
||||
}
|
||||
|
||||
credentials = {
|
||||
"model": "test-model",
|
||||
"provider": None,
|
||||
"base_url": None,
|
||||
"api_key": None,
|
||||
"api_mode": None,
|
||||
"command": None,
|
||||
"args": None,
|
||||
}
|
||||
|
||||
monkeypatch.setattr(dt, "_build_child_agent", _build_child)
|
||||
monkeypatch.setattr(dt, "_run_single_child", _blocking_child)
|
||||
monkeypatch.setattr(
|
||||
dt,
|
||||
"_resolve_delegation_credentials",
|
||||
lambda *_args, **_kwargs: credentials,
|
||||
)
|
||||
|
||||
context_tokens = set_session_vars(
|
||||
source="tui",
|
||||
session_key="session-key",
|
||||
ui_session_id="sid",
|
||||
)
|
||||
|
||||
response = None
|
||||
event = None
|
||||
try:
|
||||
dispatched = json.loads(
|
||||
dt.delegate_task(
|
||||
tasks=[
|
||||
{"goal": "first"},
|
||||
{"goal": "second"},
|
||||
{"goal": "third"},
|
||||
],
|
||||
background=True,
|
||||
parent_agent=parent,
|
||||
)
|
||||
)
|
||||
assert dispatched["status"] == "dispatched"
|
||||
assert all_children_started.wait(timeout=5)
|
||||
|
||||
response = server._handle_busy_submit(
|
||||
"r1",
|
||||
"sid",
|
||||
session,
|
||||
"follow-up",
|
||||
"ws-1",
|
||||
)
|
||||
|
||||
# The old detached-batch loop polls this parent flag every 0.5 seconds.
|
||||
time.sleep(0.7)
|
||||
release_children.set()
|
||||
event = isolated_queue.get(timeout=5)
|
||||
finally:
|
||||
release_children.set()
|
||||
clear_session_vars(context_tokens)
|
||||
ad._reset_for_tests()
|
||||
|
||||
assert response["result"]["status"] == "queued"
|
||||
assert session["queued_prompt"]["text"] == "follow-up"
|
||||
assert calls["interrupt"] == 1
|
||||
|
||||
assert event["type"] == "async_delegation"
|
||||
assert event["origin_ui_session_id"] == "sid"
|
||||
assert event["session_key"] == "session-key"
|
||||
assert [result["status"] for result in event["results"]] == [
|
||||
"completed",
|
||||
"completed",
|
||||
"completed",
|
||||
]
|
||||
assert sorted(result["summary"] for result in event["results"]) == [
|
||||
"done: first",
|
||||
"done: second",
|
||||
"done: third",
|
||||
]
|
||||
|
||||
# Exercise the same positive-proof ownership gate used by the TUI's
|
||||
# post-turn delivery path, not just event production.
|
||||
isolated_queue.put(event)
|
||||
drained = process_registry.drain_notifications(
|
||||
session_key=session.get("session_key", ""),
|
||||
owns_event=lambda candidate: server._session_owns_notification_event(
|
||||
"sid", session, candidate
|
||||
),
|
||||
)
|
||||
assert len(drained) == 1
|
||||
delivered_event, synthetic_prompt = drained[0]
|
||||
assert delivered_event is event
|
||||
assert synthetic_prompt
|
||||
assert isolated_queue.empty()
|
||||
|
|
|
|||
|
|
@ -2979,7 +2979,7 @@ def delegate_task(
|
|||
child._live_transcript_path = str(_writer.path)
|
||||
children.append((i, t, child))
|
||||
|
||||
def _execute_and_aggregate() -> dict:
|
||||
def _execute_and_aggregate(*, honor_parent_interrupt: bool = True) -> dict:
|
||||
"""Run all built children (1 or N), join on them, aggregate results,
|
||||
fire subagent_stop hooks + cost rollup, and return the combined result
|
||||
dict. Used by BOTH the synchronous path and the background runner. In
|
||||
|
|
@ -3028,7 +3028,10 @@ def delegate_task(
|
|||
|
||||
pending = set(futures.keys())
|
||||
while pending:
|
||||
if getattr(parent_agent, "_interrupt_requested", False) is True:
|
||||
if (
|
||||
honor_parent_interrupt
|
||||
and getattr(parent_agent, "_interrupt_requested", False) is True
|
||||
):
|
||||
# Parent interrupted — collect whatever finished and
|
||||
# abandon the rest. Children already received the
|
||||
# interrupt signal; we just can't wait forever.
|
||||
|
|
@ -3270,7 +3273,9 @@ def delegate_task(
|
|||
pass
|
||||
|
||||
def _batch_runner():
|
||||
return _execute_and_aggregate()
|
||||
# This batch is detached from the foreground turn. Its lifecycle is
|
||||
# owned by the async registry and cancelled only via _batch_interrupt.
|
||||
return _execute_and_aggregate(honor_parent_interrupt=False)
|
||||
|
||||
def _batch_interrupt():
|
||||
for _c in _child_agents:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue