"""Tests for PlanExecutor — 执行计划执行器"""

from __future__ import annotations

import asyncio
import pytest
from datetime import datetime, timezone
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch

from agentkit.core.plan_executor import (
    FailureAction,
    PlanExecutor,
    PlanExecutionResult,
    StepExecutionResult,
)
from agentkit.core.plan_schema import (
    ExecutionPlan,
    PlanStep,
    PlanStepStatus,
    SkillGap,
    SkillGapLevel,
)
from agentkit.core.protocol import TaskMessage, TaskResult, TaskStatus


# --- Helpers ---


def make_task(task_id: str = "t1", input_data: dict | None = None) -> TaskMessage:
    return TaskMessage(
        task_id=task_id,
        agent_name="test_agent",
        task_type="test",
        priority=1,
        input_data=input_data or {"query": "test"},
        callback_url=None,
        created_at=datetime.now(timezone.utc),
    )


def make_plan(
    steps: list[PlanStep] | None = None,
    parallel_groups: list[list[str]] | None = None,
    goal: str = "test goal",
) -> ExecutionPlan:
    if steps is None:
        steps = [PlanStep(step_id="s0", name="Step 0", description="First step")]
    if parallel_groups is None:
        parallel_groups = [[s.step_id for s in steps]]
    return ExecutionPlan(
        plan_id="p1",
        goal=goal,
        steps=steps,
        parallel_groups=parallel_groups,
        confirmed=True,
    )


class MockAgent:
    """Mock Agent for testing"""

    def __init__(
        self,
        name: str = "mock_agent",
        output_data: dict | None = None,
        should_fail: bool = False,
        fail_count: int = 0,
    ):
        self.name = name
        self.agent_type = "mock"
        self._output_data = output_data or {"result": f"output from {name}"}
        self._should_fail = should_fail
        self._fail_count = fail_count
        self._call_count = 0

    async def execute(self, task: TaskMessage) -> TaskResult:
        self._call_count += 1
        if self._should_fail:
            raise RuntimeError(f"Agent {self.name} failed")
        if self._fail_count > 0 and self._call_count <= self._fail_count:
            raise RuntimeError(f"Agent {self.name} failed (attempt {self._call_count})")
        now = datetime.now(timezone.utc)
        return TaskResult(
            task_id=task.task_id,
            agent_name=self.name,
            status=TaskStatus.COMPLETED,
            output_data=self._output_data,
            error_message=None,
            started_at=now,
            completed_at=now,
        )


class MockAgentPool:
    """Mock AgentPool for testing"""

    def __init__(
        self,
        agents: dict[str, MockAgent] | None = None,
        skill_agent_map: dict[str, MockAgent] | None = None,
        available_skills: set[str] | None = None,
    ):
        self._agents = agents or {}
        self._skill_agent_map = skill_agent_map or {}
        self._available_skills = available_skills or set(skill_agent_map.keys()) if skill_agent_map else set()
        self._created_skills: list[str] = []

    def get_agent(self, name: str) -> MockAgent | None:
        return self._agents.get(name)

    def list_agents(self) -> list[dict]:
        return [
            {"name": a.name, "agent_type": a.agent_type, "description": f"Mock agent {a.name}"}
            for a in self._agents.values()
        ]

    async def create_agent_from_skill(self, skill_name: str) -> MockAgent:
        self._created_skills.append(skill_name)
        if skill_name not in self._available_skills:
            raise RuntimeError(f"Skill '{skill_name}' not found in registry")
        if skill_name in self._skill_agent_map:
            agent = self._skill_agent_map[skill_name]
            self._agents[skill_name] = agent
            return agent
        # 不应到达这里，因为 available_skills 应包含所有可用的 skill
        raise RuntimeError(f"Skill '{skill_name}' not found in registry")


# --- Test Scenarios ---


