From bb5fddf463434dd7b8a1fe244cc35087d033c0b2 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Wed, 26 Aug 2026 23:04:12 +0100 Subject: [PATCH] feat(alerts): confirmation-family reducer with discrete parity harness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 slice 2 of docs/ALERT_ENGINE_EVOLUTION.md. ApplyDiscrete characterizes the canonical lifecycle path's match-spec semantics for connectivity / powered-state / discrete-state kinds: N consecutive matching observations activate with StartTime at the first match, one non-matching observation clears at this layer, severity follows the spec while firing, and a disabled spec resolves. The incident sub-key generalizes from metric name to state key. reducer_parity_discrete_test.go diffs the reducer against evaluateCanonicalLifecycleAlert after every step with fully simulated time on both sides, including alert StartTime on first activations. The harness again caught a real defect on first run — the confirmation start-time understatement fixed in the previous commit. Recovery confirmations and re-fire start restoration are documented as later slices. --- ...-lifecycle-contract-coverage-2026-08-26.md | 19 +- internal/alerts/reducer/discrete.go | 128 +++++++++ internal/alerts/reducer/discrete_test.go | 136 +++++++++ internal/alerts/reducer/reducer.go | 35 ++- .../alerts/reducer_parity_discrete_test.go | 268 ++++++++++++++++++ 5 files changed, 569 insertions(+), 17 deletions(-) create mode 100644 internal/alerts/reducer/discrete.go create mode 100644 internal/alerts/reducer/discrete_test.go create mode 100644 internal/alerts/reducer_parity_discrete_test.go diff --git 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 index e79e03210e..784dcb95dd 100644 --- 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 @@ -34,5 +34,20 @@ 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. +Second slice (2026-08-26): the confirmation (discrete-state) family — +N-consecutive-match activation, single-observation recovery at the +evaluator layer, spec-carried severity — characterized against +`evaluateCanonicalLifecycleAlert`. The harness again surfaced a real +defect: the confirmation maps persist only counts, so +`lifecyclePreviousState` reconstructed pending runs dated at the current +observation and activation stamped StartTime at the final confirming poll, +understating outage start by the whole confirmation window (the manager +had already fixed this once for unified incidents via +`unifiedIncidentFirstSeen` but not for the generic lifecycle path). Fixed +with `lifecycleFirstMatched`, mirroring that precedent, before the parity +slice landed. + +Remaining for later slices: recovery-confirmation gating +(`offlineRecoveryConfirmations`), re-fire-within-retention start +restoration, ack/snooze lifecycle, intent-policy interaction, and a +shadow-mode runtime feed. diff --git a/internal/alerts/reducer/discrete.go b/internal/alerts/reducer/discrete.go new file mode 100644 index 0000000000..194158fbe8 --- /dev/null +++ b/internal/alerts/reducer/discrete.go @@ -0,0 +1,128 @@ +package reducer + +// The confirmation (discrete-state) family characterizes the manager's +// canonical lifecycle path: alertspecs.Evaluate's match-spec semantics +// (connectivity, powered-state, discrete-state kinds) as wrapped by +// Manager.evaluateCanonicalLifecycleAlert. N consecutive matching +// observations activate; one non-matching observation clears at this +// layer (the manager's poll-driven offline paths add recovery +// confirmations as a separate gate, characterized in a later slice). +// Severity is carried by the rule/spec, re-derived on every observation +// while firing. + +import "time" + +// DiscreteSignal is one observation of a discrete condition on a resource: +// Matched reports whether the trigger condition was observed (offline, +// powered-state mismatch, state in the trigger set). +type DiscreteSignal struct { + ResourceID string + // Key identifies the condition, e.g. "connectivity" or a state key. + Key string + Matched bool + Severity Severity + ObservedAt time.Time +} + +// DiscreteRule is the resolved policy for one discrete condition. +type DiscreteRule struct { + // Confirmations is the number of consecutive matching observations + // required to fire. Values <= 0 mean 1, mirroring the evaluator's + // floor. (The manager's spec defaults: connectivity 3, powered-state + // 2; callers pass the resolved value.) + Confirmations int + // Disabled clears any pending or firing incident, mirroring the + // evaluator's terminal disabled transition and the manager's resolve + // of an existing alert. + Disabled bool +} + +// ApplyDiscrete advances the state for one discrete observation under one +// rule and returns the transition events, in order. Deterministic: time +// enters only through the signal's ObservedAt. +func (s *State) ApplyDiscrete(signal DiscreteSignal, rule DiscreteRule) []Event { + key := incidentKey(signal.ResourceID, signal.Key) + incident := s.incidents[key] + + event := func(eventType EventType, severity Severity) Event { + return Event{ + Type: eventType, + ResourceID: signal.ResourceID, + Key: signal.Key, + Severity: severity, + At: signal.ObservedAt, + } + } + + // A non-matching observation and a disabled rule both clear at this + // layer: pending resets entirely, firing resolves. + if rule.Disabled || !signal.Matched { + if incident == nil { + return nil + } + delete(s.incidents, key) + if incident.State == StateFiring { + return []Event{event(EventResolved, incident.Severity)} + } + return []Event{event(EventPendingCleared, "")} + } + + required := rule.Confirmations + if required <= 0 { + required = 1 + } + + // Matching while firing: stay firing, re-derive severity from the + // rule's current severity, keep the confirmation count clamped at the + // requirement — mirroring the evaluator's firing branch. + if incident != nil && incident.State == StateFiring { + previous := incident.Severity + incident.Severity = signal.Severity + incident.Confirmations = required + incident.LastObservedAt = signal.ObservedAt + if incident.Severity != previous { + return []Event{event(EventSeverityChanged, incident.Severity)} + } + return nil + } + + // Matching while pending: count consecutive matches; fire once the + // requirement is met, with StartedAt at the first matched observation + // (the manager uses FirstMatchedAt as the alert start time). + if incident != nil { + incident.Confirmations++ + incident.Severity = signal.Severity + incident.LastObservedAt = signal.ObservedAt + if incident.Confirmations < required { + return nil + } + incident.State = StateFiring + incident.Confirmations = required + incident.StartedAt = incident.PendingSince + return []Event{event(EventFired, incident.Severity)} + } + + // First matching observation. + if required <= 1 { + s.incidents[key] = &Incident{ + ResourceID: signal.ResourceID, + Key: signal.Key, + State: StateFiring, + Severity: signal.Severity, + StartedAt: signal.ObservedAt, + Confirmations: 1, + LastObservedAt: signal.ObservedAt, + } + return []Event{event(EventFired, signal.Severity)} + } + s.incidents[key] = &Incident{ + ResourceID: signal.ResourceID, + Key: signal.Key, + State: StatePending, + Severity: signal.Severity, + PendingSince: signal.ObservedAt, + Confirmations: 1, + LastObservedAt: signal.ObservedAt, + } + return []Event{event(EventPending, "")} +} diff --git a/internal/alerts/reducer/discrete_test.go b/internal/alerts/reducer/discrete_test.go new file mode 100644 index 0000000000..9de46925be --- /dev/null +++ b/internal/alerts/reducer/discrete_test.go @@ -0,0 +1,136 @@ +package reducer + +import ( + "testing" + "time" +) + +func discreteSignalAt(matched bool, severity Severity, offset time.Duration) DiscreteSignal { + return DiscreteSignal{ + ResourceID: "node-1", + Key: "connectivity", + Matched: matched, + Severity: severity, + ObservedAt: t0.Add(offset), + } +} + +func TestDiscreteFiresAfterConfirmations(t *testing.T) { + state := NewState() + rule := DiscreteRule{Confirmations: 3} + + events := state.ApplyDiscrete(discreteSignalAt(true, SeverityCritical, 0), rule) + if len(events) != 1 || events[0].Type != EventPending { + t.Fatalf("events = %+v, want pending on first match", events) + } + if events := state.ApplyDiscrete(discreteSignalAt(true, SeverityCritical, time.Minute), rule); len(events) != 0 { + t.Fatalf("second match emitted %+v", events) + } + + events = state.ApplyDiscrete(discreteSignalAt(true, SeverityCritical, 2*time.Minute), rule) + if len(events) != 1 || events[0].Type != EventFired || events[0].Severity != SeverityCritical { + t.Fatalf("events = %+v, want critical fired on third match", events) + } + + incident, ok := state.Incident("node-1", "connectivity") + if !ok || incident.State != StateFiring { + t.Fatalf("incident = %+v, want firing", incident) + } + if !incident.StartedAt.Equal(t0) { + t.Fatalf("StartedAt = %v, want first matched observation %v", incident.StartedAt, t0) + } + if incident.Confirmations != 3 { + t.Fatalf("Confirmations = %d, want clamped at 3", incident.Confirmations) + } +} + +func TestDiscreteNonMatchResetsPendingCount(t *testing.T) { + state := NewState() + rule := DiscreteRule{Confirmations: 3} + + state.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, 0), rule) + state.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, time.Minute), rule) + + events := state.ApplyDiscrete(discreteSignalAt(false, "", 2*time.Minute), rule) + if len(events) != 1 || events[0].Type != EventPendingCleared { + t.Fatalf("events = %+v, want pending cleared", events) + } + + // The count starts over: two more matches must not fire. + state.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, 3*time.Minute), rule) + if events := state.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, 4*time.Minute), rule); len(events) != 0 { + t.Fatalf("post-reset second match emitted %+v", events) + } + events = state.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, 5*time.Minute), rule) + if len(events) != 1 || events[0].Type != EventFired { + t.Fatalf("events = %+v, want fired on third consecutive match", events) + } + incident, _ := state.Incident("node-1", "connectivity") + if !incident.StartedAt.Equal(t0.Add(3 * time.Minute)) { + t.Fatalf("StartedAt = %v, want restart of the pending run", incident.StartedAt) + } +} + +func TestDiscreteSingleConfirmationFiresImmediately(t *testing.T) { + state := NewState() + rule := DiscreteRule{Confirmations: 1} + + events := state.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, 0), rule) + if len(events) != 1 || events[0].Type != EventFired { + t.Fatalf("events = %+v, want immediate fire", events) + } + events = state.ApplyDiscrete(discreteSignalAt(false, "", time.Minute), rule) + if len(events) != 1 || events[0].Type != EventResolved { + t.Fatalf("events = %+v, want resolved on single recovery", events) + } +} + +func TestDiscreteSeverityFollowsRuleWhileFiring(t *testing.T) { + state := NewState() + rule := DiscreteRule{Confirmations: 1} + + state.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, 0), rule) + events := state.ApplyDiscrete(discreteSignalAt(true, SeverityCritical, time.Minute), rule) + if len(events) != 1 || events[0].Type != EventSeverityChanged || events[0].Severity != SeverityCritical { + t.Fatalf("events = %+v, want severity change to critical", events) + } + if events := state.ApplyDiscrete(discreteSignalAt(true, SeverityCritical, 2*time.Minute), rule); len(events) != 0 { + t.Fatalf("unchanged severity emitted %+v", events) + } +} + +func TestDiscreteDisabledClearsPendingAndFiring(t *testing.T) { + pending := NewState() + rule := DiscreteRule{Confirmations: 3} + pending.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, 0), rule) + events := pending.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, time.Minute), DiscreteRule{Confirmations: 3, Disabled: true}) + if len(events) != 1 || events[0].Type != EventPendingCleared { + t.Fatalf("events = %+v, want pending cleared on disable", events) + } + + firing := NewState() + firing.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, 0), DiscreteRule{Confirmations: 1}) + events = firing.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, time.Minute), DiscreteRule{Confirmations: 1, Disabled: true}) + if len(events) != 1 || events[0].Type != EventResolved { + t.Fatalf("events = %+v, want resolved on disable", events) + } +} + +func TestDiscreteRefireRequiresFullConfirmations(t *testing.T) { + state := NewState() + rule := DiscreteRule{Confirmations: 2} + + state.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, 0), rule) + state.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, time.Minute), rule) + state.ApplyDiscrete(discreteSignalAt(false, "", 2*time.Minute), rule) + + // After recovery the confirmation run restarts from zero. + events := state.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, 3*time.Minute), rule) + if len(events) != 1 || events[0].Type != EventPending { + t.Fatalf("events = %+v, want pending after recovery", events) + } + events = state.ApplyDiscrete(discreteSignalAt(true, SeverityWarning, 4*time.Minute), rule) + if len(events) != 1 || events[0].Type != EventFired { + t.Fatalf("events = %+v, want re-fire after full confirmations", events) + } +} diff --git a/internal/alerts/reducer/reducer.go b/internal/alerts/reducer/reducer.go index ca6b6e9159..f0f771f478 100644 --- a/internal/alerts/reducer/reducer.go +++ b/internal/alerts/reducer/reducer.go @@ -64,13 +64,18 @@ const ( // 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 + ResourceID string + // Key is the incident sub-key: the metric name for the metric family, + // the discrete state key for the confirmation family. + Key string + State IncidentState + Severity Severity + PendingSince time.Time + StartedAt time.Time + LastValue float64 + // Confirmations is the consecutive-match count for the confirmation + // family; unused by the metric family. + Confirmations int LastObservedAt time.Time } @@ -89,7 +94,7 @@ const ( type Event struct { Type EventType ResourceID string - Metric string + Key string Severity Severity Value float64 At time.Time @@ -106,16 +111,16 @@ func NewState() *State { return &State{incidents: make(map[string]*Incident)} } -func incidentKey(resourceID, metric string) string { - return fmt.Sprintf("%s\x00%s", resourceID, metric) +func incidentKey(resourceID, subKey string) string { + return fmt.Sprintf("%s\x00%s", resourceID, subKey) } // Incident returns a copy of the tracked incident for the key, if any. -func (s *State) Incident(resourceID, metric string) (Incident, bool) { +func (s *State) Incident(resourceID, subKey string) (Incident, bool) { if s == nil { return Incident{}, false } - incident, ok := s.incidents[incidentKey(resourceID, metric)] + incident, ok := s.incidents[incidentKey(resourceID, subKey)] if !ok { return Incident{}, false } @@ -172,7 +177,7 @@ func (s *State) ApplyMetric(signal MetricSignal, rule MetricRule) []Event { return Event{ Type: eventType, ResourceID: signal.ResourceID, - Metric: signal.Metric, + Key: signal.Metric, Severity: severity, Value: signal.Value, At: signal.ObservedAt, @@ -215,7 +220,7 @@ func (s *State) ApplyMetric(signal MetricSignal, rule MetricRule) []Event { if incident == nil { s.incidents[key] = &Incident{ ResourceID: signal.ResourceID, - Metric: signal.Metric, + Key: signal.Metric, State: StatePending, PendingSince: signal.ObservedAt, LastValue: signal.Value, @@ -240,7 +245,7 @@ func (s *State) ApplyMetric(signal MetricSignal, rule MetricRule) []Event { severity := severityFor(signal.Value, rule.Trigger, signal.Metric) s.incidents[key] = &Incident{ ResourceID: signal.ResourceID, - Metric: signal.Metric, + Key: signal.Metric, State: StateFiring, Severity: severity, StartedAt: signal.ObservedAt, diff --git a/internal/alerts/reducer_parity_discrete_test.go b/internal/alerts/reducer_parity_discrete_test.go new file mode 100644 index 0000000000..63d9824246 --- /dev/null +++ b/internal/alerts/reducer_parity_discrete_test.go @@ -0,0 +1,268 @@ +package alerts + +import ( + "fmt" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts/reducer" + alertspecs "github.com/rcourtman/pulse-go-rewrite/internal/alerts/specs" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" +) + +// Discrete-family parity: the reducer's confirmation semantics +// (internal/alerts/reducer/discrete.go) are diffed against the live +// manager's canonical lifecycle path — buildCanonicalDiscreteStateSpec + +// evaluateCanonicalLifecycleAlert, the same shape external_probe.go uses — +// after every step. Time is fully simulated on both sides here: the +// lifecycle path takes ObservedAt from the evidence. +// +// Alert StartTime parity is asserted on first activations only. On a +// re-fire within the recently-resolved retention window the manager +// restores the ORIGINAL start time (consumeRecentlyResolvedForRefire); +// that behavior needs resolved-history state in the reducer and is left +// for a later slice. + +type discreteParityStep struct { + matched bool + severity AlertLevel + disabled bool + advance time.Duration + + wantFiring bool + wantSeverity AlertLevel + // wantStartOffset, when >= 0, asserts the firing alert's start time as + // an offset from the scenario's first observation. Use -1 to skip. + wantStartOffset time.Duration +} + +type discreteParityScenario struct { + name string + confirmations int + steps []discreteParityStep +} + +const ( + discreteParityResourceID = "parity-node-1" + discreteParityStateKey = "parity-state" +) + +type managerDiscreteParityEngine struct { + manager *Manager + clock time.Time + conf int +} + +func newManagerDiscreteParityEngine(t *testing.T, scenario discreteParityScenario, epoch time.Time) *managerDiscreteParityEngine { + t.Helper() + manager := NewManagerWithDataDir(t.TempDir(), WithoutPersistedAlertRestore()) + t.Cleanup(manager.Stop) + + manager.mu.Lock() + manager.config.Enabled = true + manager.config.FlappingEnabled = false + manager.mu.Unlock() + + return &managerDiscreteParityEngine{manager: manager, clock: epoch, conf: scenario.confirmations} +} + +func (e *managerDiscreteParityEngine) observe(t *testing.T, step discreteParityStep) { + t.Helper() + e.clock = e.clock.Add(step.advance) + + severity := step.severity + if severity == "" { + severity = AlertLevelWarning + } + spec, err := buildCanonicalDiscreteStateSpec( + discreteParityResourceID, + "parity-node", + unifiedresources.ResourceTypeAgent, + severity, + e.conf, + step.disabled, + discreteParityStateKey, + []string{"bad"}, + ) + if err != nil { + t.Fatalf("build spec: %v", err) + } + + observed := "ok" + if step.matched { + observed = "bad" + } + _, ok := e.manager.evaluateCanonicalLifecycleAlert(canonicalLifecycleAlertParams{ + Spec: spec, + Evidence: alertspecs.AlertEvidence{ + ObservedAt: e.clock, + DiscreteState: &alertspecs.DiscreteStateEvidence{ + StateKey: discreteParityStateKey, + Observed: observed, + }, + }, + Tracking: e.manager.offlineConfirmations, + TrackingKey: "parity:" + discreteParityResourceID, + AlertID: canonicalDiscreteStateStateID(discreteParityResourceID, discreteParityStateKey), + AlertType: "parity-state", + ResourceID: discreteParityResourceID, + ResourceName: "parity-node", + Instance: "parity", + Message: "parity condition observed", + AddToRecent: true, + AddToHistory: true, + RateLimit: false, + DispatchAsync: false, + }) + if !ok { + t.Fatalf("evaluateCanonicalLifecycleAlert rejected the parity spec") + } +} + +func (e *managerDiscreteParityEngine) snapshot() (bool, AlertLevel, time.Time) { + e.manager.mu.Lock() + defer e.manager.mu.Unlock() + alert, exists := e.manager.getActiveAlertNoLock( + canonicalDiscreteStateStateID(discreteParityResourceID, discreteParityStateKey)) + if !exists || alert == nil { + return false, "", time.Time{} + } + return true, alert.Level, alert.StartTime +} + +type reducerDiscreteParityEngine struct { + state *reducer.State + clock time.Time + conf int +} + +func (e *reducerDiscreteParityEngine) observe(step discreteParityStep) { + e.clock = e.clock.Add(step.advance) + severity := reducer.SeverityWarning + if step.severity == AlertLevelCritical { + severity = reducer.SeverityCritical + } + e.state.ApplyDiscrete(reducer.DiscreteSignal{ + ResourceID: discreteParityResourceID, + Key: discreteParityStateKey, + Matched: step.matched, + Severity: severity, + ObservedAt: e.clock, + }, reducer.DiscreteRule{Confirmations: e.conf, Disabled: step.disabled}) +} + +func (e *reducerDiscreteParityEngine) snapshot() (bool, AlertLevel, time.Time) { + incident, ok := e.state.Incident(discreteParityResourceID, discreteParityStateKey) + if !ok || incident.State != reducer.StateFiring { + return false, "", time.Time{} + } + return true, AlertLevel(incident.Severity), incident.StartedAt +} + +func discreteParityScenarios() []discreteParityScenario { + skip := time.Duration(-1) + return []discreteParityScenario{ + { + name: "three-confirmation activation, single recovery", + confirmations: 3, + steps: []discreteParityStep{ + {matched: true, severity: AlertLevelCritical, wantFiring: false, wantStartOffset: skip}, + {matched: true, severity: AlertLevelCritical, advance: time.Minute, wantFiring: false, wantStartOffset: skip}, + {matched: true, severity: AlertLevelCritical, advance: time.Minute, wantFiring: true, wantSeverity: AlertLevelCritical, wantStartOffset: 0}, + {matched: false, advance: time.Minute, wantFiring: false, wantStartOffset: skip}, + }, + }, + { + name: "interrupted run resets the confirmation count", + confirmations: 3, + steps: []discreteParityStep{ + {matched: true, wantFiring: false, wantStartOffset: skip}, + {matched: true, advance: time.Minute, wantFiring: false, wantStartOffset: skip}, + {matched: false, advance: time.Minute, wantFiring: false, wantStartOffset: skip}, + {matched: true, advance: time.Minute, wantFiring: false, wantStartOffset: skip}, + {matched: true, advance: time.Minute, wantFiring: false, wantStartOffset: skip}, + {matched: true, advance: time.Minute, wantFiring: true, wantSeverity: AlertLevelWarning, wantStartOffset: 3 * time.Minute}, + }, + }, + { + name: "single confirmation fires immediately and re-fires after recovery", + confirmations: 1, + steps: []discreteParityStep{ + {matched: true, wantFiring: true, wantSeverity: AlertLevelWarning, wantStartOffset: 0}, + {matched: false, advance: time.Minute, wantFiring: false, wantStartOffset: skip}, + {matched: true, advance: time.Minute, wantFiring: true, wantSeverity: AlertLevelWarning, wantStartOffset: skip}, + }, + }, + { + name: "severity follows the spec while firing", + confirmations: 1, + steps: []discreteParityStep{ + {matched: true, severity: AlertLevelWarning, wantFiring: true, wantSeverity: AlertLevelWarning, wantStartOffset: 0}, + {matched: true, severity: AlertLevelCritical, advance: time.Minute, wantFiring: true, wantSeverity: AlertLevelCritical, wantStartOffset: 0}, + {matched: true, severity: AlertLevelWarning, advance: time.Minute, wantFiring: true, wantSeverity: AlertLevelWarning, wantStartOffset: 0}, + }, + }, + { + name: "disabling the spec resolves a firing incident", + confirmations: 1, + steps: []discreteParityStep{ + {matched: true, wantFiring: true, wantSeverity: AlertLevelWarning, wantStartOffset: 0}, + {matched: true, disabled: true, advance: time.Minute, wantFiring: false, wantStartOffset: skip}, + }, + }, + } +} + +func TestReducerParityWithManagerDiscretePath(t *testing.T) { + epoch := time.Date(2026, 8, 26, 12, 0, 0, 0, time.UTC) + + for _, scenario := range discreteParityScenarios() { + scenario := scenario + t.Run(scenario.name, func(t *testing.T) { + managerEngine := newManagerDiscreteParityEngine(t, scenario, epoch) + reducerEngine := &reducerDiscreteParityEngine{ + state: reducer.NewState(), + clock: epoch, + conf: scenario.confirmations, + } + + for i, step := range scenario.steps { + managerEngine.observe(t, step) + reducerEngine.observe(step) + + managerFiring, managerLevel, managerStart := managerEngine.snapshot() + reducerFiring, reducerLevel, reducerStart := reducerEngine.snapshot() + + label := fmt.Sprintf("step %d (matched=%v severity=%s disabled=%v advance=%s)", + i, step.matched, step.severity, step.disabled, step.advance) + + if managerFiring != step.wantFiring { + t.Fatalf("%s: manager firing = %v, scenario expects %v (characterization drift)", + label, managerFiring, step.wantFiring) + } + if managerFiring && managerLevel != step.wantSeverity { + t.Fatalf("%s: manager severity = %q, scenario expects %q", label, managerLevel, step.wantSeverity) + } + if managerFiring && step.wantStartOffset >= 0 { + wantStart := epoch.Add(step.wantStartOffset) + if !managerStart.Equal(wantStart) { + t.Fatalf("%s: manager start = %v, scenario expects %v", label, managerStart, wantStart) + } + } + + 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) + } + if managerFiring && step.wantStartOffset >= 0 && !reducerStart.Equal(managerStart) { + t.Fatalf("%s: PARITY DIVERGENCE — reducer start = %v, manager = %v", + label, reducerStart, managerStart) + } + } + }) + } +}