feat: form-input library with schema parsing and endpoint

backend/app/flows/api.py

@@ -18,6 +18,7 @@ from app.settings import settings
 from app.auth.deps import get_optional_user, resolve_tenant
 
 from . import runner, store
+from . import input_schema as form_schema
 from .deploy import import_from_langflow
 from .sync import lfx_id_for, sync_from_directory
 from .langflow_client import (
@@ -46,6 +47,7 @@ from .schemas import (
     FlowImportRequest,
     FlowVersionCreate,
     FlowVersionRead,
+    InputSchemaRead,
     RunRead,
     RunRequest,
     SyncReport,
@@ -91,6 +93,7 @@ def _read_version(v) -> FlowVersionRead:
         deployed_at=v.deployed_at,
         author=v.author,
         created_at=v.created_at,
+        input_schema=v.input_schema,
     )
 
 
@@ -386,6 +389,28 @@ async def import_flow(
     )
 
 
+@flows_router.get("/{flow_id}/input-schema", response_model=InputSchemaRead)
+def get_input_schema(
+    flow_id: UUID,
+    session: Session = Depends(get_db),
+):
+    """Form schema of the latest deployed version (404 when none)."""
+    try:
+        version = store.get_latest_deployed_version(session, flow_id)
+    except (store.NotFoundError, store.InvalidError) as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+    cached = version.input_schema or form_schema.extract_form_schema(
+        version.langflow_json
+    )
+    return InputSchemaRead(
+        flow_id=flow_id,
+        version_no=version.version_no,
+        json_schema=cached.get("json_schema", {}),
+        ui_schema=cached.get("ui_schema", {}),
+        fallback=cached.get("fallback"),
+    )
+
+
 @flows_router.get("/runs/{run_id}", response_model=RunRead)
 def get_run(
     run_id: UUID,

backend/app/flows/input_schema.py

@@ -0,0 +1,147 @@
+"""Derive a web-form schema from a Langflow export.
+
+Looks for ``CustomInput*`` nodes (``data.type`` or the inner
+``display_name`` starting with ``CustomInput`` — see
+``backend/components/kirche/form_inputs.py``) and maps their ``template``
+(minus ``code``) to ``{json_schema, ui_schema}`` for RJSF:
+
+* ``name`` → property key, ``display_name`` → ``title``,
+  ``info`` → ``description``, ``value`` → ``default``,
+  ``options`` → ``enum``, ``required`` → required array
+* ``show: false`` fields are plumbing (e.g. the run envelope input), not
+  form fields, and are skipped
+* ``advanced`` fields render last via ``ui:order``
+* ``_input_type`` → JSON type + widget (unknown types fall back to string
+  so the parser never crashes on new fields)
+
+A flow without such a node gets ``{fallback: "textarea"}``: the frontend
+renders one free-text box instead of a generated form.
+"""
+
+from __future__ import annotations
+
+
+def _form_templates(langflow_json: dict) -> list[dict]:
+    """Inner node dicts (``data.node``) of all CustomInput* nodes."""
+    if not isinstance(langflow_json, dict):
+        return []
+    data = langflow_json.get("data") or {}
+    nodes = data.get("nodes") or []
+    found = []
+    for node in nodes:
+        if not isinstance(node, dict):
+            continue
+        node_data = node.get("data") or {}
+        node_type = str(node_data.get("type") or "")
+        inner = node_data.get("node") or {}
+        display = str(inner.get("display_name") or "")
+        if node_type.startswith("CustomInput") or display.startswith("CustomInput"):
+            found.append(inner)
+    return found
+
+
+def _field_schema(name: str, field: dict) -> tuple[dict, dict]:
+    """(JSON-Schema property, uiSchema entry) for one template field."""
+    input_type = str(field.get("_input_type") or "")
+    prop: dict = {}
+    ui: dict = {}
+    title = field.get("display_name") or name
+    info = field.get("info") or ""
+    if title:
+        prop["title"] = title
+    if info:
+        prop["description"] = info
+
+    options = field.get("options") or []
+    if input_type in ("StrInput", "MessageTextInput"):
+        prop["type"] = "string"
+    elif input_type == "MultilineInput":
+        prop["type"] = "string"
+        ui["ui:widget"] = "textarea"
+    elif input_type == "SecretStrInput":
+        prop["type"] = "string"
+        ui["ui:widget"] = "password"
+    elif input_type == "IntInput":
+        prop["type"] = "integer"
+    elif input_type == "FloatInput":
+        prop["type"] = "number"
+    elif input_type == "BoolInput":
+        prop["type"] = "boolean"
+    elif input_type == "DropdownInput":
+        prop["type"] = "string"
+        if options:
+            prop["enum"] = list(options)
+    elif input_type == "MultiselectInput":
+        prop["type"] = "array"
+        items: dict = {"type": "string"}
+        if options:
+            items["enum"] = list(options)
+        prop["items"] = items
+        prop["uniqueItems"] = True
+    elif input_type == "SliderInput":
+        prop["type"] = "number"
+        spec = field.get("range_spec") or {}
+        if spec.get("min") is not None:
+            prop["minimum"] = spec["min"]
+        if spec.get("max") is not None:
+            prop["maximum"] = spec["max"]
+        if spec.get("step"):
+            prop["multipleOf"] = spec["step"]
+        ui["ui:widget"] = "range"
+    elif input_type == "FileInput":
+        # Stub for now: string reference only, real upload comes later.
+        prop["type"] = "string"
+    else:
+        prop["type"] = "string"
+
+    value = field.get("value")
+    if value is not None:
+        # A default outside its enum would fail validation on touch.
+        enum = prop.get("enum")
+        if not (enum is not None and value not in enum):
+            # bool is an int subclass; keep integer defaults clean.
+            if prop.get("type") == "integer" and isinstance(value, bool):
+                pass
+            else:
+                prop["default"] = value
+    return prop, ui
+
+
+def extract_form_schema(langflow_json: dict) -> dict:
+    """``{json_schema, ui_schema}``, or with ``fallback: "textarea"``."""
+    templates = _form_templates(langflow_json)
+    if not templates:
+        return {
+            "json_schema": {"type": "object", "properties": {}},
+            "ui_schema": {},
+            "fallback": "textarea",
+        }
+    template = templates[0].get("template") or {}
+    properties: dict = {}
+    required: list[str] = []
+    ui_schema: dict = {}
+    normal: list[str] = []
+    advanced: list[str] = []
+    for key, field in template.items():
+        if key == "code" or not isinstance(field, dict):
+            continue
+        if field.get("show") is False:
+            continue
+        name = field.get("name") or key
+        prop, ui = _field_schema(name, field)
+        properties[name] = prop
+        if field.get("required"):
+            required.append(name)
+        if ui:
+            ui_schema[name] = ui
+        (advanced if field.get("advanced") else normal).append(name)
+    if advanced:
+        ui_schema["ui:order"] = normal + advanced + ["*"]
+    return {
+        "json_schema": {
+            "type": "object",
+            "properties": properties,
+            "required": required,
+        },
+        "ui_schema": ui_schema,
+    }

backend/app/flows/schemas.py

@@ -37,6 +37,18 @@ class FlowVersionRead(BaseModel):
     deployed_at: datetime | None
     author: str
     created_at: datetime
+    input_schema: dict | None = None
+
+
+class InputSchemaRead(BaseModel):
+    """Form schema of the latest deployed version (RJSF renders it)."""
+
+    flow_id: UUID
+    version_no: int
+    json_schema: dict
+    ui_schema: dict = Field(default_factory=dict)
+    # Set when the flow has no CustomInput node: render a free-text box.
+    fallback: str | None = None
 
 
 class RunRequest(BaseModel):

backend/app/flows/store.py

@@ -10,6 +10,7 @@ from sqlmodel import Session, func, select
 
 from app.models import ApprovalTask, FlowDefinition, FlowRun, FlowVersion
 
+from . import input_schema as form_schema
 from .schemas import ApprovalCreate, FlowDefinitionCreate, FlowVersionCreate
 
 
@@ -154,6 +155,9 @@ def create_version(
         version_no=(max_no or 0) + 1,
         langflow_json=data.langflow_json,
         author=data.author,
+        # Single choke point: sync, import, and API creation all cache the
+        # form schema here, so reads never re-parse.
+        input_schema=form_schema.extract_form_schema(data.langflow_json),
     )
     session.add(version)
     session.commit()

backend/app/models/flow.py

@@ -54,6 +54,9 @@ class FlowVersion(SQLModel, table=True):
     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())
+    # Cached form schema for this exact JSON (see app/flows/input_schema.py).
+    # Computed at version creation; nullable for pre-Phase-3 rows.
+    input_schema: dict | None = Field(default=None, sa_column=_json_column())
 
 
 class FlowRun(SQLModel, table=True):

backend/components/kirche/__init__.py

@@ -0,0 +1,12 @@
+"""Kirche form-input components (Langflow category ``kirche``).
+
+GUI-editable by design: plain :class:`lfx.custom.Component` subclasses with
+``lfx.io`` inputs only (stdlib + ``lfx``), so both the Langflow authoring UI
+and the LFX runtime can load them. The backend parses the exported node
+template into a web form (``app/flows/input_schema.py``); the frontend
+renders that schema with RJSF.
+"""
+
+from .form_inputs import CustomInputForm
+
+__all__ = ["CustomInputForm"]

backend/components/kirche/form_inputs.py

@@ -0,0 +1,161 @@
+"""Structured form-input component for kirche-einfach flows.
+
+A flow that starts with :class:`CustomInputForm` declares the fields of its
+web form. The API reads the exported node template
+(``template`` minus ``code``) and derives ``{json_schema, ui_schema}`` from
+the ``lfx.io`` input definitions below — one JSON-Schema property per field,
+keyed by ``name``. Field metadata maps as follows:
+
+* ``display_name`` → ``title``, ``info`` → ``description``,
+  ``value`` → ``default``, ``options`` → ``enum``
+* ``required`` → required array, ``advanced`` → ``ui:collapsed`` hint
+* ``_input_type`` → JSON type + widget (see ``app/flows/input_schema.py``)
+
+``FileInput`` is a stub for now: it carries a string reference only, real
+upload comes later.
+"""
+
+from __future__ import annotations
+
+import json
+
+from lfx.custom import Component
+from lfx.field_typing.range_spec import RangeSpec
+from lfx.io import (
+    BoolInput,
+    DropdownInput,
+    FileInput,
+    FloatInput,
+    IntInput,
+    MessageTextInput,
+    MultilineInput,
+    MultiselectInput,
+    Output,
+    SecretStrInput,
+    SliderInput,
+    StrInput,
+)
+from lfx.schema.message import Message
+
+
+def _text(value) -> str:
+    if value is None:
+        return ""
+    if hasattr(value, "text"):
+        return value.text or ""
+    return str(value)
+
+
+class CustomInputForm(Component):
+    display_name = "CustomInput Form"
+    description = "Collects structured form values into a single JSON message."
+    icon = "ClipboardList"
+
+    inputs = [
+        # Plumbing, not a form field: the API posts submitted web-form
+        # values as the run envelope (same {"payload": {...}} shape the
+        # other kirche components use); they arrive here through a ChatInput
+        # entry node and override the defaults below. Hidden in the GUI and
+        # skipped by the form parser (show=False).
+        MessageTextInput(
+            name="form_values",
+            display_name="Form values (JSON envelope)",
+            info="Submitted values as JSON; merged over the defaults",
+            value="",
+            show=False,
+        ),
+        StrInput(
+            name="full_name",
+            display_name="Full name",
+            info="Your first and last name",
+            value="",
+            required=True,
+        ),
+        MultilineInput(
+            name="notes",
+            display_name="Notes",
+            info="Anything else we should know",
+            value="",
+        ),
+        SecretStrInput(
+            name="password",
+            display_name="Password",
+            info="Optional access phrase",
+            value="",
+        ),
+        IntInput(
+            name="age",
+            display_name="Age",
+            info="Your age in years",
+            value=18,
+        ),
+        FloatInput(
+            name="amount",
+            display_name="Amount",
+            info="Amount in euros",
+            value=0.0,
+        ),
+        BoolInput(
+            name="newsletter",
+            display_name="Newsletter",
+            info="Send me the parish newsletter",
+            value=False,
+        ),
+        DropdownInput(
+            name="department",
+            display_name="Department",
+            info="Which team is this for",
+            options=["Music", "Youth", "Seniors", "Admin"],
+            value="Music",
+        ),
+        MultiselectInput(
+            name="topics",
+            display_name="Topics",
+            info="What are you interested in",
+            options=["Worship", "Outreach", "Courses"],
+            value=[],
+        ),
+        SliderInput(
+            name="satisfaction",
+            display_name="Satisfaction",
+            info="How satisfied are you (1-5)",
+            value=3,
+            range_spec=RangeSpec(min=1, max=5, step=1),
+        ),
+        FileInput(
+            name="attachment",
+            display_name="Attachment",
+            info="String reference only for now (no upload yet)",
+            file_types=["pdf", "png", "jpg"],
+        ),
+    ]
+    outputs = [
+        Output(display_name="Form Data", name="form_data", method="collect"),
+    ]
+
+    def collect(self) -> Message:
+        data = {
+            "full_name": _text(self.full_name),
+            "notes": _text(self.notes),
+            "password": _text(self.password),
+            "age": self.age,
+            "amount": self.amount,
+            "newsletter": self.newsletter,
+            "department": _text(self.department),
+            "topics": self.topics,
+            "satisfaction": self.satisfaction,
+            "attachment": _text(self.attachment),
+        }
+        # Submitted values (via form_values) win over the field defaults.
+        # Accepts the raw dict or the run envelope {"payload": {...}}.
+        try:
+            override = json.loads(_text(self.form_values) or "{}")
+        except Exception:
+            override = {}
+        if isinstance(override, dict):
+            if isinstance(override.get("payload"), dict):
+                override = override["payload"]
+            for key in data:
+                if key in override:
+                    data[key] = override[key]
+        return Message(text=json.dumps(data, ensure_ascii=False))

backend/migrations/versions/b7e2d4a1c9f3_cache_form_schema.py

@@ -0,0 +1,32 @@
+"""cache form schema per flow version
+
+Revision ID: b7e2d4a1c9f3
+Revises: a3f1c9e2b4d7
+Create Date: 2026-09-11 20:30:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel  # noqa: F401 - autogenerate renders sqlmodel.* column types
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'b7e2d4a1c9f3'
+down_revision: Union[str, Sequence[str], None] = 'a3f1c9e2b4d7'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    op.add_column(
+        "flow_versions",
+        sa.Column("input_schema", sa.JSON(), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    op.drop_column("flow_versions", "input_schema")

backend/scripts/make_flow_fixtures.py

@@ -68,8 +68,7 @@ def builtin_node(
     return node_dict
 
 
-def custom_node(
+def custom_node(    code: str,
-    code: str,
     node_id: str,
     values: dict | None = None,
     position: tuple[int, int] = (0, 0),
@@ -87,6 +86,66 @@ def custom_node(
     return node_dict
 
 
+def kirche_node(
+    node_id: str,
+    values: dict | None = None,
+    position: tuple[int, int] = (0, 0),
+) -> dict:
+    """A node from our GUI-editable category lib (backend/components/kirche).
+
+    Built with lfx's own frontend-node serialization, exactly like
+    ``Component.to_frontend_node()`` — minus its final ``Template.from_dict``
+    round-trip: lfx 1.12.x cannot re-instantiate ``MultiselectInput`` (it
+    emits ``list``, the constructor forbids extras; their own builtins fail
+    the same way). The template dict before that round-trip is complete and
+    faithful, so we use it directly and append the ``code`` field by hand
+    (mirroring the shape ``build_custom_component_template`` produces).
+    """
+    sys.path.insert(0, str(REPO_ROOT / "backend" / "components"))
+    from kirche.form_inputs import CustomInputForm
+    from lfx.template.frontend_node.custom_components import (
+        ComponentFrontendNode,
+    )
+
+    comp_name = getattr(CustomInputForm, "name", None) or "CustomInputForm"
+    inst = CustomInputForm()
+    frontend_node = ComponentFrontendNode.from_inputs(
+        **inst.get_template_config(inst)
+    )
+    inst._map_parameters_on_frontend_node(frontend_node)
+    node = frontend_node.to_dict(keep_name=False)
+    node = inst._update_template(node)
+    inst._map_parameters_on_template(node["template"])
+    if not getattr(inst, "_code", None):
+        inst.set_class_code()
+    node["template"]["code"] = {
+        "advanced": True,
+        "api_editable": False,
+        "dynamic": True,
+        "fileTypes": [],
+        "info": "",
+        "list": False,
+        "load_from_db": False,
+        "multiline": True,
+        "name": "code",
+        "password": False,
+        "placeholder": "",
+        "required": True,
+        "show": True,
+        "title_case": False,
+        "type": "code",
+        "value": inst._code,
+    }
+    node_dict = {
+        "data": {"node": node, "type": comp_name, "id": node_id},
+        "id": node_id,
+    }
+    node_dict = _normalize(node_dict, node_id, comp_name, *position)
+    for field, value in (values or {}).items():
+        node_dict["data"]["node"]["template"][field]["value"] = value
+    return node_dict
+
+
 # --------------------------------------------------------------------------
 # edges
 # --------------------------------------------------------------------------
@@ -433,12 +492,43 @@ def build_approval_handoff() -> dict:
     )
 
 
+def build_intake_form() -> dict:
+    entry = builtin_node(
+        "lfx.components.input_output.chat",
+        "ChatInput",
+        "ChatInput-entry",
+        position=(0, 200),
+    )
+    form = kirche_node(
+        "CustomInputForm-intake",
+        position=(400, 200),
+    )
+    out = builtin_node(
+        "lfx.components.input_output.chat_output",
+        "ChatOutput",
+        "ChatOutput-collected",
+        position=(800, 200),
+    )
+    nodes = [entry, form, out]
+    edges = [
+        edge(entry, "message", form, "form_values"),
+        edge(form, "form_data", out, "input_value"),
+    ]
+    return flow_doc(
+        "intake_form",
+        "Structured intake via CustomInputForm (parsed into a web form).",
+        nodes,
+        edges,
+    )
+
+
 def main() -> None:
     FIXTURE_DIR.mkdir(parents=True, exist_ok=True)
     FLOW_DIR.mkdir(parents=True, exist_ok=True)
     flows = {
         "simple_rule.json": build_simple_rule(),
         "approval_handoff.json": build_approval_handoff(),
+        "intake_form.json": build_intake_form(),
     }
     for filename, doc in flows.items():
         payload = json.dumps(doc, indent=2, ensure_ascii=False) + "\n"

backend/tests/test_form_components.py

@@ -0,0 +1,104 @@
+"""Phase 2 tests: kirche CustomInput* library + intake fixture flow."""
+
+import json
+import sys
+from pathlib import Path
+
+from lfx.custom import Component
+
+REPO_ROOT = Path(__file__).resolve().parent.parent.parent
+sys.path.insert(0, str(REPO_ROOT / "backend" / "components"))
+
+from kirche.form_inputs import CustomInputForm  # noqa: E402
+
+FIXTURES = Path(__file__).parent / "fixtures" / "flows"
+
+FORM_FIELDS = [
+    "full_name",
+    "notes",
+    "password",
+    "age",
+    "amount",
+    "newsletter",
+    "department",
+    "topics",
+    "satisfaction",
+    "attachment",
+]
+
+
+def test_component_contract():
+    assert issubclass(CustomInputForm, Component)
+    assert CustomInputForm.display_name.startswith("CustomInput")
+    inputs = {i.name: i for i in CustomInputForm.inputs}
+    # Ten form fields plus the hidden run-envelope plumbing input.
+    assert sorted(n for n in inputs if n != "form_values") == sorted(FORM_FIELDS)
+    assert inputs["form_values"].show is False
+    outputs = {o.name: o for o in CustomInputForm.outputs}
+    assert set(outputs) == {"form_data"}
+
+
+def test_collect_merges_submitted_values_over_defaults():
+    from lfx.schema.message import Message
+
+    form = CustomInputForm()
+    default = json.loads(form.collect().text)
+    assert default["full_name"] == ""
+    assert default["department"] == "Music"
+    assert isinstance(form.collect(), Message)
+
+    form.form_values = json.dumps({"full_name": "Maria", "age": 41})
+    merged = json.loads(form.collect().text)
+    assert merged["full_name"] == "Maria"
+    assert merged["age"] == 41
+    assert merged["department"] == "Music"  # default kept
+
+    # Run envelope shape {"payload": {...}} unwraps like the other components.
+    form.form_values = json.dumps({"payload": {"notes": "hi"}})
+    assert json.loads(form.collect().text)["notes"] == "hi"
+
+    form.form_values = "not json"
+    assert json.loads(form.collect().text)["full_name"] == ""
+
+
+def test_intake_fixture_shape():
+    doc = json.loads((FIXTURES / "intake_form.json").read_text())
+    nodes = {n["id"]: n for n in doc["data"]["nodes"]}
+    form = nodes["CustomInputForm-intake"]
+    assert form["data"]["type"] == "CustomInputForm"
+    assert form["data"]["node"]["display_name"].startswith("CustomInput")
+    template = form["data"]["node"]["template"]
+    for field in FORM_FIELDS:
+        assert field in template, field
+    assert template["form_values"]["show"] is False
+    assert len(template["code"]["value"]) > 200
+    edges = {(e["source"], e["target"]) for e in doc["data"]["edges"]}
+    assert ("ChatInput-entry", "CustomInputForm-intake") in edges
+    assert ("CustomInputForm-intake", "ChatOutput-collected") in edges
+
+
+async def test_intake_deploy_and_run_with_values(client, lfx_server):
+    doc = json.loads((FIXTURES / "intake_form.json").read_text())
+    created = await client.post("/api/flows", json={"name": "intake-e2e"})
+    assert created.status_code == 201, created.text
+    flow_id = created.json()["id"]
+
+    version = await client.post(
+        f"/api/flows/{flow_id}/versions",
+        json={"langflow_json": doc, "author": "test"},
+    )
+    assert version.status_code == 201, version.text
+
+    deployed = await client.post(f"/api/flows/{flow_id}/versions/1/deploy")
+    assert deployed.status_code == 200, deployed.text
+
+    run = await client.post(
+        f"/api/flows/{flow_id}/run",
+        json={"input": {"full_name": "Maria Muster", "age": 41}},
+    )
+    assert run.status_code == 200, run.text
+    assert run.json()["status"] == "completed"
+    result = json.loads(run.json()["output"]["result"])
+    assert result["full_name"] == "Maria Muster"
+    assert result["age"] == 41
+    assert result["department"] == "Music"  # default preserved

backend/tests/test_input_schema.py

@@ -0,0 +1,201 @@
+"""Phase 3 tests: form-schema extraction, sync cache, input-schema endpoint."""
+
+import json
+from pathlib import Path
+from uuid import uuid4
+
+from app.flows import input_schema as parser
+from app.flows import sync
+from app.flows.lfx_client import LfxClient
+from app.settings import settings
+
+FIXTURES = Path(__file__).parent / "fixtures" / "flows"
+
+
+def _field(name, itype, **kw):
+    base = {
+        "name": name,
+        "display_name": name.replace("_", " ").title(),
+        "info": "",
+        "value": None,
+        "required": False,
+        "advanced": False,
+        "show": True,
+        "_input_type": itype,
+    }
+    base.update(kw)
+    return base
+
+
+def _doc(*fields, node_type="CustomInputForm", display="CustomInput Form"):
+    template = {f["name"]: f for f in fields}
+    node = {
+        "data": {
+            "type": node_type,
+            "node": {"display_name": display, "template": template},
+        }
+    }
+    return {"data": {"nodes": [node], "edges": []}}
+
+
+def test_type_mapping_table():
+    doc = _doc(
+        _field("a", "StrInput", value="x"),
+        _field("b", "MessageTextInput"),
+        _field("c", "MultilineInput"),
+        _field("d", "SecretStrInput"),
+        _field("e", "IntInput", value=3),
+        _field("f", "FloatInput", value=1.5),
+        _field("g", "BoolInput", value=True),
+        _field(
+            "h",
+            "DropdownInput",
+            options=["Music", "Youth"],
+            value="Youth",
+        ),
+        _field(
+            "i",
+            "MultiselectInput",
+            options=["Worship", "Courses"],
+            value=["Worship"],
+        ),
+        _field(
+            "j",
+            "SliderInput",
+            value=3,
+            range_spec={"min": 1, "max": 5, "step": 1},
+        ),
+        _field("k", "FileInput"),
+        _field("l", "FutureFancyInput"),
+    )
+    schema = parser.extract_form_schema(doc)
+    assert "fallback" not in schema
+    props = schema["json_schema"]["properties"]
+    assert props["a"] == {"title": "A", "type": "string", "default": "x"}
+    assert props["b"]["type"] == "string"
+    assert props["e"] == {"title": "E", "type": "integer", "default": 3}
+    assert props["f"]["type"] == "number"
+    assert props["g"] == {"title": "G", "type": "boolean", "default": True}
+    assert props["h"] == {
+        "title": "H",
+        "type": "string",
+        "enum": ["Music", "Youth"],
+        "default": "Youth",
+    }
+    assert props["i"]["type"] == "array"
+    assert props["i"]["items"] == {"type": "string", "enum": ["Worship", "Courses"]}
+    assert props["i"]["uniqueItems"] is True
+    assert props["j"]["minimum"] == 1
+    assert props["j"]["maximum"] == 5
+    assert props["j"]["multipleOf"] == 1
+    assert props["k"]["type"] == "string"
+    assert props["l"]["type"] == "string"
+    ui = schema["ui_schema"]
+    assert ui["c"] == {"ui:widget": "textarea"}
+    assert ui["d"] == {"ui:widget": "password"}
+    assert ui["j"] == {"ui:widget": "range"}
+
+
+def test_dropdown_default_outside_enum_omitted():
+    doc = _doc(_field("h", "DropdownInput", options=["Music"], value=""))
+    props = parser.extract_form_schema(doc)["json_schema"]["properties"]
+    assert props["h"] == {"title": "H", "type": "string", "enum": ["Music"]}
+
+
+def test_required_advanced_hidden_code():
+    doc = _doc(
+        _field("a", "StrInput", required=True),
+        _field("b", "StrInput", advanced=True),
+        _field("c", "StrInput", show=False),
+    )
+    doc["data"]["nodes"][0]["data"]["node"]["template"]["code"] = {"value": "x"}
+    schema = parser.extract_form_schema(doc)
+    assert schema["json_schema"]["required"] == ["a"]
+    assert set(schema["json_schema"]["properties"]) == {"a", "b"}
+    assert schema["ui_schema"]["ui:order"] == ["a", "b", "*"]
+
+
+def test_display_name_match_and_fallback():
+    doc = _doc(_field("a", "StrInput"), node_type="Custom", display="CustomInput X")
+    assert set(
+        parser.extract_form_schema(doc)["json_schema"]["properties"]
+    ) == {"a"}
+    plain = {
+        "data": {
+            "nodes": [{"data": {"type": "ChatInput", "node": {"template": {}}}}],
+            "edges": [],
+        }
+    }
+    schema = parser.extract_form_schema(plain)
+    assert schema["fallback"] == "textarea"
+    assert schema["json_schema"]["properties"] == {}
+
+
+def test_real_intake_fixture_parses():
+    doc = json.loads((FIXTURES / "intake_form.json").read_text())
+    schema = parser.extract_form_schema(doc)
+    props = schema["json_schema"]["properties"]
+    assert "fallback" not in schema
+    assert set(props) == {
+        "full_name",
+        "notes",
+        "password",
+        "age",
+        "amount",
+        "newsletter",
+        "department",
+        "topics",
+        "satisfaction",
+        "attachment",
+    }
+    assert "full_name" in schema["json_schema"]["required"]
+    assert props["department"]["enum"] == ["Music", "Youth", "Seniors", "Admin"]
+    assert schema["ui_schema"]["notes"] == {"ui:widget": "textarea"}
+
+
+def _lfx() -> LfxClient:
+    return LfxClient(settings.lfx_url, api_key=settings.lfx_api_key or None)
+
+
+async def test_sync_caches_schema_per_version(session, lfx_server, tmp_path):
+    (tmp_path / "intake_form.json").write_text(
+        (FIXTURES / "intake_form.json").read_text()
+    )
+    report = await sync.sync_from_directory(session, _lfx(), tmp_path)
+    assert not report.errors, report.errors
+    from app.flows import store
+
+    defs = store.list_definitions(session)
+    definition = next(d for d in defs if d.name == "intake_form")
+    versions = store.list_versions(session, definition.id)
+    assert len(versions) == 1
+    cached = versions[0].input_schema
+    assert cached["json_schema"]["properties"]["full_name"]["type"] == "string"
+    assert "form_values" not in cached["json_schema"]["properties"]
+
+
+async def test_input_schema_endpoint(client, lfx_server):
+    assert (await client.get(f"/api/flows/{uuid4()}/input-schema")).status_code == 404
+
+    doc = json.loads((FIXTURES / "intake_form.json").read_text())
+    created = await client.post("/api/flows", json={"name": "schema-e2e"})
+    flow_id = created.json()["id"]
+    version = await client.post(
+        f"/api/flows/{flow_id}/versions", json={"langflow_json": doc}
+    )
+    assert version.status_code == 201, version.text
+    assert version.json()["input_schema"]["json_schema"]["properties"]["age"][
+        "type"
+    ] == "integer"
+    # Created but never deployed → 404.
+    assert (await client.get(f"/api/flows/{flow_id}/input-schema")).status_code == 404
+
+    deployed = await client.post(f"/api/flows/{flow_id}/versions/1/deploy")
+    assert deployed.status_code == 200, deployed.text
+    response = await client.get(f"/api/flows/{flow_id}/input-schema")
+    assert response.status_code == 200, response.text
+    body = response.json()
+    assert body["version_no"] == 1
+    assert body["fallback"] is None
+    assert set(body["json_schema"]["properties"]) >= {"full_name", "topics"}
+    assert body["ui_schema"]["password"] == {"ui:widget": "password"}