"""会商「模型自动容错（换下一个模型）」测试。

覆盖：
  - 中间件兜底/熔断 AIMessage 带可识别标记 + ``is_llm_error_message`` 检测（标记 + 正文回退）；
  - ``fallback_model_chain`` 按 config.yaml models[] 顺序排候选（当前优先、去重、开关关闭=单个）；
  - ``reason_text`` 文案；
  - 引擎 Step2 单席位异常隔离（一个席位抛错不拖垮整链/整批，仅置空交付）；
  - 网关带回的模型容错记录（switches）→ 引擎在时间线插「系统提示」气泡。

纯单元/纯逻辑：引擎用 JobProgressWriter(store=None) 干跑 + 假网关，不需要 DB / 真模型。
"""

from __future__ import annotations

import asyncio
import json
from contextlib import asynccontextmanager
from types import SimpleNamespace

import pytest

import app.gateway.roundtable_model_fallback as mf
import app.gateway.routers.multi_agent as multi_agent
import app.gateway.routers.parallel_agents as parallel_agents
from app.gateway.roundtable_model_fallback import (
    fallback_model_chain,
    is_llm_error_message,
    reason_text,
)
from app.gateway.routers._strategy_selection_seed import (
    STRATEGY_INTENT_AGENT_ID,
    STRATEGY_JUDGE_AGENT_ID,
    STRATEGY_SECURITY_AGENT_ID,
    STRATEGY_SOLUTION_AGENT_ID,
    default_strategy_agent_roles,
)
from deerflow.agents.middlewares.llm_error_handling_middleware import (
    LLM_ERROR_MARKER,
    LLMErrorHandlingMiddleware,
)
from deerflow.agents.middlewares.llm_error_handling_middleware import (
    is_llm_error_message as harness_is_llm_error_message,
)
from deerflow.agents.roundtable_orchestrator import (
    JobProgressWriter,
    SeatRef,
    run_orchestration,
)
from deerflow.agents.roundtable_orchestrator.types import LeaderTurn, ModelSwitch, SeatTurn

# ── is_llm_error_message ─────────────────────────────────────────────────────


def test_is_llm_error_message_marker_dict() -> None:
    msg = {"type": "ai", "content": "随便什么", "additional_kwargs": {LLM_ERROR_MARKER: "busy"}}
    assert is_llm_error_message(msg) == "busy"
    assert harness_is_llm_error_message(msg) == "busy"


def test_is_llm_error_message_content_fallback() -> None:
    # additional_kwargs 被剥离时，按已知兜底文案正文回退识别。
    msg = {"type": "ai", "content": "大模型本次调用未能成功完成，请稍后再试或更换模型。"}
    assert is_llm_error_message(msg) == "generic"


def test_is_llm_error_message_normal_is_none() -> None:
    assert is_llm_error_message({"type": "ai", "content": "这是正常的席位交付内容"}) is None
    assert is_llm_error_message(None) is None
    assert is_llm_error_message({"type": "ai", "content": ""}) is None


def test_parallel_agent_run_raises_instead_of_completing_llm_fallback(monkeypatch) -> None:
    fallback_message = {
        "type": "ai",
        "content": "大模型本次调用未能成功完成，请稍后再试或更换模型。",
    }

    async def fake_stream_upstream(*_args, **_kwargs):
        yield None, [f"data: {json.dumps({'messages': [fallback_message]}, ensure_ascii=False)}"]

    monkeypatch.setattr(parallel_agents, "_stream_upstream", fake_stream_upstream)

    async def collect() -> list[object]:
        return [
            item
            async for item in parallel_agents._run_one_agent(
                None,  # type: ignore[arg-type]
                "http://gateway.test",
                {"Authorization": "Bearer test"},
                "strategy-selection-intent",
                "thread-id",
                "test task",
                "",
            )
        ]

    try:
        asyncio.run(collect())
    except RuntimeError as exc:
        assert "模型调用失败" in str(exc)
    else:
        raise AssertionError("LLM fallback message must not be emitted as agent_done")


def test_parallel_agent_run_rejects_punctuation_only_delivery(monkeypatch) -> None:
    invalid_message = {"type": "ai", "content": "!" * 500}

    async def fake_stream_upstream(*_args, **_kwargs):
        yield None, [f"data: {json.dumps({'messages': [invalid_message]}, ensure_ascii=False)}"]

    monkeypatch.setattr(parallel_agents, "_stream_upstream", fake_stream_upstream)

    async def collect() -> list[object]:
        return [
            item
            async for item in parallel_agents._run_one_agent(
                None,  # type: ignore[arg-type]
                "http://gateway.test",
                {"Authorization": "Bearer test"},
                "strategy-selection-angle-security",
                "thread-id",
                "test task",
                "",
            )
        ]

    try:
        asyncio.run(collect())
    except RuntimeError as exc:
        assert "未生成有效正文" in str(exc)
    else:
        raise AssertionError("Punctuation-only output must not be emitted as agent_done")


