""" core.event ========== Generic event representation for WorldSmithAI. This module defines scheduled world-level events. Events are domain-agnostic: they do not know what "winter", "war", "breakthrough", "mana_storm", or "market_crash" means. Instead, events carry structured DSL payloads that are interpreted by the World through a generic effect-application interface. Example payload --------------- { "description": "A harsh winter reduces food and agent energy.", "effects": [ { "target": "resources", "operation": "multiply_amount", "selector": {"type": "food"}, "parameters": {"factor": 0.5} }, { "target": "agents", "operation": "increment_state", "selector": {"type": "farmer"}, "parameters": {"key": "energy", "amount": -0.2} } ] } The Event class does not implement those operations itself. The future World class will implement ``apply_event_effect(event=..., effect=...)`` and decide how structured effects mutate agents, resources, metrics, or world metadata. Minimal usage example --------------------- class MinimalWorld: step_count = 10 def apply_event_effect(self, *, event, effect): print(event.name, effect.operation) return {"success": True, "message": "Effect applied."} def record_event(self, *, event, result): print(result.to_dict()) event = Event( name="winter", trigger_step=10, payload={ "effects": [ { "target": "resources", "operation": "multiply_amount", "selector": {"type": "food"}, "parameters": {"factor": 0.5}, } ] }, ) result = event.execute(MinimalWorld()) print(result.success) """ from __future__ import annotations import logging from collections.abc import Mapping, Sequence from copy import deepcopy from dataclasses import dataclass, field from numbers import Integral from typing import Any, Protocol, TypeAlias logger = logging.getLogger(__name__) EventPayload: TypeAlias = dict[str, Any] EventMetadata: TypeAlias = dict[str, Any] _RESERVED_EFFECT_KEYS: frozenset[str] = frozenset( { "target", "operation", "selector", "parameters", "metadata", } ) @dataclass(frozen=True, slots=True) class EventEffect: """ Normalized structured effect contained in an event payload. An event may contain one or more effects. Each effect describes what kind of world object should be targeted and what generic operation should be applied. The concrete World implementation decides how to interpret each operation. Attributes ---------- target: Generic target category, such as ``"agents"``, ``"resources"``, ``"world"``, ``"metrics"``, or another future target type. operation: Generic operation name, such as ``"increment_state"``, ``"multiply_amount"``, ``"spawn_agent"``, or ``"set_metadata"``. selector: Generic selector used by the world to find affected objects. Example: ``{"type": "food"}`` or ``{"type": "scientist"}``. parameters: Operation parameters interpreted by the world. metadata: Optional metadata for logging, metrics, visualization, or narration. """ target: str operation: str selector: Mapping[str, Any] = field(default_factory=dict) parameters: Mapping[str, Any] = field(default_factory=dict) metadata: Mapping[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: """ Validate and normalize event effect fields. Raises ------ ValueError If target or operation is empty. TypeError If selector, parameters, or metadata are invalid. """ if not isinstance(self.target, str) or not self.target.strip(): raise ValueError("EventEffect.target must be a non-empty string.") if not isinstance(self.operation, str) or not self.operation.strip(): raise ValueError("EventEffect.operation must be a non-empty string.") if not isinstance(self.selector, Mapping): raise TypeError("EventEffect.selector must be a mapping.") if not isinstance(self.parameters, Mapping): raise TypeError("EventEffect.parameters must be a mapping.") if not isinstance(self.metadata, Mapping): raise TypeError("EventEffect.metadata must be a mapping.") selector = dict(self.selector) parameters = dict(self.parameters) metadata = dict(self.metadata) self._validate_string_keys(selector, "selector") self._validate_string_keys(parameters, "parameters") self._validate_string_keys(metadata, "metadata") object.__setattr__(self, "target", self.target.strip()) object.__setattr__(self, "operation", self.operation.strip()) object.__setattr__(self, "selector", selector) object.__setattr__(self, "parameters", parameters) object.__setattr__(self, "metadata", metadata) @classmethod def from_mapping(cls, raw_effect: Mapping[str, Any]) -> EventEffect: """ Create an EventEffect from a raw DSL mapping. Extra top-level keys that are not ``target``, ``operation``, ``selector``, ``parameters``, or ``metadata`` are folded into ``parameters``. This makes the DSL forgiving while preserving a clean runtime representation. Parameters ---------- raw_effect: Raw effect mapping from the event payload. Returns ------- EventEffect Normalized event effect. Raises ------ TypeError If raw_effect is not a mapping. KeyError If required fields are missing. """ if not isinstance(raw_effect, Mapping): raise TypeError("Event effect must be a mapping.") if "target" not in raw_effect: raise KeyError("Event effect is missing required key 'target'.") if "operation" not in raw_effect: raise KeyError("Event effect is missing required key 'operation'.") selector = raw_effect.get("selector", {}) parameters = raw_effect.get("parameters", {}) metadata = raw_effect.get("metadata", {}) if selector is None: selector = {} if parameters is None: parameters = {} if metadata is None: metadata = {} if not isinstance(selector, Mapping): raise TypeError("Event effect 'selector' must be a mapping.") if not isinstance(parameters, Mapping): raise TypeError("Event effect 'parameters' must be a mapping.") if not isinstance(metadata, Mapping): raise TypeError("Event effect 'metadata' must be a mapping.") extra_parameters = { key: value for key, value in raw_effect.items() if key not in _RESERVED_EFFECT_KEYS } merged_parameters = { **dict(parameters), **extra_parameters, } return cls( target=str(raw_effect["target"]), operation=str(raw_effect["operation"]), selector=dict(selector), parameters=merged_parameters, metadata=dict(metadata), ) def to_dict(self) -> dict[str, Any]: """ Convert the effect to a JSON-friendly dictionary. Returns ------- dict[str, Any] Serializable event effect. """ return { "target": self.target, "operation": self.operation, "selector": deepcopy(dict(self.selector)), "parameters": deepcopy(dict(self.parameters)), "metadata": deepcopy(dict(self.metadata)), } @staticmethod def _validate_string_keys(mapping: Mapping[str, Any], label: str) -> None: """ Validate that all mapping keys are strings. Parameters ---------- mapping: Mapping to validate. label: Human-readable mapping label for error messages. Raises ------ TypeError If any key is not a string. """ for key in mapping: if not isinstance(key, str): raise TypeError(f"EventEffect.{label} keys must be strings.") @dataclass(frozen=True, slots=True) class EventEffectResult: """ Structured result of applying one event effect. Attributes ---------- target: Effect target category. operation: Effect operation name. success: Whether this specific effect was applied successfully. message: Human-readable effect result summary. affected_count: Optional number of affected world objects. metadata: Additional structured result metadata. """ target: str operation: str success: bool message: str = "" affected_count: int | None = None metadata: Mapping[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: """ Validate and normalize effect result fields. Raises ------ ValueError If target, operation, or affected_count are invalid. TypeError If metadata is invalid. """ if not isinstance(self.target, str) or not self.target.strip(): raise ValueError("EventEffectResult.target must be a non-empty string.") if not isinstance(self.operation, str) or not self.operation.strip(): raise ValueError("EventEffectResult.operation must be a non-empty string.") if not isinstance(self.success, bool): raise TypeError("EventEffectResult.success must be a boolean.") if self.affected_count is not None: if isinstance(self.affected_count, bool) or not isinstance( self.affected_count, Integral, ): raise TypeError("EventEffectResult.affected_count must be an integer.") if self.affected_count < 0: raise ValueError("EventEffectResult.affected_count cannot be negative.") if not isinstance(self.metadata, Mapping): raise TypeError("EventEffectResult.metadata must be a mapping.") metadata = dict(self.metadata) for key in metadata: if not isinstance(key, str): raise TypeError("EventEffectResult.metadata keys must be strings.") object.__setattr__(self, "target", self.target.strip()) object.__setattr__(self, "operation", self.operation.strip()) object.__setattr__(self, "metadata", metadata) @classmethod def from_raw( cls, *, effect: EventEffect, raw_result: EventEffectResult | Mapping[str, Any] | None, ) -> EventEffectResult: """ Normalize a raw world effect result. Parameters ---------- effect: Effect that was applied. raw_result: Value returned by ``world.apply_event_effect(...)``. Returns ------- EventEffectResult Normalized effect result. Raises ------ TypeError If raw_result has an unsupported type. """ if raw_result is None: return cls( target=effect.target, operation=effect.operation, success=True, message="World effect handler returned no explicit result.", ) if isinstance(raw_result, EventEffectResult): return raw_result if isinstance(raw_result, Mapping): metadata = raw_result.get("metadata", {}) if metadata is None: metadata = {} if not isinstance(metadata, Mapping): raise TypeError("Raw effect result metadata must be a mapping.") affected_count = raw_result.get("affected_count") return cls( target=str(raw_result.get("target", effect.target)), operation=str(raw_result.get("operation", effect.operation)), success=bool(raw_result.get("success", True)), message=str(raw_result.get("message", "")), affected_count=affected_count, metadata=dict(metadata), ) raise TypeError( "world.apply_event_effect() must return EventEffectResult, " "a mapping, or None." ) def to_dict(self) -> dict[str, Any]: """ Convert the effect result to a JSON-friendly dictionary. Returns ------- dict[str, Any] Serializable effect result. """ return { "target": self.target, "operation": self.operation, "success": self.success, "message": self.message, "affected_count": self.affected_count, "metadata": deepcopy(dict(self.metadata)), } @dataclass(frozen=True, slots=True) class EventExecutionResult: """ Structured result of executing one event. Attributes ---------- event_id: Stable event identifier. event_name: Human-readable event name. trigger_step: Step at which the event becomes due. executed_step: Step at which execution was attempted. success: Whether the event execution succeeded. skipped: Whether execution was skipped because the event was not due, disabled, or exhausted. message: Human-readable execution summary. effect_results: Results for individual event effects. payload: Snapshot of the event payload at execution time. metadata: Additional execution metadata. """ event_id: str event_name: str trigger_step: int executed_step: int | None success: bool skipped: bool = False message: str = "" effect_results: tuple[EventEffectResult, ...] = field(default_factory=tuple) payload: Mapping[str, Any] = field(default_factory=dict) metadata: Mapping[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: """ Validate and normalize event execution result fields. Raises ------ ValueError If identifiers or step values are invalid. TypeError If mappings or effect results are invalid. """ if not isinstance(self.event_id, str) or not self.event_id.strip(): raise ValueError("EventExecutionResult.event_id must be non-empty.") if not isinstance(self.event_name, str) or not self.event_name.strip(): raise ValueError("EventExecutionResult.event_name must be non-empty.") if isinstance(self.trigger_step, bool) or not isinstance( self.trigger_step, Integral, ): raise TypeError("EventExecutionResult.trigger_step must be an integer.") if self.trigger_step < 0: raise ValueError("EventExecutionResult.trigger_step cannot be negative.") if self.executed_step is not None: if isinstance(self.executed_step, bool) or not isinstance( self.executed_step, Integral, ): raise TypeError("EventExecutionResult.executed_step must be an integer.") if self.executed_step < 0: raise ValueError( "EventExecutionResult.executed_step cannot be negative." ) if not isinstance(self.success, bool): raise TypeError("EventExecutionResult.success must be a boolean.") if not isinstance(self.skipped, bool): raise TypeError("EventExecutionResult.skipped must be a boolean.") if not isinstance(self.effect_results, tuple): object.__setattr__(self, "effect_results", tuple(self.effect_results)) for result in self.effect_results: if not isinstance(result, EventEffectResult): raise TypeError( "EventExecutionResult.effect_results must contain " "EventEffectResult objects." ) if not isinstance(self.payload, Mapping): raise TypeError("EventExecutionResult.payload must be a mapping.") if not isinstance(self.metadata, Mapping): raise TypeError("EventExecutionResult.metadata must be a mapping.") object.__setattr__(self, "event_id", self.event_id.strip()) object.__setattr__(self, "event_name", self.event_name.strip()) object.__setattr__(self, "trigger_step", int(self.trigger_step)) object.__setattr__( self, "executed_step", None if self.executed_step is None else int(self.executed_step), ) object.__setattr__(self, "payload", deepcopy(dict(self.payload))) object.__setattr__(self, "metadata", deepcopy(dict(self.metadata))) def to_dict(self) -> dict[str, Any]: """ Convert the execution result to a JSON-friendly dictionary. Returns ------- dict[str, Any] Serializable event execution result. """ return { "event_id": self.event_id, "event_name": self.event_name, "trigger_step": self.trigger_step, "executed_step": self.executed_step, "success": self.success, "skipped": self.skipped, "message": self.message, "effect_results": [ effect_result.to_dict() for effect_result in self.effect_results ], "payload": deepcopy(dict(self.payload)), "metadata": deepcopy(dict(self.metadata)), } class EventWorldProtocol(Protocol): """ Structural protocol for a world capable of executing event effects. The concrete World class will be implemented in ``core/world.py``. This protocol avoids importing the concrete World class here and keeps the event module loosely coupled. Required attributes ------------------- step_count: Current simulation step. Optional methods used by Event ------------------------------ apply_event_effect: Applies one normalized event effect to the world. record_event: Records the final event execution result. """ step_count: int def apply_event_effect( self, *, event: Event, effect: EventEffect, ) -> EventEffectResult | Mapping[str, Any] | None: """ Apply one event effect to the world. Parameters ---------- event: Event currently executing. effect: Normalized event effect to apply. Returns ------- EventEffectResult | Mapping[str, Any] | None Structured result, compatible mapping, or None. """ ... def record_event( self, *, event: Event, result: EventExecutionResult, ) -> None: """ Record an event execution result. Parameters ---------- event: Event that executed. result: Structured execution result. """ ... @dataclass(slots=True) class Event: """ Generic scheduled world event. Events are domain-agnostic and DSL-driven. They define when something should happen and carry a payload describing structured effects. The World decides how to apply those effects. Attributes ---------- name: Human-readable event name, such as ``"winter"`` or ``"breakthrough"``. trigger_step: First simulation step at which the event becomes due. payload: Generic DSL payload. Common keys include ``"description"`` and ``"effects"``. id: Optional stable event id. If omitted, one is derived from name and trigger step. enabled: Whether this event is active. repeat_interval: Optional positive interval for recurring events. If None, the event is one-time unless manually reset. max_executions: Maximum number of executions. Defaults to 1. Use None for unlimited. executed_steps: Steps at which this event has already executed. """ name: str trigger_step: int payload: EventPayload = field(default_factory=dict) id: str | None = None enabled: bool = True repeat_interval: int | None = None max_executions: int | None = 1 executed_steps: list[int] = field(default_factory=list) def __post_init__(self) -> None: """ Validate and normalize event fields. Raises ------ ValueError If name, id, trigger_step, repeat_interval, max_executions, or executed_steps are invalid. TypeError If payload or lifecycle fields have invalid types. """ if not isinstance(self.name, str) or not self.name.strip(): raise ValueError("Event.name must be a non-empty string.") self.name = self.name.strip() if self.id is None: self.id = self._default_event_id(self.name, self.trigger_step) elif not isinstance(self.id, str) or not self.id.strip(): raise ValueError("Event.id must be None or a non-empty string.") else: self.id = self.id.strip() self.trigger_step = self._normalize_step( self.trigger_step, label="trigger_step", ) if not isinstance(self.payload, Mapping): raise TypeError("Event.payload must be a mapping.") self.payload = deepcopy(dict(self.payload)) self._validate_string_keys(self.payload, "payload") if not isinstance(self.enabled, bool): raise TypeError("Event.enabled must be a boolean.") if self.repeat_interval is not None: self.repeat_interval = self._normalize_positive_integer( self.repeat_interval, label="repeat_interval", ) if self.max_executions is not None: self.max_executions = self._normalize_positive_integer( self.max_executions, label="max_executions", ) if not isinstance(self.executed_steps, list): raise TypeError("Event.executed_steps must be a list.") normalized_steps = [ self._normalize_step(step, label="executed_step") for step in self.executed_steps ] self.executed_steps = sorted(set(normalized_steps)) if ( self.max_executions is not None and len(self.executed_steps) > self.max_executions ): raise ValueError( "Event.executed_steps cannot exceed Event.max_executions." ) # Validate effects eagerly so malformed DSL fails at construction time. _ = self.effects @property def description(self) -> str: """ Optional human-readable event description. Returns ------- str Event description from payload, or an empty string. """ description = self.payload.get("description", "") return "" if description is None else str(description) @property def effects(self) -> tuple[EventEffect, ...]: """ Normalized event effects from payload. The payload may contain either: - ``"effects"``: a list of effect mappings - ``"effect"``: a single effect mapping Returns ------- tuple[EventEffect, ...] Normalized event effects. Raises ------ TypeError If the effects payload is malformed. """ raw_effects: Any if "effects" in self.payload: raw_effects = self.payload["effects"] elif "effect" in self.payload: raw_effects = self.payload["effect"] else: raw_effects = [] if raw_effects is None: return tuple() if isinstance(raw_effects, Mapping): return (EventEffect.from_mapping(raw_effects),) if isinstance(raw_effects, (str, bytes)): raise TypeError("Event payload effects cannot be a string.") if not isinstance(raw_effects, Sequence): raise TypeError("Event payload effects must be a sequence or mapping.") return tuple(EventEffect.from_mapping(effect) for effect in raw_effects) @property def execution_count(self) -> int: """ Number of times this event has executed. Returns ------- int Count of recorded execution steps. """ return len(self.executed_steps) def is_due(self, step_count: int) -> bool: """ Return whether this event should execute at the given step. Parameters ---------- step_count: Current simulation step. Returns ------- bool True if the event is due, otherwise False. """ current_step = self._normalize_step(step_count, label="step_count") if not self.enabled: return False if self._has_reached_execution_limit(): return False if current_step < self.trigger_step: return False if not self.executed_steps: return True if self.repeat_interval is None: return False last_execution_step = max(self.executed_steps) return current_step - last_execution_step >= self.repeat_interval def execute( self, world: EventWorldProtocol, *, force: bool = False, ) -> EventExecutionResult: """ Execute this event against a world. Execution delegates each normalized EventEffect to ``world.apply_event_effect(event=self, effect=effect)``. Parameters ---------- world: World-like object with ``step_count`` and optional event handling methods. force: If True, execute even when ``is_due(...)`` is False. Returns ------- EventExecutionResult Structured execution result. """ step_count = self._get_world_step_count(world) if step_count is None: result = self._make_execution_result( executed_step=None, success=False, skipped=True, message="Event could not execute because world.step_count is missing.", effect_results=tuple(), metadata={"reason": "missing_world_step_count"}, ) return result if not isinstance(force, bool): raise TypeError("force must be a boolean.") if not force and not self.is_due(step_count): return self._make_execution_result( executed_step=step_count, success=False, skipped=True, message=self._not_due_reason(step_count), effect_results=tuple(), metadata={"reason": "not_due"}, ) effect_results: list[EventEffectResult] = [] for effect in self.effects: try: effect_result = self._apply_effect(world=world, effect=effect) except Exception as exc: logger.exception( "Event effect failed: event_id=%s event_name=%s " "target=%s operation=%s", self.id, self.name, effect.target, effect.operation, ) effect_result = EventEffectResult( target=effect.target, operation=effect.operation, success=False, message="Event effect raised an exception.", metadata={"error": repr(exc)}, ) effect_results.append(effect_result) success = all(result.success for result in effect_results) if not effect_results: success = True message = self._execution_message( success=success, effect_results=tuple(effect_results), ) self.mark_executed(step_count) result = self._make_execution_result( executed_step=step_count, success=success, skipped=False, message=message, effect_results=tuple(effect_results), metadata={ "forced": force, "description": self.description, "effect_count": len(effect_results), }, ) self._record_result(world=world, result=result) return result def mark_executed(self, step_count: int) -> None: """ Record that this event executed at a given step. Parameters ---------- step_count: Simulation step at which execution occurred. """ normalized_step = self._normalize_step(step_count, label="step_count") if normalized_step not in self.executed_steps: self.executed_steps.append(normalized_step) self.executed_steps.sort() def reset(self) -> None: """ Clear execution history for this event. This is useful in tests, replay systems, or interactive Gradio sessions where users regenerate and rerun the same world. """ self.executed_steps.clear() def disable(self) -> None: """ Disable this event. """ self.enabled = False def enable(self) -> None: """ Enable this event. """ self.enabled = True def snapshot(self) -> dict[str, Any]: """ Return a JSON-friendly event snapshot. Returns ------- dict[str, Any] Serializable event state. """ return { "id": self.id, "name": self.name, "trigger_step": self.trigger_step, "payload": deepcopy(self.payload), "enabled": self.enabled, "repeat_interval": self.repeat_interval, "max_executions": self.max_executions, "executed_steps": list(self.executed_steps), "execution_count": self.execution_count, } def _apply_effect( self, *, world: EventWorldProtocol, effect: EventEffect, ) -> EventEffectResult: """ Apply one event effect through the world handler. Parameters ---------- world: World-like object. effect: Normalized event effect. Returns ------- EventEffectResult Structured effect result. """ handler = getattr(world, "apply_event_effect", None) if not callable(handler): return EventEffectResult( target=effect.target, operation=effect.operation, success=False, message=( "World does not implement apply_event_effect(); " "event effects cannot be applied." ), metadata={"reason": "missing_world_effect_handler"}, ) raw_result = handler(event=self, effect=effect) return EventEffectResult.from_raw(effect=effect, raw_result=raw_result) def _record_result( self, *, world: EventWorldProtocol, result: EventExecutionResult, ) -> None: """ Record the event execution result on the world if supported. Parameters ---------- world: World-like object. result: Event execution result. """ recorder = getattr(world, "record_event", None) if not callable(recorder): return try: recorder(event=self, result=result) except Exception: logger.exception( "World failed to record event result: event_id=%s event_name=%s", self.id, self.name, ) def _make_execution_result( self, *, executed_step: int | None, success: bool, skipped: bool, message: str, effect_results: tuple[EventEffectResult, ...], metadata: Mapping[str, Any] | None = None, ) -> EventExecutionResult: """ Build an EventExecutionResult for this event. Parameters ---------- executed_step: Step at which execution was attempted. success: Whether execution succeeded. skipped: Whether execution was skipped. message: Human-readable execution summary. effect_results: Per-effect execution results. metadata: Optional structured metadata. Returns ------- EventExecutionResult Normalized event execution result. """ return EventExecutionResult( event_id=self.id or self._default_event_id(self.name, self.trigger_step), event_name=self.name, trigger_step=self.trigger_step, executed_step=executed_step, success=success, skipped=skipped, message=message, effect_results=effect_results, payload=deepcopy(self.payload), metadata=dict(metadata or {}), ) def _not_due_reason(self, step_count: int) -> str: """ Return a human-readable reason for why the event is not due. Parameters ---------- step_count: Current simulation step. Returns ------- str Explanation string. """ if not self.enabled: return "Event is disabled." if self._has_reached_execution_limit(): return "Event has reached its execution limit." if step_count < self.trigger_step: return "Event trigger step has not been reached." if self.repeat_interval is None and self.executed_steps: return "One-time event has already executed." if self.repeat_interval is not None and self.executed_steps: last_execution_step = max(self.executed_steps) next_due_step = last_execution_step + self.repeat_interval return f"Recurring event is next due at step {next_due_step}." return "Event is not due." def _execution_message( self, *, success: bool, effect_results: tuple[EventEffectResult, ...], ) -> str: """ Build a human-readable execution message. Parameters ---------- success: Whether the event succeeded. effect_results: Per-effect results. Returns ------- str Execution message. """ if not effect_results: return "Event executed with no effects." if success: return f"Event executed successfully with {len(effect_results)} effect(s)." failed_count = sum(1 for result in effect_results if not result.success) return ( f"Event executed with {failed_count} failed effect(s) out of " f"{len(effect_results)}." ) def _has_reached_execution_limit(self) -> bool: """ Return whether the event reached max_executions. Returns ------- bool True if no further executions are allowed. """ return ( self.max_executions is not None and len(self.executed_steps) >= self.max_executions ) @staticmethod def _get_world_step_count(world: EventWorldProtocol) -> int | None: """ Safely read the current world step count. Parameters ---------- world: World-like object. Returns ------- int | None Current step count if available. """ step_count = getattr(world, "step_count", None) if isinstance(step_count, bool) or not isinstance(step_count, Integral): return None if step_count < 0: return None return int(step_count) @staticmethod def _default_event_id(name: str, trigger_step: int) -> str: """ Build a stable default event id. Parameters ---------- name: Event name. trigger_step: Event trigger step. Returns ------- str Default event id. """ normalized_name = "_".join(name.strip().lower().split()) return f"{normalized_name}@{trigger_step}" @staticmethod def _normalize_step(value: Any, *, label: str) -> int: """ Normalize a step value into a non-negative integer. Parameters ---------- value: Candidate step value. label: Human-readable label for error messages. Returns ------- int Normalized non-negative integer. Raises ------ TypeError If value is not an integer. ValueError If value is negative. """ if isinstance(value, bool) or not isinstance(value, Integral): raise TypeError(f"Event.{label} must be an integer.") normalized = int(value) if normalized < 0: raise ValueError(f"Event.{label} cannot be negative.") return normalized @staticmethod def _normalize_positive_integer(value: Any, *, label: str) -> int: """ Normalize a value into a positive integer. Parameters ---------- value: Candidate value. label: Human-readable label for error messages. Returns ------- int Positive integer. Raises ------ TypeError If value is not an integer. ValueError If value is not positive. """ if isinstance(value, bool) or not isinstance(value, Integral): raise TypeError(f"Event.{label} must be an integer.") normalized = int(value) if normalized <= 0: raise ValueError(f"Event.{label} must be positive.") return normalized @staticmethod def _validate_string_keys(mapping: Mapping[str, Any], label: str) -> None: """ Validate that all mapping keys are strings. Parameters ---------- mapping: Mapping to validate. label: Human-readable mapping label for error messages. Raises ------ TypeError If any key is not a string. """ for key in mapping: if not isinstance(key, str): raise TypeError(f"Event.{label} keys must be strings.") __all__ = [ "Event", "EventEffect", "EventEffectResult", "EventExecutionResult", "EventMetadata", "EventPayload", "EventWorldProtocol", ]