engine/tools.py

"""NEMOCITY event-type specs — validation/clamping for the 4 city event types.

Semantics (godseed lineage):
* out-of-range NUMBERS are CLAMPED, never rejected;
* unknown kinds are repaired via SYNONYMS (anything unknown -> house);
* malformed shapes (bad cells, missing geometry) are rejected with a terse
  observation string — but the only module that COMPOSES these events is the
  engine itself (placement/traffic), so rejections mean engine bugs, not model
  noise. The LLM never outputs coordinates.

Pure stdlib. world.py validates through `validate_call`; queue_worker uses
`resolve_kind` / `sanitize_name` / `default_name` on model-supplied fields.
"""

from __future__ import annotations

import re
import zlib
from typing import Any, Optional

from . import constants as C
from .moderation import Moderator

_moderator = Moderator()  # wordlist layers only; no judge

TOOL_NAMES: tuple[str, ...] = ("place_building", "lay_road", "apply_fix", "note")

FIX_ACTIONS = ("new_road", "upgrade_avenue")
NOTE_KINDS = ("milestone", "infill", "fix")

_FALLBACK_KIND = "house"


# ----------------------------------------------------------------- kind resolver

def resolve_kind(value: Any) -> str:
    """Map model-supplied kind text to a BUILDINGS key. Never fails: synonyms
    repair, substring rescues, anything else becomes a humble house."""
    if not isinstance(value, str):
        return _FALLBACK_KIND
    v = re.sub(r"[\s\-]+", "_", value.strip().lower()).strip("_")
    if v in C.BUILDINGS:
        return v
    if v in C.SYNONYMS:
        return C.SYNONYMS[v]
    if v.endswith("s") and v[:-1] in C.BUILDINGS:
        return v[:-1]
    for token, kind in C.SYNONYMS.items():
        if token in v:
            return kind
    for kind in C.BUILDINGS:
        if kind in v:
            return kind
    return _FALLBACK_KIND


# ------------------------------------------------------------------ name helpers

_STOPWORDS = frozenset((
    "a", "an", "the", "and", "or", "of", "for", "with", "near", "by", "at",
    "in", "on", "to", "please", "build", "make", "add", "put", "want", "like",
    "new", "some", "my", "our", "me", "us", "city", "town",
))


def sanitize_name(value: Any, default: str = "") -> str:
    """Printable, wordlist-clean, <=24 chars; falls back to `default`."""
    if not isinstance(value, str):
        return default
    name = "".join(ch for ch in value if ch.isprintable())
    name = re.sub(r"\s+", " ", name).strip()[: C.NAME_MAX_LEN].strip()
    if not name or not _moderator.check_content(name).allowed:
        return default
    return name


def default_name(kind: str, petition_text: str, wish_id: str) -> str:
    """Deterministic pleasant default derived from petition words."""
    words = [
        w for w in re.findall(r"[A-Za-z]+", str(petition_text or ""))
        if len(w) > 2 and w.lower() not in _STOPWORDS
        and w.lower() not in C.BUILDINGS and w.lower() not in C.SYNONYMS
    ]
    label = kind.replace("_", " ").title()
    if words:
        pick = words[zlib.crc32(f"{wish_id}:{kind}".encode()) % len(words)].title()
        candidate = f"{pick} {label}"[: C.NAME_MAX_LEN].strip()
        if _moderator.check_content(candidate).allowed:
            return candidate
    return f"The {label}"[: C.NAME_MAX_LEN]


def street_name_for(key: str) -> str:
    """Curated street name from a deterministic key (e.g. 'w_000005:1:road')."""
    return C.STREET_NAMES[zlib.crc32(str(key).encode()) % len(C.STREET_NAMES)]


# ------------------------------------------------------------------- primitives

def _err(msg: str) -> tuple[None, str]:
    return None, f"rejected: {msg}"


def _int_in(value: Any, lo: int, hi: int) -> Optional[int]:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        try:
            value = float(str(value).strip())
        except (ValueError, TypeError):
            return None
    return int(round(min(max(float(value), lo), hi)))


def _coerce_cell(cell: Any) -> Optional[list[int]]:
    if not isinstance(cell, (list, tuple)) or len(cell) < 2:
        return None
    cx = _int_in(cell[0], C.COORD_MIN, C.COORD_MAX)
    cz = _int_in(cell[1], C.COORD_MIN, C.COORD_MAX)
    if cx is None or cz is None:
        return None
    return [cx, cz]


def _coerce_cells(raw: Any, max_cells: int = 256) -> Optional[list[list[int]]]:
    if not isinstance(raw, (list, tuple)) or not raw:
        return None
    out: list[list[int]] = []
    seen: set[tuple[int, int]] = set()
    for c in list(raw)[:max_cells]:
        cell = _coerce_cell(c)
        if cell is None:
            return None
        if (cell[0], cell[1]) in seen:
            continue
        seen.add((cell[0], cell[1]))
        out.append(cell)
    return out


def _text(value: Any, max_len: int) -> str:
    s = "".join(ch for ch in str(value or "") if ch.isprintable())
    return re.sub(r"\s+", " ", s).strip()[:max_len]


# ------------------------------------------------------------------- validators