class TestPlanExecutorAllSuccess:
    """3 个并行步骤全部成功 → 结果正确汇总"""

    @pytest.mark.asyncio
    async def test_three_parallel_steps_all_succeed(self):
        agent_a = MockAgent("web_search", {"data": "A"})
        agent_b = MockAgent("seo_analyzer", {"data": "B"})
        agent_c = MockAgent("report_generator", {"data": "C"})

        pool = MockAgentPool(
            skill_agent_map={
                "web_search": agent_a,
                "seo_analyzer": agent_b,
                "report_generator": agent_c,
            },
        )

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", parallel_group=0, required_skills=["web_search"]),
                PlanStep(step_id="s1", name="Analyze", description="Analyze", parallel_group=0, required_skills=["seo_analyzer"]),
                PlanStep(step_id="s2", name="Report", description="Report", parallel_group=0, required_skills=["report_generator"]),
            ],
            parallel_groups=[["s0", "s1", "s2"]],
        )

        executor = PlanExecutor(agent_pool=pool)
        result = await executor.execute(plan, make_task())

        assert result.status == TaskStatus.COMPLETED
        assert len(result.completed_steps) == 3
        assert len(result.failed_steps) == 0
        assert result.step_results["s0"].status == PlanStepStatus.COMPLETED
        assert result.step_results["s1"].status == PlanStepStatus.COMPLETED
        assert result.step_results["s2"].status == PlanStepStatus.COMPLETED

    @pytest.mark.asyncio
    async def test_parallel_groups_executed_sequentially(self):
        """并行组之间串行执行，组内并行"""
        agent_a = MockAgent("search", {"step": 1})
        agent_b = MockAgent("report", {"step": 2})

        pool = MockAgentPool(
            skill_agent_map={
                "web_search": agent_a,
                "report_generator": agent_b,
            },
        )

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", parallel_group=0, required_skills=["web_search"]),
                PlanStep(step_id="s1", name="Report", description="Report", dependencies=["s0"], parallel_group=1, required_skills=["report_generator"]),
            ],
            parallel_groups=[["s0"], ["s1"]],
        )

        executor = PlanExecutor(agent_pool=pool)
        result = await executor.execute(plan, make_task())

        assert result.status == TaskStatus.COMPLETED
        assert len(result.completed_steps) == 2
        # s1 应收到 s0 的依赖结果
        assert result.step_results["s1"].status == PlanStepStatus.COMPLETED

    @pytest.mark.asyncio
    async def test_dependency_results_injected(self):
        """依赖步骤的结果应注入到后续步骤的输入中"""
        agent_a = MockAgent("search", {"search_result": "data_A"})
        agent_b = MockAgent("report", {"report_result": "data_B"})

        pool = MockAgentPool(
            skill_agent_map={
                "web_search": agent_a,
                "report_generator": agent_b,
            },
        )

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", parallel_group=0, required_skills=["web_search"]),
                PlanStep(step_id="s1", name="Report", description="Report", dependencies=["s0"], parallel_group=1, required_skills=["report_generator"]),
            ],
            parallel_groups=[["s0"], ["s1"]],
        )

        executor = PlanExecutor(agent_pool=pool)
        result = await executor.execute(plan, make_task())

        assert result.status == TaskStatus.COMPLETED
        # 验证 s1 的输入中包含 dependency_results
        # 通过检查 agent_b 收到的 task 来验证
        assert result.step_results["s0"].result == {"search_result": "data_A"}


