feat(api): durable background execution service (store + default backend)

.github/workflows/migration-validation.yml

@@ -8,6 +8,10 @@ on:
       - 'src/backend/base/langflow/services/database/service.py'
       - 'src/backend/tests/unit/alembic/**'
       - '.github/workflows/migration-validation.yml'
+      - 'src/backend/base/langflow/services/background_execution/**'
+      - 'src/backend/base/langflow/services/jobs/**'
+      - 'src/backend/base/langflow/services/job_queue/**'
+      - 'src/backend/tests/unit/background_execution/**'
 
 jobs:
   model-migration-consistency:
@@ -19,7 +23,7 @@ jobs:
         image: postgres:16
         env:
           POSTGRES_USER: langflow
-          POSTGRES_PASSWORD: langflow
+          POSTGRES_PASSWORD: langflow  # pragma: allowlist secret
           POSTGRES_DB: langflow
         ports:
           - 5432:5432
@@ -50,10 +54,63 @@ jobs:
       - name: Check model/migration consistency
         env:
           MIGRATION_VALIDATION_CI: "true"
-          LANGFLOW_TEST_DATABASE_URI: "postgresql://langflow:langflow@localhost:5432/langflow"
+          LANGFLOW_TEST_DATABASE_URI: "postgresql://langflow:langflow@localhost:5432/langflow"  # pragma: allowlist secret
         run: |
           uv run pytest src/backend/tests/unit/alembic/test_migration_execution.py -x -v
 
+  background-real-service:
+    name: Background Execution Real-Service Tests (real Postgres + Redis)
+    runs-on: ubuntu-latest
+
+    services:
+      postgres:
+        image: postgres:16
+        env:
+          POSTGRES_USER: langflow
+          POSTGRES_PASSWORD: langflow  # pragma: allowlist secret
+          POSTGRES_DB: langflow
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd="pg_isready -U langflow"
+          --health-interval=10s
+          --health-timeout=5s
+          --health-retries=5
+      redis:
+        image: redis:7
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd="redis-cli ping"
+          --health-interval=10s
+          --health-timeout=5s
+          --health-retries=5
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v6
+        with:
+          fetch-depth: 0
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v6
+
+      - name: Setup Python
+        uses: actions/setup-python@v6
+        with:
+          python-version: '3.12'
+
+      - name: Install dependencies
+        run: |
+          uv sync --extra postgresql
+
+      - name: Run real-service tests
+        env:
+          LANGFLOW_TEST_DATABASE_URI: "postgresql://langflow:langflow@localhost:5432/langflow"  # pragma: allowlist secret
+          LANGFLOW_TEST_REDIS_URL: "redis://localhost:6379/0"
+        run: |
+          uv run pytest src/backend/tests/unit/background_execution -m real_services -x -v
+
   validate-migration:
     name: Migration Pattern Validation
     runs-on: ubuntu-latest

.secrets.baseline

@@ -153,24 +153,6 @@
         "is_secret": false
       }
     ],
-    ".github/workflows/migration-validation.yml": [
-      {
-        "type": "Secret Keyword",
-        "filename": ".github/workflows/migration-validation.yml",
-        "hashed_secret": "e80c4f90316c87b6b24d03890493c8d1c7c1c99d",
-        "is_verified": false,
-        "line_number": 22,
-        "is_secret": false
-      },
-      {
-        "type": "Basic Auth Credentials",
-        "filename": ".github/workflows/migration-validation.yml",
-        "hashed_secret": "e80c4f90316c87b6b24d03890493c8d1c7c1c99d",
-        "is_verified": false,
-        "line_number": 53,
-        "is_secret": false
-      }
-    ],
     ".github/workflows/nightly_build.yml": [
       {
         "type": "Secret Keyword",
@@ -1175,15 +1157,15 @@
         "filename": "src/backend/tests/conftest.py",
         "hashed_secret": "8bb6118f8fd6935ad0876a3be34a717d32708ffd",
         "is_verified": false,
-        "line_number": 555,
+        "line_number": 559,
         "is_secret": false
       },
       {
         "type": "Secret Keyword",
         "filename": "src/backend/tests/conftest.py",
         "hashed_secret": "61fbb5a12cd7b1f1fe1624120089efc0cd299e43",
         "is_verified": false,
-        "line_number": 765,
+        "line_number": 769,
         "is_secret": false
       }
     ],
