Keep flow state across a restart, and tighten the flow typing

The engine fell back to MemoryState because there was no Redis to talk
to, so every value was lost on restart. Adds a redis service with
append-only persistence, reachable only from the backend, and points
REDIS_HOST at it.

state.py and pipeline.py leave the strict-mypy exclusion. redis-py types
its calls as sync/async unions; the casts narrow that at the points of
use rather than switching to redis.asyncio, which would make a
deliberately synchronous backend asynchronous. nodes.py stays excluded.

The .env.example CORS entry belongs with the CI work that follows.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KkmeRiyeYmVZqJVwuyHq9o

.env.example

@@ -15,7 +15,9 @@ PROJECT_NAME="Fluksio"
 STACK_NAME=fluksio-app
 
 # Backend
-BACKEND_CORS_ORIGINS="http://localhost,http://localhost:5173,http://app.localhost,https://localhost,https://localhost:5173"
+# `http://frontend` is the compose-network origin the Playwright job loads the
+# SPA from (docker/compose.ci.yml); without it every spec fails CORS at login.
+BACKEND_CORS_ORIGINS="http://localhost,http://localhost:5173,http://app.localhost,https://localhost,https://localhost:5173,http://frontend"
 SECRET_KEY=changethis
 FIRST_SUPERUSER=admin@fluksio.com
 FIRST_SUPERUSER_PASSWORD=changethis
@@ -36,6 +38,11 @@ POSTGRES_DB=app
 POSTGRES_USER=postgres
 POSTGRES_PASSWORD=changethis
 
+# Redis — flow engine state. Empty keeps the state in memory; the compose stack
+# points the backend at its own `redis` service.
+REDIS_HOST=
+REDIS_PORT=6379
+
 SENTRY_DSN=
 
 # Docker registry images

backend/app/flow/pipeline.py

@@ -32,6 +32,7 @@ class ValidationIssue(BaseModel):
         "unconnected_input",
         "missing_initial_value",
         "node_error",
+        "unauthenticated_hook",
     ]
     message: str
     flow: str = ""
@@ -243,7 +244,7 @@ class Pipeline:
     def _timestamp_key(self, msg_name: str) -> str:
         return f"__ts__:{msg_name}"
 
-    def _increment_message_versions(self, outputs: dict) -> None:
+    def _increment_message_versions(self, outputs: dict[str, Any]) -> None:
         for msg_name in outputs:
             self._state.increment(self._version_key(msg_name))
 
@@ -305,7 +306,7 @@ class Pipeline:
         if self._events is not None:
             self._events.publish(event)
 
-    def _execute_node(self, node: Node, state: StateBackend) -> dict | None:
+    def _execute_node(self, node: Node, state: StateBackend) -> dict[str, Any] | None:
         """Run one node and record its outputs. Never raises."""
         started = time.perf_counter()
         try:
@@ -378,7 +379,7 @@ class Pipeline:
 
         submitted: set[Node] = set()
         skipped: set[Node] = set()
-        node_futures: dict[Node, Future] = {}
+        node_futures: dict[Node, Future[dict[str, Any] | None]] = {}
 
         def is_ready(n: Node) -> bool:
             if in_degree[n] != 0:
@@ -416,7 +417,7 @@ class Pipeline:
         return state
 
     def run(
-        self, inputs: dict | None = None, nodes: set[Node] | None = None
+        self, inputs: dict[str, Any] | None = None, nodes: set[Node] | None = None
     ) -> StateBackend:
         """Execute the graph (or one flow's nodes) against the shared state."""
         if inputs:
@@ -426,7 +427,7 @@ class Pipeline:
 
         return self._execute_parallel(nodes, self._state, check_ready=False)
 
-    def trigger(self, node: Node, outputs: dict | None) -> StateBackend:
+    def trigger(self, node: Node, outputs: dict[str, Any] | None) -> StateBackend:
         """Publish a node's outputs and run everything downstream of it."""
         state = self._state
 

backend/app/flow/state.py

@@ -12,7 +12,7 @@ from abc import ABC, abstractmethod
 from collections.abc import Iterator
 from contextlib import contextmanager
 from threading import RLock
-from typing import Any
+from typing import Any, cast
 
 import redis
 
@@ -90,7 +90,7 @@ class StateBackend(ABC):
 
     @abstractmethod
     @contextmanager
