fleet: wire health collection so the autonomous loops actually have data

Three small follow-ups that turn the fleet management system from 'all
the pieces ship green' into 'actually produces telemetry':

cmd/worker/main.go: go workerService.RunHealth(ctx, 30s) alongside
Heartbeat. The sampler snapshots rolling 1m counters into a WorkerHealth
event via the existing event bus + codec path.

cmd/backend/main.go: background goroutine refreshes
worker_capacity_view every minute via REFRESH MATERIALIZED VIEW
CONCURRENTLY. The assignment loop, Rebalancer, Scaler, and
QuarantineEvaluator all read from the view, so it's the freshness gate
for the whole system.

internal/app/worker/event_send_email.go + health_record.go: classify
every wmail.SendResult into the right counter (auth / rate-limit /
bounce-hard / bounce-soft / success) and record SMTP latency. Falls
back to free-text message classification when the error code is
generic, so signal stays useful as new error paths are added.

End-to-end: a worker that bounces 10% of sends now lands in the
'quarantined' band within 5min of the QuarantineEvaluator tick,
auto-drains via Rebalancer, and triggers a Scaler alert if its
removal drops fleet capacity below the warning threshold.

cmd/backend/main.go

@@ -500,6 +500,26 @@ func main() {
 		// interval and writes every action to decision_log. Cancel them via
 		// the root context on shutdown.
 		decisionLogRepo := repository.NewDecisionLogRepository(primaryDB)
+
+		// Refresh worker_capacity_view every minute so the assignment loop +
+		// rebalance + scale + quarantine evaluators see fresh rolling
+		// metrics. The materialized view is what aggregates the 1h windows
+		// across all workers.
+		go func() {
+			tick := time.NewTicker(time.Minute)
+			defer tick.Stop()
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				case <-tick.C:
+					if err := workerRepository.RefreshWorkerCapacityView(ctx); err != nil {
+						log.Printf("worker_capacity_view refresh: %v", err)
+					}
+				}
+			}
+		}()
+
 		go (&fleet.Rebalancer{
 			WorkerRepo: workerRepository,
 			Decisions:  decisionLogRepo,

cmd/worker/main.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	awsconf "github.com/aws/aws-sdk-go-v2/config"
 	"github.com/google/uuid"
@@ -159,8 +160,11 @@ func main() {
 
 	workerService.InitEvents()
 
-	// Start heartbeat
+	// Start heartbeat + health sampler. RunHealth ticks every 30s, snapshots
+	// the rolling 1m counters into a WorkerHealth event, publishes via the
+	// event bus so the consumer can write a row into worker_health_samples.
 	go workerService.Heartbeat(ctx)
+	go workerService.RunHealth(ctx, 30*time.Second)
 
 	// Graceful shutdown
 	sigCh := make(chan os.Signal, 1)

internal/app/worker/event_send_email.go

@@ -61,6 +61,8 @@ func (w *WorkerService) HandleSendEmail(ctx context.Context, body any) error {
 	}
 
 	// Use unified Send method
+	w.recordSendAttempt()
+	sendStart := time.Now()
 	result := mail.Send(ctx, &wmail.SendRequest{
 		TaskID:      sendEmail.TaskID,
 		To:          sendEmail.To,
@@ -75,6 +77,8 @@ func (w *WorkerService) HandleSendEmail(ctx context.Context, body any) error {
 		IsWarmup:    sendEmail.IsWarmup,
 		WarmupToken: sendEmail.WarmupToken,
 	})
+	w.recordSendLatency(time.Since(sendStart))
+	w.recordSendOutcome(result)
 
 	if result.Success {
 		log.Info().

internal/app/worker/health_record.go

@@ -0,0 +1,67 @@
+package worker
+
+import (
+	"strings"
+	"time"
+
+	"github.com/warmbly/warmbly/internal/app/worker/wmail"
+	"github.com/warmbly/warmbly/internal/errx"
+)
+
+// recordSendAttempt + recordSendLatency + recordSendOutcome are the
+// integration points the send hot path calls. They classify a
+// wmail.SendResult into the right Record* shim so the rolling 1m
+// counters feed the QuarantineEvaluator's band classification with the
+// granularity that matters (bounce vs complaint vs auth vs rate limit).
+
+func (s *WorkerService) recordSendAttempt() {
+	s.RecordSendAttempt()
+}
+
+func (s *WorkerService) recordSendLatency(d time.Duration) {
+	s.RecordSMTPLatency(int32(d.Milliseconds()))
+}
+
+func (s *WorkerService) recordSendOutcome(result *wmail.SendResult) {
+	if result == nil {
+		return
+	}
+	if result.Success {
+		s.RecordSendSuccess()
+		return
+	}
+	if result.Error == nil {
+		return
+	}
+	switch result.Error.Code {
+	case errx.MailErrorCodeAuthenticationFailed,
+		errx.MailErrorCodeAuthorizationFailed,
+		errx.MailErrorCodeInvalidCredentials,
+		errx.MailErrorCodeGoogleAuth:
+		s.RecordAuthError()
+	case errx.MailErrorCodeRateLimitExceeded,
+		errx.MailErrorCodeSendingTooFast,
+		errx.MailErrorCodeQuotaExceeded:
+		s.RecordRateLimitError()
+	case errx.MailErrorCodeRecipientRejected,
+		errx.MailErrorCodeAccountSuspended:
+		s.RecordBounceHard()
+	case errx.MailErrorCodeServerUnreachable,
+		errx.MailErrorCodeConnectionLost:
+		s.RecordBounceSoft()
+	default:
+		// Best-effort classification on free-text — keeps the signal
+		// useful even when the error code is generic.
+		msg := strings.ToLower(result.Error.Message)
+		switch {
+		case strings.Contains(msg, "bounce") || strings.Contains(msg, "rejected"):
+			s.RecordBounceHard()
+		case strings.Contains(msg, "rate limit") || strings.Contains(msg, "throttle"):
+			s.RecordRateLimitError()
+		case strings.Contains(msg, "auth"):
+			s.RecordAuthError()
+		default:
+			s.RecordBounceSoft()
+		}
+	}
+}