[SigEvents] Add KI feature identification endpoints and refactor task to use shared service

x-pack/platform/packages/shared/kbn-streams-schema/index.ts

@@ -260,6 +260,7 @@ export { TaskStatus, type TaskResult } from './src/tasks/types';
 
 export type { GenerateDescriptionResult } from './src/api/description_generation';
 export type { IdentifyFeaturesResult, IterationResult } from './src/api/features';
+export { tokenCountSchema, iterationResultSchema } from './src/api/features';
 
 export {
   type GenerateInsightsResult,

x-pack/platform/packages/shared/kbn-streams-schema/src/api/features/index.ts

@@ -5,17 +5,34 @@
  * 2.0.
  */
 
+import { z } from '@kbn/zod/v4';
 import type { ChatCompletionTokenCount } from '@kbn/inference-common';
 import type { BaseFeature } from '../../feature';
 
-export interface IterationResult {
-  iteration: number;
-  durationMs: number;
-  state: 'success' | 'failure';
-  tokensUsed: ChatCompletionTokenCount;
-  newFeatures: Array<{ id: string; title: string }>;
-  updatedFeatures: Array<{ id: string; title: string }>;
-}
+export const tokenCountSchema = z.object({
+  prompt: z.number(),
+  completion: z.number(),
+  thinking: z.number().optional(),
+  total: z.number(),
+  cached: z.number().optional(),
+});
+
+const featureSummarySchema = z.object({
+  id: z.string(),
+  title: z.string(),
+});
+
+export const iterationResultSchema = z.object({
+  runId: z.string(),
+  iteration: z.number(),
+  durationMs: z.number(),
+  state: z.enum(['success', 'failure']),
+  tokensUsed: tokenCountSchema,
+  newFeatures: z.array(featureSummarySchema),
+  updatedFeatures: z.array(featureSummarySchema),
+});
+
+export type IterationResult = z.infer<typeof iterationResultSchema>;
 
 export interface IdentifyFeaturesResult {
   features: BaseFeature[];

x-pack/platform/packages/shared/kbn-streams-schema/src/feature.ts

@@ -73,6 +73,7 @@ export const featureSchema = baseFeatureSchema.and(
     last_seen: z.string(),
     expires_at: z.string().optional(),
     excluded_at: z.string().optional(),
+    run_id: z.string().optional(),
   })
 );
 

x-pack/platform/plugins/shared/streams/server/lib/sig_events/features/identify_computed_features.ts

@@ -0,0 +1,61 @@
+/*
+ * 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 { ElasticsearchClient } from '@kbn/core/server';
+import type { Logger } from '@kbn/logging';
+import type { Feature, Streams } from '@kbn/streams-schema';
+import { generateAllComputedFeatures } from '@kbn/streams-ai';
+import type { FeatureClient } from '../../streams/feature/feature_client';
+import { reconcileComputedFeatures } from './reconcile_features';
+
+export interface IdentifyComputedFeaturesOptions {
+  stream: Streams.all.Definition;
+  streamName: string;
+  start: number;
+  end: number;
+  esClient: ElasticsearchClient;
+  featureClient: FeatureClient;
+  logger: Logger;
+  featureTtlDays?: number;
+  runId: string;
+}
+
+export async function identifyComputedFeatures({
+  stream,
+  streamName,
+  start,
+  end,
+  esClient,
+  featureClient,
+  logger,
+  featureTtlDays,
+  runId,
+}: IdentifyComputedFeaturesOptions): Promise<Feature[]> {
+  const computedFeatures = await generateAllComputedFeatures({
+    stream,
+    start,
+    end,
+    esClient,
+    logger: logger.get('computed_features'),
+  });
+
+  const reconciledComputedFeatures = reconcileComputedFeatures({
+    computedFeatures,
+    streamName,
+    featureTtlDays,
+    runId,
+  });
+
+  if (reconciledComputedFeatures.length > 0) {
+    await featureClient.bulk(
+      streamName,
+      reconciledComputedFeatures.map((feature) => ({ index: { feature } }))
+    );
+  }
+
+  return reconciledComputedFeatures;
+}