feat(flow): add collaborative editing database foundation

Add flow.latest_operation_revision and the flow_operation table for
durable, revision-ordered operation batches. Persist forward_ops and
backward_ops with actor_user_id and actor_delegate (self/agent), plus
CRUD helpers, Alembic migration, and unit tests.

Author: HzaRashid

src/backend/base/langflow/alembic/versions/e8f1a2b3c4d5_add_flow_revision_and_flow_operation.py

@@ -0,0 +1,74 @@
+"""Add flow.latest_operation_revision and flow_operation table for collaborative editing.
+
+Phase: EXPAND
+Revision ID: e8f1a2b3c4d5
+Revises: 7c8d9e0f1a2b
+Create Date: 2026-05-28 00:00:00.000000
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from langflow.utils import migration
+
+revision: str = "e8f1a2b3c4d5"  # pragma: allowlist secret
+down_revision: str | None = "7c8d9e0f1a2b"  # pragma: allowlist secret
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+
+    with op.batch_alter_table("flow", schema=None) as batch_op:
+        if not migration.column_exists(table_name="flow", column_name="latest_operation_revision", conn=conn):
+            batch_op.add_column(
+                sa.Column("latest_operation_revision", sa.BigInteger(), server_default=sa.text("0"), nullable=False),
+            )
+            batch_op.create_index(
+                op.f("ix_flow_latest_operation_revision"),
+                ["latest_operation_revision"],
+                unique=False,
+            )
+
+    if not migration.table_exists("flow_operation", conn):
+        op.create_table(
+            "flow_operation",
+            sa.Column("id", sa.Uuid(), nullable=False),
+            sa.Column("flow_id", sa.Uuid(), nullable=False),
+            sa.Column("protocol_version", sa.Integer(), nullable=False),
+            sa.Column("revision", sa.BigInteger(), nullable=False),
+            sa.Column("client_id", sa.String(), nullable=False),
+            sa.Column("actor_user_id", sa.Uuid(), nullable=True),
+            sa.Column("actor_delegate", sa.String(), server_default="self", nullable=False),
+            sa.Column("forward_ops", sa.JSON(), nullable=False),
+            sa.Column("backward_ops", sa.JSON(), nullable=False),
+            sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.ForeignKeyConstraint(["actor_user_id"], ["user.id"], ondelete="SET NULL"),
+            sa.ForeignKeyConstraint(["flow_id"], ["flow.id"], ondelete="CASCADE"),
+            sa.PrimaryKeyConstraint("id"),
+            sa.UniqueConstraint("flow_id", "revision", name="unique_flow_operation_revision"),
+        )
+        op.create_index(op.f("ix_flow_operation_flow_id"), "flow_operation", ["flow_id"])
+        op.create_index(op.f("ix_flow_operation_actor_user_id"), "flow_operation", ["actor_user_id"])
+        op.create_index(
+            "ix_flow_operation_flow_id_created_at",
+            "flow_operation",
+            ["flow_id", "created_at"],
+        )
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+
+    if migration.table_exists("flow_operation", conn):
+        op.drop_index("ix_flow_operation_flow_id_created_at", table_name="flow_operation")
+        op.drop_index(op.f("ix_flow_operation_actor_user_id"), table_name="flow_operation")
+        op.drop_index(op.f("ix_flow_operation_flow_id"), table_name="flow_operation")
+        op.drop_table("flow_operation")
+
+    with op.batch_alter_table("flow", schema=None) as batch_op:
+        if migration.column_exists(table_name="flow", column_name="latest_operation_revision", conn=conn):
+            batch_op.drop_index(op.f("ix_flow_latest_operation_revision"))
+            batch_op.drop_column("latest_operation_revision")

src/backend/base/langflow/services/database/models/__init__.py

@@ -15,6 +15,7 @@
 from .deployment_provider_account import DeploymentProviderAccount
 from .file import File
 from .flow import Flow
+from .flow_operation import FlowOperation
 from .flow_version import FlowVersion
 from .flow_version_deployment_attachment import FlowVersionDeploymentAttachment
 from .folder import Folder
@@ -42,6 +43,7 @@
     "DeploymentProviderAccount",
     "File",
     "Flow",
+    "FlowOperation",
     "FlowVersion",
     "FlowVersionDeploymentAttachment",
     "Folder",

src/backend/base/langflow/services/database/models/flow/model.py

@@ -10,8 +10,8 @@
 from emoji import purely_emoji
 from lfx.log.logger import logger
 from pydantic import BaseModel, ValidationInfo, field_serializer, field_validator
+from sqlalchemy import BigInteger, Text, UniqueConstraint, text
 from sqlalchemy import Enum as SQLEnum
-from sqlalchemy import Text, UniqueConstraint, text
 from sqlmodel import JSON, Column, Field, Relationship, SQLModel
 
 from langflow.schema.data import Data
@@ -190,6 +190,10 @@ def validate_dt(cls, v):
 
 class Flow(FlowBase, table=True):  # type: ignore[call-arg]
     id: UUID = Field(default_factory=uuid4, primary_key=True, unique=True)
+    latest_operation_revision: int = Field(
+        default=0,
+        sa_column=Column(BigInteger, nullable=False, server_default=text("0"), index=True),
+    )
     data: dict | None = Field(default=None, sa_column=Column(JSON))
     user_id: UUID | None = Field(index=True, foreign_key="user.id", nullable=True)
     user: "User" = Relationship(back_populates="flows")

src/backend/base/langflow/services/database/models/flow_operation/__init__.py

@@ -0,0 +1,15 @@
+from .crud import (
+    create_flow_operation,
+    get_flow_operation_by_revision,
+    list_flow_operations_after_revision,
+)
+from .model import FlowOperation, FlowOperationActorDelegate, FlowOperationRead
+
+__all__ = [
+    "FlowOperation",
+    "FlowOperationActorDelegate",
+    "FlowOperationRead",
+    "create_flow_operation",
+    "get_flow_operation_by_revision",
+    "list_flow_operations_after_revision",
+]

src/backend/base/langflow/services/database/models/flow_operation/crud.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from sqlmodel import col, select
+
+from langflow.services.database.models.flow_operation.model import FlowOperation, FlowOperationActorDelegate
+
+if TYPE_CHECKING:
+    from uuid import UUID
+
+    from sqlmodel.ext.asyncio.session import AsyncSession
+
+
+async def create_flow_operation(
+    session: AsyncSession,
+    *,
+    flow_id: UUID,
+    protocol_version: int,
+    revision: int,
+    client_id: str,
+    actor_user_id: UUID,
+    actor_delegate: FlowOperationActorDelegate,
+    forward_ops: list[dict[str, Any]],
+    backward_ops: list[dict[str, Any]],
+) -> FlowOperation:
+    """Persist an accepted operation batch row.
+
+    ``actor_user_id`` must be the authenticated user that initiated the operation,
+    including agent operations started by a user. ``actor_delegate`` distinguishes
+    direct edits (``self``) from agent-mediated edits (``agent``). ``actor_user_id``
+    is nullable on the model only so existing operation rows survive user deletion
+    via ``ON DELETE SET NULL``.
+
+    This helper does not authenticate, authorize, or verify that ``actor_user_id``
+    may write to ``flow_id``. Callers must derive it from authentication, then
+    load the flow and run the appropriate authorization guard before creating an
+    operation row.
+    """
+    entry = FlowOperation(
+        flow_id=flow_id,
+        protocol_version=protocol_version,
+        revision=revision,
+        client_id=client_id,
+        actor_user_id=actor_user_id,
+        actor_delegate=actor_delegate,
+        forward_ops=forward_ops,
+        backward_ops=backward_ops,
+    )
+    session.add(entry)
+    await session.flush()
+    await session.refresh(entry)
+    return entry
+
+
+async def get_flow_operation_by_revision(
+    session: AsyncSession,
+    flow_id: UUID,
+    revision: int,
+) -> FlowOperation | None:
+    """Return an operation row by flow/revision.
+
+    Callers are responsible for flow visibility checks before exposing the row.
+    """
+    result = await session.exec(
+        select(FlowOperation).where(FlowOperation.flow_id == flow_id, FlowOperation.revision == revision)
+    )
+    return result.first()
+
+
+async def list_flow_operations_after_revision(
+    session: AsyncSession,
+    flow_id: UUID,
+    after_revision: int,
+    *,
+    page_size: int,
+) -> list[FlowOperation]:
+    """Return one page of accepted operations with revision strictly greater than ``after_revision``.
+
+    ``page_size`` is supplied by the API layer (default/max validation belong there, not in CRUD).
+    Callers are responsible for flow visibility checks before exposing rows.
+    """
+    stmt = (
+        select(FlowOperation)
+        .where(FlowOperation.flow_id == flow_id, FlowOperation.revision > after_revision)
+        .order_by(col(FlowOperation.revision).asc())
+        .limit(page_size)
+    )
+    return list((await session.exec(stmt)).all())

src/backend/base/langflow/services/database/models/flow_operation/model.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Any
+from uuid import UUID, uuid4
+
+from pydantic import BaseModel, field_serializer
+from pydantic import Field as PydanticField
+from sqlalchemy import BigInteger, Column, DateTime, ForeignKey, String, UniqueConstraint, func
+from sqlmodel import JSON, Field, SQLModel
+
+
+class FlowOperationActorDelegate(str, Enum):
+    SELF = "self"
+    AGENT = "agent"
+
+
+class FlowOperation(SQLModel, table=True):  # type: ignore[call-arg]
+    """Durable record of an accepted collaborative operation batch for a flow."""
+
+    __tablename__ = "flow_operation"
+    __mapper_args__ = {"confirm_deleted_rows": False}
+
+    id: UUID = Field(default_factory=uuid4, primary_key=True)
+    flow_id: UUID = Field(
+        sa_column=Column(ForeignKey("flow.id", ondelete="CASCADE"), nullable=False, index=True),
+    )
+    protocol_version: int = Field(nullable=False)
+    revision: int = Field(sa_column=Column(BigInteger, nullable=False))
+    client_id: str = Field(sa_column=Column(String, nullable=False))
+    actor_user_id: UUID | None = Field(
+        sa_column=Column(ForeignKey("user.id", ondelete="SET NULL"), nullable=True, index=True),
+    )
+    actor_delegate: FlowOperationActorDelegate = Field(
+        default=FlowOperationActorDelegate.SELF,
+        sa_column=Column(String, nullable=False, server_default=FlowOperationActorDelegate.SELF.value),
+    )
+    forward_ops: list[dict[str, Any]] = Field(sa_column=Column(JSON, nullable=False))
+    backward_ops: list[dict[str, Any]] = Field(sa_column=Column(JSON, nullable=False))
+    created_at: datetime = Field(
+        sa_column=Column(DateTime(timezone=True), server_default=func.now(), nullable=False),
+    )
+
+    __table_args__ = (UniqueConstraint("flow_id", "revision", name="unique_flow_operation_revision"),)
+
+
+class FlowOperationRead(BaseModel):
+    """Compact operation row for polling APIs."""
+
+    operation_id: UUID = PydanticField(description="Server-generated operation identity")
+    protocol_version: int
+    revision: int = PydanticField(ge=0)
+    client_id: str
+    actor_delegate: FlowOperationActorDelegate
+    # Nullable only for historical rows after the actor user has been deleted.
+    actor_user_id: UUID | None = None
+    forward_ops: list[dict[str, Any]]
+    created_at: datetime
+
+    @field_serializer("created_at")
+    def serialize_created_at(self, value: datetime) -> str:
+        value = value.replace(microsecond=0)
+        if value.tzinfo is None:
+            value = value.replace(tzinfo=timezone.utc)
+        return value.isoformat()