Add Memory Base API: models, migrations, service, endpoints, and tests

Introduces Memory Base (MB) — a per-flow knowledge base that auto-captures
conversation history and ingests it into a Chroma vector store on configurable
thresholds.

Backend changes:
- MemoryBase + MemoryBaseSession DB models with full CRUD
- Three Alembic migrations: base tables, merge head, phase-2 fields
  (embedding_model, preprocessing, preproc_model, preproc_instructions)
- MemoryBaseService: create/list/get/update/delete, session tracking,
  pending-message cursor logic, mismatch detection, regenerate
- ingest_memory_task: async Chroma ingestion with cursor advance on success
- REST API (/api/v1/memories): CRUD, flush, sessions, mismatch, regenerate
- Flow output hook: on_flow_output() triggers auto-capture after each run
- Plumbing in build.py, endpoints.py, workflow.py to call on_flow_output
- deps.py: expose get_memory_base_service()
- kb_helpers.py: FS/metadata helpers used by MB service

Authored-By: Debojit Kaushik <kaushik.debojit@gmail.com>

Add dedupe_key idempotency enforcement for MB ingestion jobs

Centralizes idempotency into JobService.create_job() with a null-safe check,
removing the redundant pre-flight logic from MemoryBaseService.

Key changes:
- services/jobs/exceptions.py: new DuplicateJobError(RuntimeError) —
  raised when a QUEUED/IN_PROGRESS/COMPLETED job with the same dedupe_key
  exists; FAILED/CANCELLED are retryable and are excluded
- services/jobs/__init__.py: exports DuplicateJobError
- services/jobs/service.py: null-guarded dedup query inside create_job()
  within the same session_scope as the insert (minimizes TOCTOU window)
- services/database/models/jobs/model.py: dedupe_key field -> index=True
- alembic/versions/36aa87831162: adds dedupe_key column + ix_job_dedupe_key
  index to job table with checkfirst guards
- services/memory_base/service.py: updated key format to
  "ingestion:{mb_id}:{session_id}:{first_msg_id}" for namespace isolation;
  removed _has_non_retryable_job_for_dedupe_key and _has_active_job methods
  and all call sites; DuplicateJobError catch in _maybe_trigger() for
  silent skip on auto-capture; split regenerate() catch clauses
- api/v1/memories.py: explicit DuplicateJobError catch before RuntimeError
  in flush_memory_base() for semantic clarity (both return 409)

Co-Authored-By: Debojit Kaushik <kaushik.debojit@gmail.com>

Checkpointing working version of MBs. TODO: User separation, Get messages endpoint, MB resumption midway through a chat for a session, tests.

Added messages endpoint for Memory Bases. Added pagination to sessions endpoint. Modifed messages model to include ingestion related attributes.

Added unit tests, fixed linting issues and formatting issues.

Aligned Workflows API, /run endpoint, playground to all work with Memory Bases. Created DB models asociated with tracking memory base state with sessions and jobs. Added tests, created unit tests for service and task files related to MemoryBases.

Improved concurrency of jobs, added (memory_base_id, session_id) locking to serialize jobs in case the job creation cadence moves ahead of ingestion jobs. Improved concurrency handling and moved the pending check to be live inside the ingestion job rather than a snapshot before triggering the job.

Consolidated all migrations related to memory_bases into one idempotent version and serialized all migrations form release along with memory_bases for cleanliness and maintainability.

Introduced advisory locking to address multi worker environment, added unique constraint to MB creation, added sanitization check to KB pathnames to avoid illegal directory creation.

Added partial write rollback for ChromaDB, aligned same session is  used for each job to avoid dangling advisory locks.

Author: dkaushik94

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

