"""会话引擎池:每个活跃会话一个独立 WorldEngine + 线程。 生命周期(用户选定:离开即暂停续播)—— - 用户在看(有心跳):引擎在跑,token 在烧; - 闲置超时(无心跳):存快照到 SQLite + 停引擎 + 移出内存,**不再烧 token**; - 用户回来:从快照 restore 引擎、重启线程,接着原进度演。 并发上限:活跃引擎数封顶,超了就把最久没心跳的会话挤下去(存档暂停)。 成本隔离:每会话一个 MeteredLLM(底层 LiteLLM 客户端无状态、可共享)。 """ from __future__ import annotations import asyncio import json import threading import time from pathlib import Path from genesis.obs.logging_setup import get_logger from genesis.runtime.engine import WorldEngine from genesis.runtime.persistence import restore_engine, snapshot_engine logger = get_logger("server.pool") class LiveSession: """一个用户的活世界:独立引擎线程 + 帧循环 + 叙事 + 自动存档。""" def __init__(self, sid: str, engine: WorldEngine, llm, mgr: "EngineManager", step: int = 0): self.sid, self.engine, self.llm, self.mgr = sid, engine, llm, mgr self.state = {"frame": engine.frame(step), "step": step, "running": True} self.lock = threading.Lock() self.last_beat = time.monotonic() self._restart = False self._stop = False self.thread = threading.Thread(target=self._run, daemon=True, name=f"sess-{sid[:6]}") self.thread.start() def _run(self) -> None: try: asyncio.run(self._loop()) except Exception: logger.exception("[%s] 会话引擎线程崩溃", self.sid[:6]) async def _loop(self) -> None: # 放大线程池:decide 是网络 IO(等 LLM,不占 CPU),让 7 角色 + 说书人并发思考不排队、思考连贯 import concurrent.futures asyncio.get_running_loop().set_default_executor( concurrent.futures.ThreadPoolExecutor(max_workers=10, thread_name_prefix=f"dec-{self.sid[:4]}")) while not self._stop: self.engine.start() ftask = asyncio.create_task(self._frames()) await self.engine.sched.run(self.engine.on_wake) ftask.cancel() if self._restart and not self._stop: self._restart = False self.engine = self.mgr.new_engine(self.llm, None) # 重开:全新世界 with self.lock: self.state["step"] = 0 self.state["frame"] = self.engine.frame(0) continue break async def _frames(self) -> None: last_save = 0.0 while not self._stop: await asyncio.sleep(1.0) if not self.state["running"]: if not self.engine.clock._paused: self.engine.clock.pause() continue if self.engine.clock._paused and self.engine.ending is None: self.engine.clock.resume() self.engine.maybe_fast_forward() nar = self.engine.narrator if nar is not None and nar.due(self.engine.clock.sim_seconds): nar.busy = True now_s, label, pano = self.engine.clock.sim_seconds, self.engine.clock.label(), self.engine.panorama() async def _write(): try: await asyncio.to_thread(nar.write, now_s, label, pano) finally: nar.busy = False asyncio.create_task(_write()) with self.lock: self.state["step"] += 1 self.state["frame"] = self.engine.frame(self.state["step"]) if time.monotonic() - last_save > 60: # 每分钟落一次盘(防进程崩溃丢进度) last_save = time.monotonic() self._persist() def _persist(self) -> None: try: snap = json.dumps(snapshot_engine(self.engine), ensure_ascii=False) self.mgr.store.save_snapshot(self.sid, self.state["step"], snap) except Exception: logger.exception("[%s] 存档失败", self.sid[:6]) def stop_and_save(self) -> None: self._persist() self._stop = True self.engine.sched.stop() class EngineManager: def __init__(self, world_dir, store, shared_llm_client, viz, stage, *, max_engines: int = 8, idle_timeout: float = 90.0, brain: str = "llm", rate: float = 30.0): self.world_dir = Path(world_dir) self.store = store self.shared_llm = shared_llm_client self.viz, self.stage = viz, stage self.max_engines, self.idle_timeout = max_engines, idle_timeout self.brain, self.rate = brain, rate self._live: dict[str, LiveSession] = {} self._lock = threading.RLock() threading.Thread(target=self._reaper, daemon=True, name="reaper").start() def _make_llm(self): if self.brain == "stub": # llm / langgraph 都用计量 LLM(后者亦用于兜底/反思 + 叙事/长期记忆) return None from genesis.runtime.metering import MeteredLLM return MeteredLLM(self.shared_llm) def new_engine(self, llm, snap_dict): eng = WorldEngine(self.world_dir, rate=self.rate, brain=self.brain, llm=llm, prologue=snap_dict is None) if llm is not None: from genesis.runtime.longterm import LongTerm from genesis.runtime.narrator import Narrator nlore = getattr(eng, "_lore", {}) or {} # 叙事层解耦:说书人世界设定来自 world.json 的 lore eng.narrator = Narrator(llm, roster=list(eng.minds.keys()), novel_title=nlore.get("novel_title", "《雾鸦》"), ghost=nlore.get("ghost", "阮青"), opening=nlore.get("narrator_opening")) eng.bus.taps.append(eng.narrator.tap) eng.longterm = LongTerm(eng.game_id) if snap_dict is not None: restore_engine(eng, snap_dict) return eng def ensure(self, sid: str) -> LiveSession: """确保会话引擎在跑:内存有则续,无则从快照恢复或新建。""" with self._lock: ls = self._live.get(sid) if ls is not None and not ls._stop: ls.last_beat = time.monotonic() return ls if len(self._live) >= self.max_engines: # 并发满 → 挤掉最久没心跳的 oldest = min(self._live.values(), key=lambda x: x.last_beat) logger.info("引擎池满,挤出闲置会话 %s", oldest.sid[:6]) self._evict(oldest.sid) step, snap = self.store.load_snapshot(sid) snap_dict = json.loads(snap) if snap else None llm = self._make_llm() eng = self.new_engine(llm, snap_dict) ls = LiveSession(sid, eng, llm, self, step=step or 0) self._live[sid] = ls logger.info("[%s] 会话引擎%s", sid[:6], "恢复续播" if snap_dict else "全新开局") return ls def _evict(self, sid: str) -> None: ls = self._live.pop(sid, None) if ls is not None: ls.stop_and_save() def suspend(self, sid: str) -> None: """主动暂停(关浏览器/切走时调用):立即存档停引擎,不再烧 token;回来要确认才恢复。""" with self._lock: self._evict(sid) def has_session(self, sid: str) -> bool: """会话存在(DB 有记录),不管引擎当前是否在内存跑。""" return self.store.session_exists(sid) def heartbeat(self, sid: str) -> None: ls = self._live.get(sid) if ls is not None: ls.last_beat = time.monotonic() self.store.touch(sid) def get(self, sid: str) -> LiveSession | None: ls = self._live.get(sid) return ls if (ls is not None and not ls._stop) else None def cost_total(self) -> dict: """所有活跃会话的 token 成本汇总(运维观测用)。""" calls = sum(ls.llm.stats().get("calls", 0) for ls in self._live.values() if ls.llm) yuan = sum(ls.llm.stats().get("est_cost_yuan", 0) for ls in self._live.values() if ls.llm) return {"active_sessions": len(self._live), "calls": calls, "est_cost_yuan": round(yuan, 4)} def _reaper(self) -> None: while True: time.sleep(10) now = time.monotonic() with self._lock: stale = [s for s, ls in self._live.items() if now - ls.last_beat > self.idle_timeout] for s in stale: logger.info("[%s] 闲置超时 → 存档暂停(停止烧 token)", s[:6]) self._evict(s) def shutdown(self) -> None: with self._lock: for s in list(self._live): self._evict(s)