From ab71410ccba63729a862bc460dfd9eec9c91749c Mon Sep 17 00:00:00 2001 From: rcourtman Date: Wed, 13 May 2026 03:34:50 +0100 Subject: [PATCH] add update-safety watcher substrate in internal/ai --- internal/ai/findings_update_safety.go | 229 ++++++++++++++++++ internal/ai/findings_update_safety_test.go | 258 +++++++++++++++++++++ 2 files changed, 487 insertions(+) create mode 100644 internal/ai/findings_update_safety.go create mode 100644 internal/ai/findings_update_safety_test.go diff --git a/internal/ai/findings_update_safety.go b/internal/ai/findings_update_safety.go new file mode 100644 index 000000000..5e43401aa --- /dev/null +++ b/internal/ai/findings_update_safety.go @@ -0,0 +1,229 @@ +// findings_update_safety.go implements the update-safety watcher for Docker +// containers. It snapshot-diffs ImageDigest across patrol observe trips and +// emits a reliability finding when a container's image changes unexpectedly +// (e.g. Watchtower-class auto-update). It auto-resolves via a lazy sentinel +// on the next observe trip once the container has been stable for the +// verification window with no new restarts. +package ai + +import ( + "fmt" + "sort" + "sync" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" +) + +const ( + // UpdateSafetyFindingPrefix is the dedup-key prefix for update-safety findings. + // Concrete keys are prefix + ":" + containerKey where containerKey is hostID/containerID. + UpdateSafetyFindingPrefix = "docker:image:update_divergence" + + // updateSafetyVerifyWindow is how long the watcher waits after a digest + // change, with no new restarts, before emitting a resolve sentinel. + updateSafetyVerifyWindow = 90 * time.Second + + updateSafetySource = "update-safety" + updateSafetyResolveReason = "update_safety:verified_clean" + updateSafetySnapshotCap = 2048 +) + +// resolveSentinel carries the dedup key and reason for a lazy auto-resolve. +// The caller routes it through FindingsStore.ResolveWithReason. +type resolveSentinel struct { + DedupKey string + Reason string +} + +// updateSafetySnapshot holds per-container state between observe trips. +type updateSafetySnapshot struct { + digest string // digest as of the last observe trip + restartCount int // restartCount as of the last observe trip + lastSeenAt time.Time // when this snapshot was last updated + + // Fields populated once a digest change is detected. + detectedAt time.Time // zero when no active change is being verified + priorDigest string // digest before the change + changeDigest string // digest after the change + baseRestarts int // restartCount at the moment the change was detected + lastEmittedRestarts int // restartCount at the time of the most recent emit +} + +// UpdateSafetyWatcher snapshot-diffs container image digests across patrol +// observe trips. In-memory only; no persistence needed at MVP. +type UpdateSafetyWatcher struct { + mu sync.Mutex + snapshots map[string]*updateSafetySnapshot // key: hostID/containerID + cap int +} + +// newUpdateSafetyWatcher returns an initialized watcher ready for use. +func newUpdateSafetyWatcher() *UpdateSafetyWatcher { + return &UpdateSafetyWatcher{ + snapshots: make(map[string]*updateSafetySnapshot), + cap: updateSafetySnapshotCap, + } +} + +// Observe is called on every patrol cycle with the current DockerHosts slice. +// On first call for a container it records a baseline and returns nothing. +// On subsequent calls it diffs ImageDigest and RestartCount: +// - Digest changed -> emit a reliability finding (Info or Warning). +// - Already-detected change, restarts increased -> re-emit as Warning. +// - Already-detected change, stable for >= verifyWindow, no new restarts -> emit resolve sentinel. +func (w *UpdateSafetyWatcher) Observe(hosts []models.DockerHost, now time.Time) (emit []*Finding, resolve []resolveSentinel) { + w.mu.Lock() + defer w.mu.Unlock() + + seen := make(map[string]struct{}, len(hosts)*8) + + for _, host := range hosts { + for _, c := range host.Containers { + if c.ImageDigest == "" { + continue + } + key := host.ID + "/" + c.ID + seen[key] = struct{}{} + + snap, exists := w.snapshots[key] + if !exists { + // First observation -- record baseline, emit nothing. + w.snapshots[key] = &updateSafetySnapshot{ + digest: c.ImageDigest, + restartCount: c.RestartCount, + lastSeenAt: now, + } + continue + } + snap.lastSeenAt = now + + if snap.detectedAt.IsZero() { + // State A: no change detected yet. + if c.ImageDigest == snap.digest { + snap.restartCount = c.RestartCount + continue + } + // Digest changed -- transition to state B. + snap.priorDigest = snap.digest + snap.changeDigest = c.ImageDigest + snap.baseRestarts = snap.restartCount + snap.lastEmittedRestarts = snap.restartCount + snap.detectedAt = now + snap.digest = c.ImageDigest + snap.restartCount = c.RestartCount + + severity := FindingSeverityInfo + if c.RestartCount > snap.baseRestarts { + severity = FindingSeverityWarning + snap.lastEmittedRestarts = c.RestartCount + } + emit = append(emit, buildUpdateSafetyFinding(key, host, c, snap, severity, now, now)) + continue + } + + // State B: change already detected, verifying stability. + snap.digest = c.ImageDigest + snap.restartCount = c.RestartCount + restartsAfterChange := c.RestartCount - snap.baseRestarts + + if restartsAfterChange > snap.lastEmittedRestarts-snap.baseRestarts { + // New restarts since last emission -- escalate to Warning. + snap.lastEmittedRestarts = c.RestartCount + emit = append(emit, buildUpdateSafetyFinding(key, host, c, snap, FindingSeverityWarning, snap.detectedAt, now)) + continue + } + + if now.Sub(snap.detectedAt) >= updateSafetyVerifyWindow && restartsAfterChange == 0 { + // Stable for the full window -- emit resolve sentinel and reset. + dedupKey := UpdateSafetyFindingPrefix + ":" + key + resolve = append(resolve, resolveSentinel{DedupKey: dedupKey, Reason: updateSafetyResolveReason}) + snap.detectedAt = time.Time{} + snap.priorDigest = "" + snap.changeDigest = "" + snap.baseRestarts = 0 + snap.lastEmittedRestarts = 0 + } + // Otherwise: still in window, no new restarts -- do nothing. + } + } + + w.pruneLRULocked(seen) + return emit, resolve +} + +// buildUpdateSafetyFinding constructs a Finding for a detected image change. +func buildUpdateSafetyFinding(key string, host models.DockerHost, c models.DockerContainer, snap *updateSafetySnapshot, severity FindingSeverity, detectedAt, lastSeenAt time.Time) *Finding { + name := c.Name + if name == "" { + name = c.ID + } + dedupKey := UpdateSafetyFindingPrefix + ":" + key + restartsAfterChange := c.RestartCount - snap.baseRestarts + + detail := fmt.Sprintf( + "Image digest changed from %.16s to %.16s", + snap.priorDigest, snap.changeDigest, + ) + if restartsAfterChange > 0 { + detail += fmt.Sprintf( + " Container has restarted %d time(s) since the update.", + restartsAfterChange, + ) + } + + return &Finding{ + ID: dedupKey, + Key: dedupKey, + Severity: severity, + Category: FindingCategoryReliability, + ResourceID: key, + ResourceName: name, + ResourceType: "app-container", + Node: host.Hostname, + Title: fmt.Sprintf("Container %q image updated", name), + Description: detail, + Evidence: fmt.Sprintf( + "prior_digest=%s new_digest=%s restart_count=%d", + snap.priorDigest, snap.changeDigest, c.RestartCount, + ), + Source: updateSafetySource, + DetectedAt: detectedAt, + LastSeenAt: lastSeenAt, + } +} + +// pruneLRULocked evicts state-A snapshots for unseen containers, then if +// still over cap evicts the least-recently-seen non-current-cycle entries. +// Called with w.mu held. +func (w *UpdateSafetyWatcher) pruneLRULocked(seen map[string]struct{}) { + // Remove state-A snapshots for containers no longer observed. + for k, snap := range w.snapshots { + if _, ok := seen[k]; !ok && snap.detectedAt.IsZero() { + delete(w.snapshots, k) + } + } + + if w.cap <= 0 || len(w.snapshots) <= w.cap { + return + } + + type entry struct { + key string + at time.Time + } + candidates := make([]entry, 0, len(w.snapshots)) + for k, snap := range w.snapshots { + if _, protected := seen[k]; protected { + continue + } + candidates = append(candidates, entry{key: k, at: snap.lastSeenAt}) + } + sort.Slice(candidates, func(i, j int) bool { + return candidates[i].at.Before(candidates[j].at) + }) + excess := len(w.snapshots) - w.cap + for i := 0; i < excess && i < len(candidates); i++ { + delete(w.snapshots, candidates[i].key) + } +} diff --git a/internal/ai/findings_update_safety_test.go b/internal/ai/findings_update_safety_test.go new file mode 100644 index 000000000..56885c71c --- /dev/null +++ b/internal/ai/findings_update_safety_test.go @@ -0,0 +1,258 @@ +package ai + +import ( + "fmt" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" +) + +// makeHost builds a minimal DockerHost with one container for test use. +func makeHost(hostID, cID, digest string, restarts int) models.DockerHost { + return models.DockerHost{ + ID: hostID, + Hostname: hostID + ".host", + Containers: []models.DockerContainer{ + { + ID: cID, + Name: "container-" + cID, + ImageDigest: digest, + RestartCount: restarts, + }, + }, + } +} + +func containerKey(hostID, cID string) string { return hostID + "/" + cID } + +// TestUpdateSafety_FirstObserveIsSilent verifies no findings are returned +// on the very first observe trip (baseline-only). +func TestUpdateSafety_FirstObserveIsSilent(t *testing.T) { + w := newUpdateSafetyWatcher() + now := time.Now() + hosts := []models.DockerHost{makeHost("h1", "c1", "sha256:aaa", 0)} + emit, resolve := w.Observe(hosts, now) + if len(emit) != 0 || len(resolve) != 0 { + t.Fatalf("first observe: want empty emit+resolve, got emit=%d resolve=%d", len(emit), len(resolve)) + } +} + +// TestUpdateSafety_DigestChangeEmitsFinding verifies that a changed digest on +// the second trip produces exactly one Info finding with the correct shape. +func TestUpdateSafety_DigestChangeEmitsFinding(t *testing.T) { + w := newUpdateSafetyWatcher() + now := time.Now() + hosts1 := []models.DockerHost{makeHost("h1", "c1", "sha256:aaa", 0)} + w.Observe(hosts1, now) + + hosts2 := []models.DockerHost{makeHost("h1", "c1", "sha256:bbb", 0)} + emit, resolve := w.Observe(hosts2, now.Add(5*time.Second)) + if len(emit) != 1 { + t.Fatalf("digest change: want 1 finding, got %d", len(emit)) + } + if len(resolve) != 0 { + t.Fatalf("digest change: want 0 resolves, got %d", len(resolve)) + } + f := emit[0] + if f.Category != FindingCategoryReliability { + t.Errorf("category: want %q, got %q", FindingCategoryReliability, f.Category) + } + if f.Severity != FindingSeverityInfo { + t.Errorf("severity: want %q, got %q (no restarts yet)", FindingSeverityInfo, f.Severity) + } + if f.ResourceType != "app-container" { + t.Errorf("resource_type: want %q, got %q", "app-container", f.ResourceType) + } + key := containerKey("h1", "c1") + wantDedupKey := UpdateSafetyFindingPrefix + ":" + key + if f.ID != wantDedupKey { + t.Errorf("ID: want %q, got %q", wantDedupKey, f.ID) + } + if f.Source != updateSafetySource { + t.Errorf("source: want %q, got %q", updateSafetySource, f.Source) + } +} + +// TestUpdateSafety_RestartAfterDigestEscalatesToWarning verifies that when +// RestartCount increases after a digest change, the next observe emits a +// Warning-severity finding via the same dedup key. +func TestUpdateSafety_RestartAfterDigestEscalatesToWarning(t *testing.T) { + w := newUpdateSafetyWatcher() + t0 := time.Now() + + // Trip 1: baseline. + w.Observe([]models.DockerHost{makeHost("h1", "c1", "sha256:aaa", 0)}, t0) + + // Trip 2: digest changed, no restarts yet -- should emit Info. + emit2, _ := w.Observe([]models.DockerHost{makeHost("h1", "c1", "sha256:bbb", 0)}, t0.Add(5*time.Second)) + if len(emit2) != 1 || emit2[0].Severity != FindingSeverityInfo { + t.Fatalf("trip 2: want 1 Info finding, got %v", emit2) + } + + // Trip 3: same digest, but restart count increased -- should escalate. + emit3, resolve3 := w.Observe([]models.DockerHost{makeHost("h1", "c1", "sha256:bbb", 2)}, t0.Add(10*time.Second)) + if len(emit3) != 1 { + t.Fatalf("trip 3: want 1 escalated finding, got %d", len(emit3)) + } + if emit3[0].Severity != FindingSeverityWarning { + t.Errorf("trip 3 severity: want %q, got %q", FindingSeverityWarning, emit3[0].Severity) + } + if len(resolve3) != 0 { + t.Errorf("trip 3: want 0 resolves, got %d", len(resolve3)) + } +} + +// TestUpdateSafety_StableWindowEmitsResolveSentinel verifies that after the +// verify window has elapsed with no new restarts, Observe returns a resolve +// sentinel and no new findings. +func TestUpdateSafety_StableWindowEmitsResolveSentinel(t *testing.T) { + w := newUpdateSafetyWatcher() + t0 := time.Now() + + // Trip 1: baseline. + w.Observe([]models.DockerHost{makeHost("h1", "c1", "sha256:aaa", 0)}, t0) + + // Trip 2: digest changed -- emit finding. + emit2, _ := w.Observe([]models.DockerHost{makeHost("h1", "c1", "sha256:bbb", 0)}, t0.Add(5*time.Second)) + if len(emit2) != 1 { + t.Fatalf("trip 2: want 1 finding, got %d", len(emit2)) + } + + // Trip 3: same digest, no restarts, window elapsed -- should emit resolve sentinel. + afterWindow := t0.Add(5*time.Second + updateSafetyVerifyWindow + time.Second) + emit3, resolve3 := w.Observe([]models.DockerHost{makeHost("h1", "c1", "sha256:bbb", 0)}, afterWindow) + if len(emit3) != 0 { + t.Errorf("trip 3: want 0 findings after stable window, got %d", len(emit3)) + } + if len(resolve3) != 1 { + t.Fatalf("trip 3: want 1 resolve sentinel, got %d", len(resolve3)) + } + key := containerKey("h1", "c1") + wantKey := UpdateSafetyFindingPrefix + ":" + key + if resolve3[0].DedupKey != wantKey { + t.Errorf("resolve DedupKey: want %q, got %q", wantKey, resolve3[0].DedupKey) + } + if resolve3[0].Reason != updateSafetyResolveReason { + t.Errorf("resolve Reason: want %q, got %q", updateSafetyResolveReason, resolve3[0].Reason) + } +} + +// TestUpdateSafety_EmptyDigestEmitsNothing verifies that containers with an +// empty ImageDigest (agent not yet reporting one) are silently skipped. +func TestUpdateSafety_EmptyDigestEmitsNothing(t *testing.T) { + w := newUpdateSafetyWatcher() + now := time.Now() + + hosts := []models.DockerHost{ + { + ID: "h1", + Hostname: "h1.host", + Containers: []models.DockerContainer{ + {ID: "c1", Name: "web", ImageDigest: "", RestartCount: 0}, + }, + }, + } + // Multiple trips -- should always be silent. + for i := 0; i < 3; i++ { + emit, resolve := w.Observe(hosts, now.Add(time.Duration(i)*10*time.Second)) + if len(emit) != 0 || len(resolve) != 0 { + t.Fatalf("trip %d: empty digest should emit nothing, got emit=%d resolve=%d", i+1, len(emit), len(resolve)) + } + } +} + +// TestUpdateSafety_TwoContainersDontCollide verifies that digest changes on +// two different containers produce distinct, non-colliding dedup keys. +func TestUpdateSafety_TwoContainersDontCollide(t *testing.T) { + w := newUpdateSafetyWatcher() + t0 := time.Now() + + hosts1 := []models.DockerHost{ + { + ID: "h1", + Hostname: "h1.host", + Containers: []models.DockerContainer{ + {ID: "cA", Name: "alpha", ImageDigest: "sha256:aaa", RestartCount: 0}, + {ID: "cB", Name: "beta", ImageDigest: "sha256:zzz", RestartCount: 0}, + }, + }, + } + w.Observe(hosts1, t0) + + // Change digests on both containers simultaneously. + hosts2 := []models.DockerHost{ + { + ID: "h1", + Hostname: "h1.host", + Containers: []models.DockerContainer{ + {ID: "cA", Name: "alpha", ImageDigest: "sha256:bbb", RestartCount: 0}, + {ID: "cB", Name: "beta", ImageDigest: "sha256:yyy", RestartCount: 0}, + }, + }, + } + emit, _ := w.Observe(hosts2, t0.Add(5*time.Second)) + if len(emit) != 2 { + t.Fatalf("two containers: want 2 findings, got %d", len(emit)) + } + keyA := UpdateSafetyFindingPrefix + ":h1/cA" + keyB := UpdateSafetyFindingPrefix + ":h1/cB" + ids := map[string]bool{emit[0].ID: true, emit[1].ID: true} + if !ids[keyA] || !ids[keyB] { + t.Errorf("dedup keys: want %q and %q, got %v", keyA, keyB, ids) + } +} + +// TestUpdateSafety_LRUPrunesUnseenStateAEntries verifies that state-A +// snapshots for containers that disappear from the host list are evicted. +func TestUpdateSafety_LRUPrunesUnseenStateAEntries(t *testing.T) { + w := newUpdateSafetyWatcher() + w.cap = 5 + now := time.Now() + + // Seed 5 containers in state A. + hosts := make([]models.DockerHost, 5) + for i := 0; i < 5; i++ { + hosts[i] = makeHost("h1", fmt.Sprintf("c%d", i), "sha256:aaa", 0) + } + // Flatten into a single host with all containers. + allContainers := make([]models.DockerContainer, 0, 5) + for _, h := range hosts { + allContainers = append(allContainers, h.Containers...) + } + combined := []models.DockerHost{{ID: "h1", Hostname: "h1.host", Containers: allContainers}} + w.Observe(combined, now) + if len(w.snapshots) != 5 { + t.Fatalf("want 5 snapshots after seeding, got %d", len(w.snapshots)) + } + + // Next observe: only 2 containers remain -- state-A unseen ones should be pruned. + reduced := []models.DockerHost{{ + ID: "h1", + Hostname: "h1.host", + Containers: []models.DockerContainer{ + {ID: "c0", Name: "container-c0", ImageDigest: "sha256:aaa", RestartCount: 0}, + {ID: "c1", Name: "container-c1", ImageDigest: "sha256:aaa", RestartCount: 0}, + }, + }} + w.Observe(reduced, now.Add(10*time.Second)) + if len(w.snapshots) != 2 { + t.Errorf("after pruning: want 2 snapshots, got %d", len(w.snapshots)) + } +} + +// TestUpdateSafety_WindowNotElapsedNoResolve verifies that a stable digest +// within the verify window does NOT emit a resolve sentinel. +func TestUpdateSafety_WindowNotElapsedNoResolve(t *testing.T) { + w := newUpdateSafetyWatcher() + t0 := time.Now() + + w.Observe([]models.DockerHost{makeHost("h1", "c1", "sha256:aaa", 0)}, t0) + w.Observe([]models.DockerHost{makeHost("h1", "c1", "sha256:bbb", 0)}, t0.Add(5*time.Second)) + + // Well within the window. + emit, resolve := w.Observe([]models.DockerHost{makeHost("h1", "c1", "sha256:bbb", 0)}, t0.Add(30*time.Second)) + if len(emit) != 0 || len(resolve) != 0 { + t.Errorf("within window: want silent, got emit=%d resolve=%d", len(emit), len(resolve)) + } +}