refactor: form-core wiring, JWT sessions, slowapi, packaging, frontend rewrite, test declutter

backend/app/flows/api.py

@@ -17,12 +17,13 @@ from pathlib import Path
 from typing import Annotated
 from uuid import UUID
 
-from fastapi import APIRouter, Depends, Header, HTTPException, Query
+from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request
 from fastapi.responses import StreamingResponse
 from sqlmodel import Session
 
 from app.auth.deps import get_current_user, require_admin
 from app.db import get_db
+from app.middleware import limiter, run_limit
 from app.models import User
 from app.settings import settings
 
@@ -50,6 +51,7 @@ from .schemas import (
     FlowVersionCreate,
     FlowVersionRead,
     InputSchemaRead,
+    NamedRunRead,
     RunRead,
     RunRequest,
     SyncReport,
@@ -58,6 +60,7 @@ from .sync import lfx_id_for
 
 flows_router = APIRouter(prefix="/api/flows", tags=["flows"])
 approvals_router = APIRouter(prefix="/api/approvals", tags=["approvals"])
+runs_router = APIRouter(prefix="/api/runs", tags=["runs"])
 
 logger = logging.getLogger(__name__)
 
@@ -211,7 +214,9 @@ def _resolve_version(session: Session, flow_id: UUID, version_no: int | None):
 
 
 @flows_router.post("/{flow_id}/run", response_model=RunRead)
+@limiter.limit(run_limit)
 async def run_flow(
+    request: Request,
     flow_id: UUID,
     data: RunRequest,
     session: Annotated[Session, Depends(get_db)],
@@ -279,6 +284,18 @@ async def _run_stream(
     return StreamingResponse(event_gen(), media_type="text/event-stream")
 
 
+@runs_router.get("", response_model=list[NamedRunRead])
+def list_all_runs(
+    session: Annotated[Session, Depends(get_db)],
+    user: Annotated[User, Depends(get_current_user)],
+):
+    """All runs of the session tenant with flow names (single query)."""
+    return [
+        NamedRunRead(flow_name=name, run=_read_run(run))
+        for run, name in store.list_all_runs(session, user.tenant_id)
+    ]
+
+
 @flows_router.get("/{flow_id}/runs", response_model=list[RunRead])
 def list_runs(
     flow_id: UUID,

backend/app/flows/base_client.py

@@ -2,12 +2,19 @@
 
 Consolidates base_url stripping, header building, per-request client
 construction, and HTTPError → domain-error mapping. Subclasses only
-declare their error types and endpoint calls.
+declare their error types and endpoint calls. Transport-level failures
+(connect/timeout, never 4xx) retry with exponential backoff via tenacity.
 """
 
 from __future__ import annotations
 
 import httpx
+from tenacity import (
+    retry,
+    retry_if_exception_type,
+    stop_after_attempt,
+    wait_exponential,
+)
 
 
 class BaseApiClient:
@@ -25,3 +32,13 @@ class BaseApiClient:
         return httpx.AsyncClient(
             base_url=self.base_url, timeout=self.timeout, headers=self._headers()
         )
+
+
+def transport_retry(func):  # type: ignore[no-untyped-def]
+    """Retry transport errors 3x with backoff; HTTP status errors pass through."""
+    return retry(
+        stop=stop_after_attempt(3),
+        wait=wait_exponential(multiplier=0.5, max=4),
+        retry=retry_if_exception_type((httpx.TransportError, httpx.TimeoutException)),
+        reraise=True,
+    )(func)

backend/app/flows/export.py

@@ -10,7 +10,6 @@ import hashlib
 import json
 import os
 import re
-import tempfile
 from pathlib import Path
 
 FLOWS_GLOB = "*.json"
@@ -20,18 +19,32 @@ FLOWS_GLOB = "*.json"
 #: validated, or shown in the frontend.
 IGNORE_FILENAME = ".flowsignore"
 
+# mtime-keyed cache: the watcher re-reads per event, the file rarely changes.
+_ignore_cache: dict[str, tuple[float, set[str]]] = {}
+
 
 def ignored_stems(flows_dir: str | Path) -> set[str]:
     """Stems excluded from the pipeline (empty when no ignore file)."""
+    key = str(Path(flows_dir))
+    try:
+        mtime = (Path(flows_dir) / IGNORE_FILENAME).stat().st_mtime
+    except OSError:
+        _ignore_cache.pop(key, None)
+        return set()
+    cached = _ignore_cache.get(key)
+    if cached is not None and cached[0] == mtime:
+        return cached[1]
     try:
         text = (Path(flows_dir) / IGNORE_FILENAME).read_text(encoding="utf-8")
     except OSError:
         return set()
-    return {
+    stems = {
         line.strip()
         for line in text.splitlines()
         if line.strip() and not line.strip().startswith("#")
     }
+    _ignore_cache[key] = (mtime, stems)
+    return stems
 
 
 def filename_stem(name: str) -> str:
@@ -97,25 +110,16 @@ def find_reuse_path(flows_dir: str | Path, doc: dict) -> tuple[Path | None, bool
 def write_flow_bytes(target: Path, doc: dict) -> Path:
     """Write a flow document to an explicit path atomically (tmp + rename).
 
-    The file ends up world-readable (0o644): mkstemp creates 0o600, which
-    would lock out the other containers (Langflow/LFX run as different
-    users and crash on unreadable seed files).
+    The file ends up world-readable (0o644): atomicwrites follows the
+    umask, which would lock out the other containers (Langflow/LFX run
+    as different users and crash on unreadable seed files).
     """
+    from atomicwrites import atomic_write
+
     target.parent.mkdir(parents=True, exist_ok=True)
-    fd, tmp_name = tempfile.mkstemp(
-        dir=str(target.parent), prefix=f".{target.stem}-", suffix=".tmp"
-    )
-    try:
-        with os.fdopen(fd, "wb") as tmp:
-            tmp.write(canonical_bytes(doc))
-        os.replace(tmp_name, target)
-        os.chmod(target, 0o644)
-    except BaseException:
-        try:
-            os.unlink(tmp_name)
-        except OSError:
-            pass
-        raise
+    with atomic_write(target, mode="wb", overwrite=True) as tmp:
+        tmp.write(canonical_bytes(doc))
+    os.chmod(target, 0o644)
     return target
 
 

backend/app/flows/langflow_client.py

@@ -8,7 +8,7 @@ from __future__ import annotations
 
 import httpx
 
-from .base_client import BaseApiClient
+from .base_client import BaseApiClient, transport_retry
 
 
 class LangflowError(Exception):
@@ -29,6 +29,7 @@ class LangflowClient(BaseApiClient):
     def __init__(self, base_url: str, api_key: str = "", timeout: float = 60.0) -> None:
         super().__init__(base_url, timeout=timeout, api_key=api_key)
 
+    @transport_retry
     async def list_flows(self) -> list[dict]:
         """List flows (header mode keeps payloads small)."""
         try:
@@ -44,6 +45,7 @@ class LangflowClient(BaseApiClient):
         except httpx.HTTPError as exc:
             raise LangflowUnavailable(f"Langflow unreachable: {exc}") from exc
 
+    @transport_retry
     async def export_flow(self, langflow_flow_id: str) -> dict:
         """Export one flow as Langflow JSON (FlowRead: id/name/description/data)."""
         try:
@@ -58,6 +60,7 @@ class LangflowClient(BaseApiClient):
         except httpx.HTTPError as exc:
             raise LangflowUnavailable(f"Langflow unreachable: {exc}") from exc
 
+    @transport_retry
     async def upsert_flow(self, doc: dict) -> dict:
         """Create or update a flow in Langflow by name (authoring UI refresh).
 

backend/app/flows/lfx_client.py

@@ -6,7 +6,7 @@ from dataclasses import dataclass
 
 import httpx
 
-from .base_client import BaseApiClient
+from .base_client import BaseApiClient, transport_retry
 
 
 class LfxError(Exception):
@@ -56,6 +56,7 @@ class LfxClient(BaseApiClient):
     def __init__(self, base_url: str, timeout: float = 120.0, api_key: str | None = None) -> None:
         super().__init__(base_url, timeout=timeout, api_key=api_key)
 
+    @transport_retry
     async def health(self) -> dict:
         try:
             async with self._client() as client:
@@ -65,6 +66,7 @@ class LfxClient(BaseApiClient):
         except httpx.HTTPError as exc:
             raise LfxUnavailable(f"LFX unreachable at {self.base_url}: {exc}") from exc
 
+    @transport_retry
     async def list_flows(self) -> list[dict]:
         try:
             async with self._client() as client:
@@ -76,6 +78,7 @@ class LfxClient(BaseApiClient):
         except httpx.HTTPError as exc:
             raise LfxUnavailable(f"LFX unreachable at {self.base_url}: {exc}") from exc
 
+    @transport_retry
     async def upload_flow(
         self, flow_json: dict, *, flow_id: str | None = None, replace: bool = False
     ) -> LfxUploadResult:
@@ -140,6 +143,7 @@ class LfxClient(BaseApiClient):
         except httpx.HTTPError as exc:
             raise LfxUnavailable(f"LFX unreachable at {self.base_url}: {exc}") from exc
 
+    @transport_retry
     async def flow_info(self, lfx_flow_id: str) -> dict:
         try:
             async with self._client() as client:
@@ -151,6 +155,7 @@ class LfxClient(BaseApiClient):
         except httpx.HTTPError as exc:
             raise LfxUnavailable(f"LFX unreachable at {self.base_url}: {exc}") from exc
 
+    @transport_retry
     async def delete_flow(self, lfx_flow_id: str) -> None:
         try:
             async with self._client() as client:

backend/app/flows/schemas.py

@@ -76,6 +76,13 @@ class RunRead(BaseModel):
     finished_at: datetime | None
 
 
+class NamedRunRead(BaseModel):
+    """A run with its flow name (GET /api/runs aggregate, kills N+1)."""
+
+    flow_name: str
+    run: RunRead
+
+
 class ApprovalCreate(BaseModel):
     """Callback contract: a flow reports a terminal approval state."""
 

backend/app/flows/store.py

@@ -129,12 +129,18 @@ def validate_flow_json(flow_json: dict) -> None:
         raise InvalidError("langflow_json.data must contain 'nodes' and 'edges' lists")
 
 
-def create_version(session: Session, flow_id: UUID, data: FlowVersionCreate) -> FlowVersion:
+def create_version(
+    session: Session,
+    flow_id: UUID,
+    data: FlowVersionCreate,
+    *,
+    version_no: int | None = None,
+) -> FlowVersion:
     get_definition(session, flow_id)
     validate_flow_json(data.langflow_json)
     version = FlowVersion(
         flow_id=flow_id,
-        version_no=next_version_no(session, flow_id),
+        version_no=version_no if version_no is not None else next_version_no(session, flow_id),
         langflow_json=data.langflow_json,
         author=data.author,
         # Single choke point: sync, import, and API creation all cache the
@@ -256,6 +262,18 @@ def list_runs(session: Session, flow_id: UUID, tenant_id: str | None = None) ->
     return list(session.exec(query).all())
 
 
+def list_all_runs(session: Session, tenant_id: str) -> list[tuple[FlowRun, str]]:
+    """All runs of a tenant with flow names (single query, kills N+1)."""
+    rows = session.exec(
+        select(FlowRun, FlowDefinition.name)
+        .join(FlowVersion, FlowVersion.id == FlowRun.version_id)
+        .join(FlowDefinition, FlowDefinition.id == FlowVersion.flow_id)
+        .where(FlowRun.tenant_id == tenant_id)
+        .order_by(FlowRun.created_at.desc())
+    ).all()
+    return [(run, name) for run, name in rows]
+
+
 def create_approval_task(session: Session, data: ApprovalCreate) -> ApprovalTask:
     run = session.get(FlowRun, data.run_id)
     if run is None:

backend/app/flows/sync.py

@@ -64,8 +64,8 @@ async def sync_document(
     "skipped" (content unchanged). Raises store.ConflictError on name
     clashes, store.InvalidError on invalid JSON, LfxError when deploy fails.
     With deploy=False a version is recorded without touching LFX.
+    Validation lives in store.create_version (single choke point).
     """
-    store.validate_flow_json(doc)
     name = doc.get("name") or Path(source_filename).stem
     with sync_lock:
         definition = store.get_definition_by_source_path(session, source_filename)
@@ -102,9 +102,8 @@ async def sync_document(
             session,
             definition.id,
             FlowVersionCreate(langflow_json=doc, author=author),
+            version_no=version_no,
         )
-        if version.version_no != version_no:  # pragma: no cover - lock prevents this
-            raise store.ConflictError(f"concurrent version write for flow {definition.id}")
         version = store.mark_deployed(session, version, uploaded.id)
         if latest is not None and latest.lfx_flow_id and latest.lfx_flow_id != uploaded.id:
             try: