fix(typing): drain the <=5-finding tail (exempt list 43 -> 26, 73 of 99 checked)

pyproject.toml

@@ -104,10 +104,13 @@ explicit_package_bases = true
 # config rather than the code — 769 instead of 601 when first measured, all of the difference
 # spurious. The same reason `pythonpath = ["src"]` exists for pytest above: one relocation, two
 # tools that each need telling.
-mypy_path = ["src"]
+# `tests` as well as `src`: `capability_admission` reads the recurrence-fixture roster from
+# `test_capability_set_coverage`, a real and declared dependency (see paths.TESTS_DIR). mypy_path
+# affects RESOLUTION only — the check target stays `src`, so the tests are not themselves checked.
+mypy_path = ["src", "tests"]
 
-# THE RATCHET, and the reason the check can be ON at all. 56 of the 99 modules are already clean;
-# the 43 below are not, and each is exempt BY NAME so the check can run TODAY over the
+# THE RATCHET, and the reason the check can be ON at all. 73 of the 99 modules are already clean;
+# the 26 below are not, and each is exempt BY NAME so the check can run TODAY over the
 # clean two-thirds instead of being off over everything. A blanket `ignore_errors` or a
 # `disable_error_code` list was rejected: both make the job green while checking nothing, which is
 # the defect verify.py exists to stop, and neither has anything to count.
@@ -123,48 +126,31 @@ mypy_path = ["src"]
 # module is now a red, and a module can never quietly rejoin the exempt set.
 [[tool.mypy.overrides]]
 module = [
-    "agent_auth_check",
     "capabilities",
     "capability_activation_audit",
-    "capability_admission",
     "capability_advisor",
     "capability_compiler",
-    "capability_effectiveness",
-    "capability_opportunity",
     "capability_outcome_bridge",
     "capability_propensity",
     "capability_recurrence_check",
-    "ccusage_reconcile",
-    "claims",
     "codemod_lane",
     "consumer_sync_artifact_ingest",
-    "cross_repo_lane",
     "dispatcher",
     "durability_sweep",
     "exp_abcd",
-    "exploration_backfill",
-    "exploration_collection",
-    "exploration_evidence_plan",
     "feedback",
     "issue_quality",
     "issue_readiness",
     "keepalive_evidence",
     "keepalive_outcomes",
-    "mcp_server",
     "observability_dashboard",
-    "pattern_miner",
-    "range_lane_rollout",
-    "redirect_apply",
     "redirect_sweep",
     "repo_knowledge",
-    "research_scheduler",
-    "research_subjects",
     "roles",
     "router",
     "runtime_ac",
     "runtime_ac_gate",
     "switch_review",
     "tick",
-    "verify",
 ]
 ignore_errors = true

src/agent_auth_check.py

@@ -201,7 +201,7 @@ def _render(rows: list[dict]) -> str:
 
 
 def _selftest() -> None:
