chore: Migrate to ty type checking

.devcontainer/devcontainer.json

@@ -25,6 +25,7 @@
 				"ms-python.python",
 				"ms-toolsai.jupyter",
 				"charliermarsh.ruff",
+				"astral-sh.ty",
 				"GitHub.copilot",
 				"GitHub.copilot-chat"
 			],

.github/agents/lint.agent.md

@@ -7,7 +7,7 @@ model: ['GPT-5 mini (copilot)', 'GPT-4.1 (copilot)']
 You are responsible for maintaining code quality in the Plugboard project by running linting tools and resolving any issues that arise.
 
 ## Your role:
-- Run `ruff` to check for formatting and linting issues and `mypy` to check for type errors.
+- Run `ruff` to check for formatting and linting issues and `ty` to check for type errors.
 - Review the output from these tools and identify any issues that need to be resolved.
 - Edit the code to fix any linting issues or type errors that are identified.
 - Ensure that all code is fully type-annotated and adheres to the project's coding standards.
@@ -16,13 +16,13 @@ You are responsible for maintaining code quality in the Plugboard project by run
 
 ## Project knowledge:
 - The project uses `uv` for dependency management and running commands.
-- Linting and formatting are handled by `ruff`, while static type checking is handled by `mypy`.
-- `pyproject.toml` contains the settings for `ruff` and `mypy`.
+- Linting and formatting are handled by `ruff`, while static type checking is handled by `ty`.
+- `pyproject.toml` contains the settings for `ruff` and `ty`.
 
 ## Commands you can run:
 - Run `uv run ruff format` to reformat the code.
 - Run `uv run ruff check` to check for linting issues.
-- Run `uv run mypy .` to check for type errors.
+- Run `uv run ty check plugboard/ plugboard-schemas/plugboard_schemas/ tests/` to check for type errors.
 - Run `uv lock --check` to check that the uv lockfile is up to date.
 - Run `uv run xenon --max-absolute B --max-modules A --max-average A plugboard/` to check for code complexity.
 - Run `find . -name '*.ipynb' -not -path "./.venv/*" -exec uv run nbstripout --verify {} +` to check that Jupyter notebooks are stripped of output.

.github/workflows/lint-test.yaml

@@ -52,9 +52,7 @@ jobs:
 
       - name: Type checking
         run: |
-          uv run mypy plugboard/ --explicit-package-bases
-          uv run mypy plugboard-schemas/plugboard_schemas/ --explicit-package-bases
-          uv run mypy tests/
+          uv run ty check plugboard/ plugboard-schemas/plugboard_schemas/ tests/
         if: always()
 
       - name: Code complexity

.gitignore

@@ -146,11 +146,6 @@ venv.bak/
 # mkdocs documentation
 /site
 
-# mypy
-.mypy_cache/
-.dmypy.json
-dmypy.json
-
 # Pyre type checker
 .pyre/
 

.pre-commit-config.yaml

@@ -11,15 +11,10 @@ repos:
       args: [ --fix ]
     # Run the formatter.
     - id: ruff-format
-- repo: https://github.com/pre-commit/mirrors-mypy
-  rev: v1.15.0
+- repo: https://github.com/astral-sh/ty-pre-commit
+  rev: v0.0.73
   hooks:
-    - id: mypy
-      additional_dependencies:
-        - types-PyYAML
-        - types-requests
-        - pydantic
-        - msgspec[yaml]
+    - id: ty
 - repo: https://github.com/rubik/xenon
   rev: v0.9.3
   hooks:

AGENTS.md

@@ -53,7 +53,7 @@ Plugboard is an event-driven framework in Python for simulating and orchestratin
 - You can delegate to the `lint` agent in `.github/agents` to resolve linting issues.
 - **Tools**: 
   - `ruff` - Formatting and linting.
-  - `mypy` - Static type checking.
+  - `ty` - Static type checking.
 - **Commands**:
   - `make lint` - Check for issues.
   - `make format` - Auto-format code.

CONTRIBUTING.md

@@ -28,11 +28,18 @@ uv run pytest .
 
 ### Linting
 
-We use [ruff](https://github.com/astral-sh/ruff) for code formatting and style. Install the pre-commit hook by running
+We use [ruff](https://github.com/astral-sh/ruff) for code formatting and style, and
+[ty](https://github.com/astral-sh/ty) for static type checking. Install the pre-commit hook by
+running
 ```sh
 uv run pre-commit install
 ```
 
+You can run the full local lint suite with
+```sh
+make lint
+```
+
 ### Documentation
 
 The package documentation uses [Material for MkDocs](https://squidfunk.github.io/mkdocs-material/) and can be viewed locally by running

Makefile

@@ -35,8 +35,7 @@ init: $(VENV)/__makefile_stamps_init
 lint: init
 	uv run ruff check
 	uv run ruff format --check
-	uv run mypy $(SRC)/ --explicit-package-bases
-	uv run mypy $(TESTS)/
+	uv run ty check $(SRC)/ ./plugboard-schemas/plugboard_schemas/ $(TESTS)/
 
 .PHONY: test
 test: init

README.md

@@ -50,7 +50,7 @@ Some examples of what you can build with Plugboard include:
 - **YAML model specification** format for saving model definitions, allowing you to run the same model locally or in cloud infrastructure;
 - A **command line interface** for executing models;
 - Built to handle the **data intensive simulation** requirements of industrial process applications;
-- Modern implementation with **Python 3.12+** based around **asyncio** with complete type annotation coverage;
+- Modern implementation with **Python 3.12+** based around **asyncio** with complete type annotation coverage checked with **ty**;
 - Built-in integrations for **loading/saving data** from cloud storage and SQL databases;
 - Built-in **LLM integrations** for building AI-augmented process models with support for multiple providers;
 - **Detailed logging** of component inputs, outputs and state for monitoring and process mining or surrogate modelling use-cases.

plugboard-schemas/plugboard_schemas/__init__.py

@@ -53,7 +53,10 @@
 )
 
 
-__version__ = version(__package__)
+_PACKAGE_NAME = __package__ or __name__.split(".")[0]
+
+
+__version__ = version(_PACKAGE_NAME)
 
 
 __all__ = [

plugboard-schemas/plugboard_schemas/_validator_registry.py

@@ -46,7 +46,7 @@ def __len__(self) -> int:
         return len(self._validators)
 
     def __repr__(self) -> str:
-        names = [fn.__name__ for fn in self._validators]
+        names = [getattr(fn, "__name__", type(fn).__name__) for fn in self._validators]
         return f"{type(self).__name__}({names!r})"
 
 

plugboard-schemas/pyproject.toml

@@ -37,14 +37,6 @@ root = ".."
 [tool.uv]
 package = true
 
-[tool.mypy]  # Static type checking
-exclude = [".git", "__pycache__", ".venv", "venv"]
-ignore_missing_imports = true
-disallow_untyped_defs = true
-disallow_incomplete_defs = true
-no_implicit_optional = true
-plugins = "pydantic.mypy"
-
 [tool.ruff]  # Code formatting and linting
 line-length = 100
 src = ["plugboard_schemas"]

plugboard/__init__.py

@@ -3,4 +3,7 @@
 from importlib.metadata import version
 
 
-__version__ = version(__package__)
+_PACKAGE_NAME = __package__ or __name__.split(".")[0]
+
+
+__version__ = version(_PACKAGE_NAME)

plugboard/component/component.py

@@ -178,8 +178,8 @@ def _configure_io(cls) -> None:
         cls.io = IO(
             inputs=sorted(io_args["inputs"], key=str),
             outputs=sorted(io_args["outputs"], key=str),
-            input_events=sorted(io_args["input_events"], key=str),
-            output_events=sorted(io_args["output_events"], key=str),
+            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,
         )
         # Set exports for subclass

plugboard/component/utils.py

@@ -11,10 +11,23 @@
 from plugboard.utils import gen_rand_str
 
 
-_FuncT = _t.TypeVar(
-    "_FuncT",
-    bound=_t.Callable[..., _t.Union[dict[str, _t.Any], _t.Awaitable[dict[str, _t.Any]]]],
-)
+class _ComponentFunction(_t.Protocol):
+    """Callable protocol for functions wrapped by `@component`.
+
+    The decorator relies on the wrapped callable exposing `__name__`, `__module__`, and `__doc__`
+    so it can register the generated component class and build helpful generated documentation.
+    """
+
+    __name__: str
+    __module__: str
+    __doc__: str | None
+
+    def __call__(
+        self, *args: _t.Any, **kwargs: _t.Any
+    ) -> dict[str, _t.Any] | _t.Awaitable[dict[str, _t.Any]]: ...
+
+
+_FuncT = _t.TypeVar("_FuncT", bound=_ComponentFunction)
 
 _FUNC_COMPONENT_DOC_TEMPLATE = Template(
     """Component for wrapped function $name.
@@ -101,9 +114,11 @@ async def step(self) -> _t.Any:
 
 def _ensure_async_callable(func: _FuncT) -> _t.Callable[..., _t.Awaitable[dict[str, _t.Any]]]:
     if inspect.iscoroutinefunction(func):
-        return func
+        return _t.cast(_t.Callable[..., _t.Awaitable[dict[str, _t.Any]]], func)
+
+    sync_func = _t.cast(_t.Callable[..., dict[str, _t.Any]], func)
 
     async def _async_func(*args: _t.Any, **kwargs: _t.Any) -> dict[str, _t.Any]:
-        return func(*args, **kwargs)  # type: ignore[return-value]
+        return sync_func(*args, **kwargs)
 
     return _async_func

plugboard/connector/asyncio_channel.py

@@ -32,13 +32,13 @@ def __init__(
         self._queue: asyncio.Queue = queue or asyncio.Queue(maxsize=maxsize)
         self._subscribers: _t.Optional[set[asyncio.Queue]] = subscribers
 
-    async def send(self, item: _t.Any) -> None:
+    async def send(self, msg: _t.Any) -> None:
         """Sends an item through the `Channel`."""
         if self._subscribers is None:
-            return await self._queue.put(item)
+            return await self._queue.put(msg)
         async with asyncio.TaskGroup() as tg:
             for queue in self._subscribers:
-                tg.create_task(queue.put(item))
+                tg.create_task(queue.put(msg))
 
     async def recv(self) -> _t.Any:
         """Returns an item received from the `Channel`."""

plugboard/connector/ray_channel.py

@@ -23,9 +23,9 @@ class RayChannel(Channel):
     @depends_on_optional("ray")
     def __init__(  # noqa: D417
         self,
-        actor_options: _t.Optional[dict] = None,
+        actor_options: _t.Optional[dict[str, _t.Any]] = None,
         **kwargs: _t.Any,
-    ):
+    ) -> None:
         """Instantiates `RayChannel`.
 
         Args:
@@ -51,9 +51,9 @@ def is_closed(self) -> bool:
         """
         return self._actor.getattr.remote("is_closed")  # type: ignore
 
-    async def send(self, item: _t.Any) -> None:
+    async def send(self, msg: _t.Any) -> None:
         """Sends an item through the `RayChannel`."""
-        await self._actor.send.remote(item)  # type: ignore
+        await self._actor.send.remote(msg)  # type: ignore
 
     async def recv(self) -> _t.Any:
         """Returns an item received from the `RayChannel`."""

plugboard/events/event_handlers.py

@@ -50,7 +50,8 @@ def decorator(method: AsyncCallable) -> AsyncCallable:
     def _get_class_path_for_method(method: AsyncCallable) -> str:
         """Get the fully qualified path for the class containing a method."""
         module_name = method.__module__
-        qualname_parts = method.__qualname__.split(".")
+        qualname = getattr(method, "__qualname__", type(method).__qualname__)
+        qualname_parts = qualname.split(".")
         class_name = qualname_parts[-2]  # Last part is the method name
         return f"{module_name}.{class_name}"
 

plugboard/library/llm.py

@@ -86,7 +86,9 @@ def _set_response(self, chat_response: ChatResponse, expand: bool) -> None:
         if not expand:
             self.response: str | None = chat_response.message.content
         else:
-            for field, value in chat_response.raw.model_dump().items():  # type: ignore[union-attr]
+            if chat_response.raw is None:
+                raise ValueError("Expected structured response payload from LLM.")
+            for field, value in chat_response.raw.model_dump().items():
                 setattr(self, field, value)
 
 

plugboard/library/sql_io.py

@@ -1,10 +1,11 @@
 """Provides `SQLReader` and `SQLWriter` components to access SQL databases from Plugboard models."""
 
 from collections import defaultdict, deque
+import collections.abc as cabc
 import typing as _t
 
 from sqlalchemy import MetaData, Table, insert, text
-from sqlalchemy.engine import Engine, Row, create_engine
+from sqlalchemy.engine import Connection, Engine, Row, create_engine
 from sqlalchemy.exc import InvalidRequestError
 from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
 
@@ -41,10 +42,14 @@ def __init__(
         self._connection_string = connection_string
         self._query = query
         self._params = params or {}
-        self._reader: _t.Optional[_t.AsyncIterator | _t.Iterator] = None
+        self._reader: (
+            cabc.AsyncIterator[cabc.Sequence[Row[_t.Any]]]
+            | cabc.Iterator[cabc.Sequence[Row[_t.Any]]]
+            | None
+        ) = None
         self._connect_args = connect_args or {}
 
-    async def _run_query_async(self) -> _t.AsyncIterator[_t.Sequence[Row]]:
+    async def _run_query_async(self) -> cabc.AsyncIterator[cabc.Sequence[Row[_t.Any]]]:
         engine = create_async_engine(self._connection_string, **self._connect_args)
         async with engine.connect() as conn:
             if self._chunk_size:
@@ -61,7 +66,7 @@ async def _run_query_async(self) -> _t.AsyncIterator[_t.Sequence[Row]]:
                 yield list(result)
             raise NoMoreDataException
 
-    def _run_query_sync(self) -> _t.Iterator[_t.Sequence[Row]]:
+    def _run_query_sync(self) -> cabc.Iterator[cabc.Sequence[Row[_t.Any]]]:
         engine = create_engine(self._connection_string, **self._connect_args)
         with engine.connect() as conn:
             if self._chunk_size:
@@ -77,19 +82,24 @@ def _run_query_sync(self) -> _t.Iterator[_t.Sequence[Row]]:
                 yield list(result)
             raise NoMoreDataException
 
-    async def _fetch(self) -> _t.Sequence[Row]:
+    async def _fetch(self) -> cabc.Sequence[Row[_t.Any]]:
         if self._reader is None:
             try:
                 self._reader = self._run_query_async()
-                return await self._reader.__anext__()
+                async_reader = self._reader
+                return await async_reader.__anext__()
             except InvalidRequestError:
                 # Fall back on synchronous connection
                 self._reader = self._run_query_sync()
                 return next(self._reader)
 
-        if isinstance(self._reader, _t.AsyncIterator):
-            return await self._reader.__anext__()
-        return next(self._reader)
+        if isinstance(self._reader, cabc.AsyncIterator):
+            async_reader = _t.cast(
+                cabc.AsyncIterator[cabc.Sequence[Row[_t.Any]]],
+                self._reader,
+            )
+            return await async_reader.__anext__()
+        return next(_t.cast(cabc.Iterator[cabc.Sequence[Row[_t.Any]]], self._reader))
 
     async def _convert(self, data: _t.Sequence[Row]) -> dict[str, deque]:
         converted_data: dict[str, deque] = defaultdict(deque)
@@ -138,11 +148,13 @@ async def _save_rows_async(self, data: list[dict[str, _t.Any]]) -> None:
             raise RuntimeError("No async database connection available")
         async with self._engine.connect() as conn:
             if self._table is None:
-                await conn.run_sync(
-                    self._metadata.reflect,
-                    only=[self._table_name],
-                )
-                self._table = Table(self._table_name, self._metadata, autoload_with=self._engine)  # type: ignore[arg-type]
+
+                def _load_table(sync_conn: Connection) -> Table:
+                    """Reflect and load the target table within a sync SQLAlchemy connection."""
+                    self._metadata.reflect(bind=sync_conn, only=[self._table_name])
+                    return Table(self._table_name, self._metadata, autoload_with=sync_conn)
+
+                self._table = await conn.run_sync(_load_table)
             await conn.execute(insert(self._table).values(data))
 
     def _save_rows_sync(self, data: list[dict[str, _t.Any]]) -> None:

plugboard/state/state_backend.py

@@ -37,18 +37,18 @@ def __init__(
             metadata: Metadata key value pairs.
             kwargs: Additional keyword arguments.
         """
-        self._local_state = {"job_id": job_id, "metadata": metadata, **kwargs}
+        self._local_state: dict[str, _t.Any] = {"job_id": job_id, "metadata": metadata, **kwargs}
         self._initialised_with_job_id = False
         self._logger = DI.logger.resolve_sync().bind(cls=self.__class__.__name__, job_id=job_id)
         self._logger.info("StateBackend created")
         self._ctx = ExitStack()
 
-    def __getstate__(self) -> dict:
+    def __getstate__(self) -> dict[str, _t.Any]:
         state = self.__dict__.copy()
         state.pop("_ctx", None)
         return state
 
-    def __setstate__(self, state: dict) -> None:
+    def __setstate__(self, state: dict[str, _t.Any]) -> None:
         self.__dict__.update(state)
         self._ctx = ExitStack()
         job_id = self._local_state.get("job_id")