def test_parallel_agent_run_allows_empty_content_clarification(monkeypatch) -> None:
    clarification_message = {
        "type": "ai",
        "content": "",
        "tool_calls": [
            {
                "id": "clarify-1",
                "name": "ask_clarification",
                "args": {
                    "question": "请确认策略时间范围。",
                    "clarification_type": "single_choice",
                    "options": ["短期", "长期"],
                    "allow_custom": True,
                },
            }
        ],
    }

    async def fake_stream_upstream(*_args, **_kwargs):
        yield None, [f"data: {json.dumps({'messages': [clarification_message]}, ensure_ascii=False)}"]

    monkeypatch.setattr(parallel_agents, "_stream_upstream", fake_stream_upstream)

    async def collect() -> list[object]:
        return [
            item
            async for item in parallel_agents._run_one_agent(
                None,  # type: ignore[arg-type]
                "http://gateway.test",
                {"Authorization": "Bearer test"},
                "strategy-selection-intent",
                "thread-id",
                "test task",
                "",
            )
        ]

    events = asyncio.run(collect())

    assert len(events) == 1
    event = events[0]
    assert isinstance(event, dict)
    assert event["type"] == "clarification"
    assert event["content"] == ""
    assert event["clarification"]["question"] == "请确认策略时间范围。"


def test_parallel_agent_final_values_keep_typed_reasoning_out_of_content(monkeypatch) -> None:
    final_message = {
        "type": "ai",
        "content": [
            {"type": "thinking", "text": "核对约束与证据"},
            {"type": "text", "text": "这是最终可见策略正文，包含目标、路径和验收标准。"},
        ],
    }

    async def fake_stream_upstream(*_args, **_kwargs):
        yield None, [f"data: {json.dumps({'messages': [final_message]}, ensure_ascii=False)}"]

    monkeypatch.setattr(parallel_agents, "_stream_upstream", fake_stream_upstream)

    async def collect() -> list[object]:
        return [
            item
            async for item in parallel_agents._run_one_agent(
                None,  # type: ignore[arg-type]
                "http://gateway.test",
                {"Authorization": "Bearer test"},
                "strategy-selection-angle-security",
                "thread-id",
                "test task",
                "",
            )
        ]

    events = asyncio.run(collect())

    assert len(events) == 1
    event = events[0]
    assert isinstance(event, dict)
    assert event["content"] == "这是最终可见策略正文，包含目标、路径和验收标准。"
    assert event["reasoning"] == "核对约束与证据"


def test_parallel_agent_final_values_recovers_template_injected_reasoning(monkeypatch) -> None:
    final_message = {
        "type": "ai",
        "content": "模板注入的内部推理</think>这是最终可见策略正文，包含目标、路径和验收标准。",
    }

    async def fake_stream_upstream(*_args, **_kwargs):
        yield None, [f"data: {json.dumps({'messages': [final_message]}, ensure_ascii=False)}"]

    monkeypatch.setattr(parallel_agents, "_stream_upstream", fake_stream_upstream)

    async def collect() -> list[object]:
        return [
            item
            async for item in parallel_agents._run_one_agent(
                None,  # type: ignore[arg-type]
                "http://gateway.test",
                {"Authorization": "Bearer test"},
                "strategy-selection-angle-security",
                "thread-id",
                "test task",
                "",
            )
        ]

    events = asyncio.run(collect())

    event = events[0]
    assert isinstance(event, dict)
    assert event["content"] == "这是最终可见策略正文，包含目标、路径和验收标准。"
    assert event["reasoning"] == "模板注入的内部推理"


def test_parallel_agent_substantive_content_accepts_chinese_and_english() -> None:
    assert parallel_agents._has_substantive_content("这是一份包含目标、前置条件、阶段路径和验收标准的完整策略。")
    assert parallel_agents._has_substantive_content("A complete strategy with objectives, prerequisites, milestones, and measurable outcomes.")
    assert not parallel_agents._has_substantive_content("<think>internal reasoning only</think>!!!")


def test_strategy_stage_tool_surface_allows_clarification_only_for_intent() -> None:
    assert "ask_clarification" not in parallel_agents._excluded_tools_for_agent(STRATEGY_INTENT_AGENT_ID)

    for agent_id in (
        STRATEGY_SECURITY_AGENT_ID,
        STRATEGY_JUDGE_AGENT_ID,
        STRATEGY_SOLUTION_AGENT_ID,
    ):
        assert "ask_clarification" in parallel_agents._excluded_tools_for_agent(agent_id)


def test_strategy_task_message_matches_stage_clarification_contract() -> None:
    judge_message = parallel_agents._build_task_message(
        "策略评判",
        "比较候选策略并输出结构化评分。",
        agent_id=STRATEGY_JUDGE_AGENT_ID,
    )
    assert "【本阶段禁止反问】" in judge_message
    assert "不得调用 `ask_clarification`" in judge_message
    assert "数据限制与待核验" in judge_message
    assert "不要重复输出同一正文或多个版本" in judge_message

    intent_message = parallel_agents._build_task_message(
        "意图理解",
        "整理用户需求。",
        agent_id=STRATEGY_INTENT_AGENT_ID,
    )
    assert "【需要向用户确认时】" in intent_message
    assert "调用 `ask_clarification` 工具" in intent_message
    assert "【本阶段禁止反问】" not in intent_message


