feat(alerts): deterministic metric-transition reducer with manager parity harness

Phase 1 of docs/ALERT_ENGINE_EVOLUTION.md. internal/alerts/reducer is a pure transition core for the metric-threshold family — hysteresis trigger/clear, sustained-for delay with dip reset, warning/critical severity derivation with the percentage 99-cap — characterized from Manager.checkMetric, with time entering only through the signal's ObservedAt so every sequence is deterministic and replayable. reducer_parity_test.go drives the live manager and the reducer through identical observation sequences (simulated time via the established pending-backdate trick) and fails on any divergence after every step, with the manager as the reference. The harness proved itself on its first run by catching the stale-resolve defect fixed in the previous commit. Registers coverage gap alert-lifecycle-contract-coverage. Remaining slices: offline/confirmation families, ack lifecycle, intent interaction, shadow-mode runtime feed.
2026-10-19 10:33:29 +00:00 · 2026-08-26 22:53:10 +01:00 · 2026-08-26 22:53:10 +01:00 · 7c212b906e
commit 7c212b906e
parent 20ea1ea4d3
5 changed files with 828 additions and 2 deletions
--- a/docs/release-control/v6/internal/records/alert-lifecycle-contract-coverage-2026-08-26.md
+++ b/docs/release-control/v6/internal/records/alert-lifecycle-contract-coverage-2026-08-26.md
@ -0,0 +1,38 @@
+# Alert lifecycle transitions lack a deterministic contract harness
+
+Recorded: 2026-08-26
+Origin: alerting audit (2026-08-26) finding 2 and docs/ALERT_ENGINE_EVOLUTION.md
+Phase 1.
+
+## Gap
+
+The alert manager's transition semantics (hysteresis trigger/clear,
+sustained-for delays, severity derivation, confirmation counts) are
+implemented imperatively across per-family check paths and asserted only
+by per-unit tests. There is no deterministic, engine-independent statement
+of the lifecycle contract, so regressions in the frozen transition core
+keep shipping (#1682, #1683, #1553, #1693, #1724) and cross-family
+semantic drift is invisible.
+
+## Resolution
+
+Phase 1 of docs/ALERT_ENGINE_EVOLUTION.md: a pure transition reducer
+(`internal/alerts/reducer`) that characterizes the manager's semantics
+family by family, plus a parity harness (`reducer_parity_test.go` in
+`internal/alerts`) that drives both engines through identical observation
+sequences and fails on any divergence. The reducer must characterize, not
+improve: divergences are reducer bugs unless investigation proves a
+manager defect, which is then fixed in the manager first.
+
+First slice (2026-08-26): the metric-threshold family. The harness
+immediately caught a real manager defect: `checkMetric` hysteresis
+resolution removed alerts by the legacy `<resourceID>-<metric>` ID while
+canonical-identity alerts are stored under the canonical state key and the
+legacy ID carries no alias, so the removal silently no-oped — a resolved
+notification went out while the alert stayed active and re-resolved every
+poll (guest per-disk usage alerts were the remaining production caller;
+user-visible as the stale-alert class, e.g. #1580). Fixed at the caller
+with a focused regression test before the parity slice landed.
+
+Remaining for later slices: offline/confirmation families, ack/snooze
+lifecycle, intent-policy interaction, and a shadow-mode runtime feed.
--- a/docs/release-control/v6/internal/status.json
+++ b/docs/release-control/v6/internal/status.json
@ -1,6 +1,6 @@
 {
  "version": "6.0",
-  "updated_at": "2026-08-08",
+  "updated_at": "2026-08-26",
  "scope": {
    "active_repos": [
      "pulse",
@ -9787,10 +9787,52 @@
          "kind": "file"
        }
      ]
+    },
+    {
+      "id": "alert-lifecycle-contract-coverage",
+      "summary": "Alert transition semantics (hysteresis, sustained-for delays, severity derivation, confirmations) have no deterministic engine-independent contract harness, so transition-core regressions ship undetected (#1682, #1683, #1553, #1693, #1724) and cross-family drift is invisible. Resolution is Phase 1 of docs/ALERT_ENGINE_EVOLUTION.md: a pure characterization reducer (internal/alerts/reducer) plus a parity harness driving reducer and manager through identical sequences; the first metric-family slice already surfaced and fixed a silent stale-resolve defect in checkMetric.",
+      "owner": "project-owner",
+      "status": "triaged",
+      "recorded_at": "2026-08-26",
+      "lane_ids": [
+        "L6",
+        "L13"
+      ],
+      "subsystem_ids": [
+        "alerts"
+      ],
+      "proposed_resolution": "lane-expansion",
+      "coverage_impact": 5,
+      "evidence": [
+        {
+          "repo": "pulse",
+          "path": "docs/release-control/v6/internal/records/alert-lifecycle-contract-coverage-2026-08-26.md",
+          "kind": "file"
+        },
+        {
+          "repo": "pulse",
+          "path": "docs/ALERT_ENGINE_EVOLUTION.md",
+          "kind": "file"
+        }
+      ]
    }
  ],
  "candidate_lanes": [],
-  "work_claims": [],
+  "work_claims": [
+    {
+      "id": "claude-alert-reducer-coverage-gap-alert-lifecycle-contract-coverage",
+      "agent_id": "claude-alert-reducer",
+      "summary": "Phase 1 slice: metric-family reducer + parity harness; includes checkMetric stale-resolve fix",
+      "target_id": "v6-product-lane-expansion",
+      "claimed_at": "2026-08-26T21:52:17Z",
+      "heartbeat_at": "2026-08-26T21:52:17Z",
+      "expires_at": "2026-08-27T01:52:17Z",
+      "work_item": {
+        "kind": "coverage-gap",
+        "id": "alert-lifecycle-contract-coverage"
+      }
+    }
+  ],
  "open_decisions": [],
  "source_of_truth_file": "docs/release-control/v6/internal/SOURCE_OF_TRUTH.md",
  "resolved_decisions": [
--- a/internal/alerts/reducer/reducer.go
+++ b/internal/alerts/reducer/reducer.go
@ -0,0 +1,278 @@
+// Package reducer is the deterministic alert transition core scoped by
+// docs/ALERT_ENGINE_EVOLUTION.md Phase 1. It reimplements the alert
+// manager's transition semantics as a pure function over explicit state:
+// no locks, no I/O, no wall clock — time enters only through the signal's
+// ObservedAt. That determinism is what lets the parity harness in the
+// alerts package diff this core against the live manager on identical
+// input sequences, and what will let the lifecycle contract suite run
+// exhaustively once families cut over (Phase 2).
+//
+// Scope of this slice: the metric-threshold family — hysteresis
+// trigger/clear, sustained-for delay with dip reset, and warning/critical
+// severity derivation. The semantics are characterized from
+// Manager.checkMetric, not designed fresh; behavior differences are bugs
+// here, never improvements. Notification policy (cooldown, quiet hours,
+// rate limits) is out of scope by design: it stays a downstream consumer
+// of transition events.
+package reducer
+
+import (
+	"fmt"
+	"time"
+)
+
+// MetricSignal is one observation of one metric on one resource.
+type MetricSignal struct {
+	ResourceID string
+	Metric     string
+	Value      float64
+	ObservedAt time.Time
+}
+
+// MetricRule is the resolved threshold policy for one resource+metric.
+// A Trigger <= 0 disables the rule: any pending or firing incident for the
+// key is cleared, mirroring the manager's disabled-threshold path.
+type MetricRule struct {
+	Trigger float64
+	// Clear is the hysteresis release threshold. When <= 0 it falls back to
+	// Trigger, exactly as the manager does.
+	Clear float64
+	// DelaySeconds is the sustained-above-trigger duration required before
+	// firing. Zero fires on the first exceeding observation.
+	DelaySeconds int
+}
+
+// Severity mirrors the manager's two-level model.
+type Severity string
+
+const (
+	SeverityWarning  Severity = "warning"
+	SeverityCritical Severity = "critical"
+)
+
+// IncidentState is the explicit lifecycle state of one incident.
+type IncidentState string
+
+const (
+	// StatePending: the value is above trigger but the sustained-for delay
+	// has not elapsed yet.
+	StatePending IncidentState = "pending"
+	// StateFiring: the incident is active.
+	StateFiring IncidentState = "firing"
+)
+
+// Incident is the reducer's record for one resource+metric key. Absence
+// from State means idle.
+type Incident struct {
+	ResourceID     string
+	Metric         string
+	State          IncidentState
+	Severity       Severity
+	PendingSince   time.Time
+	StartedAt      time.Time
+	LastValue      float64
+	LastObservedAt time.Time
+}
+
+// EventType enumerates transition events emitted by Apply.
+type EventType string
+
+const (
+	EventPending         EventType = "pending"
+	EventPendingCleared  EventType = "pending_cleared"
+	EventFired           EventType = "fired"
+	EventSeverityChanged EventType = "severity_changed"
+	EventResolved        EventType = "resolved"
+)
+
+// Event is one transition emitted by Apply.
+type Event struct {
+	Type       EventType
+	ResourceID string
+	Metric     string
+	Severity   Severity
+	Value      float64
+	At         time.Time
+}
+
+// State holds every incident the reducer tracks. It is not safe for
+// concurrent use; the owner serializes Apply calls.
+type State struct {
+	incidents map[string]*Incident
+}
+
+// NewState returns an empty reducer state.
+func NewState() *State {
+	return &State{incidents: make(map[string]*Incident)}
+}
+
+func incidentKey(resourceID, metric string) string {
+	return fmt.Sprintf("%s\x00%s", resourceID, metric)
+}
+
+// Incident returns a copy of the tracked incident for the key, if any.
+func (s *State) Incident(resourceID, metric string) (Incident, bool) {
+	if s == nil {
+		return Incident{}, false
+	}
+	incident, ok := s.incidents[incidentKey(resourceID, metric)]
+	if !ok {
+		return Incident{}, false
+	}
+	return *incident, true
+}
+
+// FiringCount reports how many incidents are currently firing.
+func (s *State) FiringCount() int {
+	count := 0
+	for _, incident := range s.incidents {
+		if incident.State == StateFiring {
+			count++
+		}
+	}
+	return count
+}
+
+// isPercentageMetric mirrors the manager's classification: these metrics
+// live on a 0–100 scale, which caps the derived critical threshold.
+func isPercentageMetric(metric string) bool {
+	switch metric {
+	case "cpu", "memory", "disk", "usage":
+		return true
+	default:
+		return false
+	}
+}
+
+// criticalThreshold mirrors the manager: critical is trigger+10, capped at
+// 99 for percentage metrics so escalation stays reachable at high triggers.
+func criticalThreshold(trigger float64, metric string) float64 {
+	critical := trigger + 10
+	if isPercentageMetric(metric) && critical > 99 {
+		return 99
+	}
+	return critical
+}
+
+func severityFor(value, trigger float64, metric string) Severity {
+	if value >= criticalThreshold(trigger, metric) {
+		return SeverityCritical
+	}
+	return SeverityWarning
+}
+
+// ApplyMetric advances the state for one metric observation under one rule
+// and returns the transition events, in order. It is deterministic: the
+// same state, signal, and rule always produce the same result.
+func (s *State) ApplyMetric(signal MetricSignal, rule MetricRule) []Event {
+	key := incidentKey(signal.ResourceID, signal.Metric)
+	incident := s.incidents[key]
+
+	event := func(eventType EventType, severity Severity) Event {
+		return Event{
+			Type:       eventType,
+			ResourceID: signal.ResourceID,
+			Metric:     signal.Metric,
+			Severity:   severity,
+			Value:      signal.Value,
+			At:         signal.ObservedAt,
+		}
+	}
+
+	// Disabled rule: drop pending, resolve firing. Mirrors the manager's
+	// nil/zero-trigger clear path.
+	if rule.Trigger <= 0 {
+		if incident == nil {
+			return nil
+		}
+		delete(s.incidents, key)
+		if incident.State == StateFiring {
+			return []Event{event(EventResolved, incident.Severity)}
+		}
+		return []Event{event(EventPendingCleared, "")}
+	}
+
+	if signal.Value >= rule.Trigger {
+		// Firing already: refresh value and re-derive severity. The manager
+		// recomputes level on every tick, so severity can demote as well as
+		// escalate while firing.
+		if incident != nil && incident.State == StateFiring {
+			previous := incident.Severity
+			incident.Severity = severityFor(signal.Value, rule.Trigger, signal.Metric)
+			incident.LastValue = signal.Value
+			incident.LastObservedAt = signal.ObservedAt
+			if incident.Severity != previous {
+				return []Event{event(EventSeverityChanged, incident.Severity)}
+			}
+			return nil
+		}
+
+		// Sustained-for delay: enter pending on the first exceeding
+		// observation, fire once the delay has elapsed. StartedAt is the
+		// pending entry time, matching the manager's use of the pending
+		// timestamp as the alert start time.
+		if rule.DelaySeconds > 0 {
+			if incident == nil {
+				s.incidents[key] = &Incident{
+					ResourceID:     signal.ResourceID,
+					Metric:         signal.Metric,
+					State:          StatePending,
+					PendingSince:   signal.ObservedAt,
+					LastValue:      signal.Value,
+					LastObservedAt: signal.ObservedAt,
+				}
+				return []Event{event(EventPending, "")}
+			}
+			if signal.ObservedAt.Sub(incident.PendingSince) < time.Duration(rule.DelaySeconds)*time.Second {
+				incident.LastValue = signal.Value
+				incident.LastObservedAt = signal.ObservedAt
+				return nil
+			}
+			severity := severityFor(signal.Value, rule.Trigger, signal.Metric)
+			incident.State = StateFiring
+			incident.Severity = severity
+			incident.StartedAt = incident.PendingSince
+			incident.LastValue = signal.Value
+			incident.LastObservedAt = signal.ObservedAt
+			return []Event{event(EventFired, severity)}
+		}
+
+		severity := severityFor(signal.Value, rule.Trigger, signal.Metric)
+		s.incidents[key] = &Incident{
+			ResourceID:     signal.ResourceID,
+			Metric:         signal.Metric,
+			State:          StateFiring,
+			Severity:       severity,
+			StartedAt:      signal.ObservedAt,
+			LastValue:      signal.Value,
+			LastObservedAt: signal.ObservedAt,
+		}
+		return []Event{event(EventFired, severity)}
+	}
+
+	// Below trigger.
+	if incident == nil {
+		return nil
+	}
+
+	if incident.State == StatePending {
+		// A dip below trigger resets the sustained-for delay entirely.
+		delete(s.incidents, key)
+		return []Event{event(EventPendingCleared, "")}
+	}
+
+	clear := rule.Clear
+	if clear <= 0 {
+		clear = rule.Trigger
+	}
+	if signal.Value <= clear {
+		severity := incident.Severity
+		delete(s.incidents, key)
+		return []Event{event(EventResolved, severity)}
+	}
+
+	// Hysteresis hold: between clear and trigger the incident stays firing.
+	// The manager does not refresh value or last-seen in this band; the
+	// reducer mirrors that so parity diffs stay exact.
+	return nil
+}
--- a/internal/alerts/reducer/reducer_test.go
+++ b/internal/alerts/reducer/reducer_test.go
@ -0,0 +1,211 @@
+package reducer
+
+import (
+	"testing"
+	"time"
+)
+
+var t0 = time.Date(2026, 8, 26, 12, 0, 0, 0, time.UTC)
+
+func signalAt(value float64, offset time.Duration) MetricSignal {
+	return MetricSignal{
+		ResourceID: "vm-100",
+		Metric:     "cpu",
+		Value:      value,
+		ObservedAt: t0.Add(offset),
+	}
+}
+
+func eventTypes(events []Event) []EventType {
+	types := make([]EventType, 0, len(events))
+	for _, event := range events {
+		types = append(types, event.Type)
+	}
+	return types
+}
+
+func TestFireHoldResolveWithHysteresis(t *testing.T) {
+	state := NewState()
+	rule := MetricRule{Trigger: 80, Clear: 75}
+
+	events := state.ApplyMetric(signalAt(85, 0), rule)
+	if len(events) != 1 || events[0].Type != EventFired || events[0].Severity != SeverityWarning {
+		t.Fatalf("events = %+v, want one warning fired", events)
+	}
+
+	// Between clear and trigger: hold, no events, no state refresh.
+	if events := state.ApplyMetric(signalAt(78, time.Minute), rule); len(events) != 0 {
+		t.Fatalf("hysteresis hold emitted %+v", events)
+	}
+	incident, ok := state.Incident("vm-100", "cpu")
+	if !ok || incident.State != StateFiring || incident.LastValue != 85 {
+		t.Fatalf("incident = %+v, want firing with unrefreshed value 85", incident)
+	}
+
+	events = state.ApplyMetric(signalAt(74, 2*time.Minute), rule)
+	if len(events) != 1 || events[0].Type != EventResolved {
+		t.Fatalf("events = %+v, want resolved", events)
+	}
+	if _, ok := state.Incident("vm-100", "cpu"); ok {
+		t.Fatal("incident should be gone after resolve")
+	}
+}
+
+func TestSeverityEscalatesAndDemotesWhileFiring(t *testing.T) {
+	state := NewState()
+	rule := MetricRule{Trigger: 80, Clear: 75}
+
+	state.ApplyMetric(signalAt(85, 0), rule)
+
+	events := state.ApplyMetric(signalAt(95, time.Minute), rule)
+	if len(events) != 1 || events[0].Type != EventSeverityChanged || events[0].Severity != SeverityCritical {
+		t.Fatalf("events = %+v, want escalation to critical", events)
+	}
+
+	events = state.ApplyMetric(signalAt(85, 2*time.Minute), rule)
+	if len(events) != 1 || events[0].Type != EventSeverityChanged || events[0].Severity != SeverityWarning {
+		t.Fatalf("events = %+v, want demotion to warning", events)
+	}
+}
+
+func TestFiresCriticalImmediatelyAboveCriticalThreshold(t *testing.T) {
+	state := NewState()
+	rule := MetricRule{Trigger: 80, Clear: 75}
+
+	events := state.ApplyMetric(signalAt(95, 0), rule)
+	if len(events) != 1 || events[0].Type != EventFired || events[0].Severity != SeverityCritical {
+		t.Fatalf("events = %+v, want critical fired", events)
+	}
+}
+
+func TestPercentageCriticalCapAt99(t *testing.T) {
+	state := NewState()
+	// Trigger 95 → critical would be 105, capped at 99 for cpu.
+	rule := MetricRule{Trigger: 95, Clear: 90}
+
+	events := state.ApplyMetric(signalAt(99, 0), rule)
+	if len(events) != 1 || events[0].Severity != SeverityCritical {
+		t.Fatalf("events = %+v, want critical at capped threshold", events)
+	}
+
+	// Non-percentage metric keeps trigger+10.
+	other := NewState()
+	diskWrite := MetricSignal{ResourceID: "vm-100", Metric: "diskWrite", Value: 99, ObservedAt: t0}
+	events = other.ApplyMetric(diskWrite, MetricRule{Trigger: 95})
+	if len(events) != 1 || events[0].Severity != SeverityWarning {
+		t.Fatalf("events = %+v, want warning below 105 for non-percentage metric", events)
+	}
+}
+
+func TestDelayFiresWithPendingStartAndDipResets(t *testing.T) {
+	state := NewState()
+	rule := MetricRule{Trigger: 80, Clear: 75, DelaySeconds: 120}
+
+	events := state.ApplyMetric(signalAt(90, 0), rule)
+	if len(events) != 1 || events[0].Type != EventPending {
+		t.Fatalf("events = %+v, want pending", events)
+	}
+
+	// Still inside the delay window: nothing fires.
+	if events := state.ApplyMetric(signalAt(91, time.Minute), rule); len(events) != 0 {
+		t.Fatalf("early observation emitted %+v", events)
+	}
+
+	// Dip below trigger resets the delay entirely.
+	events = state.ApplyMetric(signalAt(70, 90*time.Second), rule)
+	if len(events) != 1 || events[0].Type != EventPendingCleared {
+		t.Fatalf("events = %+v, want pending cleared", events)
+	}
+
+	// Re-enter pending; fire only after the full delay from re-entry.
+	state.ApplyMetric(signalAt(92, 100*time.Second), rule)
+	if events := state.ApplyMetric(signalAt(92, 100*time.Second+119*time.Second), rule); len(events) != 0 {
+		t.Fatalf("fired before delay elapsed: %+v", events)
+	}
+	events = state.ApplyMetric(signalAt(92, 100*time.Second+120*time.Second), rule)
+	if len(events) != 1 || events[0].Type != EventFired {
+		t.Fatalf("events = %+v, want fired after delay", events)
+	}
+	incident, _ := state.Incident("vm-100", "cpu")
+	if !incident.StartedAt.Equal(t0.Add(100 * time.Second)) {
+		t.Fatalf("StartedAt = %v, want pending entry time %v", incident.StartedAt, t0.Add(100*time.Second))
+	}
+}
+
+func TestClearFallsBackToTriggerWhenUnset(t *testing.T) {
+	state := NewState()
+	rule := MetricRule{Trigger: 80, Clear: 0}
+
+	state.ApplyMetric(signalAt(85, 0), rule)
+
+	// At exactly the trigger the fire branch wins: still firing, no events.
+	if events := state.ApplyMetric(signalAt(80, time.Minute), rule); len(events) != 0 {
+		t.Fatalf("events at trigger equality = %+v, want none", events)
+	}
+	if incident, ok := state.Incident("vm-100", "cpu"); !ok || incident.State != StateFiring {
+		t.Fatalf("incident = %+v, want still firing at trigger equality", incident)
+	}
+
+	// With Clear unset, resolve requires value <= trigger.
+	events := state.ApplyMetric(signalAt(79, 2*time.Minute), rule)
+	if len(events) != 1 || events[0].Type != EventResolved {
+		t.Fatalf("events = %+v, want resolved at value <= trigger", events)
+	}
+}
+
+func TestDisabledRuleClearsPendingAndFiring(t *testing.T) {
+	state := NewState()
+	rule := MetricRule{Trigger: 80, Clear: 75, DelaySeconds: 60}
+
+	state.ApplyMetric(signalAt(90, 0), rule)
+	events := state.ApplyMetric(signalAt(90, time.Second), MetricRule{Trigger: 0})
+	if len(events) != 1 || events[0].Type != EventPendingCleared {
+		t.Fatalf("events = %+v, want pending cleared on disable", events)
+	}
+
+	firing := NewState()
+	firing.ApplyMetric(signalAt(90, 0), MetricRule{Trigger: 80, Clear: 75})
+	events = firing.ApplyMetric(signalAt(90, time.Second), MetricRule{Trigger: 0})
+	if len(events) != 1 || events[0].Type != EventResolved {
+		t.Fatalf("events = %+v, want resolved on disable", events)
+	}
+}
+
+func TestDeterminism(t *testing.T) {
+	run := func() []EventType {
+		state := NewState()
+		rule := MetricRule{Trigger: 80, Clear: 75, DelaySeconds: 60}
+		var all []Event
+		steps := []struct {
+			value  float64
+			offset time.Duration
+		}{
+			{85, 0}, {86, 30 * time.Second}, {85, 61 * time.Second},
+			{95, 2 * time.Minute}, {78, 3 * time.Minute}, {70, 4 * time.Minute},
+		}
+		for _, step := range steps {
+			all = append(all, state.ApplyMetric(signalAt(step.value, step.offset), rule)...)
+		}
+		return eventTypes(all)
+	}
+
+	first := run()
+	second := run()
+	if len(first) != len(second) {
+		t.Fatalf("non-deterministic event counts: %v vs %v", first, second)
+	}
+	for i := range first {
+		if first[i] != second[i] {
+			t.Fatalf("non-deterministic events: %v vs %v", first, second)
+		}
+	}
+	want := []EventType{EventPending, EventFired, EventSeverityChanged, EventResolved}
+	if len(first) != len(want) {
+		t.Fatalf("events = %v, want %v", first, want)
+	}
+	for i := range want {
+		if first[i] != want[i] {
+			t.Fatalf("events = %v, want %v", first, want)
+		}
+	}
+}
--- a/internal/alerts/reducer_parity_test.go
+++ b/internal/alerts/reducer_parity_test.go
@ -0,0 +1,257 @@
+package alerts
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/alerts/reducer"
+)
+
+// The reducer parity harness drives the live manager's metric path and the
+// Phase 1 reducer (internal/alerts/reducer) through identical observation
+// sequences and diffs the outcome after every step. The reducer must
+// CHARACTERIZE Manager.checkMetric, never improve on it: any divergence is
+// a reducer bug by definition (docs/ALERT_ENGINE_EVOLUTION.md, Phase 1).
+//
+// Time is simulated. The reducer takes it via ObservedAt; the manager's
+// delay tracking uses the wall clock, so the harness backdates
+// m.pendingAlerts between steps — the same trick the existing
+// time-threshold tests use.
+
+type parityStep struct {
+	value float64
+	// advance is the simulated time between the previous observation and
+	// this one.
+	advance time.Duration
+	// disabled observes with a disabled rule (nil threshold / zero trigger).
+	disabled bool
+
+	wantFiring   bool
+	wantSeverity AlertLevel // checked only when wantFiring
+}
+
+type parityScenario struct {
+	name         string
+	trigger      float64
+	clear        float64
+	delaySeconds int
+	steps        []parityStep
+}
+
+const (
+	parityResourceID   = "parity-vm-100"
+	parityMetric       = "cpu"
+	parityResourceType = "VM"
+)
+
+func parityCanonicalStateID() string {
+	return buildCanonicalStateID(parityResourceID, "metric-threshold:"+parityMetric)
+}
+
+type managerParityEngine struct {
+	manager   *Manager
+	threshold *HysteresisThreshold
+}
+
+func newManagerParityEngine(t *testing.T, scenario parityScenario) *managerParityEngine {
+	t.Helper()
+	manager := NewManagerWithDataDir(t.TempDir(), WithoutPersistedAlertRestore())
+	t.Cleanup(manager.Stop)
+
+	manager.mu.Lock()
+	manager.config.Enabled = true
+	manager.config.MinimumDelta = 0
+	manager.config.SuppressionWindow = 0
+	manager.config.FlappingEnabled = false
+	manager.config.TimeThresholds = map[string]int{}
+	if scenario.delaySeconds > 0 {
+		manager.config.TimeThresholds["guest"] = scenario.delaySeconds
+	}
+	manager.mu.Unlock()
+
+	return &managerParityEngine{
+		manager:   manager,
+		threshold: &HysteresisThreshold{Trigger: scenario.trigger, Clear: scenario.clear},
+	}
+}
+
+func (e *managerParityEngine) observe(step parityStep) {
+	if step.advance > 0 {
+		// Simulate elapsed time for the wall-clock delay tracker.
+		e.manager.mu.Lock()
+		for key, since := range e.manager.pendingAlerts {
+			e.manager.pendingAlerts[key] = since.Add(-step.advance)
+		}
+		e.manager.mu.Unlock()
+	}
+
+	threshold := e.threshold
+	if step.disabled {
+		threshold = nil
+	}
+	e.manager.checkMetric(
+		parityResourceID, "parity-vm", "node1", "qemu/100",
+		parityResourceType, parityMetric, step.value, threshold, nil,
+	)
+}
+
+func (e *managerParityEngine) snapshot() (bool, AlertLevel) {
+	e.manager.mu.Lock()
+	defer e.manager.mu.Unlock()
+	alert, exists := e.manager.getActiveAlertNoLock(parityCanonicalStateID())
+	if !exists || alert == nil {
+		return false, ""
+	}
+	return true, alert.Level
+}
+
+type reducerParityEngine struct {
+	state *reducer.State
+	rule  reducer.MetricRule
+	clock time.Time
+}
+
+func newReducerParityEngine(scenario parityScenario) *reducerParityEngine {
+	return &reducerParityEngine{
+		state: reducer.NewState(),
+		rule: reducer.MetricRule{
+			Trigger:      scenario.trigger,
+			Clear:        scenario.clear,
+			DelaySeconds: scenario.delaySeconds,
+		},
+		clock: time.Date(2026, 8, 26, 12, 0, 0, 0, time.UTC),
+	}
+}
+
+func (e *reducerParityEngine) observe(step parityStep) {
+	e.clock = e.clock.Add(step.advance)
+	rule := e.rule
+	if step.disabled {
+		rule = reducer.MetricRule{}
+	}
+	e.state.ApplyMetric(reducer.MetricSignal{
+		ResourceID: parityResourceID,
+		Metric:     parityMetric,
+		Value:      step.value,
+		ObservedAt: e.clock,
+	}, rule)
+}
+
+func (e *reducerParityEngine) snapshot() (bool, AlertLevel) {
+	incident, ok := e.state.Incident(parityResourceID, parityMetric)
+	if !ok || incident.State != reducer.StateFiring {
+		return false, ""
+	}
+	return true, AlertLevel(incident.Severity)
+}
+
+func parityScenarios() []parityScenario {
+	return []parityScenario{
+		{
+			name:    "fire, hold in hysteresis band, resolve at clear",
+			trigger: 80, clear: 75,
+			steps: []parityStep{
+				{value: 85, wantFiring: true, wantSeverity: AlertLevelWarning},
+				{value: 78, advance: time.Minute, wantFiring: true, wantSeverity: AlertLevelWarning},
+				{value: 74, advance: time.Minute, wantFiring: false},
+			},
+		},
+		{
+			name:    "critical on fire, demote while firing, resolve",
+			trigger: 80, clear: 75,
+			steps: []parityStep{
+				{value: 95, wantFiring: true, wantSeverity: AlertLevelCritical},
+				{value: 85, advance: time.Minute, wantFiring: true, wantSeverity: AlertLevelWarning},
+				{value: 60, advance: time.Minute, wantFiring: false},
+			},
+		},
+		{
+			name:    "percentage critical cap keeps escalation reachable at high trigger",
+			trigger: 95, clear: 90,
+			steps: []parityStep{
+				{value: 99, wantFiring: true, wantSeverity: AlertLevelCritical},
+			},
+		},
+		{
+			name:    "sustained-for delay with dip reset",
+			trigger: 80, clear: 75, delaySeconds: 120,
+			steps: []parityStep{
+				{value: 90, wantFiring: false},
+				{value: 91, advance: 60 * time.Second, wantFiring: false},
+				{value: 70, advance: 30 * time.Second, wantFiring: false},
+				{value: 92, advance: 10 * time.Second, wantFiring: false},
+				{value: 92, advance: 119 * time.Second, wantFiring: false},
+				{value: 92, advance: 2 * time.Second, wantFiring: true, wantSeverity: AlertLevelCritical},
+			},
+		},
+		{
+			name:    "clear falls back to trigger when unset",
+			trigger: 80, clear: 0,
+			steps: []parityStep{
+				{value: 85, wantFiring: true, wantSeverity: AlertLevelWarning},
+				{value: 80, advance: time.Minute, wantFiring: true, wantSeverity: AlertLevelWarning},
+				{value: 79, advance: time.Minute, wantFiring: false},
+			},
+		},
+		{
+			name:    "disabling the rule resolves a firing incident",
+			trigger: 80, clear: 75,
+			steps: []parityStep{
+				{value: 85, wantFiring: true, wantSeverity: AlertLevelWarning},
+				{value: 85, advance: time.Minute, disabled: true, wantFiring: false},
+			},
+		},
+		{
+			name:    "re-fires after resolution",
+			trigger: 80, clear: 75,
+			steps: []parityStep{
+				{value: 85, wantFiring: true, wantSeverity: AlertLevelWarning},
+				{value: 70, advance: time.Minute, wantFiring: false},
+				{value: 86, advance: time.Minute, wantFiring: true, wantSeverity: AlertLevelWarning},
+			},
+		},
+	}
+}
+
+func TestReducerParityWithManagerMetricPath(t *testing.T) {
+	for _, scenario := range parityScenarios() {
+		scenario := scenario
+		t.Run(scenario.name, func(t *testing.T) {
+			managerEngine := newManagerParityEngine(t, scenario)
+			reducerEngine := newReducerParityEngine(scenario)
+
+			for i, step := range scenario.steps {
+				managerEngine.observe(step)
+				reducerEngine.observe(step)
+
+				managerFiring, managerLevel := managerEngine.snapshot()
+				reducerFiring, reducerLevel := reducerEngine.snapshot()
+
+				label := fmt.Sprintf("step %d (value=%.0f advance=%s disabled=%v)",
+					i, step.value, step.advance, step.disabled)
+
+				// The manager is the reference: first pin it to the
+				// scenario's expectation, then require the reducer to match
+				// the manager exactly.
+				if managerFiring != step.wantFiring {
+					t.Fatalf("%s: manager firing = %v, scenario expects %v (characterization drift — update the scenario, then the reducer)",
+						label, managerFiring, step.wantFiring)
+				}
+				if managerFiring && managerLevel != step.wantSeverity {
+					t.Fatalf("%s: manager severity = %q, scenario expects %q",
+						label, managerLevel, step.wantSeverity)
+				}
+
+				if reducerFiring != managerFiring {
+					t.Fatalf("%s: PARITY DIVERGENCE — reducer firing = %v, manager = %v",
+						label, reducerFiring, managerFiring)
+				}
+				if managerFiring && reducerLevel != managerLevel {
+					t.Fatalf("%s: PARITY DIVERGENCE — reducer severity = %q, manager = %q",
+						label, reducerLevel, managerLevel)
+				}
+			}
+		})
+	}
+}