"""Integration tests for Expert Team Pipeline Mode.

Covers the pipeline orchestration flow end-to-end with mocked LLM and AgentPool:
- F1: Manual team formation via @team:expert1,expert2
- F2: Default team template (@team:dev_team)
- F3: Pipeline execution (sequential phases with dependencies)
- F4: Parallel phases (no dependencies)
- F5: Phase failure and dependency failure propagation
- F6: SharedWorkspace data passing between phases
- F7: Context isolation (independent ConfigDrivenAgent per phase)
- F8: Fallback to single agent when all phases fail
- F9: Event sequence (team_formed → plan_update → phase_started → ... → team_synthesis)
- F10: TeamStatus.PLANNING state transition
- F11: Circular dependency detection
- F12: Invalid expert reference fallback
- F13: LLM decomposition failure fallback

Replaces the legacy CollaborationPlan-based tests (U9).
"""

from __future__ import annotations

import asyncio
import json
from unittest.mock import AsyncMock, MagicMock

import pytest

from agentkit.core.handoff_transport import InProcessHandoffTransport
from agentkit.core.protocol import TaskResult, TaskStatus
from agentkit.core.shared_workspace import SharedWorkspace
from agentkit.experts.config import ExpertConfig, ExpertTemplate
from agentkit.experts.orchestrator import TeamOrchestrator
from agentkit.experts.plan import PhaseStatus, PlanPhase, TeamPlan
from agentkit.experts.registry import ExpertTemplateRegistry
from agentkit.experts.router import ExpertTeamRouter
from agentkit.experts.team import ExpertTeam, TeamStatus


# ── 辅助函数 ──────────────────────────────────────────────


def _make_expert_config(
    name: str,
    persona: str = "",
    is_lead: bool = False,
    bound_skills: list[str] | None = None,
    llm: dict | None = None,
) -> ExpertConfig:
    """创建测试用 ExpertConfig"""
    return ExpertConfig(
        name=name,
        persona=persona or f"{name} expert",
        is_lead=is_lead,
        bound_skills=bound_skills or [],
        agent_type="expert",
        task_mode="llm_generate",
        prompt={"identity": f"You are {name}"},
        llm=llm,
    )


def _make_mock_agent(
    name: str,
    content: str | None = None,
    fail: bool = False,
    llm_gateway: MagicMock | None = None,
) -> MagicMock:
    """创建 mock ConfigDrivenAgent"""
    agent = MagicMock()
    if fail:
        agent.execute = AsyncMock(side_effect=RuntimeError(f"{name} execution failed"))
    else:
        agent.execute = AsyncMock(
            return_value=TaskResult(
                task_id="test",
                agent_name=name,
                status=TaskStatus.COMPLETED.value,
                output_data={"content": content or f"Result from {name}"},
                error_message=None,
                started_at=None,
                completed_at=None,
            )
        )
    agent._llm_gateway = llm_gateway
    return agent


def _make_mock_expert(
    name: str,
    is_lead: bool = False,
    is_active: bool = True,
    content: str | None = None,
    fail: bool = False,
    llm_gateway: MagicMock | None = None,
    llm: dict | None = None,
) -> MagicMock:
    """创建 mock Expert (spec=Expert)"""
    from agentkit.experts.expert import Expert

    config = _make_expert_config(name=name, is_lead=is_lead, llm=llm)
    expert = MagicMock(spec=Expert)
    expert.config = config
    expert.is_active = is_active
    expert.team_id = None
    expert.get_capabilities_summary.return_value = {
        "name": name,
        "persona": config.persona,
        "bound_skills": config.bound_skills,
        "is_lead": is_lead,
    }
    expert.agent = _make_mock_agent(name, content=content, fail=fail, llm_gateway=llm_gateway)
    return expert


def _make_mock_llm_gateway(
    phases: list[dict] | None = None,
    synthesis_content: str = "综合结果",
    decomp_fail: bool = False,
) -> MagicMock:
    """创建 mock LLM gateway.

    - phases: 提供 LLM 分解返回的阶段列表
    - decomp_fail: 分解调用抛出异常
    """
    gateway = AsyncMock()
    if decomp_fail:
        gateway.chat = AsyncMock(side_effect=RuntimeError("LLM unavailable"))
        return gateway

    if phases:
        phases_json = json.dumps(phases)
        decomp_response = MagicMock()
        decomp_response.content = phases_json
        synth_response = MagicMock()
        synth_response.content = synthesis_content
        gateway.chat = AsyncMock(side_effect=[decomp_response, synth_response])
    else:
        response = MagicMock()
        response.content = synthesis_content
        gateway.chat = AsyncMock(return_value=response)
    return gateway


