"""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 types
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 (
    SessionSource,
    build_session_key,
)
from gateway.platforms.event import MessageEvent, MessageType
from gateway.run import GatewayRunner  # 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_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_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


# ──────────────────────────────────────────────────────────────────────
# _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_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.platforms.base.merge_pending_message_event"):
            await runner._handle_active_session_busy_message(event, sk)

        parent.interrupt.assert_called_once_with("please stop")

    @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.platforms.base.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 "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.platforms.base.merge_pending_message_event"):
            await runner._handle_active_session_busy_message(event, sk)

        parent.steer.assert_called_once()
        injected = parent.steer.call_args.args[0]
        assert injected.endswith("course-correct")
        assert '"chat_id": "123"' in injected
        parent.interrupt.assert_not_called()