[anomalydetection] Simplify reporters (5)

comp/anomalydetection/AGENTS.md

@@ -93,14 +93,28 @@ and dropped before they reach observer storage.
 ## Reporter Model
 
 Reporters register through the `anomalydetection_reporters` Fx group
-(`reporter/def`). The observer subscribes each injected `Reporter` after each
-advance cycle.
-
-- **StdoutReporter** — always active in `reporter/fx`; logs correlations at
-  info on first-seen, debug for ongoing
-- **EventReporter** — created when `anomaly_detection.reporting.enabled=true`
-  AND the event-platform forwarder is available; publishes change events via
-  `reporter/impl/notify.go`
+(`reporter/def`). The observer calls each injected `Reporter.Report()` after
+every advance cycle.
+
+**Reporters hold no deduplication state.** All first-seen and recurrence logic
+lives inside each correlator via the shared `correlationEmitter` helper
+(`observer/impl/correlation_emitter.go`). Reporters iterate
+`ReportOutput.CorrelatorEvents` and dispatch directly — no per-reporter seen-map.
+
+- **StdoutReporter** — always active in `reporter/fx`; logs
+  `CorrelationDetected` events at info and ongoing active correlations at debug
+- **EventReporter** — created when `anomaly_detection.reporting.events.enabled=true`
+  AND the event-platform forwarder is available; dispatches change events for
+  `CorrelationDetected` and scorer episode events via `reporter/impl/notify.go`.
+  Not fully stateless: it carries a `retryPending` queue of `CorrelationDetected`
+  sends that failed transiently; entries are retried each advance cycle and evicted
+  after `defaultMaxRetryAttempts` consecutive failures.
+
+`ReportOutput.CorrelatorEvents` carries three event kinds:
+- `CorrelatorEventCorrelationDetected` — emitted by `TimeCluster`, `CrossSignal`,
+  `Passthrough` at first-seen (and again after a pattern goes inactive and recurs)
+- `CorrelatorEventEpisodeStarted` — emitted by `anomaly_scorer` on High entry
+- `CorrelatorEventEpisodeEnded` — emitted by `anomaly_scorer` on High exit
 
 See `reporter/reporter.allium` for the payload contract.
 

comp/anomalydetection/observer/AGENTS.md

@@ -32,7 +32,7 @@ and the testbench use the same engine.
 | File | Purpose |
 |------|---------|
 | `def/component.go` | Component interface (GetHandle, RecordSamplerDropped, DumpMetrics) |
-| `def/types.go` | Handle, View types, Detector, Correlator, StorageReader, Anomaly, etc. |
+| `def/types.go` | Handle, View types, Detector, Correlator, StorageReader, Anomaly, CorrelatorEvent, etc. |
 | `impl/engine.go` | Pipeline orchestration: ingest, advance, detect, correlate, replay |
 | `impl/storage.go` | In-memory columnar time-series storage (1s buckets, read-time aggregation) |
 | `impl/scheduler.go` | Scheduling policy: when to advance analysis |
@@ -42,7 +42,9 @@ and the testbench use the same engine.
 | `impl/log_pattern_extractor.go` | Log → virtual metrics via pattern clustering |
 | `impl/log_metrics_extractor.go` | Log → virtual metrics via regex extraction |
 | `impl/anomaly_correlator_time_cluster.go` | Default time-proximity correlator |
+| `impl/anomaly_correlator_passthrough.go` | Passthrough correlator (one ActiveCorrelation per anomaly) |
 | `impl/anomaly_scorer.go` | Unified EWMA anomaly scorer (Correlator + standalone replay) |
+| `impl/correlation_emitter.go` | Shared first-seen/recurrence helper used by all non-scorer correlators |
 | `impl/patterns/` | Tokenizer + clusterer used by log pattern extractor |
 
 ### Component catalog (defaults)
@@ -106,6 +108,34 @@ When `anomaly_detection.metrics.enabled=false`, handles wrap with
 still passes through; log-derived virtual metrics produced inside the engine
 are unaffected.
 
