[Entity Analytics][Leads generation] Lead generation EUID fixes (elastic#268666)

## Summary

Closes elastic/security-team#17246.

Four places inside the lead generation observation modules used the
entity display name as a unique identifier when querying Elasticsearch
and when joining results back to entities in memory. The display name is
not unique. The Entity Store EUID (entity.id) is.

In any deployment where two entities can share a name but have different
EUIDs, for example two local users called alice on different hosts, or
two tenants in different namespaces, the old code would either collapse
them into one bucket or fail to find a match for one of them. The test
fixtures all used unique names so nobody noticed.

The risk score path looked like it was working but only by accident. The
V2 risk score writer happens to put the EUID into host.name and
user.name on the time series index, and the lead generation lookup
happened to double prefix its key, so the two buggy strings matched.
Under the legacy writer, when entityAnalyticsEntityStoreV2 is off, the
same lookup silently returned nothing.

This change picks one rule and applies it everywhere. The EUID is the
identity. Queries filter on it, aggregations bucket by it, and the in
memory join key is just the EUID.

## What changed

1. LeadEntity now carries id as a first class field next to type and
name. entityRecordToLeadEntity reads it from entity.id. If a record
arrives without an EUID it is dropped at the boundary instead of being
downgraded to an unknown entity that would later collide with every
other unknown.

2. Behavioral analysis (alerts query) now computes the EUID as a
Painless runtime keyword field via euid.painless.getEuidRuntimeMapping
and filters and aggregates on that field. It no longer touches user.name
or host.name for identity. This mirrors the pattern already used in
calculate_esql_risk_scores.

3. Temporal state (history snapshot query) now filters and aggregates on
entity.id directly. The entity store history snapshot index already
carries entity.id, so the previous indirection through user.name and
host.name was not just wrong but also unnecessary.

4. Risk score (time series query) now filters on risk.id_field equal to
entity.id and aggregates by risk.id_value. This is the only filter that
authoritatively selects V2 shaped documents regardless of which writer
last ran. Legacy shaped documents are excluded from the lookback window,
which means the trend math is correct under both flag states and the
transitional data hazard goes away.

5. entityToKey now returns the EUID. The four call sites that used to
build inline strings like type colon name or type colon euid have been
updated to call entityToKey instead. The now dead getEntityId helper has
been removed along with its re export.


The LeadEntity type used to expose only type and name. The EUID lived
three property accesses deep inside record.entity.id. So the path of
least resistance for every new module was entity.name. The retrofit
helper getEntityId was added when the risk score module hit the EUID
boundary, but it was only adopted by that one module. The other two
modules carried on with entity.name. The tests confirmed the buggy code
rather than the requirement, because every fixture used unique names. By
adding id to LeadEntity as a peer of name, the natural ergonomic field
is now also the correct one.

## Automated checks already run

- [x] node scripts/jest
x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/lead_generation
passes (17 suites, 163 tests).
- [x] node scripts/jest on
x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/risk_score/risk_score_data_client.test.ts
passes (13 tests).
- [x] node scripts/eslint with fix on the diff reports no errors.
- [x] node scripts/type_check on the security solution tsconfig exits
clean.

New test coverage added for the failure modes the old tests masked:

- Two entities sharing a name but with distinct EUIDs receive separate
alert summaries (behavioral analysis).
- Same shape fixture for the temporal module proves only the correct
EUID is flagged for escalation, not its same named sibling.
- The data client filters on risk.id_field and excludes legacy shaped
documents from the trend.
- The risk score module produces current score observations even when no
historical buckets come back, which is the legacy writer environment.

## How to test this manually

These steps assume a local Kibana instance running against an
Elasticsearch with some Entity Store V2 data already present. If you
only have a tiny dataset, Test 1 alone is fine. Tests 2 and 3 are the
ones that actually exercise the bug class this PR fixes, so please do at
least one of them.

### Setup

1. Pull this branch and run yarn kbn bootstrap.
2. In your kibana.dev.yml make sure both flags are on:
   ```
xpack.securitySolution.enableExperimental: ['leadGenerationEnabled',
'entityAnalyticsEntityStoreV2']
   ```
3. Start Kibana and Elasticsearch and log in as a user with Security
Solution access.
4. Confirm the Entity Store has at least a handful of user and host
entities. The entity store list API or the entities UI is fine for this.

### Test 1: regression smoke test

Goal: confirm the pipeline still produces sensible leads after the
refactor.

1. Trigger lead generation. The simplest way is the API:
   ```
   POST kbn:/internal/entity_analytics/leads/generate
   ```
   This can be run from Dev Tools.
2. List the resulting leads:
   ```
   GET kbn:/internal/entity_analytics/leads
   ```
3. Pick a few leads and confirm each one has at least one observation
attached and the observations look plausible (risk score level matches
the current risk score on the entity, alert counts look right for the
lookback, and so on).
4. Watch the Kibana server logs while the pipeline runs. There should be
no new warnings or errors that did not exist on main.

Pass criteria: leads come back, observations are populated, no new log
noise.

### Test 2: composite identity, the core scenario this PR fixes

Goal: prove that two entities that share a display name but have
different EUIDs no longer collide.

1. Pick or create two Entity Store records of the same type where
entity.name is identical but entity.id differs. For example two user
records both with entity.name set to alice but with entity.id set to
user:alice@corp.com and user:alice@partner.com respectively. The exact
way to create these depends on how your local cluster is wired. If you
have entity sources running, ingest two source documents that resolve to
different EUIDs but happen to share the display name. If you do not, you
can also write the two records directly into the entity store index for
the purposes of this test.
2. Make sure the two entities differ in at least one observable way. For
example put two alerts into the alerts index pointing at one of the two
EUIDs (use the runtime field resolution in mind: the alert needs
user.name plus enough identity fields to resolve to a specific EUID),
and leave the other entity with no alerts.
3. Trigger lead generation as in Test 1.
4. Look at the observations for both entities, either via the lead
documents or directly in the lead generation observation index.

Pass criteria: each entity gets its own observations attributed to its
own EUID. The entity with alerts gets the alert based observations. The
entity without alerts does not. Before this PR both entities would
either share one bucket or one of them would lose its own observations.

Negative version of this test if you want to verify with the previous
code: check out main, repeat the same setup, and you should see one
entity getting credit for the other entity's alerts, or both getting the
same combined observations.

### Test 3: risk score module under entityAnalyticsEntityStoreV2 off

Goal: prove that the risk score path no longer silently returns zero
data when the legacy writer is the one populating the time series index.

1. Turn the V2 flag off:
   ```
   xpack.securitySolution.enableExperimental: ['leadGenerationEnabled']
   ```
leadGenerationEnabled stays on, entityAnalyticsEntityStoreV2 comes off.
2. Restart Kibana. On startup confirm in the logs that the legacy risk
scoring task was registered, not the V2 maintainer.
3. Make sure at least one entity has a current risk score on its entity
record. The current score lives on the entity record itself, not in the
time series index, so it should be there regardless of which writer is
active.
4. Trigger lead generation.

Pass criteria: current score observations such as high_risk_score,
moderate_risk_score, and privileged_high_risk are produced for entities
that qualify. Historical escalation observations (risk_escalation_24h,
risk_escalation_7d, risk_escalation_90d) are not produced, because there
are no V2 shaped time series documents in the lookback. No exceptions in
the logs. Specifically no warning about Failed to fetch time-series risk
scores.

Before this PR the same query would still run but match zero documents
(legacy shaped docs have risk.id_field set to host.name or user.name),
so the trend math would silently miss every escalation. After this PR
the same outcome is intentional and graceful rather than accidental.

### Test 4: missing EUID record is dropped at the boundary

Goal: prove that entities without an EUID no longer leak through as
unknown.

1. Manually create or temporarily edit an entity store record so
entity.id is missing or empty. Keep entity.name set.
2. Trigger lead generation with that entity included in the candidate
set.
3. Set the Kibana logger to debug level for the LeadGeneration namespace
and watch the logs.

Pass criteria: you see a single debug line that mentions skipped N
without EUID, with N at least one. The pipeline does not produce any
observations for that entity, and it does not error out.

### If you only want to do a code level review

The four spots worth focusing on:

1. observation_modules/utils.ts: entityToKey should return entity.id
only. No string concatenation.
2. The three observation modules
(behavioral_analysis_module/{data_access,module}.ts,
temporal_state_module.ts, risk_score_module.ts) should all use
entityToKey(entity) for in memory lookups. There should be no occurrence
of $\{type\}.name in any query body.
3. risk_score/risk_score_data_client.ts:
getDailyAverageRiskScoreNormSeries should filter on risk.id_field equal
to entity.id and aggregate by risk.id_value, and its returned map should
be keyed by the EUID directly (no type prefix on top of an already type
prefixed string).
4. lead_generation/entity_conversion.ts: entityRecordToLeadEntity
returns undefined when entity.id is missing, and fetchCandidateEntities
filters those out before returning the batch.

Author: abhishekbhatia1710

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/lead_generation/engine/lead_generation_engine.test.ts

@@ -17,18 +17,22 @@ jest.mock('./llm_synthesize');
 // Helpers
 // ---------------------------------------------------------------------------
 
-const createMockEntity = (name: string, type = 'user'): LeadEntity => ({
-  record: { name, type, id: `${type}-${name}` } as unknown as LeadEntity['record'],
-  type,
-  name,
-});
+const createMockEntity = (name: string, type = 'user'): LeadEntity => {
+  const id = `${type}:${name}`;
+  return {
+    record: { name, type, id } as unknown as LeadEntity['record'],
+    id,
+    type,
+    name,
+  };
+};
 
 const createMockObservation = (
   entity: LeadEntity,
   moduleId: string,
   overrides: Partial<Observation> = {}
 ): Observation => ({
-  entityId: `${entity.type}:${entity.name}`,
+  entityId: entity.id,
   moduleId,
   type: 'test_signal',
   score: 75,

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/lead_generation/engine/lead_generation_engine.ts

@@ -16,7 +16,6 @@ import type {
   ObservationModule,
 } from '../types';
 import { computeStaleness, DEFAULT_ENGINE_CONFIG } from '../types';
-import { entityToKey } from '../observation_modules/utils';
 import { llmSynthesizeBatch } from './llm_synthesize';
 
 interface LeadGenerationEngineDeps {
@@ -177,7 +176,7 @@ const scoreEntities = (
   config: LeadGenerationEngineConfig,
   moduleWeights: ReadonlyMap<string, number>
 ): ScoredEntity[] => {
-  const entityByKey = new Map(allEntities.map((e) => [entityToKey(e), e]));
+  const entityByKey = new Map(allEntities.map((e) => [e.id, e]));
   const observationsByEntity = groupObservationsByEntity(observations);
 
   return [...observationsByEntity.entries()]
@@ -297,7 +296,7 @@ const groupByObservationPattern = (scoredEntities: ScoredEntity[]): ScoredEntity
 const buildByline = (group: ScoredEntity[], observations: Observation[]): string => {
   if (group.length === 1) {
     const { entity } = group[0];
-    const entityObs = observations.filter((o) => o.entityId === entityToKey(entity));
+    const entityObs = observations.filter((o) => o.entityId === entity.id);
 
     const totalAlerts = extractNumber(entityObs, 'total_alerts');
     const distinctRules =

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/lead_generation/engine/llm_synthesize.test.ts

@@ -34,17 +34,21 @@ const { llmSynthesizeBatch } = jest.requireActual('./llm_synthesize') as {
   llmSynthesizeBatch: typeof import('./llm_synthesize').llmSynthesizeBatch;
 };
 
-const createMockEntity = (name: string, type = 'user'): LeadEntity => ({
-  record: { name, type, id: `${type}-${name}` } as unknown as LeadEntity['record'],
-  type,
-  name,
-});
+const createMockEntity = (name: string, type = 'user'): LeadEntity => {
+  const id = `${type}:${name}`;
+  return {
+    record: { name, type, id } as unknown as LeadEntity['record'],
+    id,
+    type,
+    name,
+  };
+};
 
 const createMockObservation = (
   entity: LeadEntity,
   overrides: Partial<Observation> = {}
 ): Observation => ({
-  entityId: `${entity.type}:${entity.name}`,
+  entityId: entity.id,
   moduleId: 'risk_analysis',
   type: 'high_risk_score',
   score: 80,

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/lead_generation/engine/llm_synthesize.ts

@@ -10,7 +10,6 @@ import type { InferenceChatModel } from '@kbn/inference-langchain';
 import { JsonOutputParser } from '@langchain/core/output_parsers';
 import { ChatPromptTemplate } from '@langchain/core/prompts';
 import type { LeadEntity, Observation } from '../types';
-import { entityToKey } from '../observation_modules/utils';
 
 export interface ScoredEntityInput {
   readonly entity: LeadEntity;
@@ -60,7 +59,7 @@ const formatLeadsPayload = (groups: ScoredEntityInput[][]): string => {
 
       const obsLines = group
         .flatMap((s) => {
-          const key = entityToKey(s.entity);
+          const key = s.entity.id;
           return s.observations
             .filter((o) => o.entityId === key)
             .map((obs) => {

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/lead_generation/entity_conversion.test.ts

@@ -0,0 +1,120 @@
+/*
+ * 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 { loggingSystemMock } from '@kbn/core/server/mocks';
+import type { EntityStoreCRUDClient } from '@kbn/entity-store/server';
+import { entityRecordToLeadEntity, fetchCandidateEntities } from './entity_conversion';
+
+type EntityRecord = Parameters<typeof entityRecordToLeadEntity>[0];
+
+const buildRecord = (entity: EntityRecord['entity'] | undefined): EntityRecord =>
+  ({ entity } as unknown as EntityRecord);
+
+describe('entityRecordToLeadEntity', () => {
+  it('returns a LeadEntity with id, type, and name when the record has an EUID', () => {
+    const record = buildRecord({ id: 'user:alice', type: 'user', name: 'Alice Adams' });
+
+    const lead = entityRecordToLeadEntity(record);
+
+    expect(lead).toEqual({
+      record,
+      id: 'user:alice',
+      type: 'user',
+      name: 'Alice Adams',
+    });
+  });
+
+  it('uses EngineMetadata.Type when entity.type is absent', () => {
+    const record = buildRecord({
+      id: 'host:server-01',
+      EngineMetadata: { Type: 'host' },
+      name: 'server-01',
+    } as EntityRecord['entity']);
+
+    const lead = entityRecordToLeadEntity(record);
+
+    expect(lead?.type).toBe('host');
+  });
+
+  it('falls back to id for the display name when entity.name is absent', () => {
+    const record = buildRecord({ id: 'user:bob', type: 'user' });
+
+    const lead = entityRecordToLeadEntity(record);
+
+    expect(lead?.name).toBe('user:bob');
+  });
+
+  it('returns undefined when entity.id is absent', () => {
+    const record = buildRecord({ type: 'user', name: 'Alice Adams' });
+
+    expect(entityRecordToLeadEntity(record)).toBeUndefined();
+  });
+
+  it('returns undefined when the entity object itself is missing', () => {
+    const record = buildRecord(undefined);
+
+    expect(entityRecordToLeadEntity(record)).toBeUndefined();
+  });
+
+  it('returns undefined when entity.id is an empty string', () => {
+    const record = buildRecord({ id: '', type: 'user', name: 'Alice' });
+
+    expect(entityRecordToLeadEntity(record)).toBeUndefined();
+  });
+});
+
+describe('fetchCandidateEntities', () => {
+  const logger = loggingSystemMock.createLogger();
+  const listEntities = jest.fn();
+  const crudClient = { listEntities } as unknown as EntityStoreCRUDClient;
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('filters out records that have no EUID', async () => {
+    listEntities.mockResolvedValueOnce({
+      entities: [
+        { entity: { id: 'user:alice', type: 'user', name: 'Alice' } },
+        { entity: { type: 'user', name: 'NoId' } },
+        { entity: { id: 'host:server-01', EngineMetadata: { Type: 'host' }, name: 'server-01' } },
+      ],
+      total: 3,
+    });
+
+    const result = await fetchCandidateEntities(crudClient, logger);
+
+    expect(result).toHaveLength(2);
+    expect(result.map((e) => e.id)).toEqual(['user:alice', 'host:server-01']);
+  });
+
+  it('logs the skipped-without-EUID count when non-zero', async () => {
+    listEntities.mockResolvedValueOnce({
+      entities: [
+        { entity: { id: 'user:alice', type: 'user', name: 'Alice' } },
+        { entity: { type: 'user', name: 'NoId' } },
+      ],
+      total: 2,
+    });
+
+    await fetchCandidateEntities(crudClient, logger);
+
+    expect(logger.debug).toHaveBeenCalledWith(expect.stringContaining('skipped 1 without EUID'));
+  });
+
+  it('does not mention skipped entities when all records have an EUID', async () => {
+    listEntities.mockResolvedValueOnce({
+      entities: [{ entity: { id: 'user:alice', type: 'user', name: 'Alice' } }],
+      total: 1,
+    });
+
+    await fetchCandidateEntities(crudClient, logger);
+
+    const debugCalls = logger.debug.mock.calls.flat().join(' ');
+    expect(debugCalls).not.toContain('skipped');
+  });
+});

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/lead_generation/entity_conversion.ts

@@ -18,19 +18,24 @@ type EntityStoreEntity = Awaited<
 >['entities'][number];
 
 /**
- * Convert an Entity Store V2 record into a LeadEntity, extracting the
- * convenience `type` and `name` fields from the nested `entity` object.
- * Falls back to `entity.id` (EUID) when `entity.name` is absent.
+ * Convert an Entity Store V2 record into a LeadEntity, extracting the EUID
+ * (`entity.id`) as the identity field plus the convenience `type` and `name`
+ * fields. Returns `undefined` when the record has no EUID — such records
+ * cannot be the subject of correct observations because they have no stable
+ * identity to join against.
  */
-export const entityRecordToLeadEntity = (record: EntityStoreEntity): LeadEntity => {
+export const entityRecordToLeadEntity = (record: EntityStoreEntity): LeadEntity | undefined => {
   const r = record as Record<string, unknown>;
   const entityField = r.entity as
     | { name?: string; type?: string; id?: string; EngineMetadata?: { Type?: string } }
     | undefined;
+  const id = entityField?.id;
+  if (!id) return undefined;
   return {
     record: record as Entity,
+    id,
     type: entityField?.EngineMetadata?.Type ?? entityField?.type ?? 'unknown',
-    name: entityField?.name ?? entityField?.id ?? 'unknown',
+    name: entityField?.name ?? id,
   };
 };
 
@@ -53,12 +58,17 @@ export const fetchCandidateEntities = async (
     page: 1,
   });
 
-  const leadEntities = entities.map(entityRecordToLeadEntity);
+  const leadEntities = entities
+    .map(entityRecordToLeadEntity)
+    .filter((entity): entity is LeadEntity => entity !== undefined);
+  const skipped = entities.length - leadEntities.length;
 
   logger?.debug(
-    `[LeadGeneration] Entity selection: ${total ?? leadEntities.length} total -> ${
+    `[LeadGeneration] Entity selection: ${total ?? entities.length} total -> ${
       leadEntities.length
-    } candidates (cap ${MAX_CANDIDATE_ENTITIES})`
+    } candidates (cap ${MAX_CANDIDATE_ENTITIES}${
+      skipped > 0 ? `, skipped ${skipped} without EUID` : ''
+    })`
   );
 
   return leadEntities;