"""批次 B 单测:剧情装置引擎(G13)、集会指认与终局(G14)、新可供性(G15)。""" import asyncio from genesis.runtime.cognition import Decision from genesis.runtime.engine import WorldEngine from tests.test_cognition import FakeLLM from tests.test_runtime2 import FakeTime def _engine(): ft = FakeTime() eng = WorldEngine("worlds/ravenisle", rate=600.0, time_fn=ft, brain="llm", llm=FakeLLM()) eng._ft = ft return eng def _drain_devices(eng): """把已到期的装置触发器立即点火(测试中直接驱动,绕过调度等待)。""" for token in list(eng.devices._pending): eng.devices.fire(token) # ───────────────── G13:凶案 → 胶片盒烧毁 + 残页钉尸旁 ───────────────── def test_kill_triggers_film_box_and_pinned_page(): eng = _engine() eng.state.positions.update({"江离": "断崖", "程亦深": "断崖"}) eng._execute("江离", eng.minds["江离"], Decision(action_kind="kill", action_arg="程亦深")) assert eng.state.facilities["film_boxes"].state == "act0" # 装置延迟生效,非即时 _drain_devices(eng) assert eng.state.facilities["film_boxes"].state == "act1" # 第一幕胶片烧毁 page = eng.state.items.get("pinned_程亦深") assert page is not None and page.place == "断崖" and not page.hidden # 残页钉在尸体旁 assert any("胶片烧成" in n["text"] for n in eng.narrative) seen = " ".join(eng.state.describe("餐厅", viewer="温言")) assert "第 1" in seen or "前 1" in seen or "act" not in seen # 后来者看到既成事实 def test_second_kill_burns_second_box(): eng = _engine() eng.state.positions.update({"江离": "断崖", "程亦深": "断崖"}) eng._execute("江离", eng.minds["江离"], Decision(action_kind="kill", action_arg="程亦深")) eng.state.positions.update({"江离": "灯塔", "邵铭轩": "灯塔"}) eng._execute("江离", eng.minds["江离"], Decision(action_kind="kill", action_arg="邵铭轩")) _drain_devices(eng) assert eng.state.facilities["film_boxes"].state == "act2" # ───────────────── G13:灯塔点亮 → 渡船提前 ───────────────── def test_lighthouse_lit_advances_ferry(): eng = _engine() wen = eng.minds["温言"] eng.state.positions["温言"] = "灯塔" for i in (1, 2, 3): # 凑三桶灯油 eng.state.items[f"fuel_{i}"].holder = "温言" eng._execute("温言", wen, Decision(action_kind="use", action_arg="给灯塔加油")) assert eng.state.facilities["light_tank"].state == "fueled" before = eng.ferry_eta eng._execute("温言", wen, Decision(action_kind="use", action_arg="点亮灯塔")) assert eng.state.facilities["light_tank"].state == "lit" assert eng.ferry_eta < before # 渡船提前 assert any("渡船提前" in n["text"] for n in eng.narrative) # ───────────────── G14:集会 → 投票 → 处决/终局 ───────────────── def test_assembly_vote_executes_killer_and_ends_game(): eng = _engine() for a in eng.minds: eng.state.positions[a] = "门厅" eng._execute("温言", eng.minds["温言"], Decision(action_kind="assemble")) assert eng.assembly is not None and eng.assembly["phase"] == "summon" assert "对质" in eng._bulletin() # 进入全员共识公告 eng._assembly_advance() # 召集期满 → 投票阶段 assert eng.assembly["phase"] == "vote" voters = list(eng.assembly["attendees"]) for v in voters: # 全员指认江离 eng._cast_vote(v, "门厅", "江离" if v != "江离" else "") assert "江离" in eng.executed and "江离" in eng.state.dead assert eng.ending is not None and eng.ending["kind"] == "凶手伏法" # 终局:真凶被指认 f = eng.frame(1) assert f["ending"]["kind"] == "凶手伏法" and f["executed"] == ["江离"] def test_assembly_wrong_execution_is_tragedy_not_ending(): eng = _engine() for a in eng.minds: eng.state.positions[a] = "门厅" eng._execute("唐曼", eng.minds["唐曼"], Decision(action_kind="assemble")) eng._assembly_advance() for v in list(eng.assembly["attendees"]): eng._cast_vote(v, "门厅", "白聿" if v != "白聿" else "") assert "白聿" in eng.executed and eng.ending is None # 错杀:悲剧被记录,游戏继续 assert any("可他不是凶手" in n["text"] for n in eng.narrative) def test_assembly_no_majority_disperses(): eng = _engine() for a in eng.minds: eng.state.positions[a] = "门厅" eng._execute("白聿", eng.minds["白聿"], Decision(action_kind="assemble")) eng._assembly_advance() voters = list(eng.assembly["attendees"]) for i, v in enumerate(voters): # 票数分散,无人过半 eng._cast_vote(v, "门厅", voters[(i + 1) % len(voters)]) assert not eng.executed and eng.assembly is None assert any("对质无果" in n["text"] for n in eng.narrative) def test_ferry_arrival_ending(): eng = _engine() eng._ferry_arrives() assert eng.ending is not None and eng.ending["kind"] == "凶手逍遥" # 凶手活着混在幸存者中 # ───────────────── G15:钥匙开门 + 弹琴 ───────────────── def test_master_key_unlocks_room_for_killer_only(): eng = _engine() assert eng.state.items["master_key"].holder == "江离" # 钥匙开局在凶手身上 eng.state.positions["唐曼"] = "二层走廊" eng._execute("唐曼", eng.minds["唐曼"], Decision(action_kind="use", action_arg="用钥匙打开青之房")) assert eng.state.facilities["master_door"].state == "locked" # 没钥匙开不了 eng.state.positions["江离"] = "二层走廊" eng._execute("江离", eng.minds["江离"], Decision(action_kind="use", action_arg="用钥匙打开青之房")) assert eng.state.facilities["master_door"].state == "open" assert "青之房" in eng.state.adjacent["二层走廊"] # 拓扑打通,可走进去 assert eng.state.path("二层走廊", "青之房") == ["青之房"] def test_piano_heard_castle_wide(): eng = _engine() eng.state.positions["江离"] = "琴房" heard = [] eng.bus.taps.append(lambda e: heard.append(e)) eng._execute("江离", eng.minds["江离"], Decision(action_kind="use", action_arg="弹琴")) assert any("琴声" in e.content and e.scope == "global" for e in heard) # 全堡可闻 # ───────────────── G15 通用可供性 / G27 旁听 / 会话字幕帧 ───────────────── def test_facility_verbs_bell_and_dig(): eng = _engine() jiang = eng.minds["江离"] eng.state.positions["江离"] = "荒废教堂" heard = [] eng.bus.taps.append(lambda e: heard.append(e)) eng._execute("江离", jiang, Decision(action_kind="use", action_arg="敲钟")) assert any("钟" in e.content and e.scope == "global" for e in heard) # 全岛闻钟 eng.state.positions["温言"] = "温室" eng._execute("温言", eng.minds["温言"], Decision(action_kind="use", action_arg="挖掘土畦")) assert eng.state.facilities["soil"].state == "dug" # 状态持久 assert not eng.state.items["shoot_log"].hidden # 拍摄日志出土 eng._execute("温言", eng.minds["温言"], Decision(action_kind="take", action_arg="拍摄日志")) assert eng.state.items["shoot_log"].holder == "温言" def test_eavesdrop_risk_and_reward(): eng = _engine() tang, cheng, wen = eng.minds["唐曼"], eng.minds["程亦深"], eng.minds["温言"] eng._execute("唐曼", tang, Decision(speech_to="程亦深", speech="账本的事,天亮前必须了结")) import random as _r eng.rng = _r.Random(1) # 第一抽 <0.7:偷听成功 eng._execute("温言", wen, Decision(action_kind="eavesdrop")) assert any("偷听到" in e.content for e in wen.attention._inbox) or \ any("听清了几句" in e.content for e in wen.attention._inbox) def test_frame_carries_sessions_and_ttl_exempt(): eng = _engine() tang = eng.minds["唐曼"] eng._execute("唐曼", tang, Decision(speech_to="温言", speech="跟你说件事")) f = eng.frame(1) assert f["sessions"] and f["sessions"][0]["members"] assert "跟你说件事" in f["sessions"][0]["lines"][-1] eng.clock._anchor_sim += 60 # 过 TTL f2 = eng.frame(2) me = next(a for a in f2["agents"] if a["name"] == "唐曼") assert me["speech"] == "跟你说件事" # 会话中豁免 TTL,多人对话信息可见 def test_longterm_reflect_roundtrip_with_fake_store(): from genesis.runtime.longterm import LongTerm class FakeStore: def __init__(self): self.recs = [] def add(self, rec): self.recs.append(rec) return "id" def search(self, query, agent_id, limit=20): from genesis.memory.store import SearchHit return [SearchHit(record=r, relevance=0.9) for r in self.recs if r.agent_id == agent_id][:limit] lt = LongTerm("game1", store=FakeStore()) assert lt.enabled lt.deposit("温言", "江离三次回避青之房的话题", 23.0) got = lt.recall("温言", "青之房 可疑") assert got and "青之房" in got[0] assert lt.recall("唐曼", "青之房") == [] # agent 隔离 def test_pry_open_blue_room_without_key_is_loud(): """青之房死结修复:无钥匙也能撬开(避免唐曼卡死),但全堡可闻——谁撬的成悬念。""" eng = _engine() eng.state.positions["唐曼"] = "二层走廊" heard = [] eng.bus.taps.append(lambda e: heard.append(e)) # 没钥匙用"用钥匙打开"开不了 eng._execute("唐曼", eng.minds["唐曼"], Decision(action_kind="use", action_arg="用钥匙打开青之房")) assert eng.state.facilities["master_door"].state == "locked" # 但可以撬 eng._execute("唐曼", eng.minds["唐曼"], Decision(action_kind="use", action_arg="撬开青之房")) assert eng.state.facilities["master_door"].state == "open" assert "青之房" in eng.state.adjacent["二层走廊"] # 拓扑打通,能进 assert any("撬" in e.content and e.scope == "global" for e in heard) # 撬锁声全堡可闻