[Sig Events] Per-step connector overrides for onboarding and insights discovery

x-pack/platform/plugins/shared/streams/server/lib/sig_events/tasks/insights_discovery.ts

@@ -31,6 +31,7 @@ export interface InsightsDiscoveryTaskResult {
 export interface InsightsDiscoveryTaskParams {
   /** When provided, only generate insights for these stream names. Otherwise all streams are used. */
   streamNames?: string[];
+  connectorId?: string;
 }
 
 export const STREAMS_INSIGHTS_DISCOVERY_TASK_TYPE = 'streams_insights_discovery';
@@ -47,8 +48,11 @@ export function createStreamsInsightsDiscoveryTask(taskContext: TaskContext) {
               }
               const { fakeRequest } = runContext;
 
-              const { streamNames, _task } = runContext.taskInstance
-                .params as TaskParams<InsightsDiscoveryTaskParams>;
+              const {
+                streamNames,
+                connectorId: connectorIdOverride,
+                _task,
+              } = runContext.taskInstance.params as TaskParams<InsightsDiscoveryTaskParams>;
 
               const {
                 taskClient,
@@ -62,12 +66,14 @@ export function createStreamsInsightsDiscoveryTask(taskContext: TaskContext) {
               });
 
               const taskLogger = taskContext.logger.get('insights_discovery');
-              const connectorId = await resolveConnectorForFeature({
-                searchInferenceEndpoints: taskContext.server.searchInferenceEndpoints,
-                featureId: STREAMS_SIG_EVENTS_DISCOVERY_INFERENCE_FEATURE_ID,
-                featureName: 'discovery',
-                request: fakeRequest,
-              });
+              const connectorId =
+                connectorIdOverride ??
+                (await resolveConnectorForFeature({
+                  searchInferenceEndpoints: taskContext.server.searchInferenceEndpoints,
+                  featureId: STREAMS_SIG_EVENTS_DISCOVERY_INFERENCE_FEATURE_ID,
+                  featureName: 'discovery',
+                  request: fakeRequest,
+                }));
               taskLogger.debug(`Using connector ${connectorId} for discovery`);
               const boundInferenceClient = inferenceClient.bindTo({ connectorId });
 
@@ -116,7 +122,7 @@ export function createStreamsInsightsDiscoveryTask(taskContext: TaskContext) {
 
                 await taskClient.complete<InsightsDiscoveryTaskParams, InsightsDiscoveryTaskResult>(
                   _task,
-                  { streamNames },
+                  { streamNames, connectorId: connectorIdOverride },
                   { insights, tokensUsed: result.tokens_used }
                 );
               } catch (error) {
@@ -144,7 +150,7 @@ export function createStreamsInsightsDiscoveryTask(taskContext: TaskContext) {
 
                 await taskClient.fail<InsightsDiscoveryTaskParams>(
                   _task,
-                  { streamNames },
+                  { streamNames, connectorId: connectorIdOverride },
                   errorMessage
                 );
                 return getDeleteTaskRunResult();

x-pack/platform/plugins/shared/streams/server/lib/sig_events/tasks/significant_events_queries_generation.ts

@@ -28,6 +28,7 @@ export interface SignificantEventsQueriesGenerationTaskParams {
   end: number;
   sampleDocsSize?: number;
   streamName: string;
+  connectorId?: string;
 }
 
 export const SIGNIFICANT_EVENTS_QUERIES_GENERATION_TASK_TYPE =
@@ -49,7 +50,14 @@ export function createStreamsSignificantEventsQueriesGenerationTask(taskContext:
               }
               const { fakeRequest } = runContext;
 
-              const { start, end, sampleDocsSize, streamName, _task } = runContext.taskInstance
+              const {
+                start,
+                end,
+                sampleDocsSize,
+                streamName,
+                connectorId: connectorIdOverride,
+                _task,
+              } = runContext.taskInstance
                 .params as TaskParams<SignificantEventsQueriesGenerationTaskParams>;
 
               const {
@@ -64,12 +72,14 @@ export function createStreamsSignificantEventsQueriesGenerationTask(taskContext:
               });
 
               const taskLogger = taskContext.logger.get('significant_events_queries_generation');
-              const connectorId = await resolveConnectorForFeature({
-                searchInferenceEndpoints: taskContext.server.searchInferenceEndpoints,
-                featureId: STREAMS_SIG_EVENTS_KI_QUERY_GENERATION_INFERENCE_FEATURE_ID,
-                featureName: 'query generation',
-                request: fakeRequest,
-              });
+              const connectorId =
+                connectorIdOverride ??
+                (await resolveConnectorForFeature({
+                  searchInferenceEndpoints: taskContext.server.searchInferenceEndpoints,
+                  featureId: STREAMS_SIG_EVENTS_KI_QUERY_GENERATION_INFERENCE_FEATURE_ID,
+                  featureName: 'query generation',
+                  request: fakeRequest,
+                }));
               taskLogger.debug(`Using connector ${connectorId} for rule generation`);
 
               try {
@@ -113,7 +123,11 @@ export function createStreamsSignificantEventsQueriesGenerationTask(taskContext:
                 await taskClient.complete<
                   SignificantEventsQueriesGenerationTaskParams,
                   SignificantEventsQueriesGenerationResult
-                >(_task, { start, end, sampleDocsSize, streamName }, result);
+                >(
+                  _task,
+                  { start, end, sampleDocsSize, streamName, connectorId: connectorIdOverride },
+                  result
+                );
               } catch (error) {
                 if (isDefinitionNotFoundError(error)) {
                   taskContext.logger.debug(
@@ -146,7 +160,13 @@ export function createStreamsSignificantEventsQueriesGenerationTask(taskContext:
 
                 await taskClient.fail<SignificantEventsQueriesGenerationTaskParams>(
                   _task,
-                  { start, end, sampleDocsSize, streamName },
+                  {
+                    start,
+                    end,
+                    sampleDocsSize,
+                    streamName,
+                    connectorId: connectorIdOverride,
+                  },
                   errorMessage
                 );
 

x-pack/platform/plugins/shared/streams/server/lib/tasks/task_definitions/features_identification/index.ts

@@ -335,6 +335,7 @@ export interface FeaturesIdentificationTaskParams {
   start: number;
   end: number;
   streamName: string;
+  connectorId?: string;
 }
 
 export const FEATURES_IDENTIFICATION_TASK_TYPE = 'streams_features_identification';
@@ -355,8 +356,13 @@ export function createStreamsFeaturesIdentificationTask(taskContext: TaskContext
               }
               const { fakeRequest } = runContext;
 
-              const { start, end, streamName, _task } = runContext.taskInstance
-                .params as TaskParams<FeaturesIdentificationTaskParams>;
+              const {
+                start,
+                end,
+                streamName,
+                connectorId: connectorIdOverride,
+                _task,
+              } = runContext.taskInstance.params as TaskParams<FeaturesIdentificationTaskParams>;
 
               const runId = uuid();
               const trackEmptyTelemetry = (state: 'canceled' | 'failure') => {
@@ -395,12 +401,14 @@ export function createStreamsFeaturesIdentificationTask(taskContext: TaskContext
               });
 
               const taskLogger = taskContext.logger.get('features_identification', streamName);
-              const connectorId = await resolveConnectorForFeature({
-                searchInferenceEndpoints: taskContext.server.searchInferenceEndpoints,
-                featureId: STREAMS_SIG_EVENTS_KI_EXTRACTION_INFERENCE_FEATURE_ID,
-                featureName: 'knowledge indicator extraction',
-                request: fakeRequest,
-              });
+              const connectorId =
+                connectorIdOverride ??
+                (await resolveConnectorForFeature({
+                  searchInferenceEndpoints: taskContext.server.searchInferenceEndpoints,
+                  featureId: STREAMS_SIG_EVENTS_KI_EXTRACTION_INFERENCE_FEATURE_ID,
+                  featureName: 'knowledge indicator extraction',
+                  request: fakeRequest,
+                }));
               taskLogger.debug(`Using connector ${connectorId} for knowledge indicator extraction`);
 
               let hasTrackedIteration = false;
@@ -509,7 +517,7 @@ export function createStreamsFeaturesIdentificationTask(taskContext: TaskContext
 
                 await taskClient.complete<FeaturesIdentificationTaskParams, IdentifyFeaturesResult>(
                   _task,
-                  { start, end, streamName },
+                  { start, end, streamName, connectorId: connectorIdOverride },
                   {
                     features: allFeatures,
                     durationMs,
@@ -566,7 +574,7 @@ export function createStreamsFeaturesIdentificationTask(taskContext: TaskContext
 
                 await taskClient.fail<FeaturesIdentificationTaskParams, IdentifyFeaturesResult>(
                   _task,
-                  { start, end, streamName },
+                  { start, end, streamName, connectorId: connectorIdOverride },
                   errorMessage,
                   {
                     features: [],

x-pack/platform/plugins/shared/streams/server/lib/tasks/task_definitions/onboarding.ts

@@ -42,6 +42,10 @@ export interface OnboardingTaskParams {
   to: number;
   steps: OnboardingStep[];
   saveQueries: boolean;
+  connectors?: {
+    features?: string;
+    queries?: string;
+  };
 }
 
 export const STREAMS_ONBOARDING_TASK_TYPE = 'streams_onboarding';
@@ -51,6 +55,31 @@ export function getOnboardingTaskId(streamName: string, saveQueries: boolean = t
   return saveQueries ? base : `${base}_no_save_queries`;
 }
 
+const FEATURES_IDENTIFICATION_RECENCY_MS = 12 * 60 * 60 * 1000; // 12 hours
+
+async function areFeaturesUpToDate({
+  taskClient,
+  featuresTaskId,
+}: {
+  taskClient: TaskClient<StreamsTaskType>;
+  featuresTaskId: string;
+}) {
+  const featuresTask = await taskClient.get<
+    FeaturesIdentificationTaskParams,
+    IdentifyFeaturesResult
+  >(featuresTaskId);
+
+  if (featuresTask.status !== TaskStatus.Completed) {
+    return false;
+  }
+
+  return Boolean(
+    featuresTask.last_completed_at &&
+      Date.now() - new Date(featuresTask.last_completed_at).getTime() <
+        FEATURES_IDENTIFICATION_RECENCY_MS
+  );
+}
+
 export function createStreamsOnboardingTask(taskContext: TaskContext) {
   return {
     [STREAMS_ONBOARDING_TASK_TYPE]: {
@@ -64,8 +93,8 @@ export function createStreamsOnboardingTask(taskContext: TaskContext) {
               }
               const { fakeRequest } = runContext;
 
-              const { streamName, from, to, steps, saveQueries, _task } = runContext.taskInstance
-                .params as TaskParams<OnboardingTaskParams>;
+              const { streamName, from, to, steps, saveQueries, connectors, _task } = runContext
+                .taskInstance.params as TaskParams<OnboardingTaskParams>;
 
               const { taskClient, queryClient, streamsClient } = await taskContext.getScopedClients(
                 {
@@ -83,21 +112,34 @@ export function createStreamsOnboardingTask(taskContext: TaskContext) {
                   switch (step) {
                     case OnboardingStep.FeaturesIdentification: {
                       const featuresTaskId = getFeaturesIdentificationTaskId(streamName);
-
-                      await scheduleFeaturesIdentificationTask(
-                        {
-                          start: from,
-                          end: to,
-                          streamName,
-                        },
-                        taskClient,
-                        fakeRequest
-                      );
-
-                      featuresTaskResult = await waitForSubtask<
-                        FeaturesIdentificationTaskParams,
-                        IdentifyFeaturesResult
-                      >(featuresTaskId, runContext.taskInstance.id, taskClient);
+                      const isFeaturesOnlyStep =
+                        steps.length === 1 && steps[0] === OnboardingStep.FeaturesIdentification;
+
+                      if (
+                        !isFeaturesOnlyStep &&
+                        (await areFeaturesUpToDate({ taskClient, featuresTaskId }))
+                      ) {
+                        featuresTaskResult = await taskClient.getStatus<
+                          FeaturesIdentificationTaskParams,
+                          IdentifyFeaturesResult
+                        >(featuresTaskId);
+                      } else {
+                        await scheduleFeaturesIdentificationTask(
+                          {
+                            start: from,
+                            end: to,
+                            streamName,
+                            connectorId: connectors?.features,
+                          },
+                          taskClient,
+                          fakeRequest
+                        );
+
+                        featuresTaskResult = await waitForSubtask<
+                          FeaturesIdentificationTaskParams,
+                          IdentifyFeaturesResult
+                        >(featuresTaskId, runContext.taskInstance.id, taskClient);
+                      }
 
                       if (featuresTaskResult.status !== TaskStatus.Completed) {
                         return;
@@ -111,6 +153,7 @@ export function createStreamsOnboardingTask(taskContext: TaskContext) {
                           start: from,
                           end: to,
                           streamName,
+                          connectorId: connectors?.queries,
                         },
                         taskClient,
                         fakeRequest
@@ -140,7 +183,14 @@ export function createStreamsOnboardingTask(taskContext: TaskContext) {
 
                 await taskClient.complete<OnboardingTaskParams, OnboardingResult>(
                   _task,
-                  { streamName, from, to, steps, saveQueries },
+                  {
+                    streamName,
+                    from,
+                    to,
+                    steps,
+                    saveQueries,
+                    connectors,
+                  },
                   { featuresTaskResult, queriesTaskResult }
                 );
               } catch (error) {
@@ -168,6 +218,7 @@ export function createStreamsOnboardingTask(taskContext: TaskContext) {
                     to,
                     steps,
                     saveQueries,
+                    connectors,
                   },
                   errorMessage
                 );

x-pack/platform/plugins/shared/streams/server/routes/internal/sig_events/insights/route.ts

@@ -47,6 +47,12 @@ const insightsTaskRoute = createServerRoute({
         .array(z.string())
         .describe('List of stream names to generate insights for.')
         .optional(),
+      connectorId: z
+        .string()
+        .optional()
+        .describe(
+          'Optional connector ID override. When omitted the server resolves the connector from the inference feature registry.'
+        ),
     }),
   }),
   handler: async ({ params, request, getScopedClients, server }): Promise<InsightsTaskResult> => {
@@ -67,6 +73,7 @@ const insightsTaskRoute = createServerRoute({
               taskId: STREAMS_INSIGHTS_DISCOVERY_TASK_TYPE,
               params: {
                 streamNames: body.streamNames,
+                connectorId: body.connectorId,
               },
               request,
             },

x-pack/platform/plugins/shared/streams/server/routes/internal/streams/onboarding/route.ts

@@ -53,6 +53,15 @@ export const onboardingTaskRoute = createServerRoute({
         .describe(
           'Optional list of steps to perform as part of stream onboarding in the specified sequence. By default it will execute all steps.'
         ),
+      connectors: z
+        .object({
+          features: z.string().optional().describe('Connector ID for features identification.'),
+          queries: z.string().optional().describe('Connector ID for queries generation.'),
+        })
+        .optional()
+        .describe(
+          'Optional per-step connector overrides. When omitted the server resolves connectors from the inference feature registry.'
+        ),
     }),
   }),
   handler: async ({ params, request, getScopedClients, server }): Promise<OnboardingTaskResult> => {
@@ -85,6 +94,7 @@ export const onboardingTaskRoute = createServerRoute({
                 to: body.to,
                 steps: body.steps,
                 saveQueries,
+                connectors: body.connectors,
               },
               request,
             },

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

@@ -103,6 +103,7 @@ dependsOn:
   - '@kbn/react-query'
   - '@kbn/timerange'
   - '@kbn/inference-common'
+  - '@kbn/inference-connectors'
   - '@kbn/esql'
   - '@kbn/inference-endpoint-ui-common'
   - '@kbn/stack-connectors-plugin'