def test_parallel_agent_extracts_provider_reasoning_without_mixing_answer() -> None:
    answer, reasoning = parallel_agents._extract_ai_stream_parts(
        {
            "type": "AIMessageChunk",
            "content": [
                {"type": "thinking_delta", "text": "先核对边界"},
                {"type": "text", "text": "可见正文"},
            ],
            "additional_kwargs": {"reasoning_content": "，再比较路径"},
        }
    )

    assert answer == "可见正文"
    assert reasoning == "先核对边界，再比较路径"

    answer, reasoning = parallel_agents._extract_ai_stream_parts(
        {
            "type": "AIMessageChunk",
            "content": [
                {"type": "thinking", "thinking": "结构化思考字段"},
                {"type": "text", "text": "最终正文"},
            ],
        }
    )
    assert answer == "最终正文"
    assert reasoning == "结构化思考字段"


class _EnabledStrategyRoleStore:
    async def ensure_defaults(self, _rows) -> int:
        return 0

    async def list_roles(self, *, include_disabled: bool) -> list[dict[str, object]]:
        del include_disabled
        return [dict(row) for row in default_strategy_agent_roles()]

    async def get_role(self, agent_id: str) -> dict[str, object]:
        return {"id": agent_id, "enabled": True}


class _GlobalStrategyAgentStore:
    async def get_any(self, agent_id: str) -> dict[str, object]:
        return {"id": agent_id, "user_id": None}


def _parallel_request() -> SimpleNamespace:
    return SimpleNamespace(
        app=SimpleNamespace(
            state=SimpleNamespace(
                strategy_agent_role_store=_EnabledStrategyRoleStore(),
                agent_store=_GlobalStrategyAgentStore(),
            )
        )
    )


async def _collect_parallel_events(payload: dict[str, object]) -> list[dict[str, object]]:
    response = await parallel_agents.run_parallel_agents_stream(
        _parallel_request(),
        payload,
    )
    events: list[dict[str, object]] = []
    async for chunk in response.body_iterator:
        text = chunk.decode() if isinstance(chunk, bytes) else chunk
        for line in text.splitlines():
            if line.startswith("data:"):
                events.append(json.loads(line[5:].strip()))
    return events


def test_parallel_agent_tool_progress_message_is_specific_and_bounded() -> None:
    assert (
        parallel_agents._tool_progress_message(
            "skill_view",
            {"skill_name": "industry-research"},
        )
        == "正在读取技能说明：industry-research"
    )
    assert (
        parallel_agents._tool_progress_message(
            "llmwiki_search",
            {"query": "  人工智能   产业趋势  "},
        )
        == "正在检索知识库：人工智能 产业趋势"
    )
    long_message = parallel_agents._tool_progress_message(
        "read_file",
        {"path": "/mnt/user-data/" + ("very-long-segment-" * 8)},
    )
    assert long_message.startswith("正在读取资料：/mnt/user-data/")
    assert long_message.endswith("…")
    assert len(long_message.removeprefix("正在读取资料：").removesuffix("…")) == 64


def test_parallel_agent_stream_heartbeat_precedes_terminal_event(monkeypatch) -> None:
    async def fake_run_one_agent(*_args, **_kwargs):
        await asyncio.sleep(0.035)
        yield {
            "status": "done",
            "content": "这是一份包含目标、执行路径、风险和验收标准的完整策略正文。",
            "agent_name": "strategy-selection-angle-security",
            "tool_calls": [],
        }
        # Keep the upstream iterator alive briefly after its terminal payload;
        # this used to allow a heartbeat to arrive after agent_done.
        await asyncio.sleep(0.025)

    @asynccontextmanager
    async def fake_client(*_args, **_kwargs):
        yield object()

    async def fake_create_thread(*_args, **_kwargs):
        return "thread-id"

    monkeypatch.setattr(parallel_agents, "_run_one_agent", fake_run_one_agent)
    monkeypatch.setattr(parallel_agents, "_make_loopback_client", fake_client)
    monkeypatch.setattr(parallel_agents, "_create_thread", fake_create_thread)
    monkeypatch.setattr(parallel_agents, "_loopback_base", lambda _request: "http://gateway.test")
    monkeypatch.setattr(parallel_agents, "_auth_headers", lambda _request: {})
    monkeypatch.setattr(parallel_agents, "PROGRESS_HEARTBEAT_SECONDS", 0.01)
    monkeypatch.setattr(parallel_agents, "AGENT_STREAM_IDLE_TIMEOUT_SECONDS", 1.0)

    events = asyncio.run(
        _collect_parallel_events(
            {
                "task_name": "heartbeat order",
                "brief": "show progress while a skill is silent",
                "agent_ids": ["strategy-selection-angle-security"],
            }
        )
    )
    event_types = [str(event.get("type")) for event in events]
    done_index = event_types.index("agent_done")
    progress_indexes = [index for index, event_type in enumerate(event_types) if event_type == "agent_progress"]

    assert event_types[0] == "agent_start"
    assert progress_indexes
    assert all(index < done_index for index in progress_indexes)
    assert event_types[-1] == "all_done"


