feat: Job execution status endpoint, status tracking, DB models (#11438)

* Features: job status table with related DB model, migration files, job status tracking.

Added status wrapper to handle job status updates in the DB.

Added filtration using job_id to fetch job status responses.

fix: Generate job response from vertex builds history by job id.  (#11457)

* feat: reconstruct workflow execution response from vertex_build by job_id

* [autofix.ci] apply automated fixes

* fix: use correct attribute name 'id' instead of 'vertex_id' in VertexBuildTable

* Updated the GET endpoint to return WorkflowExecutionResponse

---------

Co-authored-by: Janardan S Kavia <janardanskavia@Janardans-MacBook-Pro.local>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

Added /stop endpoint, task handling, job_id typing through job_service, job_status updates.

Consolidated migration files into one single migration.

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* [autofix.ci] apply automated fixes (attempt 3/3)

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

Author: dkaushik94

.secrets.baseline

@@ -917,15 +917,15 @@
         "filename": "src/backend/tests/conftest.py",
         "hashed_secret": "8bb6118f8fd6935ad0876a3be34a717d32708ffd",
         "is_verified": false,
-        "line_number": 427,
+        "line_number": 481,
         "is_secret": false
       },
       {
         "type": "Secret Keyword",
         "filename": "src/backend/tests/conftest.py",
         "hashed_secret": "61fbb5a12cd7b1f1fe1624120089efc0cd299e43",
         "is_verified": false,
-        "line_number": 639,
+        "line_number": 691,
         "is_secret": false
       }
     ],
@@ -1528,5 +1528,5 @@
       }
     ]
   },
-  "generated_at": "2026-01-21T13:04:28Z"
+  "generated_at": "2026-01-28T21:22:05Z"
 }

src/backend/base/langflow/alembic/migration_validator.py

@@ -351,7 +351,9 @@ def main():
     logging.basicConfig(level=logging.INFO)
     logger = logging.getLogger("migration_validator")
     if args.json:
-        logger.info(json.dumps(results, indent=2))
+        import sys as _sys
+
+        _sys.stdout.write(json.dumps(results, indent=2) + "\n")
     else:
         for result in results:
             logger.info("\n%s", "=" * 60)

src/backend/base/langflow/alembic/versions/369268b9af8b_add_job_id_to_vertex_build_create_job_.py

@@ -0,0 +1,74 @@
+"""add job_id to vertex_build, create job status table.
+
+Revision ID: 369268b9af8b
+Revises: 182e5471b900
+Create Date: 2026-01-28 13:00:52.967282
+
+Phase: EXPAND
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "369268b9af8b"  # pragma: allowlist secret
+down_revision: str | None = "182e5471b900"  # pragma: allowlist secret
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    from langflow.utils import migration
+
+    conn = op.get_bind()
+    if not migration.table_exists("job", conn):
+        op.create_table(
+            "job",
+            sa.Column("job_id", sa.Uuid(), nullable=False),
+            sa.Column("flow_id", sa.Uuid(), nullable=False),
+            sa.Column(
+                "status",
+                sa.Enum(
+                    "queued",
+                    "in_progress",
+                    "completed",
+                    "failed",
+                    "cancelled",
+                    "timed_out",
+                    name="job_status_enum",
+                ),
+                nullable=False,
+            ),
+            sa.Column("created_timestamp", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("finished_timestamp", sa.DateTime(timezone=True), nullable=True),
+            sa.PrimaryKeyConstraint("job_id"),
+        )
+        with op.batch_alter_table("job", schema=None) as batch_op:
+            batch_op.create_index(batch_op.f("ix_job_flow_id"), ["flow_id"], unique=False)
+            batch_op.create_index(batch_op.f("ix_job_job_id"), ["job_id"], unique=False)
+            batch_op.create_index(batch_op.f("ix_job_status"), ["status"], unique=False)
+
+    if not migration.column_exists("vertex_build", "job_id", conn):
+        with op.batch_alter_table("vertex_build", schema=None) as batch_op:
+            batch_op.add_column(sa.Column("job_id", sa.Uuid(), nullable=True))
+            batch_op.create_index(batch_op.f("ix_vertex_build_job_id"), ["job_id"], unique=False)
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("vertex_build", schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f("ix_vertex_build_job_id"))
+        batch_op.drop_column("job_id")
+
+    with op.batch_alter_table("job", schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f("ix_job_status"))
+        batch_op.drop_index(batch_op.f("ix_job_job_id"))
+        batch_op.drop_index(batch_op.f("ix_job_flow_id"))
+
+    op.drop_table("job")
+    # ### end Alembic commands ###

src/backend/base/langflow/api/v2/converters.py

@@ -28,6 +28,7 @@
 from lfx.schema.workflow import (
     ComponentOutput,
     ErrorDetail,
+    JobId,
     JobStatus,
     WorkflowExecutionRequest,
     WorkflowExecutionResponse,
@@ -482,7 +483,7 @@ def create_job_response(job_id: str, flow_id: str) -> WorkflowJobResponse:
 
 def create_error_response(
     flow_id: str,
-    job_id: str | None,
+    job_id: JobId,
     workflow_request: WorkflowExecutionRequest,
     error: Exception,
 ) -> WorkflowExecutionResponse: