jonas/kirche-einfach
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: jonas:74831483eb0db09059f35c08870fafc94aeae75a
Branches Tags
jonas:main
jonas:feat/declutter
..
compare: jonas:418318152a5941ce89a5ab3f3e23c786d0a2c624
Branches Tags
jonas:feat/declutter
jonas:main

No commits in common. "74831483eb0db09059f35c08870fafc94aeae75a" and "418318152a5941ce89a5ab3f3e23c786d0a2c624" have entirely different histories.
74831483eb ... 418318152a

46 changed files with 20 additions and 7407 deletions
Download patch file Download diff file Expand all files Collapse all files

11
.devcontainer/Dockerfile
View file

@ -27,14 +27,9 @@ RUN uv python install 3.12 \
# Trust the workspace bind mount (source on Windows FS -> uid mismatch)
RUN git config --system --add safe.directory /workspaces/kirche-einfach
# Pre-create named-volume targets (volumes inherit image content on first mount).
# The backend venv is created as `vscode` so VS Code's Python extension finds an
# interpreter immediately on first attach, before postCreateCommand runs `uv sync`.
USER vscode
RUN mkdir -p /workspaces/kirche-einfach/backend \
&& uv venv /workspaces/kirche-einfach/backend/.venv \
&& mkdir -p /workspaces/kirche-einfach/frontend/node_modules
USER root
# Pre-create named-volume targets (volumes inherit image content on first mount)
RUN mkdir -p /workspaces/kirche-einfach/backend/.venv \
/workspaces/kirche-einfach/frontend/node_modules
# opencode (as non-root user)
USER vscode

19
.devcontainer/Dockerfile.lfx
View file

@ -1,19 +0,0 @@
# Stateless Langflow flow runtime (LFX). Serves versioned flow JSON files
# deployed by our backend via POST /flows/upload/ (persisted in /flows).
# LFX stays on the internal compose network; our FastAPI is the only
# public surface and proxies run/stream calls.
FROM python:3.12-slim
ENV PYTHONUNBUFFERED=1 \
PIP_NO_CACHE_DIR=1
# lfx = flow runtime; requests = convenience for custom-component code.
RUN pip install "lfx==1.12.0" requests
RUN mkdir -p /flows
VOLUME /flows
EXPOSE 6004
# Serve the /flows dir (uploaded flows persist there via --flow-dir).
# Flags verified against `lfx serve --help` (lfx 1.12.0).
CMD ["lfx", "serve", "--host", "0.0.0.0", "--port", "6004", "--flow-dir", "/flows", "/flows"]

15
.devcontainer/devcontainer.json
View file