-    rows = [
+    rows: list[dict] = [
         {
             "agent": "a",
             "verdict": "OK",
@@ -279,7 +279,10 @@ def _boom(self, *a, **k):
                     raise OSError("Resource temporarily unavailable")
                 return _real_read(self, *a, **k)
 
-            Path.read_text = _boom
+            # Deliberate monkeypatch inside the selftest: the point is to make a real read FAIL
+            # so the OSError path is exercised. `method-assign` is right in general and wrong
+            # here, so the exemption is narrow and stated rather than configured away.
+            Path.read_text = _boom  # type: ignore[method-assign]
             _bad = _credential_file_state("vibe")
             assert _bad["key_present"] is None, _bad
             assert _bad["reason_class"] == "environment", _bad
@@ -288,12 +291,12 @@ def _boom(self, *a, **k):
             assert CRED_READ_ATTEMPTS >= 2, "the retry is the point; one attempt is no retry"
             assert len(_calls) == CRED_READ_ATTEMPTS, (len(_calls), CRED_READ_ATTEMPTS)
             # A file that is genuinely ABSENT stays a hard failure and is NOT retried.
-            Path.read_text = _real_read
+            Path.read_text = _real_read  # type: ignore[method-assign]
             CREDENTIAL_FILES["vibe"] = (Path(_td) / "nope.env", "MISTRAL_API_KEY")
             _gone = _credential_file_state("vibe")
             assert _gone["present"] is False and not _gone.get("reason_class"), _gone
         finally:
-            Path.read_text = _real_read
+            Path.read_text = _real_read  # type: ignore[method-assign]
             if _saved is not None:
                 CREDENTIAL_FILES["vibe"] = _saved
 

src/capability_admission.py

@@ -473,7 +473,8 @@ def admit(capability_id: str, *, path: pathlib.Path | None = None, ctx: dict | N
     if cap is None:
         raise ValueError(f"unknown capability: {capability_id}")
     ctx = ctx or _context(path)
-    checks, missing = {}, []
+    checks: dict[str, dict[str, Any]] = {}
+    missing: list[str] = []
     for name, fn in REQUIREMENTS:
         try:
             ok, detail = fn(cap, ctx)
@@ -537,7 +538,8 @@ def preflight(spec: dict) -> dict:
         "fixtures": set(),
         **_findability_context([stub["capability_id"]]),
     }
-    checks, missing = {}, []
+    checks: dict[str, dict[str, Any]] = {}
+    missing: list[str] = []
     # Caller/heartbeat/fixture cannot be verified for code that does not exist; they are reported as
     # OBLIGATIONS rather than silently skipped, because silently skipping is how they got skipped.
     obligations = {"caller_exists", "heartbeat", "fixture"}

src/capability_effectiveness.py

@@ -35,6 +35,7 @@
 import argparse
 import json
 import sys
+from typing import Any
 
 import capabilities
 import feedback
@@ -121,7 +122,7 @@ def _arm_stats(edges: list[dict]) -> dict:
     # Distinct subjects — a target counts as durable if ANY attempt on it landed durably.
     durable_subjects = {_target_of(e["run_id"]) for e in durable}
     terminal_subjects = {_target_of(e["run_id"]) for e in terminal}
-    out = {
+    out: dict[str, Any] = {
         "attributed": len(edges),
         "terminal": len(terminal),
         "durable": len(durable),

src/capability_opportunity.py

@@ -29,6 +29,7 @@
 import json
 import os
 import sys
+from collections.abc import Mapping
 
 import capabilities
 import feedback
@@ -99,7 +100,12 @@ def _role_invocation_counts(conn=None) -> dict[str, int]:
 
 
 def assess(
-    cap_id: str, cap: dict, *, task_counts: dict, role_counts: dict, env: dict | None = None
+    cap_id: str,
+    cap: dict,
+    *,
+    task_counts: dict,
+    role_counts: dict,
+    env: Mapping[str, str] | None = None,
 ) -> dict:
     """One capability: its trigger, the work that matched it, and the resulting verdict."""
     env = os.environ if env is None else env
@@ -153,7 +159,7 @@ def assess(
     }
 
 
-def report(*, path=None, env: dict | None = None) -> dict:
+def report(*, path=None, env: Mapping[str, str] | None = None) -> dict:
     caps = capabilities.load(path or capabilities.REG)
     task_counts = _task_type_counts()
     role_counts = _role_invocation_counts()

src/ccusage_reconcile.py

@@ -145,12 +145,12 @@ def _run_windows(
             skipped["unsupported_agent"] += 1
             continue
         starts = [
-            int(row.get("ts"))
+            int(row.get("ts") or 0)
             for row in run_rows
             if row.get("event") == "start" and isinstance(row.get("ts"), (int, float))
         ]
         completes = [
-            int(row.get("ts"))
+            int(row.get("ts") or 0)
             for row in run_rows
             if row.get("event") == "complete" and isinstance(row.get("ts"), (int, float))
         ]
@@ -237,7 +237,7 @@ def _match_window(
     if agent not in ATTRIBUTABLE_AGENTS:
         return None, "unsupported_agent"
     metadata = session.get("metadata") if isinstance(session.get("metadata"), dict) else {}
-    last_ts = _parse_iso_ts(metadata.get("lastActivity"))
+    last_ts = _parse_iso_ts((metadata or {}).get("lastActivity"))
     if last_ts is None:
         return None, "missing_last_activity"
     matches = [

src/claims.py

@@ -349,7 +349,8 @@ def _selftest() -> None:
         assert claim(T1, "claude") is False, "same-target collision must be blocked"
         assert claim(T2, "claude") is True
         assert claim(T1, "codex") is True, "idempotent same-agent re-claim"
-        assert holder(T1)["agent"] == "codex"
+        held = holder(T1)
+        assert held and held["agent"] == "codex"
         assert set(active_claims()) == {T1, T2}, active_claims()
         assert (
             update_metadata(T1, "codex", lane="opener", task_type="implement", pid=os.getpid())
@@ -366,7 +367,8 @@ def _selftest() -> None:
         assert holder(T1) is None
         assert claim(T1, "claude") is True
         assert release(T1, "codex") is False, "wrong-agent release must be refused"
-        assert holder(T1)["agent"] == "claude"
+        held = holder(T1)
+        assert held and held["agent"] == "claude"
 
         # stale claim owned by 'codex' (dead pid, old ts)
         stale = _claims_dir() / _slug(T3)
@@ -451,7 +453,10 @@ def _selftest() -> None:
                 }
             )
         )
-        assert holder(T3)["agent"] == "research", "any live child pid keeps a research claim held"
+        held = holder(T3)
+        assert (
+            held and held["agent"] == "research"
+        ), "any live child pid keeps a research claim held"
         release(T3)
 
         # no-meta TOCTOU guard: a fresh (unstamped) dir reads as HELD, not stale

src/cross_repo_lane.py

@@ -185,7 +185,7 @@ def validate_coordination(plan: dict[str, Any]) -> list[str]:
         coord_id = meta.get("id")
         if not _is_nonempty_string(coord_id):
             errors.append("coordination.id must be a non-empty string")
-        elif not _looks_like_slug(coord_id.strip()):
+        elif not _looks_like_slug(str(coord_id or "").strip()):
             errors.append("coordination.id must be a lowercase slug")
         for key in ("title", "goal", "source_repo"):
             if not _is_nonempty_string(meta.get(key)):
@@ -267,10 +267,10 @@ def validate_coordination(plan: dict[str, Any]) -> list[str]:
         repo = consumer.get("repo")
         if not _is_nonempty_string(repo):
             errors.append(f"{path}.repo must be a non-empty string")
-        elif not _looks_like_repo(repo):
+        elif not _looks_like_repo(str(repo or "")):
             errors.append(f"{path}.repo must look like owner/repo")
         else:
-            consumer_repos.append(repo)
+            consumer_repos.append(str(repo))
         if not _is_nonempty_string(consumer.get("reason")):
             errors.append(f"{path}.reason must be a non-empty string")
         errors.extend(
@@ -352,7 +352,7 @@ def validate_coordination(plan: dict[str, Any]) -> list[str]:
         template = prompts.get("consumer_prompt_template")
         if not _is_nonempty_string(template):
             errors.append("prompts.consumer_prompt_template must be a non-empty string")
-        elif "{repo}" not in template:
+        elif "{repo}" not in str(template or ""):
             errors.append("prompts.consumer_prompt_template must contain '{repo}'")
         if not _is_nonempty_string(prompts.get("review_prompt")):
             errors.append("prompts.review_prompt must be a non-empty string")

src/exp_abcd.py

@@ -1353,6 +1353,21 @@ def still_running(proc) -> bool:
     }
 
 
+def _bind_synthesis(fn, repo: str, exp_id: str):
+    """Bind THIS iteration's repo/exp_id, which a bare lambda in a loop would not.
+
+    Was `lambda repo=repo, exp_id=edir.name: fn(repo, exp_id)`. The default-arg trick is what made
+    it correct — it captures the current values so a later loop turn cannot rebind them — and it is
+    also what stopped mypy inferring the lambda's type. A closure over explicit parameters says the
+    same thing and is checkable.
+    """
+
+    def launch():
+        return fn(repo, exp_id)
+
+    return launch
+
+
 def followup(
     *,
     max_experiments: int = 1,
@@ -1563,9 +1578,7 @@ def launch_fn(repo=meta["repo"], exp_id=edir.name):
                     if launch_available and not promotion_inflight:
                         promotion = promotion_reconcile(
                             edir,
-                            launch_fn=lambda repo=repo, exp_id=edir.name: (
-                                (synthesize_fn or synthesize)(repo, exp_id)
-                            ),
+                            launch_fn=_bind_synthesis(synthesize_fn or synthesize, repo, edir.name),
                             completion_fn=promotion_completion_fn,
                             resume_fn=promotion_resume_fn or _resume_synthesis_promotion,
                             verify_fn=promotion_verify_fn,
@@ -1657,7 +1670,7 @@ def _winner_and_harvest(
         denom = sum(w for _, w in scs)
         means[agent] = (sum(score * w for score, w in scs) / denom) if denom else 0.0
         notes[agent] = ns
-    winner = max(means, key=means.get)
+    winner = max(means, key=lambda k: means[k])
     return {
         "winner": winner,
         "winner_mean": means[winner],

src/exploration_backfill.py

@@ -20,6 +20,7 @@
 import shutil
 import tempfile
 import time
+from collections.abc import Mapping
 from pathlib import Path
 
 import claims
@@ -505,7 +506,7 @@ def schedule_backfill(
     backlog_path: Path | None = None,
     backlog_payload: dict | None = None,
     confirm: bool = False,
-    env: dict | None = None,
+    env: Mapping[str, str] | None = None,
     prepare_fn=None,
     issue_body_fn=None,
 ) -> dict:
@@ -794,7 +795,8 @@ def fake_prepare(
         # template alone would leave the two halves free to drift apart, which is the same shape
         # as a gate whose measuring window differs from its draining window.
         assert calls[0]["exp_id"].startswith(f"{planned_job['exp_id_template']}-"), calls
-        assert claims.holder("o/r#1")["agent"] == BACKFILL_CLAIM_AGENT, claims.holder("o/r#1")
+        held = claims.holder("o/r#1")
+        assert held and held["agent"] == BACKFILL_CLAIM_AGENT, held
 
         no_progress_db = Path(tmp) / "no-progress.db"
         feedback.DB_PATH = no_progress_db

src/exploration_collection.py

@@ -26,6 +26,7 @@
 import time
 from contextlib import contextmanager
 from pathlib import Path
+from typing import cast
 
 import claims
 import dispatcher
@@ -128,6 +129,8 @@ def _filter_backlog(
 
 def _entry_is_late(task_type: str, assignment: dict) -> bool:
     for entry in (router.ROUTE_TABLE.get(task_type, {}) or {}).get("agents") or []:
+        if not isinstance(entry, dict):
+            continue
         if entry.get("agent") == assignment.get("agent") and entry.get("mode") == assignment.get(
             "mode"
         ):
@@ -421,7 +424,7 @@ def build_window(
                 require_exploration=True,
             )
             count = _exploratory_count(probe_decision)
-            seed_search["attempted"] += 1
+            seed_search["attempted"] = int(seed_search.get("attempted") or 0) + 1
             if count > best_count:
                 best_count = count
                 best_decision = probe_decision
@@ -456,7 +459,7 @@ def build_window(
         if not dry_run and rejected_assignments:
             _release_rejected_claims(rejected_assignments)
         if not dry_run and _exploratory_count(decision) < min_exploratory:
-            _release_assignments(decision.get("assignments") or [])
+            _release_assignments(cast(list, decision.get("assignments") or []))
             blocked_reasons.append(
                 "active claim race left too few direct exploration assignments to dispatch"
             )

src/exploration_evidence_plan.py

@@ -16,6 +16,7 @@
 import tempfile
 import time
 from pathlib import Path
+from typing import Any
 
 import backlog
 import capacity
@@ -182,7 +183,7 @@ def _candidate_task_types(
         if len(sample_targets[task_type]) < 3:
             sample_targets[task_type].append(item.get("target") or "")
     route_rows = {row["task_type"]: row for row in coverage.get("tasks") or []}
-    candidates = []
+    candidates: list[dict[str, Any]] = []
     for task_type in sorted(set(route_rows) | set(opener_counts) | set(outcome_counts)):
         route = route_rows.get(task_type) or {}
         opener_items = opener_counts.get(task_type, 0)

src/issue_quality.py

@@ -280,7 +280,7 @@ def parallel_fetch(keys: set[tuple[str, int]], fn) -> dict[tuple[str, int], obje
     )
     body_keys: set[tuple[str, int]] = set()
     for repo, pr_number in linked_cache:
-        issue_numbers: list[int] = linked_cache.get((repo, pr_number)) or []
+        issue_numbers = list(linked_cache.get((repo, pr_number)) or [])
         if not isinstance(issue_numbers, list):
             continue
         for issue_number in issue_numbers:
@@ -291,7 +291,7 @@ def parallel_fetch(keys: set[tuple[str, int]], fn) -> dict[tuple[str, int], obje
     body_cache = parallel_fetch(body_keys, fetch_body)
 
     for row, repo, pr_number in parsed_rows:
-        issue_numbers = linked_cache.get((repo, pr_number)) or []
+        issue_numbers = list(linked_cache.get((repo, pr_number)) or [])
         if not issue_numbers:
             skipped.append(
                 {