feat: internalize DataFlow cleaning pipelines

.github/workflows/ci.yml

@@ -72,8 +72,48 @@ jobs:
       # TODO(ratchet): once real coverage is measured for a few weeks, raise
       # this floor incrementally (e.g. 70 -> 75 -> 80) instead of jumping
       # straight to the pyproject target.
-      - name: Unit tests (with coverage report)
-        run: python -m pytest tests/unit -m "not live" --cov=backend --cov-report=term-missing --cov-fail-under=0
+      # tests/compat/dataflow pins the golden DataFlow contract and
+      # tests/integration exercises the workflow HTTP surface; both must run
+      # here or the pinned-compatibility guarantees have no regression gate.
+      # The upstream oracle stays opt-in (DATAFLOW_RUN_UPSTREAM_ORACLE=1) and
+      # self-skips in this job.
+      - name: Backend tests (with coverage report)
+        run: python -m pytest tests/unit tests/compat tests/integration -m "not live" --cov=backend --cov-report=term-missing --cov-fail-under=0
+
+  frontend-workflow-checks:
+    runs-on: ubuntu-latest
+    name: Frontend Workflow Checks
+    defaults:
+      run:
+        working-directory: frontend
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v6
+
+      # frontend/pnpm-workspace.yaml is a pnpm 10 config store (allowBuilds,
+      # overrides, no `packages` field) — pnpm 9 rejects it outright.
+      - name: Setup pnpm
+        uses: pnpm/action-setup@v4
+        with:
+          version: 10
+
+      # test-data-operator-nodes.mjs imports the real .ts sources via
+      # node:module registerHooks + stripTypeScriptTypes (needs Node >= 22.15).
+      - name: Setup Node
+        uses: actions/setup-node@v6
+        with:
+          node-version: "22"
+          cache: pnpm
+          cache-dependency-path: frontend/pnpm-lock.yaml
+
+      - name: Install dependencies
+        run: pnpm install --frozen-lockfile
+
+      - name: Data operator Canvas projection
+        run: pnpm run test:data-operator-nodes
+
+      - name: Workflow contracts
+        run: pnpm run test:workflow-contracts
 
   migrations:
     runs-on: ubuntu-latest

backend/api/v1/workflows.py

@@ -169,9 +169,12 @@ async def draft_demand_workflow(
 async def import_external_runtime_workflow(
     body: workflow_schemas.WorkflowExternalImportRequest,
 ) -> ApiResponse[workflow_schemas.WorkflowPatchResponse]:
-    """Import LangGraph/LangChain graphs as OpenCLI Admin native nodes."""
+    """Import supported external graphs as reviewable OpenCLI Admin native nodes."""
 
-    return ApiResponse.ok(import_external_workflow(body))
+    try:
+        return ApiResponse.ok(import_external_workflow(body))
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
 
 
 @router.post(

backend/schemas/workflow.py

@@ -126,7 +126,7 @@ class WorkflowParameterInterfaceField(BaseModel):
     id: str = Field(..., min_length=1)
     label: str = Field(..., min_length=1)
     groupId: str = Field(..., min_length=1)
-    type: Literal["text", "textarea", "number", "slider", "select", "boolean", "tokens"]
+    type: Literal["text", "textarea", "json", "number", "slider", "select", "boolean", "tokens"]
     binding: WorkflowParameterBinding
     description: Optional[str] = None
     order: Optional[float] = None
@@ -272,7 +272,7 @@ class WorkflowDemandDraftRequest(BaseModel):
     locale: Optional[str] = None
 
 
-ExternalWorkflowRuntime = Literal["langgraph", "langchain"]
+ExternalWorkflowRuntime = Literal["langgraph", "langchain", "dataflow"]
 
 
 class WorkflowExternalImportRequest(BaseModel):

backend/workflow/capability_projection.py

@@ -17,11 +17,13 @@
     WorkflowNodeKind,
     WorkflowRuntimeCapability,
 )
+from backend.workflow.data_operators import list_data_operator_specs
 from backend.workflow.node_registry import WORKFLOW_PRIMITIVE_IDS
 from backend.workflow.opencli_adapter_nodes import get_opencli_adapter_node_summary
 from backend.workflow.runtime_contracts import runtime_io_contract_manifest
 from backend.workflow.runtime_registry import (
     COLLECTION_OUTPUT_BINDING_ID,
+    DATA_OPERATOR_CATALOG_BINDINGS,
     DEMAND_DRAFT_BINDING_ID,
     EXTERNAL_TOOL_BINDING_ID,
     MERGE_BINDING_ID,
@@ -268,6 +270,7 @@ def _catalog_capabilities() -> list[WorkflowRuntimeCapability]:
                 probes=["typed_port_contract_registered"],
             ),
         ),
