fix(typing): drain 6 more modules (exempt list 26 -> 20, 79 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. 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
+# 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
 # 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.
@@ -127,21 +127,16 @@ mypy_path = ["src", "tests"]
 [[tool.mypy.overrides]]
 module = [
     "capabilities",
-    "capability_activation_audit",
     "capability_advisor",
     "capability_compiler",
     "capability_outcome_bridge",
     "capability_propensity",
     "capability_recurrence_check",
     "codemod_lane",
-    "consumer_sync_artifact_ingest",
     "dispatcher",
-    "durability_sweep",
     "exp_abcd",
     "feedback",
-    "issue_quality",
     "issue_readiness",
-    "keepalive_evidence",
     "keepalive_outcomes",
     "observability_dashboard",
     "redirect_sweep",
@@ -151,6 +146,5 @@ module = [
     "runtime_ac",
     "runtime_ac_gate",
     "switch_review",
-    "tick",
 ]
 ignore_errors = true

src/adversarial.py

@@ -16,6 +16,7 @@
 import json
 import re
 import sys
+from collections.abc import Mapping
 
 import dispatcher
 
@@ -110,14 +111,14 @@ def is_high_stakes(item: dict) -> bool:
     return high_stakes_reason(item) is not None
 
 
-def reviewers_from_env(env: dict | None = None) -> list[str]:
+def reviewers_from_env(env: Mapping[str, str] | None = None) -> list[str]:
     env = env or {}
     raw = env.get("ORCH_ADVERSARIAL_REVIEWERS", "")
     reviewers = [part.strip() for part in raw.split(",") if part.strip()]
     return reviewers or list(DEFAULT_REVIEWERS)
 
 
-def review_enabled(env: dict | None = None) -> bool:
+def review_enabled(env: Mapping[str, str] | None = None) -> bool:
     env = env or {}
     return env.get("ORCH_RUN_ADVERSARIAL_REVIEW") == "1"
 

src/capability_activation_audit.py

@@ -48,6 +48,7 @@
 import sys
 import time
 from pathlib import Path
+from typing import Any
 
 import backlog
 import capabilities
@@ -968,7 +969,7 @@ def heartbeat_env_gate(*, here: Path | None = None) -> dict:
     # Shell drivers live at the CHECKOUT root, not beside the modules. `here` is still honoured so
     # the selftest can point this at a synthetic tree.
     root = here or paths.checkout_root(HERE)
-    out = {
+    out: dict[str, Any] = {
         "flag": HEARTBEAT_ENV_FLAG,
         "anchor": HEARTBEAT_EXPORT_ANCHOR,
         "drivers": {},
@@ -1222,7 +1223,7 @@ def audit_capability(
                 defects.append("no_prompt_template")
                 notes.append(f"no PROMPT_TEMPLATES[{value!r}]")
             cov = label_coverage(value, label_index)
-            row.setdefault("label_coverage", {})[value] = cov
+            row.setdefault("label_coverage", {})[value] = cov  # type: ignore[index]
             if cov.get("repos_with") == 0:
                 defects.append("label_absent_from_fleet")
                 notes.append(f"no repo carries a label producing {value!r}")
@@ -2049,8 +2050,8 @@ def _selftest() -> None:
 
             # An UNREADABLE ledger must degrade to a visible note, never to an exception that
             # replaces the real assertion with a complaint about the diagnostic.
-            broken = absent_entrypoint_note(sorted(rows), path=droot / "not-a-ledger-dir")
-            assert broken == "" or "diagnostic unavailable" in broken, broken
+            broken_note = absent_entrypoint_note(sorted(rows), path=droot / "not-a-ledger-dir")
+            assert broken_note == "" or "diagnostic unavailable" in broken_note, broken_note
         finally:
             globals()["HERE"] = saved_here
     # (b) a mention inside a shell COMMENT is not a caller — matching it reported a

src/consumer_sync_artifact_ingest.py

@@ -1713,7 +1713,7 @@ def main(argv: list[str] | None = None) -> int:
         # throttled call: at LOW budget gh_capacity paces up to 10s each.
         blob_reader = BlobReader(state.setdefault("blob_digests", {}))
         needs_processing_lower = {repo.lower() for repo in needs_processing}
-        report_repos = {
+        report_repos: dict[str, dict[str, Any]] = {
             repo.lower(): {"status": "already_recorded"}
             for repo in deduped_cohort
             if repo.lower() not in needs_processing_lower

src/durability_sweep.py

@@ -18,6 +18,7 @@
 import subprocess
 import sys
 import time
+from typing import Any
 from urllib.parse import quote
 
 import feedback
@@ -291,7 +292,7 @@ def _live_revert_status(pr: dict, revert_cache: dict | None = None) -> tuple[boo
     if not repo:
         return None, "missing repo"
 
-    pr_check = (None, "missing PR number")
+    pr_check: tuple[bool | None, str] = (None, "missing PR number")
     if pr_number is not None:
         pr_check = _revert_pr_status(repo, int(pr_number), revert_cache=revert_cache)
         if pr_check[0] is True:
@@ -449,7 +450,7 @@ def sweep_durability(
     _now: int | None = None,
 ) -> dict:
     """Patch old merged+pending outcomes when their durability can be resolved with confidence."""
-    summary = {
+    summary: dict[str, Any] = {
         "checked": 0,
         "durable": 0,
         "reverted": 0,
@@ -590,7 +591,7 @@ def _verdict_for(files):
     assert _verdict_for([{"path": ".agents/x.yml"}, {"path": "app.py"}])["durability"] == "durable"
 
     # FAIL-SAFE: unknowable delivery must never manufacture a failure.
-    for unknown in (None, []):
+    for unknown in (None, []):  # type: ignore[var-annotated]  # deliberate mixed-type probe
         got = _verdict_for(unknown)
         assert got["durability"] == "durable", (unknown, got)
 

src/issue_quality.py

@@ -19,6 +19,7 @@
 import time
 from dataclasses import dataclass
 from pathlib import Path
+from typing import cast
 from urllib.parse import quote
 
 import feedback
@@ -280,7 +281,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(linked_cache.get((repo, pr_number)) or [])
+        issue_numbers = cast(list, linked_cache.get((repo, pr_number)) or [])
         if not isinstance(issue_numbers, list):
             continue
         for issue_number in issue_numbers:
@@ -291,7 +292,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 = list(linked_cache.get((repo, pr_number)) or [])
+        issue_numbers = cast(list, linked_cache.get((repo, pr_number)) or [])
         if not issue_numbers:
             skipped.append(
                 {

src/keepalive_evidence.py

@@ -18,7 +18,7 @@
 import tempfile
 import time
 from pathlib import Path
-from typing import Any
+from typing import Any, cast
 from urllib.parse import quote
 
 import feedback
@@ -374,7 +374,7 @@ def _process_signals(
     signals = []
     for work_type in sorted(grouped):
         bucket = grouped[work_type]
-        prs = list(bucket["prs"] or [])
+        prs = cast(list, bucket["prs"] or [])
         count = len(prs)
         signals.append(
             {

src/repo_knowledge.py

@@ -16,6 +16,7 @@
 import subprocess
 import sys
 from pathlib import Path
+from typing import Any
 
 import feedback
 import provision
@@ -189,7 +190,7 @@
     },
 }
 
-SEED = {
+SEED: dict[str, Any] = {
     "schema_version": SEED_SCHEMA_VERSION,
     "repos": {
         "stranske/Workflows": {
@@ -1915,7 +1916,7 @@ def approve_suggestion(
         "already_present": already,
     }
     if apply and not already:
-        item = {"text": text}
+        item: dict[str, Any] = {"text": text}
         if suggestion.get("task_type"):
             item["task_types"] = [suggestion["task_type"]]
         entry.setdefault(chosen, []).append(item)
@@ -2595,13 +2596,13 @@ def main(argv: list[str]) -> int:
     if "--validate-agents-md" in argv:
         idx = argv.index("--validate-agents-md")
         repo_path = Path(argv[idx + 1]) if len(argv) > idx + 1 else Path(".")
-        repo = argv[argv.index("--repo") + 1] if "--repo" in argv else None
+        repo_arg = argv[argv.index("--repo") + 1] if "--repo" in argv else None
         max_lines = (
             int(argv[argv.index("--max-lines") + 1])
             if "--max-lines" in argv
             else AGENTS_EXPORT_MAX_LINES
         )
-        result = validate_agents_md_export(repo_path, repo=repo, max_lines=max_lines)
+        result = validate_agents_md_export(repo_path, repo_arg=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:
@@ -2657,7 +2658,7 @@ def main(argv: list[str]) -> int:
     if "--suggest-from-docs" in argv:
         idx = argv.index("--suggest-from-docs")
         repo_path = Path(argv[idx + 1]) if len(argv) > idx + 1 else Path(".")
-        repo = argv[argv.index("--repo") + 1] if "--repo" in argv else None
+        repo_arg = argv[argv.index("--repo") + 1] if "--repo" in argv else None
         max_per_repo = int(argv[argv.index("--max") + 1]) if "--max" in argv else 10
         sections = [
             argv[i + 1] for i, arg in enumerate(argv) if arg == "--section" and i + 1 < len(argv)
@@ -2666,7 +2667,7 @@ def main(argv: list[str]) -> int:
             json.dumps(
                 suggest_from_docs(
                     repo_path,
-                    repo=repo,
+                    repo_arg=repo_arg,
                     max_per_repo=max_per_repo,
                     include_root_docs="--include-root-docs" in argv,
                     sections=sections,

src/roles.py

@@ -29,7 +29,7 @@
 import re
 import sys
 import time
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
@@ -1453,12 +1453,12 @@ def select_role_activation(
     return out
 
 
-def _shadow_gate(env: dict | None) -> bool:
+def _shadow_gate(env: Mapping[str, str] | None) -> bool:
     source = os.environ if env is None else env
     return str(source.get("ORCH_ROLE_SHADOW", "0")).strip() == "1"
 
 
-def _role_cap(env: dict | None, role_name: str) -> int:
+def _role_cap(env: Mapping[str, str] | None, role_name: str) -> int:
     source = os.environ if env is None else env
     specific = source.get(f"ORCH_{role_name.upper()}_ROLE_MAX_PER_CYCLE")
     raw = specific if specific is not None else source.get("ORCH_ROLE_MAX_PER_CYCLE", "1")
@@ -1504,7 +1504,7 @@ def activate_dispatch_roles(
     baseline_prompt: str,
     *,
     cwd: str,
-    env: dict | None = None,
+    env: Mapping[str, str] | None = None,
     cap: dict | None = None,
     dry_run: bool = False,
     prompt_runner=None,
@@ -1617,7 +1617,7 @@ def activate_tick_triage(
     items: list[dict],
     cap: dict,
     *,
-    env: dict | None = None,
+    env: Mapping[str, str] | None = None,
     dry_run: bool = False,
     runner=None,
 ) -> dict:
@@ -1714,7 +1714,7 @@ def activate_adjudicator_disagreement(
     review_status: dict | None,
     cap: dict,
     *,
-    env: dict | None = None,
+    env: Mapping[str, str] | None = None,
     dry_run: bool = False,
     runner=None,
 ) -> dict:

src/tick.py

@@ -360,7 +360,7 @@ def research_tick(
                 metadata={"target": target, "task_type": task_type},
             )
             prepare_result = prepare(
-                repo,
+                str(repo),
                 str(spec_file),
                 exp_id,
                 research_v2_arms(arms, item.get("profiles")),