diff --git a/cron/scheduler.py b/cron/scheduler.py index 825558980f7d..26b7eb517125 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -2948,6 +2948,20 @@ def run_job( platform="", chat_id="", chat_name="", + # A cron job cannot receive a completion after its turn ends. We clear the + # HERMES_SESSION_* routing keys just below, so an async delegation's + # completion event carries session_key="" — _enrich_async_delegation_routing + # cannot resolve it and _inject_watch_notification drops it ("no routing + # metadata"). And by the time a child finishes, 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, which risks routing a cron subagent's output + # into an unrelated user chat.) + # + # Declaring the channel stateless routes delegate_task to its existing + # inline/synchronous path, so results return within the job's own turn. + # See declare_stateless_channel(). Upstream: #53027, #63142. + async_delivery=False, ) _cron_delivery_vars = ( "HERMES_CRON_AUTO_DELIVER_PLATFORM", diff --git a/gateway/session_context.py b/gateway/session_context.py index 2214c2a56883..a8e7c027fecb 100644 --- a/gateway/session_context.py +++ b/gateway/session_context.py @@ -327,13 +327,40 @@ def get_session_env(name: str, default: str = "") -> str: return os.getenv(name, default) +def declare_stateless_channel() -> None: + """Declare that this session cannot receive an async background completion. + + Binds only the delivery capability, leaving every other session var unset. + Use this instead of ``set_session_vars(async_delivery=False)`` on a pure + single-process runner: ``set_session_vars`` also latches + ``_session_context_engaged`` (see above), which switches the subprocess + env bridge from "os.environ fallback" to "ContextVar-authoritative, strip on + _UNSET" in ``tools/environments/local.py``. A one-shot CLI that never engages + the session-context system must not flip that latch as a side effect of + declaring a capability. + + Callers that already build a full session context (cron's ``run_job``) get + the same state by passing ``async_delivery=False`` to ``set_session_vars``. + + A session that cannot take a late completion makes ``delegate_task`` fall + through to its existing inline/synchronous path, so subagent results are + returned within the turn instead of being dispatched to a channel that will + never deliver them. + + See NousResearch/hermes-agent#53027 and #63142. + """ + _SESSION_ASYNC_DELIVERY.set(False) + + def async_delivery_supported() -> bool: """Whether the current session can deliver a background completion later. - Returns ``False`` only when the active session was explicitly bound by a - stateless adapter (the API server) that cannot route a notification back to - the agent after the turn ends. CLI, cron, and the real gateway platforms — - and any path that never bound the contextvar — return ``True``. + Returns ``False`` when the active session was bound by a stateless channel: + an adapter that cannot route a notification back after the turn ends (the + API server), or a one-shot runner that exits after its final response + (``hermes -z``, cron — see :func:`declare_stateless_channel`). The real + gateway platforms, the interactive CLI, and any path that never bound the + contextvar return ``True``. Tools that promise async delivery (``terminal`` notify_on_complete / watch_patterns, ``delegate_task`` background=True) consult this before diff --git a/hermes_cli/oneshot.py b/hermes_cli/oneshot.py index e7db628a7c1d..0ed1dcd303a1 100644 --- a/hermes_cli/oneshot.py +++ b/hermes_cli/oneshot.py @@ -28,6 +28,7 @@ from contextlib import redirect_stderr, redirect_stdout from pathlib import Path from typing import Optional +from gateway.session_context import declare_stateless_channel from hermes_cli.fallback_config import get_fallback_chain @@ -220,6 +221,15 @@ def run_oneshot( os.environ["HERMES_YOLO_MODE"] = "1" os.environ["HERMES_ACCEPT_HOOKS"] = "1" + # One-shot prints a single final response and exits: there is no later turn + # for a detached subagent's completion to re-enter, and nothing here drains + # process_registry.completion_queue (only cli.py's interactive process_loop + # and the gateway watchers do). Left unbound, async_delivery_supported() + # defaults True, delegate_task is forced background, and every subagent + # result is discarded. Declaring the channel stateless routes delegate_task + # to its inline/synchronous path. See declare_stateless_channel(). + declare_stateless_channel() + # Redirect stderr AND stdout to devnull for the entire call tree. # We'll print the final response to the real stdout at the end. real_stdout = sys.stdout diff --git a/tests/gateway/test_async_delivery_capability.py b/tests/gateway/test_async_delivery_capability.py index 084d4dbdf32c..4f63f2ed408b 100644 --- a/tests/gateway/test_async_delivery_capability.py +++ b/tests/gateway/test_async_delivery_capability.py @@ -25,6 +25,7 @@ from gateway.session_context import ( async_delivery_supported, clear_session_vars, get_session_env, + reset_session_vars, set_session_vars, ) @@ -86,6 +87,109 @@ class TestAsyncDeliverySupported: 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 # --------------------------------------------------------------------------- diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index c6ed64febad9..632fa762db07 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -2832,10 +2832,11 @@ def delegate_task( _sync_result = _execute_and_aggregate() if isinstance(_sync_result, dict): _sync_result["note"] = ( - "background=true is not available on this endpoint (stateless " - "HTTP API — no channel to deliver a detached subagent result " - "after the turn ends), so the subagent(s) ran SYNCHRONOUSLY and " - "the result is included above." + "background=true is not available in this session — it cannot " + "receive a detached subagent result after the turn ends (a " + "one-shot runner such as `hermes -z` or a cron job, or a " + "stateless HTTP endpoint). The subagent(s) ran SYNCHRONOUSLY " + "and the result is included above." ) return json.dumps(_sync_result, ensure_ascii=False) diff --git a/tools/terminal_tool.py b/tools/terminal_tool.py index 9f7ed245cc57..57422299d0de 100644 --- a/tools/terminal_tool.py +++ b/tools/terminal_tool.py @@ -2623,9 +2623,10 @@ def terminal_tool( watch_patterns = None result_data["notify_on_complete"] = False result_data["notify_unsupported"] = ( - "notify_on_complete / watch_patterns are not available on " - "this endpoint (stateless HTTP API — no channel to deliver " - "an async completion after the turn ends). The process is " + "notify_on_complete / watch_patterns are not available in " + "this session — it cannot receive an async completion after " + "the turn ends (a one-shot runner such as `hermes -z` or a " + "cron job, or a stateless HTTP endpoint). The process is " "running in the background; retrieve its result with " "process(action='poll') or process(action='wait')." )