+        *_data_operator_capabilities(),
         _blocked_catalog(
             "intelligence.agent.summary",
             "LLM Summary",
@@ -538,6 +541,85 @@ def _catalog_capabilities() -> list[WorkflowRuntimeCapability]:
     ]
 
 
+def _data_operator_capabilities() -> list[WorkflowRuntimeCapability]:
+    specs_by_kind: dict[str, list[object]] = {}
+    for spec in list_data_operator_specs():
+        specs_by_kind.setdefault(spec.kind, []).append(spec)
+
+    rows: list[WorkflowRuntimeCapability] = []
+    for catalog_id, binding_id in DATA_OPERATOR_CATALOG_BINDINGS.items():
+        operator_kind = catalog_id.rsplit(".", 1)[-1]
+        specs = sorted(
+            specs_by_kind.get(operator_kind, []),
+            key=lambda spec: spec.operator_id,
+        )
+        if not specs:
+            continue
+        operators = [
+            {
+                "id": spec.operator_id,
+                "operatorId": spec.operator_id,
+                "kind": spec.kind,
+                "label": spec.label,
+                "description": spec.description,
+                "pack": spec.pack_id,
+                "packId": spec.pack_id,
+                "version": spec.pack_version,
+                "packVersion": spec.pack_version,
+                "status": "runnable",
+                "readiness": "ready",
+                "configKeys": list(spec.config_keys),
+            }
+            for spec in specs
+        ]
+        rows.append(
+            _capability(
+                id=catalog_id,
+                label=f"Data {operator_kind.title()}",
+                surface="catalog",
+                status="runnable",
+                backend_available=True,
+                kind="agent",
+                capability="normalize",
+                provider="workflow",
+                runtime_binding=binding_id,
+                reason="Registered versioned data operators execute on Record Candidates.",
+                tags=["data", "operator", operator_kind],
+                source="backend.workflow.data_operators",
+                manifest={
+                    **_manifest(
+                        schema=f"capability.data.{operator_kind}.v1",
+                        input_ports=[_port("in", "recordCandidate[]")],
+                        output_ports=[_port("out", "recordCandidate[]")],
+                        runtime_binding=binding_id,
+                        trace_events=[
+                            "partial:outputItemCount",
+                            "completed",
+                            "failed",
+                        ],
+                        probes=["data_operator_registry"],
+                    ),
+                    "operatorIds": list(
+                        dict.fromkeys(operator["id"] for operator in operators)
+                    ),
+                    "operators": operators,
+                    "packs": sorted({operator["packId"] for operator in operators}),
+                    "params": list(
+                        dict.fromkeys(
+                            key for spec in specs for key in spec.config_keys
+                        )
+                    ),
+                    "artifacts": [
+                        "recordCandidate[]",
+                        "metrics",
+                        "rejectedCandidateIds",
+                    ],
+                },
+            )
+        )
+    return rows
+
+
 def _manifest(
     *,
     schema: str,

backend/workflow/compiler.py

@@ -20,6 +20,10 @@
     WorkflowProjectNode,
     WorkflowRuntimePreview,
 )
+from backend.workflow.data_operators import (
+    list_data_operator_specs,
+    resolve_data_operator,
+)
 from backend.workflow.hda_templates import materialize_hda_templates
 from backend.workflow.node_registry import (
     forbidden_node_definition_keys,
@@ -29,6 +33,7 @@
 
 INTERNAL_ID_SEPARATOR = "::"
 MAX_NODE_PATH_DEPTH = 4
+_LEGACY_DATA_OPERATOR_PACK_VERSION = "1.0.0"
 
 
 @dataclass(frozen=True)
@@ -64,6 +69,22 @@ class _PortContract:
         [_PortContract("in", "input", "recordCandidate[]")],
         [_PortContract("out", "output", "recordCandidate[]")],
     ),
