308 lines
11 KiB
Python
308 lines
11 KiB
Python
"""State validator for environment / entity.data / world schema.
|
|
|
|
All mutations of `world.environment` and `entity.data` go through this module.
|
|
The orchestrator's `env_update` and `entity_update` tools use `apply_patch`.
|
|
|
|
Validation rules:
|
|
- Required fields must be present (per `environment_schema` and entity `schemas`).
|
|
- Field types must match the declared type.
|
|
- Numeric ranges enforced when `max`/`min` provided.
|
|
- Nested `object` / `array` schemas are validated recursively.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any
|
|
|
|
# Supported primitive JSON-schema type names
|
|
_PRIMITIVES = {"string", "integer", "number", "boolean"}
|
|
_PATCH_OPS = {"set", "inc", "dec", "append", "remove"}
|
|
|
|
|
|
def validate_state(state: dict[str, Any], schema_fields: list[dict]) -> tuple[bool, list[str]]:
|
|
"""Validate `state` against a list of field definitions.
|
|
|
|
Each field definition has the shape:
|
|
{
|
|
"name": "player",
|
|
"type": "object" | "array" | "string" | ...,
|
|
"required": bool,
|
|
"properties": [ ... ], # for type=object
|
|
"items": { ... }, # for type=array
|
|
"min": int, "max": int, # for numeric types
|
|
"default": <any>
|
|
}
|
|
"""
|
|
errors: list[str] = []
|
|
for field in schema_fields:
|
|
name = field.get("name")
|
|
if not name:
|
|
errors.append("Schema field missing 'name'")
|
|
continue
|
|
if name not in state:
|
|
if field.get("required"):
|
|
errors.append(f"Missing required field: {name}")
|
|
continue
|
|
_validate_value(state[name], field, path=name, errors=errors)
|
|
return (len(errors) == 0, errors)
|
|
|
|
|
|
def _validate_value(
|
|
value: Any, field_schema: dict, path: str, errors: list[str]
|
|
) -> None:
|
|
ftype = field_schema.get("type", "string")
|
|
if ftype in _PRIMITIVES:
|
|
_validate_primitive(value, ftype, field_schema, path, errors)
|
|
elif ftype == "object":
|
|
if not isinstance(value, dict):
|
|
errors.append(f"{path} must be object")
|
|
return
|
|
props = field_schema.get("properties", [])
|
|
# validate child fields
|
|
ok, child_errors = validate_state(value, props)
|
|
if not ok:
|
|
errors.extend(child_errors)
|
|
elif ftype == "array":
|
|
if not isinstance(value, list):
|
|
errors.append(f"{path} must be array")
|
|
return
|
|
items_schema = field_schema.get("items")
|
|
if items_schema:
|
|
for i, item in enumerate(value):
|
|
_validate_value(item, items_schema, f"{path}[{i}]", errors)
|
|
else:
|
|
errors.append(f"{path}: unknown type {ftype!r}")
|
|
|
|
|
|
def _validate_primitive(
|
|
value: Any, ftype: str, field_schema: dict, path: str, errors: list[str]
|
|
) -> None:
|
|
if ftype == "string":
|
|
if not isinstance(value, str):
|
|
errors.append(f"{path} must be string")
|
|
return
|
|
elif ftype == "integer":
|
|
if isinstance(value, bool) or not isinstance(value, int):
|
|
errors.append(f"{path} must be integer")
|
|
return
|
|
elif ftype == "number":
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
errors.append(f"{path} must be number")
|
|
return
|
|
elif ftype == "boolean":
|
|
if not isinstance(value, bool):
|
|
errors.append(f"{path} must be boolean")
|
|
return
|
|
# Range checks
|
|
if ftype in ("integer", "number"):
|
|
mn = field_schema.get("min")
|
|
mx = field_schema.get("max")
|
|
if mn is not None and value < mn:
|
|
errors.append(f"{path} must be >= {mn}, got {value}")
|
|
if mx is not None and value > mx:
|
|
errors.append(f"{path} must be <= {mx}, got {value}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Patch application
|
|
# ---------------------------------------------------------------------------
|
|
_PATH_TOKEN_RE = re.compile(r"\.?([^\.\[\]]+)|\[(\d+)\]")
|
|
|
|
|
|
def _split_path(path: str) -> list[tuple[str, int | None]]:
|
|
"""Split a dotted path into tokens. Supports `arr[0].field` syntax."""
|
|
tokens: list[tuple[str, int | None]] = []
|
|
for m in _PATH_TOKEN_RE.finditer(path):
|
|
if m.group(1) is not None and m.group(1) != "":
|
|
tokens.append((m.group(1), None))
|
|
elif m.group(2) is not None:
|
|
tokens.append(("", int(m.group(2))))
|
|
return tokens
|
|
|
|
|
|
def _navigate(state: Any, tokens: list[tuple[str, int | None]]) -> tuple[bool, Any, str]:
|
|
"""Walk into state along tokens. Returns (ok, value, error)."""
|
|
cur = state
|
|
for i, (key, idx) in enumerate(tokens):
|
|
if idx is not None:
|
|
if not isinstance(cur, list):
|
|
return False, None, f"cannot index into non-list at {'.'.join(t[0] for t in tokens[:i])}"
|
|
if idx >= len(cur):
|
|
return False, None, f"index {idx} out of range"
|
|
cur = cur[idx]
|
|
else:
|
|
if not isinstance(cur, dict):
|
|
return False, None, f"cannot key into non-object at {'.'.join(t[0] for t in tokens[:i])}"
|
|
if key not in cur:
|
|
return False, None, f"key {key!r} not found"
|
|
cur = cur[key]
|
|
return True, cur, ""
|
|
|
|
|
|
def _set_path(state: Any, tokens: list[tuple[str, int | None]], value: Any) -> tuple[bool, str]:
|
|
"""Set value at path, creating intermediate dicts as needed."""
|
|
if not tokens:
|
|
return False, "empty path"
|
|
cur = state
|
|
for i, (key, idx) in enumerate(tokens[:-1]):
|
|
nxt_key, nxt_idx = tokens[i + 1]
|
|
if idx is not None:
|
|
# current is list — descend by index
|
|
if not isinstance(cur, list):
|
|
return False, "cannot index non-list"
|
|
while len(cur) <= idx:
|
|
cur.append({})
|
|
cur = cur[idx]
|
|
else:
|
|
if not isinstance(cur, dict):
|
|
return False, "cannot key non-object"
|
|
if key not in cur:
|
|
cur[key] = [] if nxt_idx is not None else {}
|
|
cur = cur[key]
|
|
# last token
|
|
last_key, last_idx = tokens[-1]
|
|
if last_idx is not None:
|
|
if not isinstance(cur, list):
|
|
return False, "cannot index non-list"
|
|
while len(cur) <= last_idx:
|
|
cur.append(None)
|
|
cur[last_idx] = value
|
|
else:
|
|
if not isinstance(cur, dict):
|
|
return False, "cannot key non-object"
|
|
cur[last_key] = value
|
|
return True, ""
|
|
|
|
|
|
def _remove_path(state: Any, tokens: list[tuple[str, int | None]]) -> tuple[bool, str]:
|
|
"""Remove the value at path."""
|
|
if not tokens:
|
|
return False, "empty path"
|
|
parent_tokens = tokens[:-1]
|
|
ok, parent, err = _navigate(state, parent_tokens)
|
|
if not ok:
|
|
return False, err
|
|
last_key, last_idx = tokens[-1]
|
|
if last_idx is not None:
|
|
if not isinstance(parent, list):
|
|
return False, "cannot index non-list"
|
|
if last_idx >= len(parent):
|
|
return False, "index out of range"
|
|
parent.pop(last_idx)
|
|
else:
|
|
if not isinstance(parent, dict):
|
|
return False, "cannot key non-object"
|
|
if last_key not in parent:
|
|
return False, f"key {last_key!r} not found"
|
|
del parent[last_key]
|
|
return True, ""
|
|
|
|
|
|
def apply_patch(state: dict[str, Any], patch: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
|
|
"""Apply a patch to `state`. Returns (new_state, errors).
|
|
|
|
Patch format: `{field_path: new_value | {op: ..., by: N | value: V}}`.
|
|
Supported ops: `set` (default), `inc`, `dec`, `append`, `remove`.
|
|
|
|
The state is mutated in place — pass a deepcopy if you need to preserve the original.
|
|
"""
|
|
import copy
|
|
|
|
state = copy.deepcopy(state)
|
|
errors: list[str] = []
|
|
for path, op_spec in patch.items():
|
|
tokens = _split_path(path)
|
|
if not tokens:
|
|
errors.append(f"invalid path: {path!r}")
|
|
continue
|
|
|
|
# Determine if this is an op-dict or a direct value
|
|
if isinstance(op_spec, dict) and "op" in op_spec and op_spec["op"] in _PATCH_OPS:
|
|
op = op_spec["op"]
|
|
if op == "set":
|
|
ok, err = _set_path(state, tokens, op_spec.get("value"))
|
|
if not ok:
|
|
errors.append(f"{path}: {err}")
|
|
elif op in ("inc", "dec"):
|
|
by = op_spec.get("by", 1)
|
|
if op == "dec":
|
|
by = -by
|
|
ok, cur, err = _navigate(state, tokens)
|
|
if not ok:
|
|
# create with the delta value
|
|
ok2, err2 = _set_path(state, tokens, by)
|
|
if not ok2:
|
|
errors.append(f"{path}: {err2}")
|
|
else:
|
|
if isinstance(cur, bool) or not isinstance(cur, (int, float)):
|
|
errors.append(f"{path}: cannot {op} non-number")
|
|
else:
|
|
ok2, err2 = _set_path(state, tokens, cur + by)
|
|
if not ok2:
|
|
errors.append(f"{path}: {err2}")
|
|
elif op == "append":
|
|
value = op_spec.get("value")
|
|
ok, cur, err = _navigate(state, tokens)
|
|
if not ok:
|
|
# create empty list, then append
|
|
ok2, err2 = _set_path(state, tokens, [value])
|
|
if not ok2:
|
|
errors.append(f"{path}: {err2}")
|
|
else:
|
|
if not isinstance(cur, list):
|
|
errors.append(f"{path}: cannot append to non-list")
|
|
else:
|
|
cur.append(value)
|
|
elif op == "remove":
|
|
ok, err = _remove_path(state, tokens)
|
|
if not ok:
|
|
errors.append(f"{path}: {err}")
|
|
else:
|
|
# Direct value assignment
|
|
ok, err = _set_path(state, tokens, op_spec)
|
|
if not ok:
|
|
errors.append(f"{path}: {err}")
|
|
return state, errors
|
|
|
|
|
|
def validate_world(world_dict: dict[str, Any]) -> tuple[bool, list[str]]:
|
|
"""Top-level validation of a World dict.
|
|
|
|
Checks: presence of required keys, types of basic fields, and that
|
|
`environment` validates against `environment_schema`.
|
|
"""
|
|
errors: list[str] = []
|
|
required_top = ["name", "language", "schemas", "environment_schema", "environment"]
|
|
for k in required_top:
|
|
if k not in world_dict:
|
|
errors.append(f"Missing required world field: {k}")
|
|
|
|
# environment must validate against environment_schema
|
|
env_schema = world_dict.get("environment_schema", [])
|
|
env = world_dict.get("environment", {})
|
|
if env_schema and env:
|
|
ok, env_errors = validate_state(env, env_schema)
|
|
if not ok:
|
|
errors.extend(env_errors)
|
|
|
|
# plot_rails structure
|
|
pr = world_dict.get("plot_rails") or {}
|
|
for k in ("hooks", "current_goals", "completed_goals"):
|
|
if k not in pr:
|
|
errors.append(f"plot_rails missing key: {k}")
|
|
elif not isinstance(pr[k], list):
|
|
errors.append(f"plot_rails.{k} must be list")
|
|
|
|
# current_time format
|
|
ct = world_dict.get("current_time")
|
|
if ct:
|
|
from app.core.time_utils import GameTime
|
|
|
|
try:
|
|
GameTime.parse(ct)
|
|
except ValueError as e:
|
|
errors.append(str(e))
|
|
|
|
return (len(errors) == 0, errors)
|