from threading import Event, Lock, Timer

from mulit_agent_app.application.task_executor import TaskExecutor
from mulit_agent_app.domain.attachments import TaskAttachment
from mulit_agent_app.infrastructure.database import AgentDatabase


def test_executor_reports_a_single_completed_lifecycle(tmp_path):
    database = AgentDatabase(tmp_path / "agent.sqlite3")
    executor = TaskExecutor(database)
    events: list[dict[str, object]] = []
    completed = Event()

    def capture(event: object) -> None:
        assert isinstance(event, dict)
        events.append(event)
        if event.get("status") == "completed":
            completed.set()

    executor.task_event.connect(capture)
    executor.submit("task-1", "确认 Agent 回传结果")

    assert completed.wait(3)
    assert [event["status"] for event in events] == ["accepted", "running", "completed"]
    assert database.task_status("task-1") == "completed"


def test_executor_cancels_and_times_out_active_tasks(tmp_path):
    database = AgentDatabase(tmp_path / "agent.sqlite3")
    executor = TaskExecutor(database)
    events: list[dict[str, object]] = []
    terminal_events = Event()

    def capture(event: object) -> None:
        assert isinstance(event, dict)
        events.append(event)
        if len([item for item in events if item.get("status") in {"cancelled", "timed_out"}]) == 2:
            terminal_events.set()

    executor.task_event.connect(capture)
    executor.submit("cancel-task", "应被取消")
    Timer(0.1, lambda: executor.cancel("cancel-task", "测试取消")).start()
    executor.submit("timeout-task", "应超时", "2000-01-01T00:00:00+00:00")

    assert terminal_events.wait(1)
    assert database.task_status("cancel-task") == "cancelled"
    assert database.task_status("timeout-task") == "timed_out"
    assert [event["status"] for event in events if event["task_id"] == "cancel-task"][-1] == "cancelled"
    assert [event["status"] for event in events if event["task_id"] == "timeout-task"][-1] == "timed_out"


def test_executor_uses_the_configured_model_runner(tmp_path):
    database = AgentDatabase(tmp_path / "agent.sqlite3")
    executor = TaskExecutor(database, lambda instruction: f"真实模型输出：{instruction}")
    completed = Event()
    events: list[dict[str, object]] = []

    def capture(event: object) -> None:
        assert isinstance(event, dict)
        events.append(event)
        if event.get("status") == "completed":
            completed.set()

    executor.task_event.connect(capture)
    executor.submit("model-task", "给出一句总结")

    assert completed.wait(2)
    assert events[-1]["output"] == "真实模型输出：给出一句总结"
    assert events[-1]["execution"] == "llm"


def test_executor_keeps_each_controller_task_independent(tmp_path):
    received: list[str] = []
    database = AgentDatabase(tmp_path / "agent.sqlite3")
    executor = TaskExecutor(database, lambda instruction: received.append(instruction) or "任务结果")
    completed = Event()
    executor.task_event.connect(
        lambda event: completed.set()
        if isinstance(event, dict) and event.get("status") == "completed"
        else None
    )

    executor.submit("design-direct-task", "整理蓝色方案")

    assert completed.wait(2)
    assert received == ["整理蓝色方案"]
    assert database.task_status("design-direct-task") == "completed"


def test_executor_runs_independent_tasks_concurrently(tmp_path):
    database = AgentDatabase(tmp_path / "agent.sqlite3")
    both_started = Event()
    release = Event()
    completed = Event()
    lock = Lock()
    started = 0
    terminal = 0

    def runner(instruction: str) -> str:
        nonlocal started
        with lock:
            started += 1
            if started == 2:
                both_started.set()
        assert release.wait(2)
        return f"完成：{instruction}"

    executor = TaskExecutor(database, runner)

    def capture(event: object) -> None:
        nonlocal terminal
        if isinstance(event, dict) and event.get("status") == "completed":
            with lock:
                terminal += 1
                if terminal == 2:
                    completed.set()

    executor.task_event.connect(capture)
    executor.submit("parallel-one", "第一条")
    executor.submit("parallel-two", "第二条")

    assert both_started.wait(1), "两个独立任务应同时进入模型执行"
    release.set()
    assert completed.wait(2)
    assert database.task_status("parallel-one") == "completed"
    assert database.task_status("parallel-two") == "completed"


def test_executor_marks_desktop_filesystem_execution(tmp_path):
    database = AgentDatabase(tmp_path / "agent.sqlite3")
    executor = TaskExecutor(database)
    executor.set_full_access_enabled(False)
    executor.set_desktop_filesystem_enabled(True)
    executor._desktop_tool._desktop_path = tmp_path / "desktop"
    completed = Event()
    events: list[dict[str, object]] = []

    def capture(event: object) -> None:
        assert isinstance(event, dict)
        events.append(event)
        if event.get("status") == "completed":
            completed.set()

    executor.task_event.connect(capture)
    executor.submit("tool-task", "在电脑桌面新建 test 文件夹")

    assert completed.wait(2)
    assert events[-1]["execution"] == "desktop-filesystem"


def test_executor_defaults_to_full_access_tool_runner(tmp_path):
    database = AgentDatabase(tmp_path / "agent.sqlite3")
    calls: list[str] = []

    def runner(_instruction, _attachments, run_local):
        calls.append(run_local("Write-Output full-access-task"))
        return "已根据真实本机命令完成。"

    executor = TaskExecutor(database, runner)
    completed = Event()
    events: list[dict[str, object]] = []

    def capture(event: object) -> None:
        assert isinstance(event, dict)
        events.append(event)
        if event.get("status") == "completed":
            completed.set()

    executor.task_event.connect(capture)
    executor.submit("full-access-task", "在本机执行一项操作")

    assert completed.wait(5)
    assert calls == ["full-access-task"]
    assert events[-1]["execution"] == "full-access"


