""" Pydantic schemas for the WorldSmithAI DSL. This module defines the canonical JSON contract used by WorldSmithAI. The SLM is expected to generate JSON that validates into ``WorldSpec``. The runtime simulation engine should consume validated schemas, not raw untrusted JSON. The schema is intentionally domain-agnostic. It does not contain classes such as SheepSpec, WolfSpec, FarmSpec, ResearchSpec, CitySpec, DragonSpec, or MarketSpec. Every world is represented through generic agents, resources, events, behaviors, policies, and metadata. Example: raw_world = { "id": "research_ecosystem", "name": "Tiny Research Ecosystem", "simulation": {"steps": 50, "seed": 7}, "space": {"dimensions": 2, "bounds": [[0, 10], [0, 10]]}, "agents": [ { "id": "scientist_1", "type": "scientist", "position": [1, 1], "state": {"energy": 10, "knowledge": 1}, "memory": {}, "goals": [{"id": "publish", "importance": 2}], "behaviors": [ {"name": "study", "params": {"knowledge_gain": 1}}, {"name": "publish", "params": {"threshold": 5}} ], "policy": { "type": "rule_policy", "params": { "rules": [ {"behavior_name": "study", "score_delta": 1.0} ] } } } ], "resources": [], "events": [] } world_spec = WorldSpec.model_validate(raw_world) Future extensibility: - Add schema migrations between DSL versions. - Add formal JSON Schema export for prompt constraints. - Add registry-aware semantic validation in ``dsl.validator``. - Add DSL fragments for reusable templates. - Add compatibility profiles for small language models. - Add provenance fields for tracing generated worlds back to prompts. """ from __future__ import annotations import json import logging import math from collections.abc import Iterable, Mapping, Sequence from typing import Any, ClassVar, TypeAlias from pydantic import AliasChoices, BaseModel, ConfigDict, Field, field_validator, model_validator logger = logging.getLogger(__name__) JsonPrimitive: TypeAlias = str | int | float | bool | None JsonMapping: TypeAlias = dict[str, Any] JsonArray: TypeAlias = list[Any] JsonValue: TypeAlias = JsonPrimitive | JsonMapping | JsonArray DEFAULT_SCHEMA_VERSION = "1.0" class SchemaValidationError(ValueError): """Raised when a DSL value violates WorldSmithAI schema expectations.""" def _is_json_number(value: Any) -> bool: """Return whether a value is a JSON-compatible number. Booleans are excluded because ``bool`` is a subclass of ``int`` in Python. """ return isinstance(value, (int, float)) and not isinstance(value, bool) def _validate_json_compatible(value: Any, *, path: str = "value") -> None: """Validate that a value can be safely represented as JSON. Args: value: Candidate JSON value. path: Human-readable path used in error messages. Raises: SchemaValidationError: If the value is not JSON-compatible. """ if value is None or isinstance(value, (str, bool)): return if _is_json_number(value): if not math.isfinite(float(value)): raise SchemaValidationError(f"{path} contains a non-finite number: {value!r}") return if isinstance(value, Mapping): for key, nested_value in value.items(): if not isinstance(key, str): raise SchemaValidationError( f"{path} contains a non-string object key: {key!r}" ) _validate_json_compatible(nested_value, path=f"{path}.{key}") return if isinstance(value, list): for index, nested_value in enumerate(value): _validate_json_compatible(nested_value, path=f"{path}[{index}]") return raise SchemaValidationError( f"{path} contains a non-JSON-compatible value of type " f"{value.__class__.__name__}: {value!r}" ) def _validate_json_mapping(value: Mapping[str, Any], *, path: str) -> dict[str, Any]: """Validate and copy a JSON-compatible mapping.""" copied = dict(value) _validate_json_compatible(copied, path=path) return copied def _non_empty_string(value: str, *, field_name: str) -> str: """Validate and normalize a non-empty string field.""" normalized = str(value).strip() if not normalized: raise SchemaValidationError(f"{field_name} must not be empty") return normalized def _normalize_tags(value: Iterable[Any]) -> tuple[str, ...]: """Normalize a tag collection into a deterministic tuple of strings.""" tags = tuple(str(tag).strip() for tag in value if str(tag).strip()) return tuple(dict.fromkeys(tags)) def _validate_position(value: Sequence[float] | None, *, field_name: str) -> tuple[float, ...] | None: """Validate and normalize a spatial position.""" if value is None: return None if isinstance(value, (str, bytes)) or not isinstance(value, Sequence): raise SchemaValidationError(f"{field_name} must be a numeric sequence") if not value: raise SchemaValidationError(f"{field_name} must contain at least one component") normalized: list[float] = [] for index, component in enumerate(value): numeric_component = float(component) if not math.isfinite(numeric_component): raise SchemaValidationError( f"{field_name}[{index}] must be finite, got {component!r}" ) normalized.append(numeric_component) return tuple(normalized) def _validate_identifier_collection(values: Sequence[str], *, field_name: str) -> tuple[str, ...]: """Validate a sequence of non-empty string identifiers.""" normalized: list[str] = [] for value in values: normalized.append(_non_empty_string(str(value), field_name=field_name)) return tuple(normalized) class SchemaModel(BaseModel): """Base model for all WorldSmithAI DSL schemas. Common configuration: - Extra fields are forbidden to catch SLM hallucinated keys early. - Assignment validation is enabled to keep models valid after mutation. - Population by field name is enabled for ergonomic Python construction. """ model_config: ClassVar[ConfigDict] = ConfigDict( extra="forbid", populate_by_name=True, validate_assignment=True, arbitrary_types_allowed=False, ) def to_dict(self, *, exclude_none: bool = True) -> dict[str, Any]: """Return a JSON-friendly dictionary representation of this model.""" return self.model_dump(mode="json", exclude_none=exclude_none) def to_json_string(self, *, indent: int = 2, exclude_none: bool = True) -> str: """Return a JSON string representation of this model.""" return self.model_dump_json(indent=indent, exclude_none=exclude_none) @classmethod def from_mapping(cls, data: Mapping[str, Any]) -> Any: """Validate a model from a mapping. Args: data: Mapping to validate. Returns: An instance of the concrete schema model. """ return cls.model_validate(dict(data)) class BehaviorSpec(SchemaModel): """Declarative behavior specification. ``BehaviorSpec`` does not know how to execute a behavior. It only describes which behavior should be instantiated and with what parameters. The ``WorldFactory`` is responsible for resolving ``name`` against behavior registries. """ name: str = Field(..., description="Registry name of the behavior to instantiate.") params: JsonMapping = Field( default_factory=dict, description="Constructor parameters for the behavior.", ) enabled: bool = Field(default=True, description="Whether this behavior is active.") priority: float = Field( default=0.0, description="Optional ordering hint for policies or factories.", ) tags: tuple[str, ...] = Field( default_factory=tuple, description="Optional tags for behavior grouping or analysis.", ) metadata: JsonMapping = Field( default_factory=dict, description="Additional JSON-compatible behavior metadata.", ) @field_validator("name") @classmethod def validate_name(cls, value: str) -> str: """Validate the behavior name.""" return _non_empty_string(value, field_name="BehaviorSpec.name") @field_validator("params") @classmethod def validate_params(cls, value: Mapping[str, Any]) -> dict[str, Any]: """Validate behavior parameters.""" return _validate_json_mapping(value, path="BehaviorSpec.params") @field_validator("metadata") @classmethod def validate_metadata(cls, value: Mapping[str, Any]) -> dict[str, Any]: """Validate behavior metadata.""" return _validate_json_mapping(value, path="BehaviorSpec.metadata") @field_validator("tags", mode="before") @classmethod def validate_tags(cls, value: Any) -> tuple[str, ...]: """Normalize behavior tags.""" if value is None: return () if isinstance(value, str): return _normalize_tags((value,)) if isinstance(value, Iterable): return _normalize_tags(value) raise SchemaValidationError("BehaviorSpec.tags must be a string or iterable of strings") @property def behavior_key(self) -> str: """Return the behavior registry key.""" return self.name class PolicySpec(SchemaModel): """Declarative policy specification. The field ``type`` is intentionally generic. It may reference ``rule_policy``, ``contextual_bandit``, or future policies. """ type: str = Field( default="rule_policy", validation_alias=AliasChoices("type", "name", "policy_type"), description="Registry name of the policy to instantiate.", ) params: JsonMapping = Field( default_factory=dict, description="Constructor parameters for the policy.", ) enabled: bool = Field(default=True, description="Whether the policy is active.") metadata: JsonMapping = Field( default_factory=dict, description="Additional JSON-compatible policy metadata.", ) @field_validator("type") @classmethod def validate_type(cls, value: str) -> str: """Validate policy type.""" return _non_empty_string(value, field_name="PolicySpec.type") @field_validator("params") @classmethod def validate_params(cls, value: Mapping[str, Any]) -> dict[str, Any]: """Validate policy parameters.""" return _validate_json_mapping(value, path="PolicySpec.params") @field_validator("metadata") @classmethod def validate_metadata(cls, value: Mapping[str, Any]) -> dict[str, Any]: """Validate policy metadata.""" return _validate_json_mapping(value, path="PolicySpec.metadata") @property def policy_key(self) -> str: """Return the policy registry key.""" return self.type class AgentSpec(SchemaModel): """Declarative agent specification. Agents are generic entities. Their domain identity is expressed through the free-form ``type`` string, state, memory, goals, behaviors, and policy. """ id: str = Field(..., description="Unique agent identifier within the world.") type: str = Field(..., description="Generic agent type label.") position: tuple[float, ...] | None = Field( default=None, description="Optional numeric spatial position.", ) state: JsonMapping = Field( default_factory=dict, description="Mutable runtime state initialized on the agent.", ) memory: JsonMapping = Field( default_factory=dict, description="Mutable memory initialized on the agent.", ) goals: JsonValue = Field( default_factory=list, description="JSON-compatible goal description or collection.", ) behaviors: tuple[BehaviorSpec, ...] = Field( default_factory=tuple, description="Behavior specs attached to this agent.", ) policy: PolicySpec | None = Field( default=None, description="Optional policy used to select behaviors.", ) alive: bool = Field(default=True, description="Initial lifecycle flag.") metadata: JsonMapping = Field( default_factory=dict, description="Additional JSON-compatible agent metadata.", ) @field_validator("id") @classmethod def validate_id(cls, value: str) -> str: """Validate agent id.""" return _non_empty_string(value, field_name="AgentSpec.id") @field_validator("type") @classmethod def validate_type(cls, value: str) -> str: """Validate agent type.""" return _non_empty_string(value, field_name="AgentSpec.type") @field_validator("position", mode="before") @classmethod def validate_position(cls, value: Any) -> tuple[float, ...] | None: """Validate agent position.""" return _validate_position(value, field_name="AgentSpec.position") @field_validator("state") @classmethod def validate_state(cls, value: Mapping[str, Any]) -> dict[str, Any]: """Validate initial state.""" return _validate_json_mapping(value, path="AgentSpec.state") @field_validator("memory") @classmethod def validate_memory(cls, value: Mapping[str, Any]) -> dict[str, Any]: """Validate initial memory.""" return _validate_json_mapping(value, path="AgentSpec.memory") @field_validator("goals") @classmethod def validate_goals(cls, value: Any) -> Any: """Validate JSON-compatible goals.""" _validate_json_compatible(value, path="AgentSpec.goals") return value @field_validator("metadata") @classmethod def validate_metadata(cls, value: Mapping[str, Any]) -> dict[str, Any]: """Validate agent metadata.""" return _validate_json_mapping(value, path="AgentSpec.metadata") @property def enabled_behaviors(self) -> tuple[BehaviorSpec, ...]: """Return behavior specs that are enabled.""" return tuple(behavior for behavior in self.behaviors if behavior.enabled) @property def behavior_names(self) -> tuple[str, ...]: """Return all configured behavior names.""" return tuple(behavior.name for behavior in self.behaviors) class ResourceSpec(SchemaModel): """Declarative world resource specification. Resources are generic quantities such as food, grass, money, knowledge, mana, fuel, compute, attention, housing, energy, or any DSL-defined asset. """ id: str = Field(..., description="Unique resource identifier within the world.") type: str = Field(..., description="Generic resource type label.") amount: float = Field(default=0.0, ge=0.0, description="Initial resource amount.") position: tuple[float, ...] | None = Field( default=None, description="Optional numeric spatial position.", ) regeneration_rate: float = Field( default=0.0, ge=0.0, description="Optional deterministic regeneration amount per resource update.", ) max_amount: float | None = Field( default=None, ge=0.0, description="Optional maximum amount after regeneration.", ) metadata: JsonMapping = Field( default_factory=dict, description="Additional JSON-compatible resource metadata.", ) @field_validator("id") @classmethod def validate_id(cls, value: str) -> str: """Validate resource id.""" return _non_empty_string(value, field_name="ResourceSpec.id") @field_validator("type") @classmethod def validate_type(cls, value: str) -> str: """Validate resource type.""" return _non_empty_string(value, field_name="ResourceSpec.type") @field_validator("position", mode="before") @classmethod def validate_position(cls, value: Any) -> tuple[float, ...] | None: """Validate resource position.""" return _validate_position(value, field_name="ResourceSpec.position") @field_validator("metadata") @classmethod def validate_metadata(cls, value: Mapping[str, Any]) -> dict[str, Any]: """Validate resource metadata.""" return _validate_json_mapping(value, path="ResourceSpec.metadata") @model_validator(mode="after") def validate_amount_bounds(self) -> ResourceSpec: """Validate amount and maximum amount consistency.""" if self.max_amount is not None and self.amount > self.max_amount: raise SchemaValidationError( f"ResourceSpec.amount for resource {self.id!r} cannot exceed max_amount" ) return self class EventSpec(SchemaModel): """Declarative event specification. Events are generic scheduled effects. The runtime event implementation may interpret ``payload`` as state updates, resource changes, narrative signals, behavior triggers, or future event-system commands. """ id: str | None = Field( default=None, description="Optional stable event identifier.", ) name: str = Field(..., description="Event name.") trigger_step: int = Field( ..., ge=0, description="Simulation step on which the event should trigger.", ) payload: JsonMapping = Field( default_factory=dict, description="JSON-compatible event payload.", ) enabled: bool = Field(default=True, description="Whether this event is active.") repeat_interval: int | None = Field( default=None, gt=0, description="Optional repeat interval in steps.", ) target_agent_ids: tuple[str, ...] = Field( default_factory=tuple, description="Optional target agent ids.", ) target_resource_ids: tuple[str, ...] = Field( default_factory=tuple, description="Optional target resource ids.", ) metadata: JsonMapping = Field( default_factory=dict, description="Additional JSON-compatible event metadata.", ) @field_validator("id") @classmethod def validate_id(cls, value: str | None) -> str | None: """Validate optional event id.""" if value is None: return None return _non_empty_string(value, field_name="EventSpec.id") @field_validator("name") @classmethod def validate_name(cls, value: str) -> str: """Validate event name.""" return _non_empty_string(value, field_name="EventSpec.name") @field_validator("payload") @classmethod def validate_payload(cls, value: Mapping[str, Any]) -> dict[str, Any]: """Validate event payload.""" return _validate_json_mapping(value, path="EventSpec.payload") @field_validator("target_agent_ids", "target_resource_ids", mode="before") @classmethod def validate_target_ids(cls, value: Any) -> tuple[str, ...]: """Validate target identifiers.""" if value is None: return () if isinstance(value, str): return (_non_empty_string(value, field_name="EventSpec.target_ids"),) if isinstance(value, Iterable): return _validate_identifier_collection(tuple(value), field_name="EventSpec.target_ids") raise SchemaValidationError("EventSpec target ids must be a string or iterable of strings") @field_validator("metadata") @classmethod def validate_metadata(cls, value: Mapping[str, Any]) -> dict[str, Any]: """Validate event metadata.""" return _validate_json_mapping(value, path="EventSpec.metadata") @property def event_key(self) -> str: """Return a stable key for duplicate detection and event logs.""" return self.id or self.name class MetricSpec(SchemaModel): """Declarative metric specification. Metric specs are optional. They allow the DSL to request metrics without coupling this schema module to concrete metric implementations. """ name: str = Field(..., description="Metric registry name.") params: JsonMapping = Field(default_factory=dict, description="Metric parameters.") enabled: bool = Field(default=True, description="Whether this metric is active.") metadata: JsonMapping = Field(default_factory=dict, description="Metric metadata.") @field_validator("name") @classmethod def validate_name(cls, value: str) -> str: """Validate metric name.""" return _non_empty_string(value, field_name="MetricSpec.name") @field_validator("params") @classmethod def validate_params(cls, value: Mapping[str, Any]) -> dict[str, Any]: """Validate metric parameters.""" return _validate_json_mapping(value, path="MetricSpec.params") @field_validator("metadata") @classmethod def validate_metadata(cls, value: Mapping[str, Any]) -> dict[str, Any]: """Validate metric metadata.""" return _validate_json_mapping(value, path="MetricSpec.metadata") class SimulationSpec(SchemaModel): """Simulation-level configuration.""" steps: int = Field(default=100, ge=0, description="Number of simulation steps.") seed: int | None = Field( default=0, description="Optional deterministic seed used by policies or schedulers.", ) scheduler: str = Field( default="sequential", description="Scheduler registry key or activation mode.", ) activation: str = Field( default="sequential", description="Agent activation strategy hint.", ) collect_history: bool = Field( default=True, description="Whether runtime should collect basic history when supported.", ) metadata: JsonMapping = Field( default_factory=dict, description="Additional JSON-compatible simulation metadata.", ) @field_validator("scheduler") @classmethod def validate_scheduler(cls, value: str) -> str: """Validate scheduler name.""" return _non_empty_string(value, field_name="SimulationSpec.scheduler") @field_validator("activation") @classmethod def validate_activation(cls, value: str) -> str: """Validate activation name.""" return _non_empty_string(value, field_name="SimulationSpec.activation") @field_validator("metadata") @classmethod def validate_metadata(cls, value: Mapping[str, Any]) -> dict[str, Any]: """Validate simulation metadata.""" return _validate_json_mapping(value, path="SimulationSpec.metadata") class SpaceSpec(SchemaModel): """Optional spatial configuration for worlds with positions. The engine remains capable of non-spatial worlds. When ``space`` is absent, positions may still be used opportunistically by behaviors, visualizers, or metrics. """ dimensions: int = Field(default=2, ge=1, le=16, description="Number of spatial dimensions.") bounds: tuple[tuple[float, float], ...] | None = Field( default=None, description="Optional inclusive bounds per dimension.", ) toroidal: bool = Field( default=False, description="Whether movement may wrap around boundaries.", ) enforce_bounds: bool = Field( default=True, description="Whether schema validation should require positions within bounds.", ) metadata: JsonMapping = Field( default_factory=dict, description="Additional JSON-compatible spatial metadata.", ) @field_validator("bounds", mode="before") @classmethod def validate_bounds(cls, value: Any) -> tuple[tuple[float, float], ...] | None: """Validate spatial bounds.""" if value is None: return None if isinstance(value, (str, bytes)) or not isinstance(value, Sequence): raise SchemaValidationError("SpaceSpec.bounds must be a sequence of [min, max] pairs") normalized_bounds: list[tuple[float, float]] = [] for index, raw_pair in enumerate(value): if ( isinstance(raw_pair, (str, bytes)) or not isinstance(raw_pair, Sequence) or len(raw_pair) != 2 ): raise SchemaValidationError( f"SpaceSpec.bounds[{index}] must contain exactly two numeric values" ) lower = float(raw_pair[0]) upper = float(raw_pair[1]) if not math.isfinite(lower) or not math.isfinite(upper): raise SchemaValidationError( f"SpaceSpec.bounds[{index}] values must be finite" ) if lower >= upper: raise SchemaValidationError( f"SpaceSpec.bounds[{index}] lower bound must be less than upper bound" ) normalized_bounds.append((lower, upper)) return tuple(normalized_bounds) @field_validator("metadata") @classmethod def validate_metadata(cls, value: Mapping[str, Any]) -> dict[str, Any]: """Validate spatial metadata.""" return _validate_json_mapping(value, path="SpaceSpec.metadata") @model_validator(mode="after") def validate_dimension_consistency(self) -> SpaceSpec: """Validate that bounds match dimensionality.""" if self.bounds is not None and len(self.bounds) != self.dimensions: raise SchemaValidationError( "SpaceSpec.bounds length must equal SpaceSpec.dimensions" ) return self class WorldSpec(SchemaModel): """Top-level WorldSmithAI DSL specification. ``WorldSpec`` is the validated representation consumed by the ``WorldFactory``. It describes the initial world and simulation configuration, but it does not execute simulation logic. """ schema_version: str = Field( default=DEFAULT_SCHEMA_VERSION, description="WorldSmithAI DSL schema version.", ) id: str = Field(default="world", description="Unique world identifier.") name: str = Field(default="World", description="Human-readable world name.") description: str | None = Field(default=None, description="Optional world description.") simulation: SimulationSpec = Field( default_factory=SimulationSpec, description="Simulation configuration.", ) space: SpaceSpec | None = Field( default=None, description="Optional spatial configuration.", ) agents: tuple[AgentSpec, ...] = Field( default_factory=tuple, description="Initial agent specifications.", ) resources: tuple[ResourceSpec, ...] = Field( default_factory=tuple, description="Initial resource specifications.", ) events: tuple[EventSpec, ...] = Field( default_factory=tuple, description="Scheduled event specifications.", ) metrics: tuple[MetricSpec, ...] = Field( default_factory=tuple, description="Optional metric specifications.", ) metadata: JsonMapping = Field( default_factory=dict, description="Additional JSON-compatible world metadata.", ) @field_validator("schema_version") @classmethod def validate_schema_version(cls, value: str) -> str: """Validate schema version.""" return _non_empty_string(value, field_name="WorldSpec.schema_version") @field_validator("id") @classmethod def validate_id(cls, value: str) -> str: """Validate world id.""" return _non_empty_string(value, field_name="WorldSpec.id") @field_validator("name") @classmethod def validate_name(cls, value: str) -> str: """Validate world name.""" return _non_empty_string(value, field_name="WorldSpec.name") @field_validator("metadata") @classmethod def validate_metadata(cls, value: Mapping[str, Any]) -> dict[str, Any]: """Validate world metadata.""" return _validate_json_mapping(value, path="WorldSpec.metadata") @model_validator(mode="after") def validate_world_integrity(self) -> WorldSpec: """Validate uniqueness and optional spatial consistency.""" self._validate_unique_agent_ids() self._validate_unique_resource_ids() self._validate_unique_event_keys() self._validate_event_targets() self._validate_spatial_positions() return self @classmethod def from_json_string(cls, raw_json: str) -> WorldSpec: """Validate a world specification from a JSON string.""" data = json.loads(raw_json) if not isinstance(data, Mapping): raise SchemaValidationError("WorldSpec JSON root must be an object") return cls.model_validate(data) @classmethod def from_json_file(cls, path: str) -> WorldSpec: """Validate a world specification from a JSON file path.""" with open(path, "r", encoding="utf-8") as file: return cls.from_json_string(file.read()) def to_json_file(self, path: str, *, indent: int = 2, exclude_none: bool = True) -> None: """Write this world specification to a JSON file.""" with open(path, "w", encoding="utf-8") as file: file.write(self.to_json_string(indent=indent, exclude_none=exclude_none)) file.write("\n") @property def agent_ids(self) -> tuple[str, ...]: """Return all agent ids.""" return tuple(agent.id for agent in self.agents) @property def resource_ids(self) -> tuple[str, ...]: """Return all resource ids.""" return tuple(resource.id for resource in self.resources) @property def event_keys(self) -> tuple[str, ...]: """Return stable event keys.""" return tuple(event.event_key for event in self.events) @property def behavior_names(self) -> tuple[str, ...]: """Return all behavior names used by all agents.""" names: list[str] = [] for agent in self.agents: names.extend(agent.behavior_names) return tuple(names) @property def policy_types(self) -> tuple[str, ...]: """Return all policy types used by agents.""" return tuple( agent.policy.type for agent in self.agents if agent.policy is not None ) def get_agent(self, agent_id: str) -> AgentSpec | None: """Return an agent spec by id, if present.""" for agent in self.agents: if agent.id == agent_id: return agent return None def get_resource(self, resource_id: str) -> ResourceSpec | None: """Return a resource spec by id, if present.""" for resource in self.resources: if resource.id == resource_id: return resource return None def _validate_unique_agent_ids(self) -> None: """Validate that agent ids are unique.""" duplicates = _duplicates(self.agent_ids) if duplicates: raise SchemaValidationError(f"Duplicate agent ids: {sorted(duplicates)}") def _validate_unique_resource_ids(self) -> None: """Validate that resource ids are unique.""" duplicates = _duplicates(self.resource_ids) if duplicates: raise SchemaValidationError(f"Duplicate resource ids: {sorted(duplicates)}") def _validate_unique_event_keys(self) -> None: """Validate that event keys are unique.""" duplicates = _duplicates(self.event_keys) if duplicates: raise SchemaValidationError(f"Duplicate event keys: {sorted(duplicates)}") def _validate_event_targets(self) -> None: """Validate that event targets reference known ids.""" agent_ids = set(self.agent_ids) resource_ids = set(self.resource_ids) for event in self.events: missing_agents = [ agent_id for agent_id in event.target_agent_ids if agent_id not in agent_ids ] if missing_agents: raise SchemaValidationError( f"Event {event.event_key!r} references unknown agents: {missing_agents}" ) missing_resources = [ resource_id for resource_id in event.target_resource_ids if resource_id not in resource_ids ] if missing_resources: raise SchemaValidationError( f"Event {event.event_key!r} references unknown resources: {missing_resources}" ) def _validate_spatial_positions(self) -> None: """Validate positions against optional spatial configuration.""" if self.space is None: return for agent in self.agents: if agent.position is not None: self._validate_position_against_space( agent.position, label=f"Agent {agent.id!r}", ) for resource in self.resources: if resource.position is not None: self._validate_position_against_space( resource.position, label=f"Resource {resource.id!r}", ) def _validate_position_against_space(self, position: tuple[float, ...], *, label: str) -> None: """Validate one position against the world space settings.""" if self.space is None: return if len(position) != self.space.dimensions: raise SchemaValidationError( f"{label} position has {len(position)} dimensions, " f"but world space expects {self.space.dimensions}" ) if self.space.bounds is None or not self.space.enforce_bounds: return for index, component in enumerate(position): lower, upper = self.space.bounds[index] if component < lower or component > upper: raise SchemaValidationError( f"{label} position component {index}={component} is outside " f"bounds [{lower}, {upper}]" ) def _duplicates(values: Sequence[str]) -> set[str]: """Return duplicate values in a sequence.""" seen: set[str] = set() duplicates: set[str] = set() for value in values: if value in seen: duplicates.add(value) seen.add(value) return duplicates __all__ = [ "AgentSpec", "BehaviorSpec", "DEFAULT_SCHEMA_VERSION", "EventSpec", "JsonArray", "JsonMapping", "JsonPrimitive", "JsonValue", "MetricSpec", "PolicySpec", "ResourceSpec", "SchemaModel", "SchemaValidationError", "SimulationSpec", "SpaceSpec", "WorldSpec", ]