fix(gateway): join delegations in the turn when interim messages are off

On a messaging chat with display.interim_assistant_messages off, a top-level
delegate_task still produced two messages: a "checks are running, I don't have
the findings yet" reply and, later, the real answer. Top-level model
delegations always run in the background (ea8a8b4af8), and the tool text tells
the model to "give a one-line status and END YOUR TURN" (c89f3b8800). That
status is the turn's final answer, not interim text, so no display setting
held it back.

New key delegation.gateway_mode: auto | inline | background (alias async).
auto (default) joins the children inside the turn on a messaging chat whose
resolved interim setting is off (global, display.platforms.<p> override or
platform default). The parent waits in the tool call, gets the results there
and sends one answer. API, local and webhook surfaces keep the background
path under auto. Chats with interim messages on are unchanged: same schema
bytes, same async dispatch.

How it works:
- The gateway resolves the mode from the turn's own profile-scoped config
  when it builds a session's agent (gateway/session_execution.py), and the
  agent's first tool build freezes it (like the one-shot pin).
- An inline session carries the joined delegate_task variant the one-shot
  path already uses: "results return in this tool call", no live-control
  parameters, no END YOUR TURN.
- Dispatch follows the delegate_task schema the session carries
  (session_joins_delegation), so schema and behaviour cannot disagree. The
  schema rides the session's tools[] pin: a config change applies from the
  next session (/new), as delegation.backend does, and a cache eviction or
  restart never flips a running chat.
- Inline children run on the existing synchronous path: attached to the
  parent, so /stop and an interrupting follow-up stop them; the per-child
  heartbeat keeps the parent's activity fresh for the inactivity watchdog;
  the async completion registry is not used. Nested and one-shot delegation
  are unchanged.

Refs #95522.

gateway/session_execution.py

@@ -221,9 +221,16 @@ class GatewaySessionAgentMixin:
         if found.evicted is not None:
             self._release_evicted_agent(found.evicted)
         if agent is None:
-            agent = self._build_fresh_agent(
-                turn_route, platform_key, combined_ephemeral, max_iterations, reasoning_config, pr, skip_context_files,
-            )
+            from tools.delegate_tool_config import gateway_delegation_inline, gateway_delegation_inline_scope
+            # delegation.gateway_mode from this turn's own (profile-scoped) config and resolved interim setting;
+            # read once, by the agent's first tool build, then carried by its delegate_task schema and pin.
+            inline = gateway_delegation_inline(
+                ctx.user_config, platform_key, interim_enabled=ctx.interim_assistant_messages_enabled)
+            with gateway_delegation_inline_scope(inline):
+                agent = self._build_fresh_agent(
+                    turn_route, platform_key, combined_ephemeral, max_iterations, reasoning_config, pr,
+                    skip_context_files,
+                )
             if cache_lock and cache is not None:
                 with cache_lock:
                     # Record the snapshot's session_id with message_count so the cross-process guard

hermes_cli/config_defaults.py

