Octop/tests/unit/gateway/test_global_processor_team.py

"""tests/unit/test_global_processor_team.py"""

from __future__ import annotations

import asyncio
import json
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock

import pytest
from octop_harness.teams.inbox import InboxMessage
from octop_harness.teams.processor import ReplyEvent

from octop.infra.db.migrate import run_migrations
from octop.infra.db.pool import SqlitePool
from octop.infra.db.repos.agents import AgentRepo
from octop.infra.db.repos.sessions import SessionRepo
from octop.infra.db.repos.threads import ThreadRepo
from octop.infra.db.repos.users import UserRepo
from octop.infra.gateway.process.processor import GlobalProcessor
from octop.infra.gateway.slash.dispatcher import SlashDispatcher
from octop.infra.gateway.threads import ThreadRegistry


@pytest.fixture
def processor_env(tmp_path: Path) -> dict[str, object]:
    db = SqlitePool(tmp_path / "octop.db")
    run_migrations(db)
    UserRepo(db).create(username="u", password_hash="h", role="user")
    AgentRepo(db).create(agent_id="parent", user_id=1, name="Parent")
    AgentRepo(db).create(agent_id="child", user_id=1, name="Researcher")

    repos = MagicMock()
    repos.session_repo = SessionRepo(db)
    repos.thread_repo = ThreadRepo(db)
    repos.channel_repo = MagicMock()
    repos.audit_repo = MagicMock()
    repos.agent_repo = AgentRepo(db)
    repos.user_repo = UserRepo(db)
    repos.connector_repo = MagicMock()

    agent_manager = MagicMock()
    agent_manager.get_row.side_effect = lambda aid: AgentRepo(db).get(aid)

    from octop.infra.gateway.gateway import Gateway

    gw = Gateway(agent_manager=agent_manager, repos=repos)
    gw._channel_manager = MagicMock()

    parent_sk = ThreadRegistry.dashboard_key(agent_id="parent", user_id=1)
    gw.thread_registry._threads.insert(
        thread_id="thr_parent",
        agent_id="parent",
        user_id=1,
        channel_type="dashboard",
        session_key=parent_sk,
    )
    gw.thread_registry._sessions.upsert(
        session_key=parent_sk,
        agent_id="parent",
        user_id=1,
        channel_type="dashboard",
        chat_type="dm",
        thread_id="thr_parent",
    )

    processor = GlobalProcessor(
        agent_manager=agent_manager,
        thread_registry=gw.thread_registry,
        audit_repo=repos.audit_repo,
        agent_repo=repos.agent_repo,
        user_repo=repos.user_repo,
        connector_repo=repos.connector_repo,
        dispatcher=SlashDispatcher(),
        gateway=gw,
    )
    return {"processor": processor, "gateway": gw, "parent_sk": parent_sk}


def test_compose_followup_uses_peer_display_name(processor_env: dict) -> None:
    processor = processor_env["processor"]
    msg = InboxMessage(
        id="job-1",
        target_agent_id="child",
        source_agent_id="parent",
        source_thread_id="thr_parent",
        message="survey market",
        user_id=1,
        original_user_prompt="market size?",
    )
    text = processor.compose_followup(msg, result_text="findings", error_text=None)
    assert "Researcher" in text
    assert "findings" in text
    assert "do not" in text.lower() or "不要" in text


