mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-27 17:58:07 +00:00
fix(delegation): timeout stuck async child runners
Async background delegation can leave gateway sessions holding only a dispatched handle when the detached runner wedges before it can return and enqueue a completion. Enforce the configured child timeout in the async registry so the parent observes a terminal timeout event and the async slot is released. Constraint: Issue #60203 reports long-lived gateway processes with background child delegates that never produce completion events despite child_timeout_seconds being configured. Rejected: Relying only on _run_single_child timeout handling | it cannot finalize the async registry when the outer runner thread itself never reaches normal completion. Confidence: high Scope-risk: narrow Directive: Keep background delegation completion owned by the async registry whenever detached workers can outlive the caller's immediate control. Tested: .venv/bin/python -m pytest tests/tools/test_async_delegation.py tests/tools/test_delegate_subagent_timeout_diagnostic.py tests/tools/test_delegate.py -q Tested: .venv/bin/python -m ruff check tools/async_delegation.py tools/delegate_tool.py tests/tools/test_async_delegation.py Tested: git diff --check Not-tested: Multi-day real gateway degradation; covered with deterministic stuck-runner registry tests.
This commit is contained in:
izumi0uu
Teknium
parent
c593f7face
commit
65420cdecd
3 changed files with 370 additions and 20 deletions
|
|
@ -229,6 +229,183 @@ def test_interrupt_all_signals_running_children():
|
|||
assert evt["status"] == "interrupted"
|
||||
|
||||
|
||||
def test_async_delegation_timeout_finalizes_stuck_runner():
|
||||
gate = threading.Event()
|
||||
interrupted = {"count": 0}
|
||||
|
||||
def stuck_runner():
|
||||
gate.wait(timeout=5)
|
||||
return {"status": "completed", "summary": "too late"}
|
||||
|
||||
def interrupt_fn():
|
||||
interrupted["count"] += 1
|
||||
|
||||
res = ad.dispatch_async_delegation(
|
||||
goal="stuck child", context=None, toolsets=None, role="leaf",
|
||||
model="m", session_key="", runner=stuck_runner,
|
||||
interrupt_fn=interrupt_fn, max_async_children=1, timeout_seconds=0.1,
|
||||
)
|
||||
assert res["status"] == "dispatched"
|
||||
|
||||
evt = _drain_one(timeout=2.0)
|
||||
try:
|
||||
assert evt is not None
|
||||
assert evt["type"] == "async_delegation"
|
||||
assert evt["status"] == "timeout"
|
||||
assert evt["delegation_id"] == res["delegation_id"]
|
||||
assert evt["api_calls"] == 0
|
||||
assert "timed out after 0.1s" in evt["error"]
|
||||
assert interrupted["count"] == 1
|
||||
assert ad.active_count() == 0
|
||||
finally:
|
||||
gate.set()
|
||||
|
||||
# If the ignored runner eventually returns, it must not enqueue a second
|
||||
# completion for a delegation the watchdog already finalized.
|
||||
assert _drain_one(timeout=0.5) is None
|
||||
|
||||
|
||||
def test_async_delegation_batch_timeout_finalizes_stuck_runner():
|
||||
gate = threading.Event()
|
||||
interrupted = {"count": 0}
|
||||
|
||||
def stuck_batch():
|
||||
gate.wait(timeout=5)
|
||||
return {"results": [{"status": "completed", "summary": "too late"}]}
|
||||
|
||||
def interrupt_fn():
|
||||
interrupted["count"] += 1
|
||||
|
||||
res = ad.dispatch_async_delegation_batch(
|
||||
goals=["a", "b"], context="ctx", toolsets=None, role="leaf",
|
||||
model="m", session_key="", runner=stuck_batch,
|
||||
interrupt_fn=interrupt_fn, max_async_children=1, timeout_seconds=0.1,
|
||||
)
|
||||
assert res["status"] == "dispatched"
|
||||
|
||||
evt = _drain_one(timeout=2.0)
|
||||
try:
|
||||
assert evt is not None
|
||||
assert evt["type"] == "async_delegation"
|
||||
assert evt["status"] == "timeout"
|
||||
assert evt["is_batch"] is True
|
||||
assert evt["goals"] == ["a", "b"]
|
||||
assert evt["results"] == []
|
||||
assert "timed out after 0.1s" in evt["error"]
|
||||
assert interrupted["count"] == 1
|
||||
assert ad.active_count() == 0
|
||||
finally:
|
||||
gate.set()
|
||||
|
||||
assert _drain_one(timeout=0.5) is None
|
||||
|
||||
|
||||
def test_timeout_stays_finalizing_until_durable_persistence(tmp_path, monkeypatch):
|
||||
gate = threading.Event()
|
||||
persist_entered = threading.Event()
|
||||
allow_persist = threading.Event()
|
||||
real_persist = ad._persist_completion
|
||||
|
||||
def blocking_persist(event, result):
|
||||
persist_entered.set()
|
||||
allow_persist.wait(timeout=5)
|
||||
real_persist(event, result)
|
||||
|
||||
def stuck_runner():
|
||||
gate.wait(timeout=5)
|
||||
return {"status": "completed", "summary": "too late"}
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
monkeypatch.setattr(ad, "_persist_completion", blocking_persist)
|
||||
dispatched = ad.dispatch_async_delegation(
|
||||
goal="durable timeout", context=None, toolsets=None, role="leaf",
|
||||
model="m", session_key="owner", runner=stuck_runner,
|
||||
max_async_children=1, timeout_seconds=0.05,
|
||||
)
|
||||
|
||||
try:
|
||||
assert persist_entered.wait(timeout=2)
|
||||
assert ad.active_count() == 1
|
||||
record = next(
|
||||
item for item in ad.list_async_delegations()
|
||||
if item["delegation_id"] == dispatched["delegation_id"]
|
||||
)
|
||||
assert record["status"] == "finalizing"
|
||||
assert process_registry.completion_queue.empty()
|
||||
|
||||
allow_persist.set()
|
||||
evt = _drain_for(dispatched["delegation_id"])
|
||||
assert evt is not None
|
||||
assert evt["status"] == "timeout"
|
||||
assert ad.active_count() == 0
|
||||
durable = ad.get_durable_delegation(dispatched["delegation_id"])
|
||||
assert durable["state"] == "timeout"
|
||||
assert durable["delivery_state"] == "pending"
|
||||
finally:
|
||||
allow_persist.set()
|
||||
gate.set()
|
||||
|
||||
|
||||
def test_timeout_completion_restores_once_after_process_restart(tmp_path):
|
||||
repo = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
|
||||
env = {**os.environ, "HERMES_HOME": str(tmp_path), "PYTHONPATH": repo}
|
||||
producer = r'''
|
||||
import json
|
||||
import threading
|
||||
import time
|
||||
from tools import async_delegation as ad
|
||||
gate = threading.Event()
|
||||
r = ad.dispatch_async_delegation(
|
||||
goal="restart timeout", context=None, toolsets=None, role="leaf", model="m",
|
||||
session_key="owner-session", parent_session_id="durable-parent",
|
||||
runner=lambda: gate.wait(timeout=60), timeout_seconds=.05,
|
||||
)
|
||||
deadline = time.time() + 5
|
||||
while ad.active_count() and time.time() < deadline:
|
||||
time.sleep(.01)
|
||||
row = ad.get_durable_delegation(r["delegation_id"])
|
||||
print(json.dumps({"delegation_id": r["delegation_id"], "row": row}, sort_keys=True))
|
||||
'''
|
||||
first = subprocess.run(
|
||||
[sys.executable, "-c", producer], cwd=repo, env=env,
|
||||
text=True, capture_output=True, timeout=15, check=True,
|
||||
)
|
||||
produced = json.loads(first.stdout.strip().splitlines()[-1])
|
||||
delegation_id = produced["delegation_id"]
|
||||
assert produced["row"]["state"] == "timeout"
|
||||
assert produced["row"]["delivery_state"] == "pending"
|
||||
|
||||
consumer = r'''
|
||||
import json
|
||||
from tools.process_registry import process_registry
|
||||
evt = process_registry.completion_queue.get_nowait()
|
||||
print(json.dumps({"event": evt, "remaining": process_registry.completion_queue.qsize()}, sort_keys=True))
|
||||
'''
|
||||
second = subprocess.run(
|
||||
[sys.executable, "-c", consumer], cwd=repo, env=env,
|
||||
text=True, capture_output=True, timeout=15, check=True,
|
||||
)
|
||||
restored = json.loads(second.stdout.strip().splitlines()[-1])
|
||||
assert restored["remaining"] == 0
|
||||
assert restored["event"]["delegation_id"] == delegation_id
|
||||
assert restored["event"]["status"] == "timeout"
|
||||
assert restored["event"]["restored"] is True
|
||||
|
||||
acker = f'''
|
||||
from tools import async_delegation as ad
|
||||
assert ad.mark_completion_delivered({delegation_id!r})
|
||||
'''
|
||||
subprocess.run(
|
||||
[sys.executable, "-c", acker], cwd=repo, env=env,
|
||||
text=True, capture_output=True, timeout=15, check=True,
|
||||
)
|
||||
probe = subprocess.run(
|
||||
[sys.executable, "-c", "from tools.process_registry import process_registry; print(process_registry.completion_queue.qsize())"],
|
||||
cwd=repo, env=env, text=True, capture_output=True, timeout=15, check=True,
|
||||
)
|
||||
assert probe.stdout.strip().splitlines()[-1] == "0"
|
||||
|
||||
|
||||
def test_completed_records_pruned_to_cap():
|
||||
# Run more than the retention cap quickly; ensure list doesn't grow forever.
|
||||
for i in range(ad._MAX_RETAINED_COMPLETED + 10):
|
||||
|
|
@ -767,6 +944,45 @@ def test_delegate_task_background_batch_runs_as_one_unit(monkeypatch):
|
|||
assert _drain_one() is None
|
||||
|
||||
|
||||
def test_delegate_task_background_passes_child_timeout_to_async_registry(monkeypatch):
|
||||
import json
|
||||
from unittest.mock import MagicMock
|
||||
import tools.delegate_tool as dt
|
||||
|
||||
parent = MagicMock()
|
||||
parent._delegate_depth = 0
|
||||
parent.session_id = "sess"
|
||||
parent._interrupt_requested = False
|
||||
parent._active_children = []
|
||||
parent._active_children_lock = None
|
||||
|
||||
fake_child = MagicMock()
|
||||
fake_child._delegate_role = "leaf"
|
||||
fake_child._subagent_id = "s1"
|
||||
|
||||
creds = {
|
||||
"model": "m", "provider": None, "base_url": None, "api_key": None,
|
||||
"api_mode": None, "command": None, "args": None,
|
||||
}
|
||||
captured = {}
|
||||
|
||||
def fake_dispatch(**kwargs):
|
||||
captured.update(kwargs)
|
||||
return {"status": "dispatched", "delegation_id": "deleg_timeout"}
|
||||
|
||||
monkeypatch.setattr(dt, "_build_child_agent", lambda **kw: fake_child)
|
||||
monkeypatch.setattr(dt, "_resolve_delegation_credentials", lambda *a, **k: creds)
|
||||
monkeypatch.setattr(dt, "_get_child_timeout", lambda: 600.0)
|
||||
monkeypatch.setattr(ad, "dispatch_async_delegation_batch", fake_dispatch)
|
||||
|
||||
out = dt.delegate_task(goal="background timeout", background=True, parent_agent=parent)
|
||||
|
||||
parsed = json.loads(out)
|
||||
assert parsed["status"] == "dispatched"
|
||||
assert parsed["delegation_id"] == "deleg_timeout"
|
||||
assert captured["timeout_seconds"] == 600.0
|
||||
|
||||
|
||||
def test_model_dispatch_forces_background():
|
||||
"""The MODEL-facing dispatch path forces background=True for any top-level
|
||||
delegation (single task OR batch), and keeps it off for an orchestrator
|
||||
|
|
|
|||
|
|
@ -483,6 +483,10 @@ def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]:
|
|||
}
|
||||
|
||||
|
||||
def _format_timeout_seconds(timeout_seconds: float) -> str:
|
||||
return f"{timeout_seconds:g}"
|
||||
|
||||
|
||||
def _get_executor(max_workers: int) -> ThreadPoolExecutor:
|
||||
"""Lazily create (or grow) the shared daemon executor.
|
||||
|
||||
|
|
@ -572,6 +576,7 @@ def dispatch_async_delegation(
|
|||
origin_session_id: str = "",
|
||||
interrupt_fn: Optional[Callable[[], None]] = None,
|
||||
max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN,
|
||||
timeout_seconds: Optional[float] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Spawn ``runner`` on the daemon executor and return a handle immediately.
|
||||
|
||||
|
|
@ -624,6 +629,7 @@ def dispatch_async_delegation(
|
|||
"dispatched_at": dispatched_at,
|
||||
"completed_at": None,
|
||||
"interrupt_fn": interrupt_fn,
|
||||
"timeout_seconds": timeout_seconds,
|
||||
}
|
||||
# Capacity check and record insert under ONE lock hold — checking
|
||||
# active_count() separately would let two concurrent dispatches (e.g.
|
||||
|
|
@ -679,6 +685,7 @@ def dispatch_async_delegation(
|
|||
"status": "rejected",
|
||||
"error": f"Failed to schedule async delegation: {exc}",
|
||||
}
|
||||
_start_timeout_watchdog(delegation_id, timeout_seconds, is_batch=False)
|
||||
|
||||
logger.info(
|
||||
"Dispatched async delegation %s (session_key=%s): %s",
|
||||
|
|
@ -689,19 +696,39 @@ def dispatch_async_delegation(
|
|||
|
||||
def _finalize(delegation_id: str, result: Dict[str, Any], status: str) -> None:
|
||||
"""Mark a record complete and push the completion event onto the queue."""
|
||||
claimed = _begin_finalization(delegation_id)
|
||||
if claimed is None:
|
||||
return
|
||||
event_record, _interrupt_fn = claimed
|
||||
|
||||
_push_completion_event(event_record, result, status)
|
||||
_finish_finalization(delegation_id, status)
|
||||
|
||||
|
||||
def _begin_finalization(
|
||||
delegation_id: str,
|
||||
) -> Optional[tuple[Dict[str, Any], Optional[Callable[[], None]]]]:
|
||||
"""Atomically claim terminal delivery while keeping the record active."""
|
||||
with _records_lock:
|
||||
record = _records.get(delegation_id)
|
||||
if record is None:
|
||||
if record is None or record.get("status") != "running":
|
||||
return
|
||||
# Stay active until durable persistence and queue publication finish;
|
||||
# otherwise process shutdown can kill this daemon worker in the narrow
|
||||
# gap after status flips but before SQLite is committed.
|
||||
record["status"] = "finalizing"
|
||||
record["completed_at"] = time.time()
|
||||
interrupt_fn = record.get("interrupt_fn")
|
||||
record["interrupt_fn"] = None # drop the closure; child is done
|
||||
timer = record.pop("_timeout_timer", None)
|
||||
event_record = dict(record)
|
||||
|
||||
_push_completion_event(event_record, result, status)
|
||||
if timer is not None:
|
||||
timer.cancel()
|
||||
return event_record, interrupt_fn
|
||||
|
||||
|
||||
def _finish_finalization(delegation_id: str, status: str) -> None:
|
||||
with _records_lock:
|
||||
record = _records.get(delegation_id)
|
||||
if record is not None:
|
||||
|
|
@ -783,6 +810,7 @@ def dispatch_async_delegation_batch(
|
|||
interrupt_fn: Optional[Callable[[], None]] = None,
|
||||
max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN,
|
||||
delegation_id: Optional[str] = None,
|
||||
timeout_seconds: Optional[float] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Dispatch a WHOLE fan-out batch as ONE background unit.
|
||||
|
||||
|
|
@ -828,6 +856,7 @@ def dispatch_async_delegation_batch(
|
|||
"completed_at": None,
|
||||
"interrupt_fn": interrupt_fn,
|
||||
"is_batch": True,
|
||||
"timeout_seconds": timeout_seconds,
|
||||
}
|
||||
with _records_lock:
|
||||
running = sum(
|
||||
|
|
@ -884,6 +913,7 @@ def dispatch_async_delegation_batch(
|
|||
"status": "rejected",
|
||||
"error": f"Failed to schedule async delegation batch: {exc}",
|
||||
}
|
||||
_start_timeout_watchdog(delegation_id, timeout_seconds, is_batch=True)
|
||||
|
||||
logger.info(
|
||||
"Dispatched async delegation batch %s (%d task(s), session_key=%s)",
|
||||
|
|
@ -896,22 +926,26 @@ def _finalize_batch(
|
|||
delegation_id: str, combined: Dict[str, Any], status: str
|
||||
) -> None:
|
||||
"""Mark a batch record complete and push ONE combined completion event."""
|
||||
with _records_lock:
|
||||
record = _records.get(delegation_id)
|
||||
if record is None:
|
||||
return
|
||||
record["status"] = "finalizing"
|
||||
record["completed_at"] = time.time()
|
||||
record["interrupt_fn"] = None
|
||||
event_record = dict(record)
|
||||
claimed = _begin_finalization(delegation_id)
|
||||
if claimed is None:
|
||||
return
|
||||
event_record, _interrupt_fn = claimed
|
||||
|
||||
_push_batch_completion_event(event_record, combined, status)
|
||||
_finish_finalization(delegation_id, status)
|
||||
|
||||
|
||||
def _push_batch_completion_event(
|
||||
event_record: Dict[str, Any], combined: Dict[str, Any], status: str
|
||||
) -> None:
|
||||
"""Push a combined async-delegation batch completion event."""
|
||||
try:
|
||||
from tools.process_registry import process_registry
|
||||
except Exception as exc: # pragma: no cover
|
||||
logger.error(
|
||||
"Async delegation batch %s finished but process_registry import "
|
||||
"failed; result lost: %s",
|
||||
delegation_id, exc,
|
||||
event_record.get("delegation_id"), exc,
|
||||
)
|
||||
return
|
||||
|
||||
|
|
@ -919,7 +953,7 @@ def _finalize_batch(
|
|||
completed_at = event_record.get("completed_at") or time.time()
|
||||
evt = {
|
||||
"type": "async_delegation",
|
||||
"delegation_id": delegation_id,
|
||||
"delegation_id": event_record.get("delegation_id"),
|
||||
"session_key": event_record.get("session_key", ""),
|
||||
"origin_ui_session_id": event_record.get("origin_ui_session_id", ""),
|
||||
"origin_session_id": event_record.get("origin_session_id", ""),
|
||||
|
|
@ -951,14 +985,102 @@ def _finalize_batch(
|
|||
logger.error(
|
||||
"Async delegation batch %s: failed to enqueue completion event; "
|
||||
"result lost: %s",
|
||||
delegation_id, exc,
|
||||
event_record.get("delegation_id"), exc,
|
||||
)
|
||||
finally:
|
||||
with _records_lock:
|
||||
record = _records.get(delegation_id)
|
||||
if record is not None:
|
||||
record["status"] = status
|
||||
_prune_completed_locked()
|
||||
|
||||
|
||||
def _start_timeout_watchdog(
|
||||
delegation_id: str,
|
||||
timeout_seconds: Optional[float],
|
||||
*,
|
||||
is_batch: bool,
|
||||
) -> None:
|
||||
"""Finalize a detached delegation if its worker never returns.
|
||||
|
||||
``delegate_task(background=true)`` returns a handle immediately; the only
|
||||
user-visible result is the completion event this module enqueues later. If
|
||||
the worker thread wedges before returning, the normal ``finally`` block is
|
||||
never reached and users only see a permanent "dispatched" state. A
|
||||
configured timeout must therefore be enforced by the async registry itself,
|
||||
not only inside ``delegate_tool._run_single_child``.
|
||||
"""
|
||||
if timeout_seconds is None or timeout_seconds <= 0:
|
||||
return
|
||||
|
||||
timer = threading.Timer(
|
||||
timeout_seconds,
|
||||
_expire_delegation,
|
||||
args=(delegation_id, float(timeout_seconds)),
|
||||
kwargs={"is_batch": is_batch},
|
||||
)
|
||||
timer.daemon = True
|
||||
with _records_lock:
|
||||
record = _records.get(delegation_id)
|
||||
if record is None or record.get("status") != "running":
|
||||
return
|
||||
record["timeout_seconds"] = float(timeout_seconds)
|
||||
record["_timeout_timer"] = timer
|
||||
timer.start()
|
||||
|
||||
|
||||
def _expire_delegation(
|
||||
delegation_id: str,
|
||||
timeout_seconds: float,
|
||||
*,
|
||||
is_batch: bool,
|
||||
) -> None:
|
||||
"""Timeout a still-running async delegation and emit one completion."""
|
||||
claimed = _begin_finalization(delegation_id)
|
||||
if claimed is None:
|
||||
return
|
||||
event_record, interrupt_fn = claimed
|
||||
|
||||
completed_at = event_record.get("completed_at") or time.time()
|
||||
duration = round(
|
||||
completed_at - (event_record.get("dispatched_at") or completed_at),
|
||||
2,
|
||||
)
|
||||
timeout_text = _format_timeout_seconds(timeout_seconds)
|
||||
error = (
|
||||
f"Async delegation {delegation_id} timed out after {timeout_text}s "
|
||||
"without producing a completion event. Hermes requested interruption; "
|
||||
"the detached worker may be stuck before or inside the first model API "
|
||||
"call."
|
||||
)
|
||||
if is_batch:
|
||||
_push_batch_completion_event(
|
||||
event_record,
|
||||
{
|
||||
"results": [],
|
||||
"error": error,
|
||||
"total_duration_seconds": duration,
|
||||
},
|
||||
"timeout",
|
||||
)
|
||||
else:
|
||||
_push_completion_event(
|
||||
event_record,
|
||||
{
|
||||
"status": "timeout",
|
||||
"summary": None,
|
||||
"error": error,
|
||||
"api_calls": 0,
|
||||
"duration_seconds": duration,
|
||||
"exit_reason": "timeout",
|
||||
},
|
||||
"timeout",
|
||||
)
|
||||
_finish_finalization(delegation_id, "timeout")
|
||||
|
||||
if callable(interrupt_fn):
|
||||
try:
|
||||
interrupt_fn()
|
||||
except Exception as exc:
|
||||
logger.debug(
|
||||
"Async delegation %s timeout interrupt failed: %s",
|
||||
delegation_id,
|
||||
exc,
|
||||
)
|
||||
|
||||
|
||||
def list_async_delegations() -> List[Dict[str, Any]]:
|
||||
|
|
@ -968,7 +1090,11 @@ def list_async_delegations() -> List[Dict[str, Any]]:
|
|||
"""
|
||||
with _records_lock:
|
||||
return [
|
||||
{k: v for k, v in r.items() if k != "interrupt_fn"}
|
||||
{
|
||||
k: v
|
||||
for k, v in r.items()
|
||||
if k not in {"interrupt_fn", "_timeout_timer"}
|
||||
}
|
||||
for r in _records.values()
|
||||
]
|
||||
|
||||
|
|
@ -1066,4 +1192,11 @@ def _reset_for_tests() -> None:
|
|||
_executor = None
|
||||
_executor_max_workers = 0
|
||||
with _records_lock:
|
||||
timers = [
|
||||
timer
|
||||
for record in _records.values()
|
||||
if (timer := record.get("_timeout_timer")) is not None
|
||||
]
|
||||
_records.clear()
|
||||
for timer in timers:
|
||||
timer.cancel()
|
||||
|
|
|
|||
|
|
@ -3057,6 +3057,7 @@ def delegate_task(
|
|||
# Reuse the live-transcript directory's id (when created) so the
|
||||
# returned delegation_id matches cache/delegation/live/<id>/.
|
||||
delegation_id=live_deleg_id,
|
||||
timeout_seconds=_get_child_timeout(),
|
||||
)
|
||||
|
||||
if dispatch.get("status") == "dispatched":
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue