feat(jobs): add execution_signals table, ExecutionSignal model, SignalType enum

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/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, JobEvent
+from .jobs import ExecutionSignal, Job, JobEvent, SignalType
 from .knowledge_base import KnowledgeBaseRecord, KnowledgeBaseStatus
 from .memory_base import MemoryBase, MemoryBaseSession, MemoryBaseWorkflowRun, MessageIngestionRecord
 from .message import MessageTable
@@ -40,6 +40,7 @@ __all__ = [
     "CasbinRule",
     "Deployment",
     "DeploymentProviderAccount",
+    "ExecutionSignal",
     "File",
     "Flow",
     "FlowVersion",
@@ -58,6 +59,7 @@ __all__ = [
     "MessageTable",
     "SSOConfig",
     "SSOUserProfile",
+    "SignalType",
     "SpanTable",
     "TraceTable",
     "TransactionTable",

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

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

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

@@ -126,3 +126,48 @@ class JobEvent(SQLModel, table=True):  # type: ignore[call-arg]
         default_factory=lambda: datetime.now(timezone.utc),
         sa_column=Column(DateTime(timezone=True), nullable=False),
     )
+
+
+class SignalType(str, Enum):
+    """Cooperative control signals delivered to a running job.
+
+    Only STOP is implemented today; PAUSE/RESUME are intentionally left
+    out of the enum until they're wired so we don't ship dead values.
+    """
+
+    STOP = "stop"
+
+
+class ExecutionSignal(SQLModel, table=True):  # type: ignore[call-arg]
+    """A control signal row for a job.
+
+    The runner polls unconsumed rows at vertex boundaries and stamps
+    ``consumed_at`` once acted upon.
+    """
+
+    __tablename__ = "execution_signals"
+
+    id: UUID = Field(default_factory=uuid4, primary_key=True, index=True)
+    job_id: UUID = Field(index=True, nullable=False)
+    signal_type: SignalType = Field(
+        sa_column=Column(
+            SQLEnum(
+                SignalType,
+                name="execution_signal_type_enum",
+                values_callable=lambda obj: [item.value for item in obj],
+            ),
+            nullable=False,
+        ),
+    )
+    data: 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),
+    )
+    consumed_at: datetime | None = Field(
+        default=None,
+        sa_column=Column(DateTime(timezone=True), nullable=True),
+    )

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

@@ -44,3 +44,26 @@ def test_job_events_table_and_unique_seq(db_url):  # noqa: F811
             assert any(set(u["column_names"]) == {"job_id", "seq"} for u in unique)
     finally:
         engine.dispose()
+
+
+def test_execution_signals_table(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 "execution_signals" in inspector.get_table_names()
+            columns = {c["name"] for c in inspector.get_columns("execution_signals")}
+            assert {"id", "job_id", "signal_type", "data", "created_at", "consumed_at"} <= columns
+            indexes = inspector.get_indexes("execution_signals")
+            assert any(idx["column_names"] == ["job_id"] for idx in indexes)
+    finally:
+        engine.dispose()
+
+
+def test_signal_type_enum_has_stop():
+    from langflow.services.database.models.jobs.model import SignalType
+
+    assert SignalType.STOP.value == "stop"