vrr/Pulse
2
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-07-09 16:00:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

Stabilize dev runtime and complete PMG read-state migration

Browse source 
Refs architecture post-RC canonicalization follow-up.
This commit is contained in:
rcourtman 2026-05-21 15:15:23 +01:00
parent 2c51890d01
commit bc2adac8b5
36 changed files with 1197 additions and 117 deletions
Download patch file Download diff file Expand all files Collapse all files

16
frontend-modern/src/utils/__tests__/resourceStateAdapters.test.ts
View file

@ -192,6 +192,22 @@ describe('resourceStateAdapters nodeFromResource', () => {
expect(instance?.host).toBe('https://pmg-service.local:8006');
});
it('uses PMG host and guest URLs from canonical platform data when available', () => {
const instance = pmgInstanceFromResource(
createServiceResource('pmg', {
pmg: {
hostname: 'pmg-service.local',
hostUrl: 'https://pmg-service.local:8443',
guestUrl: 'https://mail.example.com',
instanceId: 'pmg-instance-1',
},
}),
);
expect(instance?.host).toBe('https://pmg-service.local:8443');
expect(instance?.guestURL).toBe('https://mail.example.com');
});
it('keeps PMG operator labels on local identity when governed summaries exist', () => {
const instance = pmgInstanceFromResource(
createServiceResource(

3
frontend-modern/src/utils/resourceStateAdapters.ts
View file

@ -1224,7 +1224,7 @@ export const pmgInstanceFromResource = (resource: Resource): PMGInstance | null
const platform = resourcePlatformData(resource);
const pmg = asRecord(platform?.pmg);
const hostName = getPreferredResourceHostname(resource) || resource.id;
const host = resource.platformId || `https://${hostName}:8006`;
const host = asString(pmg?.hostUrl) || resource.platformId || `https://${hostName}:8006`;
const lastSeen = toISOTime(undefined, resource.lastSeen);
const mailStats =
mapPMGMailStats(pmg?.mailStats) ||
@ -1240,6 +1240,7 @@ export const pmgInstanceFromResource = (resource: Resource): PMGInstance | null
name: getPreferredInfrastructureDisplayName(resource),
host,
guestURL:
asString(pmg?.guestUrl) ||
asString((resource as unknown as Record<string, unknown>).customURL) ||
asString((resource as unknown as Record<string, unknown>).customUrl),
status: resource.status || 'unknown',

46
internal/ai/correlation/correlation_additional_test.go
View file

@ -4,6 +4,7 @@ import (
"os"
"path/filepath"
"strings"
"sync"
"testing"
"time"
)
@ -155,6 +156,51 @@ func TestDetector_SaveLoadAndLargeFile(t *testing.T) {
}
}
func TestDetector_SaveToDiskSerializesConcurrentWrites(t *testing.T) {
dir := t.TempDir()
d := NewDetector(Config{
DataDir: dir,
MinOccurrences: 1,
RetentionWindow: time.Hour,
})
for i := 0; i < 25; i++ {
d.events = append(d.events, Event{
ID: intToStr(i),
ResourceID: "resource-" + intToStr(i),
EventType: EventAlert,
Timestamp: time.Now(),
})
}
start := make(chan struct{})
errs := make(chan error, 25)
var wg sync.WaitGroup
for i := 0; i < 25; i++ {
wg.Add(1)
go func() {
defer wg.Done()
<-start
errs <- d.saveToDisk()
}()
}
close(start)
wg.Wait()
close(errs)
for err := range errs {
if err != nil {
t.Fatalf("concurrent save failed: %v", err)
}
}
loaded := NewDetector(Config{DataDir: dir, MinOccurrences: 1})
if len(loaded.events) != len(d.events) {
t.Fatalf("loaded events = %d, want %d", len(loaded.events), len(d.events))
}
}
func TestDetector_LoadMissingAndInvalid(t *testing.T) {
dir := t.TempDir()
d := NewDetector(Config{DataDir: dir})

35
internal/ai/correlation/detector.go
View file

@ -59,6 +59,7 @@ type Correlation struct {
// Detector tracks events and detects correlations between resources
type Detector struct {
mu sync.RWMutex
saveMu sync.Mutex
events []Event
correlations map[string]*Correlation // key: sourceID:targetID:pattern
@ -454,6 +455,9 @@ func (d *Detector) saveToDisk() error {
return nil
}
d.saveMu.Lock()
defer d.saveMu.Unlock()
d.mu.RLock()
eventsSnapshot := make([]Event, len(d.events))
copy(eventsSnapshot, d.events)
@ -482,12 +486,37 @@ func (d *Detector) saveToDisk() error {
}
path := filepath.Join(d.dataDir, "ai_correlations.json")
tmpPath := path + ".tmp"
if err := os.WriteFile(tmpPath, jsonData, 0600); err != nil {
tmpFile, err := os.CreateTemp(d.dataDir, "ai_correlations-*.json.tmp")
if err != nil {
return err
}
tmpPath := tmpFile.Name()
cleanupTemp := true
defer func() {
if !cleanupTemp {
return
}
if err := os.Remove(tmpPath); err != nil && !os.IsNotExist(err) {
log.Warn().Err(err).Str("file", tmpPath).Msg("failed to remove temp correlation data file")
}
}()
return os.Rename(tmpPath, path)
if _, err := tmpFile.Write(jsonData); err != nil {
_ = tmpFile.Close()
return err
}
if err := tmpFile.Chmod(0600); err != nil {
_ = tmpFile.Close()
return err
}
if err := tmpFile.Close(); err != nil {
return err
}
if err := os.Rename(tmpPath, path); err != nil {
return err
}
cleanupTemp = false
return os.Chmod(path, 0600)
}
// loadFromDisk loads data

36
internal/ai/parity_test.go
View file

@ -582,6 +582,7 @@ func TestParityPBSInstanceFields(t *testing.T) {
func TestParityPMGInstanceFields(t *testing.T) {
now := time.Now().UTC()
queueUpdated := now.Add(-2 * time.Minute)
statsUpdated := now.Add(-5 * time.Minute)
pmg := models.PMGInstance{
ID: "pmg-1",
@ -623,12 +624,24 @@ func TestParityPMGInstanceFields(t *testing.T) {
},
},
MailStats: &models.PMGMailStats{
Timeframe: "hour",
CountTotal: 123.0,
SpamIn: 7.0,
VirusIn: 2.0,
UpdatedAt: now.Add(-5 * time.Minute),
Timeframe: "hour",
CountTotal: 123.0,
CountIn: 100.0,
CountOut: 23.0,
SpamIn: 7.0,
VirusIn: 2.0,
JunkIn: 5.0,
PregreetRejects: 4.0,
UpdatedAt: statsUpdated,
},
MailCount: []models.PMGMailCountPoint{{
Timestamp: statsUpdated,
Count: 123,
CountIn: 100,
CountOut: 23,
Timeframe: "hour",
Index: 1,
}},
ConnectionHealth: "connected",
LastSeen: now,
LastUpdated: now.Add(-1 * time.Minute),
@ -648,6 +661,7 @@ func TestParityPMGInstanceFields(t *testing.T) {
require.Equal(t, pmg.Name, v.Name())
require.Equal(t, unifiedresources.StatusOnline, v.Status())
require.Equal(t, "pmg.example.test", v.Hostname())
require.Equal(t, pmg.Host, v.HostURL())
require.Equal(t, pmg.Version, v.Version())
require.Equal(t, len(pmg.Nodes), v.NodeCount())
require.Equal(t, int64(2000), v.UptimeSeconds())
@ -660,12 +674,24 @@ func TestParityPMGInstanceFields(t *testing.T) {
require.Equal(t, pmg.MailStats.CountTotal, v.MailCountTotal())
require.Equal(t, pmg.MailStats.SpamIn, v.SpamIn())
require.Equal(t, pmg.MailStats.VirusIn, v.VirusIn())
require.Len(t, v.Nodes(), 2)
require.NotNil(t, v.Nodes()[1].QueueStatus)
require.Equal(t, int64(300), v.Nodes()[1].QueueStatus.OldestAge)
require.True(t, v.Nodes()[1].QueueStatus.UpdatedAt.Equal(queueUpdated.Add(1*time.Minute)))
require.NotNil(t, v.MailStats())
require.Equal(t, pmg.MailStats.CountIn, v.MailStats().CountIn)
require.Equal(t, pmg.MailStats.JunkIn, v.MailStats().JunkIn)
require.Equal(t, pmg.MailStats.PregreetRejects, v.MailStats().PregreetRejects)
require.True(t, v.MailStats().UpdatedAt.Equal(statsUpdated))
require.Len(t, v.MailCount(), 1)
require.Equal(t, pmg.MailCount[0].Count, v.MailCount()[0].Count)
require.Equal(t, pmg.ConnectionHealth, v.ConnectionHealth())
require.Equal(t, 0.0, v.CPUPercent())
require.Equal(t, 0.0, v.MemoryPercent())
require.Equal(t, 0.0, v.DiskPercent())
require.True(t, v.LastSeen().Equal(pmg.LastSeen))
require.Equal(t, pmg.GuestURL, v.GuestURL())
require.Equal(t, pmg.GuestURL, v.CustomURL())
}

66
internal/ai/service.go
View file

@ -562,9 +562,11 @@ func (s *Service) initDiscoveryServiceLocked() {
// Create the discovery service with config-driven settings
discoveryCfg := servicediscovery.DefaultConfig()
commandScanningEnabled := false
backgroundScanningEnabled := false
if s.cfg != nil {
discoveryCfg.Interval = s.cfg.GetDiscoveryInterval()
commandScanningEnabled = s.cfg.Enabled && s.cfg.IsDiscoveryEnabled() && s.provider != nil
backgroundScanningEnabled = commandScanningEnabled && !BackgroundAutomationDisabledForDev()
discoveryCfg.CommandScanning = commandScanningEnabled
}
@ -577,11 +579,15 @@ func (s *Service) initDiscoveryServiceLocked() {
s.discoveryService.SetReadState(s.readState)
// Start background discovery if enabled and interval is set
if commandScanningEnabled && s.cfg.GetDiscoveryInterval() > 0 {
if backgroundScanningEnabled && s.cfg.GetDiscoveryInterval() > 0 {
s.discoveryService.Start(context.Background())
log.Info().
Dur("interval", s.cfg.GetDiscoveryInterval()).
Msg("AI-powered deep discovery service started with automatic scanning")
} else if commandScanningEnabled && BackgroundAutomationDisabledForDev() {
log.Info().
Str("env", DevDisableBackgroundAIEnv).
Msg("AI-powered deep discovery service initialized; automatic scanning disabled for dev")
} else if commandScanningEnabled {
log.Info().Msg("AI-powered deep discovery service initialized (manual command scanning mode)")
} else {
@ -849,9 +855,10 @@ func (s *Service) updateInfraDiscoverySettings(cfg *config.AIConfig) {
}
enabled := s.IsEnabled() && cfg.IsDiscoveryEnabled()
backgroundEnabled := enabled && !BackgroundAutomationDisabledForDev()
interval := cfg.GetDiscoveryInterval()
if enabled && interval > 0 {
if backgroundEnabled && interval > 0 {
s.infraDiscoveryService.SetInterval(interval)
s.infraDiscoveryService.Start(context.Background())
log.Info().
@ -874,10 +881,11 @@ func (s *Service) updateDiscoverySettings(cfg *config.AIConfig) {
}
enabled := s.IsEnabled() && cfg.IsDiscoveryEnabled()
backgroundEnabled := enabled && !BackgroundAutomationDisabledForDev()
interval := cfg.GetDiscoveryInterval()
s.discoveryService.SetCommandScanningEnabled(enabled)
if enabled && interval > 0 {
if backgroundEnabled && interval > 0 {
// Update interval and ensure service is running
s.discoveryService.SetInterval(interval)
s.discoveryService.Start(context.Background())
@ -1021,6 +1029,28 @@ const (
FeatureAIAutoFix = pkglicensing.FeatureAIAutoFix
)
const DevDisableBackgroundAIEnv = "PULSE_DEV_DISABLE_BACKGROUND_AI"
// BackgroundAutomationDisabledForDev returns true when the local development
// runtime should avoid automatic AI work that can burn provider quota while a
// developer is only trying to run the app.
func BackgroundAutomationDisabledForDev() bool {
if os.Getenv("PULSE_DEV") != "true" && !strings.EqualFold(os.Getenv("NODE_ENV"), "development") {
return false
}
value := strings.TrimSpace(os.Getenv(DevDisableBackgroundAIEnv))
if value == "" {
return false
}
disabled, err := strconv.ParseBool(value)
if err != nil {
return false
}
return disabled
}
// StartPatrol starts the background patrol service
func (s *Service) StartPatrol(ctx context.Context) {
s.mu.RLock()
@ -1049,6 +1079,21 @@ func (s *Service) StartPatrol(ctx context.Context) {
patrolCfg := s.patrolConfigFromAIConfig(cfg)
patrol.SetConfig(patrolCfg)
patrol.SetProactiveMode(cfg.UseProactiveThresholds)
if BackgroundAutomationDisabledForDev() {
patrol.SetEventTriggerConfig(PatrolEventTriggerConfig{
AlertTriggersEnabled: false,
AnomalyTriggersEnabled: false,
})
if alertAnalyzer != nil {
alertAnalyzer.SetEnabled(false)
}
log.Info().
Str("env", DevDisableBackgroundAIEnv).
Msg("Pulse dev background AI disabled; Patrol scheduler and alert-triggered AI are not started")
return
}
patrol.SetEventTriggerConfig(PatrolEventTriggerConfig{
AlertTriggersEnabled: cfg.IsPatrolAlertTriggersEnabled(),
AnomalyTriggersEnabled: cfg.IsPatrolAnomalyTriggersEnabled(),
@ -1150,6 +1195,21 @@ func (s *Service) ReconfigurePatrol() {
// Update proactive threshold mode
patrol.SetProactiveMode(cfg.UseProactiveThresholds)
if BackgroundAutomationDisabledForDev() {
patrol.SetEventTriggerConfig(PatrolEventTriggerConfig{
AlertTriggersEnabled: false,
AnomalyTriggersEnabled: false,
})
patrol.Stop()
if alertAnalyzer != nil {
alertAnalyzer.SetEnabled(false)
}
log.Info().
Str("env", DevDisableBackgroundAIEnv).
Msg("Pulse dev background AI disabled; Patrol background automation stopped")
return
}
// Update event-driven patrol trigger gate
patrol.SetEventTriggerConfig(PatrolEventTriggerConfig{
AlertTriggersEnabled: cfg.IsPatrolAlertTriggersEnabled(),

36
internal/ai/service_extended_test.go
View file

@ -854,6 +854,42 @@ func TestService_PatrolManagement(t *testing.T) {
}
}
func TestService_StartPatrolHonorsDevBackgroundAIGuard(t *testing.T) {
t.Setenv("PULSE_DEV", "true")
t.Setenv(DevDisableBackgroundAIEnv, "true")
svc := NewService(nil, nil)
patrol := NewPatrolService(svc, nil)
svc.patrolService = patrol
alertAnalyzer := &mockAlertAnalyzer{enabled: true}
svc.alertTriggeredAnalyzer = alertAnalyzer
svc.licenseChecker = &mockLicenseStore{
features: map[string]bool{
FeatureAIPatrol: true,
FeatureAIAlerts: true,
},
}
svc.cfg = &config.AIConfig{
Enabled: true,
PatrolEnabled: true,
PatrolIntervalMinutes: 30,
AlertTriggeredAnalysis: true,
PatrolEventTriggersEnabled: true,
PatrolAlertTriggersEnabled: true,
PatrolAnomalyTriggersEnabled: true,
}
svc.StartPatrol(context.Background())
if alertAnalyzer.IsEnabled() {
t.Fatal("expected dev background AI guard to keep alert analyzer disabled")
}
if status := patrol.GetStatus(); status.NextPatrolAt != nil {
t.Fatalf("expected dev background AI guard to avoid scheduling patrol, got %v", status.NextPatrolAt)
}
}
func TestService_LookupNodeForVMID_Extended(t *testing.T) {
svc := NewService(nil, nil)

30
internal/ai/service_test.go
View file

@ -34,6 +34,36 @@ func (m *mockAlertAnalyzer) IsEnabled() bool { return m.en
func (m *mockAlertAnalyzer) Start() {}
func (m *mockAlertAnalyzer) Stop() {}
func TestBackgroundAutomationDisabledForDev(t *testing.T) {
t.Run("disabled in Pulse dev", func(t *testing.T) {
t.Setenv("PULSE_DEV", "true")
t.Setenv(DevDisableBackgroundAIEnv, "true")
if !BackgroundAutomationDisabledForDev() {
t.Fatal("expected dev background AI automation to be disabled")
}
})
t.Run("explicit false allows dev automation", func(t *testing.T) {
t.Setenv("PULSE_DEV", "true")
t.Setenv(DevDisableBackgroundAIEnv, "false")
if BackgroundAutomationDisabledForDev() {
t.Fatal("expected explicit false to allow dev background AI automation")
}
})
t.Run("ignored outside dev", func(t *testing.T) {
t.Setenv("PULSE_DEV", "")
t.Setenv("NODE_ENV", "production")
t.Setenv(DevDisableBackgroundAIEnv, "true")
if BackgroundAutomationDisabledForDev() {
t.Fatal("expected background AI automation guard to be dev-only")
}
})
}
func TestNewService(t *testing.T) {
svc := NewService(nil, nil)

1
internal/ai/tools/pmg_test.go
View file

@ -391,4 +391,5 @@ func TestExecuteGetMailQueuesUsesFirstInstance(t *testing.T) {
require.NoError(t, json.Unmarshal([]byte(result.Content[0].Text), &resp))
assert.Equal(t, "gateway-1", resp.Instance)
require.Len(t, resp.Queues, 1)
assert.Equal(t, int64(5), resp.Queues[0].OldestAgeSeconds)
}

2
internal/ai/tools/tools_pmg.go
View file

@ -251,7 +251,7 @@ func (e *PulseToolExecutor) executeGetMailQueues(_ context.Context, args map[str
Hold: node.QueueStatus.Hold,
Incoming: node.QueueStatus.Incoming,
Total: node.QueueStatus.Total,
OldestAgeSeconds: 0, // Age not provided in unified resource
OldestAgeSeconds: node.QueueStatus.OldestAge,
})
}
}

84
internal/alerts/active_cleanup.go
View file

@ -5,6 +5,7 @@ import (
"time"
alertspecs "github.com/rcourtman/pulse-go-rewrite/internal/alerts/specs"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
"github.com/rs/zerolog/log"
)
@ -220,7 +221,61 @@ func (m *Manager) Cleanup(maxAge time.Duration) {
}
}
// CleanupAlertsForNodes removes alerts for nodes that no longer exist.
func alertMetadataResourceType(alert *Alert) string {
if alert == nil || alert.Metadata == nil {
return ""
}
if value, ok := alert.Metadata["resourceType"].(string); ok {
return strings.ToLower(strings.TrimSpace(value))
}
return ""
}
func shouldPreserveAlertOutsideNodeCleanup(alertID string, alert *Alert) bool {
if alert == nil {
return true
}
if strings.HasPrefix(alertID, "docker-") || strings.HasPrefix(alert.ResourceID, "docker:") {
return true
}
if strings.HasPrefix(alertID, "pbs-") || alert.Type == "pbs-offline" {
return true
}
if strings.HasPrefix(alert.ResourceID, "agent:") {
return true
}
if alert.CanonicalKind == string(alertspecs.AlertSpecKindProviderIncident) {
return true
}
switch alertMetadataResourceType(alert) {
case string(unifiedresources.ResourceTypeAgent),
string(unifiedresources.ResourceTypeAppContainer),
string(unifiedresources.ResourceTypeDockerService),
string(unifiedresources.ResourceTypeK8sCluster),
string(unifiedresources.ResourceTypeK8sNode),
string(unifiedresources.ResourceTypePod),
string(unifiedresources.ResourceTypeK8sDeployment),
string(unifiedresources.ResourceTypeStorage),
string(unifiedresources.ResourceTypePBS),
string(unifiedresources.ResourceTypePMG),
string(unifiedresources.ResourceTypeCeph),
string(unifiedresources.ResourceTypePhysicalDisk),
string(unifiedresources.ResourceTypeNetworkShare),
string(unifiedresources.ResourceTypeNetworkEndpoint),
"docker",
"host",
"kubernetes",
"k8s",
"truenas",
"vmware":
return true
default:
return false
}
}
// CleanupAlertsForNodes removes stale Proxmox node-scoped alerts.
func (m *Manager) CleanupAlertsForNodes(existingNodes map[string]bool) {
m.mu.Lock()
defer m.mu.Unlock()
@ -238,33 +293,10 @@ func (m *Manager) CleanupAlertsForNodes(existingNodes map[string]bool) {
continue
}
if strings.HasPrefix(alertID, "docker-") || strings.HasPrefix(alert.ResourceID, "docker:") {
continue
}
if strings.HasPrefix(alertID, "pbs-") || alert.Type == "pbs-offline" {
continue
}
if alert.Metadata != nil {
if resourceType, _ := alert.Metadata["resourceType"].(string); resourceType == "pbs" {
continue
}
}
if alert.CanonicalKind == string(alertspecs.AlertSpecKindConnectivity) && strings.HasPrefix(alert.ResourceID, "pbs") {
continue
}
// Agent-sourced resources (Unraid, standalone Linux hosts, TrueNAS,
// any platform reached via Pulse Agent) do not appear in the
// existingNodes map — that map is built only from Proxmox nodes
// and PBS instances. Without this carve-out, every cleanup cycle
// removes the agent alert because alert.Node is empty or unmapped,
// then the next poll re-creates the alert as "new", calls
// AddAlert, and appends a fresh history row. Result observed in
// the wild: 3,980 history entries in 7 days for what were really
// a small number of persistent issues (e.g. an Unraid array
// running without parity protection generating one entry per
// 30-second poll). Agent resource IDs are namespaced as
// "agent:<id>...", so this carve-out matches that pattern.
if strings.HasPrefix(alert.ResourceID, "agent:") {
if shouldPreserveAlertOutsideNodeCleanup(alertID, alert) {
continue
}
@ -289,7 +321,7 @@ func (m *Manager) CleanupAlertsForNodes(existingNodes map[string]bool) {
}
}()
} else {
log.Info().Msg("no alerts needed cleanup")
log.Debug().Msg("no alerts needed cleanup")
}
}

78
internal/alerts/alerts_test.go
View file

@ -11372,6 +11372,84 @@ func TestCleanupAlertsForNodes(t *testing.T) {
t.Error("non-agent alert with stale node must still be removed (control case)")
}
})
t.Run("preserves non-node canonical infrastructure alerts", func(t *testing.T) {
m := newTestManager(t)
storageProviderID := buildCanonicalStateID("storage-1", "alertspec:provider-incident:test")
pmgQueueID := buildCanonicalStateID("pmg-main", "pmg-main-oldest-message")
m.mu.Lock()
m.activeAlerts[storageProviderID] = &Alert{
ID: storageProviderID,
ResourceID: "storage-1",
Type: "storage-incident",
CanonicalKind: string(alertspecs.AlertSpecKindProviderIncident),
Node: "",
}
m.activeAlerts[pmgQueueID] = &Alert{
ID: pmgQueueID,
ResourceID: "pmg-main",
Type: "message-age",
Node: "",
Metadata: map[string]interface{}{"resourceType": "pmg"},
}
m.activeAlerts["proxmox-empty-node"] = &Alert{
ID: "proxmox-empty-node",
Node: "",
}
m.mu.Unlock()
m.CleanupAlertsForNodes(map[string]bool{"pve1": true})
time.Sleep(50 * time.Millisecond)
m.mu.RLock()
_, storageStays := testLookupActiveAlert(t, m, storageProviderID)
_, pmgStays := testLookupActiveAlert(t, m, pmgQueueID)
_, proxmoxStays := testLookupActiveAlert(t, m, "proxmox-empty-node")
m.mu.RUnlock()
if !storageStays {
t.Error("canonical provider incident must survive node cleanup")
}
if !pmgStays {
t.Error("PMG alert must survive node cleanup")
}
if proxmoxStays {
t.Error("plain node-scoped alert with empty node should still be removed")
}
})
t.Run("preserves PMG alerts created by canonical stateful evaluator", func(t *testing.T) {
m := newTestManager(t)
pmgState := buildCanonicalStateID("pmg1", "pmg1-oldest-message")
m.checkPMGOldestMessage(models.PMGInstance{
ID: "pmg1",
Name: "PMG 1",
Host: "https://pmg.mock.lan:8006",
Nodes: []models.PMGNodeStatus{
{Name: "node1", QueueStatus: &models.PMGQueueStatus{OldestAge: 2400}},
},
}, PMGThresholdConfig{
OldestMessageWarnMins: 30,
OldestMessageCritMins: 60,
})
m.CleanupAlertsForNodes(map[string]bool{"pve1": true})
time.Sleep(50 * time.Millisecond)
m.mu.RLock()
alert, pmgStays := testLookupActiveAlert(t, m, pmgState)
m.mu.RUnlock()
if !pmgStays {
t.Fatal("PMG alert created by canonical stateful evaluator must survive node cleanup")
}
if got := alert.Metadata["resourceType"]; got != string(unifiedresources.ResourceTypePMG) {
t.Fatalf("resourceType = %v, want pmg", got)
}
})
}
func TestCheckZFSPoolHealth(t *testing.T) {

6
internal/alerts/canonical_lifecycle.go
View file

@ -386,6 +386,9 @@ func (m *Manager) evaluateCanonicalLifecycleAlert(params canonicalLifecycleAlert
if alert.Metadata == nil {
alert.Metadata = make(map[string]interface{}, 2)
}
if _, ok := alert.Metadata["resourceType"]; !ok && params.Spec.ResourceType != "" {
alert.Metadata["resourceType"] = string(params.Spec.ResourceType)
}
applyCanonicalIdentity(alert, params.Spec.ID, string(params.Spec.Kind))
m.preserveAlertState(storageKey, alert)
@ -508,6 +511,9 @@ func (m *Manager) evaluateCanonicalStatefulAlert(params canonicalStatefulAlertPa
if alert.Metadata == nil {
alert.Metadata = make(map[string]interface{}, 2)
}
if _, ok := alert.Metadata["resourceType"]; !ok && params.Spec.ResourceType != "" {
alert.Metadata["resourceType"] = string(params.Spec.ResourceType)
}
applyCanonicalIdentity(alert, params.Spec.ID, string(params.Spec.Kind))
m.preserveAlertState(storageKey, alert)

4
internal/alerts/unified_eval_test.go
View file

@ -7,6 +7,7 @@ import (
alertspecs "github.com/rcourtman/pulse-go-rewrite/internal/alerts/specs"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
)
func unifiedEvalBaseConfig() AlertConfig {
@ -807,6 +808,9 @@ func TestCheckPMGOldestMessageAnnotatesCanonicalSpecMetadata(t *testing.T) {
if got := alert.Metadata["canonicalSpecID"]; got != "pmg-1-oldest-message" {
t.Fatalf("canonicalSpecID = %v, want pmg-1-oldest-message", got)
}
if got := alert.Metadata["resourceType"]; got != string(unifiedresources.ResourceTypePMG) {
t.Fatalf("resourceType = %v, want pmg", got)
}
}
func TestCheckPMGNodeQueueAnnotatesCanonicalSpecMetadata(t *testing.T) {

68
internal/api/contract_test.go
View file

@ -60,6 +60,74 @@ type resourceContractSnapshot struct {
Type string
}
func TestContract_PMGInstancesEndpointUsesUnifiedReadStatePayload(t *testing.T) {
now := time.Now().UTC()
source := models.PMGInstance{
ID: "pmg-contract-1",
Name: "gateway-contract",
Host: "https://pmg-contract.example.com:8006",
GuestURL: "https://pmg-contract.example.com/quarantine",
Status: "online",
Version: "8.2",
Nodes: []models.PMGNodeStatus{{
Name: "pmg-node-contract",
Status: "online",
QueueStatus: &models.PMGQueueStatus{
Incoming: 4,
Total: 10,
OldestAge: 600,
UpdatedAt: now,
},
}},
MailStats: &models.PMGMailStats{
CountTotal: 1000,
JunkIn: 10,
PregreetRejects: 13,
UpdatedAt: now,
},
MailCount: []models.PMGMailCountPoint{{Timestamp: now, Count: 1000, Timeframe: "hour", Index: 1}},
DomainStats: []models.PMGDomainStat{{Domain: "example.com", MailCount: 100, Bytes: 2048}},
DomainStatsAsOf: now,
ConnectionHealth: "connected",
LastSeen: now,
LastUpdated: now,
}
registry := unifiedresources.NewRegistry(nil)
registry.IngestSnapshot(models.StateSnapshot{PMGInstances: []models.PMGInstance{source}})
monitor := &monitoring.Monitor{}
setUnexportedField(t, monitor, "resourceStore", unifiedresources.NewMonitorAdapter(registry))
router := &Router{monitor: monitor}
req := httptest.NewRequest(http.MethodGet, "/api/pmg/instances?id=pmg-contract-1", nil)
rec := httptest.NewRecorder()
router.handleListPMGInstances(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String())
}
var resp PMGInstancesResponse
if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil {
t.Fatalf("decode PMG contract response: %v", err)
}
if resp.Meta.Total != 1 || len(resp.Data) != 1 {
t.Fatalf("expected one PMG instance, got total=%d len=%d", resp.Meta.Total, len(resp.Data))
}
got := resp.Data[0]
if got.ID != source.ID || got.Host != source.Host || got.GuestURL != source.GuestURL {
t.Fatalf("PMG identity and URL contract not preserved: %+v", got)
}
if len(got.Nodes) != 1 || got.Nodes[0].QueueStatus == nil || got.Nodes[0].QueueStatus.OldestAge != 600 {
t.Fatalf("PMG queue contract not preserved: %+v", got.Nodes)
}
if got.MailStats == nil || got.MailStats.CountTotal != 1000 || got.MailStats.JunkIn != 10 || got.MailStats.PregreetRejects != 13 {
t.Fatalf("PMG mail stats contract not preserved: %+v", got.MailStats)
}
if len(got.MailCount) != 1 || got.MailCount[0].Index != 1 || len(got.DomainStats) != 1 || got.DomainStats[0].Bytes != 2048 {
t.Fatalf("PMG historical/domain contract not preserved: mail=%+v domains=%+v", got.MailCount, got.DomainStats)
}
}
func TestContract_MockAvailabilityTargetsUseSavedTargetPayloads(t *testing.T) {
previousMock := mock.IsMockEnabled()
if err := mock.SetEnabled(true); err != nil {

149
internal/api/pmg.go
View file

@ -8,6 +8,7 @@ import (
"github.com/rcourtman/pulse-go-rewrite/internal/config"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
"github.com/rcourtman/pulse-go-rewrite/internal/utils"
)
@ -42,8 +43,15 @@ func (r *Router) handleListPMGInstances(w http.ResponseWriter, req *http.Request
return
}
state := monitor.GetState()
instances := filterPMGInstances(state.PMGInstances, req)
readState := monitor.GetUnifiedReadStateOrSnapshot()
views := readState.PMGInstances()
mappedInstances := make([]models.PMGInstance, len(views))
for i, v := range views {
mappedInstances[i] = mapPMGInstanceViewToModel(v)
}
instances := filterPMGInstances(mappedInstances, req)
response := PMGInstancesResponse{
Data: instances,
@ -57,6 +65,143 @@ func (r *Router) handleListPMGInstances(w http.ResponseWriter, req *http.Request
}
}
// mapPMGInstanceViewToModel projects the canonical PMG read-state view onto
// the legacy API payload consumed by existing PMG surfaces.
func mapPMGInstanceViewToModel(v *unifiedresources.PMGInstanceView) models.PMGInstance {
if v == nil {
return models.PMGInstance{}
}
nodes := v.Nodes()
modelNodes := make([]models.PMGNodeStatus, len(nodes))
for i, n := range nodes {
modelNodes[i] = models.PMGNodeStatus{
Name: n.Name,
Status: n.Status,
Role: n.Role,
Uptime: n.Uptime,
LoadAvg: n.LoadAvg,
}
if n.QueueStatus != nil {
modelNodes[i].QueueStatus = &models.PMGQueueStatus{
Active: n.QueueStatus.Active,
Deferred: n.QueueStatus.Deferred,
Hold: n.QueueStatus.Hold,
Incoming: n.QueueStatus.Incoming,
Total: n.QueueStatus.Total,
OldestAge: n.QueueStatus.OldestAge,
UpdatedAt: n.QueueStatus.UpdatedAt,
}
}
}
var modelMailStats *models.PMGMailStats
if stats := v.MailStats(); stats != nil {
countTotal := stats.CountTotal
if countTotal == 0 {
countTotal = v.MailCountTotal()
}
updatedAt := stats.UpdatedAt
if updatedAt.IsZero() {
updatedAt = v.LastUpdated()
}
modelMailStats = &models.PMGMailStats{
Timeframe: stats.Timeframe,
CountTotal: countTotal,
CountIn: stats.CountIn,
CountOut: stats.CountOut,
SpamIn: stats.SpamIn,
SpamOut: stats.SpamOut,
VirusIn: stats.VirusIn,
VirusOut: stats.VirusOut,
BouncesIn: stats.BouncesIn,
BouncesOut: stats.BouncesOut,
BytesIn: stats.BytesIn,
BytesOut: stats.BytesOut,
GreylistCount: stats.GreylistCount,
JunkIn: stats.JunkIn,
AverageProcessTimeMs: stats.AverageProcessTimeMs,
RBLRejects: stats.RBLRejects,
PregreetRejects: stats.PregreetRejects,
UpdatedAt: updatedAt,
}
}
var modelQuarantine *models.PMGQuarantineTotals
if q := v.Quarantine(); q != nil {
modelQuarantine = &models.PMGQuarantineTotals{
Spam: q.Spam,
Virus: q.Virus,
Attachment: q.Attachment,
Blacklisted: q.Blacklisted,
}
}
spamDist := v.SpamDistribution()
modelSpamDist := make([]models.PMGSpamBucket, len(spamDist))
for i, b := range spamDist {
modelSpamDist[i] = models.PMGSpamBucket{
Score: b.Bucket,
Count: b.Count,
}
}
relayDomains := v.RelayDomains()
modelRelayDomains := make([]models.PMGRelayDomain, len(relayDomains))
for i, d := range relayDomains {
modelRelayDomains[i] = models.PMGRelayDomain{
Domain: d.Domain,
Comment: d.Comment,
}
}
domainStats := v.DomainStats()
modelDomainStats := make([]models.PMGDomainStat, len(domainStats))
for i, s := range domainStats {
modelDomainStats[i] = models.PMGDomainStat{
Domain: s.Domain,
MailCount: s.MailCount,
SpamCount: s.SpamCount,
VirusCount: s.VirusCount,
Bytes: s.Bytes,
}
}
host := v.HostURL()
if strings.TrimSpace(host) == "" {
host = v.Hostname()
}
guestURL := v.GuestURL()
if strings.TrimSpace(guestURL) == "" {
guestURL = v.CustomURL()
}
instanceID := v.InstanceID()
if strings.TrimSpace(instanceID) == "" {
instanceID = v.ID()
}
inst := models.PMGInstance{
ID: instanceID,
Name: v.Name(),
Host: host,
GuestURL: guestURL,
Status: string(v.Status()),
Version: v.Version(),
Nodes: modelNodes,
MailStats: modelMailStats,
MailCount: v.MailCount(),
SpamDistribution: modelSpamDist,
Quarantine: modelQuarantine,
RelayDomains: modelRelayDomains,
DomainStats: modelDomainStats,
DomainStatsAsOf: v.DomainStatsAsOf(),
ConnectionHealth: v.ConnectionHealth(),
LastSeen: v.LastSeen(),
LastUpdated: v.LastUpdated(),
}
return inst.NormalizeCollections()
}
// filterPMGInstances narrows the snapshot by optional `id=` or
// `name=` query parameters. With neither, returns the full list.
func filterPMGInstances(instances []models.PMGInstance, req *http.Request) []models.PMGInstance {

143
internal/api/pmg_test.go Normal file
View file

@ -0,0 +1,143 @@
package api
import (
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/internal/monitoring"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
)
func TestHandleListPMGInstancesUsesUnifiedReadStateWithFullPayload(t *testing.T) {
now := time.Now().UTC()
queueUpdated := now.Add(-2 * time.Minute)
statsUpdated := now.Add(-1 * time.Minute)
domainStatsAsOf := now.Add(-30 * time.Second)
source := models.PMGInstance{
ID: "pmg-1",
Name: "gateway-main",
Host: "https://pmg.example.com:8006",
GuestURL: "https://guest.example.com/pmg",
Status: "online",
Version: "8.2",
Nodes: []models.PMGNodeStatus{{
Name: "pmg-node-1",
Status: "online",
Role: "master",
Uptime: 3600,
LoadAvg: "0.10",
QueueStatus: &models.PMGQueueStatus{
Active: 1,
Deferred: 2,
Hold: 3,
Incoming: 4,
Total: 10,
OldestAge: 600,
UpdatedAt: queueUpdated,
},
}},
MailStats: &models.PMGMailStats{
Timeframe: "day",
CountTotal: 1000,
CountIn: 700,
CountOut: 300,
SpamIn: 11,
SpamOut: 2,
VirusIn: 3,
VirusOut: 4,
BouncesIn: 5,
BouncesOut: 6,
BytesIn: 7000,
BytesOut: 8000,
GreylistCount: 9,
JunkIn: 10,
AverageProcessTimeMs: 42,
RBLRejects: 12,
PregreetRejects: 13,
UpdatedAt: statsUpdated,
},
MailCount: []models.PMGMailCountPoint{{
Timestamp: statsUpdated,
Count: 1000,
CountIn: 700,
CountOut: 300,
Timeframe: "hour",
Index: 1,
}},
SpamDistribution: []models.PMGSpamBucket{{Score: "5", Count: 6}},
Quarantine: &models.PMGQuarantineTotals{Spam: 1, Virus: 2, Attachment: 3, Blacklisted: 4},
RelayDomains: []models.PMGRelayDomain{{Domain: "example.com", Comment: "primary relay"}},
DomainStats: []models.PMGDomainStat{{Domain: "example.com", MailCount: 100, SpamCount: 5, VirusCount: 1, Bytes: 2048}},
DomainStatsAsOf: domainStatsAsOf,
ConnectionHealth: "connected",
LastSeen: now,
LastUpdated: now,
}
registry := unifiedresources.NewRegistry(nil)
registry.IngestSnapshot(models.StateSnapshot{PMGInstances: []models.PMGInstance{source}})
monitor := &monitoring.Monitor{}
setUnexportedField(t, monitor, "resourceStore", unifiedresources.NewMonitorAdapter(registry))
router := &Router{monitor: monitor}
req := httptest.NewRequest(http.MethodGet, "/api/pmg/instances?id=pmg-1", nil)
rec := httptest.NewRecorder()
router.handleListPMGInstances(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String())
}
var resp PMGInstancesResponse
if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil {
t.Fatalf("decode response: %v", err)
}
if resp.Meta.Total != 1 || len(resp.Data) != 1 {
t.Fatalf("expected one PMG instance, got total=%d len=%d", resp.Meta.Total, len(resp.Data))
}
got := resp.Data[0]
if got.ID != source.ID || got.Name != source.Name || got.GuestURL != source.GuestURL || got.Status != source.Status || got.Version != source.Version {
t.Fatalf("basic PMG identity/status fields not preserved: %+v", got)
}
if got.Host != source.Host {
t.Fatalf("expected canonical host %q, got %q", source.Host, got.Host)
}
if len(got.Nodes) != 1 || got.Nodes[0].QueueStatus == nil {
t.Fatalf("expected node queue payload, got %+v", got.Nodes)
}
queue := got.Nodes[0].QueueStatus
if queue.OldestAge != 600 || !queue.UpdatedAt.Equal(queueUpdated) || queue.Incoming != 4 || queue.Total != 10 {
t.Fatalf("queue payload not preserved: %+v", queue)
}
if got.MailStats == nil {
t.Fatal("expected mail stats payload")
}
if got.MailStats.CountTotal != 1000 || got.MailStats.JunkIn != 10 || got.MailStats.PregreetRejects != 13 || !got.MailStats.UpdatedAt.Equal(statsUpdated) {
t.Fatalf("mail stats payload not preserved: %+v", got.MailStats)
}
if len(got.MailCount) != 1 || got.MailCount[0].Count != 1000 || got.MailCount[0].Index != 1 {
t.Fatalf("mail count payload not preserved: %+v", got.MailCount)
}
if len(got.SpamDistribution) != 1 || got.SpamDistribution[0].Score != "5" {
t.Fatalf("spam distribution not preserved: %+v", got.SpamDistribution)
}
if got.Quarantine == nil || got.Quarantine.Blacklisted != 4 {
t.Fatalf("quarantine payload not preserved: %+v", got.Quarantine)
}
if len(got.RelayDomains) != 1 || got.RelayDomains[0].Domain != "example.com" {
t.Fatalf("relay domains not preserved: %+v", got.RelayDomains)
}
if len(got.DomainStats) != 1 || got.DomainStats[0].Bytes != 2048 || !got.DomainStatsAsOf.Equal(domainStatsAsOf) {
t.Fatalf("domain stats not preserved: stats=%+v asOf=%v", got.DomainStats, got.DomainStatsAsOf)
}
if got.ConnectionHealth != "connected" || !got.LastSeen.Equal(now) || !got.LastUpdated.Equal(now) {
t.Fatalf("health/timestamps not preserved: %+v", got)
}
}

14
internal/api/router.go
View file

@ -2160,7 +2160,12 @@ func (r *Router) initializeAIIntelligenceServices(ctx context.Context, orgID, da
// Set patrol trigger function (triggers mini-patrol on alert events)
patrol := aiService.GetPatrolService()
if patrol != nil {
if patrol != nil && ai.BackgroundAutomationDisabledForDev() {
log.Info().
Str("env", ai.DevDisableBackgroundAIEnv).
Str("org_id", orgID).
Msg("Pulse dev background AI disabled; alert bridge patrol trigger not wired")
} else if patrol != nil {
alertBridge.SetPatrolTrigger(func(resourceID, resourceType, reason, alertType string) {
scope := ai.PatrolScope{
ResourceIDs: []string{resourceID},
@ -3579,6 +3584,13 @@ func (r *Router) WireAlertTriggeredAI() {
return
}
if ai.BackgroundAutomationDisabledForDev() {
log.Info().
Str("env", ai.DevDisableBackgroundAIEnv).
Msg("Pulse dev background AI disabled; alert-triggered AI analyzer not wired")
return
}
// 3. Connect alert-fired events to the dedicated alert-triggered analyzer.
// Patrol's event-triggered runs are owned by the canonical alert bridge /
// trigger-manager path, so this callback should not enqueue Patrol directly.

99
internal/config/watcher.go
View file

@ -26,7 +26,7 @@ var (
debounceAPITokensWrite = 250 * time.Millisecond
)
// ConfigWatcher monitors the .env file for changes and updates runtime config
// ConfigWatcher monitors runtime config files for changes and updates runtime config.
type ConfigWatcher struct {
config *Config
envPath string
@ -147,38 +147,80 @@ func (cw *ConfigWatcher) SetAPITokenReloadCallback(callback func()) {
cw.onAPITokenReload = callback
}
// Start begins watching the config file
// Start begins watching the config files.
func (cw *ConfigWatcher) Start() error {
// Watch the directory for .env
dir := filepath.Dir(cw.envPath)
err := cw.watcher.Add(dir)
if err != nil {
log.Warn().Err(err).Str("path", dir).Msg("Failed to watch config directory")
watched := 0
for _, path := range []string{cw.envPath, cw.apiTokensPath} {
ok, err := cw.addFileWatchIfPresent(path)
if err != nil {
log.Warn().Err(err).Str("path", path).Msg("Failed to watch config file")
continue
}
if ok {
watched++
}
}
if err != nil {
log.Warn().
Err(err).
Str("env_path", cw.envPath).
Dur("poll_interval", cw.pollInterval).
Msg("Falling back to polling for config changes")
if watched > 0 {
cw.wg.Add(1)
go func() {
defer cw.wg.Done()
cw.pollForChanges()
cw.watchForChanges()
}()
return nil
log.Info().
Str("env_path", cw.envPath).
Str("api_tokens_path", cw.apiTokensPath).
Dur("poll_interval", cw.pollInterval).
Msg("Started watching config files for changes")
} else {
log.Info().
Str("env_path", cw.envPath).
Str("api_tokens_path", cw.apiTokensPath).
Dur("poll_interval", cw.pollInterval).
Msg("Config files not present yet; polling for config changes")
}
cw.wg.Add(1)
go func() {
defer cw.wg.Done()
cw.watchForChanges()
cw.pollForChanges()
}()
log.Info().Str("env_path", cw.envPath).Str("api_tokens_path", cw.apiTokensPath).Msg("Started watching config files for changes")
return nil
}
func (cw *ConfigWatcher) addFileWatchIfPresent(path string) (bool, error) {
if cw == nil || cw.watcher == nil || strings.TrimSpace(path) == "" {
return false, nil
}
stat, err := watcherOsStat(path)
if err != nil {
if os.IsNotExist(err) {
return false, nil
}
return false, err
}
if stat.IsDir() {
return false, nil
}
if err := cw.watcher.Add(path); err != nil {
return false, err
}
return true, nil
}
func (cw *ConfigWatcher) rewatchFileIfPresent(path string) {
if cw == nil || cw.watcher == nil || strings.TrimSpace(path) == "" {
return
}
if _, err := watcherOsStat(path); err != nil {
return
}
_ = cw.watcher.Remove(path)
if err := cw.watcher.Add(path); err != nil {
log.Debug().Err(err).Str("path", path).Msg("Failed to re-arm config file watch")
}
}
// Stop stops the config watcher
func (cw *ConfigWatcher) Stop() {
// Use sync.Once to prevent double-close panic
@ -211,13 +253,14 @@ func (cw *ConfigWatcher) handleEvents(events <-chan fsnotify.Event, errors <-cha
}
// Check if the event is for our .env file
if filepath.Base(event.Name) == ".env" || event.Name == cw.envPath {
if configWatcherEventMatchesPath(event, cw.envPath) {
// Debounce - wait a bit for write to complete
if !cw.waitForDuration(debounceEnvWrite) {
return
}
if event.Op&(fsnotify.Write|fsnotify.Create) != 0 {
if event.Op&(fsnotify.Write|fsnotify.Create|fsnotify.Rename|fsnotify.Remove) != 0 {
cw.rewatchFileIfPresent(cw.envPath)
// Check if content actually changed to prevent restart loops on touch
newHash, err := cw.calculateFileHash(cw.envPath)
if err == nil {
@ -226,6 +269,9 @@ func (cw *ConfigWatcher) handleEvents(events <-chan fsnotify.Event, errors <-cha
continue
}
cw.lastEnvHash = newHash
if stat, statErr := watcherOsStat(cw.envPath); statErr == nil {
cw.lastModTime = stat.ModTime()
}
}
log.Info().Str("event", event.Op.String()).Msg("Detected .env file change")
@ -233,14 +279,18 @@ func (cw *ConfigWatcher) handleEvents(events <-chan fsnotify.Event, errors <-cha
}
}
if cw.apiTokensPath != "" && (filepath.Base(event.Name) == filepath.Base(cw.apiTokensPath) || event.Name == cw.apiTokensPath) {
if configWatcherEventMatchesPath(event, cw.apiTokensPath) {
// Debounce - wait longer for atomic file operations to complete
// (write to .tmp, rename to final file)
if !cw.waitForDuration(debounceAPITokensWrite) {
return
}
if event.Op&(fsnotify.Write|fsnotify.Create) != 0 {
if event.Op&(fsnotify.Write|fsnotify.Create|fsnotify.Rename|fsnotify.Remove) != 0 {
cw.rewatchFileIfPresent(cw.apiTokensPath)
if stat, statErr := watcherOsStat(cw.apiTokensPath); statErr == nil {
cw.apiTokensLastModTime = stat.ModTime()
}
log.Info().Str("event", event.Op.String()).Msg("Detected API token file change")
cw.reloadAPITokens()
}
@ -262,6 +312,13 @@ func (cw *ConfigWatcher) handleEvents(events <-chan fsnotify.Event, errors <-cha
}
}
func configWatcherEventMatchesPath(event fsnotify.Event, path string) bool {
if strings.TrimSpace(path) == "" {
return false
}
return event.Name == path || filepath.Base(event.Name) == filepath.Base(path)
}
// waitForDuration blocks for d unless shutdown is requested first.
func (cw *ConfigWatcher) waitForDuration(d time.Duration) bool {
if d <= 0 {

23
internal/config/watcher_test.go
View file

@ -107,6 +107,29 @@ func TestConfigWatcher_StopWaitsForBackgroundWorker(t *testing.T) {
}
}
func TestConfigWatcher_StartWatchesFilesWithoutDirectoryFanout(t *testing.T) {
tempDir := t.TempDir()
envPath := filepath.Join(tempDir, ".env")
apiTokensPath := filepath.Join(tempDir, "api_tokens.json")
require.NoError(t, os.WriteFile(envPath, []byte("PULSE_AUTH_USER=admin\n"), 0644))
require.NoError(t, os.WriteFile(apiTokensPath, []byte("[]"), 0644))
require.NoError(t, os.WriteFile(filepath.Join(tempDir, ".env.tmp.123"), []byte("stale"), 0644))
t.Setenv("PULSE_AUTH_CONFIG_DIR", tempDir)
cw, err := NewConfigWatcher(&Config{ConfigPath: tempDir})
require.NoError(t, err)
cw.pollInterval = time.Hour
require.NoError(t, cw.Start())
defer cw.Stop()
watched := cw.watcher.WatchList()
assert.Contains(t, watched, envPath)
assert.Contains(t, watched, apiTokensPath)
assert.NotContains(t, watched, tempDir)
assert.NotContains(t, watched, filepath.Join(tempDir, ".env.tmp.123"))
}
func TestConfigWatcher_ReloadConfig(t *testing.T) {
// Setup
tempDir := t.TempDir()

15
internal/monitoring/canonical_guardrails_test.go
View file

@ -138,6 +138,21 @@ func TestGetStateRefreshesLiveAlertSnapshots(t *testing.T) {
}
}
func TestAlertStateSyncDoesNotLogNormalResolvedStateAtInfo(t *testing.T) {
data, err := os.ReadFile("monitor_alert_sync.go")
if err != nil {
t.Fatalf("failed to read monitor_alert_sync.go: %v", err)
}
source := string(data)
if !strings.Contains(source, `log.Debug().Int("count", len(recentlyResolved)).Msg("syncing recently resolved alerts")`) {
t.Fatal("recently resolved alert sync must stay at debug level to avoid info-log churn")
}
if strings.Contains(source, `log.Info().Int("count", len(recentlyResolved)).Msg("syncing recently resolved alerts")`) {
t.Fatal("recently resolved alert sync must not log normal polling state at info level")
}
}
func TestStateBroadcastTreatsTypedNilHubAsAbsent(t *testing.T) {
data, err := os.ReadFile("monitor.go")
if err != nil {

4
internal/monitoring/monitor_alert_sync.go
View file

@ -28,8 +28,8 @@ func (m *Monitor) syncAlertsToState() {
m.state.UpdateActiveAlerts(modelAlerts)
recentlyResolved := m.alertManager.GetRecentlyResolved()
if len(recentlyResolved) > 0 {
log.Info().Int("count", len(recentlyResolved)).Msg("syncing recently resolved alerts")
if len(recentlyResolved) > 0 && logging.IsLevelEnabled(zerolog.DebugLevel) {
log.Debug().Int("count", len(recentlyResolved)).Msg("syncing recently resolved alerts")
}
m.state.UpdateRecentlyResolved(recentlyResolved)
}

14
internal/monitoring/monitor_alerts.go
View file

@ -292,16 +292,16 @@ func (m *Monitor) broadcastStateUpdate() {
func (m *Monitor) checkMockAlerts() {
defer recoverFromPanic("checkMockAlerts")
log.Info().Bool("mockEnabled", mock.IsMockEnabled()).Msg("checkMockAlerts called")
log.Debug().Bool("mockEnabled", mock.IsMockEnabled()).Msg("checkMockAlerts called")
if !mock.IsMockEnabled() {
log.Info().Msg("mock mode not enabled, skipping mock alert check")
log.Debug().Msg("mock mode not enabled, skipping mock alert check")
return
}
// Get mock state
state := mock.CurrentFixtureGraph().State
log.Info().
log.Debug().
Int("vms", len(state.VMs)).
Int("containers", len(state.Containers)).
Int("nodes", len(state.Nodes)).
@ -322,7 +322,7 @@ func (m *Monitor) checkMockAlerts() {
existingNodes[pbsInst.Host] = true
}
}
log.Info().
log.Debug().
Int("trackedNodes", len(existingNodes)).
Msg("Collecting resources for alert cleanup in mock mode")
m.alertManager.CleanupAlertsForNodes(existingNodes)
@ -367,7 +367,7 @@ func (m *Monitor) checkMockAlerts() {
}
// Check alerts for storage
log.Info().Int("storageCount", len(state.Storage)).Msg("checking storage alerts")
log.Debug().Int("storageCount", len(state.Storage)).Msg("checking storage alerts")
for _, storage := range state.Storage {
log.Debug().
Str("name", storage.Name).
@ -377,13 +377,13 @@ func (m *Monitor) checkMockAlerts() {
}
// Check alerts for PBS instances
log.Info().Int("pbsCount", len(state.PBSInstances)).Msg("checking PBS alerts")
log.Debug().Int("pbsCount", len(state.PBSInstances)).Msg("checking PBS alerts")
for _, pbsInst := range state.PBSInstances {
m.alertManager.CheckPBS(pbsInst)
}
// Check alerts for PMG instances
log.Info().Int("pmgCount", len(state.PMGInstances)).Msg("checking PMG alerts")
log.Debug().Int("pmgCount", len(state.PMGInstances)).Msg("checking PMG alerts")
for _, pmgInst := range state.PMGInstances {
m.alertManager.CheckPMG(pmgInst)
}

22
internal/unifiedresources/adapters.go
View file

@ -1328,6 +1328,8 @@ func resourceFromPMGInstance(instance models.PMGInstance) (Resource, ResourceIde
PMG: &PMGData{
InstanceID: instance.ID,
Hostname: extractHostname(instance.Host),
HostURL: strings.TrimSpace(instance.Host),
GuestURL: strings.TrimSpace(instance.GuestURL),
Version: instance.Version,
NodeCount: len(instance.Nodes),
UptimeSeconds: uptime,
@ -1360,11 +1362,13 @@ func resourceFromPMGInstance(instance models.PMGInstance) (Resource, ResourceIde
}
if n.QueueStatus != nil {
nodes[i].QueueStatus = &PMGQueueMeta{
Active: n.QueueStatus.Active,
Deferred: n.QueueStatus.Deferred,
Hold: n.QueueStatus.Hold,
Incoming: n.QueueStatus.Incoming,
Total: n.QueueStatus.Total,
Active: n.QueueStatus.Active,
Deferred: n.QueueStatus.Deferred,
Hold: n.QueueStatus.Hold,
Incoming: n.QueueStatus.Incoming,
Total: n.QueueStatus.Total,
OldestAge: n.QueueStatus.OldestAge,
UpdatedAt: n.QueueStatus.UpdatedAt,
}
}
}
@ -1375,6 +1379,7 @@ func resourceFromPMGInstance(instance models.PMGInstance) (Resource, ResourceIde
if instance.MailStats != nil {
resource.PMG.MailStats = &PMGMailStatsMeta{
Timeframe: instance.MailStats.Timeframe,
CountTotal: instance.MailStats.CountTotal,
CountIn: instance.MailStats.CountIn,
CountOut: instance.MailStats.CountOut,
SpamIn: instance.MailStats.SpamIn,
@ -1386,11 +1391,18 @@ func resourceFromPMGInstance(instance models.PMGInstance) (Resource, ResourceIde
BytesIn: instance.MailStats.BytesIn,
BytesOut: instance.MailStats.BytesOut,
GreylistCount: instance.MailStats.GreylistCount,
JunkIn: instance.MailStats.JunkIn,
RBLRejects: instance.MailStats.RBLRejects,
AverageProcessTimeMs: instance.MailStats.AverageProcessTimeMs,
PregreetRejects: instance.MailStats.PregreetRejects,
UpdatedAt: instance.MailStats.UpdatedAt,
}
}
if len(instance.MailCount) > 0 {
resource.PMG.MailCount = append([]models.PMGMailCountPoint(nil), instance.MailCount...)
}
// Populate quarantine
if instance.Quarantine != nil {
resource.PMG.Quarantine = &PMGQuarantineMeta{

27
internal/unifiedresources/canonical_identity_test.go
View file

@ -103,6 +103,33 @@ func TestRefreshCanonicalIdentityFallsBackWithoutTargets(t *testing.T) {
}
}
func TestRefreshCanonicalIdentityUsesPMGInstanceIdentity(t *testing.T) {
resource := Resource{
ID: "pmg-resource-1",
Type: ResourceTypePMG,
Name: "",
PMG: &PMGData{
InstanceID: "pmg-main",
Hostname: "mail.example",
},
}
RefreshCanonicalIdentity(&resource)
if resource.Canonical == nil {
t.Fatalf("expected canonical identity")
}
if got := resource.Canonical.DisplayName; got != "mail.example" {
t.Fatalf("displayName = %q, want mail.example", got)
}
if got := resource.Canonical.Hostname; got != "mail.example" {
t.Fatalf("hostname = %q, want mail.example", got)
}
if got := resource.Canonical.PrimaryID; got != "pmg:pmg-main" {
t.Fatalf("primaryId = %q, want pmg:pmg-main", got)
}
}
func TestRefreshCanonicalIdentityUsesAvailabilityTargetIdentity(t *testing.T) {
resource := Resource{
ID: "availability:energy-meter",

1
internal/unifiedresources/clone.go
View file

@ -241,6 +241,7 @@ func clonePMGData(in *PMGData) *PMGData {
out := *in
out.Nodes = clonePMGNodeMetaSlice(in.Nodes)
out.MailStats = clonePMGMailStatsMeta(in.MailStats)
out.MailCount = append([]models.PMGMailCountPoint(nil), in.MailCount...)
out.Quarantine = clonePMGQuarantineMeta(in.Quarantine)
out.SpamDistribution = clonePMGSpamBucketMetaSlice(in.SpamDistribution)
out.RelayDomains = clonePMGRelayDomainMetaSlice(in.RelayDomains)

1
internal/unifiedresources/monitored_system_projection.go
View file

@ -792,6 +792,7 @@ func monitoredSystemCandidateResource(candidate MonitoredSystemCandidate) *Resou
resource.PMG = &PMGData{
InstanceID: strings.TrimSpace(candidate.ResourceID),
Hostname: host,
HostURL: strings.TrimSpace(candidate.HostURL),
}
case SourceK8s:
clusterName := firstTrimmed(candidate.Name, host, candidate.ResourceID)

28
internal/unifiedresources/pbs_pmg_registry_test.go
View file

@ -55,18 +55,23 @@ func TestIngestSnapshotIncludesPBSAndPMGInstances(t *testing.T) {
ID: "pmg-1",
Name: "pmg-main",
Host: "https://pmg.example.com:8006",
GuestURL: "https://pmg.example.com/quarantine",
Status: "online",
Version: "8.2",
ConnectionHealth: "connected",
LastSeen: now,
LastUpdated: now,
MailStats: &models.PMGMailStats{
CountTotal: 1900,
BytesIn: 5_000_000,
BytesOut: 4_000_000,
SpamIn: 125,
VirusIn: 4,
CountTotal: 1900,
BytesIn: 5_000_000,
BytesOut: 4_000_000,
SpamIn: 125,
VirusIn: 4,
JunkIn: 32,
PregreetRejects: 7,
UpdatedAt: now,
},
MailCount: []models.PMGMailCountPoint{{Timestamp: now, Count: 1900, CountIn: 1200, CountOut: 700, Timeframe: "hour", Index: 1}},
Nodes: []models.PMGNodeStatus{
{
Name: "pmg-node-1",
@ -78,6 +83,7 @@ func TestIngestSnapshotIncludesPBSAndPMGInstances(t *testing.T) {
Hold: 2,
Incoming: 3,
Total: 22,
OldestAge: 1800,
UpdatedAt: now,
},
},
@ -172,4 +178,16 @@ func TestIngestSnapshotIncludesPBSAndPMGInstances(t *testing.T) {
if pmgResource.PMG == nil || pmgResource.PMG.QueueTotal != 22 {
t.Fatalf("expected PMG payload with queue totals, got %+v", pmgResource.PMG)
}
if pmgResource.PMG.HostURL != "https://pmg.example.com:8006" || pmgResource.PMG.GuestURL != "https://pmg.example.com/quarantine" {
t.Fatalf("expected PMG URL payloads, got %+v", pmgResource.PMG)
}
if len(pmgResource.PMG.Nodes) != 1 || pmgResource.PMG.Nodes[0].QueueStatus == nil || pmgResource.PMG.Nodes[0].QueueStatus.OldestAge != 1800 {
t.Fatalf("expected PMG queue oldest age, got %+v", pmgResource.PMG.Nodes)
}
if pmgResource.PMG.MailStats == nil || pmgResource.PMG.MailStats.CountTotal != 1900 || pmgResource.PMG.MailStats.JunkIn != 32 || pmgResource.PMG.MailStats.PregreetRejects != 7 {
t.Fatalf("expected full PMG mail stats payload, got %+v", pmgResource.PMG.MailStats)
}
if len(pmgResource.PMG.MailCount) != 1 || pmgResource.PMG.MailCount[0].Index != 1 {
t.Fatalf("expected PMG mail count points, got %+v", pmgResource.PMG.MailCount)
}
}

61
internal/unifiedresources/types.go
View file

@ -901,29 +901,35 @@ type PMGNodeMeta struct {
// PMGQueueMeta describes a PMG node's postfix queue status.
type PMGQueueMeta struct {
Active int `json:"active"`
Deferred int `json:"deferred"`
Hold int `json:"hold"`
Incoming int `json:"incoming"`
Total int `json:"total"`
Active int `json:"active"`
Deferred int `json:"deferred"`
Hold int `json:"hold"`
Incoming int `json:"incoming"`
Total int `json:"total"`
OldestAge int64 `json:"oldestAge,omitempty"`
UpdatedAt time.Time `json:"updatedAt,omitempty"`
}
// PMGMailStatsMeta describes PMG mail statistics.
type PMGMailStatsMeta struct {
Timeframe string `json:"timeframe"`
CountIn float64 `json:"countIn"`
CountOut float64 `json:"countOut"`
SpamIn float64 `json:"spamIn"`
SpamOut float64 `json:"spamOut"`
VirusIn float64 `json:"virusIn"`
VirusOut float64 `json:"virusOut"`
BouncesIn float64 `json:"bouncesIn"`
BouncesOut float64 `json:"bouncesOut"`
BytesIn float64 `json:"bytesIn"`
BytesOut float64 `json:"bytesOut"`
GreylistCount float64 `json:"greylistCount"`
RBLRejects float64 `json:"rblRejects"`
AverageProcessTimeMs float64 `json:"averageProcessTimeMs"`
Timeframe string `json:"timeframe"`
CountTotal float64 `json:"countTotal"`
CountIn float64 `json:"countIn"`
CountOut float64 `json:"countOut"`
SpamIn float64 `json:"spamIn"`
SpamOut float64 `json:"spamOut"`
VirusIn float64 `json:"virusIn"`
VirusOut float64 `json:"virusOut"`
BouncesIn float64 `json:"bouncesIn"`
BouncesOut float64 `json:"bouncesOut"`
BytesIn float64 `json:"bytesIn"`
BytesOut float64 `json:"bytesOut"`
GreylistCount float64 `json:"greylistCount"`
JunkIn float64 `json:"junkIn"`
RBLRejects float64 `json:"rblRejects"`
AverageProcessTimeMs float64 `json:"averageProcessTimeMs"`
PregreetRejects float64 `json:"pregreetRejects"`
UpdatedAt time.Time `json:"updatedAt,omitempty"`
}
// PMGQuarantineMeta describes PMG quarantine totals.
@ -960,6 +966,8 @@ type PMGDomainStatMeta struct {
type PMGData struct {
InstanceID string `json:"instanceId,omitempty"`
Hostname string `json:"hostname,omitempty"`
HostURL string `json:"hostUrl,omitempty"`
GuestURL string `json:"guestUrl,omitempty"`
Version string `json:"version,omitempty"`
NodeCount int `json:"nodeCount,omitempty"`
UptimeSeconds int64 `json:"uptimeSeconds,omitempty"`
@ -974,13 +982,14 @@ type PMGData struct {
ConnectionHealth string `json:"connectionHealth,omitempty"`
LastUpdated time.Time `json:"lastUpdated,omitempty"`
Nodes []PMGNodeMeta `json:"nodes,omitempty"`
MailStats *PMGMailStatsMeta `json:"mailStats,omitempty"`
Quarantine *PMGQuarantineMeta `json:"quarantine,omitempty"`
SpamDistribution []PMGSpamBucketMeta `json:"spamDistribution,omitempty"`
RelayDomains []PMGRelayDomainMeta `json:"relayDomains,omitempty"`
DomainStats []PMGDomainStatMeta `json:"domainStats,omitempty"`
DomainStatsAsOf time.Time `json:"domainStatsAsOf,omitempty"`
Nodes []PMGNodeMeta `json:"nodes,omitempty"`
MailStats *PMGMailStatsMeta `json:"mailStats,omitempty"`
MailCount []models.PMGMailCountPoint `json:"mailCount,omitempty"`
Quarantine *PMGQuarantineMeta `json:"quarantine,omitempty"`
SpamDistribution []PMGSpamBucketMeta `json:"spamDistribution,omitempty"`
RelayDomains []PMGRelayDomainMeta `json:"relayDomains,omitempty"`
DomainStats []PMGDomainStatMeta `json:"domainStats,omitempty"`
DomainStatsAsOf time.Time `json:"domainStatsAsOf,omitempty"`
}
// VMwareData contains VMware vSphere metadata for canonical agent, vm, and

58
internal/unifiedresources/views.go
View file

@ -2629,7 +2629,21 @@ func (v PMGInstanceView) Hostname() string {
if v.r == nil || v.r.PMG == nil {
return ""
}
return v.r.PMG.Hostname
return strings.TrimSpace(v.r.PMG.Hostname)
}
func (v PMGInstanceView) HostURL() string {
if v.r == nil || v.r.PMG == nil {
return ""
}
return strings.TrimSpace(v.r.PMG.HostURL)
}
func (v PMGInstanceView) GuestURL() string {
if v.r == nil || v.r.PMG == nil {
return ""
}
return strings.TrimSpace(v.r.PMG.GuestURL)
}
func (v PMGInstanceView) Version() string {
@ -2674,6 +2688,13 @@ func (v PMGInstanceView) QueueHold() int {
return v.r.PMG.QueueHold
}
func (v PMGInstanceView) QueueIncoming() int {
if v.r == nil || v.r.PMG == nil {
return 0
}
return v.r.PMG.QueueIncoming
}
func (v PMGInstanceView) QueueTotal() int {
if v.r == nil || v.r.PMG == nil {
return 0
@ -2765,6 +2786,13 @@ func (v PMGInstanceView) MailStats() *PMGMailStatsMeta {
return clonePMGMailStatsMeta(v.r.PMG.MailStats)
}
func (v PMGInstanceView) MailCount() []models.PMGMailCountPoint {
if v.r == nil || v.r.PMG == nil {
return nil
}
return append([]models.PMGMailCountPoint(nil), v.r.PMG.MailCount...)
}
func (v PMGInstanceView) Quarantine() *PMGQuarantineMeta {
if v.r == nil || v.r.PMG == nil {
return nil
@ -2779,6 +2807,34 @@ func (v PMGInstanceView) SpamDistribution() []PMGSpamBucketMeta {
return clonePMGSpamBucketMetaSlice(v.r.PMG.SpamDistribution)
}
func (v PMGInstanceView) LastUpdated() time.Time {
if v.r == nil || v.r.PMG == nil {
return time.Time{}
}
return v.r.PMG.LastUpdated
}
func (v PMGInstanceView) RelayDomains() []PMGRelayDomainMeta {
if v.r == nil || v.r.PMG == nil {
return nil
}
return clonePMGRelayDomainMetaSlice(v.r.PMG.RelayDomains)
}
func (v PMGInstanceView) DomainStats() []PMGDomainStatMeta {
if v.r == nil || v.r.PMG == nil {
return nil
}
return clonePMGDomainStatMetaSlice(v.r.PMG.DomainStats)
}
func (v PMGInstanceView) DomainStatsAsOf() time.Time {
if v.r == nil || v.r.PMG == nil {
return time.Time{}
}
return v.r.PMG.DomainStatsAsOf
}
// K8sClusterView wraps a Kubernetes cluster resource.
type K8sClusterView struct{ r *Resource }

89
internal/unifiedresources/views_test.go
View file

@ -1329,6 +1329,9 @@ func TestView_PBSAndPMGInstanceViewAccessors(t *testing.T) {
})
t.Run("PMGInstanceView", func(t *testing.T) {
queueUpdated := now.Add(-2 * time.Minute)
statsUpdated := now.Add(-1 * time.Minute)
domainStatsAsOf := now.Add(-30 * time.Second)
r := &Resource{
ID: "pmg-1",
Type: ResourceTypePMG,
@ -1339,17 +1342,67 @@ func TestView_PBSAndPMGInstanceViewAccessors(t *testing.T) {
CustomURL: "https://pmg.example/ui",
PMG: &PMGData{
Hostname: "pmg.example",
HostURL: "https://pmg.example:8006",
GuestURL: "https://pmg-guest.example:8006",
Version: "8.0",
NodeCount: 2,
UptimeSeconds: 200,
QueueActive: 10,
QueueDeferred: 20,
QueueHold: 15,
QueueIncoming: 4,
QueueTotal: 30,
MailCountTotal: 1000,
SpamIn: 11,
VirusIn: 22,
ConnectionHealth: "online",
LastUpdated: now,
Nodes: []PMGNodeMeta{{
Name: "pmg-node-1",
Status: "online",
Role: "master",
Uptime: 123,
QueueStatus: &PMGQueueMeta{
Active: 1,
Deferred: 2,
Hold: 3,
Incoming: 4,
Total: 10,
OldestAge: 55,
UpdatedAt: queueUpdated,
},
}},
MailStats: &PMGMailStatsMeta{
Timeframe: "day",
CountTotal: 1000,
CountIn: 700,
CountOut: 300,
SpamIn: 11,
SpamOut: 2,
VirusIn: 22,
VirusOut: 1,
BouncesIn: 3,
BouncesOut: 4,
BytesIn: 500,
BytesOut: 600,
GreylistCount: 7,
JunkIn: 8,
RBLRejects: 9,
AverageProcessTimeMs: 10,
PregreetRejects: 12,
UpdatedAt: statsUpdated,
},
MailCount: []models.PMGMailCountPoint{{
Timestamp: statsUpdated,
Count: 1000,
Timeframe: "hour",
Index: 1,
}},
Quarantine: &PMGQuarantineMeta{Spam: 1, Virus: 2, Attachment: 3, Blacklisted: 4},
SpamDistribution: []PMGSpamBucketMeta{{Bucket: "5", Count: 6}},
RelayDomains: []PMGRelayDomainMeta{{Domain: "example.com", Comment: "primary"}},
DomainStats: []PMGDomainStatMeta{{Domain: "example.com", MailCount: 100, SpamCount: 5, VirusCount: 1, Bytes: 2048}},
DomainStatsAsOf: domainStatsAsOf,
},
Metrics: &ResourceMetrics{
CPU: &MetricValue{Percent: 4},
@ -1361,15 +1414,39 @@ func TestView_PBSAndPMGInstanceViewAccessors(t *testing.T) {
if v.ID() != "pmg-1" || v.Name() != "pmg-a" || v.Status() != StatusWarning {
t.Fatalf("expected basic accessors to match, got id=%q name=%q status=%q", v.ID(), v.Name(), v.Status())
}
if v.Hostname() != "pmg.example" || v.Version() != "8.0" || v.NodeCount() != 2 || v.UptimeSeconds() != 200 {
t.Fatalf("expected pmg fields to match, got host=%q ver=%q nodes=%d uptime=%d", v.Hostname(), v.Version(), v.NodeCount(), v.UptimeSeconds())
if v.Hostname() != "pmg.example" || v.HostURL() != "https://pmg.example:8006" || v.GuestURL() != "https://pmg-guest.example:8006" || v.Version() != "8.0" || v.NodeCount() != 2 || v.UptimeSeconds() != 200 {
t.Fatalf("expected pmg fields to match, got host=%q hostURL=%q guestURL=%q ver=%q nodes=%d uptime=%d", v.Hostname(), v.HostURL(), v.GuestURL(), v.Version(), v.NodeCount(), v.UptimeSeconds())
}
if v.QueueActive() != 10 || v.QueueDeferred() != 20 || v.QueueHold() != 15 || v.QueueTotal() != 30 {
t.Fatalf("expected queue active/deferred/hold/total 10/20/15/30, got %d/%d/%d/%d", v.QueueActive(), v.QueueDeferred(), v.QueueHold(), v.QueueTotal())
if v.QueueActive() != 10 || v.QueueDeferred() != 20 || v.QueueHold() != 15 || v.QueueIncoming() != 4 || v.QueueTotal() != 30 {
t.Fatalf("expected queue active/deferred/hold/incoming/total 10/20/15/4/30, got %d/%d/%d/%d/%d", v.QueueActive(), v.QueueDeferred(), v.QueueHold(), v.QueueIncoming(), v.QueueTotal())
}
if v.MailCountTotal() != 1000 || v.SpamIn() != 11 || v.VirusIn() != 22 {
t.Fatalf("expected mail stats total/spam/virus 1000/11/22, got %v/%v/%v", v.MailCountTotal(), v.SpamIn(), v.VirusIn())
}
if nodes := v.Nodes(); len(nodes) != 1 || nodes[0].QueueStatus == nil || nodes[0].QueueStatus.OldestAge != 55 || !nodes[0].QueueStatus.UpdatedAt.Equal(queueUpdated) {
t.Fatalf("expected node queue details to match, got %+v", nodes)
}
if stats := v.MailStats(); stats == nil || stats.CountTotal != 1000 || stats.JunkIn != 8 || stats.PregreetRejects != 12 || !stats.UpdatedAt.Equal(statsUpdated) {
t.Fatalf("expected full mail stats, got %+v", stats)
}
if points := v.MailCount(); len(points) != 1 || points[0].Count != 1000 || points[0].Index != 1 {
t.Fatalf("expected mail count points, got %+v", points)
}
if q := v.Quarantine(); q == nil || q.Spam != 1 || q.Blacklisted != 4 {
t.Fatalf("expected quarantine details, got %+v", q)
}
if buckets := v.SpamDistribution(); len(buckets) != 1 || buckets[0].Bucket != "5" || buckets[0].Count != 6 {
t.Fatalf("expected spam distribution, got %+v", buckets)
}
if domains := v.RelayDomains(); len(domains) != 1 || domains[0].Domain != "example.com" || domains[0].Comment != "primary" {
t.Fatalf("expected relay domains, got %+v", domains)
}
if stats := v.DomainStats(); len(stats) != 1 || stats[0].Domain != "example.com" || stats[0].Bytes != 2048 {
t.Fatalf("expected domain stats, got %+v", stats)
}
if !v.LastUpdated().Equal(now) || !v.DomainStatsAsOf().Equal(domainStatsAsOf) {
t.Fatalf("expected PMG timestamps to match, got lastUpdated=%v domainStatsAsOf=%v", v.LastUpdated(), v.DomainStatsAsOf())
}
if v.ConnectionHealth() != "online" {
t.Fatalf("expected connection health %q, got %q", "online", v.ConnectionHealth())
}
@ -1386,7 +1463,7 @@ func TestView_PBSAndPMGInstanceViewAccessors(t *testing.T) {
t.Run("NilResourceAndNilPMGAreSafe", func(t *testing.T) {
var zero PMGInstanceView
if zero.QueueActive() != 0 || zero.QueueDeferred() != 0 || zero.QueueHold() != 0 || zero.QueueTotal() != 0 {
if zero.HostURL() != "" || zero.GuestURL() != "" || zero.QueueActive() != 0 || zero.QueueDeferred() != 0 || zero.QueueHold() != 0 || zero.QueueIncoming() != 0 || zero.QueueTotal() != 0 || zero.MailCount() != nil || !zero.LastUpdated().IsZero() || zero.RelayDomains() != nil || zero.DomainStats() != nil || !zero.DomainStatsAsOf().IsZero() {
t.Fatalf("expected zero queue values for nil resource, got active=%d deferred=%d hold=%d total=%d",
zero.QueueActive(), zero.QueueDeferred(), zero.QueueHold(), zero.QueueTotal())
}
@ -1394,7 +1471,7 @@ func TestView_PBSAndPMGInstanceViewAccessors(t *testing.T) {
rNoPMG := testResource(ResourceTypePMG)
rNoPMG.PMG = nil
vNoPMG := NewPMGInstanceView(rNoPMG)
if vNoPMG.QueueActive() != 0 || vNoPMG.QueueDeferred() != 0 || vNoPMG.QueueHold() != 0 || vNoPMG.QueueTotal() != 0 {
if vNoPMG.HostURL() != "" || vNoPMG.GuestURL() != "" || vNoPMG.QueueActive() != 0 || vNoPMG.QueueDeferred() != 0 || vNoPMG.QueueHold() != 0 || vNoPMG.QueueIncoming() != 0 || vNoPMG.QueueTotal() != 0 || vNoPMG.MailCount() != nil || !vNoPMG.LastUpdated().IsZero() || vNoPMG.RelayDomains() != nil || vNoPMG.DomainStats() != nil || !vNoPMG.DomainStatsAsOf().IsZero() {
t.Fatalf("expected zero queue values when PMG is nil, got active=%d deferred=%d hold=%d total=%d",
vNoPMG.QueueActive(), vNoPMG.QueueDeferred(), vNoPMG.QueueHold(), vNoPMG.QueueTotal())
}

10
scripts/hot-dev.sh
View file

@ -17,6 +17,9 @@
# PULSE_DATA_DIR=/path Override data directory
# PULSE_DEV_API_PORT=7655 Backend API port (default: 7655)
# FRONTEND_DEV_PORT=5173 Frontend dev server port (default: 5173)
# LOG_LEVEL=debug Opt into verbose backend logs (default: info)
# PULSE_DEV_DISABLE_BACKGROUND_AI=false
# Allow automatic Patrol/discovery/alert AI in dev
# HOT_DEV_BACKEND_HEALTH_STARTUP_GRACE_SECONDS=180
# Backend /api/health grace after starts/restarts
# HOT_DEV_BACKEND_UNHEALTHY_THRESHOLD=2
@ -342,12 +345,14 @@ FRONTEND_PORT=${PULSE_DEV_API_PORT}
PORT=${PULSE_DEV_API_PORT}
PULSE_DEV=true # Enable development mode features (needed for admin bypass etc)
ALLOW_ADMIN_BYPASS=1 # Allow X-Admin-Bypass header in dev mode
LOG_LEVEL="${LOG_LEVEL:-info}"
PULSE_DEV_DISABLE_BACKGROUND_AI="${PULSE_DEV_DISABLE_BACKGROUND_AI:-true}"
# Managed dev credentials default to admin/adminadminadmin unless
# HOT_DEV_AUTH_USER / HOT_DEV_AUTH_PASS explicitly override them.
PULSE_AUTH_USER="$(hot_dev_resolve_auth_user)"
PULSE_AUTH_PASS="$(hot_dev_resolve_auth_pass)"
export FRONTEND_PORT PULSE_DEV_API_PORT PORT PULSE_DEV ALLOW_ADMIN_BYPASS PULSE_AUTH_USER PULSE_AUTH_PASS
export FRONTEND_PORT PULSE_DEV_API_PORT PORT PULSE_DEV ALLOW_ADMIN_BYPASS PULSE_AUTH_USER PULSE_AUTH_PASS LOG_LEVEL PULSE_DEV_DISABLE_BACKGROUND_AI
# Data Directory Setup
# Keep one persistent data directory in both mock and real modes so real
@ -453,7 +458,8 @@ mark_backend_startup_grace() {
start_backend_process() {
mark_backend_startup_grace
LOG_LEVEL="${LOG_LEVEL:-debug}" \
LOG_LEVEL="${LOG_LEVEL:-info}" \
PULSE_DEV_DISABLE_BACKGROUND_AI="${PULSE_DEV_DISABLE_BACKGROUND_AI:-true}" \
FRONTEND_PORT="${PULSE_DEV_API_PORT:-7655}" \
PORT="${PULSE_DEV_API_PORT:-7655}" \
PULSE_DATA_DIR="${PULSE_DATA_DIR:-}" \

14
scripts/lib/hot-dev-auth.sh
View file

@ -42,6 +42,19 @@ hot_dev_single_quote() {
printf '%s' "$1" | sed "s/'/'\\\\''/g"
}
hot_dev_cleanup_env_temp_files() {
local runtime_env=$1
local env_dir
local env_base
[[ -n "${runtime_env}" ]] || return 0
env_dir="$(dirname "${runtime_env}")"
env_base="$(basename "${runtime_env}")"
[[ -d "${env_dir}" ]] || return 0
find "${env_dir}" -maxdepth 1 -type f \( -name "${env_base}.tmp.*" -o -name "${env_base}.audit.*" \) -exec rm -f {} + 2>/dev/null || true
}
hot_dev_sync_auth_env_file() {
local runtime_env=$1
local auth_user=$2
@ -49,6 +62,7 @@ hot_dev_sync_auth_env_file() {
local tmp_file
mkdir -p "$(dirname "${runtime_env}")"
hot_dev_cleanup_env_temp_files "${runtime_env}"
tmp_file="${runtime_env}.tmp.$$"
{

3
scripts/lib/hot-dev-runtime.sh
View file

@ -76,6 +76,9 @@ hot_dev_configure_network_defaults() {
ALLOWED_ORIGINS="http://${PULSE_DEV_API_HOST:-127.0.0.1}:${FRONTEND_DEV_PORT:-7655}"
ALLOWED_ORIGINS="${ALLOWED_ORIGINS},http://localhost:${FRONTEND_DEV_PORT:-7655},http://127.0.0.1:${FRONTEND_DEV_PORT:-7655}"
ALLOWED_ORIGINS="${ALLOWED_ORIGINS},http://localhost:5173,http://127.0.0.1:5173"
if [[ "${FRONTEND_DEV_HOST:-}" == "0.0.0.0" ]]; then
ALLOWED_ORIGINS="${ALLOWED_ORIGINS},http://0.0.0.0:${FRONTEND_DEV_PORT:-7655},http://0.0.0.0:5173"
fi
local ip
for ip in ${ALL_IPS:-}; do

25
scripts/tests/test-hot-dev-auth.sh
View file

@ -104,6 +104,30 @@ EOF
assert_contains "sync preserves mock settings" "${output}" "PULSE_MOCK_MODE=false"
}
test_sync_auth_env_file_removes_stale_temp_files() {
local state_dir runtime_env output
state_dir="$(make_temp_dir)"
runtime_env="${state_dir}/.env"
touch "${runtime_env}.tmp.111" "${runtime_env}.audit.222"
output="$(
HOT_DEV_AUTH_LIB="${HOT_DEV_AUTH_LIB}" \
RUNTIME_ENV_PATH="${runtime_env}" \
bash -lc '
source "${HOT_DEV_AUTH_LIB}"
hot_dev_sync_auth_env_file "${RUNTIME_ENV_PATH}" "admin" "${HOT_DEV_DEFAULT_AUTH_HASH}"
if compgen -G "${RUNTIME_ENV_PATH}.tmp.*" >/dev/null || compgen -G "${RUNTIME_ENV_PATH}.audit.*" >/dev/null; then
printf "stale_temp_remaining=yes\n"
else
printf "stale_temp_remaining=no\n"
fi
'
)"
assert_contains "sync removes stale managed temp files" "${output}" "stale_temp_remaining=no"
}
test_sync_auth_env_file_handles_managed_only_env_under_errexit() {
local state_dir runtime_env output
state_dir="$(make_temp_dir)"
@ -175,6 +199,7 @@ source "${HOT_DEV_AUTH_LIB}"
test_default_auth_contract
test_custom_auth_banner_contract
test_sync_auth_env_file_preserves_non_auth_settings
test_sync_auth_env_file_removes_stale_temp_files
test_sync_auth_env_file_handles_managed_only_env_under_errexit
test_sync_audit_signing_env_file_restores_missing_key

3
scripts/tests/test-hot-dev-runtime.sh
View file

@ -123,6 +123,8 @@ test_hot_dev_keeps_backend_launch_errors_in_debug_log() {
output="$(sed -n '1,760p' "${HOT_DEV}")"
assert_contains "hot-dev exposes backend debug log override" "${output}" "BACKEND_DEBUG_LOG=\"\${PULSE_BACKEND_LOG_FILE:-/tmp/pulse-debug.log}\""
assert_contains "hot-dev defaults backend log level to info" "${output}" "LOG_LEVEL=\"\${LOG_LEVEL:-info}\""
assert_contains "hot-dev disables background AI automation by default" "${output}" "PULSE_DEV_DISABLE_BACKGROUND_AI=\"\${PULSE_DEV_DISABLE_BACKGROUND_AI:-true}\""
assert_contains "backend launch stderr is appended to the debug log" "${output}" ">> \"\${BACKEND_DEBUG_LOG}\" 2>&1"
assert_contains "backend LOG_FILE follows debug log override" "${output}" "LOG_FILE=\"\${BACKEND_DEBUG_LOG}\""
}
@ -165,6 +167,7 @@ test_hot_dev_network_defaults_are_lan_capable() {
assert_contains "network defaults derive the websocket URL" "${output}" "ws_url=ws://192.168.50.10:7655"
assert_contains "network defaults allow the detected LAN frontend origin" "${output}" "http://192.168.50.10:5173"
assert_contains "network defaults retain loopback frontend origins" "${output}" "http://127.0.0.1:5173"
assert_contains "network defaults allow the wildcard dev origin Electron may report" "${output}" "http://0.0.0.0:5173"
}
test_hot_dev_browser_urls_distinguish_bind_and_browser_hosts() {