def test_parallel_agent_stream_idle_timeout_finishes_with_error(monkeypatch) -> None:
    async def fake_run_one_agent(*_args, **_kwargs):
        await asyncio.sleep(60)
        if False:  # pragma: no cover - keep this function an async generator
            yield None

    @asynccontextmanager
    async def fake_client(*_args, **_kwargs):
        yield object()

    async def fake_create_thread(*_args, **_kwargs):
        return "thread-id"

    monkeypatch.setattr(parallel_agents, "_run_one_agent", fake_run_one_agent)
    monkeypatch.setattr(parallel_agents, "_make_loopback_client", fake_client)
    monkeypatch.setattr(parallel_agents, "_create_thread", fake_create_thread)
    monkeypatch.setattr(parallel_agents, "_loopback_base", lambda _request: "http://gateway.test")
    monkeypatch.setattr(parallel_agents, "_auth_headers", lambda _request: {})
    monkeypatch.setattr(parallel_agents, "PROGRESS_HEARTBEAT_SECONDS", 0.005)
    monkeypatch.setattr(parallel_agents, "AGENT_STREAM_IDLE_TIMEOUT_SECONDS", 0.025)

    async def collect_with_deadline() -> list[dict[str, object]]:
        return await asyncio.wait_for(
            _collect_parallel_events(
                {
                    "task_name": "timeout",
                    "brief": "a stalled skill must not leave the page pending",
                    "agent_ids": ["strategy-selection-angle-security"],
                }
            ),
            timeout=1.0,
        )

    events = asyncio.run(collect_with_deadline())
    errors = [event for event in events if event.get("type") == "agent_error"]

    assert any(event.get("type") == "agent_progress" for event in events)
    assert len(errors) == 1
    assert "没有返回进度" in str(errors[0].get("message"))
    assert not any(event.get("type") == "agent_done" for event in events)
    assert events[-1]["type"] == "all_done"


def test_parallel_agent_stream_separates_inline_and_provider_reasoning(monkeypatch) -> None:
    async def fake_run_one_agent(*_args, **_kwargs):
        yield "event: messages"
        yield f"data: {json.dumps([{'type': 'ai', 'content': '<thi'}], ensure_ascii=False)}"
        yield "event: messages"
        yield f"data: {json.dumps([{'type': 'ai', 'content': 'nk>内部分析</think>方案正文'}], ensure_ascii=False)}"
        yield "event: messages"
        yield f"data: {json.dumps([{'type': 'ai', 'content': '', 'additional_kwargs': {'reasoning_content': '补充推理'}}], ensure_ascii=False)}"
        yield {
            "status": "done",
            "content": "<think>内部分析</think>方案正文",
            "agent_name": "strategy-selection-angle-security",
            "tool_calls": [],
        }

    @asynccontextmanager
    async def fake_client(*_args, **_kwargs):
        yield object()

    async def fake_create_thread(*_args, **_kwargs):
        return "thread-id"

    monkeypatch.setattr(parallel_agents, "_run_one_agent", fake_run_one_agent)
    monkeypatch.setattr(parallel_agents, "_make_loopback_client", fake_client)
    monkeypatch.setattr(parallel_agents, "_create_thread", fake_create_thread)
    monkeypatch.setattr(parallel_agents, "_loopback_base", lambda _request: "http://gateway.test")
    monkeypatch.setattr(parallel_agents, "_auth_headers", lambda _request: {})

    async def collect() -> list[dict[str, object]]:
        response = await parallel_agents.run_parallel_agents_stream(
            _parallel_request(),
            {
                "task_name": "test task",
                "brief": "test brief",
                "agent_ids": ["strategy-selection-angle-security"],
            },
        )
        events: list[dict[str, object]] = []
        async for chunk in response.body_iterator:
            text = chunk.decode() if isinstance(chunk, bytes) else chunk
            for line in text.splitlines():
                if line.startswith("data:"):
                    events.append(json.loads(line[5:].strip()))
        return events

    events = asyncio.run(collect())
    visible = "".join(str(event.get("text") or "") for event in events if event.get("type") == "delta")
    thinking = "".join(str(event.get("text") or "") for event in events if event.get("type") == "reasoning_delta")
    done = next(event for event in events if event.get("type") == "agent_done")

    assert visible == "方案正文"
    assert thinking == "内部分析补充推理"
    assert done["content"] == "方案正文"