def _make_team_with_experts(
    expert_names: list[str] | None = None,
    lead_name: str = "lead",
    pool: MagicMock | None = None,
    workspace: SharedWorkspace | None = None,
    transport: InProcessHandoffTransport | None = None,
) -> ExpertTeam:
    """创建包含 mock experts 的 ExpertTeam"""
    team = ExpertTeam(workspace=workspace or SharedWorkspace())
    # Use a real InProcessHandoffTransport so events can be captured
    team._handoff_transport = transport or InProcessHandoffTransport()
    if pool is not None:
        team._pool = pool

    if expert_names is None:
        expert_names = [lead_name, "member1", "member2"]

    for name in expert_names:
        is_lead = name == lead_name
        expert = _make_mock_expert(name=name, is_lead=is_lead)
        team._experts[name] = expert
        if is_lead:
            team._lead_expert_name = name

    return team


def _make_mock_pool(content_prefix: str = "Isolated") -> MagicMock:
    """创建 mock AgentPool，模拟上下文隔离的 agent 创建"""
    pool = MagicMock()
    pool.create_agent = AsyncMock(
        side_effect=lambda config: _make_mock_agent(
            config.name, content=f"{content_prefix} result from {config.name}"
        )
    )
    pool.remove_agent = AsyncMock()
    return pool


def _make_registry_with_dev_team() -> ExpertTemplateRegistry:
    """创建包含 dev_team 模板和 5 个编程专家的注册表"""
    reg = ExpertTemplateRegistry()
    for name in [
        "tech_lead",
        "frontend_engineer",
        "backend_engineer",
        "qa_engineer",
        "code_reviewer",
    ]:
        reg.register(
            ExpertTemplate(
                name=name,
                description=f"{name} expert",
                config=_make_expert_config(
                    name,
                    persona=f"{name} persona",
                    # Individual experts have empty bound_skills;
                    # only team templates use bound_skills for member list
                    bound_skills=[],
                ),
            )
        )
    # dev_team template stores members in bound_skills
    reg.register(
        ExpertTemplate(
            name="dev_team",
            description="Development team template",
            config=ExpertConfig(
                name="dev_team",
                persona="dev team",
                agent_type="expert",
                task_mode="llm_generate",
                prompt={"identity": "dev team"},
                bound_skills=[
                    "tech_lead",
                    "frontend_engineer",
                    "backend_engineer",
                    "qa_engineer",
                    "code_reviewer",
                ],
            ),
        )
    )
    return reg


async def _capture_events(transport: InProcessHandoffTransport, channel: str) -> list[dict]:
    """Capture all events sent to a channel."""
    events: list[dict] = []

    async def listener():
        async for msg in transport.listen(channel):
            events.append(msg)

    task = asyncio.create_task(listener())
    await asyncio.sleep(0.05)  # Let listener start
    return events, task


# ── Fixtures ─────────────────────────────────────────────


@pytest.fixture
def workspace():
    return SharedWorkspace()


@pytest.fixture
def registry():
    return _make_registry_with_dev_team()


# ── F1: Manual Team Formation ────────────────────────────


class TestManualTeamFormation:
    """F1: 用户通过 @team:expert1,expert2 指定专家团队"""

    def test_manual_team_with_templates(self, registry):
        """AE1: 用户通过模板名指定专家团队"""
        router = ExpertTeamRouter(registry)
        result = router.resolve("@team:tech_lead,frontend_engineer 开发登录功能")

        assert result.team_mode is True
        assert result.specified_experts == ["tech_lead", "frontend_engineer"]
        assert result.auto_compose is False
        assert result.task_content == "开发登录功能"

    def test_explicit_experts_resolved_to_configs(self, registry):
        """指定专家名解析为 ExpertConfig"""
        router = ExpertTeamRouter(registry)
        result = router.resolve("@team:frontend_engineer,backend_engineer 任务")

        configs = router.resolve_expert_configs(result.specified_experts)
        assert len(configs) == 2
        assert configs[0].name == "frontend_engineer"
        assert configs[0].is_lead is True
        assert configs[1].name == "backend_engineer"
        assert configs[1].is_lead is False

    def test_dynamic_expert_creation_for_unknown_name(self, registry):
        """未知专家名动态创建 ExpertConfig"""
        router = ExpertTeamRouter(registry)
        configs = router.resolve_expert_configs(["legal_advisor"])

        assert len(configs) == 1
        assert configs[0].name == "legal_advisor"
        assert configs[0].is_lead is True


