[anomalydetection] Unify scorer, make it a correlator (4)

Unify the anomaly scorer into a single correlator-backed implementation, move scorer event emission into the correlator path, and update the testbench/config wiring to use the dedicated anomaly_scorer namespace without regressing newer main-branch changes.

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

Author: CelianR

comp/anomalydetection/observer/AGENTS.md

@@ -42,6 +42,7 @@ 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_scorer.go` | Unified EWMA anomaly scorer (Correlator + standalone replay) |
 | `impl/patterns/` | Tokenizer + clusterer used by log pattern extractor |
 
 ### Component catalog (defaults)
@@ -58,8 +59,27 @@ Registered in `impl/component_catalog.go`. Enabled by default unless noted:
 | Detector | `cusum`, `scanmw`, `scanwelch`, `holt_residual`, `tukey_biweight` | off |
 | Correlator | `time_cluster` | on |
 | Correlator | `cross_signal`, `passthrough` | off |
+| Correlator | `anomaly_scorer` | off |
+
+Toggle detectors/correlators/extractors via `anomaly_detection.detectors.<name>.enabled` in datadog.yaml.
+
+The `anomaly_scorer` correlator has a **dedicated config namespace** under `anomaly_detection.anomaly_scorer.*` (not `detectors.*`) with an `output` sub-section controlling logs and correlation events:
+
+```yaml
+anomaly_detection:
+  anomaly_scorer:
+    enabled: true
+    alpha: 0.3
+    window_secs: 30
+    low_threshold: 0.030
+    high_threshold: 0.060
+    output:
+      logs: true
+      correlation_events: false
+      cooldown_secs: 300
+```
 
-Toggle via `anomaly_detection.detectors.<name>.enabled` in datadog.yaml.
+The scorer is also available standalone (without the engine) via `NewAnomalyScorer` in `impl/` for testbench replay.
 
 ## Key Design Decisions
 

comp/anomalydetection/observer/def/types.go

@@ -309,8 +309,8 @@ type Correlator interface {
 	Reset()
 }
 
-// ScorerConfig holds the tunable parameters for the anomaly scoring pipeline.
-type ScorerConfig struct {
+// AnomalyScorerConfig holds the tunable parameters for the anomaly scoring pipeline.
+type AnomalyScorerConfig struct {
 	// Alpha is the EWMA smoothing factor (0 < α ≤ 1). Lower = smoother.
 	Alpha float64 `json:"alpha"`
 	// SaturationK is the saturation constant k: saturation = 1−exp(−n/k).
@@ -332,16 +332,16 @@ type ScorerConfig struct {
 	// specific detector names. Each entry is [low, medium, high, xhigh] thresholds.
 	// Detectors not in this map default to level 2 (Medium) regardless of their score.
 	DetectorThresholds map[string][4]float64 `json:"detector_thresholds,omitempty"`
-	// MaxBuckets overrides the number of ScoreBucket entries retained in
+	// MaxBuckets overrides the number of AnomalyScoreBucket entries retained in
 	// ScoreState(). 0 (default) means "cap at WindowSecs", which is the
 	// correct behaviour for the live agent. Set to a large positive value
 	// (e.g. math.MaxInt64) to keep an unlimited history for offline replay.
 	MaxBuckets int64 `json:"max_buckets,omitempty"`
 }
 
-// ScoreBucket is the per-second telemetry unit emitted by the scorer.
+// AnomalyScoreBucket is the per-second telemetry unit emitted by the scorer.
 // One bucket is produced for every 1-second tick, even if it has no anomalies.
-type ScoreBucket struct {
+type AnomalyScoreBucket struct {
 	// Second is the Unix timestamp (floor) for this bucket.
 	Second int64 `json:"second"`
 	// Bins[L] is the number of deduplicated anomalies at level L (0=VeryLow … 4=XHigh).
@@ -363,17 +363,17 @@ const (
 	SeverityHigh   SeverityLevel = 2
 )
 
-// ScorerEventDirection describes whether a severity transition is an
+// AnomalyScorerEventDirection describes whether a severity transition is an
 // escalation or de-escalation.
-type ScorerEventDirection int
+type AnomalyScorerEventDirection int
 
 const (
-	// ScorerEventBoth delivers transitions in either direction (default zero value).
-	ScorerEventBoth ScorerEventDirection = 0
-	// ScorerEventEscalation delivers only transitions where ToLevel > FromLevel.
-	ScorerEventEscalation ScorerEventDirection = 1
-	// ScorerEventDeescalation delivers only transitions where ToLevel < FromLevel.
-	ScorerEventDeescalation ScorerEventDirection = 2
+	// AnomalyScorerEventBoth delivers transitions in either direction (default zero value).
+	AnomalyScorerEventBoth AnomalyScorerEventDirection = 0
+	// AnomalyScorerEventEscalation delivers only transitions where ToLevel > FromLevel.
+	AnomalyScorerEventEscalation AnomalyScorerEventDirection = 1
+	// AnomalyScorerEventDeescalation delivers only transitions where ToLevel < FromLevel.
+	AnomalyScorerEventDeescalation AnomalyScorerEventDirection = 2
 )
 
 // SeverityEvent records a severity state-machine transition.
@@ -384,72 +384,48 @@ type SeverityEvent struct {
 	FromLevel SeverityLevel `json:"from_level"`
 	// ToLevel is the state after the transition.
 	ToLevel SeverityLevel `json:"to_level"`
-	// Direction is ScorerEventEscalation when ToLevel > FromLevel, and
-	// ScorerEventDeescalation when ToLevel < FromLevel.
-	Direction ScorerEventDirection `json:"direction"`
+	// Direction is AnomalyScorerEventEscalation when ToLevel > FromLevel, and
+	// AnomalyScorerEventDeescalation when ToLevel < FromLevel.
+	Direction AnomalyScorerEventDirection `json:"direction"`
 }
 
-// ScorerListener receives severity state-machine transitions from the scorer.
-type ScorerListener interface {
+// AnomalyScorerListener receives severity state-machine transitions from the scorer.
+type AnomalyScorerListener interface {
 	OnSeverityTransition(event SeverityEvent)
 }
 
-// ScorerEventFilter selects which SeverityEvents are delivered to a listener.
+// AnomalyScorerEventFilter selects which SeverityEvents are delivered to a listener.
 // All conditions are ANDed; a nil or empty slice means "any value".
-// The zero value ScorerEventFilter{} matches every transition.
-type ScorerEventFilter struct {
+// The zero value AnomalyScorerEventFilter{} matches every transition.
+type AnomalyScorerEventFilter struct {
 	// FromLevels restricts to events whose FromLevel is in the set.
 	FromLevels []SeverityLevel
 	// ToLevels restricts to events whose ToLevel is in the set.
 	ToLevels []SeverityLevel
 	// Direction restricts by escalation or de-escalation.
-	Direction ScorerEventDirection
+	Direction AnomalyScorerEventDirection
 }
 
-// AnomalyScorerConfiguration is the single object passed to SubscribeScorer
+// AnomalyScorerConfiguration is the single object passed to Subscribe
 // when registering a listener. It bundles who to call (Listener), which
 // transitions to deliver (Filter), and per-subscription state-machine tuning.
 type AnomalyScorerConfiguration struct {
 	// Listener is called for each matching severity transition. Required.
-	Listener ScorerListener
+	Listener AnomalyScorerListener
 	// Filter controls which transitions are delivered.
-	// Zero value ScorerEventFilter{} delivers all transitions.
-	Filter ScorerEventFilter
+	// Zero value AnomalyScorerEventFilter{} delivers all transitions.
+	Filter AnomalyScorerEventFilter
 	// CooldownSecs is the minimum number of seconds that must elapse after a
 	// delivered transition before a downward (de-escalation) transition can be
 	// delivered again.
 	// Zero means no cooldown (every matching transition is delivered).
 	CooldownSecs int64
 }
 
-// ScoreState is the accumulated telemetry snapshot from the scorer.
-type ScoreState struct {
-	Buckets []ScoreBucket `json:"buckets"`
-	Config  ScorerConfig  `json:"config"`
-}
-
-// AnomalyScorer computes a smoothed anomaly intensity signal (EWMA) from the
-// stream of anomalies produced by the detection pipeline. It mirrors the
-// Correlator lifecycle: ProcessAnomaly → Advance (once per second tick) → ScoreState → Reset.
-type AnomalyScorer interface {
-	// Name returns the scorer name for debugging.
-	Name() string
-	// ProcessAnomaly feeds a raw anomaly into the scorer's current-second buffer.
-	ProcessAnomaly(a Anomaly)
-	// Advance finalises the bucket at dataTime (unix seconds) and runs the
-	// EWMA + state-machine update for that second. Callers must invoke this
-	// after each 1-second detection cycle.
-	Advance(dataTime int64)
-	// LastScore returns the most recently computed EWMA score. Returns 0
-	// before the first Advance call.
-	LastScore() float64
-	// ScoreState returns the accumulated telemetry (all buckets + events so far).
-	ScoreState() ScoreState
-	// Reset clears all internal state for reanalysis.
-	Reset()
-	// Subscribe registers a listener to receive severity transitions matching
-	// cfg.Filter. Returns an unsubscribe function. Safe to call concurrently.
-	Subscribe(cfg AnomalyScorerConfiguration) func()
+// AnomalyScoreState is the accumulated telemetry snapshot from the scorer.
+type AnomalyScoreState struct {
+	Buckets []AnomalyScoreBucket `json:"buckets"`
+	Config  AnomalyScorerConfig  `json:"config"`
 }
 
 // Reporter receives reports and displays or delivers them.

comp/anomalydetection/observer/impl/BUILD.bazel

@@ -9,7 +9,6 @@ go_library(
         "anomaly_correlator_time_cluster.go",
         "anomaly_processor_correlator.go",
         "anomaly_scorer.go",
-        "anomaly_scorer_helper.go",
         "component_catalog.go",
         "context_provider.go",
         "correlation_identity.go",
@@ -49,13 +48,10 @@ go_library(
         "//comp/anomalydetection/recorder/def",
         "//comp/anomalydetection/reporter/def",
         "//comp/core/config",
-        "//comp/core/hostname/hostnameinterface/def",
         "//comp/core/log/def",
         "//comp/core/telemetry/def",
         "//comp/core/telemetry/impl/noops",
         "//comp/def",
-        "//comp/forwarder/eventplatform/def",
-        "//pkg/logs/message",
         "//pkg/util/log",
         "//pkg/util/option",
     ],