+### Correlator-owned deduplication (`correlationEmitter`)
+
+All correlation event deduplication lives **inside each correlator**, not in reporters.
+`correlationEmitter` (`impl/correlation_emitter.go`) is a shared helper embedded in
+every non-scorer correlator (`TimeCluster`, `CrossSignal`, `Passthrough`). It tracks
+first-seen / recurrence state and produces `CorrelatorEventCorrelationDetected` events
+via `PendingEvents()`. The engine collects pending events after each `Advance` call and
+forwards them to reporters via `ReportOutput.CorrelatorEvents`.
+
+**Recurrence rule:** a pattern that leaves the active set (evicted, timeout) is
+removed from the seen-set, so it re-fires on the next occurrence. This means
+`CorrelationDetected` fires at most once per active episode, and once more each time
+the pattern vanishes and comes back.
+
+**Usage in a correlator:**
+
+```go
+// 1. In Advance — observe BEFORE evicting so batch-evicted clusters still emit.
+e.emitter.observe(e.ActiveCorrelations(), dataTime)
+// 2. In PendingEvents — drain and return.
+return e.emitter.drain()
+// 3. In Reset — clear emitter state alongside correlator state.
+e.emitter.reset()
+```
+
+The scorer uses a different path (`EpisodeStarted` / `EpisodeEnded` events) and does
+not embed a `correlationEmitter`.
+
 ## Common Pitfalls
 
 1. **Don't call engine methods from multiple goroutines.** The engine assumes

comp/anomalydetection/observer/def/types.go

@@ -288,6 +288,41 @@ type SeriesDetector interface {
 	Detect(series Series) DetectionResult
 }
 
+// CorrelatorEventKind identifies the type of a correlator lifecycle event.
+type CorrelatorEventKind int
+
+const (
+	// CorrelatorEventEpisodeStarted fires when the scorer enters High severity.
+	CorrelatorEventEpisodeStarted CorrelatorEventKind = iota + 1
+	// CorrelatorEventEpisodeEnded fires when the scorer leaves High severity.
+	CorrelatorEventEpisodeEnded
+	// CorrelatorEventCorrelationDetected fires when a correlator observes a
+	// pattern for the first time (or after it has gone inactive and recurred).
+	CorrelatorEventCorrelationDetected
+)
+
+// CorrelatorEvent is a typed lifecycle event produced by a correlator during Advance.
+// Reporters receive these via ReportOutput.CorrelatorEvents and can emit backend
+// notifications without relying on the one-shot dedup logic in ActiveCorrelations.
+// Correlators own recurrence detection and produce CorrelationDetected events via
+// a shared emitter; scorer-type correlators produce EpisodeStarted/EpisodeEnded.
+type CorrelatorEvent struct {
+	Kind CorrelatorEventKind
+	// CorrelatorName identifies the correlator that produced this event.
+	CorrelatorName string
+	// Timestamp is the data time (unix seconds) when the event occurred.
+	Timestamp int64
+	// Correlation is the pattern associated with this event.
+	// For EpisodeStarted: the newly opened episode (no end time yet).
+	// For EpisodeEnded: the closed episode with the final LastUpdated.
+	// For CorrelationDetected: the full active correlation at first-seen time.
+	Correlation ActiveCorrelation
+	// FromLevel and ToLevel carry the scorer severity transition.
+	// Populated only for EpisodeStarted/EpisodeEnded; zero for CorrelationDetected.
+	FromLevel SeverityLevel
+	ToLevel   SeverityLevel
+}
+
 // Correlator accumulates anomaly events and produces correlated patterns.
 // Correlators are stateful and cluster/filter/summarize anomaly streams.
 //
@@ -305,6 +340,11 @@ type Correlator interface {
 	Advance(dataTime int64)
 	// ActiveCorrelations returns currently detected correlation patterns.
 	ActiveCorrelations() []ActiveCorrelation
+	// PendingEvents returns and drains typed lifecycle events accumulated during
+	// the last Advance call. The caller owns the returned slice; the correlator
+	// discards it after this call. Returns nil when no events are pending.
+	// Correlators with no lifecycle events (e.g. time-cluster) always return nil.
+	PendingEvents() []CorrelatorEvent
 	// Reset clears all internal state for reanalysis.
 	Reset()
 }

comp/anomalydetection/observer/impl/BUILD.bazel

