feat: add optional types and validation for component ports

docs/usage/key-concepts.md

@@ -21,6 +21,128 @@ When implementing your own components, you will need to:
 * Optionally define an `init()` method to do any required preparatory steps before the model in run.
 * In the case of event based models, define custom [`Event`][plugboard.events.Event] subclasses and corresponding event handler methods decorated with `Event.handler`.
 
+#### Optional input and output types
+
+Declare port types by passing a mapping of names to Python types to `IOController`:
+
+```python
+from plugboard.component import Component, IOController as IO
+
+
+class Double(Component):
+    io = IO(inputs={"value": int}, outputs={"result": int})
+
+    async def step(self) -> None:
+        self.result = self.value * 2
+```
+
+Existing lists of names still work: `IO(inputs=["value"], outputs=["result"])`
+declares both ports as `typing.Any`. You can mix a list on one side with a typed
+mapping on the other, or use `Any` explicitly for individual ports. Attribute
+annotations alone do not declare port types or directions. Mapping declarations
+provide runtime metadata; add ordinary attribute annotations separately if you
+want your editor or static type checker to check `self.value` and `self.result`.
+
+Supported types are `Any`, Python classes, unions such as `int | None`, and
+`list[T]`, `dict[K, V]`, `set[T]`, `frozenset[T]`, fixed-length `tuple[T, U]`, and
+variable-length `tuple[T, ...]`. These can be nested. `None` in a declaration
+means `type(None)`. Other typing constructs, including string forward references,
+`TypeVar`, `Literal`, `Annotated`, `TypedDict`, and protocols, raise `IOSetupError`.
+Use an ordinary class for a nominal check of an array, dataframe, or other custom
+payload; its internal contents, shape, and dtype are not inspected.
+
+Ports inherited from a base component retain their types. A name-only declaration
+does not erase an inherited type. Adding a type to an inherited `Any` port is
+allowed; conflicting explicit types raise `IOSetupError`, including conflicts
+between multiple base classes.
+
+The same syntax works for ports chosen at instance construction. For example,
+this component creates its output names from a configuration list:
+
+```python
+from plugboard.schemas import ComponentArgsDict
+from typing import Unpack
+
+
+class DynamicSource(Component):
+    # A class declaration is still needed for the component lifecycle.
+    io = IO(outputs=["ready"])
+
+    def __init__(self, fields: list[str], **kwargs: Unpack[ComponentArgsDict]) -> None:
+        super().__init__(**kwargs)
+        self.io = IO(
+            outputs={"ready": bool, **{name: int for name in fields}},
+            input_events=self.io.input_events,
+            output_events=self.io.output_events,
+            namespace=self.name,
+            component=self,
+            runtime_validation="deep",
+            validate_outputs=True,
+        )
+        self._fields = fields
+
+    async def step(self) -> None:
+        self.ready = True
+        for field in self._fields:
+            setattr(self, field, 42)
+```
+
+Keep constructor arguments serializable when using YAML configuration or JSON
+state backends. The example accepts a list of names and constructs the Python
+types inside the component. Type declarations in component code also work when
+the component is selected through YAML; there is no YAML type-expression syntax.
+
+#### Type validation
+
+Process initialization checks declared output-to-input compatibility, even when
+runtime checking is disabled. For example, `Dog` output can connect to `Animal`
+input when `Dog` subclasses `Animal`; the reverse fails with `ValidationError`.
+All possible members of an output union must be accepted by the input. Mutable
+containers (`list`, `dict`, `set`) are invariant, so `list[Dog]` cannot connect to
+`list[Animal]`; `tuple` and `frozenset` allow narrower element types. Each edge of
+a fan-in or fan-out is checked. `Any` and unknown element types of bare containers
+allow a connection without proving its compatibility.
+
+Runtime value checks are **off by default**. Enable them per controller:
+
+```python
+io = IO(
+    inputs={"samples": list[float], "label": str | None},
+    outputs={"average": float},
+    runtime_validation="deep",
+    validate_outputs=True,
+)
+```
+
+| Option | Behavior |
+| --- | --- |
+| `runtime_validation="off"` | No runtime value checks; initialization still checks declarations. |
+| `runtime_validation="shallow"` | Check each supplied input's outer type, without checking container contents. |
+| `runtime_validation="deep"` | Also check supported container contents and tuple lengths. |
+| `validate_outputs=True` | Apply the selected runtime mode to outputs before buffering and sending. Defaults to `False`. |
+
+Runtime validation uses `isinstance` semantics: subclasses are accepted, including
+`bool` for `int`. There is no numeric widening or coercion: an `int` is not a
+`float`, and a string containing digits is not an `int`. Union members follow the
+same rules. Values are checked in place, without converting or copying them.
+
+Input checks cover channel messages, initial values, and direct assignments
+(including event handlers) when inputs are bound for a step. A missing input is
+not checked, while an explicitly supplied `None` must match its declared type.
+Output checks run once per port before fan-out. Failures raise `ValidationError`
+with the component, port, expected and actual types, and a nested path where
+available, and mark the component as failed. Union failures report the union at
+the point where no alternative matches. Process execution may wrap failures in an
+`ExceptionGroup` or a Ray task error.
+
+Validators are built when the controller is created and reused for every value;
+untyped ports have no validators. Deep checks visit container contents on every
+message, so their cost grows with payload size. Use shallow checks for large
+payloads when checking only their outer class is sufficient. Validation policy
+belongs to each `IOController` declaration: a subclass declaring a new controller
+should specify its desired policy, and dynamically replaced controllers should
+do the same. Class controllers pass their policy to component instances.
+
 ### Connectors
 
 Data flows between components via **Connectors**. You will need to use these to tell Plugboard how data should flow between your components. Your overall model might be very simple, for example:

