Classify incident anomalies

docs/release-control/v6/internal/subsystems/api-contracts.md

@@ -134,6 +134,9 @@ the change.
 Restart timeline entries are also a first-class contract now: `restart` change
 kinds can serialize Docker and Kubernetes restart metadata instead of being
 folded into generic state transitions.
+Incident-driven anomaly entries are also a first-class contract now:
+`metric_anomaly` change kinds can serialize canonical incident rollup changes
+instead of being flattened into generic status churn.
 For relationship changes, the `from` and `to` fields now summarize the actual
 edge(s) rather than only the parent pointer, so the API contract keeps the
 graph transition legible even before the frontend expands the related-resource

docs/release-control/v6/internal/subsystems/unified-resources.md

@@ -95,6 +95,10 @@ The change emitter now also classifies canonical restart changes for Docker
 and Kubernetes resources when restart counters increase or uptime resets, so
 the timeline can distinguish restarts from generic state transitions instead
 of flattening them into status-only noise.
+The same change emitter now also classifies canonical incident changes as
+`metric_anomaly` records when the incident rollup changes, so resource
+anomalies stay attached to the canonical incident surface instead of being
+inferred later from metric noise or alert-adjacent heuristics.
 That store also now migrates legacy `resource_changes` tables that still carry
 the pre-v6 `timestamp` column by backfilling canonical `observed_at` values,
 adding the newer `occurred_at` field, and preserving the legacy timestamp on

internal/api/contract_test.go

@@ -3866,6 +3866,64 @@ func TestContract_ResourceTimelineRestartJSONSnapshot(t *testing.T) {
 	assertJSONSnapshot(t, got, want)
 }
 