@@ -11,6 +11,7 @@ go_library(
         "anomaly_scorer.go",
         "component_catalog.go",
         "context_provider.go",
+        "correlation_emitter.go",
         "correlation_identity.go",
         "debug.go",
         "detect_digest.go",
@@ -71,6 +72,7 @@ go_test(
         "bench_main_test.go",
         "bench_storage_listing_test.go",
         "component_catalog_test.go",
+        "correlation_emitter_test.go",
         "engine_stepadvance_test.go",
         "events_test.go",
         "group_compression_test.go",

comp/anomalydetection/observer/impl/anomaly_correlator_passthrough.go

@@ -21,12 +21,14 @@ type DetectorPassthroughCorrelator struct {
 	// anomaliesByDetector groups anomalies by DetectorName.
 	anomaliesByDetector map[string][]observer.Anomaly
 	mu                  sync.RWMutex
+	emitter             correlationEmitter
 }
 
 // NewDetectorPassthroughCorrelator creates a new DetectorPassthroughCorrelator.
 func NewDetectorPassthroughCorrelator() *DetectorPassthroughCorrelator {
 	return &DetectorPassthroughCorrelator{
 		anomaliesByDetector: make(map[string][]observer.Anomaly),
+		emitter:             newCorrelationEmitter("passthrough_correlator"),
 	}
 }
 
@@ -42,31 +44,30 @@ func (c *DetectorPassthroughCorrelator) ProcessAnomaly(anomaly observer.Anomaly)
 	c.anomaliesByDetector[anomaly.DetectorName] = append(c.anomaliesByDetector[anomaly.DetectorName], anomaly)
 }
 
-// Advance is a no-op for the passthrough correlator (no windowing).
-func (c *DetectorPassthroughCorrelator) Advance(_ int64) {}
+// Advance observes the current active correlations for emission.
+func (c *DetectorPassthroughCorrelator) Advance(dataTime int64) {
+	c.mu.RLock()
+	active := c.activeCorrelationsLocked()
+	c.mu.RUnlock()
+	c.emitter.observe(active, dataTime)
+}
 
 // Reset clears all internal state for reanalysis.
 func (c *DetectorPassthroughCorrelator) Reset() {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	c.anomaliesByDetector = make(map[string][]observer.Anomaly)
+	c.emitter.reset()
 }
 
-// ActiveCorrelations returns one ActiveCorrelation per individual anomaly,
-// sorted by detector name then timestamp.
-//
-// Each anomaly becomes its own correlation with:
-//   - Pattern:     "passthrough_{detectorName}_{index}"
-//   - Title:       "Passthrough[{detectorName}]: {anomaly.Source}"
-//   - Anomalies:   single-element slice containing the original anomaly
-//   - FirstSeen/LastUpdated: the anomaly's timestamp
-//
-// This allows the scorer to evaluate each detection independently.
-func (c *DetectorPassthroughCorrelator) ActiveCorrelations() []observer.ActiveCorrelation {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
+// PendingEvents drains CorrelationDetected events accumulated during the last Advance.
+func (c *DetectorPassthroughCorrelator) PendingEvents() []observer.CorrelatorEvent {
+	return c.emitter.drain()
+}
 
-	// Sort detector names for deterministic ordering
+// activeCorrelationsLocked builds active correlations from current state.
+// Caller must hold c.mu (at least read lock).
+func (c *DetectorPassthroughCorrelator) activeCorrelationsLocked() []observer.ActiveCorrelation {
 	detectorNames := make([]string, 0, len(c.anomaliesByDetector))
 	for name := range c.anomaliesByDetector {
 		detectorNames = append(detectorNames, name)
@@ -77,7 +78,6 @@ func (c *DetectorPassthroughCorrelator) ActiveCorrelations() []observer.ActiveCo
 	for _, detName := range detectorNames {
 		anomalies := c.anomaliesByDetector[detName]
 
-		// Sort anomalies by timestamp for deterministic output
 		sorted := make([]observer.Anomaly, len(anomalies))
 		copy(sorted, anomalies)
 		sort.Slice(sorted, func(i, j int) bool { return sorted[i].Timestamp < sorted[j].Timestamp })
@@ -93,6 +93,14 @@ func (c *DetectorPassthroughCorrelator) ActiveCorrelations() []observer.ActiveCo
 			})
 		}
 	}
-
 	return result
 }
+
+// ActiveCorrelations returns one ActiveCorrelation per individual anomaly,
+// sorted by detector name then timestamp. Each anomaly becomes its own
+// correlation, allowing the scorer to evaluate each detection independently.
+func (c *DetectorPassthroughCorrelator) ActiveCorrelations() []observer.ActiveCorrelation {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.activeCorrelationsLocked()
+}

comp/anomalydetection/observer/impl/anomaly_correlator_time_cluster.go

@@ -65,6 +65,7 @@ type TimeClusterCorrelator struct {
 	nextClusterID   int
 	currentDataTime int64
 	mu              sync.RWMutex
+	emitter         correlationEmitter
 }
 
 // NewTimeClusterCorrelator creates a new TimeClusterCorrelator with the given config.
@@ -76,8 +77,8 @@ func NewTimeClusterCorrelator(config TimeClusterConfig) *TimeClusterCorrelator {
 		config.WindowSeconds = DefaultTimeClusterConfig().WindowSeconds
 	}
 	return &TimeClusterCorrelator{
-		config:   config,
-		clusters: nil,
+		config:  config,
+		emitter: newCorrelationEmitter("time_cluster_correlator"),
 	}
 }
 
@@ -201,13 +202,16 @@ func (c *TimeClusterCorrelator) mergeClusters(clusters []*timeCluster) *timeClus
 	return merged
 }
 