class TestPlanExecutorPartialFailure:
    """并行步骤中 1 个失败 → 其他步骤继续，失败步骤进入检查"""

    @pytest.mark.asyncio
    async def test_one_parallel_step_fails_others_continue(self):
        agent_a = MockAgent("search", {"data": "A"})
        agent_b = MockAgent("failing", should_fail=True)
        agent_c = MockAgent("report", {"data": "C"})

        pool = MockAgentPool(
            skill_agent_map={
                "web_search": agent_a,
                "failing_skill": agent_b,
                "report_generator": agent_c,
            },
        )

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", parallel_group=0, required_skills=["web_search"]),
                PlanStep(step_id="s1", name="Fail", description="Fail", parallel_group=0, required_skills=["failing_skill"]),
                PlanStep(step_id="s2", name="Report", description="Report", parallel_group=0, required_skills=["report_generator"]),
            ],
            parallel_groups=[["s0", "s1", "s2"]],
        )

        executor = PlanExecutor(agent_pool=pool, max_retries=0)
        result = await executor.execute(plan, make_task())

        # s0 和 s2 应成功
        assert result.step_results["s0"].status == PlanStepStatus.COMPLETED
        assert result.step_results["s2"].status == PlanStepStatus.COMPLETED
        # s1 应失败（默认策略是 SKIP）
        assert result.step_results["s1"].status == PlanStepStatus.SKIPPED
        assert len(result.completed_steps) == 2

    @pytest.mark.asyncio
    async def test_failed_step_skips_dependents(self):
        """失败步骤的依赖步骤应被跳过"""
        agent_a = MockAgent("failing", should_fail=True)
        agent_b = MockAgent("report", {"data": "B"})

        pool = MockAgentPool(
            skill_agent_map={
                "web_search": agent_a,
                "report_generator": agent_b,
            },
        )

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", parallel_group=0, required_skills=["web_search"]),
                PlanStep(step_id="s1", name="Report", description="Report", dependencies=["s0"], parallel_group=1, required_skills=["report_generator"]),
            ],
            parallel_groups=[["s0"], ["s1"]],
        )

        executor = PlanExecutor(agent_pool=pool, max_retries=0)
        result = await executor.execute(plan, make_task())

        # s0 失败后被跳过
        assert result.step_results["s0"].status == PlanStepStatus.SKIPPED
        # s1 因依赖失败也被跳过
        assert result.step_results["s1"].status == PlanStepStatus.SKIPPED
        assert len(result.skipped_steps) == 2


class TestPlanExecutorRetry:
    """步骤失败后自动重试成功"""

    @pytest.mark.asyncio
    async def test_retry_succeeds_after_initial_failure(self):
        """步骤首次失败后重试成功"""
        agent = MockAgent("search", {"data": "recovered"}, fail_count=1)

        pool = MockAgentPool(
            skill_agent_map={"web_search": agent},
        )

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", required_skills=["web_search"]),
            ],
            parallel_groups=[["s0"]],
        )

        executor = PlanExecutor(agent_pool=pool, max_retries=2)
        result = await executor.execute(plan, make_task())

        assert result.step_results["s0"].status == PlanStepStatus.COMPLETED
        assert result.step_results["s0"].retry_count == 1
        assert result.step_results["s0"].result == {"data": "recovered"}

    @pytest.mark.asyncio
    async def test_retry_exhausted_step_fails(self):
        """重试次数耗尽后步骤标记为失败"""
        agent = MockAgent("failing", should_fail=True)

        pool = MockAgentPool(
            skill_agent_map={"web_search": agent},
        )

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", required_skills=["web_search"]),
            ],
            parallel_groups=[["s0"]],
        )

        executor = PlanExecutor(agent_pool=pool, max_retries=2)
        result = await executor.execute(plan, make_task())

        assert result.step_results["s0"].status == PlanStepStatus.SKIPPED  # 默认策略是 SKIP
        assert result.step_results["s0"].retry_count == 2

    @pytest.mark.asyncio
    async def test_zero_retries_no_retry(self):
        """max_retries=0 时不重试"""
        agent = MockAgent("failing", should_fail=True)

        pool = MockAgentPool(
            skill_agent_map={"web_search": agent},
        )

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", required_skills=["web_search"]),
            ],
            parallel_groups=[["s0"]],
        )

        executor = PlanExecutor(agent_pool=pool, max_retries=0)
        result = await executor.execute(plan, make_task())

        assert result.step_results["s0"].retry_count == 0


