diff --git a/spindle/mill/mill.go b/spindle/mill/mill.go
--- a/spindle/mill/mill.go
+++ b/spindle/mill/mill.go
@@ -6,7 +6,8 @@
 	"errors"
 	"fmt"
 	"log/slog"
-	"sort"
+	"slices"
+	"strings"
 	"sync"
 	"time"
 
@@ -274,58 +275,90 @@
 		return nil, err
 	}
 
-	candidates := m.rankCandidates(engineName)
+	requiredLabels := requiredLabels(wf)
+	candidates := m.rankCandidates(engineName, requiredLabels)
 	if len(candidates) == 0 {
 		return nil, nil
 	}
-	if len(candidates) > m.cfg.TopK {
-		candidates = candidates[:m.cfg.TopK]
-	}
-
-	bidCtx, cancel := context.WithTimeout(ctx, m.cfg.BidTimeout)
-	defer cancel()
 
 	type bidResult struct {
-		sess  *millSession
-		lease *RemoteLease
-		score float64
+		sess         *millSession
+		lease        *RemoteLease
+		rank         int
+		incompatible bool
+		reason       string
 	}
 	results := make(chan bidResult, len(candidates))
-	var wg sync.WaitGroup
-	for _, sess := range candidates {
-		wg.Add(1)
-		go func(sess *millSession) {
-			defer wg.Done()
-			leaseID := m.nextLeaseID()
-			lease := newLease(leaseID, sess.nodeID, engineName)
-			wid := m.jobWid(wf)
-			msg := &millproto.Message{ReserveSeat: &millv1.ReserveSeat{
-				LeaseId:         leaseID,
-				TargetEngine:    engineName,
-				RawPipelineJson: rawPipeline,
-				RawWorkflowJson: rawWorkflow,
-				Knot:            wid.Knot,
-				Rkey:            wid.Rkey,
-				TtlSeconds:      uint32(m.cfg.ReconnectGrace / time.Second),
-			}}
-			resp, err := sess.request(bidCtx, leaseID, msg)
-			if err != nil {
+	ask := func(rank int, sess *millSession) {
+		bidCtx, cancel := context.WithTimeout(ctx, m.cfg.BidTimeout)
+		defer cancel()
+		leaseID := m.nextLeaseID()
+		lease := newLease(leaseID, sess.nodeID, engineName)
+		wid := m.jobWid(wf)
+		msg := &millproto.Message{ReserveSeat: &millv1.ReserveSeat{
+			LeaseId:         leaseID,
+			TargetEngine:    engineName,
+			RawPipelineJson: rawPipeline,
+			RawWorkflowJson: rawWorkflow,
+			Knot:            wid.Knot,
+			Rkey:            wid.Rkey,
+			TtlSeconds:      uint32(m.cfg.ReconnectGrace / time.Second),
+		}}
+		resp, err := sess.request(bidCtx, leaseID, msg)
+		if err != nil {
+			results <- bidResult{}
+			return
+		}
+		rr := resp.GetReserveResult()
+		if rr == nil {
+			results <- bidResult{}
+			return
+		}
+		if !rr.GetAccepted() {
+			if rr.GetRejectClass() == millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE {
+				results <- bidResult{sess: sess, rank: rank, incompatible: true, reason: rr.GetRejectReason()}
 				return
 			}
-			rr := resp.GetReserveResult()
-			if rr == nil || !rr.GetAccepted() {
-				return
-			}
-			results <- bidResult{sess: sess, lease: lease, score: rr.GetScore()}
-		}(sess)
+			results <- bidResult{}
+			return
+		}
+		results <- bidResult{sess: sess, lease: lease, rank: rank}
 	}
-	wg.Wait()
-	close(results)
+	window := m.cfg.TopK
+	if window <= 0 {
+		window = len(candidates)
+	}
+	next := 0
+	inFlight := 0
+	for next < len(candidates) && inFlight < window {
+		inFlight++
+		go ask(next, candidates[next])
+		next++
+	}
 
 	var winner *bidResult
 	var losers []*RemoteLease
-	for r := range results {
-		if winner == nil || r.score > winner.score {
+	var incompatible []string
+	softReject := false
+	for inFlight > 0 {
+		r := <-results
+		inFlight--
+		if r.incompatible {
+			if r.reason != "" {
+				incompatible = append(incompatible, r.reason)
+			}
+		} else if r.lease == nil {
+			softReject = true
+		}
+		if r.lease == nil {
+			for winner == nil && next < len(candidates) && inFlight < window {
+				inFlight++
+				go ask(next, candidates[next])
+				next++
+			}
+			continue
+		}
+		if winner == nil || r.rank < winner.rank {
 			if winner != nil {
 				losers = append(losers, winner.lease)
 			}
@@ -341,18 +374,22 @@
 	}
 
 	if winner == nil {
+		if len(incompatible) > 0 && !softReject {
+			return nil, fmt.Errorf("no compatible executor for %s: %s", engineName, strings.Join(incompatible, "; "))
+		}
 		return nil, nil
 	}
 	return winner.lease, nil
 }
 
-func (m *Mill) rankCandidates(engineName string) []*millSession {
+func (m *Mill) rankCandidates(engineName string, requiredLabels []string) []*millSession {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
 	type ranked struct {
 		sess  *millSession
-		seats uint32
+		worst float64
+		sum   float64
 	}
 	var rs []ranked
 	for _, s := range m.sessions {
@@ -362,13 +399,31 @@
 		if s.snapshot == nil {
 			continue
 		}
-		es, ok := s.snapshot.GetEngines()[engineName]
-		if !ok || es.GetFreeSeats() == 0 {
+		ea, ok := s.snapshot.GetEngines()[engineName]
+		if !ok || !ea.GetAvailable() {
 			continue
 		}
-		rs = append(rs, ranked{sess: s, seats: es.GetFreeSeats()})
+		if !hasLabels(s.labels, requiredLabels) {
+			continue
+		}
+		worst, sum := loadScore(ea.GetLoad())
+		rs = append(rs, ranked{sess: s, worst: worst, sum: sum})
 	}
-	sort.SliceStable(rs, func(i, j int) bool { return rs[i].seats > rs[j].seats })
+	slices.SortStableFunc(rs, func(a, b ranked) int {
+		if a.worst < b.worst {
+			return -1
+		}
+		if a.worst > b.worst {
+			return 1
+		}
+		if a.sum < b.sum {
+			return -1
+		}
+		if a.sum > b.sum {
+			return 1
+		}
+		return 0
+	})
 
 	out := make([]*millSession, len(rs))
 	for i := range rs {
@@ -377,7 +432,32 @@
 	return out
 }
 
-// --- commit + wait (RunStep) ----------------------------------------------
+func loadScore(load map[string]float64) (worst, sum float64) {
+	for _, v := range load {
+		if v > worst {
+			worst = v
+		}
+		sum += v
+	}
+	return worst, sum
+}
+
+func requiredLabels(wf *models.Workflow) []string {
+	st, ok := wf.Data.(*millWorkflowState)
+	if !ok || st == nil {
+		return nil
+	}
+	return st.RawWorkflow.RunsOn
+}
+
+func hasLabels(labels []string, required []string) bool {
+	for _, want := range required {
+		if !slices.Contains(labels, want) {
+			return false
+		}
+	}
+	return true
+}
 
 func (m *Mill) commitAndWait(ctx context.Context, wf *models.Workflow, unlocked []secrets.UnlockedSecret) error {
 	st, ok := wf.Data.(*millWorkflowState)
diff --git a/spindle/mill/mill_test.go b/spindle/mill/mill_test.go
--- a/spindle/mill/mill_test.go
+++ b/spindle/mill/mill_test.go
@@ -5,6 +5,7 @@
 	"errors"
 	"io"
 	"log/slog"
+	"strings"
 	"testing"
 	"time"
 
@@ -30,6 +31,108 @@
 			RawWorkflow:  tangled.Pipeline_Workflow{Name: name},
 			RawPipeline:  tangled.Pipeline{},
 		},
+	}
+}
+
+func testWorkflowWithRunsOn(name string, runsOn []string) *models.Workflow {
+	wf := testWorkflow(name)
+	wf.Data.(*millWorkflowState).RawWorkflow.RunsOn = runsOn
+	return wf
+}
+
+func addCandidateSession(t *testing.T, m *Mill, nodeID string, labels []string, load float64, enc messageEncoder) *millSession {
+	t.Helper()
+	if enc == nil {
+		enc = scriptedEncoder(func(*millproto.Message) error { return nil })
+	}
+	sess := newSession(nodeID, enc, slog.New(slog.NewTextHandler(io.Discard, nil)))
+	sess.labels = labels
+	sess.snapshot = &millv1.NodeSnapshot{
+		NodeId: nodeID,
+		Engines: map[string]*millv1.EngineAvailability{
+			"dummy": {Available: load < 1.0, Load: map[string]float64{"slots": load}},
+		},
+	}
+	m.mu.Lock()
+	m.sessions[nodeID] = sess
+	m.mu.Unlock()
+	return sess
+}
+
+func assertRankedNodes(t *testing.T, got []*millSession, want []string) {
+	t.Helper()
+	if len(got) != len(want) {
+		t.Fatalf("rankCandidates() returned %d candidates, want %d: got %v want %v", len(got), len(want), sessionIDs(got), want)
+	}
+	for i := range want {
+		if got[i].nodeID != want[i] {
+			t.Fatalf("rankCandidates()[%d] = %q, want %q; full order got %v want %v", i, got[i].nodeID, want[i], sessionIDs(got), want)
+		}
+	}
+}
+
+func sessionIDs(sessions []*millSession) []string {
+	out := make([]string, len(sessions))
+	for i, sess := range sessions {
+		out[i] = sess.nodeID
+	}
+	return out
+}
+
+func sameStringMultiset(a, b []string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	counts := make(map[string]int, len(a))
+	for _, s := range a {
+		counts[s]++
+	}
+	for _, s := range b {
+		if counts[s] == 0 {
+			return false
+		}
+		counts[s]--
+	}
+	return true
+}
+
+type reserveReply struct {
+	accepted    bool
+	rejectClass millv1.RejectClass
+	reason      string
+}
+
+func addReplyingCandidateSession(t *testing.T, m *Mill, nodeID string, labels []string, load float64, asked chan<- string, reply reserveReply) *millSession {
+	t.Helper()
+	var sess *millSession
+	sess = addCandidateSession(t, m, nodeID, labels, load, scriptedEncoder(func(msg *millproto.Message) error {
+		rs := msg.GetReserveSeat()
+		if rs == nil {
+			return nil
+		}
+		if asked != nil {
+			asked <- nodeID
+		}
+		sess.deliver(rs.GetLeaseId(), &millproto.Message{ReserveResult: &millv1.ReserveResult{
+			LeaseId:      rs.GetLeaseId(),
+			Accepted:     reply.accepted,
+			RejectReason: reply.reason,
+			RejectClass:  reply.rejectClass,
+		}})
+		return nil
+	}))
+	return sess
+}
+
+func drainAsked(ch <-chan string) []string {
+	var out []string
+	for {
+		select {
+		case nodeID := <-ch:
+			out = append(out, nodeID)
+		default:
+			return out
+		}
 	}
 }
 
@@ -138,6 +241,195 @@
 	}
 }
 
+func TestRankCandidatesFiltersRequiredLabelsWithANDSemantics(t *testing.T) {
+	m := New(slog.New(slog.NewTextHandler(io.Discard, nil)), Config{})
+	addCandidateSession(t, m, "linux-high", []string{"linux"}, 0.0, nil)
+	addCandidateSession(t, m, "linux-arm", []string{"linux", "arm64"}, 0.25, nil)
+	addCandidateSession(t, m, "unlabeled", nil, 0.5, nil)
+	addCandidateSession(t, m, "linux-arm-gpu", []string{"linux", "arm64", "gpu"}, 0.75, nil)
+	addCandidateSession(t, m, "linux-arm-full", []string{"linux", "arm64"}, 1.0, nil)
+
+	tests := []struct {
+		name           string
+		requiredLabels []string
+		want           []string
+	}{
+		{
+			name: "no required labels keeps old capacity ranking",
+			want: []string{"linux-high", "linux-arm", "unlabeled", "linux-arm-gpu"},
+		},
+		{
+			name:           "single required label includes every candidate carrying it",
+			requiredLabels: []string{"linux"},
+			want:           []string{"linux-high", "linux-arm", "linux-arm-gpu"},
+		},
+		{
+			name:           "all required labels must be present",
+			requiredLabels: []string{"linux", "arm64"},
+			want:           []string{"linux-arm", "linux-arm-gpu"},
+		},
+		{
+			name:           "one missing required label excludes the candidate",
+			requiredLabels: []string{"linux", "arm64", "gpu"},
+			want:           []string{"linux-arm-gpu"},
+		},
+		{
+			name:           "unknown required label leaves no candidate",
+			requiredLabels: []string{"linux", "arm64", "metal"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assertRankedNodes(t, m.rankCandidates("dummy", tt.requiredLabels), tt.want)
+		})
+	}
+}
+
+func TestPlaceWithMissingRequiredLabelsStaysPendingWithoutReserve(t *testing.T) {
+	l := slog.New(slog.NewTextHandler(io.Discard, nil))
+	m := New(l, Config{BidTimeout: 10 * time.Millisecond})
+	reserveSent := make(chan struct{}, 1)
+	addCandidateSession(t, m, "linux-only", []string{"linux"}, 0.75, scriptedEncoder(func(msg *millproto.Message) error {
+		if msg.GetReserveSeat() != nil {
+			select {
+			case reserveSent <- struct{}{}:
+			default:
+			}
+		}
+		return nil
+	}))
+	wf := testWorkflowWithRunsOn("build", []string{"linux", "arm64"})
+	wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Millisecond)
+	defer cancel()
+	_, err := m.place(ctx, "dummy", wid, wf)
+	if err != context.DeadlineExceeded {
+		t.Fatalf("place() error = %v, want DeadlineExceeded while job remains pending", err)
+	}
+	select {
+	case <-reserveSent:
+		t.Fatal("place() sent ReserveSeat to executor missing a required label")
+	default:
+	}
+}
+
+func TestPlaceKeepsAskingBelowTopKAfterIncompatibleRejects(t *testing.T) {
+	l := slog.New(slog.NewTextHandler(io.Discard, nil))
+	m := New(l, Config{TopK: 2, BidTimeout: time.Second})
+	asked := make(chan string, 4)
+	addReplyingCandidateSession(t, m, "wrong-label", []string{"linux"}, 0.0, asked, reserveReply{
+		accepted:    false,
+		rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE,
+		reason:      "wrong-label should not be asked",
+	})
+	addReplyingCandidateSession(t, m, "incompatible-a", []string{"linux", "arm64"}, 0.0, asked, reserveReply{
+		accepted:    false,
+		rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE,
+		reason:      "no runner",
+	})
+	addReplyingCandidateSession(t, m, "incompatible-b", []string{"linux", "arm64"}, 0.25, asked, reserveReply{
+		accepted:    false,
+		rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE,
+		reason:      "bad image",
+	})
+	addReplyingCandidateSession(t, m, "compatible-below-window", []string{"linux", "arm64"}, 0.5, asked, reserveReply{
+		accepted: true,
+	})
+
+	wf := testWorkflowWithRunsOn("build", []string{"linux", "arm64"})
+	wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"}
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	slot, err := m.place(ctx, "dummy", wid, wf)
+	if err != nil {
+		t.Fatalf("place() error = %v, want compatible lower-ranked executor", err)
+	}
+	defer slot.Release()
+
+	ms, ok := slot.(*millSlot)
+	if !ok {
+		t.Fatalf("place() slot type = %T, want *millSlot", slot)
+	}
+	if ms.lease.nodeID != "compatible-below-window" {
+		t.Fatalf("place() chose node %q, want compatible-below-window", ms.lease.nodeID)
+	}
+	if got, want := drainAsked(asked), []string{"incompatible-a", "incompatible-b", "compatible-below-window"}; !sameStringMultiset(got, want) {
+		t.Fatalf("ReserveSeat asked nodes = %v, want %v", got, want)
+	}
+}
+
+func TestPlaceKeepsTransientOnlyRejectsPending(t *testing.T) {
+	l := slog.New(slog.NewTextHandler(io.Discard, nil))
+	m := New(l, Config{TopK: 1, BidTimeout: 10 * time.Millisecond})
+	asked := make(chan string, 2)
+	addReplyingCandidateSession(t, m, "busy-a", []string{"linux", "arm64"}, 0.5, asked, reserveReply{
+		accepted:    false,
+		rejectClass: millv1.RejectClass_REJECT_CLASS_TRANSIENT,
+		reason:      "draining",
+	})
+	addReplyingCandidateSession(t, m, "busy-b", []string{"linux", "arm64"}, 0.75, asked, reserveReply{
+		accepted:    false,
+		rejectClass: millv1.RejectClass_REJECT_CLASS_TRANSIENT,
+		reason:      "no slot",
+	})
+
+	wf := testWorkflowWithRunsOn("build", []string{"linux", "arm64"})
+	wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"}
+	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Millisecond)
+	defer cancel()
+	_, err := m.place(ctx, "dummy", wid, wf)
+	if err != context.DeadlineExceeded {
+		t.Fatalf("place() error = %v, want DeadlineExceeded while transient rejects leave job pending", err)
+	}
+	if got, want := drainAsked(asked), []string{"busy-a", "busy-b"}; !sameStringMultiset(got, want) {
+		t.Fatalf("ReserveSeat asked nodes = %v, want %v", got, want)
+	}
+}
+
+func TestPlaceReportsOnlyEligiblePermanentIncompatibleRejects(t *testing.T) {
+	l := slog.New(slog.NewTextHandler(io.Discard, nil))
+	m := New(l, Config{TopK: 1, BidTimeout: time.Second})
+	asked := make(chan string, 3)
+	addReplyingCandidateSession(t, m, "wrong-label", []string{"linux"}, 0.25, asked, reserveReply{
+		accepted:    false,
+		rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE,
+		reason:      "wrong-label should not be asked",
+	})
+	addReplyingCandidateSession(t, m, "incompatible-a", []string{"linux", "arm64"}, 0.5, asked, reserveReply{
+		accepted:    false,
+		rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE,
+		reason:      "no qemu-system-aarch64",
+	})
+	addReplyingCandidateSession(t, m, "incompatible-b", []string{"linux", "arm64"}, 0.75, asked, reserveReply{
+		accepted:    false,
+		rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE,
+		reason:      "image arch unsupported",
+	})
+
+	wf := testWorkflowWithRunsOn("build", []string{"linux", "arm64"})
+	wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"}
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	_, err := m.place(ctx, "dummy", wid, wf)
+	if err == nil {
+		t.Fatal("place() error = nil, want permanent incompatible failure")
+	}
+	errText := err.Error()
+	for _, want := range []string{"no compatible executor for dummy", "no qemu-system-aarch64", "image arch unsupported"} {
+		if !strings.Contains(errText, want) {
+			t.Fatalf("place() error = %q, want it to contain %q", errText, want)
+		}
+	}
+	if strings.Contains(errText, "wrong-label should not be asked") {
+		t.Fatalf("place() error = %q, included a missing-label candidate as incompatible", errText)
+	}
+	if got, want := drainAsked(asked), []string{"incompatible-a", "incompatible-b"}; !sameStringMultiset(got, want) {
+		t.Fatalf("ReserveSeat asked nodes = %v, want %v", got, want)
+	}
+}
+
 func TestMaxPendingRejects(t *testing.T) {
 	l := slog.New(slog.NewTextHandler(io.Discard, nil))
 	m := New(l, Config{MaxPending: 1})
@@ -173,5 +465,33 @@
 	sess.mu.Unlock()
 	if pending != 0 {
 		t.Fatalf("request left %d pending waiters, want 0", pending)
+	}
+}
+
+func TestFallbackBidGetsFullTimeout(t *testing.T) {
+	m := New(discardLogger(), Config{TopK: 1, BidTimeout: 30 * time.Millisecond})
+	addCandidateSession(t, m, "silent", nil, 0, nil)
+	var fallback *millSession
+	fallback = addCandidateSession(t, m, "fallback", nil, 0.5, scriptedEncoder(func(msg *millproto.Message) error {
+		if reserve := msg.GetReserveSeat(); reserve != nil {
+			leaseID := reserve.GetLeaseId()
+			time.AfterFunc(10*time.Millisecond, func() {
+				fallback.deliver(leaseID, &millproto.Message{ReserveResult: &millv1.ReserveResult{
+					LeaseId:  leaseID,
+					Accepted: true,
+				}})
+			})
+		}
+		return nil
+	}))
+
+	ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
+	defer cancel()
+	lease, err := m.bid(ctx, "dummy", testWorkflow("build"))
+	if err != nil {
+		t.Fatalf("bid: %v", err)
+	}
+	if lease == nil || lease.nodeID != "fallback" {
+		t.Fatalf("bid winner = %+v, want fallback after silent incumbent times out", lease)
 	}
 }
diff --git a/spindle/mill/session.go b/spindle/mill/session.go
--- a/spindle/mill/session.go
+++ b/spindle/mill/session.go
@@ -19,6 +19,7 @@
 // request/response by lease id. We never hold a lock across a decode.
 type millSession struct {
 	nodeID string
+	labels []string
 	enc    messageEncoder
 	l      *slog.Logger
 
diff --git a/spindle/mill/executor/executor.go b/spindle/mill/executor/executor.go
--- a/spindle/mill/executor/executor.go
+++ b/spindle/mill/executor/executor.go
@@ -3,11 +3,13 @@
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"log/slog"
 	"maps"
 	"net/http"
 	"runtime"
+	"strings"
 	"sync"
 	"time"
 
@@ -34,9 +36,10 @@
 
 type Executor struct {
 	millURL string
-	secret  string
+	token   string
 	nodeID  string
 	seats   int
+	labels  []string
 
 	engines map[string]models.Engine
 	db      *db.DB
@@ -50,10 +53,9 @@
 	connMu sync.Mutex
 	enc    messageEncoder
 
-	mu          sync.Mutex
-	active      map[string]*reservation
-	draining    bool
-	activeCount int
+	mu       sync.Mutex
+	active   map[string]*reservation
+	draining bool
 }
 
 type reservation struct {
@@ -75,30 +77,18 @@
 	Encode(*millproto.Message) error
 }
 
-type reservationCleanup struct {
-	stopTail func()
-	slot     engine.WorkflowSlot
-}
-
-func (c reservationCleanup) run() {
-	if c.stopTail != nil {
-		c.stopTail()
-	}
-	if c.slot != nil {
-		c.slot.Release()
-	}
-}
-
 func New(cfg *config.Config, engines map[string]models.Engine, d *db.DB, n *notifier.Notifier, l *slog.Logger) *Executor {
 	seats := cfg.Mill.Seats
 	if seats <= 0 {
 		seats = defaultSeats
 	}
+	labels := normalizeLabels(cfg.Mill.Labels)
 	return &Executor{
 		millURL: cfg.Mill.URL,
-		secret:  cfg.Mill.SharedSecret,
+		token:   cfg.Mill.SharedSecret,
 		nodeID:  cfg.Server.Hostname,
 		seats:   seats,
+		labels:  labels,
 		engines: engines,
 		db:      d,
 		n:       n,
@@ -142,8 +132,8 @@
 
 func (e *Executor) runSession(ctx context.Context) error {
 	header := http.Header{}
-	if e.secret != "" {
-		header.Set("Authorization", "Bearer "+e.secret)
+	if e.token != "" {
+		header.Set("Authorization", "Bearer "+e.token)
 	}
 	conn, _, err := websocket.DefaultDialer.DialContext(ctx, e.millURL, header)
 	if err != nil {
@@ -165,6 +155,7 @@
 		NodeId:          e.nodeID,
 		Engines:         e.engineNames(),
 		Arch:            runtime.GOARCH,
+		Labels:          e.labels,
 		LastOffset:      e.relay.lastOffset(),
 	}}
 	if err := enc.Encode(hello); err != nil {
@@ -248,11 +239,12 @@
 // --- reserve / commit / release / cancel ----------------------------------
 
 func (e *Executor) handleReserve(ctx context.Context, rs *millv1.ReserveSeat) {
-	reject := func(reason string) {
+	reject := func(reason string, class millv1.RejectClass) {
 		e.send(&millproto.Message{ReserveResult: &millv1.ReserveResult{
 			LeaseId:      rs.GetLeaseId(),
 			Accepted:     false,
 			RejectReason: reason,
+			RejectClass:  class,
 		}})
 	}
 
@@ -260,29 +252,29 @@
 	draining := e.draining
 	e.mu.Unlock()
 	if draining {
-		reject("draining")
+		reject("draining", millv1.RejectClass_REJECT_CLASS_TRANSIENT)
 		return
 	}
 
 	realEngine, ok := e.engines[rs.GetTargetEngine()]
 	if !ok {
-		reject("unknown engine " + rs.GetTargetEngine())
+		reject("unknown engine "+rs.GetTargetEngine(), millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE)
 		return
 	}
 	slotter, ok := realEngine.(engine.WorkflowSlotter)
 	if !ok {
-		reject("engine does not support workflow slots")
+		reject("engine does not support workflow slots", millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE)
 		return
 	}
 
 	var twf tangled.Pipeline_Workflow
 	if err := json.Unmarshal([]byte(rs.GetRawWorkflowJson()), &twf); err != nil {
-		reject("bad workflow json")
+		reject("bad workflow json", millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE)
 		return
 	}
 	var tpl tangled.Pipeline
 	if err := json.Unmarshal([]byte(rs.GetRawPipelineJson()), &tpl); err != nil {
-		reject("bad pipeline json")
+		reject("bad pipeline json", millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE)
 		return
 	}
 
@@ -291,7 +283,7 @@
 
 	wf, err := realEngine.InitWorkflow(twf, tpl)
 	if err != nil {
-		reject("init workflow: " + err.Error())
+		reject("init workflow: "+err.Error(), millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE)
 		return
 	}
 	// the job skipped processPipeline, so inject TANGLED_* env here.
@@ -303,7 +295,11 @@
 	// NoWait: the executor doesn't queue locally; the mill owns the backlog.
 	slot, err := slotter.AcquireWorkflowSlot(ctx, wid, wf, engine.NoWait)
 	if err != nil {
-		reject(err.Error())
+		class := millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE
+		if errors.Is(err, engine.ErrNoWorkflowSlots) {
+			class = millv1.RejectClass_REJECT_CLASS_TRANSIENT
+		}
+		reject(err.Error(), class)
 		return
 	}
 
@@ -324,13 +320,11 @@
 
 	e.mu.Lock()
 	e.active[res.leaseID] = res
-	e.activeCount++
 	e.mu.Unlock()
 
 	e.send(&millproto.Message{ReserveResult: &millv1.ReserveResult{
 		LeaseId:  rs.GetLeaseId(),
 		Accepted: true,
-		Score:    e.bidScore(),
 	}})
 	e.pushSnapshot()
 }
@@ -380,7 +374,7 @@
 	if !ok {
 		return
 	}
-	cleanup.run()
+	cleanup()
 	e.pushSnapshot()
 }
 
@@ -394,7 +388,7 @@
 	res.cancelled = true
 	cancel := res.cancel
 	committed := res.committed
-	var cleanup reservationCleanup
+	var cleanup func()
 	if !committed {
 		cleanup = e.removeReservationLocked(res, true)
 	}
@@ -406,7 +400,7 @@
 	// committed jobs clean up when the observe loop sees the terminal row; an
 	// uncommitted cancel still needs the slot released.
 	if !committed {
-		cleanup.run()
+		cleanup()
 		e.pushSnapshot()
 	}
 }
@@ -417,46 +411,67 @@
 		return
 	}
 	e.l.Warn("reservation expired before commit", "lease", leaseID)
-	cleanup.run()
+	cleanup()
 	e.pushSnapshot()
 }
 
-func (e *Executor) takeUncommittedReservation(leaseID string, releaseSlot bool) (reservationCleanup, bool) {
+func (e *Executor) takeUncommittedReservation(leaseID string, releaseSlot bool) (func(), bool) {
 	e.mu.Lock()
 	defer e.mu.Unlock()
 	res := e.active[leaseID]
 	if res == nil || res.committed {
-		return reservationCleanup{}, false
+		return func() {}, false
 	}
 	return e.removeReservationLocked(res, releaseSlot), true
 }
 
-func (e *Executor) finishReservation(res *reservation) (reservationCleanup, bool, bool) {
+func (e *Executor) finishReservation(res *reservation) (func(), bool, bool) {
 	e.mu.Lock()
 	defer e.mu.Unlock()
 	if e.active[res.leaseID] != res {
-		return reservationCleanup{}, false, false
+		return func() {}, false, false
 	}
 	cancelled := res.cancelled
 	return e.removeReservationLocked(res, false), cancelled, true
 }
 
-func (e *Executor) removeReservationLocked(res *reservation, releaseSlot bool) reservationCleanup {
+func (e *Executor) removeReservationLocked(res *reservation, releaseSlot bool) func() {
 	delete(e.active, res.leaseID)
-	if e.activeCount > 0 {
-		e.activeCount--
-	}
 	if res.ttlTimer != nil {
 		res.ttlTimer.Stop()
 		res.ttlTimer = nil
 	}
-	cleanup := reservationCleanup{stopTail: res.stopTail}
+	stopTail := res.stopTail
 	res.stopTail = nil
+	slot := res.slot
 	if releaseSlot {
-		cleanup.slot = res.slot
 		res.slot = nil
 	}
-	return cleanup
+	return func() {
+		if stopTail != nil {
+			stopTail()
+		}
+		if releaseSlot && slot != nil {
+			slot.Release()
+		}
+	}
+}
+
+func normalizeLabels(labels []string) []string {
+	seen := make(map[string]struct{}, len(labels))
+	out := make([]string, 0, len(labels))
+	for _, label := range labels {
+		label = strings.TrimSpace(label)
+		if label == "" {
+			continue
+		}
+		if _, ok := seen[label]; ok {
+			continue
+		}
+		seen[label] = struct{}{}
+		out = append(out, label)
+	}
+	return out
 }
 
 // --- snapshots -------------------------------------------------------------
@@ -483,17 +498,21 @@
 
 func (e *Executor) pushSnapshot() {
 	e.mu.Lock()
-	free := uint32(0)
-	if !e.draining {
-		if n := e.seats - e.activeCount; n > 0 {
-			free = uint32(n)
-		}
-	}
+	draining := e.draining
+	active := len(e.active)
 	e.mu.Unlock()
 
-	engines := make(map[string]*millv1.EngineSnapshot, len(e.engines))
+	load := 0.0
+	if e.seats > 0 {
+		load = float64(active) / float64(e.seats)
+	}
+	available := !draining && active < e.seats
+	engines := make(map[string]*millv1.EngineAvailability, len(e.engines))
 	for name := range e.engines {
-		engines[name] = &millv1.EngineSnapshot{FreeSeats: free}
+		engines[name] = &millv1.EngineAvailability{
+			Available: available,
+			Load:      map[string]float64{"slots": load},
+		}
 	}
 	e.send(&millproto.Message{NodeSnapshot: &millv1.NodeSnapshot{
 		NodeId:  e.nodeID,
@@ -506,12 +525,6 @@
 	e.draining = true
 	e.mu.Unlock()
 	e.pushSnapshot()
-}
-
-func (e *Executor) bidScore() float64 {
-	e.mu.Lock()
-	defer e.mu.Unlock()
-	return float64(e.seats - e.activeCount)
 }
 
 func (e *Executor) engineNames() []string {
diff --git a/spindle/mill/executor/observe.go b/spindle/mill/executor/observe.go
--- a/spindle/mill/executor/observe.go
+++ b/spindle/mill/executor/observe.go
@@ -83,7 +83,7 @@
 	if !ok {
 		return
 	}
-	cleanup.run()
+	cleanup()
 
 	e.relayMu.Lock()
 	terminalStatus := st.Status
diff --git a/spindle/mill/executor/reserved.go b/spindle/mill/executor/reserved.go
--- a/spindle/mill/executor/reserved.go
+++ b/spindle/mill/executor/reserved.go
@@ -4,12 +4,9 @@
 	"context"
 	"fmt"
 	"sync"
-	"time"
 
-	"tangled.org/core/api/tangled"
 	"tangled.org/core/spindle/engine"
 	"tangled.org/core/spindle/models"
-	"tangled.org/core/spindle/secrets"
 )
 
 // reservedEngine wraps a real engine so that the slot acquired up-front during
@@ -19,7 +16,7 @@
 //
 // This is the only change to the execution path on an executor.
 type reservedEngine struct {
-	inner models.Engine
+	models.Engine
 	slot  engine.WorkflowSlot
 	once  sync.Once
 }
@@ -27,27 +24,7 @@
 // newReservedEngine returns a wrapper around inner that hands back slot exactly
 // once from AcquireWorkflowSlot.
 func newReservedEngine(inner models.Engine, slot engine.WorkflowSlot) models.Engine {
-	return &reservedEngine{inner: inner, slot: slot}
-}
-
-func (e *reservedEngine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) {
-	return e.inner.InitWorkflow(twf, tpl)
-}
-
-func (e *reservedEngine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error {
-	return e.inner.SetupWorkflow(ctx, wid, wf, wfLogger)
-}
-
-func (e *reservedEngine) WorkflowTimeout() time.Duration {
-	return e.inner.WorkflowTimeout()
-}
-
-func (e *reservedEngine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error {
-	return e.inner.DestroyWorkflow(ctx, wid)
-}
-
-func (e *reservedEngine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error {
-	return e.inner.RunStep(ctx, wid, w, idx, secrets, wfLogger)
+	return &reservedEngine{Engine: inner, slot: slot}
 }
 
 // AcquireWorkflowSlot hands back the pre-acquired slot exactly once. The slot
diff --git a/spindle/mill/proto/gen/mill.pb.go b/spindle/mill/proto/gen/mill.pb.go
--- a/spindle/mill/proto/gen/mill.pb.go
+++ b/spindle/mill/proto/gen/mill.pb.go
@@ -22,6 +22,55 @@
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )
 
+type RejectClass int32
+
+const (
+	RejectClass_REJECT_CLASS_UNSPECIFIED  RejectClass = 0
+	RejectClass_REJECT_CLASS_TRANSIENT    RejectClass = 1
+	RejectClass_REJECT_CLASS_INCOMPATIBLE RejectClass = 2
+)
+
+// Enum value maps for RejectClass.
+var (
+	RejectClass_name = map[int32]string{
+		0: "REJECT_CLASS_UNSPECIFIED",
+		1: "REJECT_CLASS_TRANSIENT",
+		2: "REJECT_CLASS_INCOMPATIBLE",
+	}
+	RejectClass_value = map[string]int32{
+		"REJECT_CLASS_UNSPECIFIED":  0,
+		"REJECT_CLASS_TRANSIENT":    1,
+		"REJECT_CLASS_INCOMPATIBLE": 2,
+	}
+)
+
+func (x RejectClass) Enum() *RejectClass {
+	p := new(RejectClass)
+	*p = x
+	return p
+}
+
+func (x RejectClass) String() string {
+	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
+}
+
+func (RejectClass) Descriptor() protoreflect.EnumDescriptor {
+	return file_spindle_mill_v1_mill_proto_enumTypes[0].Descriptor()
+}
+
+func (RejectClass) Type() protoreflect.EnumType {
+	return &file_spindle_mill_v1_mill_proto_enumTypes[0]
+}
+
+func (x RejectClass) Number() protoreflect.EnumNumber {
+	return protoreflect.EnumNumber(x)
+}
+
+// Deprecated: Use RejectClass.Descriptor instead.
+func (RejectClass) EnumDescriptor() ([]byte, []int) {
+	return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{0}
+}
+
 // Hello is the first frame an executor sends after dialing the mill. It
 // carries the static traits of the node plus a resume hint.
 type Hello struct {
@@ -32,10 +81,12 @@
 	NodeId string `protobuf:"bytes,2,opt,name=node_id,json=nodeId,proto3" json:"node_id,omitempty"`
 	// engine names this node can run ("microvm", "nixery").
 	Engines []string `protobuf:"bytes,3,rep,name=engines,proto3" json:"engines,omitempty"`
-	// GOARCH of the node, so the mill won't place arch-incompatible jobs.
+	// GOARCH of the node, retained as an informational trait for logs.
 	Arch string `protobuf:"bytes,4,opt,name=arch,proto3" json:"arch,omitempty"`
 	// the highest relay offset the executor believes it has sent; a resume hint.
-	LastOffset    uint64 `protobuf:"varint,5,opt,name=last_offset,json=lastOffset,proto3" json:"last_offset,omitempty"`
+	LastOffset uint64 `protobuf:"varint,5,opt,name=last_offset,json=lastOffset,proto3" json:"last_offset,omitempty"`
+	// opaque operator-defined labels used for runs_on matching.
+	Labels        []string `protobuf:"bytes,6,rep,name=labels,proto3" json:"labels,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -105,6 +156,13 @@
 	return 0
 }
 
+func (x *Hello) GetLabels() []string {
+	if x != nil {
+		return x.Labels
+	}
+	return nil
+}
+
 // Resume is the mill's reply to Hello. The executor replays every buffered
 // relay entry with offset strictly greater than ack_offset before sending new
 // ones.
@@ -152,35 +210,33 @@
 	return 0
 }
 
-// EngineSnapshot is the changing per-engine state on a node. free_seats is a
-// coarse "can you take more" hint; 0 also means draining (the mill treats a
-// draining node as a full one until it leaves). The resource fields are coarse
-// budget headroom for ranking only; the executor makes the real yes/no call in
-// ReserveResult.
-type EngineSnapshot struct {
-	state         protoimpl.MessageState `protogen:"open.v1"`
-	FreeSeats     uint32                 `protobuf:"varint,1,opt,name=free_seats,json=freeSeats,proto3" json:"free_seats,omitempty"`
-	FreeMemoryMib int64                  `protobuf:"varint,2,opt,name=free_memory_mib,json=freeMemoryMib,proto3" json:"free_memory_mib,omitempty"`
-	FreeVcpus     int64                  `protobuf:"varint,3,opt,name=free_vcpus,json=freeVcpus,proto3" json:"free_vcpus,omitempty"`
-	FreeDiskMib   int64                  `protobuf:"varint,4,opt,name=free_disk_mib,json=freeDiskMib,proto3" json:"free_disk_mib,omitempty"`
+// EngineAvailability is the changing per-engine state on a node. The executor
+// reports opaque load metrics (higher = more loaded); the mill ranks by the
+// worst load and uses the sum as a tie-breaker. The mill does not know what
+// the metric keys mean.
+type EngineAvailability struct {
+	state     protoimpl.MessageState `protogen:"open.v1"`
+	Available bool                   `protobuf:"varint,1,opt,name=available,proto3" json:"available,omitempty"`
+	// engine-defined load metrics. keys are opaque to the mill.
+	Load          map[string]float64 `protobuf:"bytes,2,rep,name=load,proto3" json:"load,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"fixed64,2,opt,name=value"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
 
-func (x *EngineSnapshot) Reset() {
-	*x = EngineSnapshot{}
+func (x *EngineAvailability) Reset() {
+	*x = EngineAvailability{}
 	mi := &file_spindle_mill_v1_mill_proto_msgTypes[2]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
 
-func (x *EngineSnapshot) String() string {
+func (x *EngineAvailability) String() string {
 	return protoimpl.X.MessageStringOf(x)
 }
 
-func (*EngineSnapshot) ProtoMessage() {}
+func (*EngineAvailability) ProtoMessage() {}
 
-func (x *EngineSnapshot) ProtoReflect() protoreflect.Message {
+func (x *EngineAvailability) ProtoReflect() protoreflect.Message {
 	mi := &file_spindle_mill_v1_mill_proto_msgTypes[2]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -192,46 +248,32 @@
 	return mi.MessageOf(x)
 }
 
-// Deprecated: Use EngineSnapshot.ProtoReflect.Descriptor instead.
-func (*EngineSnapshot) Descriptor() ([]byte, []int) {
+// Deprecated: Use EngineAvailability.ProtoReflect.Descriptor instead.
+func (*EngineAvailability) Descriptor() ([]byte, []int) {
 	return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{2}
 }
 
-func (x *EngineSnapshot) GetFreeSeats() uint32 {
+func (x *EngineAvailability) GetAvailable() bool {
 	if x != nil {
-		return x.FreeSeats
+		return x.Available
 	}
-	return 0
+	return false
 }
 
-func (x *EngineSnapshot) GetFreeMemoryMib() int64 {
+func (x *EngineAvailability) GetLoad() map[string]float64 {
 	if x != nil {
-		return x.FreeMemoryMib
+		return x.Load
 	}
-	return 0
-}
-
-func (x *EngineSnapshot) GetFreeVcpus() int64 {
-	if x != nil {
-		return x.FreeVcpus
-	}
-	return 0
-}
-
-func (x *EngineSnapshot) GetFreeDiskMib() int64 {
-	if x != nil {
-		return x.FreeDiskMib
-	}
-	return 0
+	return nil
 }
 
 // NodeSnapshot is pushed on connect, periodically, and right after any state
 // change (reserve, commit, terminal).
 type NodeSnapshot struct {
-	state         protoimpl.MessageState     `protogen:"open.v1"`
-	NodeId        string                     `protobuf:"bytes,1,opt,name=node_id,json=nodeId,proto3" json:"node_id,omitempty"`
-	Seq           uint64                     `protobuf:"varint,2,opt,name=seq,proto3" json:"seq,omitempty"`
-	Engines       map[string]*EngineSnapshot `protobuf:"bytes,3,rep,name=engines,proto3" json:"engines,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`
+	state         protoimpl.MessageState         `protogen:"open.v1"`
+	NodeId        string                         `protobuf:"bytes,1,opt,name=node_id,json=nodeId,proto3" json:"node_id,omitempty"`
+	Seq           uint64                         `protobuf:"varint,2,opt,name=seq,proto3" json:"seq,omitempty"`
+	Engines       map[string]*EngineAvailability `protobuf:"bytes,3,rep,name=engines,proto3" json:"engines,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -280,7 +322,7 @@
 	return 0
 }
 
-func (x *NodeSnapshot) GetEngines() map[string]*EngineSnapshot {
+func (x *NodeSnapshot) GetEngines() map[string]*EngineAvailability {
 	if x != nil {
 		return x.Engines
 	}
@@ -387,12 +429,11 @@
 
 // ReserveResult is the executor's accept/reject for a ReserveSeat.
 type ReserveResult struct {
-	state        protoimpl.MessageState `protogen:"open.v1"`
-	LeaseId      string                 `protobuf:"bytes,1,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"`
-	Accepted     bool                   `protobuf:"varint,2,opt,name=accepted,proto3" json:"accepted,omitempty"`
-	RejectReason string                 `protobuf:"bytes,3,opt,name=reject_reason,json=rejectReason,proto3" json:"reject_reason,omitempty"`
-	// optional bid score the mill ranks accepted leases by (higher is better).
-	Score         float64 `protobuf:"fixed64,4,opt,name=score,proto3" json:"score,omitempty"`
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	LeaseId       string                 `protobuf:"bytes,1,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"`
+	Accepted      bool                   `protobuf:"varint,2,opt,name=accepted,proto3" json:"accepted,omitempty"`
+	RejectReason  string                 `protobuf:"bytes,3,opt,name=reject_reason,json=rejectReason,proto3" json:"reject_reason,omitempty"`
+	RejectClass   RejectClass            `protobuf:"varint,4,opt,name=reject_class,json=rejectClass,proto3,enum=spindle.mill.v1.RejectClass" json:"reject_class,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -448,11 +489,11 @@
 	return ""
 }
 
-func (x *ReserveResult) GetScore() float64 {
+func (x *ReserveResult) GetRejectClass() RejectClass {
 	if x != nil {
-		return x.Score
+		return x.RejectClass
 	}
-	return 0
+	return RejectClass_REJECT_CLASS_UNSPECIFIED
 }
 
 // Secret is a single unlocked secret, sent only inside CommitLease.
@@ -1134,31 +1175,31 @@
 
 const file_spindle_mill_v1_mill_proto_rawDesc = "" +
 	"\n" +
-	"\x1aspindle/mill/v1/mill.proto\x12\x0fspindle.mill.v1\x1a\x1bbuf/validate/validate.proto\"\xa3\x01\n" +
+	"\x1aspindle/mill/v1/mill.proto\x12\x0fspindle.mill.v1\x1a\x1bbuf/validate/validate.proto\"\xbb\x01\n" +
 	"\x05Hello\x12)\n" +
 	"\x10protocol_version\x18\x01 \x01(\rR\x0fprotocolVersion\x12 \n" +
 	"\anode_id\x18\x02 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\x06nodeId\x12\x18\n" +
 	"\aengines\x18\x03 \x03(\tR\aengines\x12\x12\n" +
 	"\x04arch\x18\x04 \x01(\tR\x04arch\x12\x1f\n" +
 	"\vlast_offset\x18\x05 \x01(\x04R\n" +
-	"lastOffset\"'\n" +
+	"lastOffset\x12\x16\n" +
+	"\x06labels\x18\x06 \x03(\tR\x06labels\"'\n" +
 	"\x06Resume\x12\x1d\n" +
 	"\n" +
-	"ack_offset\x18\x01 \x01(\x04R\tackOffset\"\x9a\x01\n" +
-	"\x0eEngineSnapshot\x12\x1d\n" +
-	"\n" +
-	"free_seats\x18\x01 \x01(\rR\tfreeSeats\x12&\n" +
-	"\x0ffree_memory_mib\x18\x02 \x01(\x03R\rfreeMemoryMib\x12\x1d\n" +
-	"\n" +
-	"free_vcpus\x18\x03 \x01(\x03R\tfreeVcpus\x12\"\n" +
-	"\rfree_disk_mib\x18\x04 \x01(\x03R\vfreeDiskMib\"\xdc\x01\n" +
+	"ack_offset\x18\x01 \x01(\x04R\tackOffset\"\xae\x01\n" +
+	"\x12EngineAvailability\x12\x1c\n" +
+	"\tavailable\x18\x01 \x01(\bR\tavailable\x12A\n" +
+	"\x04load\x18\x02 \x03(\v2-.spindle.mill.v1.EngineAvailability.LoadEntryR\x04load\x1a7\n" +
+	"\tLoadEntry\x12\x10\n" +
+	"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
+	"\x05value\x18\x02 \x01(\x01R\x05value:\x028\x01\"\xe0\x01\n" +
 	"\fNodeSnapshot\x12\x17\n" +
 	"\anode_id\x18\x01 \x01(\tR\x06nodeId\x12\x10\n" +
 	"\x03seq\x18\x02 \x01(\x04R\x03seq\x12D\n" +
-	"\aengines\x18\x03 \x03(\v2*.spindle.mill.v1.NodeSnapshot.EnginesEntryR\aengines\x1a[\n" +
+	"\aengines\x18\x03 \x03(\v2*.spindle.mill.v1.NodeSnapshot.EnginesEntryR\aengines\x1a_\n" +
 	"\fEnginesEntry\x12\x10\n" +
-	"\x03key\x18\x01 \x01(\tR\x03key\x125\n" +
-	"\x05value\x18\x02 \x01(\v2\x1f.spindle.mill.v1.EngineSnapshotR\x05value:\x028\x01\"\x80\x02\n" +
+	"\x03key\x18\x01 \x01(\tR\x03key\x129\n" +
+	"\x05value\x18\x02 \x01(\v2#.spindle.mill.v1.EngineAvailabilityR\x05value:\x028\x01\"\x80\x02\n" +
 	"\vReserveSeat\x12\"\n" +
 	"\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x12,\n" +
 	"\rtarget_engine\x18\x02 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\ftargetEngine\x12*\n" +
@@ -1167,12 +1208,12 @@
 	"\x04knot\x18\x05 \x01(\tR\x04knot\x12\x12\n" +
 	"\x04rkey\x18\x06 \x01(\tR\x04rkey\x12\x1f\n" +
 	"\vttl_seconds\x18\a \x01(\rR\n" +
-	"ttlSeconds\"\x8a\x01\n" +
+	"ttlSeconds\"\xb5\x01\n" +
 	"\rReserveResult\x12\"\n" +
 	"\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x12\x1a\n" +
 	"\baccepted\x18\x02 \x01(\bR\baccepted\x12#\n" +
-	"\rreject_reason\x18\x03 \x01(\tR\frejectReason\x12\x14\n" +
-	"\x05score\x18\x04 \x01(\x01R\x05score\"0\n" +
+	"\rreject_reason\x18\x03 \x01(\tR\frejectReason\x12?\n" +
+	"\freject_class\x18\x04 \x01(\x0e2\x1c.spindle.mill.v1.RejectClassR\vrejectClass\"0\n" +
 	"\x06Secret\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
 	"\x05value\x18\x02 \x01(\tR\x05value\"d\n" +
@@ -1234,7 +1275,11 @@
 	"\fstatus_event\n" +
 	"\blog_line\n" +
 	"\x0eattempt_result\n" +
-	"\x03ack\x10\x01B0Z.tangled.org/core/spindle/mill/proto/gen;millv1b\x06proto3"
+	"\x03ack\x10\x01*f\n" +
+	"\vRejectClass\x12\x1c\n" +
+	"\x18REJECT_CLASS_UNSPECIFIED\x10\x00\x12\x1a\n" +
+	"\x16REJECT_CLASS_TRANSIENT\x10\x01\x12\x1d\n" +
+	"\x19REJECT_CLASS_INCOMPATIBLE\x10\x02B0Z.tangled.org/core/spindle/mill/proto/gen;millv1b\x06proto3"
 
 var (
 	file_spindle_mill_v1_mill_proto_rawDescOnce sync.Once
@@ -1248,48 +1293,53 @@
 	return file_spindle_mill_v1_mill_proto_rawDescData
 }
 
-var file_spindle_mill_v1_mill_proto_msgTypes = make([]protoimpl.MessageInfo, 17)
+var file_spindle_mill_v1_mill_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
+var file_spindle_mill_v1_mill_proto_msgTypes = make([]protoimpl.MessageInfo, 18)
 var file_spindle_mill_v1_mill_proto_goTypes = []any{
-	(*Hello)(nil),          // 0: spindle.mill.v1.Hello
-	(*Resume)(nil),         // 1: spindle.mill.v1.Resume
-	(*EngineSnapshot)(nil), // 2: spindle.mill.v1.EngineSnapshot
-	(*NodeSnapshot)(nil),   // 3: spindle.mill.v1.NodeSnapshot
-	(*ReserveSeat)(nil),    // 4: spindle.mill.v1.ReserveSeat
-	(*ReserveResult)(nil),  // 5: spindle.mill.v1.ReserveResult
-	(*Secret)(nil),         // 6: spindle.mill.v1.Secret
-	(*CommitLease)(nil),    // 7: spindle.mill.v1.CommitLease
-	(*Committed)(nil),      // 8: spindle.mill.v1.Committed
-	(*ReleaseLease)(nil),   // 9: spindle.mill.v1.ReleaseLease
-	(*CancelAttempt)(nil),  // 10: spindle.mill.v1.CancelAttempt
-	(*StatusEvent)(nil),    // 11: spindle.mill.v1.StatusEvent
-	(*LogLine)(nil),        // 12: spindle.mill.v1.LogLine
-	(*AttemptResult)(nil),  // 13: spindle.mill.v1.AttemptResult
-	(*Ack)(nil),            // 14: spindle.mill.v1.Ack
-	(*Message)(nil),        // 15: spindle.mill.v1.Message
-	nil,                    // 16: spindle.mill.v1.NodeSnapshot.EnginesEntry
+	(RejectClass)(0),           // 0: spindle.mill.v1.RejectClass
+	(*Hello)(nil),              // 1: spindle.mill.v1.Hello
+	(*Resume)(nil),             // 2: spindle.mill.v1.Resume
+	(*EngineAvailability)(nil), // 3: spindle.mill.v1.EngineAvailability
+	(*NodeSnapshot)(nil),       // 4: spindle.mill.v1.NodeSnapshot
+	(*ReserveSeat)(nil),        // 5: spindle.mill.v1.ReserveSeat
+	(*ReserveResult)(nil),      // 6: spindle.mill.v1.ReserveResult
+	(*Secret)(nil),             // 7: spindle.mill.v1.Secret
+	(*CommitLease)(nil),        // 8: spindle.mill.v1.CommitLease
+	(*Committed)(nil),          // 9: spindle.mill.v1.Committed
+	(*ReleaseLease)(nil),       // 10: spindle.mill.v1.ReleaseLease
+	(*CancelAttempt)(nil),      // 11: spindle.mill.v1.CancelAttempt
+	(*StatusEvent)(nil),        // 12: spindle.mill.v1.StatusEvent
+	(*LogLine)(nil),            // 13: spindle.mill.v1.LogLine
+	(*AttemptResult)(nil),      // 14: spindle.mill.v1.AttemptResult
+	(*Ack)(nil),                // 15: spindle.mill.v1.Ack
+	(*Message)(nil),            // 16: spindle.mill.v1.Message
+	nil,                        // 17: spindle.mill.v1.EngineAvailability.LoadEntry
+	nil,                        // 18: spindle.mill.v1.NodeSnapshot.EnginesEntry
 }
 var file_spindle_mill_v1_mill_proto_depIdxs = []int32{
-	16, // 0: spindle.mill.v1.NodeSnapshot.engines:type_name -> spindle.mill.v1.NodeSnapshot.EnginesEntry
-	6,  // 1: spindle.mill.v1.CommitLease.secrets:type_name -> spindle.mill.v1.Secret
-	0,  // 2: spindle.mill.v1.Message.hello:type_name -> spindle.mill.v1.Hello
-	1,  // 3: spindle.mill.v1.Message.resume:type_name -> spindle.mill.v1.Resume
-	3,  // 4: spindle.mill.v1.Message.node_snapshot:type_name -> spindle.mill.v1.NodeSnapshot
-	4,  // 5: spindle.mill.v1.Message.reserve_seat:type_name -> spindle.mill.v1.ReserveSeat
-	5,  // 6: spindle.mill.v1.Message.reserve_result:type_name -> spindle.mill.v1.ReserveResult
-	7,  // 7: spindle.mill.v1.Message.commit_lease:type_name -> spindle.mill.v1.CommitLease
-	8,  // 8: spindle.mill.v1.Message.committed:type_name -> spindle.mill.v1.Committed
-	9,  // 9: spindle.mill.v1.Message.release_lease:type_name -> spindle.mill.v1.ReleaseLease
-	10, // 10: spindle.mill.v1.Message.cancel_attempt:type_name -> spindle.mill.v1.CancelAttempt
-	11, // 11: spindle.mill.v1.Message.status_event:type_name -> spindle.mill.v1.StatusEvent
-	12, // 12: spindle.mill.v1.Message.log_line:type_name -> spindle.mill.v1.LogLine
-	13, // 13: spindle.mill.v1.Message.attempt_result:type_name -> spindle.mill.v1.AttemptResult
-	14, // 14: spindle.mill.v1.Message.ack:type_name -> spindle.mill.v1.Ack
-	2,  // 15: spindle.mill.v1.NodeSnapshot.EnginesEntry.value:type_name -> spindle.mill.v1.EngineSnapshot
-	16, // [16:16] is the sub-list for method output_type
-	16, // [16:16] is the sub-list for method input_type
-	16, // [16:16] is the sub-list for extension type_name
-	16, // [16:16] is the sub-list for extension extendee
-	0,  // [0:16] is the sub-list for field type_name
+	17, // 0: spindle.mill.v1.EngineAvailability.load:type_name -> spindle.mill.v1.EngineAvailability.LoadEntry
+	18, // 1: spindle.mill.v1.NodeSnapshot.engines:type_name -> spindle.mill.v1.NodeSnapshot.EnginesEntry
+	0,  // 2: spindle.mill.v1.ReserveResult.reject_class:type_name -> spindle.mill.v1.RejectClass
+	7,  // 3: spindle.mill.v1.CommitLease.secrets:type_name -> spindle.mill.v1.Secret
+	1,  // 4: spindle.mill.v1.Message.hello:type_name -> spindle.mill.v1.Hello
+	2,  // 5: spindle.mill.v1.Message.resume:type_name -> spindle.mill.v1.Resume
+	4,  // 6: spindle.mill.v1.Message.node_snapshot:type_name -> spindle.mill.v1.NodeSnapshot
+	5,  // 7: spindle.mill.v1.Message.reserve_seat:type_name -> spindle.mill.v1.ReserveSeat
+	6,  // 8: spindle.mill.v1.Message.reserve_result:type_name -> spindle.mill.v1.ReserveResult
+	8,  // 9: spindle.mill.v1.Message.commit_lease:type_name -> spindle.mill.v1.CommitLease
+	9,  // 10: spindle.mill.v1.Message.committed:type_name -> spindle.mill.v1.Committed
+	10, // 11: spindle.mill.v1.Message.release_lease:type_name -> spindle.mill.v1.ReleaseLease
+	11, // 12: spindle.mill.v1.Message.cancel_attempt:type_name -> spindle.mill.v1.CancelAttempt
+	12, // 13: spindle.mill.v1.Message.status_event:type_name -> spindle.mill.v1.StatusEvent
+	13, // 14: spindle.mill.v1.Message.log_line:type_name -> spindle.mill.v1.LogLine
+	14, // 15: spindle.mill.v1.Message.attempt_result:type_name -> spindle.mill.v1.AttemptResult
+	15, // 16: spindle.mill.v1.Message.ack:type_name -> spindle.mill.v1.Ack
+	3,  // 17: spindle.mill.v1.NodeSnapshot.EnginesEntry.value:type_name -> spindle.mill.v1.EngineAvailability
+	18, // [18:18] is the sub-list for method output_type
+	18, // [18:18] is the sub-list for method input_type
+	18, // [18:18] is the sub-list for extension type_name
+	18, // [18:18] is the sub-list for extension extendee
+	0,  // [0:18] is the sub-list for field type_name
 }
 
 func init() { file_spindle_mill_v1_mill_proto_init() }
@@ -1302,13 +1352,14 @@
 		File: protoimpl.DescBuilder{
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_spindle_mill_v1_mill_proto_rawDesc), len(file_spindle_mill_v1_mill_proto_rawDesc)),
-			NumEnums:      0,
-			NumMessages:   17,
+			NumEnums:      1,
+			NumMessages:   18,
 			NumExtensions: 0,
 			NumServices:   0,
 		},
 		GoTypes:           file_spindle_mill_v1_mill_proto_goTypes,
 		DependencyIndexes: file_spindle_mill_v1_mill_proto_depIdxs,
+		EnumInfos:         file_spindle_mill_v1_mill_proto_enumTypes,
 		MessageInfos:      file_spindle_mill_v1_mill_proto_msgTypes,
 	}.Build()
 	File_spindle_mill_v1_mill_proto = out.File
diff --git a/spindle/mill/proto/spindle/mill/v1/mill.proto b/spindle/mill/proto/spindle/mill/v1/mill.proto
--- a/spindle/mill/proto/spindle/mill/v1/mill.proto
+++ b/spindle/mill/proto/spindle/mill/v1/mill.proto
@@ -15,10 +15,12 @@
   string node_id = 2 [(buf.validate.field).string.min_len = 1];
   // engine names this node can run ("microvm", "nixery").
   repeated string engines = 3;
-  // GOARCH of the node, so the mill won't place arch-incompatible jobs.
+  // GOARCH of the node, retained as an informational trait for logs.
   string arch = 4;
   // the highest relay offset the executor believes it has sent; a resume hint.
   uint64 last_offset = 5;
+  // opaque operator-defined labels used for runs_on matching.
+  repeated string labels = 6;
 }
 
 // Resume is the mill's reply to Hello. The executor replays every buffered
@@ -28,16 +30,14 @@
   uint64 ack_offset = 1;
 }
 
-// EngineSnapshot is the changing per-engine state on a node. free_seats is a
-// coarse "can you take more" hint; 0 also means draining (the mill treats a
-// draining node as a full one until it leaves). The resource fields are coarse
-// budget headroom for ranking only; the executor makes the real yes/no call in
-// ReserveResult.
-message EngineSnapshot {
-  uint32 free_seats = 1;
-  int64 free_memory_mib = 2;
-  int64 free_vcpus = 3;
-  int64 free_disk_mib = 4;
+// EngineAvailability is the changing per-engine state on a node. The executor
+// reports opaque load metrics (higher = more loaded); the mill ranks by the
+// worst load and uses the sum as a tie-breaker. The mill does not know what
+// the metric keys mean.
+message EngineAvailability {
+  bool available = 1;
+  // engine-defined load metrics. keys are opaque to the mill.
+  map<string, double> load = 2;
 }
 
 // NodeSnapshot is pushed on connect, periodically, and right after any state
@@ -45,7 +45,7 @@
 message NodeSnapshot {
   string node_id = 1;
   uint64 seq = 2;
-  map<string, EngineSnapshot> engines = 3;
+  map<string, EngineAvailability> engines = 3;
 }
 
 // ReserveSeat asks an executor to hold a seat for a job. Zero secrets ride this
@@ -64,13 +64,18 @@
   uint32 ttl_seconds = 7;
 }
 
+enum RejectClass {
+  REJECT_CLASS_UNSPECIFIED = 0;
+  REJECT_CLASS_TRANSIENT = 1;
+  REJECT_CLASS_INCOMPATIBLE = 2;
+}
+
 // ReserveResult is the executor's accept/reject for a ReserveSeat.
 message ReserveResult {
   string lease_id = 1 [(buf.validate.field).string.min_len = 1];
   bool accepted = 2;
   string reject_reason = 3;
-  // optional bid score the mill ranks accepted leases by (higher is better).
-  double score = 4;
+  RejectClass reject_class = 4;
 }
 
 // Secret is a single unlocked secret, sent only inside CommitLease.