@ -1,10 +1,15 @@
{
"name": "kirche-einfach",
"dockerComposeFile": "docker-compose.yml",
"service": "app",
"runServices": ["db", "langflow", "lfx"],
"build": {
"dockerfile": "Dockerfile",
"args": {
"FORGEJO_USERNAME": "jonas"
}
},
"mounts": [
"source=${localEnv:USERPROFILE}/.ssh,target=/mnt/host-ssh,type=bind,readonly"
"source=${localEnv:USERPROFILE}/.ssh,target=/mnt/host-ssh,type=bind,readonly",
"source=kirche-einfach-venv,target=/workspaces/kirche-einfach/backend/.venv,type=volume",
"source=kirche-einfach-node-modules,target=/workspaces/kirche-einfach/frontend/node_modules,type=volume"
],
"remoteUser": "vscode",
"containerEnv": {
@ -13,7 +18,7 @@
"overrideCommand": false,
"workspaceFolder": "/workspaces/kirche-einfach",
"postCreateCommand": "sudo -E bash .devcontainer/setup.sh",
"postStartCommand": "bash .devcontainer/postStart.sh",
"postStartCommand": "mkdir -p ~/.ssh && cp /mnt/host-ssh/id_rsa /mnt/host-ssh/id_rsa.pub ~/.ssh/ && cp /mnt/host-ssh/known_hosts ~/.ssh/ 2>/dev/null; ssh-keyscan -t ed25519,rsa gitlab.jonas-miederer.com >> ~/.ssh/known_hosts 2>/dev/null; chmod 700 ~/.ssh && chmod 600 ~/.ssh/id_rsa && chmod 644 ~/.ssh/id_rsa.pub ~/.ssh/known_hosts",
"customizations": {
"vscode": {
"extensions": [

93
.devcontainer/docker-compose.yml
View file

@ -1,93 +0,0 @@
# Compose stack for the kirche-einfach devcontainer.
# VS Code attaches to the `app` service (see devcontainer.json); all values
# come from the repo-root `.env` via `env_file` (no ${} interpolation needed).
services:
app:
build:
context: .
dockerfile: Dockerfile
args:
FORGEJO_USERNAME: jonas
command: sleep infinity
volumes:
- ..:/workspaces/kirche-einfach:cached
- kirche-einfach-venv:/workspaces/kirche-einfach/backend/.venv
- kirche-einfach-node-modules:/workspaces/kirche-einfach/frontend/node_modules
env_file:
- ../.env
ports:
- "8000:8000" # FastAPI
- "3000:3000" # Vite
db:
image: postgres:16-trixie
env_file:
- ../.env
volumes:
- postgres-data:/var/lib/postgresql/data
- ./init-db.sql:/docker-entrypoint-initdb.d/init-db.sql:ro
ports:
- "5432:5432"
healthcheck:
test: ["CMD-SHELL", "pg_isready -U postgres"]
interval: 5s
timeout: 3s
retries: 10
langflow:
image: langflowai/langflow:1.12.0
depends_on:
db:
condition: service_healthy
env_file:
- ../.env
environment:
# Static values only; secrets/URLs come from ../.env via env_file
# (compose interpolates ${} from .devcontainer/.env, not the repo-root .env).
LANGFLOW_CONFIG_DIR: /app/langflow
LANGFLOW_HOST: 0.0.0.0
PORT: "7860"
LANGFLOW_AUTO_LOGIN: "false"
LANGFLOW_API_KEY_SOURCE: env
volumes:
- langflow-data:/app/langflow
ports:
- "7860:7860"
healthcheck:
test:
[
"CMD-SHELL",
"python -c \"import urllib.request; urllib.request.urlopen('http://localhost:7860/health', timeout=5)\"",
]
interval: 10s
timeout: 5s
retries: 20
start_period: 60s
lfx:
build:
context: .
dockerfile: Dockerfile.lfx
env_file:
- ../.env
volumes:
- lfx-data:/flows
ports:
- "6004:6004"
healthcheck:
test:
[
"CMD-SHELL",
"python -c \"import urllib.request; urllib.request.urlopen('http://localhost:6004/health', timeout=5)\"",
]
interval: 10s
timeout: 5s
retries: 12
start_period: 30s
volumes:
kirche-einfach-venv:
kirche-einfach-node-modules:
postgres-data:
langflow-data:
lfx-data:

7
.devcontainer/init-db.sql
View file

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

19
.devcontainer/postStart.sh
View file

@ -1,19 +0,0 @@
#!/bin/bash
# Post-start hook: copies SSH keys from the host into the container, then
# adds the git host. Never fails the devcontainer run: any missing file on
# the host (e.g. no id_rsa) must not abort the lifecycle command.
set -u
mkdir -p ~/.ssh
for f in id_rsa id_rsa.pub known_hosts; do
cp -f "/mnt/host-ssh/$f" ~/.ssh/ 2>/dev/null || true
done
ssh-keyscan -t ed25519,rsa gitlab.jonas-miederer.com >> ~/.ssh/known_hosts 2>/dev/null || true
chmod 700 ~/.ssh 2>/dev/null || true
chmod 600 ~/.ssh/id_rsa 2>/dev/null || true
chmod 644 ~/.ssh/id_rsa.pub ~/.ssh/known_hosts 2>/dev/null || true
exit 0

32
.devcontainer/setup.sh
View file

@ -13,38 +13,6 @@ echo "--- Installing Node dependencies ---"
cd /workspaces/kirche-einfach/frontend
npm install
# 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
exit 1
fi
python3 - <<'EOF'
import socket
import sys
import time
targets = [("db", 5432), ("langflow", 7860), ("lfx", 6004)]
deadline = time.time() + 180
pending = dict(targets)
while pending and time.time() < deadline:
for host, port in list(pending):
s = socket.socket()
s.settimeout(2)
try:
s.connect((host, port))
print(f"OK {host}:{port}", flush=True)
del pending[(host, port)]
except OSError:
pass
finally:
s.close()
if pending:
time.sleep(2)
if pending:
sys.exit(f"ERROR: compose services not ready: {sorted(pending)}")
EOF
echo ""
echo "=== Setup complete ==="
echo "Forgejo username: $FORGEJO_USERNAME"

28
.env.example
View file

@ -1,28 +0,0 @@
# Copy to `.env` and fill in values. `.env` is gitignored.
FORGEJO_TOKEN=
# --- databases (compose service `db`) ---
POSTGRES_USER=kirche
POSTGRES_PASSWORD=
POSTGRES_DB=kirche_einfach
DATABASE_URL=postgresql+psycopg://kirche:<password>@db:5432/kirche_einfach
TEST_DATABASE_URL=postgresql+psycopg://kirche:<password>@db:5432/kirche_einfach_test
# --- langflow (authoring, dev only) ---
LANGFLOW_URL=http://langflow:7860
LANGFLOW_SUPERUSER=langflow
LANGFLOW_SUPERUSER_PASSWORD=
LANGFLOW_SECRET_KEY=
# Single static key; validated against env (LANGFLOW_API_KEY_SOURCE=env), superuser scope.
LANGFLOW_API_KEY=sk-
LANGFLOW_DATABASE_URL=postgresql://kirche:<password>@db:5432/langflow
# --- lfx (flow runtime) ---
LFX_URL=http://lfx:6004
LFX_API_KEY=lfx-
# --- API self-reference for flow callbacks (must be reachable FROM lfx) ---
CALLBACK_BASE_URL=http://app:8000
# --- LLM (optional; tests mock the provider without it) ---
# LLM_API_KEY=

1
.gitignore vendored
View file

@ -8,7 +8,6 @@ __pycache__/
# Node
node_modules/
dist/
*.tsbuildinfo
# Environment
.env

5
AGENTS.md
View file

@ -1,5 +0,0 @@
# Agents
## Git
When committing: `git add -A && git commit -m "<message>"`. When pushing: `git push` (branch is tracking `origin/main`).

149
backend/alembic.ini
View file

@ -1,149 +0,0 @@
# A generic, single database configuration.
[alembic]
# path to migration scripts.
# this is typically a path given in POSIX (e.g. forward slashes)
# format, relative to the token %(here)s which refers to the location of this
# ini file
script_location = %(here)s/migrations
# template used to generate migration file names; The default value is %%(rev)s_%%(slug)s
# Uncomment the line below if you want the files to be prepended with date and time
# see https://alembic.sqlalchemy.org/en/latest/tutorial.html#editing-the-ini-file
# for all available tokens
# file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s
# Or organize into date-based subdirectories (requires recursive_version_locations = true)
# file_template = %%(year)d/%%(month).2d/%%(day).2d_%%(hour).2d%%(minute).2d_%%(second).2d_%%(rev)s_%%(slug)s
# sys.path path, will be prepended to sys.path if present.
# defaults to the current working directory. for multiple paths, the path separator
# is defined by "path_separator" below.
prepend_sys_path = .
# timezone to use when rendering the date within the migration file
# as well as the filename.
# If specified, requires the tzdata library which can be installed by adding
# `alembic[tz]` to the pip requirements.
# string value is passed to ZoneInfo()
# leave blank for localtime
# timezone =
# max length of characters to apply to the "slug" field
# truncate_slug_length = 40
# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false
# set to 'true' to allow .pyc and .pyo files without
# a source .py file to be detected as revisions in the
# versions/ directory
# sourceless = false
# version location specification; This defaults
# to <script_location>/versions. When using multiple version
# directories, initial revisions must be specified with --version-path.
# The path separator used here should be the separator specified by "path_separator"
# below.
# version_locations = %(here)s/bar:%(here)s/bat:%(here)s/alembic/versions
# path_separator; This indicates what character is used to split lists of file
# paths, including version_locations and prepend_sys_path within configparser
# files such as alembic.ini.
# The default rendered in new alembic.ini files is "os", which uses os.pathsep
# to provide os-dependent path splitting.
#
# Note that in order to support legacy alembic.ini files, this default does NOT
# take place if path_separator is not present in alembic.ini. If this
# option is omitted entirely, fallback logic is as follows:
#
# 1. Parsing of the version_locations option falls back to using the legacy
# "version_path_separator" key, which if absent then falls back to the legacy
# behavior of splitting on spaces and/or commas.
# 2. Parsing of the prepend_sys_path option falls back to the legacy
# behavior of splitting on spaces, commas, or colons.
#
# Valid values for path_separator are:
#
# path_separator = :
# path_separator = ;
# path_separator = space
# path_separator = newline
#
# Use os.pathsep. Default configuration used for new projects.
path_separator = os
# set to 'true' to search source files recursively
# in each "version_locations" directory
# new in Alembic version 1.10
# recursive_version_locations = false
# the output encoding used when revision files
# are written from script.py.mako
# output_encoding = utf-8
# database URL. This is consumed by the user-maintained env.py script only.
# other means of configuring database URLs may be customized within the env.py
# file.
sqlalchemy.url = driver://user:pass@localhost/dbname
[post_write_hooks]
# post_write_hooks defines scripts or Python functions that are run
# on newly generated revision scripts. See the documentation for further
# detail and examples
# format using "black" - use the console_scripts runner, against the "black" entrypoint
# hooks = black
# black.type = console_scripts
# black.entrypoint = black
# black.options = -l 79 REVISION_SCRIPT_FILENAME
# lint with attempts to fix using "ruff" - use the module runner, against the "ruff" module
# hooks = ruff
# ruff.type = module
# ruff.module = ruff
# ruff.options = check --fix REVISION_SCRIPT_FILENAME
# Alternatively, use the exec runner to execute a binary found on your PATH
# hooks = ruff
# ruff.type = exec
# ruff.executable = ruff
# ruff.options = check --fix REVISION_SCRIPT_FILENAME
# Logging configuration. This is also consumed by the user-maintained
# env.py script only.
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARNING
handlers = console
qualname =
[logger_sqlalchemy]
level = WARNING
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

36
backend/app/db.py
View file

@ -1,36 +0,0 @@
"""Database engine, sessions, and FastAPI dependency."""
from collections.abc import Iterator
from sqlalchemy import Engine
from sqlmodel import Session, SQLModel, create_engine
from .settings import settings
_engine: Engine | None = None
def get_engine(url: str | None = None) -> Engine:
"""Process-wide engine singleton (override URL in tests via settings)."""
global _engine
if _engine is None:
_engine = create_engine(url or settings.database_url)
return _engine
def reset_engine() -> None:
"""Drop the cached engine (used by the test suite between DB targets)."""
global _engine
if _engine is not None:
_engine.dispose()
_engine = None
def get_session() -> Iterator[Session]:
with Session(get_engine()) as session:
yield session
def create_all(url: str | None = None) -> None:
"""Create tables directly (dev convenience; migrations own prod schema)."""
SQLModel.metadata.create_all(get_engine(url))

0
backend/app/flows/__init__.py
View file

455
backend/app/flows/api.py
View file

@ -1,455 +0,0 @@
"""Flows + approvals HTTP API."""
from __future__ import annotations
import json
from collections.abc import AsyncIterator, Iterator
from uuid import UUID
import httpx
from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import StreamingResponse
from sqlmodel import Session
from app.db import get_engine, get_session
from app.settings import settings
from . import runner, store
from .deploy import import_from_langflow
from .langflow_client import (
LangflowClient,
LangflowError,
LangflowNotFound,
LangflowUnavailable,
)
from .lfx_client import (
LfxClient,
LfxConflict,
LfxError,
LfxInvalidFlow,
LfxNotFound,
LfxRunFailed,
LfxUnavailable,
)
from .schemas import (
ApprovalCreate,
ApprovalRead,
DecisionRead,
DecisionRequest,
FlowDefinitionCreate,
FlowDefinitionRead,
FlowImportRead,
FlowImportRequest,
FlowVersionCreate,
FlowVersionRead,
RunRead,
RunRequest,
)
flows_router = APIRouter(prefix="/api/flows", tags=["flows"])
approvals_router = APIRouter(prefix="/api/approvals", tags=["approvals"])
def get_db() -> Iterator[Session]:
yield from get_session()
def get_lfx() -> LfxClient:
return LfxClient(settings.lfx_url, api_key=settings.lfx_api_key or None)
def get_langflow() -> LangflowClient:
return LangflowClient(
settings.langflow_url, api_key=settings.langflow_api_key or ""
)
def _read_definition(d) -> FlowDefinitionRead:
return FlowDefinitionRead(
id=d.id,
name=d.name,
description=d.description,
status=d.status,
continuation_flow_id=d.continuation_flow_id,
created_at=d.created_at,
)
def _read_version(v) -> FlowVersionRead:
return FlowVersionRead(
id=v.id,
flow_id=v.flow_id,
version_no=v.version_no,
langflow_json=v.langflow_json,
lfx_flow_id=v.lfx_flow_id,
deployed_at=v.deployed_at,
author=v.author,
created_at=v.created_at,
)
def _read_run(r) -> RunRead:
return RunRead(
id=r.id,
version_id=r.version_id,
tenant_id=r.tenant_id,
status=r.status,
input=r.input,
output=r.output,
session_id=r.session_id,
created_at=r.created_at,
finished_at=r.finished_at,
)
def _read_task(t) -> ApprovalRead:
return ApprovalRead(
id=t.id,
run_id=t.run_id,
version_id=t.version_id,
tenant_id=t.tenant_id,
title=t.title,
payload=t.payload,
continuation_flow_id=t.continuation_flow_id,
approver=t.approver,
status=t.status,
decision_note=t.decision_note,
created_at=t.created_at,
decided_at=t.decided_at,
)
@flows_router.post("", response_model=FlowDefinitionRead, status_code=201)
def create_flow(data: FlowDefinitionCreate, session: Session = Depends(get_db)):
try:
return _read_definition(store.create_definition(session, data))
except store.ConflictError as exc:
raise HTTPException(status_code=409, detail=str(exc)) from exc
except store.InvalidError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
@flows_router.get("", response_model=list[FlowDefinitionRead])
def list_flows(session: Session = Depends(get_db)):
return [_read_definition(d) for d in store.list_definitions(session)]
@flows_router.get("/{flow_id}", response_model=FlowDefinitionRead)
def get_flow(flow_id: UUID, session: Session = Depends(get_db)):
try:
return _read_definition(store.get_definition(session, flow_id))
except store.NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
@flows_router.post(
"/{flow_id}/versions", response_model=FlowVersionRead, status_code=201
)
def create_version(
flow_id: UUID, data: FlowVersionCreate, session: Session = Depends(get_db)
):
try:
return _read_version(store.create_version(session, flow_id, data))
except store.NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
except store.InvalidError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
@flows_router.get("/{flow_id}/versions", response_model=list[FlowVersionRead])
def list_versions(flow_id: UUID, session: Session = Depends(get_db)):
try:
return [_read_version(v) for v in store.list_versions(session, flow_id)]
except store.NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
@flows_router.post(
"/{flow_id}/versions/{version_no}/deploy", response_model=FlowVersionRead
)
async def deploy_version(
flow_id: UUID,
version_no: int,
session: Session = Depends(get_db),
lfx: LfxClient = Depends(get_lfx),
):
try:
version = store.get_version_by_no(session, flow_id, version_no)
except store.NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
try:
uploaded = await lfx.upload_flow(
version.langflow_json, flow_id=str(version.id), replace=True
)
except LfxInvalidFlow as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
except LfxConflict as exc:
raise HTTPException(status_code=409, detail=str(exc)) from exc
except LfxUnavailable as exc:
raise HTTPException(status_code=503, detail=str(exc)) from exc
except LfxError as exc:
raise HTTPException(status_code=502, detail=str(exc)) from exc
return _read_version(store.mark_deployed(session, version, uploaded.id))
def _resolve_version(session: Session, flow_id: UUID, version_no: int | None):
if version_no is not None:
return store.get_version_by_no(session, flow_id, version_no)
return store.get_latest_deployed_version(session, flow_id)
@flows_router.post("/{flow_id}/run", response_model=RunRead)
async def run_flow(
flow_id: UUID,
data: RunRequest,
stream: bool = Query(default=False),
session: Session = Depends(get_db),
lfx: LfxClient = Depends(get_lfx),
):
try:
version = _resolve_version(session, flow_id, data.version_no)
except store.NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
except store.InvalidError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
if stream:
return await _run_stream(session, lfx, version, data)
try:
run = await runner.execute_run(
session,
lfx,
version,
payload=data.input,
tenant_id=data.tenant_id,
session_id=data.session_id,
callback_base_url=data.callback_base_url or settings.callback_base_url,
continuation_flow_id=data.continuation_flow_id,
)
except LfxNotFound as exc:
raise HTTPException(status_code=502, detail=f"LFX drift: {exc}") from exc
except LfxUnavailable as exc:
raise HTTPException(status_code=503, detail=str(exc)) from exc
except LfxError as exc:
raise HTTPException(status_code=502, detail=str(exc)) from exc
return _read_run(run)
async def _run_stream(session: Session, lfx: LfxClient, version, data: RunRequest):
if not version.lfx_flow_id:
raise HTTPException(
status_code=422, detail=f"version {version.id} is not deployed to LFX"
)
run = store.create_run(
session,
version,
input=data.input,
tenant_id=data.tenant_id,
session_id=data.session_id,
)
envelope = runner.build_envelope(
payload=data.input,
run_id=run.id,
callback_base_url=data.callback_base_url or settings.callback_base_url,
continuation_flow_id=data.continuation_flow_id,
)
run_id = run.id
lfx_flow_id = version.lfx_flow_id
session_id = data.session_id or str(run.id)
async def event_gen() -> AsyncIterator[str]:
yield f"event: run\ndata: {json.dumps({'run_id': str(run_id)})}\n\n"
try:
async with httpx.AsyncClient(
base_url=lfx.base_url, timeout=lfx.timeout, headers=lfx._headers()
) as client:
async with client.stream(
"POST",
f"/flows/{lfx_flow_id}/stream",
json={"input_value": envelope, "session_id": session_id},
) as response:
if response.status_code == 404:
raise LfxNotFound(f"flow {lfx_flow_id} unknown to LFX")
response.raise_for_status()
async for chunk in response.aiter_text():
yield chunk
except (LfxNotFound, httpx.HTTPError) as exc:
yield f"event: error\ndata: {json.dumps({'error': str(exc)})}\n\n"
# Finalize from a fresh session: the request session may be closed.
with Session(get_engine()) as fresh:
try:
fresh_run = store.get_run(fresh, run_id)
status = (
"pending_approval"
if runner._task_exists(fresh, run_id)
else "completed"
)
store.set_run_result(fresh, fresh_run, status, {"streamed": True})
except store.NotFoundError:
status = "failed"
yield f"event: run_finished\ndata: {json.dumps({'run_id': str(run_id), 'status': status})}\n\n"
return StreamingResponse(event_gen(), media_type="text/event-stream")
@flows_router.get("/{flow_id}/runs", response_model=list[RunRead])
def list_runs(
flow_id: UUID,
tenant_id: str | None = Query(default=None),
session: Session = Depends(get_db),
):
try:
return [_read_run(r) for r in store.list_runs(session, flow_id, tenant_id)]
except store.NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
@flows_router.get("/langflow/available")
async def list_langflow_flows(langflow: LangflowClient = Depends(get_langflow)):
"""Browse flows on the Langflow authoring server (admin import source)."""
try:
return await langflow.list_flows()
except LangflowUnavailable as exc:
raise HTTPException(status_code=503, detail=str(exc)) from exc
except LangflowError as exc:
raise HTTPException(status_code=502, detail=str(exc)) from exc
@flows_router.post("/import", response_model=FlowImportRead, status_code=201)
async def import_flow(
data: FlowImportRequest,
session: Session = Depends(get_db),
langflow: LangflowClient = Depends(get_langflow),
lfx: LfxClient = Depends(get_lfx),
):
"""Import a flow from Langflow: export -> version -> (deploy to LFX)."""
try:
definition, version = await import_from_langflow(
session,
langflow,
lfx,
data.langflow_flow_id,
name=data.name,
author=data.author,
deploy=data.deploy,
)
except LangflowNotFound as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
except LangflowUnavailable as exc:
raise HTTPException(status_code=503, detail=str(exc)) from exc
except LangflowError as exc:
raise HTTPException(status_code=502, detail=str(exc)) from exc
except store.ConflictError as exc:
raise HTTPException(status_code=409, detail=str(exc)) from exc
except store.InvalidError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
except LfxInvalidFlow as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
except LfxConflict as exc:
raise HTTPException(status_code=409, detail=str(exc)) from exc
except LfxUnavailable as exc:
raise HTTPException(status_code=503, detail=str(exc)) from exc
except LfxError as exc:
raise HTTPException(status_code=502, detail=str(exc)) from exc
return FlowImportRead(
definition=_read_definition(definition), version=_read_version(version)
)
@flows_router.get("/runs/{run_id}", response_model=RunRead)
def get_run(
run_id: UUID,
tenant_id: str | None = Query(default=None),
session: Session = Depends(get_db),
):
"""Read a run. With tenant_id, cross-tenant reads 404 (v1 scoping)."""
try:
run = store.get_run(session, run_id)
except store.NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
if tenant_id is not None and run.tenant_id != tenant_id:
raise HTTPException(status_code=404, detail=f"run {run_id} not found")
return _read_run(run)
@approvals_router.post("", response_model=ApprovalRead, status_code=201)
def create_approval(data: ApprovalCreate, session: Session = Depends(get_db)):
"""Flow callback contract: report a terminal approval state for a run."""
try:
return _read_task(store.create_approval_task(session, data))
except store.NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
except store.ConflictError as exc:
raise HTTPException(status_code=409, detail=str(exc)) from exc
except store.InvalidError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
@approvals_router.get("", response_model=list[ApprovalRead])
def list_approvals(
tenant_id: str | None = Query(default=None),
session: Session = Depends(get_db),
):
return [_read_task(t) for t in store.list_approval_tasks(session, tenant_id)]
@approvals_router.post("/{task_id}/decision", response_model=DecisionRead)
async def decide_approval(
task_id: UUID,
data: DecisionRequest,
session: Session = Depends(get_db),
lfx: LfxClient = Depends(get_lfx),
):
"""Decide an open task. Approve starts the continuation flow (if any)."""
try:
task = store.get_task(session, task_id)
except store.NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
# Resolve the continuation first: bad references fail before any state change.
continuation_version = None
if data.decision == "approve" and task.continuation_flow_id is not None:
try:
continuation_version = store.get_latest_deployed_version(
session, task.continuation_flow_id
)
except (store.NotFoundError, store.InvalidError) as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
try:
task = store.decide_task(
session,
task,
decision="approved" if data.decision == "approve" else "rejected",
approver=data.approver,
note=data.note,
)
except store.ConflictError as exc:
raise HTTPException(status_code=409, detail=str(exc)) from exc
# Note: the decision is committed before the continuation runs. If LFX is
# down, the task stays decided and the continuation can be started
# manually from the preserved task payload.
continuation_run = None
if continuation_version is not None:
payload = task.payload
if isinstance(payload.get("facts"), dict):
payload = payload["facts"]
try:
continuation_run = await runner.execute_run(
session,
lfx,
continuation_version,
payload=payload,
tenant_id=task.tenant_id,
session_id=f"continuation-of-{task.run_id}",
callback_base_url=data.callback_base_url or settings.callback_base_url,
)
except LfxNotFound as exc:
raise HTTPException(status_code=502, detail=f"LFX drift: {exc}") from exc
except LfxUnavailable as exc:
raise HTTPException(status_code=503, detail=str(exc)) from exc
except LfxError as exc:
raise HTTPException(status_code=502, detail=str(exc)) from exc
return DecisionRead(
task=_read_task(task),
continuation_run=_read_run(continuation_run) if continuation_run else None,
)

60
backend/app/flows/deploy.py
View file

@ -1,60 +0,0 @@
"""Authoring pipeline: Langflow export -> versioned definition -> LFX deploy."""
from __future__ import annotations
from sqlmodel import Session
from app.models import FlowDefinition, FlowVersion
from . import store
from .langflow_client import LangflowClient
from .lfx_client import LfxClient
from .schemas import FlowDefinitionCreate, FlowVersionCreate
async def import_from_langflow(
session: Session,
langflow: LangflowClient,
lfx: LfxClient,
langflow_flow_id: str,
*,
name: str | None = None,
author: str = "",
deploy: bool = True,
) -> tuple[FlowDefinition, FlowVersion]:
"""Export a flow from Langflow, store it as a new version, deploy to LFX.
Definitions are matched by name (explicit `name` wins, else the exported
name); every import creates a new immutable version.
"""
export = await langflow.export_flow(langflow_flow_id)
flow_name = name or export.get("name") or f"langflow-{langflow_flow_id}"
langflow_json = {
"id": export.get("id", langflow_flow_id),
"name": flow_name,
"description": export.get("description") or "",
"endpoint_name": export.get("endpoint_name"),
"data": export.get("data", {}),
"tags": export.get("tags", []),
"webhook": export.get("webhook", False),
"is_component": export.get("is_component", False),
}
try:
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,
definition.id,
FlowVersionCreate(langflow_json=langflow_json, author=author),
)
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

66
backend/app/flows/langflow_client.py
View file

@ -1,66 +0,0 @@
"""Async client for the Langflow authoring server (visual designer).
Only used for the authoring pipeline (list/export flows). Execution always
goes through LFX (see lfx_client.py).
"""
from __future__ import annotations
import httpx
class LangflowError(Exception):
pass
class LangflowNotFound(LangflowError):
pass
class LangflowUnavailable(LangflowError):
pass
class LangflowClient:
"""Thin async wrapper around the Langflow `/api/v1/flows` endpoints."""
def __init__(self, base_url: str, api_key: str = "", timeout: float = 60.0) -> None:
self.base_url = base_url.rstrip("/")
self.api_key = api_key
self.timeout = timeout
def _headers(self) -> dict[str, str]:
return {"x-api-key": self.api_key} if self.api_key else {}
async def list_flows(self) -> list[dict]:
"""List flows (header mode keeps payloads small)."""
try:
async with httpx.AsyncClient(
base_url=self.base_url, timeout=self.timeout, headers=self._headers()
) as client:
response = await client.get("/api/v1/flows/", params={"header_flows": "true"})
if response.status_code == 404:
raise LangflowNotFound("flow list not found")
response.raise_for_status()
data = response.json()
return data if isinstance(data, list) else data.get("items", [])
except httpx.HTTPStatusError as exc:
raise LangflowError(f"Langflow error {exc.response.status_code}") from exc
except httpx.HTTPError as exc:
raise LangflowUnavailable(f"Langflow unreachable: {exc}") from exc
async def export_flow(self, langflow_flow_id: str) -> dict:
"""Export one flow as Langflow JSON (FlowRead: id/name/description/data)."""
try:
async with httpx.AsyncClient(
base_url=self.base_url, timeout=self.timeout, headers=self._headers()
) as client:
response = await client.get(f"/api/v1/flows/{langflow_flow_id}")
if response.status_code == 404:
raise LangflowNotFound(f"flow {langflow_flow_id} not found")
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as exc:
raise LangflowError(f"Langflow error {exc.response.status_code}") from exc
except httpx.HTTPError as exc:
raise LangflowUnavailable(f"Langflow unreachable: {exc}") from exc

184
backend/app/flows/lfx_client.py
View file

@ -1,184 +0,0 @@
"""Async client for the LFX flow runtime (stateless `lfx serve` process)."""
from __future__ import annotations
from dataclasses import dataclass
import httpx
class LfxError(Exception):
"""Base error for LFX communication problems."""
class LfxNotFound(LfxError):
"""Flow id unknown to LFX (HTTP 404)."""
class LfxInvalidFlow(LfxError):
"""Flow JSON rejected by LFX (HTTP 422)."""
class LfxConflict(LfxError):
"""Flow id already registered without replace=true (HTTP 409)."""
class LfxRunFailed(LfxError):
"""Flow execution failed inside LFX (HTTP 500 or success=false)."""
class LfxUnavailable(LfxError):
"""LFX process unreachable."""
@dataclass
class LfxUploadResult:
id: str
name: str
description: str | None
run_url: str
@dataclass
class LfxRunResult:
result: str
success: bool
logs: str
type: str
component: str
class LfxClient:
"""Thin async wrapper around the LFX serve HTTP API."""
def __init__(
self, base_url: str, timeout: float = 120.0, api_key: str | None = None
) -> None:
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self.api_key = api_key
def _headers(self) -> dict[str, str]:
return {"x-api-key": self.api_key} if self.api_key else {}
def _client(self) -> httpx.AsyncClient:
return httpx.AsyncClient(
base_url=self.base_url, timeout=self.timeout, headers=self._headers()
)
async def health(self) -> dict:
try:
async with self._client() as client:
response = await client.get("/health")
response.raise_for_status()
return response.json()
except httpx.HTTPError as exc:
raise LfxUnavailable(f"LFX unreachable at {self.base_url}: {exc}") from exc
async def list_flows(self) -> list[dict]:
try:
async with self._client() as client:
response = await client.get("/flows")
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as exc:
raise self._mapped(exc) from exc
except httpx.HTTPError as exc:
raise LfxUnavailable(f"LFX unreachable at {self.base_url}: {exc}") from exc
async def upload_flow(
self, flow_json: dict, *, flow_id: str | None = None, replace: bool = False
) -> LfxUploadResult:
"""Register a Langflow-export-format flow JSON. Returns the LFX flow id."""
data = flow_json.get("data")
if not isinstance(data, dict):
raise LfxInvalidFlow("flow JSON must contain a 'data' object")
body = {
"name": flow_json.get("name", "flow"),
"description": flow_json.get("description"),
"data": data,
"replace": replace,
}
if flow_id is not None:
body["id"] = flow_id
try:
async with self._client() as client:
response = await client.post("/flows/upload/", json=body)
response.raise_for_status()
payload = response.json()
return LfxUploadResult(
id=payload["id"],
name=payload.get("name", ""),
description=payload.get("description"),
run_url=payload.get("run_url", ""),
)
except httpx.HTTPStatusError as exc:
raise self._mapped(exc) from exc
except httpx.HTTPError as exc:
raise LfxUnavailable(f"LFX unreachable at {self.base_url}: {exc}") from exc
async def run_flow(
self,
lfx_flow_id: str,
*,
input_value: str,
session_id: str | None = None,
global_vars: dict[str, str] | None = None,
) -> LfxRunResult:
"""Execute a flow synchronously; raises on failure."""
body: dict = {"input_value": input_value}
if session_id is not None:
body["session_id"] = session_id
if global_vars is not None:
body["global_vars"] = global_vars
try:
async with self._client() as client:
response = await client.post(f"/flows/{lfx_flow_id}/run", json=body)
if response.status_code == 404:
raise LfxNotFound(f"flow {lfx_flow_id} unknown to LFX")
data = response.json()
result = LfxRunResult(
result=data.get("result", ""),
success=bool(data.get("success", False)),
logs=data.get("logs", ""),
type=data.get("type", ""),
component=data.get("component", ""),
)
if response.status_code >= 400 or not result.success:
raise LfxRunFailed(f"LFX run failed: {result.result}")
return result
except httpx.HTTPError as exc:
raise LfxUnavailable(f"LFX unreachable at {self.base_url}: {exc}") from exc
async def flow_info(self, lfx_flow_id: str) -> dict:
try:
async with self._client() as client:
response = await client.get(f"/flows/{lfx_flow_id}/info")
if response.status_code == 404:
raise LfxNotFound(f"flow {lfx_flow_id} unknown to LFX")
response.raise_for_status()
return response.json()
except httpx.HTTPError as exc:
raise LfxUnavailable(f"LFX unreachable at {self.base_url}: {exc}") from exc
async def delete_flow(self, lfx_flow_id: str) -> None:
try:
async with self._client() as client:
response = await client.delete(f"/flows/{lfx_flow_id}")
if response.status_code == 404:
raise LfxNotFound(f"flow {lfx_flow_id} unknown to LFX")
response.raise_for_status()
except httpx.HTTPError as exc:
raise LfxUnavailable(f"LFX unreachable at {self.base_url}: {exc}") from exc
@staticmethod
def _mapped(exc: httpx.HTTPStatusError) -> LfxError:
status = exc.response.status_code
detail = exc.response.text[:500]
if status == 404:
return LfxNotFound(detail)
if status == 409:
return LfxConflict(detail)
if status == 422:
return LfxInvalidFlow(detail)
return LfxError(f"LFX error {status}: {detail}")

93
backend/app/flows/runner.py
View file

@ -1,93 +0,0 @@
"""Run orchestration: our DB rows + LFX execution + terminal-state mapping."""
from __future__ import annotations
import json
from uuid import UUID
from sqlmodel import Session, select
from app.models import ApprovalTask, FlowRun, FlowVersion
from . import store
from .lfx_client import LfxClient, LfxRunFailed
from .store import InvalidError
def build_envelope(
*,
payload: dict,
run_id: UUID,
callback_base_url: str,
continuation_flow_id: UUID | None,
) -> str:
return json.dumps(
{
"payload": payload,
"run_id": str(run_id),
"callback_url": callback_base_url.rstrip("/"),
"continuation_flow_id": (
str(continuation_flow_id) if continuation_flow_id else None
),
}
)
def _task_exists(session: Session, run_id: UUID) -> bool:
return (
session.exec(
select(ApprovalTask).where(
ApprovalTask.run_id == run_id, ApprovalTask.status == "open"
)
).first()
is not None
)
async def execute_run(
session: Session,
lfx: LfxClient,
version: FlowVersion,
*,
payload: dict,
tenant_id: str = "default",
session_id: str | None = None,
callback_base_url: str,
continuation_flow_id: UUID | None = None,
) -> FlowRun:
"""Run a deployed version to a terminal state (no interrupts).
The flow always terminates: either it completes, or its approval branch
POSTs back to /api/approvals (creating a task) and terminates with a
pending marker. Status is derived from that side effect, not from
result-text parsing.
"""
if not version.lfx_flow_id:
raise InvalidError(f"version {version.id} is not deployed to LFX")
run = store.create_run(
session, version, input=payload, tenant_id=tenant_id, session_id=session_id
)
envelope = build_envelope(
payload=payload,
run_id=run.id,
callback_base_url=callback_base_url,
continuation_flow_id=continuation_flow_id,
)
try:
result = await lfx.run_flow(
version.lfx_flow_id,
input_value=envelope,
session_id=session_id or str(run.id),
)
except LfxRunFailed as exc:
return store.set_run_result(
session, run, "failed", {"error": str(exc)}
)
output = {
"result": result.result,
"component": result.component,
"type": result.type,
}
if _task_exists(session, run.id):
return store.set_run_result(session, run, "pending_approval", output)
return store.set_run_result(session, run, "completed", output)

112
backend/app/flows/schemas.py
View file

@ -1,112 +0,0 @@
"""Pydantic contracts for the flows API."""
from datetime import datetime
from uuid import UUID
from pydantic import BaseModel, Field
class FlowDefinitionCreate(BaseModel):
name: str = Field(min_length=1, max_length=200)
description: str = Field(default="", max_length=2000)
continuation_flow_id: UUID | None = None
class FlowDefinitionRead(BaseModel):
id: UUID
name: str
description: str
status: str
continuation_flow_id: UUID | None
created_at: datetime
class FlowVersionCreate(BaseModel):
langflow_json: dict
author: str = Field(default="", max_length=200)
class FlowVersionRead(BaseModel):
id: UUID
flow_id: UUID
version_no: int
langflow_json: dict
lfx_flow_id: str | None
deployed_at: datetime | None
author: str
created_at: datetime
class RunRequest(BaseModel):
"""Start a flow run. `input` is the domain payload handed to the flow."""
input: dict = Field(default_factory=dict)
version_no: int | None = None
tenant_id: str = Field(default="default", max_length=200)
session_id: str | None = None
# Base URL the flow uses for callbacks (e.g. approval tasks).
# Defaults to settings.callback_base_url; tests point it at a live server.
callback_base_url: str | None = None
continuation_flow_id: UUID | None = None
class RunRead(BaseModel):
id: UUID
version_id: UUID
tenant_id: str
status: str
input: dict | None
output: dict | None
session_id: str | None
created_at: datetime
finished_at: datetime | None
class ApprovalCreate(BaseModel):
"""Callback contract: a flow reports a terminal approval state."""
run_id: UUID
title: str = Field(default="", max_length=500)
payload: dict = Field(default_factory=dict)
continuation_flow_id: UUID | None = None
class ApprovalRead(BaseModel):
id: UUID
run_id: UUID
version_id: UUID
tenant_id: str
title: str
payload: dict
continuation_flow_id: UUID | None
approver: str | None
status: str
decision_note: str
created_at: datetime
decided_at: datetime | None
class FlowImportRequest(BaseModel):
"""Import a flow from the Langflow authoring server by its Langflow id."""
langflow_flow_id: str = Field(min_length=1)
name: str | None = Field(default=None, max_length=200)
author: str = Field(default="", max_length=200)
deploy: bool = True
class FlowImportRead(BaseModel):
definition: FlowDefinitionRead
version: FlowVersionRead
class DecisionRequest(BaseModel):
decision: str = Field(pattern="^(approve|reject)$")
approver: str = Field(default="", max_length=200)
note: str = Field(default="", max_length=2000)
callback_base_url: str | None = None
class DecisionRead(BaseModel):
task: ApprovalRead
continuation_run: RunRead | None

258
backend/app/flows/store.py
View file

@ -1,258 +0,0 @@
"""Persistence operations for flow definitions, versions, runs, and tasks."""
from __future__ import annotations
from datetime import datetime, timezone
from uuid import UUID
from sqlalchemy.exc import IntegrityError
from sqlmodel import Session, func, select
from app.models import ApprovalTask, FlowDefinition, FlowRun, FlowVersion
from .schemas import ApprovalCreate, FlowDefinitionCreate, FlowVersionCreate
class NotFoundError(Exception):
pass
class ConflictError(Exception):
pass
class InvalidError(Exception):
pass
def create_definition(session: Session, data: FlowDefinitionCreate) -> FlowDefinition:
if data.continuation_flow_id is not None and session.get(
FlowDefinition, data.continuation_flow_id
) is None:
raise InvalidError("continuation_flow_id unknown")
definition = FlowDefinition(
name=data.name,
description=data.description,
continuation_flow_id=data.continuation_flow_id,
)
session.add(definition)
try:
session.commit()
except IntegrityError as exc:
session.rollback()
raise ConflictError(f"flow {data.name!r} already exists") from exc
session.refresh(definition)
return definition
def get_definition(session: Session, flow_id: UUID) -> FlowDefinition:
definition = session.get(FlowDefinition, flow_id)
if definition is None:
raise NotFoundError(f"flow {flow_id} not found")
return definition
def list_definitions(session: Session) -> list[FlowDefinition]:
return list(session.exec(select(FlowDefinition).order_by(FlowDefinition.name)).all())
def _validate_flow_json(flow_json: dict) -> None:
if not isinstance(flow_json, dict):
raise InvalidError("langflow_json must be an object")
data = flow_json.get("data")
if not isinstance(data, dict):
raise InvalidError("langflow_json must contain a 'data' object")
if not isinstance(data.get("nodes"), list) or not isinstance(data.get("edges"), list):
raise InvalidError("langflow_json.data must contain 'nodes' and 'edges' lists")
def create_version(
session: Session, flow_id: UUID, data: FlowVersionCreate
) -> FlowVersion:
get_definition(session, flow_id)
_validate_flow_json(data.langflow_json)
max_no = session.exec(
select(func.max(FlowVersion.version_no)).where(FlowVersion.flow_id == flow_id)
).one()
version = FlowVersion(
flow_id=flow_id,
version_no=(max_no or 0) + 1,
langflow_json=data.langflow_json,
author=data.author,
)
session.add(version)
session.commit()
session.refresh(version)
return version
def get_version(session: Session, version_id: UUID) -> FlowVersion:
version = session.get(FlowVersion, version_id)
if version is None:
raise NotFoundError(f"flow version {version_id} not found")
return version
def list_versions(session: Session, flow_id: UUID) -> list[FlowVersion]:
get_definition(session, flow_id)
return list(
session.exec(
select(FlowVersion)
.where(FlowVersion.flow_id == flow_id)
.order_by(FlowVersion.version_no)
).all()
)
def get_version_by_no(session: Session, flow_id: UUID, version_no: int) -> FlowVersion:
version = session.exec(
select(FlowVersion).where(
FlowVersion.flow_id == flow_id, FlowVersion.version_no == version_no
)
).first()
if version is None:
raise NotFoundError(f"flow {flow_id} has no version {version_no}")
return version
def get_latest_deployed_version(session: Session, flow_id: UUID) -> FlowVersion:
version = session.exec(
select(FlowVersion)
.where(FlowVersion.flow_id == flow_id, FlowVersion.deployed_at.is_not(None))
.order_by(FlowVersion.version_no.desc())
).first()
if version is None:
raise InvalidError(f"flow {flow_id} has no deployed version")
return version
def mark_deployed(session: Session, version: FlowVersion, lfx_flow_id: str) -> FlowVersion:
version.lfx_flow_id = lfx_flow_id
version.deployed_at = datetime.now(timezone.utc)
session.add(version)
session.commit()
session.refresh(version)
return version
def create_run(
session: Session,
version: FlowVersion,
*,
input: dict | None,
tenant_id: str = "default",
session_id: str | None = None,
) -> FlowRun:
run = FlowRun(
version_id=version.id,
tenant_id=tenant_id,
input=input,
session_id=session_id or str(version.id),
)
session.add(run)
session.commit()
session.refresh(run)
return run
def set_run_result(
session: Session, run: FlowRun, status: str, output: dict | None
) -> FlowRun:
run.status = status
run.output = output
run.finished_at = datetime.now(timezone.utc)
session.add(run)
session.commit()
session.refresh(run)
return run
def get_run(session: Session, run_id: UUID) -> FlowRun:
run = session.get(FlowRun, run_id)
if run is None:
raise NotFoundError(f"run {run_id} not found")
return run
def list_runs(
session: Session, flow_id: UUID, tenant_id: str | None = None
) -> list[FlowRun]:
get_definition(session, flow_id)
query = (
select(FlowRun)
.join(FlowVersion, FlowVersion.id == FlowRun.version_id)
.where(FlowVersion.flow_id == flow_id)
.order_by(FlowRun.created_at.desc())
)
if tenant_id is not None:
query = query.where(FlowRun.tenant_id == tenant_id)
return list(session.exec(query).all())
def create_approval_task(session: Session, data: ApprovalCreate) -> ApprovalTask:
run = session.get(FlowRun, data.run_id)
if run is None:
raise NotFoundError(f"run {data.run_id} not found")
if data.continuation_flow_id is not None and session.get(
FlowDefinition, data.continuation_flow_id
) is None:
raise InvalidError("continuation_flow_id unknown")
existing = session.exec(
select(ApprovalTask).where(
ApprovalTask.run_id == data.run_id, ApprovalTask.status == "open"
)
).first()
if existing is not None:
raise ConflictError(f"run {data.run_id} already has an open approval task")
task = ApprovalTask(
run_id=run.id,
version_id=run.version_id,
tenant_id=run.tenant_id,
title=data.title,
payload=data.payload,
continuation_flow_id=data.continuation_flow_id,
)
session.add(task)
run.status = "pending_approval"
run.finished_at = datetime.now(timezone.utc)
session.add(run)
session.commit()
session.refresh(task)
return task
def list_approval_tasks(session: Session, tenant_id: str | None = None) -> list[ApprovalTask]:
query = select(ApprovalTask).order_by(ApprovalTask.created_at.desc())
if tenant_id is not None:
query = query.where(ApprovalTask.tenant_id == tenant_id)
return list(session.exec(query).all())
def get_task(session: Session, task_id: UUID) -> ApprovalTask:
task = session.get(ApprovalTask, task_id)
if task is None:
raise NotFoundError(f"approval task {task_id} not found")
return task
def decide_task(
session: Session,
task: ApprovalTask,
*,
decision: str,
approver: str = "",
note: str = "",
) -> ApprovalTask:
"""Approve or reject an open task. Decided tasks are immutable (409)."""
if task.status != "open":
raise ConflictError(f"task {task.id} is already {task.status}")
if decision not in ("approved", "rejected"):
raise InvalidError(f"unknown decision {decision!r}")
task.status = decision
task.approver = approver
task.decision_note = note
task.decided_at = datetime.now(timezone.utc)
session.add(task)
session.commit()
session.refresh(task)
return task

4
backend/app/main.py
View file

@ -1,6 +1,5 @@
from fastapi import FastAPI
from .flows.api import approvals_router, flows_router
from .settings import settings
app = FastAPI(
@ -12,9 +11,6 @@ app = FastAPI(
openapi_url="/api/openapi.json",
)
app.include_router(flows_router)
app.include_router(approvals_router)
@app.get("/")
async def root():

3
backend/app/models/__init__.py
View file

@ -1,3 +0,0 @@
from .flow import ApprovalTask, FlowDefinition, FlowRun, FlowVersion
__all__ = ["ApprovalTask", "FlowDefinition", "FlowRun", "FlowVersion"]

90
backend/app/models/flow.py
View file

@ -1,90 +0,0 @@
"""Flow engine domain models (SQLModel tables)."""
from datetime import datetime, timezone
from uuid import UUID, uuid4
from sqlalchemy import JSON, Column, DateTime, UniqueConstraint
from sqlmodel import Field, SQLModel
def utcnow() -> datetime:
return datetime.now(timezone.utc)
def _ts_column() -> Column:
return Column(DateTime(timezone=True))
def _json_column() -> Column:
# Generic JSON: JSONB on Postgres, JSON/TEXT elsewhere.
return Column(JSON)
class FlowDefinition(SQLModel, table=True):
"""A named decision flow (e.g. 'Bauvorhaben pruefen'). Versions live separately."""
__tablename__ = "flow_definitions"
id: UUID = Field(default_factory=uuid4, primary_key=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
# 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"
)
created_at: datetime = Field(default_factory=utcnow, sa_column=_ts_column())
class FlowVersion(SQLModel, table=True):
"""An immutable version of a flow: the exact Langflow JSON that runs."""
__tablename__ = "flow_versions"
__table_args__ = (UniqueConstraint("flow_id", "version_no"),)
id: UUID = Field(default_factory=uuid4, primary_key=True)
flow_id: UUID = Field(foreign_key="flow_definitions.id", index=True)
version_no: int = Field(index=True)
langflow_json: dict = Field(default_factory=dict, sa_column=_json_column())
# Flow id under which LFX serves this version (set on deploy).
lfx_flow_id: str | None = Field(default=None, max_length=200)
deployed_at: datetime | None = Field(default=None, sa_column=_ts_column())
author: str = Field(default="", max_length=200)
created_at: datetime = Field(default_factory=utcnow, sa_column=_ts_column())
class FlowRun(SQLModel, table=True):
"""One execution of a flow version."""
__tablename__ = "flow_runs"
id: UUID = Field(default_factory=uuid4, primary_key=True)
version_id: UUID = Field(foreign_key="flow_versions.id", index=True)
tenant_id: str = Field(default="default", index=True, max_length=200)
# running | completed | pending_approval | failed
status: str = Field(default="running", index=True, max_length=30)
input: dict | None = Field(default=None, sa_column=_json_column())
output: dict | None = Field(default=None, sa_column=_json_column())
session_id: str | None = Field(default=None, max_length=200)
created_at: datetime = Field(default_factory=utcnow, sa_column=_ts_column())
finished_at: datetime | None = Field(default=None, sa_column=_ts_column())
class ApprovalTask(SQLModel, table=True):
"""Human task created by a flow that reached a terminal approval state."""
__tablename__ = "approval_tasks"
id: UUID = Field(default_factory=uuid4, primary_key=True)
run_id: UUID = Field(foreign_key="flow_runs.id", index=True)
version_id: UUID = Field(foreign_key="flow_versions.id", index=True)
tenant_id: str = Field(default="default", index=True, max_length=200)
title: str = Field(default="", max_length=500)
payload: dict = Field(default_factory=dict, sa_column=_json_column())
continuation_flow_id: UUID | None = Field(default=None)
approver: str | None = Field(default=None, max_length=200)
# open | approved | rejected
status: str = Field(default="open", index=True, max_length=20)
decision_note: str = Field(default="", max_length=2000)
created_at: datetime = Field(default_factory=utcnow, sa_column=_ts_column())
decided_at: datetime | None = Field(default=None, sa_column=_ts_column())

24
backend/app/settings.py
View file

@ -1,31 +1,11 @@
from pathlib import Path
from pydantic_settings import BaseSettings, SettingsConfigDict
# Repo-root .env (shared with the compose stack). Real environment variables
# always take precedence over .env values.
REPO_ROOT = Path(__file__).resolve().parent.parent.parent
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=REPO_ROOT / ".env", extra="ignore")
version: str = "0.1.0"
debug: bool = False
database_url: str = "postgresql+psycopg://kirche:kirche@localhost:5432/kirche_einfach"
test_database_url: str = (
"postgresql+psycopg://kirche:kirche@localhost:5432/kirche_einfach_test"
)
langflow_url: str = "http://localhost:7860"
langflow_api_key: str = ""
lfx_url: str = "http://localhost:6004"
lfx_api_key: str = ""
llm_api_key: str = ""
# Base URL flows use for callbacks into this API. Must be reachable
# FROM the LFX process (compose: http://app:8000).
callback_base_url: str = "http://localhost:8000"
model_config = {"env_prefix": "KIRCHE_"}
settings = Settings()

1
backend/migrations/README
View file

@ -1 +0,0 @@
Generic single-database configuration.

84
backend/migrations/env.py
View file

@ -1,84 +0,0 @@
from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
fileConfig(config.config_file_name)
# add your model's MetaData object here
# for 'autogenerate' support
from app.models import * # noqa: F401,F403 - register all tables
from app.settings import settings
from sqlmodel import SQLModel
target_metadata = SQLModel.metadata
# The URL always comes from settings (repo-root .env or environment),
# alembic.ini's sqlalchemy.url is only a fallback placeholder.
config.set_main_option("sqlalchemy.url", settings.database_url)
# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.
def run_migrations_offline() -> None:
"""Run migrations in 'offline' mode.
This configures the context with just a URL
and not an Engine, though an Engine is acceptable
here as well. By skipping the Engine creation
we don't even need a DBAPI to be available.
Calls to context.execute() here emit the given string to the
script output.
"""
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
"""Run migrations in 'online' mode.
In this scenario we need to create an Engine
and associate a connection with the context.
"""
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(
connection=connection, target_metadata=target_metadata
)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()

29
backend/migrations/script.py.mako
View file

@ -1,29 +0,0 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
import sqlmodel # noqa: F401 - autogenerate renders sqlmodel.* column types
${imports if imports else ""}
# revision identifiers, used by Alembic.
revision: str = ${repr(up_revision)}
down_revision: Union[str, Sequence[str], None] = ${repr(down_revision)}
branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
def upgrade() -> None:
"""Upgrade schema."""
${upgrades if upgrades else "pass"}
def downgrade() -> None:
"""Downgrade schema."""
${downgrades if downgrades else "pass"}

108
backend/migrations/versions/62ea9a5d8f9d_flow_engine_tables.py
View file

@ -1,108 +0,0 @@
"""flow engine tables
Revision ID: 62ea9a5d8f9d
Revises:
Create Date: 2026-09-06 22:27:07.129788
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
import sqlmodel
# revision identifiers, used by Alembic.
revision: str = '62ea9a5d8f9d'
down_revision: Union[str, Sequence[str], None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('flow_definitions',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('name', sqlmodel.sql.sqltypes.AutoString(length=200), nullable=False),
sa.Column('description', sqlmodel.sql.sqltypes.AutoString(length=2000), nullable=False),
sa.Column('status', sqlmodel.sql.sqltypes.AutoString(length=20), nullable=False),
sa.Column('continuation_flow_id', sa.Uuid(), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['continuation_flow_id'], ['flow_definitions.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_flow_definitions_name'), 'flow_definitions', ['name'], unique=True)
op.create_table('flow_versions',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('flow_id', sa.Uuid(), nullable=False),
sa.Column('version_no', sa.Integer(), nullable=False),
sa.Column('langflow_json', sa.JSON(), nullable=True),
sa.Column('lfx_flow_id', sqlmodel.sql.sqltypes.AutoString(length=200), nullable=True),
sa.Column('deployed_at', sa.DateTime(timezone=True), nullable=True),
sa.Column('author', sqlmodel.sql.sqltypes.AutoString(length=200), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['flow_id'], ['flow_definitions.id'], ),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('flow_id', 'version_no')
)
op.create_index(op.f('ix_flow_versions_flow_id'), 'flow_versions', ['flow_id'], unique=False)
op.create_index(op.f('ix_flow_versions_version_no'), 'flow_versions', ['version_no'], unique=False)
op.create_table('flow_runs',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('version_id', sa.Uuid(), nullable=False),
sa.Column('tenant_id', sqlmodel.sql.sqltypes.AutoString(length=200), nullable=False),
sa.Column('status', sqlmodel.sql.sqltypes.AutoString(length=30), nullable=False),
sa.Column('input', sa.JSON(), nullable=True),
sa.Column('output', sa.JSON(), nullable=True),
sa.Column('session_id', sqlmodel.sql.sqltypes.AutoString(length=200), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), nullable=True),
sa.Column('finished_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['version_id'], ['flow_versions.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_flow_runs_status'), 'flow_runs', ['status'], unique=False)
op.create_index(op.f('ix_flow_runs_tenant_id'), 'flow_runs', ['tenant_id'], unique=False)
op.create_index(op.f('ix_flow_runs_version_id'), 'flow_runs', ['version_id'], unique=False)
op.create_table('approval_tasks',
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('run_id', sa.Uuid(), nullable=False),
sa.Column('version_id', sa.Uuid(), nullable=False),
sa.Column('tenant_id', sqlmodel.sql.sqltypes.AutoString(length=200), nullable=False),
sa.Column('title', sqlmodel.sql.sqltypes.AutoString(length=500), nullable=False),
sa.Column('payload', sa.JSON(), nullable=True),
sa.Column('continuation_flow_id', sa.Uuid(), nullable=True),
sa.Column('approver', sqlmodel.sql.sqltypes.AutoString(length=200), nullable=True),
sa.Column('status', sqlmodel.sql.sqltypes.AutoString(length=20), nullable=False),
sa.Column('decision_note', sqlmodel.sql.sqltypes.AutoString(length=2000), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), nullable=True),
sa.Column('decided_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['run_id'], ['flow_runs.id'], ),
sa.ForeignKeyConstraint(['version_id'], ['flow_versions.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_approval_tasks_run_id'), 'approval_tasks', ['run_id'], unique=False)
op.create_index(op.f('ix_approval_tasks_status'), 'approval_tasks', ['status'], unique=False)
op.create_index(op.f('ix_approval_tasks_tenant_id'), 'approval_tasks', ['tenant_id'], unique=False)
op.create_index(op.f('ix_approval_tasks_version_id'), 'approval_tasks', ['version_id'], unique=False)
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_index(op.f('ix_approval_tasks_version_id'), table_name='approval_tasks')
op.drop_index(op.f('ix_approval_tasks_tenant_id'), table_name='approval_tasks')
op.drop_index(op.f('ix_approval_tasks_status'), table_name='approval_tasks')
op.drop_index(op.f('ix_approval_tasks_run_id'), table_name='approval_tasks')
op.drop_table('approval_tasks')
op.drop_index(op.f('ix_flow_runs_version_id'), table_name='flow_runs')
op.drop_index(op.f('ix_flow_runs_tenant_id'), table_name='flow_runs')
op.drop_index(op.f('ix_flow_runs_status'), table_name='flow_runs')
op.drop_table('flow_runs')
op.drop_index(op.f('ix_flow_versions_version_no'), table_name='flow_versions')
op.drop_index(op.f('ix_flow_versions_flow_id'), table_name='flow_versions')
op.drop_table('flow_versions')
op.drop_index(op.f('ix_flow_definitions_name'), table_name='flow_definitions')
op.drop_table('flow_definitions')
# ### end Alembic commands ###

18
backend/pyproject.toml
View file

@ -4,12 +4,9 @@ version = "0.1.0"
description = "Kirche Einfach - KI-gestuetzte Verwaltungsplattform fuer Kirche"
requires-python = ">=3.12"
dependencies = [
"alembic>=1.14",
"fastapi[standard]>=0.115.0",
"httpx>=0.28",
"psycopg[binary]>=3.2",
"django>=5.1",
"pydantic-settings>=2.0",
"sqlmodel>=0.0.24",
"uvicorn[standard]>=0.30.0",
]
@ -20,16 +17,3 @@ build-backend = "uv_build"
[tool.uv.build-backend]
module-name = "app"
module-root = ""
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
[dependency-groups]
dev = [
"lfx==1.12.0",
"pytest>=8",
"pytest-asyncio>=0.24",
"pytest-cov>=6",
"respx>=0.22",
]

54
backend/scripts/deploy_flow.py
View file

@ -1,54 +0,0 @@
"""Import a flow from Langflow into our backend via the running API.
Usage:
uv run python scripts/deploy_flow.py --api http://localhost:8000 \\
--langflow-id <uuid> [--name my-flow] [--author jane] [--no-deploy]
"""
from __future__ import annotations
import argparse
import json
import sys
import urllib.request
def post(api: str, path: str, body: dict) -> dict:
request = urllib.request.Request(
api.rstrip("/") + path,
data=json.dumps(body).encode(),
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(request) as response:
return {"status": response.status, "body": json.loads(response.read())}
except urllib.error.HTTPError as exc:
return {"status": exc.code, "body": exc.read().decode()[:2000]}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--api", default="http://localhost:8000")
parser.add_argument("--langflow-id", required=True)
parser.add_argument("--name", default=None)
parser.add_argument("--author", default="")
parser.add_argument("--no-deploy", action="store_true")
args = parser.parse_args()
result = post(
args.api,
"/api/flows/import",
{
"langflow_flow_id": args.langflow_id,
"name": args.name,
"author": args.author,
"deploy": not args.no_deploy,
},
)
print(json.dumps(result, indent=2, ensure_ascii=False))
return 0 if result["status"] in (200, 201) else 1
if __name__ == "__main__":
sys.exit(main())

444
backend/scripts/make_flow_fixtures.py
View file

@ -1,444 +0,0 @@
"""Generate Langflow-export-format flow JSON fixtures.
Uses lfx's own serializers (Component.to_frontend_node /
build_custom_component_template), so the output is byte-faithful to what the
Langflow UI would export. Run with the backend venv (has the `lfx` dev dep)::
uv run python scripts/make_flow_fixtures.py
Then verify::
uv run lfx validate --level 4 backend/tests/fixtures/flows/
"""
from __future__ import annotations
import importlib
import json
import sys
import uuid
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent.parent
FIXTURE_DIR = REPO_ROOT / "backend" / "tests" / "fixtures" / "flows"
HANDLE_QUOTE = "\u0153" # ReactFlow handle-id quoting used by Langflow exports
# --------------------------------------------------------------------------
# node builders
# --------------------------------------------------------------------------
def _normalize(node_dict: dict, node_id: str, comp_name: str, x: int, y: int) -> dict:
# to_frontend_node() returns an envelope {"data": {...}, "id": ...};
# build_custom_component_template() returns the bare node dict.
if "data" not in node_dict:
node_dict = {
"data": {"node": node_dict, "type": comp_name, "id": node_id},
"id": node_id,
}
else:
node_dict["id"] = node_id
node_dict["data"]["id"] = node_id
node_dict["data"]["type"] = comp_name
node_dict["data"]["showNode"] = True
node_dict["measured"] = {"width": 320, "height": 200}
node_dict["position"] = {"x": x, "y": y}
node_dict["type"] = "genericNode"
return node_dict
def builtin_node(
module_path: str,
class_name: str,
node_id: str,
values: dict | None = None,
position: tuple[int, int] = (0, 0),
) -> dict:
mod = importlib.import_module(module_path)
cls = getattr(mod, class_name)
inst = cls()
node_dict = _normalize(inst.to_frontend_node(), node_id, cls.name, *position)
for field, value in (values or {}).items():
node_dict["data"]["node"]["template"][field]["value"] = value
return node_dict
def custom_node(
code: str,
node_id: str,
values: dict | None = None,
position: tuple[int, int] = (0, 0),
) -> dict:
from lfx.custom import validate as validate_mod
from lfx.custom.utils import build_custom_component_template
cls = validate_mod.create_class(code, validate_mod.extract_class_name(code))
inst = cls()
inst._code = code
bare, _ = build_custom_component_template(inst)
node_dict = _normalize(bare, node_id, "CustomComponent", *position)
for field, value in (values or {}).items():
node_dict["data"]["node"]["template"][field]["value"] = value
return node_dict
# --------------------------------------------------------------------------
# edges
# --------------------------------------------------------------------------
def _output_def(node: dict, output_name: str) -> dict:
for out in node["data"]["node"]["outputs"]:
if out["name"] == output_name:
return out
raise KeyError(f"output {output_name!r} not in {node['id']}")
def _input_def(node: dict, input_name: str) -> dict:
field = node["data"]["node"]["template"][input_name]
return field
def edge(src: dict, src_output: str, dst: dict, dst_input: str) -> dict:
out = _output_def(src, src_output)
inp = _input_def(dst, dst_input)
source_handle = {
"dataType": src["data"]["type"],
"id": src["id"],
"name": src_output,
"output_types": out.get("types", []),
}
target_handle = {
"fieldName": dst_input,
"id": dst["id"],
"inputTypes": inp.get("input_types", []),
"type": inp.get("type", "other"),
}
sh = json.dumps(source_handle).replace('"', HANDLE_QUOTE)
th = json.dumps(target_handle).replace('"', HANDLE_QUOTE)
return {
"animated": False,
"className": "",
"data": {"sourceHandle": source_handle, "targetHandle": target_handle},
"id": f"xy-edge__{sh}-{th}",
"selected": False,
"source": src["id"],
"sourceHandle": sh,
"target": dst["id"],
"targetHandle": th,
}
# --------------------------------------------------------------------------
# flow document
# --------------------------------------------------------------------------
def flow_doc(
name: str, description: str, nodes: list[dict], edges: list[dict]
) -> dict:
return {
"data": {
"nodes": nodes,
"edges": edges,
"viewport": {"x": 0, "y": 0, "zoom": 1},
},
"description": description,
"endpoint_name": None,
"icon": None,
"icon_bg_color": None,
"id": str(uuid.uuid5(uuid.NAMESPACE_URL, f"kirche-einfach/{name}")),
"is_component": False,
"locked": False,
"mcp_enabled": False,
"name": name,
"tags": [],
"webhook": False,
}
# --------------------------------------------------------------------------
# custom component code
# --------------------------------------------------------------------------
RULE_CHECK_CODE = '''from lfx.custom import Component
from lfx.io import MessageTextInput, Output
from lfx.schema.message import Message
import json
class RuleCheckComponent(Component):
display_name = "Rule Check"
description = "Bewertet Betrag gegen Schwelle aus JSON-Eingabe."
inputs = [
MessageTextInput(name="payload", display_name="Payload"),
]
outputs = [
Output(display_name="Verdict", name="verdict", method="check"),
]
def _text(self, value) -> str:
if value is None:
return ""
if hasattr(value, "text"):
return value.text or ""
return str(value)
def check(self) -> Message:
try:
data = json.loads(self._text(self.payload) or "{}")
except Exception:
data = {}
if not isinstance(data, dict):
data = {}
# Die API ruft Flows immer mit Huelle auf
# {"payload": {...}, "run_id": ..., ...}; rohe Payloads gehen auch.
if isinstance(data.get("payload"), dict):
data = data["payload"]
try:
betrag = float(data.get("betrag", 0) or 0)
except (TypeError, ValueError):
betrag = 0.0
try:
schwelle = float(data.get("schwelle", 200000) or 200000)
except (TypeError, ValueError):
schwelle = 200000.0
return Message(text="PRUEFUNG" if betrag > schwelle else "OK")
'''
ENVELOPE_CODE = '''from lfx.custom import Component
from lfx.io import MessageTextInput, Output
from lfx.schema.message import Message
import json
class EnvelopeComponent(Component):
display_name = "Envelope"
description = "Zerlegt die Eingabe-Huelle in Payload und Laufzeit-Kontext."
inputs = [
MessageTextInput(name="raw", display_name="Raw Input"),
]
outputs = [
Output(display_name="Payload", name="payload", method="get_payload"),
Output(display_name="Run ID", name="run_id", method="get_run_id"),
Output(
display_name="Callback URL",
name="callback_url",
method="get_callback_url",
),
Output(
display_name="Continuation",
name="continuation",
method="get_continuation",
),
]
def _text(self, value) -> str:
if value is None:
return ""
if hasattr(value, "text"):
return value.text or ""
return str(value)
def _env(self) -> dict:
if not hasattr(self, "_env_cache"):
try:
self._env_cache = json.loads(self._text(self.raw) or "{}")
except Exception:
self._env_cache = {}
if not isinstance(self._env_cache, dict):
self._env_cache = {}
return self._env_cache
def get_payload(self) -> Message:
return Message(text=json.dumps(self._env().get("payload", {})))
def get_run_id(self) -> Message:
return Message(text=str(self._env().get("run_id", "")))
def get_callback_url(self) -> Message:
return Message(text=str(self._env().get("callback_url", "")))
def get_continuation(self) -> Message:
value = self._env().get("continuation_flow_id")
return Message(text=str(value) if value else "")
'''
APPROVAL_CALLBACK_CODE = '''from lfx.custom import Component
from lfx.io import MessageTextInput, Output
from lfx.schema.message import Message
class ApprovalCallbackComponent(Component):
display_name = "Approval Callback"
description = "Meldet einen Sonderfall per HTTP an unsere API und terminiert."
inputs = [
MessageTextInput(name="trigger", display_name="Trigger"),
MessageTextInput(name="run_id", display_name="Run ID"),
MessageTextInput(name="callback_url", display_name="Callback URL"),
MessageTextInput(
name="continuation_flow_id",
display_name="Continuation Flow ID",
),
MessageTextInput(name="facts", display_name="Facts"),
]
outputs = [
Output(display_name="Result", name="result", method="notify"),
]
def _text(self, value) -> str:
if value is None:
return ""
if hasattr(value, "text"):
return value.text or ""
return str(value)
def notify(self) -> Message:
import httpx
import json
trigger = self._text(self.trigger).strip()
if not trigger:
# Zweig wurde nicht genommen: kein Seiteneffekt.
return Message(text="")
try:
facts = json.loads(self._text(self.facts) or "{}")
except Exception:
facts = {}
url = self._text(self.callback_url).rstrip("/") + "/api/approvals"
body = {
"run_id": self._text(self.run_id),
"title": f"Freigabe erforderlich ({trigger})",
"payload": {"verdict": trigger, "facts": facts},
"continuation_flow_id": self._text(self.continuation_flow_id) or None,
}
try:
response = httpx.post(url, json=body, timeout=10.0)
response.raise_for_status()
except Exception as exc:
return Message(text=f"CALLBACK_FAILED: {exc}")
return Message(text="PENDING_HUMAN_APPROVAL")
'''
# --------------------------------------------------------------------------
# flows
# --------------------------------------------------------------------------
ROUTER_VALUES = {
"operator": "equals",
"match_text": "PRUEFUNG",
"true_case_message": "Sonderfall: menschliche Pruefung erforderlich",
"false_case_message": "Standardfall: automatisch freigegeben",
}
def build_simple_rule() -> dict:
entry = builtin_node(
"lfx.components.input_output.chat",
"ChatInput",
"ChatInput-entry",
position=(0, 200),
)
rule = custom_node(RULE_CHECK_CODE, "CustomComponent-rule", position=(400, 200))
router = builtin_node(
"lfx.components.flow_controls.conditional_router",
"ConditionalRouterComponent",
"ConditionalRouter-router",
values=ROUTER_VALUES,
position=(800, 200),
)
out_ok = builtin_node(
"lfx.components.input_output.chat_output",
"ChatOutput",
"ChatOutput-ok",
position=(1200, 60),
)
out_sonder = builtin_node(
"lfx.components.input_output.chat_output",
"ChatOutput",
"ChatOutput-sonder",
position=(1200, 340),
)
nodes = [entry, rule, router, out_ok, out_sonder]
edges = [
edge(entry, "message", rule, "payload"),
edge(rule, "verdict", router, "input_text"),
edge(router, "false_result", out_ok, "input_value"),
edge(router, "true_result", out_sonder, "input_value"),
]
return flow_doc(
"simple_rule",
"Regelpruefung ohne Seiteneffekt (Spike-Fixture, kein LLM).",
nodes,
edges,
)
def build_approval_handoff() -> dict:
entry = builtin_node(
"lfx.components.input_output.chat",
"ChatInput",
"ChatInput-entry",
position=(0, 300),
)
envelope = custom_node(
ENVELOPE_CODE, "CustomComponent-envelope", position=(400, 300)
)
rule = custom_node(RULE_CHECK_CODE, "CustomComponent-rule", position=(800, 300))
router = builtin_node(
"lfx.components.flow_controls.conditional_router",
"ConditionalRouterComponent",
"ConditionalRouter-router",
values=ROUTER_VALUES,
position=(1200, 300),
)
callback = custom_node(
APPROVAL_CALLBACK_CODE,
"CustomComponent-callback",
position=(1600, 160),
)
out_pending = builtin_node(
"lfx.components.input_output.chat_output",
"ChatOutput",
"ChatOutput-pending",
position=(2000, 160),
)
out_ok = builtin_node(
"lfx.components.input_output.chat_output",
"ChatOutput",
"ChatOutput-ok",
position=(2000, 440),
)
nodes = [entry, envelope, rule, router, callback, out_pending, out_ok]
edges = [
edge(entry, "message", envelope, "raw"),
edge(envelope, "payload", rule, "payload"),
edge(rule, "verdict", router, "input_text"),
edge(router, "true_result", callback, "trigger"),
edge(envelope, "run_id", callback, "run_id"),
edge(envelope, "callback_url", callback, "callback_url"),
edge(envelope, "continuation", callback, "continuation_flow_id"),
edge(envelope, "payload", callback, "facts"),
edge(callback, "result", out_pending, "input_value"),
edge(router, "false_result", out_ok, "input_value"),
]
return flow_doc(
"approval_handoff",
"Sonderfall meldet sich per HTTP an unsere API und terminiert (Spike).",
nodes,
edges,
)
def main() -> None:
FIXTURE_DIR.mkdir(parents=True, exist_ok=True)
flows = {
"simple_rule.json": build_simple_rule(),
"approval_handoff.json": build_approval_handoff(),
}
for filename, doc in flows.items():
path = FIXTURE_DIR / filename
path.write_text(json.dumps(doc, indent=2, ensure_ascii=False) + "\n")
print(f"wrote {path}")
if __name__ == "__main__":
sys.exit(main())

127
backend/scripts/smoke_e2e.py
View file

@ -1,127 +0,0 @@
"""Manual end-to-end smoke: real LFX + real API, frontend-shaped calls.
Usage (from backend/):
TEST_DATABASE_URL=postgresql+psycopg://... uv run python scripts/smoke_e2e.py
Exits non-zero on any unexpected status/body.
"""
from __future__ import annotations
import json
import os
import subprocess
import sys
import time
import urllib.request
from pathlib import Path
BACKEND = Path(__file__).resolve().parent.parent
VENV_BIN = BACKEND / ".venv" / "bin"
FIXTURES = BACKEND / "tests" / "fixtures" / "flows"
LFX_PORT = 6004
API_PORT = 8000
LFX_KEY = "lfx-759GsfJKjAeOBZ_44-cll9WdHp_yfYn_tIg7eHG38Rs"
def wait_healthy(url: str, timeout: int = 180) -> None:
deadline = time.time() + timeout
while time.time() < deadline:
try:
with urllib.request.urlopen(url, timeout=3) as r:
if r.status == 200:
return
except OSError:
time.sleep(1)
raise RuntimeError(f"never healthy: {url}")
def api(method: str, path: str, body: dict | None = None) -> tuple[int, dict | str]:
import httpx
with httpx.Client(base_url=f"http://127.0.0.1:{API_PORT}", timeout=120) as c:
r = c.request(method, path, json=body)
try:
return r.status_code, r.json()
except ValueError:
return r.status_code, r.text
def check(name: str, cond: bool, detail: str = "") -> None:
print(f"[{'OK' if cond else 'FAIL'}] {name} {detail}")
if not cond:
raise SystemExit(f"smoke failed: {name} {detail}")
def main() -> None:
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()
)
env = {
**os.environ,
"LFX_API_KEY": LFX_KEY,
"DATABASE_URL": os.environ["TEST_DATABASE_URL"],
"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(
[str(VENV_BIN / "lfx"), "serve", "--host", "127.0.0.1",
"--port", str(LFX_PORT), "--flow-dir", str(flow_dir), str(flow_dir)],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=env,
)
uvicorn = subprocess.Popen(
[str(VENV_BIN / "uvicorn"), "app.main:app", "--host", "127.0.0.1",
"--port", str(API_PORT)],
cwd=str(BACKEND), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
env=env,
)
try:
wait_healthy(f"http://127.0.0.1:{LFX_PORT}/health")
wait_healthy(f"http://127.0.0.1:{API_PORT}/health")
s, flows = api("GET", "/api/flows")
check("list flows", s == 200 and isinstance(flows, list))
s, flow = api("POST", "/api/flows", {"name": "smoke-dach"})
check("create flow", s == 201, json.dumps(flow)[:120])
fid = flow["id"]
fx = json.loads((FIXTURES / "simple_rule.json").read_text())
s, ver = api(
"POST", f"/api/flows/{fid}/versions",
{"langflow_json": fx, "author": "smoke"},
)
check("create version", s == 201, f"v{ver.get('version_no')}")
s, dep = api("POST", f"/api/flows/{fid}/versions/1/deploy")
check("deploy", s == 200 and dep.get("lfx_flow_id"), json.dumps(dep)[:160])
s, run = api(
"POST", f"/api/flows/{fid}/run",
{"input": {"betrag": 120000, "schwelle": 200000}},
)
check("run standard", s == 200 and run.get("status") == "completed",
json.dumps(run.get("output"))[:120])
s, run2 = api(
"POST", f"/api/flows/{fid}/run",
{"input": {"betrag": 999000, "schwelle": 200000}},
)
check("run sonderfall", s == 200 and "Sonderfall" in (run2.get("output") or {}).get("result", ""))
s, tasks = api("GET", "/api/approvals")
check("approvals empty (no callback flow)", s == 200 and tasks == [])
print("smoke E2E: all green")
finally:
uvicorn.terminate()
lfx.terminate()
if __name__ == "__main__":
if "TEST_DATABASE_URL" not in os.environ:
sys.exit("TEST_DATABASE_URL must be set")
main()

220
backend/tests/conftest.py
View file

@ -1,220 +0,0 @@
"""Shared pytest fixtures.
Requires TEST_DATABASE_URL (a dedicated Postgres database, e.g. the
`kirche_einfach_test` database from the compose stack). The URL is mapped
onto DATABASE_URL *before* any app module is imported, so settings and the
alembic env both target the test database. Never point this at production.
"""
import os
from pathlib import Path
import pytest
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"] = TEST_DB_URL
BACKEND_DIR = Path(__file__).resolve().parent.parent
@pytest.fixture(scope="session")
def db_engine():
"""Engine on the test DB, migrated to head once per session."""
from alembic import command as alembic_command
from alembic.config import Config
from app import db as app_db
app_db.reset_engine()
engine = app_db.get_engine()
cfg = Config()
cfg.set_main_option("script_location", str(BACKEND_DIR / "migrations"))
alembic_command.upgrade(cfg, "head")
yield engine
engine.dispose()
app_db.reset_engine()
@pytest.fixture()
def session(db_engine):
"""Function-scoped session for direct DB access in tests."""
from sqlmodel import Session
with Session(db_engine) as s:
yield s
@pytest.fixture(autouse=True)
def _clean_tables(db_engine):
"""Truncate all domain tables before and after every test (isolation)."""
from sqlmodel import SQLModel
def _truncate() -> None:
with db_engine.begin() as conn:
for table in reversed(SQLModel.metadata.sorted_tables):
conn.execute(table.delete())
_truncate()
yield
_truncate()
@pytest.fixture()
async def client():
"""ASGI test client for the FastAPI app."""
from httpx import ASGITransport, AsyncClient
from app.main import app
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as c:
yield c
@pytest.fixture(scope="session")
def lfx_server(tmp_path_factory):
"""Real `lfx serve` subprocess on a free port; points settings at it.
Serves an (initially seeded) flow dir; tests upload fixture flows through
the same deploy path production uses.
"""
import shutil
import socket
import subprocess
import sys
import time
import urllib.request
from pathlib import Path
from app.settings import settings
bindir = Path(sys.executable).parent
lfx_bin = shutil.which("lfx", path=str(bindir)) or "lfx"
sock = socket.socket()
sock.bind(("127.0.0.1", 0))
port = sock.getsockname()[1]
sock.close()
flow_dir = tmp_path_factory.mktemp("lfx-flows")
seed = (
Path(__file__).parent / "fixtures" / "flows" / "simple_rule.json"
).read_text()
(flow_dir / "seed.json").write_text(seed)
env = {
**os.environ,
# lfx serve refuses to start without a serve key; requests need it too.
"LFX_API_KEY": "test-lfx-key",
}
proc = subprocess.Popen(
[
lfx_bin,
"serve",
"--host",
"127.0.0.1",
"--port",
str(port),
"--flow-dir",
str(flow_dir),
str(flow_dir),
],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
env=env,
)
url = f"http://127.0.0.1:{port}"
deadline = time.time() + 240
healthy = False
while time.time() < deadline:
if proc.poll() is not None:
out = proc.stdout.read() if proc.stdout else ""
raise RuntimeError(f"lfx serve exited early:\n{out[-4000:]}")
try:
with urllib.request.urlopen(url + "/health", timeout=3) as response:
if response.status == 200:
healthy = True
break
except OSError:
time.sleep(1)
if not healthy:
proc.terminate()
raise RuntimeError("lfx serve did not become healthy in time")
old_url = settings.lfx_url
old_key = settings.lfx_api_key
settings.lfx_url = url
settings.lfx_api_key = "test-lfx-key"
yield url
settings.lfx_url = old_url
settings.lfx_api_key = old_key
proc.terminate()
try:
proc.wait(timeout=20)
except subprocess.TimeoutExpired:
proc.kill()
@pytest.fixture()
def override_langflow():
"""Replace the Langflow dependency with a fake (real LFX stays untouched).
Usage: override_langflow(export_doc) -- the fake export_flow() returns the
doc, or raises LangflowNotFound/Unavailable for status=404/0.
"""
from app.flows.api import get_langflow
from app.flows.langflow_client import LangflowNotFound, LangflowUnavailable
from app.main import app
def _override(export_doc=None, status=200):
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")
return export_doc
async def list_flows(self):
return [{"id": "fake-1", "name": "fake"}]
app.dependency_overrides[get_langflow] = lambda: FakeLangflow()
yield _override
app.dependency_overrides.pop(get_langflow, None)
@pytest.fixture(scope="session")
def live_server():
"""Real uvicorn server (ephemeral port) so LFX flows can call back
into our API over HTTP, exactly like production."""
import threading
import time
import uvicorn
from app.main import app
config = uvicorn.Config(app, host="127.0.0.1", port=0, log_level="warning")
server = uvicorn.Server(config)
thread = threading.Thread(target=server.run, daemon=True)
thread.start()
deadline = time.time() + 30
while not server.started and time.time() < deadline:
time.sleep(0.05)
assert server.started, "uvicorn test server did not start"
port = server.servers[0].sockets[0].getsockname()[1]
yield f"http://127.0.0.1:{port}"
server.should_exit = True
thread.join(timeout=15)

1945
backend/tests/fixtures/flows/approval_handoff.json vendored
View file

File diff suppressed because one or more lines are too long

1393
backend/tests/fixtures/flows/simple_rule.json vendored
View file

File diff suppressed because one or more lines are too long

255
backend/tests/test_approvals.py
View file

@ -1,255 +0,0 @@
"""Step 5 tests: decisions + continuations, tenant isolation, error paths."""
import json
from pathlib import Path
from uuid import UUID, uuid4
from app.flows.schemas import ApprovalCreate
FIXTURES = Path(__file__).parent / "fixtures" / "flows"
def load_fixture(name: str) -> dict:
return json.loads((FIXTURES / name).read_text())
async def _create_flow(client, name):
response = await client.post("/api/flows", json={"name": name})
assert response.status_code == 201, response.text
return response.json()
async def _version_and_deploy(client, flow_id, fixture="simple_rule.json"):
response = await client.post(
f"/api/flows/{flow_id}/versions",
json={"langflow_json": load_fixture(fixture)},
)
assert response.status_code == 201, response.text
response = await client.post(f"/api/flows/{flow_id}/versions/1/deploy")
assert response.status_code == 200, response.text
return response.json()
async def _handoff_task(client, live_server, continuation_flow_id=None):
"""Run approval_handoff over threshold; returns (run, task)."""
flow = await _create_flow(client, f"handoff-{uuid4().hex[:8]}")
await _version_and_deploy(client, flow["id"], fixture="approval_handoff.json")
body = {
"input": {"betrag": 250000, "schwelle": 200000},
"callback_base_url": live_server,
}
if continuation_flow_id is not None:
body["continuation_flow_id"] = continuation_flow_id
run = (
await client.post(f"/api/flows/{flow['id']}/run", json=body)
).json()
assert run["status"] == "pending_approval"
tasks = (await client.get("/api/approvals")).json()
task = next(t for t in tasks if t["run_id"] == run["id"])
return run, task
async def test_approve_without_continuation(client, lfx_server, live_server):
_, task = await _handoff_task(client, live_server)
response = await client.post(
f"/api/approvals/{task['id']}/decision",
json={"decision": "approve", "approver": "sachbearbeiter", "note": "ok"},
)
assert response.status_code == 200, response.text
body = response.json()
assert body["task"]["status"] == "approved"
assert body["task"]["approver"] == "sachbearbeiter"
assert body["task"]["decided_at"] is not None
assert body["continuation_run"] is None
async def test_approve_with_continuation_runs_facts(client, lfx_server, live_server):
cont = await _create_flow(client, f"cont-{uuid4().hex[:8]}")
await _version_and_deploy(client, cont["id"], fixture="simple_rule.json")
_, task = await _handoff_task(
client, live_server, continuation_flow_id=cont["id"]
)
assert task["continuation_flow_id"] == cont["id"]
response = await client.post(
f"/api/approvals/{task['id']}/decision",
json={"decision": "approve", "approver": "chef"},
)
assert response.status_code == 200, response.text
body = response.json()
assert body["task"]["status"] == "approved"
cont_run = body["continuation_run"]
assert cont_run is not None
assert cont_run["status"] == "completed"
assert cont_run["tenant_id"] == task["tenant_id"]
# Facts (betrag 250000 > Schwelle) flowed into the continuation.
assert "Sonderfall" in cont_run["output"]["result"]
async def test_reject_task(client, lfx_server, live_server):
_, task = await _handoff_task(client, live_server)
response = await client.post(
f"/api/approvals/{task['id']}/decision",
json={"decision": "reject", "approver": "chef", "note": "zu teuer"},
)
assert response.status_code == 200, response.text
body = response.json()
assert body["task"]["status"] == "rejected"
assert body["task"]["decision_note"] == "zu teuer"
assert body["continuation_run"] is None
async def test_decision_guards(client, lfx_server, live_server):
_, task = await _handoff_task(client, live_server)
url = f"/api/approvals/{task['id']}/decision"
response = await client.post(url, json={"decision": "approve"})
assert response.status_code == 200
response = await client.post(url, json={"decision": "reject"})
assert response.status_code == 409
response = await client.post(
f"/api/approvals/{uuid4()}/decision", json={"decision": "approve"}
)
assert response.status_code == 404
_, task2 = await _handoff_task(client, live_server)
response = await client.post(
f"/api/approvals/{task2['id']}/decision", json={"decision": "maybe"}
)
assert response.status_code == 422
response = await client.post(
f"/api/approvals/{task2['id']}/decision",
json={"decision": "approve", "note": "x"},
)
# continuation unset -> fine, resolves to no continuation run
assert response.status_code == 200
async def test_decision_bad_continuation_reference(client, session):
from app.flows import store as flow_store
flow = await _create_flow(client, f"badcont-{uuid4().hex[:8]}")
version = (
await client.post(
f"/api/flows/{flow['id']}/versions",
json={"langflow_json": load_fixture("simple_rule.json")},
)
).json()
run = flow_store.create_run(
session,
flow_store.get_version(session, UUID(version["id"])),
input={},
)
task = flow_store.create_approval_task(
session, ApprovalCreate(run_id=run.id, title="t")
)
# plant unknown continuation directly (store allows it pre-validation)
task.continuation_flow_id = uuid4()
session.add(task)
session.commit()
response = await client.post(
f"/api/approvals/{task.id}/decision", json={"decision": "approve"}
)
assert response.status_code == 422
async def test_tenant_isolation(client, lfx_server):
flow = await _create_flow(client, f"tenant-{uuid4().hex[:8]}")
await _version_and_deploy(client, flow["id"])
for tenant in ("pfarramt-a", "pfarramt-b"):
response = await client.post(
f"/api/flows/{flow['id']}/run",
json={"input": {"betrag": 1}, "tenant_id": tenant},
)
assert response.status_code == 200
response = await client.get(
f"/api/flows/{flow['id']}/runs", params={"tenant_id": "pfarramt-a"}
)
runs_a = response.json()
assert len(runs_a) == 1
assert runs_a[0]["tenant_id"] == "pfarramt-a"
response = await client.get(f"/api/flows/{flow['id']}/runs")
assert len(response.json()) == 2
# Cross-tenant read with scoping 404s.
response = await client.get(
f"/api/flows/runs/{runs_a[0]['id']}", params={"tenant_id": "pfarramt-b"}
)
assert response.status_code == 404
response = await client.get(
f"/api/flows/runs/{runs_a[0]['id']}", params={"tenant_id": "pfarramt-a"}
)
assert response.status_code == 200
async def test_lfx_down_maps_to_503(client, lfx_server, monkeypatch):
from app.settings import settings
flow = await _create_flow(client, f"down-{uuid4().hex[:8]}")
await client.post(
f"/api/flows/{flow['id']}/versions",
json={"langflow_json": load_fixture("simple_rule.json")},
)
monkeypatch.setattr(settings, "lfx_url", "http://127.0.0.1:9")
response = await client.post(f"/api/flows/{flow['id']}/versions/1/deploy")
assert response.status_code == 503
# Deploy for real, then break the URL again for the run call.
monkeypatch.setattr(settings, "lfx_url", lfx_server)
deployed = await client.post(f"/api/flows/{flow['id']}/versions/1/deploy")
assert deployed.status_code == 200
monkeypatch.setattr(settings, "lfx_url", "http://127.0.0.1:9")
response = await client.post(
f"/api/flows/{flow['id']}/run", json={"input": {"betrag": 1}}
)
assert response.status_code == 503
async def test_lfx_drift_maps_to_502(client, lfx_server):
from app.flows.lfx_client import LfxClient
from app.settings import settings
flow = await _create_flow(client, f"drift-{uuid4().hex[:8]}")
deployed = await _version_and_deploy(client, flow["id"])
await LfxClient(settings.lfx_url, api_key=settings.lfx_api_key or None).delete_flow(
deployed["lfx_flow_id"]
)
response = await client.post(
f"/api/flows/{flow['id']}/run", json={"input": {"betrag": 1}}
)
assert response.status_code == 502
async def test_corrupt_flow_json_deploy_maps_to_422(client, lfx_server):
flow = await _create_flow(client, f"corrupt-{uuid4().hex[:8]}")
bogus = {
"name": "bogus",
"data": {
"nodes": [
{
"id": "Nope-x1",
"data": {
"id": "Nope-x1",
"type": "Nope",
"showNode": True,
"node": {"template": {}, "outputs": []},
},
"type": "genericNode",
}
],
"edges": [],
},
}
response = await client.post(
f"/api/flows/{flow['id']}/versions", json={"langflow_json": bogus}
)
assert response.status_code == 201, response.text
response = await client.post(f"/api/flows/{flow['id']}/versions/1/deploy")
assert response.status_code == 422

109
backend/tests/test_deploy.py
View file

@ -1,109 +0,0 @@
"""Step 4 tests: Langflow import pipeline (Langflow faked, LFX real)."""
import json
from pathlib import Path
from uuid import uuid4
FIXTURES = Path(__file__).parent / "fixtures" / "flows"
LANGFLOW_ID = "11111111-2222-4333-8444-555555555555"
def _export_doc(name="importiert"):
fixture = json.loads((FIXTURES / "simple_rule.json").read_text())
return {
"id": LANGFLOW_ID,
"name": name,
"description": "aus Langflow",
"endpoint_name": None,
"data": fixture["data"],
"tags": [],
"webhook": False,
"is_component": False,
}
async def test_import_creates_definition_version_and_deploys(
client, lfx_server, override_langflow
):
override_langflow(_export_doc())
response = await client.post(
"/api/flows/import",
json={"langflow_flow_id": LANGFLOW_ID, "author": "admin"},
)
assert response.status_code == 201, response.text
body = response.json()
assert body["definition"]["name"] == "importiert"
assert body["version"]["version_no"] == 1
assert body["version"]["author"] == "admin"
assert body["version"]["lfx_flow_id"] == body["version"]["id"]
assert body["version"]["deployed_at"] is not None
assert body["version"]["langflow_json"]["data"]["nodes"]
# Imported flow actually runs on LFX.
definition_id = body["definition"]["id"]
response = await client.post(
f"/api/flows/{definition_id}/run", json={"input": {"betrag": 10}}
)
assert response.status_code == 200, response.text
assert response.json()["status"] == "completed"
async def test_import_reuses_definition_and_bumps_version(
client, lfx_server, override_langflow
):
override_langflow(_export_doc())
first = await client.post(
"/api/flows/import", json={"langflow_flow_id": LANGFLOW_ID}
)
second = await client.post(
"/api/flows/import", json={"langflow_flow_id": LANGFLOW_ID}
)
assert first.status_code == 201
assert second.status_code == 201
assert first.json()["definition"]["id"] == second.json()["definition"]["id"]
assert second.json()["version"]["version_no"] == 2
async def test_import_with_explicit_name(client, lfx_server, override_langflow):
override_langflow(_export_doc(name="langflow-name"))
response = await client.post(
"/api/flows/import",
json={"langflow_flow_id": LANGFLOW_ID, "name": "wunschname"},
)
assert response.status_code == 201, response.text
assert response.json()["definition"]["name"] == "wunschname"
async def test_import_without_deploy(client, override_langflow):
override_langflow(_export_doc())
response = await client.post(
"/api/flows/import",
json={"langflow_flow_id": LANGFLOW_ID, "deploy": False},
)
assert response.status_code == 201, response.text
body = response.json()
assert body["version"]["lfx_flow_id"] is None
assert body["version"]["deployed_at"] is None
async def test_import_langflow_not_found(client, override_langflow):
override_langflow(status=404)
response = await client.post(
"/api/flows/import", json={"langflow_flow_id": LANGFLOW_ID}
)
assert response.status_code == 404
async def test_import_langflow_unreachable(client, override_langflow):
override_langflow(status=0)
response = await client.post(
"/api/flows/import", json={"langflow_flow_id": str(uuid4())}
)
assert response.status_code == 503
async def test_list_langflow_flows(client, override_langflow):
override_langflow(_export_doc())
response = await client.get("/api/flows/langflow/available")
assert response.status_code == 200
assert response.json() == [{"id": "fake-1", "name": "fake"}]

248
backend/tests/test_flows_api.py
View file

@ -1,248 +0,0 @@
"""Step 3 tests: flows CRUD, versions, deploy, sync/stream runs, handoff E2E."""
import json
from pathlib import Path
from uuid import uuid4
FIXTURES = Path(__file__).parent / "fixtures" / "flows"
def load_fixture(name: str) -> dict:
return json.loads((FIXTURES / name).read_text())
async def _create_flow(client, name="regeltest"):
response = await client.post(
"/api/flows", json={"name": name, "description": "test flow"}
)
assert response.status_code == 201, response.text
return response.json()
async def _create_version(client, flow_id, fixture="simple_rule.json"):
response = await client.post(
f"/api/flows/{flow_id}/versions",
json={"langflow_json": load_fixture(fixture), "author": "test"},
)
assert response.status_code == 201, response.text
return response.json()
async def _deploy(client, flow_id, version_no=1):
response = await client.post(f"/api/flows/{flow_id}/versions/{version_no}/deploy")
assert response.status_code == 200, response.text
return response.json()
async def test_flow_crud(client):
created = await _create_flow(client, "crud-flow")
assert created["status"] == "active"
response = await client.get(f"/api/flows/{created['id']}")
assert response.status_code == 200
assert response.json()["name"] == "crud-flow"
response = await client.get("/api/flows")
assert response.status_code == 200
assert any(f["id"] == created["id"] for f in response.json())
response = await client.post("/api/flows", json={"name": "crud-flow"})
assert response.status_code == 409
response = await client.get(f"/api/flows/{uuid4()}")
assert response.status_code == 404
async def test_version_validation_and_numbering(client):
flow = await _create_flow(client, "versioned")
v1 = await _create_version(client, flow["id"])
assert v1["version_no"] == 1
v2 = await _create_version(client, flow["id"])
assert v2["version_no"] == 2
assert v1["deployed_at"] is None
response = await client.post(
f"/api/flows/{flow['id']}/versions", json={"langflow_json": {"nope": 1}}
)
assert response.status_code == 422
response = await client.post(
f"/api/flows/{uuid4()}/versions",
json={"langflow_json": load_fixture("simple_rule.json")},
)
assert response.status_code == 404
response = await client.get(f"/api/flows/{flow['id']}/versions")
assert [v["version_no"] for v in response.json()] == [1, 2]
async def test_deploy_and_run_standard_and_sonderfall(client, lfx_server):
flow = await _create_flow(client, "dachsanierung")
await _create_version(client, flow["id"])
deployed = await _deploy(client, flow["id"])
assert deployed["lfx_flow_id"] == deployed["id"]
assert deployed["deployed_at"] is not None
response = await client.post(
f"/api/flows/{flow['id']}/run",
json={"input": {"betrag": 120000, "schwelle": 200000}},
)
assert response.status_code == 200, response.text
run = response.json()
assert run["status"] == "completed"
assert "Standardfall" in run["output"]["result"]
assert run["finished_at"] is not None
response = await client.post(
f"/api/flows/{flow['id']}/run",
json={"input": {"betrag": 250000, "schwelle": 200000}},
)
assert response.status_code == 200, response.text
run = response.json()
assert run["status"] == "completed"
assert "Sonderfall" in run["output"]["result"]
response = await client.get(f"/api/flows/runs/{run['id']}")
assert response.status_code == 200
assert response.json()["id"] == run["id"]
async def test_run_undeployed_version_fails(client):
flow = await _create_flow(client, "undeployed")
await _create_version(client, flow["id"])
response = await client.post(
f"/api/flows/{flow['id']}/run", json={"input": {"betrag": 1}}
)
assert response.status_code == 422
async def test_run_explicit_version_no(client, lfx_server):
flow = await _create_flow(client, "explicit-version")
await _create_version(client, flow["id"])
await _create_version(client, flow["id"])
await _deploy(client, flow["id"], version_no=2)
response = await client.post(
f"/api/flows/{flow['id']}/run",
json={"input": {"betrag": 1}, "version_no": 2},
)
assert response.status_code == 200, response.text
assert response.json()["status"] == "completed"
async def test_run_stream_events(client, lfx_server):
flow = await _create_flow(client, "streamed")
await _create_version(client, flow["id"])
await _deploy(client, flow["id"])
events: list[tuple[str, str]] = []
async with client.stream(
"POST",
f"/api/flows/{flow['id']}/run?stream=true",
json={"input": {"betrag": 1}},
) as response:
assert response.status_code == 200, await response.aread()
current_event = "message"
async for line in response.aiter_lines():
if line.startswith("event:"):
current_event = line.split(":", 1)[1].strip()
elif line.startswith("data:"):
events.append((current_event, line.split(":", 1)[1].strip()))
current_event = "message"
assert events, "expected SSE events"
assert events[0][0] == "run"
first_payload = json.loads(events[0][1])
assert "run_id" in first_payload
assert events[-1][0] == "run_finished"
last_payload = json.loads(events[-1][1])
assert last_payload["status"] == "completed"
response = await client.get(f"/api/flows/runs/{first_payload['run_id']}")
assert response.status_code == 200
assert response.json()["status"] == "completed"
async def test_approval_handoff_end_to_end(client, lfx_server, live_server):
flow = await _create_flow(client, "freigabe")
await _create_version(client, flow["id"], fixture="approval_handoff.json")
await _deploy(client, flow["id"])
response = await client.post(
f"/api/flows/{flow['id']}/run",
json={
"input": {"betrag": 250000, "schwelle": 200000},
"callback_base_url": live_server,
},
)
assert response.status_code == 200, response.text
run = response.json()
assert run["status"] == "pending_approval"
assert "PENDING_HUMAN_APPROVAL" in run["output"]["result"]
response = await client.get("/api/approvals")
assert response.status_code == 200
tasks = response.json()
assert len(tasks) == 1
task = tasks[0]
assert task["run_id"] == run["id"]
assert task["status"] == "open"
assert task["payload"]["facts"]["betrag"] == 250000
async def test_approval_handoff_auto_creates_no_task(client, lfx_server, live_server):
flow = await _create_flow(client, "freigabe-auto")
await _create_version(client, flow["id"], fixture="approval_handoff.json")
await _deploy(client, flow["id"])
response = await client.post(
f"/api/flows/{flow['id']}/run",
json={
"input": {"betrag": 50000, "schwelle": 200000},
"callback_base_url": live_server,
},
)
assert response.status_code == 200, response.text
assert response.json()["status"] == "completed"
response = await client.get("/api/approvals")
assert response.status_code == 200
assert response.json() == []
async def test_approval_callback_contract(client, session):
from uuid import UUID
from app.flows import store as flow_store
flow = await _create_flow(client, "callback-target")
version = await _create_version(client, flow["id"])
version_obj = flow_store.get_version(session, UUID(version["id"]))
run = flow_store.create_run(session, version_obj, input={}, tenant_id="default")
response = await client.post(
"/api/approvals",
json={"run_id": str(run.id), "title": "bitte pruefen", "payload": {"x": 1}},
)
assert response.status_code == 201, response.text
task = response.json()
assert task["status"] == "open"
assert task["tenant_id"] == "default"
assert task["version_id"] == version["id"]
response = await client.get(f"/api/flows/runs/{run.id}")
assert response.json()["status"] == "pending_approval"
response = await client.post("/api/approvals", json={"run_id": str(run.id)})
assert response.status_code == 409
run2 = flow_store.create_run(session, version_obj, input={}, tenant_id="default")
response = await client.post(
"/api/approvals",
json={"run_id": str(run2.id), "continuation_flow_id": str(uuid4())},
)
assert response.status_code == 422
response = await client.post("/api/approvals", json={"run_id": str(uuid4())})
assert response.status_code == 404

52
backend/tests/test_langflow_client.py
View file

@ -1,52 +0,0 @@
"""LangflowClient HTTP mapping tests (Langflow mocked via respx, no LFX)."""
import httpx
import pytest
import respx
from app.flows.langflow_client import (
LangflowClient,
LangflowNotFound,
LangflowUnavailable,
)
BASE = "https://langflow.test"
def _client():
return LangflowClient(BASE, api_key="test-key")
@respx.mock
async def test_export_flow_sends_api_key_and_returns_doc():
doc = {"id": "abc", "name": "n", "data": {"nodes": [], "edges": []}}
route = respx.get(f"{BASE}/api/v1/flows/abc").mock(
return_value=httpx.Response(200, json=doc)
)
result = await _client().export_flow("abc")
assert result == doc
assert route.called
assert route.calls[0].request.headers["x-api-key"] == "test-key"
@respx.mock
async def test_export_flow_not_found():
respx.get(f"{BASE}/api/v1/flows/missing").mock(
return_value=httpx.Response(404, json={"detail": "nope"})
)
with pytest.raises(LangflowNotFound):
await _client().export_flow("missing")
@respx.mock
async def test_list_flows_unwraps_items():
respx.get(f"{BASE}/api/v1/flows/").mock(
return_value=httpx.Response(200, json={"items": [{"id": "1"}]})
)
assert await _client().list_flows() == [{"id": "1"}]
async def test_unreachable_raises_unavailable():
client = LangflowClient("https://langflow.dead", api_key="k")
with pytest.raises(LangflowUnavailable):
await client.export_flow("whatever")

82
backend/tests/test_models.py
View file

@ -1,82 +0,0 @@
"""Model-level tests: defaults, constraints, JSON roundtrip, linkage."""
import pytest
from sqlalchemy.exc import IntegrityError
from sqlmodel import select
from app.models import ApprovalTask, FlowDefinition, FlowRun, FlowVersion
def test_flow_definition_defaults(session):
d = FlowDefinition(name="bauvorhaben")
session.add(d)
session.commit()
session.refresh(d)
assert d.id is not None
assert d.status == "active"
assert d.continuation_flow_id is None
assert d.created_at is not None
assert d.created_at.tzinfo is not None
def test_flow_definition_name_unique(session):
session.add(FlowDefinition(name="doppelt"))
session.commit()
session.add(FlowDefinition(name="doppelt"))
with pytest.raises(IntegrityError):
session.commit()
session.rollback()
def test_flow_version_unique_per_flow(session):
d = FlowDefinition(name="v-flow")
session.add(d)
session.commit()
session.add(FlowVersion(flow_id=d.id, version_no=1, author="a"))
session.commit()
session.add(FlowVersion(flow_id=d.id, version_no=1, author="b"))
with pytest.raises(IntegrityError):
session.commit()
session.rollback()
session.add(FlowVersion(flow_id=d.id, version_no=2, author="b"))
session.commit()
def test_flow_version_json_roundtrip(session):
payload = {"data": {"nodes": [{"id": "n1"}], "edges": []}, "name": "x"}
d = FlowDefinition(name="json-flow")
session.add(d)
session.commit()
v = FlowVersion(flow_id=d.id, version_no=1, langflow_json=payload)
session.add(v)
session.commit()
session.refresh(v)
assert v.langflow_json == payload
assert v.deployed_at is None
assert v.lfx_flow_id is None
def test_run_and_task_linkage_with_tenant_default(session):
d = FlowDefinition(name="linked")
session.add(d)
session.commit()
v = FlowVersion(flow_id=d.id, version_no=1)
session.add(v)
session.commit()
run = FlowRun(version_id=v.id, input={"a": 1})
session.add(run)
session.commit()
task = ApprovalTask(run_id=run.id, version_id=v.id, title="bitte pruefen")
session.add(task)
session.commit()
assert run.tenant_id == "default"
assert run.status == "running"
assert task.tenant_id == "default"
assert task.status == "open"
runs = session.exec(select(FlowRun).where(FlowRun.version_id == v.id)).all()
assert len(runs) == 1
tasks = session.exec(select(ApprovalTask).where(ApprovalTask.run_id == run.id)).all()
assert len(tasks) == 1
assert tasks[0].payload == {}

37
frontend/src/App.tsx
View file

@ -1,39 +1,10 @@
import { useState } from "react";
import AdminTab from "./components/AdminTab";
import ApprovalsTab from "./components/ApprovalsTab";
import FlowsTab from "./components/FlowsTab";
type Tab = "flows" | "approvals" | "admin";
const tabStyle = (active: boolean): React.CSSProperties => ({
fontWeight: active ? "bold" : "normal",
textDecoration: active ? "underline" : "none",
});
export default function App() {
const [tab, setTab] = useState<Tab>("flows");
function App() {
return (
<div style={{ maxWidth: 900, margin: "0 auto", padding: 16 }}>
<div>
<h1>Kirche Einfach</h1>
<p>KI-gestützte Verwaltungsplattform für Kirche</p>
<nav style={{ display: "flex", gap: 12, marginBottom: 16 }}>
<button style={tabStyle(tab === "flows")} onClick={() => setTab("flows")}>
Vorgänge
</button>
<button
style={tabStyle(tab === "approvals")}
onClick={() => setTab("approvals")}
>
Freigaben
</button>
<button style={tabStyle(tab === "admin")} onClick={() => setTab("admin")}>
Administration
</button>
</nav>
{tab === "flows" && <FlowsTab />}
{tab === "approvals" && <ApprovalsTab />}
{tab === "admin" && <AdminTab />}
</div>
);
}
export default App;

138
frontend/src/api.ts
View file

@ -1,138 +0,0 @@
/* Typed client for the kirche-einfach API (served via vite /api proxy). */
export interface FlowDefinition {
id: string;
name: string;
description: string;
status: string;
continuation_flow_id: string | null;
created_at: string;
}
export interface FlowVersion {
id: string;
flow_id: string;
version_no: number;
langflow_json: unknown;
lfx_flow_id: string | null;
deployed_at: string | null;
author: string;
created_at: string;
}
export interface RunOutput {
result?: string;
component?: string;
type?: string;
error?: string;
[key: string]: unknown;
}
export interface Run {
id: string;
version_id: string;
tenant_id: string;
status: string;
input: unknown;
output: RunOutput | null;
session_id: string | null;
created_at: string;
finished_at: string | null;
}
export interface ApprovalTask {
id: string;
run_id: string;
version_id: string;
tenant_id: string;
title: string;
payload: unknown;
continuation_flow_id: string | null;
approver: string | null;
status: string;
decision_note: string;
created_at: string;
decided_at: string | null;
}
export interface Decision {
task: ApprovalTask;
continuation_run: Run | null;
}
export class ApiError extends Error {
status: number;
constructor(status: number, message: string) {
super(message);
this.status = status;
}
}
async function req<T>(path: string, init?: RequestInit): Promise<T> {
const res = await fetch(path, {
headers: { "Content-Type": "application/json" },
...init,
});
if (!res.ok) {
let detail = res.statusText;
try {
const body: unknown = await res.json();
if (
typeof body === "object" &&
body !== null &&
"detail" in body &&
typeof (body as { detail: unknown }).detail === "string"
) {
detail = (body as { detail: string }).detail;
}
} catch {
/* keep statusText */
}
throw new ApiError(res.status, detail);
}
return (await res.json()) as T;
}
export const LANGFLOW_URL =
(import.meta.env.VITE_LANGFLOW_URL as string | undefined) ??
"http://localhost:7860";
export const api = {
listFlows: () => req<FlowDefinition[]>("/api/flows"),
listVersions: (flowId: string) =>
req<FlowVersion[]>(`/api/flows/${flowId}/versions`),
deployVersion: (flowId: string, versionNo: number) =>
req<FlowVersion>(`/api/flows/${flowId}/versions/${versionNo}/deploy`, {
method: "POST",
}),
runFlow: (flowId: string, input: unknown, tenantId: string) =>
req<Run>(`/api/flows/${flowId}/run`, {
method: "POST",
body: JSON.stringify({ input, tenant_id: tenantId }),
}),
listRuns: (flowId: string) => req<Run[]>(`/api/flows/${flowId}/runs`),
listApprovals: () => req<ApprovalTask[]>("/api/approvals"),
decide: (
taskId: string,
decision: "approve" | "reject",
approver: string,
note: string,
) =>
req<Decision>(`/api/approvals/${taskId}/decision`, {
method: "POST",
body: JSON.stringify({ decision, approver, note }),
}),
importFlow: (langflowFlowId: string, name?: string) =>
req<{ definition: FlowDefinition; version: FlowVersion }>(
"/api/flows/import",
{
method: "POST",
body: JSON.stringify({
langflow_flow_id: langflowFlowId,
...(name ? { name } : {}),
}),
},
),
listLangflow: () =>
req<{ id: string; name: string }[]>("/api/flows/langflow/available"),
};

118
frontend/src/components/AdminTab.tsx
View file

@ -1,118 +0,0 @@
import { useCallback, useEffect, useState } from "react";
import {
ApiError,
LANGFLOW_URL,
api,
type FlowDefinition,
type FlowVersion,
} from "../api";
export default function AdminTab() {
const [flows, setFlows] = useState<FlowDefinition[]>([]);
const [versions, setVersions] = useState<Record<string, FlowVersion[]>>({});
const [langflowId, setLangflowId] = useState("");
const [importName, setImportName] = useState("");
const [error, setError] = useState("");
const [info, setInfo] = useState("");
const load = useCallback(async () => {
try {
const list = await api.listFlows();
setFlows(list);
const all: Record<string, FlowVersion[]> = {};
for (const f of list) {
all[f.id] = await api.listVersions(f.id);
}
setVersions(all);
} catch (e) {
setError(e instanceof ApiError ? `${e.status}: ${e.message}` : String(e));
}
}, []);
useEffect(() => {
void load();
}, [load]);
async function onImport() {
setError("");
setInfo("");
try {
const result = await api.importFlow(
langflowId,
importName.trim() ? importName.trim() : undefined,
);
setInfo(
`Importiert: ${result.definition.name} (Version ${result.version.version_no})`,
);
setLangflowId("");
setImportName("");
await load();
} catch (e) {
setError(e instanceof ApiError ? `${e.status}: ${e.message}` : String(e));
}
}
async function onDeploy(flowId: string, versionNo: number) {
setError("");
setInfo("");
try {
const version = await api.deployVersion(flowId, versionNo);
setInfo(`Deployed: Version ${version.version_no} (${version.lfx_flow_id})`);
await load();
} catch (e) {
setError(e instanceof ApiError ? `${e.status}: ${e.message}` : String(e));
}
}
return (
<div>
<h2>Administration</h2>
{error && <p style={{ color: "crimson" }}>{error}</p>}
{info && <p style={{ color: "green" }}>{info}</p>}
<p>
Flows werden visuell in Langflow entworfen:{" "}
<a href={LANGFLOW_URL} target="_blank" rel="noreferrer">
Langflow öffnen
</a>
</p>
<h3>Import aus Langflow</h3>
<div style={{ display: "flex", gap: 8, marginBottom: 12 }}>
<input
value={langflowId}
onChange={(e) => setLangflowId(e.target.value)}
placeholder="Langflow Flow-ID (UUID)"
style={{ flex: 1, fontFamily: "monospace" }}
/>
<input
value={importName}
onChange={(e) => setImportName(e.target.value)}
placeholder="Name (optional)"
/>
<button onClick={() => void onImport()} disabled={!langflowId.trim()}>
Importieren + Deployen
</button>
</div>
<h3>Flows + Versionen</h3>
{flows.length === 0 && <p>Keine Flows vorhanden.</p>}
<ul>
{flows.map((f) => (
<li key={f.id} style={{ marginBottom: 8 }}>
<strong>{f.name}</strong>
<ul>
{(versions[f.id] ?? []).map((v) => (
<li key={v.id}>
v{v.version_no} ·{" "}
{v.deployed_at ? `deployed (${v.lfx_flow_id})` : "nicht deployed"} ·{" "}
{v.author || "–"}{" "}
<button onClick={() => void onDeploy(f.id, v.version_no)}>
Deployen
</button>
</li>
))}
</ul>
</li>
))}
</ul>
</div>
);
}

95
frontend/src/components/ApprovalsTab.tsx
View file

@ -1,95 +0,0 @@
import { useCallback, useEffect, useState } from "react";
import { ApiError, api, type ApprovalTask } from "../api";
export default function ApprovalsTab() {
const [tasks, setTasks] = useState<ApprovalTask[]>([]);
const [approver, setApprover] = useState("");
const [note, setNote] = useState("");
const [error, setError] = useState("");
const [busyId, setBusyId] = useState("");
const load = useCallback(async () => {
try {
setTasks(await api.listApprovals());
} catch (e) {
setError(e instanceof ApiError ? `${e.status}: ${e.message}` : String(e));
}
}, []);
useEffect(() => {
void load();
}, [load]);
async function decide(taskId: string, decision: "approve" | "reject") {
setBusyId(taskId);
setError("");
try {
const result = await api.decide(taskId, decision, approver, note);
setTasks((prev) =>
prev.map((t) => (t.id === taskId ? result.task : t)),
);
} catch (e) {
setError(e instanceof ApiError ? `${e.status}: ${e.message}` : String(e));
} finally {
setBusyId("");
}
}
const open = tasks.filter((t) => t.status === "open");
const done = tasks.filter((t) => t.status !== "open");
return (
<div>
<h2>Freigaben</h2>
{error && <p style={{ color: "crimson" }}>{error}</p>}
<div style={{ display: "flex", gap: 8, marginBottom: 12 }}>
<input
value={approver}
onChange={(e) => setApprover(e.target.value)}
placeholder="Name (Bearbeiter)"
/>
<input
value={note}
onChange={(e) => setNote(e.target.value)}
placeholder="Notiz"
style={{ flex: 1 }}
/>
<button onClick={() => void load()}>Aktualisieren</button>
</div>
<h3>Offen ({open.length})</h3>
{open.length === 0 && <p>Keine offenen Freigaben.</p>}
<ul>
{open.map((t) => (
<li key={t.id} style={{ marginBottom: 8 }}>
<strong>{t.title || t.id}</strong> · {t.tenant_id}
<br />
<code>{JSON.stringify(t.payload)}</code>
<br />
<button
onClick={() => void decide(t.id, "approve")}
disabled={busyId === t.id}
>
Freigeben
</button>{" "}
<button
onClick={() => void decide(t.id, "reject")}
disabled={busyId === t.id}
>
Ablehnen
</button>
</li>
))}
</ul>
<h3>Entschieden ({done.length})</h3>
<ul>
{done.map((t) => (
<li key={t.id}>
<strong>{t.status}</strong> · {t.title || t.id} ·{" "}
{t.approver ?? "–"}
{t.decision_note ? ` · „${t.decision_note}“` : ""}
</li>
))}
</ul>
</div>
);
}

105
frontend/src/components/FlowsTab.tsx
View file

@ -1,105 +0,0 @@
import { useCallback, useEffect, useState } from "react";
import { ApiError, api, type FlowDefinition, type Run } from "../api";
const inputStyle: React.CSSProperties = {
width: "100%",
fontFamily: "monospace",
fontSize: 13,
};
export default function FlowsTab() {
const [flows, setFlows] = useState<FlowDefinition[]>([]);
const [flowId, setFlowId] = useState("");
const [inputJson, setInputJson] = useState(
'{\n "betrag": 120000,\n "schwelle": 200000\n}',
);
const [tenant, setTenant] = useState("default");
const [runs, setRuns] = useState<Run[]>([]);
const [busy, setBusy] = useState(false);
const [error, setError] = useState("");
const loadFlows = useCallback(async () => {
try {
const list = await api.listFlows();
setFlows(list);
const first = list[0];
if (first && !flowId) setFlowId(first.id);
} catch (e) {
setError(e instanceof ApiError ? `${e.status}: ${e.message}` : String(e));
}
}, [flowId]);
const loadRuns = useCallback(async (id: string) => {
if (!id) return;
try {
setRuns(await api.listRuns(id));
} catch (e) {
setError(e instanceof ApiError ? `${e.status}: ${e.message}` : String(e));
}
}, []);
useEffect(() => {
void loadFlows();
}, [loadFlows]);
useEffect(() => {
void loadRuns(flowId);
}, [flowId, loadRuns]);
async function onRun() {
setBusy(true);
setError("");
try {
const payload: unknown = JSON.parse(inputJson);
const run = await api.runFlow(flowId, payload, tenant);
setRuns((prev) => [run, ...prev]);
} catch (e) {
if (e instanceof SyntaxError) setError("Ungültiges JSON in der Eingabe.");
else setError(e instanceof ApiError ? `${e.status}: ${e.message}` : String(e));
} finally {
setBusy(false);
}
}
return (
<div>
<h2>Vorgang starten</h2>
{error && <p style={{ color: "crimson" }}>{error}</p>}
<div style={{ display: "flex", gap: 8, marginBottom: 8 }}>
<select value={flowId} onChange={(e) => setFlowId(e.target.value)}>
<option value="">— Flow wählen —</option>
{flows.map((f) => (
<option key={f.id} value={f.id}>
{f.name}
</option>
))}
</select>
<input
value={tenant}
onChange={(e) => setTenant(e.target.value)}
placeholder="Mandant"
style={{ width: 160 }}
/>
<button onClick={() => void onRun()} disabled={!flowId || busy}>
{busy ? "Läuft …" : "Starten"}
</button>
</div>
<textarea
rows={6}
style={inputStyle}
value={inputJson}
onChange={(e) => setInputJson(e.target.value)}
/>
<h3>Letzte Läufe</h3>
{runs.length === 0 && <p>Noch keine Läufe.</p>}
<ul>
{runs.map((r) => (
<li key={r.id}>
<strong>{r.status}</strong> · {r.tenant_id} ·{" "}
<code>{r.output?.result ?? r.output?.error ?? "–"}</code>
</li>
))}
</ul>
</div>
);
}

1
frontend/src/vite-env.d.ts vendored
View file

@ -1 +0,0 @@
/// <reference types="vite/client" />