gsadmin/pulse
1
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-09-10 02:25:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Build operational trust lifecycle foundation

Browse Source
This commit is contained in:
rcourtman
2026-07-19 00:23:27 +01:00
parent c2df4ad277
commit cf0486492e
31 changed files with 3607 additions and 198 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/release-control/v6/internal/records/operational-trust-lifecycle-evidence-notification-linkage-2026-07-19.md
+69
View File
@@ -0,0 +1,69 @@
# Operational Trust Lifecycle, Evidence, and Notification Linkage Record
Date: 2026-07-19
## Scope
This record closes the Phase 1 foundation slice in
`OPERATIONAL_TRUST_IMPLEMENTATION_SPEC.md`. It does not claim completion of
protection posture, the Patrol attention workbench, availability attachment,
governed actions, or the full Operational Trust specification.
## Canonical runtime result
1. `internal/operationaltrust/contracts.go` defines the shared typed evidence,
operational record, lifecycle transition, and notification-link contracts.
2. Evidence distinguishes complete, partial, and unavailable observations;
confirmed, inferred, and unknown confidence; sufficient, partial, denied,
and unknown permission posture; provider source; observation and ingestion
time; bounded payload references; and auditable identity correlation.
3. Alert records carry additive operational records, bounded evidence, the
latest transition, and a bounded durable transition timeline. Legacy alert
projections use explicit partial/unknown reasons instead of inventing
provider certainty.
4. Acknowledgement and unacknowledgement are lifecycle transitions, never
resolution. Resolution adds distinct recovery evidence and persists the
resolved record and timeline into alert history.
5. Canonical detector evidence is preserved as typed source evidence. Repeated
observations do not duplicate transition ids. A recurrence inside the
existing cooldown window reopens the same stable cause with a
resolved-to-open detector transition and retains the prior timeline.
6. The notification queue and audit log persist one `NotificationLink` per
included alert. Grouped delivery retains every record and transition id.
Queue retry retains the same notification id and links, while queued,
delivering, retrying, delivered, cancelled, failed, and dead-letter states
remain delivery consequences rather than operational truth.
7. Resolution notifications link to the recovery-evidence transition.
Partial cancellation of a grouped firing delivery removes only the
resolved alert and its corresponding link.
8. Existing queue databases migrate additively with JSON link columns on both
queue and audit tables. Older alert JSON remains readable, and new fields
are optional for supported clients.
9. `frontend-modern/src/types/operationalTrust.ts` is the single TypeScript
projection used by the alert API type instead of re-declaring lifecycle and
evidence vocabulary in UI features.
## Proof
The completed slice is covered by:
1. deterministic evidence, transition, and notification id tests
2. evidence limitation, permission, freshness, correlation, clone, and
validation tests
3. alert open, acknowledgement, unacknowledgement, resolution, recurrence,
JSON compatibility, history persistence, and deep-copy tests
4. grouped notification linkage, retry, delivery, cancellation rewrite,
database migration, audit persistence, destination identity, DLQ API, and
query-plan tests
5. focused Go race proof for operational lifecycle and notification linkage
6. frontend TypeScript type-check and formatting proof
7. subsystem registry, contract, and control-plane audits
## Boundary carried forward
Phase 1 supplies typed truth and durable linkage but intentionally adds no new
customer-facing operational-trust presentation. Freshness, completeness,
source explanation, protection posture, Patrol attention projection, and
action eligibility must consume these contracts in their owning later phases.
Notification queue status must never become a navigation count, alert state,
or substitute detector.
docs/release-control/v6/internal/subsystems/alerts.md
+14 -1
View File
@@ -103,10 +103,13 @@ is no longer true.
71. `frontend-modern/src/stores/websocket.ts`
72. `frontend-modern/src/utils/alerts.ts`
73. `frontend-modern/src/utils/alertsActivation.ts`
74. `internal/operationaltrust/contracts.go`
75. `internal/alerts/operational_contract.go`
## Shared Boundaries
1. `internal/proxmoxidentity/backup_identity.go` shared with `monitoring`, `storage-recovery`: Proxmox PBS backup subject identity is a shared runtime boundary for monitoring backup freshness, backup-age alert attribution, and recovery-point guest mapping.
1. `internal/operationaltrust/contracts.go` shared with `notifications`: the operational trust contract is jointly consumed by canonical alert lifecycle ownership and notification delivery linkage without making delivery state operational truth.
2. `internal/proxmoxidentity/backup_identity.go` shared with `monitoring`, `storage-recovery`: Proxmox PBS backup subject identity is a shared runtime boundary for monitoring backup freshness, backup-age alert attribution, and recovery-point guest mapping.
Alert multiline field presentation is shared with frontend-primitives:
notification, timeline, threshold ignored-prefix, and resource threshold note
editors must compose the shared `FormTextarea` primitive for label/id/help
@@ -125,6 +128,16 @@ overview must preserve active alert truth while notification delivery is
pending review or snoozed. Notification activation must not clear the browser
active-alert store, suppress resource indicators, lock threshold/history
configuration, or claim that monitoring has stopped.
Operational evidence and lifecycle identity are typed through
`internal/operationaltrust`. Evidence envelopes distinguish completeness,
confidence, permissions, freshness, correlation, and bounded provider detail.
Evidence and transition identifiers are deterministic under retry. Active and
resolved alert compatibility payloads carry an additive canonical operational
record, latest transition, and bounded evidence envelopes; legacy alert paths
must migrate through `internal/alerts/operational_contract.go` and name their
limited provenance honestly rather than inventing confirmed provider evidence.
Acknowledgement remains distinct from resolution, and every resolution
transition references recovery evidence separate from its trigger evidence.
## Extension Points
docs/release-control/v6/internal/subsystems/api-contracts.md
+17 -5
View File
@@ -47,6 +47,7 @@ product API routes free of maintainer commercial analytics.
10. `pkg/pulsecli/fleet.go`
11. `pkg/pulsecli/root.go`
12. `frontend-modern/src/types/api.ts`
12a. `frontend-modern/src/types/operationalTrust.ts`
13. `frontend-modern/src/types/actionAudit.ts`
14. `frontend-modern/src/api/actionAudit.ts`
7a. `frontend-modern/src/api/resourceActions.ts`
@@ -140,6 +141,16 @@ product API routes free of maintainer commercial analytics.
73b. `pkg/extensions/ai_autofix.go`
83. `scripts/generate-types.go`
83a. `internal/api/agent_fleet_doctor.go`
84. `internal/api/notification_queue.go`
The alert and notification transports expose the operational-trust contract
additively. Alert payloads may carry a typed operational record, latest
transition, bounded transition timeline, and evidence envelopes. Notification
queue and DLQ payloads may carry typed links back to those records and exact
transitions. Older payloads without these optional fields remain readable;
frontend consumers use `frontend-modern/src/types/operationalTrust.ts` as the
single TypeScript projection rather than recreating lifecycle or evidence
enums locally.
## Shared Boundaries
@@ -1601,11 +1612,12 @@ payload shape change when the portal presents compact client rows.
Pulse-Pro-as-page-name copy in callback titles, actions, or retry
guidance.
71. `internal/api/licensing_legacy_retry.go` shared with `cloud-paid`: the background legacy-exchange retry loop carries both API payload contract and cloud-paid entitlement boundary ownership.
72. `internal/api/notifications.go` shared with `notifications`: notification handlers are both a notification delivery control surface and a canonical API payload contract boundary.
73. `internal/api/org_handlers.go` shared with `organization-settings`: organization management handlers are both an organization settings control surface and a canonical API payload contract boundary.
74. `internal/api/org_lifecycle_handlers.go` shared with `organization-settings`: organization lifecycle handlers are both an organization settings control surface and a canonical API payload contract boundary.
75. `internal/api/payments_webhook_handlers.go` shared with `cloud-paid`: commercial payment webhook handlers carry both API payload contract and cloud-paid billing boundary ownership.
76. `internal/api/public_signup_handlers.go` shared with `cloud-paid`: hosted signup handlers carry both API payload contract and cloud-paid hosted provisioning boundary ownership.
72. `internal/api/notification_queue.go` shared with `notifications`: the notification queue and DLQ handler is both a notification delivery consequence surface and a canonical API payload boundary for operational transition links.
73. `internal/api/notifications.go` shared with `notifications`: notification handlers are both a notification delivery control surface and a canonical API payload contract boundary.
74. `internal/api/org_handlers.go` shared with `organization-settings`: organization management handlers are both an organization settings control surface and a canonical API payload contract boundary.
75. `internal/api/org_lifecycle_handlers.go` shared with `organization-settings`: organization lifecycle handlers are both an organization settings control surface and a canonical API payload contract boundary.
76. `internal/api/payments_webhook_handlers.go` shared with `cloud-paid`: commercial payment webhook handlers carry both API payload contract and cloud-paid billing boundary ownership.
77. `internal/api/public_signup_handlers.go` shared with `cloud-paid`: hosted signup handlers carry both API payload contract and cloud-paid hosted provisioning boundary ownership.
That same shared boundary also owns public hosted-signup response privacy:
syntactically valid `/api/public/signup` requests must return one generic
`202 Accepted` Pulse Account message whether provisioning/email side effects
docs/release-control/v6/internal/subsystems/notifications.md
+16 -1
View File
@@ -26,11 +26,15 @@ notification-management API surfaces.
4. `internal/notifications/webhook_enhanced.go`
5. `internal/api/notifications.go`
6. `frontend-modern/src/api/notifications.ts`
7. `internal/operationaltrust/contracts.go`
8. `internal/api/notification_queue.go`
## Shared Boundaries
1. `frontend-modern/src/api/notifications.ts` shared with `api-contracts`: the notifications frontend client is both a notification delivery control surface and a canonical API payload contract boundary.
2. `internal/api/notifications.go` shared with `api-contracts`: notification handlers are both a notification delivery control surface and a canonical API payload contract boundary.
2. `internal/api/notification_queue.go` shared with `api-contracts`: the notification queue and DLQ handler is both a notification delivery consequence surface and a canonical API payload boundary for operational transition links.
3. `internal/api/notifications.go` shared with `api-contracts`: notification handlers are both a notification delivery control surface and a canonical API payload contract boundary.
4. `internal/operationaltrust/contracts.go` shared with `alerts`: the operational trust contract is jointly consumed by canonical alert lifecycle ownership and notification delivery linkage without making delivery state operational truth.
## Extension Points
@@ -50,6 +54,17 @@ notification-management API surfaces.
2. Keep notification API transport and backend delivery proofs aligned in `registry.json`
3. Preserve explicit queue, webhook-security, and provider-delivery coverage when notification behavior changes
The persistent queue and audit log carry typed `NotificationLink` entries for
every linked alert in a grouped delivery. A link keeps one notification id,
operational record id, lifecycle transition id, cause key, destination id, and
delivery state across queue retries. Queue state is delivery evidence only:
it must never create, resolve, acknowledge, suppress, or count operational
records. Resolution notifications link to the recovery-evidence transition.
Partial cancellation of a grouped firing delivery removes only the links for
the resolved alerts, while retry, failure, cancellation, and dead-letter state
remain inspectable in the queue and audit records. Destination identities are
stable opaque routing identities and must not expose credentials.
## Current State
This subsystem now makes email, webhook, Apprise, queueing, and delivery
docs/release-control/v6/internal/subsystems/registry.json
+70 -2
View File
@@ -779,6 +779,14 @@
"cloud-paid"
]
},
{
"path": "internal/api/notification_queue.go",
"rationale": "the notification queue and DLQ handler is both a notification delivery consequence surface and a canonical API payload boundary for operational transition links",
"subsystems": [
"api-contracts",
"notifications"
]
},
{
"path": "internal/api/notifications.go",
"rationale": "notification handlers are both a notification delivery control surface and a canonical API payload contract boundary",
@@ -956,6 +964,14 @@
"monitoring"
]
},
{
"path": "internal/operationaltrust/contracts.go",
"rationale": "the operational trust contract is jointly consumed by canonical alert lifecycle ownership and notification delivery linkage without making delivery state operational truth",
"subsystems": [
"alerts",
"notifications"
]
},
{
"path": "internal/proxmoxidentity/backup_identity.go",
"rationale": "Proxmox PBS backup subject identity is a shared runtime boundary for monitoring backup freshness, backup-age alert attribution, and recovery-point guest mapping",
@@ -1899,7 +1915,8 @@
"owned_prefixes": [
"frontend-modern/src/components/Alerts/",
"frontend-modern/src/features/alerts/",
"internal/alerts/"
"internal/alerts/",
"internal/operationaltrust/"
],
"owned_files": [
"frontend-modern/src/pages/Alerts.tsx",
@@ -1988,11 +2005,28 @@
"test_prefixes": [],
"exact_files": [
"internal/alerts/migration_characterization_test.go",
"internal/alerts/operational_contract_test.go",
"internal/alerts/unified_eval_parity_test.go",
"internal/alerts/unified_eval_test.go",
"internal/alerts/unified_incidents_test.go"
]
},
{
"id": "operational-trust-contracts",
"label": "operational evidence and lifecycle contract proof",
"match_prefixes": [
"internal/operationaltrust/"
],
"match_files": [
"internal/alerts/operational_contract.go"
],
"allow_same_subsystem_tests": false,
"test_prefixes": [],
"exact_files": [
"internal/alerts/operational_contract_test.go",
"internal/operationaltrust/contracts_test.go"
]
},
{
"id": "alerts-frontend-surface",
"label": "alerts frontend surface proof",
@@ -2174,6 +2208,7 @@
"internal/alerts/guest_snapshot_test.go",
"internal/alerts/history_concurrency_test.go",
"internal/alerts/history_test.go",
"internal/alerts/operational_contract_test.go",
"internal/alerts/synology_test.go",
"internal/alerts/threshold_resolution_shared_test.go",
"internal/alerts/update_alerts_test.go"
@@ -2236,6 +2271,7 @@
"frontend-modern/src/constants/apiScopes.ts",
"frontend-modern/src/types/actionAudit.ts",
"frontend-modern/src/types/api.ts",
"frontend-modern/src/types/operationalTrust.ts",
"frontend-modern/src/utils/agentInstallCommand.ts",
"frontend-modern/src/utils/apiTokenPresentation.ts",
"frontend-modern/src/utils/infrastructureSettingsPresentation.ts",
@@ -2671,6 +2707,19 @@
"frontend-modern/src/types/api.ts"
]
},
{
"id": "frontend-operational-trust-types",
"label": "frontend operational trust type sync proof",
"match_prefixes": [],
"match_files": [
"frontend-modern/src/types/operationalTrust.ts"
],
"allow_same_subsystem_tests": false,
"test_prefixes": [],
"exact_files": [
"frontend-modern/src/types/api.ts"
]
},
{
"id": "frontend-action-audit-types",
"label": "frontend action audit type proof",
@@ -5131,7 +5180,9 @@
],
"owned_files": [
"frontend-modern/src/api/notifications.ts",
"internal/api/notifications.go"
"internal/api/notification_queue.go",
"internal/api/notifications.go",
"internal/operationaltrust/contracts.go"
],
"verification": {
"allow_same_subsystem_tests": false,
@@ -5145,16 +5196,33 @@
"match_prefixes": [],
"match_files": [
"frontend-modern/src/api/notifications.ts",
"internal/api/notification_queue.go",
"internal/api/notifications.go"
],
"allow_same_subsystem_tests": false,
"test_prefixes": [],
"exact_files": [
"frontend-modern/src/api/__tests__/notifications.test.ts",
"internal/api/notification_queue_additional_test.go",
"internal/api/notification_queue_error_test.go",
"internal/api/notifications_scope_test.go",
"internal/api/notifications_test.go"
]
},
{
"id": "notification-operational-trust-contract",
"label": "notification operational trust linkage proof",
"match_prefixes": [],
"match_files": [
"internal/operationaltrust/contracts.go"
],
"allow_same_subsystem_tests": false,
"test_prefixes": [],
"exact_files": [
"internal/notifications/queue_test.go",
"internal/operationaltrust/contracts_test.go"
]
},
{
"id": "notification-delivery-runtime",
"label": "notification delivery runtime proof",
frontend-modern/src/types/api.ts
+8
View File
@@ -1,6 +1,7 @@
// Properly typed TypeScript interfaces for Pulse API
import type { Resource } from './resource';
import type { EvidenceEnvelope, LifecycleTransition, OperationalRecord } from './operationalTrust';
export interface APITokenRecord {
id: string;
@@ -1208,6 +1209,9 @@ export interface Alert {
type: string;
level: 'warning' | 'critical';
resourceId: string;
canonicalSpecId?: string;
canonicalKind?: string;
canonicalState?: string;
resourceName: string;
node: string;
nodeDisplayName?: string;
@@ -1221,6 +1225,10 @@ export interface Alert {
ackTime?: string;
ackUser?: string;
metadata?: Record<string, unknown>;
operationalRecord?: OperationalRecord;
latestTransition?: LifecycleTransition;
transitions?: LifecycleTransition[];
evidence?: EvidenceEnvelope[];
}
export interface ResolvedAlert extends Alert {
frontend-modern/src/types/operationalTrust.ts
+128
View File
@@ -0,0 +1,128 @@
export type EvidenceCompleteness = 'complete' | 'partial' | 'unavailable';
export type EvidenceConfidence = 'confirmed' | 'inferred' | 'unknown';
export type EvidencePermissions = 'sufficient' | 'partial' | 'denied' | 'unknown';
export type EvidenceFreshness = 'fresh' | 'stale' | 'unknown';
export interface EvidenceSource {
provider: string;
collector: string;
instance?: string;
}
export interface EvidenceSubject {
resourceId?: string;
providerRef?: string;
providerScope?: string;
}
export interface EvidenceReason {
code: string;
message?: string;
}
export interface EvidencePayloadRef {
kind: string;
id: string;
}
export interface IdentityCorrelation {
rule: string;
matchedFields: Record<string, string>;
candidateCount: number;
}
export interface EvidenceEnvelope {
id: string;
source: EvidenceSource;
subject: EvidenceSubject;
observedAt: string;
ingestedAt: string;
validUntil?: string;
completeness: EvidenceCompleteness;
confidence: EvidenceConfidence;
reason?: EvidenceReason;
permissions: EvidencePermissions;
payloadRef?: EvidencePayloadRef;
correlation?: IdentityCorrelation;
}
export type OperationalState =
| 'observing'
| 'open'
| 'acknowledged'
| 'suppressed'
| 'resolving'
| 'resolved'
| 'stale'
| 'unknown';
export type OperationalSeverity = 'info' | 'warning' | 'critical' | 'unknown';
export interface OperationalAcknowledgement {
at: string;
by: string;
note?: string;
}
export interface OperationalSuppression {
at: string;
by: string;
reason: string;
expiresAt?: string;
}
export interface OperationalRecord {
id: string;
canonicalSpecId: string;
subjectResourceId: string;
state: OperationalState;
severity: OperationalSeverity;
firstObservedAt: string;
lastObservedAt: string;
stateChangedAt: string;
resolvedAt?: string;
acknowledgement?: OperationalAcknowledgement;
suppression?: OperationalSuppression;
evidenceIds: string[];
causeKey: string;
relatedResourceIds: string[];
impactSummary?: string;
recommendedNextStep?: string;
}
export type TransitionCause =
| 'detector_decision'
| 'acknowledgement'
| 'unacknowledgement'
| 'suppression'
| 'suppression_expired'
| 'recovery_evidence'
| 'collection_stale'
| 'collection_unknown';
export interface LifecycleTransition {
id: string;
operationalRecordId: string;
from: OperationalState;
to: OperationalState;
at: string;
cause: TransitionCause;
causeKey: string;
evidenceIds: string[];
reason?: string;
}
export type NotificationDeliveryState =
'queued' | 'delivering' | 'delivered' | 'retrying' | 'cancelled' | 'failed' | 'dead_letter';
export interface NotificationLink {
notificationId: string;
operationalRecordId: string;
transitionId: string;
lifecycleState: OperationalState;
causeKey: string;
destinationId: string;
deliveryState: NotificationDeliveryState;
attemptedAt?: string;
completedAt?: string;
}
internal/alerts/active_lifecycle.go
+33 -12
View File
@@ -80,10 +80,7 @@ func (m *Manager) clearAlert(alertID string) {
}
publicID := effectiveAlertID(alert, alertID)
resolvedAlert := &ResolvedAlert{
Alert: alert,
ResolvedTime: time.Now(),
}
resolvedAlert := m.newResolvedAlert(alert, time.Now(), nil)
m.addRecentlyResolvedUnlocked(resolvedAlert)
@@ -203,6 +200,23 @@ func (m *Manager) preserveAlertState(alertID string, updated *Alert) {
} else {
updated.EscalationTimes = nil
}
if existing.OperationalRecord != nil {
value := existing.OperationalRecord.Clone()
updated.OperationalRecord = &value
}
if existing.LatestTransition != nil {
value := existing.LatestTransition.Clone()
updated.LatestTransition = &value
}
for _, transition := range existing.Transitions {
updated.Transitions = appendOperationalTransition(
updated.Transitions,
transition.Clone(),
)
}
for _, envelope := range existing.Evidence {
updated.Evidence = appendOperationalEvidence(updated.Evidence, envelope.Clone())
}
log.Debug().
Str("alertID", alertID).
@@ -212,6 +226,19 @@ func (m *Manager) preserveAlertState(alertID string, updated *Alert) {
return
}
if m.historyManager != nil {
previous := m.historyManager.LatestAlertForAlert(updated)
if previous != nil &&
previous.OperationalRecord != nil &&
previous.OperationalRecord.ResolvedAt != nil &&
previous.OperationalRecord.ResolvedAt.After(time.Now().Add(-recentlyResolvedRetention)) {
if !previous.StartTime.IsZero() {
updated.StartTime = previous.StartTime
}
mergeOperationalRecurrence(updated, previous, updated.LastSeen)
}
}
if record, ok := m.getAckRecordNoLock(updated, alertID); ok && record.acknowledged {
updated.Acknowledged = true
updated.AckUser = record.user
@@ -301,10 +328,7 @@ func (m *Manager) clearResourceOfflineAlert(resourceID, resourceName, host, reso
m.removeActiveAlertNoLock(alertID)
resolvedAlert := &ResolvedAlert{
Alert: alert,
ResolvedTime: time.Now(),
}
resolvedAlert := m.newResolvedAlert(alert, time.Now(), nil)
m.addRecentlyResolvedWithPrimaryLock(resolvedAlert)
m.safeCallResolvedAlertCallback(alert, alertID, true)
@@ -356,10 +380,7 @@ func (m *Manager) clearAlertNoLock(alertID string) {
}
m.removeActiveAlertNoLock(alertID)
resolvedAlert := &ResolvedAlert{
Alert: alert,
ResolvedTime: time.Now(),
}
resolvedAlert := m.newResolvedAlert(alert, time.Now(), nil)
m.addRecentlyResolvedWithPrimaryLock(resolvedAlert)
internal/alerts/alert_store.go
+6
View File
@@ -1,5 +1,7 @@
package alerts
import "time"
func activeAlertStorageKey(alert *Alert, fallback string) string {
if fallback == "" && alert == nil {
return ""
@@ -103,6 +105,7 @@ func (m *Manager) setActiveAlertNoLock(storageKey string, alert *Alert) {
return
}
backfillCanonicalIdentity(alert)
ensureOperationalContract(alert, time.Now())
requestedKey := storageKey
storageKey = activeAlertStorageKey(alert, storageKey)
for _, staleKey := range []string{requestedKey, alert.ID, alert.CanonicalState} {
@@ -116,6 +119,9 @@ func (m *Manager) setActiveAlertNoLock(storageKey string, alert *Alert) {
}
m.activeAlerts[storageKey] = alert
m.registerActiveAlertAliasNoLock(storageKey, alert)
if m.historyManager != nil {
m.historyManager.UpdateAlertOperationalContractForAlert(alert)
}
}
func (m *Manager) unregisterActiveAlertAliasNoLock(storageKey string, alert *Alert) {
internal/alerts/canonical_lifecycle.go
+31 -8
View File
@@ -4,6 +4,7 @@ import (
"time"
alertspecs "github.com/rcourtman/pulse-go-rewrite/internal/alerts/specs"
"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
"github.com/rs/zerolog/log"
)
@@ -399,7 +400,7 @@ func (m *Manager) evaluateCanonicalLifecycleAlert(params canonicalLifecycleAlert
alert.Metadata["resourceType"] = string(params.Spec.ResourceType)
}
applyCanonicalIdentity(alert, params.Spec.ID, string(params.Spec.Kind))
applyCanonicalOperationalEvidence(alert, params.Spec, params.Evidence, time.Now())
m.preserveAlertState(storageKey, alert)
m.setActiveAlertNoLock(storageKey, alert)
if params.AddToRecent {
@@ -462,10 +463,21 @@ func (m *Manager) evaluateCanonicalLifecycleAlert(params canonicalLifecycleAlert
}
m.removeActiveAlertNoLock(storageKey)
resolvedAlert := &ResolvedAlert{
Alert: existing,
ResolvedTime: params.Evidence.ObservedAt,
recoveryEvidence, hasRecoveryEvidence := canonicalAlertEvidenceEnvelope(
params.Spec,
params.Evidence,
existing.Instance,
time.Now(),
)
var recoveryEvidenceRef *operationaltrust.EvidenceEnvelope
if hasRecoveryEvidence {
recoveryEvidenceRef = &recoveryEvidence
}
resolvedAlert := m.newResolvedAlert(
existing,
params.Evidence.ObservedAt,
recoveryEvidenceRef,
)
m.addRecentlyResolvedWithPrimaryLock(resolvedAlert)
m.safeCallResolvedAlertCallback(existing, storageKey, true)
return result, true
@@ -564,7 +576,7 @@ func (m *Manager) evaluateCanonicalStatefulAlert(params canonicalStatefulAlertPa
alert.Metadata["resourceType"] = string(params.Spec.ResourceType)
}
applyCanonicalIdentity(alert, params.Spec.ID, string(params.Spec.Kind))
applyCanonicalOperationalEvidence(alert, params.Spec, params.Evidence, time.Now())
m.preserveAlertState(storageKey, alert)
m.setActiveAlertNoLock(storageKey, alert)
if params.AddToRecent {
@@ -640,10 +652,21 @@ func (m *Manager) evaluateCanonicalStatefulAlert(params canonicalStatefulAlertPa
}
m.removeActiveAlertNoLock(storageKey)
resolvedAlert := &ResolvedAlert{
Alert: existing,
ResolvedTime: params.Evidence.ObservedAt,
recoveryEvidence, hasRecoveryEvidence := canonicalAlertEvidenceEnvelope(
params.Spec,
params.Evidence,
existing.Instance,
time.Now(),
)
var recoveryEvidenceRef *operationaltrust.EvidenceEnvelope
if hasRecoveryEvidence {
recoveryEvidenceRef = &recoveryEvidence
}
resolvedAlert := m.newResolvedAlert(
existing,
params.Evidence.ObservedAt,
recoveryEvidenceRef,
)
m.addRecentlyResolvedWithPrimaryLock(resolvedAlert)
m.safeCallResolvedAlertCallback(existing, storageKey, true)
return result, true
internal/alerts/canonical_metric.go
+16 -5
View File
@@ -6,6 +6,7 @@ import (
"time"
alertspecs "github.com/rcourtman/pulse-go-rewrite/internal/alerts/specs"
"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
"github.com/rs/zerolog/log"
)
@@ -173,7 +174,7 @@ func (m *Manager) evaluateCanonicalMetricAlert(spec alertspecs.ResourceAlertSpec
delete(m.pendingAlerts, trackingKey)
}
result, err := alertspecs.Evaluate(spec, metricPreviousState(spec, existingAlert), alertspecs.AlertEvidence{
evidence := alertspecs.AlertEvidence{
ObservedAt: observedAt,
MetricThreshold: &alertspecs.MetricThresholdEvidence{
Metric: metricType,
@@ -183,7 +184,8 @@ func (m *Manager) evaluateCanonicalMetricAlert(spec alertspecs.ResourceAlertSpec
Recovery: spec.MetricThreshold.Recovery,
Critical: spec.MetricThreshold.Critical,
},
})
}
result, err := alertspecs.Evaluate(spec, metricPreviousState(spec, existingAlert), evidence)
if err != nil {
log.Warn().
Err(err).
@@ -235,6 +237,7 @@ func (m *Manager) evaluateCanonicalMetricAlert(spec alertspecs.ResourceAlertSpec
}
applyCanonicalIdentity(alert, spec.ID, string(spec.Kind))
applyCanonicalOperationalEvidence(alert, spec, evidence, time.Now())
m.preserveAlertState(storageKey, alert)
m.setActiveAlertNoLock(storageKey, alert)
m.recentAlerts[trackingKey] = alert
@@ -291,6 +294,7 @@ func (m *Manager) evaluateCanonicalMetricAlert(spec alertspecs.ResourceAlertSpec
existingAlert.Metadata[k] = v
}
applyCanonicalIdentity(existingAlert, spec.ID, string(spec.Kind))
applyCanonicalOperationalEvidence(existingAlert, spec, evidence, time.Now())
shouldRenotify := false
if existingAlert.Acknowledged {
@@ -313,10 +317,17 @@ func (m *Manager) evaluateCanonicalMetricAlert(spec alertspecs.ResourceAlertSpec
return
}
resolvedAlert := &ResolvedAlert{
Alert: existingAlert,
ResolvedTime: observedAt,
recoveryEvidence, hasRecoveryEvidence := canonicalAlertEvidenceEnvelope(
spec,
evidence,
existingAlert.Instance,
time.Now(),
)
var recoveryEvidenceRef *operationaltrust.EvidenceEnvelope
if hasRecoveryEvidence {
recoveryEvidenceRef = &recoveryEvidence
}
resolvedAlert := m.newResolvedAlert(existingAlert, observedAt, recoveryEvidenceRef)
m.removeActiveAlertNoLock(storageKey)
m.saveActiveAlertsAsync("canonical metric resolution")
m.addRecentlyResolvedWithPrimaryLock(resolvedAlert)
internal/alerts/config_runtime.go
+1 -4
View File
@@ -549,10 +549,7 @@ func (m *Manager) reevaluateActiveAlertsLocked() {
for _, alertID := range alertsToResolve {
if alert, exists := m.getActiveAlertNoLock(alertID); exists {
resolvedAlert := &ResolvedAlert{
Alert: alert,
ResolvedTime: time.Now(),
}
resolvedAlert := m.newResolvedAlert(alert, time.Now(), nil)
trackingKey := canonicalTrackingKeyForAlert(alert)
if _, isPending := m.pendingAlerts[trackingKey]; isPending {
internal/alerts/connection.go
+1 -4
View File
@@ -239,10 +239,7 @@ func (m *Manager) clearConnectionDegradedAlert(snap ConnectionSnapshot) {
m.removeActiveAlertNoLock(alertID)
resolvedAlert := &ResolvedAlert{
Alert: alert,
ResolvedTime: time.Now(),
}
resolvedAlert := m.newResolvedAlert(alert, time.Now(), nil)
m.addRecentlyResolvedWithPrimaryLock(resolvedAlert)
m.safeCallResolvedAlertCallback(alert, alertID, true)
internal/alerts/guest.go
+1 -4
View File
@@ -441,10 +441,7 @@ func (m *Manager) clearGuestPoweredOffAlert(guestID, name string) {
m.removeActiveAlertNoLock(alertID)
downtime := time.Since(alert.StartTime)
resolvedAlert := &ResolvedAlert{
Alert: alert,
ResolvedTime: time.Now(),
}
resolvedAlert := m.newResolvedAlert(alert, time.Now(), nil)
m.addRecentlyResolvedWithPrimaryLock(resolvedAlert)
// Send recovery notification (async to avoid deadlock because callback acquires m.mu.RLock
internal/alerts/history.go
+57
View File
@@ -193,6 +193,63 @@ func (hm *HistoryManager) UpdateAlertLastSeenForAlert(alert *Alert, lastSeen tim
}
}
// UpdateAlertOperationalContractForAlert updates the most recent history
// entry with the durable evidence and lifecycle state from the active alert.
// This keeps incident history and notification linkage on the same record and
// transition identities as the runtime alert.
func (hm *HistoryManager) UpdateAlertOperationalContractForAlert(alert *Alert) {
if alert == nil {
return
}
snapshot := alert.Clone()
if snapshot == nil {
return
}
matchKey := historyIdentityKey(snapshot)
matchID := snapshot.ID
hm.mu.Lock()
defer hm.mu.Unlock()
for i := len(hm.history) - 1; i >= 0; i-- {
entry := hm.history[i].Alert.Clone()
entryKey := historyIdentityKey(entry)
if (matchKey != "" && entryKey == matchKey) ||
(matchID != "" && hm.history[i].Alert.ID == matchID) {
hm.history[i].Alert.OperationalRecord = snapshot.OperationalRecord
hm.history[i].Alert.LatestTransition = snapshot.LatestTransition
hm.history[i].Alert.Transitions = snapshot.Transitions
hm.history[i].Alert.Evidence = snapshot.Evidence
return
}
}
}
// LatestAlertForAlert returns a clone of the newest history entry matching
// the alert's canonical identity, with legacy alert ID as the compatibility
// fallback.
func (hm *HistoryManager) LatestAlertForAlert(alert *Alert) *Alert {
if alert == nil {
return nil
}
matchKey := historyIdentityKey(alert)
matchID := alert.ID
hm.mu.RLock()
defer hm.mu.RUnlock()
for i := len(hm.history) - 1; i >= 0; i-- {
entry := hm.history[i].Alert.Clone()
entryKey := historyIdentityKey(entry)
if (matchKey != "" && entryKey == matchKey) ||
(matchID != "" && hm.history[i].Alert.ID == matchID) {
return entry
}
}
return nil
}
// MigrateActiveAlert updates the most recent history entry for an in-flight
// alert when its canonical runtime identity changes, such as a guest metric
// alert moving from one node-scoped resource key to another after a VM move.
internal/alerts/metric_runtime.go
+1 -4
View File
@@ -490,10 +490,7 @@ func (m *Manager) checkMetric(resourceID, resourceName, node, instance, resource
if value <= clearThreshold {
// Threshold cleared with hysteresis - auto resolve
resolvedAlert := &ResolvedAlert{
Alert: existingAlert,
ResolvedTime: time.Now(),
}
resolvedAlert := m.newResolvedAlert(existingAlert, time.Now(), nil)
// Remove from active alerts
m.removeActiveAlertNoLock(alertID)
internal/alerts/model.go
+56 -24
View File
@@ -1,32 +1,40 @@
package alerts
import "time"
import (
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
)
// Alert represents an active alert
type Alert struct {
ID string `json:"id"`
Type string `json:"type"` // cpu, memory, disk, etc.
Level AlertLevel `json:"level"`
ResourceID string `json:"resourceId"` // guest or node ID
CanonicalSpecID string `json:"canonicalSpecId,omitempty"`
CanonicalKind string `json:"canonicalKind,omitempty"`
CanonicalState string `json:"canonicalState,omitempty"`
ResourceName string `json:"resourceName"`
Node string `json:"node"`
NodeDisplayName string `json:"nodeDisplayName,omitempty"`
Instance string `json:"instance"`
Message string `json:"message"`
Value float64 `json:"value"`
Threshold float64 `json:"threshold"`
StartTime time.Time `json:"startTime"`
LastSeen time.Time `json:"lastSeen"`
Acknowledged bool `json:"acknowledged"`
AckTime *time.Time `json:"ackTime,omitempty"`
AckUser string `json:"ackUser,omitempty"`
Metadata map[string]interface{} `json:"metadata,omitempty"`
LastNotified *time.Time `json:"lastNotified,omitempty"`
LastEscalation int `json:"lastEscalation,omitempty"`
EscalationTimes []time.Time `json:"escalationTimes,omitempty"`
ID string `json:"id"`
Type string `json:"type"` // cpu, memory, disk, etc.
Level AlertLevel `json:"level"`
ResourceID string `json:"resourceId"` // guest or node ID
CanonicalSpecID string `json:"canonicalSpecId,omitempty"`
CanonicalKind string `json:"canonicalKind,omitempty"`
CanonicalState string `json:"canonicalState,omitempty"`
ResourceName string `json:"resourceName"`
Node string `json:"node"`
NodeDisplayName string `json:"nodeDisplayName,omitempty"`
Instance string `json:"instance"`
Message string `json:"message"`
Value float64 `json:"value"`
Threshold float64 `json:"threshold"`
StartTime time.Time `json:"startTime"`
LastSeen time.Time `json:"lastSeen"`
Acknowledged bool `json:"acknowledged"`
AckTime *time.Time `json:"ackTime,omitempty"`
AckUser string `json:"ackUser,omitempty"`
Metadata map[string]interface{} `json:"metadata,omitempty"`
LastNotified *time.Time `json:"lastNotified,omitempty"`
LastEscalation int `json:"lastEscalation,omitempty"`
EscalationTimes []time.Time `json:"escalationTimes,omitempty"`
OperationalRecord *operationaltrust.OperationalRecord `json:"operationalRecord,omitempty"`
LatestTransition *operationaltrust.LifecycleTransition `json:"latestTransition,omitempty"`
Transitions []operationaltrust.LifecycleTransition `json:"transitions,omitempty"`
Evidence []operationaltrust.EvidenceEnvelope `json:"evidence,omitempty"`
}
// Clone returns a deep copy of the alert so it can be safely shared across goroutines.
@@ -55,6 +63,30 @@ func (a *Alert) Clone() *Alert {
clone.Metadata = cloneMetadata(a.Metadata)
}
if a.OperationalRecord != nil {
value := a.OperationalRecord.Clone()
clone.OperationalRecord = &value
}
if a.LatestTransition != nil {
value := a.LatestTransition.Clone()
clone.LatestTransition = &value
}
if len(a.Transitions) > 0 {
clone.Transitions = make([]operationaltrust.LifecycleTransition, len(a.Transitions))
for index := range a.Transitions {
clone.Transitions[index] = a.Transitions[index].Clone()
}
}
if len(a.Evidence) > 0 {
clone.Evidence = make([]operationaltrust.EvidenceEnvelope, len(a.Evidence))
for index := range a.Evidence {
clone.Evidence[index] = a.Evidence[index].Clone()
}
}
return &clone
}
internal/alerts/node.go
+1 -4
View File
@@ -314,10 +314,7 @@ func (m *Manager) clearNodeOfflineAlert(node models.Node) {
// Remove from active alerts
m.removeActiveAlertNoLock(alertID)
resolvedAlert := &ResolvedAlert{
Alert: alert,
ResolvedTime: time.Now(),
}
resolvedAlert := m.newResolvedAlert(alert, time.Now(), nil)
m.addRecentlyResolvedWithPrimaryLock(resolvedAlert)
// Send recovery notification (async to avoid deadlock — callback acquires m.mu.RLock
internal/alerts/operational_contract.go
+598
View File
@@ -0,0 +1,598 @@
package alerts
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"strings"
"time"
alertspecs "github.com/rcourtman/pulse-go-rewrite/internal/alerts/specs"
"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
)
const legacyAlertEvidenceReason = "legacy_alert_projection"
const legacyAlertRecoveryEvidenceReason = "legacy_alert_recovery_projection"
const maxOperationalEvidencePerAlert = 64
const maxOperationalTransitionsPerAlert = 128
func ensureOperationalContract(alert *Alert, ingestedAt time.Time) {
if alert == nil {
return
}
if ingestedAt.IsZero() {
ingestedAt = time.Now()
}
backfillCanonicalIdentity(alert)
if len(alert.Evidence) == 0 {
if envelope, ok := legacyAlertEvidenceEnvelope(alert, ingestedAt); ok {
alert.Evidence = []operationaltrust.EvidenceEnvelope{envelope}
}
}
if strings.TrimSpace(alert.ResourceID) == "" {
return
}
evidenceIDs := operationalEvidenceIDs(alert.Evidence)
state := operationalStateForAlert(alert)
severity := operationalSeverityForAlert(alert)
firstObservedAt := alert.StartTime
if firstObservedAt.IsZero() {
firstObservedAt = alert.LastSeen
}
if firstObservedAt.IsZero() {
firstObservedAt = ingestedAt
}
lastObservedAt := alert.LastSeen
if lastObservedAt.IsZero() || lastObservedAt.Before(firstObservedAt) {
lastObservedAt = firstObservedAt
}
causeKey := strings.TrimSpace(alert.CanonicalState)
if causeKey == "" {
causeKey = strings.TrimSpace(alert.ID)
}
specID := strings.TrimSpace(alert.CanonicalSpecID)
if specID == "" {
specID = "legacy-alert:" + strings.TrimSpace(alert.Type)
}
recordID := causeKey
if recordID == "" {
recordID = specID + ":" + strings.TrimSpace(alert.ResourceID)
}
record := alert.OperationalRecord
previousState := operationaltrust.OperationalObserving
hadRecord := record != nil
if hadRecord {
previousState = record.State
}
if record == nil {
record = &operationaltrust.OperationalRecord{
ID: recordID,
CanonicalSpecID: specID,
SubjectResourceID: strings.TrimSpace(alert.ResourceID),
FirstObservedAt: firstObservedAt,
StateChangedAt: firstObservedAt,
CauseKey: causeKey,
RelatedResourceIDs: []string{},
}
alert.OperationalRecord = record
}
record.ID = recordID
record.CanonicalSpecID = specID
record.SubjectResourceID = strings.TrimSpace(alert.ResourceID)
record.State = state
record.Severity = severity
record.FirstObservedAt = firstObservedAt
record.LastObservedAt = lastObservedAt
record.EvidenceIDs = evidenceIDs
record.CauseKey = causeKey
record.ImpactSummary = strings.TrimSpace(alert.Message)
if state != operationaltrust.OperationalResolved {
record.ResolvedAt = nil
}
if record.RelatedResourceIDs == nil {
record.RelatedResourceIDs = []string{}
}
if alert.Acknowledged {
acknowledgedAt := lastObservedAt
if alert.AckTime != nil && !alert.AckTime.IsZero() {
acknowledgedAt = *alert.AckTime
}
record.StateChangedAt = acknowledgedAt
if hadRecord && previousState != operationaltrust.OperationalAcknowledged &&
acknowledgedAt.Before(lastObservedAt) {
record.StateChangedAt = ingestedAt
}
record.Acknowledgement = &operationaltrust.Acknowledgement{
At: acknowledgedAt,
By: strings.TrimSpace(alert.AckUser),
}
if record.Acknowledgement.By == "" {
record.Acknowledgement.By = "unknown"
}
} else {
record.Acknowledgement = nil
if hadRecord && previousState != state {
record.StateChangedAt = ingestedAt
} else if record.StateChangedAt.IsZero() ||
record.StateChangedAt.Before(firstObservedAt) ||
record.StateChangedAt.After(lastObservedAt) {
record.StateChangedAt = firstObservedAt
}
}
if alert.LatestTransition == nil ||
alert.LatestTransition.OperationalRecordID != record.ID ||
alert.LatestTransition.To != record.State {
alert.LatestTransition = buildCurrentTransition(record, previousState)
}
if alert.LatestTransition != nil {
alert.Transitions = appendOperationalTransition(
alert.Transitions,
alert.LatestTransition.Clone(),
)
}
}
func mergeOperationalRecurrence(
target *Alert,
previous *Alert,
observedAt time.Time,
) bool {
if target == nil ||
previous == nil ||
previous.OperationalRecord == nil ||
previous.OperationalRecord.State != operationaltrust.OperationalResolved {
return false
}
currentEvidence := make([]operationaltrust.EvidenceEnvelope, len(target.Evidence))
for index := range target.Evidence {
currentEvidence[index] = target.Evidence[index].Clone()
}
acknowledged := target.Acknowledged
ackUser := target.AckUser
var originalAckTime *time.Time
if target.AckTime != nil {
value := *target.AckTime
originalAckTime = &value
}
record := previous.OperationalRecord.Clone()
target.OperationalRecord = &record
if previous.LatestTransition != nil {
transition := previous.LatestTransition.Clone()
target.LatestTransition = &transition
}
target.Transitions = nil
for _, transition := range previous.Transitions {
target.Transitions = appendOperationalTransition(
target.Transitions,
transition.Clone(),
)
}
target.Evidence = nil
for _, envelope := range previous.Evidence {
target.Evidence = appendOperationalEvidence(
target.Evidence,
envelope.Clone(),
)
}
for _, envelope := range currentEvidence {
target.Evidence = appendOperationalEvidence(target.Evidence, envelope)
}
target.Acknowledged = false
target.AckTime = nil
target.AckUser = ""
ensureOperationalContract(target, observedAt)
reopened := target.LatestTransition != nil &&
target.LatestTransition.From == operationaltrust.OperationalResolved &&
target.LatestTransition.To == operationaltrust.OperationalOpen
if !reopened || !acknowledged {
return reopened
}
reappliedAt := observedAt
target.Acknowledged = true
target.AckTime = &reappliedAt
target.AckUser = ackUser
ensureOperationalContract(target, observedAt)
target.AckTime = originalAckTime
if target.OperationalRecord != nil &&
target.OperationalRecord.Acknowledgement != nil &&
originalAckTime != nil {
target.OperationalRecord.Acknowledgement.At = *originalAckTime
}
return true
}
func newResolvedAlert(
alert *Alert,
resolvedAt time.Time,
recoveryEvidence *operationaltrust.EvidenceEnvelope,
) *ResolvedAlert {
if alert == nil {
return nil
}
if resolvedAt.IsZero() {
resolvedAt = time.Now()
}
if recoveryEvidence == nil {
if envelope, ok := legacyAlertRecoveryEvidenceEnvelope(alert, resolvedAt); ok {
recoveryEvidence = &envelope
}
}
markOperationalResolved(alert, resolvedAt, recoveryEvidence)
return &ResolvedAlert{
Alert: alert,
ResolvedTime: resolvedAt,
}
}
func markOperationalResolved(
alert *Alert,
resolvedAt time.Time,
recoveryEvidence *operationaltrust.EvidenceEnvelope,
) {
if alert == nil {
return
}
ensureOperationalContract(alert, resolvedAt)
record := alert.OperationalRecord
if record == nil || recoveryEvidence == nil || recoveryEvidence.Validate() != nil {
return
}
alert.Evidence = appendOperationalEvidence(alert.Evidence, recoveryEvidence.Clone())
recoveryEvidenceIDs := []string{recoveryEvidence.ID}
from := record.State
record.State = operationaltrust.OperationalResolved
record.StateChangedAt = resolvedAt
record.ResolvedAt = &resolvedAt
record.EvidenceIDs = operationalEvidenceIDs(alert.Evidence)
id, err := operationaltrust.NewTransitionID(
record.ID,
from,
operationaltrust.OperationalResolved,
resolvedAt,
operationaltrust.TransitionRecoveryEvidence,
record.CauseKey,
recoveryEvidenceIDs,
)
if err != nil {
return
}
alert.LatestTransition = &operationaltrust.LifecycleTransition{
ID: id,
OperationalRecordID: record.ID,
From: from,
To: operationaltrust.OperationalResolved,
At: resolvedAt,
Cause: operationaltrust.TransitionRecoveryEvidence,
CauseKey: record.CauseKey,
EvidenceIDs: recoveryEvidenceIDs,
}
alert.Transitions = appendOperationalTransition(
alert.Transitions,
alert.LatestTransition.Clone(),
)
}
func (m *Manager) newResolvedAlert(
alert *Alert,
resolvedAt time.Time,
recoveryEvidence *operationaltrust.EvidenceEnvelope,
) *ResolvedAlert {
resolved := newResolvedAlert(alert, resolvedAt, recoveryEvidence)
if resolved != nil && m != nil && m.historyManager != nil {
m.historyManager.UpdateAlertOperationalContractForAlert(alert)
}
return resolved
}
func legacyAlertEvidenceEnvelope(
alert *Alert,
ingestedAt time.Time,
) (operationaltrust.EvidenceEnvelope, bool) {
observedAt := alert.LastSeen
if observedAt.IsZero() {
observedAt = alert.StartTime
}
if observedAt.IsZero() {
observedAt = ingestedAt
}
source := operationaltrust.EvidenceSource{
Provider: "pulse",
Collector: "legacy-alert-adapter",
Instance: strings.TrimSpace(alert.Instance),
}
subject := operationaltrust.EvidenceSubject{
ResourceID: strings.TrimSpace(alert.ResourceID),
}
if subject.ResourceID == "" {
subject.ProviderRef = strings.TrimSpace(alert.ID)
if subject.ProviderRef == "" {
return operationaltrust.EvidenceEnvelope{}, false
}
subject.ProviderScope = strings.TrimSpace(alert.Instance)
if subject.ProviderScope == "" {
subject.ProviderScope = "pulse"
}
}
sourceObservationID := strings.TrimSpace(alert.ID)
if sourceObservationID == "" {
sourceObservationID = strings.TrimSpace(alert.CanonicalState)
}
if sourceObservationID == "" {
return operationaltrust.EvidenceEnvelope{}, false
}
id, err := operationaltrust.NewEvidenceID(source, subject, observedAt, sourceObservationID)
if err != nil {
return operationaltrust.EvidenceEnvelope{}, false
}
envelope := operationaltrust.EvidenceEnvelope{
ID: id,
Source: source,
Subject: subject,
ObservedAt: observedAt,
IngestedAt: ingestedAt,
Completeness: operationaltrust.EvidencePartial,
Confidence: operationaltrust.EvidenceUnknown,
Permissions: operationaltrust.EvidencePermissionsUnknown,
Reason: &operationaltrust.EvidenceReason{
Code: legacyAlertEvidenceReason,
},
}
return envelope, envelope.Validate() == nil
}
func applyCanonicalOperationalEvidence(
alert *Alert,
spec alertspecs.ResourceAlertSpec,
evidence alertspecs.AlertEvidence,
ingestedAt time.Time,
) bool {
if alert == nil {
return false
}
envelope, ok := canonicalAlertEvidenceEnvelope(spec, evidence, alert.Instance, ingestedAt)
if !ok {
return false
}
alert.Evidence = appendOperationalEvidence(alert.Evidence, envelope)
return true
}
func canonicalAlertEvidenceEnvelope(
spec alertspecs.ResourceAlertSpec,
evidence alertspecs.AlertEvidence,
instance string,
ingestedAt time.Time,
) (operationaltrust.EvidenceEnvelope, bool) {
if evidence.Envelope != nil {
envelope := evidence.Envelope.Clone()
if envelope.Subject.ResourceID != spec.ResourceID ||
!envelope.ObservedAt.Equal(evidence.ObservedAt) ||
envelope.Validate() != nil {
return operationaltrust.EvidenceEnvelope{}, false
}
return envelope, true
}
payload, err := json.Marshal(evidence)
if err != nil {
return operationaltrust.EvidenceEnvelope{}, false
}
digest := sha256.Sum256(payload)
digestID := hex.EncodeToString(digest[:])
provider := "pulse"
collector := "canonical-alert:" + string(spec.Kind)
if evidence.ProviderIncident != nil {
if value := strings.TrimSpace(evidence.ProviderIncident.Provider); value != "" {
provider = value
}
if value := strings.TrimSpace(evidence.ProviderIncident.Source); value != "" {
collector = value
}
}
source := operationaltrust.EvidenceSource{
Provider: provider,
Collector: collector,
Instance: strings.TrimSpace(instance),
}
subject := operationaltrust.EvidenceSubject{ResourceID: spec.ResourceID}
id, err := operationaltrust.NewEvidenceID(
source,
subject,
evidence.ObservedAt,
digestID,
)
if err != nil {
return operationaltrust.EvidenceEnvelope{}, false
}
envelope := operationaltrust.EvidenceEnvelope{
ID: id,
Source: source,
Subject: subject,
ObservedAt: evidence.ObservedAt,
IngestedAt: ingestedAt,
Completeness: operationaltrust.EvidenceComplete,
Confidence: operationaltrust.EvidenceConfirmed,
Permissions: operationaltrust.EvidencePermissionsSufficient,
PayloadRef: &operationaltrust.EvidencePayloadRef{
Kind: "alert-evidence-digest",
ID: digestID,
},
}
return envelope, envelope.Validate() == nil
}
func legacyAlertRecoveryEvidenceEnvelope(
alert *Alert,
resolvedAt time.Time,
) (operationaltrust.EvidenceEnvelope, bool) {
if alert == nil || strings.TrimSpace(alert.ResourceID) == "" {
return operationaltrust.EvidenceEnvelope{}, false
}
source := operationaltrust.EvidenceSource{
Provider: "pulse",
Collector: "legacy-alert-recovery-adapter",
Instance: strings.TrimSpace(alert.Instance),
}
subject := operationaltrust.EvidenceSubject{
ResourceID: strings.TrimSpace(alert.ResourceID),
}
sourceObservationID := strings.TrimSpace(alert.ID)
if sourceObservationID == "" {
sourceObservationID = strings.TrimSpace(alert.CanonicalState)
}
if sourceObservationID == "" {
return operationaltrust.EvidenceEnvelope{}, false
}
id, err := operationaltrust.NewEvidenceID(
source,
subject,
resolvedAt,
sourceObservationID+":resolved",
)
if err != nil {
return operationaltrust.EvidenceEnvelope{}, false
}
envelope := operationaltrust.EvidenceEnvelope{
ID: id,
Source: source,
Subject: subject,
ObservedAt: resolvedAt,
IngestedAt: resolvedAt,
Completeness: operationaltrust.EvidencePartial,
Confidence: operationaltrust.EvidenceUnknown,
Permissions: operationaltrust.EvidencePermissionsUnknown,
Reason: &operationaltrust.EvidenceReason{
Code: legacyAlertRecoveryEvidenceReason,
},
}
return envelope, envelope.Validate() == nil
}
func operationalStateForAlert(alert *Alert) operationaltrust.OperationalState {
if alert != nil && alert.Acknowledged {
return operationaltrust.OperationalAcknowledged
}
return operationaltrust.OperationalOpen
}
func operationalSeverityForAlert(alert *Alert) operationaltrust.OperationalSeverity {
if alert == nil {
return operationaltrust.SeverityUnknown
}
switch alert.Level {
case AlertLevelCritical:
return operationaltrust.SeverityCritical
case AlertLevelWarning:
return operationaltrust.SeverityWarning
default:
return operationaltrust.SeverityUnknown
}
}
func operationalEvidenceIDs(envelopes []operationaltrust.EvidenceEnvelope) []string {
seen := make(map[string]struct{}, len(envelopes))
result := make([]string, 0, len(envelopes))
for _, envelope := range envelopes {
id := strings.TrimSpace(envelope.ID)
if id == "" {
continue
}
if _, exists := seen[id]; exists {
continue
}
seen[id] = struct{}{}
result = append(result, id)
}
return result
}
func appendOperationalEvidence(
envelopes []operationaltrust.EvidenceEnvelope,
envelope operationaltrust.EvidenceEnvelope,
) []operationaltrust.EvidenceEnvelope {
for index := range envelopes {
if envelopes[index].ID == envelope.ID {
envelopes[index] = envelope
return envelopes
}
}
envelopes = append(envelopes, envelope)
if len(envelopes) > maxOperationalEvidencePerAlert {
envelopes = append(
[]operationaltrust.EvidenceEnvelope(nil),
envelopes[len(envelopes)-maxOperationalEvidencePerAlert:]...,
)
}
return envelopes
}
func appendOperationalTransition(
transitions []operationaltrust.LifecycleTransition,
transition operationaltrust.LifecycleTransition,
) []operationaltrust.LifecycleTransition {
for index := range transitions {
if transitions[index].ID == transition.ID {
transitions[index] = transition
return transitions
}
}
transitions = append(transitions, transition)
if len(transitions) > maxOperationalTransitionsPerAlert {
transitions = append(
[]operationaltrust.LifecycleTransition(nil),
transitions[len(transitions)-maxOperationalTransitionsPerAlert:]...,
)
}
return transitions
}
func buildCurrentTransition(
record *operationaltrust.OperationalRecord,
previousState operationaltrust.OperationalState,
) *operationaltrust.LifecycleTransition {
if record == nil {
return nil
}
from := previousState
cause := operationaltrust.TransitionDetectorDecision
at := record.FirstObservedAt
if record.State == operationaltrust.OperationalAcknowledged {
cause = operationaltrust.TransitionAcknowledgement
at = record.StateChangedAt
} else if previousState == operationaltrust.OperationalAcknowledged &&
record.State == operationaltrust.OperationalOpen {
cause = operationaltrust.TransitionUnacknowledgement
at = record.StateChangedAt
}
if from == record.State {
return nil
}
id, err := operationaltrust.NewTransitionID(
record.ID,
from,
record.State,
at,
cause,
record.CauseKey,
record.EvidenceIDs,
)
if err != nil {
return nil
}
return &operationaltrust.LifecycleTransition{
ID: id,
OperationalRecordID: record.ID,
From: from,
To: record.State,
At: at,
Cause: cause,
CauseKey: record.CauseKey,
EvidenceIDs: append([]string(nil), record.EvidenceIDs...),
}
}
internal/alerts/operational_contract_test.go
+504
View File
@@ -0,0 +1,504 @@
package alerts
import (
"encoding/json"
"testing"
"time"
alertspecs "github.com/rcourtman/pulse-go-rewrite/internal/alerts/specs"
"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
)
func TestEnsureOperationalContractBackfillsLegacyAlertHonestly(t *testing.T) {
firstObservedAt := time.Date(2026, 7, 18, 20, 0, 0, 0, time.UTC)
lastObservedAt := firstObservedAt.Add(time.Minute)
alert := &Alert{
ID: "legacy-alert",
Type: "cpu",
Level: AlertLevelWarning,
ResourceID: "resource-1",
CanonicalSpecID: "metric-threshold:cpu",
CanonicalKind: "metric-threshold",
CanonicalState: "resource-1::metric-threshold:cpu",
Message: "CPU usage exceeded threshold",
StartTime: firstObservedAt,
LastSeen: lastObservedAt,
}
ensureOperationalContract(alert, lastObservedAt.Add(time.Second))
if len(alert.Evidence) != 1 {
t.Fatalf("evidence count = %d, want 1", len(alert.Evidence))
}
if got := alert.Evidence[0].Reason; got == nil || got.Code != legacyAlertEvidenceReason {
t.Fatalf("evidence reason = %+v, want %q", got, legacyAlertEvidenceReason)
}
if got := alert.Evidence[0].Completeness; got != operationaltrust.EvidencePartial {
t.Fatalf("evidence completeness = %q, want partial", got)
}
if err := alert.Evidence[0].Validate(); err != nil {
t.Fatalf("evidence Validate() error = %v", err)
}
if alert.OperationalRecord == nil {
t.Fatal("operational record is nil")
}
if got := alert.OperationalRecord.State; got != operationaltrust.OperationalOpen {
t.Fatalf("operational state = %q, want open", got)
}
if len(alert.OperationalRecord.EvidenceIDs) != 1 ||
alert.OperationalRecord.EvidenceIDs[0] != alert.Evidence[0].ID {
t.Fatalf(
"operational evidence ids = %v, want %q",
alert.OperationalRecord.EvidenceIDs,
alert.Evidence[0].ID,
)
}
if err := alert.OperationalRecord.Validate(); err != nil {
t.Fatalf("operational record Validate() error = %v", err)
}
if alert.LatestTransition == nil {
t.Fatal("latest transition is nil")
}
if got := alert.LatestTransition.Cause; got != operationaltrust.TransitionDetectorDecision {
t.Fatalf("transition cause = %q, want detector decision", got)
}
if err := alert.LatestTransition.Validate(); err != nil {
t.Fatalf("transition Validate() error = %v", err)
}
if len(alert.Transitions) != 1 ||
alert.Transitions[0].ID != alert.LatestTransition.ID {
t.Fatalf("transition timeline = %+v, want initial transition", alert.Transitions)
}
}
func TestCanonicalAlertEvidenceBuildsConfirmedEnvelope(t *testing.T) {
observedAt := time.Date(2026, 7, 18, 20, 0, 0, 0, time.UTC)
spec := alertspecs.ResourceAlertSpec{
ID: "provider-incident:disk-1",
ResourceID: "disk-1",
ResourceType: unifiedresources.ResourceTypeStorage,
Kind: alertspecs.AlertSpecKindProviderIncident,
}
evidence := alertspecs.AlertEvidence{
ObservedAt: observedAt,
ProviderIncident: &alertspecs.ProviderIncidentEvidence{
Provider: "truenas",
NativeID: "provider-alert-7",
Source: "truenas-alert-feed",
},
}
first, ok := canonicalAlertEvidenceEnvelope(spec, evidence, "nas-a", observedAt.Add(time.Second))
if !ok {
t.Fatal("canonicalAlertEvidenceEnvelope() rejected valid evidence")
}
second, ok := canonicalAlertEvidenceEnvelope(spec, evidence, "nas-a", observedAt.Add(2*time.Second))
if !ok {
t.Fatal("canonicalAlertEvidenceEnvelope() retry rejected valid evidence")
}
if first.ID != second.ID {
t.Fatalf("canonical evidence retry id = %q, want %q", second.ID, first.ID)
}
if first.Source.Provider != "truenas" ||
first.Source.Collector != "truenas-alert-feed" ||
first.Completeness != operationaltrust.EvidenceComplete ||
first.Confidence != operationaltrust.EvidenceConfirmed {
t.Fatalf("canonical evidence = %+v", first)
}
if first.PayloadRef == nil || first.PayloadRef.Kind != "alert-evidence-digest" {
t.Fatalf("payload reference = %+v, want bounded digest", first.PayloadRef)
}
}
func TestCanonicalAlertEvidencePreservesTypedLimitations(t *testing.T) {
observedAt := time.Date(2026, 7, 18, 20, 0, 0, 0, time.UTC)
spec := alertspecs.ResourceAlertSpec{
ID: "connectivity:resource-1",
ResourceID: "resource-1",
ResourceType: unifiedresources.ResourceTypeAgent,
Kind: alertspecs.AlertSpecKindConnectivity,
}
source := operationaltrust.EvidenceSource{
Provider: "proxmox",
Collector: "connectivity-check",
}
subject := operationaltrust.EvidenceSubject{ResourceID: spec.ResourceID}
id, err := operationaltrust.NewEvidenceID(source, subject, observedAt, "check-7")
if err != nil {
t.Fatalf("NewEvidenceID() error = %v", err)
}
envelope := operationaltrust.EvidenceEnvelope{
ID: id,
Source: source,
Subject: subject,
ObservedAt: observedAt,
IngestedAt: observedAt,
Completeness: operationaltrust.EvidencePartial,
Confidence: operationaltrust.EvidenceUnknown,
Permissions: operationaltrust.EvidencePermissionsDenied,
Reason: &operationaltrust.EvidenceReason{Code: "permission_denied"},
}
evidence := alertspecs.AlertEvidence{
ObservedAt: observedAt,
Envelope: &envelope,
Connectivity: &alertspecs.ConnectivityEvidence{
Signal: "heartbeat",
Connected: false,
},
}
got, ok := canonicalAlertEvidenceEnvelope(spec, evidence, "pve-a", observedAt)
if !ok {
t.Fatal("canonicalAlertEvidenceEnvelope() rejected typed limited evidence")
}
if got.Permissions != operationaltrust.EvidencePermissionsDenied ||
got.Reason == nil ||
got.Reason.Code != "permission_denied" {
t.Fatalf("limited evidence = %+v", got)
}
mismatched := envelope.Clone()
mismatched.Subject.ResourceID = "other-resource"
evidence.Envelope = &mismatched
if _, ok := canonicalAlertEvidenceEnvelope(spec, evidence, "pve-a", observedAt); ok {
t.Fatal("canonicalAlertEvidenceEnvelope() accepted mismatched subject")
}
}
func TestEnsureOperationalContractTracksAcknowledgementWithoutResolution(t *testing.T) {
firstObservedAt := time.Date(2026, 7, 18, 20, 0, 0, 0, time.UTC)
acknowledgedAt := firstObservedAt.Add(time.Minute)
alert := &Alert{
ID: "alert-1",
Type: "offline",
Level: AlertLevelCritical,
ResourceID: "resource-1",
CanonicalSpecID: "connectivity:offline",
CanonicalState: "resource-1::connectivity:offline",
StartTime: firstObservedAt,
LastSeen: acknowledgedAt,
Acknowledged: true,
AckTime: &acknowledgedAt,
AckUser: "operator-1",
}
ensureOperationalContract(alert, acknowledgedAt)
record := alert.OperationalRecord
if record == nil {
t.Fatal("operational record is nil")
}
if record.State != operationaltrust.OperationalAcknowledged {
t.Fatalf("state = %q, want acknowledged", record.State)
}
if record.ResolvedAt != nil {
t.Fatalf("acknowledged record resolvedAt = %v, want nil", record.ResolvedAt)
}
if record.Acknowledgement == nil || record.Acknowledgement.By != "operator-1" {
t.Fatalf("acknowledgement = %+v, want operator-1", record.Acknowledgement)
}
if alert.LatestTransition == nil ||
alert.LatestTransition.Cause != operationaltrust.TransitionAcknowledgement {
t.Fatalf("latest transition = %+v, want acknowledgement", alert.LatestTransition)
}
if alert.LatestTransition.To != operationaltrust.OperationalAcknowledged {
t.Fatalf("transition to = %q, want acknowledged", alert.LatestTransition.To)
}
}
func TestEnsureOperationalContractTracksUnacknowledgement(t *testing.T) {
firstObservedAt := time.Date(2026, 7, 18, 20, 0, 0, 0, time.UTC)
acknowledgedAt := firstObservedAt.Add(time.Minute)
unacknowledgedAt := acknowledgedAt.Add(time.Minute)
alert := &Alert{
ID: "alert-1",
Type: "offline",
Level: AlertLevelCritical,
ResourceID: "resource-1",
CanonicalSpecID: "connectivity:offline",
CanonicalState: "resource-1::connectivity:offline",
StartTime: firstObservedAt,
LastSeen: firstObservedAt,
}
ensureOperationalContract(alert, firstObservedAt)
alert.Acknowledged = true
alert.AckTime = &acknowledgedAt
alert.AckUser = "operator-1"
alert.LastSeen = acknowledgedAt
ensureOperationalContract(alert, acknowledgedAt)
alert.Acknowledged = false
alert.AckTime = nil
alert.AckUser = ""
alert.LastSeen = unacknowledgedAt
ensureOperationalContract(alert, unacknowledgedAt)
if alert.OperationalRecord == nil ||
alert.OperationalRecord.State != operationaltrust.OperationalOpen {
t.Fatalf("record = %+v, want open", alert.OperationalRecord)
}
if alert.LatestTransition == nil ||
alert.LatestTransition.Cause != operationaltrust.TransitionUnacknowledgement {
t.Fatalf("latest transition = %+v, want unacknowledgement", alert.LatestTransition)
}
if len(alert.Transitions) != 3 {
t.Fatalf("transition count = %d, want open, acknowledged, open", len(alert.Transitions))
}
for _, transition := range alert.Transitions {
if err := transition.Validate(); err != nil {
t.Fatalf("transition %q Validate() error = %v", transition.ID, err)
}
}
}
func TestNewResolvedAlertAddsDistinctRecoveryEvidence(t *testing.T) {
firstObservedAt := time.Date(2026, 7, 18, 20, 0, 0, 0, time.UTC)
resolvedAt := firstObservedAt.Add(5 * time.Minute)
alert := &Alert{
ID: "alert-1",
Type: "cpu",
Level: AlertLevelWarning,
ResourceID: "resource-1",
CanonicalSpecID: "metric-threshold:cpu",
CanonicalState: "resource-1::metric-threshold:cpu",
StartTime: firstObservedAt,
LastSeen: firstObservedAt,
}
ensureOperationalContract(alert, firstObservedAt)
triggerEvidenceID := alert.Evidence[0].ID
resolved := newResolvedAlert(alert, resolvedAt, nil)
if resolved == nil || resolved.OperationalRecord == nil {
t.Fatal("resolved operational record is nil")
}
if resolved.OperationalRecord.State != operationaltrust.OperationalResolved {
t.Fatalf("resolved state = %q, want resolved", resolved.OperationalRecord.State)
}
if len(resolved.Evidence) != 2 {
t.Fatalf("resolved evidence count = %d, want trigger and recovery", len(resolved.Evidence))
}
recoveryEvidence := resolved.Evidence[1]
if recoveryEvidence.ID == triggerEvidenceID {
t.Fatal("recovery evidence reused trigger evidence id")
}
if recoveryEvidence.Reason == nil ||
recoveryEvidence.Reason.Code != legacyAlertRecoveryEvidenceReason {
t.Fatalf("recovery reason = %+v, want legacy recovery projection", recoveryEvidence.Reason)
}
if resolved.LatestTransition == nil ||
resolved.LatestTransition.Cause != operationaltrust.TransitionRecoveryEvidence {
t.Fatalf("resolved transition = %+v, want recovery evidence", resolved.LatestTransition)
}
if len(resolved.LatestTransition.EvidenceIDs) != 1 ||
resolved.LatestTransition.EvidenceIDs[0] != recoveryEvidence.ID {
t.Fatalf(
"resolution transition evidence = %v, want %q",
resolved.LatestTransition.EvidenceIDs,
recoveryEvidence.ID,
)
}
if err := resolved.OperationalRecord.Validate(); err != nil {
t.Fatalf("resolved operational record Validate() error = %v", err)
}
if err := resolved.LatestTransition.Validate(); err != nil {
t.Fatalf("resolved transition Validate() error = %v", err)
}
if len(resolved.Transitions) != 2 {
t.Fatalf("resolved transition count = %d, want trigger and resolution", len(resolved.Transitions))
}
if resolved.Transitions[0].ID == resolved.Transitions[1].ID ||
resolved.Transitions[1].ID != resolved.LatestTransition.ID {
t.Fatalf("resolved transitions = %+v, want distinct ordered timeline", resolved.Transitions)
}
}
func TestResolvedOperationalContractUpdatesHistorySnapshot(t *testing.T) {
firstObservedAt := time.Date(2026, 7, 18, 20, 0, 0, 0, time.UTC)
resolvedAt := firstObservedAt.Add(5 * time.Minute)
alert := &Alert{
ID: "alert-1",
Type: "cpu",
Level: AlertLevelWarning,
ResourceID: "resource-1",
CanonicalSpecID: "metric-threshold:cpu",
CanonicalState: "resource-1::metric-threshold:cpu",
StartTime: firstObservedAt,
LastSeen: firstObservedAt,
}
ensureOperationalContract(alert, firstObservedAt)
history := newTestHistoryManager(t)
history.AddAlert(*alert)
manager := &Manager{historyManager: history}
if resolved := manager.newResolvedAlert(alert, resolvedAt, nil); resolved == nil {
t.Fatal("newResolvedAlert() returned nil")
}
entries := history.GetAllHistory(1)
if len(entries) != 1 {
t.Fatalf("history entries = %d, want 1", len(entries))
}
entry := entries[0]
if entry.OperationalRecord == nil ||
entry.OperationalRecord.State != operationaltrust.OperationalResolved {
t.Fatalf("history operational record = %+v, want resolved", entry.OperationalRecord)
}
if entry.LatestTransition == nil ||
entry.LatestTransition.ID != alert.LatestTransition.ID {
t.Fatalf("history latest transition = %+v, want %q", entry.LatestTransition, alert.LatestTransition.ID)
}
if len(entry.Transitions) != 2 || len(entry.Evidence) != 2 {
t.Fatalf(
"history contract transitions/evidence = %d/%d, want 2/2",
len(entry.Transitions),
len(entry.Evidence),
)
}
}
func TestMergeOperationalRecurrenceReopensStableCause(t *testing.T) {
firstObservedAt := time.Date(2026, 7, 18, 20, 0, 0, 0, time.UTC)
resolvedAt := firstObservedAt.Add(5 * time.Minute)
recurredAt := resolvedAt.Add(time.Minute)
previous := &Alert{
ID: "alert-1",
Type: "cpu",
Level: AlertLevelWarning,
ResourceID: "resource-1",
CanonicalSpecID: "metric-threshold:cpu",
CanonicalState: "resource-1::metric-threshold:cpu",
StartTime: firstObservedAt,
LastSeen: firstObservedAt,
}
ensureOperationalContract(previous, firstObservedAt)
newResolvedAlert(previous, resolvedAt, nil)
recurrence := &Alert{
ID: previous.ID,
Type: previous.Type,
Level: previous.Level,
ResourceID: previous.ResourceID,
CanonicalSpecID: previous.CanonicalSpecID,
CanonicalState: previous.CanonicalState,
StartTime: previous.StartTime,
LastSeen: recurredAt,
}
ensureOperationalContract(recurrence, recurredAt)
recurrenceEvidenceID := recurrence.Evidence[0].ID
if !mergeOperationalRecurrence(recurrence, previous, recurredAt) {
t.Fatal("mergeOperationalRecurrence() did not reopen resolved cause")
}
if recurrence.OperationalRecord == nil ||
recurrence.OperationalRecord.ID != previous.OperationalRecord.ID ||
recurrence.OperationalRecord.State != operationaltrust.OperationalOpen ||
recurrence.OperationalRecord.ResolvedAt != nil {
t.Fatalf("recurrence record = %+v", recurrence.OperationalRecord)
}
if recurrence.LatestTransition == nil ||
recurrence.LatestTransition.From != operationaltrust.OperationalResolved ||
recurrence.LatestTransition.To != operationaltrust.OperationalOpen ||
recurrence.LatestTransition.Cause != operationaltrust.TransitionDetectorDecision {
t.Fatalf("recurrence transition = %+v", recurrence.LatestTransition)
}
if len(recurrence.Transitions) != 3 {
t.Fatalf("recurrence transition count = %d, want open, resolved, reopened", len(recurrence.Transitions))
}
if len(recurrence.Evidence) != 3 {
t.Fatalf("recurrence evidence count = %d, want trigger, recovery, recurrence", len(recurrence.Evidence))
}
if recurrence.Evidence[2].ID != recurrenceEvidenceID {
t.Fatalf("recurrence evidence id = %q, want %q", recurrence.Evidence[2].ID, recurrenceEvidenceID)
}
for _, transition := range recurrence.Transitions {
if err := transition.Validate(); err != nil {
t.Fatalf("transition %q Validate() error = %v", transition.ID, err)
}
}
}
func TestEnsureOperationalContractDoesNotInventCanonicalSubject(t *testing.T) {
alert := &Alert{
ID: "provider-only-alert",
Type: "provider-incident",
Instance: "provider-a",
StartTime: time.Date(2026, 7, 18, 20, 0, 0, 0, time.UTC),
}
ensureOperationalContract(alert, alert.StartTime)
if alert.OperationalRecord != nil {
t.Fatalf("operational record = %+v, want nil without canonical subject", alert.OperationalRecord)
}
if len(alert.Evidence) != 1 || alert.Evidence[0].Subject.ProviderRef == "" {
t.Fatalf("unresolved evidence = %+v, want scoped provider reference", alert.Evidence)
}
}
func TestAlertCloneDoesNotShareOperationalContractState(t *testing.T) {
now := time.Date(2026, 7, 18, 20, 0, 0, 0, time.UTC)
alert := &Alert{
ID: "alert-1",
Type: "cpu",
Level: AlertLevelWarning,
ResourceID: "resource-1",
CanonicalSpecID: "metric-threshold:cpu",
CanonicalState: "resource-1::metric-threshold:cpu",
StartTime: now,
LastSeen: now,
}
ensureOperationalContract(alert, now)
clone := alert.Clone()
clone.OperationalRecord.EvidenceIDs[0] = "mutated"
clone.LatestTransition.EvidenceIDs[0] = "mutated"
clone.Transitions[0].EvidenceIDs[0] = "mutated"
clone.Evidence[0].Reason.Code = "mutated"
if alert.OperationalRecord.EvidenceIDs[0] == "mutated" ||
alert.LatestTransition.EvidenceIDs[0] == "mutated" ||
alert.Transitions[0].EvidenceIDs[0] == "mutated" ||
alert.Evidence[0].Reason.Code == "mutated" {
t.Fatal("Alert.Clone() shared operational contract state")
}
}
func TestAlertOperationalContractJSONRemainsAdditive(t *testing.T) {
legacyPayload := []byte(`{
"id":"alert-1",
"type":"cpu",
"level":"warning",
"resourceId":"resource-1",
"resourceName":"vm-1",
"node":"node-1",
"instance":"pve-1",
"message":"CPU high",
"value":95,
"threshold":90,
"startTime":"2026-07-18T20:00:00Z",
"lastSeen":"2026-07-18T20:00:00Z",
"acknowledged":false
}`)
var alert Alert
if err := json.Unmarshal(legacyPayload, &alert); err != nil {
t.Fatalf("legacy alert unmarshal error = %v", err)
}
ensureOperationalContract(&alert, alert.LastSeen)
payload, err := json.Marshal(alert)
if err != nil {
t.Fatalf("operational alert marshal error = %v", err)
}
var roundTrip Alert
if err := json.Unmarshal(payload, &roundTrip); err != nil {
t.Fatalf("operational alert unmarshal error = %v", err)
}
if roundTrip.ID != "alert-1" ||
roundTrip.OperationalRecord == nil ||
len(roundTrip.Transitions) != 1 ||
len(roundTrip.Evidence) != 1 {
t.Fatalf("round-trip alert = %+v", roundTrip)
}
}
internal/alerts/specs/types.go
+14 -4
View File
@@ -7,6 +7,7 @@ import (
"strings"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
)
@@ -631,10 +632,11 @@ func (s DiscreteStateSpec) Validate() error {
}
type AlertEvidence struct {
ObservedAt time.Time `json:"observedAt"`
Summary string `json:"summary,omitempty"`
Attributes map[string]string `json:"attributes,omitempty"`
ParentConnected *bool `json:"parentConnected,omitempty"`
ObservedAt time.Time `json:"observedAt"`
Summary string `json:"summary,omitempty"`
Attributes map[string]string `json:"attributes,omitempty"`
ParentConnected *bool `json:"parentConnected,omitempty"`
Envelope *operationaltrust.EvidenceEnvelope `json:"envelope,omitempty"`
MetricThreshold *MetricThresholdEvidence `json:"metricThreshold,omitempty"`
SeverityThreshold *SeverityThresholdEvidence `json:"severityThreshold,omitempty"`
@@ -654,6 +656,14 @@ func (e AlertEvidence) validateForKind(kind AlertSpecKind) error {
if e.ObservedAt.IsZero() {
return fmt.Errorf("observed at is required")
}
if e.Envelope != nil {
if err := e.Envelope.Validate(); err != nil {
return fmt.Errorf("evidence envelope: %w", err)
}
if !e.Envelope.ObservedAt.Equal(e.ObservedAt) {
return fmt.Errorf("evidence envelope observation time does not match detector evidence")
}
}
payloads := 0
if e.MetricThreshold != nil {
internal/alerts/specs/types_test.go
+51
View File
@@ -5,6 +5,7 @@ import (
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
)
@@ -680,3 +681,53 @@ func TestAlertTransitionValidateRequiresKindSpecificEvidence(t *testing.T) {
t.Fatalf("unexpected error: %v", err)
}
}
func TestAlertTransitionValidateRequiresEnvelopeObservationMatch(t *testing.T) {
t.Parallel()
observedAt := time.Now().UTC()
transition := AlertTransition{
SpecID: "tank-zfs-health",
ResourceID: "storage:tank",
Kind: AlertSpecKindProviderIncident,
From: AlertStateClear,
To: AlertStateFiring,
At: observedAt,
Evidence: AlertEvidence{
ObservedAt: observedAt,
Envelope: &operationaltrust.EvidenceEnvelope{
ID: "evidence:provider-alert-1",
Source: operationaltrust.EvidenceSource{
Provider: "truenas",
Collector: "canonical-alert-evaluator",
},
Subject: operationaltrust.EvidenceSubject{
ResourceID: "storage:tank",
},
ObservedAt: observedAt,
IngestedAt: observedAt,
Completeness: operationaltrust.EvidenceComplete,
Confidence: operationaltrust.EvidenceConfirmed,
Permissions: operationaltrust.EvidencePermissionsSufficient,
},
ProviderIncident: &ProviderIncidentEvidence{
Provider: "truenas",
Code: "truenas_volume_status",
NativeID: "provider-alert-1",
},
},
}
if err := transition.Validate(); err != nil {
t.Fatalf("Validate() error = %v", err)
}
transition.Evidence.Envelope.ObservedAt = observedAt.Add(time.Minute)
err := transition.Validate()
if err == nil {
t.Fatal("expected envelope observation mismatch")
}
if !strings.Contains(err.Error(), "observation time does not match") {
t.Fatalf("unexpected error: %v", err)
}
}
internal/api/notification_queue_additional_test.go
+23 -4
View File
@@ -11,6 +11,7 @@ import (
"github.com/rcourtman/pulse-go-rewrite/internal/config"
"github.com/rcourtman/pulse-go-rewrite/internal/monitoring"
"github.com/rcourtman/pulse-go-rewrite/internal/notifications"
"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
)
func newNotificationQueueHandlers(t *testing.T) (*NotificationQueueHandlers, *notifications.NotificationQueue) {
@@ -38,10 +39,23 @@ func enqueueDLQNotification(t *testing.T, queue *notifications.NotificationQueue
t.Helper()
notification := &notifications.QueuedNotification{
ID: id,
Type: "webhook",
Status: notifications.QueueStatusDLQ,
Alerts: []*alerts.Alert{{ID: "alert-1", Type: "test"}},
ID: id,
Type: "webhook",
DestinationID: "webhook:primary",
Status: notifications.QueueStatusDLQ,
Alerts: []*alerts.Alert{{
ID: "alert-1",
Type: "test",
OperationalRecord: &operationaltrust.OperationalRecord{
ID: "record-1",
},
LatestTransition: &operationaltrust.LifecycleTransition{
ID: "transition-1",
OperationalRecordID: "record-1",
To: operationaltrust.OperationalOpen,
CauseKey: "alert-1",
},
}},
Config: json.RawMessage(`{}`),
}
if err := queue.Enqueue(notification); err != nil {
@@ -67,6 +81,11 @@ func TestNotificationQueueHandlers_GetDLQAndStats(t *testing.T) {
if len(dlq) != 1 || dlq[0].ID != "notif-1" {
t.Fatalf("DLQ = %+v, want notif-1", dlq)
}
if len(dlq[0].Links) != 1 ||
dlq[0].Links[0].TransitionID != "transition-1" ||
dlq[0].Links[0].DeliveryState != operationaltrust.NotificationDeadLetter {
t.Fatalf("DLQ operational links = %+v", dlq[0].Links)
}
req = httptest.NewRequest(http.MethodGet, "/api/notifications/queue/stats", nil)
rec = httptest.NewRecorder()
internal/notifications/notifications.go
+52 -5
View File
@@ -17,6 +17,7 @@ import (
"net/url"
"os"
"os/exec"
"sort"
"strconv"
"strings"
"sync"
@@ -1387,6 +1388,51 @@ func configJSONForNotificationDeliveryJob(job notificationDeliveryJob) ([]byte,
}
}
func notificationDestinationIDForJob(job notificationDeliveryJob) string {
parts := []string{job.Type}
switch job.Type {
case "email":
if job.EmailConfig != nil {
recipients := append([]string(nil), job.EmailConfig.To...)
sort.Strings(recipients)
parts = append(
parts,
job.EmailConfig.Provider,
job.EmailConfig.SMTPHost,
strconv.Itoa(job.EmailConfig.SMTPPort),
job.EmailConfig.From,
strings.Join(recipients, "\x00"),
)
}
case "webhook":
if job.WebhookConfig != nil {
if id := strings.TrimSpace(job.WebhookConfig.ID); id != "" {
return "webhook:" + id
}
parts = append(
parts,
job.WebhookConfig.Name,
job.WebhookConfig.Service,
job.WebhookConfig.URL,
)
}
case "apprise":
if job.AppriseConfig != nil {
targets := append([]string(nil), job.AppriseConfig.Targets...)
sort.Strings(targets)
parts = append(
parts,
string(job.AppriseConfig.Mode),
job.AppriseConfig.ServerURL,
job.AppriseConfig.ConfigKey,
strings.Join(targets, "\x00"),
)
}
}
sum := sha256.Sum256([]byte(strings.Join(parts, "\x00")))
return "destination:" + hex.EncodeToString(sum[:16])
}
func (n *NotificationManager) enqueueNotificationJobs(queue *NotificationQueue, jobs []notificationDeliveryJob) bool {
if queue == nil {
return false
@@ -1402,11 +1448,12 @@ func (n *NotificationManager) enqueueNotificationJobs(queue *NotificationQueue,
for _, bucket := range notificationQueueBucketsForJob(job, time.Now()) {
notif := &QueuedNotification{
Type: queueTypeForNotificationDeliveryJob(bucket.job),
Alerts: bucket.job.Alerts,
Config: configJSON,
MaxAttempts: 3,
NextRetryAt: bucket.nextRetryAt,
Type: queueTypeForNotificationDeliveryJob(bucket.job),
DestinationID: notificationDestinationIDForJob(bucket.job),
Alerts: bucket.job.Alerts,
Config: configJSON,
MaxAttempts: 3,
NextRetryAt: bucket.nextRetryAt,
}
if err := queue.Enqueue(notif); err != nil {
anyFailed = true
internal/notifications/notifications_test.go
+41
View File
@@ -55,6 +55,47 @@ func flushPending(n *NotificationManager) {
n.sendGroupedAlerts()
}
func TestNotificationDestinationIDIgnoresCredentialsAndTracksRouting(t *testing.T) {
base := notificationDeliveryJob{
Type: "email",
EmailConfig: &EmailConfig{
Provider: "smtp",
SMTPHost: "mail.example.test",
SMTPPort: 587,
From: "pulse@example.test",
To: []string{"ops-b@example.test", "ops-a@example.test"},
Username: "operator",
Password: "secret-a",
},
}
first := notificationDestinationIDForJob(base)
credentialsChanged := base
credentials := *base.EmailConfig
credentials.Username = "other-operator"
credentials.Password = "secret-b"
credentialsChanged.EmailConfig = &credentials
if got := notificationDestinationIDForJob(credentialsChanged); got != first {
t.Fatalf("credential-only destination id = %q, want %q", got, first)
}
orderChanged := base
reordered := *base.EmailConfig
reordered.To = []string{"ops-a@example.test", "ops-b@example.test"}
orderChanged.EmailConfig = &reordered
if got := notificationDestinationIDForJob(orderChanged); got != first {
t.Fatalf("recipient-order destination id = %q, want %q", got, first)
}
routeChanged := base
differentRoute := *base.EmailConfig
differentRoute.To = []string{"security@example.test"}
routeChanged.EmailConfig = &differentRoute
if got := notificationDestinationIDForJob(routeChanged); got == first {
t.Fatalf("routing change kept destination id %q", got)
}
}
func TestNormalizeAppriseConfig(t *testing.T) {
original := AppriseConfig{
Enabled: true,
internal/notifications/queue.go
+500 -88
View File
@@ -11,6 +11,7 @@ import (
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
"github.com/rcourtman/pulse-go-rewrite/internal/securityutil"
"github.com/rcourtman/pulse-go-rewrite/internal/utils"
"github.com/rs/zerolog/log"
@@ -24,6 +25,7 @@ const defaultQueueMaxAttempts = 3
const (
notificationAuditAlertIdentifiersColumn = "alert_identifiers"
legacyNotificationAuditAlertIdentifiersColumn = "alert_ids"
notificationOperationalLinksColumn = "operational_links"
notificationQueueDirName = "notifications"
notificationQueueFileName = "notification_queue.db"
)
@@ -42,20 +44,22 @@ const (
// QueuedNotification represents a notification in the persistent queue
type QueuedNotification struct {
ID string `json:"id"`
Type string `json:"type"` // email, webhook, apprise
Method string `json:"method,omitempty"`
Status NotificationQueueStatus `json:"status"`
Alerts []*alerts.Alert `json:"alerts"`
Config json.RawMessage `json:"config"` // EmailConfig, WebhookConfig, or AppriseConfig
Attempts int `json:"attempts"`
MaxAttempts int `json:"maxAttempts"`
LastAttempt *time.Time `json:"lastAttempt,omitempty"`
LastError *string `json:"lastError,omitempty"`
CreatedAt time.Time `json:"createdAt"`
NextRetryAt *time.Time `json:"nextRetryAt,omitempty"`
CompletedAt *time.Time `json:"completedAt,omitempty"`
PayloadBytes *int `json:"payloadBytes,omitempty"`
ID string `json:"id"`
Type string `json:"type"` // email, webhook, apprise
Method string `json:"method,omitempty"`
DestinationID string `json:"destinationId,omitempty"`
Status NotificationQueueStatus `json:"status"`
Alerts []*alerts.Alert `json:"alerts"`
Links []operationaltrust.NotificationLink `json:"links,omitempty"`
Config json.RawMessage `json:"config"` // EmailConfig, WebhookConfig, or AppriseConfig
Attempts int `json:"attempts"`
MaxAttempts int `json:"maxAttempts"`
LastAttempt *time.Time `json:"lastAttempt,omitempty"`
LastError *string `json:"lastError,omitempty"`
CreatedAt time.Time `json:"createdAt"`
NextRetryAt *time.Time `json:"nextRetryAt,omitempty"`
CompletedAt *time.Time `json:"completedAt,omitempty"`
PayloadBytes *int `json:"payloadBytes,omitempty"`
}
// NotificationQueue manages persistent notification delivery with retries and DLQ
@@ -192,9 +196,10 @@ func (nq *NotificationQueue) initSchema() error {
id TEXT PRIMARY KEY,
type TEXT NOT NULL,
method TEXT,
status TEXT NOT NULL,
alerts TEXT NOT NULL,
config TEXT NOT NULL,
status TEXT NOT NULL,
alerts TEXT NOT NULL,
operational_links TEXT NOT NULL DEFAULT '[]',
config TEXT NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
max_attempts INTEGER NOT NULL DEFAULT 3,
last_attempt INTEGER,
@@ -215,10 +220,11 @@ func (nq *NotificationQueue) initSchema() error {
notification_id TEXT NOT NULL,
type TEXT NOT NULL,
method TEXT,
status TEXT NOT NULL,
alert_identifiers TEXT,
alert_count INTEGER,
attempts INTEGER,
status TEXT NOT NULL,
alert_identifiers TEXT,
alert_count INTEGER,
operational_links TEXT NOT NULL DEFAULT '[]',
attempts INTEGER,
success BOOLEAN,
error_message TEXT,
payload_size INTEGER,
@@ -235,7 +241,19 @@ func (nq *NotificationQueue) initSchema() error {
return err
}
return nq.migrateAlertIdentifierColumns()
if err := nq.migrateAlertIdentifierColumns(); err != nil {
return err
}
if err := nq.ensureJSONColumn(
"notification_queue",
notificationOperationalLinksColumn,
); err != nil {
return err
}
return nq.ensureJSONColumn(
"notification_audit",
notificationOperationalLinksColumn,
)
}
func (nq *NotificationQueue) migrateAlertIdentifierColumns() error {
@@ -293,6 +311,98 @@ func (nq *NotificationQueue) tableColumns(table string) (map[string]bool, error)
return columns, nil
}
func (nq *NotificationQueue) ensureJSONColumn(table, column string) error {
columns, err := nq.tableColumns(table)
if err != nil {
return err
}
if columns[column] {
return nil
}
if _, err := nq.db.Exec(
`ALTER TABLE ` + table + ` ADD COLUMN ` + column + ` TEXT NOT NULL DEFAULT '[]'`,
); err != nil {
return fmt.Errorf("add %s.%s: %w", table, column, err)
}
return nil
}
func notificationLinksForAlerts(
alertsToLink []*alerts.Alert,
destinationID string,
) []operationaltrust.NotificationLink {
destinationID = strings.TrimSpace(destinationID)
if destinationID == "" {
return nil
}
links := make([]operationaltrust.NotificationLink, 0, len(alertsToLink))
for _, alert := range alertsToLink {
if alert == nil ||
alert.OperationalRecord == nil ||
alert.LatestTransition == nil {
continue
}
links = append(links, operationaltrust.NotificationLink{
OperationalRecordID: alert.OperationalRecord.ID,
TransitionID: alert.LatestTransition.ID,
LifecycleState: alert.LatestTransition.To,
CauseKey: alert.LatestTransition.CauseKey,
DestinationID: destinationID,
DeliveryState: operationaltrust.NotificationQueued,
})
}
return links
}
func notificationTransitionIDs(
links []operationaltrust.NotificationLink,
) []string {
ids := make([]string, 0, len(links))
for _, link := range links {
if id := strings.TrimSpace(link.TransitionID); id != "" {
ids = append(ids, id)
}
}
return ids
}
func normalizeNotificationLinks(
notificationID string,
destinationID string,
links []operationaltrust.NotificationLink,
state operationaltrust.NotificationDeliveryState,
attemptedAt *time.Time,
completedAt *time.Time,
) ([]operationaltrust.NotificationLink, error) {
if len(links) == 0 {
return nil, nil
}
normalized := make([]operationaltrust.NotificationLink, 0, len(links))
seen := make(map[string]struct{}, len(links))
for _, link := range links {
link = link.Clone()
link.NotificationID = notificationID
if strings.TrimSpace(link.DestinationID) == "" {
link.DestinationID = strings.TrimSpace(destinationID)
}
link.DeliveryState = state
link.AttemptedAt = attemptedAt
link.CompletedAt = completedAt
if err := link.Validate(); err != nil {
return nil, err
}
key := link.OperationalRecordID + "\x00" +
link.TransitionID + "\x00" +
link.DestinationID
if _, exists := seen[key]; exists {
continue
}
seen[key] = struct{}{}
normalized = append(normalized, link)
}
return normalized, nil
}
// Enqueue adds a notification to the queue
func (nq *NotificationQueue) Enqueue(notif *QueuedNotification) error {
if notif == nil {
@@ -310,8 +420,34 @@ func (nq *NotificationQueue) Enqueue(notif *QueuedNotification) error {
return fmt.Errorf("notification config cannot be empty")
}
if notif.CreatedAt.IsZero() {
notif.CreatedAt = time.Now()
}
if len(notif.Links) == 0 {
notif.Links = notificationLinksForAlerts(
notif.Alerts,
notif.DestinationID,
)
}
if strings.TrimSpace(notif.DestinationID) == "" && len(notif.Links) > 0 {
notif.DestinationID = strings.TrimSpace(notif.Links[0].DestinationID)
}
if notif.ID == "" {
notif.ID = fmt.Sprintf("%s-%d", notif.Type, time.Now().UnixNano())
transitionIDs := notificationTransitionIDs(notif.Links)
if len(transitionIDs) > 0 && strings.TrimSpace(notif.DestinationID) != "" {
id, err := operationaltrust.NewNotificationID(
notif.DestinationID,
notif.Type,
notif.CreatedAt,
transitionIDs,
)
if err != nil {
return fmt.Errorf("build notification id: %w", err)
}
notif.ID = id
} else {
notif.ID = fmt.Sprintf("%s-%d", notif.Type, notif.CreatedAt.UnixNano())
}
}
if notif.Status == "" {
notif.Status = QueueStatusPending
@@ -322,22 +458,35 @@ func (nq *NotificationQueue) Enqueue(notif *QueuedNotification) error {
if notif.Attempts < 0 {
notif.Attempts = 0
}
if notif.CreatedAt.IsZero() {
notif.CreatedAt = time.Now()
var err error
notif.Links, err = normalizeNotificationLinks(
notif.ID,
notif.DestinationID,
notif.Links,
notificationDeliveryStateForQueueStatus(notif.Status),
nil,
nil,
)
if err != nil {
return fmt.Errorf("normalize notification links: %w", err)
}
alertsJSON, err := json.Marshal(notif.Alerts)
if err != nil {
return fmt.Errorf("failed to marshal alerts: %w", err)
}
linksJSON, err := json.Marshal(notif.Links)
if err != nil {
return fmt.Errorf("failed to marshal operational links: %w", err)
}
nq.mu.Lock()
defer nq.mu.Unlock()
query := `
INSERT INTO notification_queue
(id, type, method, status, alerts, config, attempts, max_attempts, created_at, next_retry_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
(id, type, method, status, alerts, operational_links, config, attempts, max_attempts, created_at, next_retry_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`
var nextRetryAt *int64
@@ -352,6 +501,7 @@ func (nq *NotificationQueue) Enqueue(notif *QueuedNotification) error {
notif.Method,
notif.Status,
string(alertsJSON),
string(linksJSON),
string(notif.Config),
notif.Attempts,
notif.MaxAttempts,
@@ -383,19 +533,54 @@ func (nq *NotificationQueue) UpdateStatus(id string, status NotificationQueueSta
nq.mu.Lock()
defer nq.mu.Unlock()
now := time.Now().Unix()
var completedAt *int64
if status == QueueStatusSent || status == QueueStatusFailed || status == QueueStatusDLQ || status == QueueStatusCancelled {
completedAt = &now
}
return nq.updateNotificationStatusNoLock(id, status, errorMsg, time.Now())
}
func (nq *NotificationQueue) updateNotificationStatusNoLock(
id string,
status NotificationQueueStatus,
errorMsg string,
now time.Time,
) error {
links, err := nq.readNotificationLinksNoLock(id)
if err != nil {
return err
}
links = transitionNotificationLinks(
links,
notificationDeliveryStateForQueueStatus(status),
now,
)
linksJSON, err := json.Marshal(links)
if err != nil {
return fmt.Errorf("marshal notification links for status update: %w", err)
}
nowUnix := now.Unix()
var completedAt *int64
if status == QueueStatusSent ||
status == QueueStatusFailed ||
status == QueueStatusDLQ ||
status == QueueStatusCancelled {
completedAt = &nowUnix
}
query := `
UPDATE notification_queue
SET status = ?, last_attempt = ?, last_error = ?, completed_at = ?
SET status = ?, last_attempt = ?, last_error = ?, completed_at = ?,
operational_links = ?,
next_retry_at = CASE WHEN ? THEN NULL ELSE next_retry_at END
WHERE id = ?
`
result, err := nq.db.Exec(query, status, now, errorMsg, completedAt, id)
result, err := nq.db.Exec(
query,
status,
nowUnix,
errorMsg,
completedAt,
string(linksJSON),
completedAt != nil,
id,
)
if err != nil {
return fmt.Errorf("failed to update notification status: %w", err)
}
@@ -408,6 +593,85 @@ func (nq *NotificationQueue) UpdateStatus(id string, status NotificationQueueSta
return nil
}
func (nq *NotificationQueue) readNotificationLinksNoLock(
id string,
) ([]operationaltrust.NotificationLink, error) {
var raw string
if err := nq.db.QueryRow(
`SELECT operational_links FROM notification_queue WHERE id = ?`,
id,
).Scan(&raw); err != nil {
if err == sql.ErrNoRows {
return nil, fmt.Errorf("notification not found: %s", id)
}
return nil, fmt.Errorf("read notification links: %w", err)
}
var links []operationaltrust.NotificationLink
if err := json.Unmarshal([]byte(raw), &links); err != nil {
return nil, fmt.Errorf("unmarshal notification links: %w", err)
}
return links, nil
}
func (nq *NotificationQueue) getNotificationLinks(
id string,
) ([]operationaltrust.NotificationLink, error) {
nq.mu.RLock()
defer nq.mu.RUnlock()
return nq.readNotificationLinksNoLock(id)
}
func notificationDeliveryStateForQueueStatus(
status NotificationQueueStatus,
) operationaltrust.NotificationDeliveryState {
switch status {
case QueueStatusSending:
return operationaltrust.NotificationDelivering
case QueueStatusSent:
return operationaltrust.NotificationDelivered
case QueueStatusFailed:
return operationaltrust.NotificationFailed
case QueueStatusDLQ:
return operationaltrust.NotificationDeadLetter
case QueueStatusCancelled:
return operationaltrust.NotificationCancelled
default:
return operationaltrust.NotificationQueued
}
}
func transitionNotificationLinks(
links []operationaltrust.NotificationLink,
state operationaltrust.NotificationDeliveryState,
at time.Time,
) []operationaltrust.NotificationLink {
transitioned := make([]operationaltrust.NotificationLink, len(links))
for index := range links {
transitioned[index] = links[index].Clone()
transitioned[index].DeliveryState = state
switch state {
case operationaltrust.NotificationQueued:
transitioned[index].AttemptedAt = nil
transitioned[index].CompletedAt = nil
case operationaltrust.NotificationDelivering:
attemptedAt := at
transitioned[index].AttemptedAt = &attemptedAt
transitioned[index].CompletedAt = nil
case operationaltrust.NotificationRetrying:
transitioned[index].CompletedAt = nil
case operationaltrust.NotificationDelivered,
operationaltrust.NotificationFailed,
operationaltrust.NotificationDeadLetter,
operationaltrust.NotificationCancelled:
if transitioned[index].AttemptedAt != nil {
completedAt := at
transitioned[index].CompletedAt = &completedAt
}
}
}
return transitioned
}
// IncrementAttempt increments the attempt counter for a notification
func (nq *NotificationQueue) IncrementAttempt(id string) error {
nq.mu.Lock()
@@ -426,14 +690,29 @@ func (nq *NotificationQueue) IncrementAttemptAndSetStatus(id string, status Noti
nq.mu.Lock()
defer nq.mu.Unlock()
now := time.Now()
links, err := nq.readNotificationLinksNoLock(id)
if err != nil {
return err
}
links = transitionNotificationLinks(
links,
notificationDeliveryStateForQueueStatus(status),
now,
)
linksJSON, err := json.Marshal(links)
if err != nil {
return fmt.Errorf("marshal notification links for attempt: %w", err)
}
query := `
UPDATE notification_queue
SET attempts = attempts + 1,
status = ?,
last_attempt = ?
last_attempt = ?,
operational_links = ?
WHERE id = ?
`
_, err := nq.db.Exec(query, status, time.Now().Unix(), id)
_, err = nq.db.Exec(query, status, now.Unix(), string(linksJSON), id)
if err != nil {
return fmt.Errorf("failed to increment attempt and set status: %w", err)
}
@@ -447,7 +726,8 @@ func (nq *NotificationQueue) GetPending(limit int) ([]*QueuedNotification, error
query := `
SELECT id, type, method, status, alerts, config, attempts, max_attempts,
last_attempt, last_error, created_at, next_retry_at, completed_at, payload_bytes
last_attempt, last_error, created_at, next_retry_at, completed_at, payload_bytes,
operational_links
FROM notification_queue
WHERE status = 'pending'
AND (next_retry_at IS NULL OR next_retry_at <= ?)
@@ -484,7 +764,7 @@ func (nq *NotificationQueue) GetPending(limit int) ([]*QueuedNotification, error
// scanNotification scans a database row into a QueuedNotification
func (nq *NotificationQueue) scanNotification(rows *sql.Rows) (*QueuedNotification, error) {
var notif QueuedNotification
var alertsJSON, configJSON string
var alertsJSON, configJSON, linksJSON string
var lastAttempt, nextRetryAt, completedAt *int64
var createdAtUnix int64
@@ -503,6 +783,7 @@ func (nq *NotificationQueue) scanNotification(rows *sql.Rows) (*QueuedNotificati
&nextRetryAt,
&completedAt,
&notif.PayloadBytes,
&linksJSON,
)
if err != nil {
return nil, err
@@ -528,6 +809,12 @@ func (nq *NotificationQueue) scanNotification(rows *sql.Rows) (*QueuedNotificati
}
notif.Config = json.RawMessage(configJSON)
if err := json.Unmarshal([]byte(linksJSON), &notif.Links); err != nil {
return nil, fmt.Errorf("failed to unmarshal operational links: %w", err)
}
if len(notif.Links) > 0 {
notif.DestinationID = notif.Links[0].DestinationID
}
return &notif, nil
}
@@ -540,13 +827,33 @@ func (nq *NotificationQueue) ScheduleRetry(id string, attempt int) error {
nq.mu.Lock()
defer nq.mu.Unlock()
links, err := nq.readNotificationLinksNoLock(id)
if err != nil {
return err
}
links = transitionNotificationLinks(
links,
operationaltrust.NotificationRetrying,
time.Now(),
)
linksJSON, err := json.Marshal(links)
if err != nil {
return fmt.Errorf("marshal notification links for retry: %w", err)
}
query := `
UPDATE notification_queue
SET status = 'pending', next_retry_at = ?, last_attempt = ?
SET status = 'pending', next_retry_at = ?, last_attempt = ?,
operational_links = ?, completed_at = NULL, last_error = NULL
WHERE id = ?
`
_, err := nq.db.Exec(query, nextRetry.Unix(), time.Now().Unix(), id)
_, err = nq.db.Exec(
query,
nextRetry.Unix(),
time.Now().Unix(),
string(linksJSON),
id,
)
if err != nil {
return fmt.Errorf("failed to schedule retry: %w", err)
}
@@ -573,7 +880,8 @@ func (nq *NotificationQueue) GetDLQ(limit int) ([]*QueuedNotification, error) {
query := `
SELECT id, type, method, status, alerts, config, attempts, max_attempts,
last_attempt, last_error, created_at, next_retry_at, completed_at, payload_bytes
last_attempt, last_error, created_at, next_retry_at, completed_at, payload_bytes,
operational_links
FROM notification_queue
WHERE status = 'dlq'
ORDER BY completed_at DESC
@@ -616,20 +924,25 @@ func (nq *NotificationQueue) RecordAudit(notif *QueuedNotification, success bool
alertIdentifiers[i] = alert.ID
}
alertIdentifiersJSON, _ := json.Marshal(alertIdentifiers)
linksJSON, err := json.Marshal(notif.Links)
if err != nil {
return fmt.Errorf("failed to marshal notification links for audit: %w", err)
}
query := `
INSERT INTO notification_audit
(notification_id, type, method, status, alert_identifiers, alert_count, attempts, success, error_message, payload_size, timestamp)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
(notification_id, type, method, status, alert_identifiers, alert_count, operational_links, attempts, success, error_message, payload_size, timestamp)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`
_, err := nq.db.Exec(query,
_, err = nq.db.Exec(query,
notif.ID,
notif.Type,
notif.Method,
notif.Status,
string(alertIdentifiersJSON),
len(notif.Alerts),
string(linksJSON),
notif.Attempts,
success,
errorMsg,
@@ -781,6 +1094,15 @@ func (nq *NotificationQueue) processNotification(notif *QueuedNotification) {
Msg("Failed to increment attempt and set status")
return
}
attemptedAt := time.Now()
notif.Attempts++
notif.Status = QueueStatusSending
notif.LastAttempt = &attemptedAt
notif.Links = transitionNotificationLinks(
notif.Links,
operationaltrust.NotificationDelivering,
attemptedAt,
)
// Call processor if set
nq.mu.RLock()
@@ -792,25 +1114,12 @@ func (nq *NotificationQueue) processNotification(notif *QueuedNotification) {
err := processor(notif)
// Record audit
success := err == nil
errorMsg := ""
if err != nil {
errorMsg = err.Error()
}
if auditErr := nq.RecordAudit(notif, success, errorMsg); auditErr != nil {
log.Error().
Err(auditErr).
Str("component", "notification_queue").
Str("action", "record_audit").
Str("id", notif.ID).
Str("type", notif.Type).
Int("attempt", notif.Attempts+1).
Int("maxAttempts", notif.MaxAttempts).
Msg("Failed to record audit")
}
if success {
// Mark as sent
if err := nq.UpdateStatus(notif.ID, QueueStatusSent, ""); err != nil {
@@ -820,20 +1129,29 @@ func (nq *NotificationQueue) processNotification(notif *QueuedNotification) {
Str("action", "mark_sent").
Str("id", notif.ID).
Str("type", notif.Type).
Int("attempt", notif.Attempts+1).
Int("attempt", notif.Attempts).
Msg("Failed to update notification status to sent")
} else {
completedAt := time.Now()
notif.Status = QueueStatusSent
notif.CompletedAt = &completedAt
notif.Links = transitionNotificationLinks(
notif.Links,
operationaltrust.NotificationDelivered,
completedAt,
)
}
log.Info().
Str("component", "notification_queue").
Str("action", "send_success").
Str("id", notif.ID).
Str("type", notif.Type).
Int("attempt", notif.Attempts+1).
Int("attempt", notif.Attempts).
Int("maxAttempts", notif.MaxAttempts).
Msg("Notification sent successfully")
} else {
// Check if we should retry or move to DLQ
if notif.Attempts+1 >= notif.MaxAttempts {
if notif.Attempts >= notif.MaxAttempts {
// Move to DLQ
if dlqErr := nq.MoveToDLQ(notif.ID, errorMsg); dlqErr != nil {
log.Error().
@@ -842,45 +1160,81 @@ func (nq *NotificationQueue) processNotification(notif *QueuedNotification) {
Str("action", "move_to_dlq").
Str("id", notif.ID).
Str("type", notif.Type).
Int("attempt", notif.Attempts+1).
Int("attempt", notif.Attempts).
Int("maxAttempts", notif.MaxAttempts).
Msg("Failed to move notification to DLQ")
} else {
completedAt := time.Now()
notif.Status = QueueStatusDLQ
notif.CompletedAt = &completedAt
notif.Links = transitionNotificationLinks(
notif.Links,
operationaltrust.NotificationDeadLetter,
completedAt,
)
log.Warn().
Str("component", "notification_queue").
Str("action", "move_to_dlq").
Str("id", notif.ID).
Str("type", notif.Type).
Int("attempts", notif.Attempts+1).
Int("attempts", notif.Attempts).
Int("maxAttempts", notif.MaxAttempts).
Str("error", errorMsg).
Msg("notification moved to DLQ after max retries")
}
} else {
// Schedule retry
if retryErr := nq.ScheduleRetry(notif.ID, notif.Attempts+1); retryErr != nil {
if retryErr := nq.ScheduleRetry(notif.ID, notif.Attempts); retryErr != nil {
log.Error().
Err(retryErr).
Str("component", "notification_queue").
Str("action", "schedule_retry").
Str("id", notif.ID).
Str("type", notif.Type).
Int("attempt", notif.Attempts+1).
Int("attempt", notif.Attempts).
Int("maxAttempts", notif.MaxAttempts).
Msg("Failed to schedule retry")
} else {
notif.Status = QueueStatusPending
notif.Links = transitionNotificationLinks(
notif.Links,
operationaltrust.NotificationRetrying,
time.Now(),
)
log.Warn().
Str("component", "notification_queue").
Str("action", "schedule_retry").
Str("id", notif.ID).
Str("type", notif.Type).
Int("attempt", notif.Attempts+1).
Int("attempt", notif.Attempts).
Int("maxAttempts", notif.MaxAttempts).
Str("error", errorMsg).
Msg("notification failed, scheduled for retry")
}
}
}
if persistedLinks, linksErr := nq.getNotificationLinks(notif.ID); linksErr != nil {
log.Error().
Err(linksErr).
Str("component", "notification_queue").
Str("action", "read_links_for_audit").
Str("id", notif.ID).
Msg("Failed to read persisted notification links for audit")
} else {
notif.Links = persistedLinks
}
if auditErr := nq.RecordAudit(notif, success, errorMsg); auditErr != nil {
log.Error().
Err(auditErr).
Str("component", "notification_queue").
Str("action", "record_audit").
Str("id", notif.ID).
Str("type", notif.Type).
Int("attempt", notif.Attempts).
Int("maxAttempts", notif.MaxAttempts).
Msg("Failed to record audit")
}
}
// cleanupOldEntries removes old completed notifications
@@ -1047,7 +1401,7 @@ func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string)
defer nq.mu.Unlock()
query := `
SELECT id, type, status, alerts
SELECT id, type, status, alerts, operational_links
FROM notification_queue
WHERE status IN ('pending', 'sending')
`
@@ -1074,6 +1428,7 @@ func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string)
type queuedAlertCancellation struct {
notificationID string
remaining []*alerts.Alert
remainingLinks []operationaltrust.NotificationLink
}
var toCancelIDs []string
@@ -1086,7 +1441,14 @@ func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string)
var notifType string
var notifStatus string
var alertsJSON []byte
if err := rows.Scan(&notifID, &notifType, &notifStatus, &alertsJSON); err != nil {
var linksJSON []byte
if err := rows.Scan(
&notifID,
&notifType,
&notifStatus,
&alertsJSON,
&linksJSON,
); err != nil {
log.Error().
Err(err).
Str("component", "notification_queue").
@@ -1108,8 +1470,19 @@ func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string)
Msg("Failed to unmarshal alerts for cancellation check")
continue
}
var links []operationaltrust.NotificationLink
if err := json.Unmarshal(linksJSON, &links); err != nil {
log.Error().
Err(err).
Str("component", "notification_queue").
Str("action", "cancel_unmarshal_links").
Str("notifID", notifID).
Msg("Failed to unmarshal operational links for cancellation check")
continue
}
remainingAlerts := make([]*alerts.Alert, 0, len(queuedAlerts))
removedLinkKeys := make(map[string]struct{})
matchedAlertCount := 0
for _, alert := range queuedAlerts {
if alert == nil {
@@ -1118,6 +1491,9 @@ func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string)
}
if _, exists := alertIdentifierSet[alert.ID]; exists {
matchedAlertCount++
if alert.OperationalRecord != nil && alert.LatestTransition != nil {
removedLinkKeys[alert.OperationalRecord.ID+"\x00"+alert.LatestTransition.ID] = struct{}{}
}
continue
}
remainingAlerts = append(remainingAlerts, alert)
@@ -1134,9 +1510,18 @@ func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string)
toCancelIDs = append(toCancelIDs, notifID)
continue
}
remainingLinks := make([]operationaltrust.NotificationLink, 0, len(links))
for _, link := range links {
key := link.OperationalRecordID + "\x00" + link.TransitionID
if _, removed := removedLinkKeys[key]; removed {
continue
}
remainingLinks = append(remainingLinks, link)
}
toRewrite = append(toRewrite, queuedAlertCancellation{
notificationID: notifID,
remaining: remainingAlerts,
remainingLinks: remainingLinks,
})
}
@@ -1149,7 +1534,7 @@ func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string)
if len(toRewrite) > 0 {
updateAlertsQuery := `
UPDATE notification_queue
SET alerts = ?
SET alerts = ?, operational_links = ?
WHERE id = ?
`
for _, rewrite := range toRewrite {
@@ -1157,21 +1542,29 @@ func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string)
if err != nil {
return 0, fmt.Errorf("failed to marshal remaining alerts for %s: %w", rewrite.notificationID, err)
}
if _, err := nq.db.Exec(updateAlertsQuery, string(alertsJSON), rewrite.notificationID); err != nil {
linksJSON, err := json.Marshal(rewrite.remainingLinks)
if err != nil {
return 0, fmt.Errorf("failed to marshal remaining links for %s: %w", rewrite.notificationID, err)
}
if _, err := nq.db.Exec(
updateAlertsQuery,
string(alertsJSON),
string(linksJSON),
rewrite.notificationID,
); err != nil {
return 0, fmt.Errorf("failed to rewrite queued notification %s after alert resolution: %w", rewrite.notificationID, err)
}
}
}
if len(toCancelIDs) > 0 {
now := time.Now().Unix()
updateQuery := `
UPDATE notification_queue
SET status = ?, last_attempt = ?, last_error = ?, completed_at = ?, next_retry_at = NULL
WHERE id = ?
`
for _, notifID := range toCancelIDs {
if _, err := nq.db.Exec(updateQuery, QueueStatusCancelled, now, "Alert resolved", now, notifID); err != nil {
if err := nq.updateNotificationStatusNoLock(
notifID,
QueueStatusCancelled,
"Alert resolved",
time.Now(),
); err != nil {
log.Error().
Err(err).
Str("component", "notification_queue").
@@ -1222,28 +1615,47 @@ func (nq *NotificationQueue) CancelByTypes(types []string, reason string) error
}
query := fmt.Sprintf(`
UPDATE notification_queue
SET status = ?, last_attempt = ?, last_error = ?, completed_at = ?, next_retry_at = NULL
SELECT id
FROM notification_queue
WHERE status IN ('pending', 'sending')
AND type IN (%s)
`, strings.Join(placeholders, ","))
now := time.Now().Unix()
updateArgs := make([]any, 0, 4+len(args))
updateArgs = append(updateArgs, QueueStatusCancelled, now, reason, now)
updateArgs = append(updateArgs, args...)
result, err := nq.db.Exec(query, updateArgs...)
rows, err := nq.db.Query(query, args...)
if err != nil {
return fmt.Errorf("failed to cancel notifications by type: %w", err)
return fmt.Errorf("failed to query notifications by type: %w", err)
}
var notificationIDs []string
for rows.Next() {
var id string
if err := rows.Scan(&id); err != nil {
_ = rows.Close()
return fmt.Errorf("scan notification by type: %w", err)
}
notificationIDs = append(notificationIDs, id)
}
rowsErr := rows.Err()
_ = rows.Close()
if rowsErr != nil {
return fmt.Errorf("iterate notifications by type: %w", rowsErr)
}
rows, _ := result.RowsAffected()
if rows > 0 {
for _, id := range notificationIDs {
if err := nq.updateNotificationStatusNoLock(
id,
QueueStatusCancelled,
reason,
time.Now(),
); err != nil {
return fmt.Errorf("cancel notification %s by type: %w", id, err)
}
}
if len(notificationIDs) > 0 {
log.Info().
Str("component", "notification_queue").
Str("action", "cancel_types").
Int64("count", rows).
Int("count", len(notificationIDs)).
Strs("types", types).
Str("reason", reason).
Msg("cancelled queued notifications by type")
internal/notifications/queue_queryplan_test.go
+4 -2
View File
@@ -31,7 +31,8 @@ func TestNotificationQueueQueryPlansUseIndexes(t *testing.T) {
name: "get pending uses status index",
query: `
SELECT id, type, method, status, alerts, config, attempts, max_attempts,
last_attempt, last_error, created_at, next_retry_at, completed_at, payload_bytes
last_attempt, last_error, created_at, next_retry_at, completed_at, payload_bytes,
operational_links
FROM notification_queue
WHERE status = 'pending'
AND (next_retry_at IS NULL OR next_retry_at <= ?)
@@ -48,7 +49,8 @@ func TestNotificationQueueQueryPlansUseIndexes(t *testing.T) {
name: "get dlq uses status completed index",
query: `
SELECT id, type, method, status, alerts, config, attempts, max_attempts,
last_attempt, last_error, created_at, next_retry_at, completed_at, payload_bytes
last_attempt, last_error, created_at, next_retry_at, completed_at, payload_bytes,
operational_links
FROM notification_queue
WHERE status = 'dlq'
ORDER BY completed_at DESC
internal/notifications/queue_test.go
+189 -11
View File
@@ -12,6 +12,7 @@ import (
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
"github.com/rcourtman/pulse-go-rewrite/internal/utils"
)
@@ -518,6 +519,18 @@ func TestNewNotificationQueue_MigratesAuditAlertIdentifiersColumn(t *testing.T)
columns,
)
}
queueColumns, err := nq.tableColumns("notification_queue")
if err != nil {
t.Fatalf("tableColumns(notification_queue) failed: %v", err)
}
if !queueColumns[notificationOperationalLinksColumn] ||
!columns[notificationOperationalLinksColumn] {
t.Fatalf(
"expected migrated operational link columns, queue=%#v audit=%#v",
queueColumns,
columns,
)
}
}
func TestCancelByAlertIdentifiers_MatchingNotificationCancelled(t *testing.T) {
@@ -644,16 +657,35 @@ func TestCancelByAlertIdentifiers_MultipleAlertsPartialMatch(t *testing.T) {
// Enqueue a notification with multiple alerts (far future NextRetryAt so background processor doesn't pick it up)
futureRetry := time.Now().Add(1 * time.Hour)
notif := &QueuedNotification{
ID: "notif-multi",
Type: "webhook",
Status: QueueStatusPending,
MaxAttempts: 3,
Config: []byte(`{"url":"https://hooks.example.test"}`),
NextRetryAt: &futureRetry,
ID: "notif-multi",
Type: "webhook",
DestinationID: "webhook:primary",
Status: QueueStatusPending,
MaxAttempts: 3,
Config: []byte(`{"url":"https://hooks.example.test"}`),
NextRetryAt: &futureRetry,
Alerts: []*alerts.Alert{
{ID: "alert-1"},
{ID: "alert-2"},
{ID: "alert-3"},
{
ID: "alert-1",
OperationalRecord: &operationaltrust.OperationalRecord{ID: "record-1"},
LatestTransition: &operationaltrust.LifecycleTransition{
ID: "transition-1", To: operationaltrust.OperationalOpen, CauseKey: "cause-1",
},
},
{
ID: "alert-2",
OperationalRecord: &operationaltrust.OperationalRecord{ID: "record-2"},
LatestTransition: &operationaltrust.LifecycleTransition{
ID: "transition-2", To: operationaltrust.OperationalOpen, CauseKey: "cause-2",
},
},
{
ID: "alert-3",
OperationalRecord: &operationaltrust.OperationalRecord{ID: "record-3"},
LatestTransition: &operationaltrust.LifecycleTransition{
ID: "transition-3", To: operationaltrust.OperationalOpen, CauseKey: "cause-3",
},
},
},
}
if err := nq.Enqueue(notif); err != nil {
@@ -681,12 +713,13 @@ func TestCancelByAlertIdentifiers_MultipleAlertsPartialMatch(t *testing.T) {
var notifType string
var alertsJSON string
var linksJSON string
var configJSON string
var nextRetryAt sql.NullInt64
if err := nq.db.QueryRow(
`SELECT type, alerts, config, next_retry_at FROM notification_queue WHERE id = ?`,
`SELECT type, alerts, operational_links, config, next_retry_at FROM notification_queue WHERE id = ?`,
notif.ID,
).Scan(&notifType, &alertsJSON, &configJSON, &nextRetryAt); err != nil {
).Scan(&notifType, &alertsJSON, &linksJSON, &configJSON, &nextRetryAt); err != nil {
t.Fatalf("Failed to query rewritten notification row: %v", err)
}
if notifType != notif.Type {
@@ -706,6 +739,15 @@ func TestCancelByAlertIdentifiers_MultipleAlertsPartialMatch(t *testing.T) {
if len(remaining) != 1 || remaining[0].ID != "alert-2" {
t.Fatalf("remaining alerts = %#v, want only alert-2", remaining)
}
var remainingLinks []operationaltrust.NotificationLink
if err := json.Unmarshal([]byte(linksJSON), &remainingLinks); err != nil {
t.Fatalf("failed to unmarshal rewritten operational links: %v", err)
}
if len(remainingLinks) != 1 ||
remainingLinks[0].OperationalRecordID != "record-2" ||
remainingLinks[0].TransitionID != "transition-2" {
t.Fatalf("remaining operational links = %#v, want only alert-2 link", remainingLinks)
}
}
func TestCancelByAlertIdentifiers_SendingRowsCancelledButNotCounted(t *testing.T) {
@@ -1577,3 +1619,139 @@ func TestSetProcessorTriggersPendingDelivery(t *testing.T) {
time.Sleep(10 * time.Millisecond)
}
}
func TestNotificationQueuePersistsGroupedOperationalLinksAcrossDeliveryStates(t *testing.T) {
nq, err := NewNotificationQueue(t.TempDir())
if err != nil {
t.Fatalf("NewNotificationQueue() error = %v", err)
}
defer func() { _ = nq.Stop() }()
makeAlert := func(id, recordID, transitionID string) *alerts.Alert {
return &alerts.Alert{
ID: id,
OperationalRecord: &operationaltrust.OperationalRecord{
ID: recordID,
},
LatestTransition: &operationaltrust.LifecycleTransition{
ID: transitionID,
OperationalRecordID: recordID,
To: operationaltrust.OperationalOpen,
CauseKey: id + "-cause",
},
}
}
createdAt := time.Date(2026, 7, 18, 20, 0, 0, 123, time.UTC)
notif := &QueuedNotification{
Type: "email",
DestinationID: "destination-primary-email",
Alerts: []*alerts.Alert{
makeAlert("alert-1", "record-1", "transition-1"),
makeAlert("alert-2", "record-2", "transition-2"),
},
Config: json.RawMessage(`{}`),
CreatedAt: createdAt,
MaxAttempts: 3,
}
if err := nq.Enqueue(notif); err != nil {
t.Fatalf("Enqueue() error = %v", err)
}
if notif.ID == "" {
t.Fatal("notification id is empty")
}
if len(notif.Links) != 2 {
t.Fatalf("queued link count = %d, want 2", len(notif.Links))
}
for _, link := range notif.Links {
if link.NotificationID != notif.ID ||
link.DestinationID != notif.DestinationID ||
link.DeliveryState != operationaltrust.NotificationQueued {
t.Fatalf("queued link = %+v", link)
}
if err := link.Validate(); err != nil {
t.Fatalf("queued link Validate() error = %v", err)
}
}
if err := nq.IncrementAttemptAndSetStatus(
notif.ID,
QueueStatusSending,
); err != nil {
t.Fatalf("IncrementAttemptAndSetStatus() error = %v", err)
}
delivering, err := nq.getNotificationLinks(notif.ID)
if err != nil {
t.Fatalf("get delivering links error = %v", err)
}
for _, link := range delivering {
if link.DeliveryState != operationaltrust.NotificationDelivering ||
link.AttemptedAt == nil ||
link.CompletedAt != nil {
t.Fatalf("delivering link = %+v", link)
}
}
if err := nq.ScheduleRetry(notif.ID, 1); err != nil {
t.Fatalf("ScheduleRetry() error = %v", err)
}
retrying, err := nq.getNotificationLinks(notif.ID)
if err != nil {
t.Fatalf("get retrying links error = %v", err)
}
for index, link := range retrying {
if link.NotificationID != notif.ID ||
link.TransitionID != delivering[index].TransitionID ||
link.DeliveryState != operationaltrust.NotificationRetrying ||
link.AttemptedAt == nil ||
link.CompletedAt != nil {
t.Fatalf("retrying link = %+v", link)
}
}
if err := nq.UpdateStatus(notif.ID, QueueStatusSent, ""); err != nil {
t.Fatalf("UpdateStatus(sent) error = %v", err)
}
delivered, err := nq.getNotificationLinks(notif.ID)
if err != nil {
t.Fatalf("get delivered links error = %v", err)
}
for _, link := range delivered {
if link.NotificationID != notif.ID ||
link.DeliveryState != operationaltrust.NotificationDelivered ||
link.AttemptedAt == nil ||
link.CompletedAt == nil {
t.Fatalf("delivered link = %+v", link)
}
if err := link.Validate(); err != nil {
t.Fatalf("delivered link Validate() error = %v", err)
}
}
notif.Status = QueueStatusSent
notif.Attempts = 1
notif.Links = delivered
if err := nq.RecordAudit(notif, true, ""); err != nil {
t.Fatalf("RecordAudit() error = %v", err)
}
var auditLinksJSON string
if err := nq.db.QueryRow(
`SELECT operational_links FROM notification_audit WHERE notification_id = ?`,
notif.ID,
).Scan(&auditLinksJSON); err != nil {
t.Fatalf("query audit links: %v", err)
}
var auditLinks []operationaltrust.NotificationLink
if err := json.Unmarshal([]byte(auditLinksJSON), &auditLinks); err != nil {
t.Fatalf("unmarshal audit links: %v", err)
}
if len(auditLinks) != 2 {
t.Fatalf("audit link count = %d, want 2", len(auditLinks))
}
for index := range auditLinks {
if auditLinks[index].NotificationID != delivered[index].NotificationID ||
auditLinks[index].TransitionID != delivered[index].TransitionID ||
auditLinks[index].DeliveryState != operationaltrust.NotificationDelivered {
t.Fatalf("audit link[%d] = %+v, want %+v", index, auditLinks[index], delivered[index])
}
}
}
internal/operationaltrust/contracts.go
+791
View File
@@ -0,0 +1,791 @@
// Package operationaltrust defines the shared evidence, lifecycle, and
// notification-link contracts used by Pulse's operational read models.
package operationaltrust
import (
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"sort"
"strconv"
"strings"
"time"
)
type EvidenceCompleteness string
const (
EvidenceComplete EvidenceCompleteness = "complete"
EvidencePartial EvidenceCompleteness = "partial"
EvidenceUnavailable EvidenceCompleteness = "unavailable"
)
func (value EvidenceCompleteness) valid() bool {
switch value {
case EvidenceComplete, EvidencePartial, EvidenceUnavailable:
return true
default:
return false
}
}
type EvidenceConfidence string
const (
EvidenceConfirmed EvidenceConfidence = "confirmed"
EvidenceInferred EvidenceConfidence = "inferred"
EvidenceUnknown EvidenceConfidence = "unknown"
)
func (value EvidenceConfidence) valid() bool {
switch value {
case EvidenceConfirmed, EvidenceInferred, EvidenceUnknown:
return true
default:
return false
}
}
type EvidencePermissions string
const (
EvidencePermissionsSufficient EvidencePermissions = "sufficient"
EvidencePermissionsPartial EvidencePermissions = "partial"
EvidencePermissionsDenied EvidencePermissions = "denied"
EvidencePermissionsUnknown EvidencePermissions = "unknown"
)
func (value EvidencePermissions) valid() bool {
switch value {
case EvidencePermissionsSufficient,
EvidencePermissionsPartial,
EvidencePermissionsDenied,
EvidencePermissionsUnknown:
return true
default:
return false
}
}
type EvidenceFreshness string
const (
EvidenceFresh EvidenceFreshness = "fresh"
EvidenceStale EvidenceFreshness = "stale"
EvidenceFreshnessUnknown EvidenceFreshness = "unknown"
)
type EvidenceSource struct {
Provider string `json:"provider"`
Collector string `json:"collector"`
Instance string `json:"instance,omitempty"`
}
func (source EvidenceSource) Validate() error {
if strings.TrimSpace(source.Provider) == "" {
return errors.New("evidence source provider is required")
}
if strings.TrimSpace(source.Collector) == "" {
return errors.New("evidence source collector is required")
}
return nil
}
type EvidenceSubject struct {
ResourceID string `json:"resourceId,omitempty"`
ProviderRef string `json:"providerRef,omitempty"`
ProviderScope string `json:"providerScope,omitempty"`
}
func (subject EvidenceSubject) Validate() error {
hasResource := strings.TrimSpace(subject.ResourceID) != ""
hasProviderRef := strings.TrimSpace(subject.ProviderRef) != ""
if hasResource == hasProviderRef {
return errors.New("evidence subject requires exactly one resource id or provider reference")
}
if hasProviderRef && strings.TrimSpace(subject.ProviderScope) == "" {
return errors.New("unresolved provider evidence requires provider scope")
}
return nil
}
type EvidenceReason struct {
Code string `json:"code"`
Message string `json:"message,omitempty"`
}
func (reason EvidenceReason) Validate() error {
if strings.TrimSpace(reason.Code) == "" {
return errors.New("evidence reason code is required")
}
return nil
}
type EvidencePayloadRef struct {
Kind string `json:"kind"`
ID string `json:"id"`
}
func (ref EvidencePayloadRef) Validate() error {
if strings.TrimSpace(ref.Kind) == "" {
return errors.New("evidence payload reference kind is required")
}
if strings.TrimSpace(ref.ID) == "" {
return errors.New("evidence payload reference id is required")
}
return nil
}
type IdentityCorrelation struct {
Rule string `json:"rule"`
MatchedFields map[string]string `json:"matchedFields"`
CandidateCount int `json:"candidateCount"`
}
func (correlation IdentityCorrelation) Validate() error {
if strings.TrimSpace(correlation.Rule) == "" {
return errors.New("identity correlation rule is required")
}
if len(correlation.MatchedFields) == 0 {
return errors.New("identity correlation matched fields are required")
}
if correlation.CandidateCount != 1 {
return fmt.Errorf(
"identity correlation requires exactly one candidate, got %d",
correlation.CandidateCount,
)
}
for key, value := range correlation.MatchedFields {
if strings.TrimSpace(key) == "" || strings.TrimSpace(value) == "" {
return errors.New("identity correlation matched fields must be non-empty")
}
}
return nil
}
type EvidenceEnvelope struct {
ID string `json:"id"`
Source EvidenceSource `json:"source"`
Subject EvidenceSubject `json:"subject"`
ObservedAt time.Time `json:"observedAt"`
IngestedAt time.Time `json:"ingestedAt"`
ValidUntil *time.Time `json:"validUntil,omitempty"`
Completeness EvidenceCompleteness `json:"completeness"`
Confidence EvidenceConfidence `json:"confidence"`
Reason *EvidenceReason `json:"reason,omitempty"`
Permissions EvidencePermissions `json:"permissions"`
PayloadRef *EvidencePayloadRef `json:"payloadRef,omitempty"`
Correlation *IdentityCorrelation `json:"correlation,omitempty"`
}
func (envelope EvidenceEnvelope) Clone() EvidenceEnvelope {
clone := envelope
if envelope.ValidUntil != nil {
value := *envelope.ValidUntil
clone.ValidUntil = &value
}
if envelope.Reason != nil {
value := *envelope.Reason
clone.Reason = &value
}
if envelope.PayloadRef != nil {
value := *envelope.PayloadRef
clone.PayloadRef = &value
}
if envelope.Correlation != nil {
value := *envelope.Correlation
value.MatchedFields = cloneStringMap(envelope.Correlation.MatchedFields)
clone.Correlation = &value
}
return clone
}
func NewEvidenceID(
source EvidenceSource,
subject EvidenceSubject,
observedAt time.Time,
sourceObservationID string,
) (string, error) {
if err := source.Validate(); err != nil {
return "", err
}
if err := subject.Validate(); err != nil {
return "", err
}
if observedAt.IsZero() {
return "", errors.New("evidence observed time is required")
}
if strings.TrimSpace(sourceObservationID) == "" {
return "", errors.New("source observation id is required")
}
return stableID(
"evidence",
source.Provider,
source.Collector,
source.Instance,
subject.ResourceID,
subject.ProviderScope,
subject.ProviderRef,
observedAt.UTC().Format(time.RFC3339Nano),
sourceObservationID,
), nil
}
func (envelope EvidenceEnvelope) Validate() error {
if strings.TrimSpace(envelope.ID) == "" {
return errors.New("evidence id is required")
}
if err := envelope.Source.Validate(); err != nil {
return err
}
if err := envelope.Subject.Validate(); err != nil {
return err
}
if envelope.ObservedAt.IsZero() {
return errors.New("evidence observed time is required")
}
if envelope.IngestedAt.IsZero() {
return errors.New("evidence ingested time is required")
}
if envelope.ValidUntil != nil && envelope.ValidUntil.Before(envelope.ObservedAt) {
return errors.New("evidence validity cannot end before observation")
}
if !envelope.Completeness.valid() {
return fmt.Errorf("evidence completeness %q is invalid", envelope.Completeness)
}
if !envelope.Confidence.valid() {
return fmt.Errorf("evidence confidence %q is invalid", envelope.Confidence)
}
if !envelope.Permissions.valid() {
return fmt.Errorf("evidence permissions %q are invalid", envelope.Permissions)
}
needsReason := envelope.Completeness != EvidenceComplete ||
envelope.Confidence != EvidenceConfirmed ||
envelope.Permissions != EvidencePermissionsSufficient
if needsReason && envelope.Reason == nil {
return errors.New("limited evidence requires a typed reason")
}
if envelope.Reason != nil {
if err := envelope.Reason.Validate(); err != nil {
return err
}
}
if envelope.PayloadRef != nil {
if err := envelope.PayloadRef.Validate(); err != nil {
return err
}
}
if envelope.Correlation != nil {
if err := envelope.Correlation.Validate(); err != nil {
return err
}
}
if envelope.Confidence == EvidenceInferred && envelope.Correlation == nil {
return errors.New("inferred evidence requires identity correlation")
}
if envelope.Subject.ResourceID == "" && envelope.Correlation != nil {
return errors.New("unresolved evidence cannot claim a resolved identity correlation")
}
return nil
}
func (envelope EvidenceEnvelope) FreshnessAt(at time.Time) EvidenceFreshness {
if at.IsZero() || envelope.ObservedAt.IsZero() || envelope.ValidUntil == nil {
return EvidenceFreshnessUnknown
}
if at.After(*envelope.ValidUntil) {
return EvidenceStale
}
return EvidenceFresh
}
type OperationalState string
const (
OperationalObserving OperationalState = "observing"
OperationalOpen OperationalState = "open"
OperationalAcknowledged OperationalState = "acknowledged"
OperationalSuppressed OperationalState = "suppressed"
OperationalResolving OperationalState = "resolving"
OperationalResolved OperationalState = "resolved"
OperationalStale OperationalState = "stale"
OperationalUnknown OperationalState = "unknown"
)
func (state OperationalState) valid() bool {
switch state {
case OperationalObserving,
OperationalOpen,
OperationalAcknowledged,
OperationalSuppressed,
OperationalResolving,
OperationalResolved,
OperationalStale,
OperationalUnknown:
return true
default:
return false
}
}
type OperationalSeverity string
const (
SeverityInfo OperationalSeverity = "info"
SeverityWarning OperationalSeverity = "warning"
SeverityCritical OperationalSeverity = "critical"
SeverityUnknown OperationalSeverity = "unknown"
)
func (severity OperationalSeverity) valid() bool {
switch severity {
case SeverityInfo, SeverityWarning, SeverityCritical, SeverityUnknown:
return true
default:
return false
}
}
type Acknowledgement struct {
At time.Time `json:"at"`
By string `json:"by"`
Note string `json:"note,omitempty"`
}
func (acknowledgement Acknowledgement) Validate() error {
if acknowledgement.At.IsZero() {
return errors.New("acknowledgement time is required")
}
if strings.TrimSpace(acknowledgement.By) == "" {
return errors.New("acknowledgement actor is required")
}
return nil
}
type Suppression struct {
At time.Time `json:"at"`
By string `json:"by"`
Reason string `json:"reason"`
ExpiresAt *time.Time `json:"expiresAt,omitempty"`
}
func (suppression Suppression) Validate() error {
if suppression.At.IsZero() {
return errors.New("suppression time is required")
}
if strings.TrimSpace(suppression.By) == "" {
return errors.New("suppression actor is required")
}
if strings.TrimSpace(suppression.Reason) == "" {
return errors.New("suppression reason is required")
}
if suppression.ExpiresAt != nil && !suppression.ExpiresAt.After(suppression.At) {
return errors.New("suppression expiry must follow suppression time")
}
return nil
}
type OperationalRecord struct {
ID string `json:"id"`
CanonicalSpecID string `json:"canonicalSpecId"`
SubjectResourceID string `json:"subjectResourceId"`
State OperationalState `json:"state"`
Severity OperationalSeverity `json:"severity"`
FirstObservedAt time.Time `json:"firstObservedAt"`
LastObservedAt time.Time `json:"lastObservedAt"`
StateChangedAt time.Time `json:"stateChangedAt"`
ResolvedAt *time.Time `json:"resolvedAt,omitempty"`
Acknowledgement *Acknowledgement `json:"acknowledgement,omitempty"`
Suppression *Suppression `json:"suppression,omitempty"`
EvidenceIDs []string `json:"evidenceIds"`
CauseKey string `json:"causeKey"`
RelatedResourceIDs []string `json:"relatedResourceIds"`
ImpactSummary string `json:"impactSummary,omitempty"`
RecommendedNextStep string `json:"recommendedNextStep,omitempty"`
}
func (record OperationalRecord) Clone() OperationalRecord {
clone := record
if record.ResolvedAt != nil {
value := *record.ResolvedAt
clone.ResolvedAt = &value
}
if record.Acknowledgement != nil {
value := *record.Acknowledgement
clone.Acknowledgement = &value
}
if record.Suppression != nil {
value := *record.Suppression
if record.Suppression.ExpiresAt != nil {
expiresAt := *record.Suppression.ExpiresAt
value.ExpiresAt = &expiresAt
}
clone.Suppression = &value
}
clone.EvidenceIDs = append([]string(nil), record.EvidenceIDs...)
clone.RelatedResourceIDs = append([]string(nil), record.RelatedResourceIDs...)
return clone
}
func (record OperationalRecord) Validate() error {
if strings.TrimSpace(record.ID) == "" {
return errors.New("operational record id is required")
}
if strings.TrimSpace(record.CanonicalSpecID) == "" {
return errors.New("operational canonical spec id is required")
}
if strings.TrimSpace(record.SubjectResourceID) == "" {
return errors.New("operational subject resource id is required")
}
if !record.State.valid() {
return fmt.Errorf("operational state %q is invalid", record.State)
}
if !record.Severity.valid() {
return fmt.Errorf("operational severity %q is invalid", record.Severity)
}
if record.FirstObservedAt.IsZero() ||
record.LastObservedAt.IsZero() ||
record.StateChangedAt.IsZero() {
return errors.New("operational observation and state-change times are required")
}
if record.LastObservedAt.Before(record.FirstObservedAt) {
return errors.New("operational last observation precedes first observation")
}
if strings.TrimSpace(record.CauseKey) == "" {
return errors.New("operational cause key is required")
}
if record.Acknowledgement != nil {
if err := record.Acknowledgement.Validate(); err != nil {
return err
}
}
if record.Suppression != nil {
if err := record.Suppression.Validate(); err != nil {
return err
}
}
if record.State == OperationalAcknowledged && record.Acknowledgement == nil {
return errors.New("acknowledged operational state requires acknowledgement")
}
if record.State == OperationalSuppressed && record.Suppression == nil {
return errors.New("suppressed operational state requires suppression")
}
if record.State == OperationalResolved {
if record.ResolvedAt == nil || record.ResolvedAt.IsZero() {
return errors.New("resolved operational state requires resolution time")
}
if len(canonicalIDs(record.EvidenceIDs)) == 0 {
return errors.New("resolved operational state requires decisive evidence")
}
} else if record.ResolvedAt != nil {
return errors.New("non-resolved operational state cannot carry resolution time")
}
return nil
}
type TransitionCause string
const (
TransitionDetectorDecision TransitionCause = "detector_decision"
TransitionAcknowledgement TransitionCause = "acknowledgement"
TransitionUnacknowledgement TransitionCause = "unacknowledgement"
TransitionSuppression TransitionCause = "suppression"
TransitionSuppressionExpired TransitionCause = "suppression_expired"
TransitionRecoveryEvidence TransitionCause = "recovery_evidence"
TransitionCollectionStale TransitionCause = "collection_stale"
TransitionCollectionUnknown TransitionCause = "collection_unknown"
)
func (cause TransitionCause) valid() bool {
switch cause {
case TransitionDetectorDecision,
TransitionAcknowledgement,
TransitionUnacknowledgement,
TransitionSuppression,
TransitionSuppressionExpired,
TransitionRecoveryEvidence,
TransitionCollectionStale,
TransitionCollectionUnknown:
return true
default:
return false
}
}
type LifecycleTransition struct {
ID string `json:"id"`
OperationalRecordID string `json:"operationalRecordId"`
From OperationalState `json:"from"`
To OperationalState `json:"to"`
At time.Time `json:"at"`
Cause TransitionCause `json:"cause"`
CauseKey string `json:"causeKey"`
EvidenceIDs []string `json:"evidenceIds"`
Reason string `json:"reason,omitempty"`
}
func (transition LifecycleTransition) Clone() LifecycleTransition {
clone := transition
clone.EvidenceIDs = append([]string(nil), transition.EvidenceIDs...)
return clone
}
func NewTransitionID(
recordID string,
from OperationalState,
to OperationalState,
at time.Time,
cause TransitionCause,
causeKey string,
evidenceIDs []string,
) (string, error) {
transition := LifecycleTransition{
OperationalRecordID: recordID,
From: from,
To: to,
At: at,
Cause: cause,
CauseKey: causeKey,
EvidenceIDs: canonicalIDs(evidenceIDs),
}
if err := transition.validateWithoutID(); err != nil {
return "", err
}
parts := []string{
recordID,
string(from),
string(to),
at.UTC().Format(time.RFC3339Nano),
string(cause),
causeKey,
}
parts = append(parts, transition.EvidenceIDs...)
return stableID("transition", parts...), nil
}
func (transition LifecycleTransition) Validate() error {
if strings.TrimSpace(transition.ID) == "" {
return errors.New("lifecycle transition id is required")
}
return transition.validateWithoutID()
}
func (transition LifecycleTransition) validateWithoutID() error {
if strings.TrimSpace(transition.OperationalRecordID) == "" {
return errors.New("lifecycle operational record id is required")
}
if !transition.From.valid() {
return fmt.Errorf("lifecycle from state %q is invalid", transition.From)
}
if !transition.To.valid() {
return fmt.Errorf("lifecycle to state %q is invalid", transition.To)
}
if transition.From == transition.To {
return errors.New("lifecycle transition must change state")
}
if transition.At.IsZero() {
return errors.New("lifecycle transition time is required")
}
if !transition.Cause.valid() {
return fmt.Errorf("lifecycle transition cause %q is invalid", transition.Cause)
}
if strings.TrimSpace(transition.CauseKey) == "" {
return errors.New("lifecycle transition cause key is required")
}
switch transition.Cause {
case TransitionAcknowledgement:
if transition.To != OperationalAcknowledged {
return errors.New("acknowledgement transition must enter acknowledged state")
}
case TransitionUnacknowledgement:
if transition.From != OperationalAcknowledged || transition.To != OperationalOpen {
return errors.New("unacknowledgement transition must return acknowledged state to open")
}
case TransitionSuppression:
if transition.To != OperationalSuppressed {
return errors.New("suppression transition must enter suppressed state")
}
case TransitionSuppressionExpired:
if transition.From != OperationalSuppressed || transition.To != OperationalOpen {
return errors.New("suppression expiry must return suppressed state to open")
}
case TransitionRecoveryEvidence:
if transition.To != OperationalResolving && transition.To != OperationalResolved {
return errors.New("recovery evidence transition must enter resolving or resolved state")
}
if len(canonicalIDs(transition.EvidenceIDs)) == 0 {
return errors.New("recovery transition requires evidence")
}
case TransitionCollectionStale:
if transition.To != OperationalStale {
return errors.New("collection-stale transition must enter stale state")
}
case TransitionCollectionUnknown:
if transition.To != OperationalUnknown {
return errors.New("collection-unknown transition must enter unknown state")
}
case TransitionDetectorDecision:
if transition.To != OperationalOpen {
return errors.New("detector decision transition must enter open state")
}
if len(canonicalIDs(transition.EvidenceIDs)) == 0 {
return errors.New("detector decision requires evidence")
}
}
return nil
}
type NotificationDeliveryState string
const (
NotificationQueued NotificationDeliveryState = "queued"
NotificationDelivering NotificationDeliveryState = "delivering"
NotificationDelivered NotificationDeliveryState = "delivered"
NotificationRetrying NotificationDeliveryState = "retrying"
NotificationCancelled NotificationDeliveryState = "cancelled"
NotificationFailed NotificationDeliveryState = "failed"
NotificationDeadLetter NotificationDeliveryState = "dead_letter"
)
func (state NotificationDeliveryState) valid() bool {
switch state {
case NotificationQueued,
NotificationDelivering,
NotificationDelivered,
NotificationRetrying,
NotificationCancelled,
NotificationFailed,
NotificationDeadLetter:
return true
default:
return false
}
}
type NotificationLink struct {
NotificationID string `json:"notificationId"`
OperationalRecordID string `json:"operationalRecordId"`
TransitionID string `json:"transitionId"`
LifecycleState OperationalState `json:"lifecycleState"`
CauseKey string `json:"causeKey"`
DestinationID string `json:"destinationId"`
DeliveryState NotificationDeliveryState `json:"deliveryState"`
AttemptedAt *time.Time `json:"attemptedAt,omitempty"`
CompletedAt *time.Time `json:"completedAt,omitempty"`
}
func (link NotificationLink) Clone() NotificationLink {
clone := link
if link.AttemptedAt != nil {
value := *link.AttemptedAt
clone.AttemptedAt = &value
}
if link.CompletedAt != nil {
value := *link.CompletedAt
clone.CompletedAt = &value
}
return clone
}
func NewNotificationID(
destinationID string,
deliveryType string,
createdAt time.Time,
transitionIDs []string,
) (string, error) {
if strings.TrimSpace(destinationID) == "" {
return "", errors.New("notification destination id is required")
}
if strings.TrimSpace(deliveryType) == "" {
return "", errors.New("notification delivery type is required")
}
if createdAt.IsZero() {
return "", errors.New("notification creation time is required")
}
ids := canonicalIDs(transitionIDs)
if len(ids) == 0 {
return "", errors.New("notification transition ids are required")
}
parts := []string{
destinationID,
deliveryType,
createdAt.UTC().Format(time.RFC3339Nano),
}
parts = append(parts, ids...)
return stableID("notification", parts...), nil
}
func (link NotificationLink) Validate() error {
if strings.TrimSpace(link.NotificationID) == "" {
return errors.New("notification id is required")
}
if strings.TrimSpace(link.OperationalRecordID) == "" {
return errors.New("notification operational record id is required")
}
if strings.TrimSpace(link.TransitionID) == "" {
return errors.New("notification transition id is required")
}
if !link.LifecycleState.valid() {
return fmt.Errorf("notification lifecycle state %q is invalid", link.LifecycleState)
}
if strings.TrimSpace(link.CauseKey) == "" {
return errors.New("notification cause key is required")
}
if strings.TrimSpace(link.DestinationID) == "" {
return errors.New("notification destination id is required")
}
if !link.DeliveryState.valid() {
return fmt.Errorf("notification delivery state %q is invalid", link.DeliveryState)
}
if link.CompletedAt != nil && link.AttemptedAt == nil {
return errors.New("completed notification requires attempted time")
}
if link.AttemptedAt != nil &&
link.CompletedAt != nil &&
link.CompletedAt.Before(*link.AttemptedAt) {
return errors.New("notification completion precedes attempt")
}
return nil
}
func canonicalIDs(values []string) []string {
seen := make(map[string]struct{}, len(values))
result := make([]string, 0, len(values))
for _, value := range values {
value = strings.TrimSpace(value)
if value == "" {
continue
}
if _, exists := seen[value]; exists {
continue
}
seen[value] = struct{}{}
result = append(result, value)
}
sort.Strings(result)
return result
}
func cloneStringMap(values map[string]string) map[string]string {
if values == nil {
return nil
}
clone := make(map[string]string, len(values))
for key, value := range values {
clone[key] = value
}
return clone
}
func stableID(prefix string, parts ...string) string {
var payload strings.Builder
for _, part := range parts {
payload.WriteString(strconv.Itoa(len(part)))
payload.WriteByte(':')
payload.WriteString(part)
payload.WriteByte('|')
}
sum := sha256.Sum256([]byte(payload.String()))
return prefix + "_" + hex.EncodeToString(sum[:16])
}
internal/operationaltrust/contracts_test.go
+308
View File
@@ -0,0 +1,308 @@
package operationaltrust
import (
"strings"
"testing"
"time"
)
func TestEvidenceEnvelopeRequiresTypedLimitations(t *testing.T) {
observedAt := time.Date(2026, 7, 18, 20, 0, 0, 0, time.UTC)
validUntil := observedAt.Add(15 * time.Minute)
source := EvidenceSource{
Provider: "proxmox",
Collector: "pulse-core",
Instance: "pve-a",
}
subject := EvidenceSubject{ResourceID: "resource-123"}
id, err := NewEvidenceID(source, subject, observedAt, "provider-event-7")
if err != nil {
t.Fatalf("NewEvidenceID() error = %v", err)
}
envelope := EvidenceEnvelope{
ID: id,
Source: source,
Subject: subject,
ObservedAt: observedAt,
IngestedAt: observedAt.Add(time.Second),
ValidUntil: &validUntil,
Completeness: EvidencePartial,
Confidence: EvidenceConfirmed,
Permissions: EvidencePermissionsPartial,
}
if err := envelope.Validate(); err == nil || !strings.Contains(err.Error(), "typed reason") {
t.Fatalf("Validate() error = %v, want typed reason error", err)
}
envelope.Reason = &EvidenceReason{Code: "permission_scope_incomplete"}
if err := envelope.Validate(); err != nil {
t.Fatalf("Validate() error = %v", err)
}
if got := envelope.FreshnessAt(validUntil); got != EvidenceFresh {
t.Fatalf("FreshnessAt(validUntil) = %q, want %q", got, EvidenceFresh)
}
if got := envelope.FreshnessAt(validUntil.Add(time.Nanosecond)); got != EvidenceStale {
t.Fatalf("FreshnessAt(expired) = %q, want %q", got, EvidenceStale)
}
}
func TestContractClonesDoNotShareMutableState(t *testing.T) {
validUntil := time.Date(2026, 7, 18, 21, 0, 0, 0, time.UTC)
envelope := EvidenceEnvelope{
ValidUntil: &validUntil,
Reason: &EvidenceReason{Code: "inferred"},
Correlation: &IdentityCorrelation{
MatchedFields: map[string]string{"hostname": "db-01"},
},
}
envelopeClone := envelope.Clone()
envelopeClone.Correlation.MatchedFields["hostname"] = "db-02"
*envelopeClone.ValidUntil = validUntil.Add(time.Hour)
if envelope.Correlation.MatchedFields["hostname"] != "db-01" {
t.Fatal("evidence clone mutated source correlation")
}
if !envelope.ValidUntil.Equal(validUntil) {
t.Fatal("evidence clone mutated source validity")
}
record := OperationalRecord{
EvidenceIDs: []string{"evidence-1"},
RelatedResourceIDs: []string{"resource-2"},
Suppression: &Suppression{
ExpiresAt: &validUntil,
},
}
recordClone := record.Clone()
recordClone.EvidenceIDs[0] = "evidence-2"
recordClone.RelatedResourceIDs[0] = "resource-3"
*recordClone.Suppression.ExpiresAt = validUntil.Add(time.Hour)
if record.EvidenceIDs[0] != "evidence-1" ||
record.RelatedResourceIDs[0] != "resource-2" ||
!record.Suppression.ExpiresAt.Equal(validUntil) {
t.Fatal("operational record clone shared mutable state")
}
}
func TestEvidenceEnvelopeDistinguishesDeniedFromUnavailable(t *testing.T) {
observedAt := time.Date(2026, 7, 18, 20, 0, 0, 0, time.UTC)
source := EvidenceSource{Provider: "pbs", Collector: "recovery-adapter"}
subject := EvidenceSubject{
ProviderRef: "vm/100",
ProviderScope: "pbs-a",
}
id, err := NewEvidenceID(source, subject, observedAt, "permission-denied")
if err != nil {
t.Fatalf("NewEvidenceID() error = %v", err)
}
envelope := EvidenceEnvelope{
ID: id,
Source: source,
Subject: subject,
ObservedAt: observedAt,
IngestedAt: observedAt.Add(time.Second),
Completeness: EvidenceUnavailable,
Confidence: EvidenceUnknown,
Permissions: EvidencePermissionsDenied,
Reason: &EvidenceReason{Code: "provider_permission_denied"},
}
if err := envelope.Validate(); err != nil {
t.Fatalf("Validate() error = %v", err)
}
if got := envelope.FreshnessAt(observedAt.Add(time.Hour)); got != EvidenceFreshnessUnknown {
t.Fatalf("FreshnessAt() = %q, want %q", got, EvidenceFreshnessUnknown)
}
}
func TestInferredEvidenceRequiresAuditableSingleMatch(t *testing.T) {
observedAt := time.Date(2026, 7, 18, 20, 0, 0, 0, time.UTC)
source := EvidenceSource{Provider: "uptime-kuma", Collector: "availability-adapter"}
subject := EvidenceSubject{ResourceID: "resource-123"}
id, err := NewEvidenceID(source, subject, observedAt, "check-42")
if err != nil {
t.Fatalf("NewEvidenceID() error = %v", err)
}
envelope := EvidenceEnvelope{
ID: id,
Source: source,
Subject: subject,
ObservedAt: observedAt,
IngestedAt: observedAt,
Completeness: EvidenceComplete,
Confidence: EvidenceInferred,
Permissions: EvidencePermissionsSufficient,
Reason: &EvidenceReason{Code: "secondary_identity_match"},
}
if err := envelope.Validate(); err == nil || !strings.Contains(err.Error(), "correlation") {
t.Fatalf("Validate() error = %v, want correlation error", err)
}
envelope.Correlation = &IdentityCorrelation{
Rule: "normalized_hostname",
MatchedFields: map[string]string{"hostname": "db-01.example.test"},
CandidateCount: 2,
}
if err := envelope.Validate(); err == nil || !strings.Contains(err.Error(), "exactly one") {
t.Fatalf("Validate() error = %v, want ambiguous correlation error", err)
}
envelope.Correlation.CandidateCount = 1
if err := envelope.Validate(); err != nil {
t.Fatalf("Validate() error = %v", err)
}
}
func TestEvidenceAndTransitionIDsAreStableUnderRetry(t *testing.T) {
observedAt := time.Date(2026, 7, 18, 20, 0, 0, 123, time.UTC)
source := EvidenceSource{Provider: "proxmox", Collector: "pulse-core", Instance: "pve-a"}
subject := EvidenceSubject{ResourceID: "resource-123"}
firstEvidenceID, err := NewEvidenceID(source, subject, observedAt, "event-7")
if err != nil {
t.Fatalf("NewEvidenceID() error = %v", err)
}
secondEvidenceID, err := NewEvidenceID(source, subject, observedAt, "event-7")
if err != nil {
t.Fatalf("NewEvidenceID() retry error = %v", err)
}
if firstEvidenceID != secondEvidenceID {
t.Fatalf("evidence retry id = %q, want %q", secondEvidenceID, firstEvidenceID)
}
firstTransitionID, err := NewTransitionID(
"record-1",
OperationalObserving,
OperationalOpen,
observedAt,
TransitionDetectorDecision,
"cpu-high",
[]string{"evidence-b", firstEvidenceID, "evidence-b"},
)
if err != nil {
t.Fatalf("NewTransitionID() error = %v", err)
}
secondTransitionID, err := NewTransitionID(
"record-1",
OperationalObserving,
OperationalOpen,
observedAt,
TransitionDetectorDecision,
"cpu-high",
[]string{firstEvidenceID, "evidence-b"},
)
if err != nil {
t.Fatalf("NewTransitionID() retry error = %v", err)
}
if firstTransitionID != secondTransitionID {
t.Fatalf("transition retry id = %q, want %q", secondTransitionID, firstTransitionID)
}
}
func TestLifecycleTransitionEnforcesOperationalSemantics(t *testing.T) {
at := time.Date(2026, 7, 18, 20, 0, 0, 0, time.UTC)
if _, err := NewTransitionID(
"record-1",
OperationalOpen,
OperationalResolved,
at,
TransitionAcknowledgement,
"cpu-high",
nil,
); err == nil || !strings.Contains(err.Error(), "acknowledged") {
t.Fatalf("acknowledgement-to-resolution error = %v", err)
}
if _, err := NewTransitionID(
"record-1",
OperationalResolving,
OperationalResolved,
at,
TransitionRecoveryEvidence,
"cpu-high",
nil,
); err == nil || !strings.Contains(err.Error(), "requires evidence") {
t.Fatalf("evidenceless resolution error = %v", err)
}
if _, err := NewTransitionID(
"record-1",
OperationalUnknown,
OperationalResolved,
at,
TransitionCollectionUnknown,
"cpu-high",
[]string{"absence-only"},
); err == nil || !strings.Contains(err.Error(), "unknown state") {
t.Fatalf("unknown-to-resolution error = %v", err)
}
}
func TestOperationalRecordRequiresResolutionEvidence(t *testing.T) {
firstObserved := time.Date(2026, 7, 18, 20, 0, 0, 0, time.UTC)
resolvedAt := firstObserved.Add(5 * time.Minute)
record := OperationalRecord{
ID: "record-1",
CanonicalSpecID: "metric-threshold:cpu",
SubjectResourceID: "resource-123",
State: OperationalResolved,
Severity: SeverityWarning,
FirstObservedAt: firstObserved,
LastObservedAt: resolvedAt,
StateChangedAt: resolvedAt,
ResolvedAt: &resolvedAt,
CauseKey: "cpu-high",
}
if err := record.Validate(); err == nil || !strings.Contains(err.Error(), "decisive evidence") {
t.Fatalf("Validate() error = %v, want decisive evidence error", err)
}
record.EvidenceIDs = []string{"evidence-recovery"}
if err := record.Validate(); err != nil {
t.Fatalf("Validate() error = %v", err)
}
}
func TestNotificationLinkRequiresTransitionAndConsistentTiming(t *testing.T) {
attemptedAt := time.Date(2026, 7, 18, 20, 0, 0, 0, time.UTC)
completedAt := attemptedAt.Add(-time.Second)
link := NotificationLink{
NotificationID: "notification-1",
OperationalRecordID: "record-1",
TransitionID: "transition-1",
LifecycleState: OperationalOpen,
CauseKey: "cpu-high",
DestinationID: "email-primary",
DeliveryState: NotificationFailed,
AttemptedAt: &attemptedAt,
CompletedAt: &completedAt,
}
if err := link.Validate(); err == nil || !strings.Contains(err.Error(), "precedes") {
t.Fatalf("Validate() error = %v, want timing error", err)
}
completedAt = attemptedAt.Add(time.Second)
if err := link.Validate(); err != nil {
t.Fatalf("Validate() error = %v", err)
}
}
func TestNewNotificationIDIsStableAcrossGroupedTransitionOrder(t *testing.T) {
createdAt := time.Date(2026, 7, 18, 20, 0, 0, 0, time.UTC)
first, err := NewNotificationID(
"destination-1",
"email",
createdAt,
[]string{"transition-b", "transition-a"},
)
if err != nil {
t.Fatalf("NewNotificationID() error = %v", err)
}
second, err := NewNotificationID(
"destination-1",
"email",
createdAt,
[]string{"transition-a", "transition-b", "transition-a"},
)
if err != nil {
t.Fatalf("NewNotificationID() retry error = %v", err)
}
if first != second {
t.Fatalf("notification ids = %q and %q, want stable id", first, second)
}
}
scripts/release_control/subsystem_lookup_test.py
+6 -6
View File
@@ -2967,12 +2967,12 @@ class SubsystemLookupTest(unittest.TestCase):
self.assertEqual(
api_match["matched_contract_references"],
[
{
"heading": "## Shared Boundaries",
"path": "internal/api/access_control_handlers.go",
"line": 1233,
"heading_line": 144,
}
{
"heading": "## Shared Boundaries",
"path": "internal/api/access_control_handlers.go",
"line": 1244,
"heading_line": 155,
}
],
)