mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-25 17:18:11 +00:00
dc3e4e8428
Dispatcher-spawned Kanban workers are finite one-shot processes, so detached delegation completions can outlive their only consumer. Mark that runtime as unable to deliver async completions and reuse the synchronous delegation fallback, returning required child results before the worker exits.\n\nAlso make unsupported-session notes runtime-generic and cover the delayed-child lifecycle regression.\n\nRefs #63169
323 lines
12 KiB
Python
323 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_dispatcher_spawned_kanban_worker_is_unsupported(self, monkeypatch):
|
|
"""A one-shot Kanban worker cannot receive a detached completion
|
|
after its process exits, even when its CLI session otherwise defaults
|
|
to supporting async delivery."""
|
|
monkeypatch.setenv("HERMES_KANBAN_TASK", "t_review")
|
|
|
|
assert async_delivery_supported() is False
|
|
|
|
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
|