feat(collab): add SQLite collaboration event backplane

Introduce CollaborationEventService with a WAL SQLite mailbox keyed by
flow_id for cross-worker fanout. Register the service in deps/schema,
split event types into schemas.py, and add unit tests for publish/poll,
TTL, caps, and worker isolation. Fix service factory import inference for
lfx flow_operations.

Author: HzaRashid

src/backend/base/langflow/services/collaboration_events/__init__.py

@@ -0,0 +1,10 @@
+from langflow.services.collaboration_events.schemas import CollaborationEvent, CollaborationPollCursor
+from langflow.services.collaboration_events.service import CollaborationEventService
+from langflow.services.collaboration_events.sqlite import SQLiteCollaborationEventService
+
+__all__ = [
+    "CollaborationEvent",
+    "CollaborationEventService",
+    "CollaborationPollCursor",
+    "SQLiteCollaborationEventService",
+]

src/backend/base/langflow/services/collaboration_events/factory.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from typing_extensions import override
+
+from langflow.services.collaboration_events.sqlite import SQLiteCollaborationEventService
+from langflow.services.factory import ServiceFactory
+
+if TYPE_CHECKING:
+    from lfx.services.settings.service import SettingsService
+
+    from langflow.services.collaboration_events.service import CollaborationEventService
+
+
+class CollaborationEventServiceFactory(ServiceFactory):
+    def __init__(self) -> None:
+        super().__init__(SQLiteCollaborationEventService)
+
+    @override
+    def create(self, settings_service: SettingsService) -> CollaborationEventService:
+        return SQLiteCollaborationEventService(cache_dir=settings_service.settings.cache_dir)

src/backend/base/langflow/services/collaboration_events/schemas.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from uuid import UUID
+
+
+@dataclass(frozen=True)
+class CollaborationEvent:
+    """Opaque collaboration event stored in the cross-worker backplane."""
+
+    id: str
+    flow_id: UUID
+    created_at: float
+    type: str
+    payload: dict = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class CollaborationPollCursor:
+    """Per-worker poll position for a flow's collaboration event stream."""
+
+    created_at: float = 0.0
+    event_id: str = ""

src/backend/base/langflow/services/collaboration_events/service.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+from langflow.services.base import Service
+
+if TYPE_CHECKING:
+    from uuid import UUID
+
+    from langflow.services.collaboration_events.schemas import CollaborationEvent, CollaborationPollCursor
+
+
+class CollaborationEventService(Service, ABC):
+    """Cross-worker event backplane for collaborative flow editing.
+
+    Publishes opaque events scoped by ``flow_id``. Workers poll events for flows
+    they are actively serving and fan them out to local WebSocket rooms.
+    """
+
+    name = "collaboration_events_service"
+
+    @abstractmethod
+    def publish(self, flow_id: UUID, event_type: str, payload: dict) -> CollaborationEvent:
+        """Persist an event for cross-worker fanout."""
+
+    @abstractmethod
+    def poll(
+        self,
+        flow_id: UUID,
+        *,
+        cursor: CollaborationPollCursor | None = None,
+        limit: int | None = None,
+    ) -> tuple[list[CollaborationEvent], CollaborationPollCursor]:
+        """Return events after ``cursor`` and the updated poll cursor."""
+
+    @abstractmethod
+    def cleanup(self) -> None:
+        """Force-evict expired events. Useful for tests and ops scripts."""

src/backend/base/langflow/services/collaboration_events/sqlite.py