def test_compose_followup_team_host_asks_for_wrapup(processor_env: dict) -> None:
    processor = processor_env["processor"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    msg = InboxMessage(
        id="job-2",
        target_agent_id="child",
        source_agent_id="host",
        source_thread_id="thr_parent",
        message="survey market",
        user_id=1,
    )
    text = processor.compose_followup(msg, result_text="findings", error_text=None)
    assert "Researcher" in text
    assert "findings" in text
    assert "Member answer" in text or "成员正文" in text
    assert "do not" in text.lower() or "不要" in text


def test_compose_followup_truncates_long_member_result(processor_env: dict) -> None:
    from octop.infra.agents.teams.team_manager import _FOLLOWUP_RESULT_MAX_CHARS

    processor = processor_env["processor"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    msg = InboxMessage(
        id="job-3",
        target_agent_id="child",
        source_agent_id="host",
        source_thread_id="thr_parent",
        message="survey market",
        user_id=1,
    )
    huge = "建议采纳此项。" + ("x" * (_FOLLOWUP_RESULT_MAX_CHARS + 200))
    text = processor.compose_followup(msg, result_text=huge, error_text=None)
    assert "建议采纳此项。" in text
    assert huge not in text
    assert "…" in text
    assert len(text) < len(huge) + 800


def test_compose_followup_empty_result_placeholder(processor_env: dict) -> None:
    processor = processor_env["processor"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    msg = InboxMessage(
        id="job-4",
        target_agent_id="child",
        source_agent_id="host",
        source_thread_id="thr_parent",
        message="survey market",
        user_id=1,
    )
    text = processor.compose_followup(msg, result_text="   ", error_text=None)
    assert "(empty)" in text


@pytest.mark.asyncio
async def test_on_reply_increments_unread_on_dashboard(processor_env: dict) -> None:
    processor = processor_env["processor"]
    parent_sk = processor_env["parent_sk"]

    await processor.on_reply(
        ReplyEvent(
            inbox_id="job-1",
            status="done",
            source_agent_id="parent",
            source_thread_id="thr_parent",
            target_agent_id="child",
            user_id=1,
            reply_text="final synthesized reply",
            metadata={"session_key": parent_sk},
        )
    )

    session = processor_env["gateway"].thread_registry.get_session(parent_sk)  # type: ignore[attr-defined]
    assert session is not None
    assert session.unread_count == 1


@pytest.mark.asyncio
async def test_on_reply_pushes_watched_dashboard_thread(processor_env: dict) -> None:
    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    parent_sk = processor_env["parent_sk"]
    frames: list[dict[str, object]] = []

    async def capture(frame: dict[str, object]) -> None:
        frames.append(frame)

    gateway.ws_hub.register("conn-1", capture)
    gateway.ws_hub.subscribe("thr_parent", "conn-1")

    await processor.on_reply(
        ReplyEvent(
            inbox_id="job-watch",
            status="done",
            source_agent_id="parent",
            source_thread_id="thr_parent",
            target_agent_id="child",
            user_id=1,
            reply_text="team wrap-up in the open room",
            metadata={"session_key": parent_sk},
        )
    )

    session = gateway.thread_registry.get_session(parent_sk)  # type: ignore[attr-defined]
    assert session is not None
    assert session.unread_count == 0
    assert any(
        frame.get("type") == "token" and frame.get("content") == "team wrap-up in the open room"
        for frame in frames
    )
    assert any(frame.get("type") == "done" for frame in frames)


@pytest.mark.asyncio
async def test_on_reply_team_host_pushes_wrapup_live(processor_env: dict) -> None:
    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    parent_sk = processor_env["parent_sk"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    frames: list[dict[str, object]] = []

    async def capture(frame: dict[str, object]) -> None:
        frames.append(frame)

    gateway.ws_hub.register("conn-team-wrap", capture)
    gateway.ws_hub.subscribe("thr_parent", "conn-team-wrap")

    await processor.on_reply(
        ReplyEvent(
            inbox_id="job-team-wrap",
            status="done",
            source_agent_id="host",
            source_thread_id="thr_parent",
            target_agent_id="child",
            user_id=1,
            reply_text="睡眠建议已经给出，可以收工。",
            metadata={"session_key": parent_sk},
        )
    )

    tokens = [frame for frame in frames if frame.get("type") == "token"]
    assert tokens
    assert tokens[0].get("content") == "睡眠建议已经给出，可以收工。"
    assert tokens[0].get("team_wrapup") is True
    assert tokens[0].get("team_snapshot") is not True


@pytest.mark.asyncio
async def test_on_reply_publishes_wrapup_without_session_key(processor_env: dict) -> None:
    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    frames: list[dict[str, object]] = []

    async def capture(frame: dict[str, object]) -> None:
        frames.append(frame)

    gateway.ws_hub.register("conn-no-sk", capture)
    gateway.ws_hub.subscribe("thr_parent", "conn-no-sk")

    await processor.on_reply(
        ReplyEvent(
            inbox_id="job-no-sk",
            status="done",
            source_agent_id="host",
            source_thread_id="thr_parent",
            target_agent_id="child",
            user_id=1,
            reply_text="没有 session_key 也要进群聊",
            metadata={},
        )
    )

    tokens = [frame for frame in frames if frame.get("type") == "token"]
    assert tokens
    assert tokens[0].get("content") == "没有 session_key 也要进群聊"


@pytest.mark.asyncio
async def test_on_reply_skips_snapshot_after_live_host_wrapup(
    processor_env: dict,
) -> None:
    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    parent_sk = processor_env["parent_sk"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    processor.teams._live_host_replies.add("thr_parent")
    frames: list[dict[str, object]] = []

    async def capture(frame: dict[str, object]) -> None:
        frames.append(frame)

    gateway.ws_hub.register("conn-live-wrap", capture)
    gateway.ws_hub.subscribe("thr_parent", "conn-live-wrap")

    await processor.on_reply(
        ReplyEvent(
            inbox_id="job-live-wrap",
            status="done",
            source_agent_id="host",
            source_thread_id="thr_parent",
            target_agent_id="child",
            user_id=1,
            reply_text="可以收工。",
            metadata={"session_key": parent_sk},
        )
    )

    assert [frame for frame in frames if frame.get("type") == "token"] == []
    assert "thr_parent" not in processor.teams._live_host_replies


@pytest.mark.asyncio
async def test_stream_host_followup_marks_live_and_persists(
    processor_env: dict,
) -> None:
    from langchain_core.messages import AIMessage
    from octop_harness.request import ChatRequest

    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    frames: list[dict[str, object]] = []

    async def capture(frame: dict[str, object]) -> None:
        frames.append(frame)

    gateway.ws_hub.register("conn-host-followup", capture)
    gateway.ws_hub.subscribe("thr_parent", "conn-host-followup")

    seen_threads: list[str] = []

    async def fake_stream(_agent_id: str, request: dict[str, object]) -> object:
        seen_threads.append(str(request.get("thread_id") or ""))
        yield {"type": "token", "content": "可以"}
        yield {"type": "token", "content": "收工。"}

    processor._agent_manager.stream = fake_stream
    harness = MagicMock()
    harness.aget_history = AsyncMock(return_value=[AIMessage(content="可以收工。")])
    processor._agent_manager.get_agent.return_value = harness
    repo = MagicMock()
    repo.projection_status.return_value = "ready"
    repo.append_if_ready = MagicMock(return_value=1)
    processor.replace_thread_message_repo(repo)

    result = await processor.teams.stream_host_followup_to_room(
        ChatRequest(messages="wrap up", thread_id="thr_parent", agent_id="host"),
        room_thread_id="thr_parent",
        speaker_id="host",
    )
    assert result["team_live_streamed"] is True
    assert "thr_parent" in processor.teams._live_host_replies
    tokens = [frame for frame in frames if frame.get("type") == "token"]
    assert [frame.get("content") for frame in tokens] == ["可以", "收工。"]
    assert all(frame.get("team_wrapup") is True for frame in tokens)
    assert all(frame.get("team_snapshot") is not True for frame in frames)
    assert seen_threads == ["thr_parent"]
    done = [frame for frame in frames if frame.get("type") == "done"]
    assert done
    assert all(frame.get("team_wrapup") is True for frame in done)
    repo.append_if_ready.assert_called()


@pytest.mark.asyncio
async def test_stream_host_followup_relays_decision_custom(
    processor_env: dict,
) -> None:
    from langchain_core.messages import AIMessage
    from octop_harness.request import ChatRequest

    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    frames: list[dict[str, object]] = []

    async def capture(frame: dict[str, object]) -> None:
        frames.append(frame)

    gateway.ws_hub.register("conn-wrap-decision", capture)
    gateway.ws_hub.subscribe("thr_parent", "conn-wrap-decision")

    decision = {
        "kind": "decision",
        "model": "laya",
        "parts": [{"kind": "result", "kept": [{"name": "结果充分, 可以总结", "score": 0.91}]}],
    }

    async def fake_stream(_agent_id: str, _request: dict[str, object]) -> object:
        yield {"type": "custom", "data": decision}
        yield {"type": "token", "content": "可以收工。"}

    processor._agent_manager.stream = fake_stream
    harness = MagicMock()
    harness.aget_history = AsyncMock(return_value=[AIMessage(content="可以收工。")])
    processor._agent_manager.get_agent.return_value = harness
    repo = MagicMock()
    repo.projection_status.return_value = "ready"
    repo.append_if_ready = MagicMock(return_value=1)
    processor.replace_thread_message_repo(repo)

    await processor.teams.stream_host_followup_to_room(
        ChatRequest(messages="wrap up", thread_id="thr_parent", agent_id="host"),
        room_thread_id="thr_parent",
        speaker_id="host",
    )
    customs = [frame for frame in frames if frame.get("type") == "custom"]
    assert len(customs) == 1
    assert customs[0].get("team_wrapup") is True
    assert customs[0].get("agent_id") == "host"
    assert customs[0].get("data") == decision
    persisted = repo.append_if_ready.call_args.args[1]
    assert "octop_decision" in str(persisted)
    assert "result" in str(persisted)


@pytest.mark.asyncio
async def test_member_stream_does_not_relay_decision_custom(processor_env: dict) -> None:
    from langchain_core.messages import AIMessage
    from octop_harness.request import ChatRequest

    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    room_frames: list[dict[str, object]] = []

    async def capture_room(frame: dict[str, object]) -> None:
        room_frames.append(frame)

    gateway.ws_hub.register("conn-room-decision", capture_room)
    gateway.ws_hub.subscribe("thr_parent", "conn-room-decision")

    async def fake_stream(_agent_id: str, _request: dict[str, object]) -> object:
        yield {"type": "custom", "data": {"kind": "decision", "parts": []}}
        yield {"type": "token", "content": "成员回答"}

    processor._agent_manager.stream = fake_stream
    harness = MagicMock()
    harness.aget_history = AsyncMock(return_value=[AIMessage(content="成员回答")])
    processor._agent_manager.get_agent.return_value = harness

    await processor.teams.stream_peer_to_room(
        ChatRequest(messages="task", thread_id="thr_parent~child", agent_id="child"),
        room_thread_id="thr_parent",
        speaker_id="child",
    )
    assert [frame for frame in room_frames if frame.get("type") == "custom"] == []
    assert any(frame.get("type") == "token" for frame in room_frames)


@pytest.mark.asyncio
async def test_wrapup_request_marks_decision_followup() -> None:
    from types import SimpleNamespace

    from octop.infra.agents.teams.team_manager import _patch_inbox_wrapup

    seen: dict[str, object] = {}

    async def stream_followup(req: object, *, room_thread_id: str, speaker_id: str) -> str:
        seen["req"] = req
        seen["room"] = room_thread_id
        seen["speaker"] = speaker_id
        return "ok"

    async def original(_msg: object, _result_text: str | None, _error_text: str | None) -> str:
        return "fallback"

    inbox = SimpleNamespace(
        _synthesize_reply=original,
        _is_cancelled=lambda _msg: False,
        _processor=SimpleNamespace(
            compose_followup=lambda _msg, result_text, error_text: "wakeup",
        ),
    )
    _patch_inbox_wrapup(inbox, stream_followup=stream_followup, is_team=lambda _agent_id: True)
    msg = SimpleNamespace(
        user_id=1,
        source_thread_id="room-1",
        source_agent_id="host",
        target_agent_id="agent-research",
        message="查资料",
    )
    await inbox._synthesize_reply(msg, "成员正文", None)
    req = seen["req"]
    configurable = getattr(req, "configurable", None)
    assert configurable["octop_decision_followup"] is True
    assert configurable["peer_invoke_mode"] == "async"
    assert configurable["octop_decision_finished_member"] == "agent-research"
    assert configurable["octop_decision_task"] == "查资料"
    assert seen["room"] == "room-1"
    assert seen["speaker"] == "host"


@pytest.mark.asyncio
async def test_stream_host_followup_waits_for_dispatch_turn(
    processor_env: dict,
) -> None:
    import asyncio

    from octop_harness.request import ChatRequest

    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    processor._agent_manager.is_agent_active = MagicMock(return_value=False)
    started = asyncio.Event()

    async def fake_stream(_agent_id: str, _request: dict[str, object]) -> object:
        started.set()
        yield {"type": "token", "content": "可以收工。"}

    processor._agent_manager.stream = fake_stream
    gateway.ws_hub.mark_turn_active("thr_parent")
    task = asyncio.create_task(
        processor.teams.stream_host_followup_to_room(
            ChatRequest(messages="wrap up", thread_id="thr_parent", agent_id="host"),
            room_thread_id="thr_parent",
            speaker_id="host",
        )
    )
    await asyncio.sleep(0.08)
    assert not started.is_set()
    gateway.ws_hub.mark_turn_idle("thr_parent")
    await asyncio.wait_for(task, timeout=2)
    assert started.is_set()


@pytest.mark.asyncio
async def test_unwatched_wrapup_is_persisted_once(
    processor_env: dict,
) -> None:
    from langchain_core.messages import AIMessage
    from octop_harness.request import ChatRequest

    processor = processor_env["processor"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")

    async def fake_stream(_agent_id: str, _request: dict[str, object]) -> object:
        yield {"type": "token", "content": "可以收工。"}

    processor._agent_manager.stream = fake_stream
    harness = MagicMock()
    harness.aget_history = AsyncMock(return_value=[AIMessage(content="可以收工。")])
    processor._agent_manager.get_agent.return_value = harness
    repo = MagicMock()
    repo.projection_status.return_value = "ready"
    repo.append_if_ready = MagicMock(return_value=1)
    processor.replace_thread_message_repo(repo)

    result = await processor.teams.stream_host_followup_to_room(
        ChatRequest(messages="wrap up", thread_id="thr_parent", agent_id="host"),
        room_thread_id="thr_parent",
        speaker_id="host",
    )
    assert result["team_live_streamed"] is False
    assert "thr_parent" in processor.teams._live_host_replies
    assert repo.append_if_ready.call_count == 1

    await processor.on_reply(
        ReplyEvent(
            inbox_id="job-unwatched-wrap",
            status="done",
            source_agent_id="host",
            source_thread_id="thr_parent",
            target_agent_id="child",
            user_id=1,
            reply_text="可以收工。",
            metadata={},
        )
    )
    assert repo.append_if_ready.call_count == 1
    assert "thr_parent" not in processor.teams._live_host_replies


@pytest.mark.asyncio
async def test_stream_host_followup_pushes_wrapup_when_no_tokens(
    processor_env: dict,
) -> None:
    from langchain_core.messages import AIMessage
    from octop_harness.request import ChatRequest

    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    frames: list[dict[str, object]] = []

    async def capture(frame: dict[str, object]) -> None:
        frames.append(frame)

    gateway.ws_hub.register("conn-silent-wrap", capture)
    gateway.ws_hub.subscribe("thr_parent", "conn-silent-wrap")

    async def fake_stream(_agent_id: str, _request: dict[str, object]) -> object:
        yield {"type": "state_snapshot", "data": {}}

    processor._agent_manager.stream = fake_stream
    harness = MagicMock()
    harness.aget_history = AsyncMock(return_value=[AIMessage(content="可以收工。")])
    processor._agent_manager.get_agent.return_value = harness
    repo = MagicMock()
    repo.projection_status.return_value = "ready"
    repo.append_if_ready = MagicMock(return_value=1)
    processor.replace_thread_message_repo(repo)

    result = await processor.teams.stream_host_followup_to_room(
        ChatRequest(messages="wrap up", thread_id="thr_parent", agent_id="host"),
        room_thread_id="thr_parent",
        speaker_id="host",
    )
    assert result["team_live_streamed"] is False
    assert "thr_parent" in processor.teams._live_host_replies
    tokens = [frame for frame in frames if frame.get("type") == "token"]
    assert tokens
    assert tokens[0].get("content") == "可以收工。"
    assert tokens[0].get("team_wrapup") is True
    assert tokens[0].get("agent_id") == "host"


@pytest.mark.asyncio
async def test_record_peer_turn_persists_room_user_question(processor_env: dict) -> None:
    from langchain_core.messages import AIMessage, HumanMessage
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    repo = MagicMock()
    repo.append_if_ready = MagicMock(return_value=1)
    repo.projection_status = MagicMock(return_value="pending")
    repo.mark_projection = MagicMock()
    repo.page = MagicMock(return_value=([], False))
    processor.replace_thread_message_repo(repo)
    processor._agent_manager.get_agent.return_value = SimpleNamespace(
        aget_history=AsyncMock(return_value=[HumanMessage(content="我最近总失眠")])
    )

    await processor.record_peer_turn(
        PeerCall(
            from_agent_id="host",
            to_agent_id="child",
            user_id=1,
            message="请根据用户问题给出睡眠建议",
            source_thread_id="thr_parent",
            source_session_key=None,
        ),
        "thr_parent~child",
        {"messages": [AIMessage(content="建议早睡")]},
    )

    member_calls = [
        call for call in repo.append_if_ready.call_args_list if call.args[0] == "thr_parent~child"
    ]
    assert member_calls
    inputs = member_calls[0].args[1]
    roles = [item.role for item in inputs]
    assert any(role in {"human", "user"} for role in roles)
    assert any("请根据用户问题给出睡眠建议" in item.message_json for item in inputs)
    assert all("我最近总失眠" not in item.message_json for item in inputs)
    room_calls = [
        call for call in repo.append_if_ready.call_args_list if call.args[0] == "thr_parent"
    ]
    assert room_calls
    room_inputs = room_calls[0].args[1]
    assert any(
        item.role in {"ai", "assistant"} and "建议早睡" in item.message_json for item in room_inputs
    )
    repo.mark_projection.assert_called_with("thr_parent~child", "ready")


@pytest.mark.asyncio
async def test_record_peer_turn_keeps_sync_ask_agent_message(processor_env: dict) -> None:
    from langchain_core.messages import AIMessage, HumanMessage
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    repo = MagicMock()
    repo.append_if_ready = MagicMock(return_value=1)
    repo.projection_status = MagicMock(return_value="pending")
    repo.mark_projection = MagicMock()
    repo.page = MagicMock(return_value=([], False))
    processor.replace_thread_message_repo(repo)
    processor._agent_manager.get_agent.return_value = SimpleNamespace(
        aget_history=AsyncMock(return_value=[HumanMessage(content="用户对 A 说的话")])
    )

    await processor.record_peer_turn(
        PeerCall(
            from_agent_id="parent",
            to_agent_id="child",
            user_id=1,
            message="请根据任务给 B 的指令",
            source_thread_id="thr_parent",
            source_session_key=None,
        ),
        "thr_parent~child",
        {"messages": [AIMessage(content="工具结果")]},
    )

    member_calls = [
        call for call in repo.append_if_ready.call_args_list if call.args[0] == "thr_parent~child"
    ]
    assert member_calls
    payload = member_calls[0].args[1]
    joined = " ".join(item.message_json for item in payload)
    assert "请根据任务给 B 的指令" in joined
    assert "用户对 A 说的话" not in joined


@pytest.mark.asyncio
async def test_ensure_projection_leaves_pending_threads_with_history(
    processor_env: dict,
) -> None:
    processor = processor_env["processor"]
    repo = MagicMock()
    repo.projection_status = MagicMock(return_value="pending")
    repo.mark_projection = MagicMock()
    repo.page = MagicMock(return_value=([object()], False))
    processor.replace_thread_message_repo(repo)
    processor.teams._ensure_projection("thr_legacy")
    repo.mark_projection.assert_not_called()


@pytest.mark.asyncio
async def test_prepare_peer_session_creates_callee_thread_without_rebind(
    processor_env: dict,
) -> None:
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    parent_sk = processor_env["parent_sk"]
    prepared = await processor.prepare_peer_session(
        PeerCall(
            from_agent_id="parent",
            to_agent_id="child",
            user_id=1,
            message="hello",
            source_thread_id="thr_parent",
            source_session_key=str(parent_sk),
        )
    )
    assert prepared is not None
    assert prepared.thread_id == "thr_parent~child"
    assert prepared.session_key == "child:dashboard:1:peer:thr_parent"
    registry = processor_env["gateway"].thread_registry
    row = registry.get_thread("thr_parent~child")
    assert row is not None
    assert row.agent_id == "child"
    assert registry.get_bound_thread_id(str(parent_sk)) == "thr_parent"


@pytest.mark.asyncio
async def test_prepare_team_peer_seeds_host_assignment(processor_env: dict) -> None:
    from langchain_core.messages import HumanMessage
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    parent_sk = processor_env["parent_sk"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    processor._agent_manager.teams.member_ids.return_value = ["child"]
    repo = MagicMock()
    repo.append_if_ready = MagicMock(return_value=1)
    repo.projection_status = MagicMock(return_value="pending")
    repo.mark_projection = MagicMock()
    repo.page = MagicMock(return_value=([], False))
    processor.replace_thread_message_repo(repo)
    processor._agent_manager.get_agent.return_value = SimpleNamespace(
        aget_history=AsyncMock(return_value=[HumanMessage(content="我最近总失眠")])
    )

    await processor.prepare_peer_session(
        PeerCall(
            from_agent_id="host",
            to_agent_id="child",
            user_id=1,
            message="请给出睡眠建议",
            source_thread_id="thr_parent",
            source_session_key=str(parent_sk),
        )
    )

    repo.mark_projection.assert_called_with("thr_parent~child", "ready")
    seeded = repo.append_if_ready.call_args.args[1]
    assert seeded[0].role in {"human", "user"}
    assert "请给出睡眠建议" in seeded[0].message_json
    assert "我最近总失眠" not in seeded[0].message_json
    assert seeded[0].message_id.startswith("team-peer:thr_parent~child:")
    assert seeded[0].message_id.endswith(":human")


@pytest.mark.asyncio
async def test_prepare_team_peer_uses_isolated_session_key(processor_env: dict) -> None:
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    parent_sk = processor_env["parent_sk"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    processor._agent_manager.teams.member_ids.return_value = ["child"]
    prepared = await processor.prepare_peer_session(
        PeerCall(
            from_agent_id="host",
            to_agent_id="child",
            user_id=1,
            message="hello",
            source_thread_id="thr_parent",
            source_session_key=str(parent_sk),
            job_id="job-1",
        )
    )
    assert prepared is not None
    assert prepared.session_key == "child:dashboard:1:team:thr_parent"
    registry = processor_env["gateway"].thread_registry
    child_dm = registry.dashboard_key(agent_id="child", user_id=1)
    assert registry.get_bound_thread_id(child_dm) is None
    assert registry.get_bound_thread_id(str(parent_sk)) == "thr_parent"


@pytest.mark.asyncio
async def test_prepare_followup_seeds_new_user_question(processor_env: dict) -> None:
    from langchain_core.messages import HumanMessage
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    parent_sk = processor_env["parent_sk"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    processor._agent_manager.teams.member_ids.return_value = ["child"]
    repo = MagicMock()
    repo.append_if_ready = MagicMock(return_value=1)
    repo.projection_status = MagicMock(return_value="ready")
    repo.page = MagicMock(
        return_value=(
            [SimpleNamespace(role="human", message_json='{"content": "上一轮的问题"}')],
            False,
        )
    )
    processor.replace_thread_message_repo(repo)
    processor._agent_manager.get_agent.return_value = SimpleNamespace(
        aget_history=AsyncMock(return_value=[HumanMessage(content="请改成早睡建议")])
    )

    first = await processor.prepare_peer_session(
        PeerCall(
            from_agent_id="host",
            to_agent_id="child",
            user_id=1,
            message="继续睡眠建议",
            source_thread_id="thr_parent",
            source_session_key=str(parent_sk),
            job_id="job-1",
        )
    )
    second = await processor.prepare_peer_session(
        PeerCall(
            from_agent_id="host",
            to_agent_id="child",
            user_id=1,
            message="继续睡眠建议",
            source_thread_id="thr_parent",
            source_session_key=str(parent_sk),
            job_id="job-2",
        )
    )
    assert first is not None and second is not None
    seeded = [call.args[1][0] for call in repo.append_if_ready.call_args_list]
    assert [item.message_id for item in seeded] == [
        "team-peer:thr_parent~child:job-1:human",
        "team-peer:thr_parent~child:job-2:human",
    ]
    assert all("继续睡眠建议" in item.message_json for item in seeded)
    assert all("请改成早睡建议" not in item.message_json for item in seeded)
    assert all("上一轮的问题" not in item.message_json for item in seeded)


@pytest.mark.asyncio
async def test_prepare_team_dispatch_tracks_job_and_forces_sync(
    processor_env: dict,
) -> None:
    from octop_harness.teams.util import PeerCall

    from octop.infra.agents.teams import TeamJobTracker

    processor = processor_env["processor"]
    parent_sk = processor_env["parent_sk"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    jobs = TeamJobTracker()
    processor._agent_manager._team_jobs = jobs
    processor._agent_manager.teams.member_ids.return_value = ["child", "other"]
    prepared = await processor.prepare_peer_session(
        PeerCall(
            from_agent_id="host",
            to_agent_id="child",
            user_id=1,
            message="hello",
            source_thread_id="thr_parent",
            source_session_key=str(parent_sk),
        )
    )
    assert jobs.is_busy("host", "child")
    assert prepared is not None
    extra = getattr(prepared, "configurable", None)
    if extra is not None:
        assert extra == {
            "peer_invoke_mode": "sync",
            "team_peers": ["other"],
        }


@pytest.mark.asyncio
async def test_prepare_team_dispatch_includes_room_history(
    processor_env: dict,
) -> None:
    from langchain_core.messages import AIMessage, HumanMessage
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    parent_sk = processor_env["parent_sk"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    processor._agent_manager.teams.member_ids.return_value = ["child", "other"]
    harness = MagicMock()
    harness.aget_history = AsyncMock(
        return_value=[
            HumanMessage(content="what is the market?"),
            AIMessage(content="I will ask Researcher"),
        ]
    )
    processor._agent_manager.get_agent.return_value = harness
    prepared = await processor.prepare_peer_session(
        PeerCall(
            from_agent_id="host",
            to_agent_id="child",
            user_id=1,
            message="survey market",
            source_thread_id="thr_parent",
            source_session_key=str(parent_sk),
        )
    )
    assert prepared is not None
    assert getattr(prepared, "message", None) == "survey market"
    pair = processor.teams.peek_peer_prompt("child", "thr_parent~child")
    assert pair is not None
    question, dispatch = pair
    assert question == "survey market"
    assert "what is the market?" in dispatch
    assert "I will ask Researcher" in dispatch
    assert "Team assignment" in dispatch or "团队派工" in dispatch
    taken = processor.teams.take_peer_prompt("child", "thr_parent~child")
    assert taken == pair
    assert harness.aget_history.await_count >= 1


@pytest.mark.asyncio
async def test_prepare_team_dispatch_uses_projected_member_speech(
    processor_env: dict,
) -> None:
    import json

    from langchain_core.messages import AIMessage, HumanMessage
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    processor._agent_repo.create(agent_id="other", user_id=1, name="Analyst")
    processor._agent_manager.teams.member_ids.return_value = ["child", "other"]
    harness = MagicMock()
    harness.aget_history = AsyncMock(
        return_value=[
            HumanMessage(content="what is the market?"),
            AIMessage(content="I will ask Researcher"),
        ]
    )
    processor._agent_manager.get_agent.return_value = harness
    repo = MagicMock()
    repo.projection_status.return_value = "ready"
    repo.page.return_value = (
        [
            SimpleNamespace(
                role="human",
                message_json=json.dumps({"content": "what is the market?"}),
            ),
            SimpleNamespace(
                role="ai",
                message_json=json.dumps(
                    {
                        "content": "TAM is 12B",
                        "additional_kwargs": {"speaker_agent_id": "child"},
                    }
                ),
            ),
        ],
        False,
    )
    processor.replace_thread_message_repo(repo)

    prepared = await processor.prepare_peer_session(
        PeerCall(
            from_agent_id="host",
            to_agent_id="other",
            user_id=1,
            message="size the TAM",
            source_thread_id="thr_parent",
            source_session_key=str(processor_env["parent_sk"]),
        )
    )
    assert prepared is not None
    pair = processor.teams.peek_peer_prompt("other", "thr_parent~other")
    assert pair is not None
    question, dispatch = pair
    assert question == "size the TAM"
    assert "TAM is 12B" in dispatch
    assert "Researcher" in dispatch
    assert "what is the market?" in dispatch
    assert harness.aget_history.await_count == 0


def test_stamp_team_host_runtime_forces_async(processor_env: dict) -> None:
    processor = processor_env["processor"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    processor._agent_manager.strip_team_host_runtime_tools = MagicMock()
    request: dict[str, object] = {
        "configurable": {"session_key": "sk", "mcp_use_default": True},
        "mcp_servers": ["tencent-news__1"],
        "mcp_use_default": True,
    }
    processor.teams.stamp_host_runtime(request, "host")
    assert request["configurable"] == {
        "session_key": "sk",
        "peer_invoke_mode": "async",
    }
    assert "mcp_servers" not in request
    assert "mcp_use_default" not in request
    processor._agent_manager.strip_team_host_runtime_tools.assert_called_once_with("host")
    expert_req: dict[str, object] = {}
    processor.teams.stamp_host_runtime(expert_req, "parent")
    assert "configurable" not in expert_req


@pytest.mark.asyncio
async def test_stream_team_peer_relays_tokens_to_room(processor_env: dict) -> None:
    from langchain_core.messages import AIMessage
    from octop_harness.request import ChatRequest

    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    frames: list[dict[str, object]] = []

    async def capture(frame: dict[str, object]) -> None:
        frames.append(frame)

    gateway.ws_hub.register("conn-room", capture)
    gateway.ws_hub.subscribe("thr_parent", "conn-room")

    async def fake_stream(_agent_id: str, _request: dict[str, object]) -> object:
        yield {"type": "token", "content": "hello "}
        yield {"type": "token", "content": "doc"}
        yield {"type": "usage", "usage": {}}

    processor._agent_manager.stream = fake_stream
    harness = MagicMock()
    harness.aget_history = AsyncMock(return_value=[AIMessage(content="hello doc")])
    processor._agent_manager.get_agent.return_value = harness

    result = await processor.teams.stream_peer_to_room(
        ChatRequest(messages="task", thread_id="thr_parent~child", agent_id="child"),
        room_thread_id="thr_parent",
        speaker_id="child",
    )
    assert result["team_live_streamed"] is True
    tokens = [frame.get("content") for frame in frames if frame.get("type") == "token"]
    assert tokens == ["hello ", "doc"]
    assert all(
        frame.get("agent_id") == "child" and frame.get("agent") == "child"
        for frame in frames
        if frame.get("type") in {"token", "done"}
    )
    assert frames[-1].get("type") == "done"
    harness.aget_history.assert_awaited()


@pytest.mark.asyncio
async def test_live_member_catchup_replays_partial_then_clears(
    processor_env: dict,
) -> None:
    from octop_harness.request import ChatRequest

    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    started = asyncio.Event()
    release = asyncio.Event()

    async def fake_stream(_agent_id: str, _request: dict[str, object]) -> object:
        yield {"type": "token", "content": "hello "}
        started.set()
        await release.wait()
        yield {"type": "token", "content": "doc"}

    processor._agent_manager.stream = fake_stream
    harness = MagicMock()
    harness.aget_history = AsyncMock(return_value=[])
    processor._agent_manager.get_agent.return_value = harness

    task = asyncio.create_task(
        processor.teams.stream_peer_to_room(
            ChatRequest(messages="task", thread_id="thr_parent~child", agent_id="child"),
            room_thread_id="thr_parent",
            speaker_id="child",
        )
    )
    await asyncio.wait_for(started.wait(), timeout=2)

    caught: list[dict[str, object]] = []

    async def send(frame: dict[str, object]) -> None:
        caught.append(frame)

    await processor.teams.deliver_live_member_catchup(
        "thr_parent",
        subscribe=lambda: None,
        send=send,
    )
    assert caught == [
        {
            "type": "token",
            "content": "hello ",
            "agent_id": "child",
            "agent": "child",
            "team_snapshot": True,
            "team_live": True,
            "thread_id": "thr_parent",
        }
    ]
    release.set()
    await asyncio.wait_for(task, timeout=2)
    caught.clear()
    await processor.teams.deliver_live_member_catchup(
        "thr_parent",
        subscribe=lambda: gateway.ws_hub.subscribe("thr_parent", "late"),
        send=send,
    )
    assert caught == []


@pytest.mark.asyncio
async def test_stream_team_peer_relays_tokens_to_member_and_room(
    processor_env: dict,
) -> None:
    from langchain_core.messages import AIMessage
    from octop_harness.request import ChatRequest

    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    room_frames: list[dict[str, object]] = []
    member_frames: list[dict[str, object]] = []

    async def capture_room(frame: dict[str, object]) -> None:
        room_frames.append(frame)

    async def capture_member(frame: dict[str, object]) -> None:
        member_frames.append(frame)

    gateway.ws_hub.register("conn-room", capture_room)
    gateway.ws_hub.subscribe("thr_parent", "conn-room")
    gateway.ws_hub.register("conn-member", capture_member)
    gateway.ws_hub.subscribe("thr_parent~child", "conn-member")

    async def fake_stream(_agent_id: str, _request: dict[str, object]) -> object:
        yield {"type": "token", "content": "please rest"}
        yield {"type": "done"}

    processor._agent_manager.stream = fake_stream
    harness = MagicMock()
    harness.aget_history = AsyncMock(return_value=[AIMessage(content="please rest")])
    processor._agent_manager.get_agent.return_value = harness

    await processor.teams.stream_peer_to_room(
        ChatRequest(messages="task", thread_id="thr_parent~child", agent_id="child"),
        room_thread_id="thr_parent",
        speaker_id="child",
    )
    room_tokens = [frame for frame in room_frames if frame.get("type") == "token"]
    member_tokens = [frame for frame in member_frames if frame.get("type") == "token"]
    assert room_tokens[0].get("content") == "please rest"
    assert room_tokens[0].get("agent_id") == "child"
    assert member_tokens[0].get("content") == "please rest"
    assert member_tokens[0].get("thread_id") == "thr_parent~child"


@pytest.mark.asyncio
async def test_stream_team_peer_unwatched_is_not_live_streamed(
    processor_env: dict,
) -> None:
    from langchain_core.messages import AIMessage
    from octop_harness.request import ChatRequest

    processor = processor_env["processor"]

    async def fake_stream(_agent_id: str, _request: dict[str, object]) -> object:
        yield {"type": "token", "content": "late"}

    processor._agent_manager.stream = fake_stream
    harness = MagicMock()
    harness.aget_history = AsyncMock(return_value=[AIMessage(content="late")])
    processor._agent_manager.get_agent.return_value = harness

    result = await processor.teams.stream_peer_to_room(
        ChatRequest(messages="task", thread_id="thr_parent~child", agent_id="child"),
        room_thread_id="thr_parent",
        speaker_id="child",
    )
    assert result["team_live_streamed"] is False


@pytest.mark.asyncio
async def test_stream_team_peer_relays_reasoning_and_tools(processor_env: dict) -> None:
    from langchain_core.messages import AIMessage, HumanMessage
    from octop_harness.request import ChatRequest

    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    frames: list[dict[str, object]] = []

    async def capture(frame: dict[str, object]) -> None:
        frames.append(frame)

    gateway.ws_hub.register("conn-room", capture)
    gateway.ws_hub.subscribe("thr_parent", "conn-room")

    async def fake_stream(_agent_id: str, _request: dict[str, object]) -> object:
        yield {"type": "reasoning", "content": "check labs"}
        yield {"type": "tool_call_chunk", "id": "c1", "name": "read_file", "args": "{}"}
        yield {"type": "tool_result", "messages": [{"tool_call_id": "c1", "content": "ok"}]}
        yield {"type": "token", "content": "rest"}

    processor._agent_manager.stream = fake_stream
    harness = MagicMock()
    harness.aget_history = AsyncMock(
        return_value=[
            HumanMessage(content="please advise"),
            AIMessage(content="rest"),
        ]
    )
    processor._agent_manager.get_agent.return_value = harness

    result = await processor.teams.stream_peer_to_room(
        ChatRequest(messages="task", thread_id="thr_parent~child", agent_id="child"),
        room_thread_id="thr_parent",
        speaker_id="child",
    )
    kinds = [frame.get("type") for frame in frames]
    assert "reasoning" in kinds
    assert "tool_call_chunk" in kinds
    assert "tool_result" in kinds
    assert all(
        frame.get("agent_id") == "child" and frame.get("agent") == "child"
        for frame in frames
        if frame.get("type") in {"reasoning", "tool_call_chunk", "tool_result", "token", "done"}
    )
    messages = result["messages"]
    assert messages
    assert all(getattr(msg, "type", None) != "human" for msg in messages)
    last = messages[-1]
    assert getattr(last, "additional_kwargs", {}).get("speaker_agent_id") == "child"
    assert getattr(last, "additional_kwargs", {}).get("reasoning_content") == "check labs"


@pytest.mark.asyncio
async def test_member_tokens_stay_off_the_room_until_the_host_is_idle(
    processor_env: dict, monkeypatch: pytest.MonkeyPatch
) -> None:
    from langchain_core.messages import AIMessage
    from octop_harness.request import ChatRequest

    import octop.infra.agents.teams.team_manager as team_mod

    monkeypatch.setattr(team_mod, "_HOST_IDLE_WAIT_SEC", 0)
    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    room_frames: list[dict[str, object]] = []
    member_frames: list[dict[str, object]] = []

    async def capture_room(frame: dict[str, object]) -> None:
        room_frames.append(frame)

    async def capture_member(frame: dict[str, object]) -> None:
        member_frames.append(frame)

    gateway.ws_hub.register("conn-room", capture_room)
    gateway.ws_hub.subscribe("thr_parent", "conn-room")
    gateway.ws_hub.register("conn-member", capture_member)
    gateway.ws_hub.subscribe("thr_parent~child", "conn-member")
    gateway.ws_hub.mark_turn_active("thr_parent")
    gate = asyncio.Event()

    async def fake_stream(_agent_id: str, _request: dict[str, object]) -> object:
        yield {"type": "token", "content": "first"}
        await gate.wait()
        yield {"type": "token", "content": "second"}

    processor._agent_manager.stream = fake_stream
    harness = MagicMock()
    harness.aget_history = AsyncMock(return_value=[AIMessage(content="first second")])
    processor._agent_manager.get_agent.return_value = harness

    task = asyncio.create_task(
        processor.teams.stream_peer_to_room(
            ChatRequest(messages="task", thread_id="thr_parent~child", agent_id="child", user="1"),
            room_thread_id="thr_parent",
            speaker_id="child",
        )
    )
    for _ in range(50):
        if any(frame.get("content") == "first" for frame in member_frames):
            break
        await asyncio.sleep(0.01)
    assert any(frame.get("content") == "first" for frame in member_frames)
    assert not any(frame.get("type") == "token" for frame in room_frames)
    gateway.ws_hub.mark_turn_idle("thr_parent")
    gate.set()
    result = await task
    room_tokens = [frame.get("content") for frame in room_frames if frame.get("type") == "token"]
    assert room_tokens == ["first", "second"]
    assert all(
        frame.get("agent_id") == "child" for frame in room_frames if frame.get("type") == "token"
    )
    assert result["team_live_streamed"] is True


@pytest.mark.asyncio
async def test_member_hitl_waits_for_each_approval_before_finishing(processor_env: dict) -> None:
    from langchain_core.messages import AIMessage
    from octop_harness.request import ChatRequest

    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    room_frames: list[dict[str, object]] = []

    async def capture_room(frame: dict[str, object]) -> None:
        room_frames.append(frame)

    gateway.ws_hub.register("conn-room", capture_room)
    gateway.ws_hub.subscribe("thr_parent", "conn-room")
    resumes = {"n": 0}

    async def fake_stream(_agent_id: str, _request: dict[str, object]) -> object:
        yield {"type": "token", "content": "checking "}
        yield {
            "type": "hitl_required",
            "request": {
                "action_requests": [{"name": "execute", "args": {"command": "df"}}],
                "review_configs": [
                    {"action_name": "execute", "allowed_decisions": ["approve", "reject"]}
                ],
            },
        }

    async def fake_resume(agent_id: str, thread_id: str, decisions: list[dict[str, str]]) -> object:
        assert agent_id == "child"
        assert thread_id == "thr_parent~child"
        resumes["n"] += 1
        if resumes["n"] == 1:
            assert decisions == [{"type": "approve"}]
            yield {
                "type": "hitl_required",
                "request": {
                    "action_requests": [{"name": "write_file", "args": {"path": "note.txt"}}],
                },
            }
            return
        assert decisions == [{"type": "reject"}]
        yield {"type": "token", "content": "8G free"}

    processor._agent_manager.stream = fake_stream
    processor._agent_manager.resume_hitl = fake_resume
    harness = MagicMock()
    harness.aget_history = AsyncMock(return_value=[AIMessage(content="checking 8G free")])
    processor._agent_manager.get_agent.return_value = harness

    task = asyncio.create_task(
        processor.teams.stream_peer_to_room(
            ChatRequest(messages="task", thread_id="thr_parent~child", agent_id="child", user="1"),
            room_thread_id="thr_parent",
            speaker_id="child",
        )
    )
    answered: list[str] = []
    for _ in range(50):
        pending_id = next(
            (
                str(frame.get("pending_id"))
                for frame in room_frames
                if frame.get("type") == "hitl_required"
                and str(frame.get("pending_id")) not in answered
            ),
            "",
        )
        if pending_id:
            kind = "approve" if len(answered) == 0 else "reject"
            assert processor.teams.deliver_member_hitl(pending_id, [{"type": kind}])
            answered.append(pending_id)
            if len(answered) == 2:
                break
            continue
        assert not task.done()
        await asyncio.sleep(0.01)
    assert len(answered) == 2
    result = await task
    assert resumes["n"] == 2
    assert "checking" in result["stream_text"]
    assert "8G free" in result["stream_text"]
    assert task.done()


@pytest.mark.asyncio
async def test_member_hitl_timeout_does_not_finish_the_job(
    processor_env: dict, monkeypatch: pytest.MonkeyPatch
) -> None:
    from octop_harness.request import ChatRequest

    import octop.infra.agents.teams.team_manager as team_mod

    monkeypatch.setattr(team_mod, "_MEMBER_HITL_WAIT_SEC", 0.05)
    processor = processor_env["processor"]

    async def fake_stream(_agent_id: str, _request: dict[str, object]) -> object:
        yield {
            "type": "hitl_required",
            "request": {"action_requests": [{"name": "execute", "args": {"command": "df"}}]},
        }

    processor._agent_manager.stream = fake_stream
    processor._agent_manager.resume_hitl = AsyncMock()
    with pytest.raises(TimeoutError):
        await processor.teams.stream_peer_to_room(
            ChatRequest(messages="task", thread_id="thr_parent~child", agent_id="child", user="1"),
            room_thread_id="thr_parent",
            speaker_id="child",
        )
    processor._agent_manager.resume_hitl.assert_not_called()


@pytest.mark.asyncio
async def test_member_model_does_not_start_while_the_host_turn_is_active(
    processor_env: dict,
) -> None:
    from langchain_core.messages import AIMessage
    from octop_harness.request import ChatRequest

    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    gateway.ws_hub.mark_turn_active("thr_parent")
    started = asyncio.Event()

    async def fake_stream(_agent_id: str, _request: dict[str, object]) -> object:
        started.set()
        yield {"type": "token", "content": "after host"}

    processor._agent_manager.stream = fake_stream
    harness = MagicMock()
    harness.aget_history = AsyncMock(return_value=[AIMessage(content="after host")])
    processor._agent_manager.get_agent.return_value = harness

    task = asyncio.create_task(
        processor.teams.stream_peer_to_room(
            ChatRequest(messages="task", thread_id="thr_parent~child", agent_id="child", user="1"),
            room_thread_id="thr_parent",
            speaker_id="child",
        )
    )
    await asyncio.sleep(0.15)
    assert not started.is_set()
    gateway.ws_hub.mark_turn_idle("thr_parent")
    result = await task
    assert started.is_set()
    assert result["stream_text"] == "after host"


@pytest.mark.asyncio
async def test_replacing_the_room_turn_cancels_a_member_approval(processor_env: dict) -> None:
    from octop_harness.request import ChatRequest

    processor = processor_env["processor"]

    async def fake_stream(_agent_id: str, _request: dict[str, object]) -> object:
        yield {
            "type": "hitl_required",
            "request": {"action_requests": [{"name": "execute", "args": {"command": "df"}}]},
        }

    processor._agent_manager.stream = fake_stream
    processor._agent_manager.resume_hitl = AsyncMock()
    task = asyncio.create_task(
        processor.teams.stream_peer_to_room(
            ChatRequest(messages="task", thread_id="thr_parent~child", agent_id="child", user="1"),
            room_thread_id="thr_parent",
            speaker_id="child",
        )
    )
    for _ in range(50):
        if processor.teams._hitl_waiters:
            break
        await asyncio.sleep(0.01)
    assert processor.teams._hitl_waiters
    processor.teams.cancel_member_hitl("thr_parent")
    with pytest.raises(TimeoutError):
        await task
    processor._agent_manager.resume_hitl.assert_not_called()


@pytest.mark.asyncio
async def test_fan_in_persists_member_turn_without_dispatch_prompt(
    processor_env: dict,
) -> None:
    from langchain_core.messages import AIMessage, HumanMessage
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    repo = MagicMock()
    repo.append_if_ready = MagicMock(return_value=1)
    processor.replace_thread_message_repo(repo)

    await processor.teams.fan_in_peer_turn(
        PeerCall(
            from_agent_id="host",
            to_agent_id="child",
            user_id=1,
            message="ask",
            source_thread_id="thr_parent",
            source_session_key=None,
        ),
        [
            HumanMessage(content="internal dispatch"),
            AIMessage(content="", tool_calls=[{"name": "read_file", "id": "c1", "args": {}}]),
            AIMessage(content="please rest"),
        ],
        live_streamed=True,
    )
    repo.append_if_ready.assert_called_once()
    thread_id, inputs = repo.append_if_ready.call_args.args
    assert thread_id == "thr_parent"
    roles = [item.role for item in inputs]
    assert roles[0] in {"ai", "assistant"}
    assert any(
        item.role in {"ai", "assistant"} and "please rest" in item.message_json for item in inputs
    )
    assert any("read_file" in item.message_json for item in inputs)
    assert all("internal dispatch" not in item.message_json for item in inputs)


@pytest.mark.asyncio
async def test_fan_in_stamps_edited_files_on_room_history(
    processor_env: dict,
) -> None:
    from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    repo = MagicMock()
    repo.append_if_ready = MagicMock(return_value=1)
    processor.replace_thread_message_repo(repo)

    await processor.teams.fan_in_peer_turn(
        PeerCall(
            from_agent_id="host",
            to_agent_id="child",
            user_id=1,
            message="ask",
            source_thread_id="thr_parent",
            source_session_key=None,
        ),
        [
            HumanMessage(content="internal dispatch"),
            AIMessage(
                content="",
                tool_calls=[
                    {
                        "name": "write_file",
                        "id": "c1",
                        "args": {"path": "notes/plan.md"},
                    },
                    {
                        "name": "edit_file",
                        "id": "c2",
                        "args": {"path": "notes/plan.md"},
                    },
                    {
                        "name": "read_file",
                        "id": "c3",
                        "args": {"path": "SOUL.md"},
                    },
                ],
            ),
            ToolMessage(content="wrote plan", tool_call_id="c1", name="write_file"),
            ToolMessage(content="edited plan", tool_call_id="c2", name="edit_file"),
            ToolMessage(content="soul", tool_call_id="c3", name="read_file"),
            AIMessage(content="wrote the plan"),
        ],
        live_streamed=True,
    )
    repo.append_if_ready.assert_called_once()
    _thread_id, inputs = repo.append_if_ready.call_args.args
    roles = [item.role for item in inputs]
    assert "tool" in roles
    assert roles.count("tool") == 3
    final = next(
        item
        for item in inputs
        if item.role in {"ai", "assistant"} and "wrote the plan" in item.message_json
    )
    wire = json.loads(final.message_json)
    data = wire.get("data") if isinstance(wire, dict) else wire
    kwargs = (data or {}).get("additional_kwargs") or {}
    assert kwargs.get("speaker_agent_id") == "child"
    assert kwargs.get("edited_files") == ["notes/plan.md"]
    tool_wire = json.loads(next(item for item in inputs if item.role == "tool").message_json)
    tool_data = tool_wire.get("data") if isinstance(tool_wire, dict) else tool_wire
    assert (tool_data or {}).get("additional_kwargs", {}).get("speaker_agent_id") == "child"


@pytest.mark.asyncio
async def test_record_peer_turn_keeps_tool_trail_on_member_page(
    processor_env: dict,
) -> None:
    from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    repo = MagicMock()
    repo.append_if_ready = MagicMock(return_value=1)
    repo.projection_status = MagicMock(return_value="pending")
    repo.mark_projection = MagicMock()
    repo.page = MagicMock(return_value=([], False))
    processor.replace_thread_message_repo(repo)
    processor._agent_manager.get_agent.return_value = SimpleNamespace(
        aget_history=AsyncMock(return_value=[HumanMessage(content="用户原话")])
    )

    await processor.record_peer_turn(
        PeerCall(
            from_agent_id="host",
            to_agent_id="child",
            user_id=1,
            message="请写计划",
            source_thread_id="thr_parent",
            source_session_key=None,
            job_id="job1",
        ),
        "thr_parent~child",
        {
            "messages": [
                HumanMessage(content="请写计划"),
                AIMessage(
                    content="",
                    tool_calls=[
                        {
                            "name": "write_file",
                            "id": "c1",
                            "args": {"path": "plan.md"},
                        }
                    ],
                ),
                ToolMessage(content="ok", tool_call_id="c1", name="write_file"),
                AIMessage(content="写好了"),
            ]
        },
    )

    member_calls = [
        call for call in repo.append_if_ready.call_args_list if call.args[0] == "thr_parent~child"
    ]
    assert member_calls
    inputs = member_calls[0].args[1]
    roles = [item.role for item in inputs]
    assert "tool" in roles
    assert any(
        item.role in {"ai", "assistant"} and "write_file" in item.message_json for item in inputs
    )
    assert any(
        item.role in {"ai", "assistant"} and "写好了" in item.message_json for item in inputs
    )
    room_calls = [
        call for call in repo.append_if_ready.call_args_list if call.args[0] == "thr_parent"
    ]
    assert room_calls
    room_roles = [item.role for item in room_calls[0].args[1]]
    assert "tool" in room_roles


@pytest.mark.asyncio
async def test_fan_in_truncates_large_tool_results(
    processor_env: dict,
) -> None:
    from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
    from octop_harness.teams.util import PeerCall

    from octop.infra.agents.teams.team_manager import _HISTORY_TOOL_RESULT_MAX_CHARS

    processor = processor_env["processor"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    repo = MagicMock()
    repo.append_if_ready = MagicMock(return_value=1)
    processor.replace_thread_message_repo(repo)
    huge = "x" * (_HISTORY_TOOL_RESULT_MAX_CHARS + 500)

    await processor.teams.fan_in_peer_turn(
        PeerCall(
            from_agent_id="host",
            to_agent_id="child",
            user_id=1,
            message="ask",
            source_thread_id="thr_parent",
            source_session_key=None,
        ),
        [
            HumanMessage(content="dispatch"),
            AIMessage(
                content="",
                tool_calls=[{"name": "read_file", "id": "c1", "args": {"path": "big.md"}}],
            ),
            ToolMessage(content=huge, tool_call_id="c1", name="read_file"),
            AIMessage(content="done"),
        ],
        live_streamed=True,
    )
    _thread_id, inputs = repo.append_if_ready.call_args.args
    tool = next(item for item in inputs if item.role == "tool")
    wire = json.loads(tool.message_json)
    data = wire.get("data") if isinstance(wire, dict) else wire
    content = (data or {}).get("content") or ""
    assert len(content) <= _HISTORY_TOOL_RESULT_MAX_CHARS
    assert content.endswith("…")
    assert huge not in content


@pytest.mark.asyncio
async def test_fan_in_stamps_edited_files_when_final_text_missing(
    processor_env: dict,
) -> None:
    from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    repo = MagicMock()
    repo.append_if_ready = MagicMock(return_value=1)
    processor.replace_thread_message_repo(repo)

    await processor.teams.fan_in_peer_turn(
        PeerCall(
            from_agent_id="host",
            to_agent_id="child",
            user_id=1,
            message="ask",
            source_thread_id="thr_parent",
            source_session_key=None,
        ),
        [
            HumanMessage(content="dispatch"),
            AIMessage(
                content="",
                tool_calls=[
                    {
                        "name": "write_file",
                        "id": "c1",
                        "args": {"path": "notes/only.md"},
                    }
                ],
            ),
            ToolMessage(content="ok", tool_call_id="c1", name="write_file"),
        ],
        live_streamed=True,
    )
    _thread_id, inputs = repo.append_if_ready.call_args.args
    assert inputs
    ai = next(item for item in inputs if item.role in {"ai", "assistant"})
    wire = json.loads(ai.message_json)
    data = wire.get("data") if isinstance(wire, dict) else wire
    kwargs = (data or {}).get("additional_kwargs") or {}
    assert kwargs.get("edited_files") == ["notes/only.md"]


@pytest.mark.asyncio
async def test_fan_in_skips_snapshot_after_live_stream(processor_env: dict) -> None:
    from langchain_core.messages import AIMessage
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    frames: list[dict[str, object]] = []

    async def capture(frame: dict[str, object]) -> None:
        frames.append(frame)

    gateway.ws_hub.register("conn-room", capture)
    gateway.ws_hub.subscribe("thr_parent", "conn-room")

    await processor.teams.fan_in_peer_turn(
        PeerCall(
            from_agent_id="host",
            to_agent_id="child",
            user_id=1,
            message="ask",
            source_thread_id="thr_parent",
            source_session_key=None,
        ),
        [AIMessage(content="already streamed")],
        live_streamed=True,
    )
    tokens = [frame for frame in frames if frame.get("type") == "token"]
    assert tokens == []


@pytest.mark.asyncio
async def test_fan_in_pushes_snapshot_when_not_live(processor_env: dict) -> None:
    from langchain_core.messages import AIMessage
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    frames: list[dict[str, object]] = []

    async def capture(frame: dict[str, object]) -> None:
        frames.append(frame)

    gateway.ws_hub.register("conn-gap", capture)
    gateway.ws_hub.subscribe("thr_parent", "conn-gap")

    await processor.teams.fan_in_peer_turn(
        PeerCall(
            from_agent_id="host",
            to_agent_id="child",
            user_id=1,
            message="ask",
            source_thread_id="thr_parent",
            source_session_key=None,
        ),
        [AIMessage(content="late join")],
        live_streamed=False,
    )
    tokens = [frame for frame in frames if frame.get("type") == "token"]
    assert tokens[0].get("content") == "late join"
    assert tokens[0].get("agent_id") == "child"
    assert tokens[0].get("agent") == "child"
    assert tokens[0].get("team_snapshot") is True


def _bind_im_session(processor_env: dict, *, thread_id: str = "thr_parent") -> str:
    registry = processor_env["gateway"].thread_registry
    sk = registry.make_key(
        agent_id="host",
        channel_type="feishu",
        channel_subject_id="ou_1",
    )
    registry.ensure_thread(
        thread_id=thread_id,
        agent_id="host",
        user_id=1,
        channel_type="feishu",
        session_key=sk,
    )
    registry._sessions.upsert(
        session_key=sk,
        agent_id="host",
        user_id=1,
        channel_type="feishu",
        chat_type="dm",
        thread_id=thread_id,
        channel_id="ch-feishu-1",
    )
    return sk


@pytest.mark.asyncio
async def test_prepare_team_peer_pushes_dispatch_notice_to_im(
    processor_env: dict,
) -> None:
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    processor._agent_manager.teams.member_ids.return_value = ["child"]
    _bind_im_session(processor_env)
    pushed: list[str] = []

    async def capture_push(channel_type: str, channel_id: str, subject: object, text: str) -> None:
        pushed.append(text)

    gateway.push_text = capture_push  # type: ignore[method-assign]

    await processor.prepare_peer_session(
        PeerCall(
            from_agent_id="host",
            to_agent_id="child",
            user_id=1,
            message="look at the labs",
            source_thread_id="thr_parent",
            source_session_key=None,
            job_id="job-dispatch-im",
        )
    )
    assert len(pushed) == 1
    assert "Researcher" in pushed[0]
    assert ("请稍候" in pushed[0]) or ("please wait" in pushed[0].lower())


@pytest.mark.asyncio
async def test_fan_in_pushes_member_speech_to_im_channel(processor_env: dict) -> None:
    from langchain_core.messages import AIMessage
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    _bind_im_session(processor_env)
    pushed: list[tuple[str, str, str]] = []

    async def capture_push(channel_type: str, channel_id: str, subject: object, text: str) -> None:
        pushed.append((channel_type, channel_id, text))

    gateway.push_text = capture_push  # type: ignore[method-assign]

    await processor.teams.fan_in_peer_turn(
        PeerCall(
            from_agent_id="host",
            to_agent_id="child",
            user_id=1,
            message="ask",
            source_thread_id="thr_parent",
            source_session_key=None,
        ),
        [AIMessage(content="TAM is 12B")],
        live_streamed=True,
    )
    assert pushed == [("feishu", "ch-feishu-1", "Researcher: TAM is 12B")]


@pytest.mark.asyncio
async def test_fan_in_skips_dashboard_when_pushing_channels(processor_env: dict) -> None:
    from langchain_core.messages import AIMessage
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    pushed: list[str] = []

    async def capture_push(channel_type: str, channel_id: str, subject: object, text: str) -> None:
        pushed.append(text)

    gateway.push_text = capture_push  # type: ignore[method-assign]

    await processor.teams.fan_in_peer_turn(
        PeerCall(
            from_agent_id="host",
            to_agent_id="child",
            user_id=1,
            message="ask",
            source_thread_id="thr_parent",
            source_session_key=str(processor_env["parent_sk"]),
        ),
        [AIMessage(content="only for the wall")],
        live_streamed=True,
    )
    assert pushed == []


@pytest.mark.asyncio
async def test_on_reply_live_wrapup_still_pushes_im(processor_env: dict) -> None:
    processor = processor_env["processor"]
    gateway = processor_env["gateway"]
    processor._agent_repo.create(agent_id="host", user_id=1, name="Host", kind="team")
    processor.teams._live_host_replies.add("thr_parent")
    _bind_im_session(processor_env)
    pushed: list[str] = []

    async def capture_push(channel_type: str, channel_id: str, subject: object, text: str) -> None:
        pushed.append(text)

    gateway.push_text = capture_push  # type: ignore[method-assign]

    await processor.on_reply(
        ReplyEvent(
            inbox_id="job-live-im",
            status="done",
            source_agent_id="host",
            source_thread_id="thr_parent",
            target_agent_id="child",
            user_id=1,
            reply_text="可以收工。",
            metadata={"session_key": processor_env["parent_sk"]},
        )
    )
    assert pushed == ["【主持人总结】可以收工。"] or pushed == ["[Host wrap-up] 可以收工。"]
    assert "thr_parent" not in processor.teams._live_host_replies


@pytest.mark.asyncio
async def test_fan_in_async_non_team_writes_caller_room(processor_env: dict) -> None:
    from langchain_core.messages import AIMessage
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    repo = MagicMock()
    repo.append_if_ready = MagicMock(return_value=1)
    processor.replace_thread_message_repo(repo)

    await processor.teams.fan_in_peer_turn(
        PeerCall(
            from_agent_id="parent",
            to_agent_id="child",
            user_id=1,
            message="ask",
            source_thread_id="thr_parent",
            source_session_key=None,
            job_id="job-async",
        ),
        [AIMessage(content="from the wall")],
        live_streamed=True,
    )
    repo.append_if_ready.assert_called_once()
    thread_id, inputs = repo.append_if_ready.call_args.args
    assert thread_id == "thr_parent"
    assert any("from the wall" in item.message_json for item in inputs)


@pytest.mark.asyncio
async def test_fan_in_sync_non_team_skips_caller_room(processor_env: dict) -> None:
    from langchain_core.messages import AIMessage
    from octop_harness.teams.util import PeerCall

    processor = processor_env["processor"]
    repo = MagicMock()
    repo.append_if_ready = MagicMock(return_value=1)
    processor.replace_thread_message_repo(repo)

    await processor.teams.fan_in_peer_turn(
        PeerCall(
            from_agent_id="parent",
            to_agent_id="child",
            user_id=1,
            message="ask",
            source_thread_id="thr_parent",
            source_session_key=None,
        ),
        [AIMessage(content="tool result only")],
        live_streamed=True,
    )
    repo.append_if_ready.assert_not_called()


@pytest.mark.asyncio
async def test_on_reply_failed_releases_job(processor_env: dict) -> None:
    from octop.infra.agents.teams import TeamJobTracker

    processor = processor_env["processor"]
    parent_sk = processor_env["parent_sk"]
    jobs = TeamJobTracker()
    processor._agent_manager._team_jobs = jobs
    jobs.begin("host", "child", job_id="job-9")
    frames: list[dict] = []

    async def capture(frame: dict) -> None:
        frames.append(frame)

    gateway = processor_env["gateway"]
    gateway.ws_hub.register("failed-task-watch", capture)
    gateway.ws_hub.subscribe("thr_parent", "failed-task-watch")
    processor._thread_message_repo = MagicMock()
    processor._thread_message_repo.projection_status.return_value = "ready"
    processor.teams._thread_message_repo = processor._thread_message_repo
    await processor.on_reply(
        ReplyEvent(
            inbox_id="job-9",
            status="failed",
            source_agent_id="host",
            source_thread_id="thr_parent",
            target_agent_id="child",
            user_id=1,
            error_text="Error code: 401 - token expired",
            metadata={"session_key": parent_sk},
        )
    )
    assert not jobs.is_busy("host", "child")
    assert any(
        frame.get("type") == "error" and "失败" in str(frame.get("message")) for frame in frames
    )
    assert not any(frame.get("type") == "token" for frame in frames)
    assert frames[-1]["type"] == "done"
    inputs = processor._thread_message_repo.append_if_ready.call_args.args[1]
    assert json.loads(inputs[0].message_json)["data"]["additional_kwargs"]["octop_stream_error"]


@pytest.mark.asyncio
async def test_failed_host_wrapup_is_excluded_from_memory() -> None:
    from octop.infra.agents.teams.team_manager import _patch_inbox_wrapup

    inbox = SimpleNamespace(
        _synthesize_reply=AsyncMock(return_value="fallback"),
        _processor=SimpleNamespace(
            compose_followup=lambda *args, **kwargs: "System failure notification"
        ),
    )
    stream = AsyncMock(return_value={"messages": []})
    _patch_inbox_wrapup(inbox, stream_followup=stream, is_team=lambda _: True)
    msg = InboxMessage(
        id="failed-wrapup",
        source_agent_id="host",
        source_thread_id="room",
        target_agent_id="child",
        message="work",
        user_id=1,
    )
    await inbox._synthesize_reply(msg, None, "Authentication failed")
    request = stream.call_args.args[0]
    assert request.to_runnable_config()["configurable"]["octop_memory_capture_exclude"] is True
    configurable = request.to_runnable_config()["configurable"]
    assert configurable["octop_decision_followup"] is True
    assert configurable["peer_invoke_mode"] == "async"


def test_resolve_harness_model_auto_expert_omits_model() -> None:
    processor = GlobalProcessor(
        agent_manager=MagicMock(),
        thread_registry=MagicMock(),
        audit_repo=MagicMock(),
        agent_repo=MagicMock(),
        user_repo=MagicMock(),
        connector_repo=MagicMock(),
        dispatcher=MagicMock(),
        usage_repo=MagicMock(),
        gateway=MagicMock(),
    )
    processor._agent_manager.get_thread_model.return_value = None
    processor._agent_repo.get.return_value = MagicMock(default_model=None)
    processor._agent_manager.get_config.return_value = {}
    processor._agent_manager.providers.resolve_explicit_default_model.return_value = None

    assert (
        processor._resolve_harness_model(
            "a1",
            "t1",
            None,
            needs_multimodal=True,
        )
        is None
    )


def test_resolve_harness_model_uses_agent_default_and_upgrades_vision() -> None:
    processor = GlobalProcessor(
        agent_manager=MagicMock(),
        thread_registry=MagicMock(),
        audit_repo=MagicMock(),
        agent_repo=MagicMock(),
        user_repo=MagicMock(),
        connector_repo=MagicMock(),
        dispatcher=MagicMock(),
        usage_repo=MagicMock(),
        gateway=MagicMock(),
    )
    processor._agent_manager.get_thread_model.return_value = None
    processor._agent_repo.get.return_value = MagicMock(default_model="p/text-only")
    processor._agent_manager.get_config.return_value = {}
    processor._agent_manager.providers.resolve_explicit_default_model.return_value = "p/text-only"
    processor._agent_manager.providers.resolve_model_for_multimodal_turn.return_value = "p/vision"

    resolved = processor._resolve_harness_model(
        "a1",
        "t1",
        None,
        needs_multimodal=True,
    )
    assert resolved == "p/vision"
    processor._agent_manager.providers.resolve_model_for_multimodal_turn.assert_called_once_with(
        "p/text-only",
        needs_multimodal=True,
    )


def test_resolve_harness_model_dashboard_override_wins() -> None:
    processor = GlobalProcessor(
        agent_manager=MagicMock(),
        thread_registry=MagicMock(),
        audit_repo=MagicMock(),
        agent_repo=MagicMock(),
        user_repo=MagicMock(),
        connector_repo=MagicMock(),
        dispatcher=MagicMock(),
        usage_repo=MagicMock(),
        gateway=MagicMock(),
    )
    processor._agent_manager.get_thread_model.return_value = None
    processor._agent_repo.get.return_value = MagicMock(default_model="p/default")
    processor._agent_manager.providers.is_model_ref_usable.return_value = True
    processor._agent_manager.providers.resolve_model_for_multimodal_turn.return_value = "p/picked"

    resolved = processor._resolve_harness_model(
        "a1",
        "t1",
        {"model": "p/picked"},
        needs_multimodal=False,
    )
    assert resolved == "p/picked"
    processor._agent_manager.providers.resolve_model_for_multimodal_turn.assert_called_once_with(
        "p/picked",
        needs_multimodal=False,
    )


def test_resolve_harness_model_route_marks_dashboard_override_explicit() -> None:
    processor = GlobalProcessor(
        agent_manager=MagicMock(),
        thread_registry=MagicMock(),
        audit_repo=MagicMock(),
        agent_repo=MagicMock(),
        user_repo=MagicMock(),
        connector_repo=MagicMock(),
        dispatcher=MagicMock(),
        usage_repo=MagicMock(),
        gateway=MagicMock(),
    )
    processor._agent_manager.get_thread_model.return_value = None
    processor._agent_manager.providers.is_model_ref_usable.return_value = True
    processor._agent_manager.providers.resolve_model_for_multimodal_turn.return_value = "p/picked"

    route = processor._resolve_harness_model_route(
        "a1",
        "t1",
        {"model": "p/picked"},
        needs_multimodal=False,
    )

    assert route.model_ref == "p/picked"
    assert route.automatic is False


def test_resolve_harness_model_route_marks_fallback_automatic() -> None:
    processor = GlobalProcessor(
        agent_manager=MagicMock(),
        thread_registry=MagicMock(),
        audit_repo=MagicMock(),
        agent_repo=MagicMock(),
        user_repo=MagicMock(),
        connector_repo=MagicMock(),
        dispatcher=MagicMock(),
        usage_repo=MagicMock(),
        gateway=MagicMock(),
    )
    processor._agent_manager.get_thread_model.return_value = None
    processor._agent_repo.get.return_value = MagicMock(default_model=None)
    processor._agent_manager.get_config.return_value = {}
    processor._agent_manager.providers.resolve_explicit_default_model.return_value = None
    processor._agent_manager.resolve_fallback_model_ref.return_value = "p/auto"
    processor._agent_manager.providers.resolve_model_for_multimodal_turn.return_value = "p/auto"

    route = processor._resolve_harness_model_route(
        "a1",
        "t1",
        None,
        needs_multimodal=False,
    )

    assert route.model_ref == "p/auto"
    assert route.automatic is True


def test_resolve_harness_model_drops_unusable_dashboard_override() -> None:
    """Stale composer/expert refs must not bypass catalog usability checks."""
    processor = GlobalProcessor(
        agent_manager=MagicMock(),
        thread_registry=MagicMock(),
        audit_repo=MagicMock(),
        agent_repo=MagicMock(),
        user_repo=MagicMock(),
        connector_repo=MagicMock(),
        dispatcher=MagicMock(),
        usage_repo=MagicMock(),
        gateway=MagicMock(),
    )
    processor._agent_manager.get_thread_model.return_value = None
    processor._agent_repo.get.return_value = MagicMock(default_model="p/deleted")
    processor._agent_manager.get_config.return_value = {}
    processor._agent_manager.providers.is_model_ref_usable.return_value = False
    processor._agent_manager.providers.resolve_explicit_default_model.return_value = None

    assert (
        processor._resolve_harness_model(
            "a1",
            "t1",
            {"model": "p/deleted"},
            needs_multimodal=False,
        )
        is None
    )
    processor._agent_manager.providers.resolve_model_for_multimodal_turn.assert_not_called()


def test_composer_model_precedes_sticky_and_legacy_slash_model() -> None:
    assert (
        GlobalProcessor._model_ref_from_meta(
            "p/slash",
            {"model": "p/composer"},
            "p/sticky",
        )
        == "p/composer"
    )
    assert GlobalProcessor._model_ref_from_meta("p/slash", None, "p/sticky") == "p/sticky"


def test_stamp_stream_speaker_sets_agent_alias() -> None:
    from octop.infra.agents.teams.team_manager import (
        stamp_stream_speaker as _stamp_stream_speaker,
    )
    from octop.infra.agents.teams.team_manager import (
        stamp_team_host_chunk as _maybe_stamp_team_host,
    )

    token = _stamp_stream_speaker({"type": "token", "content": "hi"}, "host")
    assert token == {
        "type": "token",
        "content": "hi",
        "agent_id": "host",
        "agent": "host",
    }
    host_done = _stamp_stream_speaker({"type": "done"}, "host")
    assert host_done == {"type": "done"}
    member_done = _stamp_stream_speaker({"type": "done"}, "child", include_done=True)
    assert member_done["agent"] == "child"
    assert member_done["agent_id"] == "child"
    usage = _stamp_stream_speaker({"type": "usage", "usage": {}}, "host")
    assert "agent_id" not in usage
    solo = _maybe_stamp_team_host({"type": "token", "content": "hi"}, "host", False)
    assert solo == {"type": "token", "content": "hi"}
    team = _maybe_stamp_team_host({"type": "token", "content": "hi"}, "host", True)
    assert team["agent"] == "host"
    team_done = _maybe_stamp_team_host({"type": "done"}, "host", True)
    assert team_done["agent_id"] == "host"
    assert team_done["agent"] == "host"
    member = _maybe_stamp_team_host(
        {"type": "token", "content": "hi", "agent_id": "doctor", "agent": "doctor"},
        "host",
        True,
    )
    assert member["agent_id"] == "doctor"
    assert member["agent"] == "doctor"