# ── F2: Default Team Template ────────────────────────────


class TestDefaultTeamTemplate:
    """F2: @team:dev_team 调用 5 个编程专家"""

    def test_dev_team_template_expands_to_members(self, registry):
        """@team:dev_team 展开为 dev_team 模板的 bound_skills 成员列表"""
        router = ExpertTeamRouter(registry)
        result = router.resolve("@team:dev_team 开发用户系统")

        assert result.team_mode is True
        assert result.specified_experts == [
            "tech_lead",
            "frontend_engineer",
            "backend_engineer",
            "qa_engineer",
            "code_reviewer",
        ]
        assert result.auto_compose is False
        assert result.task_content == "开发用户系统"

    def test_no_experts_uses_default_template(self, registry):
        """@team 无指定专家时使用默认 dev_team 模板"""
        router = ExpertTeamRouter(registry)
        result = router.resolve("@team 开发功能")

        assert result.team_mode is True
        assert len(result.specified_experts) == 5
        assert "tech_lead" in result.specified_experts
        assert result.auto_compose is False


# ── F3: Pipeline Sequential Execution ────────────────────


class TestPipelineExecution:
    """F3: 流水线模式执行测试"""

    @pytest.mark.asyncio
    async def test_pipeline_execution_three_sequential_phases(self):
        """3 阶段（A→B→C）按序执行"""
        team = _make_team_with_experts()
        orchestrator = TeamOrchestrator(team)

        gateway = _make_mock_llm_gateway(
            phases=[
                {
                    "name": "A",
                    "assigned_expert": "lead",
                    "task_description": "阶段A",
                    "depends_on": [],
                },
                {
                    "name": "B",
                    "assigned_expert": "member1",
                    "task_description": "阶段B",
                    "depends_on": ["A"],
                },
                {
                    "name": "C",
                    "assigned_expert": "member2",
                    "task_description": "阶段C",
                    "depends_on": ["B"],
                },
            ]
        )
        team._experts["lead"].agent._llm_gateway = gateway

        result = await orchestrator.execute("串行任务")

        assert result["status"] == "completed"
        plan = result["plan"]
        assert len(plan.phases) == 3
        for ph in plan.phases:
            assert ph.status == PhaseStatus.COMPLETED

    @pytest.mark.asyncio
    async def test_pipeline_single_phase_no_llm(self):
        """无 LLM 时，任务作为单个阶段执行完成"""
        team = _make_team_with_experts()
        orchestrator = TeamOrchestrator(team)

        result = await orchestrator.execute("简单任务")

        assert result["status"] == "completed"
        assert "result" in result
        assert "phase_results" in result
        assert team.status == TeamStatus.COMPLETED

    @pytest.mark.asyncio
    async def test_pipeline_status_transitions(self):
        """执行过程中状态从 PLANNING → EXECUTING → SYNTHESIZING → COMPLETED"""
        team = _make_team_with_experts()
        orchestrator = TeamOrchestrator(team)

        statuses_seen: list[str] = []

        original_set_status = team.set_status

        def tracking_set_status(status: TeamStatus) -> None:
            statuses_seen.append(status.value)
            original_set_status(status)

        team.set_status = tracking_set_status  # type: ignore[assignment]

        await orchestrator.execute("任务")

        # PLANNING comes before EXECUTING, SYNTHESIZING, COMPLETED
        assert "planning" in statuses_seen
        assert "executing" in statuses_seen
        assert "synthesizing" in statuses_seen
        assert "completed" in statuses_seen
        assert statuses_seen.index("planning") < statuses_seen.index("executing")
        assert statuses_seen.index("executing") < statuses_seen.index("synthesizing")
        assert statuses_seen.index("synthesizing") < statuses_seen.index("completed")


# ── F4: Parallel Phases ──────────────────────────────────


