From f4452169d1b01150317543dc1b8ca08d009aa15f Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Wed, 16 Sep 2026 23:48:04 -0700 Subject: [PATCH] fix(gateway): busy-mode steer also reaches the parent's active subagents MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A parent blocked inside delegate_task only drains its steer queue after the tool returns, i.e. after the child finishes. On Telegram (busy_input_mode=steer, the /steer command, and the PRIORITY path) the gateway queued the text on that parent alone and acked "Steered into current run", so a looping delegated child never saw it (#112095). Fan the text out to running_agent._active_children — the same identity-scoped snapshot interrupt() uses — and make the ack say where it went. --- gateway/run_busy.py | 46 ++++++++++++- gateway/run_inbound.py | 2 +- .../test_busy_steer_subagent_fanout.py | 69 +++++++++++++++++++ 3 files changed, 113 insertions(+), 4 deletions(-) create mode 100644 tests/gateway/test_busy_steer_subagent_fanout.py diff --git a/gateway/run_busy.py b/gateway/run_busy.py index 2688b957e3..0753e76591 100644 --- a/gateway/run_busy.py +++ b/gateway/run_busy.py @@ -221,6 +221,40 @@ class GatewayBusySessionMixin: except Exception: return False + @staticmethod + def _steer_active_subagents(running_agent: Any, text: str) -> int: + """Queue *text* into every live child of *running_agent*; returns how many accepted it. + + A parent blocked inside ``delegate_task`` only drains its own steer queue after the tool + returns, i.e. after the child finishes — so a steer aimed at a looping child would sit + unread for the whole delegation (#112095, Telegram). Children are the parent's own + ``_active_children`` (identity-scoped, same snapshot ``interrupt()`` fans out to).""" + children = getattr(running_agent, "_active_children", None) + if not isinstance(children, (list, tuple, set)) or not children: + return 0 + lock = getattr(running_agent, "_active_children_lock", None) + try: + with lock if lock is not None else contextlib.nullcontext(): + snapshot = list(children) + except Exception: + return 0 + accepted = 0 + for child in snapshot: + steer = getattr(child, "steer", None) + if not callable(steer): + continue + try: + accepted += bool(steer(text)) + except Exception as exc: + logger.warning("Steer into subagent %r failed: %s", getattr(child, "_delegate_id", child), exc) + return accepted + + def _steer_running_agent(self, running_agent: Any, text: str) -> bool: + """``running_agent.steer(text)`` plus fan-out to its active subagents (see + :meth:`_steer_active_subagents`); True when the parent or any child queued it.""" + accepted = bool(running_agent.steer(text)) + return bool(self._steer_active_subagents(running_agent, text)) or accepted + async def _session_has_compression_in_flight(self, session_key: str) -> bool: """True when a compression lock is held for this session's id (callers demote interrupt → queue, else a follow-up against the pre-rotation parent orphans compression siblings). @@ -535,6 +569,8 @@ class GatewayBusySessionMixin: """Call ``running_agent.(text)`` (steer/redirect); False + warning on failure.""" try: call_text = self._steer_text_with_origin(text, event) if event else text + if verb == "steer": + return self._steer_running_agent(running_agent, call_text) return bool(getattr(running_agent, verb)(call_text)) except Exception as exc: logger.warning("Gateway %s failed for session %s: %s", verb, session_key, exc) @@ -615,7 +651,10 @@ class GatewayBusySessionMixin: except Exception: pass status_detail = f" ({', '.join(status_parts)})" if status_parts else "" - if is_steer_mode: + if is_steer_mode and self._agent_has_active_subagents(running_agent): + head = "⏩ Steered into current run and its active subagent(s)" + tail = ". Your message arrives after their next tool call." + elif is_steer_mode: head, tail = "⏩ Steered into current run", ". Your message arrives after the next tool call." elif is_redirect_mode: head, tail = "↪ Redirected current run", ". I'll adjust using your correction." @@ -939,14 +978,15 @@ class GatewayBusySessionMixin: if not running_agent or not hasattr(running_agent, "steer"): return _queue_fallback("No active agent — /steer queued for the next turn.") try: - accepted = running_agent.steer(self._steer_text_with_origin(steer_text, event)) + accepted = self._steer_running_agent(running_agent, self._steer_text_with_origin(steer_text, event)) except Exception as exc: logger.warning("Steer failed for session %s: %s", quick_key, exc) return f"⚠️ Steer failed: {exc}" if not accepted: return "Steer rejected (empty payload)." preview = steer_text[:60] + ("..." if len(steer_text) > 60 else "") - return f"⏩ Steer queued — arrives after the next tool call: '{preview}'" + target = "run and its active subagent(s)" if self._agent_has_active_subagents(running_agent) else "run" + return f"⏩ Steer queued into current {target} — arrives after the next tool call: '{preview}'" async def _busy_goal_command(self, event: MessageEvent, quick_key: str, source): # Control verbs are safe mid-run (state only); setting new goal text is rejected so we don't diff --git a/gateway/run_inbound.py b/gateway/run_inbound.py index a43dcaa8c4..1c65aab39a 100644 --- a/gateway/run_inbound.py +++ b/gateway/run_inbound.py @@ -625,7 +625,7 @@ class GatewayInboundMixin: steered = False if self._hm_text_only(event) and steer_text and hasattr(running_agent, "steer"): try: - steered = bool(running_agent.steer(self._steer_text_with_origin(steer_text, event))) + steered = self._steer_running_agent(running_agent, self._steer_text_with_origin(steer_text, event)) except Exception as exc: logger.warning("PRIORITY steer failed for session %s: %s", _quick_key, exc) if steered: diff --git a/tests/gateway/test_busy_steer_subagent_fanout.py b/tests/gateway/test_busy_steer_subagent_fanout.py new file mode 100644 index 0000000000..127ecf4559 --- /dev/null +++ b/tests/gateway/test_busy_steer_subagent_fanout.py @@ -0,0 +1,69 @@ +"""Busy-mode steer reaches the subagent the parent is blocked on (#112095, Telegram surface). + +A parent inside ``delegate_task`` drains its own steer queue only after the tool returns — i.e. +after the child finishes — so a steer that lands on the parent alone is never read by a looping +child. Every gateway steer entry point must fan the text out to ``_active_children`` and the ack +must say so. +""" + +import threading + +import pytest + +from gateway.config import GatewayConfig, Platform +from gateway.platforms.event import MessageEvent, MessageType +from gateway.run import GatewayRunner +from gateway.session import SessionSource + + +class _Agent: + def __init__(self, children=()): + self.payload = None + self._active_children = list(children) + self._active_children_lock = threading.Lock() + + def steer(self, text): + self.payload = text + return True + + +def _event(text="focus on rows 10-20"): + source = SessionSource(platform=Platform.TELEGRAM, chat_id="c", user_id="u", chat_type="dm") + return MessageEvent(text=text, message_type=MessageType.TEXT, source=source, message_id="m") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("route", ["busy_steer_mode", "priority", "explicit_command"]) +async def test_busy_steer_fans_out_to_active_subagents(route, tmp_path, monkeypatch): + monkeypatch.setattr("gateway.run._hermes_home", tmp_path) + runner = GatewayRunner(config=GatewayConfig()) + child_a, child_b = _Agent(), _Agent() + parent = _Agent(children=[child_a, child_b]) + runner._session_state("key").turn.agent = parent + event = _event("/steer focus on rows 10-20" if route == "explicit_command" else "focus on rows 10-20") + + if route == "busy_steer_mode": + outcome = await runner._resolve_busy_steer_or_redirect(event, "key", "steer", parent) + assert outcome.steered is True + elif route == "priority": + runner._hm_busy_steer(event, parent, "key") + else: + reply = await runner._busy_steer_command(event, "key", event.source) + assert "subagent" in reply + + assert parent.payload and parent.payload.endswith("focus on rows 10-20") + # The looping child — the one actually doing the work — gets the same text, not just the parent. + assert child_a.payload == parent.payload + assert child_b.payload == parent.payload + + +def test_busy_steer_ack_names_subagents(tmp_path, monkeypatch): + monkeypatch.setattr("gateway.run._hermes_home", tmp_path) + runner = GatewayRunner(config=GatewayConfig()) + parent = _Agent(children=[_Agent()]) + kwargs = dict(is_steer_mode=True, is_queue_mode=False, is_redirect_mode=False, + demoted_for_subagents=False, demoted_for_compression=False) + with_children = runner._compose_busy_ack_message(_event(), 0.0, None, parent, **kwargs) + without = runner._compose_busy_ack_message(_event(), 0.0, None, _Agent(), **kwargs) + assert "subagent" in with_children + assert "subagent" not in without