feat(jobs): add job_events table and JobEvent model with unique (job_id, seq)

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/services/database/models/__init__.py

@@ -19,7 +19,7 @@ from .flow_version import FlowVersion
 from .flow_version_deployment_attachment import FlowVersionDeploymentAttachment
 from .folder import Folder
 from .ingestion_run import IngestionRun, IngestionRunStatus
-from .jobs import Job
+from .jobs import Job, JobEvent
 from .knowledge_base import KnowledgeBaseRecord, KnowledgeBaseStatus
 from .memory_base import MemoryBase, MemoryBaseSession, MemoryBaseWorkflowRun, MessageIngestionRecord
 from .message import MessageTable
@@ -48,6 +48,7 @@ __all__ = [
     "IngestionRun",
     "IngestionRunStatus",
     "Job",
+    "JobEvent",
     "KnowledgeBaseRecord",
     "KnowledgeBaseStatus",
     "MemoryBase",

src/backend/base/langflow/services/database/models/jobs/__init__.py

@@ -1,3 +1,3 @@
-from .model import Job
+from .model import Job, JobEvent
 
-__all__ = ["Job"]
+__all__ = ["Job", "JobEvent"]

src/backend/base/langflow/services/database/models/jobs/model.py

@@ -1,9 +1,9 @@
 from datetime import datetime, timezone
 from enum import Enum
 from typing import Any
-from uuid import UUID
+from uuid import UUID, uuid4
 
-from sqlalchemy import JSON, Column, DateTime
+from sqlalchemy import JSON, Column, DateTime, Integer, String, UniqueConstraint
 from sqlalchemy import Enum as SQLEnum
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlmodel import Field, SQLModel
@@ -101,3 +101,28 @@ class JobBase(SQLModel):
 
 class Job(JobBase, table=True):  # type: ignore[call-arg]
     __tablename__ = "job"
+
+
+class JobEvent(SQLModel, table=True):  # type: ignore[call-arg]
+    """Durable event log for a background job.
+
+    ``seq`` is the per-job monotonic cursor used as the SSE Last-Event-ID.
+    UNIQUE(job_id, seq) enforces gap-free ordering and lets append_event
+    detect collisions.
+    """
+
+    __tablename__ = "job_events"
+    __table_args__ = (UniqueConstraint("job_id", "seq", name="uq_job_events_job_id_seq"),)
+
+    id: UUID = Field(default_factory=uuid4, primary_key=True, index=True)
+    job_id: UUID = Field(index=True, nullable=False)
+    seq: int = Field(sa_column=Column(Integer, nullable=False))
+    event_type: str = Field(sa_column=Column(String, nullable=False))
+    payload: dict[str, Any] | None = Field(
+        default=None,
+        sa_column=Column(JsonVariant, nullable=True),
+    )
+    created_at: datetime = Field(
+        default_factory=lambda: datetime.now(timezone.utc),
+        sa_column=Column(DateTime(timezone=True), nullable=False),
+    )

src/backend/tests/unit/alembic/test_background_execution_migrations.py

@@ -25,3 +25,22 @@ def test_job_has_result_and_error_columns(db_url):  # noqa: F811
 
     assert "result" in columns
     assert "error" in columns
+
+
+def test_job_events_table_and_unique_seq(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:
+            inspector = inspect(connection)
+            assert "job_events" in inspector.get_table_names()
+            columns = {c["name"] for c in inspector.get_columns("job_events")}
+            assert {"id", "job_id", "seq", "event_type", "payload", "created_at"} <= columns
+            indexes = inspector.get_indexes("job_events")
+            assert any(idx["column_names"] == ["job_id"] for idx in indexes)
+            unique = inspector.get_unique_constraints("job_events")
+            assert any(set(u["column_names"]) == {"job_id", "seq"} for u in unique)
+    finally:
+        engine.dispose()