class TestParallelPhases:
    """F4: 无依赖阶段并行执行"""

    @pytest.mark.asyncio
    async def test_parallel_phases_no_dependencies(self):
        """2 个无依赖阶段并行执行"""
        team = _make_team_with_experts()
        orchestrator = TeamOrchestrator(team)

        gateway = _make_mock_llm_gateway(
            phases=[
                {
                    "name": "A",
                    "assigned_expert": "member1",
                    "task_description": "阶段A",
                    "depends_on": [],
                },
                {
                    "name": "B",
                    "assigned_expert": "member2",
                    "task_description": "阶段B",
                    "depends_on": [],
                },
            ]
        )
        team._experts["lead"].agent._llm_gateway = gateway

        result = await orchestrator.execute("并行任务")

        assert result["status"] == "completed"
        plan = result["plan"]
        assert len(plan.phases) == 2
        for ph in plan.phases:
            assert ph.status == PhaseStatus.COMPLETED

    @pytest.mark.asyncio
    async def test_mixed_parallel_and_sequential(self):
        """混合并行和串行：A、B 并行，C 依赖 A 和 B"""
        team = _make_team_with_experts()
        orchestrator = TeamOrchestrator(team)

        gateway = _make_mock_llm_gateway(
            phases=[
                {
                    "name": "A",
                    "assigned_expert": "lead",
                    "task_description": "阶段A",
                    "depends_on": [],
                },
                {
                    "name": "B",
                    "assigned_expert": "member1",
                    "task_description": "阶段B",
                    "depends_on": [],
                },
                {
                    "name": "C",
                    "assigned_expert": "member2",
                    "task_description": "阶段C",
                    "depends_on": ["A", "B"],
                },
            ]
        )
        team._experts["lead"].agent._llm_gateway = gateway

        result = await orchestrator.execute("混合任务")

        assert result["status"] == "completed"
        plan = result["plan"]
        assert len(plan.phases) == 3
        c_phase = next(p for p in plan.phases if p.name == "C")
        assert len(c_phase.depends_on) == 2


# ── F5: Phase Failure and Dependency Propagation ─────────


class TestPhaseFailure:
    """F5: 阶段失败和依赖失败传播"""

    @pytest.mark.asyncio
    async def test_phase_dependency_failure_propagation(self):
        """阶段 B 依赖 A，A 失败时 B 标记为 FAILED"""
        team = _make_team_with_experts()
        # Make lead's agent fail
        team._experts["lead"].agent = _make_mock_agent("lead", fail=True)

        orchestrator = TeamOrchestrator(team)

        gateway = _make_mock_llm_gateway(
            phases=[
                {
                    "name": "A",
                    "assigned_expert": "lead",
                    "task_description": "阶段A",
                    "depends_on": [],
                },
                {
                    "name": "B",
                    "assigned_expert": "member1",
                    "task_description": "阶段B",
                    "depends_on": ["A"],
                },
            ]
        )
        team._experts["lead"].agent._llm_gateway = gateway

        result = await orchestrator.execute("失败传播任务")

        # Should fallback since all phases failed
        assert result["status"] == "fallback"
        plan = result["plan"]
        a_phase = next(p for p in plan.phases if p.name == "A")
        b_phase = next(p for p in plan.phases if p.name == "B")
        assert a_phase.status == PhaseStatus.FAILED
        assert b_phase.status == PhaseStatus.FAILED

    @pytest.mark.asyncio
    async def test_fallback_on_all_failure(self):
        """所有阶段失败时 fallback 到单 Agent"""
        team = _make_team_with_experts()
        # All experts fail
        for name in team._experts:
            team._experts[name].agent = _make_mock_agent(name, fail=True)

        orchestrator = TeamOrchestrator(team)

        # No LLM so single-phase path; that phase also fails → fallback
        result = await orchestrator.execute("全失败任务")

        assert result["status"] == "fallback"
        assert "result" in result
        assert result["result"] is not None


# ── F6: SharedWorkspace Data Passing ─────────────────────


