from types import SimpleNamespace from typing import Any, Callable import pytest from rich.panel import Panel from rich.text import Text from src.chat.replyer import maisaka_generator as replyer_module from src.common.data_models.reply_generation_data_models import ( GenerationMetrics, LLMCompletionResult, ReplyGenerationResult, ) from src.core.tooling import ToolExecutionResult, ToolInvocation from src.maisaka.builtin_tool.context import BuiltinToolRuntimeContext from src.maisaka.builtin_tool import reply as reply_tool_module from src.maisaka.builtin_tool import send_emoji as send_emoji_tool_module from src.maisaka.monitor_events import emit_planner_finalized from src.maisaka.reasoning_engine import MaisakaReasoningEngine from src.maisaka import runtime as runtime_module from src.maisaka.runtime import MaisakaHeartFlowChatting def test_runtime_maps_expression_config_flags_to_correct_fields(monkeypatch: pytest.MonkeyPatch) -> None: fake_chat_stream = SimpleNamespace( is_group_session=True, group_id="group-1", user_id="user-1", platform="test", ) monkeypatch.setattr( runtime_module.chat_manager, "get_session_by_session_id", lambda session_id: fake_chat_stream, ) monkeypatch.setattr(runtime_module.chat_manager, "get_session_name", lambda session_id: "测试会话") monkeypatch.setattr( runtime_module.ExpressionConfigUtils, "get_expression_config_for_chat", staticmethod(lambda session_id: (True, False, True)), ) monkeypatch.setattr(runtime_module, "ExpressionLearner", lambda session_id: SimpleNamespace()) monkeypatch.setattr(runtime_module, "JargonMiner", lambda session_id, session_name: SimpleNamespace()) monkeypatch.setattr(runtime_module, "MaisakaReasoningEngine", lambda runtime: SimpleNamespace()) monkeypatch.setattr(runtime_module, "ToolRegistry", lambda: SimpleNamespace()) monkeypatch.setattr(runtime_module, "ReplyEffectTracker", lambda **kwargs: SimpleNamespace()) monkeypatch.setattr(MaisakaHeartFlowChatting, "_register_tool_providers", lambda self: None) monkeypatch.setattr(MaisakaHeartFlowChatting, "_emit_monitor_session_start", lambda self: None) runtime = MaisakaHeartFlowChatting("session-1") assert runtime._enable_expression_use is True assert runtime._enable_expression_learning is False assert runtime._enable_jargon_learning is True class _FakeLLMResult: def __init__(self) -> None: self.response = "测试回复" self.reasoning = "先理解上下文，再给出自然回复。" self.model_name = "fake-model" self.tool_calls = [] self.prompt_tokens = 12 self.completion_tokens = 7 self.total_tokens = 19 class _FakeLegacyLLMServiceClient: def __init__(self, *args: Any, **kwargs: Any) -> None: del args del kwargs async def generate_response_with_messages(self, *, message_factory: Callable[[object], list[Any]]) -> _FakeLLMResult: assert message_factory(object()) return _FakeLLMResult() class _FakeMultimodalLLMServiceClient: def __init__(self, *args: Any, **kwargs: Any) -> None: del args del kwargs async def generate_response_with_messages(self, *, message_factory: Callable[[object], list[Any]]) -> _FakeLLMResult: assert message_factory(object()) return _FakeLLMResult() @pytest.mark.asyncio async def test_legacy_and_multimodal_replyer_monitor_detail_have_same_shape(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(replyer_module, "LLMServiceClient", _FakeLegacyLLMServiceClient) monkeypatch.setattr(replyer_module, "load_prompt", lambda *args, **kwargs: "legacy prompt") legacy_generator = replyer_module.MaisakaReplyGenerator( chat_stream=None, request_type="test_legacy", enable_visual_message=False, ) multimodal_generator = replyer_module.MaisakaReplyGenerator( chat_stream=None, request_type="test_multi", llm_client_cls=_FakeMultimodalLLMServiceClient, load_prompt_func=lambda *args, **kwargs: "multi prompt", enable_visual_message=True, ) legacy_success, legacy_result = await legacy_generator.generate_reply_with_context( stream_id="session-legacy", chat_history=[], reply_reason="测试原因", ) multimodal_success, multimodal_result = await multimodal_generator.generate_reply_with_context( stream_id="session-multi", chat_history=[], reply_reason="测试原因", ) assert legacy_success is True assert multimodal_success is True assert legacy_result.monitor_detail is not None assert multimodal_result.monitor_detail is not None assert set(legacy_result.monitor_detail.keys()) == set(multimodal_result.monitor_detail.keys()) assert set(legacy_result.monitor_detail["metrics"].keys()) == set(multimodal_result.monitor_detail["metrics"].keys()) assert legacy_result.monitor_detail["metrics"]["prompt_tokens"] == 12 assert legacy_result.monitor_detail["metrics"]["completion_tokens"] == 7 assert legacy_result.monitor_detail["metrics"]["total_tokens"] == 19 def test_legacy_replyer_builds_message_sequence_like_multimodal() -> None: legacy_generator = replyer_module.MaisakaReplyGenerator( chat_stream=None, request_type="test_legacy", enable_visual_message=False, ) legacy_prompt_loader = replyer_module.load_prompt replyer_module.load_prompt = lambda *args, **kwargs: "legacy prompt" try: session_message = replyer_module.SessionBackedMessage( raw_message=SimpleNamespace(), visible_text="[Alice]你好\n[Bob]在吗", timestamp=replyer_module.datetime.now(), source_kind="user", ) request_messages = legacy_generator._build_request_messages( chat_history=[session_message], reply_message=None, reply_reason="测试原因", stream_id="session-legacy", ) finally: replyer_module.load_prompt = legacy_prompt_loader assert len(request_messages) == 4 assert request_messages[0].role.value == "system" assert request_messages[0].get_text_content() == "legacy prompt" assert request_messages[1].role.value == "user" assert request_messages[1].get_text_content() == "[Alice]你好" assert request_messages[2].role.value == "user" assert request_messages[2].get_text_content() == "[Bob]在吗" assert request_messages[3].role.value == "user" assert "当前时间：" in request_messages[3].get_text_content() assert "【回复信息参考】" in request_messages[3].get_text_content() assert "【最新推理】\n测试原因" in request_messages[3].get_text_content() assert "请自然地回复。" in request_messages[3].get_text_content() @pytest.mark.asyncio async def test_reply_tool_puts_monitor_detail_into_metadata(monkeypatch: pytest.MonkeyPatch) -> None: fake_monitor_detail = { "prompt_text": "reply prompt", "reasoning_text": "reply reasoning", "output_text": "reply output", "metrics": {"model_name": "fake-model", "total_tokens": 10}, } fake_reply_result = ReplyGenerationResult( success=True, completion=LLMCompletionResult(response_text="测试回复"), metrics=GenerationMetrics(overall_ms=11.5), monitor_detail=fake_monitor_detail, ) class _FakeReplyer: async def generate_reply_with_context(self, **kwargs: Any) -> tuple[bool, ReplyGenerationResult]: del kwargs return True, fake_reply_result monkeypatch.setattr(reply_tool_module.replyer_manager, "get_replyer", lambda **kwargs: _FakeReplyer()) monkeypatch.setattr(reply_tool_module, "render_cli_message", lambda text: text) target_message = SimpleNamespace( message_id="msg-1", message_info=SimpleNamespace( user_info=SimpleNamespace( user_cardname="测试用户", user_nickname="测试用户", user_id="user-1", ) ), ) runtime = SimpleNamespace( _source_messages_by_id={"msg-1": target_message}, log_prefix="[test]", chat_stream=SimpleNamespace(platform=reply_tool_module.CLI_PLATFORM_NAME), session_id="session-1", _chat_history=[], _clear_force_continue_until_reply=lambda: None, _record_reply_sent=lambda: None, run_sub_agent=None, ) engine = SimpleNamespace(_get_runtime_manager=lambda: None) tool_ctx = BuiltinToolRuntimeContext(engine=engine, runtime=runtime) invocation = ToolInvocation(tool_name="reply", arguments={"msg_id": "msg-1", "set_quote": True}) result = await reply_tool_module.handle_tool(tool_ctx, invocation) assert result.success is True assert result.metadata["monitor_detail"] == fake_monitor_detail @pytest.mark.asyncio async def test_send_emoji_tool_puts_monitor_detail_into_metadata(monkeypatch: pytest.MonkeyPatch) -> None: async def _fake_build_emoji_candidate_message(emojis: list[Any]) -> object: assert emojis return SimpleNamespace() async def _fake_send_emoji_for_maisaka(**kwargs: Any) -> Any: selected_emoji, matched_emotion = await kwargs["emoji_selector"]( kwargs["requested_emotion"], kwargs["reasoning"], kwargs["context_texts"], 2, ) assert selected_emoji is not None return SimpleNamespace( success=True, message="已发送表情包：开心", emoji_base64="ZW1vamk=", description="开心", emotions=["开心", "可爱"], matched_emotion=matched_emotion or "开心", sent_message=None, ) monkeypatch.setattr(send_emoji_tool_module, "_build_emoji_candidate_message", _fake_build_emoji_candidate_message) monkeypatch.setattr(send_emoji_tool_module, "send_emoji_for_maisaka", _fake_send_emoji_for_maisaka) monkeypatch.setattr( send_emoji_tool_module.emoji_manager, "emojis", [ SimpleNamespace(description="开心,可爱", emotion=["开心", "可爱"]), SimpleNamespace(description="难过", emotion=["难过"]), ], ) async def _fake_run_sub_agent(**kwargs: Any) -> Any: del kwargs return SimpleNamespace( content='{"emoji_index": 1, "reason": "更贴合当前语气"}', prompt_tokens=9, completion_tokens=6, total_tokens=15, ) runtime = SimpleNamespace( _chat_history=[], log_prefix="[test]", session_id="session-emoji", run_sub_agent=_fake_run_sub_agent, ) engine = SimpleNamespace(last_reasoning_content="用户刚刚表达了开心情绪") tool_ctx = BuiltinToolRuntimeContext(engine=engine, runtime=runtime) invocation = ToolInvocation(tool_name="send_emoji", arguments={"emotion": "开心"}) result = await send_emoji_tool_module.handle_tool(tool_ctx, invocation) assert result.success is True assert result.metadata["monitor_detail"]["prompt_text"] assert result.metadata["monitor_detail"]["reasoning_text"] == "更贴合当前语气" assert result.metadata["monitor_detail"]["metrics"]["total_tokens"] == 15 assert any( section["title"] == "表情发送结果" for section in result.metadata["monitor_detail"]["extra_sections"] ) @pytest.mark.asyncio async def test_emit_planner_finalized_broadcasts_new_protocol(monkeypatch: pytest.MonkeyPatch) -> None: captured: dict[str, Any] = {} async def _fake_broadcast(event: str, data: dict[str, Any]) -> None: captured["event"] = event captured["data"] = data monkeypatch.setattr("src.maisaka.monitor_events._broadcast", _fake_broadcast) await emit_planner_finalized( session_id="session-1", cycle_id=3, timing_request_messages=[{"role": "user", "content": "先看看要不要继续"}], timing_selected_history_count=3, timing_tool_count=1, timing_action="continue", timing_content="继续", timing_tool_calls=[SimpleNamespace(call_id="timing-call-1", func_name="continue", args={})], timing_tool_results=["- continue [成功]: 继续执行"], timing_prompt_tokens=40, timing_completion_tokens=5, timing_total_tokens=45, timing_duration_ms=11.2, planner_request_messages=[{"role": "user", "content": "你好"}], planner_selected_history_count=5, planner_tool_count=2, planner_content="先查询再回复", planner_tool_calls=[SimpleNamespace(call_id="call-1", func_name="reply", args={"msg_id": "m1"})], planner_prompt_tokens=100, planner_completion_tokens=30, planner_total_tokens=130, planner_duration_ms=88.5, tools=[ { "tool_call_id": "call-1", "tool_name": "reply", "tool_args": {"msg_id": "m1"}, "success": True, "duration_ms": 22.0, "summary": "- reply [成功]: 已回复", "detail": {"output_text": "测试回复"}, } ], time_records={"planner": 0.1, "tool_calls": 0.2}, agent_state="stop", ) assert captured["event"] == "planner.finalized" payload = captured["data"] assert payload["timing_gate"]["result"]["action"] == "continue" assert payload["timing_gate"]["result"]["tool_results"] == ["- continue [成功]: 继续执行"] assert payload["request"]["messages"][0]["content"] == "你好" assert payload["request"]["tool_count"] == 2 assert payload["planner"]["tool_calls"][0]["id"] == "call-1" assert payload["tools"][0]["detail"]["output_text"] == "测试回复" assert payload["final_state"]["agent_state"] == "stop" @pytest.mark.asyncio async def test_emit_planner_finalized_supports_timing_only_cycle(monkeypatch: pytest.MonkeyPatch) -> None: captured: dict[str, Any] = {} async def _fake_broadcast(event: str, data: dict[str, Any]) -> None: captured["event"] = event captured["data"] = data monkeypatch.setattr("src.maisaka.monitor_events._broadcast", _fake_broadcast) await emit_planner_finalized( session_id="session-2", cycle_id=7, timing_request_messages=[{"role": "user", "content": "先别回"}], timing_selected_history_count=2, timing_tool_count=1, timing_action="no_reply", timing_content="当前不适合继续", timing_tool_calls=[SimpleNamespace(call_id="timing-call-2", func_name="no_reply", args={})], timing_tool_results=["- no_reply [成功]: 暂停当前对话"], timing_prompt_tokens=18, timing_completion_tokens=4, timing_total_tokens=22, timing_duration_ms=6.5, planner_request_messages=None, planner_selected_history_count=None, planner_tool_count=None, planner_content=None, planner_tool_calls=None, planner_prompt_tokens=None, planner_completion_tokens=None, planner_total_tokens=None, planner_duration_ms=None, tools=[], time_records={"timing_gate": 0.02}, agent_state="stop", ) assert captured["event"] == "planner.finalized" payload = captured["data"] assert payload["timing_gate"]["result"]["action"] == "no_reply" assert payload["planner"] is None assert payload["request"] is None def test_reasoning_engine_build_tool_monitor_result_keeps_non_reply_tool_without_detail() -> None: engine = object.__new__(MaisakaReasoningEngine) tool_call = SimpleNamespace(call_id="call-2", func_name="query_memory") invocation = ToolInvocation(tool_name="query_memory", arguments={"query": "Alice"}) result = ToolExecutionResult(tool_name="query_memory", success=True, content="查询成功") tool_result = engine._build_tool_monitor_result(tool_call, invocation, result, duration_ms=18.6) assert tool_result["tool_call_id"] == "call-2" assert tool_result["tool_name"] == "query_memory" assert tool_result["tool_args"] == {"query": "Alice"} assert tool_result["detail"] is None def test_runtime_build_tool_detail_panels_renders_reply_monitor_detail() -> None: runtime = object.__new__(MaisakaHeartFlowChatting) runtime.session_id = "session-1" panels = runtime._build_tool_detail_cards( [ { "tool_call_id": "call-reply-1", "tool_name": "reply", "tool_args": {"msg_id": "m1"}, "success": True, "duration_ms": 20.5, "summary": "- reply [成功]: 已回复", "detail": { "prompt_text": "reply prompt", "reasoning_text": "reply reasoning", "output_text": "reply output", "metrics": { "model_name": "fake-model", "prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15, "prompt_ms": 2.1, "llm_ms": 18.4, "overall_ms": 20.5, }, }, } ], stage_title="工具调用", ) assert len(panels) == 1 assert isinstance(panels[0], Panel) def test_runtime_filter_redundant_tool_results_keeps_only_non_detailed_summary() -> None: filtered_results = MaisakaHeartFlowChatting._filter_redundant_tool_results( tool_results=[ "- reply [成功]: 已回复", "- query_memory [成功]: 查询到 2 条记录", ], tool_detail_results=[ { "summary": "- reply [成功]: 已回复", "detail": {"output_text": "测试回复"}, } ], ) assert filtered_results == ["- query_memory [成功]: 查询到 2 条记录"] def test_runtime_build_tool_detail_panels_uses_prompt_access_panel(monkeypatch: pytest.MonkeyPatch) -> None: runtime = object.__new__(MaisakaHeartFlowChatting) runtime.session_id = "session-link" captured: dict[str, Any] = {} def _fake_build_text_access_panel(content: str, **kwargs: Any) -> str: captured["content"] = content captured["kwargs"] = kwargs return "PROMPT_LINK" monkeypatch.setattr( "src.maisaka.runtime.PromptCLIVisualizer.build_text_access_panel", _fake_build_text_access_panel, ) panels = runtime._build_tool_detail_cards( [ { "tool_call_id": "call-reply-2", "tool_name": "reply", "tool_args": {"msg_id": "m2"}, "success": True, "duration_ms": 12.0, "summary": "- reply [成功]: 已回复", "detail": { "prompt_text": "reply prompt link", "output_text": "reply output", }, } ], stage_title="工具调用", ) assert len(panels) == 1 assert captured["content"] == "reply prompt link" assert captured["kwargs"]["chat_id"] == "session-link" assert captured["kwargs"]["request_kind"] == "replyer" def test_runtime_build_tool_detail_panels_uses_emotion_prompt_access_panel(monkeypatch: pytest.MonkeyPatch) -> None: runtime = object.__new__(MaisakaHeartFlowChatting) runtime.session_id = "session-emotion" captured: dict[str, Any] = {} def _fake_build_text_access_panel(content: str, **kwargs: Any) -> str: captured["content"] = content captured["kwargs"] = kwargs return "EMOTION_PROMPT_LINK" monkeypatch.setattr( "src.maisaka.runtime.PromptCLIVisualizer.build_text_access_panel", _fake_build_text_access_panel, ) panels = runtime._build_tool_detail_cards( [ { "tool_call_id": "call-emoji-1", "tool_name": "send_emoji", "tool_args": {"emotion": "开心"}, "success": True, "duration_ms": 15.0, "summary": "- send_emoji [成功]: 已发送表情包", "detail": { "prompt_text": "emotion prompt link", "output_text": '{"emoji_index": 1}', }, } ], stage_title="工具调用", ) assert len(panels) == 1 assert captured["content"] == "emotion prompt link" assert captured["kwargs"]["chat_id"] == "session-emotion" assert captured["kwargs"]["request_kind"] == "emotion" def test_runtime_build_tool_detail_cards_uses_structured_prompt_messages_with_images( monkeypatch: pytest.MonkeyPatch, ) -> None: runtime = object.__new__(MaisakaHeartFlowChatting) runtime.session_id = "session-image" captured: dict[str, Any] = {} def _fake_build_prompt_access_panel(messages: list[Any], **kwargs: Any) -> str: captured["messages"] = messages captured["kwargs"] = kwargs return "IMAGE_PROMPT_LINK" def _fake_build_text_access_panel(content: str, **kwargs: Any) -> str: captured["text_content"] = content captured["text_kwargs"] = kwargs return "TEXT_PROMPT_LINK" monkeypatch.setattr( "src.maisaka.runtime.PromptCLIVisualizer.build_prompt_access_panel", _fake_build_prompt_access_panel, ) monkeypatch.setattr( "src.maisaka.runtime.PromptCLIVisualizer.build_text_access_panel", _fake_build_text_access_panel, ) panels = runtime._build_tool_detail_cards( [ { "tool_call_id": "call-reply-image-1", "tool_name": "reply", "tool_args": {"msg_id": "m3"}, "success": True, "duration_ms": 22.0, "summary": "- reply [成功]: 已回复", "detail": { "prompt_text": "reply prompt image", "request_messages": [ { "role": "user", "content": ["前缀文本", ["png", "ZmFrZQ=="]], } ], "output_text": "reply output", }, } ], stage_title="工具调用", ) assert len(panels) == 1 assert "messages" in captured assert "text_content" not in captured assert captured["kwargs"]["chat_id"] == "session-image" assert captured["kwargs"]["request_kind"] == "replyer" def test_runtime_render_context_usage_panel_merges_timing_and_planner(monkeypatch: pytest.MonkeyPatch) -> None: runtime = object.__new__(MaisakaHeartFlowChatting) runtime.session_id = "session-merged" runtime.session_name = "测试聊天流" runtime._max_context_size = 20 printed: list[Any] = [] monkeypatch.setattr("src.maisaka.runtime.console.print", lambda renderable: printed.append(renderable)) runtime._render_context_usage_panel( cycle_id=12, timing_selected_history_count=3, timing_prompt_tokens=15, timing_action="continue", timing_response="继续执行", planner_selected_history_count=5, planner_prompt_tokens=42, planner_response="先查询再回复", ) assert len(printed) == 1 outer_panel = printed[0] assert isinstance(outer_panel, Panel) renderables = list(outer_panel.renderable.renderables) assert isinstance(renderables[0], Text) assert "聊天流名称：测试聊天流" in renderables[0].plain assert "聊天流ID：session-merged" in renderables[0].plain assert len(renderables) == 3