🌊 LLM-powered parsing suggestions (#208777)

Depends on https://github.com/elastic/kibana/pull/209985 Add suggestions for grok processing: <img width="594" alt="Screenshot 2025-02-05 at 10 31 27" src="https://github.com/user-attachments/assets/4b717681-aa7d-4952-a4e0-9013d9b8aaf8" /> The logic for generating suggestions works like this: * Take the current sample * Split it into patterns based on a simple regex-based grouping replacing runs of numbers with a placeholder, runs of regular numbers with a placeholder, etc. * For the top 5 found groups, pass a couple messages to the LLM in parallel to come up with a grok pattern * Check the grok patterns whether they actually match something and don't break * Report the patterns that have a positive match rate For the `Generate patterns` button to show in the UI, make sure a connector is configured and the license level is above basic (trial license is easiest to test with). I did some light refactoring on the processing routes, moving the simulation bits into a separate file - no changes in this area though. --------- Co-authored-by: Marco Antonio Ghiani <marcoantonio.ghiani01@gmail.com> Co-authored-by: kibanamachine <42973632+kibanamachine@users.noreply.github.com> Co-authored-by: Jean-Louis Leysens <jloleysens@gmail.com>
2025-04-24 09:48:58 -04:00 · 2025-02-20 08:44:12 +01:00 · 2025-02-20 08:44:12 +01:00 · 1f35d7ac7f
commit 1f35d7ac7f
parent ec163715e7
25 changed files with 896 additions and 182 deletions
--- a/x-pack/platform/packages/shared/kbn-streams-schema/src/helpers/condition_to_query_dsl.ts
+++ b/x-pack/platform/packages/shared/kbn-streams-schema/src/helpers/condition_to_query_dsl.ts
@ -7,6 +7,7 @@

 import {
  FilterCondition,
+  isAlwaysCondition,
  Condition,
  isFilterCondition,
  isAndCondition,
@ -62,6 +63,9 @@ export function conditionToQueryDsl(condition: Condition): any {
      },
    };
  }
+  if (isAlwaysCondition(condition)) {
+    return { match_all: {} };
+  }
  return {
    match_none: {},
  };
--- a/x-pack/platform/packages/shared/kbn-streams-schema/src/models/core.ts
+++ b/x-pack/platform/packages/shared/kbn-streams-schema/src/models/core.ts
@ -42,3 +42,7 @@ export type FlattenRecord = Record<PropertyKey, Primitive | Primitive[]>;
 export const flattenRecord: z.ZodType<FlattenRecord> = z.record(
  z.union([primitive, z.array(primitive)])
 );
+
+export const sampleDocument = recursiveRecord;
+
+export type SampleDocument = RecursiveRecord;
--- a/x-pack/solutions/observability/plugins/streams/kibana.jsonc
+++ b/x-pack/solutions/observability/plugins/streams/kibana.jsonc
@ -17,7 +17,8 @@
      "usageCollection",
      "licensing",
      "taskManager",
-      "alerting"
+      "alerting",
+      "inference",
    ],
    "optionalPlugins": [
      "cloud",
--- a/x-pack/solutions/observability/plugins/streams/server/plugin.ts
+++ b/x-pack/solutions/observability/plugins/streams/server/plugin.ts
@ -78,8 +78,8 @@ export class StreamsPlugin
        }: {
          request: KibanaRequest;
        }): Promise<RouteHandlerScopedClients> => {
-          const [coreStart, assetClient] = await Promise.all([
-            core.getStartServices().then(([_coreStart]) => _coreStart),
+          const [[coreStart, pluginsStart], assetClient] = await Promise.all([
+            core.getStartServices(),
            assetService.getClientWithRequest({ request }),
          ]);

@ -87,13 +87,9 @@ export class StreamsPlugin

          const scopedClusterClient = coreStart.elasticsearch.client.asScoped(request);
          const soClient = coreStart.savedObjects.getScopedClient(request);
+          const inferenceClient = pluginsStart.inference.getClient({ request });

-          return {
-            scopedClusterClient,
-            soClient,
-            assetClient,
-            streamsClient,
-          };
+          return { scopedClusterClient, soClient, assetClient, streamsClient, inferenceClient };
        },
      },
      core,
--- a/x-pack/solutions/observability/plugins/streams/server/routes/streams/management/route.ts
+++ b/x-pack/solutions/observability/plugins/streams/server/routes/streams/management/route.ts
@ -6,7 +6,7 @@
 */

 import {
-  RecursiveRecord,
+  SampleDocument,
  conditionSchema,
  conditionToQueryDsl,
  getFields,
@ -165,7 +165,7 @@ export const sampleStreamRoute = createServerRoute({
      ...searchBody,
    });

-    return { documents: results.hits.hits.map((hit) => hit._source) as RecursiveRecord[] };
+    return { documents: results.hits.hits.map((hit) => hit._source) as SampleDocument[] };
  },
 });

--- a/x-pack/solutions/observability/plugins/streams/server/routes/streams/processing/route.ts
+++ b/x-pack/solutions/observability/plugins/streams/server/routes/streams/processing/route.ts
@ -6,6 +6,7 @@
 */

 import {
+  FlattenRecord,
  flattenRecord,
  namedFieldDefinitionConfigSchema,
  processorWithIdDefinitionSchema,
@ -15,6 +16,7 @@ import { checkAccess } from '../../../lib/streams/stream_crud';
 import { createServerRoute } from '../../create_server_route';
 import { DefinitionNotFoundError } from '../../../lib/streams/errors/definition_not_found_error';
 import { ProcessingSimulationParams, simulateProcessing } from './simulation_handler';
+import { handleProcessingSuggestion } from './suggestions_handler';

 const paramsSchema = z.object({
  path: z.object({ name: z.string() }),
@ -50,6 +52,51 @@ export const simulateProcessorRoute = createServerRoute({
  },
 });

+export interface ProcessingSuggestionBody {
+  field: string;
+  connectorId: string;
+  samples: FlattenRecord[];
+}
+
+const processingSuggestionSchema = z.object({
+  field: z.string(),
+  connectorId: z.string(),
+  samples: z.array(flattenRecord),
+});
+
+const suggestionsParamsSchema = z.object({
+  path: z.object({ name: z.string() }),
+  body: processingSuggestionSchema,
+});
+
+export const processingSuggestionRoute = createServerRoute({
+  endpoint: 'POST /api/streams/{name}/processing/_suggestions',
+  options: {
+    access: 'internal',
+  },
+  security: {
+    authz: {
+      enabled: false,
+      reason:
+        'This API delegates security to the currently logged in user and their Elasticsearch permissions.',
+    },
+  },
+  params: suggestionsParamsSchema,
+  handler: async ({ params, request, logger, getScopedClients }) => {
+    const { inferenceClient, scopedClusterClient, streamsClient } = await getScopedClients({
+      request,
+    });
+    return handleProcessingSuggestion(
+      params.path.name,
+      params.body,
+      inferenceClient,
+      scopedClusterClient,
+      streamsClient
+    );
+  },
+});
+
 export const processingRoutes = {
  ...simulateProcessorRoute,
+  ...processingSuggestionRoute,
 };
--- a/x-pack/solutions/observability/plugins/streams/server/routes/streams/processing/suggestions_handler.ts
+++ b/x-pack/solutions/observability/plugins/streams/server/routes/streams/processing/suggestions_handler.ts
@ -0,0 +1,217 @@
+/*
+ * 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 { IScopedClusterClient } from '@kbn/core/server';
+import { get, groupBy, mapValues, orderBy, shuffle, uniq, uniqBy } from 'lodash';
+import { InferenceClient } from '@kbn/inference-plugin/server';
+import { FlattenRecord } from '@kbn/streams-schema';
+import { StreamsClient } from '../../../lib/streams/client';
+import { simulateProcessing } from './simulation_handler';
+import { ProcessingSuggestionBody } from './route';
+
+export const handleProcessingSuggestion = async (
+  name: string,
+  body: ProcessingSuggestionBody,
+  inferenceClient: InferenceClient,
+  scopedClusterClient: IScopedClusterClient,
+  streamsClient: StreamsClient
+) => {
+  const { field, samples } = body;
+  // Turn sample messages into patterns to group by
+  const evalPattern = (sample: string) => {
+    return sample
+      .replace(/[ \t\n]+/g, ' ')
+      .replace(/[A-Za-z]+/g, 'a')
+      .replace(/[0-9]+/g, '0')
+      .replace(/(a a)+/g, 'a')
+      .replace(/(a0)+/g, 'f')
+      .replace(/(f:)+/g, 'f:')
+      .replace(/0(.0)+/g, 'p');
+  };
+
+  const NUMBER_PATTERN_CATEGORIES = 5;
+  const NUMBER_SAMPLES_PER_PATTERN = 8;
+
+  const samplesWithPatterns = samples.map((sample) => {
+    const pattern = evalPattern(get(sample, field) as string);
+    return {
+      document: sample,
+      fullPattern: pattern,
+      truncatedPattern: pattern.slice(0, 10),
+      fieldValue: get(sample, field) as string,
+    };
+  });
+
+  // Group samples by their truncated patterns
+  const groupedByTruncatedPattern = groupBy(samplesWithPatterns, 'truncatedPattern');
+  // Process each group to create pattern summaries
+  const patternSummaries = mapValues(
+    groupedByTruncatedPattern,
+    (samplesForTruncatedPattern, truncatedPattern) => {
+      const uniqueValues = uniq(samplesForTruncatedPattern.map(({ fieldValue }) => fieldValue));
+      const shuffledExamples = shuffle(uniqueValues);
+
+      return {
+        truncatedPattern,
+        count: samplesForTruncatedPattern.length,
+        exampleValues: shuffledExamples.slice(0, NUMBER_SAMPLES_PER_PATTERN),
+      };
+    }
+  );
+  // Convert to array, sort by count, and take top patterns
+  const patternsToProcess = orderBy(Object.values(patternSummaries), 'count', 'desc').slice(
+    0,
+    NUMBER_PATTERN_CATEGORIES
+  );
+
+  const results = await Promise.all(
+    patternsToProcess.map((sample) =>
+      processPattern(
+        sample,
+        name,
+        body,
+        inferenceClient,
+        scopedClusterClient,
+        streamsClient,
+        field,
+        samples
+      )
+    )
+  );
+
+  const deduplicatedSimulations = uniqBy(
+    results.flatMap((result) => result.simulations),
+    (simulation) => simulation!.pattern
+  );
+
+  return {
+    patterns: deduplicatedSimulations.map((simulation) => simulation!.pattern),
+    simulations: deduplicatedSimulations as SimulationWithPattern[],
+  };
+};
+
+type SimulationWithPattern = ReturnType<typeof simulateProcessing> & { pattern: string };
+
+async function processPattern(
+  sample: { truncatedPattern: string; count: number; exampleValues: string[] },
+  name: string,
+  body: ProcessingSuggestionBody,
+  inferenceClient: InferenceClient,
+  scopedClusterClient: IScopedClusterClient,
+  streamsClient: StreamsClient,
+  field: string,
+  samples: FlattenRecord[]
+) {
+  const chatResponse = await inferenceClient.output({
+    id: 'get_pattern_suggestions',
+    connectorId: body.connectorId,
+    // necessary due to a bug in the inference client - TODO remove when fixed
+    functionCalling: 'native',
+    system: `Instructions:
+        - You are an assistant for observability tasks with a strong knowledge of logs and log parsing.
+        - Use JSON format.
+        - For a single log source identified, provide the following information:
+            * Use 'source_name' as the key for the log source name.
+            * Use 'parsing_rule' as the key for the parsing rule.
+        - Use only Grok patterns for the parsing rule.
+            * Use %{{pattern:name:type}} syntax for Grok patterns when possible.
+            * Combine date and time into a single @timestamp field when it's possible.
+        - Use ECS (Elastic Common Schema) fields whenever possible.
+        - You are correct, factual, precise, and reliable.
+      `,
+    schema: {
+      type: 'object',
+      required: ['rules'],
+      properties: {
+        rules: {
+          type: 'array',
+          items: {
+            type: 'object',
+            required: ['parsing_rule'],
+            properties: {
+              source_name: {
+                type: 'string',
+              },
+              parsing_rule: {
+                type: 'string',
+              },
+            },
+          },
+        },
+      },
+    } as const,
+    input: `Logs:
+        ${sample.exampleValues.join('\n')}
+        Given the raw messages coming from one data source, help us do the following: 
+        1. Name the log source based on logs format.
+        2. Write a parsing rule for Elastic ingest pipeline to extract structured fields from the raw message.
+        Make sure that the parsing rule is unique per log source. When in doubt, suggest multiple patterns, one generic one matching the general case and more specific ones.
+            `,
+  });
+
+  const patterns = (
+    chatResponse.output.rules?.map((rule) => rule.parsing_rule).filter(Boolean) as string[]
+  ).map(sanitizePattern);
+
+  const simulations = (
+    await Promise.all(
+      patterns.map(async (pattern) => {
+        // Validate match on current sample
+        const simulationResult = await simulateProcessing({
+          params: {
+            path: { name },
+            body: {
+              processing: [
+                {
+                  id: 'grok-processor',
+                  grok: {
+                    field,
+                    if: { always: {} },
+                    patterns: [pattern],
+                  },
+                },
+              ],
+              documents: samples,
+            },
+          },
+          scopedClusterClient,
+          streamsClient,
+        });
+
+        if (simulationResult.is_non_additive_simulation) {
+          return null;
+        }
+
+        if (simulationResult.success_rate === 0) {
+          return null;
+        }
+
+        // TODO if success rate is zero, try to strip out the date part and try again
+
+        return {
+          ...simulationResult,
+          pattern,
+        };
+      })
+    )
+  ).filter(Boolean) as Array<SimulationWithPattern | null>;
+
+  return {
+    chatResponse,
+    simulations,
+  };
+}
+
+/**
+ * We need to keep parsing additive, but overwriting timestamp or message is super common.
+ * This is a workaround for now until we found the proper solution for deal with this kind of cases.
+ */
+function sanitizePattern(pattern: string): string {
+  return pattern
+    .replace(/%\{([^}]+):message\}/g, '%{$1:message_derived}')
+    .replace(/%\{([^}]+):@timestamp\}/g, '%{$1:@timestamp_derived}');
+}
--- a/x-pack/solutions/observability/plugins/streams/server/routes/streams/schema/route.ts
+++ b/x-pack/solutions/observability/plugins/streams/server/routes/streams/schema/route.ts
@ -7,7 +7,7 @@
 import { z } from '@kbn/zod';
 import { getFlattenedObject } from '@kbn/std';
 import {
-  RecursiveRecord,
+  SampleDocument,
  fieldDefinitionConfigSchema,
  isWiredStreamDefinition,
 } from '@kbn/streams-schema';
