fix(typing): drain 21 modules off the mypy exempt list (64 -> 43)

pyproject.toml

@@ -106,8 +106,8 @@ explicit_package_bases = true
 # tools that each need telling.
 mypy_path = ["src"]
 
-# THE RATCHET, and the reason the check can be ON at all. 35 of the 99 modules are already clean;
-# the 64 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. 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
 # 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,10 +123,7 @@ mypy_path = ["src"]
 # module is now a red, and a module can never quietly rejoin the exempt set.
 [[tool.mypy.overrides]]
 module = [
-    "adversarial",
     "agent_auth_check",
-    "backlog",
-    "cadence_registry",
     "capabilities",
     "capability_activation_audit",
     "capability_admission",
@@ -137,41 +134,27 @@ module = [
     "capability_outcome_bridge",
     "capability_propensity",
     "capability_recurrence_check",
-    "capability_targets",
     "ccusage_reconcile",
     "claims",
     "codemod_lane",
     "consumer_sync_artifact_ingest",
-    "consumer_sync_shadow",
     "cross_repo_lane",
     "dispatcher",
     "durability_sweep",
-    "epic_lane",
-    "evidence_acquisition",
-    "execution_profiles",
     "exp_abcd",
     "exploration_backfill",
     "exploration_collection",
     "exploration_evidence_plan",
-    "exploration_review",
-    "features",
     "feedback",
-    "gh_capacity",
-    "human_calibration",
     "issue_quality",
     "issue_readiness",
-    "judge_reliability",
     "keepalive_evidence",
     "keepalive_outcomes",
-    "keepalive_shadow",
-    "ledger_reconcile",
     "mcp_server",
     "observability_dashboard",
-    "partitioned_review",
     "pattern_miner",
     "range_lane_rollout",
     "redirect_apply",
-    "redirect_shadow",
     "redirect_sweep",
     "repo_knowledge",
     "research_scheduler",
@@ -180,12 +163,8 @@ module = [
     "router",
     "runtime_ac",
     "runtime_ac_gate",
-    "strategy_experiment",
     "switch_review",
-    "synthesis_promotion",
     "tick",
-    "ux_review",
     "verify",
-    "watch",
 ]
 ignore_errors = true

scripts/ci_lint_baseline.py

@@ -41,6 +41,7 @@
 
 import argparse
 import json
+import pathlib
 import re
 import shutil
 import subprocess
@@ -191,6 +192,34 @@ def measure_black() -> dict:
     }
 
 
+def _measure_behind_exempt_list(target: str) -> tuple[int, int]:
+    """Findings hidden by `[[tool.mypy.overrides]] ignore_errors`, and how many modules hide them.
+
+    Run against a COPY of pyproject.toml with that override removed, so the real config is never
+    mutated even transiently -- a crash mid-measure must not leave the repo's gate config edited.
+    """
+    import re
+    import tempfile
+
+    text = (REPO / "pyproject.toml").read_text(encoding="utf-8")
+    stripped = re.sub(
+        r"\n\[\[tool\.mypy\.overrides\]\]\nmodule = \[[\s\S]*?\]\nignore_errors = true\n",
+        "\n",
+        text,
+    )
+    exempt = len(re.findall(r'^    "[a-z_]+",$', text, re.M))
+    if stripped == text:
+        return 0, 0
+    with tempfile.TemporaryDirectory(prefix="lint-baseline-") as td:
+        cfg = pathlib.Path(td) / "pyproject.toml"
+        cfg.write_text(stripped, encoding="utf-8")
+        out = _run(
+            ["mypy", "--config-file", str(cfg), "--exclude", GATE_EXCLUDE_RUFF, target]
+        ).stdout
+    match = re.search(r"Found (\d+) errors?", out)
+    return (int(match.group(1)) if match else 0), exempt
+
+
 def measure_mypy() -> dict:
     """`mypy --config-file pyproject.toml --exclude .workflows-lib src` -- the Gate's own command.
 
@@ -209,12 +238,20 @@ def measure_mypy() -> dict:
         codes[match.group(1)] = codes.get(match.group(1), 0) + 1
     match = re.search(r"Found (\d+) errors? in (\d+) files?", out)
     blocking = int(match.group(1)) if match else len(codes)
+    # BOTH NUMBERS, or this report stops being able to see its own subject. Once the per-module
+    # exempt list landed, the gate-facing count became 0 -- true, and useless for tracking the
+    # drain, because the findings behind the list vanished from the only report that counts them.
+    # So measure a SECOND time with the `ignore_errors` override stripped: that is the number the
+    # campaign is against, and `mypy_exempt_max` in .verify-floor.json is what stops it growing.
+    behind_ratchet, exempt_modules = _measure_behind_exempt_list(target)
     setup_abort = "errors prevented further checking" in out
     return {
         "check": "typecheck-mypy",
         "command": " ".join(cmd),
         "blocking": blocking,
         "blocking_unit": "errors",
+        "behind_exempt_list": behind_ratchet,
+        "exempt_modules": exempt_modules,
         # No `mypy --fix` exists. This 0 is a fact about the toolchain, not a judgement about
         # how hard the work is.
         "drainable": 0,
@@ -303,6 +340,22 @@ def render(report: dict) -> str:
             f"  {entry['check']:<16}{blocking:>22}{entry['drainable']:>12}   "
             f"{entry['drain'][:60]}{'  <-- ' + verdict if verdict else ''}"
         )
+    # The exempt list is the campaign's subject, so it gets a line of its own. Without it the
+    # table reads "typecheck-mypy 0 errors" and the work behind the list is invisible — the exact
+    # both-numbers rule this report exists to serve.
+    for entry in report["checks"]:
+        behind = entry.get("behind_exempt_list") or 0
+        if behind:
+            lines.append("")
+            lines.append(
+                f"  {entry['check']}: {entry['blocking']} at the gate, but {behind} finding(s) "
+                f"remain behind {entry.get('exempt_modules', 0)} per-module `ignore_errors` "
+                f"exemption(s)."
+            )
+            lines.append(
+                "    That list may only SHRINK — `mypy_exempt_max` in .verify-floor.json fails if "
+                "it grows. Drain it by typing a module and deleting its line."
+            )
     lines.append("")
     for entry in report["checks"]:
         if entry.get("by_code"):

src/adversarial.py

@@ -300,7 +300,7 @@ def aggregate_veto(
             )
         if (unexamined_min or 0) > 0:
             reasons.append(
-                f"at least {unexamined_min} of {int(findings_submitted)} submitted findings received no "
+                f"at least {unexamined_min} of {int(findings_submitted or 0)} submitted findings received no "
                 f"verdict (one verdict settles at most one finding, and verdicts are unattributed)"
             )
         out["inconclusive_reason"] = (

src/backlog.py

@@ -32,6 +32,7 @@
 import sys
 import time
 from pathlib import Path
+from typing import Any
 
 HANDOFF = Path(os.environ.get("HANDOFF_DIR", Path.home() / ".codex" / "handoff"))
 SENTINEL = HANDOFF / "lane-handoff.json"

src/cadence_registry.py

@@ -347,7 +347,9 @@ def inspect_cadence(
         elif success_ts is None:
             success_status = "missing"
         else:
-            success_status = "stale" if success_age > stale_after_s else "fresh"
+            # `success_ts is None` was ruled out above, so the age is a real int here; stating it
+            # keeps the comparison typed without changing the branch logic.
+            success_status = "stale" if int(success_age or 0) > stale_after_s else "fresh"
         try:
             failure_count = int(failure_path.read_text().strip()) if failure_ts is not None else 0
         except (OSError, ValueError):

src/capability_admission.py

@@ -46,6 +46,7 @@
 import os
 import pathlib
 import re
+from typing import Any
 
 import capabilities
 import env_prereq

src/capability_advisor.py

@@ -37,6 +37,7 @@
 import re
 import sys
 import time
+from typing import Any
 
 import capabilities
 import env_prereq

src/capability_targets.py

@@ -203,7 +203,8 @@ def _matches(binding: Mapping[str, Any], trigger: Mapping[str, Any]) -> bool:
     if kind == "role":
         role = roles.get_role(binding["context"]["role_name"])
         selector = role.selector or {}
-        value = trigger.get(selector.get("field"))
+        field = selector.get("field")
+        value = trigger.get(field) if isinstance(field, str) else None
         expected = selector.get("value")
         return (
             value == expected if selector.get("operator") == "equals" else value in (expected or [])

src/consumer_sync_shadow.py

@@ -237,24 +237,24 @@ def validate_consumer_sync_plan(raw: Any) -> dict[str, Any]:
         if not isinstance(removal, dict) or set(removal) != REMOVAL_FIELDS:
             reasons.append(f"invalid_removal_fields:{index}")
             continue
-        target = _safe_path(removal.get("target"))
-        if target is None:
+        removal_target = _safe_path(removal.get("target"))
+        if removal_target is None:
             reasons.append(f"unsafe_removal_target:{index}")
             continue
         fingerprint = str(removal.get("effect_fingerprint") or "")
         if not SHA256_RE.fullmatch(fingerprint):
             reasons.append(f"invalid_removal_effect_fingerprint:{index}")
-        if fingerprint != _stable_hash("consumer-sync-removal-effect", {"target": target}):
+        if fingerprint != _stable_hash("consumer-sync-removal-effect", {"target": removal_target}):
             reasons.append(f"removal_effect_identity_mismatch:{index}")
-        if target in target_owners:
-            reasons.append(f"duplicate_target:{target}")
-        target_owners[target] = f"removal:{index}"
+        if removal_target in target_owners:
+            reasons.append(f"duplicate_target:{removal_target}")
+        target_owners[removal_target] = f"removal:{index}"
         description = str(removal.get("description") or "").strip()
         if not description:
             reasons.append(f"invalid_removal_description:{index}")
         removals.append(
             {
-                "target": target,
+                "target": removal_target,
                 "description": description,
                 "effect_fingerprint": fingerprint,
             }

src/epic_lane.py

@@ -198,7 +198,7 @@ def validate_plan(plan: dict[str, Any]) -> list[str]:
         if not _is_nonempty_string(task_id):
             errors.append(f"{path}.id must be a non-empty string")
         else:
-            normalized_id = task_id.strip()
+            normalized_id = str(task_id).strip()
             if normalized_id in seen:
                 errors.append(f"{path}.id duplicates {normalized_id!r}")
             seen.add(normalized_id)

src/evidence_acquisition.py

@@ -37,6 +37,7 @@
 import json
 import os
 import time
+from collections.abc import Mapping
 from pathlib import Path
 from typing import Any
 
@@ -49,7 +50,7 @@
 LIVE_FLAG = "ORCH_EVIDENCE_ACQUISITION"
 
 
-def live_enabled(env: dict | None = None) -> bool:
+def live_enabled(env: Mapping[str, str] | None = None) -> bool:
     """True only when the documented default-off switch is explicitly set to 1."""
     env = os.environ if env is None else env
     return str(env.get(LIVE_FLAG, "")).strip() == "1"

src/execution_profiles.py

@@ -385,7 +385,7 @@ def select_profile(
         values = scores or {}
         selected = min(eligible, key=lambda pid: (-float(values.get(pid, 0.0)), pid))
         probability = 1.0
-    body = {
+    body: dict[str, Any] = {
         "schema_version": 2,
         "task_type": task_type,
         "target": target,

src/exploration_evidence_plan.py

@@ -222,8 +222,8 @@ def _candidate_task_types(
     candidates.sort(
         key=lambda row: (
             not row["recommended"],
-            -row["opener_backlog_items"],
-            -row["recent_outcome_rows"],
+            -int(row["opener_backlog_items"]),
+            -int(row["recent_outcome_rows"]),
             row["task_type"],
         )
     )

src/exploration_review.py

@@ -41,6 +41,8 @@ def _route_agents(task_type: str, route_table: dict | None = None) -> list[str]:
     spec = (route_table or router.ROUTE_TABLE).get(task_type) or {}
     out: list[str] = []
     for row in spec.get("agents") or []:
+        if not isinstance(row, dict):
+            continue
         agent = row.get("agent")
         if not agent or agent in out or agent in router.BACKUP_AGENTS:
             continue

src/features.py

@@ -18,6 +18,7 @@
 import os
 import time
 from pathlib import Path
+from typing import Any
 
 ORCH = Path(__file__).resolve().parent
 REG = Path(os.environ.get("ORCH_FEATURES_PATH", ORCH / "experiments" / "features.json"))

src/gh_capacity.py

@@ -35,6 +35,7 @@
 import sys
 import time
 from pathlib import Path
+from typing import Any
 
 HANDOFF = Path(os.environ.get("HANDOFF_DIR", Path.home() / ".codex" / "handoff"))
 GH_LEDGER = (
@@ -200,7 +201,9 @@ def gh_run(args: list[str], *, resource: str = "core", timeout_s: int = 30, runn
     except Exception:
         return 1, None
     headers, body = _split_headers_body(getattr(r, "stdout", "") or "")
-    _append_ledger([_ratelimit_row_from_headers(headers, fallback_resource=resource)])
+    row = _ratelimit_row_from_headers(headers, fallback_resource=resource)
+    if row is not None:
+        _append_ledger([row])
     parsed = None
     if body:
         try:
@@ -301,7 +304,11 @@ def throttle_if_enabled(resource: str, **kw) -> dict | None:
 def build(*, runner=subprocess.run) -> dict:
     """Probe + snapshot the tracked resources (what `gh_capacity.py` with no args writes/prints)."""
     resources = probe(runner=runner)
-    out = {"generated_at": int(time.time()), "probe_ok": resources is not None, "resources": {}}
+    out: dict[str, Any] = {
+        "generated_at": int(time.time()),
+        "probe_ok": resources is not None,
+        "resources": {},
+    }
     for name in TRACKED:
         st, meta = state(name)
         out["resources"][name] = {"state": st, **meta}

src/human_calibration.py

@@ -346,7 +346,7 @@ def compute(
     raw_mae = None
     if pairs:
         raw_mae = sum(
-            abs(float(row["proxy_score"]) - float(row["human_score"])) for row in pairs
+            abs(float(str(row["proxy_score"])) - float(str(row["human_score"]))) for row in pairs
         ) / len(pairs)
     return {
         "generated_at": generated_at or int(time.time()),

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 = linked_cache.get((repo, pr_number)) or []
+        issue_numbers: list[int] = linked_cache.get((repo, pr_number)) or []
         if not isinstance(issue_numbers, list):
             continue
         for issue_number in issue_numbers:

src/judge_reliability.py

@@ -14,6 +14,7 @@
 import json
 import time
 from collections import defaultdict
+from typing import Any
 
 import feedback
 

src/keepalive_shadow.py

@@ -35,6 +35,7 @@
 import subprocess
 import sys
 from pathlib import Path
+from typing import Any
 
 import redirect_policy
 
@@ -161,7 +162,7 @@ def synthesize_report(signals: dict) -> dict:
 
     state_raw = str(signals.get("pr_state") or "open").lower()
     last_changes = bool(signals.get("last_has_changes"))
-    drift = {"severity": "none", "findings": []}
+    drift: dict[str, Any] = {"severity": "none", "findings": []}
 
     if state_raw in ("closed", "merged"):
         state = "exited"

src/ledger_reconcile.py

@@ -478,7 +478,7 @@ def record_completion(
             # caller left EMPTY -- an explicitly supplied resolved_model always wins, because the
             # caller may have provenance this reader cannot see.
             probed = adapters.cli_reported_model(
-                agent, target[len("offload:") :], started_ts=started_ts, log_file=log_file
+                agent, str(target)[len("offload:") :], started_ts=started_ts, log_file=log_file
             )
             probe_reason = probed.get("reason")
             if probed.get("model"):