@pytest.mark.parametrize(
    ("buffered_content", "expected_error"),
    [
        ("大模型本次调用未能成功完成，请稍后再试或更换模型。", "模型调用失败"),
        ("", "未生成有效正文"),
        ("<think>internal reasoning only</think>", "未生成有效正文"),
        ("!" * 500, "未生成有效正文"),
    ],
)
def test_parallel_agent_truncated_stream_rejects_invalid_buffer(
    monkeypatch,
    buffered_content: str,
    expected_error: str,
) -> None:
    async def fake_run_one_agent(*_args, **_kwargs):
        yield "event: messages"
        yield f"data: {json.dumps([{'type': 'ai', 'content': buffered_content}], ensure_ascii=False)}"

    @asynccontextmanager
    async def fake_client(*_args, **_kwargs):
        yield object()

    async def fake_create_thread(*_args, **_kwargs):
        return "thread-id"

    monkeypatch.setattr(parallel_agents, "_run_one_agent", fake_run_one_agent)
    monkeypatch.setattr(parallel_agents, "_make_loopback_client", fake_client)
    monkeypatch.setattr(parallel_agents, "_create_thread", fake_create_thread)
    monkeypatch.setattr(parallel_agents, "_loopback_base", lambda _request: "http://gateway.test")
    monkeypatch.setattr(parallel_agents, "_auth_headers", lambda _request: {})

    async def collect() -> list[dict[str, object]]:
        response = await parallel_agents.run_parallel_agents_stream(
            _parallel_request(),
            {
                "task_name": "test task",
                "brief": "test brief",
                "agent_ids": ["strategy-selection-intent"],
            },
        )
        events: list[dict[str, object]] = []
        async for chunk in response.body_iterator:
            text = chunk.decode() if isinstance(chunk, bytes) else chunk
            for line in text.splitlines():
                if line.startswith("data:"):
                    events.append(json.loads(line[5:].strip()))
        return events

    events = asyncio.run(collect())
    assert not any(event.get("type") == "agent_done" for event in events)
    errors = [event for event in events if event.get("type") == "agent_error"]
    assert len(errors) == 1
    assert expected_error in str(errors[0].get("message"))


def test_parallel_agent_plain_text_streams_each_chunk_without_probe_delay(
    monkeypatch,
) -> None:
    chunks = [
        "This visible strategy ",
        "includes objectives and execution paths, ",
        "plus measurable acceptance criteria.",
    ]

    async def fake_run_one_agent(*_args, **_kwargs):
        for content in chunks:
            yield "event: messages"
            yield f"data: {json.dumps([{'type': 'ai', 'content': content}], ensure_ascii=False)}"

    @asynccontextmanager
    async def fake_client(*_args, **_kwargs):
        yield object()

    async def fake_create_thread(*_args, **_kwargs):
        return "thread-id"

    monkeypatch.setattr(parallel_agents, "_run_one_agent", fake_run_one_agent)
    monkeypatch.setattr(parallel_agents, "_make_loopback_client", fake_client)
    monkeypatch.setattr(parallel_agents, "_create_thread", fake_create_thread)
    monkeypatch.setattr(
        parallel_agents,
        "_loopback_base",
        lambda _request: "http://gateway.test",
    )
    monkeypatch.setattr(parallel_agents, "_auth_headers", lambda _request: {})
    monkeypatch.setattr(
        parallel_agents,
        "_should_probe_implicit_reasoning",
        lambda *_args: False,
    )

    async def collect() -> list[dict[str, object]]:
        response = await parallel_agents.run_parallel_agents_stream(
            _parallel_request(),
            {
                "task_name": "plain stream",
                "brief": "plain answer chunks must stay live",
                "agent_ids": ["strategy-selection-intent"],
            },
        )
        events: list[dict[str, object]] = []
        async for chunk in response.body_iterator:
            text = chunk.decode() if isinstance(chunk, bytes) else chunk
            for line in text.splitlines():
                if line.startswith("data:"):
                    events.append(json.loads(line[5:].strip()))
        return events

    events = asyncio.run(collect())
    assert [event["text"] for event in events if event.get("type") == "delta"] == chunks
    done = next(event for event in events if event.get("type") == "agent_done")
    assert done["content"] == "".join(chunks)


def test_parallel_agent_typed_reasoning_unlocks_live_answer_channel(monkeypatch) -> None:
    chunks = [
        "This visible strategy ",
        "streams after typed reasoning, ",
        "with objectives and acceptance criteria.",
    ]

    async def fake_run_one_agent(*_args, **_kwargs):
        yield "event: messages"
        yield f"data: {json.dumps([{'type': 'ai', 'content': '', 'additional_kwargs': {'reasoning_content': 'live reasoning'}}])}"
        for content in chunks:
            yield "event: messages"
            yield f"data: {json.dumps([{'type': 'ai', 'content': content}])}"

    @asynccontextmanager
    async def fake_client(*_args, **_kwargs):
        yield object()

    async def fake_create_thread(*_args, **_kwargs):
        return "thread-id"

    monkeypatch.setattr(parallel_agents, "_run_one_agent", fake_run_one_agent)
    monkeypatch.setattr(parallel_agents, "_make_loopback_client", fake_client)
    monkeypatch.setattr(parallel_agents, "_create_thread", fake_create_thread)
    monkeypatch.setattr(parallel_agents, "_loopback_base", lambda _request: "http://gateway.test")
    monkeypatch.setattr(parallel_agents, "_auth_headers", lambda _request: {})
    monkeypatch.setattr(parallel_agents, "_should_probe_implicit_reasoning", lambda *_args: True)

    async def collect() -> list[dict[str, object]]:
        response = await parallel_agents.run_parallel_agents_stream(
            _parallel_request(),
            {
                "task_name": "typed reasoning stream",
                "brief": "answer chunks must stay live",
                "agent_ids": ["strategy-selection-intent"],
            },
        )
        events: list[dict[str, object]] = []
        async for chunk in response.body_iterator:
            text = chunk.decode() if isinstance(chunk, bytes) else chunk
            for line in text.splitlines():
                if line.startswith("data:"):
                    events.append(json.loads(line[5:].strip()))
        return events

    events = asyncio.run(collect())
    assert [event["text"] for event in events if event.get("type") == "delta"] == chunks
    assert "".join(str(event.get("text") or "") for event in events if event.get("type") == "reasoning_delta") == "live reasoning"


