Spaces:
Sleeping
Sleeping
| """Butterfly game logic: the state machine between the browser and the world model. | |
| One live world per GPU. The game wraps the StreamingHost with: | |
| * anchors — up to 3 named .wsave savepoints (the timeline tree) | |
| * recording — after an anchor, the played segment becomes that anchor's | |
| "original timeline" (ghost) : frames + latents + actions | |
| * duel — restore an anchor with a re-rolled noise stream (chaos!) and | |
| replay the same duration; per-block similarity vs the ghost | |
| is the score. Beating the do-nothing chaos baseline = skill. | |
| * dream life — the RoPE table budget (360 latent frames) is the session clock. | |
| Pure-python; transport-agnostic (the FastAPI layer just forwards messages). | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import os | |
| from dataclasses import dataclass, field | |
| from typing import Any, Dict, List, Optional | |
| import torch | |
| MAX_ANCHORS = 3 | |
| GRADES = ((0.35, "S"), (0.20, "A"), (0.05, "B"), (-1e9, "C")) | |
| # Butterfly grading: amplification of your ONE flap vs a full re-roll of fate. | |
| BFLY_GRADES = ((1.25, "S"), (0.8, "A"), (0.35, "B"), (-1e9, "C")) | |
| class Ghost: | |
| """The original timeline recorded after an anchor.""" | |
| frames: List[torch.Tensor] = field(default_factory=list) # uint8 [12,H,W,3] per block | |
| latents: List[torch.Tensor] = field(default_factory=list) # [16,3,44,80] per block | |
| actions: List[Dict[str, Any]] = field(default_factory=list) | |
| class DuelState: | |
| anchor_id: str | |
| block: int = 0 | |
| dists: List[float] = field(default_factory=list) | |
| class ButterflyState: | |
| """One-flap divergence race: same fate (saved RNG), one changed action.""" | |
| anchor_id: str | |
| block: int = 0 | |
| flap_block: Optional[int] = None | |
| dists: List[float] = field(default_factory=list) # from the flap onward | |
| # Mission「守梦远征」— the default objective. Explore N blocks (they become the | |
| # original timeline), then fate re-rolls (THE TEAR) and you must hold the world: | |
| # a block whose similarity-vs-ghost falls to 0 (worse than doing nothing against | |
| # chaos) is a strike; MISSION_STRIKES strikes tear the dream. Survive the full | |
| # hold and the dream is held — graded by the usual duel score. | |
| MISSION_EXPLORE_BLOCKS = 10 | |
| MISSION_STRIKES = 3 | |
| # Tear line, calibrated on the 184931 e2e measurement: a player who merely | |
| # mimics their ghost sits at similarity ~0 (score 0.043) and must NOT strike | |
| # out; a strike means actively pushing the world further than chaos itself | |
| # (dist > 1.25x the measured chaos baseline). | |
| MISSION_TEAR_LINE = -0.25 | |
| class MissionState: | |
| phase: str = "explore" # explore | hold | |
| # None until the first step: capturing at t=0 is impossible (the streaming | |
| # VAE's conv cache only exists after the first decode — 184937 root cause), | |
| # so the home anchor drops lazily after the player's first move. | |
| anchor_id: Optional[str] = None | |
| strikes: int = 0 | |
| class ButterflyGame: | |
| def __init__(self, host, timeline, save_dir: str, | |
| chaos_baseline_path: Optional[str] = None): | |
| self.host = host | |
| self.timeline = timeline | |
| self.save_dir = save_dir | |
| self.mode = "idle" # idle | free | duel | butterfly | |
| self.anchors: List[str] = [] # save_ids, oldest first | |
| self.ghosts: Dict[str, Ghost] = {} | |
| self.duel: Optional[DuelState] = None | |
| self.bfly: Optional[ButterflyState] = None | |
| self.mission: Optional[MissionState] = None | |
| self.chaos_curve = self._load_chaos_baseline(chaos_baseline_path) | |
| # -- helpers ------------------------------------------------------------- | |
| def _load_chaos_baseline(path: Optional[str]) -> List[float]: | |
| """Mean noise-fork latent-MSE curve from the M2 run (per pixel frame).""" | |
| if path and os.path.exists(path): | |
| with open(path) as fh: | |
| return json.load(fh)["noise"]["curve"] | |
| return [] | |
| def _chaos_mean(self, n_blocks: int) -> float: | |
| """Baseline mean latent distance over a duel horizon of n blocks.""" | |
| if not self.chaos_curve: | |
| return 0.25 # fallback: retro-fit from M2 plateau | |
| horizon = min(len(self.chaos_curve), n_blocks * 12) | |
| return sum(self.chaos_curve[:horizon]) / max(1, horizon) | |
| def blocks_left(self) -> int: | |
| return (self.host.max_latent_frames - self.host.cursor["latent_frame"]) \ | |
| // self.host.num_frame_per_block | |
| def _status(self) -> Dict[str, Any]: | |
| return {"mode": self.mode, "blocks_left": self.blocks_left, | |
| "cursor": self.host.cursor, "anchors": self.anchors, | |
| "current": getattr(self, "_current_anchor", None)} | |
| # -- commands ------------------------------------------------------------ | |
| def new_game(self, image, seed: int) -> Dict[str, Any]: | |
| self.host.prime(image, seed=seed) | |
| self.mode = "free" | |
| self.anchors, self.ghosts, self.duel, self.bfly = [], {}, None, None | |
| self.mission = None # a fresh dream owes nothing to an old mission | |
| self._current_anchor = None | |
| return {"type": "started", **self._status()} | |
| def step(self, action: Dict[str, Any]): | |
| """Advance one block. Returns (frames_live, ghost_frame_or_None, msg).""" | |
| if self.mode == "idle": | |
| raise RuntimeError("start a game first") | |
| if self.blocks_left <= 0: | |
| self.mode = "idle" | |
| msg = {"type": "collapsed", **self._status()} | |
| if self.mission is not None: | |
| msg = self._mission_msg(msg) | |
| return None, None, msg | |
| flap = bool(action.get("flap")) | |
| action = {"keys": action.get("keys", []), "mouse": action.get("mouse", [0.0, 0.0])} | |
| if self.mode == "butterfly": | |
| b = self.bfly | |
| if flap and b.flap_block is None: | |
| b.flap_block = b.block # the one wing-beat: the player's action goes in | |
| else: | |
| action = self.ghosts[b.anchor_id].actions[b.block] # faithful replay | |
| result = self.host.step(action) | |
| live_u8 = (result.frames.permute(0, 2, 3, 1) * 255).clamp(0, 255) \ | |
| .to(torch.uint8).cpu() # JPEG encoding is host-side | |
| ghost_u8, msg = None, None | |
| if self.mode == "duel": | |
| d = self.duel | |
| ghost = self.ghosts[d.anchor_id] | |
| # pixel-space MSE in [0,1] — same space & scale as the M2 chaos baseline | |
| dist = float(((live_u8.float() - ghost.frames[d.block].float()) / 255.0) | |
| .pow(2).mean()) | |
| d.dists.append(dist) | |
| ghost_u8 = ghost.frames[d.block] | |
| d.block += 1 | |
| similarity = self._similarity(dist) | |
| if d.block >= len(ghost.latents): | |
| msg = self._end_duel() | |
| else: | |
| msg = {"type": "duel_tick", "block": d.block, | |
| "total": len(ghost.latents), "dist": dist, | |
| "similarity": similarity, **self._status()} | |
| elif self.mode == "butterfly": | |
| b = self.bfly | |
| ghost = self.ghosts[b.anchor_id] | |
| dist = float(((live_u8.float() - ghost.frames[b.block].float()) / 255.0) | |
| .pow(2).mean()) | |
| if b.flap_block is not None: | |
| b.dists.append(dist) | |
| ghost_u8 = ghost.frames[b.block] | |
| b.block += 1 | |
| base = self._chaos_mean(max(1, b.block)) | |
| similarity = max(0.0, 1.0 - dist / max(base, 1e-9)) | |
| if b.block >= len(ghost.latents): | |
| msg = self._end_butterfly() | |
| else: | |
| msg = {"type": "butterfly_tick", "block": b.block, | |
| "total": len(ghost.latents), "dist": dist, | |
| "similarity": similarity, | |
| "flapped": b.flap_block is not None, | |
| "flap_block": b.flap_block, **self._status()} | |
| else: | |
| # free mode: record into the current anchor's ghost | |
| if self._current_anchor is not None: | |
| g = self.ghosts[self._current_anchor] | |
| g.frames.append(live_u8) | |
| g.latents.append(result.latents.float().cpu()) | |
| g.actions.append(action) | |
| msg = {"type": "stepped", "stats": result.stats, **self._status()} | |
| if self.mission is not None and self.mission.anchor_id is None \ | |
| and self.mode == "free": | |
| # first move done -> the streaming caches exist -> drop home | |
| m, self.mission = self.mission, None # bypass the mission guard | |
| try: | |
| a = self.drop_anchor("home") | |
| finally: | |
| self.mission = m | |
| m.anchor_id = a["save_id"] | |
| msg["home_anchor"] = a["save_id"] | |
| if self.mission is not None: | |
| msg = self._mission_msg(msg) | |
| return live_u8, ghost_u8, msg | |
| def _similarity(self, dist: float) -> float: | |
| base = self._chaos_mean(max(1, self.duel.block + 1)) | |
| return max(0.0, 1.0 - dist / max(base, 1e-9)) | |
| def drop_anchor(self, label: str) -> Dict[str, Any]: | |
| from savepoint.state import save_wsave | |
| self._mission_guard() | |
| if self.mode != "free": | |
| raise RuntimeError("anchors only in free play") | |
| if len(self.anchors) >= MAX_ANCHORS: | |
| raise RuntimeError(f"all {MAX_ANCHORS} anchor slots used") | |
| state = self.host.capture(mode="state") | |
| node = self.timeline.add( | |
| path="pending", latent_frame=self.host.cursor["latent_frame"], | |
| mode="state", parent_id=self._current_anchor, label=label) | |
| fname = f"{node.save_id}.wsave" | |
| save_wsave(state, os.path.join(self.save_dir, fname)) | |
| self.timeline.nodes[node.save_id].path = fname | |
| self.timeline._flush() | |
| self.anchors.append(node.save_id) | |
| self.ghosts[node.save_id] = Ghost() | |
| self._current_anchor = node.save_id | |
| return {"type": "anchored", "save_id": node.save_id, | |
| "bytes": os.path.getsize(os.path.join(self.save_dir, fname)), | |
| **self._status()} | |
| def rewind(self, anchor_id: str) -> Dict[str, Any]: | |
| from savepoint.state import load_wsave | |
| self._mission_guard() | |
| if anchor_id not in self.ghosts: | |
| raise KeyError(f"unknown anchor {anchor_id}") | |
| self.host.restore(load_wsave(os.path.join( | |
| self.save_dir, self.timeline.nodes[anchor_id].path))) | |
| # rewinding truncates that anchor's ghost: a new original timeline begins | |
| self.ghosts[anchor_id] = Ghost() | |
| self._current_anchor = anchor_id | |
| self.mode = "free" | |
| self.duel = None | |
| self.bfly = None | |
| return {"type": "rewound", "anchor": anchor_id, **self._status()} | |
| def start_duel(self, anchor_id: str, seed: int) -> Dict[str, Any]: | |
| from savepoint.state import load_wsave | |
| self._mission_guard() | |
| ghost = self.ghosts.get(anchor_id) | |
| if not ghost or not ghost.latents: | |
| raise RuntimeError("play some blocks after this anchor first — " | |
| "the duel needs an original timeline to defend") | |
| self.host.restore(load_wsave(os.path.join( | |
| self.save_dir, self.timeline.nodes[anchor_id].path))) | |
| self.host._rng.manual_seed(seed) # re-roll chance: chaos begins | |
| self.mode = "duel" | |
| self.duel = DuelState(anchor_id=anchor_id) | |
| self.bfly = None | |
| self._current_anchor = anchor_id | |
| return {"type": "duel_started", "anchor": anchor_id, | |
| "total": len(ghost.latents), | |
| "ghost_actions": ghost.actions, **self._status()} | |
| def _end_duel(self) -> Dict[str, Any]: | |
| d = self.duel | |
| mean_d = sum(d.dists) / len(d.dists) | |
| base = self._chaos_mean(len(d.dists)) | |
| score = max(-1.0, min(1.0, 1.0 - mean_d / max(base, 1e-9))) | |
| grade = next(g for thr, g in GRADES if score >= thr) | |
| self.mode = "free" | |
| self.duel = None | |
| # the duel branch is now the live timeline; its anchor ghost resets | |
| self.ghosts[d.anchor_id] = Ghost() | |
| return {"type": "duel_end", "score": score, "grade": grade, | |
| "mean_dist": mean_d, "chaos_baseline": base, | |
| "dists": d.dists, **self._status()} | |
| # -- butterfly「扇翅」 ---------------------------------------------------- | |
| def start_butterfly(self, anchor_id: str) -> Dict[str, Any]: | |
| from savepoint.state import load_wsave | |
| self._mission_guard() | |
| ghost = self.ghosts.get(anchor_id) | |
| if not ghost or not ghost.latents: | |
| raise RuntimeError("play some blocks after this anchor first — " | |
| "the flap needs an original timeline to disturb") | |
| # Same fate: the .wsave brings the RNG back, so a faithful replay is | |
| # bit-exact and every ripple on screen is caused by the one flap alone. | |
| self.host.restore(load_wsave(os.path.join( | |
| self.save_dir, self.timeline.nodes[anchor_id].path))) | |
| self.mode = "butterfly" | |
| self.duel = None | |
| self.bfly = ButterflyState(anchor_id=anchor_id) | |
| self._current_anchor = anchor_id | |
| return {"type": "butterfly_started", "anchor": anchor_id, | |
| "total": len(ghost.latents), | |
| "ghost_actions": ghost.actions, **self._status()} | |
| def _end_butterfly(self) -> Dict[str, Any]: | |
| b = self.bfly | |
| self.mode = "free" | |
| self.bfly = None | |
| # the flapped branch is now the live timeline; the anchor's ghost resets | |
| self.ghosts[b.anchor_id] = Ghost() | |
| if b.flap_block is None or not b.dists: | |
| return {"type": "butterfly_end", "score": 0.0, "grade": "C", | |
| "amp": 0.0, "flap_block": None, "mean_dist": 0.0, | |
| "chaos_baseline": self._chaos_mean(1), "dists": [], | |
| **self._status()} | |
| mean_d = sum(b.dists) / len(b.dists) | |
| base = self._chaos_mean(len(b.dists)) | |
| amp = mean_d / max(base, 1e-9) | |
| grade = next(g for thr, g in BFLY_GRADES if amp >= thr) | |
| return {"type": "butterfly_end", "score": amp, "grade": grade, | |
| "amp": amp, "flap_block": b.flap_block, "mean_dist": mean_d, | |
| "chaos_baseline": base, "dists": b.dists, **self._status()} | |
| # -- mission「守梦远征」— the default objective --------------------------- | |
| def start_mission(self, image, seed: int) -> Dict[str, Any]: | |
| """New dream with an explicit goal: explore, survive the tear, hold.""" | |
| self.mission = None # clear any stale mission before reset | |
| self.new_game(image, seed) | |
| self.mission = MissionState() # home anchor drops after the first move | |
| return {"type": "mission_started", "anchor": None, | |
| "explore_blocks": MISSION_EXPLORE_BLOCKS, | |
| "strikes_allowed": MISSION_STRIKES, **self._status()} | |
| def mission_tear(self, seed: int) -> Dict[str, Any]: | |
| """Fate re-rolls; the hold phase begins.""" | |
| m = self.mission | |
| if not m or m.phase != "explore": | |
| raise RuntimeError("no mission is exploring") | |
| ghost = self.ghosts.get(m.anchor_id) | |
| if not ghost or not ghost.latents: | |
| raise RuntimeError("walk a little first — the tear needs a timeline to threaten") | |
| self.mission = None # internal duel start bypasses the guard | |
| try: | |
| msg = self.start_duel(m.anchor_id, seed) | |
| finally: | |
| self.mission = m | |
| m.phase, m.strikes = "hold", 0 | |
| return {**msg, "type": "mission_tear", | |
| "strikes_allowed": MISSION_STRIKES} | |
| def abandon_mission(self) -> Dict[str, Any]: | |
| m = self.mission | |
| if not m: | |
| raise RuntimeError("no mission to abandon") | |
| self.mission = None | |
| if self.mode == "duel" and self.duel and self.duel.anchor_id == m.anchor_id: | |
| self.mode, self.duel = "free", None | |
| self.ghosts[m.anchor_id] = Ghost() | |
| return {"type": "mission_abandoned", **self._status()} | |
| def _mission_guard(self) -> None: | |
| if self.mission is not None: | |
| raise RuntimeError("the mission holds this dream — " | |
| "hold it to the end or abandon it first") | |
| def _mission_msg(self, msg: Dict[str, Any]) -> Dict[str, Any]: | |
| """Overlay mission progress / verdicts onto the base message stream.""" | |
| m = self.mission | |
| if m.anchor_id is None: # pre-anchor step (shouldn't happen) | |
| return msg | |
| if msg["type"] == "stepped" and m.phase == "explore": | |
| played = len(self.ghosts[m.anchor_id].actions) | |
| tear_in = max(0, MISSION_EXPLORE_BLOCKS - played) | |
| msg["mission"] = {"phase": "explore", "tear_in": tear_in, | |
| "tear_now": tear_in == 0} | |
| elif msg["type"] == "duel_tick" and m.phase == "hold": | |
| base = self._chaos_mean(max(1, msg["block"])) | |
| sim_raw = 1.0 - msg["dist"] / max(base, 1e-9) | |
| if sim_raw < MISSION_TEAR_LINE: | |
| m.strikes += 1 | |
| if m.strikes >= MISSION_STRIKES: | |
| d = self.duel | |
| self.mode, self.duel, self.mission = "free", None, None | |
| self.ghosts[d.anchor_id] = Ghost() | |
| mean_d = sum(d.dists) / len(d.dists) | |
| return {"type": "mission_end", "won": False, "reason": "torn", | |
| "strikes": MISSION_STRIKES, "held": d.block, | |
| "total": msg["total"], "mean_dist": mean_d, | |
| "chaos_baseline": self._chaos_mean(len(d.dists)), | |
| **self._status()} | |
| msg["mission"] = {"phase": "hold", "strikes": m.strikes, | |
| "strikes_allowed": MISSION_STRIKES, | |
| "sim_raw": sim_raw, | |
| "held": msg["block"], "total": msg["total"]} | |
| elif msg["type"] == "duel_end" and m.phase == "hold": | |
| # the final block ends the duel before its tick — judge it here too | |
| dists = msg.get("dists") or [] | |
| if dists: | |
| base = self._chaos_mean(len(dists)) | |
| if 1.0 - dists[-1] / max(base, 1e-9) < MISSION_TEAR_LINE: | |
| m.strikes += 1 | |
| self.mission = None | |
| if m.strikes >= MISSION_STRIKES: | |
| msg = {**msg, "type": "mission_end", "won": False, | |
| "reason": "torn", "strikes": m.strikes, | |
| "held": len(dists), "total": len(dists)} | |
| else: | |
| msg = {**msg, "type": "mission_end", "won": True, | |
| "reason": "held", "strikes": m.strikes} | |
| elif msg["type"] == "collapsed": | |
| self.mission = None | |
| msg = {**msg, "type": "mission_end", "won": False, | |
| "reason": "collapsed", "strikes": m.strikes} | |
| return msg | |
| # -- relay「传梦」 -------------------------------------------------------- | |
| def export_info(self, anchor_id: str) -> Dict[str, Any]: | |
| node = self.timeline.nodes.get(anchor_id) | |
| if node is None: | |
| raise KeyError(f"unknown anchor {anchor_id}") | |
| path = os.path.join(self.save_dir, node.path) | |
| return {"type": "export", "save_id": anchor_id, "path": path, | |
| "bytes": os.path.getsize(path), "label": node.label, | |
| "latent_frame": node.latent_frame} | |
| def import_save(self, path: str, label: str = "") -> Dict[str, Any]: | |
| import shutil | |
| from savepoint.state import load_wsave | |
| self._mission_guard() | |
| if self.mode in ("duel", "butterfly"): | |
| raise RuntimeError("finish the current round first") | |
| if len(self.anchors) >= MAX_ANCHORS: | |
| raise RuntimeError(f"all {MAX_ANCHORS} anchor slots used") | |
| path = os.path.expanduser(path) | |
| if not os.path.isfile(path): | |
| raise FileNotFoundError(path) | |
| state = load_wsave(path) # validate before adopting someone's dream | |
| node = self.timeline.add( | |
| path="pending", latent_frame=int(state.cursor.get("latent_frame", 0)), | |
| mode=state.mode, parent_id=None, label=label or "relayed dream") | |
| fname = f"{node.save_id}.wsave" | |
| dest = os.path.join(self.save_dir, fname) | |
| if os.path.abspath(path) != os.path.abspath(dest): | |
| shutil.copy2(path, dest) | |
| self.timeline.nodes[node.save_id].path = fname | |
| self.timeline._flush() | |
| self.anchors.append(node.save_id) | |
| self.ghosts[node.save_id] = Ghost() | |
| return {"type": "imported", "save_id": node.save_id, | |
| "bytes": os.path.getsize(dest), "label": node.label, | |
| **self._status()} | |