docs: rewrite README and AGENTS.md, add admin sync button

.devcontainer/Dockerfile.lfx

@@ -8,7 +8,7 @@ ENV PYTHONUNBUFFERED=1 \
     PIP_NO_CACHE_DIR=1
 
 # lfx = flow runtime; requests = convenience for custom-component code.
-RUN pip install "lfx==1.12.0" requests
+RUN pip install lfx requests
 
 RUN mkdir -p /flows
 VOLUME /flows

.devcontainer/docker-compose.yml

@@ -19,26 +19,8 @@ services:
       - "8000:8000" # FastAPI
       - "3000:3000" # Vite
 
-  db:
-    image: postgres:16-trixie
-    env_file:
-      - ../.env
-    volumes:
-      - postgres-data:/var/lib/postgresql/data
-      - ./init-db.sql:/docker-entrypoint-initdb.d/init-db.sql:ro
-    ports:
-      - "5432:5432"
-    healthcheck:
-      test: ["CMD-SHELL", "pg_isready -U postgres"]
-      interval: 5s
-      timeout: 3s
-      retries: 10
-
   langflow:
-    image: langflowai/langflow:1.12.0
+    image: langflowai/langflow:latest
-    depends_on:
-      db:
-        condition: service_healthy
     env_file:
       - ../.env
     environment:
@@ -49,6 +31,9 @@ services:
       PORT: "7860"
       LANGFLOW_AUTO_LOGIN: "false"
       LANGFLOW_API_KEY_SOURCE: env
+      # SQLite file inside the langflow-data volume (absolute path required
+      # by Langflow); overrides any postgres URL in ../.env.
+      LANGFLOW_DATABASE_URL: sqlite:////app/langflow/langflow.db
     volumes:
       - langflow-data:/app/langflow
     ports:
@@ -85,9 +70,23 @@ services:
       retries: 12
       start_period: 30s
 
+  # Optional nightly auto-updates of :latest images. Not part of the default
+  # dev startup (no container updates needed while iterating).
+  watchtower:
+    image: nickfedor/watchtower:latest
+    profiles: ["updates"]
+    volumes:
+      - /var/run/docker.sock:/var/run/docker.sock
+    environment:
+      WATCHTOWER_LABEL_ENABLE: "true"
+      WATCHTOWER_SCHEDULE: "0 0 4 * * *"
+      WATCHTOWER_CLEANUP: "true"
+      WATCHTOWER_ROLLING_RESTART: "true"
+    labels:
+      com.centurylinklabs.watchtower.enable: "true"
+
 volumes:
   kirche-einfach-venv:
   kirche-einfach-node-modules:
-  postgres-data:
   langflow-data:
   lfx-data:

.devcontainer/init-db.sql

@@ -1,7 +0,0 @@
--- Runs once on first init of the postgres-data volume.
--- The main app database ($POSTGRES_DB) is created by the image itself.
--- Langflow (authoring UI) gets its own database on the same instance.
-CREATE DATABASE langflow;
-
--- Isolated database for the pytest suite (migrated via Alembic in conftest).
-CREATE DATABASE kirche_einfach_test;

.devcontainer/setup.sh

@@ -13,7 +13,7 @@ echo "--- Installing Node dependencies ---"
 cd /workspaces/kirche-einfach/frontend
 npm install
 
-# Compose services (db, langflow, lfx) must be up; fail loudly otherwise.
+# Compose services (langflow, lfx) must be up; fail loudly otherwise.
 echo "--- Waiting for compose services ---"
 if [ ! -f /workspaces/kirche-einfach/.env ]; then
   echo "ERROR: /workspaces/kirche-einfach/.env is missing. Copy .env.example to .env and fill in secrets." >&2
@@ -24,7 +24,7 @@ import socket
 import sys
 import time
 
-targets = [("db", 5432), ("langflow", 7860), ("lfx", 6004)]
+targets = [("langflow", 7860), ("lfx", 6004)]
 deadline = time.time() + 180
 pending = dict(targets)
 while pending and time.time() < deadline:
@@ -45,6 +45,14 @@ if pending:
     sys.exit(f"ERROR: compose services not ready: {sorted(pending)}")
 EOF
 
+# Migrate the SQLite schema and reconcile backend/flows into the DB + LFX.
+echo "--- Migrating database ---"
+cd /workspaces/kirche-einfach/backend
+uv run alembic upgrade head
+
+echo "--- Syncing canonical flows ---"
+uv run python scripts/sync_flows.py --sync || echo "WARN: flow sync failed (check backend/flows and LFX)"
+
 echo ""
 echo "=== Setup complete ==="
 echo "Forgejo username: $FORGEJO_USERNAME"

.env.example

@@ -1,12 +1,14 @@
 # Copy to `.env` and fill in values. `.env` is gitignored.
 FORGEJO_TOKEN=
 
-# --- databases (compose service `db`) ---
+# --- app database (SQLite, default if unset) ---
-POSTGRES_USER=kirche
+# Override only if you want a different engine/URL.
-POSTGRES_PASSWORD=
+# DATABASE_URL=sqlite:///backend/data/kirche.db
-POSTGRES_DB=kirche_einfach
+
-DATABASE_URL=postgresql+psycopg://kirche:<password>@db:5432/kirche_einfach
+# --- canonical flow definitions ---
-TEST_DATABASE_URL=postgresql+psycopg://kirche:<password>@db:5432/kirche_einfach_test
+# FLOWS_DIR=backend/flows
+# Auto-reconcile backend/flows into DB + LFX while the API runs.
+# FLOWS_WATCH_ENABLED=true
 
 # --- langflow (authoring, dev only) ---
 LANGFLOW_URL=http://langflow:7860
@@ -15,7 +17,7 @@ LANGFLOW_SUPERUSER_PASSWORD=
 LANGFLOW_SECRET_KEY=
 # Single static key; validated against env (LANGFLOW_API_KEY_SOURCE=env), superuser scope.
 LANGFLOW_API_KEY=sk-
-LANGFLOW_DATABASE_URL=postgresql://kirche:<password>@db:5432/langflow
+# Langflow stores flows in its own SQLite file when unset.
 
 # --- lfx (flow runtime) ---
 LFX_URL=http://lfx:6004

.gitignore

@@ -14,6 +14,12 @@ dist/
 .env
 .env.local
 
+# Runtime SQLite databases
+backend/data/
+*.db
+*.db-shm
+*.db-wal
+
 # IDE
 .vscode/
 .idea/

AGENTS.md

@@ -1,5 +1,44 @@
 # Agents
 
+Kirche Einfach — AI-assisted church administration. FastAPI backend (`backend/`, Python 3.12, SQLModel, SQLite) + React 19/Vite frontend (`frontend/`). Flow definitions live as JSON files in `backend/flows/` (source of truth); the DB is derived by the sync engine.
+
+## Commands
+
+Backend (run from `backend/`):
+- `uv sync` — install dependencies
+- `uv run pytest` — full test suite (throwaway SQLite file, no external DB needed)
+- `uv run alembic upgrade head` — migrate the dev database (`backend/data/`, gitignored)
+- `uv run python scripts/sync_flows.py --sync` — reconcile `backend/flows/` into DB + LFX
+- `uv run uvicorn app.main:app --port 8000` — run the API (paths resolve from repo root, CWD-independent)
+
+Frontend (run from `frontend/`): `npm install`, `npm run dev`, `npm run build` (typecheck + build).
+
+## Definition of done
+
+- New behavior has tests; `uv run pytest` is green before reporting completion.
+- `npm run build` is green when frontend code changed.
+- Schema changes ship as a new Alembic revision — never edit an applied migration.
+
+## Architecture rules
+
+- `backend/flows/*.json` is the source of truth (one Langflow export per file). Never write flow JSON anywhere else.
+- Only `backend/app/flows/sync.py` reconciles files into the DB. CLI, `POST /api/flows/sync`, and the embedded watcher must call it — never duplicate its logic.
+- All DB access goes through `backend/app/flows/store.py`. Keep API handlers thin.
+- Talk to LFX only via `LfxClient`, to Langflow only via `LangflowClient`.
+- Flow JSON must stay a valid Langflow export (`store` requires `data.nodes`/`data.edges`).
+
+```python
+from app.flows.sync import sync_from_directory
+report = await sync_from_directory(session, lfx, settings.flows_dir)
+```
+
+## Boundaries
+
+- Never commit secrets. `.env` is gitignored; only `.env.example` is tracked.
+- Never edit `backend/data/*` (runtime SQLite files) or `__pycache__`.
+- Never add a second file-watcher or a second sync path.
+- Deleting `backend/flows/<name>.json` deletes the flow including its runs from the DB (by design).
+
 ## Git
 
