agent endpoints: P2C worker balancing, routing-miss sentinel, api-key-scoped attach

- worker selection uses power-of-two-choices over live in-flight streams
  (InflightStreams / SpareStreams) instead of weighted-random over a
  reported load; the conn pick stays least-loaded. In-flight is observed
  locally so it is correct for adopted wires, and the reported Load hook is
  dropped from the endpoint path.
- the front tags its own routing misses (X-Livekit-Endpoint-Miss) so a relay
  caller can distinguish them from a worker app's own 404/405 and retry past
  them across the remaining candidates.
- a data-wire attach is bound to the connecting grant's api key, so a leaked
  attach token cannot bind a conn to another project's worker.
- remove the dead scope-hook machinery (it only fed the removed bus resolve)
  and rename the request identity from "scope" to apiKey.

pkg/agent/endpoint/conn.go

@@ -122,6 +122,17 @@ func (c *DataConn) OpenStreams() int {
 	return len(c.streams)
 }
 
+// SpareStreams reports how many more streams this conn can accept before its
+// per-conn cap; never negative.
+func (c *DataConn) SpareStreams() int {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if n := int(c.params.MaxStreamsPerConn) - len(c.streams); n > 0 {
+		return n
+	}
+	return 0
+}
+
 // HasCapacity reports whether a new stream may be opened.
 func (c *DataConn) HasCapacity() bool {
 	c.mu.Lock()

pkg/agent/endpoint/front.go

@@ -32,6 +32,24 @@ import (
 	"github.com/livekit/protocol/utils/guid"
 )
 
+// HeaderEndpointMiss marks a response the front produced ITSELF - a routing
+// miss (no matching route, wrong method, auth required, or no local capacity) -
+// as distinct from a response the worker's app returned through the bridge. A
+// relay caller keys its cross-node retry on this header, so a worker's own 404
+// (e.g. GET /users/999 for a missing user) is never mistaken for "this node
+// can't serve the path" and re-relayed. The value is the miss kind, so the
+// caller can surface the most informative aggregate status. The header is set
+// only on the private relay listener (see MarkMisses) and stripped by the relay
+// caller, so it never reaches a client.
+const HeaderEndpointMiss = "X-Livekit-Endpoint-Miss"
+
+const (
+	MissNotFound         = "notfound"
+	MissMethodNotAllowed = "methodnotallowed"
+	MissUnauthenticated  = "unauthenticated"
+	MissUnavailable      = "unavailable"
+)
+
 const (
 	// PathPrefix is the public route namespace: /agents/{deployment}/{path...}
 	PathPrefix = "/agents/"
@@ -45,15 +63,15 @@ const (
 	maxAttempts = 3
 )
 
-// ScopeResolver maps an inbound request to its project scope. It returns the
-// api key the request is authorized for (empty when unauthenticated) - the
-// service layer implements it from validated grants.
-type ScopeResolver func(r *http.Request) (apiKey string, authenticated bool)
+// APIKeyResolver maps an inbound request to the api key it is authorized for
+// (empty when unauthenticated) - the service layer implements it from validated
+// grants.
+type APIKeyResolver func(r *http.Request) (apiKey string, authenticated bool)
 
 type Front struct {
-	registry     *Registry
-	resolveScope ScopeResolver
-	logger       logger.Logger
+	registry      *Registry
+	resolveAPIKey APIKeyResolver
+	logger        logger.Logger
 
 	// fallback is consulted when nothing local can serve the request (no
 	// candidates, no route match, or every match without capacity); a
@@ -62,21 +80,25 @@ type Front struct {
 	fallback Fallback
 	// see WithSingleKeyFallback
 	singleKeyFallback bool
+	// see MarkMisses: set on the private relay listener so a relay caller can
+	// tell a routing miss from a worker-app response
+	markMisses bool
 }
 
-func NewFront(registry *Registry, resolveScope ScopeResolver, log logger.Logger) *Front {
+func NewFront(registry *Registry, resolveAPIKey APIKeyResolver, log logger.Logger) *Front {
 	return &Front{
-		registry:     registry,
-		resolveScope: resolveScope,
-		logger:       log.WithComponent("agents.endpoint"),
+		registry:      registry,
+		resolveAPIKey: resolveAPIKey,
+		logger:        log.WithComponent("agents.endpoint"),
 	}
 }
 
 // FallbackRequest describes a request nothing local could serve. The request
 // body is untouched when the fallback runs.
 type FallbackRequest struct {
-	// Scope is the project identity the front resolved (api key in OSS)
-	Scope         string
+	// APIKey is the identity the front resolved the request to (empty when
+	// unauthenticated)
+	APIKey        string
 	Authenticated bool
 	Deployment    string
 	// Path within the deployment, '/'-rooted
@@ -96,9 +118,28 @@ func (f *Front) WithFallback(fb Fallback) *Front {
 	return f
 }
 
+// MarkMisses tags the front's own routing-miss responses with HeaderEndpointMiss
+// so a relay caller can distinguish them from worker-app responses. Set it on
+// the private relay listener only; the public front must not (the header would
+// leak to clients, and its misses are final anyway).
+func (f *Front) MarkMisses() *Front {
+	f.markMisses = true
+	return f
+}
+
+// writeMiss writes a front-originated miss, tagging it with the kind when this
+// front marks misses (the relay listener) so the relay caller can retry past it
+// and aggregate the most informative status.
+func (f *Front) writeMiss(w http.ResponseWriter, status int, kind, msg string) {
+	if f.markMisses {
+		w.Header().Set(HeaderEndpointMiss, kind)
+	}
+	http.Error(w, msg, status)
+}
+
 // WithSingleKeyFallback resolves unauthenticated requests to the registry's
-// single api key when the scope resolver yields none. Self-hosted convenience
-// only: a multi-tenant front must never guess a scope from what happens to be
+// single api key when the resolver yields none. Self-hosted convenience only: a
+// multi-tenant front must never guess an api key from what happens to be
 // registered.
 func (f *Front) WithSingleKeyFallback() *Front {
 	f.singleKeyFallback = true
@@ -123,7 +164,7 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	isWS := isWebSocketUpgrade(r)
 
-	apiKey, authenticated := f.resolveScope(r)
+	apiKey, authenticated := f.resolveAPIKey(r)
 	if apiKey == "" && f.singleKeyFallback {
 		// unauthenticated: OSS serves public routes when the worker fleet
 		// belongs to a single key
@@ -131,14 +172,14 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 	if apiKey == "" {
 		w.Header().Set("WWW-Authenticate", "Bearer")
-		http.Error(w, "authentication required", http.StatusUnauthorized)
+		f.writeMiss(w, http.StatusUnauthorized, MissUnauthenticated, "authentication required")
 		return
 	}
 
 	candidates := f.registry.Candidates(apiKey, deployment)
 	if len(candidates) == 0 && f.fallback == nil {
 		w.Header().Set("Retry-After", "1")
-		http.Error(w, "no workers available for deployment", http.StatusServiceUnavailable)
+		f.writeMiss(w, http.StatusServiceUnavailable, MissUnavailable, "no workers available for deployment")
 		return
 	}
 
@@ -167,14 +208,14 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	if route == nil && f.fallback != nil {
 		// nothing local can serve: hand off before the local status mapping
 		if f.fallback(w, r, &FallbackRequest{
-			Scope: apiKey, Authenticated: authenticated,
+			APIKey: apiKey, Authenticated: authenticated,
 			Deployment: deployment, Path: path, WebSocket: isWS,
 		}) {
 			return
 		}
 		if len(candidates) == 0 && !restricted && !partial {
 			w.Header().Set("Retry-After", "1")
-			http.Error(w, "no workers available for deployment", http.StatusServiceUnavailable)
+			f.writeMiss(w, http.StatusServiceUnavailable, MissUnavailable, "no workers available for deployment")
 			return
 		}
 	}
@@ -182,9 +223,9 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		switch {
 		case restricted:
 			w.Header().Set("WWW-Authenticate", "Bearer")
-			http.Error(w, "authentication required", http.StatusUnauthorized)
+			f.writeMiss(w, http.StatusUnauthorized, MissUnauthenticated, "authentication required")
 		case partial:
-			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+			f.writeMiss(w, http.StatusMethodNotAllowed, MissMethodNotAllowed, "method not allowed")
 		default:
 			for _, reg := range candidates {
 				if alt, ok := reg.Manifest.RedirectSlashes(path, r.Method, isWS); ok {
@@ -194,7 +235,7 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 					return
 				}
 			}
-			http.NotFound(w, r)
+			f.writeMiss(w, http.StatusNotFound, MissNotFound, "not found")
 		}
 		return
 	}
@@ -204,7 +245,7 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	attempted := make(map[*Registration]bool)
 	for attempt := 0; attempt < maxAttempts; attempt++ {
-		reg := pickWeighted(matched, attempted)
+		reg := pickWorker(matched, attempted)
 		if reg == nil {
 			break
 		}
@@ -223,7 +264,7 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	// retryable.
 	if bodyConsumed == 0 && f.fallback != nil {
 		if f.fallback(w, r, &FallbackRequest{
-			Scope: apiKey, Authenticated: authenticated,
+			APIKey: apiKey, Authenticated: authenticated,
 			Deployment: deployment, Path: path, WebSocket: isWS,
 		}) {
 			return
@@ -231,38 +272,49 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 
 	w.Header().Set("Retry-After", "1")
-	http.Error(w, "no worker could serve the request", http.StatusServiceUnavailable)
+	f.writeMiss(w, http.StatusServiceUnavailable, MissUnavailable, "no worker could serve the request")
 }
 
-// pickWeighted is capacity-weighted random over non-draining workers with
-// attached data connections.
-func pickWeighted(regs []*Registration, ignore map[*Registration]bool) *Registration {
-	var sum float32
-	weights := make([]float32, len(regs))
-	for i, reg := range regs {
+// pickWorker chooses a worker by the power of two choices: sample two eligible
+// registrations at random and take the one with fewer in-flight streams (least
+// outstanding requests). This approximates optimal load spreading without global
+// coordination or the herding of exact least-loaded, and - unlike a
+// self-reported load - the in-flight count is observed here on this node's own
+// data conns, so it is accurate for adopted (satellite) registrations too.
+// Eligible = not already attempted, has attached conns, not draining.
+func pickWorker(regs []*Registration, ignore map[*Registration]bool) *Registration {
+	eligible := regs[:0:0]
+	for _, reg := range regs {
 		if ignore[reg] || reg.AttachedConns() == 0 {
 			continue
 		}
 		if reg.Draining != nil && reg.Draining() {
 			continue
 		}
-		w := float32(1)
-		if reg.Load != nil {
-			w = max(0.01, 1-reg.Load())
-		}
-		weights[i] = w
-		sum += w
+		eligible = append(eligible, reg)
 	}
-	if sum == 0 {
+	if len(eligible) == 0 {
 		return nil
 	}
-	target := rand.Float32() * sum
-	for i, reg := range regs {
-		if target -= weights[i]; weights[i] > 0 && target <= 0 {
-			return reg
-		}
+	return eligible[p2c(len(eligible), func(i int) int { return eligible[i].InflightStreams() })]
+}
+
+// p2c returns the index of the less-loaded of two distinct random draws from
+// [0,n) (n >= 1). With n == 2 both are always sampled, so it is exact; larger n
+// trades a little optimality for O(1) work and no herding.
+func p2c(n int, load func(int) int) int {
+	if n == 1 {
+		return 0
 	}
-	return nil
+	i := rand.IntN(n)
+	j := rand.IntN(n - 1)
+	if j >= i { // fold to a distinct second draw
+		j++
+	}
+	if load(i) <= load(j) {
+		return i
+	}
+	return j
 }
 
 // bridge runs one attempt against one worker. done means a response (or abort)

pkg/agent/endpoint/pick_test.go

@@ -0,0 +1,42 @@
+// Copyright 2026 LiveKit, Inc.
+
+package endpoint
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestP2CChoice(t *testing.T) {
+	// a single candidate is always chosen
+	require.Equal(t, 0, p2c(1, func(int) int { return 42 }))
+
+	// with two candidates both are always drawn, so the lower-load one wins
+	// deterministically regardless of the random draw
+	for i := 0; i < 100; i++ {
+		require.Equal(t, 1, p2c(2, func(i int) int { return []int{5, 2}[i] }))
+		require.Equal(t, 0, p2c(2, func(i int) int { return []int{2, 5}[i] }))
+	}
+
+	// larger n: the draws are always distinct and in range
+	for i := 0; i < 500; i++ {
+		idx := p2c(5, func(int) int { return 0 })
+		require.GreaterOrEqual(t, idx, 0)
+		require.Less(t, idx, 5)
+	}
+
+	// one clearly-idle candidate among four busy ones: whenever it is one of the
+	// two draws it wins, so it is picked far more than its 1/n share and the busy
+	// ones are relieved (this is the whole point of power-of-two-choices)
+	loads := []int{0, 100, 100, 100, 100}
+	low := 0
+	for i := 0; i < 4000; i++ {
+		if p2c(len(loads), func(i int) int { return loads[i] }) == 0 {
+			low++
+		}
+	}
+	// P(idle drawn) = 1 - (4/5)(3/4) = 0.4, and it always wins when drawn
+	require.Greater(t, low, 1000)
+	require.Less(t, low, 2400)
+}

pkg/agent/endpoint/registry.go

@@ -63,9 +63,9 @@ type Registration struct {
 	Settings  Settings
 	Logger    logger.Logger
 
-	// Load and Draining are provided by the control-plane layer that owns the
-	// worker (reported load rides UpdateWorkerStatus on the control connection).
-	Load     func() float32
+	// Draining is provided by the control-plane layer that owns the worker; a
+	// draining worker takes no new streams. Worker selection uses live in-flight
+	// streams, not a reported load, so no load hook is needed here.
 	Draining func() bool
 
 	// pendingAttaches counts slots reserved by validated-but-unadopted wires so
@@ -199,6 +199,36 @@ func (r *Registration) AttachedConns() int {
 	return len(r.conns)
 }
 
+// InflightStreams reports the open streams across this registration's data
+// conns - the requests currently being served from this node. It is the
+// least-outstanding-requests signal for worker selection: a per-node view (each
+// node counts only the streams it opened), which is exactly what a node needs
+// to spread its own traffic, and it needs no load reported by the worker.
+func (r *Registration) InflightStreams() int {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	n := 0
+	for _, c := range r.conns {
+		n += c.OpenStreams()
+	}
+	return n
+}
+
+// SpareStreams reports the total spare stream capacity across this
+// registration's data conns on this node - how many more concurrent requests it
+// can accept. It is the node's live, locally-observed serving headroom for the
+// worker; being measured on the conns this node holds, it is accurate for
+// adopted (satellite) registrations too, which report no load.
+func (r *Registration) SpareStreams() int {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	n := 0
+	for _, c := range r.conns {
+		n += c.SpareStreams()
+	}
+	return n
+}
+
 func (r *Registration) close() {
 	r.mu.Lock()
 	if r.closed {
@@ -221,20 +251,6 @@ type Registry struct {
 	mu    sync.Mutex
 	regs  map[string]*Registration // by worker id
 	byKey map[regKey][]*Registration
-
-	// scope hooks fire when a scope (api key / project) gains its first or
-	// loses its last registration; the Remote uses them to (de)register the
-	// resolve topic on the bus
-	scopeCounts   map[string]int
-	onScopeActive func(scope string)
-	onScopeIdle   func(scope string)
-
-	// hookMu serializes scope hook invocations. Transitions are decided against
-	// the registry's CURRENT state under hookMu, never from values computed
-	// earlier: a reconnect's activation racing the old connection's idle must
-	// not leave a live scope without its resolve topic.
-	hookMu        sync.Mutex
-	scopeNotified map[string]bool
 }
 
 type regKey struct {
@@ -244,62 +260,8 @@ type regKey struct {
 
 func NewRegistry() *Registry {
 	return &Registry{
-		regs:          make(map[string]*Registration),
-		byKey:         make(map[regKey][]*Registration),
-		scopeCounts:   make(map[string]int),
-		scopeNotified: make(map[string]bool),
-	}
-}
-
-// notifyScope reconciles the scope's hook state with the registry's current
-// truth. Racing register/deregister notifications converge instead of
-// interleaving.
-func (g *Registry) notifyScope(scope string) {
-	g.hookMu.Lock()
-	defer g.hookMu.Unlock()
-	g.mu.Lock()
-	active := g.scopeCounts[scope] > 0
-	activeHook, idleHook := g.onScopeActive, g.onScopeIdle
-	g.mu.Unlock()
-	if activeHook == nil && idleHook == nil {
-		return
-	}
-	if g.scopeNotified[scope] == active {
-		return
-	}
-	if active {
-		g.scopeNotified[scope] = true
-		activeHook(scope)
-	} else {
-		delete(g.scopeNotified, scope)
-		idleHook(scope)
-	}
-}
-
-// SetScopeHooks installs the scope activation callbacks, replaying currently
-// active scopes so a resolver layered on after registrations still registers
-// its topics. Multi-node deployments use this to answer endpoint resolves only
-// while they hold registrations for a scope.
-func (g *Registry) SetScopeHooks(active, idle func(scope string)) {
-	g.hookMu.Lock()
-	defer g.hookMu.Unlock()
-	g.mu.Lock()
-	g.onScopeActive = active
-	g.onScopeIdle = idle
-	scopes := make([]string, 0, len(g.scopeCounts))
-	for scope, n := range g.scopeCounts {
-		if n > 0 {
-			scopes = append(scopes, scope)
-		}
-	}
-	g.mu.Unlock()
-	if active != nil {
-		for _, s := range scopes {
-			if !g.scopeNotified[s] {
-				g.scopeNotified[s] = true
-				active(s)
-			}
-		}
+		regs:  make(map[string]*Registration),
+		byKey: make(map[regKey][]*Registration),
 	}
 }
 
@@ -321,13 +283,8 @@ func (g *Registry) Register(r *Registration) error {
 	}
 	g.regs[r.WorkerID] = r
 	g.byKey[key] = append(g.byKey[key], r)
-	g.scopeCounts[r.APIKey]++
 	g.mu.Unlock()
-	g.notifyScope(r.APIKey)
 	if old != nil {
-		if old.APIKey != r.APIKey {
-			g.notifyScope(old.APIKey)
-		}
 		old.close()
 	}
 	return nil
@@ -347,10 +304,6 @@ func (g *Registry) removeLocked(r *Registration) {
 			g.byKey[key] = regs
 		}
 	}
-	g.scopeCounts[r.APIKey]--
-	if g.scopeCounts[r.APIKey] == 0 {
-		delete(g.scopeCounts, r.APIKey)
-	}
 }
 
 // Deregister removes exactly this registration; it is a no-op when a newer
@@ -363,31 +316,40 @@ func (g *Registry) Deregister(r *Registration) {
 	}
 	g.removeLocked(r)
 	g.mu.Unlock()
-	g.notifyScope(r.APIKey)
 	r.close()
 }
 
 // ValidateAttach checks an attach without adopting the connection, so the
-// attach response can be written before stream frames may flow.
-func (g *Registry) ValidateAttach(workerID, instanceID, token string) error {
+// attach response can be written before stream frames may flow. apiKey is the
+// project the connecting grant is authorized for; it must match the
+// registration's, so a leaked attach token is useless without a grant for the
+// worker's own project (defense in depth over the token secret).
+func (g *Registry) ValidateAttach(workerID, instanceID, apiKey, token string) error {
 	g.mu.Lock()
 	r, ok := g.regs[workerID]
 	g.mu.Unlock()
 	if !ok {
 		return ErrUnknownWorker
 	}
+	if apiKey != r.APIKey {
+		return ErrAttachRejected // wrong project: terminal, never adopt-retried
+	}
 	return r.validateAttach(instanceID, token)
 }
 
 // BeginAttach validates an attach and reserves a pool slot; the caller writes
-// the ack and then Completes (or Aborts) the ticket.
-func (g *Registry) BeginAttach(workerID, instanceID, token string) (*AttachTicket, error) {
+// the ack and then Completes (or Aborts) the ticket. apiKey binds the attach to
+// the connecting grant's project (see ValidateAttach).
+func (g *Registry) BeginAttach(workerID, instanceID, apiKey, token string) (*AttachTicket, error) {
 	g.mu.Lock()
 	r, ok := g.regs[workerID]
 	g.mu.Unlock()
 	if !ok {
 		return nil, ErrUnknownWorker
 	}
+	if apiKey != r.APIKey {
+		return nil, ErrAttachRejected // wrong project: terminal, never adopt-retried
+	}
 	return r.beginAttach(instanceID, token)
 }
 

pkg/agent/endpoint/registry_test.go

@@ -34,8 +34,11 @@ func TestRegistrySupersede(t *testing.T) {
 	require.NoError(t, g.Register(newReg))
 
 	require.Equal(t, []*Registration{newReg}, g.Candidates("key", "production"))
-	require.Error(t, g.ValidateAttach("AW_1", "i-1", "ATT_i-1"), "old epoch must not attach")
-	require.NoError(t, g.ValidateAttach("AW_1", "i-2", "ATT_i-2"))
+	require.Error(t, g.ValidateAttach("AW_1", "i-1", "key", "ATT_i-1"), "old epoch must not attach")
+	require.NoError(t, g.ValidateAttach("AW_1", "i-2", "key", "ATT_i-2"))
+	// a grant for another project cannot attach even with the right token
+	require.ErrorIs(t, g.ValidateAttach("AW_1", "i-2", "other", "ATT_i-2"), ErrAttachRejected,
+		"wrong-project grant must not attach")
 
 	// the old control connection tears down after the new one registered
 	g.Deregister(oldReg)

pkg/service/agentservice.go

@@ -144,8 +144,8 @@ type AgentService struct {
 type AgentHandler struct {
 	agentServer rpc.AgentInternalServer
 	// the server's only configured api key, when exactly one exists: the
-	// unauthenticated scope for public endpoints regardless of which node holds
-	// the workers
+	// unauthenticated identity for public endpoints regardless of which node
+	// holds the workers
 	singleAPIKey string
 	mu           sync.Mutex
 	logger       logger.Logger
@@ -222,14 +222,14 @@ func NewAgentService(
 }
 
 // EndpointFront is the /agents/{deployment}/{path...} handler backed by this
-// node's attached workers. Project scope comes from validated grants when a
+// node's attached workers. The api key comes from validated grants when a
 // token is present; unauthenticated requests reach public endpoints only.
 func (s *AgentService) EndpointFront() http.Handler {
 	front := endpoint.NewFront(s.endpointRegistry, func(r *http.Request) (string, bool) {
 		if claims := GetGrants(r.Context()); claims != nil {
 			return GetAPIKey(r.Context()), true
 		}
-		// unauthenticated: with a single configured key the scope is
+		// unauthenticated: with a single configured key the api key is
 		// unambiguous even when this node holds no registrations (multi-node)
 		return s.singleAPIKey, false
 	}, s.logger)
@@ -251,8 +251,10 @@ func (s *AgentService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		if attach {
 			// data wire: no registration handshake, no signal loop; the wire
 			// speaks AgentHttp.Frame exclusively and the endpoint mux owns it
-			// after a successful attach
-			HandleEndpointAttach(s.endpointRegistry, NewEndpointWireConn(conn), s.wireParams(), nil)
+			// after a successful attach. The connecting grant's api key binds the
+			// attach to its own project (a leaked attach token alone cannot bind a
+			// data conn to another project's worker).
+			HandleEndpointAttach(s.endpointRegistry, NewEndpointWireConn(conn), s.wireParams(), nil, GetAPIKey(r.Context()))
 			return
 		}
 
@@ -346,14 +348,17 @@ func (s *AgentService) wireParams() endpoint.WireParams {
 // registry - behind a load balancer the wire may land on a node other than the
 // one holding the registration. The adopter may install a local registration
 // (e.g. a satellite fetched from the holder); afterwards the attach is retried
-// locally once. nil rejects unknown workers.
-type AttachAdopter func(a *livekit.AgentHttp_AttachDataConnection) error
+// locally once. apiKey is the identity the connecting grant is authorized for:
+// the adopter must refuse to install a registration for any other project, so
+// an unentitled grant cannot adopt (or even probe) another project's worker.
+// nil rejects unknown workers.
+type AttachAdopter func(a *livekit.AgentHttp_AttachDataConnection, apiKey string) error
 
 // HandleEndpointAttach adopts a worker-dialed data wire: the first frame is
 // Attach on stream 0, validated against the registration's epoch and token; the
 // response carries this node's wire parameters. Shared by OSS and cloud
 // servers.
-func HandleEndpointAttach(registry *endpoint.Registry, wire endpoint.WireConn, params endpoint.WireParams, adopt AttachAdopter) {
+func HandleEndpointAttach(registry *endpoint.Registry, wire endpoint.WireConn, params endpoint.WireParams, adopt AttachAdopter, apiKey string) {
 	if err := wire.SetReadDeadline(time.Now().Add(agent.RegisterTimeout)); err != nil {
 		_ = wire.Close()
 		return
@@ -389,14 +394,14 @@ func HandleEndpointAttach(registry *endpoint.Registry, wire endpoint.WireConn, p
 	// the slot is reserved before the ack is written (never a success ack for a
 	// wire that then loses the cap race) and adopted only after it, so the
 	// worker cannot observe stream frames ahead of the attach outcome
-	ticket, err := registry.BeginAttach(a.GetWorkerId(), a.GetInstanceId(), a.GetAttachToken())
+	ticket, err := registry.BeginAttach(a.GetWorkerId(), a.GetInstanceId(), apiKey, a.GetAttachToken())
 	if (errors.Is(err, endpoint.ErrUnknownWorker) || errors.Is(err, endpoint.ErrWrongEpoch)) && adopt != nil {
-		if aerr := adopt(a); aerr != nil {
+		if aerr := adopt(a, apiKey); aerr != nil {
 			_ = respond(aerr.Error())
 			_ = wire.Close()
 			return
 		}
-		ticket, err = registry.BeginAttach(a.GetWorkerId(), a.GetInstanceId(), a.GetAttachToken())
+		ticket, err = registry.BeginAttach(a.GetWorkerId(), a.GetInstanceId(), apiKey, a.GetAttachToken())
 	}
 	if err != nil {
 		_ = respond(err.Error())
@@ -516,7 +521,6 @@ func (h *AgentHandler) registerEndpoints(w *agent.Worker) *endpoint.Registration
 			DataConnCount: settings.GetDataConnectionCount(),
 		},
 		Logger:   w.Logger(),
-		Load:     w.Load,
 		Draining: w.Draining,
 	}
 	if err := h.endpointRegistry.Register(reg); err != nil {