"""说书人全景视角 + 预思考流水线的单测。""" import asyncio from genesis.runtime.bus import Event from genesis.runtime.cognition import Decision from genesis.runtime.engine import WorldEngine from genesis.runtime.narrator import Narrator from tests.test_cognition import FakeLLM from tests.test_runtime2 import FakeTime def test_narrator_sees_session_dialogue_but_not_private_reveals(): nar = Narrator(FakeLLM()) nar.tap(Event(etype="speech", content="唐曼说:「钥匙在谁手里?」", scope="private", to=["温言"], salience=0.5)) # 会话对白:可见 nar.tap(Event(etype="ambient", content="你亲眼目睹江离对程亦深下了杀手!!", scope="private", to=["温言"], salience=1.0)) # 单独揭示:屏蔽 nar.tap(Event(etype="feedback", content="你拿到了阮青的日记", scope="private", to=["唐曼"], salience=0.4)) # 动作反馈:屏蔽 assert len(nar.buffer) == 1 and "钥匙" in nar.buffer[0] def test_narrator_prompt_carries_panorama(): llm = FakeLLM('{"prose":"x","summary":"y"}') nar = Narrator(llm) nar.tap(Event(etype="speech", content="a说:「b」", place="门厅", salience=0.5)) nar.write(700.0, "第1天 22:10", panorama="门厅:顾长风、白聿(在交谈);书房:温言。暴雨;灯火尚明") user = llm.calls[0][1]["content"] assert "【此刻全景】" in user and "书房:温言" in user # 全景数据仍喂入(供选最有戏的一笔) sys = llm.calls[0][0]["content"] assert "60-120" in sys # 瘦身:每段短而紧,不再铺全景 def test_engine_panorama_format(): ft = FakeTime() eng = WorldEngine("worlds/ravenisle", rate=600.0, time_fn=ft, brain="llm", llm=FakeLLM()) eng.state.positions.update({"温言": "书房", "江离": "书房"}) pano = eng.panorama() assert "书房:江离、温言" in pano and "暴雨" in pano def test_prethink_overlaps_action_and_reuses_decision(): """动作未完成时预想下一步;动作完成时直接用预想结果,不再次调 LLM。""" llm = FakeLLM('{"thought":"接着去酒窖","intent":"查酒窖",' '"speech":null,"action":{"kind":"move","arg":"酒窖"},"minutes":5}') async def main(): ft = FakeTime() eng = WorldEngine("worlds/ravenisle", rate=600.0, time_fn=ft, brain="llm", llm=llm) wen = eng.minds["温言"] eng.state.positions["温言"] = "书房" wen.goals = [{"desc": "查清剪报", "status": "doing", "note": ""}] # 跳过首调的目标自举 wen.actor.start_do("整理剪报", 1200) # 20 分钟的动作 n0 = len(llm.calls) ft.t += 1.75 # 600×:推进 1050s,动作只剩 150s(真实预想时机) await eng.on_wake("温言", "prethink") # 动作还剩时间 → 预想 assert wen.pending is not None and len(llm.calls) == n0 + 1 assert wen.actor.current is not None # 动作未被打断 ft.t += 0.3 # 动作到点 await eng.on_wake("温言", "tick") # 完成 → 直接用预想,不再调 LLM assert len(llm.calls) == n0 + 1 assert wen.pending is None assert wen.actor.current is not None and wen.actor.current.kind == "move" assert wen.actor.current.dest == "酒窖" asyncio.run(main()) def test_interrupt_invalidates_pending(): llm = FakeLLM('{"thought":"x","intent":"y","speech":null,' '"action":{"kind":"idle","arg":""},"minutes":5}') async def main(): ft = FakeTime() eng = WorldEngine("worlds/ravenisle", rate=600.0, time_fn=ft, brain="llm", llm=llm) wen = eng.minds["温言"] wen.actor.start_do("翻找", 1200) await eng.on_wake("温言", "prethink") assert wen.pending is not None await eng.on_wake("温言", "interrupt:ambient") # 突发 → 预想作废,重新评估 assert wen.pending is None asyncio.run(main()) # ───────────────── 本批新增:干预/下钻/节律/紧急成段/MBTI ───────────────── def test_godmode_inject_whisper_and_detail(): from tests.test_cognition import FakeLLM from genesis.runtime.engine import WorldEngine eng = WorldEngine("worlds/ravenisle", rate=600.0, time_fn=FakeTime(), brain="llm", llm=FakeLLM()) n = eng.inject_event("一声枪响从灯塔方向传来!") assert n == 7 and any("枪响" in x["text"] for x in eng.narrative) assert eng.whisper("温言", "去查暗房") and not eng.whisper("不存在的人", "x") assert any("击中你" in e.content for e in eng.minds["温言"].attention._inbox) eng.set_world_goal("让猜疑发酵") assert "让猜疑发酵" in eng._bulletin() d = eng.agent_detail("温言") assert d["name"] == "温言" and "relations" in d and "soma" in d def test_social_rhythm_and_mbti_traits(): from genesis.runtime.bulletin import social_rhythm from genesis.runtime.cognition import mbti_traits assert "早饭" in social_rhythm(7.5) and social_rhythm(15.0) == "" t = mbti_traits("ISTJ") assert "独处不慌" in t and "计划" in t assert "结伴" in mbti_traits("ENFP") def test_narrator_urgent_event_writes_immediately(): from genesis.runtime.narrator import Narrator from tests.test_cognition import FakeLLM nar = Narrator(FakeLLM('{"prose":"x","summary":"y"}')) nar.tap(Event(etype="speech", content="闲聊", place="门厅", salience=0.4)) assert not nar.due(10.0) # 普通事件等窗口 nar.tap(Event(etype="ambient", content="惨叫划破夜空!", scope="global", salience=0.95)) assert nar.due(10.0) # 凶案级 → 立即成段 nar.write(10.0, "第1天 23:00") assert not nar.urgent def test_decorated_occupants_do_not_break_sessions(): """G18 装饰串(动作/神色)只进 prompt,不影响会话在场校验。""" from tests.test_cognition import FakeLLM from genesis.runtime.cognition import Decision from genesis.runtime.engine import WorldEngine eng = WorldEngine("worlds/ravenisle", rate=600.0, time_fn=FakeTime(), brain="llm", llm=FakeLLM()) tang, wen = eng.minds["唐曼"], eng.minds["温言"] wen.actor.start_do("翻看剪报", 600) # 温言带着动作(会被装饰) eng._execute("唐曼", tang, Decision(speech_to="温言", speech="聊两句?")) snap = eng._snapshot("唐曼", tang) assert snap.dialogue_with == "温言" # 会话仍正确识别(未被装饰串破坏) assert any("翻看剪报" in o for o in snap.occupants) # 社会信号可见 def test_prologue_landing_sequence(): """序幕:码头登岛 → 走进门厅 → 回房安顿 → 放映室落座 → 放映事件 → 心智接管。""" import asyncio as _a from tests.test_cognition import FakeLLM from genesis.runtime.engine import WorldEngine async def main(): ft = FakeTime() eng = WorldEngine("worlds/ravenisle", rate=600.0, time_fn=ft, brain="llm", llm=FakeLLM(), prologue=True) eng.start() assert all(p == "码头" for p in eng.state.positions.values()) # 从码头开始 assert any("序幕" in n["text"] for n in eng.narrative) task = _a.create_task(eng.sched.run(eng.on_wake)) for _ in range(300): # 推进 ~2 模拟小时(序幕全程,留足余量) ft.t += 0.04 await _a.sleep(0.025) if not eng.prologue: break await _a.sleep(0.3) eng.sched.stop() await _a.wait_for(task, timeout=2) return eng eng = asyncio.run(main()) assert not eng.prologue # 序幕已结束 assert any("放映机自己转了起来" in n["text"] for n in eng.narrative) seated = sum(1 for p in eng.state.positions.values() if p == "放映室") assert seated >= 5, f"开演时应多数人已落座,实际:{eng.state.positions}" # 放映事件进了每个人的收件箱(心智接管后第一次思考就会看到) one = next(iter(eng.minds.values())) assert any("杀青宴" in e.content for e in one.attention._inbox)