@@ -7272,5 +7254,5 @@
       }
     ]
   },
-  "generated_at": "2026-07-08T10:30:03Z"
+  "generated_at": "2026-07-08T18:19:43Z"
 }

Makefile

@@ -168,6 +168,13 @@ unit_tests: ## run unit tests
 unit_tests_looponfail:
 	@make unit_tests args="-f"
 
+real_services_tests: ## run tests that need real service instances (needs LANGFLOW_TEST_DATABASE_URI + LANGFLOW_TEST_REDIS_URL)
+	@uv sync --frozen
+	uv run pytest src/backend/tests/unit \
+	--ignore=src/backend/tests/integration \
+	--ignore=src/backend/tests/unit/template \
+	-m real_services -ra $(args)
+
 lfx_tests: ## run lfx package unit tests
 	@echo 'Running LFX Package Tests...'
 	@cd src/lfx && \

pyproject.toml

@@ -270,7 +270,8 @@ markers = [
     "unit: Unit tests",
     "integration: Integration tests",
     "slow: Slow-running tests",
-    "security: Security regression tests (IDOR, auth, access control)"
+    "security: Security regression tests (IDOR, auth, access control)",
+    "real_services: Tests that need real service instances (real SQLite + real Postgres + real Redis)"
 ]
 asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "function"

src/backend/base/langflow/alembic/versions/185482a2d715_add_result_error_to_job.py

