ci: add blocking Python type checks

.github/workflows/ci.yml

@@ -51,6 +51,20 @@ jobs:
         run: make check-dependency-licenses
 
 
+  typecheck:
+    name: Type checking
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+      - uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0
+        with:
+          version-file: "pyproject.toml"
+          python-version: "3.12"
+          enable-cache: true
+      - run: make install-ci
+      - run: make typecheck
+
   test:
     name: Tests (Python ${{ matrix.python-version }})
     runs-on: ubuntu-latest
@@ -173,7 +187,7 @@ jobs:
   status:
     name: CI Status
     if: always()
-    needs: [static, test, packaging, e2e, container]
+    needs: [static, typecheck, test, packaging, e2e, container]
     runs-on: ubuntu-latest
     timeout-minutes: 5
     env:

.pre-commit-config.yaml

@@ -46,6 +46,12 @@ repos:
         language: system
         pass_filenames: false
         always_run: true
+      - id: typecheck
+        name: type checking (Python 3.11)
+        entry: make typecheck
+        language: system
+        pass_filenames: false
+        always_run: true
       - id: spdx-headers
         name: SPDX headers (staged files)
         entry: uv run python scripts/update_license_headers.py

CONTRIBUTING.md

@@ -280,9 +280,23 @@ Your PR title must follow [Conventional Commits](https://www.conventionalcommits
 the settings in the [contribution policy](docs/engineering/contribution-policy.md),
 the title check alone does not guarantee the commit subject on `main`.
 
+## Type checking
+
+`make typecheck` runs the locked `ty` version against shipped `src/usersim`,
+repository gate scripts, and installed CLI e2e helpers. The checker targets
+Python 3.11, the supported minimum, even when the development interpreter is
+newer. Ordinary unit-test fixtures are outside the initial type-check scope.
+
+Type checking runs in pre-commit, `make check-all`, the native validation
+bundle, and CI. Keep annotations consistent with runtime contracts; the
+existing tests still establish behavior. Dynamic Pydantic schema construction
+and upstream library annotation limitations have local, rule-specific
+suppressions with explanations. New suppressions need the same justification;
+do not disable a rule globally to make the gate pass.
+
 ## What CI runs
 
-`CI Status` aggregates static checks, all supported Python versions, packaging,
+`CI Status` aggregates static checks, type checking, all supported Python versions, packaging,
 installed CLI e2e tests, and CPU container tests. Missing, skipped, cancelled, or
 failed mandatory jobs fail that status. Run the same Make targets locally; see
 [local and CI validation](docs/engineering/ci.md) for commands, diagnostics, and

Makefile

@@ -60,7 +60,7 @@ COV_FAIL_UNDER ?= 84
 
 .DEFAULT_GOAL := help
 
-.PHONY: help bootstrap check-tools check-lock install-dev install-ci install-kernel health-check check-notebooks-clean lint lint-fix format format-check check-all \
+.PHONY: typecheck help bootstrap check-tools check-lock install-dev install-ci install-kernel health-check check-notebooks-clean lint lint-fix format format-check check-all \
         check-ci-status check-hooks check-all-fix validate-native validate-ci test test-fast test-e2e coverage smoke dist check-dist check-release-tag check-extensions check-doc-links clean-notebooks check-license-headers update-license-headers check-dependency-licenses clean
 
 
@@ -99,6 +99,9 @@ install-ci: check-tools  ## Sync exactly what CI needs, enforcing the lockfile
 
 # --- lint / format ---------------------------------------------------------
 
+typecheck: check-tools  ## Check shipped code, gate scripts and e2e helpers against Python 3.11
+	$(UV) run --locked ty check --error-on-warning
+
 lint:  ## Check lint rules (no changes written)
 	$(UV) run ruff check --output-format=full $(PY_ALL)
 	@echo "Lint clean. Run 'make lint-fix' to auto-fix issues."
@@ -113,7 +116,7 @@ format-check:  ## Check formatting (no changes written)
 format:  ## Reformat in place
 	$(UV) run ruff format $(PY_ALL)
 
-check-all: check-tools check-lock lint format-check check-license-headers check-doc-links check-notebooks-clean check-hooks  ## Run ordinary static gates (hooks may rewrite files)
+check-all: check-tools check-lock typecheck lint format-check check-license-headers check-doc-links check-notebooks-clean check-hooks  ## Run ordinary static gates (hooks may rewrite files)
 
 check-hooks:  ## Run hooks on every tracked file; review rewrites and rerun after failure
 	$(UV) run pre-commit run --all-files --show-diff-on-failure

docs/engineering/ci.md

@@ -6,7 +6,7 @@ bundle, including distribution checks and installed CLI tests.
 
 | Target | Checks |
 | --- | --- |
-| `make check-all` | Static checks defined by the repository. |
+| `make check-all` | Static checks, including the configured type-check scope. |
 | `make check-dependency-licenses` | Dependency license policy. |
 | `make coverage` | Ordinary test suite with an 84% minimum total coverage. |
 | `make smoke` | Offline configuration, registry, asset, and reporting self-check. |
@@ -27,6 +27,18 @@ Python 3.12 through `make test-e2e`. It retains failure diagnostics under
 `WHEEL_DIR=dist` after archive validation so the e2e test installs that same
 wheel without rebuilding.
 
+## Type checking
+
+`make typecheck` uses the locked `ty==0.0.69` against `src/usersim`, `scripts`,
+and `tests_e2e`, targeting Python 3.11 with `src` as the module root. The
+dedicated CI job runs this target after locked setup; hooks, static validation,
+and release validation inherit it through `check-all`.
+Ordinary unit-test fixtures are outside this initial scope.
+
+Keep annotations aligned with runtime behavior. Rule-specific suppressions
+cover documented runtime schema construction or upstream annotation limits;
+do not disable a rule across the repository.
+
 ## Coverage
 
 The Python 3.12 CI matrix job runs `make coverage`; the other Python versions
@@ -54,7 +66,7 @@ setup stops before XML is produced, inspect the failed step's logs.
 
 ## Mandatory jobs and aggregate status
 
-`CI Status` waits for static checks, the complete Python test matrix, packaging,
+`CI Status` waits for static checks, the dedicated `Type checking` job, the complete Python test matrix, packaging,
 installed CLI e2e, and CPU container tests. Every job must report `success`.
 Failures, cancellation, missing results, malformed results, and skipped jobs fail
 the aggregate; there are no intentional job skips. Conditional coverage steps

pyproject.toml

@@ -90,6 +90,7 @@ dev = [
     "ruff>=0.14,<0.15",
     "pytest-cov>=7.0,<8",
     "pre-commit>=4.0,<5",
+    "ty==0.0.69",
     # Distributes the test suite across cores. It is CPU-bound with every LLM
     # call mocked, so it parallelises well. `make test` passes -n and fails
     # without this.
@@ -233,3 +234,10 @@ exclude_lines = [
     "if TYPE_CHECKING:",
     "raise NotImplementedError",
 ]
+
+[tool.ty.src]
+include = ["src/usersim", "scripts", "tests_e2e"]
+
+[tool.ty.environment]
+python-version = "3.11"
+root = ["src"]

scripts/check_ci_results.py

@@ -9,7 +9,7 @@
 import os
 import sys
 
-REQUIRED_JOBS = frozenset({"static", "test", "packaging", "e2e", "container"})
+REQUIRED_JOBS = frozenset({"static", "typecheck", "test", "packaging", "e2e", "container"})
 
 
 def validate_results(raw: str) -> list[str]:

src/usersim/asset_gen/financial_services/pipeline.py

@@ -35,7 +35,7 @@
 import uuid
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Any, Iterator
+from typing import Any, Iterator, cast
 
 from pydantic import BaseModel, Field
 
@@ -1403,6 +1403,8 @@ def _embed_documents(
             time.sleep(delay)
 
     # ORDERED seed + reset_index preserves row order, so ids align by position.
+    # A completed embedding attempt assigns result; failures re-raise above.
+    result = cast(pd.DataFrame, result)
     id_to_vec = {doc_id: r["body_embedding"]["embeddings"][0] for doc_id, (_, r) in zip(edf["id"], result.iterrows())}
     for ib in institutions:
         ib.embeddings = {d["id"]: id_to_vec[d["id"]] for d in ib.documents if d["id"] in id_to_vec}

src/usersim/cli/_async.py

@@ -12,24 +12,28 @@
 
 import asyncio
 import concurrent.futures
-from collections.abc import Coroutine
-from typing import Any, TypeVar
+from collections.abc import Awaitable
+from typing import TypeVar
 
 _T = TypeVar("_T")
 
 
-def run_coroutine(coro: Coroutine[Any, Any, _T]) -> _T:
-    """Run ``coro`` to completion and return its result.
+def run_coroutine(coro: Awaitable[_T]) -> _T:
+    """Run a scorer awaitable to completion and return its result.
 
     A notebook already owns the loop on its thread, and ``asyncio.run``
     refuses to nest inside one. Where that is the case the work moves to a
     thread with a loop of its own, which is the only way a synchronous call
     can wait for a coroutine from inside a live loop.
     """
+
+    async def await_result() -> _T:
+        return await coro
+
     try:
         asyncio.get_running_loop()
     except RuntimeError:
-        return asyncio.run(coro)
+        return asyncio.run(await_result())
 
     with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
-        return pool.submit(asyncio.run, coro).result()
+        return pool.submit(lambda: asyncio.run(await_result())).result()

src/usersim/cli/_env.py

@@ -11,6 +11,7 @@
 
 import logging
 import os
+from collections.abc import Iterable
 from pathlib import Path
 
 logger = logging.getLogger("usersim.cli.env")
@@ -69,6 +70,6 @@ def describe_loaded_env(paths: list[Path]) -> str:
     return f"read {names} (existing environment variables take precedence)"
 
 
-def missing_keys(names: object) -> list[str]:
+def missing_keys(names: Iterable[str]) -> list[str]:
     """Return which of ``names`` are absent or empty in the environment."""
-    return [n for n in names if not os.environ.get(str(n))]  # type: ignore[union-attr]
+    return [n for n in names if not os.environ.get(str(n))]

src/usersim/cli/_inputs.py

@@ -30,7 +30,7 @@
 import tempfile
 import time
 from collections import Counter
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
@@ -123,7 +123,7 @@ def _plain(value: Any) -> Any:
 
 
 def prepare_episode_inputs(
-    rows: list[Mapping[str, Any]],
+    rows: Sequence[Mapping[str, Any]],
     *,
     assets_dir: str | Path | None = None,
     source: str | Path | None = None,

src/usersim/cli/_models.py

@@ -57,7 +57,7 @@
 import tomllib
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Iterable
+from typing import Any, Iterable, Mapping
 
 from usersim.cli._errors import ConfigError
 
@@ -396,7 +396,7 @@ def apply_parallel_overrides(config: ModelsConfig, overrides: dict[str, int]) ->
 
 def apply_model_overrides(
     config: ModelsConfig,
-    overrides: dict[str, str | dict[str, Any]],
+    overrides: Mapping[str, str | dict[str, Any]],
 ) -> ModelsConfig:
     """Return a new ``ModelsConfig`` with ``model`` strings (and optionally
     other per-model knobs) swapped per alias.
@@ -1002,7 +1002,9 @@ def smoke_test_models(
         if key_pair in seen:
             # Inherit the prior result for the same (model, provider).
             prior_alias = next(
-                a for a, (_, m) in results.items() if (config.get(a).model, config.get(a).provider) == key_pair
+                a
+                for a, (_, m) in results.items()
+                if (prior_spec := config.get(a)) is not None and (prior_spec.model, prior_spec.provider) == key_pair
             )
             results[spec.alias] = results[prior_alias]
             continue

src/usersim/cli/_persona_language.py

@@ -145,7 +145,8 @@ def _column_value_counts(
     frequencies = tuple(
         sorted(
             (str(entry["values"]), int(entry["counts"]))
-            for entry in pc.value_counts(table[column]).to_pylist()
+            # Arrow installs its compute kernels dynamically; the attribute is public.
+            for entry in pc.value_counts(table[column]).to_pylist()  # ty: ignore[unresolved-attribute]
             if entry["values"] is not None
         )
     )

src/usersim/cli/_pipeline.py

@@ -418,7 +418,8 @@ def build_simulator_config_builder(
     )
 
     config_builder.add_column(
-        ConversationSimulatorConfig(
+        # Registered plugin configs are accepted at runtime, outside DD's closed type union.
+        ConversationSimulatorConfig(  # ty: ignore[invalid-argument-type]
             name="conversation_messages",
             persona_column="persona",
             probe_type_column="probe_type",
@@ -672,7 +673,8 @@ def build_evaluator_config_builder(
     _warn_on_undispatched_scorers(scorers or [])
 
     config_builder.add_column(
-        TrajectoryEvaluatorConfig(
+        # Registered plugin configs are accepted at runtime, outside DD's closed type union.
+        TrajectoryEvaluatorConfig(  # ty: ignore[invalid-argument-type]
             name=output_column,
             judges=judge_specs,
             axes=axes,

src/usersim/cli/select.py

@@ -227,7 +227,7 @@ def run(args: argparse.Namespace) -> int:
     if args.stratify_by and args.stratify_by.strip().lower() != "none":
         stratify_by = tuple(c.strip() for c in args.stratify_by.split(",") if c.strip())
     # --run accepts a single id, 'latest', or a comma-separated list.
-    runs: object = args.run
+    runs: str | list[str] | None = args.run
     if isinstance(args.run, str) and "," in args.run:
         runs = [r.strip() for r in args.run.split(",") if r.strip()]
 

src/usersim/cli/simulate.py

@@ -35,6 +35,7 @@
 from collections import Counter
 from fractions import Fraction
 from pathlib import Path
+from typing import cast
 
 from usersim.cli._pipeline import resolve_toolset_seed_for_mix, verify_probe_assets
 
@@ -738,6 +739,8 @@ def _simulate_to_parquet(
         logger.info("run=%s is the bare legacy layout; no run manifest written", run_id)
         return 0
 
+    run_id = cast(str, run_id)  # At least one successful save_sampled_rows assigned it.
+
     # Write the run manifest. ``started_at_epoch`` is when the invocation
     # that made the run started, kept in the run's record. First-writer-wins:
     # a resume into a run that already has a manifest is a noop.

src/usersim/engine/core/behavioral.py

@@ -15,7 +15,7 @@
 import json
 import random
 import re
-from typing import Any
+from typing import Any, TypedDict
 
 _OCEAN_TRAITS = ("openness", "conscientiousness", "extraversion", "agreeableness", "neuroticism")
 
@@ -596,7 +596,14 @@ def format_disclosure_instructions(style: str) -> str:
 # Sim2Real: User interaction style (Option C + G)
 # ---------------------------------------------------------------------------
 
-INTERACTION_STYLES = {
+
+class _InteractionStyle(TypedDict):
+    base_frustration: int
+    escalation_rate: float
+    description: str
+
+
+INTERACTION_STYLES: dict[str, _InteractionStyle] = {
     "confrontational": {
         "base_frustration": 2,
         "escalation_rate": 2.0,
@@ -715,7 +722,7 @@ def compute_frustration_level(
     base = info["base_frustration"]
     rate = info["escalation_rate"]
 
-    frustration = base
+    frustration: float = base
     if assistant_failures >= 2:
         frustration += assistant_failures * rate
 

src/usersim/engine/core/finance_bank.py

@@ -244,7 +244,7 @@ def verify_identity_schema(
     }
 
 
-def framework_primitive_tool(name: str) -> dict[str, Any] | None:
+def framework_primitive_tool(name: str | None) -> dict[str, Any] | None:
     """Canonical ``{description, parameters}`` for a framework primitive.
 
     Returns ``None`` for non-primitive (per-institution DB) tools. Callers get a

src/usersim/engine/core/manifest.py

@@ -201,7 +201,7 @@ class ScopeInfo:
     #: (``hi_Deva_IN`` filters on "\u0939\u093f\u0902\u0926\u0940",
     #: ``hi_Latn_IN`` on "Hindi"). Which people were eligible is the
     #: thing that decides whether two runs are comparable.
-    persona_language_filters: dict[str, dict[str, list[str]]] = field(default_factory=dict)
+    persona_language_filters: Mapping[str, Mapping[str, Sequence[str]]] = field(default_factory=dict)
 
 
 @dataclass(frozen=True)