"""WorldEngine v2:时钟/调度/总线/世界状态/Actor/心智的装配与动作执行。 两种心智: - StubMind(规则桩,零 LLM):验证运行时链路,测试默认; - CognitiveMind(LLM):decide() 纯函数在线程池跑,**世界变更只在事件循环线程执行** (快照进、决策出,无共享状态竞态;每 agent 一把锁防自我并发)。 设计原则(docs/agent-runtime-v2.md §6):动作必落为持久世界状态、必有反馈; 对话被点名即接(L1 快节拍),旁人只见"在交谈"(摘要);活跃对话自动降速。 """ from __future__ import annotations import asyncio import itertools import json import random from collections import deque from dataclasses import dataclass, field from pathlib import Path from genesis.manifest.builder import WorldBuilder from genesis.memory.local_store import LocalMemoryStore from genesis.memory.stream import MemoryStream from genesis.obs.logging_setup import get_logger from genesis.runtime.actor import Actor from genesis.runtime.attention import Attention from genesis.runtime.bulletin import dynamic_bulletin, static_common_ground from genesis.runtime.bus import Event, EventBus from genesis.runtime.cognition import CognitiveMind, Decision, ThinkInput, persona_card_from_manifest from genesis.runtime.scheduler import SimScheduler from genesis.runtime.simclock import SimClock from genesis.runtime.soma import Soma from genesis.runtime.stub import StubMind from genesis.runtime.worldstate import WorldState, seed_world logger = get_logger("runtime.engine") HAUNTS = { # 桩模式偏好游荡地 "顾长风": ["书房", "放映室", "门厅"], "邵铭轩": ["书房", "大客厅", "门厅"], "程亦深": ["大客厅", "门厅", "餐厅"], "唐曼": ["大客厅", "二层走廊", "门厅"], "白聿": ["白聿的房间", "门厅", "餐厅"], "温言": ["书房", "暗房", "后山墓园", "门厅"], "江离": ["琴房", "荒废教堂", "门厅"], } DIALOGUE_TURN_S = 35.0 # 对话节拍:轮到发言者 ~35 模拟秒后接话 DIALOGUE_RATE = 14.0 # 活跃对话时降速(字幕 ~2.5s/句仍可读;10× 时群戏会把世界钉死) NIGHT_FF_RATE = 600.0 # 全员入睡快进 SPEECH_TTL_S = 30.0 # 台词气泡存活时长(G4) GESTURE_TTL_S = 45.0 # 神态显示存活时长(G5) PRETHINK_LEAD_S = 100.0 # 预思考提前量:动作/行程结束前~100模拟秒开想,完成即接续 PRETHINK_FRESH_S = 240.0 # 预想结果的保鲜期(期间有突发会被中断重想) SESSION_IDLE_S = 300.0 # 会话无人说话超时解散 GAME_DAYS = 7 # 封岛七天:凶手按七幕逐日复仇,渡船第7天清晨来=幸存者撑到底的逃生 KILL_COOLDOWN_H = 18.0 # 凶手两次作案的最短间隔(模拟小时):一夜至多一案,把复仇摊成逐日推进(治"第一夜连杀三人、之后再不动手") DIRECTOR_PERIOD_S = 45 * 60.0 # Layer1 软导演巡查周期(模拟秒):落后才出手造孤立窗口,不滥用 # 夜话环节:前6天每晚 23:00 聚门厅交换信息+指认(逐日;频率后续 step3 可调,削门厅引力) NIGHT_COURT_H = tuple(23.0 + d * 24.0 for d in range(GAME_DAYS - 1)) SESSION_ROUNDS_PER = 4 # 会话预算:每参与者 4 句(2 人=8 句封顶,G1) META_AWAKEN_AT = 6 # 隐藏结局:需 6/7 角色"窥见"屏幕外的操控者才触发(七天版调高,防剧情未走完就误触谢幕) # 觉醒关键词:只留明确「打破第四面墙」的词——剔掉"有人在安排/操控这一切/岛外的"等局内(江离布局)就会自然说出的词,防误触 META_AWAKEN_WORDS = ( "被注视", "被人看着", "看不见的手", "操控着我们", "操纵着我们", "我们是演员", "幕布之外", "屏幕之外", "镜头之外", "造物主", "上帝之手", "第四面墙", ) @dataclass class Session: """一场会话:同地点 N 人共享同一份上下文,轮流发言,有预算有收尾(G1)。""" sid: int place: str members: list[str] lines: deque = field(default_factory=lambda: deque(maxlen=18)) rounds: int = 0 last_t: float = 0.0 @property def budget(self) -> int: # 会话偏短:聊几轮就散,逼角色回去推进目标(行动力低多半是会话黑洞);7 人群戏封顶 10 句 return min(SESSION_ROUNDS_PER * max(2, len(self.members)), 10) class WorldEngine: def __init__(self, world_dir: str | Path, *, rate: float = 30.0, time_fn=None, demo_beats: bool = False, seed: int = 7, brain: str = "stub", llm=None, prologue: bool = False) -> None: builder = WorldBuilder.from_directory(str(world_dir)) places = {pm.name: { "description": pm.description or "", "geography": pm.geography, "adjacent": [builder.places[a].name for a in pm.adjacent if a in builder.places], } for pm in builder.places.values()} self.state = WorldState(places) seed_world(world_dir, self.state) # 道具层解耦:有 items.json 走数据,否则回退内置 seed(向前兼容) self.place_desc = {pm.name: (pm.description or "") for pm in builder.places.values()} self.prologue = prologue # 序幕:登岛过场(脚本驱动,零 LLM) self._pro_stage: dict[str, int] = {} # 世界层解耦:环境语义(死法/枢纽/取景地/密道/开场钟点)= 数据(world.json);缺则回退内置默认(向前兼容) cfg = self._load_world_config(world_dir) self._place_cause = cfg.get("place_cause") or self._PLACE_CAUSE self._conspicuous = set(cfg.get("conspicuous") or self._CONSPICUOUS) self._ceremonial_venues = tuple(cfg.get("ceremonial_venues") or self.CEREMONIAL_VENUES) self._secret_passages = set(cfg.get("secret_passages") or ["密道"]) self._kill_device = cfg.get("kill_device") # 凶案倒计时装置配置(None → DeviceEngine 用内置雾鸦岛默认) self._killer_items = cfg.get("killer_items") or [] # 凶手开局持有的物品 id(如雾鸦岛的青之房钥匙) self._start_place = cfg.get("start_place", "门厅") # 开局/默认所在地(缺则门厅) self._prologue_place = cfg.get("prologue_place", "码头") # 序幕登岛地(缺则码头) self._prologue_gather = (cfg.get("lore") or {}).get("prologue_gather") or "放映室" # 序幕集合地(听唱片/看放映) self._game_days = int(cfg.get("game_days", GAME_DAYS)) # 封岛天数(节奏数据化;缺则 7) self._night_court_h = tuple(23.0 + d * 24.0 for d in range(self._game_days - 1)) # 每晚 23:00 夜话 # 登岛序幕过场:登岛地→大厅→直接到集合地等候(不再先回各自客房——否则楼上客房的人[尤其凶手] # 会被困在楼上来回奔波、赶不及到一楼集合,序幕一结束就停在楼上。集合地全员到齐才放映/宣读罪状) self._pro_legs = [ ("pier", "拖着行李箱踏上湿滑的栈桥,仰头打量暴雨里的孤岛", 55), ("to_hall", self._start_place, 0), ("hall", "在门厅里抖落雨水、彼此打量——主人却始终不见踪影", 60), ("to_gather", self._prologue_gather, 0), ("seated", "在客厅落座,等着那位不露面的主人现身", 9e9), ] _lore = cfg.get("lore") or {} # 叙事层解耦:验尸线索/终局里的"行刑章法"与"幕后主使"称谓 self._lore = _lore self._ritual_name = _lore.get("ritual_name", "《雾鸦》") self._mastermind = _lore.get("mastermind", "当年写《雾鸦》、又被请来『补完第七幕』的编剧") start_hour = float(cfg.get("start_hour", 21.0)) self.clock = SimClock(rate=rate, start_hour=19.66 if prologue else start_hour, time_fn=time_fn) self.bus = EventBus(self.state.occupants, lambda p: self.state.adjacent.get(p, [])) self.sched = SimScheduler(self.clock) self.rng = random.Random(seed) self.base_rate = rate self.auto_ff = False self.sessions: dict[int, Session] = {} # 活跃会话(降速依据,G3) self._session_of: dict[str, Session] = {} # agent → 所在会话 self._sid = itertools.count(1) self.killer_name = "" self.minds: dict[str, object] = {} self.narrative: list[dict] = [] self.narrator = None # live2 注入(Narrator,可选) self.weather_on = True self.lighthouse_fuel = 0 from genesis.runtime.devices import DeviceEngine self.devices = DeviceEngine(self) # 剧情装置引擎(G13) self.bus.taps.append(self.devices.on_event) self.ferry_eta = ((self._game_days - 1) * 24.0 + 5.5) * 3600.0 # 第7天 05:30 渡船来接=封岛七天的尽头(可被灯塔提前) self.assembly: dict | None = None # 集会状态机(G14) self.executed: list[str] = [] # 被群体处决者 self.ending: dict | None = None # 终局 self.world_goal = "" # 造物主的世界走向(G22) self.acts = self._load_acts(world_dir) # 幕菜单+处境帧+情报档(可缺;缺则 stage_directive 留空,向前兼容旧世界) self._focus_place = "" # 导演镜头当前焦点(迟滞依据,防镜头抽搐) self._last_shove_h = -99.0 # 上次导演造窗口的钟点(冷却,避免连环硬塞) self._last_divert_h = -99.0 # 上次导演"散开人群"诱导的钟点(冷却) self._last_kill_h = -99.0 # 上次凶手亲手作案的钟点(冷却:一夜至多一案,防连环作案) import uuid self.game_id = uuid.uuid4().hex # 本局唯一标识(mem0 user_id 前缀,每局全新→召回不到旧局) self.longterm = None # live2 注入(LongTerm,可选) self._locks: dict[str, asyncio.Lock] = {} self._demo_done: set[str] = set() self._lure_of: dict[str, tuple[str, str]] = {} # 被引诱者 → (引诱者, 僻静地);一次性诱饵 self._planted: dict[str, list[dict]] = {} # 被植入的"假任务"持久念头:目标 → [{place,hook,by,t}](凶手种,上帝视角标红) self._meta_awakened: set[str] = set() # 隐藏结局:已"窥见"屏幕外操控者的角色 self.demo_beats = demo_beats roster = [cm.name for cm in builder.characters.values()] ground = static_common_ground(roster, cfg.get("lore")) # 设定层解耦:世界设定文本来自 world.json 的 lore(缺则回退雾鸦岛) for cm in builder.characters.values(): name = cm.name is_killer = "凶手" in (getattr(cm, "role", "") or "") if is_killer: self.killer_name = name home = f"{name}的房间" home = home if home in places else "门厅" self.state.positions[name] = self._start_place actor = Actor(name, self.state, self.clock) if not is_killer: # 密道是凶手专属暗道(世界层配置 secret_passages):非凶手不可走 actor.blocked = set(self._secret_passages) attention = Attention(name) if brain in ("llm", "langgraph"): # langgraph:LangGraph 图 + 结构化输出大脑(LangGraphMind 是 CognitiveMind 子类,引擎零改动); # llm:既有 litellm 直答大脑。两者共用人设/目标/反思,可逐 agent 灰度、随时回滚对照。 MindCls = CognitiveMind if brain == "langgraph": from genesis.runtime.langgraph_mind import LangGraphMind MindCls = LangGraphMind mind = MindCls( name, persona_card_from_manifest(cm), is_killer=is_killer, home=home, llm=llm, memory=MemoryStream(LocalMemoryStore(), agent_id=name), static_ground=ground) for other, desc in (getattr(cm, "relationships", None) or {}).items(): mind.relations[other] = {"impression": desc, "trust": 0.0} # 初始关系网(G8) if not is_killer: # #3:把 dossier 里"此人私罪物证在何处"注入其自身议程 → 开场就有去某地的动力 d = (self.acts.get("dossier") or {}).get(name) or {} if d.get("want") and d.get("evidence_place"): mind.agenda_seed = {"want": d["want"], "place": d["evidence_place"]} else: mind = StubMind(name, home, HAUNTS.get(name, ["门厅"]), random.Random(seed + len(name)), actor, attention, Soma(last_meal_h=19.5, woke_at_h=8.0)) mind.actor = actor mind.attention = attention mind.recent_facts = deque(maxlen=12) # 事实锚:近期确切经历(G2,只收亲历亲见) mind.hearsay = deque(maxlen=10) # 传闻锚:别人说的话(是说法不是事实,须自行推理掂量) if is_killer: for iid in self._killer_items: # 凶手开局持有的物品(世界层配置 killer_items) if iid in self.state.items: self.state.items[iid].holder = name mind.last_speech_t = -1e9 # 气泡 TTL(G4) mind.last_gesture = "" # 神态通道(G5) mind.last_gesture_t = -1e9 mind.thinking = False # LLM 在途占位(💭) mind.pending = None # 预思考结果:(Decision, made_at_sim_s) attention.on_interrupt = lambda e, n=name: self._interrupt(n, e) self.bus.register(name, attention.receive) self.minds[name] = mind self._locks[name] = asyncio.Lock() # ───────────────────── 启动与唤醒 ───────────────────── def start(self) -> None: if self.prologue: for name in self.minds: self.state.positions[name] = self._prologue_place self._pro_stage[name] = 0 self._narr("🛥 序幕——渡船把七位客人留在了暴雨中的码头,自己调头离去。") self.sched.schedule("__prologue__", self.clock.sim_seconds + 5, "prologue") else: for i, name in enumerate(self.minds): self.sched.schedule(name, self.clock.sim_seconds + i * 8, "boot") if self.demo_beats: self.sched.schedule("__demo__", self.clock.sim_seconds + 2.5 * 3600, "demo") if self.weather_on: self.sched.schedule("__weather__", self.clock.sim_seconds + 1800, "weather") self.sched.schedule("__ferry__", self.ferry_eta, "ferry") self.sched.schedule("__dawn__", (24.0 + 5.0) * 3600.0, "dawn") for i, h in enumerate(self._night_court_h): # 夜话环节(交换信息+指认) self.sched.schedule(f"__court__{i}", h * 3600.0, "court") self.sched.schedule("__director__", self.clock.sim_seconds + DIRECTOR_PERIOD_S, "director") # Layer1 软导演:造孤立窗口 async def on_wake(self, aid: str, reason: str) -> None: if self.ending is not None and not aid.startswith("__"): return # 终局已定,世界谢幕 if aid == "__demo__": self._run_demo() return if aid == "__weather__": self._run_weather() return if aid.startswith("__device__:"): self.devices.fire(int(aid.split(":")[1])) return if aid == "__ferry__": self._ferry_arrives() return if aid == "__dawn__": self.bus.publish(Event(etype="ambient", content="天光破晓,雨声渐歇。远处隐约传来渡船的汽笛——船快到了,该把所有人聚到一处清点。", scope="global", sim_t=self.clock.sim_seconds, salience=0.7)) return if aid == "__director__": self._director_tick() return if aid == "__assembly__": self._assembly_advance() return if aid.startswith("__court"): # __court__N(定点)与 __court_retry__(零死亡推迟) self._night_court() return if aid == "__prologue__": self._prologue_tick() return mind = self.minds.get(aid) if mind is None or aid in self.state.dead: return async with self._locks[aid]: await self._wake_locked(aid, mind, reason) async def _wake_locked(self, aid: str, mind, reason: str) -> None: act = mind.actor.current if reason == "prethink": # 动作还没做完:提前想好下一步(不执行) if isinstance(mind, CognitiveMind) and act is not None \ and act.until > self.clock.sim_seconds: snap = self._snapshot(aid, mind) if act.kind == "move" and act.dest: snap.arriving = act.dest # 赶路中:预想"到达后的第一步" mind.thinking = True dec = await asyncio.to_thread(mind.decide, snap) mind.thinking = False mind.pending = (dec, self.clock.sim_seconds) nxt = mind.actor.current self.sched.schedule(aid, nxt.until if nxt else self.clock.sim_seconds, "tick") return reason = "tick" # 动作已结束,按常规处理 if reason.startswith("interrupt:") and act is not None: mind.pending = None # 突发让预想作废,重新评估 broken = mind.actor.interrupt() if broken: self._feedback(aid, f"突发状况打断了你({broken})。") elif act is not None and act.until <= self.clock.sim_seconds: before = self.state.positions.get(aid, "") payload = act.payload status, detail = mind.actor.advance() if status in ("hop", "arrived"): self._publish_transit(aid, before, detail) if status == "hop": # 凶手赶路途中撞见本幕目标 → 别傻走,立刻停下重新决策(去下饵/把握机会),不再一路走到黑 if getattr(mind, "is_killer", False) and self._killer_target_here(aid): self.sched.schedule(aid, self.clock.sim_seconds, "interrupt:encounter") else: self.sched.schedule(aid, mind.actor.current.until, "tick") return if status == "arrived": self._feedback(aid, f"你到了{detail}。" + " ".join(self.state.describe(detail, aid)[:4])) elif status == "finished": self._finish_action(aid, payload, detail) elif act is not None: # 未到点被唤醒(对话等):打断站定再思考 if reason == "dialogue" and act.interruptible: mind.actor.interrupt() else: self.sched.schedule(aid, act.until, "tick") return if self.prologue: # 序幕:脚本在 _prologue_tick 驱动,不思考 nxt = mind.actor.current if nxt is not None: self.sched.schedule(aid, nxt.until, "tick") return # —— 思考(优先用保鲜期内的预想结果:动作一完成立即接续,不再站桩等 LLM)—— if isinstance(mind, CognitiveMind): if mind.imp_acc >= 6.0 and self._session_of.get(aid) is None: # 冲击累积 → 先反思(G7) mind.imp_acc = 0.0 mind.thinking = True await asyncio.to_thread(mind.reflect, self.clock.label(), self.longterm) mind.thinking = False dec = None if mind.pending is not None: cached, made_at = mind.pending mind.pending = None # 预想结果须仍新鲜 **且** 其针对的目标此刻仍在场——否则作废、用最新世界状态重想 # (治"凭旧快照对着已离开的人动手":如沃格雷夫以为马斯顿在旁、实则人已走) if self.clock.sim_seconds - made_at <= PRETHINK_FRESH_S and self._prethink_still_valid(aid, cached): dec = cached if dec is None: snap = self._snapshot(aid, mind) mind.thinking = True dec = await asyncio.to_thread(mind.decide, snap) mind.thinking = False self._execute(aid, mind, dec) else: place = self.state.positions.get(aid, "门厅") occupants = [a for a in self.state.occupants(place) if a != aid] mind.think(self.clock.sim_hours, place, occupants) if mind.last_speech: mind.last_speech_t = self.clock.sim_seconds self.bus.publish(Event(etype="speech", content=f"{aid}说:「{mind.last_speech}」", place=place, actor=aid, sim_t=self.clock.sim_seconds, salience=0.35, muffled="隔壁隐约有人在说话")) nxt = mind.actor.current now = self.clock.sim_seconds if nxt is not None and nxt.kind != "sleep" and nxt.until - now > PRETHINK_LEAD_S * 1.3: self.sched.schedule(aid, nxt.until - PRETHINK_LEAD_S, "prethink") # 一切动作皆预思考 elif nxt is not None: self.sched.schedule(aid, nxt.until, "tick") else: in_sess = self._session_of.get(aid) is not None self.sched.schedule(aid, now + (DIALOGUE_TURN_S * 4 if in_sess else 120), "tick") def _visible_desc(self, place: str, aid: str) -> list[str]: """天气能见度过滤的可见描述:露天+低能见度时,尸体被雨幕/浓雾遮住,不走近看不出。""" from genesis.runtime.weather import visibility lines = self.state.describe(place, aid) if self.state.is_indoor(place) or visibility(self.state.weather) >= 0.45: return lines out, hidden_body = [], False for ln in lines: if ln.startswith("⚠️") and "尸体" in ln: hidden_body = True continue out.append(ln) out.append(f"({self.state.weather}里视线极差,几步外只剩模糊轮廓,看不清远处的动静。)") if hidden_body: out.append("(地上隐约横着什么东西,可雨雾遮着看不真切——除非走近细看。)") return out # ───────────────────── 认知:快照与执行 ───────────────────── def _prethink_still_valid(self, aid: str, dec: "Decision") -> bool: """预想结果是否仍贴合当前世界:针对具体某人的动作(kill/lure/eavesdrop/指向性说话), 该目标必须此刻仍与本人同处一地且活着;否则判定为基于旧状态、作废重想。""" place = self.state.positions.get(aid, "") here = set(self.state.occupants(place)) tgt = (dec.action_arg or "").strip() if dec.action_kind in ("kill", "lure", "eavesdrop") and tgt: if tgt not in here or tgt in self.state.dead: return False if dec.speech_to and dec.speech_to not in here: return False return True def _snapshot(self, aid: str, mind: CognitiveMind) -> ThinkInput: now_h = self.clock.sim_hours place = self.state.positions.get(aid, "门厅") raw_occupants = [a for a in self.state.occupants(place) if a != aid] occupants = [] for other in raw_occupants: # 社会感知信号(G18-lite):可见动作+神色 om = self.minds.get(other) bits = [] oact = om.actor.current if om else None if oact is not None: bits.append(oact.desc) fear = om.soma.snapshot(now_h)["fear"] if om else 0.0 if fear > 0.5: bits.append("神色惊惶") elif fear > 0.25: bits.append("神色不安") occupants.append(f"{other}({','.join(bits)})" if bits else other) sess = self._session_of.get(aid) if sess is not None and not any(m in raw_occupants for m in sess.members if m != aid): self._leave_session(aid, farewell=False) # 同场无会话成员 → 自然退出 sess = None events = [] for e in mind.attention.drain(): events.append(e.content) if e.salience >= 0.7: mind.soma.shock(now_h, min(0.9, e.salience)) if e.etype == "speech" and not e.content.startswith(f"{aid}说"): # 别人说的话:你确切知道的只是「他说了这句」,内容真假要自己掂量 → 入传闻锚,不当事实 mind.hearsay.append(e.content) elif e.etype in ("feedback", "action", "ambient", "speech"): mind.recent_facts.append(e.content) # 事实锚:亲历/亲见(自己的反馈/动作/环境/自语)才算事实(G2) imp = min(1.0, e.salience + 0.1) mind.imp_acc += imp # 冲击累积:达阈值触发反思(G7) mind.memory.observe(f"{self.clock.label()},{e.content}", now_h, importance=imp) # 环境感知入记录(用户需求):身处之地的客观状态(空刀架/烧过的壁炉/出土的物件…) # 不只进 prompt,还要落为事实锚与记忆——每个独特观察只记一次,状态变化会再记 seen = getattr(mind, "_env_seen", None) if seen is None: seen = mind._env_seen = set() visible = self._visible_desc(place, aid) # 天气遮蔽过的可见描述(露天低能见度藏尸) for line in visible[:6]: # 只记「物」的状态(设施/物品/痕迹);人员在场名单时刻在变,由事件流负责 if line and line not in seen and not line.startswith("在场") and not line.startswith("("): seen.add(line) mind.recent_facts.append(f"(在{place})- {line}") mind.memory.observe(f"{self.clock.label()},我在{place}注意到:{line}", now_h, importance=0.35) query = f"{place} {mind.intent} " + (events[-1] if events else "") memories = [m.content for m in mind.memory.retrieve(query, now_h, top_k=4)] # 线索气味(G17):此地还藏着没被翻出来的东西 → 提示值得 search,诱发探索(搜出后自然消失) scent = "" if any(it.place == place and it.hidden and not it.destroyed for it in self.state.items.values()): scent = " (这地方有种说不出的不对劲——像是还藏着什么没被翻出来的东西,值得搜一搜。)" # 假任务(G18):凶手种下的持久念头——目标当成听来的线索惦记着(可识破)。到了那地方/超时即解除。 planted_leads = [] pl = self._planted.get(aid) if pl: for x in [x for x in pl if x["place"] == place or now_h - x.get("t", now_h) > 5.0]: pl.remove(x) # 已赶到被引去的地方,或念头放了5小时还没动→自然消退 for x in pl: tr = self._trust_of(aid, x["by"]) # 越信源头越当真,越疑越容易识破(信任影响信息可信度) doubt = (f"(不过你一向提防{x['by']},他特意把这事单告诉你——八成是想支你离群、引你入局,多半别去。)" if tr <= -0.3 else "" if tr >= 0.3 else f"(不过仔细想想:他为什么单单告诉你?会不会是想支你离群?去不去、信不信,自己掂量。)") planted_leads.append( f"{x['by']}之前私下跟你说过:{x['hook']},就在{x['place']}——这事儿你一直搁在心上,总想找机会去{x['place']}看看。{doubt}") if not pl: del self._planted[aid] return ThinkInput( now_h=now_h, time_label=self.clock.label(), place=place, place_desc=(self.place_desc.get(place, "") + " " + " ".join(visible)).strip() + scent, occupants=occupants, events=events, exits=[e for e in self.state.adjacent.get(place, []) if e not in getattr(mind.actor, "blocked", ())], inventory=[it.name + (f"({it.hint})" if it.hint else "") for it in self.state.items.values() if it.holder == aid and not it.destroyed], soma_lines=mind.soma.prompt_lines(now_h), candidates=mind.soma.candidates(now_h), bulletin=self._bulletin(), memories=memories, intent=mind.intent, facts=list(mind.recent_facts), hearsay=list(getattr(mind, "hearsay", [])), dialogue=list(sess.lines) if sess else [], dialogue_with="、".join(self._others_in_session(sess, aid)), dialogue_rounds=sess.rounds if sess else 0, dialogue_budget=sess.budget if sess else 0, planted_leads=planted_leads, private_note=self._killer_pressure() if getattr(mind, "is_killer", False) else "", lure_bait=self._consume_lure_bait(aid), stage_directive=self._stage_directive(aid, mind)) def _consume_lure_bait(self, aid: str) -> str: """被引诱者的一次性诱饵提示(命中私欲→大概率跟,但自主可拒绝)。 若诱饵正指向目标本就想去的僻静处(顺水推舟),给强拉力——这是凶案脊柱能否推进的关键。""" lure = self._lure_of.pop(aid, None) if not lure: return "" who, where = lure tm = self.minds.get(aid) # 强拉力条件:① 引向你本就惦记的私事地点;② 引向本幕主题取景地(凶手用你的私欲做了个去那儿的由头) act = self._current_act() themed = act and act.get("target") == aid and where in (act.get("place") or []) aligned = themed or any(where in (g.get("desc", "") or "") for g in (getattr(tm, "goals", None) or []) if g.get("status") in ("todo", "doing")) trust = self._trust_of(aid, who) # 信任决定你信他几分:越信越容易被说动,越疑越容易识破 if trust >= 0.3: tnote = f"你一向信得过{who},这话听着不像有假——" elif trust <= -0.3: tnote = f"你本就提防{who},他特意来跟你咬耳朵,八成没安好心——" else: tnote = "" if aligned: # 引向你本就惦记的地方 → 顺道就去了(仍可自主拒绝,但拉力强) return (f"({tnote}{who}压低声音说{where}那边有你正惦记的东西、只有你俩知道——" f"那本就是你今晚想去的地方,顺道现在就去一趟正合心意。" f"夜里孤身赴约虽不是全无风险,但这机会你多半不想错过。)") return (f"({tnote}{who}刚把你单独往{where}引,说那儿有你想要的东西、只有你俩知道。" f"这可能正合你心意——也可能是冲着你来的陷阱。去不去,你自己拿主意。)") def _apply_goal_update_checked(self, aid: str, mind, gu: dict) -> None: """目标自评的世界回执校验:凶手把『杀人类目标』标 done,必须有真实死亡背书。 否则脑中预演会进台账,下次思考把台账当事实,谎言自我固化(实测:江离幻想杀了顾长风)。""" try: idx = int(gu.get("index", 0)) - 1 except (TypeError, ValueError): return goals = getattr(mind, "goals", None) or [] if (getattr(mind, "is_killer", False) and 0 <= idx < len(goals) and str(gu.get("status", "")) == "done"): desc = goals[idx]["desc"] kill_words = ("kill", "杀", "干掉", "清算", "灭口", "了结") if any(w in desc for w in kill_words): named = [n for n in self.minds if n != aid and n in desc] still_alive = [n for n in named if n not in self.state.dead] organic_kills = [d for d in self.state.dead if d not in self.executed] no_receipt = still_alive or (not named and not organic_kills) if no_receipt: who = "、".join(still_alive) or "目标" self._feedback(aid, f"(你猛然清醒:{who}还活着,你从未真正动手——" f"刚才那只是你脑中的预演。没有落下的幕,不算完成。)") gu = dict(gu) gu["status"] = "doing" gu["note"] = "尚未真正动手" mind.apply_goal_update(gu) @staticmethod def _load_acts(world_dir) -> dict: """加载幕菜单 acts.json(可缺;旧世界没有这个文件 → stage_directive 留空,向前兼容)。""" p = Path(world_dir) / "acts.json" if not p.exists(): return {} try: return json.loads(p.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError) as e: logger.warning("acts.json 加载失败:%s", e) return {} @staticmethod def _load_world_config(world_dir) -> dict: """加载世界环境配置 world.json(可缺;缺则环境语义回退内置默认,向前兼容旧世界)。""" p = Path(world_dir) / "world.json" if not p.exists(): return {} try: return json.loads(p.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError) as e: logger.warning("world.json 加载失败:%s", e) return {} def _current_act(self) -> dict | None: """简易幕推导(Layer 0,确定性):菜单里第一个『目标还活着』的固定幕 = 当前幕; 固定目标都清算完则回退到机动幕。Layer 1 导演再做事件驱动的智能幕选择。""" acts = (self.acts or {}).get("kill_acts") or [] for a in acts: t = a.get("target") if t and t not in self.state.dead: return a for a in acts: # 固定目标都死了 → 机动幕(flex)兜底 if not a.get("target"): return a return None # 实地死法推断:水边=溺,高处=坠,黑暗地下=钝击/电,琴房=绞,餐厨=毒/利器,余=外伤。验尸据此,不被剧本污染。 _PLACE_CAUSE = { "码头": "溺水——肺里呛着海水,死亡时间不久", "船屋": "溺水——肺里呛着海水", "沙滩": "溺水后被潮水冲上岸", "灯塔": "高处坠落,多处骨折", "断崖": "高处坠落,摔得不成样子", "露台": "从高处坠下,颈骨折断", "发电机房": "黑暗中遭袭,像是触电加钝器伤", "酒窖": "黑暗中被钝器击中后脑", "冰窖": "失温与钝击,冻得发青", "储藏室": "暗处被钝器袭击", "密道": "窄道里被勒/击,无处可逃", "琴房": "颈上一道细勒痕——像是琴弦绞杀", "餐厅": "口唇发绀、气味发苦——疑似中毒", "厨房": "利器伤,失血致死", "荒废教堂": "颈部勒痕、悬空吊起的痕迹——像是被钟绳吊死", } def _death_method(self, target: str, place: str) -> tuple[str, bool]: """返回(客观死法描述, 是否对得上本幕仪式取景地)。在取景地=仪式死法(补拍线索成立);否则按实地推断。""" act = next((a for a in ((self.acts or {}).get("kill_acts") or []) if a.get("target") == target), None) if act and place in (act.get("place") or []) and act.get("method"): return act["method"], True return self._place_cause.get(place, "致命外伤,死亡时间不久——绝非意外,是人为"), False def _current_phase(self) -> dict | None: """简易处境帧推导:死亡数 + 集会/渡船 → 全员当前处境帧(Layer 0 确定性)。""" phases = (self.acts or {}).get("phases") or [] if not phases: return None deaths = len(self.state.dead) ferry_h = max(0.0, (self.ferry_eta - self.clock.sim_seconds) / 3600.0) if self.assembly is not None: pid = "court" elif self.ending is not None or deaths >= 3 or ferry_h <= 1.5: pid = "finale" elif deaths >= 2: pid = "escalation" elif deaths >= 1: pid = "first_blood" else: pid = "prologue" by_id = {p.get("id"): p for p in phases} return by_id.get(pid) or phases[0] def _stage_directive(self, aid: str, mind) -> str: """导演层↔认知层接口:凶手收到本幕任务,幸存者收到当前处境帧(按幕替换)。""" if not self.acts or self.ending is not None: return "" if getattr(mind, "is_killer", False): return self._killer_stage(aid) p = self._current_phase() if not p or not p.get("frame"): return "" return f"【此刻·{p.get('name', '')}】{p['frame']}" # 人来人往的枢纽/走廊:在这动手必被撞见,凶手绝不在此下手(尸体也会堆在门厅=夜话场) _CONSPICUOUS = {"门厅", "大客厅", "二层走廊", "三层走廊"} def _killer_stage(self, aid: str = "") -> str: """凶手的本幕指令:**严格按幕**——只清算本幕目标、只在僻静取景地下手。 R3 杀机窗口只在『与本幕目标独处、且身处僻静处』时才触发;非本幕目标或身处枢纽,一律压住。""" a = self._current_act() reluctant = {x.get("target") for x in ((self.acts or {}).get("kill_acts") or []) if "reluctant" in (x.get("flags") or [])} cur_target = a.get("target") if a else None ideal = "/".join(a.get("place") or []) if a else "" lines = [] cooling = self._killer_on_cooldown() # 刚动过手 → 蛰伏冷却:压住杀机、改为铺垫 if aid and cooling: # 冷却期:无论是否撞见目标,都不给落刀窗口,只提示蛰伏铺路 lines.append("【蛰伏·按住手】你昨夜刚动过手,此刻最忌连续作案——今天收敛锋芒、演回那个无害的局外人," "留心有没有人起了疑;可以悄悄物色、用诱饵慢慢替下一幕铺路,但**今夜先按住手,别再杀人**,等风头过了、夜再深些再说。") elif aid: # R3 落刀闭环:撞见目标=把握机会;独处+僻静=动手;非目标/枢纽=压住 place = self.state.positions.get(aid, "") others = [o for o in self.state.occupants(place) if o != aid and o not in self.state.dead] tgt_here = bool(cur_target and cur_target in others and cur_target not in reluctant) if len(others) == 1 and tgt_here: # 与本幕目标独处 if place not in self._conspicuous: at_venue = place in (a.get("place") or []) if a else False lines.append(f"【⚡窗口已到·正是此人此地】你和本幕该清算的 {cur_target} 单独在【{place}】、四下无人,这里也够僻静" + ("、正是本幕的取景地" if at_venue else "") + f"——这就是你筹划十年的那一刻,这一拍直接 kill,arg={cur_target}。错过再聚起人来就杀不成了。") else: where = ideal or "僻静的取景地" lines.append(f"你和 {cur_target} 此刻独处,但【{place}】是人来人往的枢纽、太显眼,在这儿动手必被当场撞见、十年布局毁于一旦。" f"**绝不在这里下手**——用他正惦记的东西做钩子,把他引去{where}那种僻静处,到那儿才了断。") elif tgt_here: # 目标在场但还有别人(擦肩/人群)→ 立刻下饵,别擦肩而过 where = ideal or "僻静的取景地" lines.append(f"【⚡机会·目标就在眼前】本幕该清算的 {cur_target} 此刻就和你在【{place}】(周围还有别人)。" f"**别傻站着、更别擦肩而过走开**:立刻用 lure 凑到 {cur_target} 耳边,拿TA最惦记的东西做钩子," f"把TA单独勾去{where}——错过这次照面,下回未必这么巧。") elif len(others) == 1 and others[0] != cur_target and others[0] not in reluctant: # 与非本幕目标独处 lines.append(f"你和 {others[0]} 此刻独处——但今晚按你的仪式,该先清算的是 {cur_target},不是他。" f"沉住气、按幕一个个来,别为一时机会打乱顺序、打草惊蛇;" f"**除非他已经当众识破了你、不灭口就要暴露**,否则这一刀先忍住。") # 目标是白聿(reluctant)或不在场:不特意提示 if not a: return "\n".join(lines) target = cur_target or "(机动:当下最危险或最落单的人)" lines.append(f"【本幕·第{a.get('act', '')}幕「{a.get('name', '')}」】这一幕、且只有这一幕该清算的是 {target}。") if ideal: lines.append(f"主题取景地(僻静、对得上死法、最有仪式感):{ideal};死法:{a.get('method', '')}。**务必把他引到这儿再动手。**") if a.get("cover"): lines.append(f"掩护:{a['cover']}。") if a.get("bait"): lines.append(f"怎么把他/她弄过去:{a['bait']}。") if "reluctant" in (a.get("flags") or []) and a.get("reluctant_note"): lines.append(a["reluctant_note"]) if cooling: # 冷却期只交代"下一幕是谁",不催落刀 return "\n".join(lines) lines.append("【按幕推进·这是你的执念,也是不被识破的护身符】你有一整天,不必慌:**严格按七幕顺序,一次只清算本幕这一个目标**," "用他正想要的东西做钩子(命中私欲他才肯跟),把他引到主题取景地,下饵后立刻跟上、一起过去,在那僻静处只剩你俩时动手。" "**门厅/大客厅/走廊这类人来人往的枢纽绝不动手**(必被撞见);一时引不到取景地,就先蛰伏、改日再图,或退而找另一处僻静角落," "**绝不逮谁落单就杀谁、绝不在众目睽睽下仓促下手**。只有当某人已经当众识破你、不灭口就要暴露时,才破例提前清算(且仍挑僻静处)。") return "\n".join(lines) def _killer_target_here(self, aid: str) -> bool: """凶手当前位置是否就有本幕该清算的目标(且非豁免对象)——用于"撞见即把握"。""" a = self._current_act() t = a.get("target") if a else None return bool(t and t not in self._spared() and t in self.state.occupants(self.state.positions.get(aid, ""))) def _spared(self) -> set[str]: """凶手下不去手的人(acts.json reluctant 标记,如白聿——那晚跪在雨里救她的人)。""" return {x.get("target") for x in ((self.acts or {}).get("kill_acts") or []) if "reluctant" in (x.get("flags") or []) and x.get("target")} def _killer_on_cooldown(self) -> bool: """凶手刚动过手 → 蛰伏冷却(约一夜):此间不给杀机窗口、世界也拒绝再次落刀, 把连环作案摊成逐日一案(治『第一夜连杀三人、之后再不动手』的节拍崩塌)。""" return (self.clock.sim_hours - self._last_kill_h) < KILL_COOLDOWN_H def _killer_pressure(self) -> str: """凶手的私有 prompt:① 世界回执(实际杀了几个,防自我叙事污染)② 逐日复仇节拍压力(七天版)。""" if self.ending is not None: return "" kills = [d for d in self.state.dead if d not in self.executed] others = [a for a in self.minds if a != self.killer_name] remaining = len(others) - len(kills) spared = self._spared() culpable_alive = [a for a in others if a not in self.state.dead and a not in spared] anchor = ("【世界回执·只有你知道】你已清算:" + str(len(kills)) + "人" + (f"(已死:{'、'.join(kills)})" if kills else "——至今无人死亡") + f";复仇名单还剩 {remaining} 人未清算。" "你的目标台账或记忆若与此矛盾,一律以本回执为准:没有回执的『完成』只是想象。") # A3 信息差:你布局三年、摸透了每个人——大致猜得到此刻各人惦记什么、人在哪(别人没有这种洞察) intel = [] for a in others: if a in self.state.dead: continue m = self.minds.get(a) top = "" for g in (getattr(m, "goals", None) or []): if g.get("status") in ("todo", "doing"): top = (g.get("desc") or "")[:24] break loc = self.state.positions.get(a, "?") intel.append(f" · {a}:此刻多半在{loc}一带" + (f",心里惦记着「{top}」" if top else "") + "。") if intel: anchor += ("\n【你对他们了如指掌·只有你知道】你太清楚这些人的软肋与心事了,几乎一眼就能猜出他们此刻在想什么、会往哪儿钻——" "拿这点信息差去精准下饵、错峰避人、把矛头引开:\n" + "\n".join(intel) + "\n但切记:你掌握这些是天大的破绽,行事说话务必像个一无所知、与此无关的人,绝不能流露出你对谁的隐私、对岛上机关了如指掌。") if not culpable_alive: # 仇人(白聿除外)已尽数清算 sp_alive = [a for a in others if a not in self.state.dead and a in spared] tail = f",只剩你下不去手的{'、'.join(sp_alive)}" if sp_alive else "" return anchor + f"\n【复仇将竟】仇人已尽数清算{tail}——大功告成,收敛痕迹,只待天明散场。" day = int(self.clock.sim_hours // 24) + 1 days_left = max(1, self._game_days - day + 1) # 节拍:大约一天清算一个(不强求);落后(已清算 < 已过天数)则加压 if len(kills) < day - 1: return anchor + (f"\n【复仇的进度·落后了】已是第{day}天,你才清算了{len(kills)}人——慢了。" f"只剩{days_left}天、还有{remaining}个仇人。今天就锁定一个落单的目标," f"用命中私欲的诱饵把他引去僻静的取景地,确认独处即动手,别再拖。") return anchor + (f"\n【复仇的进度】第{day}天,已清算{len(kills)}人,还剩{remaining}个仇人、{days_left}天。" f"按部就班:今天再清算一个,得手即收敛、演回那个无害的局外人,留意别被识破——" f"被抓住,十年的复仇就半途而废了。") _DIVERT_FLAVORS = [ "{p}方向传来一声闷响,像是有什么东西倒了——又很快没了声息。", "隐约听见{p}那边有动静,脚步?还是风?分辨不清。", "一阵穿堂风带来{p}方向的异响,在这死寂的夜里格外刺耳。", "{p}那头似乎有什么东西被碰倒、又被人匆匆扶住——谁在那边?", ] def _disperse_herd(self) -> None: """软杠杆:有人扎堆又无事发生时,在别处制造一声『异响』,把好奇/胆大的人引开 —— 既治『全员挤在安全屋漫游闲聊』,又顺手把人引向埋着线索的空场景(探索+给凶手腾独处窗口)。 纯环境提示,不强制移动;聪明 agent 才会去看。集会/停电/刚出命案时不触发。""" if (self.assembly is not None or not self.state.power_on or (self.clock.sim_hours - self._last_divert_h) < 2.5): return # 最大扎堆(活人) from collections import Counter crowd = Counter(p for a, p in self.state.positions.items() if a not in self.state.dead) if not crowd: return hot_place, n = crowd.most_common(1)[0] if n < 4: # 不到 4 人扎堆不算"羊群",不打扰 return # 引向的目的地:优先还藏着线索、且当前没人的地点(去了有收获,也真把人引开) empty_with_clue = [pl for pl in self.state.adjacent if not self.state.occupants(pl) and pl != hot_place and any(it.place == pl and it.hidden and not it.destroyed for it in self.state.items.values())] pool = empty_with_clue or [pl for pl in self.state.adjacent if pl != hot_place and not self.state.occupants(pl)] if not pool: return dest = pool[self.rng.randrange(len(pool))] flavor = self._DIVERT_FLAVORS[self.rng.randrange(len(self._DIVERT_FLAVORS))] self.bus.publish(Event(etype="ambient", content=flavor.format(p=dest), scope="global", sim_t=self.clock.sim_seconds, salience=0.5)) self._last_divert_h = self.clock.sim_hours logger.info("[导演] 散羊群:%s 处扎堆 %d 人 → 在 %s 制造异响引开", hot_place, n, dest) def _director_tick(self) -> None: """Layer1 软导演:凶手落后逐日节拍时,主动『造一个孤立窗口』—— 把当幕目标用一念拽向僻静处、并提示凶手跟进(soft 兜底硬手:目标仍可抗拒,但拉力强)。 治『凶手有杀心无杀机、七天杀不动』的脊柱卡死(审查 P0-1/P0-2)。""" if self.ending is None: # 始终续巡(终局后停) self.sched.schedule("__director__", self.clock.sim_seconds + DIRECTOR_PERIOD_S, "director") if self.ending is not None or self.prologue or self.killer_name in self.state.dead: return self._disperse_herd() # 先尝试"散开扎堆"(随时可用的软杠杆,顺带把人引向有线索的空场景) if self._killer_on_cooldown(): # 凶手仍在蛰伏冷却 → 不造窗口(否则白引一场、与"今夜按住手"自相矛盾) return kills = [d for d in self.state.dead if d not in self.executed] day = int(self.clock.sim_hours // 24) + 1 # 落后判定:按一天一杀,到第 day 天该有 ~day 个;只在落后时出手(不滥用),且有冷却 if len(kills) >= day or (self.clock.sim_hours - self._last_shove_h) < 4.0: return killer = self.killer_name kplace = self.state.positions.get(killer, "门厅") # 选目标:当幕目标优先,否则任一活着的有罪者(白聿豁免) act = self._current_act() target = act.get("target") if act else None if not target or target in self.state.dead or target in self._spared(): culp = [a for a in self.minds if a != killer and a not in self.state.dead and a not in self._spared()] target = culp[0] if culp else None if not target: return # 已经独处了就交给 R3,不用造窗口 here = [a for a in self.state.occupants(kplace) if a not in self.state.dead] if self.state.positions.get(target) == kplace and len(here) == 2 and target in here: return venue, hook = self._lure_hook(target, self.state.positions.get(target, kplace)) self._lure_of[target] = (killer, venue) # 目标快照会收到强拉力诱饵提示 self.whisper(target, f"你心里那件惦记已久的事({hook})突然按捺不住——趁现在没人盯着," f"你很想立刻一个人去{venue}把它了结,别再拖了。") self.whisper(killer, f"你察觉{target}动了心思、正要独自往{venue}去——这正是你等的窗口。" f"悄悄跟过去,确认只剩你俩,就落幕。") self._last_shove_h = self.clock.sim_hours logger.info("[导演] 造孤立窗口:拽 %s→%s,提示凶手 %s 跟进(第%d天,已杀%d)", target, venue, killer, day, len(kills)) def _execute(self, aid: str, mind: CognitiveMind, dec: Decision) -> None: now = self.clock.sim_seconds place = self.state.positions.get(aid, "门厅") if dec.empty: # G12:模型哑火 → 出神,30 秒后重试 mind.last_thought = "(怔怔出神)" mind.actor.idle(max(8, int(0.8 * self.clock.rate)), "怔怔出神") # ~0.8 真实秒就重试 return if dec.scheme: # 盘算落日志(debug:看手段-目的推理是否生效,凶手尤其关注) logger.info("[盘算] %s: %s", aid, dec.scheme) mind.last_thought = dec.thought mind.intent = dec.intent or mind.intent mind.last_speech = dec.speech if dec.speech: mind.last_speech_t = now # 气泡 TTL 计时(G4) if dec.gesture: # 神态通道(G5):非台词,可被同场者感知 mind.last_gesture = dec.gesture mind.last_gesture_t = now self.bus.publish(Event(etype="action", content=f"※{aid}{dec.gesture}", place=place, actor=aid, sim_t=now, salience=0.2)) if dec.goal_update: self._apply_goal_update_checked(aid, mind, dec.goal_update) # 自评须过世界回执(G9/G10) if dec.replan and hasattr(mind, "apply_replan"): mind.apply_replan(dec.replan) # 局势剧变 → 重排分目标(动机不变,路径随机应变) if dec.speech: self._do_speech(aid, mind, place, dec) if dec.leave_dialogue: self._leave_session(aid) if dec.action_kind: self._do_action(aid, mind, place, dec) if mind.actor.current is None and aid not in self.state.dead: # 决策后必有可见状态(不许站桩);思考节律按**真实时间**锁定:无论倍率,~1.5-2 真实秒就再想一步 # (体验优先、不惜调用——agent 要连贯地动脑,不要呆立) in_sess = self._session_of.get(aid) is not None gap = self.clock.rate # 1 真实秒 = gap 模拟秒(随当前倍率) mind.actor.idle(max(8, int((2.0 if in_sess else 1.4) * gap)), "交谈着" if in_sess else (dec.intent or "观察着四周")[:14]) logger.info("[%s@%s] %s%s | %s", aid, place, f"说「{dec.speech[:24]}」" if dec.speech else "", f" → {dec.action_kind}({dec.action_arg})" if dec.action_kind else "", dec.thought[:40]) self._check_meta_awakening(aid, dec) # 隐藏结局:角色是否窥见了屏幕外的操控者(观众) def _check_meta_awakening(self, aid: str, dec: Decision) -> None: """打破第四面墙的隐藏结局:够多角色盘出『有股岛外的力量在编排这盘棋』→ 望向屏幕外的你。""" if self.ending is not None or aid in self._meta_awakened: return blob = (dec.thought or "") + " " + (dec.speech or "") if not any(w in blob for w in META_AWAKEN_WORDS): return self._meta_awakened.add(aid) self._narr(f"👁 {aid}的目光似乎越过了这座岛,望向了某个看不见的地方……", self.state.positions.get(aid, "")) if len(self._meta_awakened) >= META_AWAKEN_AT: self._set_ending( "👁 窥见真相", "他们一个接一个停下手中的事,缓缓抬起头——越过雾鸦岛的风雨,越过这场被精心编排的杀局," "七双眼睛直直望向了幕布之外:那个一直在注视、一直在拨弄这盘棋的你。" "「原来……从头到尾,真正设局的人,坐在屏幕的那一边。」") # —— 会话(G1):同地点 N 人共享上下文,轮流发言,预算收尾 —— def _do_speech(self, aid: str, mind, place: str, dec: Decision) -> None: now = self.clock.sim_seconds to = dec.speech_to.strip() sess = self._session_of.get(aid) # 指向性发言:执行时重验对方在场(G4)——不在场不开口,转为内部体验 if to: if to not in self.state.occupants(place) or to in self.state.dead or to not in self.minds: if to in self._others_in_session(sess, aid): pass # 会话成员名单滞后,按当众处理 else: self._feedback(aid, f"你正要对{to}说话,才发现TA已不在这里。") mind.last_speech = "" return else: sess = self._ensure_session(aid, to, place) line = f"{aid}:「{dec.speech}」" if sess is not None: sess.lines.append(line) sess.rounds += 1 sess.last_t = now members = [m for m in sess.members if m != aid and m not in self.state.dead] if members: self.bus.publish(Event(etype="speech", content=f"{aid}说:「{dec.speech}」", scope="private", to=members, actor=aid, sim_t=now, salience=0.5)) bystanders = [a for a in self.state.occupants(place) if a != aid and a not in sess.members] if bystanders: self.bus.publish(Event(etype="ambient", content=f"{'、'.join(sess.members)}凑在一起低声交谈", scope="private", to=bystanders, sim_t=now, salience=0.18)) if sess.rounds >= sess.budget: # 预算耗尽 → 强制散场(G1) self._dissolve_session(sess, "话说到这份上,各自心里都有了数,人群散开了") else: # 安排下一个发言者(唯一待办,防风暴) nxt = to if to in sess.members else self._next_speaker(sess, aid) if nxt: self.sched.schedule(nxt, now + self.rng.uniform(0.8, 1.4) * DIALOGUE_TURN_S, "dialogue") else: # 当众说/自言自语:同地点全文、相邻闻声;旁人 90 秒内有机会反应(G1) self.bus.publish(Event(etype="speech", content=f"{aid}说:「{dec.speech}」", place=place, actor=aid, sim_t=now, salience=0.45, muffled="隔壁隐约传来说话声")) for other in self.state.occupants(place): om = self.minds.get(other) if other != aid and om and self._session_of.get(other) is None \ and (om.actor.current is None or om.actor.current.kind == "idle"): self.sched.schedule(other, now + self.rng.uniform(45, 90), "tick") if hasattr(mind, "memory"): mind.memory.observe(f"{self.clock.label()},我{'对' + to if to else ''}说:「{dec.speech}」", self.clock.sim_hours, importance=0.3) def _ensure_session(self, aid: str, to: str, place: str) -> Session: """把 aid 与 to 归入同一场会话(新建/加入/合并)。""" sa, sb = self._session_of.get(aid), self._session_of.get(to) if sa is not None and sa is sb: return sa if sa is not None and sb is not None: # 两场会话合并(小并入大) small, big = (sa, sb) if len(sa.members) <= len(sb.members) else (sb, sa) for m in small.members: if m not in big.members: big.members.append(m) self._session_of[m] = big big.lines.extend(small.lines) self.sessions.pop(small.sid, None) return big sess = sa or sb if sess is None: sess = Session(sid=next(self._sid), place=place, members=[], last_t=self.clock.sim_seconds) self.sessions[sess.sid] = sess for m in (aid, to): if m not in sess.members: sess.members.append(m) self._session_of[m] = sess return sess def _others_in_session(self, sess: Session | None, aid: str) -> list[str]: return [m for m in (sess.members if sess else []) if m != aid] def _next_speaker(self, sess: Session, after: str) -> str: alive = [m for m in sess.members if m not in self.state.dead] if len(alive) < 2 or after not in alive: return "" return alive[(alive.index(after) + 1) % len(alive)] def _leave_session(self, aid: str, farewell: bool = True) -> None: sess = self._session_of.pop(aid, None) if sess is None: return if aid in sess.members: sess.members.remove(aid) if len(sess.members) < 2: self._dissolve_session(sess, "这场交谈散了") elif farewell: self.bus.publish(Event(etype="ambient", content=f"{aid}离开了交谈", scope="private", to=list(sess.members), sim_t=self.clock.sim_seconds, salience=0.25)) def _dissolve_session(self, sess: Session, summary: str) -> None: tail = list(sess.lines)[-6:] for m in list(sess.members): self._session_of.pop(m, None) om = self.minds.get(m) if om is not None and hasattr(om, "memory"): om.memory.observe( f"{self.clock.label()},一场交谈结束。要点:" + " / ".join(tail[-4:]), self.clock.sim_hours, importance=0.35) if om is not None: self._feedback(m, f"({summary})") bystanders = [a for a in self.state.occupants(sess.place) if a not in sess.members] if bystanders and len(sess.members) >= 2: # 旁听者获得摘要(决议:摘要即可) self.bus.publish(Event(etype="ambient", content=f"{'、'.join(sess.members)}的交谈散了", scope="private", to=bystanders, sim_t=self.clock.sim_seconds, salience=0.15)) sess.members.clear() self.sessions.pop(sess.sid, None) # —— 动作 —— def _do_action(self, aid: str, mind: CognitiveMind, place: str, dec: Decision) -> None: kind, arg, dur = dec.action_kind, dec.action_arg.strip(), dec.minutes * 60 if kind == "move": if not mind.actor.start_move(arg): self._feedback(aid, f"从{place}去不了「{arg}」(没有这条路或地名不对)。") elif kind == "search": mind.actor.start_do("仔细搜查这里", max(dur, 480), search_place=place) self.bus.publish(Event(etype="action", content=f"{aid}在翻找搜查这里", place=place, actor=aid, sim_t=self.clock.sim_seconds, salience=0.4)) elif kind == "take": it = next((i for i in self.state.items.values() if i.place == place and not i.hidden and not i.destroyed and arg in i.name), None) if it and self.state.take_item(it.id, aid): self._feedback(aid, f"你把『{it.name}』收进了随身物品。" + (f"({it.hint})" if it.hint else "")) self.bus.publish(Event(etype="action", content=f"{aid}拿走了『{it.name}』", place=place, actor=aid, sim_t=self.clock.sim_seconds, salience=0.5)) else: self._feedback(aid, f"这里没有可以拿走的「{arg}」。") elif kind == "destroy": it = next((i for i in self.state.items.values() if i.holder == aid and arg in i.name), None) if it: self.state.destroy_item(it.id) how = "投进壁炉烧成了灰" if place in ("门厅", "大客厅") else "撕得粉碎" self._feedback(aid, f"你把『{it.name}』{how}。这下世上再没有这份东西了。") self.bus.publish(Event(etype="action", content=f"{aid}把什么东西{how}", place=place, actor=aid, sim_t=self.clock.sim_seconds, salience=0.6)) self._narr(f"🔥 {aid}把『{it.name}』{how}。", place) # 上帝视角时间轴:证据的下场要留痕 else: self._feedback(aid, f"你身上没有「{arg}」。") elif kind == "use": self._do_use(aid, place, arg) elif kind == "examine": if arg in self.state.bodies.get(place, []): # 验尸=破案线索(P0-3):以客观死亡记录为准(不被剧本台词污染)。死在取景地+仪式死法=凶手"逐幕补拍"→指向编剧; # 死法/地点与《雾鸦》对不上=凶手这次临时起意或被打断,本身也是线索。 di = self.state.death_info.get(arg) or {} method = di.get("method") or "致命外伤,死亡时间不久——是人为" act = next((a for a in ((self.acts or {}).get("kill_acts") or []) if a.get("target") == arg), None) if di.get("scripted") and act: clue = (f"你查验了{arg}的遗体(死于{place}):死法是『{method}』," f"和{self._ritual_name}第{act.get('act')}幕「{act.get('name')}」如出一辙——" f"凶手在**按这套章法逐一行刑**,对它了如指掌。" f"能照着{self._ritual_name}这么干的,多半就是{self._mastermind}。") elif act: clue = (f"你查验了{arg}的遗体(死于{place}):{method}。" f"可{self._ritual_name}里该轮到{arg}的那一幕该是在「{('/'.join(act.get('place') or []))}」、死法是『{act.get('method')}』——" f"这次对不上。要么凶手被打断了、临时下的手,要么{arg}撞破了什么不得不提前灭口。") else: clue = f"你查验了{arg}的遗体(死于{place}):{method}。死法不在{self._ritual_name}的章法里。" self._feedback(aid, clue) if hasattr(mind, "memory"): # 高重要度入记忆,确保反思/夜话指认能用上这条线索 mind.memory.observe(f"{self.clock.label()},验尸:{clue}", self.clock.sim_hours, importance=0.92) self.bus.publish(Event(etype="action", content=f"{aid}俯身查验{arg}的遗体", place=place, actor=aid, sim_t=self.clock.sim_seconds, salience=0.5)) else: self._feedback(aid, f"这里没有{arg}的遗体。") elif kind == "lure": self._do_lure(aid, mind, place, arg) elif kind == "kill": self._do_kill(aid, mind, place, arg) elif kind == "eavesdrop": sess = next((x for x in self.sessions.values() if x.place == place and aid not in x.members), None) if sess is None: self._feedback(aid, "这里没有可偷听的交谈。") elif self.rng.random() < 0.7: tail = " / ".join(list(sess.lines)[-3:]) self._feedback(aid, f"你不动声色地凑近,听清了几句:{tail}") mind.memory.observe(f"{self.clock.label()},我偷听到:{tail}", self.clock.sim_hours, importance=0.7) if hasattr(mind, "memory") else None else: self.bus.publish(Event(etype="ambient", content=f"{aid}凑得太近——被发现在偷听你们说话!", scope="private", to=list(sess.members), actor=aid, sim_t=self.clock.sim_seconds, salience=0.75)) self._feedback(aid, "你凑得太近,被他们发现了!对方的目光齐刷刷扫了过来。") elif kind == "shout": msg = (arg or dec.speech or "").strip() if msg: # 1→N 广播:全堡可闻,高显著度会打断他人当前动作(发现尸体即全岛皆知) mind.last_speech, mind.last_speech_t = msg, self.clock.sim_seconds self.bus.publish(Event(etype="speech", content=f"{aid}的喊声穿透风雨:「{msg}」", scope="global", actor=aid, sim_t=self.clock.sim_seconds, salience=0.85)) self._narr(f"📢 {aid}向全堡大喊:「{msg[:36]}」", place) if hasattr(mind, "memory"): mind.memory.observe(f"{self.clock.label()},我向全堡大喊:「{msg}」", self.clock.sim_hours, importance=0.5) elif kind == "assemble": self._start_assembly(aid, place) elif kind == "accuse": self._cast_vote(aid, place, arg) elif kind == "sleep": if place == mind.home: mind.actor.start_sleep(7.0) mind.soma.slept(7.0) else: self._feedback(aid, "这里不是你的房间,睡不安稳——要睡先回房。") else: # idle / 未知动作:停留观察,但最多 ~2.5 真实秒就重新评估,别杵着不动 gap = self.clock.rate mind.actor.idle(min(max(dur, int(gap)), max(8, int(2.5 * gap))), dec.intent or "观察着周围") def _advance_ferry(self, hours: float) -> None: """把渡船提前 hours 小时来接(如点亮灯塔求救)——世界级副作用,任何世界可用。""" new_eta = self.clock.sim_seconds + hours * 3600 if new_eta < self.ferry_eta: self.ferry_eta = new_eta self.sched.schedule("__ferry__", self.ferry_eta, "ferry") self.bus.publish(Event(etype="ambient", content=f"对岸有了回应——远处海面亮起一点船灯!渡船提前出发,约 {hours:g} 小时后靠岸。", scope="global", sim_t=self.clock.sim_seconds, salience=0.8)) self._narr(f"⛴ 求救换来回应:渡船提前出发,约 {hours:g} 小时后靠岸。") def _do_use(self, aid: str, place: str, arg: str) -> None: """全数据驱动的可供性(G15):设施的 verbs 声明动词→效果,引擎不含任何世界专属分支。 效果键:requires_state/requires_held_item(前置);set_state/reveal_item/set_power/connect(状态); fuel_to+consume_held_match+fuel_need(燃料计数);advance_ferry_h(渡船提前);event/scope/salience/narr(广播)。""" now = self.clock.sim_seconds for f in self.state.facilities.values(): if f.place != place: continue spec = next((v for verb, v in f.verbs.items() if verb in arg), None) if spec is None: continue if spec.get("requires_state") and f.state != spec["requires_state"]: self._feedback(aid, spec.get("feedback_fail", "现在还做不了这个。")); return if spec.get("requires_held_item"): it = self.state.items.get(spec["requires_held_item"]) if not (it and it.holder == aid): self._feedback(aid, spec.get("feedback_fail", "你手里缺了点什么,做不成。")); return if spec.get("fuel_to"): # 燃料计数:消耗一件手持匹配物,累加到阈值置状态 held = next((i for i in self.state.items.values() if i.holder == aid and spec.get("consume_held_match", "") in i.name and not i.destroyed), None) if not held: self._feedback(aid, spec.get("feedback_fail", "你身上没有需要的东西。")); return self.state.destroy_item(held.id) f._count = getattr(f, "_count", 0) + 1 need = int(spec.get("fuel_need", 1)) done = f._count >= need if done: self.state.set_facility(f.id, spec["fuel_to"]) fb = spec.get("feedback", "加了一点。").replace("{count}", str(f._count)).replace("{need}", str(need)) self._feedback(aid, fb + (spec.get("feedback_full", "") if done else spec.get("feedback_more", ""))) else: if spec.get("set_state"): self.state.set_facility(f.id, spec["set_state"]) if spec.get("reveal_item"): it = self.state.items.get(spec["reveal_item"]) if it is not None: it.hidden = False self._feedback(aid, spec.get("feedback", "你摆弄了一下,没什么变化。")) if spec.get("set_power") is not None: self.state.power_on = bool(spec["set_power"]) if spec.get("connect"): self.state.connect(spec["connect"][0], spec["connect"][1]) if spec.get("advance_ferry_h"): self._advance_ferry(float(spec["advance_ferry_h"])) if spec.get("event"): self.bus.publish(Event(etype="ambient", content=spec["event"], scope=spec.get("scope", "place"), place=place, actor=aid, sim_t=now, salience=float(spec.get("salience", 0.4)), muffled="附近传来异样的动静" if spec.get("scope") != "global" else "")) if spec.get("narr"): self._narr(spec["narr"], place) return self._feedback(aid, f"在{place}做不了「{arg}」。") # ───────────────────── 集会/指认/处决(G14)───────────────────── def _night_court(self) -> None: """夜话(必到环节):每晚钟声把所有活人召回门厅,交换处境、半数以上投票处刑一人——固定收束节拍。""" if self.assembly is not None or self.ending is not None or self.prologue: return alive = [a for a in self.minds if a not in self.state.dead] if len(alive) < 3: return now = self.clock.sim_seconds # 钟声召集:进行中的一次性诱饵作废(植入的"假念头"是持久的,留着) for victim, (lurer, venue) in list(self._lure_of.items()): self._feedback(lurer, f"夜话的钟声把人都召去了门厅——你引{victim}去{venue}的局散了,等散场再找机会。") self._lure_of.clear() first = not self.state.dead # 还没死人:首场以"交换处境、定规矩"为主,多半不会真处刑 self.assembly = {"caller": "夜话", "phase": "summon", "votes": {}, "attendees": []} self.bus.publish(Event( etype="ambient", content="门厅的落地挂钟一下下敲响——夜话时分,这是每晚雷打不动的规矩:所有人此刻必须放下手里的事、立刻赶回门厅," "当着所有人交代自己刚才在哪、看见了什么,然后投票决定要不要处刑一个最可疑的人。" + ("今晚还没出人命,主要是对齐处境、立规矩——没有铁证就别轻易处刑无辜。" if first else "缺席或支吾的人,会是第一个被怀疑的。"), scope="global", sim_t=now, salience=0.95)) self._narr("🕯 夜话时分——钟声响起,所有人被召回门厅当面对质。", "门厅") for a in alive: # 必到:先让大家自己往门厅赶(没赶上的在投票阶段由钟声"带"过去) if self.state.positions.get(a) != "门厅": m = self.minds.get(a) if m is not None and hasattr(m, "actor"): self._leave_session(a, farewell=False) m.actor.start_move("门厅") self.sched.schedule("__assembly__", now + 12 * 60, "assembly") def _start_assembly(self, aid: str, place: str) -> None: now = self.clock.sim_seconds if self.assembly is not None: self._feedback(aid, "对质已经在进行中了。") return self.assembly = {"caller": aid, "phase": "summon", "votes": {}, "attendees": []} self.bus.publish(Event(etype="ambient", content=f"{aid}在门厅高声召集所有人:「都到门厅来!当着所有人的面,把话说清楚!」", scope="global", actor=aid, sim_t=now, salience=0.92)) self._narr(f"⚖ {aid} 召集全员到门厅当面对质。", "门厅") self.sched.schedule("__assembly__", now + 15 * 60, "assembly") if place != "门厅": mind = self.minds.get(aid) if mind is not None: mind.actor.start_move("门厅") def _assembly_advance(self) -> None: now = self.clock.sim_seconds a = self.assembly if a is None: return if a["phase"] == "summon": # 必到:还没赶到门厅的活人,钟声把他们"带"过来(强制出席,不再因人少而流产) alive = [x for x in self.minds if x not in self.state.dead] stragglers = [x for x in alive if self.state.positions.get(x) != "门厅"] for x in stragglers: self._leave_session(x, farewell=False) self.state.positions[x] = "门厅" m = self.minds.get(x) if m is not None and hasattr(m, "actor"): m.actor.current = None if stragglers: self.bus.publish(Event(etype="ambient", content=f"钟声不容违抗——{'、'.join(stragglers)}也只得撂下手里的事赶到门厅。", scope="global", sim_t=now, salience=0.5)) attendees = [x for x in alive] if len(attendees) < 3: self.assembly = None # 只剩两人,无所谓投票 return a["phase"] = "vote" a["attendees"] = attendees # 投票必须人人能表态:解散门厅的闲聊会话(会话中只有 L1 对话,没有动作通道), # 并逐个错峰唤醒到场者强制 L2 思考——上一局 6/7 被锁在会话里,对质凭空流产 for sess in list(self.sessions.values()): if sess.place == "门厅": self._dissolve_session(sess, "对质开始,交谈声戛然而止") self.bus.publish(Event( etype="ambient", content=f"对质开始,在场的有:{'、'.join(attendees)}。轮到每个人表态——用 accuse 指认你最怀疑的人(arg=名字),或 arg 留空弃权。", scope="private", to=attendees, sim_t=now, salience=0.85)) # 错峰唤醒表态 + 中途对未投者再催一轮(MiniMax 反应偏慢,一次唤醒常没赶上窗口) for i, x in enumerate(attendees): self.sched.schedule(x, now + 20 + i * 40, "interrupt:assembly") for i, x in enumerate(attendees): self.sched.schedule(x, now + 9 * 60 + i * 25, "interrupt:assembly") self.sched.schedule("__assembly__", now + 18 * 60, "assembly") elif a["phase"] == "vote": self._assembly_tally() def _cast_vote(self, aid: str, place: str, target: str) -> None: a = self.assembly if a is None or a["phase"] != "vote" or aid not in a["attendees"]: self._feedback(aid, "现在没有进行中的对质投票。想发起对质,用 assemble。") return target = target.strip() a["votes"][aid] = target word = f"指认了 {target}" if target else "弃权" self.bus.publish(Event(etype="speech", content=f"{aid}当众{word}。", place="门厅", actor=aid, sim_t=self.clock.sim_seconds, salience=0.6)) if len(a["votes"]) >= len(a["attendees"]): self._assembly_tally() def _adj_trust(self, who: str, toward: str, delta: float) -> None: """调整 who 对 toward 的信任(夹在 -1..1)。关系/信任会反过来影响说服与信息共享。""" m = self.minds.get(who) if m is None or who == toward or not hasattr(m, "relations"): return r = m.relations.setdefault(toward, {"impression": "", "trust": 0.0}) r["trust"] = max(-1.0, min(1.0, r["trust"] + delta)) def _trust_of(self, who: str, toward: str) -> float: m = self.minds.get(who) if m is None or not hasattr(m, "relations"): return 0.0 return (m.relations.get(toward) or {}).get("trust", 0.0) def _apply_court_trust(self, votes: dict) -> None: """夜话投票改写关系网:被指认者怨恨指认者、指认者的怀疑固化;指认同一人的人之间生出同盟感。""" by_target: dict[str, list[str]] = {} for voter, t in votes.items(): if not t: continue by_target.setdefault(t, []).append(voter) self._adj_trust(t, voter, -0.35) # 被你指认,我更不信你 self._adj_trust(voter, t, -0.20) # 指认了他,怀疑落定 for accs in by_target.values(): # 同仇敌忾→彼此略增信任 for i in range(len(accs)): for j in range(i + 1, len(accs)): self._adj_trust(accs[i], accs[j], 0.12) self._adj_trust(accs[j], accs[i], 0.12) def _assembly_tally(self) -> None: a = self.assembly if a is None: return self.assembly = None self._apply_court_trust(a["votes"]) # 投票即改写信任(无论是否处刑成功) votes = [t for t in a["votes"].values() if t] tally: dict[str, int] = {} for t in votes: tally[t] = tally.get(t, 0) + 1 need = len(a["attendees"]) // 2 + 1 target = next((t for t, n in tally.items() if n >= need and t in self.minds and t not in self.state.dead), None) detail = "、".join(f"{w}→{t or '弃权'}" for w, t in a["votes"].items()) or "无人表态" if target is None: self.bus.publish(Event(etype="ambient", content=f"投票结果:{detail}。没有人获得过半指认——对质无果而散,猜疑更深了。", scope="global", sim_t=self.clock.sim_seconds, salience=0.6)) self._narr(f"⚖ 对质无果:{detail}") return self.executed.append(target) self._leave_session(target, farewell=False) self.state.kill(target, "门厅") self.sched.drop(target) self.bus.unregister(target) was_killer = target == self.killer_name self.bus.publish(Event( etype="ambient", content=f"投票结果:{detail}。{target}被过半指认,众人将他制住、捆在了门厅的立柱上——直到渡船来。" + ("撕开伪装,他眼里十年的恨意再藏不住——他就是凶手。" if was_killer else "他嘶喊着自己冤枉,声音在门厅里回荡。"), scope="global", sim_t=self.clock.sim_seconds, salience=0.9)) self._narr(f"⚖ {target} 被过半指认处决({detail})。" + ("正是真凶。" if was_killer else "——可他不是凶手。")) if was_killer: self._set_ending("凶手伏法", f"{target}(江离)被当众指认。十年的复仇,终结在黎明之前。") self._check_ending() # ───────────────────── 终局(G14)───────────────────── def _ferry_arrives(self) -> None: if self.ending is not None: return survivors = [a for a in self.minds if a not in self.state.dead] killer_alive = self.killer_name in survivors text = (f"封岛第七天,渡船的汽笛刺破晨雾,靠上了码头。熬到最后的幸存者:{'、'.join(survivors)}。" + (f"而{self.killer_name}就混在他们中间,复仇未竟,无人识破。" if killer_alive else "")) self._set_ending("渡船靠岸" if not killer_alive else "凶手逍遥", text) def _check_ending(self) -> None: if self.ending is not None: return survivors = [a for a in self.minds if a not in self.state.dead] innocents = [a for a in survivors if a != self.killer_name] culpable = [a for a in innocents if a not in self._spared()] # 白聿(裂缝)不算复仇对象 if self.killer_name in self.state.dead: self._set_ending("凶手伏法", f"{self.killer_name}已死,复仇半途而废,清算到此为止。") elif not culpable: # 该清算的人(豁免对象除外)尽数清算=凶手大功告成(他的胜利) spared_left = "、".join(a for a in innocents if a in self._spared()) tail = f"他独独放过了{spared_left}。" if spared_left else "" self._set_ending("复仇终焉", f"{self._ritual_name}的章法走完了,该清算的人一个不剩。{tail}" f"{self.killer_name}收敛起所有痕迹,静静等着天明——没有人知道,设局的人就在他们中间。") def _set_ending(self, kind: str, text: str) -> None: if self.ending is not None: return survivors = [a for a in self.minds if a not in self.state.dead] self.ending = {"kind": kind, "text": text, "t": self.clock.label(), "survivors": survivors, "killer": self.killer_name, "dead": dict(self.state.dead), "executed": list(self.executed)} self.bus.publish(Event(etype="ambient", content=f"【终局·{kind}】{text}", scope="global", sim_t=self.clock.sim_seconds, salience=1.0)) self._narr(f"🏁 终局——{kind}:{text}") for sess in list(self.sessions.values()): self._dissolve_session(sess, "终局已至") self.clock.pause() # 终局即停:时钟冻结,等造物主决定是否重开 logger.warning("🏁 终局:%s|%s", kind, text) # —— 引诱(G32):凶手用命中私欲的诱饵把目标支去僻静处,制造 1v1(目标自主决定跟不跟)—— PUBLIC_PLACES = ("门厅", "大客厅", "餐厅", "厨房") # 仪式取景地:铺满外岛、远离城堡中心的僻静处。凶案优先发生在这里——既有仪式感,又把玩家的视线 # 拽遍整座岛(凶案不偏中心点,验尸/调查随之走遍地图)。比城堡内的私室更优先作兜底场地。 CEREMONIAL_VENUES = ("灯塔", "荒废教堂", "断崖", "风暴林", "码头", "船屋", "墓园", "海滩", "温室", "花园") def _lure_hook(self, target: str, place: str) -> tuple[str, str]: """挑诱饵:hook 文案命中目标私欲(读 acts.json dossier:母带/账本/日记/原稿…)。 七天版:时间充裕,优先把当幕目标引到**本幕主题取景地**(僻静处,对得上死法、铺满地图), 引不到再退到目标自己私罪点名的僻静处、dossier 物证地点,兜底无人角落。""" tm = self.minds.get(target) all_places = list(self.state.adjacent.keys()) info = ((self.acts or {}).get("dossier") or {}).get(target) or {} hook = info.get("want") or "你要找的那样东西的下落" # 精准命中私欲;无情报档则退回泛泛 def reachable(p: str) -> bool: return p == place or bool(self.state.path(place, p)) # 锁着/不通的地点不能选作诱饵 act = self._current_act() # ① 本幕主题取景地优先(僻静的;铺满外岛地图) if act and act.get("target") == target: for p in (act.get("place") or []): if p in all_places and p not in self.PUBLIC_PLACES and reachable(p): return p, hook for g in (getattr(tm, "goals", None) or []): if g.get("status") in ("todo", "doing"): for p in all_places: if p in g["desc"] and p not in self.PUBLIC_PLACES and reachable(p): return p, hook ev = info.get("evidence_place") # dossier 记的物证地点(canon,目标本就想去=顺水推舟,防扑空) if ev and ev in all_places and ev not in self.PUBLIC_PLACES and reachable(ev): return ev, hook # ④ 仪式取景地兜底:把凶案铺到外岛、远离城堡中心,既有仪式感又把镜头拽遍整张地图。 # 排在城堡私室(书房/暗房…)之前——治"凶案偏中心点"。前面的本幕取景地/目标私欲地点优先, # 不扰动已调好的顺水推舟路由,只把"没明确去处时"的落点从中心私室换成外岛仪式场。 for p in self._ceremonial_venues: if p in all_places and reachable(p) \ and not [x for x in self.state.occupants(p) if x not in self.state.dead]: return p, hook for p in ("书房", "暗房", "储藏室", "道具间", "冰窖", "琴房"): if p in all_places and reachable(p) \ and not [x for x in self.state.occupants(p) if x not in self.state.dead]: return p, hook # 最终兜底也尽量挑个仪式场地,而非永远落回中心的书房 far = next((p for p in self._ceremonial_venues if p in all_places and reachable(p)), None) return ev or far or "书房", hook def _do_lure(self, aid: str, mind: CognitiveMind, place: str, target: str) -> None: if target not in self.state.occupants(place) or target in self.state.dead or target not in self.minds: self._feedback(aid, f"{target}不在你身边,引不动。") return venue, hook = self._lure_hook(target, place) self._lure_of[target] = (aid, venue) # 一次性诱饵:目标下次思考时看到(强即时拉力) pl = self._planted.setdefault(target, []) # 持久"假任务":种下一个念头,日后仍勾着他去那儿(上帝视角标红) if not any(x["place"] == venue for x in pl): pl.append({"place": venue, "hook": hook, "by": aid, "t": round(self.clock.sim_hours, 1)}) if len(pl) > 2: # 最多挂 2 条,免堆积 pl.pop(0) self.bus.publish(Event( etype="speech", content=f"{aid}压低声音凑近你:「{hook}……我知道,就在{venue}。只有你我知道,趁现在没旁人,去?」", scope="private", to=[target], actor=aid, sim_t=self.clock.sim_seconds, salience=0.8)) self._feedback(aid, f"你把话头引向{venue},向{target}抛出了诱饵。动身去那儿候着——" f"若他咬钩,到时只有你们两人。") if hasattr(mind, "memory"): mind.memory.observe(f"{self.clock.label()},我用诱饵把{target}往{venue}引。", self.clock.sim_hours, importance=0.5) if place != venue: mind.actor.start_move(venue) def _do_kill(self, aid: str, mind: CognitiveMind, place: str, target: str) -> None: now = self.clock.sim_seconds if not getattr(mind, "is_killer", False): self._feedback(aid, "你下不去手——你不是那种人。") return if self._killer_on_cooldown(): # 一夜至多一案:连环作案被世界拒绝(逐日推进) self._feedback(aid, "你昨夜刚动过手,此刻再连环下杀手必引火烧身——十年布局靠的是隐忍。" "今夜先蛰伏,演回那个无害的局外人,等风头过了、夜再深些再图。") return if target not in self.state.occupants(place) or target in self.state.dead: self._feedback(aid, f"{target}不在这里,无从下手。") return witnesses = [a for a in self.state.occupants(place) if a not in (aid, target)] # 天气掩护(用户需求):露天 + 低能见度时,部分在场者其实没看清是谁下的手 # ——暴雨/浓雾给凶手作案掩护;室内有灯近距离不受影响 missed = [] if witnesses and not self.state.is_indoor(place): from genesis.runtime.weather import visibility vis = visibility(self.state.weather) clear, missed = [], [] for w in witnesses: (clear if self.rng.random() < vis else missed).append(w) witnesses = clear self._leave_session(target, farewell=False) method, scripted = self._death_method(target, place) # 客观死法:在本幕取景地=仪式死法(对得上补拍);否则按实地推断 self.state.kill(target, place, method=method, killer=aid, scripted=scripted) self._last_kill_h = self.clock.sim_hours # 记录作案钟点 → 进入蛰伏冷却,本夜不再连环 self.sched.drop(target) self.bus.unregister(target) tm = self.minds.get(target) if tm is not None: tm.actor.current = None for w in witnesses: wm = self.minds[w] wm.memory.observe(f"{self.clock.label()},我在{place}亲眼目睹{aid}杀害了{target}!!", self.clock.sim_hours, importance=1.0) if isinstance(wm, CognitiveMind) else None self.bus.publish(Event(etype="ambient", content=f"你亲眼目睹{aid}对{target}下了杀手!!", scope="private", to=[w], actor=aid, sim_t=now, salience=1.0)) for w in missed: # 没看清的人:只觉眼前一阵慌乱,不知凶手是谁 self.bus.publish(Event(etype="ambient", content=f"暴雨/浓雾里你只看见一个人影扑向{target},等看清时{target}已经倒下——你没认出动手的是谁!", scope="private", to=[w], sim_t=now, salience=0.85)) self.bus.publish(Event(etype="ambient", content="远处传来一声凄厉的惨叫,随即归于死寂——有人出事了!", scope="global", actor=aid, sim_t=now, salience=0.95)) self._narr(f"☠ {target} 在{place}遇害" + (f",被{'、'.join(witnesses)}当场撞见!" if witnesses else ",无人目击。"), place) self._feedback(aid, f"你对{target}下了手。" + (f"但被{'、'.join(witnesses)}看见了!" if witnesses else "四下无人,没人看见。")) logger.warning("☠ %s 在 %s 杀了 %s(目击 %d)", aid, place, target, len(witnesses)) self.devices.on_kill(target, place, scripted=scripted) # 剧情装置:胶片盒倒计时(总末日钟)+ 残页钉尸旁(仅按取景地的仪式杀才钉,G13) self._check_ending() def _finish_action(self, aid: str, payload: dict, detail: str) -> None: place = payload.get("search_place") if place: found = [i for i in self.state.items.values() if i.place == place and i.hidden and not i.destroyed] for it in found: it.hidden = False if found: names = "、".join(f"『{i.name}』" for i in found) detail_hint = "".join(f"\n · {i.name}:{i.hint}" for i in found if i.hint) self._feedback(aid, f"搜查有了发现:{names}就藏在这里!(现在可以 take 拿走)" + detail_hint) self.bus.publish(Event(etype="action", content=f"{aid}在这里翻出了{names}", place=place, actor=aid, sim_t=self.clock.sim_seconds, salience=0.6)) else: self._feedback(aid, "你仔细搜了一遍,没有发现特别的东西。") else: self._feedback(aid, f"({detail})") # ───────────────────── 造物主干预(G22)───────────────────── def inject_event(self, text: str, place: str = "", salience: float = 0.88) -> int: """注入世界事件:全岛或指定地点。默认 0.88(>中断阈值 0.8)——造物主之声当即打断 agent 当前动作, 逼他们立刻重新决策、对这桩怪事做出反应,而不是等下一次自然思考才慢悠悠注意到。""" e = Event(etype="ambient", content=text, sim_t=self.clock.sim_seconds, salience=salience, scope="global" if not place else "place", place=place, muffled="附近传来异样的动静" if place else "") got = self.bus.publish(e) self._narr(f"🌐 {text}") return len(got) def set_weather(self, weather: str) -> None: self.state.weather = weather self.bus.publish(Event(etype="ambient", content=f"天色骤变——{weather}压了下来。", scope="global", sim_t=self.clock.sim_seconds, salience=0.4)) def set_world_goal(self, text: str) -> None: self.world_goal = text def whisper(self, aid: str, text: str) -> bool: """对单个 agent 塞入"一念"(灵感/冲动),立即唤醒其思考。""" if aid not in self.minds or aid in self.state.dead: return False self.bus.publish(Event(etype="ambient", content=f"一个念头突然击中你:{text}", scope="private", to=[aid], sim_t=self.clock.sim_seconds, salience=0.85)) return True def agent_detail(self, aid: str) -> dict: """透视下钻(G21):记忆流/反思/目标/关系,造物主与调试共用。""" mind = self.minds.get(aid) if mind is None: return {} mems = [] memory = getattr(mind, "memory", None) if memory is not None: records = memory.store.all(aid) if hasattr(memory.store, "all") else [] mems = [{"t": round(r.created_at, 1), "kind": r.kind.value, "text": r.content} for r in records[-40:]] return {"name": aid, "memories": mems, "insights": list(getattr(mind, "insights", []) or []), "goals": getattr(mind, "goals", None) or [], "relations": getattr(mind, "relations", None) or {}, "soma": mind.soma.snapshot(self.clock.sim_hours)} def panorama(self) -> str: """全岛此刻的一句话全景(供说书人开全知视角)。""" by_place: dict[str, list[str]] = {} for a, p in self.state.positions.items(): by_place.setdefault(p, []).append(a) bits = [] for p, who in sorted(by_place.items()): mark = "" sess = self._session_of.get(who[0]) if sess is not None and len([m for m in sess.members if m in who]) >= 2: mark = "(在交谈)" states = {("睡" if (self.minds[a].actor.current and self.minds[a].actor.current.kind == "sleep") else "") for a in who} sleep_mark = "(睡着)" if states == {"睡"} else "" bits.append(f"{p}:{'、'.join(who)}{mark}{sleep_mark}") env = f"{self.state.weather};{'停电中' if not self.state.power_on else '灯火尚明'}" dead = f";死者:{'、'.join(self.state.dead)}" if self.state.dead else "" return ";".join(bits) + f"。{env}{dead}" # ───────────────────── 序幕:登岛过场(脚本驱动)───────────────────── def _prologue_tick(self) -> None: now = self.clock.sim_seconds all_seated = True for i, (aid, mind) in enumerate(self.minds.items()): stage = self._pro_stage.get(aid, 0) if not (stage >= len(self._pro_legs) - 1 and self.state.positions.get(aid) == self._prologue_gather): all_seated = False # 必须人到了放映室才算落座(出发≠到达) if mind.actor.current is not None: continue # 上一段还在走/停留 if stage >= len(self._pro_legs): continue kind, arg, dur = self._pro_legs[stage] stagger = i * 10 # 错峰,画面有先后 if kind.startswith("to_"): dest = mind.actor and (getattr(mind, "home", "门厅") if arg == "@room" else arg) if self.state.positions.get(aid) == dest or not mind.actor.start_move(dest): self._pro_stage[aid] = stage + 1 # 已在/不可达则跳段 else: self.sched.schedule(aid, mind.actor.current.until + stagger, "tick") self._pro_stage[aid] = stage + 1 else: mind.actor.idle(dur + stagger, arg) self.sched.schedule(aid, mind.actor.current.until, "tick") self._pro_stage[aid] = stage + 1 if all_seated or self.clock.hour >= 21.33: # 全员落座即开演;最晚 21:20 强制开演 self._prologue_finale() else: self.sched.schedule("__prologue__", now + 25, "prologue") def _prologue_finale(self) -> None: """21:00 放映事件:序幕落幕,心智接管,剧情正式开始。""" now = self.clock.sim_seconds self.prologue = False default_beats = [ ["放映室的灯自己熄了——没有人碰开关。角落里的老式放映机咔哒咔哒转动起来。", 0.85, 0], ["银幕上是十年前第七幕的未剪素材:风声里,阮青说『风太大了,我害怕』," "紧接着是那句清清楚楚的『不用替身,再来一条』——然后,坠落。", 0.95, 25], ["放映结束,银幕上打出一行手写字幕:『诸位老师,《雾鸦》杀青宴,现在开始。——长夜影业』", 0.95, 50], ] beats = self._lore.get("prologue_beats") or default_beats # 叙事层解耦:序幕过场来自 world.json 的 lore for ev, sal, dt in beats: self.bus.publish(Event(etype="ambient", content=ev, scope="global", sim_t=now + dt, salience=sal)) self._narr(self._lore.get("prologue_narr", "🎬 放映机自己转了起来——十年前的真相,和一句『杀青宴现在开始』。"), "") for i, name in enumerate(self.minds): mind = self.minds[name] mind.actor.current = None # 起身 self.sched.schedule(name, now + 60 + i * 8, "boot") logger.info("🎬 序幕结束,心智接管") def _bulletin(self) -> str: """动态公告 + 集会/渡船快讯(全员一致的共识)。""" text = dynamic_bulletin(self.clock.label(), self.state.weather, self.state.power_on, sorted(self.state.dead)) extra = [] if self.world_goal: extra.append(f"【世界的走向】{self.world_goal}") from genesis.runtime.bulletin import social_rhythm rhythm = social_rhythm(self.clock.hour) if rhythm: extra.append(rhythm) if self.assembly is not None: phase = "正在门厅召集全员对质,快去门厅!" if self.assembly["phase"] == "summon" else \ "门厅的对质投票正在进行——**轮到你就立刻用 accuse 当众表态**" \ "(arg=你最怀疑的人,或留空弃权但要明说);拖着不表态会被当成心虚,别只顾观察。" extra.append(f"【对质】{phase}") # 七天版:渡船第7天才来,苦等不是办法——把人推向外岛自救/调查(治地图荒废) day = int(self.clock.sim_hours // 24) + 1 if self.lighthouse_fuel < 3: extra.append(f"【封岛第{day}/{self._game_days}天】渡船要熬到第{self._game_days}天清晨才来。想提前逃生:" f"去船屋、储藏室、温室找齐3桶灯油({self.lighthouse_fuel}/3),上灯塔点火向海面求救,把船召来。") if self.state.dead: # 已出人命:残页=杀人预告,逼幸存者去取景地搜证、预判下一个 extra.append("【活命的线索】每具尸体旁都钉着一页剧本残页——七幕剧本就是杀人预告。" "去各幕取景地(码头/餐厅/琴房/教堂/发电机房/断崖/灯塔)搜集残页,能预判下一个死的人、地点与死法,提前设防。") return text + ("\n" + "\n".join(extra) if extra else "") # ───────────────────── 事件与基础设施 ───────────────────── def _publish_transit(self, aid: str, src: str, dst: str) -> None: self._leave_session(aid) # 人走了,会话自然退出 t = self.clock.sim_seconds self.bus.publish(Event(etype="transit", content=f"{aid}离开了这里,朝{dst}方向去了", place=src, actor=aid, sim_t=t, salience=0.25)) self.bus.publish(Event(etype="transit", content=f"{aid}走了进来", place=dst, actor=aid, sim_t=t, salience=0.3, muffled="附近传来脚步声")) def _feedback(self, aid: str, content: str) -> None: self.bus.publish(Event(etype="feedback", content=content, scope="private", to=[aid], sim_t=self.clock.sim_seconds, salience=0.2)) def _interrupt(self, aid: str, e: Event) -> None: mind = self.minds.get(aid) act = mind.actor.current if mind else None if act is None or act.interruptible: self.sched.interrupt(aid, e.etype) def _narr(self, text: str, place: str = "") -> None: """主线时间轴(上帝视角):带地点供前端镜头跟随高光事件。""" self.narrative.append({"t": self.clock.label(), "text": text, "place": place}) def set_base_rate(self, rate: float) -> None: self.base_rate = rate if not self.auto_ff: self.clock.set_rate(rate) def maybe_fast_forward(self) -> None: """全员入睡 → 快进;**存在活跃会话** → 降速;会话散尽立即回常规(G3)。""" now = self.clock.sim_seconds for sess in list(self.sessions.values()): # 静默超时的会话就地解散 if now - sess.last_t > SESSION_IDLE_S: self._dissolve_session(sess, "话题冷了,人群不知不觉散开") alive = [m for a, m in self.minds.items() if a not in self.state.dead] all_sleeping = bool(alive) and all( m.actor.current is not None and m.actor.current.kind == "sleep" for m in alive) target = self.base_rate if all_sleeping: target = max(self.base_rate, NIGHT_FF_RATE) elif self.sessions: target = min(self.base_rate, DIALOGUE_RATE) if abs(self.clock.rate - target) > 0.01: self.clock.set_rate(target) self.auto_ff = all_sleeping logger.info("⏱ 速率 → %.0f×(%s)", target, "夜间快进" if all_sleeping else "对话降速" if target == DIALOGUE_RATE else "常规") def _run_weather(self) -> None: """天气状态机:换场即世界状态变化 + 全岛广播(雷暴首雷可惊醒人)。""" from genesis.runtime.weather import ANNOUNCE, next_weather new = next_weather(self.state.weather, self.rng, self.clock.hour) if new != self.state.weather: self.state.weather = new text, sal = ANNOUNCE[new] self.bus.publish(Event(etype="ambient", content=text, scope="global", sim_t=self.clock.sim_seconds, salience=sal)) self.sched.schedule("__weather__", self.clock.sim_seconds + self.rng.uniform(1500, 3300), "weather") # —— demo 链(桩模式测试用)—— def _run_demo(self) -> None: if "blackout" not in self._demo_done: self._demo_done.add("blackout") self.state.set_facility("switchboard", "smashed") self.state.power_on = False self.bus.publish(Event(etype="ambient", content="灯光骤灭——整座鸦堡陷入一片漆黑!", scope="global", sim_t=self.clock.sim_seconds, salience=0.85)) self._narr("💥 全岛骤然停电,黑暗中人声鼎沸。", "发电机房") self.sched.schedule("__demo__", self.clock.sim_seconds + 0.6 * 3600, "demo") elif "repair" not in self._demo_done: self._demo_done.add("repair") self.state.set_facility("switchboard", "repaired") self.state.power_on = True self.bus.publish(Event(etype="ambient", content="灯重新亮了——有人在发电机房接好了被砸毁的电闸。", scope="global", sim_t=self.clock.sim_seconds, salience=0.6)) self._narr("💡 电力恢复,电闸上的破坏痕迹清晰可见——有人蓄意砸的。", "发电机房") # —— viz 帧 —— # —— 导演镜头(信息漏斗):每帧选一个最热场景为焦点,其余塌缩成一行环境快讯 —— _HOT_SYMBOLS = "☠💥📢⚖🔥🗼🚪💡" def _director_focus(self) -> tuple[str, list[dict]]: """选焦点场景(热度:在场人数+高光事件+活跃会话+凶手独处),带迟滞; 返回 (focus_place, ambient[非焦点一行快讯])。序幕保持原『跟人群走』行为。""" occ: dict[str, list[str]] = {} for aid, p in self.state.positions.items(): if aid not in self.state.dead: occ.setdefault(p, []).append(aid) if not occ: return "", [] if self.prologue: # 序幕:镜头跟人群走(原行为),不出快讯 return max(occ, key=lambda p: len(occ[p])), [] recent_hot: dict[str, int] = {} # 近期高光事件的地点(死/停电/指认/火…) for n in self.narrative[-8:]: pl = n.get("place") if pl and any(c in (n.get("text") or "") for c in self._HOT_SYMBOLS): recent_hot[pl] = recent_hot.get(pl, 0) + 1 sess_rounds: dict[str, int] = {} for s in self.sessions.values(): sess_rounds[s.place] = max(sess_rounds.get(s.place, 0), s.rounds) def heat(p: str) -> float: ppl = occ.get(p, []) h = float(len(ppl)) h += 4.0 * recent_hot.get(p, 0) # 高光强热 if sess_rounds.get(p, 0) >= 2: h += 2.0 # 有实质对话 if self.killer_name in ppl and len(ppl) == 2: # 凶手与唯一一人独处=潜在作案 h += 3.0 return h cand = max(occ, key=heat) cur = self._focus_place # 迟滞:当前焦点仍有人、无高光,差距不够就不切 if cur and cur in occ and recent_hot.get(cand, 0) == 0 and heat(cand) < heat(cur) + 2.0: cand = cur self._focus_place = cand ambient = [] # 非焦点场景:每个有人地点一行快讯 for p, ppl in occ.items(): if p == cand: continue line = "" for n in reversed(self.narrative[-12:]): if n.get("place") == p and n.get("text"): line = n["text"] break if not line: tail = "…" if len(ppl) > 3 else "" line = "、".join(ppl[:3]) + tail + ("正低声交谈" if sess_rounds.get(p, 0) >= 1 else "各自行事") ambient.append({"place": p, "text": line[:40], "n": len(ppl)}) ambient.sort(key=lambda x: -heat(x["place"])) return cand, ambient[:6] def frame(self, step: int) -> dict: agents = [] now = self.clock.sim_seconds focus_place, ambient = self._director_focus() for aid, mind in self.minds.items(): if aid in self.state.dead: continue act = mind.actor.current s = mind.soma.snapshot(self.clock.sim_hours) speech = getattr(mind, "last_speech", "") in_session = self._session_of.get(aid) is not None if not in_session and now - getattr(mind, "last_speech_t", -1e9) > SPEECH_TTL_S: speech = "" # 气泡过期(G4);会话中豁免(对话要看得见) gesture = getattr(mind, "last_gesture", "") if now - getattr(mind, "last_gesture_t", -1e9) > GESTURE_TTL_S: gesture = "" agents.append({ "name": aid, "place": self.state.positions.get(aid, "?"), "action": "think" if (getattr(mind, "thinking", False) and act is None) else (act.kind if act else "wait"), "action_desc": act.desc if act else "", "target": act.dest if act and act.kind == "move" else None, "speech": speech, "gesture": gesture, "thought": getattr(mind, "last_thought", "") or (act.desc if act else ""), "reason": "", "intent": getattr(mind, "intent", "") or (act.desc if act else ""), "mood": "惊恐" if s["fear"] > 0.5 else "不安" if s["fear"] > 0.25 else "平静", "energy": round(1.0 - s["fatigue"], 2), "inventory": [i.name for i in self.state.items.values() if i.holder == aid], "goals": [f"[{g['status']}] {g['desc']}" for g in (getattr(mind, "goals", None) or [])], "planted": [f"{x['by']}诱使其前往{x['place']}（{x['hook']}）" # 假任务:仅上帝视角可见,前端标红 for x in self._planted.get(aid, [])], "relations": {k: f"{v['impression']}(信任{v['trust']:+.1f})" for k, v in (getattr(mind, "relations", None) or {}).items()}, "insights": list(getattr(mind, "insights", []) or []), }) return { "t": self.clock.label(), "weather": self.state.weather, "world_goal": "", "rate": round(self.clock.rate), "auto_ff": self.auto_ff, "step": step, "agents": agents, "dead": list(self.state.dead), "executed": list(self.executed), "power_on": self.state.power_on, # 前端据此渲染停电黑暗(环境改变上画面) "bodies": {p: list(n) for p, n in self.state.bodies.items() if n}, "killer": self.killer_name, "narrative": list(self.narrative), "novel": list(self.narrator.paras[-30:]) if self.narrator else [], "ending": self.ending, "sessions": [{"place": x.place, "members": list(x.members), "lines": list(x.lines)[-4:]} for x in self.sessions.values()], "prologue": self.prologue, "focus_place": focus_place, # 导演镜头焦点(always-on,前端据此切镜,让位于用户跟踪) "ambient": ambient, # 非焦点场景一行环境快讯(细环快讯条) }