from __future__ import annotations from typing import Any import pytest from core.agentscope.runtime.tasks import _build_recent_context_messages from schemas.agent.forwarded_props import RuntimeMode from schemas.agent.runtime_config import MessageContextConfig class _StubContextCache: def __init__(self, messages: list[dict[str, object]]) -> None: self._messages = messages async def get(self, **_: object) -> list[dict[str, object]]: return self._messages class _StubAttachmentCache: async def get(self, **_: object) -> None: return None async def set(self, **_: object) -> None: return None @pytest.mark.asyncio async def test_build_recent_context_messages_accepts_snake_case_ganzhi( monkeypatch: pytest.MonkeyPatch, ) -> None: metadata_payload: dict[str, Any] = { "run_id": "run_1", "agent_output": { "status": "success", "sign_level": "中上签", "conclusion": ["结论"], "focus_points": ["重点"], "advice": ["建议"], "keywords": ["一", "二", "三"], "answer": "这是回答", "divination_derived": { "question": "问题", "question_type": "运势", "divination_method": "自动起卦", "divination_time": "2026-04-10T12:00:00Z", "binary_code": "101010", "changed_binary_code": "010101", "gua_name": "乾为天", "upper_name": "乾", "lower_name": "乾", "target_gua_name": "坤为地", "world_position": 3, "response_position": 6, "has_changing_yao": True, "ganzhi": { "year_gan_zhi": "甲子", "month_gan_zhi": "乙丑", "day_gan_zhi": "丙寅", "time_gan_zhi": "丁卯", "year_kong_wang": "戌亥", "month_kong_wang": "申酉", "day_kong_wang": "午未", "time_kong_wang": "辰巳", "yue_jian": "子月", "ri_chen": "寅日", "yue_po": "午火", "ri_chong": "申金", }, "wu_xing_statuses": {"金": "旺"}, "yao_info_list": [ { "position": 1, "spirit_name": "青龙", "relation_name": "兄弟", "tigan_name": "甲", "element_name": "木", "is_yang": True, "is_changing": False, "special_mark": "", }, { "position": 2, "spirit_name": "朱雀", "relation_name": "子孙", "tigan_name": "乙", "element_name": "火", "is_yang": False, "is_changing": False, "special_mark": "", }, { "position": 3, "spirit_name": "勾陈", "relation_name": "妻财", "tigan_name": "丙", "element_name": "土", "is_yang": True, "is_changing": True, "special_mark": "世", }, { "position": 4, "spirit_name": "腾蛇", "relation_name": "官鬼", "tigan_name": "丁", "element_name": "金", "is_yang": False, "is_changing": False, "special_mark": "", }, { "position": 5, "spirit_name": "白虎", "relation_name": "父母", "tigan_name": "戊", "element_name": "水", "is_yang": True, "is_changing": False, "special_mark": "应", }, { "position": 6, "spirit_name": "玄武", "relation_name": "兄弟", "tigan_name": "己", "element_name": "木", "is_yang": False, "is_changing": True, "special_mark": "", }, ], "target_yao_info_list": [], "fushen_positions": [], "fushen_info_list": [], }, }, } cache = _StubContextCache( messages=[ { "role": "assistant", "content": "fallback", "metadata": metadata_payload, } ] ) monkeypatch.setattr( "core.agentscope.runtime.tasks.create_context_messages_cache", lambda: cache, ) monkeypatch.setattr( "core.agentscope.runtime.tasks.create_attachment_content_cache", lambda: _StubAttachmentCache(), ) converted = await _build_recent_context_messages( session=None, thread_id="thread_1", runtime_mode=RuntimeMode.CHAT, context_config=MessageContextConfig(), ) assert len(converted) == 1 content = converted[0].content assert isinstance(content, str) assert "[assistant_context]" in content assert "gua_name:" in content