class TestPlanExecutorPlanAdjustment:
    """步骤失败后调整计划（跳过/替换）继续执行"""

    @pytest.mark.asyncio
    async def test_skip_action_skips_step(self):
        """on_step_failed 返回 SKIP 时跳过步骤"""
        agent = MockAgent("failing", should_fail=True)

        pool = MockAgentPool(
            skill_agent_map={"web_search": agent},
        )

        async def on_failed(step, result):
            return FailureAction.SKIP

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", required_skills=["web_search"]),
            ],
            parallel_groups=[["s0"]],
        )

        executor = PlanExecutor(agent_pool=pool, max_retries=0, on_step_failed=on_failed)
        result = await executor.execute(plan, make_task())

        assert result.step_results["s0"].status == PlanStepStatus.SKIPPED
        assert result.adjusted is True

    @pytest.mark.asyncio
    async def test_replace_action_skips_original(self):
        """on_step_failed 返回 REPLACE 时标记原步骤为 SKIPPED"""
        agent = MockAgent("failing", should_fail=True)

        pool = MockAgentPool(
            skill_agent_map={"web_search": agent},
        )

        async def on_failed(step, result):
            return FailureAction.REPLACE

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", required_skills=["web_search"]),
            ],
            parallel_groups=[["s0"]],
        )

        executor = PlanExecutor(agent_pool=pool, max_retries=0, on_step_failed=on_failed)
        result = await executor.execute(plan, make_task())

        assert result.step_results["s0"].status == PlanStepStatus.SKIPPED
        assert result.adjusted is True

    @pytest.mark.asyncio
    async def test_abort_action_skips_all_remaining(self):
        """on_step_failed 返回 ABORT 时跳过所有剩余步骤"""
        agent_a = MockAgent("failing", should_fail=True)
        agent_b = MockAgent("report", {"data": "B"})

        pool = MockAgentPool(
            skill_agent_map={
                "web_search": agent_a,
                "report_generator": agent_b,
            },
        )

        async def on_failed(step, result):
            return FailureAction.ABORT

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", required_skills=["web_search"]),
                PlanStep(step_id="s1", name="Report", description="Report", required_skills=["report_generator"]),
            ],
            parallel_groups=[["s0"], ["s1"]],
        )

        executor = PlanExecutor(agent_pool=pool, max_retries=0, on_step_failed=on_failed)
        result = await executor.execute(plan, make_task())

        # The failed step should remain FAILED (not SKIPPED)
        assert result.step_results["s0"].status == PlanStepStatus.FAILED
        # Remaining steps should be SKIPPED
        assert result.step_results["s1"].status == PlanStepStatus.SKIPPED
        assert result.adjusted is True


class TestPlanExecutorHumanIntervention:
    """执行中请求人工介入后继续"""

    @pytest.mark.asyncio
    async def test_human_intervention_then_skip(self):
        """人工介入后选择跳过"""
        agent = MockAgent("failing", should_fail=True)

        pool = MockAgentPool(
            skill_agent_map={"web_search": agent},
        )

        async def on_failed(step, result):
            return FailureAction.REQUEST_HUMAN

        async def on_human(step, result):
            return FailureAction.SKIP

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", required_skills=["web_search"]),
            ],
            parallel_groups=[["s0"]],
        )

        executor = PlanExecutor(
            agent_pool=pool,
            max_retries=0,
            on_step_failed=on_failed,
            on_human_intervention=on_human,
        )
        result = await executor.execute(plan, make_task())

        assert result.step_results["s0"].status == PlanStepStatus.SKIPPED
        assert result.human_intervention_requested is True
        assert result.adjusted is True

    @pytest.mark.asyncio
    async def test_human_intervention_without_callback(self):
        """请求人工介入但无回调时标记 human_intervention"""
        agent = MockAgent("failing", should_fail=True)

        pool = MockAgentPool(
            skill_agent_map={"web_search": agent},
        )

        async def on_failed(step, result):
            return FailureAction.REQUEST_HUMAN

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", required_skills=["web_search"]),
            ],
            parallel_groups=[["s0"]],
        )

        executor = PlanExecutor(
            agent_pool=pool,
            max_retries=0,
            on_step_failed=on_failed,
        )
        result = await executor.execute(plan, make_task())

        assert result.human_intervention_requested is True


