chore: LLM provider env for Langflow, persistent sessions, Fernet key doc

.env.example

@@ -17,6 +17,11 @@ LANGFLOW_SUPERUSER=langflow
 # Long random values (check_prod.py enforces length). Never leave the
 # placeholders: the authoring UI is internet-facing.
 LANGFLOW_SUPERUSER_PASSWORD=
+# MUST be a valid Fernet key (44 chars, urlsafe-base64, 32 bytes) — any
+# other value breaks variable decryption ("Fernet key must be 32
+# url-safe base64-encoded bytes") and provider keys cannot be saved.
+# Generate with:
+#   python -c "import base64,os;print(base64.urlsafe_b64encode(os.urandom(32)).decode())"
 LANGFLOW_SECRET_KEY=
 # Single static key; validated against env (LANGFLOW_API_KEY_SOURCE=env), superuser scope.
 LANGFLOW_API_KEY=sk-
@@ -52,3 +57,20 @@ SMTP_PASSWORD=
 
 # --- LLM (optional; tests mock the provider without it) ---
 # LLM_API_KEY=
+
+# --- LLM provider für Langflow (Authoring-UI + LFX-Runtime) ---
+# Langflow erzeugt daraus automatisch Global Variables (Credential-Typ);
+# LFX liest dieselben Werte zur Laufzeit. Eine in der UI angelegte Global
+# Variable mit gleichem Namen hat Vorrang, die Env-Variable ist Fallback.
+# OpenAI-kompatibler Endpoint (auch selbst gehostet); ohne eigenen Provider
+# leer lassen. Beispiel: https://openrouter.ai/api/v1
+OPENAI_API_KEY=
+OPENAI_BASE_URL=
+
+# --- Langflow-Login-Sitzung ---
+# Refresh-Token-Gültigkeit in Sekunden (Default 604800 = 7 Tage).
+# 7776000 = 90 Tage: einmal einloggen, Container-Restarts übersteht die
+# Session, solange LANGFLOW_SECRET_KEY stabil bleibt. In Prod kürzer wählen.
+# LANGFLOW_REFRESH_TOKEN_EXPIRE_SECONDS=7776000
+# Erstellt Global Variables aus Env (Credential-Typ), restart-fest:
+# LANGFLOW_VARIABLES_TO_GET_FROM_ENVIRONMENT=OPENAI_API_KEY,OPENAI_BASE_URL

CHANGELOG.md

@@ -5,7 +5,37 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+### Added
+- Flow-Sync ist jetzt ein manueller Full-Pass: `POST /api/flows/sync`
+  (Button "Aus Langflow synchronisieren" in der Verwaltung) holt erst alle
+  Flows aus Langflow nach `backend/flows/*.json` und deployed sie danach
+  automatisch (Report mit `pulled`-Liste). Der Hintergrund-Beobachter und der
+  Startup-Sync bleiben dateibasiert. Die Verwaltung zeigt eine deutsche
+  Schritt-Anleitung direkt beim Sync-Button.
+- LLM-Provider-Umgebung für Langflow: `OPENAI_API_KEY` + `OPENAI_BASE_URL`
+  in `.env` (OpenAI-kompatible Endpoints) erzeugen automatisch
+  restart-feste Global Variables in Langflow; LFX liest dieselben Werte.
+- Langflow-Sitzungen halten länger: `LANGFLOW_REFRESH_TOKEN_EXPIRE_SECONDS`
+  (Default weiter 7 Tage, Dev-Beispiel 90) + stabile JWT-Signatur via
+  `LANGFLOW_SECRET_KEY` — kein Re-Login nach Container-Restarts.
 ### Fixed
