deerflow-code/offline-backend-20260512/backend/tests/test_roundtable_seat_modes.py

"""每席位「推理深度覆盖」（seat_modes）在进程内网关 run_seat 的解析测试。

语义（与前端 useStep2Orchestration「覆盖优先、否则跟随链条级 selectedMode」逐席位对齐）：
- 席位命中 seat_modes → 用该档（_SEAT_MODE_PARAMS）派生的 thinking/reasoning/subagent **覆盖**
  作业级 seat_*（不论作业级是什么）；
- 席位未命中 → 跟随作业级 seat_thinking_enabled / seat_reasoning_effort / seat_subagent_enabled；
- 作业级仍为 None → 落 policy("seat") 默认。

只验证传给 _run_agent_turn 的三个运行参数，故把 _run_agent_turn / policy / 文本解析全部打桩，
不触发真实 LLM / 沙箱。
"""
from __future__ import annotations

from types import SimpleNamespace

import pytest


def test_roundtable_tool_projection_excludes_internal_write_handshake() -> None:
    from app.gateway.roundtable_inprocess_gateway import _tool_calls_since_last_human

    messages = [
        {"type": "human", "content": "写报告"},
        {
            "type": "ai",
            "tool_calls": [
                {
                    "id": "begin-1",
                    "name": "begin_write_file",
                    "args": {"path": "/tmp/report.md"},
                }
            ],
        },
        {
            "type": "ai",
            "tool_calls": [
                {
                    "id": "write-1",
                    "name": "write_file",
                    "args": {"path": "/tmp/report.md", "content": "# report"},
                }
            ],
        },
    ]

    calls = _tool_calls_since_last_human(messages)

    assert [call["name"] for call in calls] == ["write_file"]
    assert calls[0]["args"]["content"] == "# report"


@pytest.fixture
def _gw(monkeypatch):
    """构造一个被打桩的 InProcessRoundtableGateway，返回 (build, captured)。

    build(seat_modes=..., seat_thinking_enabled=..., seat_reasoning_effort=..., seat_subagent_enabled=...)
    返回 gateway 实例；captured 记录最近一次 _run_agent_turn 收到的 thinking/reasoning/subagent。
    """
    import app.gateway.roundtable_inprocess_gateway as gw

    captured: dict[str, object] = {}

    async def _fake_run_agent_turn(*args, **kwargs):
        captured["thinking_enabled"] = kwargs.get("thinking_enabled")
        captured["reasoning_effort"] = kwargs.get("reasoning_effort")
        captured["subagent_enabled"] = kwargs.get("subagent_enabled")
        return []

    # policy("seat") 默认：thinking=True / reasoning="medium"（= 现有席位默认 pro 行为）。
    monkeypatch.setattr(
        gw,
        "roundtable_run_policy",
        lambda role: SimpleNamespace(excluded_tools=[], skill_stop_names=[], thinking_enabled=True, reasoning_effort="medium"),
    )
    monkeypatch.setattr(gw, "_run_agent_turn", _fake_run_agent_turn)
    # 技能强化与本测试无关 → 原样返回任务，避免触发 load_agent_config / skill storage。
    monkeypatch.setattr(gw, "append_seat_skill_directive", lambda agent_id, task, **k: task)
    monkeypatch.setattr(gw, "_latest_ai_text", lambda messages: "")
    monkeypatch.setattr(gw, "_tool_calls_since_last_human", lambda messages: [])

    def _build(**kwargs):
        return gw.InProcessRoundtableGateway(SimpleNamespace(state=SimpleNamespace()), user_id="u1", agents=[], **kwargs)

    return _build, captured


async def _run(gateway):
    await gateway.run_seat(thread_ids={"seat-1": "tid-1"}, agent_id="seat-1", task="任务", model="m")


@pytest.mark.asyncio
async def test_seat_mode_override_wins_over_job_level(_gw):
    """席位配了 flash → 覆盖作业级 pro（thinking off / minimal / 无子代理）。"""
    build, captured = _gw
    gateway = build(
        seat_modes={"seat-1": "flash"},
        seat_thinking_enabled=True,
        seat_reasoning_effort="high",
        seat_subagent_enabled=True,
    )
    await _run(gateway)
    assert captured["thinking_enabled"] is False
    assert captured["reasoning_effort"] == "minimal"
    assert captured["subagent_enabled"] is False


@pytest.mark.asyncio
async def test_ultra_override_enables_subagent(_gw):
    build, captured = _gw
    gateway = build(seat_modes={"seat-1": "ultra"})
    await _run(gateway)
    assert captured["thinking_enabled"] is True
    assert captured["reasoning_effort"] == "high"
    assert captured["subagent_enabled"] is True


@pytest.mark.asyncio
async def test_no_override_follows_job_level(_gw):
    """席位未命中 seat_modes → 跟随作业级 seat_*。"""
    build, captured = _gw
    gateway = build(
        seat_modes={"other": "flash"},
        seat_thinking_enabled=True,
        seat_reasoning_effort="low",
        seat_subagent_enabled=False,
    )
    await _run(gateway)
    assert captured["thinking_enabled"] is True
    assert captured["reasoning_effort"] == "low"
    assert captured["subagent_enabled"] is False


@pytest.mark.asyncio
async def test_no_override_no_job_level_falls_back_to_policy(_gw):
    """既无席位覆盖、作业级也为 None → 落 policy 默认（thinking=True / medium）。"""
    build, captured = _gw
    gateway = build(seat_modes={})
    await _run(gateway)
    assert captured["thinking_enabled"] is True
    assert captured["reasoning_effort"] == "medium"
    assert captured["subagent_enabled"] is False


@pytest.mark.asyncio
async def test_unknown_mode_value_ignored_falls_through(_gw):
    """非法档位值（前端不会下发，防御）→ 安全忽略，回退作业级/policy。"""
    build, captured = _gw
    gateway = build(seat_modes={"seat-1": "bogus"}, seat_reasoning_effort="low", seat_thinking_enabled=True)
    await _run(gateway)
    assert captured["thinking_enabled"] is True
    assert captured["reasoning_effort"] == "low"


def test_chain_seat_schema_keeps_human_validation_checkpoint():
    """Chain-seat JSON must preserve the frontend Step 2 pause configuration."""
    from app.gateway.routers.roundtable_chains import ChainSeatSchema

    assert ChainSeatSchema(agent_id="seat-1").human_validation is False
    assert ChainSeatSchema(agent_id="seat-1", human_validation=True).model_dump()[
        "human_validation"
    ] is True