.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 requests
+RUN pip install "lfx==1.12.0" requests
 
 RUN mkdir -p /flows
 VOLUME /flows

.devcontainer/docker-compose.yml

@@ -19,8 +19,26 @@ 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:latest
+    image: langflowai/langflow:1.12.0
+    depends_on:
+      db:
+        condition: service_healthy
     env_file:
       - ../.env
     environment:
@@ -31,9 +49,6 @@ 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:
@@ -70,23 +85,9 @@ 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

@@ -0,0 +1,7 @@
+-- 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 (langflow, lfx) must be up; fail loudly otherwise.
+# Compose services (db, 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 = [("langflow", 7860), ("lfx", 6004)]
+targets = [("db", 5432), ("langflow", 7860), ("lfx", 6004)]
 deadline = time.time() + 180
 pending = dict(targets)
 while pending and time.time() < deadline:
@@ -45,14 +45,6 @@ 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,14 +1,12 @@
 # Copy to `.env` and fill in values. `.env` is gitignored.
 FORGEJO_TOKEN=
 
-# --- app database (SQLite, default if unset) ---
+# --- databases (compose service `db`) ---
-# Override only if you want a different engine/URL.
+POSTGRES_USER=kirche
-# DATABASE_URL=sqlite:///backend/data/kirche.db
+POSTGRES_PASSWORD=
-
+POSTGRES_DB=kirche_einfach
-# --- canonical flow definitions ---
+DATABASE_URL=postgresql+psycopg://kirche:<password>@db:5432/kirche_einfach
-# FLOWS_DIR=backend/flows
+TEST_DATABASE_URL=postgresql+psycopg://kirche:<password>@db:5432/kirche_einfach_test
-# Auto-reconcile backend/flows into DB + LFX while the API runs.
-# FLOWS_WATCH_ENABLED=true
 
 # --- langflow (authoring, dev only) ---
 LANGFLOW_URL=http://langflow:7860
@@ -17,7 +15,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 stores flows in its own SQLite file when unset.
+LANGFLOW_DATABASE_URL=postgresql://kirche:<password>@db:5432/langflow
 
 # --- lfx (flow runtime) ---
 LFX_URL=http://lfx:6004

.gitignore

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

AGENTS.md

@@ -1,44 +1,5 @@
 # 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>"` (conventional commits: `feat:`, `fix:`, `chore:`, `docs:`). When pushing: `git push` (branch is tracking `origin/main`).
+When committing: `git add -A && git commit -m "<message>"`. When pushing: `git push` (branch is tracking `origin/main`).

README.md

@@ -1,103 +1,17 @@
 # Kirche Einfach
 
-AI-assisted administration platform for church organizations.
+KI-gestützte Verwaltungsplattform für Kirche.
 
-Staff describe a request in plain words. The platform guides them through the
+## Stack
-process, checks the rules automatically, and asks a human only when something
-needs judgment. Every step is documented and traceable.
 
-## Features
+- **Backend:** Python 3.12, FastAPI, Django, uv
-
+- **Frontend:** React 19, TypeScript, Vite
-- **Visual flow authoring** — design decision flows in Langflow, then bring
+- **Dev:** Devcontainer, opencode
-  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
 
-1. Copy `.env.example` to `.env` and fill in the secrets.
+Open in VS Code → "Reopen in Container".
-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`.
 
-Production (self-hosted, internal use — no TLS/reverse proxy included):
+Backend: `http://localhost:8000`
-
+Frontend: `http://localhost:3000`
-```sh
+API Docs: `http://localhost:8000/api/docs`
-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

@@ -1,21 +0,0 @@
-# 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,7 +1,6 @@
 """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
@@ -11,23 +10,11 @@ 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:
-        url = url or settings.database_url
+        _engine = create_engine(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,7 +4,6 @@ from __future__ import annotations
 
 import json
 from collections.abc import AsyncIterator, Iterator
-from pathlib import Path
 from uuid import UUID
 
 import httpx
@@ -17,7 +16,6 @@ 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,
@@ -46,7 +44,6 @@ from .schemas import (
     FlowVersionRead,
     RunRead,
     RunRequest,
-    SyncReport,
 )
 
 flows_router = APIRouter(prefix="/api/flows", tags=["flows"])
@@ -73,7 +70,6 @@ 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,
     )
@@ -138,18 +134,6 @@ 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:
@@ -194,12 +178,8 @@ 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,
+            version.langflow_json, flow_id=str(version.id), replace=True
-            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 -> canonical file -> version -> LFX deploy."""
+"""Authoring pipeline: Langflow export -> versioned definition -> LFX deploy."""
 
 from __future__ import annotations
 
 from sqlmodel import Session
 
 from app.models import FlowDefinition, FlowVersion