class TestPlanExecutorStepTimeout:
    """步骤超时处理"""

    @pytest.mark.asyncio
    async def test_step_timeout_fails_step(self):
        """步骤超时后标记为失败"""

        class SlowAgent:
            name = "slow_agent"
            agent_type = "mock"

            async def execute(self, task):
                await asyncio.sleep(10)  # 模拟超时

        pool = MockAgentPool(
            skill_agent_map={"web_search": SlowAgent()},
        )

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", required_skills=["web_search"]),
            ],
            parallel_groups=[["s0"]],
        )

        executor = PlanExecutor(agent_pool=pool, max_retries=0, step_timeout=0.1)
        result = await executor.execute(plan, make_task())

        assert result.step_results["s0"].status == PlanStepStatus.SKIPPED
        assert "timed out" in (result.step_results["s0"].error or "")


class TestPlanExecutorNoAgentAvailable:
    """Agent 不可用时的处理"""

    @pytest.mark.asyncio
    async def test_no_agent_for_step(self):
        """步骤所需 Agent 不可用时步骤失败"""
        pool = MockAgentPool(skill_agent_map={})

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", required_skills=["nonexistent_skill"]),
            ],
            parallel_groups=[["s0"]],
        )

        executor = PlanExecutor(agent_pool=pool, max_retries=0)
        result = await executor.execute(plan, make_task())

        assert result.step_results["s0"].status == PlanStepStatus.SKIPPED
        assert "No agent available" in (result.step_results["s0"].error or "")


class TestPlanExecutorStepComplete:
    """步骤完成回调"""

    @pytest.mark.asyncio
    async def test_on_step_complete_callback_called(self):
        """步骤完成后调用回调"""
        agent = MockAgent("search", {"data": "A"})
        pool = MockAgentPool(skill_agent_map={"web_search": agent})

        completed_steps: list[tuple[str, PlanStepStatus]] = []

        async def on_complete(step, result):
            completed_steps.append((step.step_id, result.status))

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", required_skills=["web_search"]),
            ],
            parallel_groups=[["s0"]],
        )

        executor = PlanExecutor(agent_pool=pool, on_step_complete=on_complete)
        result = await executor.execute(plan, make_task())

        assert len(completed_steps) == 1
        assert completed_steps[0] == ("s0", PlanStepStatus.COMPLETED)


class TestPlanExecutionResult:
    """PlanExecutionResult 属性测试"""

    def test_completed_steps(self):
        result = PlanExecutionResult(
            plan_id="p1",
            step_results={
                "s0": StepExecutionResult(step_id="s0", status=PlanStepStatus.COMPLETED, result={"a": 1}),
                "s1": StepExecutionResult(step_id="s1", status=PlanStepStatus.FAILED, error="err"),
                "s2": StepExecutionResult(step_id="s2", status=PlanStepStatus.SKIPPED, error="skip"),
            },
            status=TaskStatus.COMPLETED,
            total_duration_ms=100.0,
        )
        assert result.completed_steps == ["s0"]
        assert result.failed_steps == ["s1"]
        assert result.skipped_steps == ["s2"]

    def test_empty_results(self):
        result = PlanExecutionResult(
            plan_id="p1",
            step_results={},
            status=TaskStatus.COMPLETED,
            total_duration_ms=0.0,
        )
        assert result.completed_steps == []
        assert result.failed_steps == []
        assert result.skipped_steps == []


class TestPlanExecutorOverallStatus:
    """整体状态判定"""

    @pytest.mark.asyncio
    async def test_all_completed(self):
        agent = MockAgent("search", {"data": "A"})
        pool = MockAgentPool(skill_agent_map={"web_search": agent})

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", required_skills=["web_search"]),
            ],
            parallel_groups=[["s0"]],
        )

        executor = PlanExecutor(agent_pool=pool)
        result = await executor.execute(plan, make_task())

        assert result.status == TaskStatus.COMPLETED

    @pytest.mark.asyncio
    async def test_all_skipped(self):
        agent = MockAgent("failing", should_fail=True)
        pool = MockAgentPool(skill_agent_map={"web_search": agent})

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", required_skills=["web_search"]),
            ],
            parallel_groups=[["s0"]],
        )

        executor = PlanExecutor(agent_pool=pool, max_retries=0)
        result = await executor.execute(plan, make_task())

        # 默认策略 SKIP → 全部跳过 → FAILED（没有实际完成的步骤）
        assert result.status == TaskStatus.FAILED

    @pytest.mark.asyncio
    async def test_empty_plan(self):
        pool = MockAgentPool()
        plan = make_plan(steps=[], parallel_groups=[])

        executor = PlanExecutor(agent_pool=pool)
        result = await executor.execute(plan, make_task())

        assert result.status == TaskStatus.COMPLETED
        assert len(result.step_results) == 0