-When committing: `git add -A && git commit -m "<message>"`. When pushing: `git push` (branch is tracking `origin/main`).
+When committing: `git add -A && git commit -m "<message>"` (conventional commits: `feat:`, `fix:`, `chore:`, `docs:`). When pushing: `git push` (branch is tracking `origin/main`).

README.md

@@ -1,17 +1,103 @@
 # Kirche Einfach
 
-KI-gestützte Verwaltungsplattform für Kirche.
+AI-assisted administration platform for church organizations.
 
-## Stack
+Staff describe a request in plain words. The platform guides them through the
+process, checks the rules automatically, and asks a human only when something
+needs judgment. Every step is documented and traceable.
 
-- **Backend:** Python 3.12, FastAPI, Django, uv
+## Features
-- **Frontend:** React 19, TypeScript, Vite
+
-- **Dev:** Devcontainer, opencode
+- **Visual flow authoring** — design decision flows in Langflow, then bring
+  them into the platform.
+- **Flows as files** — each flow lives as one JSON file under
+  `backend/flows/`. The directory is the source of truth: committed to git,
+  reviewable, diffable.
+- **Automatic sync** — the API watches `backend/flows/` and reconciles every
+  change into the database and the runtime: new files become flows, edits
+  become new versions, deleted files remove the flow (including its runs).
+  Same on every startup; also triggerable via `POST /api/flows/sync` or
+  `uv run python scripts/sync_flows.py --sync`.
+- **Versioned runtime** — every change creates an immutable version; versions
+  deploy to the lightweight LFX runtime and can run synchronously or streamed.
+- **Approvals** — a flow can pause and ask a human for a decision. Approving
+  can automatically start a follow-up ("continuation") flow.
+- **Web UI** — three tabs: *Vorgänge* (flows and runs), *Freigaben*
+  (approvals), *Administration* (import, sync, deploy).
+
+## Infrastructure
+
+```mermaid
+flowchart LR
+    subgraph Repo["Git repo — source of truth"]
+        FLOWS["backend/flows/*.json<br/>one Langflow export per flow"]
+    end
+    subgraph Stack["Compose stack (dev + prod)"]
+        UI["frontend — React UI (3000)"]
+        API["app — FastAPI (8000)<br/>+ embedded flow-watcher"]
+        DB[("app.db — SQLite file")]
+        LF["langflow — authoring UI (7860)<br/>langflow.db — SQLite file"]
+        LFX["lfx — flow runtime (6004)"]
+        WT["watchtower — nightly image updates"]
+    end
+    UI --> API
+    FLOWS -->|watch + sync| API
+    API -->|definitions · versions · runs · approvals| DB
+    API -->|pull / export| LF
+    API -->|upload · run · unregister| LFX
+    LFX -->|approval callbacks| API
+    WT -.->|updates :latest images| LF
+```
+
+| Service        | Purpose                              | URL                        |
+| -------------- | ------------------------------------ | -------------------------- |
+| React frontend | Web UI (Vite)                        | http://localhost:3000      |
+| FastAPI backend| API + flow orchestration             | http://localhost:8000      |
+| API docs       | Swagger UI                           | http://localhost:8000/api/docs |
+| Langflow       | Visual flow authoring                | http://localhost:7860      |
+| LFX            | Flow runtime (internal, no direct UI)| localhost:6004             |
+| SQLite files   | `backend/data/kirche.db` (app), Langflow keeps its own file | — |
+
+Why SQLite: a flow run needs only a handful of single-row statements, and
+Langflow itself runs on SQLite by default — so the whole stack works without
+a database server. Postgres can return later by changing `DATABASE_URL`
+(migrations stay portable).
 
 ## Getting Started
 
-Open in VS Code → "Reopen in Container".
+1. Copy `.env.example` to `.env` and fill in the secrets.
+2. Open the repo in VS Code → "Reopen in Container". Setup installs
+   dependencies, migrates the database, and syncs `backend/flows/`.
+3. Run the backend from `backend/`: `uv run uvicorn app.main:app --port 8000`.
+   Run the frontend from `frontend/`: `npm run dev`.
 
-Backend: `http://localhost:8000`
+Production (self-hosted, internal use — no TLS/reverse proxy included):
-Frontend: `http://localhost:3000`
+
-API Docs: `http://localhost:8000/api/docs`
+```sh
+docker compose -f compose.prod.yml up -d --build
+```
+
+## Development
+
+Backend (from `backend/`):
+
+- `uv run pytest` — full test suite (throwaway SQLite file, no external DB).
+- `uv run alembic upgrade head` — migrate the dev database.
+- `uv run python scripts/sync_flows.py --sync` — reconcile `backend/flows/`.
+- `uv run python scripts/sync_flows.py --pull` — export all Langflow flows
+  into `backend/flows/` first.
+- `uv run python scripts/deploy_flow.py --langflow-id <uuid>` — import one
+  Langflow flow via the API.
+- `uv run python scripts/make_flow_fixtures.py` — regenerate the bundled
+  flows (writes test fixtures and `backend/flows/`).
+- `uv run python scripts/smoke_e2e.py` — manual end-to-end smoke.
+
+Frontend (from `frontend/`): `npm install`, `npm run dev`, `npm run build`.
+
+Flow workflow: design in Langflow → export into `backend/flows/<name>.json`
+(`--pull`, or UI download) → commit → the watcher deploys it automatically.
+Deleting a file removes the flow including its runs (git keeps the JSON).
+
+Container updates: registry images (`langflowai/langflow:latest`)
+auto-update nightly via Watchtower (`nickfedor/watchtower`). Locally built
+images (`app`, `lfx`, `frontend`) update with `docker compose build --pull`.

backend/Dockerfile

@@ -0,0 +1,21 @@
+# Production image for the FastAPI backend.
+FROM python:3.12-slim
+
+ENV PYTHONUNBUFFERED=1 \
+    PIP_NO_CACHE_DIR=1
+
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
+
+WORKDIR /srv/backend
+COPY pyproject.toml alembic.ini ./
+COPY app ./app
+COPY migrations ./migrations
+COPY scripts ./scripts
+# No lockfile is committed; dependencies resolve at build time.
+RUN uv sync --no-dev
+
+ENV FLOWS_WATCH_ENABLED=true
+EXPOSE 8000
+
+# Migrate on start, then serve (watcher reconciles backend/flows automatically).
+CMD ["sh", "-c", ".venv/bin/alembic upgrade head && exec .venv/bin/uvicorn app.main:app --host 0.0.0.0 --port 8000"]

backend/app/db.py

@@ -1,6 +1,7 @@
 """Database engine, sessions, and FastAPI dependency."""
 
 from collections.abc import Iterator
+from pathlib import Path
 
 from sqlalchemy import Engine
 from sqlmodel import Session, SQLModel, create_engine
@@ -10,11 +11,23 @@ from .settings import settings
 _engine: Engine | None = None
 
 
+def _engine_kwargs(url: str) -> dict:
+    if url.startswith("sqlite"):
+        # FastAPI runs sync endpoints in a threadpool; connections are shared
+        # across threads, so defer the same-thread check. File-based SQLite
+        # needs an existing parent dir (SQLite won't create it).
+        return {"connect_args": {"check_same_thread": False}}
+    return {}
+
+
 def get_engine(url: str | None = None) -> Engine:
     """Process-wide engine singleton (override URL in tests via settings)."""
     global _engine
     if _engine is None:
-        _engine = create_engine(url or settings.database_url)
+        url = url or settings.database_url
+        if url.startswith("sqlite:///") and not url.startswith("sqlite:///:memory:"):
+            Path(url.removeprefix("sqlite:///")).parent.mkdir(parents=True, exist_ok=True)
+        _engine = create_engine(url, **_engine_kwargs(url))
     return _engine
 
 

backend/app/flows/api.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import json
 from collections.abc import AsyncIterator, Iterator
+from pathlib import Path
 from uuid import UUID
 
 import httpx
@@ -16,6 +17,7 @@ from app.settings import settings
 
 from . import runner, store
 from .deploy import import_from_langflow