+func TestContract_ResourceTimelineAnomalyJSONSnapshot(t *testing.T) {
+	now := time.Date(2026, 3, 18, 17, 12, 0, 0, time.UTC)
+	payload := struct {
+		ResourceID    string                            `json:"resourceId"`
+		RecentChanges []unifiedresources.ResourceChange `json:"recentChanges"`
+		Count         int                               `json:"count"`
+	}{
+		ResourceID: "storage:1",
+		RecentChanges: []unifiedresources.ResourceChange{
+			{
+				ID:            "chg-anomaly-1",
+				ObservedAt:    now,
+				OccurredAt:    &now,
+				ResourceID:    "storage:1",
+				Kind:          unifiedresources.ChangeAnomaly,
+				From:          "none",
+				To:            "capacity_runway_low[warning]:PBS datastore archive is READ_ONLY",
+				SourceType:    unifiedresources.SourcePulseDiff,
+				SourceAdapter: unifiedresources.AdapterProxmox,
+				Confidence:    unifiedresources.ConfidenceHigh,
+				Reason:        "resource incident changed",
+				Metadata: map[string]any{
+					"changedFields": []string{"incidents"},
+				},
+			},
+		},
+		Count: 1,
+	}
+
+	got, err := json.Marshal(payload)
+	if err != nil {
+		t.Fatalf("marshal resource timeline anomaly response: %v", err)
+	}
+
+	const want = `{
+		"resourceId":"storage:1",
+		"recentChanges":[
+			{
+				"id":"chg-anomaly-1",
+				"observedAt":"2026-03-18T17:12:00Z",
+				"occurredAt":"2026-03-18T17:12:00Z",
+				"resourceId":"storage:1",
+				"kind":"metric_anomaly",
+				"from":"none",
+				"to":"capacity_runway_low[warning]:PBS datastore archive is READ_ONLY",
+				"sourceType":"pulse_diff",
+				"sourceAdapter":"proxmox_adapter",
+				"confidence":"high",
+				"reason":"resource incident changed",
+				"metadata":{"changedFields":["incidents"]}
+			}
+		],
+		"count":1
+	}`
+
+	assertJSONSnapshot(t, got, want)
+}
+
 func TestContract_ResourceFacetsJSONSnapshot(t *testing.T) {
 	now := time.Date(2026, 3, 18, 17, 0, 0, 0, time.UTC)
 	payload := struct {

internal/unifiedresources/change_emission.go

@@ -113,6 +113,11 @@ func buildResourceChange(before Resource, beforeOK bool, after Resource, afterOK
 		change.From = resourceRestartSummary(before)
 		change.To = resourceRestartSummary(after)
 		change.Reason = "resource restart detected"
+	case resourceIncidentChanged(before, after):
+		change.Kind = ChangeAnomaly
+		change.From = resourceIncidentSummary(before)
+		change.To = resourceIncidentSummary(after)
+		change.Reason = "resource incident changed"
 	case before.Status != after.Status || dockerCommandChanged(before, after) || dockerUpdateStatusChanged(before, after) || proxmoxLifecycleChanged(before, after):
 		change.Kind = ChangeStateTransition
 		change.From = resourceStateSummary(before)
@@ -176,6 +181,9 @@ func resourceChangedFields(before, after Resource) []string {
 	if !reflect.DeepEqual(before.Identity, after.Identity) {
 		changed = append(changed, "identity")
 	}
+	if resourceIncidentChanged(before, after) {
+		changed = append(changed, "incidents")
+	}
 	if dockerRestartChanged(before, after) {
 		changed = append(changed, "docker.restartCount", "docker.uptimeSeconds")
 	}
@@ -302,6 +310,68 @@ func resourceRestartSummary(resource Resource) string {
 	return strings.Join(parts, "|")
 }
 
+func resourceIncidentChanged(before, after Resource) bool {
+	return resourceIncidentFingerprint(before.Incidents) != resourceIncidentFingerprint(after.Incidents)
+}
+
+func resourceIncidentSummary(resource Resource) string {
+	return resourceIncidentSummaryFromSlice(resource.Incidents)
+}
+
+func resourceIncidentSummaryFromSlice(incidents []ResourceIncident) string {
+	if len(incidents) == 0 {
+		return "none"
+	}
+
+	labels := make([]string, 0, len(incidents))
+	for _, incident := range incidents {
+		labels = append(labels, resourceIncidentLabel(incident))
+	}
+
+	sort.Strings(labels)
+	if len(labels) == 1 {
+		return labels[0]
+	}
+	if len(labels) <= 3 {
+		return strings.Join(labels, ", ")
+	}
+	return fmt.Sprintf("%d incidents", len(labels))
+}
+
+func resourceIncidentFingerprint(incidents []ResourceIncident) string {
+	if len(incidents) == 0 {
+		return "none"
+	}
+
+	labels := make([]string, 0, len(incidents))
+	for _, incident := range incidents {
+		labels = append(labels, fmt.Sprintf("%s|%s|%s|%s|%s",
+			strings.TrimSpace(incident.Provider),
+			strings.TrimSpace(incident.NativeID),
+			strings.TrimSpace(incident.Code),
+			strings.TrimSpace(string(incident.Severity)),
+			strings.TrimSpace(incident.Summary),
+		))
+	}
+
+	sort.Strings(labels)
+	return strings.Join(labels, "||")
+}
+
+func resourceIncidentLabel(incident ResourceIncident) string {
+	code := strings.TrimSpace(incident.Code)
+	if code == "" {
+		code = "incident"
+	}
+	if severity := strings.TrimSpace(string(incident.Severity)); severity != "" {
+		code += fmt.Sprintf("[%s]", severity)
+	}
+	if summary := strings.TrimSpace(incident.Summary); summary != "" {
+		code += fmt.Sprintf(":%s", summary)
+	}
+	return code
+}
+
 func resourceRelationSummary(resource Resource) string {
 	if summary := resourceRelationshipSummary(resource.Relationships); summary != "" {
 		return summary

internal/unifiedresources/change_emission_test.go

@@ -190,6 +190,39 @@ func TestBuildResourceChange_ClassifiesKubernetesRestartChange(t *testing.T) {
 	}
 }
 
+func TestBuildResourceChange_ClassifiesIncidentAnomalyChange(t *testing.T) {
+	before := Resource{
+		ID:     "storage:1",
+		Type:   ResourceTypeStorage,
+		Name:   "storage-1",
+		Status: StatusOnline,
+	}
+	after := before
+	after.Incidents = []ResourceIncident{{
+		Provider: "pbs",
+		NativeID: "datastore:capacity_runway_low",
+		Code:     "capacity_runway_low",
+		Severity: "warning",
+		Source:   "pbs",
+		Summary:  "PBS datastore archive is READ_ONLY",
+	}}
+	refreshResourceIncidentRollup(&after)
+
+	change := buildResourceChange(before, true, after, true, time.Now().UTC(), nil, SourcePulseDiff, "")
+	if change == nil {
+		t.Fatal("expected incident anomaly change, got nil")
+	}
+	if change.Kind != ChangeAnomaly {
+		t.Fatalf("Kind = %q, want %q", change.Kind, ChangeAnomaly)
+	}
+	if change.From != "none" || change.To != "capacity_runway_low[warning]:PBS datastore archive is READ_ONLY" {
+		t.Fatalf("From/To = %q/%q, want incident summaries", change.From, change.To)
+	}
+	if !sameStringSet(mustChangedFields(t, change), []string{"incidents"}) {
+		t.Fatalf("changedFields = %+v, want incidents", mustChangedFields(t, change))
+	}
+}
+
 func TestBuildResourceChange_ClassifiesConfigUpdate(t *testing.T) {
 	before := Resource{
 		ID:     "vm:1",

internal/unifiedresources/code_standards_test.go

@@ -290,8 +290,11 @@ func TestResourceChangeEmissionCoversRelationshipAndCapabilityChanges(t *testing
 	requiredSnippets := []string{
 		"relatedResourceIDs(change.ResourceID, before, after)",
 		"case resourceRestartChanged(before, after):",
+		"case resourceIncidentChanged(before, after):",
 		"if !reflect.DeepEqual(before.Relationships, after.Relationships) {",
 		"changed = append(changed, \"relationships\")",
+		"if resourceIncidentChanged(before, after) {",
+		"changed = append(changed, \"incidents\")",
 		"if dockerRestartChanged(before, after) {",
 		"changed = append(changed, \"docker.restartCount\", \"docker.uptimeSeconds\")",
 		"if kubernetesRestartChanged(before, after) {",
@@ -300,6 +303,7 @@ func TestResourceChangeEmissionCoversRelationshipAndCapabilityChanges(t *testing
 		"changed = append(changed, \"capabilities\")",
 		"resourceRelationshipSummary(relationships []ResourceRelationship)",
 		"resourceRestartSummary(resource Resource) string",
+		"resourceIncidentSummary(resource Resource) string",
 	}
 	for _, snippet := range requiredSnippets {
 		if !strings.Contains(source, snippet) {