class TestPlanExecutorStepStateMachine:
    """步骤状态机：PENDING → RUNNING → COMPLETED/FAILED"""

    @pytest.mark.asyncio
    async def test_step_transitions_to_running_then_completed(self):
        """步骤状态正确转换"""
        agent = MockAgent("search", {"data": "A"})
        pool = MockAgentPool(skill_agent_map={"web_search": agent})

        step = PlanStep(step_id="s0", name="Search", description="Search", required_skills=["web_search"])
        plan = make_plan(steps=[step], parallel_groups=[["s0"]])

        executor = PlanExecutor(agent_pool=pool)
        result = await executor.execute(plan, make_task())

        # 执行完成后步骤状态应为 COMPLETED
        assert step.status == PlanStepStatus.COMPLETED

    @pytest.mark.asyncio
    async def test_step_transitions_to_failed_on_error(self):
        agent = MockAgent("failing", should_fail=True)
        pool = MockAgentPool(skill_agent_map={"web_search": agent})

        step = PlanStep(step_id="s0", name="Search", description="Search", required_skills=["web_search"])
        plan = make_plan(steps=[step], parallel_groups=[["s0"]])

        executor = PlanExecutor(agent_pool=pool, max_retries=0)
        result = await executor.execute(plan, make_task())

        # 默认策略 SKIP → 步骤最终状态为 SKIPPED
        assert step.status == PlanStepStatus.SKIPPED


class TestPlanExecutorDependencyInjection:
    """依赖结果注入"""

    @pytest.mark.asyncio
    async def test_dependency_results_format(self):
        """依赖结果格式正确"""
        agent_a = MockAgent("search", {"search_data": "result_A"})
        agent_b = MockAgent("report", {"report_data": "result_B"})

        received_inputs: list[dict] = []

        class CapturingAgent:
            name = "report_agent"
            agent_type = "mock"

            async def execute(self, task: TaskMessage) -> TaskResult:
                received_inputs.append(dict(task.input_data))
                now = datetime.now(timezone.utc)
                return TaskResult(
                    task_id=task.task_id,
                    agent_name=self.name,
                    status=TaskStatus.COMPLETED,
                    output_data={"report": "done"},
                    error_message=None,
                    started_at=now,
                    completed_at=now,
                )

        pool = MockAgentPool(
            skill_agent_map={
                "web_search": agent_a,
                "report_generator": CapturingAgent(),
            },
        )

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", parallel_group=0, required_skills=["web_search"]),
                PlanStep(step_id="s1", name="Report", description="Report", dependencies=["s0"], parallel_group=1, required_skills=["report_generator"]),
            ],
            parallel_groups=[["s0"], ["s1"]],
        )

        executor = PlanExecutor(agent_pool=pool)
        result = await executor.execute(plan, make_task())

        # s1 的输入应包含 dependency_results
        assert len(received_inputs) == 1
        assert "dependency_results" in received_inputs[0]
        assert "s0" in received_inputs[0]["dependency_results"]
        assert received_inputs[0]["dependency_results"]["s0"]["status"] == "completed"

    @pytest.mark.asyncio
    async def test_step_metadata_in_input(self):
        """步骤元信息应注入到输入中"""
        agent = MockAgent("search", {"data": "A"})
        pool = MockAgentPool(skill_agent_map={"web_search": agent})

        received_inputs: list[dict] = []

        class CapturingAgent:
            name = "search_agent"
            agent_type = "mock"

            async def execute(self, task: TaskMessage) -> TaskResult:
                received_inputs.append(dict(task.input_data))
                now = datetime.now(timezone.utc)
                return TaskResult(
                    task_id=task.task_id,
                    agent_name=self.name,
                    status=TaskStatus.COMPLETED,
                    output_data={"result": "done"},
                    error_message=None,
                    started_at=now,
                    completed_at=now,
                )

        pool2 = MockAgentPool(
            skill_agent_map={"web_search": CapturingAgent()},
        )

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search Step", description="Do the search", required_skills=["web_search"]),
            ],
            parallel_groups=[["s0"]],
        )

        executor = PlanExecutor(agent_pool=pool2)
        result = await executor.execute(plan, make_task())

        assert len(received_inputs) == 1
        assert received_inputs[0]["step_name"] == "Search Step"
        assert received_inputs[0]["step_description"] == "Do the search"


