fix(worker): harden control-plane lifecycle

- reconnect safely after token rotation and retry leased results
- reject malformed tasks and remove production cluster debug mutation
- validate environment files and require immutable container images

BREAKING CHANGE: Docker install, deploy, and Compose now require an
immutable repository@sha256 image reference.

internal/distworker/client.go

@@ -2,12 +2,15 @@ package distworker
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"net/url"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/gorilla/websocket"
@@ -49,12 +52,16 @@ func NewClient(endpoint, authToken string) *Client {
 
 // postJSON is a helper for sending JSON POST requests
 func (c *Client) postJSON(path string, payload interface{}) (*http.Response, error) {
+	return c.postJSONContext(context.Background(), path, payload)
+}
+
+func (c *Client) postJSONContext(ctx context.Context, path string, payload interface{}) (*http.Response, error) {
 	body, err := json.Marshal(payload)
 	if err != nil {
 		return nil, err
 	}
 
-	httpReq, err := http.NewRequest("POST", c.endpoint+path, bytes.NewReader(body))
+	httpReq, err := http.NewRequestWithContext(ctx, "POST", c.endpoint+path, bytes.NewReader(body))
 	if err != nil {
 		return nil, err
 	}
@@ -91,10 +98,10 @@ func (c *Client) Heartbeat(req wire.HeartbeatRequest) error {
 // bearer token for this worker. The current bearer is used for
 // authentication; the response carries the freshly issued token.
 //
-// Returns the new token string. The main app invalidates the old
-// token immediately.
-func (c *Client) RotateToken() (string, error) {
-	resp, err := c.postJSON("/api/internal/workers/rotate-token", struct{}{})
+// Returns the new token string. The main app invalidates the old token
+// immediately.
+func (c *Client) RotateToken(ctx context.Context) (string, error) {
+	resp, err := c.postJSONContext(ctx, "/api/internal/workers/rotate-token", struct{}{})
 	if err != nil {
 		return "", err
 	}
@@ -156,6 +163,12 @@ func (c *Client) ReportResults(req wire.ResultsRequest) error {
 
 // WorkerSocket connects to the websocket task channel.
 func (c *Client) WorkerSocket() (*websocket.Conn, error) {
+	return c.WorkerSocketContext(context.Background())
+}
+
+// WorkerSocketContext connects to the websocket task channel, cancelling the
+// dial when the caller's context ends.
+func (c *Client) WorkerSocketContext(ctx context.Context) (*websocket.Conn, error) {
 	endpoint := strings.TrimRight(c.endpoint, "/")
 	wsURL := endpoint
 	if !strings.HasSuffix(wsURL, "/worker") && !strings.HasSuffix(wsURL, "/api/worker") {
@@ -175,7 +188,33 @@ func (c *Client) WorkerSocket() (*websocket.Conn, error) {
 	q.Set("token", c.authToken)
 	u.RawQuery = q.Encode()
 
-	conn, resp, err := websocket.DefaultDialer.Dial(u.String(), nil)
+	dialer := *websocket.DefaultDialer
+	var (
+		connMu   sync.Mutex
+		dialConn net.Conn
+	)
+	stopClose := context.AfterFunc(ctx, func() {
+		connMu.Lock()
+		if dialConn != nil {
+			_ = dialConn.Close()
+		}
+		connMu.Unlock()
+	})
+	defer stopClose()
+	dialer.NetDialContext = func(dialCtx context.Context, network, address string) (net.Conn, error) {
+		conn, err := (&net.Dialer{}).DialContext(dialCtx, network, address)
+		if err != nil {
+			return nil, err
+		}
+		connMu.Lock()
+		dialConn = conn
+		if ctx.Err() != nil {
+			_ = conn.Close()
+		}
+		connMu.Unlock()
+		return conn, nil
+	}
+	conn, resp, err := dialer.DialContext(ctx, u.String(), nil)
 	if err != nil && resp != nil {
 		defer resp.Body.Close() //nolint:errcheck
 		body, _ := io.ReadAll(resp.Body)

internal/distworker/notification.go

@@ -25,14 +25,27 @@ import (
 // silently dropping it as the production plan forbids).
 //
 //nolint:gocritic // task model is shared with the rest of the dispatcher; keep by-value
-func (r *Runner) ExecuteNotification(ctx context.Context, task models.Task) wire.NotificationResultReport {
+func (r *Runner) ExecuteNotification(ctx context.Context, task models.Task) (report wire.NotificationResultReport) {
 	start := time.Now()
-
-	report := wire.NotificationResultReport{
+	method := ""
+	report = wire.NotificationResultReport{
 		JobID:      task.JobID,
 		Status:     wire.NotificationResultPermanent,
 		DurationMs: 0,
 	}
+	defer func() {
+		if recover() != nil {
+			report.Status = wire.NotificationResultPermanent
+			report.ProviderResponse = nil
+			report.RetryAfterSeconds = nil
+			report.Error = stringPtr("notification executor panic")
+		}
+		report.DurationMs = int(time.Since(start) / time.Millisecond)
+		r.recordDelegatedNotification(report.JobID, method, report)
+	}()
+	if r.notificationExecutor != nil {
+		return r.notificationExecutor(ctx, task)
+	}
 
 	if len(task.Payload) == 0 {
 		report.Status = wire.NotificationResultPermanent
@@ -50,6 +63,7 @@ func (r *Runner) ExecuteNotification(ctx context.Context, task models.Task) wire
 	if nt.JobID == "" {
 		nt.JobID = task.JobID
 	}
+	method = nt.Method
 	if nt.MessageID == 0 && task.MessageID != nil {
 		nt.MessageID = *task.MessageID
 	}

internal/distworker/notification_test.go

@@ -2,7 +2,10 @@ package distworker
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"fmt"
+	"sync"
 	"testing"
 	"time"
 
@@ -228,6 +231,177 @@ func TestExecuteNotification_ReportsJobIDFromPayload(t *testing.T) {
 	assert.Equal(t, "outer-job", report.JobID)
 }
 
+func TestDelegatedNotificationAuditCoexistsWithSelfcheckWithoutSecrets(t *testing.T) {
+	const (
+		recipient        = "ops-private@example.com"
+		renderedBody     = "private rendered body"
+		credentialSecret = "smtp-password-secret"
+		hookURL          = "https://hooks.example.com/private"
+		authorization    = "Bearer private-authorization"
+		providerResponse = "private provider response"
+	)
+	r := runnerWithCreds(&wire.NotificationCredentials{
+		SMTP: []wire.SMTPCredential{{Password: credentialSecret}},
+	})
+	r.notifyResults = make(chan notifyResultEnvelope, 1)
+	r.RecordNotification(&NotificationRow{
+		Kind: "email", Channel: notificationChannelSMTP, Subject: "selfcheck", Body: "main API down", OK: true,
+	})
+	r.executeAndForwardNotification(wire.NotificationTask{
+		JobID: "delegated-job", Method: "sms", Subject: authorization,
+		BodyText: renderedBody, BodyHTML: providerResponse,
+		Contact: wire.NotificationContact{Kind: "sms", Value: recipient, Name: hookURL},
+	})
+
+	rows := r.RecentNotifications(10)
+	require.Len(t, rows, 2)
+	assert.Equal(t, "selfcheck", rows[0].Subject)
+	delegated := rows[1]
+	assert.Equal(t, "delegated-job", delegated.JobID)
+	assert.Equal(t, "sms", delegated.Method)
+	assert.Equal(t, wire.NotificationResultPermanent, delegated.Status)
+	assert.Empty(t, delegated.Subject)
+	assert.Empty(t, delegated.Body)
+	assert.Empty(t, delegated.Error)
+	assert.False(t, delegated.At.IsZero())
+
+	stored, err := json.Marshal(delegated)
+	require.NoError(t, err)
+	for _, secret := range []string{recipient, renderedBody, credentialSecret, hookURL, authorization, providerResponse} {
+		assert.NotContains(t, string(stored), secret)
+	}
+}
+
+func TestDelegatedNotificationAuditRecordsOneTerminalRow(t *testing.T) {
+	const secret = "private-notification-secret"
+	validTask := func(jobID string) wire.NotificationTask {
+		return wire.NotificationTask{
+			JobID: jobID, LeaseToken: "lease", Method: "sms", Subject: secret, BodyText: secret,
+			Contact: wire.NotificationContact{Value: secret, Name: secret},
+		}
+	}
+	tests := []struct {
+		name   string
+		invoke func(*Runner)
+		method string
+	}{
+		{
+			name:   "normal result",
+			invoke: func(r *Runner) { r.executeAndForwardNotification(validTask("normal")) },
+			method: "sms",
+		},
+		{
+			name: "invalid deadline",
+			invoke: func(r *Runner) {
+				task := validTask("invalid-deadline")
+				deadline := "not-a-timestamp"
+				task.Deadline = &deadline
+				r.executeAndForwardNotification(task)
+			},
+			method: "sms",
+		},
+		{
+			name: "expired deadline",
+			invoke: func(r *Runner) {
+				task := validTask("expired-deadline")
+				deadline := time.Now().Add(-time.Second).Format(time.RFC3339Nano)
+				task.Deadline = &deadline
+				r.executeAndForwardNotification(task)
+			},
+			method: "sms",
+		},
+		{
+			name: "malformed payload",
+			invoke: func(r *Runner) {
+				r.ExecuteNotification(context.Background(), models.Task{JobID: "malformed-payload", Payload: []byte(`{"subject":"` + secret)})
+			},
+		},
+		{
+			name: "executor panic",
+			invoke: func(r *Runner) {
+				r.notificationExecutor = func(context.Context, models.Task) wire.NotificationResultReport {
+					panic(secret)
+				}
+				r.ExecuteNotification(context.Background(), models.Task{JobID: "panic", Payload: []byte(`{}`)})
+			},
+		},
+		{
+			name: "malformed envelope",
+			invoke: func(r *Runner) {
+				r.enqueueTaskMessage(wire.WorkerMessage{TaskEnvelope: &wire.TaskEnvelope{
+					Type: "invalid", JobID: "bad-envelope", Notify: ptrNotificationTask(validTask("bad-envelope")),
+				}})
+			},
+			method: "sms",
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			r := runnerWithCreds(&wire.NotificationCredentials{})
+			r.notifyResults = make(chan notifyResultEnvelope, 1)
+			tc.invoke(r)
+
+			rows := r.RecentNotifications(10)
+			require.Len(t, rows, 1)
+			assert.Equal(t, wire.NotificationResultPermanent, rows[0].Status)
+			assert.Equal(t, tc.method, rows[0].Method)
+			stored, err := json.Marshal(rows[0])
+			require.NoError(t, err)
+			assert.NotContains(t, string(stored), secret)
+		})
+	}
+}
+
+func TestDelegatedNotificationAuditConcurrentRowsAreSecretFree(t *testing.T) {
+	const attempts = 32
+	const secret = "concurrent-private-secret"
+	r := runnerWithCreds(&wire.NotificationCredentials{})
+	var wg sync.WaitGroup
+	for i := 0; i < attempts; i++ {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			r.ExecuteNotification(context.Background(), models.Task{
+				JobID:   fmt.Sprintf("job-%d", i),
+				Payload: []byte(`{"job_id":"job-` + fmt.Sprint(i) + `","method":"sms","body_text":"` + secret + `","contact":{"value":"` + secret + `"}}`),
+			})
+		}(i)
+	}
+	wg.Wait()
+
+	rows := r.RecentNotifications(attempts)
+	require.Len(t, rows, attempts)
+	for _, row := range rows {
+		stored, err := json.Marshal(row)
+		require.NoError(t, err)
+		assert.NotContains(t, string(stored), secret)
+		assert.Equal(t, wire.NotificationResultPermanent, row.Status)
+	}
+}
+
+func TestInvalidNotificationDeadlineIsNotExecuted(t *testing.T) {
+	r := runnerWithCreds(&wire.NotificationCredentials{})
+	r.notifyResults = make(chan notifyResultEnvelope, 1)
+	called := false
+	r.notificationExecutor = func(context.Context, models.Task) wire.NotificationResultReport {
+		called = true
+		return wire.NotificationResultReport{Status: wire.NotificationResultDelivered}
+	}
+	deadline := "not-a-timestamp"
+	r.executeAndForwardNotification(wire.NotificationTask{
+		JobID: "invalid-deadline", LeaseToken: "lease", Method: "sms", Deadline: &deadline,
+	})
+
+	assert.False(t, called)
+	env := <-r.notifyResults
+	assert.Equal(t, wire.NotificationResultPermanent, env.report.Status)
+	require.NotNil(t, env.report.Error)
+	assert.Equal(t, "invalid notification deadline", *env.report.Error)
+	require.Len(t, r.RecentNotifications(10), 1)
+}
+
+func ptrNotificationTask(task wire.NotificationTask) *wire.NotificationTask { return &task }
+
 // guard against time import being unused if the above compile-time helpers
 // are dropped in a future refactor.
 var _ = time.Second

internal/distworker/results.go

@@ -14,10 +14,7 @@ import (
 const recentResultsSize = 200
 
 // recentNotificationsSize mirrors recentResultsSize for emitted
-// notifications. Phase 1 only writes selfcheck alerts to this
-// buffer (the main app's notification flow still lives in the main
-// app); the buffer is shape-stable so future phases can append
-// without changing the page contract.
+// notifications.
 const recentNotificationsSize = 100
 
 // ResultRow is one row from the worker's in-memory result ring
@@ -35,9 +32,9 @@ type ResultRow struct {
 	At         time.Time
 }
 
-// NotificationRow is one row from the worker's notification ring
-// buffer. Phase 1 only fills this from selfcheck alerts; the row
-// shape is forward-compatible with main-app-issued notifications.
+// NotificationRow is one row from the worker's notification ring buffer.
+// Delegated rows use only JobID, Method, Status, DurationMs, and At. They
+// deliberately omit delivery inputs and provider output.
 type NotificationRow struct {
 	Kind    string // "email", "telegram_private", "telegram_group"
 	Channel string
@@ -46,6 +43,11 @@ type NotificationRow struct {
 	OK      bool
 	Error   string
 	At      time.Time
+
+	JobID      string
+	Method     string
+	Status     string
+	DurationMs int
 }
 
 // resultBuffer is a thread-safe FIFO ring buffer of ResultRow. The

internal/distworker/runner.go

@@ -25,6 +25,8 @@ const (
 	// minQueueCapacity is the lower bound for the bounded job/result
 	// channels so that a small pool still has some backpressure headroom.
 	minQueueCapacity = 16
+
+	malformedTaskEnvelopeError = "malformed_task_envelope"
 )
 
 // jobPool is the minimal interface the Runner needs from a worker pool. It
@@ -43,6 +45,11 @@ type resultEnvelope struct {
 	reports []wire.CheckResultReport
 }
 
+type metricEnvelope struct {
+	generation uint64
+	report     wire.ServerMetricReport
+}
+
 // Runner manages the worker execution loop.
 //
 // Concurrency design:
@@ -61,23 +68,32 @@ type resultEnvelope struct {
 //     websocket writes thread-safe and removes the per-task goroutine
 //     that previously blocked pool execution while holding the mutex.
 type Runner struct {
-	config         *Config
-	client         *Client
-	pool           jobPool
-	concurrency    int64
-	maxConcurrency int
-	jobQueue       chan wire.CheckJob
-	results        chan resultEnvelope
-	notifyQueue    chan wire.NotificationTask
-	notifyResults  chan notifyResultEnvelope
-	metricResults  chan wire.ServerMetricReport
-	stopCh         chan struct{}
-	wg             sync.WaitGroup
-	started        atomic.Bool
-	queueDepth     int64
-	activeCount    int64
-	notifyDepth    int64
-	notifyActive   int64
+	config           *Config
+	client           *Client
+	pool             jobPool
+	concurrency      int64
+	maxConcurrency   int
+	jobQueue         chan wire.CheckJob
+	results          chan resultEnvelope
+	notifyQueue      chan wire.NotificationTask
+	notifyResults    chan notifyResultEnvelope
+	metricResults    chan metricEnvelope
+	stopCh           chan struct{}
+	wg               sync.WaitGroup
+	controlWG        sync.WaitGroup
+	backgroundWG     sync.WaitGroup
+	lifecycleMu      sync.Mutex
+	controlCtx       context.Context
+	controlCancel    context.CancelFunc
+	backgroundCtx    context.Context
+	backgroundCancel context.CancelFunc
+	rotationCtx      context.Context
+	rotationCancel   context.CancelFunc
+	started          atomic.Bool
+	queueDepth       int64
+	activeCount      int64
+	notifyDepth      int64
+	notifyActive     int64
 
 	// credentialsMu guards credentials during init/config refresh.
 	credentialsMu sync.RWMutex
@@ -121,15 +137,15 @@ type Runner struct {
 	masterStatusUp *bool
 	masterStatusAt time.Time
 
-	// selfcheckCancel terminates the periodic selfcheck goroutine started
-	// by Start(). Nil until Start runs.
-	selfcheckCancel context.CancelFunc
-
 	// executor is the function the pool runs for each job. It is a
 	// field so tests can swap it for a deterministic stub without
 	// touching the websocket plumbing.
 	executor func(payload interface{}) interface{}
 
+	// notificationExecutor is a test seam for the notification task
+	// boundary. Production uses ExecuteNotification's normal delivery path.
+	notificationExecutor func(context.Context, models.Task) wire.NotificationResultReport
+
 	// resultsBuf holds the most recent result rows. The webapp
 	// reads from it via RecentResults(n). Cleared by Stop so the
 	// ring does not leak between worker runs.
@@ -152,26 +168,40 @@ type Runner struct {
 	// workerID / regionCode / workerVersion / workerCaps are
 	// captured from the most recent init/config websocket message
 	// so the webapp can render them read-only.
-	workerIDMu    sync.RWMutex
-	workerID      string
-	regionMu      sync.RWMutex
-	regionCode    string
-	versionMu     sync.RWMutex
-	workerVersion string
-	capsMu        sync.RWMutex
-	workerCaps    []string
-	serverID      atomic.Int64
+	workerIDMu           sync.RWMutex
+	workerID             string
+	regionMu             sync.RWMutex
+	regionCode           string
+	versionMu            sync.RWMutex
+	workerVersion        string
+	capsMu               sync.RWMutex
+	workerCaps           []string
+	serverID             atomic.Int64
+	metricGeneration     atomic.Uint64
+	nextMetricGeneration atomic.Uint64
 
-	// clientMu guards swap of the http/websocket client during
-	// token rotation. The websocket loop reads r.client under the
-	// lock; RotateToken swaps a fresh client in under the lock
-	// before closing the old connection.
-	clientMu sync.Mutex
+	// clientMu guards the control-plane client and its active websocket.
+	// Rotation swaps the client and closes only controlConn, leaving the
+	// runner's global stop signal and local subsystems untouched.
+	clientMu       sync.Mutex
+	controlConn    *websocket.Conn
+	controlWriteMu *sync.Mutex
+	reconnectCh    chan struct{}
+	rotationMu     sync.Mutex
 
-	// closeOnce guards Close against double-close on the websocket
-	// from RotateToken. Phase 1 has a single websocket; RotateToken
-	// closes it so the reconnect loop picks up the new token.
-	closeOnce sync.Once
+	// outbox holds messages removed from a per-connection writer queue that
+	// could not be written before its websocket closed.
+	outboxMu   sync.Mutex
+	outbox     []wire.WorkerMessage
+	outboxWake chan struct{}
+
+	// beforeControlWrite is a test seam used to hold a dequeued message while
+	// a connection rotates.
+	beforeControlWrite func()
+
+	// beforeTokenCommit lets tests make shutdown win between a successful HTTP
+	// rotation response and the lifecycle-protected in-memory commit.
+	beforeTokenCommit func()
 }
 
 // NewRunner creates a new worker runner. Config is taken by pointer to
@@ -183,10 +213,21 @@ func NewRunner(cfg *Config) *Runner {
 	if maxConc <= 0 {
 		maxConc = DefaultMaxConcurrency
 	}
+	controlCtx, controlCancel := context.WithCancel(context.Background())
+	backgroundCtx, backgroundCancel := context.WithCancel(context.Background())
+	rotationCtx, rotationCancel := context.WithCancel(context.Background())
 	return &Runner{
 		config:           cfg,
 		maxConcurrency:   maxConc,
 		stopCh:           make(chan struct{}),
+		reconnectCh:      make(chan struct{}, 1),
+		outboxWake:       make(chan struct{}, 1),
+		controlCtx:       controlCtx,
+		controlCancel:    controlCancel,
+		backgroundCtx:    backgroundCtx,
+		backgroundCancel: backgroundCancel,
+		rotationCtx:      rotationCtx,
+		rotationCancel:   rotationCancel,
 		resultsBuf:       newResultBuffer(),
 		notificationsBuf: newNotificationBuffer(),
 		peerCache:        newPeerCache(),
@@ -196,23 +237,32 @@ func NewRunner(cfg *Config) *Runner {
 // Start begins the worker execution
 func (r *Runner) Start() error {
 	log.Println("worker: starting...")
+	r.lifecycleMu.Lock()
 
 	if r.config.URL == "" || r.config.Token == "" {
+		r.lifecycleMu.Unlock()
 		return fmt.Errorf("RSMON_URL and RSMON_TOKEN must be set")
 	}
 
 	if !r.started.CompareAndSwap(false, true) {
+		r.lifecycleMu.Unlock()
 		return fmt.Errorf("worker: runner already started")
 	}
+	if r.stopped() {
+		r.lifecycleMu.Unlock()
+		return fmt.Errorf("worker: runner stopped")
+	}
 
+	r.clientMu.Lock()
 	r.client = NewClient(r.config.URL, r.config.Token)
+	r.clientMu.Unlock()
 
 	queueCap := r.queueCapacity()
 	r.jobQueue = make(chan wire.CheckJob, queueCap)
 	r.results = make(chan resultEnvelope, queueCap)
 	r.notifyQueue = make(chan wire.NotificationTask, queueCap)
 	r.notifyResults = make(chan notifyResultEnvelope, queueCap)
-	r.metricResults = make(chan wire.ServerMetricReport, queueCap)
+	r.metricResults = make(chan metricEnvelope, queueCap)
 
 	if r.executor == nil {
 		r.executor = r.defaultExecuteJob
@@ -237,31 +287,48 @@ func (r *Runner) Start() error {
 		go r.notifyDispatcher()
 	}
 
-	// Start websocket task loop
-	go r.websocketLoop()
-	go r.serverMetricLoop()
+	// Start websocket task loop.
+	r.controlWG.Add(1)
+	go func() {
+		defer r.controlWG.Done()
+		r.websocketLoop()
+	}()
+	r.backgroundWG.Add(1)
+	go func() {
+		defer r.backgroundWG.Done()
+		r.serverMetricLoop(r.backgroundCtx)
+	}()
 
 	// Start periodic selfcheck loop. This probes the main API and, on
 	// sustained unreachability, notifies system contacts directly via
 	// the cached credentials. Lifetimes of selfcheck goroutines are
 	// bound to stopCh (and the explicit cancel, kept for symmetry).
-	selfcheckCtx, selfcheckCancel := context.WithCancel(context.Background())
-	r.selfcheckCancel = selfcheckCancel
-	go r.startSelfcheck(selfcheckCtx)
+	r.backgroundWG.Add(1)
+	go func() {
+		defer r.backgroundWG.Done()
+		r.startSelfcheck(r.backgroundCtx)
+	}()
 
 	// Start the peer poller. It refreshes r.peerCache with each
 	// peer's latest /api/peer/status verdict. The selfcheck
 	// consumes the cache to drive consensus. The poller is bound
 	// to selfcheckCtx so it shuts down together with the
 	// selfcheck loop on Stop.
-	go r.peerPollerLoop(selfcheckCtx)
+	r.backgroundWG.Add(1)
+	go func() {
+		defer r.backgroundWG.Done()
+		r.peerPollerLoop(r.backgroundCtx)
+	}()
+	r.lifecycleMu.Unlock()
 
 	// Wait for stop signal
 	<-r.stopCh
 	log.Println("worker: shutting down...")
-	if r.selfcheckCancel != nil {
-		r.selfcheckCancel()
+	if r.backgroundCancel != nil {
+		r.backgroundCancel()
 	}
+	r.controlWG.Wait()
+	r.backgroundWG.Wait()
 
 	// Dispatchers exit via stopCh. Do not close jobQueue here: the websocket
 	// reader can still be unwinding and may otherwise race with a send.
@@ -271,21 +338,42 @@ func (r *Runner) Start() error {
 	// otherwise an in-flight pool.Process would panic.
 	r.pool.Close()
 
-	// Close results so any future writers exit promptly. (At this point
-	// the websocket connection is also gone, so this is just defensive.)
-	close(r.results)
-
 	return nil
 }
 
 // Stop gracefully stops the worker
 func (r *Runner) Stop() {
+	r.lifecycleMu.Lock()
 	select {
 	case <-r.stopCh:
 		// already closed
 	default:
 		close(r.stopCh)
 	}
+	if r.controlCancel != nil {
+		r.controlCancel()
+	}
+	if r.backgroundCancel != nil {
+		r.backgroundCancel()
+	}
+	if r.rotationCancel != nil {
+		r.rotationCancel()
+	}
+	r.clientMu.Lock()
+	conn := r.controlConn
+	r.clientMu.Unlock()
+	if conn != nil {
+		_ = conn.Close()
+	}
+	r.lifecycleMu.Unlock()
+
+	// Cancel and close before waiting for rotation. A rotation can be waiting
+	// on controlWriteMu while a blocked writer needs that close to return.
+	r.rotationMu.Lock()
+	r.rotationMu.Unlock()
+	r.controlWG.Wait()
+	r.backgroundWG.Wait()
+	r.wg.Wait()
 }
 
 // Enqueue submits a job to the worker pool. It returns false if the runner
@@ -295,6 +383,9 @@ func (r *Runner) Enqueue(job wire.CheckJob) bool { //nolint:lll,gocritic // wire
 	if r.jobQueue == nil {
 		return false
 	}
+	if r.stopped() {
+		return false
+	}
 	select {
 	case r.jobQueue <- job:
 		atomic.AddInt64(&r.queueDepth, 1)
@@ -318,6 +409,9 @@ func (r *Runner) EnqueueNotification(task wire.NotificationTask) bool {
 	if r.notifyQueue == nil {
 		return false
 	}
+	if r.stopped() {
+		return false
+	}
 	select {
 	case r.notifyQueue <- task:
 		atomic.AddInt64(&r.notifyDepth, 1)
@@ -350,23 +444,30 @@ func (r *Runner) notifyDispatcher() {
 // shape, runs the executor, and pushes the result into notifyResults. The
 // writer goroutine picks it up and serializes the websocket write.
 func (r *Runner) executeAndForwardNotification(task wire.NotificationTask) { //nolint:gocritic // wire payload is shared
-	if r.notifyResults == nil {
-		return
-	}
-	deadline := time.Now().Add(models.DefaultNotificationExecutionTimeout)
+	started := time.Now()
+	deadline := started.Add(models.DefaultNotificationExecutionTimeout)
+	var taskDeadline *time.Time
 	if task.Deadline != nil {
-		if taskDeadline, err := time.Parse(time.RFC3339Nano, *task.Deadline); err == nil && taskDeadline.Before(deadline) {
-			deadline = taskDeadline
+		parsedDeadline, err := time.Parse(time.RFC3339Nano, *task.Deadline)
+		if err != nil {
+			r.completeNotification(task, notificationPermanentReport(task, "invalid notification deadline", started))
+			return
+		}
+		taskDeadline = &parsedDeadline
+		if !parsedDeadline.After(started) {
+			r.completeNotification(task, notificationPermanentReport(task, "notification deadline expired", started))
+			return
+		}
+		if parsedDeadline.Before(deadline) {
+			deadline = parsedDeadline
 		}
 	}
 	ctx, cancel := context.WithDeadline(context.Background(), deadline)
 	defer cancel()
 	payload, _ := json.Marshal(task)
 	dbTask := models.Task{JobID: task.JobID, LeaseToken: task.LeaseToken, Payload: payload}
-	if task.Deadline != nil {
-		if deadline, err := time.Parse(time.RFC3339Nano, *task.Deadline); err == nil {
-			dbTask.Deadline = &deadline
-		}
+	if taskDeadline != nil {
+		dbTask.Deadline = taskDeadline
 	}
 	if task.MessageID != 0 {
 		msgID := task.MessageID
@@ -377,15 +478,33 @@ func (r *Runner) executeAndForwardNotification(task wire.NotificationTask) { //n
 		_ = ev
 	}
 	report := r.ExecuteNotification(ctx, dbTask)
-	env := notifyResultEnvelope{task: task, report: report}
+	r.forwardNotificationResult(task, report)
+}
 
-	select {
-	case <-r.stopCh:
+func notificationPermanentReport(task wire.NotificationTask, message string, started time.Time) wire.NotificationResultReport {
+	return wire.NotificationResultReport{
+		JobID:      task.JobID,
+		LeaseToken: task.LeaseToken,
+		MessageID:  task.MessageID,
+		Status:     wire.NotificationResultPermanent,
+		Error:      stringPtr(message),
+		DurationMs: int(time.Since(started) / time.Millisecond),
+	}
+}
+
+// completeNotification records the terminal local outcome before forwarding
+// the result. Rejected tasks use this path because they do not enter the executor.
+func (r *Runner) completeNotification(task wire.NotificationTask, report wire.NotificationResultReport) {
+	r.recordDelegatedNotification(report.JobID, task.Method, report)
+	r.forwardNotificationResult(task, report)
+}
+
+func (r *Runner) forwardNotificationResult(task wire.NotificationTask, report wire.NotificationResultReport) {
+	if r.notifyResults == nil || r.stopped() {
 		return
-	default:
 	}
 	select {
-	case r.notifyResults <- env:
+	case r.notifyResults <- notifyResultEnvelope{task: task, report: report}:
 	case <-r.stopCh:
 	}
 }
@@ -482,6 +601,7 @@ func (r *Runner) websocketLoop() {
 
 		select {
 		case <-time.After(3 * time.Second):
+		case <-r.reconnectCh:
 		case <-r.stopCh:
 			return
 		}
@@ -489,29 +609,75 @@ func (r *Runner) websocketLoop() {
 }
 
 func (r *Runner) runWebsocket() error {
-	conn, err := r.client.WorkerSocket()
+	var writeMu sync.Mutex
+	r.clientMu.Lock()
+	client := r.client
+	r.clientMu.Unlock()
+	if client == nil {
+		return fmt.Errorf("worker: client not initialized")
+	}
+
+	conn, err := client.WorkerSocketContext(r.controlCtx)
 	if err != nil {
 		return err
 	}
-	defer conn.Close() //nolint:errcheck
+
+	// A rotation can complete while the websocket dial is in flight. Do not
+	// install a connection authenticated with the superseded token.
+	r.clientMu.Lock()
+	if r.client != client || r.stopped() {
+		r.clientMu.Unlock()
+		_ = conn.Close()
+		return nil
+	}
+	r.controlConn = conn
+	r.controlWriteMu = &writeMu
+	metricGeneration := r.nextMetricGeneration.Add(1)
+	r.metricGeneration.Store(metricGeneration)
+	r.clientMu.Unlock()
+	defer func() {
+		r.clientMu.Lock()
+		if r.controlConn == conn {
+			r.controlConn = nil
+			r.controlWriteMu = nil
+			r.metricGeneration.CompareAndSwap(metricGeneration, 0)
+		}
+		r.clientMu.Unlock()
+		_ = conn.Close()
+	}()
 	log.Println("worker: websocket connected")
 
-	var writeMu sync.Mutex
 	done := make(chan struct{})
+	var doneOnce sync.Once
+	closeDone := func() { doneOnce.Do(func() { close(done) }) }
+	var connectionWG sync.WaitGroup
+	defer func() {
+		closeDone()
+		_ = conn.Close()
+		connectionWG.Wait()
+	}()
 
 	// Heartbeat goroutine — shares writeMu with the writer.
-	go r.heartbeat(conn, &writeMu, done)
+	connectionWG.Add(1)
+	go func() {
+		defer connectionWG.Done()
+		r.heartbeat(conn, &writeMu, done)
+	}()
 
 	// Single writer goroutine for this connection: serializes result
 	// and heartbeat writes through writeMu so websocket.WriteJSON is
 	// never called concurrently. The dispatcher loop feeds it via the
 	// bounded results channel.
-	go r.writer(conn, &writeMu, done)
+	connectionWG.Add(1)
+	go func() {
+		defer connectionWG.Done()
+		r.writer(conn, &writeMu, done, metricGeneration)
+	}()
 
 	for {
 		var msg wire.WorkerMessage
 		if err := conn.ReadJSON(&msg); err != nil {
-			close(done)
+			closeDone()
 			return err
 		}
 		if (msg.Kind == "init" || msg.Kind == "config") && msg.Init != nil {
@@ -522,32 +688,113 @@ func (r *Runner) runWebsocket() error {
 			continue
 		}
 		if !r.enqueueTaskMessage(msg) {
-			close(done)
+			closeDone()
 			return nil
 		}
 	}
 }
 
-// enqueueTaskMessage prefers the v2 envelope over sibling legacy fields. Some
-// rollout frames contain both check representations; executing the first match
-// only keeps a current runner from running one check twice.
+// enqueueTaskMessage prefers the v2 envelope over sibling legacy fields. An
+// envelope is accepted only when it selects exactly one matching payload with
+// the same non-empty outer and inner job IDs. Invalid envelopes never fall
+// back to a sibling legacy payload, which could otherwise execute a task the
+// control plane did not intend to send.
 func (r *Runner) enqueueTaskMessage(msg wire.WorkerMessage) bool { //nolint:gocritic // wire envelope is the dispatcher boundary
+	if msg.TaskEnvelope != nil {
+		return r.enqueueTaskEnvelope(msg.TaskEnvelope)
+	}
+
 	switch {
-	case msg.TaskEnvelope != nil && msg.TaskEnvelope.Type == wire.TaskTypeNotification && msg.TaskEnvelope.Notify != nil:
-		return r.EnqueueNotification(*msg.TaskEnvelope.Notify)
-	case msg.TaskEnvelope != nil && msg.TaskEnvelope.Type == wire.TaskTypeCheck && msg.TaskEnvelope.Job != nil:
-		return r.Enqueue(*msg.TaskEnvelope.Job)
 	case msg.NotificationTask != nil:
+		if msg.NotificationTask.JobID == "" || msg.NotificationTask.LeaseToken == "" {
+			return true
+		}
 		log.Printf("worker: received websocket notification task %s method=%s", msg.NotificationTask.JobID, msg.NotificationTask.Method)
 		return r.EnqueueNotification(*msg.NotificationTask)
 	case msg.Task != nil:
+		if msg.Task.JobID == "" || msg.Task.LeaseToken == "" {
+			return true
+		}
 		log.Printf("worker: received websocket task %s", msg.Task.JobID)
+		if !checkexec.SupportsKind(msg.Task.Kind) {
+			return r.enqueueUnsupportedCheck(*msg.Task)
+		}
 		return r.Enqueue(*msg.Task)
 	default:
 		return true
 	}
 }
 
+func (r *Runner) enqueueTaskEnvelope(envelope *wire.TaskEnvelope) bool {
+	if (envelope.Job == nil) == (envelope.Notify == nil) {
+		return true
+	}
+	if envelope.Job != nil {
+		if !matchingEnvelopeJobID(envelope.JobID, envelope.Job.JobID) || envelope.Job.LeaseToken == "" {
+			return true
+		}
+		if envelope.Type != wire.TaskTypeCheck {
+			return r.enqueueFailedCheck(*envelope.Job, malformedTaskEnvelopeError)
+		}
+		if !checkexec.SupportsKind(envelope.Job.Kind) {
+			return r.enqueueUnsupportedCheck(*envelope.Job)
+		}
+		return r.Enqueue(*envelope.Job)
+	}
+	if !matchingEnvelopeJobID(envelope.JobID, envelope.Notify.JobID) || envelope.Notify.LeaseToken == "" {
+		return true
+	}
+	if envelope.Type != wire.TaskTypeNotification {
+		return r.enqueueFailedNotification(*envelope.Notify, malformedTaskEnvelopeError)
+	}
+	return r.EnqueueNotification(*envelope.Notify)
+}
+
+func matchingEnvelopeJobID(outer, inner string) bool {
+	return outer != "" && outer == inner
+}
+
+func (r *Runner) enqueueUnsupportedCheck(job wire.CheckJob) bool {
+	return r.enqueueFailedCheck(job, "unsupported_kind: "+job.Kind)
+}
+
+func (r *Runner) enqueueFailedCheck(job wire.CheckJob, errorCode string) bool {
+	if r.results == nil || r.stopped() {
+		return false
+	}
+	report := wire.CheckResultReport{
+		JobID:      job.JobID,
+		CheckID:    job.CheckID,
+		MonitorID:  job.MonitorID,
+		State:      "FAIL",
+		Error:      stringPtr(errorCode),
+		DurationMs: 0,
+	}
+	select {
+	case r.results <- resultEnvelope{job: job, reports: []wire.CheckResultReport{report}}:
+		return true
+	case <-r.stopCh:
+		return false
+	}
+}
+
+func (r *Runner) enqueueFailedNotification(task wire.NotificationTask, errorCode string) bool {
+	if r.stopped() {
+		return false
+	}
+	r.completeNotification(task, notificationPermanentReport(task, errorCode, time.Now()))
+	return true
+}
+
+func (r *Runner) stopped() bool {
+	select {
+	case <-r.stopCh:
+		return true
+	default:
+		return false
+	}
+}
+
 func (r *Runner) heartbeat(conn *websocket.Conn, writeMu *sync.Mutex, done <-chan struct{}) {
 	ticker := time.NewTicker(heartbeatInterval)
 	defer ticker.Stop()
@@ -601,23 +848,39 @@ func (r *Runner) LastHeartbeatAck() time.Time {
 	return r.lastHeartbeatAt
 }
 
-func (r *Runner) writer(conn *websocket.Conn, writeMu *sync.Mutex, done <-chan struct{}) {
+func (r *Runner) writer(conn *websocket.Conn, writeMu *sync.Mutex, done <-chan struct{}, metricGeneration uint64) {
 	for {
+		if message, ok := r.takeOutbox(); ok {
+			if err := r.writeControlMessage(conn, writeMu, message); err != nil {
+				r.requeueOutbox(message)
+				_ = conn.Close()
+				return
+			}
+			continue
+		}
+
 		select {
 		case env, ok := <-r.results:
 			if !ok {
 				return
 			}
-			writeMu.Lock()
 			for i := range env.reports {
 				report := &env.reports[i]
 				report.LeaseToken = env.job.LeaseToken
-				if err := conn.WriteJSON(wire.WorkerMessage{Kind: "result", Result: report}); err != nil {
+				message := wire.WorkerMessage{Kind: "result", Result: report}
+				if err := r.writeControlMessage(conn, writeMu, message); err != nil {
+					remaining := []wire.WorkerMessage{message}
+					for j := i + 1; j < len(env.reports); j++ {
+						next := env.reports[j]
+						next.LeaseToken = env.job.LeaseToken
+						remaining = append(remaining, wire.WorkerMessage{Kind: "result", Result: &next})
+					}
+					r.requeueOutbox(remaining...)
 					log.Printf(
 						"worker: failed to report result job=%s check=%d kind=%s state=%s: %v",
 						report.JobID, report.CheckID, env.job.Kind, report.State, err,
 					)
-					writeMu.Unlock()
+					_ = conn.Close()
 					return
 				}
 				log.Printf(
@@ -626,32 +889,37 @@ func (r *Runner) writer(conn *websocket.Conn, writeMu *sync.Mutex, done <-chan s
 				)
 				r.resultsBuf.add(resultRowFromReport(&env, report, time.Now().UTC()))
 			}
-			writeMu.Unlock()
 		case env, ok := <-r.notifyResults:
 			if !ok {
 				return
 			}
-			writeMu.Lock()
-			if err := conn.WriteJSON(wire.WorkerMessage{Kind: "result", NotificationResult: &env.report}); err != nil {
+			message := wire.WorkerMessage{Kind: "result", NotificationResult: &env.report}
+			if err := r.writeControlMessage(conn, writeMu, message); err != nil {
+				r.requeueOutbox(message)
 				log.Printf(
 					"worker: failed to report notification result job=%s method=%s status=%s: %v",
 					env.report.JobID, env.task.Method, env.report.Status, err,
 				)
-				writeMu.Unlock()
+				_ = conn.Close()
 				return
 			}
 			log.Printf(
 				"worker: completed notification job=%s method=%s status=%s message=%d duration_ms=%d",
 				env.report.JobID, env.task.Method, env.report.Status, env.task.MessageID, env.report.DurationMs,
 			)
-			writeMu.Unlock()
-		case report := <-r.metricResults:
-			writeMu.Lock()
-			if err := conn.WriteJSON(wire.WorkerMessage{Kind: "result", ServerMetric: &report}); err != nil {
-				writeMu.Unlock()
+		case metric := <-r.metricResults:
+			if metric.generation != metricGeneration {
+				continue
+			}
+			message := wire.WorkerMessage{Kind: "result", ServerMetric: &metric.report}
+			if err := r.writeControlMessage(conn, writeMu, message); err != nil {
+				// Metrics are periodic snapshots without a lease or idempotency key.
+				// Dropping a failed snapshot avoids duplicate control-plane inserts;
+				// the next collection tick supplies a fresh replacement.
+				_ = conn.Close()
 				return
 			}
-			writeMu.Unlock()
+		case <-r.outboxWake:
 		case <-done:
 			return
 		case <-r.stopCh:
@@ -660,6 +928,39 @@ func (r *Runner) writer(conn *websocket.Conn, writeMu *sync.Mutex, done <-chan s
 	}
 }
 
+func (r *Runner) takeOutbox() (wire.WorkerMessage, bool) {
+	r.outboxMu.Lock()
+	defer r.outboxMu.Unlock()
+	if len(r.outbox) == 0 {
+		return wire.WorkerMessage{}, false
+	}
+	message := r.outbox[0]
+	r.outbox = r.outbox[1:]
+	return message, true
+}
+
+func (r *Runner) requeueOutbox(messages ...wire.WorkerMessage) {
+	if len(messages) == 0 {
+		return
+	}
+	r.outboxMu.Lock()
+	r.outbox = append(messages, r.outbox...)
+	r.outboxMu.Unlock()
+	select {
+	case r.outboxWake <- struct{}{}:
+	default:
+	}
+}
+
+func (r *Runner) writeControlMessage(conn *websocket.Conn, writeMu *sync.Mutex, message wire.WorkerMessage) error {
+	writeMu.Lock()
+	defer writeMu.Unlock()
+	if r.beforeControlWrite != nil {
+		r.beforeControlWrite()
+	}
+	return conn.WriteJSON(message)
+}
+
 func (r *Runner) applyInit(init *wire.WorkerInit) {
 	if init.Concurrency > 0 && init.Concurrency != r.Concurrency() {
 		size := init.Concurrency
@@ -881,9 +1182,6 @@ func (r *Runner) RecentResults(n int) []ResultRow {
 }
 
 // RecentNotifications returns the most recent n notification rows.
-// Phase 1 only fills this buffer from selfcheck alerts via
-// RecordNotification; the main-app-issued notifications still live
-// in the main app's database.
 func (r *Runner) RecentNotifications(n int) []NotificationRow {
 	if r == nil || r.notificationsBuf == nil {
 		return nil
@@ -892,8 +1190,7 @@ func (r *Runner) RecentNotifications(n int) []NotificationRow {
 }
 
 // RecordNotification appends one row to the notification ring buffer.
-// Called from selfcheck.sendSystemAlert so the webapp /notifications
-// page can show what the worker emitted. Safe before Start.
+// Called from selfcheck.sendSystemAlert. Safe before Start.
 func (r *Runner) RecordNotification(n *NotificationRow) {
 	if r == nil || r.notificationsBuf == nil {
 		return
@@ -904,12 +1201,24 @@ func (r *Runner) RecordNotification(n *NotificationRow) {
 	r.notificationsBuf.add(n)
 }
 
+func (r *Runner) recordDelegatedNotification(jobID, method string, report wire.NotificationResultReport) {
+	r.RecordNotification(&NotificationRow{
+		JobID:      jobID,
+		Method:     method,
+		Status:     report.Status,
+		DurationMs: report.DurationMs,
+		At:         time.Now().UTC(),
+	})
+}
+
 // Token returns the current bearer token. The webapp settings page
 // masks this for display.
 func (r *Runner) Token() string {
 	if r == nil || r.config == nil {
 		return ""
 	}
+	r.clientMu.Lock()
+	defer r.clientMu.Unlock()
 	return r.config.Token
 }
 
@@ -983,48 +1292,95 @@ func (r *Runner) HTTPConfig() HTTPConfig {
 //
 // Returns the new token string. On any failure the old token and
 // client are kept untouched.
-func (r *Runner) RotateToken(_ context.Context) (string, error) {
+func (r *Runner) RotateToken(ctx context.Context) (string, error) {
 	if r == nil || r.config == nil {
 		return "", fmt.Errorf("worker: runner not initialized")
 	}
-	if r.client == nil {
+
+	// The endpoint authenticates with the current token, so concurrent
+	// rotations must not mint replacements from the same stale client.
+	r.rotationMu.Lock()
+	defer r.rotationMu.Unlock()
+	if r.stopped() {
+		return "", fmt.Errorf("worker: runner stopped")
+	}
+	requestCtx, cancelRequest := context.WithCancel(ctx)
+	rotationDone := make(chan struct{})
+	defer func() {
+		close(rotationDone)
+		cancelRequest()
+	}()
+	go func() {
+		select {
+		case <-r.rotationCtx.Done():
+			cancelRequest()
+		case <-rotationDone:
+		}
+	}()
+
+	r.clientMu.Lock()
+	client := r.client
+	r.clientMu.Unlock()
+	if client == nil {
 		return "", fmt.Errorf("worker: client not yet started")
 	}
-	newToken, err := r.client.RotateToken()
+	newToken, err := client.RotateToken(requestCtx)
 	if err != nil {
 		return "", err
 	}
+	if r.beforeTokenCommit != nil {
+		r.beforeTokenCommit()
+	}
+
+	// Stop and the post-HTTP commit share lifecycleMu. Once Stop has closed
+	// stopCh, this rotation cannot install or report a replacement token.
+	r.lifecycleMu.Lock()
+	r.clientMu.Lock()
+	if r.stopped() {
+		r.clientMu.Unlock()
+		r.lifecycleMu.Unlock()
+		return "", fmt.Errorf("worker: runner stopped")
+	}
 	if newToken == "" || newToken == r.config.Token {
+		r.clientMu.Unlock()
+		r.lifecycleMu.Unlock()
 		return "", fmt.Errorf("worker: rotate-token returned unchanged or empty token")
 	}
 
-	// Swap config + client under lock so a concurrent heartbeat
-	// cannot race with the rotation.
-	r.clientMu.Lock()
+	// Swap client and detach only the active control-plane connection. stopCh
+	// remains exclusively owned by Stop, so dispatch, metrics, selfcheck, peer
+	// polling, and services started alongside the runner keep running.
 	r.config.Token = newToken
-	oldClient := r.client
 	r.client = NewClient(r.config.URL, newToken)
+	conn := r.controlConn
+	writeMu := r.controlWriteMu
 	r.clientMu.Unlock()
+	r.lifecycleMu.Unlock()
 
 	// Stamp the rotation time so the settings page can show it.
 	r.tokenRotatedMu.Lock()
 	r.tokenRotatedAt = time.Now().UTC()
 	r.tokenRotatedMu.Unlock()
 
-	// Force the websocket loop to reconnect with the new token. The
-	// old connection's next heartbeat will fail with 401; closing
-	// the connection now shortens that window.
-	r.closeOnce.Do(func() {
-		// Close the underlying websocket by triggering the runner's
-		// normal stop path; the websocketLoop goroutine will reconnect
-		// after we re-arm stopCh. This is the cleanest way to drive
-		// the loop without exposing internals.
-		select {
-		case <-r.stopCh:
-		default:
-			close(r.stopCh)
-		}
-	})
-	_ = oldClient // client has no Close; the websocket layer owns it.
+	// Interrupt the current connection before waiting for its writer. A stalled
+	// WriteJSON holds writeMu; closing the socket makes that write fail so the
+	// writer can requeue its dequeued envelope and release the mutex.
+	if conn != nil {
+		_ = conn.Close()
+	}
+	if writeMu != nil {
+		writeMu.Lock()
+		writeMu.Unlock()
+	}
+	r.lifecycleMu.Lock()
+	stopped := r.stopped()
+	r.lifecycleMu.Unlock()
+	if stopped {
+		return "", fmt.Errorf("worker: runner stopped")
+	}
+	select {
+	case r.reconnectCh <- struct{}{}:
+	default:
+	}
 	return newToken, nil
 }

internal/distworker/runner_protocol_test.go

@@ -11,9 +11,162 @@ import (
 
 func TestEnqueueTaskMessagePrefersEnvelopeOverLegacyCheck(t *testing.T) {
 	r := &Runner{jobQueue: make(chan wire.CheckJob, 2), stopCh: make(chan struct{})}
-	message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{Type: wire.TaskTypeCheck, Job: &wire.CheckJob{JobID: "v2"}}, Task: &wire.CheckJob{JobID: "v1"}}
+	message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{Type: wire.TaskTypeCheck, JobID: "v2", Job: &wire.CheckJob{JobID: "v2", LeaseToken: "lease-2", Kind: "http"}}, Task: &wire.CheckJob{JobID: "v1"}}
 	require.True(t, r.enqueueTaskMessage(message))
 	job := <-r.jobQueue
 	assert.Equal(t, "v2", job.JobID)
 	assert.Empty(t, r.jobQueue, "the sibling legacy field must not run a second check")
 }
+
+func TestEnqueueTaskMessageRejectsMismatchedEnvelopeJobIDs(t *testing.T) {
+	r := &Runner{jobQueue: make(chan wire.CheckJob, 1), results: make(chan resultEnvelope, 1), stopCh: make(chan struct{})}
+	message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{
+		Type:  wire.TaskTypeCheck,
+		JobID: "outer-job",
+		Job:   &wire.CheckJob{JobID: "inner-job", LeaseToken: "lease-1", Kind: "http"},
+	}}
+
+	require.True(t, r.enqueueTaskMessage(message))
+	assert.Empty(t, r.jobQueue)
+	assert.Empty(t, r.results, "a task without an unambiguous job ID cannot be reported safely")
+}
+
+func TestEnqueueTaskMessageRejectsMissingEnvelopeJobID(t *testing.T) {
+	r := &Runner{jobQueue: make(chan wire.CheckJob, 1), results: make(chan resultEnvelope, 1), stopCh: make(chan struct{})}
+	message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{
+		Type: wire.TaskTypeCheck,
+		Job:  &wire.CheckJob{JobID: "inner-job", LeaseToken: "lease-1", Kind: "http"},
+	}}
+
+	require.True(t, r.enqueueTaskMessage(message))
+	assert.Empty(t, r.jobQueue)
+	assert.Empty(t, r.results, "a missing outer ID cannot be reported safely")
+}
+
+func TestEnqueueTaskMessageReportsMalformedCheckEnvelopeWithMatchingJobID(t *testing.T) {
+	r := &Runner{jobQueue: make(chan wire.CheckJob, 1), results: make(chan resultEnvelope, 1), stopCh: make(chan struct{})}
+	message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{
+		Type:  wire.TaskTypeNotification,
+		JobID: "job-1",
+		Job:   &wire.CheckJob{JobID: "job-1", LeaseToken: "lease-1", CheckID: 4, MonitorID: 5, Kind: "http"},
+	}}
+
+	require.True(t, r.enqueueTaskMessage(message))
+	assert.Empty(t, r.jobQueue)
+	env := <-r.results
+	require.Len(t, env.reports, 1)
+	assert.Equal(t, "job-1", env.reports[0].JobID)
+	assert.Equal(t, "FAIL", env.reports[0].State)
+	require.NotNil(t, env.reports[0].Error)
+	assert.Equal(t, malformedTaskEnvelopeError, *env.reports[0].Error)
+	assert.Empty(t, r.results)
+}
+
+func TestEnqueueTaskMessageRejectsAmbiguousEnvelope(t *testing.T) {
+	r := &Runner{
+		jobQueue:      make(chan wire.CheckJob, 1),
+		results:       make(chan resultEnvelope, 1),
+		notifyQueue:   make(chan wire.NotificationTask, 1),
+		notifyResults: make(chan notifyResultEnvelope, 1),
+		stopCh:        make(chan struct{}),
+	}
+	message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{
+		Type:   wire.TaskTypeCheck,
+		JobID:  "job-1",
+		Job:    &wire.CheckJob{JobID: "job-1", LeaseToken: "lease-1", Kind: "http"},
+		Notify: &wire.NotificationTask{JobID: "job-1", LeaseToken: "lease-1"},
+	}}
+
+	require.True(t, r.enqueueTaskMessage(message))
+	assert.Empty(t, r.jobQueue)
+	assert.Empty(t, r.results)
+	assert.Empty(t, r.notifyQueue)
+	assert.Empty(t, r.notifyResults)
+}
+
+func TestEnqueueTaskMessageReportsUnsupportedCheckKind(t *testing.T) {
+	r := &Runner{jobQueue: make(chan wire.CheckJob, 1), results: make(chan resultEnvelope, 1), stopCh: make(chan struct{})}
+	message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{
+		Type:  wire.TaskTypeCheck,
+		JobID: "job-1",
+		Job:   &wire.CheckJob{JobID: "job-1", LeaseToken: "lease-1", CheckID: 4, MonitorID: 5, Kind: "rkn"},
+	}}
+
+	require.True(t, r.enqueueTaskMessage(message))
+	assert.Empty(t, r.jobQueue, "unsupported work must never reach the executor")
+	env := <-r.results
+	require.Len(t, env.reports, 1)
+	report := env.reports[0]
+	assert.Equal(t, "job-1", report.JobID)
+	assert.Equal(t, int64(4), report.CheckID)
+	assert.Equal(t, int64(5), report.MonitorID)
+	assert.Equal(t, "FAIL", report.State)
+	require.NotNil(t, report.Error)
+	assert.Equal(t, "unsupported_kind: rkn", *report.Error)
+	assert.Empty(t, r.results, "each rejected task must generate one terminal result")
+}
+
+func TestEnqueueTaskMessageReportsMalformedNotificationForInvalidType(t *testing.T) {
+	r := &Runner{notifyQueue: make(chan wire.NotificationTask, 1), notifyResults: make(chan notifyResultEnvelope, 1), stopCh: make(chan struct{})}
+	message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{
+		Type:   "unknown",
+		JobID:  "job-1",
+		Notify: &wire.NotificationTask{JobID: "job-1", LeaseToken: "lease-1", MessageID: 9, Method: "email"},
+	}}
+
+	require.True(t, r.enqueueTaskMessage(message))
+	assert.Empty(t, r.notifyQueue)
+	env := <-r.notifyResults
+	assert.Equal(t, "job-1", env.report.JobID)
+	assert.Equal(t, "lease-1", env.report.LeaseToken)
+	assert.Equal(t, int64(9), env.report.MessageID)
+	assert.Equal(t, wire.NotificationResultPermanent, env.report.Status)
+	require.NotNil(t, env.report.Error)
+	assert.Equal(t, malformedTaskEnvelopeError, *env.report.Error)
+	assert.Empty(t, r.notifyResults)
+}
+
+func TestEnqueueTaskMessageRejectsEmptyLeaseWithoutSideEffects(t *testing.T) {
+	cases := []struct {
+		name    string
+		message wire.WorkerMessage
+	}{
+		{
+			name: "envelope check",
+			message: wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{
+				Type: wire.TaskTypeCheck, JobID: "job-1", Job: &wire.CheckJob{JobID: "job-1", Kind: "http"},
+			}},
+		},
+		{
+			name: "envelope notification",
+			message: wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{
+				Type: wire.TaskTypeNotification, JobID: "job-1", Notify: &wire.NotificationTask{JobID: "job-1", Method: "email"},
+			}},
+		},
+		{
+			name:    "legacy check",
+			message: wire.WorkerMessage{Kind: "task", Task: &wire.CheckJob{JobID: "job-1", Kind: "http"}},
+		},
+		{
+			name:    "legacy notification",
+			message: wire.WorkerMessage{Kind: "task", NotificationTask: &wire.NotificationTask{JobID: "job-1", Method: "email"}},
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			r := &Runner{
+				jobQueue:      make(chan wire.CheckJob, 1),
+				results:       make(chan resultEnvelope, 1),
+				notifyQueue:   make(chan wire.NotificationTask, 1),
+				notifyResults: make(chan notifyResultEnvelope, 1),
+				stopCh:        make(chan struct{}),
+			}
+
+			require.True(t, r.enqueueTaskMessage(tc.message))
+			assert.Empty(t, r.jobQueue)
+			assert.Empty(t, r.results)
+			assert.Empty(t, r.notifyQueue)
+			assert.Empty(t, r.notifyResults)
+		})
+	}
+}

internal/distworker/runner_test.go

@@ -1,6 +1,9 @@
 package distworker
 
 import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
 	"strconv"
 	"sync"
 	"sync/atomic"
@@ -8,6 +11,7 @@ import (
 	"time"
 
 	"github.com/Jeffail/tunny"
+	"github.com/gorilla/websocket"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -174,6 +178,835 @@ func TestEnqueueRespectsBackpressure(t *testing.T) {
 	}
 }
 
+func TestStopRejectsTerminalResultsWithoutClosingResultChannel(t *testing.T) {
+	r := NewRunner(&Config{MaxConcurrency: 1})
+	r.results = make(chan resultEnvelope, 1)
+	r.Stop()
+
+	assert.False(t, r.enqueueFailedCheck(wire.CheckJob{JobID: "job-1", LeaseToken: "lease-1"}, "unsupported_kind: rkn"))
+	assert.Empty(t, r.results)
+
+	select {
+	case r.results <- resultEnvelope{}:
+	default:
+		t.Fatal("results channel should remain open after Stop")
+	}
+}
+
+func TestRotateTokenReconnectsWithoutStoppingRunner(t *testing.T) {
+	const (
+		oldToken = "old-token"
+		newToken = "new-token"
+	)
+
+	connections := make(chan string, 2)
+	results := make(chan wire.WorkerMessage, 1)
+	upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		switch req.URL.Path {
+		case "/api/internal/workers/rotate-token":
+			if req.Header.Get("Authorization") != "Bearer "+oldToken {
+				http.Error(w, "unexpected rotation token", http.StatusUnauthorized)
+				return
+			}
+			_ = json.NewEncoder(w).Encode(struct {
+				AuthToken string `json:"auth_token"`
+			}{AuthToken: newToken})
+		case "/worker":
+			conn, err := upgrader.Upgrade(w, req, nil)
+			if err != nil {
+				return
+			}
+			defer conn.Close()
+			token := req.URL.Query().Get("token")
+			connections <- token
+			if token == oldToken {
+				_, _, _ = conn.ReadMessage() // Rotation must close this connection.
+				return
+			}
+			if token != newToken {
+				return
+			}
+			if conn.WriteJSON(wire.WorkerMessage{Kind: "task", Task: &wire.CheckJob{
+				JobID: "after-rotation", LeaseToken: "lease", CheckID: 1, Kind: "http",
+			}}) != nil {
+				return
+			}
+			for {
+				var message wire.WorkerMessage
+				if err := conn.ReadJSON(&message); err != nil {
+					return
+				}
+				if message.Kind == "result" && message.Result != nil && message.Result.JobID == "after-rotation" {
+					results <- message
+					return
+				}
+			}
+		default:
+			http.NotFound(w, req)
+		}
+	}))
+	defer server.Close()
+
+	r := NewRunner(&Config{URL: server.URL, Token: oldToken, MaxConcurrency: 1})
+	r.executor = func(payload interface{}) interface{} {
+		job := payload.(wire.CheckJob)
+		return []wire.CheckResultReport{{JobID: job.JobID, CheckID: job.CheckID, State: "OK"}}
+	}
+	startDone := make(chan error, 1)
+	go func() { startDone <- r.Start() }()
+	t.Cleanup(func() {
+		r.Stop()
+		select {
+		case err := <-startDone:
+			require.NoError(t, err)
+		case <-time.After(time.Second):
+			t.Fatal("runner did not stop")
+		}
+	})
+
+	select {
+	case token := <-connections:
+		require.Equal(t, oldToken, token)
+	case <-time.After(time.Second):
+		t.Fatal("worker did not establish its initial control connection")
+	}
+
+	gotToken, err := r.RotateToken(t.Context())
+	require.NoError(t, err)
+	require.Equal(t, newToken, gotToken)
+	require.Equal(t, newToken, r.Token())
+
+	select {
+	case token := <-connections:
+		require.Equal(t, newToken, token)
+	case <-time.After(time.Second):
+		t.Fatal("worker did not reconnect with the replacement token")
+	}
+	select {
+	case result := <-results:
+		require.NotNil(t, result.Result)
+		assert.Equal(t, "OK", result.Result.State)
+	case <-time.After(time.Second):
+		t.Fatal("runner did not execute work after token rotation")
+	}
+	assert.False(t, r.stopped(), "rotation must not stop runner-owned subsystems")
+}
+
+func TestStopClosesAndJoinsIdleControlConnection(t *testing.T) {
+	closed := make(chan struct{}, 1)
+	connected := make(chan struct{}, 1)
+	upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		if req.URL.Path != "/worker" {
+			http.NotFound(w, req)
+			return
+		}
+		conn, err := upgrader.Upgrade(w, req, nil)
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+		connected <- struct{}{}
+		_, _, _ = conn.ReadMessage()
+		closed <- struct{}{}
+	}))
+	defer server.Close()
+
+	r := NewRunner(&Config{URL: server.URL, Token: "token", MaxConcurrency: 1})
+	startDone := make(chan error, 1)
+	go func() { startDone <- r.Start() }()
+	t.Cleanup(func() { r.Stop() })
+
+	select {
+	case <-time.After(time.Second):
+		t.Fatal("worker did not establish idle control connection")
+	case <-connected:
+	}
+	r.Stop()
+	select {
+	case <-closed:
+	case <-time.After(time.Second):
+		t.Fatal("Stop did not close the idle control connection")
+	}
+	select {
+	case err := <-startDone:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("Stop did not join the control loop")
+	}
+}
+
+func TestStopCancelsDialInProgress(t *testing.T) {
+	dialStarted := make(chan struct{})
+	allowUpgrade := make(chan struct{})
+	upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		if req.URL.Path != "/worker" {
+			http.NotFound(w, req)
+			return
+		}
+		close(dialStarted)
+		<-allowUpgrade
+		_, _ = upgrader.Upgrade(w, req, nil)
+	}))
+	defer func() {
+		close(allowUpgrade)
+		server.Close()
+	}()
+
+	r := NewRunner(&Config{URL: server.URL, Token: "token", MaxConcurrency: 1})
+	startDone := make(chan error, 1)
+	go func() { startDone <- r.Start() }()
+	select {
+	case <-dialStarted:
+	case <-time.After(time.Second):
+		t.Fatal("worker did not begin websocket dial")
+	}
+	r.Stop()
+	select {
+	case err := <-startDone:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("Stop did not join a canceled websocket dial")
+	}
+}
+
+func TestStartAndStopRegisterControlLoopSafely(t *testing.T) {
+	server := httptest.NewServer(http.NotFoundHandler())
+	defer server.Close()
+	for i := 0; i < 25; i++ {
+		r := NewRunner(&Config{URL: server.URL, Token: "token", MaxConcurrency: 1})
+		startDone := make(chan error, 1)
+		stopDone := make(chan struct{})
+		go func() { startDone <- r.Start() }()
+		go func() {
+			r.Stop()
+			close(stopDone)
+		}()
+		select {
+		case <-stopDone:
+		case <-time.After(time.Second):
+			t.Fatal("Stop did not complete")
+		}
+		select {
+		case <-startDone:
+		case <-time.After(time.Second):
+			t.Fatal("Start did not return after concurrent Stop")
+		}
+	}
+}
+
+func TestRotateTokenPreservesDequeuedResult(t *testing.T) {
+	const (
+		oldToken = "old-token"
+		newToken = "new-token"
+	)
+	connected := make(chan string, 2)
+	delivered := make(chan wire.WorkerMessage, 1)
+	upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		switch req.URL.Path {
+		case "/api/internal/workers/rotate-token":
+			_ = json.NewEncoder(w).Encode(struct {
+				AuthToken string `json:"auth_token"`
+			}{AuthToken: newToken})
+		case "/worker":
+			conn, err := upgrader.Upgrade(w, req, nil)
+			if err != nil {
+				return
+			}
+			defer conn.Close()
+			token := req.URL.Query().Get("token")
+			connected <- token
+			if token == oldToken {
+				var message wire.WorkerMessage
+				if conn.ReadJSON(&message) == nil {
+					delivered <- message
+				}
+				return
+			}
+			if token != newToken {
+				return
+			}
+			var message wire.WorkerMessage
+			if conn.ReadJSON(&message) == nil {
+				delivered <- message
+			}
+		}
+	}))
+	defer server.Close()
+
+	r := NewRunner(&Config{URL: server.URL, Token: oldToken, MaxConcurrency: 1})
+	enteredWrite := make(chan struct{})
+	releaseWrite := make(chan struct{})
+	var once sync.Once
+	r.beforeControlWrite = func() {
+		once.Do(func() {
+			close(enteredWrite)
+			<-releaseWrite
+		})
+	}
+	startDone := make(chan error, 1)
+	go func() { startDone <- r.Start() }()
+	t.Cleanup(func() {
+		r.Stop()
+		select {
+		case <-startDone:
+		case <-time.After(time.Second):
+			t.Fatal("runner did not stop")
+		}
+	})
+
+	select {
+	case token := <-connected:
+		require.Equal(t, oldToken, token)
+	case <-time.After(time.Second):
+		t.Fatal("worker did not establish its initial control connection")
+	}
+	r.results <- resultEnvelope{job: wire.CheckJob{JobID: "result", LeaseToken: "lease"}, reports: []wire.CheckResultReport{{JobID: "result", State: "OK"}}}
+	select {
+	case <-enteredWrite:
+	case <-time.After(time.Second):
+		t.Fatal("writer did not dequeue result")
+	}
+	rotated := make(chan error, 1)
+	go func() {
+		_, err := r.RotateToken(t.Context())
+		rotated <- err
+	}()
+	close(releaseWrite)
+	require.NoError(t, <-rotated)
+
+	select {
+	case message := <-delivered:
+		require.NotNil(t, message.Result)
+		assert.Equal(t, "result", message.Result.JobID)
+		assert.Equal(t, "lease", message.Result.LeaseToken)
+	case <-time.After(time.Second):
+		t.Fatal("dequeued result was lost during rotation")
+	}
+}
+
+func TestRotateTokenSerializesWithStop(t *testing.T) {
+	rotationStarted := make(chan struct{})
+	releaseHandler := make(chan struct{})
+	upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		switch req.URL.Path {
+		case "/api/internal/workers/rotate-token":
+			close(rotationStarted)
+			<-releaseHandler
+		case "/worker":
+			conn, err := upgrader.Upgrade(w, req, nil)
+			if err == nil {
+				defer conn.Close()
+				_, _, _ = conn.ReadMessage()
+			}
+		}
+	}))
+	defer func() {
+		close(releaseHandler)
+		server.Close()
+	}()
+
+	r := NewRunner(&Config{URL: server.URL, Token: "old-token", MaxConcurrency: 1})
+	startDone := make(chan error, 1)
+	go func() { startDone <- r.Start() }()
+	t.Cleanup(func() { r.Stop() })
+
+	// Wait for Start to install its client before beginning rotation.
+	deadline := time.After(time.Second)
+	for {
+		r.clientMu.Lock()
+		started := r.client != nil
+		r.clientMu.Unlock()
+		if started {
+			break
+		}
+		select {
+		case <-deadline:
+			t.Fatal("runner did not start")
+		default:
+			time.Sleep(time.Millisecond)
+		}
+	}
+
+	rotated := make(chan error, 1)
+	go func() {
+		_, err := r.RotateToken(t.Context())
+		rotated <- err
+	}()
+	select {
+	case <-rotationStarted:
+	case <-time.After(time.Second):
+		t.Fatal("rotation request did not start")
+	}
+	stopped := make(chan struct{})
+	go func() {
+		r.Stop()
+		close(stopped)
+	}()
+	require.Error(t, <-rotated, "Stop must cancel an in-flight rotation request")
+	select {
+	case <-stopped:
+	case <-time.After(time.Second):
+		t.Fatal("Stop did not complete after canceling rotation")
+	}
+	select {
+	case err := <-startDone:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("runner did not stop")
+	}
+	_, err := r.RotateToken(t.Context())
+	require.Error(t, err, "rotation cannot succeed after shutdown")
+}
+
+func TestStopUnblocksRotationWaitingForWriter(t *testing.T) {
+	upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
+	connected := make(chan struct{}, 1)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		switch req.URL.Path {
+		case "/api/internal/workers/rotate-token":
+			_ = json.NewEncoder(w).Encode(struct {
+				AuthToken string `json:"auth_token"`
+			}{AuthToken: "new-token"})
+		case "/worker":
+			conn, err := upgrader.Upgrade(w, req, nil)
+			if err != nil {
+				return
+			}
+			defer conn.Close()
+			connected <- struct{}{}
+			_, _, _ = conn.ReadMessage()
+		}
+	}))
+	defer server.Close()
+
+	r := NewRunner(&Config{URL: server.URL, Token: "old-token", MaxConcurrency: 1})
+	writeBlocked := make(chan struct{})
+	var once sync.Once
+	r.beforeControlWrite = func() {
+		once.Do(func() {
+			close(writeBlocked)
+			<-r.controlCtx.Done()
+		})
+	}
+	rotationReady := make(chan struct{})
+	r.beforeTokenCommit = func() { close(rotationReady) }
+	startDone := make(chan error, 1)
+	go func() { startDone <- r.Start() }()
+	t.Cleanup(func() { r.Stop() })
+	select {
+	case <-connected:
+	case <-time.After(time.Second):
+		t.Fatal("worker did not connect")
+	}
+	r.results <- resultEnvelope{job: wire.CheckJob{JobID: "blocked", LeaseToken: "lease"}, reports: []wire.CheckResultReport{{JobID: "blocked", State: "OK"}}}
+	select {
+	case <-writeBlocked:
+	case <-time.After(time.Second):
+		t.Fatal("writer did not block")
+	}
+	rotated := make(chan error, 1)
+	go func() {
+		_, err := r.RotateToken(t.Context())
+		rotated <- err
+	}()
+	select {
+	case <-rotationReady:
+	case <-time.After(time.Second):
+		t.Fatal("rotation did not reach writer serialization")
+	}
+	stopped := make(chan struct{})
+	go func() {
+		r.Stop()
+		close(stopped)
+	}()
+	select {
+	case <-stopped:
+	case <-time.After(time.Second):
+		t.Fatal("Stop deadlocked behind rotation waiting for writer")
+	}
+	select {
+	case <-rotated:
+	case <-time.After(time.Second):
+		t.Fatal("rotation did not unblock after Stop closed the connection")
+	}
+	select {
+	case err := <-startDone:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("runner did not stop")
+	}
+}
+
+func TestStopWinsBeforePostHTTPRotationCommit(t *testing.T) {
+	commitReady := make(chan struct{})
+	releaseCommit := make(chan struct{})
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		if req.URL.Path != "/api/internal/workers/rotate-token" {
+			http.NotFound(w, req)
+			return
+		}
+		_ = json.NewEncoder(w).Encode(struct {
+			AuthToken string `json:"auth_token"`
+		}{AuthToken: "new-token"})
+	}))
+	defer server.Close()
+
+	r := NewRunner(&Config{URL: server.URL, Token: "old-token", MaxConcurrency: 1})
+	r.client = NewClient(server.URL, "old-token")
+	r.beforeTokenCommit = func() {
+		close(commitReady)
+		<-releaseCommit
+	}
+	rotated := make(chan error, 1)
+	go func() {
+		_, err := r.RotateToken(t.Context())
+		rotated <- err
+	}()
+	select {
+	case <-commitReady:
+	case <-time.After(time.Second):
+		t.Fatal("rotation did not reach post-HTTP commit")
+	}
+	stopped := make(chan struct{})
+	go func() {
+		r.Stop()
+		close(stopped)
+	}()
+	select {
+	case <-r.stopCh:
+	case <-time.After(time.Second):
+		t.Fatal("Stop did not win lifecycle ownership")
+	}
+	close(releaseCommit)
+	require.Error(t, <-rotated, "rotation cannot succeed after Stop wins")
+	select {
+	case <-stopped:
+	case <-time.After(time.Second):
+		t.Fatal("Stop did not finish")
+	}
+	assert.Equal(t, "old-token", r.Token())
+}
+
+func TestRotateTokenUnchangedResponseReleasesLifecycle(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		_ = json.NewEncoder(w).Encode(struct {
+			AuthToken string `json:"auth_token"`
+		}{AuthToken: "old-token"})
+	}))
+	defer server.Close()
+
+	r := NewRunner(&Config{URL: server.URL, Token: "old-token", MaxConcurrency: 1})
+	r.client = NewClient(server.URL, "old-token")
+	_, err := r.RotateToken(t.Context())
+	require.Error(t, err)
+
+	stopped := make(chan struct{})
+	go func() {
+		r.Stop()
+		close(stopped)
+	}()
+	select {
+	case <-stopped:
+	case <-time.After(time.Second):
+		t.Fatal("Stop deadlocked after unchanged rotation response")
+	}
+	_, err = r.RotateToken(t.Context())
+	require.Error(t, err, "later rotation must observe shutdown")
+}
+
+func TestRotateTokenClosesStalledWriterBeforeWaiting(t *testing.T) {
+	upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
+	connected := make(chan struct{}, 1)
+	connectionClosed := make(chan struct{})
+	resent := make(chan wire.WorkerMessage, 1)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		switch req.URL.Path {
+		case "/api/internal/workers/rotate-token":
+			_ = json.NewEncoder(w).Encode(struct {
+				AuthToken string `json:"auth_token"`
+			}{AuthToken: "new-token"})
+		case "/worker":
+			conn, err := upgrader.Upgrade(w, req, nil)
+			if err != nil {
+				return
+			}
+			defer conn.Close()
+			if req.URL.Query().Get("token") == "old-token" {
+				connected <- struct{}{}
+				_, _, _ = conn.ReadMessage()
+				connectionClosed <- struct{}{}
+				return
+			}
+			var message wire.WorkerMessage
+			if conn.ReadJSON(&message) == nil {
+				resent <- message
+			}
+		}
+	}))
+	defer server.Close()
+
+	r := NewRunner(&Config{URL: server.URL, Token: "old-token", MaxConcurrency: 1})
+	writerBlocked := make(chan struct{})
+	var once sync.Once
+	r.beforeControlWrite = func() {
+		once.Do(func() {
+			close(writerBlocked)
+			<-connectionClosed
+		})
+	}
+	startDone := make(chan error, 1)
+	go func() { startDone <- r.Start() }()
+	t.Cleanup(func() { r.Stop() })
+	select {
+	case <-connected:
+	case <-time.After(time.Second):
+		t.Fatal("worker did not connect")
+	}
+	r.results <- resultEnvelope{job: wire.CheckJob{JobID: "stalled", LeaseToken: "lease"}, reports: []wire.CheckResultReport{{JobID: "stalled", State: "OK"}}}
+	select {
+	case <-writerBlocked:
+	case <-time.After(time.Second):
+		t.Fatal("writer did not stall")
+	}
+	rotated := make(chan error, 1)
+	go func() {
+		_, err := r.RotateToken(t.Context())
+		rotated <- err
+	}()
+	select {
+	case err := <-rotated:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("rotation waited for stalled writer before closing its connection")
+	}
+	select {
+	case message := <-resent:
+		require.NotNil(t, message.Result)
+		assert.Equal(t, "stalled", message.Result.JobID)
+	case <-time.After(time.Second):
+		t.Fatal("failed check result was not requeued after rotation")
+	}
+	r.Stop()
+	select {
+	case err := <-startDone:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("runner did not stop")
+	}
+}
+
+func TestWriterDropsFailedServerMetricSnapshot(t *testing.T) {
+	upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
+	upgraded := make(chan struct{})
+	closeServer := make(chan struct{})
+	serverClosed := make(chan struct{})
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		conn, err := upgrader.Upgrade(w, req, nil)
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+		close(upgraded)
+		<-closeServer
+		close(serverClosed)
+	}))
+	defer server.Close()
+
+	conn, err := NewClient(server.URL, "token").WorkerSocket()
+	require.NoError(t, err)
+	defer conn.Close()
+	select {
+	case <-upgraded:
+	case <-time.After(time.Second):
+		t.Fatal("websocket did not connect")
+	}
+
+	r := NewRunner(&Config{})
+	r.metricResults = make(chan metricEnvelope, 1)
+	enteredWrite := make(chan struct{})
+	releaseWrite := make(chan struct{})
+	r.beforeControlWrite = func() {
+		close(enteredWrite)
+		<-releaseWrite
+	}
+	done := make(chan struct{})
+	writerDone := make(chan struct{})
+	go func() {
+		var writeMu sync.Mutex
+		r.writer(conn, &writeMu, done, 1)
+		close(writerDone)
+	}()
+	r.metricResults <- metricEnvelope{generation: 1, report: wire.ServerMetricReport{ServerID: 1}}
+	select {
+	case <-enteredWrite:
+	case <-time.After(time.Second):
+		t.Fatal("writer did not dequeue metric snapshot")
+	}
+	close(closeServer)
+	select {
+	case <-serverClosed:
+	case <-time.After(time.Second):
+		t.Fatal("server did not close websocket")
+	}
+	_ = conn.Close()
+	close(releaseWrite)
+	select {
+	case <-writerDone:
+	case <-time.After(time.Second):
+		t.Fatal("writer did not return after failed metric write")
+	}
+	_, replayable := r.takeOutbox()
+	assert.False(t, replayable, "failed metric snapshot must not enter result outbox")
+}
+
+func TestWriterRequeuesFailedNotificationResult(t *testing.T) {
+	upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
+	upgraded := make(chan struct{})
+	closeServer := make(chan struct{})
+	serverClosed := make(chan struct{})
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		conn, err := upgrader.Upgrade(w, req, nil)
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+		close(upgraded)
+		<-closeServer
+		close(serverClosed)
+	}))
+	defer server.Close()
+
+	conn, err := NewClient(server.URL, "token").WorkerSocket()
+	require.NoError(t, err)
+	defer conn.Close()
+	select {
+	case <-upgraded:
+	case <-time.After(time.Second):
+		t.Fatal("websocket did not connect")
+	}
+
+	r := NewRunner(&Config{})
+	r.notifyResults = make(chan notifyResultEnvelope, 1)
+	enteredWrite := make(chan struct{})
+	releaseWrite := make(chan struct{})
+	r.beforeControlWrite = func() {
+		close(enteredWrite)
+		<-releaseWrite
+	}
+	done := make(chan struct{})
+	writerDone := make(chan struct{})
+	go func() {
+		var writeMu sync.Mutex
+		r.writer(conn, &writeMu, done, 0)
+		close(writerDone)
+	}()
+	r.notifyResults <- notifyResultEnvelope{report: wire.NotificationResultReport{JobID: "notification", LeaseToken: "lease"}}
+	select {
+	case <-enteredWrite:
+	case <-time.After(time.Second):
+		t.Fatal("writer did not dequeue notification result")
+	}
+	close(closeServer)
+	select {
+	case <-serverClosed:
+	case <-time.After(time.Second):
+		t.Fatal("server did not close websocket")
+	}
+	_ = conn.Close()
+	close(releaseWrite)
+	select {
+	case <-writerDone:
+	case <-time.After(time.Second):
+		t.Fatal("writer did not return after failed notification write")
+	}
+	message, replayable := r.takeOutbox()
+	require.True(t, replayable, "failed notification result must enter result outbox")
+	require.NotNil(t, message.NotificationResult)
+	assert.Equal(t, "notification", message.NotificationResult.JobID)
+}
+
+func TestMetricGenerationRejectsDisconnectedAndSendsFreshMetric(t *testing.T) {
+	upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
+	firstClosed := make(chan struct{})
+	secondConnected := make(chan struct{})
+	received := make(chan wire.WorkerMessage, 1)
+	var connections atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		conn, err := upgrader.Upgrade(w, req, nil)
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+		if connections.Add(1) == 1 {
+			close(firstClosed)
+			return
+		}
+		close(secondConnected)
+		var message wire.WorkerMessage
+		if conn.ReadJSON(&message) == nil {
+			received <- message
+		}
+	}))
+	defer server.Close()
+
+	r := NewRunner(&Config{URL: server.URL, Token: "token", MaxConcurrency: 1})
+	startDone := make(chan error, 1)
+	go func() { startDone <- r.Start() }()
+	t.Cleanup(func() { r.Stop() })
+	select {
+	case <-firstClosed:
+	case <-time.After(time.Second):
+		t.Fatal("worker did not establish initial websocket")
+	}
+	// Wait until the old connection has fully torn down, then try to enqueue a
+	// metric in the old-drain/disconnected interleaving.
+	deadline := time.After(time.Second)
+	for r.metricGeneration.Load() != 0 {
+		select {
+		case <-deadline:
+			t.Fatal("old control generation did not clear")
+		default:
+			time.Sleep(time.Millisecond)
+		}
+	}
+	staleCount := 1
+	r.enqueueMetric(wire.ServerMetricReport{ServerID: 1, ProcessCount: &staleCount})
+	assert.Empty(t, r.metricResults, "disconnected metric must not enter bounded channel")
+	select {
+	case r.reconnectCh <- struct{}{}:
+	default:
+	}
+	select {
+	case <-secondConnected:
+	case <-time.After(time.Second):
+		t.Fatal("worker did not reconnect")
+	}
+	deadline = time.After(time.Second)
+	for r.metricGeneration.Load() == 0 {
+		select {
+		case <-deadline:
+			t.Fatal("new control generation did not install")
+		default:
+			time.Sleep(time.Millisecond)
+		}
+	}
+	// This enqueue occurs immediately after the new connection installation.
+	freshCount := 2
+	r.enqueueMetric(wire.ServerMetricReport{ServerID: 1, ProcessCount: &freshCount})
+	select {
+	case message := <-received:
+		require.NotNil(t, message.ServerMetric)
+		assert.Equal(t, 2, *message.ServerMetric.ProcessCount)
+	case <-time.After(time.Second):
+		t.Fatal("fresh metric was not sent after reconnect")
+	}
+}
+
 func TestApplyInitResizesPool(t *testing.T) {
 	executor := func(payload interface{}) interface{} {
 		return []wire.CheckResultReport{}

internal/distworker/server_metrics.go

@@ -1,6 +1,7 @@
 package distworker
 
 import (
+	"context"
 	"os"
 	"path/filepath"
 	"sort"
@@ -35,32 +36,41 @@ const (
 // the real Linux /proc and statfs collector; unsupported platforms return no
 // report rather than fabricated values. The worker has no control-plane DB
 // access and forwards snapshots on its authenticated websocket.
-func (r *Runner) serverMetricLoop() {
+func (r *Runner) serverMetricLoop(ctx context.Context) {
 	ticker := time.NewTicker(serverMetricInterval)
 	defer ticker.Stop()
 	for {
 		select {
-		case <-r.stopCh:
+		case <-ctx.Done():
 			return
 		case <-ticker.C:
 			serverID := r.serverID.Load()
-			if serverID == 0 || r.metricResults == nil {
+			if serverID == 0 {
 				continue
 			}
 			report, ok := collectServerMetric(serverID)
 			if !ok {
 				continue
 			}
-			select {
-			case r.metricResults <- report:
-			case <-r.stopCh:
-				return
-			default:
-			}
+			r.enqueueMetric(report)
 		}
 	}
 }
 
+func (r *Runner) enqueueMetric(report wire.ServerMetricReport) {
+	if r.metricResults == nil {
+		return
+	}
+	generation := r.metricGeneration.Load()
+	if generation == 0 || r.metricGeneration.Load() != generation {
+		return
+	}
+	select {
+	case r.metricResults <- metricEnvelope{generation: generation, report: report}:
+	default:
+	}
+}
+
 func collectServerMetric(serverID int64) (wire.ServerMetricReport, bool) {
 	memTotal, memAvailable, load1, load5, load15, uptime, ok := readLinuxHostMetrics("/proc")
 	if !ok {