@@ -0,0 +1,44 @@
+"""add result and error columns to job table.
+
+Revision ID: 185482a2d715
+Revises: b7c4d8e9f012
+Create Date: 2026-06-03 10:00:00.000000
+
+Phase: EXPAND
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = "185482a2d715"  # pragma: allowlist secret
+down_revision: str | None = "c3e7a1b9d2f4"  # pragma: allowlist secret
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    existing_columns = {col["name"] for col in sa.inspect(conn).get_columns("job")}
+
+    with op.batch_alter_table("job", schema=None) as batch_op:
+        if "result" not in existing_columns:
+            batch_op.add_column(sa.Column("result", _JSON, nullable=True))
+        if "error" not in existing_columns:
+            batch_op.add_column(sa.Column("error", _JSON, nullable=True))
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    existing_columns = {col["name"] for col in sa.inspect(conn).get_columns("job")}
+
+    with op.batch_alter_table("job", schema=None) as batch_op:
+        if "error" in existing_columns:
+            batch_op.drop_column("error")
+        if "result" in existing_columns:
+            batch_op.drop_column("result")

src/backend/base/langflow/alembic/versions/8ce44e4858c6_add_execution_signals_table.py

@@ -0,0 +1,58 @@
+"""add execution_signals table for cooperative job control.
+
+Revision ID: 8ce44e4858c6
+Revises: b026885b89c8
+Create Date: 2026-06-03 10:10:00.000000
+
+Phase: EXPAND
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = "8ce44e4858c6"  # pragma: allowlist secret
+down_revision: str | None = "b026885b89c8"  # pragma: allowlist secret
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
+
+
+def upgrade() -> None:
+    from langflow.utils import migration
+
+    conn = op.get_bind()
+    if not migration.table_exists("execution_signals", conn):
+        op.create_table(
+            "execution_signals",
+            sa.Column("id", sa.Uuid(), nullable=False),
+            sa.Column("job_id", sa.Uuid(), nullable=False),
+            sa.Column(
+                "signal_type",
+                sa.Enum("stop", name="execution_signal_type_enum"),
+                nullable=False,
+            ),
+            sa.Column("data", _JSON, nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("consumed_at", sa.DateTime(timezone=True), nullable=True),
+            sa.PrimaryKeyConstraint("id"),
+        )
+        with op.batch_alter_table("execution_signals", schema=None) as batch_op:
+            batch_op.create_index(batch_op.f("ix_execution_signals_id"), ["id"], unique=False)
+            batch_op.create_index(batch_op.f("ix_execution_signals_job_id"), ["job_id"], unique=False)
+
+
+def downgrade() -> None:
+    from langflow.utils import migration
+
+    conn = op.get_bind()
+    if migration.table_exists("execution_signals", conn):
+        op.drop_table("execution_signals")
+    # Drop the postgres enum type explicitly; sqlite has no standalone enum type.
+    bind = op.get_bind()
+    if bind.dialect.name == "postgresql":
+        sa.Enum(name="execution_signal_type_enum").drop(bind, checkfirst=True)

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

@@ -0,0 +1,51 @@
+"""add job_events table for durable background-job event log.
+
+Revision ID: b026885b89c8
+Revises: 185482a2d715
+Create Date: 2026-06-03 10:05:00.000000
+
+Phase: EXPAND
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = "b026885b89c8"  # pragma: allowlist secret
+down_revision: str | None = "185482a2d715"  # pragma: allowlist secret
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
+
+
+def upgrade() -> None:
+    from langflow.utils import migration
+
+    conn = op.get_bind()
+    if not migration.table_exists("job_events", conn):
+        op.create_table(
+            "job_events",
+            sa.Column("id", sa.Uuid(), nullable=False),
+            sa.Column("job_id", sa.Uuid(), nullable=False),
+            sa.Column("seq", sa.Integer(), nullable=False),
+            sa.Column("event_type", sa.String(), nullable=False),
+            sa.Column("payload", _JSON, nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.PrimaryKeyConstraint("id"),
+            sa.UniqueConstraint("job_id", "seq", name="uq_job_events_job_id_seq"),
+        )
+        with op.batch_alter_table("job_events", schema=None) as batch_op:
+            batch_op.create_index(batch_op.f("ix_job_events_id"), ["id"], unique=False)
+            batch_op.create_index(batch_op.f("ix_job_events_job_id"), ["job_id"], unique=False)
+
+
+def downgrade() -> None:
+    from langflow.utils import migration
+
+    conn = op.get_bind()
+    if migration.table_exists("job_events", conn):
+        op.drop_table("job_events")

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

@@ -8,6 +8,7 @@
 from fastapi import BackgroundTasks, HTTPException, Response
 from lfx.graph.graph.base import Graph
 from lfx.graph.utils import log_vertex_build
+from lfx.graph.vertex.base import Vertex
 from lfx.log.logger import logger
 from lfx.schema.legacy_render import project_payload_to_v1
 from lfx.schema.schema import InputValueRequest
@@ -394,6 +395,7 @@ async def generate_flow_events(
     source_flow_id: uuid.UUID | None = None,
     run_id: str | None = None,
     track_job_status: bool = True,
+    tweaks: dict | None = None,
 ) -> None:
     """Generate events for flow building process.
 
@@ -422,6 +424,22 @@ async def build_graph_and_get_order() -> tuple[list[str], list[str], Graph]:
             async with session_scope() as fresh_session:
                 graph = await create_graph(fresh_session, flow_id_str, flow_name)
 
+            # Apply request tweaks to the built graph. The sync path applies
+            # tweaks before Graph construction; the streaming/background path
+            # builds from the DB (or request data), so tweaks must be applied
+            # to the built graph here or they are silently dropped. We use
+            # ``update_raw_params`` rather than the lfx ``process_tweaks_on_graph``
+            # helper because that helper only sets ``vertex.params`` and does not
+            # persist the override to runtime (mirrors the workaround in
+            # ``lfx.base.tools.run_flow._process_tweaks_on_graph``).
+            if tweaks:
+                for vertex in graph.vertices:
+                    if not (isinstance(vertex, Vertex) and isinstance(vertex.id, str)):
+                        continue
+                    if node_tweaks := tweaks.get(vertex.id):
+                        node_tweaks = {k: v for k, v in node_tweaks.items() if k != "code"}
+                        vertex.update_raw_params(node_tweaks, overwrite=True)
+
             graph.set_run_id(build_run_id)
             first_layer = sort_vertices(graph)
 
@@ -786,16 +804,21 @@ async def _run_vertex_build() -> None:
     # 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)
+    # Gated on ``track_job_status`` for the same reason as the job row above:
+    # when a caller owns the run's lifecycle (the v2 durable background path
+    # passes ``track_job_status=False`` and fires this hook itself with the
+    # durable job_id), firing here too would double-capture the flow output.
+    if track_job_status:
+        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()))