feat: run pinned pipelines and publish outputs

README.md

@@ -2,9 +2,9 @@
 
 GovOPlaN Dataflow defines and runs governed tabular transformation pipelines.
 Power users can work with the same pipeline as a graphical node graph or as a
-constrained SQL query. Every saved definition change produces an immutable revision, and
+constrained SQL query. Every saved definition change produces an immutable
-every preview records diagnostics and reproducibility metadata without storing
+revision, and every preview records diagnostics and reproducibility metadata
-the previewed row contents.
+without storing the previewed row contents.
 
 ## Boundary
 
@@ -55,6 +55,14 @@ result-byte, graph-node, and response-row bounds. Saved previews record the
 pipeline revision, executor version, source fingerprints, node diagnostics,
 and output summary, but not source or result rows.
 
+Saved revisions can also be started through the versioned
+`dataflow.runLifecycle` capability or the Run dialog. The first runner is
+synchronous and bounded. It records an idempotent terminal run, pins the
+pipeline revision, and can atomically publish a complete result through
+`datasources.publication`. Publication is rejected when a source or result was
+truncated; larger asynchronous and artifact-backed execution belongs to a
+subsequent runner slice.
+
 ## Development
 
 ```bash

src/govoplan_dataflow/backend/db/models.py

@@ -73,6 +73,12 @@ class DataflowPipelineRevision(Base, TimestampMixin):
 class DataflowRun(Base, TimestampMixin):
     __tablename__ = "dataflow_runs"
     __table_args__ = (
+        UniqueConstraint(
+            "tenant_id",
+            "pipeline_id",
+            "idempotency_key",
+            name="uq_dataflow_run_idempotency",
+        ),
         Index("ix_dataflow_runs_tenant_status", "tenant_id", "status"),
         Index("ix_dataflow_runs_pipeline_created", "pipeline_id", "created_at"),
     )
@@ -93,11 +99,35 @@ class DataflowRun(Base, TimestampMixin):
     status: Mapped[str] = mapped_column(String(30), nullable=False, index=True)
     executor_version: Mapped[str] = mapped_column(String(40), nullable=False)
     definition_hash: Mapped[str] = mapped_column(String(64), nullable=False)
+    idempotency_key: Mapped[str | None] = mapped_column(
+        String(255),
+        nullable=True,
+        index=True,
+    )
+    request_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    request_: Mapped[dict[str, Any]] = mapped_column(
+        "request",
+        JSON,
+        default=dict,
+        nullable=False,
+    )
     source_fingerprints: Mapped[list[dict[str, Any]]] = mapped_column(JSON, default=list, nullable=False)
     result_schema: Mapped[list[dict[str, Any]]] = mapped_column(JSON, default=list, nullable=False)
     diagnostics: Mapped[list[dict[str, Any]]] = mapped_column(JSON, default=list, nullable=False)
     input_row_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
     output_row_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
+    output_publication_ref: Mapped[str | None] = mapped_column(
+        String(500),
+        nullable=True,
+    )
+    output_datasource_ref: Mapped[str | None] = mapped_column(
+        String(500),
+        nullable=True,
+    )
+    output_materialization_ref: Mapped[str | None] = mapped_column(
+        String(500),
+        nullable=True,
+    )
     started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
     finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
     error: Mapped[str | None] = mapped_column(Text)

src/govoplan_dataflow/backend/manifest.py

@@ -6,6 +6,7 @@ from govoplan_core.core.module_guards import (
     drop_table_retirement_provider,
     persistent_table_uninstall_guard,
 )
+from govoplan_core.core.dataflows import CAPABILITY_DATAFLOW_RUN_LIFECYCLE
 from govoplan_core.core.modules import (
     DocumentationTopic,
     FrontendModule,
@@ -21,6 +22,7 @@ from govoplan_core.core.modules import (
 from govoplan_core.core.datasources import (
     CAPABILITY_DATASOURCE_CATALOGUE,
     CAPABILITY_DATASOURCE_LIFECYCLE,
+    CAPABILITY_DATASOURCE_PUBLICATION,
 )
 from govoplan_core.db.base import Base
 from govoplan_dataflow.backend.db import models as dataflow_models
@@ -143,6 +145,12 @@ def _dataflow_router(context: ModuleContext):
     return router
 
 
+def _run_provider(context: ModuleContext):
+    from govoplan_dataflow.backend.service import SqlDataflowRunLifecycleProvider
+
+    return SqlDataflowRunLifecycleProvider(registry=context.registry)
+
+
 def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
     return {
         "dataflow_pipelines": (
@@ -180,6 +188,7 @@ manifest = ModuleManifest(
     optional_capabilities=(
         CAPABILITY_DATASOURCE_CATALOGUE,
         CAPABILITY_DATASOURCE_LIFECYCLE,
+        CAPABILITY_DATASOURCE_PUBLICATION,
     ),
     provides_interfaces=(
         ModuleInterfaceProvider(name="dataflow.pipeline_catalog", version=MODULE_VERSION),
@@ -200,6 +209,12 @@ manifest = ModuleManifest(
             version_max_exclusive="1.0.0",
             optional=True,
         ),
+        ModuleInterfaceRequirement(
+            name="datasources.publication",
+            version_min="0.1.0",
+            version_max_exclusive="1.0.0",
+            optional=True,
+        ),
     ),
     permissions=PERMISSIONS,
     role_templates=ROLE_TEMPLATES,
@@ -226,6 +241,9 @@ manifest = ModuleManifest(
         ),
     ),
     route_factory=_dataflow_router,
+    capability_factories={
+        CAPABILITY_DATAFLOW_RUN_LIFECYCLE: _run_provider,
+    },
     tenant_summary_providers=(_tenant_summary,),
     migration_spec=MigrationSpec(
         module_id=MODULE_ID,

src/govoplan_dataflow/backend/migrations/versions/b8e3c6a1f4d9_v0114_dataflow_runs.py

@@ -0,0 +1,83 @@
+"""v0.1.14 Dataflow run lifecycle
+
+Revision ID: b8e3c6a1f4d9
+Revises: d4f7a1c8e2b0
+Create Date: 2026-07-28 00:00:00.000000
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "b8e3c6a1f4d9"
+down_revision = "d4f7a1c8e2b0"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("dataflow_runs") as batch_op:
+        batch_op.add_column(
+            sa.Column("idempotency_key", sa.String(length=255), nullable=True)
+        )
+        batch_op.add_column(
+            sa.Column("request_hash", sa.String(length=64), nullable=True)
+        )
+        batch_op.add_column(
+            sa.Column(
+                "request",
+                sa.JSON(),
+                nullable=False,
+                server_default=sa.text("'{}'"),
+            )
+        )
+        batch_op.add_column(
+            sa.Column(
+                "output_publication_ref",
+                sa.String(length=500),
+                nullable=True,
+            )
+        )
+        batch_op.add_column(
+            sa.Column(
+                "output_datasource_ref",
+                sa.String(length=500),
+                nullable=True,
+            )
+        )
+        batch_op.add_column(
+            sa.Column(
+                "output_materialization_ref",
+                sa.String(length=500),
+                nullable=True,
+            )
+        )
+        batch_op.create_unique_constraint(
+            "uq_dataflow_run_idempotency",
+            ["tenant_id", "pipeline_id", "idempotency_key"],
+        )
+    op.create_index(
+        op.f("ix_dataflow_runs_idempotency_key"),
+        "dataflow_runs",
+        ["idempotency_key"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        op.f("ix_dataflow_runs_idempotency_key"),
+        table_name="dataflow_runs",
+    )
+    with op.batch_alter_table("dataflow_runs") as batch_op:
+        batch_op.drop_constraint(
+            "uq_dataflow_run_idempotency",
+            type_="unique",
+        )
+        batch_op.drop_column("output_materialization_ref")
+        batch_op.drop_column("output_datasource_ref")
+        batch_op.drop_column("output_publication_ref")
+        batch_op.drop_column("request")
+        batch_op.drop_column("request_hash")
+        batch_op.drop_column("idempotency_key")

src/govoplan_dataflow/backend/router.py

@@ -5,6 +5,10 @@ from sqlalchemy.orm import Session
 
 from govoplan_core.audit.logging import audit_event
 from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
+from govoplan_core.core.dataflows import (
+    DataflowPublicationTarget,
+    DataflowRunRequest,
+)
 from govoplan_core.core.datasources import (
     DatasourceAccessError,
     DatasourceDescriptor,
@@ -32,6 +36,9 @@ from govoplan_dataflow.backend.schemas import (
     PipelineListResponse,
     PipelinePreviewRequest,
     PipelinePreviewResponse,
+    PipelineRunCreateRequest,
+    PipelineRunListResponse,
+    PipelineRunResponse,
     PipelineResponse,
     PipelineSqlResponse,
     PipelineUpdateRequest,
@@ -48,12 +55,17 @@ from govoplan_dataflow.backend.service import (
     DataflowValidationError,
     compile_sql_draft,
     create_pipeline,
+    cancel_pipeline_run,
     delete_pipeline,
     get_pipeline,
+    get_pipeline_run,
+    list_pipeline_runs,
     list_pipelines,
     pipeline_response,
+    pipeline_run_response,
     preview_pipeline,
     render_graph_sql,
+    start_pipeline_run,
     update_pipeline,
     validate_draft,
 )
@@ -167,6 +179,7 @@ def _source_response(source: DatasourceDescriptor) -> TabularSourceResponse:
         source_name=source.source_name,
         name=source.name,
         description=source.description,
+        mode=source.mode,
         columns=[
             TabularSourceColumnResponse(
                 name=column.name,
@@ -421,6 +434,152 @@ def api_delete_pipeline(
     return PipelineDeleteResponse(deleted=True, pipeline_id=pipeline.id)
 
 
+@router.get(
+    "/pipelines/{pipeline_id}/runs",
+    response_model=PipelineRunListResponse,
+)
+def api_list_pipeline_runs(
+    pipeline_id: str,
+    limit: int = 50,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> PipelineRunListResponse:
+    _require_any_scope(principal, READ_SCOPE, RUN_SCOPE, ADMIN_SCOPE)
+    try:
+        runs = list_pipeline_runs(
+            session,
+            tenant_id=principal.tenant_id,
+            pipeline_id=pipeline_id,
+            limit=limit,
+        )
+        return PipelineRunListResponse(
+            runs=[pipeline_run_response(session, run) for run in runs]
+        )
+    except DataflowError as exc:
+        raise _http_error(exc) from exc
+
+
+@router.post(
+    "/pipelines/{pipeline_id}/runs",
+    response_model=PipelineRunResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+def api_start_pipeline_run(
+    pipeline_id: str,
+    payload: PipelineRunCreateRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> PipelineRunResponse:
+    _require_any_scope(principal, RUN_SCOPE, ADMIN_SCOPE)
+    try:
+        pipeline = get_pipeline(
+            session,
+            tenant_id=principal.tenant_id,
+            pipeline_id=pipeline_id,
+        )
+        publication = (
+            DataflowPublicationTarget(
+                target_datasource_ref=payload.publication.target_datasource_ref,
+                name=payload.publication.name,
+                source_name=payload.publication.source_name,
+                description=payload.publication.description,
+                freeze=payload.publication.freeze,
+                frozen_label=payload.publication.frozen_label,
+                set_current=payload.publication.set_current,
+                metadata=payload.publication.metadata,
+            )
+            if payload.publication
+            else None
+        )
+        run, replayed = start_pipeline_run(
+            session,
+            tenant_id=principal.tenant_id,
+            actor_id=_actor_id(principal),
+            principal=principal,
+            registry=get_registry(),
+            request=DataflowRunRequest(
+                pipeline_ref=f"pipeline:{pipeline.id}",
+                revision=payload.revision or pipeline.current_revision,
+                idempotency_key=payload.idempotency_key,
+                row_limit=payload.row_limit,
+                publication=publication,
+            ),
+        )
+    except DataflowError as exc:
+        raise _http_error(exc) from exc
+    audit_event(
+        session,
+        tenant_id=principal.tenant_id,
+        user_id=getattr(principal.user, "id", None),
+        api_key_id=principal.api_key_id,
+        action=(
+            "dataflow.run.replayed"
+            if replayed
+            else f"dataflow.run.{run.status}"
+        ),
+        object_type="dataflow_run",
+        object_id=run.id,
+        details={
+            "pipeline_id": run.pipeline_id,
+            "revision_id": run.pipeline_revision_id,
+            "status": run.status,
+            "output_datasource_ref": run.output_datasource_ref,
+            "output_materialization_ref": run.output_materialization_ref,
+        },
+    )
+    response = pipeline_run_response(session, run, replayed=replayed)
+    session.commit()
+    return response
+
+
+@router.get("/runs/{run_id}", response_model=PipelineRunResponse)
+def api_get_pipeline_run(
+    run_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> PipelineRunResponse:
+    _require_any_scope(principal, READ_SCOPE, RUN_SCOPE, ADMIN_SCOPE)
+    try:
+        run = get_pipeline_run(
+            session,
+            tenant_id=principal.tenant_id,
+            run_ref=f"dataflow-run:{run_id}",
+        )
+        return pipeline_run_response(session, run)
+    except DataflowError as exc:
+        raise _http_error(exc) from exc
+
+
+@router.post("/runs/{run_id}/cancel", response_model=PipelineRunResponse)
+def api_cancel_pipeline_run(
+    run_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> PipelineRunResponse:
+    _require_any_scope(principal, RUN_SCOPE, ADMIN_SCOPE)
+    try:
+        run = cancel_pipeline_run(
+            session,
+            tenant_id=principal.tenant_id,
+            run_ref=f"dataflow-run:{run_id}",
+        )
+    except DataflowError as exc:
+        raise _http_error(exc) from exc
+    audit_event(
+        session,
+        tenant_id=principal.tenant_id,
+        user_id=getattr(principal.user, "id", None),
+        api_key_id=principal.api_key_id,
+        action="dataflow.run.cancelled",
+        object_type="dataflow_run",
+        object_id=run.id,
+        details={"pipeline_id": run.pipeline_id},
+    )
+    response = pipeline_run_response(session, run)
+    session.commit()
+    return response
+
+
 @router.post("/validate", response_model=PipelineValidationResponse)
 def api_validate_pipeline(
     payload: PipelineDraftRequest,

src/govoplan_dataflow/backend/schemas.py

@@ -158,6 +158,68 @@ class PipelinePreviewResponse(BaseModel):
     executor_version: str
 
 
+class PipelineRunPublicationRequest(BaseModel):
+    target_datasource_ref: str | None = Field(default=None, max_length=500)
+    name: str | None = Field(default=None, min_length=1, max_length=300)
+    source_name: str | None = Field(
+        default=None,
+        min_length=1,
+        max_length=120,
+        pattern=r"^[A-Za-z_][A-Za-z0-9_]*$",
+    )
+    description: str | None = Field(default=None, max_length=4_000)
+    freeze: bool = False
+    frozen_label: str | None = Field(default=None, max_length=300)
+    set_current: bool = True
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+    @model_validator(mode="after")
+    def validate_target(self) -> "PipelineRunPublicationRequest":
+        if self.target_datasource_ref:
+            return self
+        if not self.name or not self.source_name:
+            raise ValueError(
+                "New publication targets require name and source_name."
+            )
+        return self
+
+
+class PipelineRunCreateRequest(BaseModel):
+    revision: int | None = Field(default=None, ge=1)
+    idempotency_key: str = Field(min_length=1, max_length=255)
+    row_limit: int = Field(default=500, ge=1, le=500)
+    publication: PipelineRunPublicationRequest | None = None
+
+
+class PipelineRunResponse(BaseModel):
+    ref: str
+    pipeline_id: str
+    revision: int
+    run_type: str
+    status: Literal["running", "succeeded", "failed", "cancelled"]
+    idempotency_key: str | None
+    definition_hash: str
+    executor_version: str
+    source_fingerprints: list[dict[str, Any]]
+    result_schema: list[PreviewColumn]
+    diagnostics: list[DataflowDiagnostic]
+    input_row_count: int
+    output_row_count: int
+    output_publication_ref: str | None
+    output_datasource_ref: str | None
+    output_materialization_ref: str | None
+    error: str | None
+    started_at: datetime
+    finished_at: datetime | None
+    created_by: str | None
+    created_at: datetime
+    replayed: bool = False
+
+
+class PipelineRunListResponse(BaseModel):
+    runs: list[PipelineRunResponse]
+
+
 class PipelineDeleteResponse(BaseModel):
     deleted: bool
     pipeline_id: str
@@ -210,6 +272,7 @@ class TabularSourceResponse(BaseModel):
     source_name: str
     name: str
     description: str | None
+    mode: Literal["live", "cached", "static"]
     columns: list[TabularSourceColumnResponse]
     schema_version: str
     fingerprint: str

src/govoplan_dataflow/backend/service.py

@@ -1,15 +1,26 @@
 from __future__ import annotations
 
+import hashlib
+import json
 from dataclasses import dataclass
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.dataflows import (
+    DataflowRunConflictError,
+    DataflowRunDescriptor,
+    DataflowRunError,
+    DataflowRunNotFoundError,
+    DataflowRunRequest,
+)
 from govoplan_core.core.datasources import (
     DatasourceError,
+    DatasourcePublicationRequest,
     DatasourceReadRequest,
     datasource_catalogue,
+    datasource_publication,
 )
 from govoplan_core.db.base import utcnow
 from govoplan_dataflow.backend.db.models import (
@@ -34,6 +45,7 @@ from govoplan_dataflow.backend.schemas import (
     PipelinePreviewResponse,
     PipelineResponse,
     PipelineRevisionResponse,
+    PipelineRunResponse,
     PipelineSqlResponse,
     PipelineUpdateRequest,
     PipelineValidationResponse,
@@ -45,15 +57,15 @@ from govoplan_dataflow.backend.sql_compiler import (
 )
 
 
-class DataflowError(RuntimeError):
+class DataflowError(DataflowRunError):
     pass
 
 
-class DataflowNotFoundError(DataflowError):
+class DataflowNotFoundError(DataflowError, DataflowRunNotFoundError):
     pass
 
 
-class DataflowConflictError(DataflowError):
+class DataflowConflictError(DataflowError, DataflowRunConflictError):
     pass
 
 
@@ -495,6 +507,363 @@ def preview_pipeline(
     )
 
 
+def list_pipeline_runs(
+    session: Session,
+    *,
+    tenant_id: str,
+    pipeline_id: str | None = None,
+    limit: int = 100,
+) -> list[DataflowRun]:
+    statement = (
+        select(DataflowRun)
+        .where(DataflowRun.tenant_id == tenant_id)
+        .order_by(DataflowRun.created_at.desc(), DataflowRun.id.desc())
+        .limit(max(1, min(int(limit), 100)))
+    )
+    if pipeline_id:
+        get_pipeline(
+            session,
+            tenant_id=tenant_id,
+            pipeline_id=pipeline_id,
+        )
+        statement = statement.where(DataflowRun.pipeline_id == pipeline_id)
+    return list(session.scalars(statement))
+
+
+def get_pipeline_run(
+    session: Session,
+    *,
+    tenant_id: str,
+    run_ref: str,
+) -> DataflowRun:
+    run_id = _strip_ref(run_ref, "dataflow-run:")
+    if not run_id:
+        raise DataflowNotFoundError("Dataflow run not found")
+    run = session.scalar(
+        select(DataflowRun).where(
+            DataflowRun.id == run_id,
+            DataflowRun.tenant_id == tenant_id,
+        )
+    )
+    if run is None:
+        raise DataflowNotFoundError("Dataflow run not found")
+    return run
+
+
+def start_pipeline_run(
+    session: Session,
+    *,
+    tenant_id: str,
+    actor_id: str | None,
+    principal: ApiPrincipal,
+    registry: object | None,
+    request: DataflowRunRequest,
+) -> tuple[DataflowRun, bool]:
+    pipeline_id = _strip_ref(request.pipeline_ref, "pipeline:")
+    if not pipeline_id:
+        raise DataflowNotFoundError("Dataflow pipeline not found")
+    pipeline = get_pipeline(
+        session,
+        tenant_id=tenant_id,
+        pipeline_id=pipeline_id,
+    )
+    revision = get_pipeline_revision(
+        session,
+        pipeline=pipeline,
+        revision=request.revision,
+    )
+    idempotency_key = request.idempotency_key.strip()
+    if not idempotency_key or len(idempotency_key) > 255:
+        raise DataflowConflictError(
+            "A Dataflow run idempotency key of at most 255 characters is required."
+        )
+    if request.row_limit < 1 or request.row_limit > 500:
+        raise DataflowConflictError(
+            "The bounded Dataflow runner supports between 1 and 500 output rows."
+        )
+    request_hash = _run_request_hash(request)
+    existing = session.scalar(
+        select(DataflowRun).where(
+            DataflowRun.tenant_id == tenant_id,
+            DataflowRun.pipeline_id == pipeline.id,
+            DataflowRun.idempotency_key == idempotency_key,
+        )
+    )
+    if existing is not None:
+        if existing.request_hash != request_hash:
+            raise DataflowConflictError(
+                "The Dataflow run idempotency key was already used with "
+                "different parameters."
+            )
+        return existing, True
+
+    graph = PipelineGraph.model_validate(revision.graph)
+    started_at = utcnow()
+    run = DataflowRun(
+        tenant_id=tenant_id,
+        pipeline_id=pipeline.id,
+        pipeline_revision_id=revision.id,
+        run_type="published" if request.publication else "run",
+        status="running",
+        executor_version=EXECUTOR_VERSION,
+        definition_hash=revision.content_hash,
+        idempotency_key=idempotency_key,
+        request_hash=request_hash,
+        request_=_run_request_payload(request),
+        source_fingerprints=[],
+        result_schema=[],
+        diagnostics=[],
+        input_row_count=0,
+        output_row_count=0,
+        started_at=started_at,
+        created_by=actor_id,
+    )
+    session.add(run)
+    session.flush()
+
+    try:
+        result = execute_preview(
+            graph,
+            row_limit=request.row_limit,
+            source_resolver=_datasource_source_resolver(
+                session=session,
+                principal=principal,
+                registry=registry,
+            ),
+        )
+        if request.publication and (
+            result.truncated
+            or any(
+                bool(item.get("truncated"))
+                for item in result.source_fingerprints
+            )
+        ):
+            raise PipelineExecutionError(
+                "The bounded runner cannot publish a truncated result or a "
+                "result calculated from truncated source data."
+            )
+
+        run.source_fingerprints = result.source_fingerprints
+        run.result_schema = [
+            item.model_dump(mode="json") for item in result.columns
+        ]
+        run.diagnostics = [
+            item.model_dump(mode="json") for item in result.diagnostics
+        ]
+        run.input_row_count = result.input_row_count
+        run.output_row_count = result.total_rows
+
+        if request.publication:
+            publisher = datasource_publication(registry)
+            if publisher is None:
+                raise PipelineExecutionError(
+                    "Publishing Dataflow output requires the Datasources "
+                    "publication capability."
+                )
+            target = request.publication
+            publication = publisher.publish_rows(
+                session,
+                principal,
+                request=DatasourcePublicationRequest(
+                    producer_module="dataflow",
+                    producer_run_ref=f"dataflow-run:{run.id}",
+                    idempotency_key=f"{pipeline.id}:{idempotency_key}",
+                    rows=tuple(dict(row) for row in result.rows),
+                    target_datasource_ref=target.target_datasource_ref,
+                    name=target.name or f"{pipeline.name} output",
+                    source_name=target.source_name,
+                    description=target.description,
+                    freeze=target.freeze,
+                    frozen_label=target.frozen_label,
+                    set_current=target.set_current,
+                    provenance={
+                        "pipeline_ref": f"pipeline:{pipeline.id}",
+                        "pipeline_revision": revision.revision,
+                        "definition_hash": revision.content_hash,
+                        "source_fingerprints": result.source_fingerprints,
+                    },
+                    metadata={
+                        **dict(target.metadata),
+                        "dataflow_run_ref": f"dataflow-run:{run.id}",
+                    },
+                ),
+            )
+            run.output_publication_ref = publication.ref
+            run.output_datasource_ref = publication.datasource.ref
+            run.output_materialization_ref = publication.materialization.ref
+        run.status = "succeeded"
+        run.finished_at = utcnow()
+        run.error = None
+    except (DatasourceError, PipelineExecutionError) as exc:
+        run.status = "failed"
+        run.finished_at = utcnow()
+        run.error = str(exc)
+        diagnostics = list(getattr(exc, "diagnostics", ()))
+        diagnostics.append(
+            DataflowDiagnostic(
+                severity="error",
+                code="run.execution",
+                message=str(exc),
+                node_id=getattr(exc, "node_id", None),
+            )
+        )
+        run.diagnostics = [
+            item.model_dump(mode="json") for item in diagnostics
+        ]
+        run.source_fingerprints = list(
+            getattr(exc, "source_fingerprints", ())
+        )
+        run.input_row_count = int(getattr(exc, "input_row_count", 0))
+    session.flush()
+    return run, False
+
+
+def cancel_pipeline_run(
+    session: Session,
+    *,
+    tenant_id: str,
+    run_ref: str,
+) -> DataflowRun:
+    run = get_pipeline_run(
+        session,
+        tenant_id=tenant_id,
+        run_ref=run_ref,
+    )
+    if run.status not in {"queued", "running"}:
+        raise DataflowConflictError(
+            f"Dataflow run is already {run.status} and cannot be cancelled."
+        )
+    run.status = "cancelled"
+    run.finished_at = utcnow()
+    run.error = "Cancelled by request."
+    session.flush()
+    return run
+
+
+def pipeline_run_response(
+    session: Session,
+    run: DataflowRun,
+    *,
+    replayed: bool = False,
+) -> PipelineRunResponse:
+    revision = session.get(DataflowPipelineRevision, run.pipeline_revision_id)
+    if revision is None:
+        raise DataflowNotFoundError("Dataflow pipeline revision not found")
+    return PipelineRunResponse(
+        ref=f"dataflow-run:{run.id}",
+        pipeline_id=run.pipeline_id,
+        revision=revision.revision,
+        run_type=run.run_type,
+        status=run.status,  # type: ignore[arg-type]
+        idempotency_key=run.idempotency_key,
+        definition_hash=run.definition_hash,
+        executor_version=run.executor_version,
+        source_fingerprints=list(run.source_fingerprints),
+        result_schema=list(run.result_schema),
+        diagnostics=list(run.diagnostics),
+        input_row_count=run.input_row_count,
+        output_row_count=run.output_row_count,
+        output_publication_ref=run.output_publication_ref,
+        output_datasource_ref=run.output_datasource_ref,
+        output_materialization_ref=run.output_materialization_ref,
+        error=run.error,
+        started_at=run.started_at,
+        finished_at=run.finished_at,
+        created_by=run.created_by,
+        created_at=run.created_at,
+        replayed=replayed,
+    )
+
+
+def pipeline_run_descriptor(
+    session: Session,
+    run: DataflowRun,
+    *,
+    replayed: bool = False,
+) -> DataflowRunDescriptor:
+    revision = session.get(DataflowPipelineRevision, run.pipeline_revision_id)
+    if revision is None:
+        raise DataflowNotFoundError("Dataflow pipeline revision not found")
+    return DataflowRunDescriptor(
+        ref=f"dataflow-run:{run.id}",
+        pipeline_ref=f"pipeline:{run.pipeline_id}",
+        revision=revision.revision,
+        status=run.status,
+        definition_hash=run.definition_hash,
+        executor_version=run.executor_version,
+        input_row_count=run.input_row_count,
+        output_row_count=run.output_row_count,
+        output_publication_ref=run.output_publication_ref,
+        output_datasource_ref=run.output_datasource_ref,
+        output_materialization_ref=run.output_materialization_ref,
+        error=run.error,
+        started_at=run.started_at,
+        finished_at=run.finished_at,
+        replayed=replayed,
+        metadata={
+            "run_type": run.run_type,
+            "source_fingerprints": list(run.source_fingerprints),
+            "diagnostics": list(run.diagnostics),
+        },
+    )
+
+
+class SqlDataflowRunLifecycleProvider:
+    def __init__(self, *, registry: object | None = None) -> None:
+        self._registry = registry
+
+    def start_run(
+        self,
+        session: object,
+        principal: object,
+        *,
+        request: DataflowRunRequest,
+    ) -> DataflowRunDescriptor:
+        db, api_principal = _run_context(session, principal)
+        run, replayed = start_pipeline_run(
+            db,
+            tenant_id=api_principal.tenant_id,
+            actor_id=_principal_actor_id(api_principal),
+            principal=api_principal,
+            registry=self._registry,
+            request=request,
+        )
+        return pipeline_run_descriptor(db, run, replayed=replayed)
+
+    def get_run(
+        self,
+        session: object,
+        principal: object,
+        *,
+        run_ref: str,
+    ) -> DataflowRunDescriptor | None:
+        db, api_principal = _run_context(session, principal)
+        try:
+            run = get_pipeline_run(
+                db,
+                tenant_id=api_principal.tenant_id,
+                run_ref=run_ref,
+            )
+        except DataflowNotFoundError:
+            return None
+        return pipeline_run_descriptor(db, run)
+
+    def cancel_run(
+        self,
+        session: object,
+        principal: object,
+        *,
+        run_ref: str,
+    ) -> DataflowRunDescriptor:
+        db, api_principal = _run_context(session, principal)
+        run = cancel_pipeline_run(
+            db,
+            tenant_id=api_principal.tenant_id,
+            run_ref=run_ref,
+        )
+        return pipeline_run_descriptor(db, run)
+
+
 def normalize_definition(
     *,
     graph: PipelineGraph,
@@ -541,6 +910,109 @@ def _source_nodes(
     return nodes
 
 
+def _datasource_source_resolver(
+    *,
+    session: Session,
+    principal: ApiPrincipal,
+    registry: object | None,
+):
+    provider = datasource_catalogue(registry)
+
+    def resolve_source(node: GraphNode, limit: int) -> ResolvedSource:
+        if provider is None:
+            raise PipelineExecutionError(
+                "Datasource-backed execution requires the Datasources "
+                "catalogue capability.",
+                node_id=node.id,
+            )
+        try:
+            resolved = provider.read_datasource(
+                session,
+                principal,
+                request=DatasourceReadRequest(
+                    datasource_ref=str(node.config["source_ref"]),
+                    consistency=str(
+                        node.config.get("consistency") or "current"
+                    ),  # type: ignore[arg-type]
+                    limit=limit,
+                    expected_fingerprint=_clean_optional(
+                        node.config.get("expected_fingerprint")
+                    ),
+                ),
+            )
+        except DatasourceError as exc:
+            raise PipelineExecutionError(str(exc), node_id=node.id) from exc
+        return ResolvedSource(
+            rows=tuple(dict(row) for row in resolved.rows),
+            source_ref=resolved.datasource.ref,
+            provider=resolved.datasource.provider or "datasources",
+            fingerprint=resolved.datasource.fingerprint,
+            total_rows=resolved.total_rows,
+            truncated=resolved.truncated,
+        )
+
+    return resolve_source
+
+
+def _run_context(
+    session: object,
+    principal: object,
+) -> tuple[Session, ApiPrincipal]:
+    if not isinstance(session, Session):
+        raise TypeError("Dataflow run providers require a SQLAlchemy session.")
+    if not isinstance(principal, ApiPrincipal):
+        raise DataflowConflictError("A tenant API principal is required.")
+    if not principal.tenant_id:
+        raise DataflowConflictError("A tenant API principal is required.")
+    return session, principal
+
+
+def _principal_actor_id(principal: ApiPrincipal) -> str | None:
+    return principal.account_id or principal.membership_id or principal.identity_id
+
+
+def _strip_ref(value: str, prefix: str) -> str | None:
+    cleaned = str(value or "").strip()
+    if not cleaned:
+        return None
+    if cleaned.startswith(prefix):
+        return cleaned[len(prefix) :]
+    return cleaned if ":" not in cleaned else None
+
+
+def _run_request_payload(request: DataflowRunRequest) -> dict[str, object]:
+    target = request.publication
+    return {
+        "pipeline_ref": request.pipeline_ref,
+        "revision": request.revision,
+        "row_limit": request.row_limit,
+        "publication": (
+            {
+                "target_datasource_ref": target.target_datasource_ref,
+                "name": target.name,
+                "source_name": target.source_name,
+                "description": target.description,
+                "freeze": target.freeze,
+                "frozen_label": target.frozen_label,
+                "set_current": target.set_current,
+                "metadata": dict(target.metadata),
+            }
+            if target
+            else None
+        ),
+    }
+
+
+def _run_request_hash(request: DataflowRunRequest) -> str:
+    encoded = json.dumps(
+        _run_request_payload(request),
+        sort_keys=True,
+        separators=(",", ":"),
+        default=str,
+    )
+    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+
+
 def _clean_optional(value: str | None) -> str | None:
     if value is None:
         return None
@@ -553,16 +1025,23 @@ __all__ = [
     "DataflowError",
     "DataflowNotFoundError",
     "DataflowValidationError",
+    "SqlDataflowRunLifecycleProvider",
+    "cancel_pipeline_run",
     "compile_sql_draft",
     "create_pipeline",
     "delete_pipeline",
     "get_pipeline",
     "get_pipeline_revision",
+    "get_pipeline_run",
+    "list_pipeline_runs",
     "list_pipelines",
     "normalize_definition",
     "pipeline_response",
+    "pipeline_run_descriptor",
+    "pipeline_run_response",
     "preview_pipeline",
     "render_graph_sql",
+    "start_pipeline_run",
     "update_pipeline",
     "validate_draft",
 ]

tests/test_manifest.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import unittest
 
+from govoplan_core.core.dataflows import CAPABILITY_DATAFLOW_RUN_LIFECYCLE
 from govoplan_core.core.modules import ModuleManifest
 from govoplan_dataflow.backend.manifest import get_manifest
 
@@ -19,6 +20,10 @@ class DataflowManifestTests(unittest.TestCase):
         self.assertEqual(manifest.frontend.package_name, "@govoplan/dataflow-webui")
         self.assertIsNotNone(manifest.route_factory)
         self.assertIsNotNone(manifest.migration_spec)
+        self.assertIn(
+            CAPABILITY_DATAFLOW_RUN_LIFECYCLE,
+            manifest.capability_factories,
+        )
         self.assertEqual(
             {
                 "dataflow.pipeline_catalog",

tests/test_migrations.py

@@ -24,7 +24,7 @@ class DataflowMigrationTests(unittest.TestCase):
             try:
                 with engine.connect() as connection:
                     self.assertIn(
-                        "d4f7a1c8e2b0",
+                        "b8e3c6a1f4d9",
                         set(MigrationContext.configure(connection).get_current_heads()),
                     )
                     self.assertEqual(

tests/test_service.py

@@ -5,6 +5,18 @@ import unittest
 from sqlalchemy import create_engine, func, select
 from sqlalchemy.orm import Session, sessionmaker
 
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.dataflows import (
+    DataflowPublicationTarget,
+    DataflowRunRequest,
+)
+from govoplan_core.core.datasources import (
+    CAPABILITY_DATASOURCE_PUBLICATION,
+    DatasourceDescriptor,
+    DatasourceMaterialization,
+    DatasourcePublicationResult,
+)
 from govoplan_core.db.base import Base
 from govoplan_dataflow.backend.db.models import (
     DataflowPipeline,
@@ -28,6 +40,7 @@ from govoplan_dataflow.backend.service import (
     get_pipeline,
     list_pipelines,
     preview_pipeline,
+    start_pipeline_run,
     update_pipeline,
 )
 
@@ -71,6 +84,61 @@ def sample_graph(*, minimum: int = 10) -> PipelineGraph:
     )
 
 
+def principal(tenant_id: str = "tenant-1") -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-1",
+            membership_id="membership-1",
+            tenant_id=tenant_id,
+            scopes=frozenset(),
+        ),
+        account=object(),
+        user=object(),
+    )
+
+
+class FakePublicationProvider:
+    def __init__(self) -> None:
+        self.requests = []
+
+    def publish_rows(self, _session, _principal, *, request):
+        self.requests.append(request)
+        descriptor = DatasourceDescriptor(
+            ref=request.target_datasource_ref or "datasource:output-1",
+            source_name=request.source_name or "existing_output",
+            name=request.name or "Existing output",
+            kind="custom",
+            mode="static",
+            shape="tabular",
+            fingerprint="published-fingerprint",
+        )
+        return DatasourcePublicationResult(
+            ref="publication:publication-1",
+            status="published",
+            datasource=descriptor,
+            materialization=DatasourceMaterialization(
+                ref="materialization:materialization-1",
+                datasource_ref=descriptor.ref,
+                revision=1,
+                state="published",
+                fingerprint=descriptor.fingerprint,
+            ),
+        )
+
+
+class FakeRegistry:
+    def __init__(self, publication_provider: FakePublicationProvider) -> None:
+        self.publication_provider = publication_provider
+
+    def has_capability(self, name: str) -> bool:
+        return name == CAPABILITY_DATASOURCE_PUBLICATION
+
+    def capability(self, name: str):
+        if not self.has_capability(name):
+            raise KeyError(name)
+        return self.publication_provider
+
+
 class DataflowServiceTests(unittest.TestCase):
     def setUp(self) -> None:
         self.engine = create_engine("sqlite:///:memory:")
@@ -263,6 +331,109 @@ class DataflowServiceTests(unittest.TestCase):
         self.assertEqual(run.source_fingerprints, response.source_fingerprints)
         self.assertEqual("preview.execution", run.diagnostics[-1]["code"])
 
+    def test_pinned_run_publishes_once_and_replays_idempotently(self) -> None:
+        pipeline = self._create()
+        publication_provider = FakePublicationProvider()
+        request = DataflowRunRequest(
+            pipeline_ref=f"pipeline:{pipeline.id}",
+            revision=1,
+            idempotency_key="monthly-2026-07",
+            publication=DataflowPublicationTarget(
+                name="Monthly result",
+                source_name="monthly_result",
+                freeze=True,
+                frozen_label="July 2026",
+            ),
+        )
+
+        first, first_replayed = start_pipeline_run(
+            self.session,
+            tenant_id="tenant-1",
+            actor_id="user-1",
+            principal=principal(),
+            registry=FakeRegistry(publication_provider),
+            request=request,
+        )
+        replay, replayed = start_pipeline_run(
+            self.session,
+            tenant_id="tenant-1",
+            actor_id="user-1",
+            principal=principal(),
+            registry=FakeRegistry(publication_provider),
+            request=request,
+        )
+        self.session.commit()
+
+        self.assertFalse(first_replayed)
+        self.assertTrue(replayed)
+        self.assertEqual(first.id, replay.id)
+        self.assertEqual("succeeded", first.status)
+        self.assertEqual("datasource:output-1", first.output_datasource_ref)
+        self.assertEqual(
+            "materialization:materialization-1",
+            first.output_materialization_ref,
+        )
+        self.assertEqual(
+            [{"id": 2, "amount": 15}, {"id": 3, "amount": 25}],
+            list(publication_provider.requests[0].rows),
+        )
+        self.assertEqual(1, len(publication_provider.requests))
+
+    def test_run_idempotency_key_rejects_changed_parameters(self) -> None:
+        pipeline = self._create()
+        publication_provider = FakePublicationProvider()
+        registry = FakeRegistry(publication_provider)
+        start_pipeline_run(
+            self.session,
+            tenant_id="tenant-1",
+            actor_id="user-1",
+            principal=principal(),
+            registry=registry,
+            request=DataflowRunRequest(
+                pipeline_ref=f"pipeline:{pipeline.id}",
+                revision=1,
+                idempotency_key="stable-key",
+                row_limit=100,
+            ),
+        )
+
+        with self.assertRaises(DataflowConflictError):
+            start_pipeline_run(
+                self.session,
+                tenant_id="tenant-1",
+                actor_id="user-1",
+                principal=principal(),
+                registry=registry,
+                request=DataflowRunRequest(
+                    pipeline_ref=f"pipeline:{pipeline.id}",
+                    revision=1,
+                    idempotency_key="stable-key",
+                    row_limit=10,
+                ),
+            )
+
+    def test_publication_without_datasources_finishes_as_failed_run(self) -> None:
+        pipeline = self._create()
+        run, _ = start_pipeline_run(
+            self.session,
+            tenant_id="tenant-1",
+            actor_id="user-1",
+            principal=principal(),
+            registry=None,
+            request=DataflowRunRequest(
+                pipeline_ref=f"pipeline:{pipeline.id}",
+                revision=1,
+                idempotency_key="missing-publisher",
+                publication=DataflowPublicationTarget(
+                    name="Output",
+                    source_name="output",
+                ),
+            ),
+        )
+
+        self.assertEqual("failed", run.status)
+        self.assertIn("Datasources publication capability", run.error)
+
 
 if __name__ == "__main__":
     unittest.main()

webui/src/api/dataflow.ts

@@ -146,6 +146,7 @@ export type TabularSource = {
   source_name: string;
   name: string;
   description?: string | null;
+  mode: "live" | "cached" | "static";
   columns: TabularSourceColumn[];
   schema_version: string;
   fingerprint: string;
@@ -161,6 +162,31 @@ export type TabularSourceCatalogue = {
   sources: TabularSource[];
 };
 
+export type PipelineRun = {
+  ref: string;
+  pipeline_id: string;
+  revision: number;
+  run_type: string;
+  status: "running" | "succeeded" | "failed" | "cancelled";
+  idempotency_key?: string | null;
+  definition_hash: string;
+  executor_version: string;
+  source_fingerprints: Record<string, unknown>[];
+  result_schema: PreviewColumn[];
+  diagnostics: DataflowDiagnostic[];
+  input_row_count: number;
+  output_row_count: number;
+  output_publication_ref?: string | null;
+  output_datasource_ref?: string | null;
+  output_materialization_ref?: string | null;
+  error?: string | null;
+  started_at: string;
+  finished_at?: string | null;
+  created_by?: string | null;
+  created_at: string;
+  replayed: boolean;
+};
+
 export async function listDataflowNodeTypes(settings: ApiSettings): Promise<NodeTypeDefinition[]> {
   const response = await apiFetch<{ nodes: NodeTypeDefinition[] }>(
     settings,
@@ -280,3 +306,43 @@ export function previewDataflowPipeline(
     body: JSON.stringify(payload)
   });
 }
+
+export async function listDataflowPipelineRuns(
+  settings: ApiSettings,
+  pipelineId: string
+): Promise<PipelineRun[]> {
+  const response = await apiFetch<{ runs: PipelineRun[] }>(
+    settings,
+    `/api/v1/dataflow/pipelines/${encodeURIComponent(pipelineId)}/runs?limit=25`
+  );
+  return response.runs;
+}
+
+export function runDataflowPipeline(
+  settings: ApiSettings,
+  pipelineId: string,
+  payload: {
+    revision: number;
+    idempotency_key: string;
+    row_limit?: number;
+    publication?: {
+      target_datasource_ref?: string | null;
+      name?: string | null;
+      source_name?: string | null;
+      description?: string | null;
+      freeze?: boolean;
+      frozen_label?: string | null;
+      set_current?: boolean;
+      metadata?: Record<string, unknown>;
+    } | null;
+  }
+): Promise<PipelineRun> {
+  return apiFetch(
+    settings,
+    `/api/v1/dataflow/pipelines/${encodeURIComponent(pipelineId)}/runs`,
+    {
+      method: "POST",
+      body: JSON.stringify(payload)
+    }
+  );
+}

webui/src/features/dataflow/DataflowPage.tsx

@@ -8,6 +8,7 @@ import {
 import {
   CheckCircle2,
   Code2,
+  DatabaseZap,
   Network,
   Play,
   Plus,
@@ -28,6 +29,7 @@ import {
   LoadingFrame,
   SegmentedControl,
   StatusBadge,
+  ToggleSwitch,
   hasScope,
   isApiError,
   useUnsavedChanges,
@@ -42,9 +44,11 @@ import {
   createDataflowSourceSnapshot,
   deleteDataflowPipeline,
   listDataflowNodeTypes,
+  listDataflowPipelineRuns,
   listDataflowPipelines,
   listDataflowSources,
   previewDataflowPipeline,
+  runDataflowPipeline,
   renderDataflowSql,
   updateDataflowPipeline,
   validateDataflowPipeline,
@@ -56,6 +60,7 @@ import {
   type Pipeline,
   type PipelineGraphNode,
   type PipelinePreview,
+  type PipelineRun,
   type TabularSource
 } from "../../api/dataflow";
 import DataflowCanvas, { updateGraphNode } from "./DataflowCanvas";
@@ -96,6 +101,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
   const [resultTab, setResultTab] = useState<ResultTab>("preview");
   const [deleteOpen, setDeleteOpen] = useState(false);
   const [snapshotOpen, setSnapshotOpen] = useState(false);
+  const [runOpen, setRunOpen] = useState(false);
   const [nodeLibrary, setNodeLibrary] = useState<NodeTypeDefinition[]>(FALLBACK_NODE_LIBRARY);
   const [sources, setSources] = useState<TabularSource[]>([]);
   const [sourceCatalogueAvailable, setSourceCatalogueAvailable] = useState(false);
@@ -533,6 +539,13 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
                   <Button variant="primary" onClick={() => void runPreview()} disabled={working || !canRun}>
                     <Play size={16} /> Preview
                   </Button>
+                  <Button
+                    onClick={() => setRunOpen(true)}
+                    disabled={working || !canRun || dirty || !draft.id || !draft.currentRevision}
+                    title={dirty ? "Save the pipeline before starting a pinned run." : undefined}
+                  >
+                    <DatabaseZap size={16} /> Run
+                  </Button>
                   <IconButton
                     label="Discard changes"
                     icon={<RotateCcw size={16} />}
@@ -724,10 +737,265 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
           setSelectedNodeId(node.id);
         }}
       />
+      <RunPipelineDialog
+        open={runOpen}
+        settings={settings}
+        pipeline={draft?.id && draft.currentRevision ? {
+          id: draft.id,
+          name: draft.name,
+          revision: draft.currentRevision
+        } : null}
+        sources={sources}
+        onClose={() => setRunOpen(false)}
+        onCompleted={(run) => {
+          if (run.status === "succeeded") {
+            setSuccess(
+              run.output_datasource_ref
+                ? `Run published ${run.output_row_count} rows to ${run.output_datasource_ref}.`
+                : `Run completed with ${run.output_row_count} output rows.`
+            );
+          } else {
+            setError(run.error || "Dataflow run failed.");
+          }
+          if (run.output_datasource_ref) {
+            void listDataflowSources(settings).then((catalogue) => {
+              setSources(catalogue.sources);
+              setSourceCatalogueAvailable(catalogue.available);
+              setSourceCatalogueWritable(catalogue.writable);
+            });
+          }
+        }}
+      />
     </main>
   );
 }
 
+type RunMode = "run" | "publish";
+
+function RunPipelineDialog({
+  open,
+  settings,
+  pipeline,
+  sources,
+  onClose,
+  onCompleted
+}: {
+  open: boolean;
+  settings: ApiSettings;
+  pipeline: { id: string; name: string; revision: number } | null;
+  sources: TabularSource[];
+  onClose: () => void;
+  onCompleted: (run: PipelineRun) => void;
+}) {
+  const [mode, setMode] = useState<RunMode>("run");
+  const [targetRef, setTargetRef] = useState("new");
+  const [name, setName] = useState("");
+  const [sourceName, setSourceName] = useState("");
+  const [description, setDescription] = useState("");
+  const [freeze, setFreeze] = useState(false);
+  const [frozenLabel, setFrozenLabel] = useState("");
+  const [idempotencyKey, setIdempotencyKey] = useState("");
+  const [runs, setRuns] = useState<PipelineRun[]>([]);
+  const [loadingRuns, setLoadingRuns] = useState(false);
+  const [busy, setBusy] = useState(false);
+  const [error, setError] = useState("");
+  const publicationTargets = sources.filter((source) =>
+    source.mode !== "live" && source.capabilities.includes("read")
+  );
+
+  useEffect(() => {
+    if (!open || !pipeline) return;
+    setMode("run");
+    setTargetRef("new");
+    setName(`${pipeline.name} output`);
+    setSourceName(sourceNameFromFile(`${pipeline.name}_output`));
+    setDescription("");
+    setFreeze(false);
+    setFrozenLabel("");
+    setIdempotencyKey(crypto.randomUUID());
+    setError("");
+    setLoadingRuns(true);
+    void listDataflowPipelineRuns(settings, pipeline.id)
+      .then(setRuns)
+      .catch((loadError) => setError(apiErrorMessage(loadError)))
+      .finally(() => setLoadingRuns(false));
+  }, [open, pipeline?.id, pipeline?.revision, settings]);
+
+  const start = async () => {
+    if (!pipeline) return;
+    setBusy(true);
+    setError("");
+    try {
+      const run = await runDataflowPipeline(settings, pipeline.id, {
+        revision: pipeline.revision,
+        idempotency_key: idempotencyKey,
+        row_limit: 500,
+        publication: mode === "publish" ? {
+          target_datasource_ref: targetRef === "new" ? null : targetRef,
+          name: targetRef === "new" ? name.trim() : null,
+          source_name: targetRef === "new" ? sourceName.trim() : null,
+          description: targetRef === "new" ? description.trim() || null : null,
+          freeze,
+          frozen_label: freeze ? frozenLabel.trim() || null : null,
+          set_current: true
+        } : null
+      });
+      setRuns((current) => [
+        run,
+        ...current.filter((item) => item.ref !== run.ref)
+      ]);
+      onCompleted(run);
+      setIdempotencyKey(crypto.randomUUID());
+      if (run.status !== "succeeded") {
+        setError(run.error || "Dataflow run failed.");
+      }
+    } catch (runError) {
+      setError(apiErrorMessage(runError));
+    } finally {
+      setBusy(false);
+    }
+  };
+
+  const targetComplete = mode === "run"
+    || targetRef !== "new"
+    || Boolean(name.trim() && sourceName.trim());
+
+  return (
+    <Dialog
+      open={open}
+      title={`Run ${pipeline?.name ?? "pipeline"}`}
+      className="dataflow-run-dialog"
+      onClose={() => {
+        if (!busy) onClose();
+      }}
+      footer={(
+        <>
+          <Button onClick={onClose} disabled={busy}>Close</Button>
+          <Button
+            variant="primary"
+            onClick={() => void start()}
+            disabled={busy || !pipeline || !targetComplete}
+          >
+            <Play size={16} /> {busy ? "Running..." : "Run revision"}
+          </Button>
+        </>
+      )}
+    >
+      <div className="dataflow-run-dialog-content">
+        {error ? (
+          <DismissibleAlert tone="danger" resetKey={error}>
+            {error}
+          </DismissibleAlert>
+        ) : null}
+        <div className="dataflow-run-controls">
+          <SegmentedControl<RunMode>
+            ariaLabel="Run output"
+            options={[
+              { id: "run", label: "Run only" },
+              { id: "publish", label: "Publish result" }
+            ]}
+            value={mode}
+            onChange={setMode}
+          />
+          <span className="dataflow-run-revision">
+            Revision {pipeline?.revision ?? "-"}
+          </span>
+        </div>
+        {mode === "publish" ? (
+          <div className="dataflow-run-publication-fields">
+            <FormField label="Target">
+              <select
+                value={targetRef}
+                onChange={(event) => setTargetRef(event.target.value)}
+              >
+                <option value="new">New datasource</option>
+                {publicationTargets.map((source) => (
+                  <option key={source.ref} value={source.ref}>
+                    {source.name} ({source.mode})
+                  </option>
+                ))}
+              </select>
+            </FormField>
+            {targetRef === "new" ? (
+              <>
+                <FormField label="Name">
+                  <input
+                    value={name}
+                    onChange={(event) => setName(event.target.value)}
+                  />
+                </FormField>
+                <FormField label="Logical source name">
+                  <input
+                    value={sourceName}
+                    onChange={(event) => setSourceName(event.target.value)}
+                    pattern="[A-Za-z_][A-Za-z0-9_]*"
+                  />
+                </FormField>
+                <FormField label="Description">
+                  <input
+                    value={description}
+                    onChange={(event) => setDescription(event.target.value)}
+                  />
+                </FormField>
+              </>
+            ) : null}
+            <div className="dataflow-run-freeze">
+              <ToggleSwitch
+                label="Freeze published state"
+                checked={freeze}
+                onChange={setFreeze}
+              />
+            </div>
+            {freeze ? (
+              <FormField label="Frozen-state label">
+                <input
+                  value={frozenLabel}
+                  onChange={(event) => setFrozenLabel(event.target.value)}
+                />
+              </FormField>
+            ) : null}
+          </div>
+        ) : null}
+        <section className="dataflow-run-history">
+          <div className="dataflow-run-history-heading">
+            <strong>Recent runs</strong>
+            <small>{loadingRuns ? "Loading..." : `${runs.length} shown`}</small>
+          </div>
+          <div>
+            <table>
+              <thead>
+                <tr>
+                  <th>Started</th>
+                  <th>Revision</th>
+                  <th>Status</th>
+                  <th>Rows</th>
+                  <th>Output</th>
+                </tr>
+              </thead>
+              <tbody>
+                {runs.map((run) => (
+                  <tr key={run.ref}>
+                    <td>{new Date(run.started_at).toLocaleString()}</td>
+                    <td>{run.revision}</td>
+                    <td><StatusBadge status={run.status} label={run.status} /></td>
+                    <td>{run.output_row_count.toLocaleString()}</td>
+                    <td title={run.output_materialization_ref ?? ""}>
+                      {run.output_datasource_ref ?? "Run evidence"}
+                    </td>
+                  </tr>
+                ))}
+                {!loadingRuns && !runs.length ? (
+                  <tr><td colSpan={5}>No runs yet</td></tr>
+                ) : null}
+              </tbody>
+            </table>
+          </div>
+        </section>
+      </div>
+    </Dialog>
+  );
+}
+
 function SourceSnapshotDialog({
   open,
   settings,

webui/src/styles/dataflow.css

@@ -601,6 +601,106 @@
   width: min(720px, calc(100vw - 32px));
 }
 
+.dataflow-run-dialog {
+  width: min(860px, calc(100vw - 32px));
+}
+
+.dataflow-run-dialog-content {
+  display: grid;
+  gap: 14px;
+  max-height: min(680px, calc(100vh - 210px));
+  overflow: hidden;
+}
+
+.dataflow-run-controls,
+.dataflow-run-history-heading {
+  display: flex;
+  align-items: center;
+  justify-content: space-between;
+  gap: 10px;
+}
+
+.dataflow-run-revision,
+.dataflow-run-history-heading small {
+  color: var(--muted);
+  font-size: 11px;
+}
+
+.dataflow-run-publication-fields {
+  display: grid;
+  grid-template-columns: repeat(2, minmax(0, 1fr));
+  gap: 12px;
+  padding: 12px 0;
+  border-top: var(--border-line);
+  border-bottom: var(--border-line);
+}
+
+.dataflow-run-publication-fields input,
+.dataflow-run-publication-fields select {
+  margin-top: 5px;
+  padding: 7px 8px;
+}
+
+.dataflow-run-freeze {
+  display: flex;
+  min-height: 54px;
+  align-items: end;
+  padding-bottom: 4px;
+}
+
+.dataflow-run-history {
+  display: flex;
+  min-height: 190px;
+  flex-direction: column;
+  overflow: hidden;
+}
+
+.dataflow-run-history-heading {
+  min-height: 36px;
+  flex: 0 0 auto;
+  border-bottom: var(--border-line);
+}
+
+.dataflow-run-history-heading strong {
+  color: var(--text-strong);
+  font-size: 12px;
+}
+
+.dataflow-run-history > div:last-child {
+  min-height: 0;
+  flex: 1 1 auto;
+  overflow: auto;
+}
+
+.dataflow-run-history table {
+  width: 100%;
+  border-collapse: collapse;
+  font-size: 11px;
+}
+
+.dataflow-run-history th,
+.dataflow-run-history td {
+  max-width: 230px;
+  border-bottom: var(--border-line);
+  padding: 7px 9px;
+  overflow: hidden;
+  text-align: left;
+  text-overflow: ellipsis;
+  white-space: nowrap;
+}
+
+.dataflow-run-history th {
+  position: sticky;
+  z-index: 1;
+  top: 0;
+  background: var(--panel-soft);
+  color: var(--text-strong);
+}
+
+.dataflow-run-history tbody tr:hover {
+  background: var(--primary-soft);
+}
+
 .dataflow-source-dialog-fields {
   display: grid;
   grid-template-columns: 1fr 1fr;
@@ -942,4 +1042,8 @@
   .dataflow-results {
     flex-basis: 220px;
   }
+
+  .dataflow-run-publication-fields {
+    grid-template-columns: 1fr;
+  }
 }