fix(reporter): restore telemetry semantics, align allium spec with stateless reporter model

- emittedCounter now counts only CorrelationDetected events (not episode events),
  matching the pre-refactor "new pattern first-seen" semantics
- ongoingCounter now fires only when at least one active pattern was not newly
  detected this cycle, matching the pre-refactor "ongoing but not new" semantics
- ongoing debug log now excludes patterns that were newly detected this cycle
- EventReporter is fully stateless (no seen_patterns, no retry queue); all events
  are at-most-once — forwarder failure logs and discards the event
- reporter.allium updated: removed seen_patterns/previously_active/retry_queue,
  replaced PublishNewCorrelation+DropExpiredSeenPatterns with PublishCorrelatorEvents,
  updated ObserverReportBoundary surface to use correlator_events instead of
  correlation_history, added CorrelatorEvent external entity and CorrelatorEventKind enum

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: CelianR

comp/anomalydetection/reporter/impl/reporter.go

@@ -103,18 +103,31 @@ func (r *stdoutReporter) Name() string { return "stdout_reporter" }
 func (r *stdoutReporter) Report(output reporterdef.ReportOutput) bool {
 	emitted := false
 
+	// Build the set of newly-detected patterns from this cycle so they can be
+	// excluded from the "ongoing" telemetry path below.
+	newlyDetected := make(map[string]struct{}, len(output.CorrelatorEvents))
+
 	// Log all correlator events at info level and drive the emitted counter.
-	if r.stdoutEnabled {
-		for _, ce := range output.CorrelatorEvents {
-			switch ce.Kind {
-			case observerdef.CorrelatorEventEpisodeStarted:
+	// emittedCounter counts only CorrelationDetected events (new pattern first-seen
+	// or recurrence); episode events are not counted.
+	for _, ce := range output.CorrelatorEvents {
+		switch ce.Kind {
+		case observerdef.CorrelatorEventEpisodeStarted:
+			if r.stdoutEnabled {
 				r.logger.Infof("[observer] scorer episode started: scorer=%s pattern=%s t=%d",
 					ce.CorrelatorName, ce.Correlation.Pattern, ce.Timestamp)
-			case observerdef.CorrelatorEventEpisodeEnded:
+			}
+		case observerdef.CorrelatorEventEpisodeEnded:
+			if r.stdoutEnabled {
 				r.logger.Infof("[observer] scorer episode ended: scorer=%s pattern=%s t=%d duration=%ds",
 					ce.CorrelatorName, ce.Correlation.Pattern, ce.Timestamp,
 					ce.Correlation.LastUpdated-ce.Correlation.FirstSeen)
-			case observerdef.CorrelatorEventCorrelationDetected:
+			}
+		case observerdef.CorrelatorEventCorrelationDetected:
+			newlyDetected[ce.Correlation.Pattern] = struct{}{}
+			r.emittedCounter.Add(1)
+			emitted = true
+			if r.stdoutEnabled {
 				r.logger.Infof("[observer] anomaly detection report: pattern=%s title=%q members=%d",
 					ce.Correlation.Pattern, ce.Correlation.Title, len(ce.Correlation.Members))
 				if r.stdoutVerbose {
@@ -127,23 +140,24 @@ func (r *stdoutReporter) Report(output reporterdef.ReportOutput) bool {
 			}
 		}
 	}
-	for range output.CorrelatorEvents {
-		r.emittedCounter.Add(1)
-		emitted = true
-	}
-
-	// Ongoing counter: at least one correlation is currently active.
-	if len(output.ActiveCorrelations) > 0 {
-		r.ongoingCounter.Add(1)
-	}
 
-	// Debug log for ongoing correlations.
-	if r.stdoutEnabled {
-		for _, ac := range output.ActiveCorrelations {
-			r.logger.Debugf("[observer] ongoing anomaly correlation: pattern=%s members=%d",
-				ac.Pattern, len(ac.Members))
+	// Ongoing counter: fires when at least one active correlation was already
+	// seen in a prior cycle (i.e. not newly detected this cycle). This mirrors
+	// the pre-refactor semantics where ongoingCounter incremented once per
+	// advance that had any pattern not in the freshly-emitted set.
+	hasOngoing := false
+	for _, ac := range output.ActiveCorrelations {
+		if _, isNew := newlyDetected[ac.Pattern]; !isNew {
+			if r.stdoutEnabled {
+				r.logger.Debugf("[observer] ongoing anomaly correlation: pattern=%s members=%d",
+					ac.Pattern, len(ac.Members))
+			}
+			hasOngoing = true
 		}
 	}
+	if hasOngoing {
+		r.ongoingCounter.Add(1)
+	}
 
 	// Debug log for raw new anomalies detected this cycle.
 	if r.stdoutEnabled {

comp/anomalydetection/reporter/impl/reporter_event.go

@@ -12,9 +12,13 @@ import (
 )
 
 // EventReporter sends Datadog events for correlator lifecycle events.
-// It is a stateless forwarder: all deduplication and recurrence logic lives
-// inside each correlator via the correlationEmitter helper. Reporters simply
-// iterate output.CorrelatorEvents and dispatch to the appropriate sender method.
+// Deduplication and recurrence logic live inside each correlator via the
+// correlationEmitter helper. The reporter forwards each CorrelatorEvent to the
+// appropriate sender method.
+//
+// All events are at-most-once: a transient forwarder failure logs an error and
+// discards the event. The correlator has already drained the event from its
+// pending queue so there is no replay mechanism.
 //
 // It implements reporterdef.StorageConsumer so the observer can inject engine
 // storage post-construction for windowed log-rate annotations in change messages.
@@ -44,8 +48,8 @@ func (r *EventReporter) SetStorage(storage observerdef.StorageReader) {
 //   - EpisodeStarted / EpisodeEnded  → sendEpisodeEvent (scorer severity transitions)
 //   - CorrelationDetected            → send (cluster/pattern first-seen, emitter-deduplicated)
 //
-// Events are at-most-once: a transient forwarder error drops the event (the
-// correlator already drained it). This matches the existing scorer episode model.
+// All events are at-most-once: a transient forwarder error logs the failure and
+// discards the event. The correlator has already drained it from its pending queue.
 func (r *EventReporter) Report(output reporterdef.ReportOutput) bool {
 	emitted := false
 	for _, ce := range output.CorrelatorEvents {