-// Advance evicts old clusters past the retention window (reporters pull state via ActiveCorrelations).
+// Advance evicts old clusters past the retention window.
+// The emitter is observed BEFORE eviction so that batch clusters evicted in the
+// same cycle still produce a CorrelationDetected event.
 func (c *TimeClusterCorrelator) Advance(dataTime int64) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if dataTime > c.currentDataTime {
 		c.currentDataTime = dataTime
 	}
+	c.emitter.observe(c.activeCorrelationsLocked(), dataTime)
 	c.evictOldClustersLocked()
 }
 
@@ -219,6 +223,7 @@ func (c *TimeClusterCorrelator) Reset() {
 	c.clusters = c.clusters[:0]
 	c.nextClusterID = 0
 	c.currentDataTime = 0
+	c.emitter.reset()
 }
 
 // evictOldClustersLocked removes clusters whose latest timestamp is outside the window.
@@ -309,13 +314,15 @@ func (c *TimeClusterCorrelator) GetExtraData() interface{} {
 	return c.GetClusters()
 }
 
-// ActiveCorrelations returns clusters as active correlation patterns.
-func (c *TimeClusterCorrelator) ActiveCorrelations() []observer.ActiveCorrelation {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
+// PendingEvents drains CorrelationDetected events accumulated during the last Advance.
+func (c *TimeClusterCorrelator) PendingEvents() []observer.CorrelatorEvent {
+	return c.emitter.drain()
+}
 
+// activeCorrelationsLocked builds the active correlations from current clusters.
+// Caller must hold c.mu (at least read lock).
+func (c *TimeClusterCorrelator) activeCorrelationsLocked() []observer.ActiveCorrelation {
 	var result []observer.ActiveCorrelation
-
 	for _, cluster := range c.clusters {
 		if c.config.MinClusterSize > 0 && len(cluster.anomalies) < c.config.MinClusterSize {
 			continue
@@ -329,14 +336,18 @@ func (c *TimeClusterCorrelator) ActiveCorrelations() []observer.ActiveCorrelatio
 			LastUpdated: cluster.maxTimestamp,
 		})
 	}
-
-	// Sort by cluster size (largest first), then by time
 	sort.Slice(result, func(i, j int) bool {
 		if len(result[i].Anomalies) != len(result[j].Anomalies) {
 			return len(result[i].Anomalies) > len(result[j].Anomalies)
 		}
 		return result[i].FirstSeen < result[j].FirstSeen
 	})
-
 	return result
 }
+
+// ActiveCorrelations returns clusters as active correlation patterns.
+func (c *TimeClusterCorrelator) ActiveCorrelations() []observer.ActiveCorrelation {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.activeCorrelationsLocked()
+}

comp/anomalydetection/observer/impl/anomaly_processor_correlator.go

@@ -72,6 +72,7 @@ type CrossSignalCorrelator struct {
 	buffer             []timestampedAnomaly
 	activeCorrelations map[string]*observer.ActiveCorrelation
 	currentDataTime    int64 // latest data timestamp seen
+	emitter            correlationEmitter
 }
 
 // NewCorrelator creates a new CrossSignalCorrelator with the given config.
@@ -82,9 +83,8 @@ func NewCorrelator(config CorrelatorConfig) *CrossSignalCorrelator {
 	}
 	return &CrossSignalCorrelator{
 		config:             config,
-		buffer:             nil,
 		activeCorrelations: make(map[string]*observer.ActiveCorrelation),
-		currentDataTime:    0,
+		emitter:            newCorrelationEmitter("cross_signal_correlator"),
 	}
 }
 
@@ -179,13 +179,16 @@ func (c *CrossSignalCorrelator) Advance(dataTime int64) {
 			delete(c.activeCorrelations, name)
 		}
 	}
+
+	c.emitter.observe(c.ActiveCorrelations(), dataTime)
 }
 
 // Reset clears all internal state for reanalysis.
 func (c *CrossSignalCorrelator) Reset() {
 	c.buffer = nil
 	c.activeCorrelations = make(map[string]*observer.ActiveCorrelation)
 	c.currentDataTime = 0
+	c.emitter.reset()
 }
 
 // patternMatches checks if all required sources for a pattern are present.
@@ -231,6 +234,11 @@ func (c *CrossSignalCorrelator) getBuffer() []timestampedAnomaly {
 	return c.buffer
 }
 
+// PendingEvents drains CorrelationDetected events accumulated during the last Advance.
+func (c *CrossSignalCorrelator) PendingEvents() []observer.CorrelatorEvent {
+	return c.emitter.drain()
+}
+
 // ActiveCorrelations returns a copy of the currently active correlation patterns.
 func (c *CrossSignalCorrelator) ActiveCorrelations() []observer.ActiveCorrelation {
 	result := make([]observer.ActiveCorrelation, 0, len(c.activeCorrelations))