[WorkplaceAI] Enhance CRUD API with workflows & onechat (elastic#246561)

## Summary

Closes [[WorkplaceAI] Add CRUD API for managing data
connectors](elastic/search-team#12059)

This adds capability to create/delete workflows & tools. Added some
idempotency logic on DELETE so that we handle:
* case where workflows or tools or ksc was already deleted - gracefully
move on with deletion of any remaining resources if there any left,
otherwise report success
* partial cascade deletes - don't delete SO, but update it with
remaining resources IDs (that we failed to delete)

Also added unit tests and more logging, and refactored into helper
functions for readability and to keep code DRYer.

## Testing

```
yarn test:jest x-pack/platform/plugins/shared/data_connectors/server/routes/
```

Or spin up local kibana and test out endpoints in dev console

```
GET kbn:/api/data_connectors
GET kbn:/api/data_connectors/9932f24c-9974-4c8d-9a10-79eef5368201

DELETE kbn:/api/data_connectors
DELETE kbn:/api/data_connectors/9932f24c-9974-4c8d-9a10-79eef5368201


POST kbn:/api/data_connectors
{
    "type": "notion",
    "name": "My beloved Notion",
    "token": "..."
}
```

### Checklist

Check the PR satisfies following conditions. 

Reviewers should verify this PR satisfies this list as well.

- [ ] Any text added follows [EUI's writing
guidelines](https://elastic.github.io/eui/#/guidelines/writing), uses
sentence case text and includes [i18n
support](https://github.com/elastic/kibana/blob/main/src/platform/packages/shared/kbn-i18n/README.md)
- [ ]
[Documentation](https://www.elastic.co/guide/en/kibana/master/development-documentation.html)
was added for features that require explanation or tutorials
- [ ] [Unit or functional
tests](https://www.elastic.co/guide/en/kibana/master/development-tests.html)
were updated or added to match the most common scenarios
- [ ] If a plugin configuration key changed, check if it needs to be
allowlisted in the cloud and added to the [docker
list](https://github.com/elastic/kibana/blob/main/src/dev/build/tasks/os_packages/docker_generator/resources/base/bin/kibana-docker)
- [ ] This was checked for breaking HTTP API changes, and any breaking
changes have been approved by the breaking-change committee. The
`release_note:breaking` label should be applied in these situations.
- [ ] [Flaky Test
Runner](https://ci-stats.kibana.dev/trigger_flaky_test_runner/1) was
used on any tests changed
- [x] The PR description includes the appropriate Release Notes section,
and the correct `release_note:*` label is applied per the
[guidelines](https://www.elastic.co/guide/en/kibana/master/contributing.html#kibana-release-notes-process)
- [x] Review the [backport
guidelines](https://docs.google.com/document/d/1VyN5k91e5OVumlc0Gb9RPa3h1ewuPE705nRtioPiTvY/edit?usp=sharing)
and apply applicable `backport:*` labels.

---------

Co-authored-by: kibanamachine <42973632+kibanamachine@users.noreply.github.com>
Co-authored-by: Sean Story <sean.story@elastic.co>

Author: 3 people

src/platform/plugins/shared/workflows_management/server/workflows_management/routes/delete_workflows_bulk.test.ts

@@ -47,7 +47,12 @@ describe('DELETE /api/workflows', () => {
     });
 
     it('should delete multiple workflows successfully', async () => {
-      workflowsApi.deleteWorkflows = jest.fn().mockResolvedValue(undefined);
+      const mockResult = {
+        total: 3,
+        deleted: 3,
+        failures: [],
+      };
+      workflowsApi.deleteWorkflows = jest.fn().mockResolvedValue(mockResult);
 
       const mockContext = {};
       const mockRequest = {
@@ -66,11 +71,16 @@ describe('DELETE /api/workflows', () => {
         'default',
         mockRequest
       );
-      expect(mockResponse.ok).toHaveBeenCalledWith();
+      expect(mockResponse.ok).toHaveBeenCalledWith({ body: mockResult });
     });
 
     it('should handle empty ids array gracefully', async () => {
-      workflowsApi.deleteWorkflows = jest.fn().mockResolvedValue(undefined);
+      const mockResult = {
+        total: 3,
+        deleted: 3,
+        failures: [],
+      };
+      workflowsApi.deleteWorkflows = jest.fn().mockResolvedValue(mockResult);
 
       const mockContext = {};
       const mockRequest = {
@@ -85,7 +95,7 @@ describe('DELETE /api/workflows', () => {
       await routeHandler(mockContext, mockRequest, mockResponse);
 
       expect(workflowsApi.deleteWorkflows).toHaveBeenCalledWith([], 'default', mockRequest);
-      expect(mockResponse.ok).toHaveBeenCalledWith();
+      expect(mockResponse.ok).toHaveBeenCalledWith({ body: mockResult });
     });
 
     it('should handle API errors gracefully', async () => {
@@ -113,7 +123,12 @@ describe('DELETE /api/workflows', () => {
     });
 
     it('should work with different space contexts', async () => {
-      workflowsApi.deleteWorkflows = jest.fn().mockResolvedValue(undefined);
+      const mockResult = {
+        total: 2,
+        deleted: 2,
+        failures: [],
+      };
+      workflowsApi.deleteWorkflows = jest.fn().mockResolvedValue(mockResult);
       mockSpaces.getSpaceId = jest.fn().mockReturnValue('custom-space');
 
       const mockContext = {};
@@ -133,7 +148,7 @@ describe('DELETE /api/workflows', () => {
         'custom-space',
         mockRequest
       );
-      expect(mockResponse.ok).toHaveBeenCalledWith();
+      expect(mockResponse.ok).toHaveBeenCalledWith({ body: mockResult });
     });
 
     it('should handle Elasticsearch connection errors', async () => {

src/platform/plugins/shared/workflows_management/server/workflows_management/routes/delete_workflows_bulk.ts

@@ -34,8 +34,8 @@ export function registerDeleteWorkflowsBulkRoute({
       try {
         const { ids } = request.body as { ids: string[] };
         const spaceId = spaces.getSpaceId(request);
-        await api.deleteWorkflows(ids, spaceId, request);
-        return response.ok();
+        const result = await api.deleteWorkflows(ids, spaceId, request);
+        return response.ok({ body: result });
       } catch (error) {
         return handleRouteError(response, error);
       }

src/platform/plugins/shared/workflows_management/server/workflows_management/workflows_management_api.ts

@@ -52,6 +52,15 @@ export interface GetWorkflowsParams {
   _full?: boolean;
 }
 
+export interface DeleteWorkflowsResponse {
+  total: number;
+  deleted: number;
+  failures: Array<{
+    id: string;
+    error: string;
+  }>;
+}
+
 export interface GetWorkflowExecutionLogsParams {
   executionId: string;
   stepExecutionId?: string;
@@ -170,7 +179,7 @@ export class WorkflowsManagementApi {
     workflowIds: string[],
     spaceId: string,
     request: KibanaRequest
-  ): Promise<void> {
+  ): Promise<DeleteWorkflowsResponse> {
     return this.workflowsService.deleteWorkflows(workflowIds, spaceId);
   }
 

src/platform/plugins/shared/workflows_management/server/workflows_management/workflows_management_service.test.ts

@@ -82,6 +82,7 @@ describe('WorkflowsService', () => {
       index: jest.fn().mockResolvedValue({ _id: 'test-id' }),
       update: jest.fn().mockResolvedValue({ _id: 'test-id' }),
       delete: jest.fn().mockResolvedValue({ _id: 'test-id' }),
+      bulk: jest.fn().mockResolvedValue({ items: [] }),
     } as any;
 
     mockLogger = loggerMock.create();
@@ -1193,14 +1194,27 @@ steps:
           total: { value: 1 },
         },
       } as any);
-      mockEsClient.index.mockResolvedValue({ _id: 'test-workflow-id' } as any);
+      mockEsClient.bulk.mockResolvedValue({
+        items: [
+          {
+            index: {
+              _id: 'test-workflow-id',
+              status: 200,
+            },
+          },
+        ],
+      } as any);
 
-      await service.deleteWorkflows(['test-workflow-id'], 'default');
+      const result = await service.deleteWorkflows(['test-workflow-id'], 'default');
+
+      expect(result).toEqual({
+        total: 1,
+        deleted: 1,
+        failures: [],
+      });
 
       expect(mockEsClient.search).toHaveBeenCalledWith(
         expect.objectContaining({
-          index: '.workflows-workflows',
-          allow_no_indices: true,
           query: {
             bool: {
               must: [{ ids: { values: ['test-workflow-id'] } }, { term: { spaceId: 'default' } }],
@@ -1209,18 +1223,16 @@ steps:
           size: 1,
         })
       );
-      expect(mockEsClient.index).toHaveBeenCalledWith(
-        expect.objectContaining({
-          id: 'test-workflow-id',
-          index: '.workflows-workflows',
-          document: expect.objectContaining({
-            enabled: false,
-            deleted_at: expect.any(Date),
-          }),
-          refresh: 'wait_for',
-          require_alias: true,
-        })
-      );
+      // Verify bulk was called with correct structure
+      const bulkCall = mockEsClient.bulk.mock.calls[0][0];
+      expect(bulkCall.index).toBe('.workflows-workflows');
+      expect(bulkCall.operations).toBeDefined();
+      expect(bulkCall.operations).toHaveLength(2); // metadata + document
+      expect(bulkCall.operations![0]).toEqual({ index: { _id: 'test-workflow-id' } });
+      expect(bulkCall.operations![1]).toMatchObject({
+        enabled: false,
+        deleted_at: expect.any(Date),
+      });
     });
 
     it('should handle not found workflows gracefully', async () => {
@@ -1231,7 +1243,60 @@ steps:
         },
       } as any);
 
-      await expect(service.deleteWorkflows(['non-existent-id'], 'default')).resolves.not.toThrow();
+      const result = await service.deleteWorkflows(['non-existent-id'], 'default');
+
+      expect(result).toEqual({
+        total: 1,
+        deleted: 1,
+        failures: [],
+      });
+    });
+
+    it('should handle partial failures when deleting multiple workflows', async () => {
+      const mockWorkflowDocument2 = {
+        _id: 'workflow-2',
+        _source: {
+          ...mockWorkflowDocument._source,
+          name: 'Test Workflow 2',
+        },
+      };
+
+      // Mock search to return both workflows
+      mockEsClient.search.mockResolvedValue({
+        hits: {
+          hits: [{ ...mockWorkflowDocument, _id: 'workflow-1' }, mockWorkflowDocument2],
+          total: { value: 2 },
+        },
+      } as any);
+
+      // Mock bulk operation where one succeeds and one fails
+      mockEsClient.bulk.mockResolvedValue({
+        items: [
+          {
+            index: {
+              _id: 'workflow-1',
+              status: 200,
+            },
+          },
+          {
+            index: {
+              _id: 'workflow-2',
+              status: 500,
+              error: {
+                reason: 'Database error',
+              },
+            },
+          },
+        ],
+      } as any);
+
+      const result = await service.deleteWorkflows(['workflow-1', 'workflow-2'], 'default');
+
+      expect(result).toEqual({
+        total: 2,
+        deleted: 1,
+        failures: [{ id: 'workflow-2', error: 'Database error' }],
+      });
     });
   });
 

src/platform/plugins/shared/workflows_management/server/workflows_management/workflows_management_service.ts

@@ -55,6 +55,7 @@ import { searchStepExecutions } from './lib/search_step_executions';
 import { searchWorkflowExecutions } from './lib/search_workflow_executions';
 
 import type {
+  DeleteWorkflowsResponse,
   GetAvailableConnectorsResponse,
   GetStepExecutionParams,
   GetWorkflowsParams,
@@ -468,47 +469,74 @@ export class WorkflowsService {
     }
   }
 
-  public async deleteWorkflows(ids: string[], spaceId: string): Promise<void> {
+  public async deleteWorkflows(ids: string[], spaceId: string): Promise<DeleteWorkflowsResponse> {
     if (!this.workflowStorage) {
       throw new Error('WorkflowsService not initialized');
     }
 
     const now = new Date();
+    const failures: Array<{ id: string; error: string }> = [];
+    const client = this.workflowStorage.getClient();
 
-    // Soft delete by setting deleted_at timestamp
-    for (const id of ids) {
+    // Bulk fetch all workflows in a single search call
+    const searchResponse = await client.search({
+      query: {
+        bool: {
+          must: [{ ids: { values: ids } }, { term: { spaceId } }],
+        },
+      },
+      size: ids.length,
+      track_total_hits: false,
+    });
+
+    // Build bulk operations for all found workflows
+    const bulkOperations = searchResponse.hits.hits.map((hit) => ({
+      index: {
+        _id: hit._id,
+        document: {
+          ...hit._source,
+          deleted_at: now,
+          enabled: false,
+        },
+      },
+    }));
+
+    // Bulk update all found workflows in a single call
+    if (bulkOperations.length > 0) {
       try {
-        // Check if workflow exists and belongs to the correct space
-        const searchResponse = await this.workflowStorage.getClient().search({
-          query: {
-            bool: {
-              must: [{ ids: { values: [id] } }, { term: { spaceId } }],
-            },
-          },
-          size: 1,
-          track_total_hits: false,
+        const bulkResponse = await client.bulk({
+          operations: bulkOperations,
         });
 
-        if (searchResponse.hits.hits.length > 0) {
-          const existingDocument = searchResponse.hits.hits[0];
-          const updatedData = {
-            ...existingDocument._source,
-            deleted_at: now,
-            enabled: false,
-          };
-
-          await this.workflowStorage.getClient().index({
-            id,
-            document: updatedData,
-          });
-        }
+        // Process bulk response to track successes and failures
+        bulkResponse.items.forEach((item) => {
+          const operation = item.index;
+          if (operation?.error) {
+            failures.push({
+              id: operation._id ?? 'unknown',
+              error:
+                typeof operation.error === 'object' && 'reason' in operation.error
+                  ? operation.error.reason ?? JSON.stringify(operation.error)
+                  : JSON.stringify(operation.error),
+            });
+          }
+        });
       } catch (error) {
-        if (error.statusCode !== 404) {
-          throw error;
-        }
-        // Ignore not found errors for soft delete
+        // If the entire bulk operation fails, mark all as failed
+        bulkOperations.forEach((op) => {
+          failures.push({
+            id: op.index._id ?? 'unknown',
+            error: error instanceof Error ? error.message : String(error),
+          });
+        });
       }
     }
+
+    return {
+      total: ids.length,
+      deleted: ids.length - failures.length,
+      failures,
+    };
   }
 
   public async getWorkflows(params: GetWorkflowsParams, spaceId: string): Promise<WorkflowListDto> {

x-pack/platform/plugins/shared/data_connectors/kibana.jsonc

@@ -10,7 +10,7 @@
     "browser": true,
     "server": true,
     "configPath": ["xpack", "dataConnectors"],
-    "requiredPlugins": ["actions", "dataSourcesRegistry"],
+    "requiredPlugins": ["actions", "dataSourcesRegistry", "agentBuilder", "workflowsManagement"],
     "optionalPlugins": [],
     "requiredBundles": ["kibanaReact"],
     "extraPublicDirs": [

x-pack/platform/plugins/shared/data_connectors/moon.yml

@@ -31,6 +31,16 @@ dependsOn:
   - '@kbn/connector-specs'
   - '@kbn/actions-plugin'
   - '@kbn/core-saved-objects-common'
+  - '@kbn/core-saved-objects-api-server-mocks'
+  - '@kbn/logging-mocks'
+  - '@kbn/core-http-server-mocks'
+  - '@kbn/core-saved-objects-api-server'
+  - '@kbn/core-http-server'
+  - '@kbn/logging'
+  - '@kbn/core-saved-objects-utils-server'
+  - '@kbn/workflows-management-plugin'
+  - '@kbn/agent-builder-common'
+  - '@kbn/agent-builder-plugin'
 tags:
   - plugin
   - prod

x-pack/platform/plugins/shared/data_connectors/server/data_sources/notion/workflows.ts

@@ -21,7 +21,7 @@ inputs:
       - "data_source"
 steps:
   - name: search-page-by-title
-    type: notion.searchPageByTitle
+    type: notion.searchPageOrDSByTitle
     connector-id: ${stackConnectorId}
     with:
       query: "\${{inputs.query_string}}"