+- Flow-Sync aus Langflow: neue Flows blieben unsichtbar, weil LFX-Uploads
+  mit `Permission denied` scheiterten (named Volume root-owned, LFX läuft
+  rootless). `lfx-entrypoint.sh` chownt `/flows` per gosu vor dem Start;
+  zusätzlich Pull-Dedup (wiederverwendet gleichnamige/identische Dateien
+  statt Bindestrich-Duplikate wie `simple-rule.json`, die jede Sync-Meldung
+  mit Namenskonflikten füllten).
+- Sync-Dateinamen behalten die Langflow-Namen (`test_flow.json` statt
+  `test-flow.json`; nur dateisystem-reservierte Zeichen werden ersetzt).
+- Gezogene Flow-Dateien sind world-readable (0o644): `mkstemp` erzeugte
+  0o600-Dateien, die Langflow beim Start abstürzen ließen
+  (`Permission denied` in `/app/flows`, Crash-Loop) und der API Fehler
+  (`unreadable`) verursachten.
+- `LANGFLOW_SECRET_KEY` muss ein gültiger Fernet-Key sein (44 Zeichen,
+  urlsafe-base64): ein Hex-String brach die Variablen-Verschlüsselung mit
+  `Fernet key must be 32 url-safe base64-encoded bytes` (500 beim Anlegen
+  von Global Variables — Provider-Keys ließen sich nicht speichern).
+  `.env.example` dokumentiert die Erzeugung.
 - LFX runtime user gets a home directory (`useradd -m`, `HOME` set):
   without one every flow run failed with "Settings service not found".
 ### Security

Dockerfile.lfx

@@ -16,11 +16,18 @@ RUN pip install "lfx~=${LFX_VERSION}" requests
 # Non-root runtime: uploads persist into the mounted volume. The user needs
 # a home directory: lfx resolves settings/cache paths from $HOME, and
 # without one every flow run fails ("Settings service not found").
+# gosu drops privileges in the entrypoint after chowning /flows: the named
+# volume arrives root-owned and the server user could otherwise never write
+# uploads (every deploy would fail with "Permission denied").
 ENV HOME=/home/lfx
