mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-21 16:18:55 +00:00
3d9be27895
run_agent._dispatch_delegate_task forces background=True for every top-level
delegation, and async_delivery_supported() returns True for any session that
never binds the capability. On runners that cannot receive a completion after
their turn ends, that combination silently discards every subagent result: the
model gets a dispatch handle, ends its turn, and reports 'waiting for results'.
Two such runners never bind the capability:
* hermes -z (one-shot) prints one final response and exits. It bypasses cli.py,
so nothing drains process_registry.completion_queue (only the interactive
process_loop and the gateway watchers do).
* cron run_job clears the HERMES_SESSION_* routing keys, so a completion event
carries session_key="" — _enrich_async_delegation_routing cannot resolve it
and _inject_watch_notification drops it ("no routing metadata"). By then
run_job has already shipped the job's final response via _deliver_result;
there is no turn left to re-enter. Worse, get_current_session_key() can fall
back to the ambient os.environ HERMES_SESSION_KEY, so a cron subagent's output
can be routed into an unrelated user chat rather than merely dropped.
Add declare_stateless_channel() and bind it in both runners, routing
delegate_task to its existing inline/synchronous path — the same fallback the
stateless HTTP adapter already relies on, and the fix suggested in #63142. The
helper binds only the capability: set_session_vars() would also latch
_session_context_engaged, which a pure single-process one-shot must not trigger.
Also correct two agent-facing strings that hardcoded 'stateless HTTP API' as the
only channel without async delivery (delegate_tool, terminal_tool); they now name
the actual condition.
Repro (before): hermes -z 'Use delegate_task to spawn a subagent that replies
BANANA. Report its reply.' -> "Waiting for the subagent's response...", exit 0,
no BANANA. After: BANANA is returned in-turn.
Fixes #53027
Fixes #63142
315 lines
12 KiB
Python
315 lines
12 KiB
Python
"""Tests for the async-delivery capability gate (issue #10760).
|
|
|
|
Stateless request/response adapters (the API server / WebUI path) cannot route
|
|
a background completion back to the agent after a turn ends — there is no
|
|
persistent channel and ``APIServerAdapter.send()`` is a no-op stub. So tools
|
|
that promise async delivery (``terminal`` notify_on_complete / watch_patterns,
|
|
``delegate_task`` background=True) must refuse the promise on that path instead
|
|
of silently registering a watcher that never fires.
|
|
|
|
This is wired through:
|
|
- ``BasePlatformAdapter.supports_async_delivery`` (default True)
|
|
- ``APIServerAdapter.supports_async_delivery = False``
|
|
- ``gateway.session_context._SESSION_ASYNC_DELIVERY`` contextvar +
|
|
``async_delivery_supported()`` helper, bound per-session.
|
|
|
|
These are behavior/invariant tests (how the capability relates to the channel),
|
|
not snapshots of a current value.
|
|
"""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from gateway.session_context import (
|
|
async_delivery_supported,
|
|
clear_session_vars,
|
|
get_session_env,
|
|
reset_session_vars,
|
|
set_session_vars,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Capability helper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestAsyncDeliverySupported:
|
|
def test_default_unbound_is_supported(self):
|
|
"""CLI / cron / unaware paths never bind the var -> supported."""
|
|
assert async_delivery_supported() is True
|
|
|
|
def test_set_true_is_supported(self):
|
|
tokens = set_session_vars(
|
|
platform="telegram",
|
|
chat_id="123",
|
|
session_key="telegram:private:123",
|
|
async_delivery=True,
|
|
)
|
|
try:
|
|
assert async_delivery_supported() is True
|
|
# Platform metadata stays readable alongside the capability.
|
|
assert get_session_env("HERMES_SESSION_PLATFORM") == "telegram"
|
|
finally:
|
|
clear_session_vars(tokens)
|
|
|
|
def test_set_false_is_unsupported(self):
|
|
tokens = set_session_vars(
|
|
platform="api_server",
|
|
chat_id="sess1",
|
|
session_key="sess1",
|
|
async_delivery=False,
|
|
)
|
|
try:
|
|
assert async_delivery_supported() is False
|
|
# Platform must still be readable for routing/diagnostics even
|
|
# though delivery is unsupported.
|
|
assert get_session_env("HERMES_SESSION_PLATFORM") == "api_server"
|
|
finally:
|
|
clear_session_vars(tokens)
|
|
|
|
def test_omitted_arg_defaults_supported(self):
|
|
"""Back-compat: callers that don't pass async_delivery stay supported."""
|
|
tokens = set_session_vars(platform="discord", chat_id="9")
|
|
try:
|
|
assert async_delivery_supported() is True
|
|
finally:
|
|
clear_session_vars(tokens)
|
|
|
|
def test_clear_resets_to_default_supported(self):
|
|
"""A cleared context must fall back to default-supported, NOT be
|
|
mistaken for an opted-out stateless adapter."""
|
|
tokens = set_session_vars(
|
|
platform="api_server", session_key="s1", async_delivery=False
|
|
)
|
|
assert async_delivery_supported() is False
|
|
clear_session_vars(tokens)
|
|
assert async_delivery_supported() is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stateless runners — issues #53027 / #63142
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDeclareStatelessChannel:
|
|
"""``hermes -z`` and cron cannot receive a completion after their turn ends.
|
|
|
|
Cron clears the ``HERMES_SESSION_*`` routing keys, so an async delegation's
|
|
completion event carries ``session_key=""`` and the gateway watcher drops it
|
|
for lack of routing metadata; either way the job's final response has already
|
|
shipped. One-shot simply exits. Both must bind the capability, or
|
|
``delegate_task`` is forced background and every subagent result is lost.
|
|
"""
|
|
|
|
def test_declare_stateless_channel_disables_async_delivery(self):
|
|
from gateway.session_context import declare_stateless_channel
|
|
|
|
reset_session_vars() # don't assume ambient contextvar state
|
|
assert async_delivery_supported() is True
|
|
try:
|
|
declare_stateless_channel()
|
|
assert async_delivery_supported() is False
|
|
finally:
|
|
reset_session_vars()
|
|
|
|
def test_declare_does_not_engage_full_session_context(self):
|
|
"""The helper binds ONLY the capability.
|
|
|
|
``set_session_vars`` latches ``_session_context_engaged``, which flips the
|
|
subprocess env bridge to ContextVar-authoritative. A pure single-process
|
|
one-shot must not trigger that as a side effect of declaring a capability.
|
|
"""
|
|
from gateway import session_context as sc
|
|
|
|
reset_session_vars()
|
|
engaged_before = sc._session_context_engaged
|
|
try:
|
|
sc.declare_stateless_channel()
|
|
assert sc._session_context_engaged is engaged_before
|
|
finally:
|
|
reset_session_vars()
|
|
|
|
|
|
class TestStatelessChannelForcesSyncDelegation:
|
|
"""The behavioral contract: a stateless channel must run delegations INLINE.
|
|
|
|
This is the regression that #53027 / #63142 describe — a background dispatch
|
|
on a channel that can never deliver the completion.
|
|
"""
|
|
|
|
def test_background_delegation_runs_inline_when_channel_is_stateless(
|
|
self, monkeypatch
|
|
):
|
|
import tools.delegate_tool as dt
|
|
from gateway.session_context import declare_stateless_channel
|
|
|
|
class _Parent:
|
|
_delegate_depth = 0
|
|
_subagent_id = None
|
|
|
|
fake_child = type("C", (), {"_subagent_id": "s1"})()
|
|
dispatched = []
|
|
|
|
def _fake_dispatch(*a, **kw):
|
|
dispatched.append(kw)
|
|
return {"delegation_id": "deleg_x"}
|
|
|
|
def _child(task_index, goal, child=None, parent_agent=None, **kw):
|
|
return {
|
|
"task_index": 0, "status": "completed", "summary": f"done: {goal}",
|
|
"api_calls": 1, "duration_seconds": 0.1, "model": "m",
|
|
"exit_reason": "completed",
|
|
}
|
|
|
|
creds = {
|
|
"model": "m", "provider": None, "base_url": None, "api_key": None,
|
|
"api_mode": None, "command": None, "args": None,
|
|
}
|
|
monkeypatch.setattr(dt, "_build_child_agent", lambda **kw: fake_child)
|
|
monkeypatch.setattr(dt, "_run_single_child", _child)
|
|
monkeypatch.setattr(dt, "_resolve_delegation_credentials", lambda *a, **k: creds)
|
|
monkeypatch.setattr(
|
|
"tools.async_delegation.dispatch_async_delegation_batch", _fake_dispatch
|
|
)
|
|
|
|
reset_session_vars()
|
|
try:
|
|
declare_stateless_channel()
|
|
out = dt.delegate_task(
|
|
goal="review the spec", background=True, parent_agent=_Parent()
|
|
)
|
|
finally:
|
|
reset_session_vars()
|
|
|
|
parsed = json.loads(out)
|
|
# The whole point: NOT dispatched to a channel that can't deliver.
|
|
assert not dispatched, "stateless channel must not dispatch a detached child"
|
|
assert parsed.get("status") != "dispatched"
|
|
# The caller gets the actual work product, in-turn.
|
|
assert "results" in parsed
|
|
assert "done: review the spec" in json.dumps(parsed)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Adapter capability flag
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestAdapterCapabilityFlag:
|
|
def test_base_default_true(self):
|
|
from gateway.platforms.base import BasePlatformAdapter
|
|
|
|
assert BasePlatformAdapter.supports_async_delivery is True
|
|
|
|
def test_api_server_false(self):
|
|
from gateway.platforms.api_server import APIServerAdapter
|
|
|
|
assert APIServerAdapter.supports_async_delivery is False
|
|
|
|
def test_api_server_bind_chokepoint_hardwires_no_delivery(self):
|
|
"""Every API-server agent-entry path binds through
|
|
_bind_api_server_session, which hardwires async_delivery=False — a new
|
|
route physically cannot reintroduce the silent no-op (#10760)."""
|
|
from gateway.platforms.api_server import APIServerAdapter
|
|
from gateway.session_context import clear_session_vars, get_session_env
|
|
|
|
tokens = APIServerAdapter._bind_api_server_session(
|
|
chat_id="c1", session_key="sk1", session_id="sid1"
|
|
)
|
|
try:
|
|
assert async_delivery_supported() is False
|
|
assert get_session_env("HERMES_SESSION_PLATFORM") == "api_server"
|
|
finally:
|
|
clear_session_vars(tokens)
|
|
|
|
def test_api_server_binding_does_not_outlive_turn(self):
|
|
"""The no-delivery decision is request-scoped, NOT stuck to the session.
|
|
After clear, a session resumed on a delivering interface re-binds fresh
|
|
and is NOT blocked."""
|
|
from gateway.platforms.api_server import APIServerAdapter
|
|
from gateway.session_context import clear_session_vars
|
|
|
|
# Turn 1: same session over the API server -> blocked.
|
|
tokens = APIServerAdapter._bind_api_server_session(session_key="shared-key")
|
|
assert async_delivery_supported() is False
|
|
clear_session_vars(tokens)
|
|
|
|
# Turn 2: SAME session_key resumed on a delivering interface (CLI/gateway)
|
|
# -> supported. The earlier False did not follow the session.
|
|
tokens = set_session_vars(
|
|
platform="telegram",
|
|
session_key="shared-key",
|
|
async_delivery=True,
|
|
)
|
|
try:
|
|
assert async_delivery_supported() is True
|
|
finally:
|
|
clear_session_vars(tokens)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# terminal_tool: refuses to register a watcher on unsupported sessions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestTerminalNotifyGate:
|
|
@pytest.fixture(autouse=True)
|
|
def _clean_watchers(self):
|
|
from tools.process_registry import process_registry
|
|
|
|
process_registry.pending_watchers = []
|
|
yield
|
|
process_registry.pending_watchers = []
|
|
|
|
def _run_bg(self, command):
|
|
from tools.terminal_tool import terminal_tool
|
|
|
|
return json.loads(
|
|
terminal_tool(command=command, background=True, notify_on_complete=True)
|
|
)
|
|
|
|
def test_api_server_skips_watcher_and_notes(self):
|
|
from tools.process_registry import process_registry
|
|
|
|
tokens = set_session_vars(
|
|
platform="api_server", chat_id="s1", session_key="s1", async_delivery=False
|
|
)
|
|
try:
|
|
d = self._run_bg("sleep 30 && echo DONE")
|
|
finally:
|
|
clear_session_vars(tokens)
|
|
|
|
assert d.get("notify_on_complete") is False
|
|
assert d.get("notify_unsupported"), "must explain the limitation"
|
|
assert "poll" in d["notify_unsupported"].lower()
|
|
assert len(process_registry.pending_watchers) == 0
|
|
|
|
def test_gateway_registers_watcher(self):
|
|
from tools.process_registry import process_registry
|
|
|
|
tokens = set_session_vars(
|
|
platform="telegram",
|
|
chat_id="123",
|
|
thread_id="7",
|
|
user_id="u1",
|
|
session_key="telegram:private:123",
|
|
async_delivery=True,
|
|
)
|
|
try:
|
|
d = self._run_bg("sleep 30 && echo DONE")
|
|
finally:
|
|
clear_session_vars(tokens)
|
|
|
|
assert d.get("notify_on_complete") is True
|
|
assert not d.get("notify_unsupported")
|
|
assert len(process_registry.pending_watchers) == 1
|
|
assert process_registry.pending_watchers[0]["platform"] == "telegram"
|
|
|
|
def test_cli_stays_supported(self):
|
|
"""CLI delivers via the in-process completion_queue: notify stays on,
|
|
no false 'unsupported' note, and no pending_watcher (empty platform)."""
|
|
from tools.process_registry import process_registry
|
|
|
|
d = self._run_bg("sleep 30 && echo DONE")
|
|
assert d.get("notify_on_complete") is True
|
|
assert not d.get("notify_unsupported")
|
|
# No platform bound -> no gateway watcher, but completion_queue still fires.
|
|
assert len(process_registry.pending_watchers) == 0
|