"""NEMOCITY world state — the deterministic engine owns ALL facts. The world is an append-only ordered list of Features (events). City-at-any- moment is a pure function of that list; the JS renderer derives all geometry from it and engine/city.py derives the Python views. The LLM never touches this module's internals: the engine composes event args (placement/traffic own all coordinates), validation clamps/repairs via engine/tools.py, the deterministic seed is crc32(f"{wish_id}:{call_index}"), and a terse observation string goes back to the planner loop. Wire/internal names keep godseed's wish_id/wish_* verbatim (zero-risk reuse); user-facing copy says "petition" elsewhere. Pure stdlib, synchronous, no pydantic. """ from __future__ import annotations import copy import re import time import zlib from dataclasses import dataclass from typing import Any, Iterable, Optional from . import summary as _summary from . import tools as _tools _FEATURE_ID_RE = re.compile(r"^e_(\d+)$") _EPITAPH_KEEP = 64 # in-memory tail; the full archive lives in traces/wishes.jsonl @dataclass class Feature: """One immutable city event. Canonical JSON shape per ARCHITECTURE.md.""" id: str wish_id: str tool: str args: dict seed: int t: float def to_dict(self) -> dict: return { "id": self.id, "wish_id": self.wish_id, "tool": self.tool, "args": copy.deepcopy(self.args), "seed": self.seed, "t": self.t, } @classmethod def from_dict(cls, d: dict) -> "Feature": return cls( id=d["id"], wish_id=d["wish_id"], tool=d["tool"], args=copy.deepcopy(d.get("args") or {}), seed=int(d.get("seed", 0)), t=d.get("t", 0), ) def _seed_for(wish_id: str, call_index: int) -> int: return zlib.crc32(f"{wish_id}:{call_index}".encode("utf-8")) & 0x7FFFFFFF def _observation(feature: Feature) -> str: """Terse 'ok' observation the planner can record.""" t, a = feature.tool, feature.args if t == "place_building": return ( f"ok: {a['kind']} '{a['name']}' placed at ({a['cx']},{a['cz']}), " f"{a['floors']} floors" ) if t == "lay_road": name = a.get("name") or "unnamed" return f"ok: {len(a['cells'])} cells of {a['klass']} laid ({name})" if t == "apply_fix": return f"ok: {a['action']} applied on {a['name']} ({len(a['cells'])} cells)" if t == "note": return "ok: noted on the city ledger" return "ok" # pragma: no cover — every known tool is handled above class World: """Append-only event list + derived views. Single-threaded (asyncio).""" def __init__(self) -> None: self._features: list[Feature] = [] self._epitaphs: list[str] = [] self._next_index: int = 0 self._last_signature: Optional[tuple] = None # ------------------------------------------------------------- loading @classmethod def load(cls, features: Iterable[Any], epitaphs: Optional[Iterable[str]] = None) -> "World": """Rebuild a world from persisted event dicts (or Feature objects). Persisted data is trusted verbatim — values (including seeds) are NOT re-validated, so genesis and replayed traces stay byte-identical. """ world = cls() for f in features or (): feature = f if isinstance(f, Feature) else Feature.from_dict(f) world._features.append(feature) world._next_index = world._derive_next_index() world._epitaphs = [str(e)[:120] for e in (epitaphs or ()) if e][-_EPITAPH_KEEP:] return world def _derive_next_index(self) -> int: best = -1 for f in self._features: m = _FEATURE_ID_RE.match(f.id) if m: best = max(best, int(m.group(1))) return best + 1 if best >= 0 else len(self._features) # ------------------------------------------------------------- mutation def apply( self, wish_id: str, call_index: int, call: Any, t: Optional[float] = None, ) -> tuple[Optional[Feature], str]: """Validate and apply one event. Returns (feature, observation) on success, (None, rejection) otherwise. Out-of-range numbers are clamped; unknown tools/shapes are rejected with a terse observation string. The world only changes on success. """ if not isinstance(call, dict): return None, "rejected: malformed call" tool = call.get("tool") if isinstance(tool, str): tool = tool.strip().lower() canonical, err = _tools.validate_call(tool, call.get("args")) if err: return None, err # An identical consecutive call within one wish is refused with a nudge # (small models repeat themselves; the city does not). signature = (str(wish_id), tool, repr(sorted(canonical.items()))) if signature == self._last_signature: return None, "rejected: already done; change something, or say done" self._last_signature = signature feature = Feature( id=f"e_{self._next_index:06d}", wish_id=str(wish_id), tool=tool, args=canonical, seed=_seed_for(str(wish_id), int(call_index)), t=time.time() if t is None else t, ) self._next_index += 1 self._features.append(feature) return feature, _observation(feature) def record_epitaph(self, text: Any) -> None: """Record a granted petition's blurb (the ticker shows the last few).""" if text: self._epitaphs.append(str(text)[:120]) if len(self._epitaphs) > _EPITAPH_KEEP: del self._epitaphs[: -_EPITAPH_KEEP] # ------------------------------------------------------------- views @property def features(self) -> tuple[Feature, ...]: return tuple(self._features) @property def epitaphs(self) -> tuple[str, ...]: return tuple(self._epitaphs) @property def version(self) -> int: return len(self._features) @property def epoch(self) -> int: """Number of distinct granted petitions (genesis is epoch 0).""" seen: set[str] = set() for f in self._features: if f.wish_id != "genesis": seen.add(f.wish_id) return len(seen) def state_dict(self) -> dict: """Full world state for /api/state and SSE hello consumers.""" return { "version": self.version, "epoch": self.epoch, "features": [f.to_dict() for f in self._features], } def summary(self, now_s: Optional[float] = None) -> str: """Compact (<600 chars) description for the LLM prompt.""" return _summary.summarize(self, now_s)