class TestSharedWorkspacePassing:
    """F6: 阶段间通过 SharedWorkspace 传递数据"""

    @pytest.mark.asyncio
    async def test_phase_output_written_to_workspace(self, workspace):
        """阶段 A 的输出写入 SharedWorkspace"""
        team = _make_team_with_experts(workspace=workspace)
        orchestrator = TeamOrchestrator(team)

        gateway = _make_mock_llm_gateway(
            phases=[
                {
                    "name": "A",
                    "assigned_expert": "lead",
                    "task_description": "阶段A",
                    "depends_on": [],
                },
                {
                    "name": "B",
                    "assigned_expert": "member1",
                    "task_description": "阶段B",
                    "depends_on": ["A"],
                },
            ]
        )
        team._experts["lead"].agent._llm_gateway = gateway

        result = await orchestrator.execute("数据传递任务")

        assert result["status"] == "completed"
        plan = result["plan"]

        # Phase A output should be in workspace
        a_phase = next(p for p in plan.phases if p.name == "A")
        output_key = f"{plan.id}/phase/{a_phase.id}/output"
        data = await workspace.read(output_key)
        assert data is not None
        assert "Result from lead" in data.get("value", "")

    @pytest.mark.asyncio
    async def test_dependency_output_read_by_dependent(self, workspace):
        """阶段 B 读取阶段 A 的输出"""
        team = _make_team_with_experts(workspace=workspace)
        orchestrator = TeamOrchestrator(team)

        gateway = _make_mock_llm_gateway(
            phases=[
                {
                    "name": "A",
                    "assigned_expert": "lead",
                    "task_description": "阶段A",
                    "depends_on": [],
                },
                {
                    "name": "B",
                    "assigned_expert": "member1",
                    "task_description": "阶段B",
                    "depends_on": ["A"],
                },
            ]
        )
        team._experts["lead"].agent._llm_gateway = gateway

        # Track what input_data member1's agent receives
        received_inputs: list[dict] = []
        original_execute = team._experts["member1"].agent.execute

        async def tracking_execute(task_msg):
            received_inputs.append(task_msg.input_data)
            return await original_execute(task_msg)

        team._experts["member1"].agent.execute = tracking_execute

        await orchestrator.execute("依赖读取任务")

        # member1 should have received dependency_outputs in input_data
        assert len(received_inputs) > 0
        b_input = received_inputs[0]
        assert "dependency_outputs" in b_input
        assert "A" in b_input["dependency_outputs"]


# ── F7: Context Isolation ────────────────────────────────


class TestContextIsolation:
    """F7: 上下文隔离 - 每个阶段创建独立 ConfigDrivenAgent"""

    @pytest.mark.asyncio
    async def test_isolated_agent_created_per_phase(self):
        """每个阶段通过 AgentPool 创建独立的 agent"""
        pool = _make_mock_pool()
        team = _make_team_with_experts(pool=pool)
        orchestrator = TeamOrchestrator(team)

        gateway = _make_mock_llm_gateway(
            phases=[
                {
                    "name": "A",
                    "assigned_expert": "lead",
                    "task_description": "阶段A",
                    "depends_on": [],
                },
                {
                    "name": "B",
                    "assigned_expert": "member1",
                    "task_description": "阶段B",
                    "depends_on": [],
                },
            ]
        )
        team._experts["lead"].agent._llm_gateway = gateway

        await orchestrator.execute("隔离任务")

        # Pool.create_agent should have been called for each phase
        assert pool.create_agent.call_count >= 2

    @pytest.mark.asyncio
    async def test_isolated_agent_cleaned_up_after_phase(self):
        """阶段完成后清理临时 agent"""
        pool = _make_mock_pool()
        team = _make_team_with_experts(pool=pool)
        orchestrator = TeamOrchestrator(team)

        gateway = _make_mock_llm_gateway(
            phases=[
                {
                    "name": "A",
                    "assigned_expert": "lead",
                    "task_description": "阶段A",
                    "depends_on": [],
                },
            ]
        )
        team._experts["lead"].agent._llm_gateway = gateway

        await orchestrator.execute("清理任务")

        # All temp agents should be cleaned up
        assert len(orchestrator._temp_agents) == 0
        assert pool.remove_agent.call_count >= 1


# ── F8: Event Sequence ───────────────────────────────────


