feat: flow publish & versioning

src/backend/base/DEVELOPMENT-GUIDE.md

@@ -0,0 +1,54 @@
+## Flow versioning (checkpointing)
+
+Langflow uses `FlowVersion` to store snapshots of flow edits.
+
+When you mutate a flow (particularly `Flow.data`), you must use `save_flow_checkpoint`. This function handles both the checkpoint creation (if `Flow.data` changes) and the update of the Flow object in the database.
+
+### Required
+**ALWAYS use `save_flow_checkpoint` to update flow data.**
+`save_flow_checkpoint` updates the Flow row in the database and creates a new checkpoint (if needed) in the FlowVersion table in the same transaction.
+
+`save_flow_checkpoint` compares the `update_data` you pass it against the *current data* in the database. It will:
+1. Fetch the current flow from the database.
+2. Compare the new data against the stored data.
+3. If `Flow.data` (the graph) has changed, create a new `FlowVersion` entry.
+4. Update the `Flow` object with the new values from `update_data`.
+5. Return the updated `Flow` object.
+
+### Example: Updating a Flow
+
+#### ✅ DO THIS
+Pass the session (optional), user ID, flow ID, and the dictionary of updates to `save_flow_checkpoint`.
+
+```python
+# 1. Prepare your new data (e.g. from a FlowUpdate model)
+# update_data = flow_update.model_dump(exclude_unset=True, exclude_none=True)
+update_data = {"data": {...}, "name": "New Name", "description": "New Desc"}
+
+# 2. Checkpoint and Update
+# save_flow_checkpoint updates the Flow row and creates a checkpoint (if needed)
+# and returns the updated row in the Flow table
+db_flow = await save_flow_checkpoint(
+    session=session,
+    flow_id=flow_id,
+    user_id=user.id,
+    update_data=update_data
+)
+
+# 3. Flush/Refresh if needed (e.g. to get updated timestamps or IDs)
+await session.flush()
+await session.refresh(db_flow)
+```
+
+#### ❌ DO NOT DO THIS
+Do not update the database object manually before calling checkpoint, and do not expect `save_flow_checkpoint` to only handle versioning without updating the flow. (save_flow_checkpoint does both).
+
+```python
+# 1. Update the DB object first (BAD!)
+flow.data = new_flow_data
+session.add(flow)
+
+# 2. Checkpoint too late or with wrong arguments
+# The session might already see flow.data as the "current" state, missing the change.
+await save_flow_checkpoint(...)
+```

src/backend/base/langflow/agentic/utils/flow_component.py

@@ -7,7 +7,7 @@
 from lfx.log.logger import logger
 
 from langflow.helpers.flow import get_flow_by_id_or_endpoint_name
-from langflow.services.database.models.flow.model import Flow
+from langflow.helpers.flow_version import save_flow_checkpoint
 from langflow.services.deps import session_scope
 
 
@@ -265,19 +265,10 @@ async def update_component_field_value(
 
         # Update the flow in the database
         async with session_scope() as session:
-            # Get the database flow object
-            db_flow = await session.get(Flow, UUID(flow_id_str))
+            db_flow = await save_flow_checkpoint(
+                session=session, user_id=user_id, flow_id=flow_id_str, update_data={"data": flow_data}
+            )
 
-            if not db_flow:
-                return {"error": f"Flow {flow_id_str} not found in database", "success": False}
-
-            # Verify user has permission
-            if str(db_flow.user_id) != str(user_id):
-                return {"error": "User does not have permission to update this flow", "success": False}
-
-            # Update the flow data
-            db_flow.data = flow_data
-            session.add(db_flow)
             await session.commit()
             await session.refresh(db_flow)
 

src/backend/base/langflow/alembic/versions/0098a9ce61cd_add_flow_version_and_flow_publish_.py

@@ -0,0 +1,82 @@
+"""add flow_version and flow_publish tables, and latest_version column to flow table.
+Migration: Add flow_version and flow_publish tables and add latest_version column to flow table.
+Phase: EXPAND
+Revision ID: 0098a9ce61cd
+Revises: 182e5471b900
+Create Date: 2026-01-04 18:02:32.002916
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel
+from sqlalchemy.engine.reflection import Inspector
+from langflow.utils import migration
+
+
+# revision identifiers, used by Alembic.
+revision: str = '0098a9ce61cd'
+down_revision: Union[str, None] = '182e5471b900'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('flow_version',
+    sa.Column('id', sa.Uuid(), nullable=False),
+    sa.Column('user_id', sa.Uuid(), nullable=True),
+    sa.Column('flow_id', sa.Uuid(), nullable=False),
+    sa.Column('flow_data', sa.JSON(), nullable=True),
+    sa.Column('version', sa.Integer(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), nullable=False),
+    sa.ForeignKeyConstraint(['flow_id'], ['flow.id'], ondelete='CASCADE'),
+    sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
+    sa.PrimaryKeyConstraint('id'),
+    sa.UniqueConstraint('flow_id', 'version', name='unique_flow_version'),
+    sa.UniqueConstraint('id')
+    )
+    with op.batch_alter_table('flow_version', schema=None) as batch_op:
+        batch_op.create_index('flow_version_index', ['flow_id', 'version'], unique=False)
+
+    op.create_table('flow_publish',
+    sa.Column('id', sa.Uuid(), nullable=False),
+    sa.Column('user_id', sa.Uuid(), nullable=True),
+    sa.Column('flow_id', sa.Uuid(), nullable=False),
+    sa.Column('flow_version_id', sa.Uuid(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), nullable=False),
+    sa.Column('publish_state', sa.Enum('PENDING', 'FAILED', 'SUCCESS', 'REMOVED', name='publishstateenum'), nullable=True),
+    sa.Column('publish_provider', sa.Enum('S3', name='publishproviderenum'), nullable=True),
+    sa.ForeignKeyConstraint(['flow_id'], ['flow.id'], ),
+    sa.ForeignKeyConstraint(['flow_version_id'], ['flow_version.id'], ),
+    sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
+    sa.PrimaryKeyConstraint('id'),
+    sa.UniqueConstraint('flow_id', 'flow_version_id', name='at_most_one_publish_per_flow_version'),
+    sa.UniqueConstraint('id')
+    )
+    with op.batch_alter_table('flow_publish', schema=None) as batch_op:
+        batch_op.create_index('flow_id_index', ['flow_id'], unique=False)
+
+    with op.batch_alter_table('flow', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('latest_version', sa.Integer(), server_default=sa.text('0'), nullable=False))
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('flow', schema=None) as batch_op:
+        batch_op.drop_column('latest_version')
+
+    with op.batch_alter_table('flow_publish', schema=None) as batch_op:
+        batch_op.drop_index('flow_id_index')
+
+    op.drop_table('flow_publish')
+    with op.batch_alter_table('flow_version', schema=None) as batch_op:
+        batch_op.drop_index('flow_version_index')
+
+    op.drop_table('flow_version')
+    # ### end Alembic commands ###