def test_parallel_agent_truncated_close_only_stream_separates_reasoning(monkeypatch) -> None:
    async def fake_run_one_agent(*_args, **_kwargs):
        yield "event: messages"
        yield f"data: {json.dumps([{'type': 'ai', 'content': 'internal '}], ensure_ascii=False)}"
        yield "event: messages"
        yield f"data: {json.dumps([{'type': 'ai', 'content': 'reasoning</thi'}], ensure_ascii=False)}"
        yield "event: messages"
        yield f"data: {json.dumps([{'type': 'ai', 'content': 'nk>This visible strategy includes objectives, execution paths, and acceptance criteria.'}], ensure_ascii=False)}"

    @asynccontextmanager
    async def fake_client(*_args, **_kwargs):
        yield object()

    async def fake_create_thread(*_args, **_kwargs):
        return "thread-id"

    monkeypatch.setattr(parallel_agents, "_run_one_agent", fake_run_one_agent)
    monkeypatch.setattr(parallel_agents, "_make_loopback_client", fake_client)
    monkeypatch.setattr(parallel_agents, "_create_thread", fake_create_thread)
    monkeypatch.setattr(parallel_agents, "_loopback_base", lambda _request: "http://gateway.test")
    monkeypatch.setattr(parallel_agents, "_auth_headers", lambda _request: {})
    monkeypatch.setattr(
        parallel_agents,
        "_should_probe_implicit_reasoning",
        lambda *_args: True,
    )

    async def collect() -> list[dict[str, object]]:
        response = await parallel_agents.run_parallel_agents_stream(
            _parallel_request(),
            {
                "task_name": "test task",
                "brief": "test brief",
                "agent_ids": ["strategy-selection-intent"],
            },
        )
        events: list[dict[str, object]] = []
        async for chunk in response.body_iterator:
            text = chunk.decode() if isinstance(chunk, bytes) else chunk
            for line in text.splitlines():
                if line.startswith("data:"):
                    events.append(json.loads(line[5:].strip()))
        return events

    events = asyncio.run(collect())
    done = next(event for event in events if event.get("type") == "agent_done")
    reasoning = "".join(str(event.get("text") or "") for event in events if event.get("type") == "reasoning_delta")
    visible = "".join(str(event.get("text") or "") for event in events if event.get("type") == "delta")
    assert done["content"] == ("This visible strategy includes objectives, execution paths, and acceptance criteria.")
    assert visible == done["content"]
    assert reasoning == "internal reasoning"


# ── 中间件给兜底 AIMessage 打标记 ────────────────────────────────────────────


def _middleware() -> LLMErrorHandlingMiddleware:
    app_config = SimpleNamespace(circuit_breaker=SimpleNamespace(failure_threshold=5, recovery_timeout_sec=30))
    return LLMErrorHandlingMiddleware(app_config=app_config)


def test_middleware_marks_error_message() -> None:
    mw = _middleware()

    def _boom(_req):
        raise RuntimeError("insufficient_quota: 余额不足")  # quota → 非重试 → 立即兜底

    out = mw.wrap_model_call(None, _boom)
    assert harness_is_llm_error_message(out) == "quota"
    assert out.additional_kwargs.get(LLM_ERROR_MARKER) == "quota"


# ── fallback_model_chain ─────────────────────────────────────────────────────


def _patch_models(monkeypatch, names: list[str]) -> None:
    cfg = SimpleNamespace(models=[SimpleNamespace(name=n) for n in names])
    monkeypatch.setattr(mf, "get_app_config", lambda: cfg)


def test_chain_current_first_then_config_order(monkeypatch) -> None:
    _patch_models(monkeypatch, ["m1", "m2", "m3"])
    monkeypatch.delenv("ROUNDTABLE_MODEL_FALLBACK", raising=False)
    monkeypatch.delenv("ROUNDTABLE_MODEL_FALLBACK_MAX", raising=False)
    assert fallback_model_chain("m2") == ["m2", "m1", "m3"]