+    "intelligence.data.generate": (
+        [_PortContract("in", "input", "recordCandidate[]")],
+        [_PortContract("out", "output", "recordCandidate[]")],
+    ),
+    "intelligence.data.filter": (
+        [_PortContract("in", "input", "recordCandidate[]")],
+        [_PortContract("out", "output", "recordCandidate[]")],
+    ),
+    "intelligence.data.evaluate": (
+        [_PortContract("in", "input", "recordCandidate[]")],
+        [_PortContract("out", "output", "recordCandidate[]")],
+    ),
+    "intelligence.data.refine": (
+        [_PortContract("in", "input", "recordCandidate[]")],
+        [_PortContract("out", "output", "recordCandidate[]")],
+    ),
     "intelligence.flow.merge": (
         [
             _PortContract("in1", "input", "recordCandidate[]"),
@@ -316,6 +337,7 @@ def _validate_project(project: WorkflowProject) -> list[WorkflowCompileError]:
             )
 
         errors.extend(_validate_node_origin(node, ["nodes", node.id]))
+        errors.extend(_validate_data_operator_node(node, ["nodes", node.id]))
 
     errors.extend(_validate_typed_edges(project.nodes, project.edges, path_prefix=["edges"]))
     errors.extend(_cycle_errors(project))
@@ -366,6 +388,103 @@ def _validate_node_origin(
     return errors
 
 
+def _validate_data_operator_node(
+    node: WorkflowProjectNode,
+    path_prefix: list[str],
+) -> list[WorkflowCompileError]:
+    catalog_id = _read_string((node.ui or {}).get("catalogId"))
+    prefix = "intelligence.data."
+    expected_kind = (
+        catalog_id.removeprefix(prefix)
+        if catalog_id and catalog_id.startswith(prefix)
+        else None
+    )
+    operator_id = _read_string(node.params.get("operatorId"))
+    if expected_kind not in {"generate", "filter", "evaluate", "refine"}:
+        if "operatorId" not in node.params:
+            return []
+        return [
+            WorkflowCompileError(
+                code="data_operator_catalog_required",
+                message=(
+                    f'Workflow node "{node.id}" declares params.operatorId but does '
+                    "not reference a registered intelligence.data catalog node"
+                ),
+                node_id=node.id,
+                path=[*path_prefix, "ui", "catalogId"],
+            )
+        ]
+    if operator_id is None:
+        return [
+            WorkflowCompileError(
+                code="missing_data_operator_id",
+                message=f'Workflow data node "{node.id}" requires params.operatorId',
+                node_id=node.id,
+                path=[*path_prefix, "params", "operatorId"],
+            )
+        ]
+
+    pack_version_provided = "packVersion" in node.params
+    requested_pack_version = _read_string(node.params.get("packVersion"))
+    if pack_version_provided and requested_pack_version is None:
+        return [
+            WorkflowCompileError(
+                code="unsupported_data_operator_version",
+                message=(
+                    f'Workflow data node "{node.id}" requires params.packVersion '
+                    "to be a non-empty string when provided"
+                ),
+                node_id=node.id,
+                path=[*path_prefix, "params", "packVersion"],
+            )
+        ]
+    resolved_pack_version = (
+        requested_pack_version or _LEGACY_DATA_OPERATOR_PACK_VERSION
+    )
+    spec = resolve_data_operator(operator_id, resolved_pack_version)
+    if spec is None:
+        if any(
+            registered.id == operator_id
+            for registered in list_data_operator_specs()
+        ):
+            return [
+                WorkflowCompileError(
+                    code="unsupported_data_operator_version",
+                    message=(
+                        f'Workflow data node "{node.id}" references unsupported '
+                        f'version "{resolved_pack_version}" of operator '
+                        f'"{operator_id}"'
+                    ),
+                    node_id=node.id,
+                    path=[*path_prefix, "params", "packVersion"],
+                )
+            ]
+        return [
+            WorkflowCompileError(
+                code="unknown_data_operator",
+                message=(
+                    f'Workflow data node "{node.id}" references unknown operator '
+                    f'"{operator_id}"'
+                ),
+                node_id=node.id,
+                path=[*path_prefix, "params", "operatorId"],
+            )
+        ]
+    if spec.kind != expected_kind:
+        return [
+            WorkflowCompileError(
+                code="data_operator_kind_mismatch",
+                message=(
+                    f'Data operator "{operator_id}" has kind "{spec.kind}", but node '
+                    f'"{node.id}" requires "{expected_kind}"'
+                ),
+                node_id=node.id,
+                path=[*path_prefix, "params", "operatorId"],
+            )
+        ]
+    return []
+
+
 def _validate_typed_edges(
     nodes: list[WorkflowProjectNode],
     edges: list[WorkflowProjectEdge],
@@ -1004,6 +1123,7 @@ def _validate_package_internals(
                 internal_path_prefix,
             )
         )
+        errors.extend(_validate_data_operator_node(internal_node, internal_path_prefix))
         if _is_structural_container(internal_node):
             errors.extend(
                 _validate_package_internals(