def _validate_place_building(args: dict):
    kind = resolve_kind(args.get("kind"))
    spec = C.BUILDINGS[kind]
    cx = _int_in(args.get("cx"), C.COORD_MIN, C.COORD_MAX - spec["w"] + 1)
    cz = _int_in(args.get("cz"), C.COORD_MIN, C.COORD_MAX - spec["d"] + 1)
    if cx is None or cz is None or args.get("cx") is None or args.get("cz") is None:
        return _err("place_building needs cx, cz")
    lo, hi = spec["floors"]
    floors = _int_in(args.get("floors"), lo, hi)
    if floors is None:
        floors = lo
    hue = _int_in(args.get("hue"), 0, 360)
    if hue is None:
        hue = 35
    variant = _int_in(args.get("variant"), 0, 7)
    if variant is None:
        variant = 0
    name = sanitize_name(args.get("name"), default=f"The {kind.replace('_', ' ').title()}")
    return {
        "kind": kind, "name": name, "cx": cx, "cz": cz,
        "w": spec["w"], "d": spec["d"], "floors": floors,
        "hue": hue, "variant": variant,
    }, None


def _validate_lay_road(args: dict):
    cells = _coerce_cells(args.get("cells"))
    if cells is None:
        return _err("lay_road needs cells [[cx,cz],...]")
    klass = str(args.get("klass") or "street").strip().lower()
    if klass not in C.ROAD_CLASSES:
        klass = "street"
    name = sanitize_name(args.get("name"), default="")
    out = {"cells": cells, "klass": klass}
    if name:
        out["name"] = name
    return out, None


def _coerce_metrics(raw: Any) -> dict:
    out: dict[str, Any] = {}
    if isinstance(raw, dict):
        for k, v in list(raw.items())[:6]:
            key = _text(k, 24)
            if not key:
                continue
            if isinstance(v, bool):
                continue
            if isinstance(v, (int, float)):
                num = round(float(v), 2)
                out[key] = int(num) if num.is_integer() else num
            elif isinstance(v, str):
                out[key] = _text(v, 48)
    return out


def _validate_apply_fix(args: dict):
    action = str(args.get("action") or "").strip().lower()
    if action not in FIX_ACTIONS:
        return _err(f"unknown fix action '{_text(args.get('action'), 24)}'")
    cells = _coerce_cells(args.get("cells"))
    if cells is None:
        return _err("apply_fix needs cells [[cx,cz],...]")
    klass = str(args.get("klass") or "avenue").strip().lower()
    if klass not in C.ROAD_CLASSES:
        klass = "avenue"
    name = sanitize_name(args.get("name"), default=street_name_for(repr(cells[0])))
    diagnosis = _text(args.get("diagnosis"), 200)
    return {
        "action": action, "cells": cells, "klass": klass, "name": name,
        "diagnosis": diagnosis,
        "metrics_before": _coerce_metrics(args.get("metrics_before")),
        "metrics_predicted": _coerce_metrics(args.get("metrics_predicted")),
    }, None


def _validate_note(args: dict):
    text = _text(args.get("text"), 140)
    if not text:
        return _err("note needs text")
    if not _moderator.check_content(text).allowed:
        return _err("those words may not be posted on the city ledger")
    kind = str(args.get("kind") or "milestone").strip().lower()
    if kind not in NOTE_KINDS:
        kind = "milestone"
    return {"text": text, "kind": kind}, None


_VALIDATORS = {
    "place_building": _validate_place_building,
    "lay_road": _validate_lay_road,
    "apply_fix": _validate_apply_fix,
    "note": _validate_note,
}


def validate_call(tool: Any, args: Any) -> tuple[Optional[dict], Optional[str]]:
    """Validate one event. Returns (canonical_args, None) or (None, rejection)."""
    if not isinstance(tool, str) or tool not in _VALIDATORS:
        return _err(f"unknown tool '{_text(tool, 32)}'")
    if args is None:
        args = {}
    if not isinstance(args, dict):
        return _err("args must be an object")
    return _VALIDATORS[tool](args)


def as_dict() -> dict:
    """Plain-JSON view of the event surface (prompt builders may render this)."""
    return {
        "place_building": {
            "kind": {"type": "enum", "values": list(C.BUILDINGS)},
            "name": {"type": "string", "max_len": C.NAME_MAX_LEN},
            "cx": {"type": "int", "min": C.COORD_MIN, "max": C.COORD_MAX},
            "cz": {"type": "int", "min": C.COORD_MIN, "max": C.COORD_MAX},
            "floors": {"type": "int", "note": "clamped to the kind's range"},
            "hue": {"type": "int", "min": 0, "max": 360},
        },
        "lay_road": {
            "cells": {"type": "cells", "note": "[[cx,cz],...] engine-routed only"},
            "klass": {"type": "enum", "values": list(C.ROAD_CLASSES)},
            "name": {"type": "string", "max_len": C.NAME_MAX_LEN},
        },
        "apply_fix": {
            "action": {"type": "enum", "values": list(FIX_ACTIONS)},
            "cells": {"type": "cells"},
            "klass": {"type": "enum", "values": list(C.ROAD_CLASSES)},
            "diagnosis": {"type": "string", "max_len": 200},
        },
        "note": {
            "text": {"type": "string", "max_len": 140},
            "kind": {"type": "enum", "values": list(NOTE_KINDS)},
        },
    }