def test_chain_empty_current_uses_config_order(monkeypatch) -> None:
    _patch_models(monkeypatch, ["m1", "m2", "m3"])
    monkeypatch.delenv("ROUNDTABLE_MODEL_FALLBACK", raising=False)
    assert fallback_model_chain("") == ["m1", "m2", "m3"]
    assert fallback_model_chain(None) == ["m1", "m2", "m3"]


def test_chain_unknown_current_kept_first(monkeypatch) -> None:
    _patch_models(monkeypatch, ["m1", "m2"])
    monkeypatch.delenv("ROUNDTABLE_MODEL_FALLBACK", raising=False)
    assert fallback_model_chain("custom") == ["custom", "m1", "m2"]


def test_chain_disabled_single(monkeypatch) -> None:
    _patch_models(monkeypatch, ["m1", "m2", "m3"])
    monkeypatch.setenv("ROUNDTABLE_MODEL_FALLBACK", "0")
    assert fallback_model_chain("m2") == ["m2"]


def test_chain_cap(monkeypatch) -> None:
    _patch_models(monkeypatch, ["m1", "m2", "m3", "m4"])
    monkeypatch.delenv("ROUNDTABLE_MODEL_FALLBACK", raising=False)
    monkeypatch.setenv("ROUNDTABLE_MODEL_FALLBACK_MAX", "2")
    assert fallback_model_chain("m1") == ["m1", "m2"]


def test_chain_no_config_returns_current(monkeypatch) -> None:
    _patch_models(monkeypatch, [])
    monkeypatch.delenv("ROUNDTABLE_MODEL_FALLBACK", raising=False)
    assert fallback_model_chain("m9") == ["m9"]
    assert fallback_model_chain("") == []


def test_reason_text() -> None:
    assert reason_text("busy") == "服务繁忙"
    assert reason_text("empty") == "未产出有效内容"
    assert reason_text("timeout") == "长时间无响应（卡死）"
    assert reason_text(None) == "调用失败"
    assert reason_text("不存在的码") == "调用失败"


@pytest.mark.asyncio
async def test_thinking_only_seat_does_not_switch_model(monkeypatch) -> None:
    """Reasoning-only output is an incomplete delivery, not a model failure."""
    calls: list[str] = []

    async def fake_stream(_client, _base, _headers, _thread_id, body):
        calls.append(body["context"]["model_name"])
        yield None, [
            'data: {"messages":[{"type":"ai","content":"<think>仍在分析约束</think>"}]}'
        ]

    monkeypatch.setattr(multi_agent, "_stream_upstream", fake_stream)
    monkeypatch.setattr(
        multi_agent,
        "fallback_model_chain",
        lambda _model: ["slow-thinking-model", "backup-model"],
    )
    monkeypatch.setattr(
        multi_agent,
        "append_seat_skill_directive",
        lambda _agent_id, task, **_kwargs: task,
    )
    monkeypatch.setattr(multi_agent, "_spawn_detached_broadcast", lambda *_args, **_kwargs: None)

    frames = [
        frame
        async for frame in multi_agent._special_run(
            client=object(),
            base="http://gateway",
            headers={},
            agent_threads={"seat-a": "seat-thread"},
            agent_name="seat-a",
            new_message="完成分析",
            skill_stop_names=[],
            model_name="slow-thinking-model",
        )
    ]

    assert calls == ["slow-thinking-model"]
    assert not any(
        isinstance(frame, dict) and frame.get("status") == "model_switch"
        for frame in frames
    )
    assert frames[-1]["status"] == "invalid_delivery"
    assert frames[-1]["reason"] == "thinking_only"


@pytest.mark.asyncio
async def test_real_model_error_still_switches_to_next_model(monkeypatch) -> None:
    calls: list[str] = []

    async def fake_stream(_client, _base, _headers, _thread_id, body):
        model = body["context"]["model_name"]
        calls.append(model)
        message = (
            {
                "type": "ai",
                "content": "大模型服务繁忙，请稍后再试。",
                "additional_kwargs": {LLM_ERROR_MARKER: "busy"},
            }
            if model == "broken-model"
            else {"type": "ai", "content": "这是备用模型完成的有效交付正文。"}
        )
        yield None, [f"data: {json.dumps({'messages': [message]}, ensure_ascii=False)}"]

    monkeypatch.setattr(multi_agent, "_stream_upstream", fake_stream)
    monkeypatch.setattr(
        multi_agent,
        "fallback_model_chain",
        lambda _model: ["broken-model", "backup-model"],
    )
    monkeypatch.setattr(
        multi_agent,
        "append_seat_skill_directive",
        lambda _agent_id, task, **_kwargs: task,
    )
    monkeypatch.setattr(multi_agent, "_spawn_detached_broadcast", lambda *_args, **_kwargs: None)

    frames = [
        frame
        async for frame in multi_agent._special_run(
            client=object(),
            base="http://gateway",
            headers={},
            agent_threads={"seat-a": "seat-thread"},
            agent_name="seat-a",
            new_message="完成分析",
            skill_stop_names=[],
            model_name="broken-model",
        )
    ]

    assert calls == ["broken-model", "backup-model"]
    assert any(
        isinstance(frame, dict) and frame.get("status") == "model_switch"
        for frame in frames
    )
    assert frames[-1]["model_used"] == "backup-model"
    assert frames[-1]["content"] == "这是备用模型完成的有效交付正文。"


