"""第 3 批单测:JSON 解析、决策映射、对话流、击杀链、搜查发现(FakeLLM,零真实调用)。""" import asyncio from genesis.runtime.cognition import CognitiveMind, Decision, parse_json_block from genesis.runtime.engine import WorldEngine class FakeLLM: """脚本桩:按系统词内容路由回复。""" def __init__(self, default: str = '{"thought":"看看再说","intent":"观察",' '"speech":null,"action":{"kind":"idle","arg":""},"minutes":8}'): self.default = default self.calls: list[list[dict]] = [] def complete(self, messages, **kw): self.calls.append(messages) return self.default def _engine(llm=None) -> WorldEngine: from tests.test_runtime2 import FakeTime ft = FakeTime() eng = WorldEngine("worlds/ravenisle", rate=600.0, time_fn=ft, brain="llm", llm=llm or FakeLLM()) eng._ft = ft return eng # ───────────────── 解析稳健性 ───────────────── def test_parse_json_block_robust(): assert parse_json_block('好的```json\n{"a": 1,}\n```')["a"] == 1 assert parse_json_block('{"thought":"x","action":{"kind":"move","arg":"书房"}}')["action"]["arg"] == "书房" assert parse_json_block("说不出 JSON") == {} def test_to_decision_mapping(): m = CognitiveMind("温言", "card", is_killer=False, home="温言的房间", llm=FakeLLM(), memory=None, static_ground="") d = m._to_decision('{"thought":"去拿原稿","intent":"销毁罪证",' '"speech":{"to":"","text":"我去书房找本书"},' '"action":{"kind":"move","arg":"书房"},"minutes":6}') assert d.action_kind == "move" and d.action_arg == "书房" assert d.speech == "我去书房找本书" and d.speech_to == "" assert d.intent == "销毁罪证" and 2 <= d.minutes <= 10 assert m._to_decision("乱码").action_kind == "" # 解析失败 → 观望,不抛异常 # ───────────────── 会话流:点名→入会话→第三人加入→预算→散场 ───────────────── def test_session_roundtrip_bystander_and_dissolve(): eng = _engine() tang, cheng, wen = eng.minds["唐曼"], eng.minds["程亦深"], eng.minds["温言"] eng._execute("唐曼", tang, Decision(thought="套话", intent="试探程亦深", speech_to="程亦深", speech="你那晚到底喝了多少?")) sess = eng._session_of["唐曼"] assert eng._session_of.get("程亦深") is sess # 同一场会话 assert any("唐曼说" in e.content for e in cheng.attention._inbox) # 成员全文入耳 assert any("低声交谈" in e.content for e in wen.attention._inbox) # 旁人只知在交谈(摘要) eng.maybe_fast_forward() from genesis.runtime.engine import DIALOGUE_RATE assert eng.clock.rate <= DIALOGUE_RATE # 有会话 → 自动降速 # 第三人插话 → 并入同一场会话(群聊,G1) eng._execute("温言", wen, Decision(speech_to="唐曼", speech="你们在聊什么?")) assert eng._session_of.get("温言") is sess and len(sess.members) == 3 # 程亦深接话并告辞 → 余 2 人会话仍在 eng._execute("程亦深", cheng, Decision(speech_to="唐曼", speech="我先回房了", leave_dialogue=True)) assert "程亦深" not in sess.members and len(sess.members) == 2 # 预算耗尽 → 强制散场,成员留有交谈记忆 for _ in range(sess.budget): eng._execute("唐曼", tang, Decision(speech_to="温言", speech="再说两句")) if not sess.members: break assert not eng.sessions # 会话解散 eng.maybe_fast_forward() assert eng.clock.rate == eng.base_rate # 散场即恢复常规倍率(G3) hits = tang.memory.retrieve("交谈 要点", 23.0, top_k=3) assert any("交谈" in h.content for h in hits) def test_directed_speech_to_absent_is_suppressed(): """G4:对方已离场 → 不开口,转为内部体验,世上不会出现隔空喊话。""" eng = _engine() tang = eng.minds["唐曼"] eng.state.positions["程亦深"] = "灯塔" # 人在半个岛之外 delivered = [] eng.bus.taps.append(lambda e: delivered.append(e)) eng._execute("唐曼", tang, Decision(speech_to="程亦深", speech="你给我站住!")) assert all("站住" not in e.content for e in delivered if e.scope != "private") assert tang.last_speech == "" # 气泡也不显示 assert any("已不在这里" in e.content for e in delivered if e.scope == "private") def test_speech_ttl_and_gesture_channel(): """G4 气泡过期 + G5 台词/神态分离。""" eng = _engine() tang = eng.minds["唐曼"] eng._execute("唐曼", tang, Decision(speech="大家冷静!", gesture="把烛台举高了些")) f = eng.frame(1) me = next(a for a in f["agents"] if a["name"] == "唐曼") assert me["speech"] == "大家冷静!" and me["gesture"] == "把烛台举高了些" eng.clock._anchor_sim += 40 # 拨快 40 模拟秒 f2 = eng.frame(2) me2 = next(a for a in f2["agents"] if a["name"] == "唐曼") assert me2["speech"] == "" # 30 秒后气泡消失 assert me2["gesture"] == "把烛台举高了些" # 神态 45 秒存活 def test_empty_decision_becomes_musing(): """G12:模型哑火 → 出神占位,不是静止呆人。""" eng = _engine() tang = eng.minds["唐曼"] eng._execute("唐曼", tang, Decision()) assert tang.last_thought == "(怔怔出神)" assert tang.actor.current is not None and tang.actor.current.desc == "怔怔出神" # ───────────────── 击杀链:状态/尸体/惨叫/目击 ───────────────── def test_kill_flow_objective_and_broadcast(): eng = _engine() jiang = eng.minds["江离"] eng.state.positions.update({"江离": "琴房", "程亦深": "琴房", "温言": "书房"}) eng._execute("江离", jiang, Decision(action_kind="kill", action_arg="程亦深")) # LLM 把 target 放在 arg 也要能杀:引擎按 arg 解析 assert "程亦深" in eng.state.dead and eng.state.bodies["琴房"] == ["程亦深"] assert any("惨叫" in e.content for e in eng.minds["温言"].attention._inbox) # 全岛闻声 assert any("☠" in n["text"] for n in eng.narrative) # 非凶手杀人会被拒绝 eng.state.positions.update({"白聿": "酒窖", "唐曼": "酒窖"}) eng._execute("白聿", eng.minds["白聿"], Decision(action_kind="kill", action_arg="唐曼")) assert "唐曼" not in eng.state.dead def test_kill_with_witness_marks_memory(): eng = _engine() jiang = eng.minds["江离"] # 室内现场:有灯近距离,目击者必定看清(露天低能见度的漏看另有专测) eng.state.positions.update({"江离": "琴房", "邵铭轩": "琴房", "温言": "琴房"}) eng._execute("江离", jiang, Decision(action_kind="kill", action_arg="邵铭轩")) hits = eng.minds["温言"].memory.retrieve("目睹 杀害", 23.0, top_k=3) assert any("亲眼目睹" in h.content for h in hits) # 目击者顶级记忆 # ───────────────── 搜查:隐藏物品被翻出 → 可拿走 ───────────────── def test_search_reveals_hidden_items_then_take(): eng = _engine() wen = eng.minds["温言"] eng.state.positions["温言"] = "书房" eng._execute("温言", wen, Decision(action_kind="search", minutes=10)) assert wen.actor.current is not None and "搜查" in wen.actor.current.desc eng._finish_action("温言", wen.actor.current.payload, "") assert not eng.state.items["draft"].hidden # 报道原稿被翻出 eng._execute("温言", wen, Decision(action_kind="take", action_arg="报道原稿")) assert eng.state.items["draft"].holder == "温言" eng._execute("温言", wen, Decision(action_kind="destroy", action_arg="报道原稿")) assert eng.state.items["draft"].destroyed # 罪证从世上消失 # ───────────────── 认知端到端:FakeLLM 驱动引擎,决策真的来自"思考" ───────────────── def test_engine_with_fake_llm_moves_agents(): llm = FakeLLM('{"thought":"先去书房看看","intent":"查个究竟",' '"speech":null,"action":{"kind":"move","arg":"书房"},"minutes":5}') async def main(): from tests.test_runtime2 import FakeTime ft = FakeTime() eng = WorldEngine("worlds/ravenisle", rate=600.0, time_fn=ft, brain="llm", llm=llm) eng.start() task = asyncio.create_task(eng.sched.run(eng.on_wake)) for _ in range(40): ft.t += 0.4 await asyncio.sleep(0.05) eng.sched.stop() await asyncio.wait_for(task, timeout=2) return eng eng = asyncio.run(main()) assert len(llm.calls) >= 7 # 每个人都真的思考过 assert sum(1 for p in eng.state.positions.values() if p == "书房") >= 5 # 决策驱动移动 # 目标自举调用如今也带地图常识(事实锚),定位正式 L2 决策调用须看动作协议 full = next(c[0]["content"] for c in llm.calls if "你每次输出一个 JSON" in c[0]["content"]) assert "雾鸦岛地图常识" in full # 共识层注入 assert "你每次输出一个 JSON" in full # 动作协议注入