chore(alerting_v2): simplify and clean up dispatcher

- tick_summary.ts: use stages.at(-1)?.counts ?? ZERO_COUNTS instead of
  manual length-guard + index arithmetic
- state_counts.ts: replace Object.fromEntries(entries.map(…)) in
  toSpanLabels with a direct property-assignment loop (no intermediate
  arrays)
- queries.ts: fuse the two alertEpisodes traversals in
  getAlertEpisodeSuppressionsQueries into a single for...of loop that
  builds minLastEventTimestamp and the uniquePairKeySet together
- types.ts: export CLEAN_HALT_REASONS next to DispatcherHaltReason so
  the watermark-advancement policy lives at the type boundary
- dispatcher.ts: use CLEAN_HALT_REASONS.has() in
  extractAdvanceableWatermark; adding a new controlled halt reason now
  requires only one change

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: kdelemme

x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/dispatcher.ts

@@ -17,6 +17,7 @@ import {
   type DispatcherPipelineResult,
 } from './execution_pipeline';
 import { buildTickSummary, emitTickSummary, startHrtime } from './telemetry';
+import { CLEAN_HALT_REASONS } from './types';
 import type { DispatcherExecutionParams, DispatcherExecutionResult } from './types';
 
 export interface DispatcherServiceContract {
@@ -99,12 +100,8 @@ export class DispatcherService implements DispatcherServiceContract {
  * data loss.
  */
 function extractAdvanceableWatermark(result: DispatcherPipelineResult): string | undefined {
-  if (
-    !result.completed &&
-    result.haltReason !== 'no_episodes' &&
-    result.haltReason !== 'no_actions'
-  ) {
-    return undefined;
-  }
-  return result.finalState.nextEventWatermark;
+  const isClean =
+    result.completed ||
+    (result.haltReason !== undefined && CLEAN_HALT_REASONS.has(result.haltReason));
+  return isClean ? result.finalState.nextEventWatermark : undefined;
 }

x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/queries.ts

@@ -145,20 +145,16 @@ export const getAlertEpisodeSuppressionsQueries = (
 ): EsqlRequest[] => {
   if (alertEpisodes.length === 0) return [];
 
-  const minLastEventTimestamp =
-    alertEpisodes.reduce<string | undefined>((min, ep) => {
-      const parsedTimestamp = new Date(ep.last_event_timestamp);
-      if (Number.isNaN(parsedTimestamp.getTime())) {
-        return min;
-      }
-
-      const normalizedTimestamp = parsedTimestamp.toISOString();
-      return min === undefined || normalizedTimestamp < min ? normalizedTimestamp : min;
-    }, undefined) ?? new Date(0).toISOString();
-
-  const uniquePairKeys = [
-    ...new Set(alertEpisodes.map((ep) => `${ep.rule_id}${PAIR_SEPARATOR}${ep.group_hash}`)),
-  ];
+  let minLastEventTimestamp: string | undefined;
+  const uniquePairKeySet = new Set<string>();
+  for (const ep of alertEpisodes) {
+    if (minLastEventTimestamp === undefined || ep.last_event_timestamp < minLastEventTimestamp) {
+      minLastEventTimestamp = ep.last_event_timestamp;
+    }
+    uniquePairKeySet.add(`${ep.rule_id}${PAIR_SEPARATOR}${ep.group_hash}`);
+  }
+  const uniquePairKeys = [...uniquePairKeySet];
+  const minTs = minLastEventTimestamp ?? new Date(0).toISOString();
 
   return chunkInClauseLiterals(uniquePairKeys).map((chunk) => {
     const pairValues = chunk.map((key) => esql.str(key));
@@ -167,7 +163,7 @@ export const getAlertEpisodeSuppressionsQueries = (
         | EVAL _pair_key = CONCAT(rule_id, ${PAIR_SEPARATOR}, group_hash)
         | WHERE _pair_key IN (${pairValues})
         | WHERE action_type IN ("ack", "unack", "deactivate", "activate", "snooze", "unsnooze")
-        | WHERE action_type != "snooze" OR expiry > ${minLastEventTimestamp}::datetime
+        | WHERE action_type != "snooze" OR expiry > ${minTs}::datetime
         | INLINE STATS
             last_snooze_action = LAST(action_type, @timestamp) WHERE action_type IN ("snooze", "unsnooze")
             BY rule_id, group_hash

x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/telemetry/state_counts.ts

@@ -34,7 +34,9 @@ export function computeStateCounts(state: DispatcherPipelineState): DispatcherSt
  * APM-facing shapes are allowed to drift independently.
  */
 export function toSpanLabels(counts: DispatcherStageCounts): Record<string, number> {
-  return Object.fromEntries(
-    Object.entries(counts).map(([key, value]) => [`count_${key}`, value as number])
-  );
+  const labels: Record<string, number> = {};
+  for (const key of Object.keys(counts) as Array<keyof DispatcherStageCounts>) {
+    labels[`count_${key}`] = counts[key];
+  }
+  return labels;
 }

x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/telemetry/tick_summary.ts

@@ -56,7 +56,7 @@ export function buildTickSummary({
     completed,
     halt_reason: haltReason,
     stages,
-    totals: stages.length === 0 ? ZERO_COUNTS : stages[stages.length - 1].counts,
+    totals: stages.at(-1)?.counts ?? ZERO_COUNTS,
   };
 }
 

x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/types.ts

@@ -199,6 +199,16 @@ export type DispatcherStepHaltReason = 'no_episodes' | 'no_actions';
  */
 export type DispatcherHaltReason = DispatcherStepHaltReason | 'step_error';
 
+/**
+ * Halt reasons that allow the event watermark to advance. Defined here,
+ * next to the halt-reason types, so adding a new controlled reason only
+ * requires a single change.
+ */
+export const CLEAN_HALT_REASONS: ReadonlySet<DispatcherHaltReason> = new Set<DispatcherHaltReason>([
+  'no_episodes',
+  'no_actions',
+]);
+
 export type DispatcherStepOutput =
   | { type: 'continue'; data?: Partial<Omit<DispatcherPipelineState, 'input'>> }
   | {