[Automatic Import] Handle empty categorization results (elastic#210420)

x-pack/platform/plugins/shared/automatic_import/__jest__/fixtures/categorization.ts

@@ -40,8 +40,7 @@ export const categorizationInitialPipeline: Pipeline = {
 export const categorizationExpectedResults = {
   docs: [
     {
-      key: 'value',
-      anotherKey: 'anotherValue',
+      event: { type: ['change'], category: ['database'] },
     },
   ],
   pipeline: {

x-pack/platform/plugins/shared/automatic_import/server/graphs/categorization/graph.test.ts

@@ -31,6 +31,7 @@ import {
   ActionsClientChatOpenAI,
   ActionsClientSimpleChatModel,
 } from '@kbn/langchain/server/language_models';
+import { handleValidatePipeline } from '../../util/graph';
 
 const model = new FakeLLM({
   response: "I'll callback later.",
@@ -41,6 +42,7 @@ jest.mock('./review');
 jest.mock('./categorization');
 jest.mock('./invalid');
 jest.mock('./stable');
+jest.mock('../../util/graph');
 
 jest.mock('../../util/pipeline', () => ({
   testPipeline: jest.fn(),
@@ -66,6 +68,39 @@ describe('runCategorizationGraph', () => {
     // We do not care about ES in these tests, the mock is just to prevent errors.
 
     // After this is triggered, the mock of TestPipeline will trigger the expected error, to route to error handler
+    (handleValidatePipeline as jest.Mock)
+      .mockResolvedValueOnce({
+        previousPipelineResults: [],
+        pipelineResults: [{ event: { type: ['creation'], category: ['database'] } }],
+        lastExecutedChain: 'validate_pipeline',
+      })
+      .mockResolvedValueOnce({
+        previousPipelineResults: [],
+        errors: [{ error: 'Sample error message 1' }],
+        pipelineResults: [{ event: { type: ['change'], category: ['database'] } }],
+        lastExecutedChain: 'validate_pipeline',
+      })
+      .mockResolvedValueOnce({
+        previousPipelineResults: [],
+        errors: [],
+        pipelineResults: [{ event: { type: ['change'], category: ['database'] } }],
+        lastExecutedChain: 'validate_pipeline',
+      })
+      .mockResolvedValueOnce({
+        previousPipelineResults: [],
+        pipelineResults: [{ event: { type: ['change'], category: ['database'] } }],
+        lastExecutedChain: 'validate_pipeline',
+      })
+      .mockResolvedValueOnce({
+        previousPipelineResults: [],
+        pipelineResults: [{ event: { type: ['change'], category: ['database'] } }],
+        lastExecutedChain: 'validate_pipeline',
+      })
+      .mockResolvedValueOnce({
+        previousPipelineResults: [],
+        pipelineResults: [{ event: { type: ['change'], category: ['database'] } }],
+        lastExecutedChain: 'validate_pipeline',
+      });
     (handleCategorization as jest.Mock).mockImplementation(async () => {
       const currentProcessors = await mockInvokeCategorization();
       const processors = {
@@ -145,6 +180,11 @@ describe('runCategorizationGraph', () => {
         finalized: false,
         lastExecutedChain: 'handleUpdateStableSamples',
       })
+      .mockResolvedValueOnce({
+        stableSamples: [],
+        finalized: false,
+        lastExecutedChain: 'handleUpdateStableSamples',
+      })
       .mockResolvedValueOnce({
         stableSamples: [0],
         finalized: false,
@@ -178,12 +218,13 @@ describe('runCategorizationGraph', () => {
       throw Error(`getCategorizationGraph threw an error: ${error}`);
     }
 
-    expect(response.results).toStrictEqual(categorizationExpectedResults);
-
     // Check if the functions were called
+    expect(handleValidatePipeline).toHaveBeenCalled();
     expect(handleCategorization).toHaveBeenCalled();
     expect(handleErrors).toHaveBeenCalled();
     expect(handleInvalidCategorization).toHaveBeenCalled();
     expect(handleReview).toHaveBeenCalled();
+
+    expect(response.results).toStrictEqual(categorizationExpectedResults);
   });
 });

x-pack/platform/plugins/shared/automatic_import/server/graphs/categorization/graph.ts

@@ -208,7 +208,7 @@ export async function getCategorizationGraph({ client, model }: CategorizationGr
     .addEdge(START, 'modelInput')
     .addEdge('modelOutput', END)
     .addEdge('modelInput', 'handleValidatePipeline')
-    .addEdge('handleCategorization', 'handleValidatePipeline')
+    .addEdge('handleCategorization', 'handleCategorizationValidation')
     .addEdge('handleInvalidCategorization', 'handleValidatePipeline')
     .addEdge('handleErrors', 'handleValidatePipeline')
     .addEdge('handleReview', 'handleValidatePipeline')