[sigevents][kis] Unify features and queries in the same data stream

src/platform/packages/shared/kbn-es-mappings/src/mappings.ts

@@ -13,6 +13,8 @@ import type {
   DateMapping,
   KeywordMapping,
   TextMapping,
+  MatchOnlyTextMapping,
+  SemanticTextMapping,
   DateNanosMapping,
   IntegerMapping,
   LongMapping,
@@ -36,6 +38,28 @@ export function object<T>(def: WithoutTypeField<ObjectMapping<T>>): ObjectMappin
   return merge(defaults, def);
 }
 
+export function matchOnlyText(def?: WithoutTypeField<MatchOnlyTextMapping>): MatchOnlyTextMapping {
+  const defaults: MatchOnlyTextMapping = omitUnsetKeys(
+    {
+      type: 'match_only_text',
+    },
+    def
+  );
+
+  return merge(defaults, def);
+}
+
+export function semanticText(def?: WithoutTypeField<SemanticTextMapping>): SemanticTextMapping {
+  const defaults: SemanticTextMapping = omitUnsetKeys(
+    {
+      type: 'semantic_text',
+    },
+    def
+  );
+
+  return merge(defaults, def);
+}
+
 export function text(def?: WithoutTypeField<TextMapping>): TextMapping {
   const defaults: TextMapping = omitUnsetKeys(
     {

src/platform/packages/shared/kbn-es-mappings/src/types.ts

@@ -24,6 +24,8 @@ export type StrictMappingTypeMapping = Strict<api.MappingTypeMapping>;
 export type AnyMapping = Strict<api.MappingProperty>;
 export type KeywordMapping = Strict<api.MappingKeywordProperty>;
 export type TextMapping = Strict<api.MappingTextProperty>;
+export type MatchOnlyTextMapping = Strict<api.MappingMatchOnlyTextProperty>;
+export type SemanticTextMapping = Strict<api.MappingSemanticTextProperty>;
 export type DateMapping = Strict<api.MappingDateProperty>;
 export type DateNanosMapping = Strict<api.MappingDateNanosProperty>;
 export type LongMapping = Strict<api.MappingLongNumberProperty>;
@@ -46,6 +48,8 @@ type AllMappingPropertyType = Required<api.MappingProperty>['type'];
 type SupportedMappingPropertyType = AllMappingPropertyType &
   (
     | 'text'
+    | 'match_only_text'
+    | 'semantic_text'
     | 'integer'
     | 'keyword'
     | 'boolean'
@@ -81,6 +85,10 @@ export type ToPrimitives<O extends { properties: Record<string, MappingProperty>
             : string
           : T extends 'text'
           ? string
+          : T extends 'match_only_text'
+          ? string
+          : T extends 'semantic_text'
+          ? string
           : T extends 'integer'
           ? number
           : T extends 'long'

x-pack/platform/packages/shared/kbn-streams-ai/src/knowledge_indicators/search.test.ts

@@ -10,17 +10,16 @@ import { searchKnowledgeIndicators } from './search';
 
 function makeFeature(overrides: Partial<Feature> = {}): Feature {
   return {
-    uuid: 'feature-uuid',
     id: 'feature-id',
     stream_name: 'logs.test',
     type: 'dataset_analysis',
     description: 'Feature description',
     properties: {},
     confidence: 90,
-    status: 'active',
-    last_seen: new Date().toISOString(),
+    deleted: false,
+    run_id: 'run-id',
     ...overrides,
-  };
+  } as Feature;
 }
 
 function makeStreamQuery(overrides: Partial<StreamQuery> = {}): StreamQuery {
@@ -46,9 +45,6 @@ describe('searchKnowledgeIndicators', () => {
           rule_backed: true,
           rule_id: 'rule-1',
           stream_name: 'logs.test',
-          'asset.uuid': 'asset-uuid',
-          'asset.type': 'query',
-          'asset.id': 'asset-id',
         },
       ],
     });
@@ -67,9 +63,6 @@ describe('searchKnowledgeIndicators', () => {
           rule_backed: false,
           rule_id: 'rule-1',
           stream_name: 'logs.test',
-          'asset.uuid': 'asset-uuid',
-          'asset.type': 'query',
-          'asset.id': 'asset-id',
         },
       ]
     );
@@ -142,27 +135,20 @@ describe('searchKnowledgeIndicators', () => {
         makeFeature({ id: 'f1', confidence: 10 }),
         makeFeature({ id: 'f2', confidence: 20 }),
       ],
-      getQueries: async (): Promise<QueryLink[]> =>
-        [
-          {
-            query: makeStreamQuery({ id: 'q1' }),
-            rule_backed: true,
-            rule_id: 'rule-1',
-            stream_name: 'logs.test',
-            'asset.uuid': 'asset-uuid',
-            'asset.type': 'query',
-            'asset.id': 'asset-id',
-          },
-          {
-            query: makeStreamQuery({ id: 'q2' }),
-            rule_backed: true,
-            rule_id: 'rule-2',
-            stream_name: 'logs.test',
-            'asset.uuid': 'asset-uuid',
-            'asset.type': 'query',
-            'asset.id': 'asset-id',
-          },
-        ] as QueryLink[],
+      getQueries: async (): Promise<QueryLink[]> => [
+        {
+          query: makeStreamQuery({ id: 'q1' }),
+          rule_backed: true,
+          rule_id: 'rule-1',
+          stream_name: 'logs.test',
+        },
+        {
+          query: makeStreamQuery({ id: 'q2' }),
+          rule_backed: true,
+          rule_id: 'rule-2',
+          stream_name: 'logs.test',
+        },
+      ],
     });
 
     expect(res.knowledge_indicators).toHaveLength(2);

x-pack/platform/packages/shared/kbn-streams-ai/workflows/partition_stream/feature_tool.test.ts

@@ -22,7 +22,6 @@ describe('partitionStream features_tool', () => {
   const sampleFeatures: Feature[] = [
     {
       id: 'feature-1',
-      uuid: 'uuid-1',
       stream_name: 'logs',
       type: 'technology',
       title: 'Node.js',
@@ -32,12 +31,11 @@ describe('partitionStream features_tool', () => {
       evidence: ['require statement', 'package.json reference'],
       tags: ['backend', 'javascript'],
       meta: { detected_at: '2024-01-01' },
-      status: 'active',
-      last_seen: '2024-01-15T00:00:00.000Z',
-    },
+      deleted: false,
+      run_id: 'run-1',
+    } as Feature,
     {
       id: 'feature-2',
-      uuid: 'uuid-2',
       stream_name: 'logs',
       type: 'entity',
       title: 'API Gateway',
@@ -47,9 +45,9 @@ describe('partitionStream features_tool', () => {
       evidence: ['HTTP routing patterns'],
       tags: ['api', 'gateway'],
       meta: {},
-      status: 'stale',
-      last_seen: '2024-01-14T00:00:00.000Z',
-    },
+      deleted: false,
+      run_id: 'run-1',
+    } as Feature,
   ];
 
   describe('toFeatureForLlmContext', () => {
@@ -76,7 +74,6 @@ describe('partitionStream features_tool', () => {
     it('should preserve nested properties objects', () => {
       const featureWithComplexProperties: Feature = {
         id: 'feature-3',
-        uuid: 'uuid-3',
         stream_name: 'logs',
         type: 'infrastructure',
         title: 'PostgreSQL',
@@ -92,9 +89,9 @@ describe('partitionStream features_tool', () => {
         evidence: ['connection string'],
         tags: ['database', 'sql'],
         meta: { tables: ['users', 'orders'] },
-        status: 'active',
-        last_seen: '2024-01-15T00:00:00.000Z',
-      };
+        deleted: false,
+        run_id: 'run-1',
+      } as Feature;
 
       const result = toFeatureForLlmContext(featureWithComplexProperties);
 

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

@@ -229,7 +229,6 @@ export {
   type BaseFeature,
   type IdentifiedFeature,
   type IgnoredFeature,
-  type FeatureStatus,
   DATASET_ANALYSIS_FEATURE_TYPE,
   LOG_SAMPLES_FEATURE_TYPE,
   LOG_PATTERNS_FEATURE_TYPE,
@@ -247,7 +246,6 @@ export {
   baseFeatureSchema,
   identifiedFeatureSchema,
   ignoredFeatureSchema,
-  featureStatusSchema,
 } from './src/feature';
 
 export { FeatureAccumulator } from './src/feature_accumulator';
@@ -301,6 +299,7 @@ export {
   type Verdict,
   sigEventSchema,
   type SigEvent,
+  type KnowledgeIndicator,
 } from './src/sig_events';
 export type { OnboardingResult } from './src/onboarding';
 export { OnboardingStep } from './src/onboarding';

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

@@ -9,10 +9,6 @@ import { z } from '@kbn/zod/v4';
 import { isEqual, uniq } from 'lodash';
 import { conditionSchema, type Condition } from '@kbn/streamlang';
 
-const featureStatus = ['active', 'stale', 'expired'] as const;
-export const featureStatusSchema = z.enum(featureStatus);
-export type FeatureStatus = z.infer<typeof featureStatusSchema>;
-
 export const DATASET_ANALYSIS_FEATURE_TYPE = 'dataset_analysis' as const;
 export const LOG_SAMPLES_FEATURE_TYPE = 'log_samples' as const;
 export const LOG_PATTERNS_FEATURE_TYPE = 'log_patterns' as const;
@@ -74,14 +70,26 @@ export const ignoredFeatureSchema = z.object({
 
 export type IgnoredFeature = z.infer<typeof ignoredFeatureSchema>;
 
+/**
+ * Server-side feature shape on the unified knowledge indicators data stream.
+ *
+ * Note: as part of the unified KI data stream migration, the legacy
+ * `uuid`, `status`, `last_seen`, and `expires_at` fields have been removed.
+ * Identity is now `(stream.name, type, id)` and revisions are append-only.
+ * `updated_at` is read-only at the domain layer — it is derived from the
+ * latest revision's `@timestamp` when reading and is not a property of the
+ * write payload (`BaseFeature`).
+ *
+ * `excluded` is a root-level marker on the storage doc surfaced here so the
+ * UI can split active vs. excluded features when the caller opts into
+ * `include_excluded=true`. Soft deletes use tombstone revisions (`deleted:
+ * true` on identity-only docs), not this field.
+ */
 export const featureSchema = baseFeatureSchema.and(
   z.object({
-    uuid: z.string(),
-    status: featureStatusSchema,
-    last_seen: z.string(),
-    expires_at: z.string().optional(),
-    excluded_at: z.string().optional(),
     run_id: z.string().optional(),
+    excluded: z.boolean().optional(),
+    updated_at: z.string().optional(),
   })
 );
 

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

@@ -7,28 +7,33 @@
 
 import { isDuplicateFeature, type BaseFeature, type Feature } from './feature';
 
+/**
+ * Identity key used by the accumulator. With the unified KI data stream,
+ * features are identified by `(stream.name, id)`; within a single stream
+ * the `id` is unique, so the accumulator keys on `id` directly.
+ */
 export class FeatureAccumulator {
-  private readonly byUuid = new Map<string, Feature>();
+  private readonly byId = new Map<string, Feature>();
   private readonly byLowerId = new Map<string, Feature>();
   private readonly fromStorage = new Set<string>();
 
   constructor(initialFeatures: Feature[] = []) {
     for (const f of initialFeatures) {
       this.add(f);
-      this.fromStorage.add(f.uuid);
+      this.fromStorage.add(f.id);
     }
   }
 
   add(feature: Feature) {
-    this.byUuid.set(feature.uuid, feature);
+    this.byId.set(feature.id, feature);
     this.byLowerId.set(feature.id.toLowerCase(), feature);
   }
 
   update(feature: Feature) {
-    if (!this.byUuid.has(feature.uuid)) {
+    if (!this.byId.has(feature.id)) {
       return;
     }
-    this.byUuid.set(feature.uuid, feature);
+    this.byId.set(feature.id, feature);
     this.byLowerId.set(feature.id.toLowerCase(), feature);
   }
 
@@ -40,19 +45,19 @@ export class FeatureAccumulator {
   }
 
   isStoredFeature(feature: Feature): boolean {
-    return this.fromStorage.has(feature.uuid);
+    return this.fromStorage.has(feature.id);
   }
 
-  promoteFromStorage(featureUuid: string) {
-    this.fromStorage.delete(featureUuid);
+  promoteFromStorage(featureId: string) {
+    this.fromStorage.delete(featureId);
   }
 
   getAll(): Feature[] {
-    return Array.from(this.byUuid.values());
+    return Array.from(this.byId.values());
   }
 
   getDiscovered(): Feature[] {
-    return this.getAll().filter((f) => !this.fromStorage.has(f.uuid));
+    return this.getAll().filter((f) => !this.fromStorage.has(f.id));
   }
 
   getTopRanked(limit: number): Feature[] {
@@ -67,6 +72,6 @@ export class FeatureAccumulator {
   }
 
   public get length(): number {
-    return this.byUuid.size;
+    return this.byId.size;
   }
 }

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

@@ -9,6 +9,7 @@ import { z } from '@kbn/zod/v4';
 import type { QueryDslQueryContainer } from '@elastic/elasticsearch/lib/api/types';
 import { NonEmptyString } from '@kbn/zod-helpers/v4';
 import { primitive } from '../shared/record_types';
+import type { Feature } from '../feature';
 import type { SignificantEventsResponse } from '../api/significant_events';
 
 export interface EsqlQuery {
@@ -115,14 +116,33 @@ export interface QueriesOccurrencesGetResponse {
   total_occurrences: number;
 }
 
+/**
+ * Wire shape for a query knowledge indicator. Identity is `(stream_name, query.id)`.
+ *
+ * Note: as part of the unified KI data stream migration, the legacy
+ * `asset.uuid` / `asset.id` / `asset.type` fields have been removed —
+ * the document `_id` is server-generated per revision and not surfaced.
+ */
 export interface QueryLink {
-  'asset.uuid': string;
-  'asset.type': 'query';
-  'asset.id': string;
   query: StreamQuery;
   stream_name: string;
   /** Whether a Kibana rule exists for this query. */
   rule_backed: boolean;
   /** The deterministic ID of the Kibana rule associated with this query. */
   rule_id: string;
+  /**
+   * ISO timestamp of the latest revision in storage. Bumped by every write,
+   * including no-op `syncQueries` reconciliation passes. Read-only at the
+   * domain layer.
+   */
+  updated_at?: string;
 }
+
+/**
+ * Unified knowledge indicator on the wire. Server callers use this when they
+ * need to handle both feature and query indicators uniformly. The discriminator
+ * is the root `type` field.
+ */
+export type KnowledgeIndicator =
+  | { type: 'feature'; feature: Feature }
+  | { type: 'query'; query: QueryLink };

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

@@ -9,3 +9,4 @@ export { detectionSchema, type Detection } from './detections';
 export { discoverySchema, type Discovery } from './discoveries';
 export { verdictSchema, type Verdict } from './verdicts';
 export { sigEventSchema, type SigEvent } from './events';
+export type { KnowledgeIndicator } from '../queries';