engine/world.py

"""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)