[SigEvents] Add model setting tab

packages/kbn-check-saved-objects-cli/current_fields.json

@@ -1261,6 +1261,7 @@
   ],
   "spaces-usage-stats": [],
   "stream-prompts": [],
+  "streams-significant-events-settings": [],
   "synthetics-dynamic-settings": [],
   "synthetics-monitor": [
     "alert",

packages/kbn-check-saved-objects-cli/current_mappings.json

@@ -4135,6 +4135,10 @@
     "dynamic": false,
     "properties": {}
   },
+  "streams-significant-events-settings": {
+    "dynamic": false,
+    "properties": {}
+  },
   "synthetics-dynamic-settings": {
     "dynamic": false,
     "properties": {}

src/core/packages/saved-objects/server-internal/src/object_types/index.ts

@@ -11,4 +11,4 @@ export { registerCoreObjectTypes } from './registration';
 
 // set minimum number of registered saved objects to ensure no object types are removed after 8.8
 // declared in internal implementation explicitly to prevent unintended changes.
-export const SAVED_OBJECT_TYPES_COUNT = 150 as const;
+export const SAVED_OBJECT_TYPES_COUNT = 151 as const;

src/core/server/integration_tests/ci_checks/saved_objects/check_registered_types.test.ts

@@ -185,6 +185,7 @@ describe('checking migration metadata changes on all registered SO types', () =>
         "space": "a3792f667888f8a976a742c941bd6db28d027604eccc854986a5ec2515f44483",
         "spaces-usage-stats": "f692e8d75a0ac9187bfbf237e1b5aa6f9a028cf5569a3df2f9e0e96e625b6215",
         "stream-prompts": "25d761054f4e6c60362ef73da19e1bee4c22bcf79e858bdb11ef3eb7f433c66c",
+        "streams-significant-events-settings": "698cd5c78851582ab24fa131b9c44be69f1f8c27e79ee0d7cbb07d30d66550e8",
         "synthetics-dynamic-settings": "30883b9647f119e1d4529f49bc0fee459c56f62489993cd7f2397ffcb8608e63",
         "synthetics-monitor": "19af677892a5eac45b3d1903c083cb222d8898d69ee545eaf079522b0bbbb85b",
         "synthetics-monitor-multi-space": "9159e968b4c011ee11fe7887572e883e0928c4b4b0924d672c77003cfdf01fe0",
@@ -1211,6 +1212,10 @@ describe('checking migration metadata changes on all registered SO types', () =>
         "stream-prompts|10.2.0: a79e6825ccd534ba98f6b8ce45357da62f162efe3fbad1d0c2c7d3c0f8d1e48c",
         "stream-prompts|10.1.0: ba35cffdf735265c1b4bc2652c4ea22c74e81fafb58a5c67eba53a4ec3e821bc",
         "=======================================================================================",
+        "streams-significant-events-settings|global: 4ed764f62f5d97deb97b50db2c41718ffd13b391",
+        "streams-significant-events-settings|mappings: e1b10e5bec060a176469a5e9a4f80c94e23abcd7",
+        "streams-significant-events-settings|schemas: da39a3ee5e6b4b0d3255bfef95601890afd80709",
+        "=====================================================================================",
         "synthetics-dynamic-settings|global: 79542e96ca9347188221391c3c7157ee5e8daefc",
         "synthetics-dynamic-settings|mappings: e1b10e5bec060a176469a5e9a4f80c94e23abcd7",
         "synthetics-dynamic-settings|schemas: da39a3ee5e6b4b0d3255bfef95601890afd80709",
@@ -1487,6 +1492,7 @@ describe('checking migration metadata changes on all registered SO types', () =>
         "space": "10.2.0",
         "spaces-usage-stats": "10.0.0",
         "stream-prompts": "10.3.0",
+        "streams-significant-events-settings": "10.0.0",
         "synthetics-dynamic-settings": "10.0.0",
         "synthetics-monitor": "10.2.0",
         "synthetics-monitor-multi-space": "10.0.0",
@@ -1647,6 +1653,7 @@ describe('checking migration metadata changes on all registered SO types', () =>
         "space": "10.2.0",
         "spaces-usage-stats": "7.14.1",
         "stream-prompts": "10.3.0",
+        "streams-significant-events-settings": "0.0.0",
         "synthetics-dynamic-settings": "0.0.0",
         "synthetics-monitor": "10.2.0",
         "synthetics-monitor-multi-space": "10.0.0",

src/core/server/integration_tests/saved_objects/registration/type_registrations.test.ts

@@ -188,6 +188,7 @@ const previouslyRegisteredTypes = [
   'workplace_search_telemetry',
   'gap_auto_fill_scheduler',
   'trial-companion-nba-milestone',
+  'streams-significant-events-settings',
 ].sort();
 
 describe('SO type registrations', () => {

x-pack/platform/plugins/shared/streams/server/lib/saved_objects/register_saved_objects.ts

@@ -8,7 +8,9 @@
 import type { SavedObjectsServiceSetup } from '@kbn/core/server';
 
 import { getStreamsPromptsSavedObject } from './significant_events/prompts_config';
+import { getStreamsSignificantEventsSettingsSavedObject } from './significant_events/model_settings_config';
 
 export const registerStreamsSavedObjects = (savedObjectsService: SavedObjectsServiceSetup) => {
   savedObjectsService.registerType(getStreamsPromptsSavedObject());
+  savedObjectsService.registerType(getStreamsSignificantEventsSettingsSavedObject());
 };

x-pack/platform/plugins/shared/streams/server/lib/saved_objects/significant_events/model_settings_config.ts

@@ -0,0 +1,40 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import type { SavedObjectsType } from '@kbn/core/server';
+import { schema, type TypeOf } from '@kbn/config-schema';
+
+export const STREAMS_SIGNIFICANT_EVENTS_SETTINGS_SO_TYPE = 'streams-significant-events-settings';
+
+export const streamsSignificantEventsSettingsSOAttributesV1 = schema.object({
+  connectorIdKnowledgeIndicatorExtraction: schema.maybe(schema.string()),
+  connectorIdRuleGeneration: schema.maybe(schema.string()),
+  connectorIdDiscovery: schema.maybe(schema.string()),
+});
+
+export type ModelSettingsConfigAttributes = TypeOf<
+  typeof streamsSignificantEventsSettingsSOAttributesV1
+>;
+
+const SINGLETON_ID = 'streams-significant-events-settings';
+
+export const getStreamsSignificantEventsSettingsSavedObject = (): SavedObjectsType => {
+  return {
+    name: STREAMS_SIGNIFICANT_EVENTS_SETTINGS_SO_TYPE,
+    hidden: false,
+    namespaceType: 'multiple',
+    mappings: {
+      dynamic: false,
+      properties: {},
+    },
+    management: {
+      importableAndExportable: true,
+    },
+  };
+};
+
+export const STREAMS_SIGNIFICANT_EVENTS_SETTINGS_SINGLETON_ID = SINGLETON_ID;

x-pack/platform/plugins/shared/streams/server/lib/saved_objects/significant_events/model_settings_config_client.ts

@@ -0,0 +1,94 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import type { Logger, SavedObjectsClientContract } from '@kbn/core/server';
+import {
+  STREAMS_SIGNIFICANT_EVENTS_SETTINGS_SO_TYPE,
+  STREAMS_SIGNIFICANT_EVENTS_SETTINGS_SINGLETON_ID,
+} from './model_settings_config';
+import type { ModelSettingsConfigAttributes } from './model_settings_config';
+
+/**
+ * Raw model settings as stored or returned by the API.
+ * Each property is undefined when no saved object exists or the value was never set.
+ */
+export interface ModelSettings {
+  connectorIdKnowledgeIndicatorExtraction?: string;
+  connectorIdRuleGeneration?: string;
+  connectorIdDiscovery?: string;
+}
+
+export interface ModelSettingsConfigClient {
+  getSettings(): Promise<ModelSettings>;
+  updateSettings(settings: Partial<ModelSettingsConfigAttributes>): Promise<void>;
+}
+
+export class ModelSettingsConfigClientImpl implements ModelSettingsConfigClient {
+  constructor(
+    private readonly soClient: SavedObjectsClientContract,
+    private readonly logger: Logger
+  ) {}
+
+  async getSettings(): Promise<ModelSettings> {
+    let attributes: ModelSettingsConfigAttributes | null = null;
+    try {
+      const data = await this.soClient.get<ModelSettingsConfigAttributes>(
+        STREAMS_SIGNIFICANT_EVENTS_SETTINGS_SO_TYPE,
+        STREAMS_SIGNIFICANT_EVENTS_SETTINGS_SINGLETON_ID
+      );
+      attributes = data.attributes;
+    } catch (err) {
+      if (
+        (err as { output?: { statusCode?: number } })?.output?.statusCode === 404 ||
+        (err as { statusCode?: number })?.statusCode === 404
+      ) {
+        this.logger.debug(
+          `No saved settings found for ${STREAMS_SIGNIFICANT_EVENTS_SETTINGS_SINGLETON_ID}`
+        );
+      } else {
+        throw err;
+      }
+    }
+
+    if (!attributes) {
+      return {
+        connectorIdKnowledgeIndicatorExtraction: undefined,
+        connectorIdRuleGeneration: undefined,
+        connectorIdDiscovery: undefined,
+      };
+    }
+
+    const toOptional = (v: string | undefined) => (v != null && v.trim() !== '' ? v : undefined);
+    return {
+      connectorIdKnowledgeIndicatorExtraction: toOptional(
+        attributes.connectorIdKnowledgeIndicatorExtraction
+      ),
+      connectorIdRuleGeneration: toOptional(attributes.connectorIdRuleGeneration),
+      connectorIdDiscovery: toOptional(attributes.connectorIdDiscovery),
+    };
+  }
+
+  async updateSettings(settings: Partial<ModelSettingsConfigAttributes>): Promise<void> {
+    const current = await this.getSettings();
+    const updates = Object.fromEntries(
+      Object.entries(settings).filter(([, v]) => v !== undefined)
+    ) as Partial<ModelSettingsConfigAttributes>;
+    const merged: ModelSettings = { ...current, ...updates };
+    const toWrite = Object.fromEntries(
+      Object.entries(merged).filter(([, v]) => v !== undefined)
+    ) as ModelSettingsConfigAttributes;
+
+    await this.soClient.create<ModelSettingsConfigAttributes>(
+      STREAMS_SIGNIFICANT_EVENTS_SETTINGS_SO_TYPE,
+      toWrite,
+      {
+        id: STREAMS_SIGNIFICANT_EVENTS_SETTINGS_SINGLETON_ID,
+        overwrite: true,
+      }
+    );
+  }
+}

x-pack/platform/plugins/shared/streams/server/lib/saved_objects/significant_events/model_settings_config_service.ts

@@ -0,0 +1,21 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import type { Logger, SavedObjectsClientContract } from '@kbn/core/server';
+import type { ModelSettingsConfigClient } from './model_settings_config_client';
+import { ModelSettingsConfigClientImpl } from './model_settings_config_client';
+
+export type { ModelSettings, ModelSettingsConfigClient } from './model_settings_config_client';
+
+export class ModelSettingsConfigService {
+  constructor(private readonly logger: Logger) {}
+
+  getClient({ soClient }: { soClient: SavedObjectsClientContract }): ModelSettingsConfigClient {
+    const clientLogger = this.logger.get('model-settings-config-client');
+    return new ModelSettingsConfigClientImpl(soClient, clientLogger);
+  }
+}

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

@@ -20,6 +20,7 @@ import { getDeleteTaskRunResult } from '@kbn/task-manager-plugin/server/task';
 import type { LogMeta } from '@kbn/logging';
 import { getErrorMessage } from '../../streams/errors/parse_error';
 import { formatInferenceProviderError } from '../../../routes/utils/create_connector_sse_error';
+import { resolveConnectorId } from '../../../routes/utils/resolve_connector_id';
 import type { TaskContext } from '.';
 import type { TaskParams } from '../types';
 import { PromptsConfigService } from '../../saved_objects/significant_events/prompts_config_service';
@@ -29,7 +30,6 @@ import { isDefinitionNotFoundError } from '../../streams/errors/definition_not_f
 import type { StreamsFeaturesIdentifiedProps } from '../../telemetry';
 
 export interface FeaturesIdentificationTaskParams {
-  connectorId: string;
   start: number;
   end: number;
   streamName: string;
@@ -53,7 +53,7 @@ export function createStreamsFeaturesIdentificationTask(taskContext: TaskContext
               }
 
               const taskStart = Date.now();
-              const { connectorId, start, end, streamName, _task } = runContext.taskInstance
+              const { start, end, streamName, _task } = runContext.taskInstance
                 .params as TaskParams<FeaturesIdentificationTaskParams>;
 
               const telemetryProps: StreamsFeaturesIdentifiedProps = {
@@ -76,10 +76,21 @@ export function createStreamsFeaturesIdentificationTask(taskContext: TaskContext
                 streamsClient,
                 inferenceClient,
                 soClient,
+                modelSettingsClient,
+                uiSettingsClient,
               } = await taskContext.getScopedClients({
                 request: runContext.fakeRequest,
               });
 
+              const taskLogger = taskContext.logger.get('features_identification');
+              const settings = await modelSettingsClient.getSettings();
+              const connectorId = await resolveConnectorId({
+                connectorId: settings.connectorIdKnowledgeIndicatorExtraction,
+                uiSettingsClient,
+                logger: taskLogger,
+              });
+              taskLogger.debug(`Using connector ${connectorId} for knowledge indicator extraction`);
+
               try {
                 const [stream, { featurePromptOverride }] = await Promise.all([
                   streamsClient.getStream(streamName),
@@ -183,7 +194,7 @@ export function createStreamsFeaturesIdentificationTask(taskContext: TaskContext
 
                 await taskClient.complete<FeaturesIdentificationTaskParams, IdentifyFeaturesResult>(
                   _task,
-                  { connectorId, start, end, streamName },
+                  { start, end, streamName },
                   { features }
                 );
 
@@ -232,7 +243,7 @@ export function createStreamsFeaturesIdentificationTask(taskContext: TaskContext
 
                 await taskClient.fail<FeaturesIdentificationTaskParams>(
                   _task,
-                  { connectorId, start, end, streamName },
+                  { start, end, streamName },
                   errorMessage
                 );
 

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

@@ -18,14 +18,14 @@ import type { TaskParams } from '../types';
 import { generateInsights } from '../../significant_events/insights/generate_insights';
 import { getErrorMessage } from '../../streams/errors/parse_error';
 import { formatInferenceProviderError } from '../../../routes/utils/create_connector_sse_error';
+import { resolveConnectorId } from '../../../routes/utils/resolve_connector_id';
 
 export interface InsightsDiscoveryTaskResult {
   insights: Insight[];
   tokensUsed: ChatCompletionTokenCount;
 }
 
 export interface InsightsDiscoveryTaskParams {
-  connectorId: string;
   /** When provided, only generate insights for these stream names. Otherwise all streams are used. */
   streamNames?: string[];
 }
@@ -43,7 +43,7 @@ export function createStreamsInsightsDiscoveryTask(taskContext: TaskContext) {
                 throw new Error('Request is required to run this task');
               }
 
-              const { connectorId, streamNames, _task } = runContext.taskInstance
+              const { streamNames, _task } = runContext.taskInstance
                 .params as TaskParams<InsightsDiscoveryTaskParams>;
 
               const {
@@ -53,10 +53,20 @@ export function createStreamsInsightsDiscoveryTask(taskContext: TaskContext) {
                 inferenceClient,
                 queryClient,
                 insightClient,
+                modelSettingsClient,
+                uiSettingsClient,
               } = await taskContext.getScopedClients({
                 request: runContext.fakeRequest,
               });
 
+              const taskLogger = taskContext.logger.get('insights_discovery');
+              const settings = await modelSettingsClient.getSettings();
+              const connectorId = await resolveConnectorId({
+                connectorId: settings.connectorIdDiscovery,
+                uiSettingsClient,
+                logger: taskLogger,
+              });
+              taskLogger.debug(`Using connector ${connectorId} for discovery`);
               const boundInferenceClient = inferenceClient.bindTo({ connectorId });
 
               try {
@@ -66,7 +76,7 @@ export function createStreamsInsightsDiscoveryTask(taskContext: TaskContext) {
                   esClient: scopedClusterClient.asCurrentUser,
                   inferenceClient: boundInferenceClient,
                   signal: runContext.abortController.signal,
-                  logger: taskContext.logger.get('insights_discovery'),
+                  logger: taskLogger,
                   streamNames,
                 });
 
@@ -104,7 +114,7 @@ export function createStreamsInsightsDiscoveryTask(taskContext: TaskContext) {
 
                 await taskClient.complete<InsightsDiscoveryTaskParams, InsightsDiscoveryTaskResult>(
                   _task,
-                  { connectorId, streamNames },
+                  { streamNames },
                   { insights, tokensUsed: result.tokens_used }
                 );
               } catch (error) {
@@ -128,7 +138,7 @@ export function createStreamsInsightsDiscoveryTask(taskContext: TaskContext) {
 
                 await taskClient.fail<InsightsDiscoveryTaskParams>(
                   _task,
-                  { connectorId, streamNames },
+                  { streamNames },
                   errorMessage
                 );
                 return getDeleteTaskRunResult();