From 27945c9ffc36317a3ebecfcbffe7eb3dcc3e3ce4 Mon Sep 17 00:00:00 2001
From: ogabrielluiz <gabriel@langflow.org>
Date: Wed, 3 Jun 2026 23:07:24 -0300
Subject: [PATCH] feat(jobs): add durable result and error columns to job table

---
 .../185482a2d715_add_result_error_to_job.py   | 44 +++++++++++++++++++
 .../services/database/models/jobs/model.py    | 13 ++++++
 .../test_background_execution_migrations.py   | 27 ++++++++++++
 3 files changed, 84 insertions(+)
 create mode 100644 src/backend/base/langflow/alembic/versions/185482a2d715_add_result_error_to_job.py
 create mode 100644 src/backend/tests/unit/alembic/test_background_execution_migrations.py

diff --git a/src/backend/base/langflow/alembic/versions/185482a2d715_add_result_error_to_job.py b/src/backend/base/langflow/alembic/versions/185482a2d715_add_result_error_to_job.py
new file mode 100644
index 0000000000..798936984b
--- /dev/null
+++ b/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")
diff --git a/src/backend/base/langflow/services/database/models/jobs/model.py b/src/backend/base/langflow/services/database/models/jobs/model.py
index e38c93d61b..86ce5332f1 100644
--- a/src/backend/base/langflow/services/database/models/jobs/model.py
+++ b/src/backend/base/langflow/services/database/models/jobs/model.py
@@ -85,6 +85,19 @@ class JobBase(SQLModel):
         sa_column=Column(JsonVariant, nullable=True),
     )
 
+    # Durable terminal payloads for background workflow runs. result holds
+    # the final output blob on COMPLETED; error holds {type, message, ...}
+    # on FAILED/TIMED_OUT. Both nullable so non-workflow jobs and in-flight
+    # rows stay readable.
+    result: dict[str, Any] | None = Field(
+        default=None,
+        sa_column=Column(JsonVariant, nullable=True),
+    )
+    error: dict[str, Any] | None = Field(
+        default=None,
+        sa_column=Column(JsonVariant, nullable=True),
+    )
+
 
 class Job(JobBase, table=True):  # type: ignore[call-arg]
     __tablename__ = "job"
diff --git a/src/backend/tests/unit/alembic/test_background_execution_migrations.py b/src/backend/tests/unit/alembic/test_background_execution_migrations.py
new file mode 100644
index 0000000000..03ab8192dc
--- /dev/null
+++ b/src/backend/tests/unit/alembic/test_background_execution_migrations.py
@@ -0,0 +1,27 @@
+"""Structure tests for the background-execution migrations.
+
+Covers job.result/error columns, the job_events table and execution_signals
+table. Runs on sqlite and (when configured) postgres.
+"""
+
+from __future__ import annotations
+
+from alembic import command
+from sqlalchemy import create_engine, inspect
+
+from .test_migration_execution import _engine_url, _make_alembic_cfg, db_url  # noqa: F401
+
+
+def test_job_has_result_and_error_columns(db_url):  # noqa: F811
+    alembic_cfg = _make_alembic_cfg(db_url)
+    command.upgrade(alembic_cfg, "head")
+
+    engine = create_engine(_engine_url(db_url))
+    try:
+        with engine.connect() as connection:
+            columns = {c["name"] for c in inspect(connection).get_columns("job")}
+    finally:
+        engine.dispose()
+
+    assert "result" in columns
+    assert "error" in columns