fix(typing): drain 4 more modules (exempt list 20 -> 16, 83 of 99 checked)

pyproject.toml

@@ -109,8 +109,8 @@ explicit_package_bases = true
 # 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. 79 of the 99 modules are already clean;
-# the 20 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. 83 of the 99 modules are already clean;
+# the 16 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.
@@ -132,19 +132,15 @@ module = [
     "capability_outcome_bridge",
     "capability_propensity",
     "capability_recurrence_check",
-    "codemod_lane",
     "dispatcher",
-    "exp_abcd",
     "feedback",
     "issue_readiness",
     "keepalive_outcomes",
     "observability_dashboard",
     "redirect_sweep",
-    "repo_knowledge",
     "roles",
     "router",
     "runtime_ac",
     "runtime_ac_gate",
-    "switch_review",
 ]
 ignore_errors = true

src/codemod_lane.py

@@ -183,7 +183,7 @@ def validate_campaign(campaign: dict[str, Any]) -> list[str]:
         campaign_id = meta.get("id")
         if not _is_nonempty_string(campaign_id):
             errors.append("campaign.id must be a non-empty string")
-        elif not _looks_like_slug(campaign_id.strip()):
+        elif not _looks_like_slug(str(campaign_id or "").strip()):
             errors.append("campaign.id must be a lowercase slug")
         if not _is_nonempty_string(meta.get("title")):
             errors.append("campaign.title must be a non-empty string")
@@ -333,7 +333,10 @@ def _recipe_dry_run_command(
 
     if tool == "ast-grep":
         if _is_nonempty_string(rule_file):
-            return " ".join(["ast-grep", "scan", "--rule", shlex.quote(rule_file.strip())]), None
+            return (
+                " ".join(["ast-grep", "scan", "--rule", shlex.quote(str(rule_file or "").strip())]),
+                None,
+            )
         if _is_nonempty_string(match):
             return (
                 " ".join(
@@ -343,7 +346,7 @@ def _recipe_dry_run_command(
                         "-l",
                         shlex.quote(language),
                         "-p",
-                        shlex.quote(match.strip()),
+                        shlex.quote(str(match or "").strip()),
                     ]
                 ),
                 None,
@@ -356,8 +359,8 @@ def _recipe_dry_run_command(
             " ".join(
                 [
                     "comby",
-                    shlex.quote(match.strip()),
-                    shlex.quote(rewrite.strip()),
+                    shlex.quote(str(match or "").strip()),
+                    shlex.quote(str(rewrite or "").strip()),
                     shlex.quote(target),
                     "-matcher",
                     shlex.quote(matcher),
@@ -375,7 +378,7 @@ def _recipe_dry_run_command(
                     [
                         "jscodeshift",
                         "-t",
-                        shlex.quote(rule_file.strip()),
+                        shlex.quote(str(rule_file or "").strip()),
                         "--dry",
                         "--print",
                         shlex.quote(target),
@@ -384,14 +387,14 @@ def _recipe_dry_run_command(
                 None,
             )
         if _is_nonempty_string(command_template):
-            cmd = command_template.strip()
+            cmd = str(command_template or "").strip()
             if _template_is_safe_dry_run(cmd) and _has_dry_run_marker(cmd):
                 return cmd, None
             return None, "jscodeshift command_template omitted because it is not clearly dry-run"
 
     if tool == "openrewrite":
         if _is_nonempty_string(command_template):
-            cmd = command_template.strip()
+            cmd = str(command_template or "").strip()
             if _template_is_safe_dry_run(cmd) and _has_dry_run_marker(cmd):
                 return cmd, None
             return (
@@ -403,15 +406,15 @@ def _recipe_dry_run_command(
                 " ".join(
                     [
                         "./mvnw",
-                        f"-Drewrite.activeRecipes={shlex.quote(rule_file.strip())}",
+                        f"-Drewrite.activeRecipes={shlex.quote(str(rule_file or "").strip())}",
                         "rewrite:dryRun",
                     ]
                 ),
                 None,
             )
 
     if tool == "custom" and _is_nonempty_string(command_template):
-        cmd = command_template.strip()
+        cmd = str(command_template or "").strip()
         if _template_is_safe_dry_run(cmd) and _has_dry_run_marker(cmd):
             return cmd, None
         return None, "custom command_template omitted because it is not clearly safe dry-run"

src/feedback.py

@@ -34,6 +34,7 @@
 import sys
 import time
 from pathlib import Path
+from typing import Any
 
 import execution_profiles
 
@@ -675,7 +676,7 @@ def _sanitize_completion_value(value, redacted: list[str], path: str):
             for index, item in enumerate(items[:MAX_COMPLETION_LIST_ITEMS])
         ]
     if isinstance(value, dict):
-        clean = {}
+        clean: dict[str, Any] = {}
         for key, item in sorted(value.items(), key=lambda pair: str(pair[0])):
             name = str(key)
             child_path = f"{path}.{name}" if path else name
@@ -719,7 +720,7 @@ def _sanitize_artifact_refs(value, redacted: list[str]) -> list[dict]:
         unsafe = sorted(set(row) - allowed)
         if unsafe:
             redacted.extend(f"artifact_refs[{index}].{key}" for key in unsafe)
-        clean = {}
+        clean: dict[str, Any] = {}
         for key in sorted(allowed & set(row)):
             item = row.get(key)
             if item in (None, ""):
@@ -760,7 +761,7 @@ def _sanitize_completion_payload(payload: dict | None) -> tuple[dict, str, int]:
     raw = dict(payload or {})
     redacted: list[str] = []
     rejection_codes = []
-    clean = {}
+    clean: dict[str, Any] = {}
     for key, value in sorted(raw.items()):
         name = str(key)
         if name not in COMPLETION_PAYLOAD_FIELDS:
@@ -2574,7 +2575,7 @@ def _record_outcome_in_conn(
             # episode stays ineligible -- the honest outcome, not a gap to paper over with a
             # constant. (A caller-supplied gate id would be better still; this uses evidence
             # record_outcome already holds rather than inventing a parameter.)
-            verification_payload = dict(payload)
+            verification_payload: dict[str, Any] = dict(payload)
             if stored_ci:
                 verification_payload["acceptance_gate_ids"] = ["ci"]
             _record_completion_event_in_conn(
@@ -2818,7 +2819,7 @@ def role_activation_metrics(*, conn: sqlite3.Connection | None = None) -> dict:
             role_runs = int(
                 c.execute("SELECT COUNT(*) FROM runs WHERE role_name=?", (role,)).fetchone()[0]
             )
-            out = selector[role]
+            out: dict[str, Any] = selector[role]
             out["role_runs"] = role_runs
             out["linked"] = int(linked or 0)
             out["durable"] = int(durable or 0)
@@ -3327,7 +3328,7 @@ def canonical_rank(row: tuple) -> tuple:
                         derived_subject = research_subjects.subject_identity_from_hash(
                             canonical_target,
                             canonical_task_type,
-                            raw_spec_hash,
+                            str(raw_spec_hash or ""),
                             base_sha,
                             normalized_arms,
                             profile_value,
@@ -3629,7 +3630,7 @@ def _record_execution_attempt_in_conn(
 ) -> None:
     role = validate_operation_role(operation_role)
     if role == "worker":
-        resolved_model = validate_resolved_worker_model(resolved_model)
+        validated_model = validate_resolved_worker_model(resolved_model)
     try:
         ordinal = max(1, int(attempt_ordinal or 1))
     except (TypeError, ValueError):
@@ -3821,7 +3822,11 @@ def complete_profile_attempt(
     """
     if not str(resolved_model or "").strip():
         raise ValueError("profile completion requires actually reported resolved_model")
-    resolved_model = validate_resolved_worker_model(resolved_model)
+    # A SEPARATE BINDING, not a reassignment: `validate_resolved_worker_model` returns None for a
+    # rejected adapter tag, and the write below must still see that None. Reassigning `resolved_model`
+    # (typed `str`) widened it, and coercing the None away — the first fix attempted here — turned a
+    # deliberate refusal into an empty string, which broke three provenance tests.
+    validated_model: str | None = validate_resolved_worker_model(resolved_model)
     attempt_id = f"attempt:profile:{run_id}"
     with _conn() as c:
         row = c.execute(
@@ -3846,7 +3851,7 @@ def complete_profile_attempt(
             "WHERE attempt_id=?",
             (
                 resolved_provider,
-                resolved_model,
+                validated_model,
                 status,
                 int(completed_ts or time.time()),
                 attempt_id,
@@ -4785,7 +4790,7 @@ def relearn_quality(task_type_priors: dict, window_days: int = 120) -> int:
             for m in metrics
         }
 
-        def _effective(cell_agent: str, s: dict, m: str) -> tuple[float, str]:
+        def _effective(cell_agent: str, s: dict, m: str) -> tuple[float | None, str]:
             if s[m]:
                 return float(s[m]), "m"
             if cell_agent in agent_mean[m]:

src/issue_readiness.py

@@ -52,6 +52,7 @@
 import re
 import subprocess
 import sys
+from typing import Any
 
 import backlog
 
@@ -206,7 +207,9 @@ def out(verdict, reason):
 
 def assess(issues: list[dict]) -> dict:
     """Classify a list of issues and roll up. Exclusions are COUNTED, never silently dropped."""
-    rows, counts, reasons = [], {v: 0 for v in VERDICTS}, {}
+    rows: list = []
+    counts = {v: 0 for v in VERDICTS}
+    reasons: dict[str, Any] = {}
     for issue in issues:
         verdict = classify_issue(issue)
         repo = (issue.get("repository") or {}).get("name") or issue.get("repo") or "?"
@@ -454,13 +457,15 @@ def apply_ready(rows: list[dict], *, dry_run: bool = True) -> dict:
     r"|\b\d{4}-\d{2}-\d{2}\b|\(run \d+\)|#\d+\b",
     re.IGNORECASE,
 )
-BOT_AUTHOR = re.compile(r"\[bot\]$|^app/|^(renovate|dependabot|github-actions)$", re.IGNORECASE)
+BOT_AUTHOR: re.Pattern[str] = re.compile(
+    r"\[bot\]$|^app/|^(renovate|dependabot|github-actions)$", re.IGNORECASE
+)
 _TITLE_NOISE = re.compile(
     r"[\U0001F300-\U0001FAFF\u2600-\u27BF]|#\d+|\b\d{4}-\d{2}-\d{2}\b|\b\d+\b"
 )
 
 
-def normalize_title(title: str) -> str:
+def normalize_title(title: str | None) -> str:
     """Strip emoji, dates and numbers so the same recurring tracker matches across repos."""
     return re.sub(r"\s+", " ", _TITLE_NOISE.sub("", str(title or ""))).strip().lower()
 
@@ -934,7 +939,9 @@ def _selftest() -> None:
     # one hardcoded variant silently under-applies. Resolution must follow the repo, and a repo
     # with no ready label at all must produce an error rather than a skipped no-op.
     spellings = {"spaced": "status: ready", "tight": "status:ready", "none": None}
-    live_cache = repo_ready_label.__defaults__[0]  # the module's own memo, primed for the test
+    live_cache = (repo_ready_label.__defaults__ or (None,))[
+        0
+    ]  # the module's own memo, primed for the test
     saved_run, calls = subprocess.run, []
     try:
         live_cache.update(spellings)
@@ -948,7 +955,7 @@ class R:
                 calls.append((cmd[cmd.index("--repo") + 1], cmd[cmd.index("--add-label") + 1]))
             return R()
 
-        subprocess.run = fake_run
+        subprocess.run = fake_run  # type: ignore[assignment]  # deliberate selftest monkeypatch
         out = apply_ready(rows, dry_run=False)
     finally:
         subprocess.run = saved_run
@@ -1117,7 +1124,9 @@ def _iss(repo, num, title, labels=(), author="app/github-actions", body="x"):
     # DELIBERATE BREAK -> REVERT on the guard: without it, a testgen issue is stolen by codemod.
     _saved_classify = backlog.classify
     try:
-        backlog.classify = lambda _labels: "implement"  # pretend there is never a signal
+        backlog.classify = (
+            lambda _labels: "implement"
+        )  # pretend there is never a signal  # type: ignore[assignment]  # deliberate selftest monkeypatch
         stolen = task_label_for(dict(camp, labels=[{"name": "testing"}]))
         assert stolen == "refactor", "break did not change behaviour — test is vacuous"
     finally:

src/repo_knowledge.py

@@ -2602,7 +2602,7 @@ def main(argv: list[str]) -> int:
             if "--max-lines" in argv
             else AGENTS_EXPORT_MAX_LINES
         )
-        result = validate_agents_md_export(repo_path, repo_arg=repo_arg, max_lines=max_lines)
+        result = validate_agents_md_export(repo_path, repo=repo_arg, max_lines=max_lines)
         print(json.dumps(result, indent=2) if "--json" in argv else result)
         return 0 if result["ok"] else 1
     if "--suggest-from-snapshot" in argv:
@@ -2667,7 +2667,7 @@ def main(argv: list[str]) -> int:
             json.dumps(
                 suggest_from_docs(
                     repo_path,
-                    repo_arg=repo_arg,
+                    repo=repo_arg,
                     max_per_repo=max_per_repo,
                     include_root_docs="--include-root-docs" in argv,
                     sections=sections,

src/switch_review.py

@@ -37,6 +37,7 @@
 import subprocess
 import sys
 import time
+from collections.abc import Mapping
 from pathlib import Path
 
 import capabilities
@@ -89,7 +90,7 @@ def _capability_heartbeat(event_type: str = "invocation") -> None:
 MIRROR_DIR = Path(os.environ.get("ORCH_MIRROR", Path.home() / ".codex" / "orchestrator-mirror"))
 
 
-def stale_runners(*, now: int | None = None, mirror: Path | None = None) -> list[dict]:
+def stale_runners(*, now: float | None = None, mirror: Path | None = None) -> list[dict]:
     """Long-lived processes running mirror code OLDER than the mirror on disk.
 
     WHY THIS IS THE SAME DEFECT CLASS AS A HELD SWITCH. `orch-sync-mirror.sh` is treated as the
@@ -119,7 +120,7 @@ def stale_runners(*, now: int | None = None, mirror: Path | None = None) -> list
         ).stdout
     except (OSError, subprocess.SubprocessError):
         return []
-    now = float(now if now is not None else time.time())
+    resolved_now = float(now if now is not None else time.time())
     stale: list[dict] = []
     for line in out.splitlines():
         parts = line.strip().split(None, 2)
@@ -129,7 +130,7 @@ def stale_runners(*, now: int | None = None, mirror: Path | None = None) -> list
         age = _etime_seconds(etime)
         if age is None:
             continue
-        started = now - age
+        started = resolved_now - age
         if started >= newest:
             continue
         stale.append(
@@ -167,7 +168,7 @@ def _etime_seconds(etime: str) -> int | None:
     return days * 86400 + nums[0] * 3600 + nums[1] * 60 + nums[2]
 
 
-def review(*, now: int | None = None, env: dict | None = None, path=None) -> dict:
+def review(*, now: int | None = None, env: Mapping[str, str] | None = None, path=None) -> dict:
     """Which held-or-idle switches are due for an owner decision, and why."""
     import capability_recurrence_check as rc
 
@@ -241,7 +242,9 @@ def review(*, now: int | None = None, env: dict | None = None, path=None) -> dic
 
 def raise_questions(rep: dict, *, dry_run: bool = True) -> dict:
     """Record ONE non-blocking, auto-expiring owner question per due switch."""
-    raised, deduped, errors = [], [], []
+    raised: list = []
+    deduped: list = []
+    errors: list = []
     for row in rep["held_off"] + rep["on_but_idle"]:
         flag, cap_id, state = row["flag"], row["capability"], row["state"]
         if state == "off":