class TestEventSequence:
    """F8: 事件顺序正确"""

    @pytest.mark.asyncio
    async def test_event_sequence_pipeline_execution(self):
        """事件顺序：team_formed → plan_update → phase_started → phase_completed → team_synthesis"""
        transport = InProcessHandoffTransport()
        team = _make_team_with_experts(transport=transport)
        orchestrator = TeamOrchestrator(team)

        gateway = _make_mock_llm_gateway(
            phases=[
                {
                    "name": "A",
                    "assigned_expert": "lead",
                    "task_description": "阶段A",
                    "depends_on": [],
                },
            ]
        )
        team._experts["lead"].agent._llm_gateway = gateway

        events, listener_task = await _capture_events(transport, team.team_channel)

        await orchestrator.execute("事件序列任务")

        # Stop listener
        transport.close()
        try:
            await asyncio.wait_for(listener_task, timeout=0.5)
        except (asyncio.TimeoutError, Exception):
            listener_task.cancel()

        event_types = [e.get("type") for e in events]

        # Verify required events are present in order
        assert "team_formed" in event_types
        assert "plan_update" in event_types
        assert "phase_started" in event_types
        assert "phase_completed" in event_types
        assert "team_synthesis" in event_types

        # Verify order
        assert event_types.index("team_formed") < event_types.index("plan_update")
        assert event_types.index("plan_update") < event_types.index("phase_started")
        assert event_types.index("phase_started") < event_types.index("phase_completed")
        assert event_types.index("phase_completed") < event_types.index("team_synthesis")

    @pytest.mark.asyncio
    async def test_phase_failed_event_emitted_on_failure(self):
        """阶段失败时发出 phase_failed 事件"""
        transport = InProcessHandoffTransport()
        team = _make_team_with_experts(transport=transport)
        # Make all agents fail to trigger phase_failed
        for name in team._experts:
            team._experts[name].agent = _make_mock_agent(name, fail=True)
        orchestrator = TeamOrchestrator(team)

        events, listener_task = await _capture_events(transport, team.team_channel)

        await orchestrator.execute("失败事件任务")

        transport.close()
        try:
            await asyncio.wait_for(listener_task, timeout=0.5)
        except (asyncio.TimeoutError, Exception):
            listener_task.cancel()

        event_types = [e.get("type") for e in events]
        # phase_started should be emitted before failure
        assert "phase_started" in event_types


# ── F9: TeamStatus.PLANNING Transition ───────────────────


class TestTeamStatusPlanning:
    """F9: create_team 后状态为 PLANNING"""

    @pytest.mark.asyncio
    async def test_create_team_sets_planning_status(self):
        """create_team 后设置 PLANNING 状态"""
        pool = _make_mock_pool()
        team = ExpertTeam(pool=pool)

        lead_config = _make_expert_config("lead", is_lead=True)
        member_config = _make_expert_config("member")

        await team.create_team(lead_config, [member_config])

        assert team.status == TeamStatus.PLANNING

    @pytest.mark.asyncio
    async def test_orchestrator_sets_planning_before_decomposition(self):
        """orchestrator.execute 在分解任务前设置 PLANNING"""
        team = _make_team_with_experts()
        orchestrator = TeamOrchestrator(team)

        statuses_during_execution: list[TeamStatus] = []
        original_set_status = team.set_status

        def tracking_set_status(status: TeamStatus) -> None:
            statuses_during_execution.append(status)
            original_set_status(status)

        team.set_status = tracking_set_status  # type: ignore[assignment]

        await orchestrator.execute("任务")

        # First status set should be PLANNING
        assert TeamStatus.PLANNING in statuses_during_execution


# ── F10: Circular Dependency Detection ───────────────────


class TestCircularDependency:
    """F10: 循环依赖检测"""

    @pytest.mark.asyncio
    async def test_circular_dependency_triggers_fallback(self):
        """A→B→A 循环依赖 → fallback 到单 Agent"""
        team = _make_team_with_experts()
        orchestrator = TeamOrchestrator(team)

        # Phases with circular dependency: A depends on B, B depends on A
        # We need to craft LLM response that creates circular deps via name resolution
        # Since depends_on is resolved by name → id, and ids are auto-generated,
        # we craft phases where A depends on B and B depends on A
        gateway = _make_mock_llm_gateway(
            phases=[
                {
                    "name": "A",
                    "assigned_expert": "lead",
                    "task_description": "阶段A",
                    "depends_on": ["B"],
                },
                {
                    "name": "B",
                    "assigned_expert": "member1",
                    "task_description": "阶段B",
                    "depends_on": ["A"],
                },
            ]
        )
        team._experts["lead"].agent._llm_gateway = gateway

        result = await orchestrator.execute("循环依赖任务")

        # Should fallback due to ValueError from topological_sort
        assert result["status"] == "fallback"

    def test_topological_sort_raises_on_cycle(self):
        """topological_sort() 检测到循环依赖时抛出 ValueError"""
        plan = TeamPlan(task="test", lead_expert="lead")
        a = PlanPhase(name="A", assigned_expert="lead")
        b = PlanPhase(name="B", assigned_expert="member1")
        a.depends_on = [b.id]
        b.depends_on = [a.id]
        plan.phases = [a, b]

        with pytest.raises(ValueError, match="Circular dependency"):
            plan.topological_sort()