@@ -1355,8 +1355,7 @@ DEFAULT_CONFIG = {
         # children fail only from real errors (API, tools, iteration budget). A progressing child (including one
         # waiting on a multi-minute completion) restarts the window; a frozen one is caught.
         "child_timeout_seconds": 0,
-        # Subagent effort: "ultra" | "max" | "xhigh" | "high" | "medium" | "low" | "minimal" |
-        # "none" (empty = inherit)
+        # Subagent effort: "ultra" | "max" | "xhigh" | "high" | "medium" | "low" | "minimal" | "none" (empty = inherit)
         "reasoning_effort": "",
         # Max parallel children per batch AND max concurrent background delegation units; async
         # dispatches beyond it run synchronously. Floor 1, no ceiling.
@@ -1364,6 +1363,7 @@ DEFAULT_CONFIG = {
         # Background fan-outs return as ONE message when the whole call finishes. true = each task
         # (or `group`) returns on its own as it finishes — more new turns for the orchestrator.
         "independent_completions": False,
+        "gateway_mode": "auto",  # gateway: inline (join in the turn) | background | auto = inline where interim msgs are off
         # Orchestrator role controls. Depth floored at 1, no ceiling; each level multiplies cost.
         "max_spawn_depth": 1,  # 1 = flat, 2 = orchestrator→leaf, 3+ = deeper
         "orchestrator_enabled": True,  # kill switch for role="orchestrator"

run_agent.py

@@ -1343,9 +1343,10 @@ class AIAgent(
     def _dispatch_delegate_task(self, function_args: dict) -> str:
         """Single call site for delegate_task dispatch; new DELEGATE_TASK_SCHEMA fields are added only here."""
         from tools.delegate_tool import _strip_model_hidden_task_fields, delegate_task as _delegate_task
-        # Top-level MODEL delegations always run in the background (handle returned, results re-enter as
-        # messages). An ORCHESTRATOR SUBAGENT (depth > 0) stays synchronous — it needs results in-turn and
-        # owns no gateway session. The schema-level `background` param is intentionally ignored.
+        # Top-level MODEL delegations run in the background (handle returned, results re-enter as messages),
+        # unless the session's delegate_task is the joined variant (one-shot, delegation.gateway_mode inline).
+        # An ORCHESTRATOR SUBAGENT (depth > 0) stays synchronous — it needs results in-turn and owns no gateway
+        # session. The schema-level `background` param is intentionally ignored.
         return _delegate_task(
             goal=function_args.get("goal"), context=function_args.get("context"),
             tasks=_strip_model_hidden_task_fields(function_args.get("tasks")),

tests/gateway/test_gateway_delegation_mode.py

@@ -0,0 +1,318 @@
+"""Gateway delegation joins inside the turn when interim assistant messages are off.
+
+Top-level ``delegate_task`` on a messaging gateway runs in the background, and its tool text tells
+the model to "give a one-line status and END YOUR TURN". With ``display.interim_assistant_messages``
+off that status is the turn's FINAL answer, so the chat still receives a "checks are running, no
+findings yet" message and the real answer later, as a second message. ``delegation.gateway_mode``
+(default ``auto``) now joins the children inside the turn on such sessions: the model waits for them
+in the tool call and the chat gets one completed answer. Related: #95522, #141778.
+
+Real ``GatewayRunner._run_agent`` + real ``AIAgent`` + real child ``AIAgent`` against a loopback
+model. The model obeys whatever the ``delegate_task`` tool tells it: given a dispatch handle it ends
+its turn with a status, given the child's result it answers. The adapter is a recording transport
+driven by the base adapter's own ``_process_message_background`` (typing loop + final send).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import threading
+import time
+from pathlib import Path
+
+import pytest
+
+import hermes_yaml as yaml
+from gateway.config import Platform, PlatformConfig, StreamingConfig
+from gateway.platforms.base import BasePlatformAdapter, SendResult
+from gateway.platforms.event import MessageEvent, MessageType
+from gateway.session import SessionSource
+from tests.e2e.core.history._helpers import NO_BACKGROUND_REVIEW, OFFLINE_CONFIG
+from tests.fakes.fake_llm_provider import FakeLLMServer, Hang, Text, ToolCall, write_hermes_home
+from tests.gateway.test_run_progress_topics import _make_runner
+from tests.waiting import wait_until
+
+GOAL = "Audit the billing module for stale invoices"
+FINDING = "BILLING-OK-7731"
+PAUSE = "Checks are running in parallel, I don't have the findings yet."
+ANSWER = f"Done. The billing audit came back clean: {FINDING}"
+END_TURN = "END YOUR TURN"
+
+
+class _Chat(BasePlatformAdapter):
+    """Recording transport: every outbound message lands in ``sent``."""
+
+    SUPPORTS_MESSAGE_EDITING = True
+
+    def __init__(self, platform: Platform):
+        super().__init__(PlatformConfig(enabled=True, token="***"), platform)
+        self.sent: list[str] = []
+        self.typing = 0
+
+    async def connect(self, *, is_reconnect=False):
+        return True
+
+    async def disconnect(self):
+        return None
+
+    async def send(self, chat_id, content, reply_to=None, metadata=None):
+        self.sent.append(content)
+        return SendResult(success=True, message_id=f"m{len(self.sent)}")
+
+    async def edit_message(self, chat_id, message_id, content, **_):
+        self.sent.append(content)
+        return SendResult(success=True, message_id=message_id)
+
+    async def send_typing(self, chat_id, metadata=None):
+        self.typing += 1
+
+    async def get_chat_info(self, chat_id):
+        return {"id": chat_id}
+
+
+def _first_user_text(record: dict) -> str:
+    messages = record["body"].get("messages", [])
+    return str(next((m.get("content") for m in messages if m.get("role") == "user"), ""))
+
+
+def _model(child_reply):
+    """A model that follows the tool contract: a dispatch handle gets a status, a result gets the answer."""
+
+    def respond(request):
+        if GOAL in _first_user_text(request):  # the child's own conversation
+            return child_reply
+        last = request["body"]["messages"][-1]
+        if last.get("role") == "tool":
+            return Text(ANSWER if FINDING in str(last.get("content")) else PAUSE)
+        return ToolCall("delegate_task", {"tasks": [{"goal": GOAL, "context": "Repo at /srv/billing."}]})
+
+    def aux(request):
+        return child_reply if GOAL in _first_user_text(request) else Text("Fake summary.")
+
+    return respond, aux
+
+
+def _delegate_def(request: dict) -> dict:
+    return next(t["function"] for t in request["tools"] if t["function"]["name"] == "delegate_task")
+
+
+@pytest.fixture
+def dispatches(monkeypatch):
+    """Every hand-off to the async completion registry (the background path)."""
+    from tools import async_delegation
+
+    calls = []
+    real = async_delegation.dispatch_async_delegation_batch
+
+    def recording(**kwargs):
+        calls.append(kwargs)
+        return real(**kwargs)
+
+    monkeypatch.setattr(async_delegation, "dispatch_async_delegation_batch", recording)
+    return calls
+
+
+async def _turn(monkeypatch, config: dict, *, platform=Platform.WHATSAPP, model=None, during=None):
+    """One inbound message through the real gateway turn; returns (chat, the parent's model requests)."""
+    import gateway.run as gateway_run
+    from tools import async_delegation
+
+    home = Path(os.environ["HERMES_HOME"])
+    chat = _Chat(platform)
+    source = SessionSource(platform=platform, chat_id="c1", chat_type="dm", user_id="u1")
+    event = MessageEvent(text="How is billing doing?", source=source, message_type=MessageType.TEXT)
+    session_key = f"agent:main:{platform.value}:dm:c1"
+    respond, aux = model or _model(Text(f"Audit finished: {FINDING}"))
+    base = {
+        "tools": {"tool_search": {"enabled": "off"}},
+        "platform_toolsets": {platform.value: ["delegation", "todo"]},
+    }
+    # Tool progress lines are their own operator knob (display.tool_progress); a quiet chat turns them off.
+    merged = {**base, **config, "display": {"tool_progress": "off", **config.get("display", {})}}
+    with FakeLLMServer(respond, aux=aux) as srv:
+        write_hermes_home(home, srv.base_url, extra_config=OFFLINE_CONFIG + NO_BACKGROUND_REVIEW
+                          + yaml.safe_dump(merged))
+        monkeypatch.setattr(gateway_run, "_hermes_home", home)
+        monkeypatch.setattr(gateway_run, "_resolve_runtime_agent_kwargs", lambda: {
+            "api_key": "sk-fake-e2e", "base_url": srv.base_url, "provider": "custom", "api_mode": "chat_completions"})
+        runner = _make_runner(chat)
+        runner.config.streaming = StreamingConfig(enabled=False)
+
+        async def handler(ev):
+            result = await runner._run_agent(message=ev.text, context_prompt="", history=[], source=source,
+                                             session_id="s1", session_key=session_key)
+            return None if result.get("already_sent") else result["final_response"]
+
+        chat.set_message_handler(handler)
+        turn = asyncio.ensure_future(chat._process_message_background(event, session_key))
+        if during is not None:
+            await during()
+        await asyncio.wait_for(turn, timeout=60)
+        # A background unit must settle before its provider goes away.
+        await asyncio.to_thread(wait_until, lambda: async_delegation.active_count() == 0, 30, 0.02, "async units")
+    parent_requests = [r["body"] for r in srv.requests
+                       if r["kind"] == "main" and GOAL not in _first_user_text(r)]
+    return chat, parent_requests
+
+
+def _interim(value) -> dict:
+    return {"display": {"interim_assistant_messages": value}}
+
+
+@pytest.mark.asyncio
+async def test_interim_off_joins_the_children_and_the_chat_gets_one_completed_answer(monkeypatch, dispatches):
+    chat, requests = await _turn(monkeypatch, _interim(False))
+
+    assert chat.sent == [ANSWER]  # one message, written after the child's result
+    assert chat.typing  # the typing indicator is what the user sees while the children run
+    delegate = _delegate_def(requests[0])
+    assert END_TURN not in delegate["description"] and "BETWEEN your turns" not in delegate["description"]
+    assert "results return in this tool call" in delegate["description"]
+    assert not {"action", "subagent_id", "message"} & set(delegate["parameters"]["properties"])
+    tool_result = requests[-1]["messages"][-1]
+    assert tool_result["role"] == "tool" and FINDING in tool_result["content"]
+    assert END_TURN not in tool_result["content"] and '"dispatched"' not in tool_result["content"]
+    assert dispatches == []  # the async completion registry is never used
+
+
+@pytest.mark.asyncio
+async def test_interim_on_keeps_the_background_path_and_its_schema_byte_identical(monkeypatch, dispatches):
+    import model_tools
+
+    chat, requests = await _turn(monkeypatch, {})
+
+    assert chat.sent == [PAUSE]  # today's behaviour: the status is the turn's answer
+    assert len(dispatches) == 1
+    delegate = _delegate_def(requests[0])
+    assert END_TURN in delegate["description"]
+    registry_delegate = next(t["function"] for t in model_tools.get_tool_definitions(
+        enabled_toolsets=["delegation", "todo"], quiet_mode=True) if t["function"]["name"] == "delegate_task")
+    assert delegate == registry_delegate
+    assert '"dispatched"' in requests[-1]["messages"][-1]["content"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("config", "inline"), [
+    ({**_interim(True), "delegation": {"gateway_mode": "inline"}}, True),
+    ({**_interim(False), "delegation": {"gateway_mode": "background"}}, False),
+    ({**_interim(False), "delegation": {"gateway_mode": "async"}}, False),
+    ({**_interim(False), "delegation": {"gateway_mode": "auto"}}, True),
+], ids=["inline-overrides-interim-on", "background-overrides-interim-off", "async-alias", "explicit-auto"])
+async def test_an_explicit_gateway_mode_wins_over_the_interim_setting(monkeypatch, dispatches, config, inline):
+    chat, requests = await _turn(monkeypatch, config)
+
+    assert chat.sent == ([ANSWER] if inline else [PAUSE])
+    assert (END_TURN in _delegate_def(requests[0])["description"]) is not inline
+    assert len(dispatches) == (0 if inline else 1)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("platform", "inline"), [(Platform.WHATSAPP, True), (Platform.TELEGRAM, False)])
+async def test_a_platform_interim_override_decides_for_that_platform_only(monkeypatch, dispatches, platform, inline):
+    config = {"display": {"interim_assistant_messages": True,
+                          "platforms": {"whatsapp": {"interim_assistant_messages": False}}}}
+    chat, _ = await _turn(monkeypatch, config, platform=platform)
+
+    assert chat.sent == ([ANSWER] if inline else [PAUSE])
+    assert len(dispatches) == (0 if inline else 1)
+
+
+@pytest.mark.asyncio
+async def test_a_platform_override_can_turn_interim_back_on_under_a_global_off(monkeypatch, dispatches):
+    config = {"display": {"interim_assistant_messages": False,
+                          "platforms": {"whatsapp": {"interim_assistant_messages": True}}}}
+    chat, _ = await _turn(monkeypatch, config)
+
+    assert chat.sent == [PAUSE] and len(dispatches) == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("hard", [True, False], ids=["stop-command", "interrupting-follow-up"])
+async def test_stop_during_the_inline_wait_cancels_the_children(monkeypatch, dispatches, hard):
+    """``/stop`` hard-interrupts the running parent and a follow-up message under the default
+    ``busy_input_mode: interrupt`` soft-interrupts it; inline children are attached to the parent, so either
+    reaches the child's in-flight request and the turn ends without waiting for the provider."""
+    from tools import delegate_tool
+
+    parents, stops = [], []
+    real_build = delegate_tool._build_child_agent
+
+    def recording_build(*args, **kwargs):
+        child = real_build(*args, **kwargs)
+        parents.append(kwargs.get("parent_agent"))
+        for name in ("interrupt", "hard_interrupt"):  # instance attributes win the static lookup too
+            real = getattr(child, name)
+            setattr(child, name, lambda *a, _real=real, _name=name, **k: (stops.append(_name), _real(*a, **k))[1])
+        return child
+
+    monkeypatch.setattr(delegate_tool, "_build_child_agent", recording_build)
+    child_waiting = threading.Event()
+    respond, _ = _model(Text("unused"))
+
+    def routed(request):
+        if GOAL in _first_user_text(request):
+            if not child_waiting.is_set():  # the first child's request never gets an answer
+                child_waiting.set()
+                return Hang(seconds=120)
+            return Text(f"Audit finished: {FINDING}")
+        return respond(request)
+
+    async def stop_once_the_child_waits():
+        assert await asyncio.to_thread(child_waiting.wait, 30), "the child never reached its provider"
+        if hard:
+            parents[0].interrupt(None, hard_cancel=True)  # what /stop does to the running agent
+        else:
+            parents[0].interrupt("actually, check payroll instead")  # a follow-up in interrupt mode
+
+    started = time.monotonic()
+    chat, _ = await _turn(monkeypatch, _interim(False), model=(routed, lambda req: Text("Fake summary.")),
+                          during=stop_once_the_child_waits)
+
+    assert stops  # the stop reached the child through the parent's interrupt fan-out
+    assert time.monotonic() - started < 60  # the turn did not sit out the child's 120 s provider hang
+    if hard:
+        assert all(FINDING not in text for text in chat.sent)
+    else:  # the follow-up then runs as the next turn, which delegates afresh and answers once
+        assert chat.sent[-1] == ANSWER
+    assert dispatches == []
+
+
+def test_each_profile_resolves_its_own_mode_under_the_multiplexer(monkeypatch, tmp_path):
+    """A -> B -> A through ``_profile_scope_for_source`` (the seam every turn enters): the mode comes from the
+    routed profile's own config, never from whichever profile resolved last."""
+    from agent import secret_scope
+    from tools.delegate_tool_config import gateway_delegation_inline
+
+    root = tmp_path / "hermes"
+    beta = root / "profiles" / "beta"
+    beta.mkdir(parents=True)
+    (root / "config.yaml").write_text(yaml.safe_dump(_interim(False)))
+    (beta / "config.yaml").write_text(yaml.safe_dump(_interim(True)))
+    monkeypatch.setenv("HERMES_HOME", str(root))
+    monkeypatch.setattr("hermes_constants.get_default_hermes_root", lambda: root)
+    prev = secret_scope.is_multiplex_active()
+    secret_scope.set_multiplex_active(True)
+    try:
+        runner = _make_runner(_Chat(Platform.WHATSAPP))
+        runner.config.multiplex_profiles = True
+        source_a = SessionSource(platform=Platform.WHATSAPP, chat_id="D1", chat_type="dm", profile="default")
+        source_b = SessionSource(platform=Platform.WHATSAPP, chat_id="D2", chat_type="dm", profile="beta")
+        seen = []
+        for source in (source_a, source_b, source_a):
+            with runner._profile_scope_for_source(source):  # the inputs a fresh agent's build resolves from
+                disp = runner._run_agent_display_settings(source)
+                seen.append(gateway_delegation_inline(
+                    disp.user_config, disp.platform_key, interim_enabled=disp.interim_assistant_messages_enabled))
+    finally:
+        secret_scope.set_multiplex_active(prev)
+    assert seen == [True, False, True]
+
+
+@pytest.mark.parametrize("platform_key", ["api_server", "local", "webhook"])
+def test_auto_never_joins_on_a_surface_that_is_not_a_chat(platform_key):
+    """``auto`` is about a pause message reaching a person: API, local and webhook turns keep today's path."""
+    from tools.delegate_tool_config import gateway_delegation_inline
+
+    assert gateway_delegation_inline(_interim(False), platform_key, interim_enabled=False) is False
+    assert gateway_delegation_inline({"delegation": {"gateway_mode": "inline"}}, platform_key, interim_enabled=True)

tests/tools/test_delegate_gateway_inline.py

@@ -0,0 +1,165 @@
+"""A session's ``delegate_task`` schema decides whether its children join in the call.
+
+The gateway builds a session's agent inside ``gateway_delegation_inline_scope`` when that session resolved
+``delegation.gateway_mode`` to inline (``auto`` + interim assistant messages off, or ``inline``). The tool build
+then carries the joined ``delegate_task`` variant (results return in the call, no live control), and dispatch
+follows the variant the model actually sees, so schema and behaviour cannot disagree. The variant rides the
+session's tools[] pin like every other schema byte: a config edit applies from the next session (``/new``),
+the rule ``delegation.backend`` already follows, and a cache eviction or restart never flips a live chat.
+"""
+
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+from gateway.session_context import set_session_vars
+from tools.process_registry import process_registry
+
+LIVE_CONTROL = {"action", "subagent_id", "message"}
+
+
+@pytest.fixture(autouse=True)
+def _clean_queue_and_context(monkeypatch):
+    import gateway.session_context as sc
+
+    monkeypatch.delenv("HERMES_SINGLE_QUERY_SESSION", raising=False)
+    yield
+    for var in sc._VAR_MAP.values():
+        var.set(sc._UNSET)
+    sc._SESSION_ASYNC_DELIVERY.set(sc._UNSET)
+    sc._SESSION_HISTORY_DELIVERY.set(sc._UNSET)
+    while not process_registry.completion_queue.empty():
+        process_registry.completion_queue.get_nowait()
+
+
+def _built(inline: bool):
+    from agent.agent_init import _load_tools
+    from tools.delegate_tool_config import gateway_delegation_inline_scope
+
+    agent = SimpleNamespace(platform="whatsapp", quiet_mode=True)
+    with gateway_delegation_inline_scope(inline):
+        _load_tools(agent, ["hermes-cli"], None)
+    return agent
+
+
+def _delegate(agent) -> dict:
+    return next(t["function"] for t in agent.tools if t["function"]["name"] == "delegate_task")
+
+
+def test_the_build_scope_picks_the_schema_and_the_agent_keeps_it():
+    import model_tools
+    from tools.delegate_tool import oneshot_delegate_function
+    from tools.mcp_tool_agent import refresh_agent_mcp_tools
+
+    background = _built(inline=False)
+    registry = next(t["function"] for t in model_tools.get_tool_definitions(
+        enabled_toolsets=["hermes-cli"], disabled_toolsets=background.disabled_toolsets, quiet_mode=True)
+        if t["function"]["name"] == "delegate_task")
+    assert _delegate(background) == registry  # byte-identical to main
+
+    inline = _built(inline=True)
+    assert _delegate(inline) == oneshot_delegate_function(registry)
+    assert "END YOUR TURN" not in _delegate(inline)["description"]
+    assert not LIVE_CONTROL & set(_delegate(inline)["parameters"]["properties"])
+    assert "clarify" in {t["function"]["name"] for t in inline.tools}  # nothing else of the one-shot footprint
+
+    # A later rebuild outside any gateway scope (explicit reload, compaction) keeps the session's variant.
+    refresh_agent_mcp_tools(inline, quiet_mode=True, content_aware=True)
+    assert _delegate(inline) == oneshot_delegate_function(registry)
+
+
+def test_a_session_keeps_its_mode_until_a_new_session():
+    """Operator flips interim back on mid-session, then the cached agent is evicted: the fresh agent is built
+    for background, but the session pin restores the joined schema the model has been seeing, and dispatch
+    follows the pin. A new session (no pin) takes the new setting."""
+    from tools.delegate_tool import session_joins_delegation
+    from tools.mcp_tool_agent import restore_agent_tool_prefix, tool_pin_version
+
+    first = _built(inline=True)
+    fresh = _built(inline=False)
+    restore_agent_tool_prefix(fresh, {"version": tool_pin_version(), "tools": first.tools})
+    assert _delegate(fresh) == _delegate(first)
+    assert session_joins_delegation(fresh) is True
+    assert session_joins_delegation(_built(inline=False)) is False
+
+    # And the other way round: a background session stays background after interim is turned off.
+    background = _built(inline=False)
+    rebuilt = _built(inline=True)
+    restore_agent_tool_prefix(rebuilt, {"version": tool_pin_version(), "tools": background.tools})
+    assert session_joins_delegation(rebuilt) is False
+
+
+def _parent(tools, depth=0):
+    parent = MagicMock()
+    parent._delegate_depth = depth
+    parent.session_id = "sess"
+    parent._interrupt_requested = False
+    parent._active_children = []
+    parent._active_children_lock = None
+    parent._oneshot_session = False
+    parent.tools = tools
+    return parent
+
+
+@pytest.fixture
+def fast_children(monkeypatch):
+    import tools.delegate_tool as dt
+
+    def build_child(**kw):
+        child = MagicMock()
+        child._delegate_role, child._subagent_id = "leaf", "s1"
+        return child
+
+    def run_child(task_index, goal, child=None, parent_agent=None, **kw):
+        return {"task_index": task_index, "status": "completed", "summary": f"done: {goal}",
+                "api_calls": 1, "duration_seconds": 0.1, "model": "m", "exit_reason": "completed"}
+
+    creds = dict.fromkeys(("provider", "base_url", "api_key", "api_mode", "command", "args"), None)
+    monkeypatch.setattr(dt, "_build_child_agent", build_child)
+    monkeypatch.setattr(dt, "_run_single_child", run_child)
+    monkeypatch.setattr(dt, "_resolve_delegation_credentials", lambda *a, **k: {**creds, "model": "m"})
+    set_session_vars(platform="whatsapp", chat_id="c1", session_key="agent:main:whatsapp:dm:c1", session_id="sess")
+    return dt
+
+
+def test_dispatch_follows_the_schema_the_session_carries(fast_children, monkeypatch):
+    from tools import async_delegation
+
+    dispatched = []
+    monkeypatch.setattr(async_delegation, "dispatch_async_delegation_batch",
+                        lambda **kw: dispatched.append(kw) or {"status": "dispatched", "delegation_id": "d1"})
+
+    joined = json.loads(fast_children.delegate_task(
+        goal="audit", background=True, parent_agent=_parent(_built(inline=True).tools)))
+    assert joined["results"][0]["summary"] == "done: audit"
+    assert "note" not in joined and dispatched == []  # no "end your turn", no async registry
+
+    detached = json.loads(fast_children.delegate_task(
+        goal="audit", background=True, parent_agent=_parent(_built(inline=False).tools)))
+    assert detached["status"] == "dispatched" and "END YOUR TURN" in detached["note"]
+    assert len(dispatched) == 1
+
+
+def test_nested_and_one_shot_delegation_keep_their_own_paths(fast_children, monkeypatch):
+    # An orchestrator child (depth > 0) was always synchronous; a one-shot session keeps its explanatory note.
+    monkeypatch.setattr(fast_children, "_get_max_spawn_depth", lambda: 2)
+    nested = json.loads(fast_children.delegate_task(
+        goal="audit", background=False, parent_agent=_parent(_built(inline=False).tools, depth=1)))
+    assert nested["results"][0]["status"] == "completed" and "note" not in nested
+
+    oneshot = _parent(_built(inline=True).tools)
+    oneshot._oneshot_session = True
+    result = json.loads(fast_children.delegate_task(goal="audit", background=True, parent_agent=oneshot))
+    assert "SYNCHRONOUSLY" in result["note"]
+
+
+def test_live_control_degrades_to_plain_answers_in_an_inline_session(fast_children):
+    parent = _parent(_built(inline=True).tools)
+    listed = json.loads(fast_children.delegate_task(action="list", parent_agent=parent))
+    assert isinstance(listed, dict)
+    stopped = json.loads(fast_children.delegate_task(action="stop", subagent_id="nope", parent_agent=parent))
+    assert "error" in stopped

tools/AGENTS.md

@@ -152,7 +152,7 @@ and the stall clock arms when the runner starts, so a queued unit is never judge
 no `delegate_task`, `clarify`, `memory`, `send_message`, `cronjob`; keeps `execute_code`) and
 `orchestrator` (keeps `delegate_task`; gated by `delegation.orchestrator_enabled`, bounded by
 `delegation.max_spawn_depth`, default 2). Config knobs under `delegation:`:
-`max_concurrent_children, independent_completions, max_spawn_depth, child_timeout_seconds, orchestrator_enabled,
+`max_concurrent_children, independent_completions, gateway_mode, max_spawn_depth, child_timeout_seconds, orchestrator_enabled,
 subagent_auto_approve, inherit_mcp_toolsets, max_iterations, backend, backends`. **Routes:** `delegation.backend` picks a
 delegation provider (`agent/delegation_provider.py`, `ctx.register_delegation_provider`; its settings live in
 `delegation.backends.<name>`) that answers each task's route (`tools/delegate_tool_routes.py`: model/provider fields

tools/delegate_tool.py

@@ -872,6 +872,20 @@ def oneshot_delegate_function(function: dict) -> dict:
     return {**function, "description": description, "parameters": {**params, "properties": properties}}
 
 
+def session_joins_delegation(agent: Any) -> bool:
+    """Whether *agent*'s top-level delegations join in the call because the ``delegate_task`` the model sees is the
+    joined variant (a gateway session with ``delegation.gateway_mode`` inline). Dispatch follows the schema bytes,
+    which ride the session's tools[] pin: a config change can never leave the model reading "results return in
+    this tool call" while children detach, or the reverse. One-shot sessions keep their own fallback (and note)."""
+    if getattr(agent, "_oneshot_session", None) is True:
+        return False
+    for tool in getattr(agent, "tools", None) or ():
+        function = tool.get("function") if isinstance(tool, dict) else None
+        if isinstance(function, dict) and function.get("name") == "delegate_task":
+            return _ONESHOT_DESCRIPTION_DELIVERY in (function.get("description") or "")
+    return False
+
+
 # --- Registry ---
 from tools.registry import registry, tool_error
 

tools/delegate_tool_config.py

@@ -4,7 +4,9 @@ from __future__ import annotations
 
 import logging
 import os
-from typing import Any, Dict, List, NamedTuple, Optional
+from contextlib import contextmanager
+from contextvars import ContextVar
+from typing import Any, Dict, Iterator, List, NamedTuple, Optional
 from utils import base_url_hostname, is_truthy_value
 from hermes_cli.fallback_config import scoped_fallback_chain
 
@@ -112,6 +114,46 @@ def _get_independent_completions() -> bool:
     completion messages that land as each finishes. Off = one consolidated message when the whole call is done."""
     return is_truthy_value(_cfg().get("independent_completions", False))
 
+# delegation.gateway_mode: "background" (alias "async") always detaches; "inline" joins in the turn; anything else is
+# "auto". Read from the gateway turn's own (profile-scoped) user config, not from this process's config.
+_GATEWAY_MODES = {"inline": True, "background": False, "async": False}
+
+def gateway_delegation_inline(user_config: Any, platform_key: str, *, interim_enabled: bool) -> bool:
+    """Whether a gateway session joins its top-level delegations inside the turn (``delegation.gateway_mode``).
+
+    ``auto`` (default) joins on a human chat whose interim assistant messages are off (the gateway's resolved
+    ``display.interim_assistant_messages``, platform overrides included): there the "status, end your turn" reply
+    the background path asks for would be delivered as the turn's answer, then the real answer as a second message.
+    API, local and webhook surfaces keep the background path under ``auto``."""
+    from gateway.session_context import NON_MESSAGING_SESSION_SURFACES
+    section = user_config.get("delegation") if isinstance(user_config, dict) else None
+    mode = str(section.get("gateway_mode") or "auto").strip().lower() if isinstance(section, dict) else "auto"
+    if mode in _GATEWAY_MODES:
+        return _GATEWAY_MODES[mode]
+    return not interim_enabled and str(platform_key or "").lower() not in NON_MESSAGING_SESSION_SURFACES
+
+# Set by the gateway around building a session's agent; read once by that agent's first tool build.
+_GATEWAY_DELEGATION_INLINE: ContextVar[bool] = ContextVar("hermes_gateway_delegation_inline", default=False)
+
+@contextmanager
+def gateway_delegation_inline_scope(inline: bool) -> Iterator[None]:
+    """Build an agent whose session joins its delegations in the call (see :func:`agent_joins_delegation`)."""
+    token = _GATEWAY_DELEGATION_INLINE.set(bool(inline))
+    try:
+        yield
+    finally:
+        _GATEWAY_DELEGATION_INLINE.reset(token)
+
+def agent_joins_delegation(agent: Any) -> bool:
+    """The gateway's inline decision, frozen on *agent* at its first tool build (like ``agent_is_oneshot``), so
+    later rebuilds outside the build scope keep the same ``delegate_task`` bytes. Dispatch does not read this: it
+    follows the schema the session carries (``tools.delegate_tool.session_joins_delegation``)."""
+    cached = getattr(agent, "_delegation_inline", None)
+    if not isinstance(cached, bool):
+        cached = _GATEWAY_DELEGATION_INLINE.get()
+        agent._delegation_inline = cached
+    return cached
+
 def _get_worktree_isolation() -> bool:
     """delegation.worktree_isolation (bool, default False): each child gets its own
     git worktree off the parent's HEAD so parallel children never contend for one

tools/delegate_tool_dispatch.py

@@ -486,10 +486,11 @@ def _dispatch_background(batch: _Batch) -> str:
 
 def _run_batch(batch: _Batch, background: bool) -> str:
     """Tool result JSON: a dispatch handle (background) or the joined combined results."""
+    from tools.delegate_tool import session_joins_delegation
     # Every unit of this call shares task_list, so its last joined unit emits the call's one row.
     begin_delegation_run(
         batch.task_list, subagents=len(batch.children), depth=getattr(batch.parent_agent, "_delegate_depth", 0) + 1,
     )
-    if background:
+    if background and not session_joins_delegation(batch.parent_agent):
         return _dispatch_background(batch)
     return json.dumps(_execute_and_aggregate(batch), ensure_ascii=False)

toolsets.py

@@ -474,13 +474,15 @@ def agent_tool_drops(agent: Any) -> frozenset:
 
 
 def agent_tool_surface(agent: Any, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    """*tools* as *agent* carries them: :func:`agent_tool_drops` removed and, in a one-shot session, the
-    one-shot schema variants (``agent.oneshot_footprint.oneshot_tool_schemas``). Applied to FRESH definitions (load,
-    MCP refresh) before any merge: a pinned or carried entry keeps the bytes already sent."""
+    """*tools* as *agent* carries them: :func:`agent_tool_drops` removed and, in a one-shot session or a gateway
+    session that joins its delegations in the turn (``delegation.gateway_mode``), the joined ``delegate_task``
+    variant (``agent.oneshot_footprint.oneshot_tool_schemas``). Applied to FRESH definitions (load, MCP refresh)
+    before any merge: a pinned or carried entry keeps the bytes already sent."""
     from agent.oneshot_footprint import agent_is_oneshot, oneshot_tool_schemas
+    from tools.delegate_tool_config import agent_joins_delegation
     drops = agent_tool_drops(agent)
     tools = [tool for tool in tools if (tool.get("function") or {}).get("name") not in drops] if drops else list(tools)
-    return oneshot_tool_schemas(tools) if agent_is_oneshot(agent) else tools
+    return oneshot_tool_schemas(tools) if agent_is_oneshot(agent) or agent_joins_delegation(agent) else tools
 
 
 def validate_toolset(name: str) -> bool:

website/docs/user-guide/configuration.md

@@ -3089,6 +3089,7 @@ delegation:
   max_spawn_depth: 1                        # Delegation tree depth cap (1-3, clamped). 1 = flat (default): parent spawns leaves that cannot delegate. 2 = orchestrator children can spawn leaf grandchildren. 3 = three levels.
   orchestrator_enabled: true                # Global kill switch. When false, role="orchestrator" is ignored and every child is forced to leaf regardless of max_spawn_depth.
   oneshot_max_children: 2                   # Total subagents a one-shot run (hermes chat -q / --oneshot) may spawn; 0 = unlimited. Interactive and gateway sessions are never capped by this.
+  gateway_mode: auto                        # Gateway sessions: auto | inline | background. auto = join children inside the turn (one answer) on a messaging chat whose display.interim_assistant_messages is off. See Subagent Delegation → Messaging chats without interim messages.
 ```
 
 **Delegation strategy (`backend`):** `delegation.backend` picks the strategy plugin that decides each subagent's route. `default` follows the keys below (or the parent's route when they are empty). Plugin strategies such as `model_map` (a named model per task) and `profile_routes` (a Hermes profile per task) read their settings from `delegation.backends.<backend name>`. A backend whose plugin is missing or disabled fails closed: `delegate_task` returns a tool error naming it and the fix (`/delegation default`, or `hermes config set delegation.backend default`) instead of running children on another route. A backend route without its own `fallback_providers` uses `delegation.fallback_providers`. See [Subagent Delegation → Delegation strategies](features/delegation.md#delegation-strategies-delegationbackend).

website/docs/user-guide/features/delegation.md

@@ -606,7 +606,8 @@ parent in the same turn. Batch children still run in parallel, subject to
 final response before the CLI exits.
 
 - **Automatic joining:** no opt-in or background-mode override is needed.
-  Interactive TTY chat and messaging sessions keep background delegation.
+  Interactive TTY chat keeps background delegation; messaging sessions follow
+  `delegation.gateway_mode` (see below).
 - **Per-input policy:** the CLI records finite consumption with the admitted prompt,
   including when resuming an interactive session. It does not change that session's
   creation policy, sibling viewers, or the daemon environment.
@@ -623,6 +624,40 @@ Closing the CLI detaches rather than terminating its gateway-owned turn. Child
 execution is still process-local to that owner: use a durable scheduler for work
 that must survive an owner restart.
 
+## Messaging chats without interim messages
+
+On a messaging platform, a background delegation tells the parent to send a one-line
+status and end its turn. When `display.interim_assistant_messages` is off for that
+platform, that status is the turn's final answer, so the chat would receive a "working
+on it" message and then the real answer as a second message. `delegation.gateway_mode`
+controls this:
+
+```yaml
+delegation:
+  gateway_mode: auto   # auto (default) | inline | background
+```
+
+- **`auto`:** a messaging chat whose resolved `display.interim_assistant_messages` is off
+  (global value, `display.platforms.<platform>` override or the platform's default) joins
+  its children inside the turn. The parent waits in the `delegate_task` call, gets the
+  results there, and sends one completed answer. Chats with interim messages on, the API
+  server, local sessions and webhooks keep background delegation.
+- **`inline`:** every gateway session joins its children inside the turn.
+- **`background`** (alias `async`): every gateway session keeps background delegation.
+
+An inline session's `delegate_task` says results return in the call and leaves out the
+live-control parameters (`action` list/steer/stop, `subagent_id`, `message`), the same
+schema as a one-shot run. While it waits, the typing indicator stays on and the turn
+behaves like any long tool call. `/stop` stops the children, and so does a follow-up
+message under the default `busy_input_mode: interrupt` (each child returns an interrupted
+result with its partial output); under `queue` the follow-up waits for the turn.
+`display.long_running_notifications` still governs the periodic "still working" notice.
+Nested delegation is unchanged.
+
+The mode is part of a session's tool schema, which is cached with the conversation. A
+change to `gateway_mode` or to `interim_assistant_messages` applies from the next session
+(`/new`); a running chat keeps the mode it started with, also across gateway restarts.
+
 ## Lifetime and Durability
 
 :::warning Background completion durability is not durable execution
@@ -709,6 +744,7 @@ delegation:
   max_iterations: 250                       # Max turns per child (default: 250)
   # max_concurrent_children: 10             # Parallel children per batch (default: 10)
   # independent_completions: false          # true = each task/group returns as it finishes (default: one message per call)
+  # gateway_mode: auto                      # auto | inline | background (see Messaging chats without interim messages)
   # worktree_isolation: false               # Give each child its own git worktree (see Worktree Isolation above)
   # max_spawn_depth: 1                      # Tree depth (floor 1, no ceiling, default 1 = flat). Raise to 2 to allow orchestrator children to spawn leaves; 3+ for deeper trees.
   # orchestrator_enabled: true              # Disable to force all children to leaf role.