""" Runtime object factory for WorldSmithAI. This module converts validated DSL specifications into live runtime objects: World, Agent, Resource, Event, Behavior, Policy, and Scheduler instances. The factory is intentionally domain-agnostic. It does not know about sheep, wolves, farms, dragons, scientists, markets, civilizations, power grids, transport networks, or research ecosystems. It only resolves generic DSL records through registries and runtime constructors. Important compatibility notes: - DSL positions are JSON arrays, but the runtime world may expect NumPy arrays with attributes such as ``.size``. - Runtime Agent and Resource positions are always converted to NumPy arrays. - Runtime behavior IDs are generated automatically. - Runtime event IDs are always non-null because core.world.add_event may store events by ``event.id``. - World collections are initialized as dictionaries when add_* methods expect mapping-style storage. """ from __future__ import annotations import copy import inspect import logging from collections.abc import Mapping, Sequence from dataclasses import dataclass, field from importlib import import_module from types import MappingProxyType from typing import Any import numpy as np from core.agent import Agent from core.behavior import Behavior from core.event import Event from core.resource import Resource from core.scheduler import Scheduler from core.world import World from dsl.schema import ( AgentSpec, BehaviorSpec, EventSpec, MetricSpec, PolicySpec, ResourceSpec, WorldSpec, ) from dsl.validator import ( ValidationConfig, ValidationReport, ValidationSeverity, validate_world_spec, ) from policies.base_policy import BasePolicy logger = logging.getLogger(__name__) DEFAULT_BEHAVIOR_MODULES: tuple[str, ...] = ( "behaviors.movement", "behaviors.consume", "behaviors.trade", "behaviors.attack", "behaviors.research", "behaviors.collaboration", "behaviors.transport", "behaviors.construction", "behaviors.governance", "behaviors.market", "behaviors.adoption", "behaviors.planning", "behaviors.memory", ) DEFAULT_POLICY_MODULES: tuple[str, ...] = ( "policies.rule_policy", "policies.contextual_bandit", ) BEHAVIOR_REGISTRY_ATTRIBUTES: tuple[str, ...] = ("BEHAVIOR_REGISTRY",) POLICY_REGISTRY_ATTRIBUTES: tuple[str, ...] = ( "POLICY_REGISTRY", "RULE_POLICY_REGISTRY", ) _CORE_AGENT_ALIASES: Mapping[str, tuple[str, ...]] = MappingProxyType( { "id": ("agent_id",), "type": ("agent_type", "kind"), } ) _CORE_RESOURCE_ALIASES: Mapping[str, tuple[str, ...]] = MappingProxyType( { "id": ("resource_id",), "type": ("resource_type", "kind"), "amount": ("quantity",), } ) _CORE_EVENT_ALIASES: Mapping[str, tuple[str, ...]] = MappingProxyType( { "id": ("event_id",), "name": ("type", "event_type"), "trigger_step": ("step", "at_step"), "payload": ("data",), } ) _CORE_WORLD_ALIASES: Mapping[str, tuple[str, ...]] = MappingProxyType({}) _CORE_SCHEDULER_ALIASES: Mapping[str, tuple[str, ...]] = MappingProxyType({}) class WorldFactoryError(RuntimeError): """Base exception raised by the runtime world factory.""" class RegistryResolutionError(WorldFactoryError): """Raised when a behavior, policy, or runtime class cannot be resolved.""" class ConstructorBindingError(WorldFactoryError): """Raised when DSL parameters cannot be bound to a constructor.""" class RuntimeObjectBuildError(WorldFactoryError): """Raised when a runtime object cannot be built.""" class WorldFactoryValidationError(WorldFactoryError): """Raised when semantic DSL validation fails before world construction.""" def __init__(self, report: ValidationReport) -> None: """Initialize the error from a validation report.""" self.report = report super().__init__(report.summary()) @dataclass(frozen=True) class RegistryLoadReport: """Report describing dynamically loaded registry entries.""" registry: Mapping[str, Any] import_errors: Mapping[str, str] = field(default_factory=dict) @property def names(self) -> tuple[str, ...]: """Return sorted registry names.""" return tuple(sorted(str(name) for name in self.registry.keys())) def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly registry load report.""" return { "names": list(self.names), "count": len(self.registry), "import_errors": copy.deepcopy(dict(self.import_errors)), } @dataclass(frozen=True) class WorldBuildReport: """Structured report generated during world construction.""" world_id: str agent_ids: tuple[str, ...] resource_ids: tuple[str, ...] event_keys: tuple[str, ...] behavior_registry_count: int policy_registry_count: int import_errors: Mapping[str, str] = field(default_factory=dict) validation_report: ValidationReport | None = None metadata: Mapping[str, Any] = field(default_factory=dict) @property def valid(self) -> bool: """Return whether the optional validation report is valid.""" return self.validation_report is None or self.validation_report.is_valid def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly build report.""" return { "world_id": self.world_id, "agent_ids": list(self.agent_ids), "resource_ids": list(self.resource_ids), "event_keys": list(self.event_keys), "agent_count": len(self.agent_ids), "resource_count": len(self.resource_ids), "event_count": len(self.event_keys), "behavior_registry_count": self.behavior_registry_count, "policy_registry_count": self.policy_registry_count, "import_errors": copy.deepcopy(dict(self.import_errors)), "validation_report": None if self.validation_report is None else self.validation_report.to_dict(), "metadata": copy.deepcopy(dict(self.metadata)), } @dataclass(frozen=True) class WorldBuildResult: """Factory result containing the runtime world and build report.""" world: World report: WorldBuildReport @dataclass class WorldFactoryConfig: """Configuration for ``WorldFactory``.""" validate_before_build: bool = True strict_unknown_behaviors: bool = True strict_unknown_policies: bool = True strict_constructor_params: bool = True include_disabled_behaviors: bool = False include_disabled_events: bool = False attach_default_policy: bool = True default_policy_type: str = "rule_policy" attach_scheduler: bool = True attach_dsl_spec_to_world: bool = True behavior_modules: tuple[str, ...] = DEFAULT_BEHAVIOR_MODULES policy_modules: tuple[str, ...] = DEFAULT_POLICY_MODULES behavior_registry: Mapping[str, Any] | None = None policy_registry: Mapping[str, Any] | None = None agent_class: type[Agent] = Agent resource_class: type[Resource] = Resource event_class: type[Event] = Event world_class: type[World] = World scheduler_class: type[Scheduler] | None = Scheduler world_uses_mapping_collections: bool = True @dataclass class GenericScheduledEvent: """Fallback event object used when ``core.event.Event`` is not instantiable.""" name: str trigger_step: int payload: Mapping[str, Any] = field(default_factory=dict) id: str | None = None enabled: bool = True repeat_interval: int | None = None target_agent_ids: tuple[str, ...] = () target_resource_ids: tuple[str, ...] = () metadata: Mapping[str, Any] = field(default_factory=dict) def execute(self, world: World) -> dict[str, Any]: """Execute the event in a domain-agnostic way.""" return { "event": self.name, "event_id": self.id, "trigger_step": self.trigger_step, "enabled": self.enabled, "payload": copy.deepcopy(dict(self.payload)), "target_agent_ids": list(self.target_agent_ids), "target_resource_ids": list(self.target_resource_ids), "world_step": getattr(world, "step_count", None), } @dataclass class WorldFactory: """Factory that converts ``WorldSpec`` into a runtime ``World``.""" config: WorldFactoryConfig = field(default_factory=WorldFactoryConfig) _behavior_registry: dict[str, Any] = field(default_factory=dict, init=False, repr=False) _policy_registry: dict[str, Any] = field(default_factory=dict, init=False, repr=False) _registry_import_errors: dict[str, str] = field(default_factory=dict, init=False, repr=False) def __post_init__(self) -> None: """Load configured behavior and policy registries.""" self._behavior_registry = self._build_behavior_registry() self._policy_registry = self._build_policy_registry() @property def behavior_registry(self) -> Mapping[str, Any]: """Return a read-only view of the behavior registry.""" return MappingProxyType(self._behavior_registry) @property def policy_registry(self) -> Mapping[str, Any]: """Return a read-only view of the policy registry.""" return MappingProxyType(self._policy_registry) @property def registry_import_errors(self) -> Mapping[str, str]: """Return registry import errors collected during initialization.""" return MappingProxyType(self._registry_import_errors) def create_world(self, spec: WorldSpec | Mapping[str, Any]) -> World: """Create a runtime world from a validated or mappable world spec.""" return self.create_world_result(spec).world def create_world_result(self, spec: WorldSpec | Mapping[str, Any]) -> WorldBuildResult: """Create a world and return both runtime object and build report.""" world_spec = self._coerce_world_spec(spec) validation_report = self._validate_before_build(world_spec) space_dimensions = ( int(world_spec.space.dimensions) if world_spec.space is not None else None ) resources = tuple( self.create_resource(resource_spec, dimensions=space_dimensions) for resource_spec in world_spec.resources ) events = tuple( self.create_event(event_spec) for event_spec in world_spec.events if event_spec.enabled or self.config.include_disabled_events ) agents = tuple( self.create_agent(agent_spec, dimensions=space_dimensions) for agent_spec in world_spec.agents ) metrics = tuple( self.create_metric_config(metric_spec) for metric_spec in world_spec.metrics ) world = self._create_world_object( spec=world_spec, agents=agents, resources=resources, events=events, metrics=metrics, ) if self.config.attach_scheduler: self._attach_scheduler(world, world_spec) report = WorldBuildReport( world_id=world_spec.id, agent_ids=tuple(agent.id for agent in world_spec.agents), resource_ids=tuple(resource.id for resource in world_spec.resources), event_keys=tuple(event.event_key for event in world_spec.events), behavior_registry_count=len(self._behavior_registry), policy_registry_count=len(self._policy_registry), import_errors=copy.deepcopy(self._registry_import_errors), validation_report=validation_report, metadata={ "world_name": world_spec.name, "schema_version": world_spec.schema_version, "simulation_steps": world_spec.simulation.steps, }, ) logger.info( "Created world %s with %s agent(s), %s resource(s), and %s event(s)", world_spec.id, len(agents), len(resources), len(events), ) return WorldBuildResult(world=world, report=report) def create_agent( self, spec: AgentSpec, *, dimensions: int | None = None, ) -> Agent: """Create a runtime ``Agent`` from an ``AgentSpec``.""" behavior_map: dict[str, Behavior] = {} for behavior_index, behavior_spec in enumerate(spec.behaviors): if not behavior_spec.enabled and not self.config.include_disabled_behaviors: continue behavior = self.create_behavior( behavior_spec, owner_agent_id=spec.id, behavior_index=behavior_index, ) behavior_id = str( getattr(behavior, "id", "") or self._behavior_runtime_id( behavior_spec, owner_agent_id=spec.id, behavior_index=behavior_index, ) ) if behavior_id in behavior_map: behavior_id = f"{behavior_id}#{behavior_index}" behavior_map[behavior_id] = behavior policy = self.create_policy_for_agent(spec) runtime_position = self._agent_runtime_position( spec.position, dimensions=dimensions, ) params = { "id": spec.id, "type": spec.type, "position": runtime_position.copy(), "state": copy.deepcopy(spec.state), "memory": copy.deepcopy(spec.memory), "goals": copy.deepcopy(spec.goals), "behaviors": behavior_map, "policy": policy, "alive": bool(spec.alive), "metadata": copy.deepcopy(spec.metadata), } agent = self._instantiate_runtime_object( self.config.agent_class, params, aliases=_CORE_AGENT_ALIASES, strict=False, path=f"agents.{spec.id}", item_kind="agent", ) self._safe_setattr(agent, "id", spec.id) self._safe_setattr(agent, "type", spec.type) self._force_setattr(agent, "position", runtime_position.copy()) self._safe_setattr(agent, "state", copy.deepcopy(spec.state)) self._safe_setattr(agent, "memory", copy.deepcopy(spec.memory)) self._safe_setattr(agent, "goals", copy.deepcopy(spec.goals)) self._safe_setattr(agent, "behaviors", behavior_map) self._safe_setattr(agent, "policy", policy) self._safe_setattr(agent, "alive", bool(spec.alive)) self._safe_setattr(agent, "metadata", copy.deepcopy(spec.metadata)) self._safe_setattr(agent, "dsl_spec", spec) self._ensure_agent_position_array(agent, dimensions=dimensions) return agent def create_behavior( self, spec: BehaviorSpec, *, owner_agent_id: str | None = None, behavior_index: int | None = None, ) -> Behavior: """Create a runtime behavior object from a ``BehaviorSpec``. The DSL behavior spec has this shape: {"name": "attack", "params": {...}} Runtime behavior objects should receive a deterministic runtime id, but the DSL behavior name should not be passed as a constructor argument named ``name`` because many behavior classes expose ``name`` as a read-only or class-level registry attribute. This method supports both behavior implementation styles used in the codebase: 1. Newer dataclass-style behaviors that accept params directly as constructor kwargs. 2. Rich core.behavior-style behaviors that expect: Behavior(id="...", parameters={...}) The second form is especially important for rich movement/conflict modules. """ target = self._resolve_registry_item( registry=self._behavior_registry, name=spec.name, strict=self.config.strict_unknown_behaviors, item_kind="behavior", path=f"behavior:{spec.name}", ) runtime_behavior_id = self._behavior_runtime_id( spec, owner_agent_id=owner_agent_id, behavior_index=behavior_index, ) raw_params = copy.deepcopy(spec.params) constructor_attempts: tuple[tuple[str, dict[str, Any], bool], ...] = ( ( "direct_params", { **copy.deepcopy(raw_params), "id": runtime_behavior_id, }, self.config.strict_constructor_params, ), ( "parameters_dict", { "id": runtime_behavior_id, "parameters": copy.deepcopy(raw_params), }, False, ), ( "behavior_id_parameters_dict", { "behavior_id": runtime_behavior_id, "parameters": copy.deepcopy(raw_params), }, False, ), ) behavior: Any | None = None attempt_errors: list[str] = [] for attempt_name, attempt_params, strict in constructor_attempts: try: behavior = self._instantiate_runtime_object( target, attempt_params, aliases={}, strict=strict, path=f"behavior:{spec.name}.params", item_kind="behavior", ) break except Exception as exc: attempt_errors.append( f"{attempt_name}: {exc.__class__.__name__}: {exc}" ) logger.debug( "Behavior construction attempt failed for behavior=%s attempt=%s", spec.name, attempt_name, exc_info=True, ) if behavior is None: raise RuntimeObjectBuildError( "Could not instantiate behavior " f"{spec.name!r} after {len(constructor_attempts)} constructor attempt(s). " f"Attempt errors: {attempt_errors}" ) if not isinstance(behavior, Behavior) and not callable(getattr(behavior, "execute", None)): raise RuntimeObjectBuildError( f"Behavior {spec.name!r} did not produce an executable behavior object" ) self._safe_setattr(behavior, "id", runtime_behavior_id) self._safe_setattr(behavior, "behavior_id", runtime_behavior_id) # Do not require this to succeed. Some behavior classes expose ``name`` as # a read-only/class-level registry attribute. self._safe_setattr(behavior, "name", spec.name) # These are safe diagnostics/metadata attachments. They may fail on highly # slotted/frozen classes, so _safe_setattr intentionally does not raise. self._safe_setattr(behavior, "enabled", bool(spec.enabled)) self._safe_setattr(behavior, "priority", float(spec.priority)) self._safe_setattr(behavior, "tags", tuple(spec.tags)) self._safe_setattr(behavior, "metadata", copy.deepcopy(spec.metadata)) self._safe_setattr(behavior, "dsl_spec", spec) # Rich behavior modules such as movement.py and attack.py use # ``self.parameters`` internally. Preserve the original DSL params there # whenever the object supports it. existing_parameters = getattr(behavior, "parameters", None) if not isinstance(existing_parameters, Mapping) or not existing_parameters: self._safe_setattr(behavior, "parameters", copy.deepcopy(raw_params)) return behavior def create_policy_for_agent(self, agent_spec: AgentSpec) -> BasePolicy | None: """Create the policy configured for an agent.""" policy_spec = agent_spec.policy if policy_spec is None and self.config.attach_default_policy and agent_spec.behaviors: policy_spec = PolicySpec(type=self.config.default_policy_type, params={}) if policy_spec is None: return None return self.create_policy(policy_spec) def create_policy(self, spec: PolicySpec) -> BasePolicy: """Create a runtime policy object from a ``PolicySpec``.""" target = self._resolve_registry_item( registry=self._policy_registry, name=spec.type, strict=self.config.strict_unknown_policies, item_kind="policy", path=f"policy:{spec.type}", ) params = copy.deepcopy(spec.params) params.setdefault("enabled", bool(spec.enabled)) if spec.metadata: params.setdefault("metadata", copy.deepcopy(spec.metadata)) policy = self._instantiate_runtime_object( target, params, aliases={}, strict=self.config.strict_constructor_params, path=f"policy:{spec.type}.params", item_kind="policy", ) if not isinstance(policy, BasePolicy) and not callable(getattr(policy, "choose_action", None)): raise RuntimeObjectBuildError( f"Policy {spec.type!r} did not produce a valid policy object" ) self._safe_setattr(policy, "enabled", bool(spec.enabled)) self._safe_setattr(policy, "metadata", copy.deepcopy(spec.metadata)) self._safe_setattr(policy, "dsl_spec", spec) return policy def create_resource( self, spec: ResourceSpec, *, dimensions: int | None = None, ) -> Resource: """Create a runtime ``Resource`` from a ``ResourceSpec``.""" metadata = copy.deepcopy(spec.metadata) metadata.setdefault("regeneration_rate", spec.regeneration_rate) metadata.setdefault("max_amount", spec.max_amount) runtime_position = self._resource_runtime_position( spec.position, dimensions=dimensions, ) params = { "id": spec.id, "type": spec.type, "amount": float(spec.amount), "position": runtime_position.copy(), "metadata": metadata, "regeneration_rate": float(spec.regeneration_rate), "max_amount": spec.max_amount, } resource = self._instantiate_runtime_object( self.config.resource_class, params, aliases=_CORE_RESOURCE_ALIASES, strict=False, path=f"resources.{spec.id}", item_kind="resource", ) self._safe_setattr(resource, "id", spec.id) self._safe_setattr(resource, "type", spec.type) self._safe_setattr(resource, "amount", float(spec.amount)) self._force_setattr(resource, "position", runtime_position.copy()) self._safe_setattr(resource, "metadata", metadata) self._safe_setattr(resource, "regeneration_rate", float(spec.regeneration_rate)) self._safe_setattr(resource, "max_amount", spec.max_amount) self._safe_setattr(resource, "dsl_spec", spec) self._ensure_resource_position_array(resource, dimensions=dimensions) return resource def create_event(self, spec: EventSpec) -> Event | GenericScheduledEvent: """Create a runtime event object from an ``EventSpec``.""" runtime_event_id = spec.event_key params = { "id": runtime_event_id, "name": spec.name, "trigger_step": int(spec.trigger_step), "payload": copy.deepcopy(spec.payload), "enabled": bool(spec.enabled), "repeat_interval": spec.repeat_interval, "target_agent_ids": tuple(spec.target_agent_ids), "target_resource_ids": tuple(spec.target_resource_ids), "metadata": copy.deepcopy(spec.metadata), } event_class = self.config.event_class use_fallback = inspect.isabstract(event_class) if use_fallback: event: Event | GenericScheduledEvent = GenericScheduledEvent(**params) else: try: event = self._instantiate_runtime_object( event_class, params, aliases=_CORE_EVENT_ALIASES, strict=False, path=f"events.{spec.event_key}", item_kind="event", ) except WorldFactoryError: logger.debug( "Falling back to GenericScheduledEvent for event %s", spec.event_key, exc_info=True, ) event = GenericScheduledEvent(**params) self._safe_setattr(event, "id", runtime_event_id) self._safe_setattr(event, "name", spec.name) self._safe_setattr(event, "trigger_step", int(spec.trigger_step)) self._safe_setattr(event, "payload", copy.deepcopy(spec.payload)) self._safe_setattr(event, "enabled", bool(spec.enabled)) self._safe_setattr(event, "repeat_interval", spec.repeat_interval) self._safe_setattr(event, "target_agent_ids", tuple(spec.target_agent_ids)) self._safe_setattr(event, "target_resource_ids", tuple(spec.target_resource_ids)) self._safe_setattr(event, "metadata", copy.deepcopy(spec.metadata)) self._safe_setattr(event, "dsl_spec", spec) self._ensure_event_id(event) return event def create_metric_config(self, spec: MetricSpec) -> dict[str, Any]: """Create a metric configuration record.""" return { "name": spec.name, "params": copy.deepcopy(spec.params), "enabled": bool(spec.enabled), "metadata": copy.deepcopy(spec.metadata), } def _create_world_object( self, *, spec: WorldSpec, agents: Sequence[Agent], resources: Sequence[Resource], events: Sequence[Event | GenericScheduledEvent], metrics: Sequence[Mapping[str, Any]], ) -> World: """Instantiate and populate the runtime world object.""" agent_collection = { str(getattr(agent, "id")): agent for agent in agents } resource_collection = { str(getattr(resource, "id")): resource for resource in resources } event_collection = { str(getattr(event, "id")): event for event in events if getattr(event, "id", None) is not None } metric_collection = { str(metric["name"]): copy.deepcopy(dict(metric)) for metric in metrics if "name" in metric } params = { "agents": agent_collection if self.config.world_uses_mapping_collections else tuple(agents), "resources": resource_collection if self.config.world_uses_mapping_collections else tuple(resources), "events": event_collection, "metrics": metric_collection, "step_count": 0, "metadata": copy.deepcopy(spec.metadata), } try: world = self._instantiate_runtime_object( self.config.world_class, params, aliases=_CORE_WORLD_ALIASES, strict=False, path=f"world:{spec.id}", item_kind="world", ) except WorldFactoryError: logger.debug("Falling back to no-argument World constructor", exc_info=True) world = self.config.world_class() self._attach_world_metadata(world, spec) self._install_agents(world, agents) self._install_resources(world, resources) self._install_events(world, events) self._safe_setattr(world, "metrics", metric_collection) self._safe_setattr(world, "step_count", 0) return world def _install_agents(self, world: World, agents: Sequence[Agent]) -> None: """Install agents into the world using adders when available.""" self._safe_setattr(world, "agents", {} if self.config.world_uses_mapping_collections else []) add_agent = getattr(world, "add_agent", None) if callable(add_agent): for agent in agents: self._ensure_agent_position_array(agent) add_agent(agent) return if self.config.world_uses_mapping_collections: for agent in agents: self._ensure_agent_position_array(agent) self._safe_setattr( world, "agents", {str(getattr(agent, "id")): agent for agent in agents}, ) else: for agent in agents: self._ensure_agent_position_array(agent) self._safe_setattr(world, "agents", list(agents)) def _install_resources(self, world: World, resources: Sequence[Resource]) -> None: """Install resources into the world using adders when available.""" self._safe_setattr(world, "resources", {} if self.config.world_uses_mapping_collections else []) add_resource = getattr(world, "add_resource", None) if callable(add_resource): for resource in resources: self._ensure_resource_position_array(resource) add_resource(resource) return if self.config.world_uses_mapping_collections: for resource in resources: self._ensure_resource_position_array(resource) self._safe_setattr( world, "resources", {str(getattr(resource, "id")): resource for resource in resources}, ) else: for resource in resources: self._ensure_resource_position_array(resource) self._safe_setattr(world, "resources", list(resources)) def _install_events( self, world: World, events: Sequence[Event | GenericScheduledEvent], ) -> None: """Install events into the world using adders when available. core.world.add_event may assign events by id: self.events[event.id] = event Therefore ``world.events`` must be mapping-like before add_event(). """ self._safe_setattr(world, "events", {}) add_event = getattr(world, "add_event", None) if callable(add_event): for event in events: self._ensure_event_id(event) add_event(event) return self._safe_setattr( world, "events", { str(getattr(event, "id")): event for event in events if getattr(event, "id", None) is not None }, ) def _attach_scheduler(self, world: World, spec: WorldSpec) -> None: """Attach a scheduler to the world when a scheduler class is configured.""" scheduler_class = self.config.scheduler_class if scheduler_class is None: return params = { "world": world, "activation": spec.simulation.activation, "scheduler": spec.simulation.scheduler, "seed": spec.simulation.seed, } try: scheduler = self._instantiate_runtime_object( scheduler_class, params, aliases=_CORE_SCHEDULER_ALIASES, strict=False, path=f"world:{spec.id}.scheduler", item_kind="scheduler", ) except WorldFactoryError: logger.debug("Falling back to no-argument Scheduler constructor", exc_info=True) scheduler = scheduler_class() self._safe_setattr(scheduler, "world", world) self._safe_setattr(scheduler, "activation", spec.simulation.activation) self._safe_setattr(scheduler, "scheduler", spec.simulation.scheduler) self._safe_setattr(scheduler, "seed", spec.simulation.seed) self._safe_setattr(world, "scheduler", scheduler) def _attach_world_metadata(self, world: World, spec: WorldSpec) -> None: """Attach DSL metadata and simulation configuration to the runtime world.""" self._safe_setattr(world, "id", spec.id) self._safe_setattr(world, "name", spec.name) self._safe_setattr(world, "description", spec.description) self._safe_setattr(world, "metadata", copy.deepcopy(spec.metadata)) self._safe_setattr(world, "simulation", spec.simulation) self._safe_setattr(world, "simulation_config", spec.simulation.to_dict()) self._safe_setattr(world, "space", spec.space) self._safe_setattr(world, "space_config", None if spec.space is None else spec.space.to_dict()) self._safe_setattr(world, "schema_version", spec.schema_version) if spec.space is not None: self._safe_setattr(world, "dimensions", int(spec.space.dimensions)) self._safe_setattr(world, "dimension", int(spec.space.dimensions)) self._safe_setattr(world, "space_dimensions", int(spec.space.dimensions)) if self.config.attach_dsl_spec_to_world: self._safe_setattr(world, "dsl_spec", spec) def _validate_before_build(self, spec: WorldSpec) -> ValidationReport | None: """Optionally run semantic validation before object construction.""" if not self.config.validate_before_build: return None validation_config = ValidationConfig( behavior_registry=self._behavior_registry, policy_registry=self._policy_registry, require_known_behaviors=self.config.strict_unknown_behaviors, require_known_policies=self.config.strict_unknown_policies, validate_constructor_params=self.config.strict_constructor_params, unknown_registry_item_severity=ValidationSeverity.ERROR if self.config.strict_unknown_behaviors or self.config.strict_unknown_policies else ValidationSeverity.WARNING, constructor_param_severity=ValidationSeverity.ERROR if self.config.strict_constructor_params else ValidationSeverity.WARNING, ) report = validate_world_spec(spec, config=validation_config) if not report.is_valid: raise WorldFactoryValidationError(report) return report def _coerce_world_spec(self, spec: WorldSpec | Mapping[str, Any]) -> WorldSpec: """Return a ``WorldSpec`` from a spec-like object.""" if isinstance(spec, WorldSpec): return spec if isinstance(spec, Mapping): return WorldSpec.model_validate(dict(spec)) raise TypeError(f"Expected WorldSpec or mapping, got {spec.__class__.__name__}") def _build_behavior_registry(self) -> dict[str, Any]: """Build the behavior registry from defaults and custom entries.""" registry: dict[str, Any] = {} module_report = self._load_module_registries( self.config.behavior_modules, BEHAVIOR_REGISTRY_ATTRIBUTES, base_class=Behavior, ) registry.update(module_report.registry) self._registry_import_errors.update(module_report.import_errors) if self.config.behavior_registry is not None: registry.update({ str(key): value for key, value in self.config.behavior_registry.items() }) return self._with_class_aliases(registry) def _build_policy_registry(self) -> dict[str, Any]: """Build the policy registry from defaults and custom entries.""" registry: dict[str, Any] = {} module_report = self._load_module_registries( self.config.policy_modules, POLICY_REGISTRY_ATTRIBUTES, base_class=BasePolicy, ) registry.update(module_report.registry) self._registry_import_errors.update(module_report.import_errors) if self.config.policy_registry is not None: registry.update({ str(key): value for key, value in self.config.policy_registry.items() }) return self._with_class_aliases(registry) def _load_module_registries( self, module_names: Sequence[str], registry_attributes: Sequence[str], *, base_class: type[Any], ) -> RegistryLoadReport: """Load registry mappings from modules and discover named subclasses.""" registry: dict[str, Any] = {} import_errors: dict[str, str] = {} for module_name in module_names: try: module = import_module(module_name) except Exception as exc: import_errors[str(module_name)] = f"{exc.__class__.__name__}: {exc}" continue for attribute_name in registry_attributes: raw_registry = getattr(module, attribute_name, None) if isinstance(raw_registry, Mapping): registry.update({ str(key): value for key, value in raw_registry.items() }) registry.update(self._discover_named_subclasses(module, base_class=base_class)) return RegistryLoadReport(registry=registry, import_errors=import_errors) @staticmethod def _discover_named_subclasses(module: Any, *, base_class: type[Any]) -> dict[str, Any]: """Discover subclasses with a string ``name`` attribute from a module.""" discovered: dict[str, Any] = {} for attribute_value in vars(module).values(): if not inspect.isclass(attribute_value): continue if attribute_value is base_class: continue try: is_subclass = issubclass(attribute_value, base_class) except TypeError: is_subclass = False if not is_subclass: continue name = getattr(attribute_value, "name", None) if isinstance(name, str) and name: discovered.setdefault(name, attribute_value) return discovered @staticmethod def _with_class_aliases(registry: Mapping[str, Any]) -> dict[str, Any]: """Add class-name and ``name`` aliases to registry entries.""" expanded = {str(key): value for key, value in registry.items()} for key, value in list(expanded.items()): class_name = getattr(value, "__name__", None) if class_name: expanded.setdefault(str(class_name), value) runtime_name = getattr(value, "name", None) if isinstance(runtime_name, str) and runtime_name: expanded.setdefault(runtime_name, value) expanded.setdefault(str(key).lower(), value) return expanded def _resolve_registry_item( self, *, registry: Mapping[str, Any], name: str, strict: bool, item_kind: str, path: str, ) -> Any: """Resolve a registry item by exact or case-insensitive name.""" if name in registry: return registry[name] lower_name = name.lower() if lower_name in registry: return registry[lower_name] for registered_name, item in registry.items(): if registered_name.lower() == lower_name: return item message = ( f"Unknown {item_kind} {name!r} at {path}. " f"Known {item_kind}s: {sorted(str(key) for key in registry.keys())}" ) if strict: raise RegistryResolutionError(message) logger.warning(message) if item_kind == "behavior": return self._noop_behavior_factory(name) return self._noop_policy_factory(name) def _instantiate_runtime_object( self, target: Any, params: Mapping[str, Any], *, aliases: Mapping[str, Sequence[str]], strict: bool, path: str, item_kind: str, ) -> Any: """Instantiate a runtime object using signature-aware keyword binding.""" if not callable(target): try: return copy.deepcopy(target) except Exception as exc: raise RuntimeObjectBuildError( f"Registered {item_kind} at {path} is not callable and could not be copied" ) from exc kwargs = self._bind_constructor_kwargs( target, params, aliases=aliases, strict=strict, path=path, item_kind=item_kind, ) try: return target(**kwargs) except TypeError as exc: raise ConstructorBindingError( f"Could not instantiate {item_kind} at {path}: {exc}" ) from exc except Exception as exc: raise RuntimeObjectBuildError( f"Constructor for {item_kind} at {path} failed: {exc}" ) from exc def _bind_constructor_kwargs( self, target: Any, params: Mapping[str, Any], *, aliases: Mapping[str, Sequence[str]], strict: bool, path: str, item_kind: str, ) -> dict[str, Any]: """Filter and alias params according to a target constructor signature.""" working = copy.deepcopy(dict(params)) signature_info = self._constructor_signature_info(target) if signature_info is None: return working allowed_params, required_params, accepts_var_keyword = signature_info if accepts_var_keyword: return working for canonical_name, alias_names in aliases.items(): if canonical_name not in working: continue if canonical_name in allowed_params: continue for alias_name in alias_names: if alias_name in allowed_params and alias_name not in working: working[alias_name] = working[canonical_name] break unknown_params = set(working.keys()) - allowed_params if unknown_params and strict: raise ConstructorBindingError( f"Unknown constructor parameter(s) for {item_kind} at {path}: " f"{sorted(unknown_params)}. Allowed parameters: {sorted(allowed_params)}" ) filtered = { key: value for key, value in working.items() if key in allowed_params } missing_required = required_params - set(filtered.keys()) if missing_required: raise ConstructorBindingError( f"Missing required constructor parameter(s) for {item_kind} at {path}: " f"{sorted(missing_required)}" ) return filtered @staticmethod def _constructor_signature_info(target: Any) -> tuple[set[str], set[str], bool] | None: """Return constructor binding information for a callable target.""" try: signature = inspect.signature(target) except (TypeError, ValueError): return None allowed_params: set[str] = set() required_params: set[str] = set() accepts_var_keyword = False for name, parameter in signature.parameters.items(): if name == "self": continue if parameter.kind is inspect.Parameter.VAR_KEYWORD: accepts_var_keyword = True continue if parameter.kind is inspect.Parameter.VAR_POSITIONAL: continue if parameter.kind not in { inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY, }: continue allowed_params.add(name) if parameter.default is inspect.Parameter.empty: required_params.add(name) return allowed_params, required_params, accepts_var_keyword @staticmethod def _behavior_runtime_id( spec: BehaviorSpec, *, owner_agent_id: str | None, behavior_index: int | None, ) -> str: """Return a deterministic runtime id for a behavior spec.""" metadata_id = None if isinstance(spec.metadata, Mapping): metadata_id = spec.metadata.get("id") or spec.metadata.get("behavior_id") if metadata_id is not None and str(metadata_id).strip(): return str(metadata_id).strip() owner = owner_agent_id or "world" index = 0 if behavior_index is None else int(behavior_index) return f"{owner}:behavior:{index}:{spec.name}" @staticmethod def _agent_runtime_position( position: Any, *, dimensions: int | None = None, ) -> np.ndarray: """Return a NumPy position vector compatible with runtime agents.""" return WorldFactory._runtime_position( position, dimensions=dimensions, object_label="Agent", ) @staticmethod def _resource_runtime_position( position: Any, *, dimensions: int | None = None, ) -> np.ndarray: """Return a NumPy position vector compatible with runtime resources.""" return WorldFactory._runtime_position( position, dimensions=dimensions, object_label="Resource", ) @staticmethod def _runtime_position( position: Any, *, dimensions: int | None = None, object_label: str = "Object", ) -> np.ndarray: """Return a NumPy runtime position vector.""" if position is None: target_dimensions = ( int(dimensions) if dimensions is not None and int(dimensions) > 0 else 2 ) return np.zeros(target_dimensions, dtype=np.float64) try: array = np.asarray(position, dtype=np.float64).reshape(-1) except (TypeError, ValueError) as exc: raise RuntimeObjectBuildError( f"{object_label} position must be numeric, got {position!r}" ) from exc if array.size == 0: target_dimensions = ( int(dimensions) if dimensions is not None and int(dimensions) > 0 else 2 ) return np.zeros(target_dimensions, dtype=np.float64) if not np.all(np.isfinite(array)): raise RuntimeObjectBuildError( f"{object_label} position must contain only finite values, got {position!r}" ) if dimensions is not None and int(dimensions) > 0: target_dimensions = int(dimensions) if int(array.size) != target_dimensions: raise RuntimeObjectBuildError( f"{object_label} position has {int(array.size)} dimension(s), " f"but world space expects {target_dimensions}" ) return array.astype(np.float64, copy=True) def _ensure_agent_position_array( self, agent: Agent, *, dimensions: int | None = None, ) -> None: """Ensure agent.position is a NumPy array before adding it to World.""" raw_position = getattr(agent, "position", None) runtime_position = self._agent_runtime_position( raw_position, dimensions=dimensions, ) self._force_setattr(agent, "position", runtime_position.copy()) repaired_position = getattr(agent, "position", None) if not hasattr(repaired_position, "size"): raise RuntimeObjectBuildError( "Agent.position could not be converted to a NumPy array before " f"world installation. Agent={getattr(agent, 'id', None)!r}, " f"position_type={type(repaired_position).__name__}" ) def _ensure_resource_position_array( self, resource: Resource, *, dimensions: int | None = None, ) -> None: """Ensure resource.position is a NumPy array before adding it to World.""" raw_position = getattr(resource, "position", None) runtime_position = self._resource_runtime_position( raw_position, dimensions=dimensions, ) self._force_setattr(resource, "position", runtime_position.copy()) repaired_position = getattr(resource, "position", None) if not hasattr(repaired_position, "size"): raise RuntimeObjectBuildError( "Resource.position could not be converted to a NumPy array before " f"world installation. Resource={getattr(resource, 'id', None)!r}, " f"position_type={type(repaired_position).__name__}" ) def _ensure_event_id(self, event: Event | GenericScheduledEvent) -> None: """Ensure an event has a stable id before installation into World.""" event_id = getattr(event, "id", None) if event_id is not None and str(event_id).strip(): self._safe_setattr(event, "id", str(event_id).strip()) return name = getattr(event, "name", None) trigger_step = getattr(event, "trigger_step", 0) fallback_id = f"{name or 'event'}:{trigger_step}" self._safe_setattr(event, "id", fallback_id) @staticmethod def _safe_setattr(instance: Any, name: str, value: Any) -> bool: """Set an attribute defensively.""" try: setattr(instance, name, value) return True except Exception: logger.debug( "Could not set attribute %s on %s", name, instance.__class__.__name__, exc_info=True, ) return False @staticmethod def _force_setattr(instance: Any, name: str, value: Any) -> None: """Set an attribute, using object.__setattr__ as a fallback.""" try: setattr(instance, name, value) return except Exception: pass try: object.__setattr__(instance, name, value) return except Exception as exc: raise RuntimeObjectBuildError( f"Could not set attribute {name!r} on {instance.__class__.__name__}" ) from exc @staticmethod def _noop_behavior_factory(name: str) -> type[Behavior]: """Create a no-op behavior class for permissive unknown behavior mode.""" class NoOpBehavior(Behavior): """Fallback behavior used only when unknown behaviors are permissive.""" behavior_name = name def check_preconditions(self, agent: Agent, world: World) -> bool: """Return False so unknown behavior never executes.""" del agent, world return False def execute(self, agent: Agent, world: World) -> dict[str, Any]: """Return a structured no-op result.""" del world return { "behavior": self.behavior_name, "actor_id": getattr(agent, "id", None), "success": False, "details": {"reason": "unknown_behavior_noop"}, } NoOpBehavior.name = name # type: ignore[attr-defined] return NoOpBehavior @staticmethod def _noop_policy_factory(name: str) -> type[BasePolicy]: """Create a no-op policy class for permissive unknown policy mode.""" class NoOpPolicy(BasePolicy): """Fallback policy used only when unknown policies are permissive.""" policy_name_value = name def choose_action( self, agent: Agent, world: World, behaviors: Sequence[Behavior] | Mapping[str, Behavior] | None = None, ) -> Behavior | None: """Choose no behavior.""" del agent, world, behaviors return None NoOpPolicy.name = name # type: ignore[attr-defined] return NoOpPolicy def create_world( spec: WorldSpec | Mapping[str, Any], *, config: WorldFactoryConfig | None = None, ) -> World: """Convenience function that creates a runtime world from a spec.""" return WorldFactory(config=config or WorldFactoryConfig()).create_world(spec) def create_world_result( spec: WorldSpec | Mapping[str, Any], *, config: WorldFactoryConfig | None = None, ) -> WorldBuildResult: """Convenience function that creates a world and returns its build report.""" return WorldFactory(config=config or WorldFactoryConfig()).create_world_result(spec) __all__ = [ "BEHAVIOR_REGISTRY_ATTRIBUTES", "DEFAULT_BEHAVIOR_MODULES", "DEFAULT_POLICY_MODULES", "GenericScheduledEvent", "ConstructorBindingError", "POLICY_REGISTRY_ATTRIBUTES", "RegistryLoadReport", "RegistryResolutionError", "RuntimeObjectBuildError", "WorldBuildReport", "WorldBuildResult", "WorldFactory", "WorldFactoryConfig", "WorldFactoryError", "WorldFactoryValidationError", "create_world", "create_world_result", ]