@pytest.mark.asyncio
async def test_background_thinking_only_does_not_switch_model(monkeypatch) -> None:
    import app.gateway.roundtable_inprocess_gateway as gateway_module

    calls: list[str] = []

    async def fake_run_agent_turn(*_args, **kwargs):
        calls.append(kwargs["model"])
        return [{"type": "ai", "content": "<think>继续进行长思考</think>"}]

    monkeypatch.setattr(gateway_module, "_run_agent_turn", fake_run_agent_turn)
    monkeypatch.setattr(
        gateway_module,
        "fallback_model_chain",
        lambda _model: ["slow-thinking-model", "backup-model"],
    )
    gateway = gateway_module.InProcessRoundtableGateway(
        SimpleNamespace(state=SimpleNamespace()),
        user_id="u1",
        agents=[],
    )

    messages, switches, used_model = await gateway._run_turn_with_fallback(
        thread_id="seat-thread",
        agent_name="seat-a",
        message="完成分析",
        requested_model="slow-thinking-model",
        excluded_tools=[],
        allowed_tools=None,
        skill_stop_names=[],
        thinking_enabled=True,
        reasoning_effort="medium",
        recursion_limit=1000,
        stage="step2_seat",
    )

    assert calls == ["slow-thinking-model"]
    assert switches == []
    assert used_model == "slow-thinking-model"
    assert messages[-1]["content"] == "<think>继续进行长思考</think>"


# ── 引擎：单席位异常隔离 + 模型容错记录 → 系统提示气泡 ──────────────────────────

_AGENTS = [SeatRef("seat-a", "情报"), SeatRef("seat-b", "环境"), SeatRef("seat-c", "方案")]


def _writer() -> JobProgressWriter:
    return JobProgressWriter(None, job_id="job-x", agents=_AGENTS)


def _thread_ids() -> dict[str, str]:
    return {a.agent_id: f"tid-{a.agent_id}" for a in _AGENTS} | {"roundtable-coordinator": "tid-co"}


class _ChainGateway:
    """chain 模式假网关：可指定某席位抛异常 / 带回模型容错记录。"""

    def __init__(self, *, fail_agent: str | None = None, switch_agent: str | None = None) -> None:
        self.fail_agent = fail_agent
        self.switch_agent = switch_agent
        self.seen: list[str] = []

    async def run_leader(
        self,
        *,
        thread_ids,
        coordinator_name,
        message,
        model,
        finalize_only: bool = False,
    ) -> LeaderTurn:
        return LeaderTurn(final_text="总控说明", dispatched=[])

    async def run_seat(self, *, thread_ids, agent_id, task, model) -> SeatTurn:
        self.seen.append(agent_id)
        if agent_id == self.fail_agent:
            raise RuntimeError("boom: 模型连接被打满")
        switches = []
        if agent_id == self.switch_agent:
            switches = [ModelSwitch(failed_model="m1", next_model="m2", reason_label="服务繁忙")]
        return SeatTurn(final_text=f"{agent_id} 的交付", switches=switches)

    async def create_report_thread(self) -> str:  # pragma: no cover
        raise AssertionError("enable_report=False 不应绘制报告")


async def _run_chain(gw: _ChainGateway) -> tuple[object, JobProgressWriter]:
    writer = _writer()
    result = await run_orchestration(
        gateway=gw,
        progress=writer,
        agents=_AGENTS,
        thread_ids=_thread_ids(),
        coordinator_name="roundtable-coordinator",
        model="m1",
        seed_message="种子任务",
        mode="chain",
        enable_report=False,
    )
    return result, writer


def test_chain_seat_failure_isolated() -> None:
    gw = _ChainGateway(fail_agent="seat-b")
    result, writer = asyncio.run(_run_chain(gw))
    # 单席位失败不拖垮整链：仍走完三个席位并收口为 done。
    assert result.status == "done"
    assert gw.seen == ["seat-a", "seat-b", "seat-c"]
    # 失败席位的交付气泡为空，其余正常。
    seat_dlgs = {d["agentId"]: d["content"] for d in writer.dialogues() if d["role"] == "seat"}
    assert seat_dlgs["seat-b"] == ""
    assert seat_dlgs["seat-a"] == "seat-a 的交付"
    assert seat_dlgs["seat-c"] == "seat-c 的交付"


def test_model_switch_emits_system_notice() -> None:
    gw = _ChainGateway(switch_agent="seat-b")
    _result, writer = asyncio.run(_run_chain(gw))
    systems = [d for d in writer.dialogues() if d["role"] == "system"]
    assert len(systems) == 1
    assert "模型容错" in systems[0]["content"]
    assert "服务繁忙" in systems[0]["content"]
    assert "m2" in systems[0]["content"]