plugboard/component/component.py

@@ -90,14 +90,22 @@ def __init__(
             except IOSetupError:
                 pass
         self.io = IO(
-            inputs=self.__class__.io.inputs,
-            outputs=self.__class__.io.outputs,
+            inputs={
+                name: self.__class__.io.input_types.get(name, _t.Any)
+                for name in self.__class__.io.inputs
+            },
+            outputs={
+                name: self.__class__.io.output_types.get(name, _t.Any)
+                for name in self.__class__.io.outputs
+            },
             initial_values=self._initial_values,
             input_events=self.__class__.io.input_events,
             output_events=self.__class__.io.output_events,
             event_field_coverage=self.__class__.io.event_field_coverage,
             namespace=self.name,
             component=self,
+            runtime_validation=self.__class__.io.runtime_validation,
+            validate_outputs=self.__class__.io.validate_outputs,
         )
         self.resources = resources or Resource(
             cpu=self.__class__.resources.cpu,
@@ -174,13 +182,36 @@ def _configure_io(cls) -> None:
         # Get all parent classes that are Component subclasses
         io_args, exports = cls._get_aggregated_io_args()
         event_field_coverage = cls._get_event_field_coverage()
+        port_types: dict[str, dict[str, _t.Any]] = {"inputs": {}, "outputs": {}}
+        for base in reversed(cls.__mro__):
+            base_io = base.__dict__.get("io")
+            if not isinstance(base_io, IO):
+                continue
+            for direction, types in port_types.items():
+                for field, hint in getattr(base_io, f"{direction[:-1]}_types").items():
+                    previous = types.get(field, _t.Any)
+                    if previous is not _t.Any and hint is not _t.Any and previous != hint:
+                        raise IOSetupError(
+                            f"Conflicting inherited types for {cls.__name__}.{field}: "
+                            f"{previous!r} and {hint!r}"
+                        )
+                    if hint is not _t.Any or field not in types:
+                        types[field] = hint
+        runtime_validation = cls.io.runtime_validation
+        validate_outputs = cls.io.validate_outputs
         # Set io arguments for subclass
         cls.io = IO(
-            inputs=sorted(io_args["inputs"], key=str),
-            outputs=sorted(io_args["outputs"], key=str),
+            inputs={
+                name: port_types["inputs"].get(name, _t.Any) for name in sorted(io_args["inputs"])
+            },
+            outputs={
+                name: port_types["outputs"].get(name, _t.Any) for name in sorted(io_args["outputs"])
+            },
             input_events=_t.cast(list[_t.Type[Event]], sorted(io_args["input_events"], key=str)),
             output_events=_t.cast(list[_t.Type[Event]], sorted(io_args["output_events"], key=str)),
             event_field_coverage=event_field_coverage,
+            runtime_validation=runtime_validation,
+            validate_outputs=validate_outputs,
         )
         # Set exports for subclass
         cls.exports = sorted(set(exports))
@@ -364,15 +395,15 @@ async def _wrapper() -> None:
                 await self._set_status(Status.RUNNING, publish=not self._is_running)
                 await self._io_read_with_status_check()
                 await self._handle_events()
-                self._bind_inputs()
-                if self._can_step:
-                    try:
+                try:
+                    self._bind_inputs()
+                    if self._can_step:
                         await self._step()
-                    except Exception as e:
-                        await self._set_status(Status.FAILED)
-                        self._logger.exception("Component step failed")
-                        raise e
-                self._bind_outputs()
+                    self._bind_outputs()
+                except Exception:
+                    await self._set_status(Status.FAILED)
+                    self._logger.exception("Component step failed")
+                    raise
                 await self.io.write()
                 self._reset_input_trackers()
                 await self._set_status(Status.WAITING, publish=not self._is_running)
@@ -473,6 +504,8 @@ def _bind_inputs(self) -> None:
         # Consume input data directly assigned and read from channels and reset to empty values
         received_inputs = dict(self.io.buf_fields[_io_key_in].flush())
         self._field_inputs.update(received_inputs)
+        if self.io._input_validators:
+            self.io.validate_inputs(self._field_inputs)
         # Check if all input fields have been set
         self._field_inputs_ready = all(k in self._field_inputs for k in self.io.inputs)
         for field in self.io.inputs:
@@ -489,6 +522,8 @@ def _bind_outputs(self) -> None:
         if not self._can_step:
             return
         output_data = {field: getattr(self, field, None) for field in self.io.outputs}
+        if self.io._output_validators:
+            self.io.validate_output_values(output_data)
         self.io.buf_fields[_io_key_out].put(output_data.items())
 
     async def _handle_events(self) -> None:

plugboard/component/io_controller.py

@@ -7,6 +7,13 @@
 from functools import cache, cached_property
 import typing as _t
 
+from plugboard.component.port_types import (
+    PortDeclaration,
+    RuntimeValidation,
+    compile_validator,
+    normalize_ports,
+    type_name,
+)
 from plugboard.connector import AsyncioChannel, Channel, Connector
 from plugboard.events import Event, StopEvent
 from plugboard.exceptions import ChannelClosedError, IOStreamClosedError
@@ -29,22 +36,44 @@
 
 
 class IOController:
-    """`IOController` manages input/output to/from components."""
+    """`IOController` manages input/output to/from components.
+
+    Args:
+        inputs: Port names (implicitly Any), or a mapping of names to Python types.
+        outputs: Port names (implicitly Any), or a mapping of names to Python types.
+        runtime_validation: Input value checking: off (default), shallow (outer type),
+            or deep (including built-in container contents). Values are never coerced.
+        validate_outputs: Also check outputs before buffering, using runtime_validation.
+            Defaults to False. Each output is checked once before fan-out.
+    """
 
     def __init__(
         self,
-        inputs: _t.Optional[_t.Any] = None,
-        outputs: _t.Optional[_t.Any] = None,
+        inputs: PortDeclaration | None = None,
+        outputs: PortDeclaration | None = None,
         initial_values: _t.Optional[dict[str, _t.Iterable]] = None,
         input_events: _t.Optional[list[_t.Type[Event]]] = None,
         output_events: _t.Optional[list[_t.Type[Event]]] = None,
         event_field_coverage: _t.Optional[dict[str, list[str]]] = None,
         namespace: str = IO_NS_UNSET,
         component: _t.Optional[Component] = None,
+        *,
+        runtime_validation: RuntimeValidation = "off",
+        validate_outputs: bool = False,
     ) -> None:
         self.namespace = namespace
-        self.inputs = inputs or []
-        self.outputs = outputs or []
+        self.input_types = normalize_ports(inputs)
+        self.output_types = normalize_ports(outputs)
+        self.inputs = list(self.input_types)
+        self.outputs = list(self.output_types)
+        if runtime_validation not in ("off", "shallow", "deep"):
+            raise ValueError(f"Unknown runtime validation mode: {runtime_validation!r}")
+        self.runtime_validation = runtime_validation
+        self.validate_outputs = validate_outputs
+        self._input_validators = self._compile_validators(self.input_types)
+        self._output_validators = (
+            self._compile_validators(self.output_types) if validate_outputs else {}
+        )
         self.initial_values = initial_values or {}
         self.input_events = input_events or []
         self.output_events = output_events or []
@@ -87,6 +116,28 @@ def __init__(
         self._received_events_lock = asyncio.Lock()
         self._has_received_events = asyncio.Event()
 
+    def _compile_validators(
+        self, types: dict[str, _t.Any]
+    ) -> dict[str, _t.Callable[[_t.Any, str], None]]:
+        if self.runtime_validation == "off":
+            return {}
+        return {
+            field: compile_validator(hint, deep=self.runtime_validation == "deep")
+            for field, hint in types.items()
+            if hint is not _t.Any
+        }
+
+    def validate_inputs(self, values: dict[str, _t.Any]) -> None:
+        """Check supplied inputs, including event assignments and initial values."""
+        for field, validate in self._input_validators.items():
+            if field in values:
+                validate(values[field], f"{self.namespace}.{field} (input)")
+
+    def validate_output_values(self, values: dict[str, _t.Any]) -> None:
+        """Check each typed output once before buffering and fan-out."""
+        for field, validate in self._output_validators.items():
+            validate(values[field], f"{self.namespace}.{field} (output)")
+
     @property
     def is_closed(self) -> bool:
         """Returns `True` if the `IOController` is closed, `False` otherwise."""
@@ -418,6 +469,15 @@ def dict(self) -> dict[str, _t.Any]:  # noqa: D102
             "output_events": [e.safe_type() for e in self.output_events],
             "initial_values": {k: list(v) for k, v in self._initial_values.items()},
             "event_field_coverage": {k: list(v) for k, v in self.event_field_coverage.items()},
+            # Live types/validators stay out of state backend serialization.
+            **{
+                key: {field: type_name(hint) for field, hint in types.items()}
+                for key, types in (
+                    ("input_types", self.input_types),
+                    ("output_types", self.output_types),
+                )
+                if any(hint is not _t.Any for hint in types.values())
+            },
         }