def test_executor_rejects_prose_only_answer_for_a_local_action(tmp_path):
    database = AgentDatabase(tmp_path / "agent.sqlite3")
    executor = TaskExecutor(database, lambda _instruction: "我会在桌面创建文件夹。")
    failed = Event()
    events: list[dict[str, object]] = []

    def capture(event: object) -> None:
        assert isinstance(event, dict)
        events.append(event)
        if event.get("status") == "failed":
            failed.set()

    executor.task_event.connect(capture)
    executor.submit("no-tool-action", "在桌面创建一个 test 文件夹")

    assert failed.wait(3)
    assert "未执行任何操作" in str(events[-1]["error"])


def test_executor_does_not_fall_back_to_llm_when_desktop_tool_is_not_applied(tmp_path):
    database = AgentDatabase(tmp_path / "agent.sqlite3")
    calls: list[str] = []
    executor = TaskExecutor(database, lambda instruction: calls.append(instruction) or "不应调用模型")
    executor.set_full_access_enabled(False)
    completed = Event()
    events: list[dict[str, object]] = []

    def capture(event: object) -> None:
        assert isinstance(event, dict)
        events.append(event)
        if event.get("status") == "failed":
            completed.set()

    executor.task_event.connect(capture)
    executor.submit("missing-tool", "在电脑桌面新建 missing-tool-test 文件夹")

    assert completed.wait(2)
    assert calls == []
    assert events[-1]["execution"] == "desktop-filesystem"
    assert "未部署或尚未应用" in str(events[-1]["error"])


def test_executor_streams_tool_progress_while_the_task_runs(tmp_path):
    database = AgentDatabase(tmp_path / "agent.sqlite3")

    def runner(_instruction, _attachments, toolbox):
        return f"结果：{toolbox.call('computer_time', {'timezone': 'UTC'})}"

    executor = TaskExecutor(database, runner)
    completed = Event()
    events: list[dict[str, object]] = []

    def capture(event: object) -> None:
        assert isinstance(event, dict)
        events.append(event)
        if event.get("status") == "completed":
            completed.set()

    executor.task_event.connect(capture)
    executor.submit("progress-task", "看看现在几点")

    assert completed.wait(10)
    progress = [str(event["message"]) for event in events if event.get("status") == "progress"]
    assert any(message.startswith("调用 computer_time") for message in progress), progress
    assert any(message.startswith("computer_time 返回：") for message in progress), progress
    # Progress must arrive before the terminal event so the UI can stream it live.
    statuses = [event.get("status") for event in events]
    assert statuses.index("progress") < statuses.index("completed")


def test_executor_ignores_progress_after_the_task_ended(tmp_path):
    database = AgentDatabase(tmp_path / "agent.sqlite3")
    executor = TaskExecutor(database, lambda instruction: "完成")
    completed = Event()
    events: list[dict[str, object]] = []

    def capture(event: object) -> None:
        assert isinstance(event, dict)
        events.append(event)
        if event.get("status") == "completed":
            completed.set()

    executor.task_event.connect(capture)
    executor.submit("finished-task", "总结")
    assert completed.wait(5)
    reported = len(events)

    executor._report_progress("finished-task")("迟到的进展")

    assert len(events) == reported, "结束后的任务不应再产生中间步骤"


def test_late_model_result_cannot_overwrite_a_cancelled_terminal_state(tmp_path):
    database = AgentDatabase(tmp_path / "agent.sqlite3")
    started, release, returned, cancelled = Event(), Event(), Event(), Event()

    def runner(_instruction: str) -> str:
        started.set()
        assert release.wait(2)
        returned.set()
        return "这是取消后才返回的模型结果"

    executor = TaskExecutor(database, runner)
    executor.task_event.connect(
        lambda event: cancelled.set()
        if isinstance(event, dict) and event.get("status") == "cancelled"
        else None
    )
    executor.submit("late-result", "等待模型返回")
    assert started.wait(2)
    executor.cancel("late-result", "测试取消")
    assert cancelled.wait(1)
    release.set()
    assert returned.wait(1)
    assert database.task_status("late-result") == "cancelled"


def test_executor_passes_the_conversation_history_to_the_provider(tmp_path):
    database = AgentDatabase(tmp_path / "agent.sqlite3")
    received: list[tuple[tuple[str, str], ...]] = []

    def runner(_instruction, _attachments, _toolbox, history):
        received.append(history)
        return "已复制到 D:\\target"

    executor = TaskExecutor(database, runner)
    completed = Event()
    executor.task_event.connect(
        lambda event: completed.set() if isinstance(event, dict) and event.get("status") == "completed" else None
    )

    executor.submit(
        "follow-up-task",
        "把它复制到 D:\\target",
        history=(("user", "下载到 D:\\demo"), ("assistant", "已下载到 D:\\demo\\a.zip")),
    )

    assert completed.wait(5)
    assert received == [(("user", "下载到 D:\\demo"), ("assistant", "已下载到 D:\\demo\\a.zip"))]


def test_image_task_never_falls_back_to_text_only_model(tmp_path):
    database = AgentDatabase(tmp_path / "agent.sqlite3")
    calls: list[str] = []
    executor = TaskExecutor(database, lambda instruction: calls.append(instruction) or "不应调用")
    finished = Event()
    events: list[dict[str, object]] = []
    executor.task_event.connect(lambda event: (events.append(event), finished.set()) if isinstance(event, dict) and event.get("status") == "failed" else None)
    executor.submit("image-without-vision", "分析图片", attachments=(TaskAttachment("x.png", "image/png", "data:image/png;base64,aQ=="),))
    assert finished.wait(2)
    assert calls == []
    assert "未被降级" in str(events[-1]["error"])