-from app.settings import settings
 
-from . import export, store, sync
+from . import store
 from .langflow_client import LangflowClient
 from .lfx_client import LfxClient
+from .schemas import FlowDefinitionCreate, FlowVersionCreate
 
 
 async def import_from_langflow(
@@ -22,31 +22,39 @@ async def import_from_langflow(
     author: str = "",
     deploy: bool = True,
 ) -> tuple[FlowDefinition, FlowVersion]:
-    """Export a flow from Langflow, persist it to backend/flows, then sync.
+    """Export a flow from Langflow, store it as a new version, deploy to LFX.
 
-    The canonical file is written first so the directory stays the source of
+    Definitions are matched by name (explicit `name` wins, else the exported
-    truth; sync_document() derives the definition + version from it.
+    name); every import creates a new immutable version.
     """
-    exported = await langflow.export_flow(langflow_flow_id)
+    export = await langflow.export_flow(langflow_flow_id)
-    flow_name = name or exported.get("name") or f"langflow-{langflow_flow_id}"
+    flow_name = name or export.get("name") or f"langflow-{langflow_flow_id}"
     langflow_json = {
-        "id": exported.get("id", langflow_flow_id),
+        "id": export.get("id", langflow_flow_id),
         "name": flow_name,
-        "description": exported.get("description") or "",
+        "description": export.get("description") or "",
-        "endpoint_name": exported.get("endpoint_name"),
+        "endpoint_name": export.get("endpoint_name"),
-        "data": exported.get("data", {}),
+        "data": export.get("data", {}),
-        "tags": exported.get("tags", []),
+        "tags": export.get("tags", []),
-        "webhook": exported.get("webhook", False),
+        "webhook": export.get("webhook", False),
-        "is_component": exported.get("is_component", False),
+        "is_component": export.get("is_component", False),
     }
-    path = export.write_flow_file(settings.flows_dir, langflow_json, name=flow_name)
+    try:
-    definition, version, _action = await sync.sync_document(
+        definition = next(
+            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,
-        lfx,
+        definition.id,
-        langflow_json,
+        FlowVersionCreate(langflow_json=langflow_json, author=author),
-        source_filename=path.name,
-        author=author,
-        deploy=deploy,
     )
-    assert version is not None  # sync_document always versions on import
+    if deploy:
+        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

@@ -1,71 +0,0 @@
-"""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,8 +17,6 @@ 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
 
@@ -112,16 +110,3 @@ 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,82 +56,7 @@ def list_definitions(session: Session) -> list[FlowDefinition]:
     return list(session.exec(select(FlowDefinition).order_by(FlowDefinition.name)).all())
 
 
-def get_definition_by_source_path(
+def _validate_flow_json(flow_json: dict) -> None:
-    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")
@@ -145,7 +70,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

@@ -1,274 +0,0 @@
-"""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,58 +1,8 @@
-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",
@@ -60,7 +10,6 @@ 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,9 +29,6 @@ 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,13 +13,10 @@ class Settings(BaseSettings):
     version: str = "0.1.0"
     debug: bool = False
 
-    # App database (SQLite file). Override via DATABASE_URL for other engines.
+    database_url: str = "postgresql+psycopg://kirche:kirche@localhost:5432/kirche_einfach"
-    database_url: str = str(REPO_ROOT / "backend" / "data" / "kirche.db")
+    test_database_url: str = (
-
+        "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

@@ -1,32 +0,0 @@
-"""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",
+    "lfx==1.12.0",
     "pytest>=8",
     "pytest-asyncio>=0.24",
     "pytest-cov>=6",

backend/scripts/make_flow_fixtures.py

@@ -6,10 +6,6 @@ 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/
@@ -25,7 +21,6 @@ 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
 
@@ -435,16 +430,14 @@ 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():
-        payload = json.dumps(doc, indent=2, ensure_ascii=False) + "\n"
+        path = FIXTURE_DIR / filename
-        for target in (FIXTURE_DIR / filename, FLOW_DIR / filename):
+        path.write_text(json.dumps(doc, indent=2, ensure_ascii=False) + "\n")
-            target.write_text(payload)
+        print(f"wrote {path}")
-            print(f"wrote {target}")
 
 
 if __name__ == "__main__":

backend/scripts/smoke_e2e.py

@@ -1,8 +1,7 @@
 """Manual end-to-end smoke: real LFX + real API, frontend-shaped calls.
 
 Usage (from backend/):
-    uv run python scripts/smoke_e2e.py
+    TEST_DATABASE_URL=postgresql+psycopg://... 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.
 """
 
@@ -12,7 +11,7 @@ import json
 import os
 import subprocess
 import sys
-import tempfile
+import time
 import urllib.request
 from pathlib import Path
 
@@ -55,31 +54,17 @@ 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": db_url,
+        "DATABASE_URL": os.environ["TEST_DATABASE_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

@@ -1,89 +0,0 @@
-"""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.
 
-Tests run on a throwaway SQLite file (no external database required).
+Requires TEST_DATABASE_URL (a dedicated Postgres database, e.g. the
-`DATABASE_URL` and `FLOWS_DIR` are pointed at a session temp dir *before* any
+`kirche_einfach_test` database from the compose stack). The URL is mapped
-app module is imported, so settings pick them up at import time. The app
+onto DATABASE_URL *before* any app module is imported, so settings and the
-schema is migrated to head once per session via Alembic.
+alembic env both target the test database. Never point this at production.
 """
 
 import os
-import tempfile
 from pathlib import Path
 
 import pytest
 
-TEST_AREA = Path(tempfile.mkdtemp(prefix="kirche-tests-"))
+TEST_DB_URL = os.environ.get("TEST_DATABASE_URL")
+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"] = f"sqlite:///{TEST_AREA / 'test.db'}"
+os.environ["DATABASE_URL"] = TEST_DB_URL
-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,10 +115,6 @@ 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,9 +2,7 @@
 
 import json
 from pathlib import Path
-from uuid import UUID, uuid4
+from uuid import uuid4
-
-from app.flows.sync import lfx_id_for
 
 FIXTURES = Path(__file__).parent / "fixtures" / "flows"
 LANGFLOW_ID = "11111111-2222-4333-8444-555555555555"
@@ -37,9 +35,7 @@ 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"] == lfx_id_for(
+    assert body["version"]["lfx_flow_id"] == body["version"]["id"]
-        UUID(body["definition"]["id"]), 1
-    )
     assert body["version"]["deployed_at"] is not None
     assert body["version"]["langflow_json"]["data"]["nodes"]
 
@@ -52,7 +48,7 @@ async def test_import_creates_definition_version_and_deploys(
     assert response.json()["status"] == "completed"
 
 
-async def test_import_same_content_is_idempotent(
+async def test_import_reuses_definition_and_bumps_version(
     client, lfx_server, override_langflow
 ):
     override_langflow(_export_doc())
@@ -65,44 +61,9 @@ async def test_import_same_content_is_idempotent(
     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

@@ -1,39 +0,0 @@
-"""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,9 +2,7 @@
 
 import json
 from pathlib import Path
-from uuid import UUID, uuid4
+from uuid import uuid4
-
-from app.flows.sync import lfx_id_for
 
 FIXTURES = Path(__file__).parent / "fixtures" / "flows"
 
@@ -83,7 +81,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"] == lfx_id_for(UUID(flow["id"]), 1)
+    assert deployed["lfx_flow_id"] == deployed["id"]
     assert deployed["deployed_at"] is not None
 
     response = await client.post(

backend/tests/test_models.py

@@ -16,9 +16,7 @@ def test_flow_definition_defaults(session):
     assert d.status == "active"
     assert d.continuation_flow_id is None
     assert d.created_at is not None
-    # Timestamps are written as UTC. SQLite does not round-trip tzinfo, so
+    assert d.created_at.tzinfo is not None
-    # 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

@@ -1,157 +0,0 @@
-"""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

@@ -1,122 +0,0 @@
-# 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

@@ -1,13 +0,0 @@
-# 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,7 +5,6 @@ export interface FlowDefinition {
   name: string;
   description: string;
   status: string;
-  source_path: string | null;
   continuation_flow_id: string | null;
   created_at: string;
 }
@@ -61,14 +60,6 @@ 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) {
@@ -144,8 +135,4 @@ 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,26 +64,6 @@ 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>
@@ -113,19 +93,11 @@ 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,14 +12,4 @@ 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,
-      },
-    },
-  },
 });