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-hard-proof:
+    name: Background Execution Hard-Proof (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 hard-proof 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 hard_proof -x -v
+
   validate-migration:
     name: Migration Pattern Validation
     runs-on: ubuntu-latest

.secrets.baseline

@@ -2831,15 +2831,15 @@
         "filename": "src/backend/tests/conftest.py",
         "hashed_secret": "8bb6118f8fd6935ad0876a3be34a717d32708ffd",
         "is_verified": false,
-        "line_number": 480,
+        "line_number": 484,
         "is_secret": false
       },
       {
         "type": "Secret Keyword",
         "filename": "src/backend/tests/conftest.py",
         "hashed_secret": "61fbb5a12cd7b1f1fe1624120089efc0cd299e43",
         "is_verified": false,
-        "line_number": 690,
+        "line_number": 694,
         "is_secret": false
       }
     ],
@@ -9281,5 +9281,5 @@
       }
     ]
   },
-  "generated_at": "2026-05-29T22:28:26Z"
+  "generated_at": "2026-06-04T01:58:46Z"
 }

Makefile

@@ -168,6 +168,13 @@ unit_tests: ## run unit tests
 unit_tests_looponfail:
 	@make unit_tests args="-f"
 
+hard_proof_tests: ## run real-instance hard-proof tests (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 hard_proof -ra $(args)
+
 lfx_tests: ## run lfx package unit tests
 	@echo 'Running LFX Package Tests...'
 	@cd src/lfx && \

pyproject.toml

@@ -204,7 +204,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)",
+    "hard_proof: Real-instance proof tests (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 = "b7c4d8e9f012"  # 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

@@ -55,6 +55,32 @@
 STREAMING_ACTIVITY_REFRESH_S = 10.0
 
 
+def _output_meta_for_vertex(graph: Graph, vertex_id: str) -> dict:
+    """Authoritative per-output metadata for the v2 ``output`` stream event.
+
+    Sourced from the real graph vertex (the only place ``display_name`` /
+    ``is_output`` / declared output types are authoritative) so the streamed
+    ``OutputEvent`` matches the sync ``outputs[id]`` ``ComponentOutput``. Shipped as
+    an additive ``output_meta`` key on ``end_vertex``; existing consumers read
+    ``build_data`` and ignore this. ``is_terminal`` mirrors the sync ``outputs`` set
+    so the stream emits an ``output`` event for exactly the same components.
+    """
+    vertex = graph.get_vertex(vertex_id)
+    output_types = vertex.outputs[0].get("types", []) if (vertex.outputs and len(vertex.outputs) > 0) else []
+    try:
+        terminal_ids = set(graph.get_terminal_nodes())
+    except AttributeError:
+        terminal_ids = {v.id for v in graph.vertices if not graph.successor_map.get(v.id, [])}
+    return {
+        "component_id": vertex.id,
+        "display_name": vertex.display_name or vertex.vertex_type,
+        "vertex_type": vertex.vertex_type,
+        "is_output": bool(vertex.is_output),
+        "is_terminal": vertex_id in terminal_ids,
+        "output_types": output_types,
+    }
+
+
 def _log_component_input_telemetry(
     vertex,
     vertex_id: str,
@@ -615,7 +641,9 @@ async def build_vertices(
             msg = f"Error serializing vertex build response: {exc}"
             raise ValueError(msg) from exc
 
-        event_manager.on_end_vertex(data={"build_data": build_data})
+        event_manager.on_end_vertex(
+            data={"build_data": build_data, "output_meta": _output_meta_for_vertex(graph, vertex_id)}
+        )
 
         if vertex_build_response.valid and vertex_build_response.next_vertices_ids:
             tasks = []

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

@@ -39,6 +39,7 @@
 from langflow.api.v2 import files_router as files_router_v2
 from langflow.api.v2 import mcp_router as mcp_router_v2
 from langflow.api.v2 import registration_router as registration_router_v2
+from langflow.api.v2 import workflow_public_router as workflow_public_router_v2
 from langflow.api.v2 import workflow_router as workflow_router_v2
 
 router_v1 = APIRouter(
@@ -124,6 +125,7 @@ def _include_agentic_router():
 router_v2.include_router(mcp_router_v2)
 router_v2.include_router(registration_router_v2)
 router_v2.include_router(workflow_router_v2)
+router_v2.include_router(workflow_public_router_v2)
 
 router = APIRouter(
     prefix="/api",

src/backend/base/langflow/api/utils/__init__.py

@@ -45,6 +45,7 @@
     build_graph_from_db_no_cache,
     cascade_delete_flow,
     scope_session_to_namespace,
+    validate_public_files,
     verify_public_flow_and_get_user,
 )
 
@@ -90,5 +91,6 @@
     "remove_api_keys",
     "scope_session_to_namespace",
     "validate_is_component",
+    "validate_public_files",
     "verify_public_flow_and_get_user",
 ]