think
16px
820px
← all files

Task 3 report — Evaluator loop + main.go wiring + shared warning-eval helper

Plan: docs/superpowers/plans/2026-07-09-ews-notifications.md, Task 3 (final task of the EWS Notifications plan).
Branch: feat/ews-notifications.
Status: DONE.

Files

  • internal/api/predict.go — new EvaluateEarlyWarnings(ctx, pool) ([]predict.Warning, error), new EvaluateAllWarnings(ctx, pool) ([]predict.Warning, error); handleEarlyWarnings refactored to call EvaluateEarlyWarnings.
  • internal/api/security.go — new EvaluateSecurityWarnings(ctx, pool) ([]predict.Warning, error); handleSecurityWarnings refactored to call it.
  • internal/api/predict_test.go — TestEvaluateAllWarningsIsUnionOfEarlyAndSecurity (new).
  • internal/api/security_test.go — TestSecurityWarningsShape (new file; no shape test existed for this handler before).
  • internal/notify/evaluator.go (new) — Evaluator{Interval, Eval, Dispatcher, Channels, SendTimeout, now}, Run(ctx) error, unexported tick(ctx), unexported dispatch(ctx, ch, n).
  • internal/notify/evaluator_test.go (new) — 4 test functions.
  • cmd/observatory/main.go — notifier wiring after the DCGM sampler block.

Part A — shared eval helpers (no duplication)

Extracted the query+eval bodies exactly as specified:

  • EvaluateEarlyWarnings(ctx, pool): queue-wait forecast, GPU-utilisation forecast, error-rate anomaly — reuses the existing earlyWarn* consts, store.QueryMetricSeries/QuerySeries, errorRateByBucket, unchanged.
  • EvaluateSecurityWarnings(ctx, pool): actor rollup + body-access stats → predict.EvaluateSecurity — reuses store.QueryActors/QueryBodyAccessStats, actorLabel, isSyntheticActor, unchanged.
  • EvaluateAllWarnings(ctx, pool): early..., security... concatenation; an error from either propagates (no partial result).

Both handlers now just call the shared func for warnings and separately compute their own from/to/bucket for the response's window/lookback metadata (per the plan: "the window fields are computed in the handler"). JSON response shape is unchanged for both routes.

Judgment call (worth flagging): EvaluateEarlyWarnings's signature, as specified in the task, takes no bucket parameter — it always evaluates over the fixed earlyWarnBucket/earlyWarnLookback consts, reusing "the existing earlyWarn* consts" as instructed. Previously, handleEarlyWarnings threaded a client-supplied ?bucket= override into the actual store queries (defaulting to earlyWarnBucket). After this refactor, ?bucket= is still parsed/validated and still labels the response's "bucket" field (so a bad value still 400s, and the field is still present) — but the actual rule evaluation always runs over earlyWarnBucket, regardless of the query param. No existing test exercises a non-default bucket for this endpoint (TestEarlyWarningsShape uses ?bucket=5m, which is the default), so nothing regresses, but a client that was relying on ?bucket=1m actually changing the queried granularity would now see it only reflected in the label. This is arguably a correctness improvement too: the rules' thresholds (earlyWarnRecentBuckets, the forecast horizon) are tuned against the fixed 5-minute bucket and were already not meaningfully re-tunable via the query param. Flagging for the whole-branch review per the plan's "byte-identical" instruction, in case a live client depends on the old override behavior.

New test TestEvaluateAllWarningsIsUnionOfEarlyAndSecurity proves the union via a real (not length-only) multiset-equality check against a live test DB, so it can't pass by a length coincidence. New TestSecurityWarningsShape locks in handleSecurityWarnings' response shape (no prior test covered it at all before this task).

Part B — internal/notify/evaluator.go

Evaluator matches the specified struct exactly: Interval, Eval func(context.Context) ([]predict.Warning, error), Dispatcher *Dispatcher, Channels []Channel, SendTimeout time.Duration, unexported now func() time.Time for test clock injection.

  • Run(ctx) error: time.NewTicker(Interval); each fire calls tick(ctx); returns nil on ctx.Done().
  • Waits one Interval before the first tick (does not evaluate immediately on start) — see judgment call below.
  • tick(ctx): calls Eval(ctx) — on error, logs and returns (never stops the loop, never panics). On success, calls Dispatcher.Decide(now, warnings); if toSend is empty, returns (nothing to send). Otherwise builds one Notification{toSend, escalated, now} and calls dispatch for every channel.
  • dispatch(ctx, ch, n): per attempt, a fresh context.WithTimeout(ctx, SendTimeout); up to sendAttempts=2 attempts with a sendRetryBackoff=50ms pause between them; any failure (including exhausting both attempts) is logged via log.Printf, never returned/propagated — one channel's failure can never block or skip another channel, and never stops the loop.

