Check the four integration nodes under strict mypy

mqtt, http, influx and delay carried the prototype's annotations and were
excluded wholesale. Annotating them leaves nothing to exclude: what the two
untyped dependencies need is an implicit-reexport allowance for
influxdb_client and the import-site ignore croniter already gets elsewhere.

InfluxDbNode.inject now takes the durable flag its supertype passes, and the
cron loop reads its stop event once instead of dereferencing a field that
stop_cron() may have cleared.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01H7LwYgJfpkbLCTeiAf8U4A

backend/app/flow/nodes/delay.py

@@ -106,7 +106,7 @@ class DelayNode(Node):
         mapping: dict[str, str] = {}
         cron: str | None = None
 
-    def __init__(self, params: dict[str, Any] | None = None, **kwargs):
+    def __init__(self, params: dict[str, Any] | None = None, **kwargs: Any) -> None:
         cfg = self.Params.model_validate(params or {})
         self.delay = cfg.delay
         self.interval = cfg.interval
@@ -114,8 +114,8 @@ class DelayNode(Node):
         self.cron_expr = cfg.cron
         self.ts = 0.0
         self.last_input: dict[str, Any] = {}
-        self._cron_task = None
+        self._cron_task: asyncio.Task[None] | None = None
-        self._stop_cron = None
+        self._stop_cron: asyncio.Event | None = None
 
         super().__init__(f=self._f, params=params, **kwargs)
 
@@ -128,7 +128,7 @@ class DelayNode(Node):
             for i, o in zip(self.input_ports, self.output_ports, strict=False)
         ]
 
-    def _f(self, params, **kwargs):
+    def _f(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None:
         """Forward messages with optional delay, rate-limiting, and alarm."""
 
         logger.info("[%s] Received %s", self.name, kwargs)
@@ -220,12 +220,12 @@ class DelayNode(Node):
         self._stop_cron = None
         logger.info("Stopped cron for node '%s'", self.name)
 
-    async def _cron_loop(self):
+    async def _cron_loop(self) -> None:
         """Background loop that sleeps until the next cron tick and triggers."""
         from datetime import datetime
 
         try:
-            from croniter import croniter
+            from croniter import croniter  # type: ignore[import-untyped]
         except ImportError:
             logger.error(
                 "croniter package is required for cron scheduling. "
@@ -243,7 +243,9 @@ class DelayNode(Node):
 
         cron = croniter(self.cron_expr, datetime.now())
 
-        while not (self._stop_cron and self._stop_cron.is_set()):
+        # Walrus so the body can use the event without re-reading a field
+        # stop_cron() may have cleared in the meantime.
+        while (stop := self._stop_cron) is not None and not stop.is_set():
             try:
                 # Compute seconds until next tick
                 next_dt = cron.get_next(datetime)
@@ -259,7 +261,7 @@ class DelayNode(Node):
 
                 # Sleep until next tick (wake up on stop signal)
                 try:
-                    await asyncio.wait_for(self._stop_cron.wait(), timeout=wait_seconds)
+                    await asyncio.wait_for(stop.wait(), timeout=wait_seconds)
                     # If we get here, stop was requested
                     break
                 except asyncio.TimeoutError:
@@ -283,7 +285,7 @@ class DelayNode(Node):
                 )
                 raise
 
-    async def _trigger_cron(self):
+    async def _trigger_cron(self) -> None:
         """Emit data into the pipeline on a cron tick."""
         if self._pipeline is None:
             logger.warning(

backend/app/flow/nodes/http.py

@@ -204,7 +204,7 @@ class HttpNode(Node):
         )
 
     @staticmethod
-    def _noop_trigger(params: dict, **kwargs) -> dict | None:
+    def _noop_trigger(params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None:
         """
         No-op function for trigger mode nodes.
 
@@ -213,7 +213,9 @@ class HttpNode(Node):
         """
         return None
 
-    def _sender_handler(self, params: dict, **kwargs) -> dict | None:
+    def _sender_handler(
+        self, params: dict[str, Any], **kwargs: Any
+    ) -> dict[str, Any] | None:
         """
         Send HTTP request with pipeline data (sender mode).
 
@@ -329,6 +331,7 @@ class HttpNode(Node):
 
             try:
                 # Parse request data
+                data: dict[str, Any]
                 if self.method == "GET":
                     data = dict(request.query_params)
                 else:  # POST

backend/app/flow/nodes/influx.py

@@ -9,7 +9,7 @@ from typing import Any
 from pydantic import BaseModel, ConfigDict, Field
 
 from app.flow.messages import MessageSpec
-from app.flow.nodes.base import Node
+from app.flow.nodes.base import Node, NodeResult
 
 logger = logging.getLogger(__name__)
 
@@ -200,7 +200,7 @@ class InfluxDbNode(Node):
             name=name,
         )
 
-    def _handler(self, params: dict, **kwargs) -> dict | None:
+    def _handler(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None:
         """
         Handle incoming data - write to InfluxDB and optionally query.
 
@@ -224,7 +224,7 @@ class InfluxDbNode(Node):
 
         return None
 
-    def _write_points(self, data: dict) -> None:
+    def _write_points(self, data: dict[str, Any]) -> None:
         """
         Write data points to InfluxDB using configuration from ``writes``.
 
@@ -327,7 +327,7 @@ class InfluxDbNode(Node):
             logger.error("InfluxDB write error in node '%s': %s", self.name, e)
             raise
 
-    def _query_data(self) -> dict:
+    def _query_data(self) -> dict[str, Any]:
         """
         Query data from InfluxDB based on provides configuration.
 
@@ -399,7 +399,7 @@ class InfluxDbNode(Node):
         self,
         measurement: str,
         field: str,
-        tags: dict,
+        tags: dict[str, Any],
         time_range: str,
         aggregation: str,
     ) -> str:
@@ -452,7 +452,7 @@ class InfluxDbNode(Node):
 
         return "\n".join(query_parts)
 
-    def _extract_query_result(self, tables, spec: MessageSpec) -> Any:
+    def _extract_query_result(self, tables: Any, spec: MessageSpec) -> Any:
         """
         Extract a single value from query result tables.
 
@@ -473,7 +473,9 @@ class InfluxDbNode(Node):
 
         return None
 
-    def inject(self, outputs: dict | None = None) -> dict | None:
+    def inject(
+        self, outputs: dict[str, Any] | None = None, durable: bool | None = None
+    ) -> NodeResult:
         """
         Inject queried data into the pipeline.
 
@@ -495,4 +497,4 @@ class InfluxDbNode(Node):
                 return None
             outputs = self._query_data()
 
-        return self._pipeline.trigger(self, self._to_messages(outputs))
+        return self._pipeline.trigger(self, self._to_messages(outputs), durable=durable)

backend/app/flow/nodes/mqtt.py

@@ -190,11 +190,11 @@ class MqttNode(Node):
         self.keepalive = cfg.keepalive
 
         # Runtime state
-        self._subscription_task: asyncio.Task | None = None
+        self._subscription_task: asyncio.Task[None] | None = None
         self._mqtt_client = None
         self._stop_event: asyncio.Event | None = None
-        self._publish_queue: asyncio.Queue[dict] | None = None
+        self._publish_queue: asyncio.Queue[dict[str, Any]] | None = None
-        self._publisher_task: asyncio.Task | None = None
+        self._publisher_task: asyncio.Task[None] | None = None
         self._loop: asyncio.AbstractEventLoop | None = None
 
         # Set default name based on mode and topics
@@ -228,7 +228,9 @@ class MqttNode(Node):
         )
 
     @staticmethod
