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

### What does this PR do?

Unifies the anomaly scorer into a single `anomalyScorer` struct implementing the `Correlator` interface in a single file. We can now benchmark the scorer. This removes the helper reporting logs, it's merged into this struct.

This contains the main score logic and also the subscribe logic which is used to create events (high severity = event).

> [!NOTE]
> We use "the severity becomes high" and "the severity isn't high anymore" as start / end of correlation. This will be reworked in #52493 to directly emit event at the start.

The config can now enable/disable logs and reporter events (telemetry always enabled if component enabled).

New config keys for output:
```yaml
anomaly_detection:
  anomaly_scorer:
    enabled: true
    output:
      logs: true
      correlation_events: false
```

#### Diagram

```mermaid
flowchart LR
    IN["ProcessAnomaly(a)"] --> PEND["pending map"]

    PEND --> ADV

    subgraph ADV["Advance(t)  — mu held"]
        EWMA["merge → evict → bucket → EWMA"]
    end

    ADV -->|"[]secEWMA\nmu released"| SUBS

    subgraph SUBS["Per-subscription state machines  — subsMu"]
        FSM["Low / Medium / High FSM\n+ cooldown per sub"]
        FSM -->|"transition"| CB["Listener.OnSeverityTransition(evt)"]
    end

    CB -->|"self-sub\n(internal watcher)"| IW["gauges · logs · episode tracking"]

    IW --> AC["ActiveCorrelations()"]
    ADV --> SC["ScoreState() / LastScore()"]
    SUBS --> SB["Subscribe(cfg) → func()\nexposes via SubscribeScorer on Component"]
```

### Motivation

Nice score for kafka scenario, no false positives (using bocpd+tukey_biweight+holt_residual):

<img width="531" height="304" alt="Screenshot 2026-06-19 at 11 20 44" src="https://github.com/user-attachments/assets/ace05855-359a-4064-b42c-994b6baf5a1e" />

### Describe how you validated your changes

### Additional Notes

Co-authored-by: celian.raimbault <celian.raimbault@datadoghq.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",
     ],