Judgment call — wait-one-interval vs. immediate-on-start: implemented "wait one interval" (ticker fires first after Interval elapses, matching time.NewTicker's documented behavior; no manual first-eval-then-tick). Rationale: the Dispatcher's in-memory dedup/escalation state does not survive a process restart (documented in dispatcher.go), so an immediate eval-on-start would mean every restart instantly re-notifies every currently-active warning as a fresh "first sighting" burst, at the exact moment other boot-time work (migrations, ingester catch-up) is also happening. Waiting one interval means the first (re-)notification only happens once the loop has run a real, steady-state evaluation — the tradeoff is up to one eval_interval_s (default 300s) of delay on the very first notification after a restart, which is acceptable for a background alerting loop.

Testability

tick is directly callable (lowercase, but deliberately not gated behind the ticker) so tests drive cycles deterministically — no time.Sleep on wall-clock anywhere in evaluator_test.go; the one test that exercises Run (TestEvaluatorRunReturnsPromptlyOnContextCancel) cancels the context before starting so it returns immediately regardless of Interval (set to 1h to guarantee no real tick could fire first).

4 test functions in evaluator_test.go:
- TestEvaluatorTickDispatchesNewWarningOnceAndFiltersSeverity — one critical + one info warning; only the critical one (above min_severity: warning) is dispatched; a second tick with the same warnings sends nothing new (dedup via the real Dispatcher, not a stub).
- TestEvaluatorTickEvalErrorSkipsWithoutPanicOrSend — Eval returns an error; tick does not panic and dispatches nothing.
- TestEvaluatorTickChannelFailureDoesNotBlockOtherChannelsOrLoop — one channel whose Send always errors, another that succeeds: the good channel still receives; the loop is still usable for a subsequent tick afterward.
- TestEvaluatorRunReturnsPromptlyOnContextCancel — Run returns nil promptly when ctx is already canceled.

Part C — cmd/observatory/main.go wiring

After the DCGM sampler block: if nc := cfg.Notifications; nc.Enabled && (nc.Email.Enabled || nc.Webhook.Enabled) { ... }. Inside: builds []notify.Channel from whichever of Email/Webhook is enabled (via notify.NewEmailChannel/NewWebhookChannel), constructs Evaluator{Interval: EvalIntervalS seconds, Eval: closure over api.EvaluateAllWarnings(ctx, pool), Dispatcher: notify.NewDispatcher(nc), Channels, SendTimeout: notifySendTimeout (10s)}, logs one startup line naming the enabled channels + cadence, and go evaluator.Run(ingestCtx).

When disabled/absent (cfg.Notifications zero-value or enabled: false, or enabled: true with no channel — the last case can't actually reach main.go since config.Load's validateNotifications already fails loud on it): the whole if is skipped — no goroutine started, nothing logged, byte-identical startup to before this feature existed. Verified: TestRunFailsFastWhenIngesterCannotStart in cmd/observatory/main_test.go constructs a bare *config.Config{} (so Notifications is zero-value/disabled) and still passes unchanged.

notify never imports internal/api — Eval is injected as a closure built in main.go, so there is no import cycle (api → predict/store; notify → predict/config; main → api, notify).

Gates run

Note: this sandbox's Docker is remote (DOCKER_HOST=tcp://dind:2375), so TEST_PG_DSN must point at dind:5434, not localhost:5434/the scripts' default — otherwise every DB-gated test silently t.Skip()s and the suite looks green without ever touching the DB. Ran with:

export TEST_PG_DSN="postgres://postgres:test@dind:5434/postgres"
go build ./...        # clean
go vet ./...          # clean
go test ./... -count=1
  ok    datahive.id/ahu-ai-observatory/cmd/observatory   1.8s
  ok    datahive.id/ahu-ai-observatory/internal/api       2.0s   (DB-backed, not skipped)
  ok    datahive.id/ahu-ai-observatory/internal/config    0.02s
  ok    datahive.id/ahu-ai-observatory/internal/ingest    17.2s  (DB-backed, not skipped)
  ok    datahive.id/ahu-ai-observatory/internal/notify    0.09s
  ok    datahive.id/ahu-ai-observatory/internal/predict   0.01s
  ok    datahive.id/ahu-ai-observatory/internal/store     1.2s   (DB-backed, not skipped)
gofmt -l .            # no output (already formatted)

All 7 packages pass, including the DB-gated internal/api, internal/store, cmd/observatory, internal/ingest suites (confirmed actually exercised against timescale/timescaledb:latest-pg16 via ./scripts/testdb.sh, not silently skipped).

Commit

feat(notify): background evaluator loop + main wiring (dormant unless configured) — this closes out the EWS Notifications plan (Tasks 1, 2, 3 all done). Whole-branch adversarial review, merge to master, live deploy config, and progress-sheet update are the plan's "Post-execution (controller)" steps — out of scope for this task.

Adversarial-review fixes (post-review pass, 2026-07-09)

Commit: dff4191 — fix(notify): fingerprint on stable subject (no storm) + validate dedup window + bounded SMTP + recover in loop (review fixes)
Status: DONE — all gates green (go build/go vet/gofmt -l clean; go test -race ./internal/notify/; full go test ./... with TEST_PG_DSN=postgres://postgres:test@dind:5434/postgres + scripts/testdb.sh, all 7 packages incl. DB-gated ones pass).

Per-finding

  • C1 (CRITICAL) — fixed. Added Subject string \json:"subject,omitempty"`topredict.Warning; set it on every per-actor security warning (sec_error_probing/sec_egress_outlier/sec_volume_outlier/sec_off_hours→a.Label) and the body-access watcher (sec_body_access→ba.Actor); early-warning fleet-singletons leave it "" (JSON byte-identical via omitempty).notify.Fingerprintnow hashesID + "|" + Metric + "|" + Subject` only — all evidence-value hashing removed. Storm defeated; actor collisions resolved.
  • I2 (IMPORTANT) — fixed. validateNotifications now rejects dedup_window_s < 0 (0 still defaults to 3600 in defaulting, unchanged).
  • M5 (folded into I2) — fixed. Added maxNotificationIntervalS = 86400 cap; validateNotifications rejects eval_interval_s and dedup_window_s above it, so time.Duration(s)*time.Second can't overflow to a negative Duration and panic time.NewTicker.
  • I3 (IMPORTANT) — fixed. Rewrote email.go: send is now synchronous + ctx-bounded (no goroutine racing ctx). realSMTPSend dials via net.Dialer.DialContext(ctx), sets conn.SetDeadline(ctx deadline) (fallback 30s if the ctx has none) so ALL SMTP IO is bounded, then STARTTLS(when configured)/Auth/MAIL/RCPT/DATA/Quit, closing the conn on every error path. No goroutine/socket leak on a silent server.
  • I4 (IMPORTANT) — fixed. evaluator.go: safeEval/safeDecide recover panics in Eval/Dispatcher.Decide (log + skip tick); dispatch has a per-channel recover (log channel + skip that channel, siblings still run). Send is synchronous post-I3, so no extra goroutine to guard.
  • M7 (MINOR) — fixed. main.go notifier startup logs a WARNING (env-var NAME only, never a value; never fails startup) when email is enabled + username set but Email.PasswordEnv resolves empty, and when Webhook.TokenEnv is set-but-empty (misnamed-env case). Optional webhook token with no env named stays silent.
  • M6 (OPTIONAL) — skipped. Making the reported window match the queried one requires changing both EvaluateEarlyWarnings/EvaluateSecurityWarnings signatures (return or accept from/to) and updating EvaluateAllWarnings + the handlers + api tests — a signature complication for a sub-millisecond cosmetic gain. Skipped per the finding's own "skip if it complicates the signature" guidance.

C1 regression tests (proving dedup holds under changing evidence)

  • internal/notify/notify_test.go: TestFingerprintStableUnderChangingEvidence (same ID/Metric/Subject, changing Evidence values+order → same fingerprint), TestFingerprintDistinctForStableIdentity (distinct Subject/ID/Metric → distinct fingerprints, even with identical evidence). Replaced the old TestFingerprintStableAcrossIdenticalWarnings / TestFingerprintDistinctForDifferingEvidence whose premise (hash the evidence) was the bug.
  • internal/notify/dispatcher_test.go: TestDispatcherDedupHoldsUnderChangingEvidence — same logical warning across 6 ticks with a new evidence value each tick → sent ticks [0,3] (fresh, then escalate-once), NOT [0,1,2,3,4,5]. Verified this test FAILS against the pre-fix evidence-hashing fingerprint (produced the [0 1 2 3 4 5] storm) and PASSES after the fix.
  • Config: TestNotificationsNegativeDedupWindowRejected (I2), TestNotificationsHugeEvalIntervalRejected + TestNotificationsHugeDedupWindowRejected (M5).
  • Evaluator (I4): TestEvaluatorTickRecoversFromEvalPanic, TestEvaluatorTickRecoversFromChannelSendPanic (bad channel listed first, good channel still receives).
  • Email (I3): TestEmailChannelSendIsCtxBoundedAgainstSilentServer — real realSMTPSend against a real silent TCP listener returns an error within the ctx bound, never hangs. Existing message-assertion tests (To/Subject/body/[ESCALATED]/password-absent) updated to the new func(ctx, smtpDelivery) error sender seam.

Judgment calls

  • Email sender seam reshape. Old seam was sender func(addr, auth, from, to, msg) error (smtp.SendMail-shaped) raced in a goroutine. Replaced with sender func(ctx, smtpDelivery) error where smtpDelivery carries addr/host/startTLS/auth/from/to/msg. Password is still resolved via config.ResolveSecret at send time in Send and passed as a built smtp.Auth (never as a raw string in the delivery struct, never reaches a fake, never logged). NewEmailChannel now always wires realSMTPSend (the STARTTLS-vs-plaintext choice moved into the per-send startTLS bool instead of swapping the sender); dropped the separate sendMailPlain.
  • M7 webhook scope. Only warn when a token env var was NAMED but resolves empty (a real misconfig); silent when no token intended, since a webhook token is legitimately optional.
  • M6 skipped as above.