# ── F11: Invalid Expert Reference Fallback ───────────────


class TestInvalidExpertReference:
    """F11: 阶段引用不存在的专家 → fallback 到 lead"""

    @pytest.mark.asyncio
    async def test_invalid_expert_reference_falls_back_to_lead(self):
        """阶段引用不存在的专家时 fallback 到 lead expert"""
        team = _make_team_with_experts()
        orchestrator = TeamOrchestrator(team)

        gateway = _make_mock_llm_gateway(
            phases=[
                {
                    "name": "A",
                    "assigned_expert": "nonexistent_expert",
                    "task_description": "阶段A",
                    "depends_on": [],
                },
            ]
        )
        team._experts["lead"].agent._llm_gateway = gateway

        result = await orchestrator.execute("无效专家任务")

        # Should complete (fallback to lead for the phase)
        assert result["status"] == "completed"
        plan = result["plan"]
        a_phase = plan.phases[0]
        # assigned_expert should have been reassigned to lead
        assert a_phase.assigned_expert == "lead"
        assert a_phase.status == PhaseStatus.COMPLETED


# ── F12: LLM Decomposition Failure ───────────────────────


class TestLLMDecompositionFailure:
    """F12: LLM 分解失败 → fallback 到单阶段"""

    @pytest.mark.asyncio
    async def test_llm_decomposition_failure_falls_back_to_single_phase(self):
        """LLM 返回无效 JSON → fallback 到单阶段"""
        team = _make_team_with_experts()
        orchestrator = TeamOrchestrator(team)

        # Gateway that returns invalid JSON for decomposition
        gateway = AsyncMock()
        decomp_response = MagicMock()
        decomp_response.content = "This is not JSON"
        synth_response = MagicMock()
        synth_response.content = "综合结果"
        gateway.chat = AsyncMock(side_effect=[decomp_response, synth_response])
        team._experts["lead"].agent._llm_gateway = gateway

        result = await orchestrator.execute("无效JSON任务")

        # Should complete as single phase
        assert result["status"] == "completed"
        plan = result["plan"]
        assert len(plan.phases) == 1

    @pytest.mark.asyncio
    async def test_llm_unavailable_falls_back_to_single_phase(self):
        """LLM 不可用时 fallback 到单阶段"""
        team = _make_team_with_experts()
        orchestrator = TeamOrchestrator(team)

        gateway = _make_mock_llm_gateway(decomp_fail=True)
        team._experts["lead"].agent._llm_gateway = gateway

        result = await orchestrator.execute("LLM不可用任务")

        assert result["status"] == "completed"
        plan = result["plan"]
        assert len(plan.phases) == 1
        assert plan.phases[0].assigned_expert == "lead"


# ── F13: Decentralized Collaboration (HandoffTransport) ──


