Gradually introduce mypy type checking (Phases 1-3)

.github/workflows/lint.yml

@@ -24,3 +24,11 @@ jobs:
 
       - name: ruff format check
         run: ruff format --check .
+
+      # Pin to match the pre-commit mirrors-mypy rev and the pyproject dev
+      # group so CI, local commits, and `uv sync` never run different mypys.
+      - name: Install mypy
+        run: pip install mypy==2.1.0
+
+      - name: mypy
+        run: mypy .

.pre-commit-config.yaml

@@ -5,3 +5,12 @@ repos:
       - id: ruff
         args: [--fix]
       - id: ruff-format
+
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    # Keep in sync with the mypy pin in pyproject.toml dev group and
+    # .github/workflows/lint.yml. Run `pre-commit autoupdate` to bump.
+    rev: v2.1.0
+    hooks:
+      - id: mypy
+        pass_filenames: false
+        args: ["."]

CONTRIBUTING.md

@@ -2,14 +2,13 @@
 
 ## Code quality toolchain
 
-This project uses [ruff](https://docs.astral.sh/ruff/) for linting and formatting, and [pre-commit](https://pre-commit.com/) to enforce this automatically before every commit.
-
-> **Note:** Type checking with mypy is not yet enforced. See the tracking issue for gradually
-> introducing mypy across the codebase.
+This project uses [ruff](https://docs.astral.sh/ruff/) for linting and formatting, and
+[mypy](https://mypy-lang.org/) for type checking, enforced automatically before every
+commit via [pre-commit](https://pre-commit.com/).
 
 ### First-time setup
 
-Install the dev dependencies (includes ruff and pre-commit):
+Install the dev dependencies (includes ruff, mypy, and pre-commit):
 
 ```bash
 uv sync
@@ -21,7 +20,8 @@ Then install the pre-commit hooks:
 pre-commit install
 ```
 
-From this point on, ruff runs automatically on staged files whenever you `git commit`.
+From this point on, ruff runs on staged files and mypy runs across the whole
+project whenever you `git commit`.
 
 ### Running manually
 
@@ -46,6 +46,27 @@ To update hook versions to their latest releases:
 pre-commit autoupdate
 ```
 
+### Type checking
+
+Run mypy the same way pre-commit and CI do:
+
+```bash
+mypy .
+```
+
+The codebase is being typed incrementally (see issue #46). Modules with
+pre-existing errors are listed in the `[[tool.mypy.overrides]]` block in
+`pyproject.toml` with `ignore_errors = true`, so `mypy .` passes today even
+though not every module is fully typed yet.
+
+Policy for working with this list:
+
+- **New modules** must pass `mypy .` cleanly — do not add them to the
+  overrides list.
+- **PRs that substantively touch a module on the overrides list** should fix
+  that module's type errors and remove it from the list as part of the
+  change.
+
 ### CI
 
-Every pull request runs the lint job (`.github/workflows/lint.yml`), which checks ruff lint and formatting across the entire project. Fix any failures locally with `pre-commit run --all-files` before pushing.
+Every pull request runs the lint job (`.github/workflows/lint.yml`), which checks ruff lint and formatting and runs mypy across the entire project. Fix any failures locally with `pre-commit run --all-files` before pushing.

gently/app/agent.py

@@ -79,7 +79,7 @@ def __init__(
         model: str = settings.models.main,
         microscope_client=None,
         session_id: str | None = None,
-        store: FileStore = None,
+        store: FileStore | None = None,
         no_api: bool = False,
     ):
         """
@@ -411,7 +411,7 @@ def enter_resolution_mode(self) -> str:
         logger.info("Entered resolution mode")
         return "Resolution mode active. Determining what this session is for."
 
-    def exit_resolution_mode(self, outcome: str = None) -> str:
+    def exit_resolution_mode(self, outcome: str | None = None) -> str:
         """Leave resolution mode for run mode.
 
         Called by resolution tools (attach_session_to_plan,
@@ -490,7 +490,7 @@ def exit_plan_mode(self) -> str:
 
     # ===== Prompt & System Prompt =====
 
-    def _update_system_prompt(self, context_summary: str = None):
+    def _update_system_prompt(self, context_summary: str | None = None):
         """Rebuild system prompt via PromptManager."""
         self.system_prompt = self.prompts.update_system_prompt(
             self.experiment,
@@ -1065,7 +1065,9 @@ async def handle_message_stream(self, user_message: str):
             if acquired:
                 lock.release()
 
-    async def run_wake_turn(self, wake_note: str, trigger: str = None, interactive: bool = False):
+    async def run_wake_turn(
+        self, wake_note: str, trigger: str | None = None, interactive: bool = False
+    ):
         """Drive one autonomous (no-user) turn for the wake-router.
 
         Runs through the normal streaming pipeline (so it acquires the turn-lock

gently/app/benchmark.py

@@ -138,9 +138,9 @@ async def run_benchmark(
     exposure_ms: float = 10.0,
     warmup: int = 1,
     # Legacy parameters (ignored, kept for API compat with bridge)
-    n_volumes: int = None,
-    n_slices: int = None,
-    n_warmup: int = None,
+    n_volumes: int | None = None,
+    n_slices: int | None = None,
+    n_warmup: int | None = None,
     progress_fn: callable | None = None,
 ) -> BenchmarkResults:
     """

gently/app/orchestration/timelapse.py

@@ -195,7 +195,7 @@ def __init__(
 
     async def start(
         self,
-        embryo_ids: list[str],
+        embryo_ids: list[str] | None = None,
         stop_condition: str = "manual",
         base_interval_seconds: float = 120.0,
         condition_value: Any = None,
@@ -739,7 +739,7 @@ async def _run_loop(self):
                 },
             )
 
-    async def _acquire_embryo(self, embryo_state: EmbryoState, round_time: datetime = None):
+    async def _acquire_embryo(self, embryo_state: EmbryoState, round_time: datetime | None = None):
         """Acquire a single volume for one embryo
 
         Parameters
@@ -2510,7 +2510,7 @@ def _check_interval_rules(
                 result,
             )
 
-    def _finalize_perception_run(self, status: str = "completed", error_message: str = None):
+    def _finalize_perception_run(self, status: str = "completed", error_message: str | None = None):
         """Mark the perception run as finished in FileStore."""
         if self._store and self._perception_run_id is not None:
             try:
@@ -2575,7 +2575,7 @@ async def _run_detector(
         volume,
         embryo_state: EmbryoState,
         detector_name: str,
-        volume_uids: dict = None,
+        volume_uids: dict | None = None,
     ):
         """Run a role-declared Detector (Phase 2) and persist + emit results.
 
@@ -2791,7 +2791,7 @@ async def _run_perception(
         timepoint: int,
         volume,
         embryo_state: EmbryoState,
-        volume_uids: dict = None,
+        volume_uids: dict | None = None,
     ):
         """Run the per-role detector on the acquired volume and emit results.
 

gently/app/tools/acquisition_tools.py

@@ -9,7 +9,7 @@
 
 import numpy as np
 
-from gently.harness.tools.helpers import get_embryo_or_error
+from gently.harness.tools.helpers import ctx_get, get_embryo_or_error
 from gently.harness.tools.registry import ToolCategory, ToolExample, tool
 
 logger = logging.getLogger(__name__)
@@ -44,12 +44,12 @@ async def acquire_volume(
     embryo_id: str,
     num_slices: int = 50,
     exposure_ms: float = 10.0,
-    z_buffer_um: float = None,
-    context: dict = None,
+    z_buffer_um: float | None = None,
+    context: dict | None = None,
 ) -> str:
     """Acquire single volume - moves to embryo first, uses calibration"""
-    agent = context.get("agent")
-    client = context.get("client")
+    agent = ctx_get(context, "agent")
+    client = ctx_get(context, "client")
 
     if not agent:
         return "Error: No agent context"
@@ -237,15 +237,15 @@ async def acquire_volume(
     ],
 )
 async def capture_lightsheet(
-    piezo_position: float = None,
+    piezo_position: float | None = None,
     galvo_position: float = 0.0,
-    embryo_id: str = None,
+    embryo_id: str | None = None,
     show: bool = True,
-    context: dict = None,
+    context: dict | None = None,
 ) -> str:
     """Capture and optionally display a single lightsheet image"""
-    client = context.get("client")
-    agent = context.get("agent")
+    client = ctx_get(context, "client")
+    agent = ctx_get(context, "agent")
 
     try:
         embryo = None
@@ -355,10 +355,10 @@ async def capture_lightsheet(
         ToolExample("Capture all embryos", {}),
     ],
 )
-async def batch_lightsheet(galvo_position: float = 0.0, context: dict = None) -> str:
+async def batch_lightsheet(galvo_position: float = 0.0, context: dict | None = None) -> str:
     """Capture lightsheet images from all embryos and show them in the web UI"""
-    agent = context.get("agent")
-    client = context.get("client")
+    agent = ctx_get(context, "agent")
+    client = ctx_get(context, "client")
 
     if not agent or not client:
         return "Error: Agent or microscope not available"

gently/app/tools/analysis_tools.py

@@ -18,7 +18,7 @@ async def analyze_volume(
     analysis_prompt: str,
     use_recent_context: bool = False,
     timepoint: int | None = None,
-    context: dict = None,
+    context: dict | None = None,
 ) -> str:
     """Analyze embryo volume with Claude Vision"""
     agent, err = require_agent(context)
@@ -60,7 +60,7 @@ async def analyze_volume(
 def get_recent_perceptions(
     embryo_id: str | None = None,
     n: int = 5,
-    context: dict = None,
+    context: dict | None = None,
 ) -> str:
     """Read live per-embryo perception state from the perception sessions.
 

gently/app/tools/calibration_tools.py

@@ -14,7 +14,7 @@
 
 from gently.analysis.core import AdaptiveSweepState, FitFunction, fit_focus_curve  # noqa: E402
 from gently.harness.state import CalibrationPrior  # noqa: E402
-from gently.harness.tools.helpers import get_embryo_or_error  # noqa: E402
+from gently.harness.tools.helpers import ctx_get, get_embryo_or_error  # noqa: E402
 from gently.harness.tools.registry import ToolCategory, ToolExample, tool  # noqa: E402
 from gently.ui.web.plots import (  # noqa: E402
     generate_calibration_summary_plot,
@@ -462,7 +462,7 @@ async def hybrid_focus_selection(
     agent,
     embryo_id: str,
     fft_confidence_threshold: float = 0.85,
-) -> tuple[int, str, float]:
+) -> tuple[int, str, float | None]:
     """
     Two-stage focus selection: FFT first, Vision if ambiguous.
 
@@ -696,8 +696,8 @@ async def fast_calibrate_embryo(
     """
     from gently.hardware.dispim.claude_client import AsyncClaudeClient
 
-    agent = context.get("agent")
-    client = context.get("client")
+    agent = ctx_get(context, "agent")
+    client = ctx_get(context, "client")
 
     if not agent:
         return False, "Error: No agent context", 0
@@ -946,15 +946,15 @@ async def fast_calibrate_embryo(
 async def calibrate_embryo(
     embryo_id: str,
     skip_edge_detection: bool = False,
-    galvo_top: float = None,
-    galvo_bottom: float = None,
+    galvo_top: float | None = None,
+    galvo_bottom: float | None = None,
     edge_step: float = 0.05,
     edge_max_range: float = 0.5,
     edge_tolerance_deg: float = 0.20,
     inset_fraction: float = 0.4,
     z_buffer_um: float = 25.0,
     use_v04_plan: bool = False,
-    context: dict = None,
+    context: dict | None = None,
 ) -> str:
     """Run piezo-galvo calibration with Claude vision edge detection.
 
@@ -999,8 +999,8 @@ async def calibrate_embryo(
     from gently.analysis.core import calculate_focus_score
     from gently.hardware.dispim.claude_client import AsyncClaudeClient
 
-    agent = context.get("agent")
-    client = context.get("client")
+    agent = ctx_get(context, "agent")
+    client = ctx_get(context, "client")
 
     if not agent:
         return "Error: No agent context"
@@ -1396,13 +1396,13 @@ async def check_embryo_at_position(galvo_pos: float) -> bool:
     ],
 )
 async def calibrate_all_embryos(
-    embryo_ids: list[str] = None,
+    embryo_ids: list[str] | None = None,
     skip_edge_detection: bool = False,
     z_buffer_um: float = 25.0,
-    context: dict = None,
+    context: dict | None = None,
 ) -> str:
     """Calibrate all embryos sequentially with Claude vision"""
-    agent = context.get("agent")
+    agent = ctx_get(context, "agent")
 
     if not agent:
         return "Error: No agent context"
@@ -1512,7 +1512,7 @@ def apply_calibration_to_embryos(
     source_embryo_id: str,
     target_embryo_ids: list[str] | None = None,
     overwrite_existing: bool = True,
-    context: dict = None,
+    context: dict | None = None,
 ) -> str:
     """Broadcast one embryo's calibration to others.