diff --git a/CHANGELOG.md b/CHANGELOG.md index ffec58f..bdf48f5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,13 @@ 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. ### Fixed - LFX runtime user gets a home directory (`useradd -m`, `HOME` set): without one every flow run failed with "Settings service not found". diff --git a/README.md b/README.md index 30fde11..0fca6c9 100644 --- a/README.md +++ b/README.md @@ -122,9 +122,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/.json` -(`--pull`, or UI download) → commit → the watcher deploys it automatically. -Deleting a file removes the flow including its runs (git keeps the JSON). +Flow workflow: design in Langflow → press "Aus Langflow synchronisieren" +in Verwaltung (manual full pass: Langflow → `backend/flows/.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) diff --git a/backend/app/flows/api.py b/backend/app/flows/api.py index f47ee17..eaca507 100644 --- a/backend/app/flows/api.py +++ b/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) diff --git a/backend/app/flows/deploy.py b/backend/app/flows/deploy.py index 535d436..ee990ef 100644 --- a/backend/app/flows/deploy.py +++ b/backend/app/flows/deploy.py @@ -2,16 +2,60 @@ 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) + path = export.write_flow_file(flows_dir, doc, name=doc["name"]) + 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 +72,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 = { - "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) + langflow_json = normalize_exported_flow(exported, langflow_flow_id, name=name) + path = export.write_flow_file(settings.flows_dir, langflow_json, name=langflow_json["name"]) definition, version, _action = await sync.sync_document( session, lfx, diff --git a/backend/app/flows/schemas.py b/backend/app/flows/schemas.py index d32dfcb..f2616a4 100644 --- a/backend/app/flows/schemas.py +++ b/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) diff --git a/backend/scripts/sync_flows.py b/backend/scripts/sync_flows.py index 5db9670..8eaeff1 100644 --- a/backend/scripts/sync_flows.py +++ b/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.langflow_client import LangflowClient, LangflowError +from app.flows import sync +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() - 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} + return await pull_all_from_langflow(client, flows_dir) async def run_sync(flows_dir: Path) -> SyncReport: diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 6af8fd5..4f8c4e7 100644 --- a/backend/tests/conftest.py +++ b/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() diff --git a/backend/tests/test_sync.py b/backend/tests/test_sync.py index bbff7ca..5b1fc58 100644 --- a/backend/tests/test_sync.py +++ b/backend/tests/test_sync.py @@ -118,20 +118,102 @@ 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_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 diff --git a/frontend/src/api.ts b/frontend/src/api.ts index f0af0f7..e693c20 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -66,6 +66,7 @@ export interface SyncReport { updated: string[]; deleted: string[]; skipped: string[]; + pulled: string[]; errors: Record; } diff --git a/frontend/src/components/AdminTab.tsx b/frontend/src/components/AdminTab.tsx index 9c068de..c695386 100644 --- a/frontend/src/components/AdminTab.tsx +++ b/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() { + + + Holt alle Flows aus Langflow nach backend/flows und deployed sie + danach automatisch. + {flows.length === 0 ? ( Keine Flows vorhanden. diff --git a/frontend/src/components/FlowHelp.tsx b/frontend/src/components/FlowHelp.tsx new file mode 100644 index 0000000..ed568cb --- /dev/null +++ b/frontend/src/components/FlowHelp.tsx @@ -0,0 +1,92 @@ +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 ( + + + {open && ( +
+ + So kommt ein neuer Flow in die Webapp + + + + + 1. In Langflow entwerfen und speichern:{" "} + + + Langflow öffnen + + . Die bestehenden Flows sind dort bereits geladen. Name vergeben, speichern. + + + + 2. Sync-Button unten klicken:{" "} + + holt alle Flows aus Langflow nach{" "} + + backend/flows/*.json + {" "} + und deployed sie danach automatisch (neue Version nur bei geändertem Inhalt, + sonst übersprungen). + + + + 3. Prüfen:{" "} + + 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. + + + + Regeln + + + + Der Dateiname wird aus dem Flow-Namen erzeugt (Kleinbuchstaben, + Bindestriche). Umbenennen erzeugt eine neue Datei. + + + Eine gelöschte Datei löscht den Flow inklusive seiner Läufe + (Datei bleibt in Git erhalten). + + + 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. + + + + Formular-Felder + + + 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. + +
+ )} + + ); +}