langflow-ai
diff --git a/‎.github/workflows/migration-validation.yml‎
Lines changed: 59 additions & 2 deletions b/‎.github/workflows/migration-validation.yml‎
Lines changed: 59 additions & 2 deletions
diff --git a/‎.secrets.baseline‎
Lines changed: 2 additions & 20 deletions b/‎.secrets.baseline‎
Lines changed: 2 additions & 20 deletions
diff --git a/‎Makefile‎
Lines changed: 7 additions & 0 deletions b/‎Makefile‎
Lines changed: 7 additions & 0 deletions
diff --git a/‎pyproject.toml‎
Lines changed: 2 additions & 1 deletion b/‎pyproject.toml‎
Lines changed: 2 additions & 1 deletion
diff --git a/‎src/backend/base/langflow/alembic/versions/185482a2d715_add_result_error_to_job.py‎
Lines changed: 44 additions & 0 deletions b/‎src/backend/base/langflow/alembic/versions/185482a2d715_add_result_error_to_job.py‎
Lines changed: 44 additions & 0 deletions
diff --git a/‎src/backend/base/langflow/alembic/versions/8ce44e4858c6_add_execution_signals_table.py‎
Lines changed: 58 additions & 0 deletions b/‎src/backend/base/langflow/alembic/versions/8ce44e4858c6_add_execution_signals_table.py‎
Lines changed: 58 additions & 0 deletions
diff --git a/‎src/backend/base/langflow/alembic/versions/9d5e24d777bf_merge_mcp_and_execution_signals_heads.py‎
Lines changed: 24 additions & 0 deletions b/‎src/backend/base/langflow/alembic/versions/9d5e24d777bf_merge_mcp_and_execution_signals_heads.py‎
Lines changed: 24 additions & 0 deletions
diff --git a/‎src/backend/base/langflow/alembic/versions/b026885b89c8_add_job_events_table.py‎
Lines changed: 51 additions & 0 deletions b/‎src/backend/base/langflow/alembic/versions/b026885b89c8_add_job_events_table.py‎
Lines changed: 51 additions & 0 deletions
@@ -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
 
@@ -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
       }
     ],
 
@@ -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 && \
 
@@ -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"
 
@@ -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")
@@ -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)
@@ -0,0 +1,24 @@
+"""merge_mcp_and_execution_signals_heads
+
+Revision ID: 9d5e24d777bf
+Revises: 247308ce2598, 8ce44e4858c6
+Create Date: 2026-07-10 12:00:00.000000
+
+Phase: EXPAND
+"""
+
+from collections.abc import Sequence
+
+# revision identifiers, used by Alembic.
+revision: str = "9d5e24d777bf"  # pragma: allowlist secret
+down_revision: str | Sequence[str] | None = ("247308ce2598", "8ce44e4858c6")  # pragma: allowlist secret
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass
@@ -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")