Background subagents interrupted or lost on new user message

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #32076 fix(gateway): protect in-flight subagents from busy-mode interrupts

  merged to main

  files: gateway/run.py, tests/gateway/test_subagent_protection_30170.py

  Show the diff

  • gateway/run.py

    @@ -3034,6 +3034,44 @@ def _snapshot_running_agents(self) -> Dict[str, Any]:
                 if agent is not _AGENT_PENDING_SENTINEL
             }
     
    +    @staticmethod
    +    def _agent_has_active_subagents(running_agent: Any) -> bool:
    +        """Return True when *running_agent* is currently driving subagents
    +        via the ``delegate_task`` tool.
    +
    +        Background (#30170): ``AIAgent.interrupt()`` cascades through the
    +        parent's ``_active_children`` list and calls ``interrupt()`` on
    +        every child synchronously, which aborts in-flight subagent work
    +        and produces a fallback cascade with no actionable signal.
    +        Demoting ``busy_input_mode='interrupt'`` to ``queue`` semantics
    +        whenever this helper returns True protects subagent work from
    +        conversational follow-ups while leaving the explicit ``/stop``
    +        path (which goes through ``_interrupt_and_clear_session``)
    +        untouched. Safe-by-default: returns False on any attribute or
    +        lock error so a missing/broken parent never blocks the existing
    +        interrupt path.
    +        """
    +        if running_agent is None or running_agent is _AGENT_PENDING_SENTINEL:
    +            return False
    +        children = getattr(running_agent, "_active_children", None)
    +        # AIAgent always initialises this as a concrete list (see
    +        # agent/agent_init.py). Reject anything that isn't a real
    +        # collection — this guards against ``MagicMock()._active_children``
    +        # auto-creating a truthy stub in tests and triggering the demotion
    +        # against an agent that doesn't actually have subagents.
    +        if not isinstance(children, (list, tuple, set)):
    +            return False
    +        if not children:
    +            return False
    +        lock = getattr(running_agent, "_active_children_lock", None)
    +        try:
    +            if lock is not None:
    +                with lock:
    +                    return bool(children)
    +            return bool(children)
    +        except Exception:
    +            return False
    +
         def _queue_or_replace_pending_event(self, session_key: str, event: MessageEvent) -> None:
             adapter = self.adapters.get(event.source.platform)
             if not adapter:
    @@ -3105,6 +3143,25 @@ async def _handle_active_session_busy_message(self, event: MessageEvent, session
             # queueing + interrupting.  If the agent isn't running yet
             # (sentinel) or lacks steer(), or the payload is empty, fall back
             # to queue semantics so nothing is lost.
    +        # #30170 — Subagent protection. ``AIAgent.interrupt()`` cascades
    +        # to every entry in the parent's ``_active_children`` list and
    +        # aborts in-flight ``delegate_task`` work. Demote ``interrupt``
    +        # to ``queue`` when the parent is currently driving subagents so
    +        # a conversational follow-up doesn't destroy minutes of subagent
    +        # work. Explicit ``/stop`` and ``/new`` slash commands go through
    +        # ``_interrupt_and_clear_session`` and are unaffected — the
    +        # operator still has a way to force-cancel everything.
    +        demoted_for_subagents = (
    +            effective_mode == "interrupt"
    +            and self._agent_has_active_subagents(running_agent)
    +        )
    +        if demoted_for_subagents:
    +            logger.info(
    +                "Demoting busy_input_mode 'interrupt' to 'queue' for session %s "
    +                "because the running agent has active subagents (#30170)",
    +                session_key,
    +            )
    +            effective_mode = "queue"
             steered = False
             if effective_mode == "steer":
                 steer_text = (event.text or "").strip()
    @@ -3192,6 +3249,14 @@ async def _handle_active_session_busy_message(self, event: MessageEvent, session
                     f"⏩ Steered into current run{status_detail}. "
                     f"Your message arrives after the next tool call."
                 )
    +        elif is_queue_mode and demoted_for_subagents:
    +            # #30170 — explain the demotion so the user knows their
    +            # follow-up didn't accidentally kill the subagent and
    +            # discovers `/stop` as the explicit escape hatch.
    +            message = (
    +                f"⏳ Subagent working{status_detail} — your message is queued for "
    +                f"when it finishes (use /stop to cancel everything)."
    +            )
             elif is_queue_mode:
                 message = (
                     f"⏳ Queued for the next turn{status_detail}. "
    @@ -7246,6 +7311,22 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
                     logger.debug("PRIORITY steer-fallback-to-queue for session %s", _quick_key)
                     self._queue_or_replace_pending_event(_quick_key, event)
                     return None
    +            # #30170 — Subagent protection (PRIORITY path). Same rationale
    +            # as ``_handle_active_session_busy_message``: an interrupt
    +            # cascades through ``_active_children`` and aborts in-flight
    +            # delegate_task work. Demote to queue semantics when the
    +            # parent is currently driving subagents so a conversational
    +            # follow-up doesn't destroy minutes of subagent progress.
    +            # /stop reaches its dedicated handler above, so the operator
    +            # still has a clean escape hatch.
    +            if self._agent_has_active_subagents(running_agent):
    +                logger.info(
    +                    "PRIORITY interrupt demoted to queue for session %s "
    +                    "because the running agent has active subagents (#30170)",
    +                    _quick_key,
    +                )
    +                self._queue_or_replace_pending_event(_quick_key, event)
    +                return None
                 logger.debug("PRIORITY interrupt for session %s", _quick_key)
                 running_agent.interrupt(event.text)
                 # NOTE: self._pending_messages was write-only (never consumed).

  • tests/gateway/test_subagent_protection_30170.py

    @@ -0,0 +1,348 @@
    +"""Regression tests for #30170.
    +
    +#30170: Sending a message while ``delegate_task`` is running killed the
    +subagent because the gateway always called ``running_agent.interrupt()``
    +on the parent, which then cascaded synchronously through
    +``AIAgent._active_children`` and aborted every in-flight subagent. The
    +reporter (and the linked Phase-1 spec) asked for the gateway to demote
    +``busy_input_mode='interrupt'`` to ``queue`` semantics whenever the
    +parent is currently driving subagents, while leaving explicit ``/stop``
    +and ``/new`` slash commands untouched.
    +
    +These tests pin down the gateway-side guard introduced for #30170:
    +
    +* ``GatewayRunner._agent_has_active_subagents`` correctly recognises
    +  parents that own real children, without false-positives from a
    +  ``MagicMock()._active_children`` auto-attribute, missing locks, or
    +  the ``_AGENT_PENDING_SENTINEL`` placeholder.
    +* ``_handle_active_session_busy_message`` demotes the interrupt mode to
    +  queue semantics (no ``interrupt()`` call, message merged into the
    +  pending queue, ack reflects the demotion) when the parent has active
    +  subagents.
    +* The ``queue`` and ``steer`` configured modes still behave exactly as
    +  before — the guard is interrupt-only.
    +"""
    +
    +from __future__ import annotations
    +
    +import sys
    +import threading
    +import time
    +import types
    +from typing import Any
    +from unittest.mock import AsyncMock, MagicMock, patch
    +
    +import pytest
    +
    +# ──────────────────────────────────────────────────────────────────────
    +# Minimal stubs so gateway imports cleanly (mirrors test_busy_session_ack)
    +# ──────────────────────────────────────────────────────────────────────
    +_tg = types.ModuleType("telegram")
    +_tg.constants = types.ModuleType("telegram.constants")
    +_ct = MagicMock()
    +_ct.SUPERGROUP = "supergroup"
    +_ct.GROUP = "group"
    +_ct.PRIVATE = "private"
    +_tg.constants.ChatType = _ct
    +sys.modules.setdefault("telegram", _tg)
    +sys.modules.setdefault("telegram.constants", _tg.constants)
    +sys.modules.setdefault("telegram.ext", types.ModuleType("telegram.ext"))
    +
    +from gateway.platforms.base import (  # noqa: E402
    +    MessageEvent,
    +    MessageType,
    +    SessionSource,
    +    build_session_key,
    +)
    +from gateway.run import GatewayRunner, _AGENT_PENDING_SENTINEL  # noqa: E402
    +
    +
    +# ──────────────────────────────────────────────────────────────────────
    +# Builders (parallel to tests/gateway/test_busy_session_ack.py)
    +# ──────────────────────────────────────────────────────────────────────
    +def _make_event(text: str = "hello", chat_id: str = "123") -> MessageEvent:
    +    source = SessionSource(
    +        platform=MagicMock(value="telegram"),
    +        chat_id=chat_id,
    +        chat_type="private",
    +        user_id="user1",
    +    )
    +    return MessageEvent(
    +        text=text,
    +        message_type=MessageType.TEXT,
    +        source=source,
    +        message_id="msg1",
    +    )
    +
    +
    +def _make_runner() -> GatewayRunner:
    +    runner = object.__new__(GatewayRunner)
    +    runner._running_agents = {}
    +    runner._running_agents_ts = {}
    +    runner._pending_messages = {}
    +    runner._busy_ack_ts = {}
    +    runner._draining = False
    +    runner.adapters = {}
    +    runner.config = MagicMock()
    +    runner.session_store = None
    +    runner.hooks = MagicMock()
    +    runner.hooks.emit = AsyncMock()
    +    runner.pairing_store = MagicMock()
    +    runner.pairing_store.is_approved.return_value = True
    +    runner._is_user_authorized = lambda _source: True
    +    return runner
    +
    +
    +def _make_adapter() -> MagicMock:
    +    adapter = MagicMock()
    +    adapter._pending_messages = {}
    +    adapter._send_with_retry = AsyncMock()
    +    adapter.config = MagicMock()
    +    adapter.config.extra = {}
    +    adapter.platform = MagicMock(value="telegram")
    +    return adapter
    +
    +
    +def _make_parent_with_subagents(
    +    *, children: int = 1, with_lock: bool = True
    +) -> MagicMock:
    +    """A MagicMock shaped like an AIAgent that currently owns *children* subagents."""
    +    parent = MagicMock()
    +    parent._active_children = [MagicMock() for _ in range(children)]
    +    parent._active_children_lock = threading.Lock() if with_lock else None
    +    parent.get_activity_summary.return_value = {
    +        "api_call_count": 7,
    +        "max_iterations": 60,
    +        "current_tool": "delegate_task",
    +    }
    +    return parent
    +
    +
    +def _make_parent_no_subagents() -> MagicMock:
    +    """A MagicMock shaped like an AIAgent that is NOT delegating."""
    +    parent = MagicMock()
    +    parent._active_children = []
    +    parent._active_children_lock = threading.Lock()
    +    parent.get_activity_summary.return_value = {
    +        "api_call_count": 3,
    +        "max_iterations": 60,
    +        "current_tool": "terminal",
    +    }
    +    return parent
    +
    +
    +# ──────────────────────────────────────────────────────────────────────
    +# _agent_has_active_subagents
    +# ──────────────────────────────────────────────────────────────────────
    +class TestAgentHasActiveSubagents:
    +    """The detection helper must be both precise and defensive."""
    +
    +    def test_returns_false_for_none(self) -> None:
    +        assert GatewayRunner._agent_has_active_subagents(None) is False
    +
    +    def test_returns_false_for_pending_sentinel(self) -> None:
    +        assert (
    +            GatewayRunner._agent_has_active_subagents(_AGENT_PENDING_SENTINEL)
    +            is False
    +        )
    +
    +    def test_returns_false_when_attribute_missing(self) -> None:
    +        """Production AIAgents always have _active_children, but the helper
    +        must not blow up on test stubs or partial mocks."""
    +
    +        class StubAgent:
    +            pass
    +
    +        assert GatewayRunner._agent_has_active_subagents(StubAgent()) is False
    +
    +    def test_returns_false_for_empty_list(self) -> None:
    +        assert (
    +            GatewayRunner._agent_has_active_subagents(_make_parent_no_subagents())
    +            is False
    +        )
    +
    +    def test_returns_true_for_single_child(self) -> None:
    +        assert (
    +            GatewayRunner._agent_has_active_subagents(_make_parent_with_subagents())
    +            is True
    +        )
    +
    +    def test_returns_true_for_many_children(self) -> None:
    +        assert (
    +            GatewayRunner._agent_has_active_subagents(
    +                _make_parent_with_subagents(children=5)
    +            )
    +            is True
    +        )
    +
    +    def test_works_without_lock(self) -> None:
    +        """``_active_children_lock`` is optional in test stubs."""
    +        assert (
    +            GatewayRunner._agent_has_active_subagents(
    +                _make_parent_with_subagents(with_lock=False)
    +            )
    +            is True
    +        )
    +
    +    def test_rejects_truthy_non_collection_attribute(self) -> None:
    +        """The MagicMock auto-attribute regression. ``MagicMock()._active_children``
    +        is itself a truthy MagicMock — without the isinstance guard, the
    +        helper would falsely report subagents on every test mock."""
    +        parent = MagicMock()  # no explicit _active_children setup
    +        assert GatewayRunner._agent_has_active_subagents(parent) is False
    +
    +    @pytest.mark.parametrize(
    +        "container",
    +        [(MagicMock(),), {MagicMock()}, [MagicMock()]],
    +        ids=["tuple", "set", "list"],
    +    )
    +    def test_accepts_list_tuple_set(self, container: Any) -> None:
    +        parent = MagicMock()
    +        parent._active_children = container
    +        parent._active_children_lock = threading.Lock()
    +        assert GatewayRunner._agent_has_active_subagents(parent) is True
    +
    +
    +# ──────────────────────────────────────────────────────────────────────
    +# _handle_active_session_busy_message — interrupt demotion
    +# ──────────────────────────────────────────────────────────────────────
    +class TestBusyHandlerDemotesInterruptForSubagents:
    +    """The Phase-1 fix from #30170: parent.interrupt() must NOT fire when
    +    the parent is currently driving subagents."""
    +
    +    @pytest.mark.asyncio
    +    async def test_does_not_call_interrupt_when_subagents_active(self) -> None:
    +        runner = _make_runner()
    +        runner._busy_input_mode = "interrupt"
    +        adapter = _make_adapter()
    +        event = _make_event(text="follow up while subagent runs")
    +        sk = build_session_key(event.source)
    +        parent = _make_parent_with_subagents()
    +        runner._running_agents[sk] = parent
    +        runner.adapters[event.source.platform] = adapter
    +
    +        with patch("gateway.run.merge_pending_message_event") as merge_mock:
    +            handled = await runner._handle_active_session_busy_message(event, sk)
    +
    +        assert handled is True
    +        parent.interrupt.assert_not_called()
    +        # Message must still be queued so it gets picked up on the next turn.
    +        merge_mock.assert_called_once()
    +
    +    @pytest.mark.asyncio
    +    async def test_ack_explains_the_demotion(self) -> None:
    +        """The user-visible ack must mention the subagent context AND
    +        the `/stop` escape hatch so the operator can self-correct."""
    +        runner = _make_runner()
    +        runner._busy_input_mode = "interrupt"
    +        adapter = _make_adapter()
    +        event = _make_event(text="hi mid-delegation")
    +        sk = build_session_key(event.source)
    +        parent = _make_parent_with_subagents()
    +        runner._running_agents[sk] = parent
    +        runner._running_agents_ts[sk] = time.time() - 120
    +        runner.adapters[event.source.platform] = adapter
    +
    +        with patch("gateway.run.merge_pending_message_event"):
    +            await runner._handle_active_session_busy_message(event, sk)
    +
    +        adapter._send_with_retry.assert_called_once()
    +        content = adapter._send_with_retry.call_args.kwargs.get("content", "")
    +        assert "Subagent working" in content
    +        assert "queued" in content.lower()
    +        assert "/stop" in content
    +        assert "Interrupting" not in content
    +
    +    @pytest.mark.asyncio
    +    async def test_interrupt_still_fires_when_no_subagents(self) -> None:
    +        """Regression-guard the other direction: with no subagents the
    +        demotion must NOT trigger and behaviour must be byte-identical
    +        to the pre-#30170 interrupt path."""
    +        runner = _make_runner()
    +        runner._busy_input_mode = "interrupt"
    +        adapter = _make_adapter()
    +        event = _make_event(text="please stop")
    +        sk = build_session_key(event.source)
    +        parent = _make_parent_no_subagents()
    +        runner._running_agents[sk] = parent
    +        runner.adapters[event.source.platform] = adapter
    +
    +        with patch("gateway.run.merge_pending_message_event"):
    +            await runner._handle_active_session_busy_message(event, sk)
    +
    +        parent.interrupt.assert_called_once_with("please stop")
    +        content = adapter._send_with_retry.call_args.kwargs.get("content", "")
    +        assert "Interrupting" in content
    +        assert "Subagent" not in content
    +
    +    @pytest.mark.asyncio
    +    async def test_queue_mode_unchanged_with_subagents(self) -> None:
    +        """Configured ``queue`` mode is already subagent-safe; the new
    +        guard must not change its behaviour or its ack text."""
    +        runner = _make_runner()
    +        runner._busy_input_mode = "queue"
    +        adapter = _make_adapter()
    +        event = _make_event(text="queued during delegate")
    +        sk = build_session_key(event.source)
    +        parent = _make_parent_with_subagents()
    +        runner._running_agents[sk] = parent
    +        runner.adapters[event.source.platform] = adapter
    +
    +        with patch("gateway.run.merge_pending_message_event"):
    +            await runner._handle_active_session_busy_message(event, sk)
    +
    +        parent.interrupt.assert_not_called()
    +        content = adapter._send_with_retry.call_args.kwargs.get("content", "")
    +        # The vanilla queue copy — NOT the #30170 "Subagent working" copy,
    +        # because the user explicitly asked for queue mode.
    +        assert "Queued for the next turn" in content
    +        assert "respond once the current task finishes" in content
    +        assert "Subagent working" not in content
    +
    +    @pytest.mark.asyncio
    +    async def test_steer_mode_still_routes_through_running_agent_steer(
    +        self,
    +    ) -> None:
    +        """Configured ``steer`` mode must reach ``running_agent.steer()``
    +        even when subagents are active — the #30170 demotion is
    +        interrupt-specific so it doesn't accidentally disable steer."""
    +        runner = _make_runner()
    +        runner._busy_input_mode = "steer"
    +        adapter = _make_adapter()
    +        event = _make_event(text="course-correct")
    +        sk = build_session_key(event.source)
    +        parent = _make_parent_with_subagents()
    +        parent.steer = MagicMock(return_value=True)
    +        runner._running_agents[sk] = parent
    +        runner.adapters[event.source.platform] = adapter
    +
    +        with patch("gateway.run.merge_pending_message_event"):
    +            await runner._handle_active_session_busy_message(event, sk)
    +
    +        parent.steer.assert_called_once_with("course-correct")
    +        parent.interrupt.assert_not_called()
    +
    +    @pytest.mark.asyncio
    +    async def test_pending_sentinel_does_not_demote(self) -> None:
    +        """The placeholder ``_AGENT_PENDING_SENTINEL`` is not a real
    +        agent — the guard must not treat it as having subagents.
    +        Otherwise we'd permanently queue messages for sessions that
    +        haven't actually started running yet."""
    +        runner = _make_runner()
    +        runner._busy_input_mode = "interrupt"
    +        adapter = _make_adapter()
    +        event = _make_event(text="follow up before start")
    +        sk = build_session_key(event.source)
    +        runner._running_agents[sk] = _AGENT_PENDING_SENTINEL
    +        runner.adapters[event.source.platform] = adapter
    +
    +        with patch("gateway.run.merge_pending_message_event"):
    +            handled = await runner._handle_active_session_busy_message(event, sk)
    +
    +        assert handled is True
    +        # Sentinel can't be interrupted (no .interrupt to call) — verify
    +        # that the helper still returns the "interrupting" copy because
    +        # demotion did NOT fire (and the sentinel branch in the real
    +        # handler just skips the interrupt call silently).
    +        content = adapter._send_with_retry.call_args.kwargs.get("content", "")
    +        assert "Subagent working" not in content

• #65040 fix(tui): preserve background delegation on busy submit

  merged to main

  files: tests/test_tui_gateway_queue_on_busy.py, tools/delegate_tool.py

• #72409 fix(desktop): preserve live background subagents across message.start (salvage #67005)

  merged to main

  files: apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts, apps/desktop/src/store/subagents.test.ts, apps/desktop/src/store/subagents.ts, contributors/emails/[redacted:email]