fix: coordinate event-driven mesh refreshes

cmd/meshd/main.go

@@ -90,7 +90,8 @@ Up flags:
   --tun [name]      Create a real TUN device (default; asks sudo when needed)
   --no-tun          Use userspace mode without OS routes
   --port <n>        WireGuard UDP listen port (default: auto)
-  --poll-interval   DWN poll interval (default: 30s)
+  --poll-interval <dur>
+                    Override refresh cadence (fallback: 30s; healthy: 5m)
   --wait-timeout <dur>
                     How long to wait for dashboard approval (default: 15m)
   --no-wait         Exit immediately when the join request is still pending
@@ -3590,7 +3591,7 @@ func parseUpFlags(args []string) upFlags {
 			}
 		case "--poll-interval":
 			if i+1 < len(args) {
-				if d, err := time.ParseDuration(args[i+1]); err == nil {
+				if d, err := time.ParseDuration(args[i+1]); err == nil && d > 0 {
 					f.pollInterval = d
 				}
 				i++
@@ -4249,7 +4250,7 @@ func cmdUp(ctx context.Context, args []string, flagProfile string) error {
 	daemonSrv := daemon.NewServer(socketPath, func() daemon.Status {
 		routing := eng.RoutingStatus()
 		meshSnapshot := eng.MeshSnapshot()
-		selfStatus, peerStatuses, snapshotStatus := daemonStatusesFromMeshSnapshot(meshSnapshot)
+		selfStatus, peerStatuses, snapshotStatus := daemonStatusesFromMeshSnapshot(meshSnapshot, eng.RefreshHealth())
 		if meshSnapshot == nil {
 			// Preserve the legacy tray/status view before the first materialized
 			// snapshot attempt. The peer-list fast path requires Snapshot+Self
@@ -4321,22 +4322,86 @@ func cmdUp(ctx context.Context, args []string, flagProfile string) error {
 	return nil
 }
 
-func daemonStatusesFromMeshSnapshot(snapshot *engine.MeshSnapshot) (*daemon.PeerStatus, []daemon.PeerStatus, *daemon.SnapshotStatus) {
-	if snapshot == nil {
+func daemonStatusesFromMeshSnapshot(snapshot *engine.MeshSnapshot, refresh *engine.RefreshCoordinatorHealth) (*daemon.PeerStatus, []daemon.PeerStatus, *daemon.SnapshotStatus) {
+	if snapshot == nil && refresh == nil {
 		return nil, nil, nil
 	}
 
 	var self *daemon.PeerStatus
-	if snapshot.Self != nil {
-		status := daemonPeerStatus(*snapshot.Self)
-		self = &status
+	var peers []daemon.PeerStatus
+	status := &daemon.SnapshotStatus{}
+	if snapshot != nil {
+		if snapshot.Self != nil {
+			peerStatus := daemonPeerStatus(*snapshot.Self)
+			self = &peerStatus
+		}
+		peers = daemonPeerStatuses(snapshot.Peers)
+		status.Generation = snapshot.Generation
+		status.RefreshedAt = daemonSnapshotTimestamp(snapshot.RefreshedAt)
+		status.LastAttemptAt = daemonSnapshotTimestamp(snapshot.LastAttemptAt)
+		status.LastError = snapshot.LastError
+	}
+	applyDaemonRefreshHealth(status, snapshot, refresh)
+	return self, peers, status
+}
+
+func applyDaemonRefreshHealth(status *daemon.SnapshotStatus, snapshot *engine.MeshSnapshot, refresh *engine.RefreshCoordinatorHealth) {
+	if refresh == nil {
+		return
+	}
+
+	status.State = daemonRefreshState(snapshot, refresh)
+	status.Mode = string(refresh.Mode)
+	status.InFlight = refresh.InFlight
+	status.Pending = daemonRefreshReasons(refresh.PendingReasons)
+	status.Paused = refresh.Paused
+	status.LastReasons = daemonRefreshReasons(refresh.LastReasons)
+	status.LastDurationMS = refresh.LastDuration.Milliseconds()
+	status.ConsecutiveFailures = refresh.ConsecutiveFailures
+	status.LastSuccessAt = daemonSnapshotTimestamp(refresh.LastSuccessAt)
+	status.RetryNotBefore = daemonSnapshotTimestamp(refresh.RetryNotBefore)
+	status.NextAttemptAt = daemonSnapshotTimestamp(refresh.NextAttemptAt)
+	if !refresh.LastAttemptAt.IsZero() {
+		status.LastAttemptAt = daemonSnapshotTimestamp(refresh.LastAttemptAt)
+	}
+	if status.LastError == "" {
+		status.LastError = refresh.LastError
+	}
+	if len(refresh.Streams) != 0 {
+		status.Streams = make(map[string]daemon.SnapshotStreamStatus, len(refresh.Streams))
+		for stream, health := range refresh.Streams {
+			status.Streams[string(stream)] = daemon.SnapshotStreamStatus{
+				Covered:  health.Covered,
+				Live:     health.Live,
+				Repaired: health.Repaired,
+			}
+		}
+	}
+}
+
+func daemonRefreshState(snapshot *engine.MeshSnapshot, refresh *engine.RefreshCoordinatorHealth) string {
+	hasSuccessfulRefresh := !refresh.LastSuccessAt.IsZero() || (snapshot != nil && snapshot.Generation > 0)
+	if !hasSuccessfulRefresh {
+		if refresh.ConsecutiveFailures > 0 || refresh.LastError != "" || refresh.Paused {
+			return "degraded"
+		}
+		return "starting"
+	}
+	if !refresh.Running || refresh.Paused || !refresh.StreamsHealthy || refresh.ConsecutiveFailures > 0 || refresh.LastError != "" {
+		return "degraded"
+	}
+	return "healthy"
+}
+
+func daemonRefreshReasons(reasons []engine.RefreshReason) []string {
+	if len(reasons) == 0 {
+		return nil
 	}
-	return self, daemonPeerStatuses(snapshot.Peers), &daemon.SnapshotStatus{
-		Generation:    snapshot.Generation,
-		RefreshedAt:   daemonSnapshotTimestamp(snapshot.RefreshedAt),
-		LastAttemptAt: daemonSnapshotTimestamp(snapshot.LastAttemptAt),
-		LastError:     snapshot.LastError,
+	result := make([]string, len(reasons))
+	for i, reason := range reasons {
+		result[i] = string(reason)
 	}
+	return result
 }
 
 func daemonPeerStatuses(peers []engine.PeerSnapshot) []daemon.PeerStatus {

cmd/meshd/main_test.go

@@ -821,6 +821,17 @@ func TestParseUpFlagsForeground(t *testing.T) {
 	}
 }
 
+func TestParseUpFlagsPollIntervalOverride(t *testing.T) {
+	if got := parseUpFlags([]string{"--poll-interval", "2m"}).pollInterval; got != 2*time.Minute {
+		t.Fatalf("poll interval = %v, want 2m", got)
+	}
+	for _, value := range []string{"0s", "-1s", "invalid"} {
+		if got := parseUpFlags([]string{"--poll-interval", value}).pollInterval; got != 0 {
+			t.Fatalf("poll interval for %q = %v, want unset", value, got)
+		}
+	}
+}
+
 func TestShouldReexecWithSudoForTun(t *testing.T) {
 	t.Setenv(trayConnectEnv, "")
 	if !shouldReexecWithSudoForTun(upFlags{}, 501, "darwin") {

cmd/meshd/peer_list_local_test.go

@@ -219,7 +219,7 @@ func TestDaemonStatusesFromMeshSnapshot(t *testing.T) {
 		}},
 	}
 
-	self, peers, freshness := daemonStatusesFromMeshSnapshot(snapshot)
+	self, peers, freshness := daemonStatusesFromMeshSnapshot(snapshot, nil)
 	if self == nil {
 		t.Fatal("self status = nil")
 	}
@@ -240,11 +240,121 @@ func TestDaemonStatusesFromMeshSnapshot(t *testing.T) {
 		t.Fatalf("freshness = %+v", freshness)
 	}
 
-	if self, peers, freshness := daemonStatusesFromMeshSnapshot(nil); self != nil || peers != nil || freshness != nil {
+	if self, peers, freshness := daemonStatusesFromMeshSnapshot(nil, nil); self != nil || peers != nil || freshness != nil {
 		t.Fatalf("nil snapshot = (%+v, %+v, %+v), want nils", self, peers, freshness)
 	}
 }
 
+func TestDaemonRefreshHealthMapping(t *testing.T) {
+	base := time.Date(2026, 7, 11, 12, 0, 0, 987654321, time.FixedZone("offset", -7*60*60))
+	snapshot := &engine.MeshSnapshot{
+		Generation:    12,
+		RefreshedAt:   base,
+		LastAttemptAt: base,
+	}
+	health := &engine.RefreshCoordinatorHealth{
+		Running:        true,
+		Paused:         true,
+		InFlight:       true,
+		Mode:           engine.RefreshModeFallback,
+		StreamsHealthy: false,
+		Streams: map[engine.RefreshStream]engine.RefreshStreamHealth{
+			engine.RefreshStreamTopology: {Covered: true, Live: true, Repaired: true},
+			engine.RefreshStreamDelivery: {Covered: true, Live: false, Repaired: false},
+		},
+		PendingReasons:      []engine.RefreshReason{engine.RefreshReasonDelivery, engine.RefreshReasonEndpoint},
+		ConsecutiveFailures: 2,
+		LastAttemptAt:       base.Add(2 * time.Minute),
+		LastSuccessAt:       base.Add(-time.Minute),
+		LastReasons:         []engine.RefreshReason{engine.RefreshReasonTopology},
+		LastDuration:        1500*time.Millisecond + 250*time.Microsecond,
+		LastError:           "refresh failed",
+		RetryNotBefore:      base.Add(3 * time.Minute),
+		NextAttemptAt:       base.Add(3 * time.Minute),
+	}
+
+	_, _, got := daemonStatusesFromMeshSnapshot(snapshot, health)
+	if got == nil {
+		t.Fatal("snapshot status = nil")
+	}
+	if got.State != "degraded" || got.Mode != "fallback" || !got.InFlight || !got.Paused {
+		t.Fatalf("scheduler state = %+v", got)
+	}
+	if !reflect.DeepEqual(got.Pending, []string{"delivery", "endpoint"}) ||
+		!reflect.DeepEqual(got.LastReasons, []string{"topology"}) {
+		t.Fatalf("refresh reasons = pending %v, last %v", got.Pending, got.LastReasons)
+	}
+	if got.LastDurationMS != 1500 || got.ConsecutiveFailures != 2 || got.LastError != "refresh failed" {
+		t.Fatalf("refresh failure status = %+v", got)
+	}
+	if got.LastAttemptAt != health.LastAttemptAt.UTC().Format(time.RFC3339Nano) ||
+		got.LastSuccessAt != health.LastSuccessAt.UTC().Format(time.RFC3339Nano) ||
+		got.RetryNotBefore != health.RetryNotBefore.UTC().Format(time.RFC3339Nano) ||
+		got.NextAttemptAt != health.NextAttemptAt.UTC().Format(time.RFC3339Nano) {
+		t.Fatalf("refresh timestamps = %+v", got)
+	}
+	if len(got.Streams) != 2 || !got.Streams["topology"].Covered || !got.Streams["topology"].Live ||
+		!got.Streams["topology"].Repaired || !got.Streams["delivery"].Covered || got.Streams["delivery"].Live ||
+		got.Streams["delivery"].Repaired {
+		t.Fatalf("stream status = %+v", got.Streams)
+	}
+}
+
+func TestDaemonRefreshState(t *testing.T) {
+	tests := []struct {
+		name     string
+		snapshot *engine.MeshSnapshot
+		health   engine.RefreshCoordinatorHealth
+		want     string
+	}{
+		{
+			name: "starting before first success",
+			health: engine.RefreshCoordinatorHealth{
+				Running: true,
+				Mode:    engine.RefreshModeFallback,
+			},
+			want: "starting",
+		},
+		{
+			name:     "healthy after success with repaired streams",
+			snapshot: &engine.MeshSnapshot{Generation: 1},
+			health: engine.RefreshCoordinatorHealth{
+				Running:        true,
+				Mode:           engine.RefreshModeHealthy,
+				StreamsHealthy: true,
+			},
+			want: "healthy",
+		},
+		{
+			name:     "degraded while subscriptions use fallback",
+			snapshot: &engine.MeshSnapshot{Generation: 1},
+			health: engine.RefreshCoordinatorHealth{
+				Running:        true,
+				Mode:           engine.RefreshModeFallback,
+				StreamsHealthy: false,
+			},
+			want: "degraded",
+		},
+		{
+			name: "degraded when initial refresh fails",
+			health: engine.RefreshCoordinatorHealth{
+				Running:             true,
+				ConsecutiveFailures: 1,
+				LastError:           "unavailable",
+			},
+			want: "degraded",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := daemonRefreshState(tc.snapshot, &tc.health); got != tc.want {
+				t.Fatalf("daemonRefreshState() = %q, want %q", got, tc.want)
+			}
+		})
+	}
+}
+
 func TestCmdPeerListUsesDaemonSnapshotBeforeIdentity(t *testing.T) {
 	stateDir := t.TempDir()
 	t.Setenv("MESHD_STATE_DIR", stateDir)

internal/daemon/daemon.go

@@ -84,10 +84,32 @@ type PeerStatus struct {
 // SnapshotStatus describes the freshness of the daemon's materialized mesh
 // snapshot. Timestamp fields are encoded as RFC3339Nano strings.
 type SnapshotStatus struct {
-	Generation    uint64 `json:"generation"`
-	RefreshedAt   string `json:"refreshedAt,omitempty"`
-	LastAttemptAt string `json:"lastAttemptAt,omitempty"`
-	LastError     string `json:"lastError,omitempty"`
+	Generation          uint64                          `json:"generation"`
+	RefreshedAt         string                          `json:"refreshedAt,omitempty"`
+	LastAttemptAt       string                          `json:"lastAttemptAt,omitempty"`
+	LastError           string                          `json:"lastError,omitempty"`
+	State               string                          `json:"state,omitempty"`
+	Mode                string                          `json:"mode,omitempty"`
+	InFlight            bool                            `json:"inFlight,omitempty"`
+	Pending             []string                        `json:"pending,omitempty"`
+	Paused              bool                            `json:"paused,omitempty"`
+	LastReasons         []string                        `json:"lastReasons,omitempty"`
+	LastDurationMS      int64                           `json:"lastDurationMs,omitempty"`
+	ConsecutiveFailures int                             `json:"consecutiveFailures,omitempty"`
+	LastSuccessAt       string                          `json:"lastSuccessAt,omitempty"`
+	RetryNotBefore      string                          `json:"retryNotBefore,omitempty"`
+	NextAttemptAt       string                          `json:"nextAttemptAt,omitempty"`
+	Streams             map[string]SnapshotStreamStatus `json:"streams,omitempty"`
+}
+
+// SnapshotStreamStatus describes whether one invalidation stream is
+// authorized, connected, and repaired through its end-of-stored-events
+// barrier. False values are intentionally retained when the stream is present
+// so operators can distinguish missing coverage from a legacy daemon.
+type SnapshotStreamStatus struct {
+	Covered  bool `json:"covered"`
+	Live     bool `json:"live"`
+	Repaired bool `json:"repaired"`
 }
 
 type peerAuthorizedContextKey struct{}

internal/daemon/status_snapshot_test.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"net"
 	"net/http"
+	"reflect"
 	"testing"
 	"time"
 )
@@ -34,10 +35,25 @@ func TestStatusEndpointRichSnapshot(t *testing.T) {
 		LastSeen:       "2026-07-11T17:40:06.987654321Z",
 	}
 	wantSnapshot := SnapshotStatus{
-		Generation:    42,
-		RefreshedAt:   "2026-07-11T17:40:06.987654321Z",
-		LastAttemptAt: "2026-07-11T17:40:07.123456789Z",
-		LastError:     "refresh deferred by rate limit",
+		Generation:          42,
+		RefreshedAt:         "2026-07-11T17:40:06.987654321Z",
+		LastAttemptAt:       "2026-07-11T17:40:07.123456789Z",
+		LastError:           "refresh deferred by rate limit",
+		State:               "degraded",
+		Mode:                "fallback",
+		InFlight:            true,
+		Pending:             []string{"delivery", "endpoint"},
+		Paused:              true,
+		LastReasons:         []string{"topology"},
+		LastDurationMS:      1250,
+		ConsecutiveFailures: 3,
+		LastSuccessAt:       "2026-07-11T17:39:06.987654321Z",
+		RetryNotBefore:      "2026-07-11T17:40:09.123456789Z",
+		NextAttemptAt:       "2026-07-11T17:40:09.123456789Z",
+		Streams: map[string]SnapshotStreamStatus{
+			"delivery": {Covered: true, Live: false, Repaired: false},
+			"topology": {Covered: true, Live: true, Repaired: true},
+		},
 	}
 
 	srv := NewServer(sock, func() Status {
@@ -68,7 +84,7 @@ func TestStatusEndpointRichSnapshot(t *testing.T) {
 	if status.Self == nil || *status.Self != wantSelf {
 		t.Fatalf("Self = %+v, want %+v", status.Self, wantSelf)
 	}
-	if status.Snapshot == nil || *status.Snapshot != wantSnapshot {
+	if status.Snapshot == nil || !reflect.DeepEqual(*status.Snapshot, wantSnapshot) {
 		t.Fatalf("Snapshot = %+v, want %+v", status.Snapshot, wantSnapshot)
 	}
 }
@@ -113,6 +129,23 @@ func TestClientGetStatusLegacyResponse(t *testing.T) {
 	}
 }
 
+func TestSnapshotStatusLegacyJSONShape(t *testing.T) {
+	snapshot := SnapshotStatus{
+		Generation:    7,
+		RefreshedAt:   "2026-07-11T17:40:06Z",
+		LastAttemptAt: "2026-07-11T17:40:07Z",
+		LastError:     "temporary failure",
+	}
+	got, err := json.Marshal(snapshot)
+	if err != nil {
+		t.Fatalf("Marshal() error: %v", err)
+	}
+	want := `{"generation":7,"refreshedAt":"2026-07-11T17:40:06Z","lastAttemptAt":"2026-07-11T17:40:07Z","lastError":"temporary failure"}`
+	if string(got) != want {
+		t.Fatalf("legacy SnapshotStatus JSON = %s, want %s", got, want)
+	}
+}
+
 func TestPeerStatusLegacyJSONShape(t *testing.T) {
 	peer := PeerStatus{Name: "peer-a", MeshIP: "10.200.0.2", Online: true}
 	got, err := json.Marshal(peer)