class TestPlanExecutorExistingAgent:
    """通过 get_agent 获取已有 Agent"""

    @pytest.mark.asyncio
    async def test_fallback_to_existing_agent(self):
        """Skill 创建失败时回退到池中已有 Agent"""
        agent = MockAgent("existing_agent", {"data": "existing"})

        pool = MockAgentPool(
            agents={"s0": agent},  # 按步骤 ID 注册
            skill_agent_map={},  # 无 Skill 映射
            available_skills=set(),  # 无可用 Skill
        )

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", required_skills=["web_search"]),
            ],
            parallel_groups=[["s0"]],
        )

        executor = PlanExecutor(agent_pool=pool, max_retries=0)
        result = await executor.execute(plan, make_task())

        assert result.step_results["s0"].status == PlanStepStatus.COMPLETED
        assert result.step_results["s0"].result == {"data": "existing"}

    @pytest.mark.asyncio
    async def test_fallback_to_skill_name_agent(self):
        """Skill 创建失败时尝试用 Skill 名称获取 Agent"""
        agent = MockAgent("web_search", {"data": "by_skill_name"})

        pool = MockAgentPool(
            agents={"web_search": agent},
            skill_agent_map={},  # create_agent_from_skill 会失败
            available_skills=set(),  # 无可用 Skill
        )

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", required_skills=["web_search"]),
            ],
            parallel_groups=[["s0"]],
        )

        executor = PlanExecutor(agent_pool=pool, max_retries=0)
        result = await executor.execute(plan, make_task())

        assert result.step_results["s0"].status == PlanStepStatus.COMPLETED


class TestPlanExecutorCascadingSkip:
    """级联跳过：失败步骤的依赖链全部跳过"""

    @pytest.mark.asyncio
    async def test_cascading_skip(self):
        agent_a = MockAgent("failing", should_fail=True)
        agent_b = MockAgent("report", {"data": "B"})
        agent_c = MockAgent("final", {"data": "C"})

        pool = MockAgentPool(
            skill_agent_map={
                "web_search": agent_a,
                "report_generator": agent_b,
                "final_skill": agent_c,
            },
        )

        plan = make_plan(
            steps=[
                PlanStep(step_id="s0", name="Search", description="Search", parallel_group=0, required_skills=["web_search"]),
                PlanStep(step_id="s1", name="Report", description="Report", dependencies=["s0"], parallel_group=1, required_skills=["report_generator"]),
                PlanStep(step_id="s2", name="Final", description="Final", dependencies=["s1"], parallel_group=2, required_skills=["final_skill"]),
            ],
            parallel_groups=[["s0"], ["s1"], ["s2"]],
        )

        executor = PlanExecutor(agent_pool=pool, max_retries=0)
        result = await executor.execute(plan, make_task())

        # s0 失败 → s1 和 s2 都应被跳过
        assert result.step_results["s0"].status == PlanStepStatus.SKIPPED
        assert result.step_results["s1"].status == PlanStepStatus.SKIPPED
        assert result.step_results["s2"].status == PlanStepStatus.SKIPPED
        assert "failed dependency" in (result.step_results["s1"].error or "")
        assert "failed dependency" in (result.step_results["s2"].error or "")