-RUN useradd -r -m -s /bin/false lfx && mkdir -p /flows && chown lfx /flows
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends gosu \
+    && rm -rf /var/lib/apt/lists/* \
+    && useradd -r -m -s /bin/false lfx && mkdir -p /flows && chown lfx /flows
+COPY lfx-entrypoint.sh /usr/local/bin/lfx-entrypoint.sh
 VOLUME /flows
 EXPOSE 6004
-USER lfx
+ENTRYPOINT ["sh", "/usr/local/bin/lfx-entrypoint.sh"]
 
 # Serve the /flows dir (uploaded flows persist there via --flow-dir).
 # No positional path: with one, lfx>=1.12.1 exits when the dir holds no

README.md

@@ -89,7 +89,11 @@ a database server. Postgres can return later by changing `DATABASE_URL`
 
 `.env` holds only ubiquitous variables (identical value in dev and prod:
 secrets, service URLs). Dev-only variables travel via `devcontainer.json`
-(`APP_ENV=dev`); paths resolve from the repo root in code.
+(`APP_ENV=dev`); paths resolve from the repo root in code. LLM provider
+keys for Langflow are plain env vars (`OPENAI_API_KEY`,
+`OPENAI_BASE_URL`) — Langflow turns them into restart-proof global
+variables, LFX reads them at runtime. `LANGFLOW_SECRET_KEY` must be a
+valid Fernet key (see `.env.example`).
 
 ## Getting Started
 
@@ -122,9 +126,12 @@ Frontend (node lives in the `frontend` container):
 - `docker compose exec frontend npm run build` — typecheck + build.
 - Without a docker socket: VS Code "Attach to Running Container" → frontend.
 
-Flow workflow: design in Langflow → export into `backend/flows/<name>.json`
+Flow workflow: design in Langflow → press "Aus Langflow synchronisieren"
-(`--pull`, or UI download) → commit → the watcher deploys it automatically.
+in Verwaltung (manual full pass: Langflow → `backend/flows/<name>.json` →
-Deleting a file removes the flow including its runs (git keeps the JSON).
+DB + LFX deploy) → commit. File-only alternatives: `scripts/sync_flows.py
+--pull --sync`, UI download, or the single-flow import by Langflow ID. The
+watcher deploys file edits automatically. Deleting a file removes the flow
+including its runs (git keeps the JSON).
 
 ## Production (self-hosted — TLS/reverse proxy required, not included)
 

backend/app/flows/api.py

@@ -29,7 +29,7 @@ from app.settings import settings
 
 from . import input_schema as form_schema
 from . import runner, store
-from .deploy import import_from_langflow
+from .deploy import import_from_langflow, pull_all_from_langflow
 from .langflow_client import (
     LangflowClient,
     LangflowError,
@@ -160,13 +160,25 @@ def list_flows(session: Annotated[Session, Depends(get_db)]):
 async def sync_flows(
     session: Annotated[Session, Depends(get_db)],
     lfx: Annotated[LfxClient, Depends(get_lfx)],
+    langflow: Annotated[LangflowClient, Depends(get_langflow)],
     _admin: Annotated[User, Depends(require_admin)],
 ):
-    """Reconcile backend/flows/*.json into the DB + LFX (add/update/delete)."""
+    """Manual full pass: Langflow -> backend/flows/*.json -> DB + LFX."""
     try:
-        return await sync_from_directory(session, lfx, Path(settings.flows_dir))
+        pulled_result = await pull_all_from_langflow(langflow, Path(settings.flows_dir))
+    except LangflowUnavailable as exc:
+        logger.warning("langflow unavailable: %s", exc)
+        raise HTTPException(status_code=503, detail="authoring server unavailable") from exc
+    except LangflowError as exc:
+        logger.warning("langflow error: %s", exc)
+        raise HTTPException(status_code=502, detail="authoring server error") from exc
+    try:
+        report = await sync_from_directory(session, lfx, Path(settings.flows_dir))
     except FileNotFoundError as exc:
         raise HTTPException(status_code=500, detail=str(exc)) from exc
+    report.pulled = pulled_result["pulled"]
+    report.errors = {**pulled_result["errors"], **report.errors}
+    return report
 
 
 @flows_router.get("/{flow_id}", response_model=FlowDefinitionRead)

backend/app/flows/deploy.py

@@ -2,16 +2,65 @@
 
 from __future__ import annotations
 
+from pathlib import Path
+
 from sqlmodel import Session
 
 from app.models import FlowDefinition, FlowVersion
 from app.settings import settings
 
 from . import export, sync
-from .langflow_client import LangflowClient
+from .langflow_client import LangflowClient, LangflowError
 from .lfx_client import LfxClient
 
 
+def normalize_exported_flow(exported: dict, langflow_flow_id: str, name: str | None = None) -> dict:
+    """Reduce a Langflow FlowRead to the canonical fields we persist."""
+    flow_name = name or exported.get("name") or f"langflow-{langflow_flow_id}"
+    return {
+        "id": exported.get("id", langflow_flow_id),
+        "name": flow_name,
+        "description": exported.get("description") or "",
+        "endpoint_name": exported.get("endpoint_name"),
+        "data": exported.get("data", {}),
+        "tags": exported.get("tags", []),
+        "webhook": exported.get("webhook", False),
+        "is_component": exported.get("is_component", False),
+    }
+
+
+async def pull_all_from_langflow(
+    langflow: LangflowClient, flows_dir: str | Path
+) -> dict[str, list[str] | dict[str, str]]:
+    """Export every Langflow flow to its canonical file.
+
+    Returns {"pulled": [...stems], "errors": {langflow_id: reason}}.
+    Per-flow failures are collected, never raised; listing failures raise
+    (caller maps them to 503/502).
+    """
+    flows_dir = Path(flows_dir)
+    headers = await langflow.list_flows()
+    pulled: list[str] = []
+    errors: dict[str, str] = {}
+    for header in headers:
+        flow_id = str(header.get("id", ""))
+        if not flow_id:
+            continue
+        try:
+            exported = await langflow.export_flow(flow_id)
+            doc = normalize_exported_flow(exported, flow_id)
+            reuse, identical = export.find_reuse_path(flows_dir, doc)
+            if identical and reuse is not None:
+                pulled.append(reuse.stem)
+                continue
+            target = reuse or export.flow_path(flows_dir, export.filename_stem(doc["name"]))
+            path = export.write_flow_bytes(target, doc)
+            pulled.append(path.stem)
+        except (LangflowError, OSError, ValueError) as exc:
+            errors[flow_id] = str(exc)
+    return {"pulled": sorted(pulled), "errors": errors}
+
+
 async def import_from_langflow(
     session: Session,
     langflow: LangflowClient,
@@ -28,18 +77,8 @@ async def import_from_langflow(
     truth; sync_document() derives the definition + version from it.
     """
     exported = await langflow.export_flow(langflow_flow_id)
-    flow_name = name or exported.get("name") or f"langflow-{langflow_flow_id}"
+    langflow_json = normalize_exported_flow(exported, langflow_flow_id, name=name)
-    langflow_json = {
+    path = export.write_flow_file(settings.flows_dir, langflow_json, name=langflow_json["name"])
-        "id": exported.get("id", langflow_flow_id),
-        "name": flow_name,
-        "description": exported.get("description") or "",
-        "endpoint_name": exported.get("endpoint_name"),
-        "data": exported.get("data", {}),
-        "tags": exported.get("tags", []),
-        "webhook": exported.get("webhook", False),
-        "is_component": exported.get("is_component", False),
-    }
-    path = export.write_flow_file(settings.flows_dir, langflow_json, name=flow_name)
     definition, version, _action = await sync.sync_document(
         session,
         lfx,

backend/app/flows/export.py

@@ -16,10 +16,17 @@ from pathlib import Path
 FLOWS_GLOB = "*.json"
 
 
-def slugify(name: str) -> str:
+def filename_stem(name: str) -> str:
-    """Map a flow name to a safe filename stem (lowercase, [a-z0-9-])."""
+    """Map a flow name to a filename stem, keeping the original name.
-    slug = re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")
+
-    return slug or "flow"
+    Only filesystem-reserved characters (``/ \\ : * ? " < > |`` and control
+    chars) become ``-``; leading/trailing whitespace and trailing dots are
+    stripped (Windows). Case, spaces, underscores and umlauts are preserved:
+    ``test_flow`` stays ``test_flow``.
+    """
+    stem = re.sub(r'[\\/:*?"<>|\x00-\x1f]', "-", name).strip()
+    stem = stem.rstrip(".")
+    return stem or "flow"
 
 
 def canonical_bytes(doc: dict) -> bytes:
@@ -36,21 +43,55 @@ 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:
+def _read_doc(path: Path) -> dict | None:
-    """Write a flow document to backend/flows atomically (tmp + rename).
+    try:
+        doc = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return None
+    return doc if isinstance(doc, dict) else None
 
-    The filename derives from `name` (or the document's own name). Returns
+
-    the written path.
+def find_reuse_path(flows_dir: str | Path, doc: dict) -> tuple[Path | None, bool]:
+    """Existing file to reuse for `doc`: (path, identical).
+
+    Matches identical content first (any filename), then the same flow name.
+    Returns (None, False) when no candidate exists (caller writes a slug file).
+    This keeps pulls idempotent: a flow seeded from `simple_rule.json` is
+    updated in place instead of duplicated as `simple-rule.json`.
     """
     flows_dir = Path(flows_dir)
-    flows_dir.mkdir(parents=True, exist_ok=True)
+    if not flows_dir.is_dir():
-    stem = slugify(name or doc.get("name") or "flow")
+        return None, False
-    target = flow_path(flows_dir, stem)
+    wanted_hash = content_hash(doc)
-    fd, tmp_name = tempfile.mkstemp(dir=str(flows_dir), prefix=f".{stem}-", suffix=".tmp")
+    wanted_name = doc.get("name")
+    named: Path | None = None
+    for path in iter_flow_files(flows_dir):
+        existing = _read_doc(path)
+        if existing is None:
+            continue
+        if content_hash(existing) == wanted_hash:
+            return path, True
+        if named is None and wanted_name and existing.get("name") == wanted_name:
+            named = path
+    return (named, False) if named is not None else (None, False)
+
+
+def write_flow_bytes(target: Path, doc: dict) -> Path:
+    """Write a flow document to an explicit path atomically (tmp + rename).
+
+    The file ends up world-readable (0o644): mkstemp creates 0o600, which
+    would lock out the other containers (Langflow/LFX run as different
+    users and crash on unreadable seed files).
+    """
+    target.parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp_name = tempfile.mkstemp(
+        dir=str(target.parent), prefix=f".{target.stem}-", suffix=".tmp"
+    )
     try:
         with os.fdopen(fd, "wb") as tmp:
             tmp.write(canonical_bytes(doc))
         os.replace(tmp_name, target)
+        os.chmod(target, 0o644)
     except BaseException:
         try:
             os.unlink(tmp_name)
@@ -60,6 +101,18 @@ def write_flow_file(flows_dir: str | Path, doc: dict, *, name: str | None = None
     return target
 
 
+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 = filename_stem(name or doc.get("name") or "flow")
+    return write_flow_bytes(flow_path(flows_dir, stem), doc)
+
+
 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

@@ -127,13 +127,15 @@ class DecisionRead(BaseModel):
 
 
 class SyncReport(BaseModel):
-    """Result of reconciling backend/flows/*.json into DB + LFX.
+    """Result of the manual full pass: Langflow -> files -> DB + LFX.
 
-    Lists hold filename stems; errors maps filename to reason.
+    Lists hold filename stems; errors maps filename (or Langflow id on
+    pull failures) 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)
+    pulled: list[str] = Field(default_factory=list)
     errors: dict[str, str] = Field(default_factory=dict)

backend/scripts/sync_flows.py

@@ -18,8 +18,9 @@ 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 import sync
-from app.flows.langflow_client import LangflowClient, LangflowError
+from app.flows.deploy import pull_all_from_langflow
+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
@@ -28,19 +29,7 @@ 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()
+    return await pull_all_from_langflow(client, flows_dir)
-    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 (LangflowError, OSError, ValueError) as exc:
-            errors[flow_id] = str(exc)
-    return {"pulled": sorted(pulled), "errors": errors}
 
 
 async def run_sync(flows_dir: Path) -> SyncReport:

backend/tests/conftest.py

@@ -218,22 +218,36 @@ def override_langflow():
 
     Usage: override_langflow(export_doc) -- the fake export_flow() returns the
     doc, or raises LangflowNotFound/Unavailable for status=404/0.
+    Pass flows=[...] + docs={id: doc} for multi-flow pull tests, or
+    list_status=0/500 to fail the listing. Default list is empty so
+    file-only sync tests stay hermetic.
     """
     from app.flows.api import get_langflow
-    from app.flows.langflow_client import LangflowNotFound, LangflowUnavailable
+    from app.flows.langflow_client import LangflowError, LangflowNotFound, LangflowUnavailable
     from app.main import app
 
-    def _override(export_doc=None, status=200):
+    def _override(export_doc=None, status=200, flows=None, docs=None, list_status=200):
+        if flows is None:
+            flows = [{"id": "fake-1", "name": "fake"}] if export_doc is not None else []
+
         class FakeLangflow:
             async def export_flow(self, flow_id):
                 if status == 404:
                     raise LangflowNotFound(f"flow {flow_id} not found")
                 if status == 0:
                     raise LangflowUnavailable("Langflow unreachable")
+                if docs is not None:
+                    if flow_id in docs:
+                        return docs[flow_id]
+                    raise LangflowNotFound(f"flow {flow_id} not found")
                 return export_doc
 
             async def list_flows(self):
-                return [{"id": "fake-1", "name": "fake"}]
+                if list_status == 0:
+                    raise LangflowUnavailable("Langflow unreachable")
+                if list_status == 500:
+                    raise LangflowError("Langflow error 500")
+                return flows
 
         app.dependency_overrides[get_langflow] = lambda: FakeLangflow()
 

backend/tests/test_export.py

@@ -5,11 +5,15 @@ import json
 from app.flows import export
 
 
-def test_slugify():
+def test_filename_stem_keeps_original_names():
-    assert export.slugify("Dachsanierung 200k") == "dachsanierung-200k"
+    assert export.filename_stem("simple_rule") == "simple_rule"
-    assert export.slugify("  A/B\\C  ") == "a-b-c"
+    assert export.filename_stem("test_flow") == "test_flow"
-    assert export.slugify("simple_rule") == "simple-rule"
+    assert export.filename_stem("Gehaltserhöhung") == "Gehaltserhöhung"
-    assert export.slugify("") == "flow"
+    assert export.filename_stem("Dachsanierung 200k") == "Dachsanierung 200k"
+    assert export.filename_stem("  A/B\\C  ") == "A-B-C"
+    assert export.filename_stem('a:b*c?d"e<f>g|h') == "a-b-c-d-e-f-g-h"
+    assert export.filename_stem("trailing...") == "trailing"
+    assert export.filename_stem("") == "flow"
 
 
 def test_canonical_bytes_stable_under_key_reorder():
@@ -22,7 +26,7 @@ def test_canonical_bytes_stable_under_key_reorder():
 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 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()
@@ -37,3 +41,42 @@ def test_iter_flow_files_only_top_level_json(tmp_path):
     sub.mkdir()
     (sub / "c.json").write_text("{}")
     assert export.iter_flow_files(tmp_path) == [tmp_path / "a.json"]
+
+
+def test_find_reuse_path_matches_identical_content_first(tmp_path):
+    doc = {"name": "Simple Rule", "data": {"nodes": [], "edges": []}}
+    other = tmp_path / "other-name.json"
+    other.write_text(export.canonical_bytes(doc).decode())
+    path, identical = export.find_reuse_path(tmp_path, dict(doc))
+    assert (path, identical) == (other, True)
+
+
+def test_find_reuse_path_matches_same_name(tmp_path):
+    old = {"name": "Simple Rule", "data": {"nodes": [], "edges": []}, "extra": 1}
+    (tmp_path / "simple_rule.json").write_text(json.dumps(old))
+    new = {"name": "Simple Rule", "data": {"nodes": [], "edges": []}}
+    path, identical = export.find_reuse_path(tmp_path, new)
+    assert (path, identical) == (tmp_path / "simple_rule.json", False)
+
+
+def test_find_reuse_path_no_match_and_missing_dir(tmp_path):
+    (tmp_path / "a.json").write_text(json.dumps({"name": "Other"}))
+    assert export.find_reuse_path(tmp_path, {"name": "Neu"}) == (None, False)
+    assert export.find_reuse_path(tmp_path / "gibts-nicht", {"name": "Neu"}) == (None, False)
+
+
+def test_find_reuse_path_skips_unreadable_files(tmp_path):
+    (tmp_path / "kaputt.json").write_text("{kein json")
+    assert export.find_reuse_path(tmp_path, {"name": "Neu"}) == (None, False)
+
+
+def test_write_flow_file_is_world_readable(tmp_path):
+    import os
+    import stat
+
+    doc = {"name": "Lesbar", "data": {"nodes": [], "edges": []}}
+    path = export.write_flow_file(tmp_path, doc)
+    mode = stat.S_IMODE(os.stat(path).st_mode)
+    # Read bits for user/group/other (mkstemp defaults to 0o600; on Windows
+    # chmod maps to 0o666, so only assert the read bits, not the exact mode).
+    assert mode & 0o444 == 0o444, f"expected world-readable, got {oct(mode)}"

backend/tests/test_sync.py

@@ -118,20 +118,127 @@ async def test_sync_rename_updates_definition_name(session, lfx_server, tmp_path
     assert definition.name == "umbenannt"
 
 
-async def test_sync_endpoint_returns_report(client, lfx_server, tmp_path, monkeypatch, login):
+async def test_sync_endpoint_returns_report(
+    client, lfx_server, tmp_path, monkeypatch, login, override_langflow
+):
     await login()
+    override_langflow()  # empty Langflow: file-only pass
     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["pulled"] == []
     assert body["errors"] == {}
 
     response = await client.post("/api/flows/sync")
     assert response.json()["skipped"] == ["simple_rule"]
 
 
+def _langflow_doc(name: str, langflow_id: str) -> dict:
+    doc = load_fixture("simple_rule.json")
+    return {
+        "id": langflow_id,
+        "name": name,
+        "description": "aus Langflow",
+        "endpoint_name": None,
+        "data": doc["data"],
+        "tags": [],
+        "webhook": False,
+        "is_component": False,
+    }
+
+
+async def test_sync_endpoint_pulls_from_langflow_then_syncs(
+    client, lfx_server, tmp_path, monkeypatch, login, override_langflow
+):
+    await login()
+    monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
+    langflow_id = "22222222-3333-4444-8555-666666666666"
+    override_langflow(
+        flows=[{"id": langflow_id, "name": "Gehaltserhöhung"}],
+        docs={langflow_id: _langflow_doc("Gehaltserhöhung", langflow_id)},
+    )
+    response = await client.post("/api/flows/sync")
+    assert response.status_code == 200, response.text
+    body = response.json()
+    assert body["pulled"] == ["Gehaltserhöhung"]
+    assert body["added"] == ["Gehaltserhöhung"]
+    assert body["errors"] == {}
+    assert (tmp_path / "Gehaltserhöhung.json").is_file()
+
+    # Second pass: same content pulls again but syncs nothing new.
+    response = await client.post("/api/flows/sync")
+    assert response.status_code == 200, response.text
+    body = response.json()
+    assert body["pulled"] == ["Gehaltserhöhung"]
+    assert body["skipped"] == ["Gehaltserhöhung"]
+
+
+async def test_sync_endpoint_pull_reuses_same_named_file(
+    client, lfx_server, tmp_path, monkeypatch, login, override_langflow
+):
+    """A pulled flow seeded from simple_rule.json updates it in place
+    instead of duplicating it as simple-rule.json (slug drift)."""
+    await login()
+    monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
+    _seed_dir(tmp_path)  # fixture shape, name "simple_rule"
+    langflow_id = "44444444-5555-4666-8777-888888888888"
+    override_langflow(
+        flows=[{"id": langflow_id, "name": "simple_rule"}],
+        docs={langflow_id: _langflow_doc("simple_rule", langflow_id)},
+    )
+    response = await client.post("/api/flows/sync")
+    assert response.status_code == 200, response.text
+    body = response.json()
+    assert body["pulled"] == ["simple_rule"]
+    assert body["added"] == ["simple_rule"]
+    assert body["errors"] == {}
+    assert not (tmp_path / "simple-rule.json").exists()
+
+    response = await client.post("/api/flows/sync")
+    assert response.json()["skipped"] == ["simple_rule"]
+
+
+async def test_sync_endpoint_pull_error_continues(
+    client, lfx_server, tmp_path, monkeypatch, login, override_langflow
+):
+    await login()
+    monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
+    good_id = "33333333-4444-4555-8666-777777777777"
+    override_langflow(
+        flows=[{"id": good_id, "name": "Gut"}, {"id": "kaputt-id", "name": "Kaputt"}],
+        docs={good_id: _langflow_doc("Gut", good_id)},
+    )
+    response = await client.post("/api/flows/sync")
+    assert response.status_code == 200, response.text
+    body = response.json()
+    assert body["pulled"] == ["Gut"]
+    assert body["added"] == ["Gut"]
+    assert "kaputt-id" in body["errors"]
+
+
+async def test_sync_endpoint_langflow_unreachable(
+    client, lfx_server, tmp_path, monkeypatch, login, override_langflow
+):
+    await login()
+    monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
+    override_langflow(flows=[], list_status=0)
+    response = await client.post("/api/flows/sync")
+    assert response.status_code == 503
+
+
+async def test_sync_endpoint_langflow_error(
+    client, lfx_server, tmp_path, monkeypatch, login, override_langflow
+):
+    await login()
+    monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
+    override_langflow(flows=[], list_status=500)
+    response = await client.post("/api/flows/sync")
+    assert response.status_code == 502
+
+
 async def test_watch_forever_triggers_on_file_change(tmp_path):
     import contextlib
 

frontend/src/api.ts

@@ -66,6 +66,7 @@ export interface SyncReport {
   updated: string[];
   deleted: string[];
   skipped: string[];
+  pulled: string[];
   errors: Record<string, string>;
 }
 

frontend/src/components/AdminTab.tsx

@@ -24,6 +24,7 @@ import {
   type FlowDefinition,
   type FlowVersion,
 } from "../api";
+import FlowHelp from "./FlowHelp";
 import StatusBadge from "./StatusBadge";
 
 export default function AdminTab() {
@@ -89,6 +90,7 @@ export default function AdminTab() {
     try {
       const r = await api.syncFlows();
       const parts = [
+        `Langflow: ${r.pulled.length}`,
         `+${r.added.length}`,
         `~${r.updated.length}`,
         `-${r.deleted.length}`,
@@ -147,6 +149,7 @@ export default function AdminTab() {
           </Button>
         </Group>
       </Card>
+      <FlowHelp />
       <Card withBorder padding="md" mb="md">
         <Group>
           <Button
@@ -154,9 +157,13 @@ export default function AdminTab() {
             leftSection={<IconRefresh size={16} />}
             onClick={() => void onSync()}
           >
-            Aus backend/flows synchronisieren
+            Aus Langflow synchronisieren
           </Button>
         </Group>
+        <Text size="sm" c="dimmed" mt="xs">
+          Holt alle Flows aus Langflow nach backend/flows und deployed sie
+          danach automatisch.
+        </Text>
       </Card>
       {flows.length === 0 ? (
         <Text c="dimmed">Keine Flows vorhanden.</Text>

frontend/src/components/FlowHelp.tsx

@@ -0,0 +1,93 @@
+import { useState } from "react";
+import {
+  Anchor,
+  Button,
+  Card,
+  List,
+  Text,
+  Title,
+} from "@mantine/core";
+import { IconChevronDown, IconChevronUp, IconExternalLink } from "@tabler/icons-react";
+import { LANGFLOW_URL } from "../api";
+
+export default function FlowHelp() {
+  const [open, setOpen] = useState(false);
+
+  return (
+    <Card withBorder padding="md" mb="md">
+      <Button
+        variant="subtle"
+        leftSection={open ? <IconChevronUp size={16} /> : <IconChevronDown size={16} />}
+        onClick={() => setOpen((v) => !v)}
+      >
+        {open ? "Anleitung verbergen" : "Anleitung: Neuer Flow in Langflow"}
+      </Button>
+      {open && (
+      <div>
+        <Title order={4} mt="sm" mb="xs">
+          So kommt ein neuer Flow in die Webapp
+        </Title>
+        <List size="sm" spacing="xs">
+          <List.Item>
+            <Text component="span" fw={600}>
+              1. In Langflow entwerfen und speichern:{" "}
+            </Text>
+            <Anchor href={LANGFLOW_URL} target="_blank" rel="noreferrer">
+              Langflow öffnen <IconExternalLink size={14} />
+            </Anchor>
+            . Die bestehenden Flows sind dort bereits geladen. Name vergeben, speichern.
+          </List.Item>
+          <List.Item>
+            <Text component="span" fw={600}>
+              2. Sync-Button unten klicken:{" "}
+            </Text>
+            holt alle Flows aus Langflow nach{" "}
+            <Text component="span" ff="monospace">
+              backend/flows/*.json
+            </Text>{" "}
+            und deployed sie danach automatisch (neue Version nur bei geändertem Inhalt,
+            sonst übersprungen).
+          </List.Item>
+          <List.Item>
+            <Text component="span" fw={600}>
+              3. Prüfen:{" "}
+            </Text>
+            Tabelle unten zeigt Version, Autor und Status. Danach erscheint der Flow
+            automatisch auf der Seite Workflows mit seinem Formular. Flows ohne
+            Formular-Knoten zeigen ein Freitext-Feld.
+          </List.Item>
+        </List>
+        <Title order={5} mt="md" mb="xs">
+          Regeln
+        </Title>
+        <List size="sm" spacing="xs">
+          <List.Item>
+            Die Datei heißt wie der Flow in Langflow (z. B. test_flow.json).
+            Umbenennen in Langflow erzeugt eine neue Datei; die alte bleibt
+            bestehen und muss bei Bedarf gelöscht werden.
+          </List.Item>
+          <List.Item>
+            Eine gelöschte Datei löscht den Flow inklusive seiner Läufe
+            (Datei bleibt in Git erhalten).
+          </List.Item>
+          <List.Item>
+            Fehler stehen in der Sync-Meldung (z. B. ungültiges Flow-JSON).
+            Der automatische Beobachter im Hintergrund syncet nur Dateien,
+            Langflow holt nur dieser Button.
+          </List.Item>
+        </List>
+        <Title order={5} mt="md" mb="xs">
+          Formular-Felder
+        </Title>
+        <Text size="sm">
+          Für ein Web-Formular braucht der Flow einen CustomInputForm-Knoten: jeder
+          Eintrag dort ist ein Formularfeld (Name = Feld-Key, Anzeigename = Label).
+          Die Verbindung ChatInput nach form_values bitte behalten — darüber kommen
+          die ausgefüllten Werte an. Unbekannte Feldtypen fallen automatisch auf ein
+          Textfeld zurück.
+        </Text>
+      </div>
+      )}
+    </Card>
+  );
+}

lfx-entrypoint.sh

@@ -0,0 +1,9 @@
+#!/bin/sh
+# LFX entrypoint (runs as root, drops privileges for the server).
+# The named volume arrives root-owned (Docker creates it as root and the
+# image USER cannot chown it) — without this every flow upload fails with
+# "Permission denied" and synced flows stay version-less (invisible in UI).
+set -e
+
+chown -R lfx:lfx /flows
+exec gosu lfx "$@"