""" Parser utilities for the WorldSmithAI DSL. This module converts raw DSL input into validated ``WorldSpec`` objects. It is designed for realistic SLM output, which may contain Markdown fences, explanatory text, or minor structural variations. The parser does not instantiate runtime objects and does not execute arbitrary code. It only: 1. extracts JSON-like content, 2. parses JSON into Python data, 3. applies conservative structural normalization, 4. validates the result with ``WorldSpec``. Example: raw_output = ''' Here is the world: ```json { "id": "tiny_farm", "agents": [ { "id": "farmer_1", "type": "farmer", "behaviors": ["move", {"name": "harvest"}], "policy": "rule_policy" } ], "resources": [] } ``` ''' spec = parse_world_spec(raw_output) print(spec.id) Future extensibility: - Add schema-version migrations. - Add YAML support if project dependencies allow it. - Add stricter SLM repair modes with explicit diagnostics. - Add streaming parsers for large generated worlds. - Add parser telemetry for hackathon demos and debugging. """ from __future__ import annotations import copy import json import logging import re from collections.abc import Mapping, Sequence from dataclasses import dataclass, field from pathlib import Path from typing import Any from pydantic import ValidationError from dsl.schema import ( AgentSpec, BehaviorSpec, EventSpec, MetricSpec, PolicySpec, ResourceSpec, SchemaValidationError, SimulationSpec, SpaceSpec, WorldSpec, ) logger = logging.getLogger(__name__) class DSLParseError(ValueError): """Raised when raw DSL input cannot be parsed into a ``WorldSpec``. The original exception is retained as ``cause`` when available so callers can inspect or log the underlying failure without exposing low-level details in the UI. """ def __init__( self, message: str, *, cause: BaseException | None = None, diagnostics: Mapping[str, Any] | None = None, ) -> None: """Initialize the parsing error.""" super().__init__(message) self.cause = cause self.diagnostics = dict(diagnostics or {}) @dataclass(frozen=True) class ParseResult: """Structured result returned by ``WorldDSLParser.parse_result``. Attributes: spec: Validated world specification. source_kind: Human-readable source kind such as ``json_string`` or ``mapping``. normalized: Whether conservative structural normalization was applied. diagnostics: Parser diagnostics useful for debugging and UI feedback. """ spec: WorldSpec source_kind: str normalized: bool diagnostics: Mapping[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly summary of the parse result.""" return { "world_id": self.spec.id, "world_name": self.spec.name, "schema_version": self.spec.schema_version, "source_kind": self.source_kind, "normalized": self.normalized, "agent_count": len(self.spec.agents), "resource_count": len(self.spec.resources), "event_count": len(self.spec.events), "behavior_count": len(self.spec.behavior_names), "diagnostics": copy.deepcopy(dict(self.diagnostics)), } @dataclass class WorldDSLParser: """Parser for WorldSmithAI DSL input. The parser is intentionally independent from runtime world objects. It can be safely used in the LLM layer, CLI entry points, tests, Gradio callbacks, or batch example loading. """ allow_markdown_fences: bool = True allow_surrounding_text: bool = True normalize_common_shapes: bool = True require_json_object_root: bool = True def parse(self, raw_input: str | bytes | Mapping[str, Any] | WorldSpec) -> WorldSpec: """Parse raw input into a validated ``WorldSpec``. Args: raw_input: A ``WorldSpec``, mapping, JSON string, bytes, Markdown fenced JSON, or text containing a JSON object. Returns: Validated ``WorldSpec``. Raises: DSLParseError: If parsing or schema validation fails. """ return self.parse_result(raw_input).spec def parse_result(self, raw_input: str | bytes | Mapping[str, Any] | WorldSpec) -> ParseResult: """Parse raw input and return a structured ``ParseResult``.""" if isinstance(raw_input, WorldSpec): return ParseResult( spec=raw_input, source_kind="world_spec", normalized=False, diagnostics={"message": "input_already_validated"}, ) if isinstance(raw_input, Mapping): data = copy.deepcopy(dict(raw_input)) normalized_data, normalized = self._normalize_if_enabled(data) spec = self._validate_world_spec(normalized_data, source_kind="mapping") return ParseResult( spec=spec, source_kind="mapping", normalized=normalized, diagnostics={"input_type": "mapping"}, ) if isinstance(raw_input, bytes): try: text = raw_input.decode("utf-8") except UnicodeDecodeError as exc: raise DSLParseError("DSL bytes input must be valid UTF-8", cause=exc) from exc return self._parse_text_result(text, source_kind="bytes") if isinstance(raw_input, str): return self._parse_text_result(raw_input, source_kind="json_string") raise DSLParseError( "Unsupported DSL input type", diagnostics={"input_type": raw_input.__class__.__name__}, ) def parse_json_string(self, raw_json: str) -> WorldSpec: """Parse a JSON string, Markdown fenced JSON, or text containing JSON.""" return self.parse(raw_json) def parse_mapping(self, data: Mapping[str, Any]) -> WorldSpec: """Parse a Python mapping into a validated ``WorldSpec``.""" return self.parse(data) def parse_file(self, path: str | Path) -> WorldSpec: """Parse a JSON DSL file from disk.""" return self.parse_file_result(path).spec def parse_file_result(self, path: str | Path) -> ParseResult: """Parse a JSON DSL file and return a structured parse result.""" file_path = Path(path) try: text = file_path.read_text(encoding="utf-8") except OSError as exc: raise DSLParseError( f"Could not read DSL file: {file_path}", cause=exc, diagnostics={"path": str(file_path)}, ) from exc result = self._parse_text_result(text, source_kind="file") return ParseResult( spec=result.spec, source_kind="file", normalized=result.normalized, diagnostics={ **dict(result.diagnostics), "path": str(file_path), }, ) def extract_json_text(self, text: str) -> str: """Extract JSON text from raw text. The method first tries the full text. If that fails, it optionally checks Markdown code fences and then searches for the first balanced JSON object. """ stripped = text.strip() if not stripped: raise DSLParseError("DSL input is empty") if self._looks_like_json_object(stripped): return stripped if self.allow_markdown_fences: fenced = self._extract_from_markdown_fence(stripped) if fenced is not None: return fenced if self.allow_surrounding_text: balanced = self._extract_first_balanced_json_object(stripped) if balanced is not None: return balanced raise DSLParseError( "Could not find a JSON object in DSL input", diagnostics={ "allow_markdown_fences": self.allow_markdown_fences, "allow_surrounding_text": self.allow_surrounding_text, }, ) def loads(self, text: str) -> dict[str, Any]: """Load JSON text into a mapping. Args: text: Raw JSON object string. Returns: Parsed dictionary. Raises: DSLParseError: If JSON decoding fails or root is not an object. """ try: data = json.loads(text) except json.JSONDecodeError as exc: raise DSLParseError( self._json_error_message(exc), cause=exc, diagnostics={ "line": exc.lineno, "column": exc.colno, "position": exc.pos, }, ) from exc if self.require_json_object_root and not isinstance(data, Mapping): raise DSLParseError( "World DSL root must be a JSON object", diagnostics={"root_type": data.__class__.__name__}, ) if not isinstance(data, Mapping): raise DSLParseError( "World DSL parser expected a mapping root", diagnostics={"root_type": data.__class__.__name__}, ) return dict(data) def _parse_text_result(self, text: str, *, source_kind: str) -> ParseResult: """Parse textual input and return a structured parse result.""" json_text = self.extract_json_text(text) data = self.loads(json_text) normalized_data, normalized = self._normalize_if_enabled(data) spec = self._validate_world_spec(normalized_data, source_kind=source_kind) return ParseResult( spec=spec, source_kind=source_kind, normalized=normalized, diagnostics={ "input_length": len(text), "json_length": len(json_text), "extracted_json": json_text != text.strip(), }, ) def _normalize_if_enabled(self, data: Mapping[str, Any]) -> tuple[dict[str, Any], bool]: """Normalize common SLM output shapes when enabled.""" copied = copy.deepcopy(dict(data)) if not self.normalize_common_shapes: return copied, False normalized = normalize_world_mapping(copied) return normalized, normalized != copied def _validate_world_spec(self, data: Mapping[str, Any], *, source_kind: str) -> WorldSpec: """Validate normalized data as ``WorldSpec``.""" try: return WorldSpec.model_validate(dict(data)) except ValidationError as exc: raise DSLParseError( "World DSL failed schema validation", cause=exc, diagnostics={ "source_kind": source_kind, "errors": _format_pydantic_errors(exc), }, ) from exc except SchemaValidationError as exc: raise DSLParseError( str(exc), cause=exc, diagnostics={"source_kind": source_kind}, ) from exc except ValueError as exc: raise DSLParseError( "World DSL contains invalid values", cause=exc, diagnostics={"source_kind": source_kind}, ) from exc @staticmethod def _looks_like_json_object(text: str) -> bool: """Return whether text appears to be a JSON object.""" return text.startswith("{") and text.endswith("}") @staticmethod def _extract_from_markdown_fence(text: str) -> str | None: """Extract JSON content from the first Markdown fenced block.""" fence_pattern = re.compile( r"```(?:json|JSON|javascript|js|)\s*(?P.*?)```", re.DOTALL, ) match = fence_pattern.search(text) if match is None: return None body = match.group("body").strip() return body or None @staticmethod def _extract_first_balanced_json_object(text: str) -> str | None: """Extract the first balanced JSON object from arbitrary text. This scanner respects JSON strings and escaped characters, so braces inside strings do not break extraction. """ start_index = text.find("{") if start_index < 0: return None depth = 0 in_string = False escaped = False for index in range(start_index, len(text)): char = text[index] if escaped: escaped = False continue if char == "\\" and in_string: escaped = True continue if char == '"': in_string = not in_string continue if in_string: continue if char == "{": depth += 1 elif char == "}": depth -= 1 if depth == 0: return text[start_index : index + 1] return None @staticmethod def _json_error_message(error: json.JSONDecodeError) -> str: """Return a concise JSON parse error message.""" return ( f"Invalid JSON at line {error.lineno}, column {error.colno}: " f"{error.msg}" ) def normalize_world_mapping(data: Mapping[str, Any]) -> dict[str, Any]: """Normalize common SLM-generated DSL shapes. This function is conservative. It does not infer semantics or repair unknown behavior names. It only converts common structural variants into the canonical shape expected by ``WorldSpec``. Supported normalizations: - top-level ``world`` wrapper - singular aliases like ``agent`` -> ``agents`` - mapping collections converted to lists - behavior strings converted to ``{"name": value}`` - policy strings converted to ``{"type": value}`` - common aliases like ``agent_type`` -> ``type`` """ normalized = copy.deepcopy(dict(data)) if isinstance(normalized.get("world"), Mapping): world_wrapper = dict(normalized.pop("world")) for key, value in normalized.items(): world_wrapper.setdefault(key, value) normalized = world_wrapper _apply_top_level_aliases(normalized) normalized["agents"] = _normalize_collection( normalized.get("agents", ()), id_field="id", ) normalized["resources"] = _normalize_collection( normalized.get("resources", ()), id_field="id", ) normalized["events"] = _normalize_collection( normalized.get("events", ()), id_field="id", ) if "metrics" in normalized: normalized["metrics"] = _normalize_collection( normalized.get("metrics", ()), id_field="name", ) normalized["agents"] = [ normalize_agent_mapping(agent) for agent in normalized.get("agents", ()) ] normalized["resources"] = [ normalize_resource_mapping(resource) for resource in normalized.get("resources", ()) ] normalized["events"] = [ normalize_event_mapping(event) for event in normalized.get("events", ()) ] if "metrics" in normalized: normalized["metrics"] = [ normalize_metric_mapping(metric) for metric in normalized.get("metrics", ()) ] if "simulation" in normalized and isinstance(normalized["simulation"], Mapping): normalized["simulation"] = normalize_simulation_mapping(normalized["simulation"]) if "space" in normalized and isinstance(normalized["space"], Mapping): normalized["space"] = normalize_space_mapping(normalized["space"]) if "metadata" in normalized and normalized["metadata"] is None: normalized["metadata"] = {} return normalized def normalize_agent_mapping(agent: Mapping[str, Any]) -> dict[str, Any]: """Normalize one agent mapping into canonical schema shape.""" normalized = copy.deepcopy(dict(agent)) _rename_key(normalized, "agent_id", "id") _rename_key(normalized, "agent_type", "type") _rename_key(normalized, "kind", "type") _rename_key(normalized, "location", "position") _rename_key(normalized, "pos", "position") if "state" not in normalized: normalized["state"] = {} if "memory" not in normalized: normalized["memory"] = {} if "goals" not in normalized: normalized["goals"] = [] if "behaviors" not in normalized: normalized["behaviors"] = [] normalized["behaviors"] = [ normalize_behavior_spec(behavior) for behavior in _normalize_collection( normalized.get("behaviors", ()), id_field="name", ) ] policy = normalized.get("policy") if policy is not None: normalized["policy"] = normalize_policy_spec(policy) if "metadata" in normalized and normalized["metadata"] is None: normalized["metadata"] = {} return normalized def normalize_resource_mapping(resource: Mapping[str, Any]) -> dict[str, Any]: """Normalize one resource mapping into canonical schema shape.""" normalized = copy.deepcopy(dict(resource)) _rename_key(normalized, "resource_id", "id") _rename_key(normalized, "resource_type", "type") _rename_key(normalized, "kind", "type") _rename_key(normalized, "quantity", "amount") _rename_key(normalized, "location", "position") _rename_key(normalized, "pos", "position") _rename_key(normalized, "regen_rate", "regeneration_rate") _rename_key(normalized, "capacity", "max_amount") if "metadata" in normalized and normalized["metadata"] is None: normalized["metadata"] = {} return normalized def normalize_event_mapping(event: Mapping[str, Any]) -> dict[str, Any]: """Normalize one event mapping into canonical schema shape.""" normalized = copy.deepcopy(dict(event)) _rename_key(normalized, "event_id", "id") _rename_key(normalized, "type", "name") _rename_key(normalized, "step", "trigger_step") _rename_key(normalized, "at_step", "trigger_step") _rename_key(normalized, "data", "payload") if "payload" not in normalized: normalized["payload"] = {} if "targets" in normalized: targets = normalized.pop("targets") if isinstance(targets, Mapping): normalized.setdefault("target_agent_ids", targets.get("agents", targets.get("agent_ids", ()))) normalized.setdefault("target_resource_ids", targets.get("resources", targets.get("resource_ids", ()))) elif isinstance(targets, Sequence) and not isinstance(targets, (str, bytes)): normalized.setdefault("target_agent_ids", list(targets)) if "metadata" in normalized and normalized["metadata"] is None: normalized["metadata"] = {} return normalized def normalize_metric_mapping(metric: Mapping[str, Any]) -> dict[str, Any]: """Normalize one metric mapping into canonical schema shape.""" normalized = copy.deepcopy(dict(metric)) _rename_key(normalized, "type", "name") _rename_key(normalized, "metric", "name") if "params" not in normalized: params = { key: value for key, value in normalized.items() if key not in {"name", "enabled", "metadata"} } if params: normalized = { "name": normalized.get("name"), "enabled": normalized.get("enabled", True), "metadata": normalized.get("metadata", {}), "params": params, } if "metadata" in normalized and normalized["metadata"] is None: normalized["metadata"] = {} return normalized def normalize_behavior_spec(behavior: Any) -> dict[str, Any]: """Normalize behavior input into a canonical behavior spec mapping.""" if isinstance(behavior, str): return {"name": behavior, "params": {}} if not isinstance(behavior, Mapping): raise DSLParseError( "Behavior entries must be strings or objects", diagnostics={"behavior_type": behavior.__class__.__name__}, ) normalized = copy.deepcopy(dict(behavior)) _rename_key(normalized, "type", "name") _rename_key(normalized, "behavior", "name") _rename_key(normalized, "config", "params") _rename_key(normalized, "kwargs", "params") if "params" not in normalized: reserved_keys = {"name", "enabled", "priority", "tags", "metadata"} params = { key: value for key, value in normalized.items() if key not in reserved_keys } if params and "name" in normalized: normalized = { "name": normalized["name"], "params": params, "enabled": normalized.get("enabled", True), "priority": normalized.get("priority", 0.0), "tags": normalized.get("tags", ()), "metadata": normalized.get("metadata", {}), } else: normalized["params"] = {} if "metadata" in normalized and normalized["metadata"] is None: normalized["metadata"] = {} return normalized def normalize_policy_spec(policy: Any) -> dict[str, Any]: """Normalize policy input into a canonical policy spec mapping.""" if isinstance(policy, str): return {"type": policy, "params": {}} if not isinstance(policy, Mapping): raise DSLParseError( "Policy must be a string or object", diagnostics={"policy_type": policy.__class__.__name__}, ) normalized = copy.deepcopy(dict(policy)) _rename_key(normalized, "name", "type") _rename_key(normalized, "policy_type", "type") _rename_key(normalized, "config", "params") _rename_key(normalized, "kwargs", "params") if "params" not in normalized: reserved_keys = {"type", "enabled", "metadata"} params = { key: value for key, value in normalized.items() if key not in reserved_keys } if params and "type" in normalized: normalized = { "type": normalized["type"], "params": params, "enabled": normalized.get("enabled", True), "metadata": normalized.get("metadata", {}), } else: normalized["params"] = {} if "metadata" in normalized and normalized["metadata"] is None: normalized["metadata"] = {} return normalized def normalize_simulation_mapping(simulation: Mapping[str, Any]) -> dict[str, Any]: """Normalize simulation configuration aliases.""" normalized = copy.deepcopy(dict(simulation)) _rename_key(normalized, "num_steps", "steps") _rename_key(normalized, "n_steps", "steps") _rename_key(normalized, "random_seed", "seed") _rename_key(normalized, "activation_mode", "activation") if "metadata" in normalized and normalized["metadata"] is None: normalized["metadata"] = {} return normalized def normalize_space_mapping(space: Mapping[str, Any]) -> dict[str, Any]: """Normalize space configuration aliases.""" normalized = copy.deepcopy(dict(space)) _rename_key(normalized, "dim", "dimensions") _rename_key(normalized, "dims", "dimensions") _rename_key(normalized, "wrap", "toroidal") _rename_key(normalized, "wraparound", "toroidal") if "metadata" in normalized and normalized["metadata"] is None: normalized["metadata"] = {} return normalized def parse_world_spec(raw_input: str | bytes | Mapping[str, Any] | WorldSpec) -> WorldSpec: """Parse raw input into a validated ``WorldSpec`` using default parser settings.""" return WorldDSLParser().parse(raw_input) def parse_world_spec_result(raw_input: str | bytes | Mapping[str, Any] | WorldSpec) -> ParseResult: """Parse raw input into a structured ``ParseResult`` using default settings.""" return WorldDSLParser().parse_result(raw_input) def parse_world_json(raw_json: str) -> WorldSpec: """Parse a raw JSON string or SLM response into ``WorldSpec``.""" return WorldDSLParser().parse_json_string(raw_json) def parse_world_file(path: str | Path) -> WorldSpec: """Parse a world DSL JSON file into ``WorldSpec``.""" return WorldDSLParser().parse_file(path) def world_spec_to_json(spec: WorldSpec, *, indent: int = 2, exclude_none: bool = True) -> str: """Serialize a ``WorldSpec`` to a JSON string.""" return spec.to_json_string(indent=indent, exclude_none=exclude_none) def world_spec_to_dict(spec: WorldSpec, *, exclude_none: bool = True) -> dict[str, Any]: """Serialize a ``WorldSpec`` to a JSON-friendly dictionary.""" return spec.to_dict(exclude_none=exclude_none) def _normalize_collection(value: Any, *, id_field: str) -> list[Any]: """Normalize list-like or mapping-like DSL collections. If a collection is supplied as a mapping, values become entries and the mapping key is used as ``id_field`` when the entry does not already define one. """ if value is None: return [] if isinstance(value, Mapping): items: list[Any] = [] for key in sorted(value.keys(), key=str): item = copy.deepcopy(value[key]) if isinstance(item, Mapping): mapped_item = dict(item) mapped_item.setdefault(id_field, str(key)) items.append(mapped_item) else: items.append({id_field: str(key), "value": item}) return items if isinstance(value, Sequence) and not isinstance(value, (str, bytes)): return list(value) return [value] def _apply_top_level_aliases(data: dict[str, Any]) -> None: """Apply conservative aliases to top-level world data.""" _rename_key(data, "world_id", "id") _rename_key(data, "title", "name") _rename_key(data, "config", "simulation") if "agent" in data and "agents" not in data: data["agents"] = [data.pop("agent")] if "resource" in data and "resources" not in data: data["resources"] = [data.pop("resource")] if "event" in data and "events" not in data: data["events"] = [data.pop("event")] def _rename_key(data: dict[str, Any], old_key: str, new_key: str) -> None: """Rename a key if present and the destination is absent.""" if old_key in data and new_key not in data: data[new_key] = data.pop(old_key) def _format_pydantic_errors(error: ValidationError) -> list[dict[str, Any]]: """Convert Pydantic validation errors into compact diagnostics.""" formatted: list[dict[str, Any]] = [] for item in error.errors(): location = ".".join(str(part) for part in item.get("loc", ())) formatted.append( { "path": location, "message": item.get("msg"), "type": item.get("type"), "input": _safe_error_input(item.get("input")), } ) return formatted def _safe_error_input(value: Any) -> Any: """Return a small JSON-friendly representation of invalid input.""" if value is None or isinstance(value, (str, int, float, bool)): return value if isinstance(value, Mapping): keys = list(value.keys()) return {"type": "object", "keys": [str(key) for key in keys[:10]]} if isinstance(value, Sequence) and not isinstance(value, (str, bytes)): return {"type": "array", "length": len(value)} return {"type": value.__class__.__name__, "repr": repr(value)[:200]} __all__ = [ "DSLParseError", "ParseResult", "WorldDSLParser", "parse_world_file", "parse_world_json", "parse_world_spec", "parse_world_spec_result", "world_spec_to_dict", "world_spec_to_json", "normalize_agent_mapping", "normalize_behavior_spec", "normalize_event_mapping", "normalize_metric_mapping", "normalize_policy_spec", "normalize_resource_mapping", "normalize_simulation_mapping", "normalize_space_mapping", "normalize_world_mapping", ]