@@ -0,0 +1,235 @@
+"""add_memory_base_schema
+
+Consolidates all Memory Base schema changes into a single migration:
+  - job.dedupe_key (nullable String) + ix_job_dedupe_key
+  - message.run_id (nullable UUID) + ix_message_run_id
+  - message.is_output (bool, default false)
+  - memory_base table + ix_memory_base_flow_id + ix_memory_base_user_id
+  - memory_base_session table + three indexes
+  - message_ingestion_record table + three indexes
+  - memory_base_workflow_run table + two indexes
+
+Phase: EXPAND
+
+Revision ID: mb00a1b2c3d4
+Revises: d306e5c17c41
+Create Date: 2026-04-14 00:00:00.000000
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from langflow.utils import migration
+
+# revision identifiers, used by Alembic.
+revision: str = "mb00a1b2c3d4"  # pragma: allowlist secret
+down_revision: str | None = "d306e5c17c41"  # pragma: allowlist secret
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+
+    # ------------------------------------------------------------------ #
+    #  job.dedupe_key                                                     #
+    # ------------------------------------------------------------------ #
+    inspector = sa.inspect(conn)
+    existing_job_indexes = {idx["name"] for idx in inspector.get_indexes("job")}
+    with op.batch_alter_table("job", schema=None) as batch_op:
+        if not migration.column_exists("job", "dedupe_key", conn):
+            batch_op.add_column(sa.Column("dedupe_key", sa.String(), nullable=True))
+        if "ix_job_dedupe_key" not in existing_job_indexes:
+            batch_op.create_index(batch_op.f("ix_job_dedupe_key"), ["dedupe_key"], unique=False)
+
+    # ------------------------------------------------------------------ #
+    #  message.run_id + message.is_output                                 #
+    # ------------------------------------------------------------------ #
+    with op.batch_alter_table("message", schema=None) as batch_op:
+        if not migration.column_exists("message", "run_id", conn):
+            batch_op.add_column(sa.Column("run_id", sa.Uuid(), nullable=True))
+        if not migration.column_exists("message", "is_output", conn):
+            batch_op.add_column(sa.Column("is_output", sa.Boolean(), nullable=False, server_default=sa.text("false")))
+
+    existing_message_indexes = {idx["name"] for idx in sa.inspect(conn).get_indexes("message")}
+    if "ix_message_run_id" not in existing_message_indexes:
+        op.create_index("ix_message_run_id", "message", ["run_id"])
+
+    # ------------------------------------------------------------------ #
+    #  memory_base                                                         #
+    # ------------------------------------------------------------------ #
+    if not migration.table_exists("memory_base", conn):
+        op.create_table(
+            "memory_base",
+            sa.Column("id", sa.Uuid(), nullable=False),
+            sa.Column("name", sa.String(), nullable=False),
+            sa.Column("flow_id", sa.Uuid(), nullable=False),
+            sa.Column("user_id", sa.Uuid(), nullable=False),
+            sa.Column("threshold", sa.Integer(), nullable=False, server_default=sa.text("50")),
+            sa.Column("auto_capture", sa.Boolean(), nullable=False, server_default=sa.text("true")),
+            sa.Column("embedding_model", sa.String(), nullable=False, server_default=sa.text("''")),
+            sa.Column("preprocessing", sa.Boolean(), nullable=False, server_default=sa.text("false")),
+            sa.Column("preproc_model", sa.String(), nullable=True),
+            sa.Column("preproc_instructions", sa.String(), nullable=True),
+            sa.Column("kb_name", sa.String(), nullable=False),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.PrimaryKeyConstraint("id"),
+            sa.UniqueConstraint("user_id", "name", name="uq_memory_base_user_name"),
+            sa.Index("ix_memory_base_flow_id", "flow_id"),
+            sa.Index("ix_memory_base_user_id", "user_id"),
+        )
+
+    # ------------------------------------------------------------------ #
+    #  memory_base_session                                                 #
+    # ------------------------------------------------------------------ #
+    if not migration.table_exists("memory_base_session", conn):
+        op.create_table(
+            "memory_base_session",
+            sa.Column("id", sa.Uuid(), nullable=False),
+            sa.Column(
+                "memory_base_id",
+                sa.Uuid(),
+                sa.ForeignKey("memory_base.id", ondelete="CASCADE"),
+                nullable=False,
+            ),
+            sa.Column("session_id", sa.String(), nullable=False),
+            sa.Column("cursor_id", sa.Uuid(), nullable=True),
+            sa.Column("total_processed", sa.Integer(), nullable=False, server_default=sa.text("0")),
+            sa.Column("last_sync_at", sa.DateTime(timezone=True), nullable=True),
+            sa.PrimaryKeyConstraint("id"),
+            sa.UniqueConstraint("memory_base_id", "session_id", name="uq_memory_base_session"),
+        )
+        op.create_index("ix_memory_base_session_memory_base_id", "memory_base_session", ["memory_base_id"])
+        op.create_index("ix_memory_base_session_session_id", "memory_base_session", ["session_id"])
+        op.create_index(
+            "ix_memory_base_session_lookup",
+            "memory_base_session",
+            ["memory_base_id", "session_id"],
+        )
+
+    # ------------------------------------------------------------------ #
+    #  message_ingestion_record                                            #
+    # ------------------------------------------------------------------ #
+    if not migration.table_exists("message_ingestion_record", conn):
+        op.create_table(
+            "message_ingestion_record",
+            sa.Column("id", sa.Uuid(), nullable=False),
+            sa.Column(
+                "message_id",
+                sa.Uuid(),
+                sa.ForeignKey("message.id", ondelete="CASCADE"),
+                nullable=False,
+            ),
+            sa.Column(
+                "memory_base_id",
+                sa.Uuid(),
+                sa.ForeignKey("memory_base.id", ondelete="CASCADE"),
+                nullable=False,
+            ),
+            sa.Column(
+                "job_id",
+                sa.Uuid(),
+                sa.ForeignKey("job.job_id", ondelete="SET NULL"),
+                nullable=True,
+            ),
+            sa.Column("session_id", sa.String(), nullable=False),
+            sa.Column("ingested_at", sa.DateTime(timezone=True), nullable=False),
+            sa.PrimaryKeyConstraint("id"),
+            sa.UniqueConstraint(
+                "message_id",
+                "session_id",
+                "memory_base_id",
+                name="uq_mir_message_session_mb",
+            ),
+        )
+        op.create_index("ix_mir_message_id", "message_ingestion_record", ["message_id"])
+        op.create_index("ix_mir_job_id", "message_ingestion_record", ["job_id"])
+        op.create_index(
+            "ix_mir_memory_base_session",
+            "message_ingestion_record",
+            ["memory_base_id", "session_id"],
+        )
+
+    # ------------------------------------------------------------------ #
+    #  memory_base_workflow_run                                            #
+    # ------------------------------------------------------------------ #
+    if not migration.table_exists("memory_base_workflow_run", conn):
+        op.create_table(
+            "memory_base_workflow_run",
+            sa.Column("id", sa.Uuid(), nullable=False),
+            sa.Column(
+                "memory_base_id",
+                sa.Uuid(),
+                sa.ForeignKey("memory_base.id", ondelete="CASCADE"),
+                nullable=False,
+            ),
+            sa.Column("session_id", sa.String(), nullable=False),
+            sa.Column(
+                "workflow_job_id",
+                sa.Uuid(),
+                sa.ForeignKey("job.job_id", ondelete="SET NULL"),
+                nullable=True,
+            ),
+            sa.Column(
+                "ingestion_job_id",
+                sa.Uuid(),
+                sa.ForeignKey("job.job_id", ondelete="SET NULL"),
+                nullable=True,
+            ),
+            sa.Column("recorded_at", sa.DateTime(timezone=True), nullable=False),
+            sa.PrimaryKeyConstraint("id"),
+            sa.UniqueConstraint(
+                "memory_base_id",
+                "session_id",
+                "workflow_job_id",
+                name="uq_mbwr_mb_session_wf_job",
+            ),
+        )
+        op.create_index("ix_mbwr_mb_session", "memory_base_workflow_run", ["memory_base_id", "session_id"])
+        op.create_index("ix_mbwr_ingestion_job_id", "memory_base_workflow_run", ["ingestion_job_id"])
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+
+    # Children first (FK dependencies) ----------------------------------- #
+    if migration.table_exists("memory_base_workflow_run", conn):
+        op.drop_index("ix_mbwr_ingestion_job_id", table_name="memory_base_workflow_run")
+        op.drop_index("ix_mbwr_mb_session", table_name="memory_base_workflow_run")
+        op.drop_table("memory_base_workflow_run")
+
+    if migration.table_exists("message_ingestion_record", conn):
+        op.drop_index("ix_mir_memory_base_session", table_name="message_ingestion_record")
+        op.drop_index("ix_mir_job_id", table_name="message_ingestion_record")
+        op.drop_index("ix_mir_message_id", table_name="message_ingestion_record")
+        op.drop_table("message_ingestion_record")
+
+    if migration.table_exists("memory_base_session", conn):
+        op.drop_index("ix_memory_base_session_lookup", table_name="memory_base_session")
+        op.drop_index("ix_memory_base_session_session_id", table_name="memory_base_session")
+        op.drop_index("ix_memory_base_session_memory_base_id", table_name="memory_base_session")
+        op.drop_table("memory_base_session")
+
+    if migration.table_exists("memory_base", conn):
+        op.drop_index("ix_memory_base_user_id", table_name="memory_base")
+        op.drop_index("ix_memory_base_flow_id", table_name="memory_base")
+        op.drop_table("memory_base")
+
+    # Message column/index ----------------------------------------------- #
+    existing_message_indexes = {idx["name"] for idx in sa.inspect(conn).get_indexes("message")}
+    if "ix_message_run_id" in existing_message_indexes:
+        op.drop_index("ix_message_run_id", table_name="message")
+    with op.batch_alter_table("message", schema=None) as batch_op:
+        if migration.column_exists("message", "is_output", conn):
+            batch_op.drop_column("is_output")
+        if migration.column_exists("message", "run_id", conn):
+            batch_op.drop_column("run_id")
+
+    # Job column/index --------------------------------------------------- #
+    with op.batch_alter_table("job", schema=None) as batch_op:
+        existing_job_indexes = {idx["name"] for idx in sa.inspect(conn).get_indexes("job")}
+        if "ix_job_dedupe_key" in existing_job_indexes:
+            batch_op.drop_index(batch_op.f("ix_job_dedupe_key"))
+        if migration.column_exists("job", "dedupe_key", conn):
+            batch_op.drop_column("dedupe_key")

src/backend/base/langflow/api/build.py

@@ -29,7 +29,15 @@
 from langflow.schema.message import ErrorMessage
 from langflow.schema.schema import OutputValue
 from langflow.services.database.models.flow.model import Flow
-from langflow.services.deps import get_chat_service, get_telemetry_service, session_scope
+from langflow.services.database.models.jobs.model import JobType
+from langflow.services.deps import (
+    get_chat_service,
+    get_job_service,
+    get_memory_base_service,
+    get_task_service,
+    get_telemetry_service,
+    session_scope,
+)
 from langflow.services.job_queue.service import JobQueueNotFoundError, JobQueueService
 from langflow.services.telemetry.schema import ComponentInputsPayload, ComponentPayload, PlaygroundPayload
 
@@ -527,35 +535,81 @@ async def build_vertices(
         event_manager.on_error(data=error_message.data)
         raise
 
+    # Create a WORKFLOW job record so memory-base on_flow_output can track this run.
+    # Best-effort: failures here must never break the build path.
+    _build_job_svc = None
+    _build_run_id: uuid.UUID | None = None
+    try:
+        _build_run_id = uuid.UUID(graph.run_id) if graph.run_id else None
+        if _build_run_id is not None:
+            _build_job_svc = get_job_service()
+            await _build_job_svc.create_job(
+                job_id=_build_run_id,
+                flow_id=flow_id,
+                user_id=current_user.id,
+                job_type=JobType.WORKFLOW,
+            )
+    except Exception:  # noqa: BLE001
+        await logger.awarning(
+            "Failed to create workflow job for /build — memory base tracking disabled for flow %s",
+            flow_id,
+            exc_info=True,
+        )
+        _build_job_svc = None
+
     event_manager.on_vertices_sorted(data={"ids": ids, "to_run": vertices_to_run})
 
     vertex_timedeltas: list[float] = []
     event_manager.on_build_start(data={})
-    tasks = []
-    for vertex_id in ids:
-        task = asyncio.create_task(build_vertices(vertex_id, graph, event_manager, vertex_timedeltas))
-        tasks.append(task)
-    try:
-        await asyncio.gather(*tasks)
-    except asyncio.CancelledError:
-        background_tasks.add_task(graph.end_all_traces_in_context())
-        raise
-    except Exception as e:
-        await logger.aerror(f"Error building vertices: {e}")
-        custom_component = graph.get_vertex(vertex_id).custom_component
-        trace_name = getattr(custom_component, "trace_name", None)
-        error_message = ErrorMessage(
-            flow_id=flow_id,
-            exception=e,
-            session_id=graph.session_id,
-            trace_name=trace_name,
-        )
-        event_manager.on_error(data=error_message.data)
-        raise
+
+    async def _run_vertex_build() -> None:
+        tasks = []
+        for vertex_id in ids:
+            task = asyncio.create_task(build_vertices(vertex_id, graph, event_manager, vertex_timedeltas))
+            tasks.append(task)
+        try:
+            await asyncio.gather(*tasks)
+        except asyncio.CancelledError:
+            background_tasks.add_task(graph.end_all_traces_in_context())
+            raise
+        except Exception as e:
+            await logger.aerror(f"Error building vertices: {e}")
+            custom_component = graph.get_vertex(vertex_id).custom_component
+            trace_name = getattr(custom_component, "trace_name", None)
+            error_message = ErrorMessage(
+                flow_id=flow_id,
+                exception=e,
+                session_id=graph.session_id,
+                trace_name=trace_name,
+            )
+            event_manager.on_error(data=error_message.data)
+            raise
+
+    if _build_job_svc and _build_run_id:
+        await _build_job_svc.execute_with_status(_build_run_id, _run_vertex_build)
+    else:
+        await _run_vertex_build()
 
     build_duration = sum(vertex_timedeltas)
     event_manager.on_end(data={"build_duration": build_duration})
     await graph.end_all_traces()
+
+    # Fire memory-base auto-capture hook — non-blocking background effect.
+    # Must use fire_and_forget_task (not background_tasks.add_task) because
+    # generate_flow_events runs as an asyncio task; by the time the flow
+    # finishes, FastAPI has already drained the background_tasks queue and any
+    # tasks added after that point are silently dropped.
+    try:
+        _run_id_uuid = uuid.UUID(graph.run_id) if graph.run_id else None  # type-cast only; same run_id set on graph
+        await get_task_service().fire_and_forget_task(
+            get_memory_base_service().on_flow_output,
+            flow_id=flow_id,
+            session_id=graph.session_id or str(flow_id),
+            job_id=_run_id_uuid,
+        )
+    except (RuntimeError, ValueError, OSError):
+        await logger.awarning("Memory base hook scheduling failed for flow %s", flow_id, exc_info=True)
+
     await event_manager.queue.put((None, None, time.time()))
 
 

src/backend/base/langflow/api/router.py

@@ -15,6 +15,7 @@
     login_router,
     mcp_projects_router,
     mcp_router,
+    memories_router,
     model_options_router,
     models_router,
     monitor_router,
@@ -68,6 +69,7 @@ def include_deployment_router(target_router: APIRouter) -> None:
 router_v1.include_router(projects_router)
 router_v1.include_router(starter_projects_router)
 router_v1.include_router(knowledge_bases_router)
+router_v1.include_router(memories_router)
 router_v1.include_router(mcp_router)
 router_v1.include_router(voice_mode_router)
 router_v1.include_router(mcp_projects_router)