-    def lock(self, timeout: float = 10.0):
+    def lock(self, timeout: float = 10.0) -> Iterator[None]:
         """
         Acquire a lock for thread-safe operations.
 
@@ -211,7 +211,7 @@ class MemoryState(StateBackend):
             self._data.update(mapping)
 
     @contextmanager
-    def lock(self, timeout: float = 10.0):
+    def lock(self, timeout: float = 10.0) -> Iterator[None]:
         """Acquire the internal lock."""
         acquired = self._lock.acquire(timeout=timeout)
         if not acquired:
@@ -224,7 +224,7 @@ class MemoryState(StateBackend):
     def increment(self, key: str) -> int:
         """Atomically increment a counter and return the new value."""
         with self._lock:
-            value = self._data.get(key, 0)
+            value: int = self._data.get(key, 0)
             new_value = value + 1
             self._data[key] = new_value
             return new_value
@@ -313,8 +313,10 @@ class RedisState(StateBackend):
         """Deserialize value from storage."""
         return json.loads(data) if data else None
 
+    # redis-py types every command as a sync/async union; this is the
+    # synchronous client, so the results are narrowed where they are consumed.
     def get(self, key: str, default: Any = None) -> Any:
-        data = self._client.get(self._key(key))
+        data = cast(bytes | None, self._client.get(self._key(key)))
         return self._deserialize(data) if data is not None else default
 
     def set(self, key: str, value: Any) -> None:
@@ -332,7 +334,10 @@ class RedisState(StateBackend):
         pattern = f"{self._namespace}:*"
         cursor = 0
         while True:
-            cursor, keys = self._client.scan(cursor, match=pattern, count=100)
+            cursor, keys = cast(
+                tuple[int, list[bytes]],
+                self._client.scan(cursor, match=pattern, count=100),
+            )
             if keys:
                 self._client.delete(*keys)
             if cursor == 0:
@@ -345,7 +350,10 @@ class RedisState(StateBackend):
         result = []
         cursor = 0
         while True:
-            cursor, keys = self._client.scan(cursor, match=pattern, count=100)
+            cursor, keys = cast(
+                tuple[int, list[bytes]],
+                self._client.scan(cursor, match=pattern, count=100),
+            )
             for k in keys:
                 key_str = k.decode("utf-8")
                 # Skip the lock key
@@ -369,7 +377,7 @@ class RedisState(StateBackend):
         pipe.execute()
 
     @contextmanager
-    def lock(self, timeout: float = 10.0):
+    def lock(self, timeout: float = 10.0) -> Iterator[None]:
         """
         Acquire a distributed lock using Redis.
 
@@ -402,13 +410,13 @@ class RedisState(StateBackend):
         :rtype: bool
         """
         try:
-            return self._client.ping()
+            return bool(self._client.ping())
         except redis.ConnectionError:
             return False
 
     def increment(self, key: str) -> int:
         """Atomically increment a counter using Redis INCR."""
-        return self._client.incr(self._key(key))
+        return cast(int, self._client.incr(self._key(key)))
 
     def get_multi(self, keys: list[str]) -> dict[str, Any]:
         """Get multiple values atomically using Redis MGET."""
@@ -416,7 +424,7 @@ class RedisState(StateBackend):
             return {}
 
         full_keys = [self._key(k) for k in keys]
-        values = self._client.mget(full_keys)
+        values = cast(list[bytes | None], self._client.mget(full_keys))
 
         result = {}
         for key, value in zip(keys, values, strict=True):

backend/pyproject.toml

@@ -44,11 +44,11 @@ build-backend = "hatchling.build"
 strict = true
 exclude = ["venv", ".venv", "alembic"]
 
-# app/flow/nodes.py still carries the prototype's node classes, and state.py's
-# redis calls type as sync/async unions under the current stubs. Both are being
-# revisited; every module around them is checked strictly.
+# app/flow/nodes.py still carries the prototype's node classes: the integration
+# nodes need annotations of their own, plus stubs for croniter and influxdb_client.
+# It is being revisited per integration; every module around it is checked strictly.
 [[tool.mypy.overrides]]
-module = ["app.flow.nodes", "app.flow.pipeline", "app.flow.state"]
+module = ["app.flow.nodes"]
 ignore_errors = true
 
 [tool.ruff]

docker/compose.dev.yml

@@ -45,6 +45,9 @@ services:
     ports:
       - "127.0.0.1:8080:8080"
 
+  redis:
+    restart: "no"
+
   backend:
     restart: "no"
     ports:

docker/compose.yml

@@ -29,6 +29,27 @@ services:
       - POSTGRES_USER=${POSTGRES_USER?Variable not set}
       - POSTGRES_DB=${POSTGRES_DB?Variable not set}
 
+  redis:
+    image: redis:8.2-alpine
+    container_name: fluksio-redis
+    restart: always
+    security_opt:
+      - no-new-privileges:true
+    # Flow state, not a cache: append-only persistence keeps the last value of
+    # every message across a restart.
+    command: redis-server --appendonly yes
+    # Deliberately not on `proxy` and not published: only the backend talks to it.
+    networks:
+      - default
+    healthcheck:
+      test: ["CMD", "redis-cli", "ping"]
+      interval: 10s
+      retries: 5
+      start_period: 10s
+      timeout: 5s
+    volumes:
+      - app-redis-data:/data
+
   adminer:
     image: adminer
     container_name: fluksio-adminer
@@ -94,6 +115,9 @@ services:
       db:
         condition: service_healthy
         restart: true
+      redis:
+        condition: service_healthy
+        restart: true
       prestart:
         condition: service_completed_successfully
     env_file:
@@ -116,6 +140,8 @@ services:
       - POSTGRES_USER=${POSTGRES_USER?Variable not set}
       - POSTGRES_PASSWORD=${POSTGRES_PASSWORD?Variable not set}
       - SENTRY_DSN=${SENTRY_DSN}
+      # Flow state survives a restart in Redis; without a host it stays in memory.
+      - REDIS_HOST=redis
       # Flows are files in a git repository; secrets sit encrypted beside it.
       - FLOWS_DIR=/data/flows
       - SECRETS_FILE=/data/secrets.enc
@@ -183,6 +209,7 @@ services:
 
 volumes:
   app-db-data:
+  app-redis-data:
   app-flow-data:
 
 networks: