engine/world.py

"""GODSEED world state — the deterministic engine owns ALL facts.

The world is an append-only ordered list of Features. World-at-any-moment is
a pure function of that list; the JS renderer derives all geometry from it.
Sky/weather are last-write-wins for the convenience views (state_dict) but
every set_sky / set_weather is stored as a feature so replay works.

The LLM never touches this module's internals: it proposes tool calls, the
engine validates (clamp numbers / reject unknown tools & enums), assigns the
deterministic seed `crc32(f"{wish_id}:{call_index}") & 0x7fffffff`, appends,
and returns a terse observation string.

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"^f_(\d+)$")

# Convenience default when no set_weather feature exists yet (genesis has none).
DEFAULT_WEATHER = {"kind": "clear", "intensity": 0.0}

_EPITAPH_KEEP = 64  # in-memory tail; the full archive lives in traces/wishes.jsonl


@dataclass
class Feature:
    """One immutable creation 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 _coord(args: dict) -> str:
    return f"({round(args['lat'])},{round(args['lon'])})"


def _observation(feature: Feature) -> str:
    """Terse 'ok' observation the model can react to on the next turn."""
    t, a = feature.tool, feature.args
    if t == "raise_terrain":
        return f"ok: mountains risen at {_coord(a)}"
    if t == "lower_terrain":
        return f"ok: ground lowered at {_coord(a)}"
    if t == "spawn_flora":
        return f"ok: {a['kind']} spawned at {_coord(a)}"
    if t == "place_structure":
        return f"ok: {a['kind']} placed at {_coord(a)}"
    if t == "place_water":
        if a["kind"] == "stream":
            return f"ok: stream traced through {len(a['path'])} points"
        center = a["path"][0]
        return f"ok: pool placed at ({round(center[0])},{round(center[1])})"
    if t == "set_weather":
        return f"ok: weather set to {a['kind']}"
    if t == "set_sky":
        return f"ok: sky set to {a['palette']}"
    if t == "inscribe_wish":
        return f"ok: wish inscribed in {a['style']}"
    if t == "spawn_life":
        return f"ok: {a['count']} {a['kind']} stirring at {_coord(a)}"
    if t == "build_district":
        return f"ok: a district rises near {_coord(a)}"
    if t == "place_road":
        return f"ok: a road laid through {len(a['path'])} points"
    return "ok"  # pragma: no cover — every known tool is handled above


class World:
    """Append-only feature 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 feature 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 tool call.

        Returns (feature, observation) on success, (None, rejection) otherwise.
        Out-of-range numbers are clamped; unknown tools/enums 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
        # Small models repeat themselves; the land does not. An identical
        # consecutive call within one wish is refused with a nudge.
        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"f_{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 wish's epitaph (summary shows the last 2)."""
        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 wishes (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)

    @property
    def sky(self) -> Optional[dict]:
        """Last-write-wins set_sky args (genesis always sets one)."""
        for f in reversed(self._features):
            if f.tool == "set_sky":
                return copy.deepcopy(f.args)
        return None

    @property
    def weather(self) -> dict:
        """Last-write-wins set_weather args; calm 'clear' before any is set."""
        for f in reversed(self._features):
            if f.tool == "set_weather":
                return copy.deepcopy(f.args)
        return dict(DEFAULT_WEATHER)

    def state_dict(self) -> dict:
        """Full world state for /api/state and SSE hello consumers."""
        return {
            "version": self.version,
            "epoch": self.epoch,
            "sky": self.sky,
            "weather": self.weather,
            "features": [f.to_dict() for f in self._features],
        }

    def summary(self) -> str:
        """Compact (<600 chars) description for the LLM prompt."""
        return _summary.summarize(self)