chore(mypy): type services/runtime/misc (42 -> 27 in src/) (#1280)

Part of #1133. Typing-only changes, no runtime behavior change:

- pyproject mypy overrides: add `regex` / `regex.*` (third-party, no stubs)
  -> clears notification_matcher import-untyped
- numpy_semantic: TYPE_CHECKING-import np / NearestNeighbors and annotate
  `_matrix` / `_index` fields (they are lazy-imported at runtime)
- scheduler/service: rename loop-bound `all_active` to `active_pipelines`
  in the pipeline branch so it is not re-typed from list[SearchQuery]
- runtime/worker: annotate `bot_payload` as dict[str, Any] (was inferred
  bool-only from the initial {"configured": False})
- models.PipelineGraph.from_json: narrow param to str | dict[str, Any]
  and assert after json.loads so the .get() calls type-check
- config: targeted type: ignore[call-arg] on DatabaseConfig() — pydantic
  Field(gt=0) defaults are mis-seen as required by the mypy plugin
- agent/tools/_registry: replace direct attribute access on duck-typed
  `object | None` client_pool with getattr/cast in the three pool helpers
  (clients / connected_phones / _pool_reports_connections)


Claude-Session: https://claude.ai/code/session_01G3ExCZyXkbUpTkRRrQrFA2

Co-authored-by: axisrow <axisrow@users.noreply.github.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: axisrow

pyproject.toml

@@ -174,6 +174,8 @@ module = [
     "langdetect",
     "openpyxl",
     "openpyxl.*",
+    "regex",
+    "regex.*",
     "sklearn.*",
     "telethon",
     "telethon.*",

src/agent/tools/_registry.py

@@ -259,7 +259,9 @@ def connected_phones_from_pool(client_pool: object | None) -> set[str]:
     if isinstance(instance_attrs, dict) and "connected_phones" in instance_attrs:
         connected_phones = instance_attrs["connected_phones"]
     elif callable(getattr(type(client_pool), "connected_phones", None)):
-        connected_phones = client_pool.connected_phones
+        # type(client_pool).connected_phones is callable (a property/method); access it,
+        # giving mypy a typed handle without weakening the duck-typed signature.
+        connected_phones = cast(Any, client_pool).connected_phones
 
     if callable(connected_phones):
         try:
@@ -270,7 +272,7 @@ def connected_phones_from_pool(client_pool: object | None) -> set[str]:
             return {str(phone) for phone in phones}
 
     try:
-        clients = client_pool.clients
+        clients = getattr(client_pool, "clients", {})
     except Exception:
         clients = {}
     if isinstance(clients, dict):
@@ -321,7 +323,7 @@ def _pool_reports_connections(client_pool: object | None) -> bool:
     if callable(getattr(type(client_pool), "connected_phones", None)):
         return True
     try:
-        return isinstance(client_pool.clients, dict)
+        return isinstance(getattr(client_pool, "clients", None), dict)
     except Exception:
         return False
 

src/config.py

@@ -191,7 +191,7 @@ class AppConfig(BaseModel):
     web: WebConfig = WebConfig()
     scheduler: SchedulerConfig = SchedulerConfig()
     notifications: NotificationsConfig = NotificationsConfig()
-    database: DatabaseConfig = DatabaseConfig()
+    database: DatabaseConfig = DatabaseConfig()  # type: ignore[call-arg]  # Field(gt=0) defaults are mis-seen as required
     llm: LLMConfig = LLMConfig()
     agent: AgentConfig = AgentConfig()
     security: SecurityConfig = SecurityConfig()

src/models.py

@@ -667,12 +667,14 @@ def to_json(self) -> str:
         )
 
     @classmethod
-    def from_json(cls, data: str | dict) -> "PipelineGraph":
+    def from_json(cls, data: str | dict[str, Any]) -> "PipelineGraph":
         """Собрать граф из JSON-строки или уже разобранного dict (валидирует узлы
         и рёбра через их модели)."""
         import json
         if isinstance(data, str):
             data = json.loads(data)
+        # json.loads returns Any; the str branch is gone, so narrow to the parsed dict.
+        assert isinstance(data, dict)
         nodes = [PipelineNode.model_validate(n) for n in data.get("nodes", [])]
         edges = [PipelineEdge.model_validate(e) for e in data.get("edges", [])]
         return cls(nodes=nodes, edges=edges)

src/runtime/worker.py

@@ -6,6 +6,7 @@
 from dataclasses import asdict
 from datetime import datetime, timezone
 from inspect import isawaitable
+from typing import Any
 
 from src.config import AppConfig
 from src.database import DatabaseBusyError
@@ -251,7 +252,7 @@ async def _publish_collection_queue_status_snapshot(container, now: datetime) ->
 
 async def _notification_target_status_payload(container, stop_event: asyncio.Event | None) -> dict:
     target_status = await container.notification_target_service.describe_target()
-    bot_payload = {"configured": False}
+    bot_payload: dict[str, Any] = {"configured": False}
     if target_status.state == "available":
         try:
             bot = await NotificationService(

src/scheduler/service.py

@@ -549,8 +549,8 @@ async def get_potential_jobs(self) -> list[dict]:
                 logger.exception("Error fetching search queries for potential jobs")
         if self._pipeline_bundle:
             try:
-                all_active = await self._pipeline_bundle.get_all(active_only=True)
-                for p in all_active:
+                active_pipelines = await self._pipeline_bundle.get_all(active_only=True)
+                for p in active_pipelines:
                     if p.id is not None and p.is_active:
                         # Only content_generate_ is a periodic job now (#835/2). Do NOT advertise
                         # pipeline_run_ as a togglable potential job: sync_pipeline_jobs always

src/search/numpy_semantic.py

@@ -6,7 +6,8 @@
 logger = logging.getLogger(__name__)
 
 if TYPE_CHECKING:
-    pass
+    import numpy as np
+    from sklearn.neighbors import NearestNeighbors
 
 
 class NumpySemanticIndex:
@@ -19,8 +20,8 @@ class NumpySemanticIndex:
 
     def __init__(self) -> None:
         self._ids: list[int] = []
-        self._matrix = None  # numpy ndarray, shape (N, dims), lazy import
-        self._index = None  # sklearn.neighbors.NearestNeighbors, lazy import
+        self._matrix: np.ndarray | None = None  # shape (N, dims), lazy import
+        self._index: NearestNeighbors | None = None  # lazy import
 
     def load(self, embeddings: list[tuple[int, list[float]]]) -> None:
         """Load pre-computed embeddings.  ``embeddings`` is a list of (message_id, vector)."""