feat: embedded flow-watcher in API lifespan with startup sync

backend/app/main.py

@@ -1,8 +1,58 @@
+import asyncio
+import logging
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from pathlib import Path
+
 from fastapi import FastAPI
 
 from .flows.api import approvals_router, flows_router
+from .flows.sync import run_sync_blocking, watch_forever
 from .settings import settings
 
+logger = logging.getLogger(__name__)
+
+
+def _blocking_startup_sync() -> None:
+    """One-shot reconciliation at startup (fresh session, never fatal)."""
+    try:
+        report = run_sync_blocking(Path(settings.flows_dir))
+        logger.info(
+            "startup flow sync: +%d ~%d -%d skipped=%d errors=%d",
+            len(report.added),
+            len(report.updated),
+            len(report.deleted),
+            len(report.skipped),
+            len(report.errors),
+        )
+        for filename, reason in report.errors.items():
+            logger.warning("startup flow sync: %s: %s", filename, reason)
+    except FileNotFoundError as exc:
+        logger.error("startup flow sync skipped: %s", exc)
+    except Exception:  # noqa: BLE001 - API must start even if sync fails
+        logger.exception("startup flow sync failed")
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI) -> AsyncIterator[None]:
+    task: asyncio.Task | None = None
+    stop: asyncio.Event | None = None
+    if settings.flows_watch_enabled:
+        await asyncio.to_thread(_blocking_startup_sync)
+        stop = asyncio.Event()
+        task = asyncio.create_task(
+            watch_forever(stop, Path(settings.flows_dir)), name="flows-watcher"
+        )
+    yield
+    if task is not None and stop is not None:
+        stop.set()
+        task.cancel()
+        try:
+            await task
+        except (asyncio.CancelledError, Exception):  # noqa: BLE001 - shutdown
+            pass
+
+
 app = FastAPI(
     title="Kirche Einfach API",
     description="KI-gestützte Verwaltungsplattform für Kirche",
@@ -10,6 +60,7 @@ app = FastAPI(
     docs_url="/api/docs",
     redoc_url="/api/redoc",
     openapi_url="/api/openapi.json",
+    lifespan=lifespan,
 )
 
 app.include_router(flows_router)

backend/tests/conftest.py

@@ -17,6 +17,9 @@ TEST_AREA = Path(tempfile.mkdtemp(prefix="kirche-tests-"))
 # Must precede all `app.*` imports: Settings is instantiated at import time.
 os.environ["DATABASE_URL"] = f"sqlite:///{TEST_AREA / 'test.db'}"
 os.environ["FLOWS_DIR"] = str(TEST_AREA / "flows")
+# The embedded watcher + startup sync stay off in tests; sync is exercised
+# directly (sync_from_directory) with per-test tmp dirs instead.
+os.environ["FLOWS_WATCH_ENABLED"] = "false"
 
 BACKEND_DIR = Path(__file__).resolve().parent.parent
 

backend/tests/test_sync.py

@@ -1,7 +1,9 @@
 """Sync engine tests: backend/flows/*.json -> DB + LFX (LFX real, DB sqlite)."""
 
+import asyncio
 import json
 import shutil
+import time
 from pathlib import Path
 from uuid import UUID
 
@@ -9,6 +11,7 @@ import pytest
 
 from app.flows import store, sync
 from app.flows.lfx_client import LfxClient, LfxNotFound
+from app.flows.schemas import SyncReport
 from app.flows.sync import lfx_id_for
 from app.settings import settings
 
@@ -127,3 +130,28 @@ async def test_sync_endpoint_returns_report(client, lfx_server, tmp_path, monkey
 
     response = await client.post("/api/flows/sync")
     assert response.json()["skipped"] == ["simple_rule"]
+
+
+async def test_watch_forever_triggers_on_file_change(tmp_path):
+    import contextlib
+
+    calls: list[str] = []
+    stop = asyncio.Event()
+
+    def run():
+        calls.append("sync")
+        return SyncReport()
+
+    task = asyncio.create_task(sync.watch_forever(stop, tmp_path, run_sync=run))
+    try:
+        await asyncio.sleep(0.5)  # let the watcher arm
+        (tmp_path / "neu.json").write_text("{}")
+        deadline = time.time() + 15
+        while not calls and time.time() < deadline:
+            await asyncio.sleep(0.1)
+        assert calls == ["sync"]
+    finally:
+        stop.set()
+        task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await task