@@ -0,0 +1,193 @@
+from __future__ import annotations
+
+import json
+import sqlite3
+import tempfile
+import threading
+import time
+import uuid
+from pathlib import Path
+from uuid import UUID
+
+from langflow.services.collaboration_events.schemas import CollaborationEvent, CollaborationPollCursor
+from langflow.services.collaboration_events.service import CollaborationEventService
+
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS collaboration_events (
+    id         TEXT NOT NULL,
+    flow_id    TEXT NOT NULL,
+    created_at REAL NOT NULL,
+    type       TEXT NOT NULL,
+    payload    TEXT NOT NULL DEFAULT '{}',
+    expires_at REAL NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_collab_events_flow_created_id
+    ON collaboration_events(flow_id, created_at, id);
+CREATE INDEX IF NOT EXISTS idx_collab_events_expires ON collaboration_events(expires_at);
+"""
+
+
+class SQLiteCollaborationEventService(CollaborationEventService):
+    """SQLite WAL-backed collaboration event mailbox keyed by ``flow_id``.
+
+    Uses stdlib ``sqlite3`` so multiple uvicorn/gunicorn workers on the same host
+    can share one cache-dir database file. This is polling-based fanout, not
+    pub/sub.
+    """
+
+    TTL_SECONDS: float = 120.0
+    MAX_EVENTS_PER_FLOW: int = 1000
+    DEFAULT_POLL_LIMIT: int = 200
+
+    def __init__(self, cache_dir: str | Path | None = None) -> None:
+        if cache_dir is None:
+            cache_dir = Path(tempfile.gettempdir()) / "langflow_collaboration_events"
+        cache_dir = Path(cache_dir)
+        cache_dir.mkdir(parents=True, exist_ok=True)
+        self._db_path = cache_dir / "collaboration_events.sqlite"
+        self._conn = sqlite3.connect(
+            str(self._db_path),
+            isolation_level=None,
+            check_same_thread=False,
+            timeout=5.0,
+        )
+        # SQLite locks coordinate access across connections/processes. This lock
+        # coordinates threads in this worker that share the same connection object.
+        self._lock = threading.Lock()
+        with self._lock:
+            self._conn.execute("PRAGMA journal_mode=WAL")
+            self._conn.execute("PRAGMA synchronous=NORMAL")
+            self._conn.execute("PRAGMA busy_timeout=5000")
+            self._conn.executescript(_SCHEMA)
+
+    def publish(self, flow_id: UUID, event_type: str, payload: dict) -> CollaborationEvent:
+        flow_id_key = str(flow_id)
+        if not event_type:
+            msg = "event_type is required"
+            raise ValueError(msg)
+        if payload is None:
+            msg = "payload must be a dict"
+            raise TypeError(msg)
+
+        event_id = str(uuid.uuid4())
+        event_payload = dict(payload)
+        payload_json = json.dumps(event_payload)
+
+        with self._lock:
+            self._conn.execute("BEGIN IMMEDIATE")
+            try:
+                now = time.time()
+                event = CollaborationEvent(
+                    id=event_id,
+                    flow_id=flow_id,
+                    created_at=now,
+                    type=event_type,
+                    payload=event_payload,
+                )
+                expires_at = now + self.TTL_SECONDS
+                self._purge_expired_locked(now)
+                self._conn.execute(
+                    """
+                    INSERT INTO collaboration_events
+                        (id, flow_id, created_at, type, payload, expires_at)
+                    VALUES (?, ?, ?, ?, ?, ?)
+                    """,
+                    (event_id, flow_id_key, now, event_type, payload_json, expires_at),
+                )
+                self._enforce_per_flow_cap_locked(flow_id_key)
+                self._conn.execute("COMMIT")
+            except Exception:
+                self._conn.execute("ROLLBACK")
+                raise
+
+        return event
+
+    def poll(
+        self,
+        flow_id: UUID,
+        *,
+        cursor: CollaborationPollCursor | None = None,
+        limit: int | None = None,
+    ) -> tuple[list[CollaborationEvent], CollaborationPollCursor]:
+        flow_id_key = str(flow_id)
+
+        poll_limit = self.DEFAULT_POLL_LIMIT if limit is None else limit
+        if poll_limit < 1:
+            msg = "limit must be at least 1"
+            raise ValueError(msg)
+
+        cursor = cursor or CollaborationPollCursor()
+        # Match SQLite's cross-worker granularity: cleanup and read are separate statements,
+        # not one local-only critical section that blocks same-worker publishes.
+        with self._lock:
+            now = time.time()
+            self._purge_expired_locked(now)
+
+        with self._lock:
+            now = time.time()
+            rows = self._conn.execute(
+                """
+                SELECT id, created_at, type, payload
+                FROM collaboration_events
+                WHERE flow_id = ?
+                  AND expires_at >= ?
+                  AND (
+                        created_at > ?
+                     OR (created_at = ? AND id > ?)
+                  )
+                ORDER BY created_at ASC, id ASC
+                LIMIT ?
+                """,
+                (
+                    flow_id_key,
+                    now,
+                    cursor.created_at,
+                    cursor.created_at,
+                    cursor.event_id,
+                    poll_limit,
+                ),
+            ).fetchall()
+
+        events = [
+            CollaborationEvent(
+                id=row[0],
+                flow_id=flow_id,
+                created_at=row[1],
+                type=row[2],
+                payload=json.loads(row[3]),
+            )
+            for row in rows
+        ]
+
+        if not events:
+            return [], cursor
+
+        last = events[-1]
+        return events, CollaborationPollCursor(created_at=last.created_at, event_id=last.id)
+
+    def cleanup(self) -> None:
+        with self._lock:
+            now = time.time()
+            self._purge_expired_locked(now)
+
+    async def teardown(self) -> None:
+        with self._lock:
+            self._conn.close()
+
+    def _purge_expired_locked(self, now: float) -> None:
+        self._conn.execute("DELETE FROM collaboration_events WHERE expires_at < ?", (now,))
+
+    def _enforce_per_flow_cap_locked(self, flow_id: str) -> None:
+        # SQLite: LIMIT -1 means no row cap; OFFSET skips the N newest rows so we delete older ones.
+        self._conn.execute(
+            """
+            DELETE FROM collaboration_events
+            WHERE rowid IN (
+                SELECT rowid FROM collaboration_events
+                WHERE flow_id = ?
+                ORDER BY created_at DESC, id DESC
+                LIMIT -1 OFFSET ?
+            )
+            """,
+            (flow_id, self.MAX_EVENTS_PER_FLOW),
+        )

src/backend/base/langflow/services/deps.py

@@ -280,6 +280,12 @@ def get_flow_events_service():
     return get_service(ServiceType.FLOW_EVENTS_SERVICE, FlowEventsServiceFactory())
 
 
+def get_collaboration_events_service():
+    from langflow.services.collaboration_events.factory import CollaborationEventServiceFactory
+
+    return get_service(ServiceType.COLLABORATION_EVENTS_SERVICE, CollaborationEventServiceFactory())
+
+
 def get_flow_operation_service() -> BaseFlowOperationService:
     """Retrieves the FlowOperationService instance from the service manager."""
     from lfx.services.flow_operations.factory import FlowOperationServiceFactory

src/backend/base/langflow/services/factory.py

@@ -76,8 +76,8 @@ def import_all_services_into_a_dict():
         try:
             service_name = ServiceType(service_type).value.replace("_service", "")
 
-            # Special handling for mcp_composer which is now in lfx module
-            if service_name == "mcp_composer":
+            # Services implemented in the lfx package (not langflow.services.*)
+            if service_name in {"mcp_composer", "flow_operations"}:
                 module_name = f"lfx.services.{service_name}.service"
             else:
                 module_name = f"langflow.services.{service_name}.service"

src/backend/base/langflow/services/schema.py

@@ -23,5 +23,6 @@ class ServiceType(str, Enum):
     MCP_COMPOSER_SERVICE = "mcp_composer_service"
     JOB_SERVICE = "jobs_service"
     FLOW_EVENTS_SERVICE = "flow_events_service"
+    COLLABORATION_EVENTS_SERVICE = "collaboration_events_service"
     FLOW_OPERATIONS_SERVICE = "flow_operations_service"
     MEMORY_BASE_SERVICE = "memory_base_service"