-    def _noop_subscriber(params: dict, **kwargs) -> dict | None:
+    def _noop_subscriber(
+        params: dict[str, Any], **kwargs: Any
+    ) -> dict[str, Any] | None:
         """
         No-op function for subscriber mode nodes.
 
@@ -245,7 +247,9 @@ class MqttNode(Node):
         """
         return None
 
-    def _publisher_handler(self, params: dict, **kwargs) -> dict | None:
+    def _publisher_handler(
+        self, params: dict[str, Any], **kwargs: Any
+    ) -> dict[str, Any] | None:
         """
         Publish pipeline data to MQTT topic (publisher mode).
 
@@ -277,7 +281,9 @@ class MqttNode(Node):
             )
         return None
 
-    def _enqueue(self, queue: asyncio.Queue[dict], data: dict) -> None:
+    def _enqueue(
+        self, queue: asyncio.Queue[dict[str, Any]], data: dict[str, Any]
+    ) -> None:
         """Queue a payload, dropping the oldest when the broker cannot keep up."""
         if queue.full():
             try:
@@ -317,7 +323,7 @@ class MqttNode(Node):
                     self.report_health("down", str(exc))
                     raise
 
-    async def _publish_once(self, data: dict) -> None:
+    async def _publish_once(self, data: dict[str, Any]) -> None:
         """Connect, publish, disconnect — the unstarted node's path."""
         import aiomqtt
 
@@ -331,7 +337,7 @@ class MqttNode(Node):
         ) as client:
             await self._publish_with(client, data)
 
-    async def _publish_with(self, client: Any, data: dict) -> None:
+    async def _publish_with(self, client: Any, data: dict[str, Any]) -> None:
         """
         Publish messages to their mapped MQTT topics.
 

backend/pyproject.toml

@@ -46,18 +46,17 @@ build-backend = "hatchling.build"
 [tool.mypy]
 strict = true
 exclude = ["venv", ".venv", "alembic"]
+# influxdb_client's Point builder carries no annotations; calling it is not a
+# reason to stop checking the caller (app/flow/nodes/influx.py).
+untyped_calls_exclude = ["influxdb_client"]
 
-# The integration nodes still carry the prototype's annotations, and croniter and
+# Every module is checked strictly. influxdb_client is typed but re-exports its
-# influxdb_client ship no stubs. Shrinking one module at a time as each integration
+# names implicitly, which strict mode refuses to follow. (croniter, the other
-# is revisited; the rest of the package, and everything around it, is checked strictly.
+# untyped dependency, is ignored at its two import sites.)
 [[tool.mypy.overrides]]
-module = [
+module = ["influxdb_client"]
-    "app.flow.nodes.delay",
+implicit_reexport = true
-    "app.flow.nodes.http",
+
-    "app.flow.nodes.influx",
-    "app.flow.nodes.mqtt",
-]
-ignore_errors = true
 
 [tool.ruff]
 target-version = "py310"