+from .sync import lfx_id_for, sync_from_directory
 from .langflow_client import (
     LangflowClient,
     LangflowError,
@@ -44,6 +46,7 @@ from .schemas import (
     FlowVersionRead,
     RunRead,
     RunRequest,
+    SyncReport,
 )
 
 flows_router = APIRouter(prefix="/api/flows", tags=["flows"])
@@ -70,6 +73,7 @@ def _read_definition(d) -> FlowDefinitionRead:
         name=d.name,
         description=d.description,
         status=d.status,
+        source_path=d.source_path,
         continuation_flow_id=d.continuation_flow_id,
         created_at=d.created_at,
     )
@@ -134,6 +138,18 @@ def list_flows(session: Session = Depends(get_db)):
     return [_read_definition(d) for d in store.list_definitions(session)]
 
 
+@flows_router.post("/sync", response_model=SyncReport)
+async def sync_flows(
+    session: Session = Depends(get_db),
+    lfx: LfxClient = Depends(get_lfx),
+):
+    """Reconcile backend/flows/*.json into the DB + LFX (add/update/delete)."""
+    try:
+        return await sync_from_directory(session, lfx, Path(settings.flows_dir))
+    except FileNotFoundError as exc:
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+
 @flows_router.get("/{flow_id}", response_model=FlowDefinitionRead)
 def get_flow(flow_id: UUID, session: Session = Depends(get_db)):
     try:
@@ -178,8 +194,12 @@ async def deploy_version(
     except store.NotFoundError as exc:
         raise HTTPException(status_code=404, detail=str(exc)) from exc
     try:
+        # Deterministic LFX id: same (definition, version) always maps to the
+        # same registration, no matter which path deployed it.
         uploaded = await lfx.upload_flow(
-            version.langflow_json, flow_id=str(version.id), replace=True
+            version.langflow_json,
+            flow_id=lfx_id_for(flow_id, version_no),
+            replace=True,
         )
     except LfxInvalidFlow as exc:
         raise HTTPException(status_code=422, detail=str(exc)) from exc

backend/app/flows/deploy.py

@@ -1,15 +1,15 @@
-"""Authoring pipeline: Langflow export -> versioned definition -> LFX deploy."""
+"""Authoring pipeline: Langflow export -> canonical file -> version -> LFX deploy."""
 
 from __future__ import annotations
 
 from sqlmodel import Session
 
 from app.models import FlowDefinition, FlowVersion
+from app.settings import settings
 
-from . import store
+from . import export, store, sync
 from .langflow_client import LangflowClient
 from .lfx_client import LfxClient
-from .schemas import FlowDefinitionCreate, FlowVersionCreate
 
 
 async def import_from_langflow(
@@ -22,39 +22,31 @@ async def import_from_langflow(
     author: str = "",
     deploy: bool = True,
 ) -> tuple[FlowDefinition, FlowVersion]:
-    """Export a flow from Langflow, store it as a new version, deploy to LFX.
+    """Export a flow from Langflow, persist it to backend/flows, then sync.
 
-    Definitions are matched by name (explicit `name` wins, else the exported
+    The canonical file is written first so the directory stays the source of
-    name); every import creates a new immutable version.
+    truth; sync_document() derives the definition + version from it.
     """
-    export = await langflow.export_flow(langflow_flow_id)
+    exported = await langflow.export_flow(langflow_flow_id)
-    flow_name = name or export.get("name") or f"langflow-{langflow_flow_id}"
+    flow_name = name or exported.get("name") or f"langflow-{langflow_flow_id}"
     langflow_json = {
-        "id": export.get("id", langflow_flow_id),
+        "id": exported.get("id", langflow_flow_id),
         "name": flow_name,
-        "description": export.get("description") or "",
+        "description": exported.get("description") or "",
-        "endpoint_name": export.get("endpoint_name"),
+        "endpoint_name": exported.get("endpoint_name"),
-        "data": export.get("data", {}),
+        "data": exported.get("data", {}),
-        "tags": export.get("tags", []),
+        "tags": exported.get("tags", []),
-        "webhook": export.get("webhook", False),
+        "webhook": exported.get("webhook", False),
-        "is_component": export.get("is_component", False),
+        "is_component": exported.get("is_component", False),
     }
-    try:
+    path = export.write_flow_file(settings.flows_dir, langflow_json, name=flow_name)
-        definition = next(
+    definition, version, _action = await sync.sync_document(
-            d for d in store.list_definitions(session) if d.name == flow_name
-        )
-    except StopIteration:
-        definition = store.create_definition(
-            session, FlowDefinitionCreate(name=flow_name)
-        )
-    version = store.create_version(
         session,
-        definition.id,
+        lfx,
-        FlowVersionCreate(langflow_json=langflow_json, author=author),
+        langflow_json,
+        source_filename=path.name,
+        author=author,
+        deploy=deploy,
     )
-    if deploy:
+    assert version is not None  # sync_document always versions on import
-        uploaded = await lfx.upload_flow(
-            version.langflow_json, flow_id=str(version.id), replace=True
-        )
-        version = store.mark_deployed(session, version, uploaded.id)
     return definition, version

backend/app/flows/export.py

@@ -0,0 +1,71 @@
+"""Canonical flow files: backend/flows/<name>.json is the source of truth.
+
+One Langflow export per file, committed to git. This module only reads and
+writes the directory; reconciliation into the DB lives in sync.py.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+import tempfile
+from pathlib import Path
+
+FLOWS_GLOB = "*.json"
+
+
+def slugify(name: str) -> str:
+    """Map a flow name to a safe filename stem (lowercase, [a-z0-9-])."""
+    slug = re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")
+    return slug or "flow"
+
+
+def canonical_bytes(doc: dict) -> bytes:
+    """Deterministic serialization: stable bytes for stable git diffs."""
+    return (json.dumps(doc, indent=2, ensure_ascii=False, sort_keys=True) + "\n").encode(
+        "utf-8"
+    )
+
+
+def content_hash(doc: dict) -> str:
+    """Hash of the parsed document (formatting-independent)."""
+    return hashlib.sha256(canonical_bytes(doc)).hexdigest()
+
+
+def flow_path(flows_dir: str | Path, stem: str) -> Path:
+    return Path(flows_dir) / f"{stem}.json"
+
+
+def write_flow_file(
+    flows_dir: str | Path, doc: dict, *, name: str | None = None
+) -> Path:
+    """Write a flow document to backend/flows atomically (tmp + rename).
+
+    The filename derives from `name` (or the document's own name). Returns
+    the written path.
+    """
+    flows_dir = Path(flows_dir)
+    flows_dir.mkdir(parents=True, exist_ok=True)
+    stem = slugify(name or doc.get("name") or "flow")
+    target = flow_path(flows_dir, stem)
+    fd, tmp_name = tempfile.mkstemp(
+        dir=str(flows_dir), prefix=f".{stem}-", suffix=".tmp"
+    )
+    try:
+        with os.fdopen(fd, "wb") as tmp:
+            tmp.write(canonical_bytes(doc))
+        os.replace(tmp_name, target)
+    except BaseException:
+        try:
+            os.unlink(tmp_name)
+        except OSError:
+            pass
+        raise
+    return target
+
+
+def iter_flow_files(flows_dir: str | Path) -> list[Path]:
+    """Sorted list of candidate flow files (top level only)."""
+    return sorted(Path(flows_dir).glob(FLOWS_GLOB))

backend/app/flows/schemas.py

@@ -17,6 +17,8 @@ class FlowDefinitionRead(BaseModel):
     name: str
     description: str
     status: str
+    # backend/flows filename this definition syncs from (None if API-created).
+    source_path: str | None = None
     continuation_flow_id: UUID | None
     created_at: datetime
 
@@ -110,3 +112,16 @@ class DecisionRequest(BaseModel):
 class DecisionRead(BaseModel):
     task: ApprovalRead
     continuation_run: RunRead | None
+
+
+class SyncReport(BaseModel):
+    """Result of reconciling backend/flows/*.json into DB + LFX.
+
+    Lists hold filename stems; errors maps filename to reason.
+    """
+
+    added: list[str] = Field(default_factory=list)
+    updated: list[str] = Field(default_factory=list)
+    deleted: list[str] = Field(default_factory=list)
+    skipped: list[str] = Field(default_factory=list)
+    errors: dict[str, str] = Field(default_factory=dict)

backend/app/flows/store.py

@@ -56,7 +56,82 @@ def list_definitions(session: Session) -> list[FlowDefinition]:
     return list(session.exec(select(FlowDefinition).order_by(FlowDefinition.name)).all())
 
 
-def _validate_flow_json(flow_json: dict) -> None:
+def get_definition_by_source_path(
+    session: Session, source_path: str
+) -> FlowDefinition | None:
+    """Find the definition synced from a given backend/flows filename."""
+    return session.exec(
+        select(FlowDefinition).where(FlowDefinition.source_path == source_path)
+    ).first()
+
+
+def set_definition_name(
+    session: Session, definition: FlowDefinition, name: str
+) -> FlowDefinition:
+    """Rename a definition and record its source file. Name clashes 409."""
+    definition.name = name
+    session.add(definition)
+    try:
+        session.commit()
+    except IntegrityError as exc:
+        session.rollback()
+        raise ConflictError(f"flow {name!r} already exists") from exc
+    session.refresh(definition)
+    return definition
+
+
+def next_version_no(session: Session, flow_id: UUID) -> int:
+    max_no = session.exec(
+        select(func.max(FlowVersion.version_no)).where(FlowVersion.flow_id == flow_id)
+    ).one()
+    return (max_no or 0) + 1
+
+
+def delete_definition_cascade(
+    session: Session, definition: FlowDefinition
+) -> dict[str, int]:
+    """Remove a definition and its whole subtree (tasks, runs, versions).
+
+    Used when the canonical file disappears: the directory is the source of
+    truth, so the derived registry follows. Commits once.
+    """
+    version_ids = list(
+        session.exec(
+            select(FlowVersion.id).where(FlowVersion.flow_id == definition.id)
+        ).all()
+    )
+    task_count = 0
+    run_count = 0
+    if version_ids:
+        tasks = list(
+            session.exec(
+                select(ApprovalTask).where(ApprovalTask.version_id.in_(version_ids))
+            ).all()
+        )
+        runs = list(
+            session.exec(
+                select(FlowRun).where(FlowRun.version_id.in_(version_ids))
+            ).all()
+        )
+        task_count, run_count = len(tasks), len(runs)
+        for row in (*tasks, *runs):
+            session.delete(row)
+        for version in session.exec(
+            select(FlowVersion).where(FlowVersion.flow_id == definition.id)
+        ).all():
+            session.delete(version)
+    session.delete(definition)
+    session.commit()
+    return {
+        "definitions": 1,
+        "versions": len(version_ids),
+        "runs": run_count,
+        "tasks": task_count,
+    }
+
+
+def validate_flow_json(flow_json: dict) -> None:
+    """Reject anything that is not a Langflow-export-shaped document."""
     if not isinstance(flow_json, dict):
         raise InvalidError("langflow_json must be an object")
     data = flow_json.get("data")
@@ -70,7 +145,7 @@ def create_version(
     session: Session, flow_id: UUID, data: FlowVersionCreate
 ) -> FlowVersion:
     get_definition(session, flow_id)
-    _validate_flow_json(data.langflow_json)
+    validate_flow_json(data.langflow_json)
     max_no = session.exec(
         select(func.max(FlowVersion.version_no)).where(FlowVersion.flow_id == flow_id)
     ).one()

backend/app/flows/sync.py

@@ -0,0 +1,274 @@
+"""Reconciliation: backend/flows/*.json (source of truth) -> DB + LFX.
+
+The ONLY place that derives registry state from the canonical directory.
+Triggers (CLI, POST /api/flows/sync, embedded watcher) call
+sync_from_directory(); single-document callers (e.g. /import) use
+sync_document(). Everything is async (the LFX client is async) and guarded
+by a module lock so concurrent triggers serialize on SQLite's single writer.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import threading
+import uuid
+from collections.abc import Callable
+from pathlib import Path
+from uuid import UUID
+
+from sqlmodel import Session
+from watchfiles import DefaultFilter, awatch
+
+from app.db import get_engine
+from app.models import FlowDefinition, FlowVersion
+from app.settings import settings
+
+from . import export, store
+from .lfx_client import LfxClient, LfxError
+from .schemas import FlowDefinitionCreate, FlowVersionCreate, SyncReport
+
+logger = logging.getLogger(__name__)
+
+# Serializes all writers in this process (sync engine + manual API writes).
+sync_lock = threading.Lock()
+
+
+def lfx_id_for(definition_id: UUID, version_no: int) -> str:
+    """Deterministic LFX id (a real UUID: LFX validates the format).
+
+    Same (definition, version) always maps to the same id, so re-syncs and
+    crash recovery overwrite instead of orphaning LFX registrations.
+    """
+    return str(
+        uuid.uuid5(uuid.NAMESPACE_URL, f"kirche-einfach/{definition_id}/v{version_no}")
+    )
+
+
+def _latest_version(session: Session, flow_id: UUID) -> FlowVersion | None:
+    versions = store.list_versions(session, flow_id)
+    return versions[-1] if versions else None
+
+
+async def sync_document(
+    session: Session,
+    lfx: LfxClient,
+    doc: dict,
+    *,
+    source_filename: str,
+    author: str = "",
+    deploy: bool = True,
+) -> tuple[FlowDefinition, FlowVersion | None, str]:
+    """Reconcile one parsed document. Returns (definition, version, action).
+
+    Actions: "added" (new definition + version), "updated" (new version),
+    "skipped" (content unchanged). Raises store.ConflictError on name
+    clashes, store.InvalidError on invalid JSON, LfxError when deploy fails.
+    With deploy=False a version is recorded without touching LFX.
+    """
+    store.validate_flow_json(doc)
+    name = doc.get("name") or Path(source_filename).stem
+    with sync_lock:
+        definition = store.get_definition_by_source_path(session, source_filename)
+        if definition is None:
+            definition = store.create_definition(
+                session, FlowDefinitionCreate(name=name)
+            )
+            definition.source_path = source_filename
+            session.add(definition)
+            session.commit()
+            session.refresh(definition)
+            action = "added"
+        else:
+            action = "updated"
+            if definition.name != name:
+                store.set_definition_name(session, definition, name)
+
+        latest = _latest_version(session, definition.id)
+        if (
+            latest is not None
+            and export.content_hash(latest.langflow_json)
+            == export.content_hash(doc)
+        ):
+            return definition, latest, "skipped"
+
+        if not deploy:
+            version = store.create_version(
+                session,
+                definition.id,
+                FlowVersionCreate(langflow_json=doc, author=author),
+            )
+            return definition, version, action
+
+        version_no = store.next_version_no(session, definition.id)
+        lfx_id = lfx_id_for(definition.id, version_no)
+        uploaded = await lfx.upload_flow(doc, flow_id=lfx_id, replace=True)
+        version = store.create_version(
+            session,
+            definition.id,
+            FlowVersionCreate(langflow_json=doc, author=author),
+        )
+        if version.version_no != version_no:  # pragma: no cover - lock prevents this
+            raise store.ConflictError(
+                f"concurrent version write for flow {definition.id}"
+            )
+        version = store.mark_deployed(session, version, uploaded.id)
+        if (
+            latest is not None
+            and latest.lfx_flow_id
+            and latest.lfx_flow_id != uploaded.id
+        ):
+            try:
+                await lfx.delete_flow(latest.lfx_flow_id)
+            except LfxError as exc:
+                logger.warning(
+                    "LFX unregister %s failed (continuing): %s",
+                    latest.lfx_flow_id,
+                    exc,
+                )
+        return definition, version, action
+
+
+async def delete_definition_tree(
+    session: Session, lfx: LfxClient, definition: FlowDefinition
+) -> dict[str, int]:
+    """Remove a definition whose canonical file is gone.
+
+    LFX registrations go first (best-effort, fail-closed for new runs), the
+    DB subtree (tasks, runs, versions, definition) after. Deterministic LFX
+    ids make any crash in between self-healing on the next sync.
+    """
+    for version in store.list_versions(session, definition.id):
+        if version.lfx_flow_id:
+            try:
+                await lfx.delete_flow(version.lfx_flow_id)
+            except LfxError as exc:
+                logger.warning(
+                    "LFX unregister %s failed (continuing): %s",
+                    version.lfx_flow_id,
+                    exc,
+                )
+    with sync_lock:
+        return store.delete_definition_cascade(session, definition)
+
+
+async def sync_from_directory(
+    session: Session, lfx: LfxClient, flows_dir: str | Path
+) -> SyncReport:
+    """Reconcile the whole canonical directory. Per-file problems land in
+    report.errors (never raised); a missing directory raises FileNotFoundError
+    so misconfiguration fails loudly instead of wiping the registry.
+    """
+    flows_dir = Path(flows_dir)
+    report = SyncReport()
+    if not flows_dir.is_dir():
+        raise FileNotFoundError(f"flows directory missing: {flows_dir}")
+    files = {path.name: path for path in export.iter_flow_files(flows_dir)}
+    for filename, path in files.items():
+        try:
+            doc = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, ValueError) as exc:
+            report.errors[filename] = f"unreadable: {exc}"
+            continue
+        if not isinstance(doc, dict):
+            report.errors[filename] = "top-level JSON must be an object"
+            continue
+        try:
+            _, _, action = await sync_document(
+                session, lfx, doc, source_filename=filename
+            )
+        except Exception as exc:  # noqa: BLE001 - reconciliation is total
+            report.errors[filename] = str(exc)
+            continue
+        getattr(report, action).append(path.stem)
+    known = {
+        definition.source_path
+        for definition in store.list_definitions(session)
+        if definition.source_path
+    }
+    for stale in sorted(known - set(files)):
+        definition = store.get_definition_by_source_path(session, stale)
+        if definition is None:  # pragma: no cover - defensive
+            continue
+        try:
+            await delete_definition_tree(session, lfx, definition)
+        except Exception as exc:  # noqa: BLE001 - reconciliation is total
+            report.errors[stale] = str(exc)
+            continue
+        report.deleted.append(Path(stale).stem)
+    return report
+
+
+def default_session_factory() -> Callable[[], Session]:
+    return lambda: Session(get_engine())
+
+
+def default_lfx_factory() -> Callable[[], LfxClient]:
+    return lambda: LfxClient(
+        settings.lfx_url, api_key=settings.lfx_api_key or None
+    )
+
+
+def run_sync_blocking(
+    flows_dir: str | Path,
+    *,
+    session_factory: Callable[[], Session] | None = None,
+    lfx_factory: Callable[[], LfxClient] | None = None,
+) -> SyncReport:
+    """Blocking one-shot sync for worker threads (lifespan, watcher)."""
+    session = (session_factory or default_session_factory())()
+    try:
+        lfx = (lfx_factory or default_lfx_factory())()
+        return asyncio.run(sync_from_directory(session, lfx, Path(flows_dir)))
+    finally:
+        session.close()
+
+
+class _JsonFilter(DefaultFilter):
+    def __call__(self, change, path: str) -> bool:  # type: ignore[override]
+        return super().__call__(change, path) and path.endswith(".json")
+
+
+async def watch_forever(
+    stop: asyncio.Event,
+    flows_dir: str | Path,
+    run_sync: Callable[[], SyncReport] | None = None,
+) -> None:
+    """Run until `stop` is set: sync on every batch of .json changes.
+
+    All failures are logged, never raised — a broken file must not kill the
+    watcher. `run_sync` is injectable for tests.
+    """
+    flows_dir = Path(flows_dir)
+    run = run_sync or (lambda: run_sync_blocking(flows_dir))
+    while not stop.is_set() and not flows_dir.is_dir():
+        logger.error("flows dir missing, retrying in 30s: %s", flows_dir)
+        try:
+            await asyncio.wait_for(stop.wait(), timeout=30)
+        except asyncio.TimeoutError:
+            pass
+    if stop.is_set():
+        return
+    logger.info("watching %s", flows_dir)
+    async for _ in awatch(
+        flows_dir, watch_filter=_JsonFilter(), debounce=1000, step=100
+    ):
+        if stop.is_set():
+            break
+        try:
+            report = await asyncio.to_thread(run)
+        except FileNotFoundError as exc:
+            logger.error("flow sync failed: %s", exc)
+            continue
+        except Exception:  # noqa: BLE001 - watcher must survive
+            logger.exception("flow sync failed")
+            continue
+        logger.info(
+            "flow sync: +%d ~%d -%d skipped=%d errors=%d",
+            len(report.added),
+            len(report.updated),
+            len(report.deleted),
+            len(report.skipped),
+            len(report.errors),
+        )

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/app/models/flow.py

@@ -29,6 +29,9 @@ class FlowDefinition(SQLModel, table=True):
     name: str = Field(index=True, unique=True, max_length=200)
     description: str = Field(default="", max_length=2000)
     status: str = Field(default="active", max_length=20)  # active | archived
+    # Filename under backend/flows/ this definition is synced from
+    # (e.g. "simple_rule.json"). None for definitions created via the API.
+    source_path: str | None = Field(default=None, max_length=255)
     # Flow to run after a human approval on this flow's runs (nullable).
     continuation_flow_id: UUID | None = Field(
         default=None, foreign_key="flow_definitions.id"

backend/app/settings.py

@@ -13,10 +13,13 @@ class Settings(BaseSettings):
     version: str = "0.1.0"
     debug: bool = False
 
-    database_url: str = "postgresql+psycopg://kirche:kirche@localhost:5432/kirche_einfach"
+    # App database (SQLite file). Override via DATABASE_URL for other engines.
-    test_database_url: str = (
+    database_url: str = str(REPO_ROOT / "backend" / "data" / "kirche.db")
-        "postgresql+psycopg://kirche:kirche@localhost:5432/kirche_einfach_test"
+
-    )
+    # Canonical flow definitions (source of truth): one Langflow export per file.
+    flows_dir: str = str(REPO_ROOT / "backend" / "flows")
+    # Auto-reconcile backend/flows into the DB + LFX while the API is running.
+    flows_watch_enabled: bool = True
 
     langflow_url: str = "http://localhost:7860"
     langflow_api_key: str = ""

backend/migrations/versions/d932b6993f9a_add_source_path_to_flow_definitions.py

@@ -0,0 +1,32 @@
+"""add source_path to flow_definitions
+
+Revision ID: d932b6993f9a
+Revises: 62ea9a5d8f9d
+Create Date: 2026-09-07 08:40:48.986026
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel  # noqa: F401 - autogenerate renders sqlmodel.* column types
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'd932b6993f9a'
+down_revision: Union[str, Sequence[str], None] = '62ea9a5d8f9d'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    op.add_column(
+        "flow_definitions",
+        sa.Column("source_path", sa.String(length=255), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    op.drop_column("flow_definitions", "source_path")

backend/pyproject.toml

@@ -7,10 +7,10 @@ dependencies = [
     "alembic>=1.14",
     "fastapi[standard]>=0.115.0",
     "httpx>=0.28",
-    "psycopg[binary]>=3.2",
     "pydantic-settings>=2.0",
     "sqlmodel>=0.0.24",
     "uvicorn[standard]>=0.30.0",
+    "watchfiles>=0.24",
 ]
 
 [build-system]
@@ -27,7 +27,7 @@ testpaths = ["tests"]
 
 [dependency-groups]
 dev = [
-    "lfx==1.12.0",
+    "lfx",
     "pytest>=8",
     "pytest-asyncio>=0.24",
     "pytest-cov>=6",

backend/scripts/make_flow_fixtures.py

@@ -6,6 +6,10 @@ Langflow UI would export. Run with the backend venv (has the `lfx` dev dep)::
 
     uv run python scripts/make_flow_fixtures.py
 
+Writes to two places:
+- backend/tests/fixtures/flows/ (test fixtures, never deployed as-is)
+- backend/flows/ (canonical source of truth, synced into DB + LFX)
+
 Then verify::
 
     uv run lfx validate --level 4 backend/tests/fixtures/flows/
@@ -21,6 +25,7 @@ from pathlib import Path
 
 REPO_ROOT = Path(__file__).resolve().parent.parent.parent
 FIXTURE_DIR = REPO_ROOT / "backend" / "tests" / "fixtures" / "flows"
+FLOW_DIR = REPO_ROOT / "backend" / "flows"
 
 HANDLE_QUOTE = "\u0153"  # ReactFlow handle-id quoting used by Langflow exports
 
@@ -430,14 +435,16 @@ def build_approval_handoff() -> dict:
 
 def main() -> None:
     FIXTURE_DIR.mkdir(parents=True, exist_ok=True)
+    FLOW_DIR.mkdir(parents=True, exist_ok=True)
     flows = {
         "simple_rule.json": build_simple_rule(),
         "approval_handoff.json": build_approval_handoff(),
     }
     for filename, doc in flows.items():
-        path = FIXTURE_DIR / filename
+        payload = json.dumps(doc, indent=2, ensure_ascii=False) + "\n"
-        path.write_text(json.dumps(doc, indent=2, ensure_ascii=False) + "\n")
+        for target in (FIXTURE_DIR / filename, FLOW_DIR / filename):
-        print(f"wrote {path}")
+            target.write_text(payload)
+            print(f"wrote {target}")
 
 
 if __name__ == "__main__":

backend/scripts/smoke_e2e.py

@@ -1,7 +1,8 @@
 """Manual end-to-end smoke: real LFX + real API, frontend-shaped calls.
 
 Usage (from backend/):
-    TEST_DATABASE_URL=postgresql+psycopg://... uv run python scripts/smoke_e2e.py
+    uv run python scripts/smoke_e2e.py
+Runs the API against a throwaway SQLite file and an ephemeral LFX pass.
 Exits non-zero on any unexpected status/body.
 """
 
@@ -11,7 +12,7 @@ import json
 import os
 import subprocess
 import sys
-import time
+import tempfile
 import urllib.request
 from pathlib import Path
 
@@ -54,17 +55,31 @@ def check(name: str, cond: bool, detail: str = "") -> None:
 
 
 def main() -> None:
+    tmp = Path(tempfile.mkdtemp(prefix="kirche-smoke-"))
     flow_dir = Path("/tmp/smoke-e2e-flows")
     flow_dir.mkdir(exist_ok=True)
     (flow_dir / "seed.json").write_text(
         (FIXTURES / "simple_rule.json").read_text()
     )
+    # Migrate a throwaway SQLite schema to head before starting the API.
+    from alembic import command as alembic_command
+    from alembic.config import Config
+
+    db_url = f"sqlite:///{tmp / 'smoke.db'}"
+    cfg = Config()
+    cfg.set_main_option("script_location", str(BACKEND / "migrations"))
+    cfg.set_main_option("sqlalchemy.url", db_url)
+    alembic_command.upgrade(cfg, "head")
+
     env = {
         **os.environ,
+        # lfx prefers LANGFLOW_API_KEY over LFX_API_KEY; the repo .env may set
+        # it. Clear it so the serve key below is the one we send.
+        "LANGFLOW_API_KEY": "",
         "LFX_API_KEY": LFX_KEY,
-        "DATABASE_URL": os.environ["TEST_DATABASE_URL"],
+        "DATABASE_URL": db_url,
+        "FLOWS_WATCH_ENABLED": "false",
         "LFX_URL": f"http://127.0.0.1:{LFX_PORT}",
-        "LFX_API_KEY": LFX_KEY,
         "CALLBACK_BASE_URL": f"http://127.0.0.1:{API_PORT}",
     }
     lfx = subprocess.Popen(

backend/scripts/sync_flows.py

@@ -0,0 +1,89 @@
+"""Reconcile backend/flows/*.json into the DB + LFX.
+
+Usage (from backend/, venv active):
+    uv run python scripts/sync_flows.py --sync [--dir backend/flows]
+    uv run python scripts/sync_flows.py --pull [--sync]  # export all Langflow flows to files first
+
+Exit codes: 0 ok, 1 sync errors, 2 flows dir missing.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import sys
+from pathlib import Path
+
+from sqlmodel import Session
+
+from app.db import get_engine, reset_engine
+from app.flows import export, sync
+from app.flows.langflow_client import LangflowClient
+from app.flows.lfx_client import LfxClient
+from app.flows.schemas import SyncReport
+from app.settings import settings
+
+
+async def pull_from_langflow(flows_dir: Path) -> dict:
+    """Export every Langflow flow to its own canonical file. Returns a report."""
+    client = LangflowClient(settings.langflow_url, api_key=settings.langflow_api_key)
+    headers = await client.list_flows()
+    pulled: list[str] = []
+    errors: dict[str, str] = {}
+    for header in headers:
+        flow_id = str(header.get("id", ""))
+        try:
+            doc = await client.export_flow(flow_id)
+            name = doc.get("name") or flow_id
+            path = export.write_flow_file(flows_dir, doc, name=name)
+            pulled.append(path.stem)
+        except Exception as exc:  # noqa: BLE001 - pull-all is total
+            errors[flow_id] = str(exc)
+    return {"pulled": sorted(pulled), "errors": errors}
+
+
+async def run_sync(flows_dir: Path) -> SyncReport:
+    reset_engine()
+    session = Session(get_engine())
+    try:
+        lfx = LfxClient(settings.lfx_url, api_key=settings.lfx_api_key or None)
+        return await sync.sync_from_directory(session, lfx, flows_dir)
+    finally:
+        session.close()
+
+
+async def main_async(args) -> int:
+    flows_dir = Path(args.dir or settings.flows_dir)
+    if args.pull:
+        pulled = await pull_from_langflow(flows_dir)
+        print(json.dumps(pulled, indent=2, ensure_ascii=False))
+        if pulled["errors"]:
+            return 1
+    if args.sync or not args.pull:
+        try:
+            report = await run_sync(flows_dir)
+        except FileNotFoundError as exc:
+            print(f"ERROR: {exc}", file=sys.stderr)
+            return 2
+        print(report.model_dump_json(indent=2))
+        return 1 if report.errors else 0
+    return 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--pull", action="store_true",
+                        help="export all Langflow flows to backend/flows first")
+    parser.add_argument("--sync", action="store_true",
+                        help="reconcile backend/flows into DB + LFX")
+    parser.add_argument("--dir", default=None,
+                        help="flows directory (default: settings.flows_dir)")
+    args = parser.parse_args()
+    if not args.pull and not args.sync:
+        args.sync = True
+    return asyncio.run(main_async(args))
+
+
+if __name__ == "__main__":
+    sys.exit(main())

backend/tests/conftest.py

@@ -1,25 +1,25 @@
 """Shared pytest fixtures.
 
-Requires TEST_DATABASE_URL (a dedicated Postgres database, e.g. the
+Tests run on a throwaway SQLite file (no external database required).
-`kirche_einfach_test` database from the compose stack). The URL is mapped
+`DATABASE_URL` and `FLOWS_DIR` are pointed at a session temp dir *before* any
-onto DATABASE_URL *before* any app module is imported, so settings and the
+app module is imported, so settings pick them up at import time. The app
-alembic env both target the test database. Never point this at production.
+schema is migrated to head once per session via Alembic.
 """
 
 import os
+import tempfile
 from pathlib import Path
 
 import pytest
 
-TEST_DB_URL = os.environ.get("TEST_DATABASE_URL")
+TEST_AREA = Path(tempfile.mkdtemp(prefix="kirche-tests-"))
-if not TEST_DB_URL:
-    raise RuntimeError(
-        "TEST_DATABASE_URL is not set. Refusing to run tests without "
-        "an explicit test database (see .env.example)."
-    )
 
 # Must precede all `app.*` imports: Settings is instantiated at import time.
-os.environ["DATABASE_URL"] = TEST_DB_URL
+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
 
@@ -115,6 +115,10 @@ def lfx_server(tmp_path_factory):
     env = {
         **os.environ,
         # lfx serve refuses to start without a serve key; requests need it too.
+        # get_api_key() prefers LANGFLOW_API_KEY over LFX_API_KEY, and the repo
+        # .env may put a real Langflow key in the environment — clear it so the
+        # test key below is the one the client sends.
+        "LANGFLOW_API_KEY": "",
         "LFX_API_KEY": "test-lfx-key",
     }
     proc = subprocess.Popen(

backend/tests/test_deploy.py

@@ -2,7 +2,9 @@
 
 import json
 from pathlib import Path
-from uuid import uuid4
+from uuid import UUID, uuid4
+
+from app.flows.sync import lfx_id_for
 
 FIXTURES = Path(__file__).parent / "fixtures" / "flows"
 LANGFLOW_ID = "11111111-2222-4333-8444-555555555555"
@@ -35,7 +37,9 @@ async def test_import_creates_definition_version_and_deploys(
     assert body["definition"]["name"] == "importiert"
     assert body["version"]["version_no"] == 1
     assert body["version"]["author"] == "admin"
-    assert body["version"]["lfx_flow_id"] == body["version"]["id"]
+    assert body["version"]["lfx_flow_id"] == lfx_id_for(
+        UUID(body["definition"]["id"]), 1
+    )
     assert body["version"]["deployed_at"] is not None
     assert body["version"]["langflow_json"]["data"]["nodes"]
 
@@ -48,7 +52,7 @@ async def test_import_creates_definition_version_and_deploys(
     assert response.json()["status"] == "completed"
 
 
-async def test_import_reuses_definition_and_bumps_version(
+async def test_import_same_content_is_idempotent(
     client, lfx_server, override_langflow
 ):
     override_langflow(_export_doc())
@@ -61,9 +65,44 @@ async def test_import_reuses_definition_and_bumps_version(
     assert first.status_code == 201
     assert second.status_code == 201
     assert first.json()["definition"]["id"] == second.json()["definition"]["id"]
+    # Unchanged content creates no new version.
+    assert second.json()["version"]["version_no"] == 1
+
+
+async def test_import_changed_content_bumps_version(
+    client, lfx_server, override_langflow
+):
+    override_langflow(_export_doc())
+    first = await client.post(
+        "/api/flows/import", json={"langflow_flow_id": LANGFLOW_ID}
+    )
+    assert first.status_code == 201
+    changed = _export_doc()
+    changed["description"] = "neue Beschreibung"
+    override_langflow(changed)
+    second = await client.post(
+        "/api/flows/import", json={"langflow_flow_id": LANGFLOW_ID}
+    )
+    assert second.status_code == 201
     assert second.json()["version"]["version_no"] == 2
 
 
+async def test_import_writes_canonical_file(
+    client, lfx_server, override_langflow, tmp_path, monkeypatch
+):
+    from app.settings import settings
+
+    monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
+    override_langflow(_export_doc())
+    response = await client.post(
+        "/api/flows/import", json={"langflow_flow_id": LANGFLOW_ID}
+    )
+    assert response.status_code == 201, response.text
+    written = tmp_path / "importiert.json"
+    assert written.is_file()
+    assert json.loads(written.read_text())["name"] == "importiert"
+
+
 async def test_import_with_explicit_name(client, lfx_server, override_langflow):
     override_langflow(_export_doc(name="langflow-name"))
     response = await client.post(

backend/tests/test_export.py

@@ -0,0 +1,39 @@
+"""Unit tests for canonical flow files (export.py). No DB or LFX needed."""
+
+import json
+
+from app.flows import export
+
+
+def test_slugify():
+    assert export.slugify("Dachsanierung 200k") == "dachsanierung-200k"
+    assert export.slugify("  A/B\\C  ") == "a-b-c"
+    assert export.slugify("simple_rule") == "simple-rule"
+    assert export.slugify("") == "flow"
+
+
+def test_canonical_bytes_stable_under_key_reorder():
+    doc = {"name": "x", "data": {"nodes": [], "edges": []}, "tags": ["b", "a"]}
+    shuffled = {"tags": ["b", "a"], "data": {"edges": [], "nodes": []}, "name": "x"}
+    assert export.canonical_bytes(shuffled) == export.canonical_bytes(doc)
+    assert export.content_hash(doc) == export.content_hash(shuffled)
+
+
+def test_write_and_read_roundtrip_is_byte_stable(tmp_path):
+    doc = {"name": "My Flow", "data": {"nodes": [], "edges": []}}
+    path = export.write_flow_file(tmp_path, doc)
+    assert path == tmp_path / "my-flow.json"
+    assert json.loads(path.read_text()) == doc
+    # Rewriting identical content is byte-identical (no phantom git diffs).
+    before = path.read_bytes()
+    export.write_flow_file(tmp_path, doc)
+    assert path.read_bytes() == before
+
+
+def test_iter_flow_files_only_top_level_json(tmp_path):
+    (tmp_path / "a.json").write_text("{}")
+    (tmp_path / "b.txt").write_text("{}")
+    sub = tmp_path / "sub"
+    sub.mkdir()
+    (sub / "c.json").write_text("{}")
+    assert export.iter_flow_files(tmp_path) == [tmp_path / "a.json"]

backend/tests/test_flows_api.py

@@ -2,7 +2,9 @@
 
 import json
 from pathlib import Path
-from uuid import uuid4
+from uuid import UUID, uuid4
+
+from app.flows.sync import lfx_id_for
 
 FIXTURES = Path(__file__).parent / "fixtures" / "flows"
 
@@ -81,7 +83,7 @@ async def test_deploy_and_run_standard_and_sonderfall(client, lfx_server):
     flow = await _create_flow(client, "dachsanierung")
     await _create_version(client, flow["id"])
     deployed = await _deploy(client, flow["id"])
-    assert deployed["lfx_flow_id"] == deployed["id"]
+    assert deployed["lfx_flow_id"] == lfx_id_for(UUID(flow["id"]), 1)
     assert deployed["deployed_at"] is not None
 
     response = await client.post(

backend/tests/test_models.py

@@ -16,7 +16,9 @@ def test_flow_definition_defaults(session):
     assert d.status == "active"
     assert d.continuation_flow_id is None
     assert d.created_at is not None
-    assert d.created_at.tzinfo is not None
+    # Timestamps are written as UTC. SQLite does not round-trip tzinfo, so
+    # after a refresh the value is calendar-time UTC without tzinfo.
+    assert d.created_at.year >= 2026
 
 
 def test_flow_definition_name_unique(session):

backend/tests/test_sync.py

@@ -0,0 +1,157 @@
+"""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
+
+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
+
+FIXTURES = Path(__file__).parent / "fixtures" / "flows"
+
+
+def load_fixture(name: str) -> dict:
+    return json.loads((FIXTURES / name).read_text())
+
+
+def _lfx() -> LfxClient:
+    return LfxClient(settings.lfx_url, api_key=settings.lfx_api_key or None)
+
+
+def _seed_dir(tmp_path: Path, name: str = "simple_rule.json") -> Path:
+    shutil.copy(FIXTURES / name, tmp_path / name)
+    return tmp_path
+
+
+async def test_sync_adds_new_flow(session, lfx_server, tmp_path):
+    _seed_dir(tmp_path)
+    report = await sync.sync_from_directory(session, _lfx(), tmp_path)
+    assert report.added == ["simple_rule"]
+    assert report.errors == {}
+    definition = store.get_definition_by_source_path(session, "simple_rule.json")
+    assert definition is not None
+    assert definition.name == "simple_rule"
+    versions = store.list_versions(session, definition.id)
+    assert [v.version_no for v in versions] == [1]
+    assert versions[0].lfx_flow_id == lfx_id_for(definition.id, 1)
+    assert versions[0].deployed_at is not None
+
+
+async def test_sync_skips_unchanged(session, lfx_server, tmp_path):
+    _seed_dir(tmp_path)
+    first = await sync.sync_from_directory(session, _lfx(), tmp_path)
+    assert first.added == ["simple_rule"]
+    second = await sync.sync_from_directory(session, _lfx(), tmp_path)
+    assert second.skipped == ["simple_rule"]
+    assert second.added == second.updated == []
+    definition = store.get_definition_by_source_path(session, "simple_rule.json")
+    assert len(store.list_versions(session, definition.id)) == 1
+
+
+async def test_sync_updates_on_change_and_replaces_lfx(session, lfx_server, tmp_path):
+    _seed_dir(tmp_path)
+    await sync.sync_from_directory(session, _lfx(), tmp_path)
+    definition = store.get_definition_by_source_path(session, "simple_rule.json")
+    old_lfx_id = lfx_id_for(definition.id, 1)
+
+    doc = load_fixture("simple_rule.json")
+    doc["description"] = "geaendert"
+    (tmp_path / "simple_rule.json").write_text(json.dumps(doc))
+
+    report = await sync.sync_from_directory(session, _lfx(), tmp_path)
+    assert report.updated == ["simple_rule"]
+    versions = store.list_versions(session, definition.id)
+    assert [v.version_no for v in versions] == [1, 2]
+    assert versions[1].lfx_flow_id == lfx_id_for(definition.id, 2)
+    # Superseded LFX registration is gone.
+    with pytest.raises(LfxNotFound):
+        await _lfx().flow_info(old_lfx_id)
+
+
+async def test_sync_deletes_missing_file_with_cascade(session, lfx_server, tmp_path):
+    _seed_dir(tmp_path)
+    await sync.sync_from_directory(session, _lfx(), tmp_path)
+    definition = store.get_definition_by_source_path(session, "simple_rule.json")
+    version = store.list_versions(session, definition.id)[0]
+    run = store.create_run(session, version, input={"a": 1}, tenant_id="default")
+    lfx_id = lfx_id_for(definition.id, 1)
+
+    (tmp_path / "simple_rule.json").unlink()
+    report = await sync.sync_from_directory(session, _lfx(), tmp_path)
+    assert report.deleted == ["simple_rule"]
+    assert store.get_definition_by_source_path(session, "simple_rule.json") is None
+    with pytest.raises(store.NotFoundError):
+        store.get_run(session, run.id)
+    with pytest.raises(LfxNotFound):
+        await _lfx().flow_info(lfx_id)
+
+
+async def test_sync_reports_invalid_json(session, lfx_server, tmp_path):
+    (tmp_path / "kaputt.json").write_text("{kein json")
+    report = await sync.sync_from_directory(session, _lfx(), tmp_path)
+    assert "kaputt.json" in report.errors
+    assert report.added == []
+    assert store.list_definitions(session) == []
+
+
+async def test_sync_missing_dir_raises(session, lfx_server, tmp_path):
+    with pytest.raises(FileNotFoundError):
+        await sync.sync_from_directory(session, _lfx(), tmp_path / "gibts-nicht")
+
+
+async def test_sync_rename_updates_definition_name(session, lfx_server, tmp_path):
+    _seed_dir(tmp_path)
+    await sync.sync_from_directory(session, _lfx(), tmp_path)
+    doc = load_fixture("simple_rule.json")
+    doc["name"] = "umbenannt"
+    (tmp_path / "simple_rule.json").write_text(json.dumps(doc))
+    report = await sync.sync_from_directory(session, _lfx(), tmp_path)
+    assert report.updated == ["simple_rule"]
+    definition = store.get_definition_by_source_path(session, "simple_rule.json")
+    assert definition.name == "umbenannt"
+
+
+async def test_sync_endpoint_returns_report(client, lfx_server, tmp_path, monkeypatch):
+    monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
+    _seed_dir(tmp_path)
+    response = await client.post("/api/flows/sync")
+    assert response.status_code == 200, response.text
+    body = response.json()
+    assert body["added"] == ["simple_rule"]
+    assert body["errors"] == {}
+
+    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

compose.prod.yml

@@ -0,0 +1,122 @@
+# Minimal self-hosted deployment (internal use: no TLS / reverse proxy).
+#
+# Flows are the bind-mounted ./backend/flows directory (source of truth);
+# the embedded API watcher reconciles them into the DB + LFX automatically.
+# Registry images (:latest) auto-update nightly via watchtower; locally
+# built images (app, lfx, frontend) update via `docker compose build --pull`.
+#
+# Usage: docker compose -f compose.prod.yml up -d --build
+services:
+  app:
+    build:
+      context: ./backend
+      dockerfile: Dockerfile
+    env_file:
+      - .env
+    environment:
+      DATABASE_URL: sqlite:////srv/backend/data/kirche.db
+      FLOWS_DIR: /srv/backend/flows
+      LFX_URL: http://lfx:6004
+      LANGFLOW_URL: http://langflow:7860
+      CALLBACK_BASE_URL: http://app:8000
+    volumes:
+      - ./backend/flows:/srv/backend/flows
+      - appdata:/srv/backend/data
+    ports:
+      - "8000:8000"
+    depends_on:
+      lfx:
+        condition: service_healthy
+      langflow:
+        condition: service_healthy
+    restart: unless-stopped
+    healthcheck:
+      test:
+        [
+          "CMD-SHELL",
+          "python -c \"import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=5)\"",
+        ]
+      interval: 30s
+      timeout: 5s
+      retries: 3
+      start_period: 30s
+
+  frontend:
+    build:
+      context: ./frontend
+      dockerfile: Dockerfile
+    ports:
+      - "3000:3000"
+    depends_on:
+      app:
+        condition: service_healthy
+    restart: unless-stopped
+
+  langflow:
+    image: langflowai/langflow:latest
+    env_file:
+      - .env
+    environment:
+      LANGFLOW_CONFIG_DIR: /app/langflow
+      LANGFLOW_HOST: 0.0.0.0
+      PORT: "7860"
+      LANGFLOW_AUTO_LOGIN: "false"
+      LANGFLOW_API_KEY_SOURCE: env
+      LANGFLOW_DATABASE_URL: sqlite:////app/langflow/langflow.db
+    volumes:
+      - langflow-data:/app/langflow
+    ports:
+      - "7860:7860"
+    labels:
+      com.centurylinklabs.watchtower.enable: "true"
+    restart: unless-stopped
+    healthcheck:
+      test:
+        [
+          "CMD-SHELL",
+          "python -c \"import urllib.request; urllib.request.urlopen('http://localhost:7860/health', timeout=5)\"",
+        ]
+      interval: 30s
+      timeout: 5s
+      retries: 20
+      start_period: 60s
+
+  lfx:
+    build:
+      context: .devcontainer
+      dockerfile: Dockerfile.lfx
+    env_file:
+      - .env
+    volumes:
+      - lfx-data:/flows
+    ports:
+      - "6004:6004"
+    restart: unless-stopped
+    healthcheck:
+      test:
+        [
+          "CMD-SHELL",
+          "python -c \"import urllib.request; urllib.request.urlopen('http://localhost:6004/health', timeout=5)\"",
+        ]
+      interval: 30s
+      timeout: 5s
+      retries: 12
+      start_period: 30s
+
+  watchtower:
+    image: nickfedor/watchtower:latest
+    volumes:
+      - /var/run/docker.sock:/var/run/docker.sock
+    environment:
+      WATCHTOWER_LABEL_ENABLE: "true"
+      WATCHTOWER_SCHEDULE: "0 0 4 * * *"
+      WATCHTOWER_CLEANUP: "true"
+      WATCHTOWER_ROLLING_RESTART: "true"
+    labels:
+      com.centurylinklabs.watchtower.enable: "true"
+    restart: unless-stopped
+
+volumes:
+  appdata:
+  langflow-data:
+  lfx-data:

frontend/Dockerfile

@@ -0,0 +1,13 @@
+# Production image for the React UI (build + Vite preview; internal use).
+FROM node:22-slim
+
+WORKDIR /srv/frontend
+COPY package.json package-lock.json ./
+RUN npm ci
+COPY . ./
+RUN npm run build
+
+ENV VITE_API_PROXY_TARGET=http://app:8000
+EXPOSE 3000
+
+CMD ["npm", "run", "preview", "--", "--host", "0.0.0.0", "--port", "3000"]

frontend/src/api.ts

@@ -5,6 +5,7 @@ export interface FlowDefinition {
   name: string;
   description: string;
   status: string;
+  source_path: string | null;
   continuation_flow_id: string | null;
   created_at: string;
 }
@@ -60,6 +61,14 @@ export interface Decision {
   continuation_run: Run | null;
 }
 
+export interface SyncReport {
+  added: string[];
+  updated: string[];
+  deleted: string[];
+  skipped: string[];
+  errors: Record<string, string>;
+}
+
 export class ApiError extends Error {
   status: number;
   constructor(status: number, message: string) {
@@ -135,4 +144,8 @@ export const api = {
     ),
   listLangflow: () =>
     req<{ id: string; name: string }[]>("/api/flows/langflow/available"),
+  syncFlows: () =>
+    req<SyncReport>("/api/flows/sync", {
+      method: "POST",
+    }),
 };

frontend/src/components/AdminTab.tsx

@@ -64,6 +64,26 @@ export default function AdminTab() {
     }
   }
 
+  async function onSync() {
+    setError("");
+    setInfo("");
+    try {
+      const r = await api.syncFlows();
+      const parts = [
+        `+${r.added.length}`,
+        `~${r.updated.length}`,
+        `-${r.deleted.length}`,
+      ].join(" ");
+      const errs = Object.entries(r.errors)
+        .map(([f, e]) => `${f}: ${e}`)
+        .join("; ");
+      setInfo(`Sync ${parts}${errs ? ` — Fehler: ${errs}` : ""}`);
+      await load();
+    } catch (e) {
+      setError(e instanceof ApiError ? `${e.status}: ${e.message}` : String(e));
+    }
+  }
+
   return (
     <div>
       <h2>Administration</h2>
@@ -93,11 +113,19 @@ export default function AdminTab() {
         </button>
       </div>
       <h3>Flows + Versionen</h3>
+      <div style={{ marginBottom: 12 }}>
+        <button onClick={() => void onSync()}>
+          Aus backend/flows synchronisieren
+        </button>
+      </div>
       {flows.length === 0 && <p>Keine Flows vorhanden.</p>}
       <ul>
         {flows.map((f) => (
           <li key={f.id} style={{ marginBottom: 8 }}>
-            <strong>{f.name}</strong>
+            <strong>{f.name}</strong>{" "}
+            <span style={{ fontFamily: "monospace", fontSize: "0.85em" }}>
+              {f.source_path ? `(${f.source_path})` : "(API)"}
+            </span>
             <ul>
               {(versions[f.id] ?? []).map((v) => (
                 <li key={v.id}>

frontend/vite.config.ts

@@ -12,4 +12,14 @@ export default defineConfig({
       },
     },
   },
+  preview: {
+    port: 3000,
+    proxy: {
+      "/api": {
+        // Overridable at container start (prod compose: http://app:8000).
+        target: process.env.VITE_API_PROXY_TARGET ?? "http://localhost:8000",
+        changeOrigin: true,
+      },
+    },
+  },
 });