diff --git a/src/backend/base/langflow/alembic/versions/8ce44e4858c6_add_execution_signals_table.py b/src/backend/base/langflow/alembic/versions/8ce44e4858c6_add_execution_signals_table.py new file mode 100644 index 0000000000..ede4bb7eca --- /dev/null +++ b/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) diff --git a/src/backend/base/langflow/services/database/models/__init__.py b/src/backend/base/langflow/services/database/models/__init__.py index 63f0f5c875..a5745c7c64 100644 --- a/src/backend/base/langflow/services/database/models/__init__.py +++ b/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", diff --git a/src/backend/base/langflow/services/database/models/jobs/__init__.py b/src/backend/base/langflow/services/database/models/jobs/__init__.py index 546eea0fe9..0a63b05e8b 100644 --- a/src/backend/base/langflow/services/database/models/jobs/__init__.py +++ b/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"] 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 72d690a549..3903ea7421 100644 --- a/src/backend/base/langflow/services/database/models/jobs/model.py +++ b/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), + ) diff --git a/src/backend/tests/unit/alembic/test_background_execution_migrations.py b/src/backend/tests/unit/alembic/test_background_execution_migrations.py index b2b4ea09a0..27ecf7066f 100644 --- a/src/backend/tests/unit/alembic/test_background_execution_migrations.py +++ b/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"