class TestDecentralizedCollaboration:
    """F13: 专家间直接协作（HandoffTransport）"""

    @pytest.mark.asyncio
    async def test_expert_direct_handoff(self):
        """AE4: Expert A 请求 Expert B 协助"""
        transport = InProcessHandoffTransport()
        channel = "expert:analyst:handoff"

        messages: list[dict] = []

        async def listener():
            async for msg in transport.listen(channel):
                messages.append(msg)
                break

        task = asyncio.create_task(listener())
        await asyncio.sleep(0.05)

        await transport.send(
            channel,
            {
                "source_expert": "analyst",
                "target_expert": "researcher",
                "task": "需要行业数据",
                "type": "assist_request",
            },
        )

        await asyncio.wait_for(task, timeout=2.0)

        assert len(messages) == 1
        assert messages[0]["source_expert"] == "analyst"
        assert messages[0]["type"] == "assist_request"

        transport.close()

    @pytest.mark.asyncio
    async def test_team_channel_broadcast(self):
        """团队频道消息广播到所有专家"""
        transport = InProcessHandoffTransport()
        channel = "team:test-team"

        consumer1_msgs: list[dict] = []
        consumer2_msgs: list[dict] = []

        async def consumer1():
            async for msg in transport.listen(channel):
                consumer1_msgs.append(msg)
                if len(consumer1_msgs) >= 1:
                    break

        async def consumer2():
            async for msg in transport.listen(channel):
                consumer2_msgs.append(msg)
                if len(consumer2_msgs) >= 1:
                    break

        t1 = asyncio.create_task(consumer1())
        t2 = asyncio.create_task(consumer2())
        await asyncio.sleep(0.05)

        await transport.send(channel, {"type": "chat", "content": "hello"})

        await asyncio.wait_for(asyncio.gather(t1, t2), timeout=2.0)

        assert len(consumer1_msgs) == 1
        assert len(consumer2_msgs) == 1

        transport.close()


# ── F14: User Intervention ───────────────────────────────


class TestUserIntervention:
    """F14: 用户干预消息广播"""

    @pytest.mark.asyncio
    async def test_user_intervention_broadcast(self):
        """AE3: 用户干预消息到达所有专家"""
        transport = InProcessHandoffTransport()
        channel = "team:intervention-test"

        received: list[dict] = []

        async def listener():
            async for msg in transport.listen(channel):
                received.append(msg)
                if len(received) >= 1:
                    break

        task = asyncio.create_task(listener())
        await asyncio.sleep(0.05)

        await transport.send(
            channel,
            {"type": "user_intervention", "content": "重点看成本优化"},
        )

        await asyncio.wait_for(task, timeout=2.0)

        assert len(received) == 1
        assert received[0]["type"] == "user_intervention"

        transport.close()


# ── F15: Team Dissolution ────────────────────────────────


class TestTeamDissolution:
    """F15: 团队解散和输出保留"""

    @pytest.mark.asyncio
    async def test_dissolution_preserves_outputs(self, workspace):
        """R36: 临时 Expert 输出在 SharedWorkspace 中保留"""
        await workspace.write("team:test:analyst:result", {"report": "analysis result"}, "analyst")

        data = await workspace.read("team:test:analyst:result")
        assert data is not None
        assert data["value"]["report"] == "analysis result"

    @pytest.mark.asyncio
    async def test_dissolution_sets_status(self):
        """团队解散后状态为 DISSOLVED"""
        pool = _make_mock_pool()
        team = ExpertTeam(pool=pool)

        lead_config = _make_expert_config("lead", is_lead=True)
        await team.create_team(lead_config)

        await team.dissolve()

        assert team.status == TeamStatus.DISSOLVED

    @pytest.mark.asyncio
    async def test_dissolution_clears_experts(self):
        """团队解散后 experts 列表为空"""
        pool = _make_mock_pool()
        team = ExpertTeam(pool=pool)

        lead_config = _make_expert_config("lead", is_lead=True)
        member_config = _make_expert_config("member")
        await team.create_team(lead_config, [member_config])

        assert len(team.experts) == 2

        await team.dissolve()

        assert len(team.experts) == 0
        assert team.lead_expert is None


# ── F16: Dynamic Expert Management ───────────────────────


class TestDynamicExpertManagement:
    """F16: 动态添加和移除专家"""

    @pytest.mark.asyncio
    async def test_add_expert_by_template(self, registry):
        """通过模板名添加专家"""
        router = ExpertTeamRouter(registry)
        result = router.resolve("@team:tech_lead 分析报告")

        configs = router.resolve_expert_configs(result.specified_experts)
        assert len(configs) == 1
        assert configs[0].name == "tech_lead"
        assert configs[0].is_lead is True

    @pytest.mark.asyncio
    async def test_add_expert_dynamic(self, registry):
        """未知专家名动态创建配置"""
        router = ExpertTeamRouter(registry)
        configs = router.resolve_expert_configs(["legal_advisor"])

        assert len(configs) == 1
        assert configs[0].name == "legal_advisor"
        assert "legal_advisor" in configs[0].persona
