feat(api/v2): durable + distributed background execution service for workflows

.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/__main__.py

@@ -833,6 +833,65 @@ def print_banner(host: str, port: int, protocol: str) -> None:
             logger.info(f"Open Langflow: {protocol}://{access_host}:{port}")
 
 
+@app.command()
+def worker(
+    log_level: str | None = typer.Option(None, help="Logging level."),
+    env_file: Path | None = typer.Option(None, help="Path to the .env file."),
+    idle_block_ms: int = typer.Option(1000, help="Claim blocking-pop window in milliseconds."),
+) -> None:
+    """Run a Langflow background worker that drains the redis job-claim queue.
+
+    Requires LANGFLOW_JOB_QUEUE_TYPE=redis. Each worker claims queued jobs, runs
+    them through the background JobRunner (publishing live frames to redis
+    Streams so any API replica can reattach), and releases the lease. Run as many
+    worker processes as you need horizontal capacity for.
+    """
+    if env_file:
+        load_dotenv(env_file, override=True)
+    if log_level:
+        configure(log_level=log_level)
+
+    async def _run_worker() -> None:
+        from langflow.services.background_execution.worker import build_worker, run_worker_loop
+
+        await initialize_services()
+        settings = get_settings_service().settings
+        if not settings.background_backend_is_scaled:
+            typer.echo("LANGFLOW_JOB_QUEUE_TYPE must be 'redis' to run a worker.")
+            raise typer.Exit(code=1)
+
+        from uuid import uuid4
+
+        from langflow.services.deps import get_job_service
+
+        # Process-unique owner so this worker's heartbeats are attributable, and
+        # a periodic watchdog so a dead worker's in-flight job is reaped under a
+        # steady fleet WITHOUT requiring a restart.
+        owner = f"worker:{os.getpid()}:{uuid4().hex[:8]}"
+        backend, worker_runner, teardown = await build_worker(owner=owner)
+        stop_event = asyncio.Event()
+        loop = asyncio.get_running_loop()
+        for sig in (signal.SIGTERM, signal.SIGINT):
+            with suppress(NotImplementedError):
+                loop.add_signal_handler(sig, stop_event.set)
+        logger.info("Langflow worker started; draining the redis job-claim queue.")
+        try:
+            await run_worker_loop(
+                backend,
+                worker_runner,
+                stop_event=stop_event,
+                idle_block_ms=idle_block_ms,
+                job_service=get_job_service(),
+                owner=owner,
+                lease_ttl_s=settings.background_lease_ttl_s,
+                watchdog_interval_s=settings.background_watchdog_interval_s,
+            )
+        finally:
+            await teardown()
+
+    asyncio.run(_run_worker())
+
+
 @app.command()
 def superuser(
     username: str = typer.Option(

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/v2/adapters/__init__.py

@@ -76,6 +76,16 @@ def terminal_error_type(self) -> str | None:
         via event type; the buffer task falls back to other signals.
         """
 
+    def is_durable(self, event_type: str) -> bool:
+        """True when a frame of ``event_type`` must be persisted to the durable log.
+
+        Durable frames (milestones) are appended to ``job_events`` so a
+        reattaching client can rebuild state after the live bus is gone.
+        Ephemeral frames (token deltas) are published to the live bus only.
+        Unknown types default to ephemeral — a new milestone must opt in
+        explicitly rather than silently bloat the durable log.
+        """
+
 
 StreamAdapterFactory = Callable[[StreamAdapterContext], StreamAdapter]