""" Semantic validator for the WorldSmithAI DSL. The schema layer validates JSON shape. This module validates semantic consistency: known behavior names, known policies, constructor parameters, references to agents/resources, policy rule targets, event targets, and other cross-object relationships. The validator does not instantiate runtime World objects. It is safe to run after parsing and before ``WorldFactory``. Example: from dsl.parser import parse_world_spec from dsl.validator import assert_valid_world_spec spec = parse_world_spec(raw_llm_output) assert_valid_world_spec(spec) Future extensibility: - Add version-specific semantic validation. - Add plugin registry discovery. - Add metric registry validation. - Add structured repair suggestions for SLM output. - Add validation profiles for strict production mode versus permissive demo mode. """ from __future__ import annotations import copy import inspect import logging from collections.abc import Iterable, Mapping, MutableSequence, Sequence from dataclasses import dataclass, field from enum import Enum from importlib import import_module from typing import Any from pydantic import ValidationError from dsl.schema import AgentSpec, BehaviorSpec, EventSpec, MetricSpec, PolicySpec, ResourceSpec, WorldSpec 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", ) AGENT_REFERENCE_KEYS: frozenset[str] = frozenset( { "agent_id", "target_agent_id", "owner_agent_id", "source_agent_id", "collector_agent_id", "arbitrator_agent_id", "market_agent_id", "station_agent_id", "recipient_agent_id", "counterparty_agent_id", } ) RESOURCE_REFERENCE_KEYS: frozenset[str] = frozenset( { "resource_id", "target_resource_id", "source_resource_id", "station_resource_id", } ) BEHAVIOR_REFERENCE_KEYS: frozenset[str] = frozenset( { "behavior_name", "selected_behavior_name", } ) class ValidationSeverity(str, Enum): """Validation issue severity levels.""" ERROR = "error" WARNING = "warning" INFO = "info" class DSLValidationError(ValueError): """Raised when a semantic validation report contains errors.""" def __init__(self, report: ValidationReport) -> None: """Initialize the exception with a validation report.""" self.report = report super().__init__(report.summary()) @dataclass(frozen=True) class ValidationIssue: """A single semantic validation issue.""" code: str message: str path: str severity: ValidationSeverity = ValidationSeverity.ERROR details: Mapping[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly representation of the issue.""" return { "code": self.code, "message": self.message, "path": self.path, "severity": self.severity.value, "details": copy.deepcopy(dict(self.details)), } @dataclass class ValidationReport: """Semantic validation report for a world specification.""" world_id: str issues: list[ValidationIssue] = field(default_factory=list) def add( self, *, code: str, message: str, path: str, severity: ValidationSeverity = ValidationSeverity.ERROR, details: Mapping[str, Any] | None = None, ) -> None: """Add a validation issue to the report.""" self.issues.append( ValidationIssue( code=code, message=message, path=path, severity=severity, details=details or {}, ) ) @property def errors(self) -> tuple[ValidationIssue, ...]: """Return all error-level issues.""" return tuple(issue for issue in self.issues if issue.severity is ValidationSeverity.ERROR) @property def warnings(self) -> tuple[ValidationIssue, ...]: """Return all warning-level issues.""" return tuple(issue for issue in self.issues if issue.severity is ValidationSeverity.WARNING) @property def infos(self) -> tuple[ValidationIssue, ...]: """Return all info-level issues.""" return tuple(issue for issue in self.issues if issue.severity is ValidationSeverity.INFO) @property def is_valid(self) -> bool: """Return whether the report has no errors.""" return not self.errors def raise_for_errors(self) -> None: """Raise ``DSLValidationError`` if this report contains errors.""" if not self.is_valid: raise DSLValidationError(self) def summary(self) -> str: """Return a concise human-readable report summary.""" return ( f"WorldSpec {self.world_id!r} validation: " f"{len(self.errors)} error(s), " f"{len(self.warnings)} warning(s), " f"{len(self.infos)} info(s)" ) def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly report representation.""" return { "world_id": self.world_id, "valid": self.is_valid, "error_count": len(self.errors), "warning_count": len(self.warnings), "info_count": len(self.infos), "issues": [issue.to_dict() for issue in self.issues], } @dataclass(frozen=True) class RegistryLoadResult: """Result of loading behavior or policy registries from modules.""" registry: Mapping[str, Any] import_errors: Mapping[str, str] = field(default_factory=dict) @property def names(self) -> frozenset[str]: """Return registered names.""" return frozenset(str(name) for name in self.registry.keys()) def known_names_with_class_aliases(self) -> frozenset[str]: """Return registry keys plus registered class names when available.""" names: set[str] = set(str(name) for name in self.registry.keys()) for value in self.registry.values(): class_name = getattr(value, "__name__", None) if class_name: names.add(str(class_name)) return frozenset(names) @dataclass class ValidationConfig: """Configuration for ``DSLSemanticValidator``.""" behavior_registry: Mapping[str, Any] | None = None policy_registry: Mapping[str, Any] | None = None behavior_modules: tuple[str, ...] = DEFAULT_BEHAVIOR_MODULES policy_modules: tuple[str, ...] = DEFAULT_POLICY_MODULES load_default_registries: bool = True require_known_behaviors: bool = True require_known_policies: bool = True validate_constructor_params: bool = True validate_references: bool = True validate_policy_rule_references: bool = True unknown_registry_item_severity: ValidationSeverity = ValidationSeverity.ERROR constructor_param_severity: ValidationSeverity = ValidationSeverity.ERROR unresolved_reference_severity: ValidationSeverity = ValidationSeverity.ERROR missing_policy_severity: ValidationSeverity = ValidationSeverity.WARNING registry_import_error_severity: ValidationSeverity = ValidationSeverity.WARNING suspicious_configuration_severity: ValidationSeverity = ValidationSeverity.WARNING known_metric_names: tuple[str, ...] = () def load_registry_from_modules( module_names: Sequence[str], registry_attributes: Sequence[str], ) -> RegistryLoadResult: """Load registry mappings from modules. Import failures are captured as diagnostics instead of being raised. This makes validation usable during development while still surfacing registry problems in the report. """ 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: # pragma: no cover - defensive for user code modules import_errors[str(module_name)] = f"{exc.__class__.__name__}: {exc}" continue for attribute in registry_attributes: raw_registry = getattr(module, attribute, None) if isinstance(raw_registry, Mapping): registry.update({str(key): value for key, value in raw_registry.items()}) return RegistryLoadResult(registry=registry, import_errors=import_errors) def load_default_behavior_registry() -> RegistryLoadResult: """Load default behavior registries from the configured behavior modules.""" return load_registry_from_modules(DEFAULT_BEHAVIOR_MODULES, BEHAVIOR_REGISTRY_ATTRIBUTES) def load_default_policy_registry() -> RegistryLoadResult: """Load default policy registries from the configured policy modules.""" return load_registry_from_modules(DEFAULT_POLICY_MODULES, POLICY_REGISTRY_ATTRIBUTES) @dataclass class DSLSemanticValidator: """Semantic validator for ``WorldSpec``. The validator is registry-aware and can be used with default registries or custom plugin registries. """ config: ValidationConfig = field(default_factory=ValidationConfig) _behavior_load_result: RegistryLoadResult = field(init=False, repr=False) _policy_load_result: RegistryLoadResult = field(init=False, repr=False) def __post_init__(self) -> None: """Load registries according to the validator configuration.""" if self.config.behavior_registry is not None: self._behavior_load_result = RegistryLoadResult( registry={str(key): value for key, value in self.config.behavior_registry.items()} ) elif self.config.load_default_registries: self._behavior_load_result = load_registry_from_modules( self.config.behavior_modules, BEHAVIOR_REGISTRY_ATTRIBUTES, ) else: self._behavior_load_result = RegistryLoadResult(registry={}) if self.config.policy_registry is not None: self._policy_load_result = RegistryLoadResult( registry={str(key): value for key, value in self.config.policy_registry.items()} ) elif self.config.load_default_registries: self._policy_load_result = load_registry_from_modules( self.config.policy_modules, POLICY_REGISTRY_ATTRIBUTES, ) else: self._policy_load_result = RegistryLoadResult(registry={}) @property def behavior_registry(self) -> Mapping[str, Any]: """Return the behavior registry used by this validator.""" return self._behavior_load_result.registry @property def policy_registry(self) -> Mapping[str, Any]: """Return the policy registry used by this validator.""" return self._policy_load_result.registry @property def known_behavior_names(self) -> frozenset[str]: """Return known behavior registry names and class-name aliases.""" return self._behavior_load_result.known_names_with_class_aliases() @property def known_policy_names(self) -> frozenset[str]: """Return known policy registry names and class-name aliases.""" return self._policy_load_result.known_names_with_class_aliases() def validate(self, spec: WorldSpec | Mapping[str, Any]) -> ValidationReport: """Validate a ``WorldSpec`` or mapping semantically. Args: spec: Validated ``WorldSpec`` or raw mapping that can be validated into ``WorldSpec``. Returns: Semantic validation report. """ world_spec = self._coerce_world_spec(spec) report = ValidationReport(world_id=world_spec.id) self._validate_registry_loading(report) self._validate_world_shape(world_spec, report) self._validate_agents(world_spec, report) self._validate_resources(world_spec, report) self._validate_events(world_spec, report) self._validate_metrics(world_spec, report) return report def assert_valid(self, spec: WorldSpec | Mapping[str, Any]) -> WorldSpec: """Validate and raise when semantic errors are present. Returns the original/coerced ``WorldSpec`` on success. """ world_spec = self._coerce_world_spec(spec) report = self.validate(world_spec) report.raise_for_errors() return world_spec 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): try: return WorldSpec.model_validate(dict(spec)) except ValidationError as exc: report = ValidationReport(world_id="") report.add( code="schema_validation_failed", message="WorldSpec failed Pydantic schema validation before semantic validation", path="$", severity=ValidationSeverity.ERROR, details={"errors": exc.errors()}, ) raise DSLValidationError(report) from exc report = ValidationReport(world_id="") report.add( code="unsupported_spec_type", message="Semantic validator expected WorldSpec or mapping", path="$", severity=ValidationSeverity.ERROR, details={"type": spec.__class__.__name__}, ) raise DSLValidationError(report) def _validate_registry_loading(self, report: ValidationReport) -> None: """Add diagnostics for registry import failures or empty registries.""" for module_name, error in self._behavior_load_result.import_errors.items(): report.add( code="behavior_registry_import_failed", message=f"Could not import behavior registry module {module_name!r}", path="registries.behaviors", severity=self.config.registry_import_error_severity, details={"module": module_name, "error": error}, ) for module_name, error in self._policy_load_result.import_errors.items(): report.add( code="policy_registry_import_failed", message=f"Could not import policy registry module {module_name!r}", path="registries.policies", severity=self.config.registry_import_error_severity, details={"module": module_name, "error": error}, ) if self.config.require_known_behaviors and not self.behavior_registry: report.add( code="behavior_registry_empty", message="No behavior registry entries are available for semantic validation", path="registries.behaviors", severity=ValidationSeverity.WARNING, ) if self.config.require_known_policies and not self.policy_registry: report.add( code="policy_registry_empty", message="No policy registry entries are available for semantic validation", path="registries.policies", severity=ValidationSeverity.WARNING, ) def _validate_world_shape(self, spec: WorldSpec, report: ValidationReport) -> None: """Validate high-level world consistency.""" if not spec.agents: report.add( code="world_has_no_agents", message="World has no agents; simulation may run but no agent behavior will occur", path="agents", severity=self.config.suspicious_configuration_severity, ) if spec.simulation.steps == 0: report.add( code="simulation_has_zero_steps", message="Simulation is configured for zero steps", path="simulation.steps", severity=ValidationSeverity.INFO, ) if spec.space is not None and spec.space.bounds is not None: for index, bounds in enumerate(spec.space.bounds): if bounds[0] >= bounds[1]: report.add( code="invalid_space_bounds", message="Space lower bound must be less than upper bound", path=f"space.bounds[{index}]", severity=ValidationSeverity.ERROR, details={"bounds": bounds}, ) def _validate_agents(self, spec: WorldSpec, report: ValidationReport) -> None: """Validate agents, behaviors, and policies.""" agent_ids = set(spec.agent_ids) resource_ids = set(spec.resource_ids) agent_types = {agent.type for agent in spec.agents} resource_types = {resource.type for resource in spec.resources} seen_agent_ids: set[str] = set() for agent_index, agent in enumerate(spec.agents): agent_path = f"agents[{agent_index}]" if agent.id in seen_agent_ids: report.add( code="duplicate_agent_id", message=f"Duplicate agent id {agent.id!r}", path=f"{agent_path}.id", severity=ValidationSeverity.ERROR, ) seen_agent_ids.add(agent.id) if not agent.behaviors: report.add( code="agent_has_no_behaviors", message=f"Agent {agent.id!r} has no behaviors", path=f"{agent_path}.behaviors", severity=self.config.suspicious_configuration_severity, ) if agent.behaviors and agent.policy is None: report.add( code="agent_has_behaviors_without_policy", message=( f"Agent {agent.id!r} has behaviors but no policy. " "This is valid only if Agent.step executes behaviors directly." ), path=f"{agent_path}.policy", severity=self.config.missing_policy_severity, ) self._validate_agent_behaviors( spec=spec, agent=agent, agent_path=agent_path, agent_ids=agent_ids, resource_ids=resource_ids, agent_types=agent_types, resource_types=resource_types, report=report, ) if agent.policy is not None: self._validate_policy( spec=spec, agent=agent, policy=agent.policy, path=f"{agent_path}.policy", report=report, ) def _validate_agent_behaviors( self, *, spec: WorldSpec, agent: AgentSpec, agent_path: str, agent_ids: set[str], resource_ids: set[str], agent_types: set[str], resource_types: set[str], report: ValidationReport, ) -> None: """Validate all behaviors attached to one agent.""" seen_behavior_labels: dict[str, int] = {} for behavior_index, behavior in enumerate(agent.behaviors): behavior_path = f"{agent_path}.behaviors[{behavior_index}]" seen_behavior_labels[behavior.name] = seen_behavior_labels.get(behavior.name, 0) + 1 self._validate_behavior_name(behavior, behavior_path, report) self._validate_constructor_params( registry=self.behavior_registry, registry_item_name=behavior.name, params=behavior.params, path=f"{behavior_path}.params", report=report, item_kind="behavior", ) if self.config.validate_references: self._validate_param_references( params=behavior.params, path=f"{behavior_path}.params", agent_ids=agent_ids, resource_ids=resource_ids, known_behavior_names=self.known_behavior_names, agent_types=agent_types, resource_types=resource_types, report=report, ) for behavior_name, count in sorted(seen_behavior_labels.items()): if count > 1: report.add( code="duplicate_behavior_name_on_agent", message=( f"Agent {agent.id!r} has {count} behavior specs with name " f"{behavior_name!r}; this is allowed but policies may treat them as one arm" ), path=f"{agent_path}.behaviors", severity=ValidationSeverity.INFO, details={"behavior_name": behavior_name, "count": count}, ) def _validate_behavior_name( self, behavior: BehaviorSpec, path: str, report: ValidationReport, ) -> None: """Validate that a behavior name is known when registry checking is enabled.""" if not self.config.require_known_behaviors: return if not self.behavior_registry: return if behavior.name not in self.known_behavior_names: report.add( code="unknown_behavior", message=f"Unknown behavior name {behavior.name!r}", path=f"{path}.name", severity=self.config.unknown_registry_item_severity, details={ "behavior_name": behavior.name, "known_behavior_names": sorted(self.known_behavior_names), }, ) def _validate_policy( self, *, spec: WorldSpec, agent: AgentSpec, policy: PolicySpec, path: str, report: ValidationReport, ) -> None: """Validate one policy spec.""" if self.config.require_known_policies and self.policy_registry: if policy.type not in self.known_policy_names: report.add( code="unknown_policy", message=f"Unknown policy type {policy.type!r}", path=f"{path}.type", severity=self.config.unknown_registry_item_severity, details={ "policy_type": policy.type, "known_policy_names": sorted(self.known_policy_names), }, ) self._validate_constructor_params( registry=self.policy_registry, registry_item_name=policy.type, params=policy.params, path=f"{path}.params", report=report, item_kind="policy", ) if self.config.validate_policy_rule_references: self._validate_policy_rule_references( spec=spec, agent=agent, policy=policy, path=path, report=report, ) def _validate_policy_rule_references( self, *, spec: WorldSpec, agent: AgentSpec, policy: PolicySpec, path: str, report: ValidationReport, ) -> None: """Validate behavior references inside rule-policy style params.""" rules = policy.params.get("rules") if not isinstance(rules, Sequence) or isinstance(rules, (str, bytes)): return agent_behavior_names = set(agent.behavior_names) known_names = self.known_behavior_names if not known_names: known_names = frozenset(spec.behavior_names) for rule_index, raw_rule in enumerate(rules): if not isinstance(raw_rule, Mapping): report.add( code="invalid_policy_rule", message="Policy rule must be an object", path=f"{path}.params.rules[{rule_index}]", severity=ValidationSeverity.ERROR, details={"rule_type": raw_rule.__class__.__name__}, ) continue target_names = self._extract_rule_behavior_names(raw_rule) for target_name in target_names: rule_target_path = f"{path}.params.rules[{rule_index}]" if known_names and target_name not in known_names: report.add( code="rule_targets_unknown_behavior", message=f"Policy rule targets unknown behavior {target_name!r}", path=rule_target_path, severity=self.config.unknown_registry_item_severity, details={ "target_behavior": target_name, "known_behavior_names": sorted(known_names), }, ) if target_name not in agent_behavior_names: report.add( code="rule_targets_missing_agent_behavior", message=( f"Policy rule targets behavior {target_name!r}, but agent " f"{agent.id!r} does not define that behavior" ), path=rule_target_path, severity=self.config.suspicious_configuration_severity, details={ "target_behavior": target_name, "agent_behavior_names": sorted(agent_behavior_names), }, ) def _validate_resources(self, spec: WorldSpec, report: ValidationReport) -> None: """Validate resources.""" seen_resource_ids: set[str] = set() for resource_index, resource in enumerate(spec.resources): path = f"resources[{resource_index}]" if resource.id in seen_resource_ids: report.add( code="duplicate_resource_id", message=f"Duplicate resource id {resource.id!r}", path=f"{path}.id", severity=ValidationSeverity.ERROR, ) seen_resource_ids.add(resource.id) if resource.max_amount is not None and resource.amount > resource.max_amount: report.add( code="resource_amount_exceeds_max", message=f"Resource {resource.id!r} amount exceeds max_amount", path=f"{path}.amount", severity=ValidationSeverity.ERROR, details={ "amount": resource.amount, "max_amount": resource.max_amount, }, ) def _validate_events(self, spec: WorldSpec, report: ValidationReport) -> None: """Validate event references and scheduling consistency.""" agent_ids = set(spec.agent_ids) resource_ids = set(spec.resource_ids) seen_event_keys: set[str] = set() for event_index, event in enumerate(spec.events): path = f"events[{event_index}]" event_key = event.event_key if event_key in seen_event_keys: report.add( code="duplicate_event_key", message=f"Duplicate event key {event_key!r}", path=path, severity=ValidationSeverity.ERROR, ) seen_event_keys.add(event_key) self._validate_event_targets(event, path, agent_ids, resource_ids, report) if event.trigger_step >= spec.simulation.steps and event.repeat_interval is None: report.add( code="event_never_triggers", message=( f"Event {event_key!r} triggers at step {event.trigger_step}, " f"but simulation has only {spec.simulation.steps} step(s)" ), path=f"{path}.trigger_step", severity=self.config.suspicious_configuration_severity, details={ "trigger_step": event.trigger_step, "simulation_steps": spec.simulation.steps, }, ) if self.config.validate_references: self._validate_param_references( params=event.payload, path=f"{path}.payload", agent_ids=agent_ids, resource_ids=resource_ids, known_behavior_names=self.known_behavior_names, agent_types={agent.type for agent in spec.agents}, resource_types={resource.type for resource in spec.resources}, report=report, ) def _validate_event_targets( self, event: EventSpec, path: str, agent_ids: set[str], resource_ids: set[str], report: ValidationReport, ) -> None: """Validate explicit event target ids.""" for target_index, agent_id in enumerate(event.target_agent_ids): if agent_id not in agent_ids: report.add( code="event_unknown_agent_target", message=f"Event {event.event_key!r} references unknown agent {agent_id!r}", path=f"{path}.target_agent_ids[{target_index}]", severity=self.config.unresolved_reference_severity, details={"agent_id": agent_id}, ) for target_index, resource_id in enumerate(event.target_resource_ids): if resource_id not in resource_ids: report.add( code="event_unknown_resource_target", message=f"Event {event.event_key!r} references unknown resource {resource_id!r}", path=f"{path}.target_resource_ids[{target_index}]", severity=self.config.unresolved_reference_severity, details={"resource_id": resource_id}, ) def _validate_metrics(self, spec: WorldSpec, report: ValidationReport) -> None: """Validate optional metric specs.""" if not self.config.known_metric_names: return known_metrics = set(self.config.known_metric_names) for metric_index, metric in enumerate(spec.metrics): self._validate_metric(metric, f"metrics[{metric_index}]", known_metrics, report) def _validate_metric( self, metric: MetricSpec, path: str, known_metrics: set[str], report: ValidationReport, ) -> None: """Validate one metric spec against configured known metric names.""" if metric.name not in known_metrics: report.add( code="unknown_metric", message=f"Unknown metric name {metric.name!r}", path=f"{path}.name", severity=self.config.unknown_registry_item_severity, details={"known_metric_names": sorted(known_metrics)}, ) def _validate_constructor_params( self, *, registry: Mapping[str, Any], registry_item_name: str, params: Mapping[str, Any], path: str, report: ValidationReport, item_kind: str, ) -> None: """Validate params against a registered class/function signature.""" if not self.config.validate_constructor_params: return target = registry.get(registry_item_name) if target is None: return signature_info = _constructor_signature_info(target) if signature_info is None: return allowed_params, required_params, accepts_var_keyword = signature_info if not accepts_var_keyword: unknown_params = sorted(set(params.keys()) - allowed_params) if unknown_params: report.add( code=f"unknown_{item_kind}_constructor_params", message=( f"{item_kind.title()} {registry_item_name!r} received unknown " f"constructor parameter(s): {unknown_params}" ), path=path, severity=self.config.constructor_param_severity, details={ "unknown_params": unknown_params, "allowed_params": sorted(allowed_params), }, ) runtime_injected_params: set[str] = set() if item_kind == "behavior": runtime_injected_params.update({"id", "behavior_id", "name"}) if item_kind == "policy": runtime_injected_params.update({"enabled", "metadata"}) missing_required = sorted((required_params - set(params.keys())) - runtime_injected_params) if missing_required: report.add( code=f"missing_{item_kind}_constructor_params", message=( f"{item_kind.title()} {registry_item_name!r} is missing required " f"constructor parameter(s): {missing_required}" ), path=path, severity=self.config.constructor_param_severity, details={ "missing_required_params": missing_required, "allowed_params": sorted(allowed_params), "runtime_injected_params": sorted(runtime_injected_params), }, ) def _validate_param_references( self, *, params: Mapping[str, Any], path: str, agent_ids: set[str], resource_ids: set[str], known_behavior_names: frozenset[str], agent_types: set[str], resource_types: set[str], report: ValidationReport, ) -> None: """Validate common id references inside arbitrary params.""" for key, value, value_path in _walk_mapping(params, base_path=path): key_text = str(key) if _is_agent_reference_key(key_text): self._validate_reference_value( value=value, known_ids=agent_ids, path=value_path, report=report, code="unknown_agent_reference", label="agent", ) continue if _is_agent_reference_collection_key(key_text): self._validate_reference_collection( value=value, known_ids=agent_ids, path=value_path, report=report, code="unknown_agent_reference", label="agent", ) continue if _is_resource_reference_key(key_text): self._validate_reference_value( value=value, known_ids=resource_ids, path=value_path, report=report, code="unknown_resource_reference", label="resource", ) continue if _is_resource_reference_collection_key(key_text): self._validate_reference_collection( value=value, known_ids=resource_ids, path=value_path, report=report, code="unknown_resource_reference", label="resource", ) continue if _is_behavior_reference_key(key_text): self._validate_reference_value( value=value, known_ids=set(known_behavior_names), path=value_path, report=report, code="unknown_behavior_reference", label="behavior", skip_when_no_known_ids=True, ) continue if _is_behavior_reference_collection_key(key_text): self._validate_reference_collection( value=value, known_ids=set(known_behavior_names), path=value_path, report=report, code="unknown_behavior_reference", label="behavior", skip_when_no_known_ids=True, ) continue if key_text in {"target_types", "agent_types"}: self._validate_type_collection( value=value, known_types=agent_types, path=value_path, report=report, code="unknown_agent_type_reference", label="agent type", ) continue if key_text in {"resource_types", "target_resource_types"}: self._validate_type_collection( value=value, known_types=resource_types, path=value_path, report=report, code="unknown_resource_type_reference", label="resource type", ) def _validate_reference_value( self, *, value: Any, known_ids: set[str], path: str, report: ValidationReport, code: str, label: str, skip_when_no_known_ids: bool = False, ) -> None: """Validate one id-like reference value.""" if value is None: return if skip_when_no_known_ids and not known_ids: return if isinstance(value, str): if value not in known_ids: report.add( code=code, message=f"Unknown {label} reference {value!r}", path=path, severity=self.config.unresolved_reference_severity, details={ "reference": value, "known_ids": sorted(known_ids), }, ) return report.add( code=f"invalid_{label.replace(' ', '_')}_reference_type", message=f"{label.title()} reference must be a string or null", path=path, severity=self.config.unresolved_reference_severity, details={"value_type": value.__class__.__name__}, ) def _validate_reference_collection( self, *, value: Any, known_ids: set[str], path: str, report: ValidationReport, code: str, label: str, skip_when_no_known_ids: bool = False, ) -> None: """Validate a collection of id-like references.""" if value is None: return if skip_when_no_known_ids and not known_ids: return if isinstance(value, str): values = (value,) elif isinstance(value, Sequence) and not isinstance(value, (str, bytes)): values = tuple(value) else: report.add( code=f"invalid_{label.replace(' ', '_')}_reference_collection_type", message=f"{label.title()} reference collection must be a string or sequence", path=path, severity=self.config.unresolved_reference_severity, details={"value_type": value.__class__.__name__}, ) return for index, item in enumerate(values): item_path = f"{path}[{index}]" self._validate_reference_value( value=item, known_ids=known_ids, path=item_path, report=report, code=code, label=label, skip_when_no_known_ids=skip_when_no_known_ids, ) def _validate_type_collection( self, *, value: Any, known_types: set[str], path: str, report: ValidationReport, code: str, label: str, ) -> None: """Validate type labels used by target-type filters.""" if value is None or not known_types: return if isinstance(value, str): values = (value,) elif isinstance(value, Sequence) and not isinstance(value, (str, bytes)): values = tuple(str(item) for item in value) else: report.add( code=f"invalid_{label.replace(' ', '_')}_collection_type", message=f"{label.title()} collection must be a string or sequence", path=path, severity=self.config.suspicious_configuration_severity, details={"value_type": value.__class__.__name__}, ) return for index, item in enumerate(values): if item not in known_types: report.add( code=code, message=f"Unknown {label} {item!r}; no initial world object has this type", path=f"{path}[{index}]", severity=self.config.suspicious_configuration_severity, details={ "reference": item, "known_types": sorted(known_types), }, ) @staticmethod def _extract_rule_behavior_names(rule: Mapping[str, Any]) -> tuple[str, ...]: """Extract behavior names targeted by a policy rule mapping.""" names: list[str] = [] behavior_name = rule.get("behavior_name") if behavior_name is not None: names.append(str(behavior_name)) behavior_names = rule.get("behavior_names") if isinstance(behavior_names, str): names.append(behavior_names) elif isinstance(behavior_names, Iterable): names.extend(str(name) for name in behavior_names) return tuple(dict.fromkeys(name for name in names if name)) def validate_world_spec( spec: WorldSpec | Mapping[str, Any], *, config: ValidationConfig | None = None, ) -> ValidationReport: """Validate a world specification semantically and return a report.""" return DSLSemanticValidator(config=config or ValidationConfig()).validate(spec) def assert_valid_world_spec( spec: WorldSpec | Mapping[str, Any], *, config: ValidationConfig | None = None, ) -> WorldSpec: """Validate a world specification and raise when errors are present.""" return DSLSemanticValidator(config=config or ValidationConfig()).assert_valid(spec) def _constructor_signature_info(target: Any) -> tuple[set[str], set[str], bool] | None: """Return constructor signature information for a registry target. Returns: ``(allowed_params, required_params, accepts_var_keyword)`` or ``None`` when the signature cannot be inspected. """ 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 def _walk_mapping( value: Mapping[str, Any], *, base_path: str, ) -> tuple[tuple[str, Any, str], ...]: """Walk nested mappings and return ``(key, value, path)`` entries. Lists are traversed when they contain mappings. Scalar list values are not interpreted as named fields because they have no key. """ results: list[tuple[str, Any, str]] = [] def walk(current: Any, current_path: str) -> None: if isinstance(current, Mapping): for key, nested_value in current.items(): nested_path = f"{current_path}.{key}" results.append((str(key), nested_value, nested_path)) walk(nested_value, nested_path) return if isinstance(current, Sequence) and not isinstance(current, (str, bytes)): for index, item in enumerate(current): if isinstance(item, Mapping): walk(item, f"{current_path}[{index}]") walk(value, base_path) return tuple(results) def _is_agent_reference_key(key: str) -> bool: """Return whether a key names a single agent id reference.""" return key in AGENT_REFERENCE_KEYS or key.endswith("_agent_id") def _is_agent_reference_collection_key(key: str) -> bool: """Return whether a key names a collection of agent id references.""" return key in {"agent_ids", "target_agent_ids"} or key.endswith("_agent_ids") def _is_resource_reference_key(key: str) -> bool: """Return whether a key names a single resource id reference.""" return key in RESOURCE_REFERENCE_KEYS or key.endswith("_resource_id") def _is_resource_reference_collection_key(key: str) -> bool: """Return whether a key names a collection of resource id references.""" return key in {"resource_ids", "target_resource_ids"} or key.endswith("_resource_ids") def _is_behavior_reference_key(key: str) -> bool: """Return whether a key names a single behavior reference.""" return key in BEHAVIOR_REFERENCE_KEYS or key.endswith("_behavior_name") def _is_behavior_reference_collection_key(key: str) -> bool: """Return whether a key names a collection of behavior references.""" return key in {"behavior_names", "target_behavior_names"} or key.endswith("_behavior_names") __all__ = [ "AGENT_REFERENCE_KEYS", "BEHAVIOR_REFERENCE_KEYS", "BEHAVIOR_REGISTRY_ATTRIBUTES", "DEFAULT_BEHAVIOR_MODULES", "DEFAULT_POLICY_MODULES", "DSLValidationError", "DSLSemanticValidator", "POLICY_REGISTRY_ATTRIBUTES", "RESOURCE_REFERENCE_KEYS", "RegistryLoadResult", "ValidationConfig", "ValidationIssue", "ValidationReport", "ValidationSeverity", "assert_valid_world_spec", "load_default_behavior_registry", "load_default_policy_registry", "load_registry_from_modules", "validate_world_spec", ]