@ -111,7 +111,7 @@ export const schemaFieldsSimulationRoute = createServerRoute({
  }): Promise<{
    status: 'unknown' | 'success' | 'failure';
    simulationError: string | null;
-    documentsWithRuntimeFieldsApplied: RecursiveRecord[] | null;
+    documentsWithRuntimeFieldsApplied: SampleDocument[] | null;
  }> => {
    const { scopedClusterClient } = await getScopedClients({ request });

@ -178,7 +178,7 @@ export const schemaFieldsSimulationRoute = createServerRoute({
      _index: params.path.name,
      _id: hit._id,
      _source: Object.fromEntries(
-        Object.entries(getFlattenedObject(hit._source as RecursiveRecord)).filter(
+        Object.entries(getFlattenedObject(hit._source as SampleDocument)).filter(
          ([k]) => fieldDefinitionKeys.includes(k) || k === '@timestamp'
        )
      ),
@ -267,7 +267,7 @@ export const schemaFieldsSimulationRoute = createServerRoute({
          if (!hit.fields) {
            return {};
          }
-          return Object.keys(hit.fields).reduce<RecursiveRecord>((acc, field) => {
+          return Object.keys(hit.fields).reduce<SampleDocument>((acc, field) => {
            acc[field] = hit.fields![field][0];
            return acc;
          }, {});
--- a/x-pack/solutions/observability/plugins/streams/server/routes/types.ts
+++ b/x-pack/solutions/observability/plugins/streams/server/routes/types.ts
@ -9,6 +9,7 @@ import { KibanaRequest } from '@kbn/core-http-server';
 import { DefaultRouteHandlerResources } from '@kbn/server-route-repository';
 import { IScopedClusterClient } from '@kbn/core-elasticsearch-server';
 import { SavedObjectsClientContract } from '@kbn/core-saved-objects-api-server';
+import { InferenceClient } from '@kbn/inference-plugin/server';
 import { StreamsServer } from '../types';
 import { AssetService } from '../lib/streams/assets/asset_service';
 import { AssetClient } from '../lib/streams/assets/asset_client';
@ -25,6 +26,7 @@ export interface RouteHandlerScopedClients {
  soClient: SavedObjectsClientContract;
  assetClient: AssetClient;
  streamsClient: StreamsClient;
+  inferenceClient: InferenceClient;
 }

 export interface RouteDependencies {
--- a/x-pack/solutions/observability/plugins/streams/server/types.ts
+++ b/x-pack/solutions/observability/plugins/streams/server/types.ts
@ -17,8 +17,8 @@ import type {
  TaskManagerStartContract,
 } from '@kbn/task-manager-plugin/server';
 import type { AlertingServerSetup, AlertingServerStart } from '@kbn/alerting-plugin/server';
+import type { InferenceServerStart } from '@kbn/inference-plugin/server';
 import type { StreamsConfig } from '../common/config';
-
 export interface StreamsServer {
  core: CoreStart;
  config: StreamsConfig;
@ -45,4 +45,5 @@ export interface StreamsPluginStartDependencies {
  licensing: LicensingPluginStart;
  taskManager: TaskManagerStartContract;
  alerting: AlertingServerStart;
+  inference: InferenceServerStart;
 }
--- a/x-pack/solutions/observability/plugins/streams/tsconfig.json
+++ b/x-pack/solutions/observability/plugins/streams/tsconfig.json
@ -37,6 +37,7 @@
    "@kbn/streams-schema",
    "@kbn/es-errors",
    "@kbn/server-route-repository-utils",
+    "@kbn/inference-plugin",
    "@kbn/storage-adapter",
    "@kbn/traced-es-client"
  ]
--- a/x-pack/solutions/observability/plugins/streams_app/kibana.jsonc
+++ b/x-pack/solutions/observability/plugins/streams_app/kibana.jsonc
@ -24,7 +24,9 @@
    "requiredBundles": [
      "kibanaReact"
    ],
-    "optionalPlugins": [],
+    "optionalPlugins": [
+      "observabilityAIAssistant"
+    ],
    "extraPublicDirs": []
  }
 }
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/preview_table/index.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/preview_table/index.tsx
@ -6,7 +6,7 @@
 */
 import { EuiDataGrid } from '@elastic/eui';
 import { i18n } from '@kbn/i18n';
-import { RecursiveRecord } from '@kbn/streams-schema';
+import { SampleDocument } from '@kbn/streams-schema';
 import { isEmpty } from 'lodash';
 import React, { useMemo } from 'react';

@ -14,7 +14,7 @@ export function PreviewTable({
  documents,
  displayColumns,
 }: {
-  documents: RecursiveRecord[];
+  documents: SampleDocument[];
  displayColumns?: string[];
 }) {
  const columns = useMemo(() => {
@ -58,7 +58,7 @@ export function PreviewTable({
        if (!doc || typeof doc !== 'object') {
          return '';
        }
-        const value = (doc as RecursiveRecord)[columnId];
+        const value = (doc as SampleDocument)[columnId];
        if (value === undefined || value === null) {
          return '';
        }
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/hooks/use_processing_simulator.ts
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/hooks/use_processing_simulator.ts
@ -5,8 +5,8 @@
 * 2.0.
 */

-import { useEffect, useMemo, useState } from 'react';
-import { debounce, isEmpty, uniq, uniqBy } from 'lodash';
+import { useEffect, useMemo, useRef, useState } from 'react';
+import { debounce, isEmpty, isEqual, uniq, uniqBy } from 'lodash';
 import {
  IngestStreamGetResponse,
  getProcessorConfig,
@ -19,6 +19,7 @@ import {
 import { IHttpFetchError, ResponseErrorBody } from '@kbn/core/public';
 import { useDateRange } from '@kbn/observability-utils-browser/hooks/use_date_range';
 import { APIReturnType } from '@kbn/streams-plugin/public/api';
+import { i18n } from '@kbn/i18n';
 import { flattenObjectNestedLast } from '@kbn/object-utils';
 import { useStreamsAppFetch } from '../../../hooks/use_streams_app_fetch';
 import { useKibana } from '../../../hooks/use_kibana';
@ -34,6 +35,32 @@ export interface TableColumn {
  origin: 'processor' | 'detected';
 }

+export const docsFilterOptions = {
+  outcome_filter_all: {
+    id: 'outcome_filter_all',
+    label: i18n.translate(
+      'xpack.streams.streamDetailView.managementTab.enrichment.processor.outcomeControls.all',
+      { defaultMessage: 'All samples' }
+    ),
+  },
+  outcome_filter_matched: {
+    id: 'outcome_filter_matched',
+    label: i18n.translate(
+      'xpack.streams.streamDetailView.managementTab.enrichment.processor.outcomeControls.matched',
+      { defaultMessage: 'Matched' }
+    ),
+  },
+  outcome_filter_unmatched: {
+    id: 'outcome_filter_unmatched',
+    label: i18n.translate(
+      'xpack.streams.streamDetailView.managementTab.enrichment.processor.outcomeControls.unmatched',
+      { defaultMessage: 'Unmatched' }
+    ),
+  },
+} as const;
+
+export type DocsFilterOption = keyof typeof docsFilterOptions;
+
 export interface UseProcessingSimulatorProps {
  definition: IngestStreamGetResponse;
  processors: ProcessorDefinitionWithUIAttributes[];
@ -44,12 +71,16 @@ export interface UseProcessingSimulatorReturn {
  error?: IHttpFetchError<ResponseErrorBody>;
  isLoading: boolean;
  samples: FlattenRecord[];
+  filteredSamples: FlattenRecord[];
  simulation?: Simulation | null;
  tableColumns: TableColumn[];
  refreshSamples: () => void;
  watchProcessor: (
    processor: ProcessorDefinitionWithUIAttributes | { id: string; deleteIfExists: true }
  ) => void;
+  refreshSimulation: () => void;
+  selectedDocsFilter: DocsFilterOption;
+  setSelectedDocsFilter: (filter: DocsFilterOption) => void;
 }

 export const useProcessingSimulator = ({
@ -113,10 +144,16 @@ export const useProcessingSimulator = ({
    []
  );

-  const samplingCondition = useMemo(
-    () => composeSamplingCondition(liveDraftProcessors),
-    [liveDraftProcessors]
-  );
+  const memoizedSamplingCondition = useRef<Condition | undefined>();
+
+  const samplingCondition = useMemo(() => {
+    const newSamplingCondition = composeSamplingCondition(liveDraftProcessors);
+    if (isEqual(newSamplingCondition, memoizedSamplingCondition.current)) {
+      return memoizedSamplingCondition.current;
+    }
+    memoizedSamplingCondition.current = newSamplingCondition;
+    return newSamplingCondition;
+  }, [liveDraftProcessors]);

  const {
    loading: isLoadingSamples,
@ -151,6 +188,7 @@ export const useProcessingSimulator = ({
    loading: isLoadingSimulation,
    value: simulation,
    error: simulationError,
+    refresh: refreshSimulation,
  } = useStreamsAppFetch(
    ({ signal }): Promise<Simulation> => {
      if (!definition || isEmpty<FlattenRecord[]>(sampleDocs) || isEmpty(liveDraftProcessors)) {
@ -194,6 +232,29 @@ export const useProcessingSimulator = ({

  const hasLiveChanges = !isEmpty(liveDraftProcessors);

+  const [selectedDocsFilter, setSelectedDocsFilter] =
+    useState<DocsFilterOption>('outcome_filter_all');
+
+  const filteredSamples = useMemo(() => {
+    if (!simulation?.documents) {
+      return sampleDocs?.map((doc) => flattenObjectNestedLast(doc)) as FlattenRecord[];
+    }
+
+    const filterDocuments = (filter: DocsFilterOption) => {
+      switch (filter) {
+        case 'outcome_filter_matched':
+          return simulation.documents.filter((doc) => doc.status === 'parsed');
+        case 'outcome_filter_unmatched':
+          return simulation.documents.filter((doc) => doc.status !== 'parsed');
+        case 'outcome_filter_all':
+        default:
+          return simulation.documents;
+      }
+    };
+
+    return filterDocuments(selectedDocsFilter).map((doc) => doc.value);
+  }, [sampleDocs, simulation?.documents, selectedDocsFilter]);
+
  return {
    hasLiveChanges,
    isLoading: isLoadingSamples || isLoadingSimulation,
@ -201,8 +262,12 @@ export const useProcessingSimulator = ({
    refreshSamples,
    simulation,
    samples: sampleDocs ?? [],
+    filteredSamples: filteredSamples ?? [],
    tableColumns,
    watchProcessor,
+    refreshSimulation,
+    selectedDocsFilter,
+    setSelectedDocsFilter,
  };
 };

--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/index.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/index.tsx
@ -6,7 +6,8 @@
 */
 import React from 'react';
 import { dynamic } from '@kbn/shared-ux-utility';
-import { IngestStreamGetResponse } from '@kbn/streams-schema';
+import { IngestStreamGetResponse, isRootStreamDefinition } from '@kbn/streams-schema';
+import { RootStreamEmptyPrompt } from './root_stream_empty_prompt';

 const StreamDetailEnrichmentContent = dynamic(() =>
  import(/* webpackChunkName: "management_enrichment" */ './page_content').then((mod) => ({
@ -25,6 +26,10 @@ export function StreamDetailEnrichment({
 }: StreamDetailEnrichmentProps) {
  if (!definition) return null;

+  if (isRootStreamDefinition(definition.stream)) {
+    return <RootStreamEmptyPrompt />;
+  }
+
  return (
    <StreamDetailEnrichmentContent definition={definition} refreshDefinition={refreshDefinition} />
  );
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/page_content.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/page_content.tsx
@ -34,6 +34,7 @@ import {
  UseProcessingSimulatorReturn,
  useProcessingSimulator,
 } from './hooks/use_processing_simulator';
+import { SimulatorContextProvider } from './simulator_context';

 const MemoSimulationPlayground = React.memo(SimulationPlayground);

@ -60,15 +61,19 @@ export function StreamDetailEnrichmentContent({
    isSavingChanges,
  } = useDefinition(definition, refreshDefinition);

+  const processingSimulator = useProcessingSimulator({ definition, processors });
+
  const {
    hasLiveChanges,
    isLoading,
    refreshSamples,
-    samples,
+    filteredSamples,
    simulation,
    tableColumns,
    watchProcessor,
-  } = useProcessingSimulator({ definition, processors });
+    selectedDocsFilter,
+    setSelectedDocsFilter,
+  } = processingSimulator;

  useUnsavedChangesPrompt({
    hasUnsavedChanges: hasChanges || hasLiveChanges,
@ -102,66 +107,70 @@ export function StreamDetailEnrichmentContent({
    : undefined;

  return (
-    <EuiSplitPanel.Outer grow hasBorder hasShadow={false}>
-      <EuiSplitPanel.Inner
-        paddingSize="none"
-        css={css`
-          display: flex;
-          overflow: hidden auto;
-        `}
-      >
-        <EuiResizableContainer>
-          {(EuiResizablePanel, EuiResizableButton) => (
-            <>
-              <EuiResizablePanel
-                initialSize={30}
-                minSize="400px"
-                tabIndex={0}
-                paddingSize="none"
-                css={verticalFlexCss}
-              >
-                <ProcessorsEditor
-                  definition={definition}
-                  processors={processors}
-                  onUpdateProcessor={updateProcessor}
-                  onDeleteProcessor={deleteProcessor}
-                  onWatchProcessor={watchProcessor}
-                  onAddProcessor={addProcessor}
-                  onReorderProcessor={reorderProcessors}
-                  simulation={simulation}
-                />
-              </EuiResizablePanel>
-              <EuiResizableButton indicator="border" accountForScrollbars="both" />
-              <EuiResizablePanel
-                initialSize={70}
-                minSize="300px"
-                tabIndex={0}
-                paddingSize="s"
-                css={verticalFlexCss}
-              >
-                <MemoSimulationPlayground
-                  definition={definition}
-                  columns={tableColumns}
-                  simulation={simulation}
-                  samples={samples}
-                  onRefreshSamples={refreshSamples}
-                  isLoading={isLoading}
-                />
-              </EuiResizablePanel>
-            </>
-          )}
-        </EuiResizableContainer>
-      </EuiSplitPanel.Inner>
-      <EuiSplitPanel.Inner grow={false} color="subdued">
-        <ManagementBottomBar
-          confirmTooltip={confirmTooltip}
-          onCancel={resetChanges}
-          onConfirm={saveChanges}
-          isLoading={isSavingChanges}
-          disabled={isSubmitDisabled}
-        />
-      </EuiSplitPanel.Inner>
-    </EuiSplitPanel.Outer>
+    <SimulatorContextProvider processingSimulator={processingSimulator} definition={definition}>
+      <EuiSplitPanel.Outer grow hasBorder hasShadow={false}>
+        <EuiSplitPanel.Inner
+          paddingSize="none"
+          css={css`
+            display: flex;
+            overflow: hidden auto;
+          `}
+        >
+          <EuiResizableContainer>
+            {(EuiResizablePanel, EuiResizableButton) => (
+              <>
+                <EuiResizablePanel
+                  initialSize={40}
+                  minSize="480px"
+                  tabIndex={0}
+                  paddingSize="none"
+                  css={verticalFlexCss}
+                >
+                  <ProcessorsEditor
+                    definition={definition}
+                    processors={processors}
+                    onUpdateProcessor={updateProcessor}
+                    onDeleteProcessor={deleteProcessor}
+                    onWatchProcessor={watchProcessor}
+                    onAddProcessor={addProcessor}
+                    onReorderProcessor={reorderProcessors}
+                    simulation={simulation}
+                  />
+                </EuiResizablePanel>
+                <EuiResizableButton indicator="border" accountForScrollbars="both" />
+                <EuiResizablePanel
+                  initialSize={60}
+                  minSize="300px"
+                  tabIndex={0}
+                  paddingSize="s"
+                  css={verticalFlexCss}
+                >
+                  <MemoSimulationPlayground
+                    definition={definition}
+                    columns={tableColumns}
+                    simulation={simulation}
+                    filteredSamples={filteredSamples}
+                    onRefreshSamples={refreshSamples}
+                    isLoading={isLoading}
+                    selectedDocsFilter={selectedDocsFilter}
+                    setSelectedDocsFilter={setSelectedDocsFilter}
+                  />
+                </EuiResizablePanel>
+              </>
+            )}
+          </EuiResizableContainer>
+        </EuiSplitPanel.Inner>
+        <EuiSplitPanel.Inner grow={false} color="subdued">
+          <ManagementBottomBar
+            confirmTooltip={confirmTooltip}
+            onCancel={resetChanges}
+            onConfirm={saveChanges}
+            isLoading={isSavingChanges}
+            disabled={isSubmitDisabled}
+          />
+        </EuiSplitPanel.Inner>
+      </EuiSplitPanel.Outer>
+    </SimulatorContextProvider>
  );
 }

--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/processor_outcome_preview.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/processor_outcome_preview.tsx
@ -5,7 +5,7 @@
 * 2.0.
 */

-import React, { useMemo, useState } from 'react';
+import React, { useMemo } from 'react';
 import { useDateRange } from '@kbn/observability-utils-browser/hooks/use_date_range';
 import {
  EuiFlexGroup,
@ -18,58 +18,43 @@ import {
 } from '@elastic/eui';
 import { i18n } from '@kbn/i18n';
 import { TimeRange } from '@kbn/es-query';
-import { flattenObjectNestedLast } from '@kbn/object-utils';
 import { isEmpty } from 'lodash';
-import { RecursiveRecord } from '@kbn/streams-schema';
+import { SampleDocument } from '@kbn/streams-schema';
 import { useKibana } from '../../hooks/use_kibana';
 import { StreamsAppSearchBar, StreamsAppSearchBarProps } from '../streams_app_search_bar';
 import { PreviewTable } from '../preview_table';
-import { TableColumn, UseProcessingSimulatorReturn } from './hooks/use_processing_simulator';
+import {
+  DocsFilterOption,
+  TableColumn,
+  UseProcessingSimulatorReturn,
+  docsFilterOptions,
+} from './hooks/use_processing_simulator';
 import { AssetImage } from '../asset_image';

 interface ProcessorOutcomePreviewProps {
  columns: TableColumn[];
  isLoading: UseProcessingSimulatorReturn['isLoading'];
  simulation: UseProcessingSimulatorReturn['simulation'];
-  samples: UseProcessingSimulatorReturn['samples'];
+  filteredSamples: UseProcessingSimulatorReturn['samples'];
  onRefreshSamples: UseProcessingSimulatorReturn['refreshSamples'];
+  selectedDocsFilter: UseProcessingSimulatorReturn['selectedDocsFilter'];
+  setSelectedDocsFilter: UseProcessingSimulatorReturn['setSelectedDocsFilter'];
 }

 export const ProcessorOutcomePreview = ({
  columns,
  isLoading,
  simulation,
-  samples,
+  filteredSamples,
  onRefreshSamples,
+  selectedDocsFilter,
+  setSelectedDocsFilter,
 }: ProcessorOutcomePreviewProps) => {
  const { dependencies } = useKibana();
  const { data } = dependencies.start;

  const { timeRange, setTimeRange } = useDateRange({ data });

-  const [selectedDocsFilter, setSelectedDocsFilter] =
-    useState<DocsFilterOption>('outcome_filter_all');
-
-  const simulationDocuments = useMemo(() => {
-    if (!simulation?.documents) {
-      return samples.map((doc) => flattenObjectNestedLast(doc)) as RecursiveRecord[];
-    }
-
-    const filterDocuments = (filter: DocsFilterOption) => {
-      switch (filter) {
-        case 'outcome_filter_matched':
-          return simulation.documents.filter((doc) => doc.status === 'parsed');
-        case 'outcome_filter_unmatched':
-          return simulation.documents.filter((doc) => doc.status !== 'parsed');
-        case 'outcome_filter_all':
-        default:
-          return simulation.documents;
-      }
-    };
-
-    return filterDocuments(selectedDocsFilter).map((doc) => doc.value);
-  }, [samples, simulation?.documents, selectedDocsFilter]);
-
  const tableColumns = useMemo(() => {
    switch (selectedDocsFilter) {
      case 'outcome_filter_unmatched':
@ -97,38 +82,12 @@ export const ProcessorOutcomePreview = ({
        />
      </EuiFlexItem>
      <EuiSpacer size="m" />
-      <OutcomePreviewTable documents={simulationDocuments} columns={tableColumns} />
+      <OutcomePreviewTable documents={filteredSamples} columns={tableColumns} />
      {isLoading && <EuiProgress size="xs" color="accent" position="absolute" />}
    </>
  );
 };

-const docsFilterOptions = {
-  outcome_filter_all: {
-    id: 'outcome_filter_all',
-    label: i18n.translate(
-      'xpack.streams.streamDetailView.managementTab.enrichment.processor.outcomeControls.all',
-      { defaultMessage: 'All samples' }
-    ),
-  },
-  outcome_filter_matched: {
-    id: 'outcome_filter_matched',
-    label: i18n.translate(
-      'xpack.streams.streamDetailView.managementTab.enrichment.processor.outcomeControls.matched',
-      { defaultMessage: 'Matched' }
-    ),
-  },
-  outcome_filter_unmatched: {
-    id: 'outcome_filter_unmatched',
-    label: i18n.translate(
-      'xpack.streams.streamDetailView.managementTab.enrichment.processor.outcomeControls.unmatched',
-      { defaultMessage: 'Unmatched' }
-    ),
-  },
-} as const;
-
-type DocsFilterOption = keyof typeof docsFilterOptions;
-
 interface OutcomeControlsProps {
  docsFilter: DocsFilterOption;
  timeRange: TimeRange;
@ -211,7 +170,7 @@ const OutcomeControls = ({
 };

 interface OutcomePreviewTableProps {
-  documents: RecursiveRecord[];
+  documents: SampleDocument[];
  columns: string[];
 }

--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/processors/grok/grok_ai_suggestions.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/processors/grok/grok_ai_suggestions.tsx
@ -0,0 +1,329 @@
+/*
+ * 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 React, { useCallback, useState } from 'react';
+import {
+  EuiBadge,
+  EuiButton,
+  EuiButtonIcon,
+  EuiCallOut,
+  EuiCodeBlock,
+  EuiContextMenuItem,
+  EuiContextMenuPanel,
+  EuiFlexGroup,
+  EuiFlexItem,
+  EuiPopover,
+  EuiText,
+  useGeneratedHtmlId,
+} from '@elastic/eui';
+import { i18n } from '@kbn/i18n';
+import { useWatch, useFormContext } from 'react-hook-form';
+import { FlattenRecord, IngestStreamGetResponse } from '@kbn/streams-schema';
+import type { FindActionResult } from '@kbn/actions-plugin/server';
+import { UseGenAIConnectorsResult } from '@kbn/observability-ai-assistant-plugin/public/hooks/use_genai_connectors';
+import { useAbortController, useBoolean } from '@kbn/react-hooks';
+import { useKibana } from '../../../../hooks/use_kibana';
+import { GrokFormState, ProcessorFormState } from '../../types';
+import { UseProcessingSimulatorReturn } from '../../hooks/use_processing_simulator';
+import { useSimulatorContext } from '../../simulator_context';
+
+const RefreshButton = ({
+  generatePatterns,
+  connectors,
+  selectConnector,
+  currentConnector,
+  isLoading,
+}: {
+  generatePatterns: () => void;
+  selectConnector?: UseGenAIConnectorsResult['selectConnector'];
+  connectors?: FindActionResult[];
+  currentConnector?: string;
+  isLoading: boolean;
+}) => {
+  const [isPopoverOpen, { off: closePopover, toggle: togglePopover }] = useBoolean(false);
+  const splitButtonPopoverId = useGeneratedHtmlId({
+    prefix: 'splitButtonPopover',
+  });
+
+  return (
+    <EuiFlexGroup responsive={false} gutterSize="xs" alignItems="center">
+      <EuiFlexItem grow={false}>
+        <EuiButton
+          size="s"
+          iconType="sparkles"
+          data-test-subj="streamsAppGrokAiSuggestionsRefreshSuggestionsButton"
+          onClick={generatePatterns}
+          isLoading={isLoading}
+          disabled={currentConnector === undefined}
+        >
+          {i18n.translate(
+            'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.refreshSuggestions',
+            {
+              defaultMessage: 'Generate patterns',
+            }
+          )}
+        </EuiButton>
+      </EuiFlexItem>
+      {connectors && connectors.length > 1 && (
+        <EuiFlexItem grow={false}>
+          <EuiPopover
+            id={splitButtonPopoverId}
+            isOpen={isPopoverOpen}
+            button={
+              <EuiButtonIcon
+                data-test-subj="streamsAppGrokAiPickConnectorButton"
+                onClick={togglePopover}
+                display="base"
+                size="s"
+                iconType="boxesVertical"
+                aria-label={i18n.translate('xpack.streams.refreshButton.euiButtonIcon.moreLabel', {
+                  defaultMessage: 'More',
+                })}
+              />
+            }
+          >
+            <EuiContextMenuPanel
+              size="s"
+              items={connectors.map((connector) => (
+                <EuiContextMenuItem
+                  key={connector.id}
+                  icon={connector.id === currentConnector ? 'check' : 'empty'}
+                  onClick={() => {
+                    selectConnector?.(connector.id);
+                    closePopover();
+                  }}
+                >
+                  {connector.name}
+                </EuiContextMenuItem>
+              ))}
+            />
+          </EuiPopover>
+        </EuiFlexItem>
+      )}
+    </EuiFlexGroup>
+  );
+};
+
+function useAiEnabled() {
+  const { dependencies } = useKibana();
+  const { observabilityAIAssistant } = dependencies.start;
+
+  const aiAssistantEnabled = observabilityAIAssistant?.service.isEnabled();
+
+  const genAiConnectors = observabilityAIAssistant?.useGenAIConnectors();
+
+  return aiAssistantEnabled && (genAiConnectors?.connectors || []).length > 0;
+}
+
+function InnerGrokAiSuggestions({
+  refreshSimulation,
+  filteredSamples,
+  definition,
+}: {
+  refreshSimulation: UseProcessingSimulatorReturn['refreshSimulation'];
+  filteredSamples: FlattenRecord[];
+  definition: IngestStreamGetResponse;
+}) {
+  const { dependencies } = useKibana();
+  const {
+    streams: { streamsRepositoryClient },
+    observabilityAIAssistant,
+  } = dependencies.start;
+
+  const fieldValue = useWatch<ProcessorFormState, 'field'>({ name: 'field' });
+  const form = useFormContext<GrokFormState>();
+
+  const genAiConnectors = observabilityAIAssistant?.useGenAIConnectors();
+  const currentConnector = genAiConnectors?.selectedConnector;
+
+  const [isLoadingSuggestions, setSuggestionsLoading] = useState(false);
+  const [suggestionsError, setSuggestionsError] = useState<Error | undefined>();
+  const [suggestions, setSuggestions] = useState<
+    { patterns: string[]; simulations: any[] } | undefined
+  >();
+  const [blocklist, setBlocklist] = useState<Set<string>>(new Set());
+
+  const abortController = useAbortController();
+
+  const refreshSuggestions = useCallback(() => {
+    if (!currentConnector) {
+      setSuggestions({ patterns: [], simulations: [] });
+      return;
+    }
+    setSuggestionsLoading(true);
+    setSuggestionsError(undefined);
+    setSuggestions(undefined);
+    streamsRepositoryClient
+      .fetch('POST /api/streams/{name}/processing/_suggestions', {
+        signal: abortController.signal,
+        params: {
+          path: { name: definition.stream.name },
+          body: {
+            field: fieldValue,
+            connectorId: currentConnector,
+            samples: filteredSamples,
+          },
+        },
+      })
+      .then((response) => {
+        setSuggestions(response);
+        setSuggestionsLoading(false);
+      })
+      .catch((error) => {
+        setSuggestionsError(error);
+        setSuggestionsLoading(false);
+      });
+  }, [
+    abortController.signal,
+    currentConnector,
+    definition.stream.name,
+    fieldValue,
+    filteredSamples,
+    streamsRepositoryClient,
+  ]);
+
+  let content: React.ReactNode = null;
+
+  if (suggestionsError) {
+    content = <EuiCallOut color="danger">{suggestionsError.message}</EuiCallOut>;
+  }
+
+  const currentPatterns = form.getValues().patterns;
+
+  const filteredSuggestions = suggestions?.patterns
+    .map((pattern, i) => ({
+      pattern,
+      success_rate: suggestions.simulations[i].success_rate,
+    }))
+    .filter(
+      (suggestion) =>
+        !blocklist.has(suggestion.pattern) &&
+        !currentPatterns.some(({ value }) => value === suggestion.pattern)
+    );
+
+  if (suggestions && !suggestions.patterns.length) {
+    content = (
+      <>
+        <EuiCallOut color="primary">
+          {i18n.translate(
+            'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.noSuggestions',
+            { defaultMessage: 'No suggested patterns found' }
+          )}{' '}
+        </EuiCallOut>
+      </>
+    );
+  } else if (filteredSuggestions && !filteredSuggestions.length) {
+    // if all suggestions are in the blocklist or already in the patterns, just show the generation button, but no message
+    content = null;
+  }
+
+  if (filteredSuggestions && filteredSuggestions.length) {
+    content = (
+      <EuiFlexGroup direction="column" gutterSize="m">
+        <EuiText size="xs">
+          {i18n.translate(
+            'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.suggestions',
+            {
+              defaultMessage: 'Generated patterns',
+            }
+          )}
+        </EuiText>
+        {filteredSuggestions.map((suggestion) => {
+          return (
+            <EuiFlexGroup responsive={false} wrap={false} key={suggestion.pattern}>
+              <EuiFlexItem grow basis="0">
+                <EuiFlexGroup direction="column" gutterSize="s" alignItems="flexStart">
+                  <EuiCodeBlock paddingSize="s">{suggestion.pattern}</EuiCodeBlock>
+                  <EuiBadge color="hollow">
+                    {i18n.translate(
+                      'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.matchRate',
+                      {
+                        defaultMessage: 'Match rate: {matchRate}%',
+                        values: {
+                          matchRate: (suggestion.success_rate * 100).toFixed(2),
+                        },
+                      }
+                    )}
+                  </EuiBadge>
+                </EuiFlexGroup>
+              </EuiFlexItem>
+              <EuiFlexItem grow={false}>
+                <EuiFlexGroup direction="column" gutterSize="s" alignItems="flexStart">
+                  <EuiButtonIcon
+                    onClick={() => {
+                      const currentState = form.getValues();
+                      const hasNoPatterns =
+                        !currentState.patterns || !currentState.patterns.some(({ value }) => value);
+                      form.clearErrors('patterns');
+                      if (hasNoPatterns) {
+                        form.setValue('patterns', [{ value: suggestion.pattern }]);
+                      } else {
+                        form.setValue('patterns', [
+                          ...currentState.patterns,
+                          { value: suggestion.pattern },
+                        ]);
+                      }
+                      refreshSimulation();
+                    }}
+                    data-test-subj="streamsAppGrokAiSuggestionsButton"
+                    iconType="plusInCircle"
+                    aria-label={i18n.translate(
+                      'xpack.streams.grokAiSuggestions.euiButtonIcon.addPatternLabel',
+                      { defaultMessage: 'Add pattern' }
+                    )}
+                  />
+                  <EuiButtonIcon
+                    onClick={() => {
+                      setBlocklist(new Set([...blocklist, suggestion.pattern]));
+                    }}
+                    data-test-subj="hideSuggestionButton"
+                    iconType="cross"
+                    aria-label={i18n.translate(
+                      'xpack.streams.grokAiSuggestions.euiButtonIcon.hidePatternSuggestionLabel',
+                      { defaultMessage: 'Hide pattern suggestion' }
+                    )}
+                  />
+                </EuiFlexGroup>
+              </EuiFlexItem>
+            </EuiFlexGroup>
+          );
+        })}
+      </EuiFlexGroup>
+    );
+  }
+  return (
+    <>
+      {content != null && (
+        <EuiFlexGroup direction="column" gutterSize="m">
+          {content}
+        </EuiFlexGroup>
+      )}
+      <EuiFlexItem grow={false}>
+        <EuiFlexGroup direction="row" gutterSize="m" justifyContent="flexStart" alignItems="center">
+          <RefreshButton
+            isLoading={isLoadingSuggestions}
+            generatePatterns={refreshSuggestions}
+            connectors={genAiConnectors?.connectors}
+            selectConnector={genAiConnectors?.selectConnector}
+            currentConnector={currentConnector}
+          />
+        </EuiFlexGroup>
+      </EuiFlexItem>
+    </>
+  );
+}
+
+export function GrokAiSuggestions() {
+  const isAiEnabled = useAiEnabled();
+  const props = useSimulatorContext();
+
+  if (!isAiEnabled || !props.filteredSamples.length) {
+    return null;
+  }
+
+  return <InnerGrokAiSuggestions {...props} />;
+}
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/processors/grok/grok_patterns_editor.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/processors/grok/grok_patterns_editor.tsx
@ -5,7 +5,6 @@
 * 2.0.
 */

-import React from 'react';
 import {
  useFormContext,
  useFieldArray,
@ -25,8 +24,10 @@ import {
  EuiButtonIcon,
 } from '@elastic/eui';
 import { i18n } from '@kbn/i18n';
+import React from 'react';
 import { SortableList } from '../../sortable_list';
 import { GrokFormState } from '../../types';
+import { GrokAiSuggestions } from './grok_ai_suggestions';

 export const GrokPatternsEditor = () => {
  const {
@ -59,41 +60,45 @@ export const GrokPatternsEditor = () => {
  const getRemovePatternHandler = (id: number) => (fields.length > 1 ? () => remove(id) : null);

  return (
-    <EuiFormRow
-      label={i18n.translate(
-        'xpack.streams.streamDetailView.managementTab.enrichment.processor.grokEditorLabel',
-        { defaultMessage: 'Grok patterns editor' }
-      )}
-    >
-      <EuiPanel color="subdued" paddingSize="none">
-        <SortableList onDragItem={handlerPatternDrag}>
-          {fieldsWithError.map((field, idx) => (
-            <DraggablePatternInput
-              key={field.id}
-              field={field}
-              idx={idx}
-              onRemove={getRemovePatternHandler(idx)}
-              inputProps={register(`patterns.${idx}.value`, {
-                required: i18n.translate(
-                  'xpack.streams.streamDetailView.managementTab.enrichment.processor.grokEditorRequiredError',
-                  { defaultMessage: 'A pattern is required.' }
-                ),
-              })}
-            />
-          ))}
-        </SortableList>
+    <>
+      <EuiFormRow
+        label={i18n.translate(
+          'xpack.streams.streamDetailView.managementTab.enrichment.processor.grokEditorLabel',
+          { defaultMessage: 'Grok patterns editor' }
+        )}
+      >
+        <EuiPanel color="subdued" paddingSize="none">
+          <SortableList onDragItem={handlerPatternDrag}>
+            {fieldsWithError.map((field, idx) => (
+              <DraggablePatternInput
+                key={field.id}
+                field={field}
+                idx={idx}
+                onRemove={getRemovePatternHandler(idx)}
+                inputProps={register(`patterns.${idx}.value`, {
+                  required: i18n.translate(
+                    'xpack.streams.streamDetailView.managementTab.enrichment.processor.grokEditorRequiredError',
+                    { defaultMessage: 'A pattern is required.' }
+                  ),
+                })}
+              />
+            ))}
+          </SortableList>
+        </EuiPanel>
+      </EuiFormRow>
+      <EuiFlexGroup justifyContent="spaceBetween" gutterSize="s" wrap>
+        <GrokAiSuggestions />
        <EuiButtonEmpty
          data-test-subj="streamsAppGrokPatternsEditorAddPatternButton"
          onClick={handleAddPattern}
-          iconType="plusInCircle"
        >
          {i18n.translate(
            'xpack.streams.streamDetailView.managementTab.enrichment.processor.grokEditor.addPattern',
            { defaultMessage: 'Add pattern' }
          )}
        </EuiButtonEmpty>
-      </EuiPanel>
-    </EuiFormRow>
+      </EuiFlexGroup>
+    </>
  );
 };

--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/processors/index.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/processors/index.tsx
@ -38,9 +38,9 @@ import {
  isDissectProcessor,
 } from '../utils';
 import { useDiscardConfirm } from '../../../hooks/use_discard_confirm';
-import { UseDefinitionReturn } from '../hooks/use_definition';
 import { ProcessorMetrics, UseProcessingSimulatorReturn } from '../hooks/use_processing_simulator';
 import { ProcessorErrors, ProcessorMetricBadges } from './processor_metrics';
+import { UseDefinitionReturn } from '../hooks/use_definition';

 export interface ProcessorPanelProps {
  definition: IngestStreamGetResponse;
@ -95,9 +95,9 @@ export function AddProcessorPanel({
  };

  const handleCancel = () => {
+    closePanel();
    methods.reset();
    onWatchProcessor({ id: 'draft', deleteIfExists: true });
-    closePanel();
  };

  const handleOpen = () => {
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/simulation_playground.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/simulation_playground.tsx
@ -17,12 +17,23 @@ interface SimulationPlaygroundProps {
  columns: TableColumn[];
  isLoading: UseProcessingSimulatorReturn['isLoading'];
  simulation: UseProcessingSimulatorReturn['simulation'];
-  samples: UseProcessingSimulatorReturn['samples'];
+  filteredSamples: UseProcessingSimulatorReturn['filteredSamples'];
+  selectedDocsFilter: UseProcessingSimulatorReturn['selectedDocsFilter'];
+  setSelectedDocsFilter: UseProcessingSimulatorReturn['setSelectedDocsFilter'];
  onRefreshSamples: UseProcessingSimulatorReturn['refreshSamples'];
 }

 export const SimulationPlayground = (props: SimulationPlaygroundProps) => {
-  const { definition, columns, isLoading, simulation, samples, onRefreshSamples } = props;
+  const {
+    definition,
+    columns,
+    isLoading,
+    simulation,
+    filteredSamples,
+    onRefreshSamples,
+    setSelectedDocsFilter,
+    selectedDocsFilter,
+  } = props;

  const tabs = {
    dataPreview: {
@ -64,8 +75,10 @@ export const SimulationPlayground = (props: SimulationPlaygroundProps) => {
          columns={columns}
          isLoading={isLoading}
          simulation={simulation}
-          samples={samples}
+          filteredSamples={filteredSamples}
          onRefreshSamples={onRefreshSamples}
+          selectedDocsFilter={selectedDocsFilter}
+          setSelectedDocsFilter={setSelectedDocsFilter}
        />
      )}
      {selectedTabId === 'detectedFields' &&
--- a/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/simulator_context.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/components/stream_detail_enrichment/simulator_context.tsx
@ -0,0 +1,46 @@
+/*
+ * 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 React, { useMemo } from 'react';
+import { createContext } from 'react';
+import { IngestStreamGetResponse } from '@kbn/streams-schema';
+import { UseProcessingSimulatorReturn } from './hooks/use_processing_simulator';
+
+export const context = createContext<SimulatorContextValue | undefined>(undefined);
+
+export interface SimulatorContextValue extends UseProcessingSimulatorReturn {
+  definition: IngestStreamGetResponse;
+}
+
+export function SimulatorContextProvider({
+  processingSimulator,
+  definition,
+
+  children,
+}: {
+  processingSimulator: UseProcessingSimulatorReturn;
+  definition: IngestStreamGetResponse;
+  children: React.ReactNode;
+}) {
+  const contextValue = useMemo(() => {
+    return {
+      definition,
+      ...processingSimulator,
+    };
+  }, [definition, processingSimulator]);
+  return <context.Provider value={contextValue}>{children}</context.Provider>;
+}
+
+export function useSimulatorContext() {
+  const ctx = React.useContext(context);
+  if (!ctx) {
+    throw new Error(
+      'useStreamsEnrichmentContext must be used within a StreamsEnrichmentContextProvider'
+    );
+  }
+  return ctx;
+}
--- a/x-pack/solutions/observability/plugins/streams_app/public/hooks/queries/use_async_sample.tsx
+++ b/x-pack/solutions/observability/plugins/streams_app/public/hooks/queries/use_async_sample.tsx
@ -8,7 +8,7 @@
 import { useEffect, useMemo, useState } from 'react';
 import {
  Condition,
-  RecursiveRecord,
+  SampleDocument,
  WiredStreamGetResponse,
  conditionToQueryDsl,
  getFields,
@ -38,7 +38,7 @@ export const useAsyncSample = (options: Options) => {
  // Documents
  const [isLoadingDocuments, toggleIsLoadingDocuments] = useToggle(false);
  const [documentsError, setDocumentsError] = useState();
-  const [documents, setDocuments] = useState<RecursiveRecord[]>([]);
+  const [documents, setDocuments] = useState<SampleDocument[]>([]);

  // Document counts / percentage
  const [isLoadingDocumentCounts, toggleIsLoadingDocumentCounts] = useToggle(false);
--- a/x-pack/solutions/observability/plugins/streams_app/public/types.ts
+++ b/x-pack/solutions/observability/plugins/streams_app/public/types.ts
@ -19,6 +19,10 @@ import type { SharePublicSetup, SharePublicStart } from '@kbn/share-plugin/publi
 import type { SavedObjectTaggingPluginStart } from '@kbn/saved-objects-tagging-plugin/public';
 import { NavigationPublicStart } from '@kbn/navigation-plugin/public/types';
 import { FieldsMetadataPublicStart } from '@kbn/fields-metadata-plugin/public';
+import {
+  ObservabilityAIAssistantPublicSetup,
+  ObservabilityAIAssistantPublicStart,
+} from '@kbn/observability-ai-assistant-plugin/public';
 /* eslint-disable @typescript-eslint/no-empty-interface*/

 export interface ConfigSchema {}
@ -30,6 +34,7 @@ export interface StreamsAppSetupDependencies {
  observabilityShared: ObservabilitySharedPluginSetup;
  unifiedSearch: {};
  share: SharePublicSetup;
+  observabilityAIAssistant?: ObservabilityAIAssistantPublicSetup;
 }

 export interface StreamsAppStartDependencies {
@ -42,6 +47,7 @@ export interface StreamsAppStartDependencies {
  savedObjectsTagging: SavedObjectTaggingPluginStart;
  navigation: NavigationPublicStart;
  fieldsMetadata: FieldsMetadataPublicStart;
+  observabilityAIAssistant?: ObservabilityAIAssistantPublicStart;
 }

 export interface StreamsAppPublicSetup {}
--- a/x-pack/solutions/observability/plugins/streams_app/tsconfig.json
+++ b/x-pack/solutions/observability/plugins/streams_app/tsconfig.json
@ -49,14 +49,16 @@
    "@kbn/react-field",
    "@kbn/shared-ux-utility",
    "@kbn/unsaved-changes-prompt",
-    "@kbn/object-utils",
    "@kbn/deeplinks-analytics",
    "@kbn/dashboard-plugin",
    "@kbn/react-kibana-mount",
    "@kbn/fields-metadata-plugin",
+    "@kbn/observability-ai-assistant-plugin",
+    "@kbn/actions-plugin",
    "@kbn/datemath",
    "@kbn/dataset-quality-plugin",
    "@kbn/search-types",
-    "@kbn/traced-es-client"
+    "@kbn/object-utils",
+    "@kbn/traced-es-client",
  ]
 }