agent endpoint: the thread grant, scoped to the agent's endpoints

livekit/protocol at the head of PR 1843: the agent session is a thread. The
preamble carries thread_id and thread_grant, the forged header is
X-LiveKit-Thread-Grant, and a grant that lists endpoints opens only those: the
first path segment under the deployment, 403 outside it before any worker sees
the request. The local protocol replace goes with the pin.

go.mod

@@ -21,8 +21,8 @@ require (
 	github.com/jxskiss/base62 v1.1.0
 	github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731
 	github.com/livekit/mediatransportutil v0.0.0-20260821083140-f234b534b095
-	github.com/livekit/protocol v1.52.2-0.20261006063934-0761f135e913
-	github.com/livekit/psrpc v0.8.1
+	github.com/livekit/protocol v1.52.2-0.20261008023442-60589e83886b
+	github.com/livekit/psrpc v0.8.2
 	github.com/mackerelio/go-osstat v0.2.8
 	github.com/magefile/mage v1.17.2
 	github.com/mitchellh/go-homedir v1.1.0
@@ -156,7 +156,7 @@ require (
 	golang.org/x/tools v0.51.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20261005182115-fad411399dd8 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20261005182115-fad411399dd8 // indirect
-	google.golang.org/grpc v1.84.0 // indirect
+	google.golang.org/grpc v1.84.0
 )
 
 tool (

go.sum

@@ -166,10 +166,10 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT
 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ=
 github.com/livekit/mediatransportutil v0.0.0-20260821083140-f234b534b095 h1:BcliKAXoMhl/nWmzQweQ5kmh4Qqagxl4s3Z5pvM/7AY=
 github.com/livekit/mediatransportutil v0.0.0-20260821083140-f234b534b095/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU=
-github.com/livekit/protocol v1.52.2-0.20261006063934-0761f135e913 h1:LopYipWFeZbKrjRFSYgX7GBJx6FBQ2mvtcoZU4+ynJM=
-github.com/livekit/protocol v1.52.2-0.20261006063934-0761f135e913/go.mod h1:EJRuXu4hCsje+SiDnPGJ9U2wpFbwTl6bzC4ZxPSdRec=
-github.com/livekit/psrpc v0.8.1 h1:aeEnC2WRWB2tzNpmrifog8B+DD9+0K2Fk8xT/0kMuxg=
-github.com/livekit/psrpc v0.8.1/go.mod h1:6KnU8vULoizX0Er9NNfeKJDV8ug87j3fuo19M0bhj6Q=
+github.com/livekit/protocol v1.52.2-0.20261008023442-60589e83886b h1:BX9CZ2H6evJPiyr9l7mz3zQhZk0u5vIuTWj6v7kPWh4=
+github.com/livekit/protocol v1.52.2-0.20261008023442-60589e83886b/go.mod h1:k7v95/DNhMkslMeBxdQ+lQ6gKc8hMNGYB7t0+6sEGU4=
+github.com/livekit/psrpc v0.8.2 h1:7tbkhBJoYlK5Csfkdv1XYbZjyAQeaEZgupvIXNCv7Dc=
+github.com/livekit/psrpc v0.8.2/go.mod h1:6KnU8vULoizX0Er9NNfeKJDV8ug87j3fuo19M0bhj6Q=
 github.com/livekit/webrtc-pion/v4 v4.2.18-warp.6 h1:UaGYlQ/Ma73LN12xPDlH3OBYNfOEDTVOtHdRZbHg2dY=
 github.com/livekit/webrtc-pion/v4 v4.2.18-warp.6/go.mod h1:Kxgo6xwFzVK38XVlQTBT1zj2JU5G7R0R0pa5w5d9d/Q=
 github.com/mackerelio/go-osstat v0.2.8 h1:I2duicTaCGWoM53XwAwA9OIe1inu0xnVs8/pqOWWVr4=

pkg/agent/endpoint/attempt.go

@@ -34,10 +34,10 @@ import (
 	"github.com/livekit/protocol/utils/guid"
 )
 
-// sessionGrantHeader is the header a worker reached without a node accepts the
+// threadGrantHeader is the header a worker reached without a node accepts the
 // grant on. Through a node the grant travels in the preamble, so this header is
 // reserved to the node and stripped from every forwarded request.
-const sessionGrantHeader = "X-LiveKit-Session-Grant"
+const threadGrantHeader = "X-LiveKit-Thread-Grant"
 
 // attempt is the per-request state shared across worker attempts.
 type attempt struct {
@@ -48,15 +48,15 @@ type attempt struct {
 	target    string
 	requestID string
 	granted   bool
-	// grant is the caller's verified agent session grant, nil when the token
+	// grant is the caller's verified agent thread grant, nil when the token
 	// carried none. Every kind of route is told: the worker enforces it.
-	grant *livekit.AgentHttp_AgentSessionGrant
-	// sessionID is the agent session the front placed the request in; empty
-	// when it placed none. Only a text-kind declaration is told.
-	sessionID string
-	body      io.Reader
-	pools     *bridgePools
-	preamble  *livekit.AgentHttp_StreamPreamble
+	grant *livekit.AgentHttp_AgentThreadGrant
+	// threadID is the thread the front placed the request in; empty when it
+	// placed none. Only a text-kind declaration is told.
+	threadID string
+	body     io.Reader
+	pools    *bridgePools
+	preamble *livekit.AgentHttp_StreamPreamble
 	// the serialized canonical request head, built once and replayed verbatim
 	head []byte
 	// client body bytes consumed when this attempt began
@@ -107,11 +107,11 @@ func (a *attempt) newPreamble() *livekit.AgentHttp_StreamPreamble {
 	clientAddr, _, _ := net.SplitHostPort(a.req.RemoteAddr)
 
 	p := &livekit.AgentHttp_StreamPreamble{
-		RequestId:    a.requestID,
-		Authorized:   a.granted,
-		SessionGrant: a.grant,
-		ClientAddr:   clientAddr,
-		Scheme:       scheme,
+		RequestId:   a.requestID,
+		Authorized:  a.granted,
+		ThreadGrant: a.grant,
+		ClientAddr:  clientAddr,
+		Scheme:      scheme,
 	}
 	return p
 }
@@ -123,9 +123,9 @@ func (a *attempt) newPreamble() *livekit.AgentHttp_StreamPreamble {
 func (a *attempt) setRoute(w routeWorker) {
 	a.preamble.Route = w.raw
 	a.preamble.Kind = w.kind
-	a.preamble.SessionId = ""
+	a.preamble.ThreadId = ""
 	if w.kind == livekit.AgentHttp_AEK_TEXT {
-		a.preamble.SessionId = a.sessionID
+		a.preamble.ThreadId = a.threadID
 	}
 }
 
@@ -141,15 +141,15 @@ func (a *attempt) requestHeader() http.Header {
 		h.Set("Connection", "Upgrade")
 		h.Set("Upgrade", "websocket")
 	}
-	// identity travels as preamble.authorized and preamble.session_grant; the
+	// identity travels as preamble.authorized and preamble.thread_grant; the
 	// token must never reach the worker
 	if isLiveKitBearer(h.Get("Authorization")) {
 		h.Del("Authorization")
 	}
-	// the session grant is the node's to assert, from the verified token. A
+	// the thread grant is the node's to assert, from the verified token. A
 	// worker reached directly, with no node in front, may still read this
 	// header, so a client must not be able to plant one here.
-	h.Del(sessionGrantHeader)
+	h.Del(threadGrantHeader)
 	// the signalling namespace is the node's to write: a client able to set
 	// x-lk-completion could tell the worker a body it cut short was whole
 	wire.StripReservedHeaders(h)

pkg/agent/endpoint/front.go

@@ -61,8 +61,8 @@ const (
 	// worker cannot hold a client open by trickling them forever.
 	maxInformationalHeads = 8
 
-	// DefaultSessionPeek bounds the request body prefix a SessionResolver sees.
-	DefaultSessionPeek = 4 << 10
+	// DefaultThreadPeek bounds the request body prefix a ThreadResolver sees.
+	DefaultThreadPeek = 4 << 10
 
 	maxResponseHeadSize = 1 << 20
 	responseBufSize     = 8 << 10
@@ -119,10 +119,11 @@ type Access struct {
 	// curried on the deployment it resolved. nil means local misses are final.
 	Fallback Fallback
 	Level    AccessLevel
-	// SessionGrant is the verified token's agentSession claim, nil when the
+	// ThreadGrant is the verified token's agentThread claim, nil when the
 	// token carried none or there was no token. It reaches the worker in the
-	// preamble; the token itself never does.
-	SessionGrant *auth.AgentSessionGrant
+	// preamble; the token itself never does. Its endpoints, when it lists any,
+	// are checked here: a route outside them is forbidden.
+	ThreadGrant *auth.AgentThreadGrant
 }
 
 // AccessResolver maps an inbound request, plus the agent and deployment its URL
@@ -141,15 +142,15 @@ type Front struct {
 // false leaves them to the URL.
 type Identity func(r *http.Request) (agentName, deployment string, ok bool)
 
-// SessionResolver places a request on an AEK_TEXT route in an agent session.
-// It runs once per request, after access is resolved and before a worker is
-// picked, and only when a matched declaration is of the text kind: that kind is
-// the worker's consent to the front reading its body. body is the first bytes
-// of the request body, at most SessionPeek; the worker receives the whole body
-// regardless. Reporting false leaves the session to the worker, which resolves
-// it from the body itself. Cloud installs one backed by the session registry;
+// ThreadResolver places a request on an AEK_TEXT route in a thread. It runs
+// once per request, after access is resolved and before a worker is picked,
+// and only when a matched declaration is of the text kind: that kind is the
+// worker's consent to the front reading its body. body is the first bytes of
+// the request body, at most ThreadPeek; the worker receives the whole body
+// regardless. Reporting false leaves the thread to the worker, which resolves
+// it from the body itself. Cloud installs one backed by the threads service;
 // OSS leaves it nil, and a text route is then forwarded unchanged.
-type SessionResolver func(r *http.Request, agentName, deployment string, body []byte) (sessionID string, ok bool)
+type ThreadResolver func(r *http.Request, agentName, deployment string, body []byte) (threadID string, ok bool)
 
 // FrontParams configures a Front. Fields are read on every request once the
 // Front is serving, so none may change after construction.
@@ -158,12 +159,12 @@ type FrontParams struct {
 	Logger        logger.Logger
 	Identity      Identity
 
-	// ResolveSession places text-kind requests in an agent session. nil
-	// forwards them unchanged.
-	ResolveSession SessionResolver
-	// SessionPeek bounds the body prefix ResolveSession sees. 0 means
-	// DefaultSessionPeek.
-	SessionPeek int
+	// ResolveThread places text-kind requests in a thread. nil forwards them
+	// unchanged.
+	ResolveThread ThreadResolver
+	// ThreadPeek bounds the body prefix ResolveThread sees. 0 means
+	// DefaultThreadPeek.
+	ThreadPeek int
 
 	// MaxTunnelsPerWorker caps one worker's concurrent WebSocket tunnels. 0
 	// means half its session's MaxStreams.
@@ -278,9 +279,16 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	var sessionID string
-	if f.params.ResolveSession != nil && declaresText(matched) {
-		peek, err := peekBody(r, f.sessionPeek())
+	// a grant that lists endpoints covers this one or none: the endpoint is the
+	// first segment of the path under the deployment
+	if access.ThreadGrant != nil && !access.ThreadGrant.AllowsEndpoint(endpointOf(path)) {
+		http.Error(w, "forbidden", http.StatusForbidden)
+		return
+	}
+
+	var threadID string
+	if f.params.ResolveThread != nil && declaresText(matched) {
+		peek, err := peekBody(r, f.threadPeek())
 		if err != nil {
 			// the client's body failed before any worker saw the request, so
 			// there is nothing to place and nothing to retry
@@ -288,8 +296,8 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			http.Error(w, "bad request", http.StatusBadRequest)
 			return
 		}
-		if id, ok := f.params.ResolveSession(r, agentName, deployment, peek); ok {
-			sessionID = id
+		if id, ok := f.params.ResolveThread(r, agentName, deployment, peek); ok {
+			threadID = id
 		}
 	}
 
@@ -300,8 +308,8 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		target:    requestTarget(escPath, r.URL.RawQuery),
 		requestID: reqID,
 		granted:   granted,
-		grant:     access.SessionGrant.ToProto(),
-		sessionID: sessionID,
+		grant:     access.ThreadGrant.ToProto(),
+		threadID:  threadID,
 		webSocket: isWebSocketUpgrade(r),
 		pools:     f.pools,
 	}
@@ -363,15 +371,22 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	f.writeUnavailable(w, "no worker could serve the request")
 }
 
-func (f *Front) sessionPeek() int {
-	if f.params.SessionPeek > 0 {
-		return f.params.SessionPeek
+func (f *Front) threadPeek() int {
+	if f.params.ThreadPeek > 0 {
+		return f.params.ThreadPeek
 	}
-	return DefaultSessionPeek
+	return DefaultThreadPeek
+}
+
+// endpointOf is the endpoint a decoded path addresses, its first segment: the
+// name every door of the agent is reached under, {agent_name}/{endpoint}.
+func endpointOf(path string) string {
+	endpoint, _, _ := strings.Cut(strings.TrimPrefix(path, "/"), "/")
+	return endpoint
 }
 
 // declaresText reports whether any declaration the request matched is of the
-// text kind. Declarers of one route may disagree, so the session is resolved
+// text kind. Declarers of one route may disagree, so the thread is resolved
 // once and handed only to the workers that asked for it (setRoute).
 func declaresText(matched []routeWorker) bool {
 	for _, w := range matched {

pkg/agent/endpoint/front_test.go

@@ -473,16 +473,16 @@ func postText(t *testing.T, ts *httptest.Server, path, body string) {
 }
 
 // the preamble carries the kind the picked worker declared the route with, and
-// with no resolver installed a text route is forwarded unchanged: no session,
+// with no resolver installed a text route is forwarded unchanged: no thread,
 // whole body.
 func TestFrontTextRouteWithoutResolverPassesThrough(t *testing.T) {
 	ts, seen := textFront(t, FrontParams{})
-	body := strings.Repeat("x", 3*DefaultSessionPeek)
+	body := strings.Repeat("x", 3*DefaultThreadPeek)
 
 	postText(t, ts, "/a2a", body)
 	got := <-seen
 	require.Equal(t, livekit.AgentHttp_AEK_TEXT, got.pre.GetKind())
-	require.Empty(t, got.pre.GetSessionId())
+	require.Empty(t, got.pre.GetThreadId())
 	require.Equal(t, body, got.body)
 
 	postText(t, ts, "/plain", body)
@@ -494,25 +494,25 @@ func TestFrontTextRouteWithoutResolverPassesThrough(t *testing.T) {
 // a resolver sees the request head and a bounded prefix of the body; its id
 // reaches the text-kind worker in the preamble, and the worker still gets the
 // whole body.
-func TestFrontSessionResolverPlacesTextRoute(t *testing.T) {
+func TestFrontThreadResolverPlacesTextRoute(t *testing.T) {
 	type call struct {
 		method, path, channel, agent, deployment string
 		peek                                     string
 	}
 	calls := make(chan call, 1)
 	ts, seen := textFront(t, FrontParams{
-		SessionPeek: 8,
-		ResolveSession: func(r *http.Request, agentName, deployment string, body []byte) (string, bool) {
+		ThreadPeek: 8,
+		ResolveThread: func(r *http.Request, agentName, deployment string, body []byte) (string, bool) {
 			calls <- call{r.Method, r.URL.Path, r.Header.Get("X-Channel"), agentName, deployment, string(body)}
 			return "AT_1", true
 		},
 	})
-	body := `{"jsonrpc":"2.0","params":` + strings.Repeat("x", 2*DefaultSessionPeek) + `}`
+	body := `{"jsonrpc":"2.0","params":` + strings.Repeat("x", 2*DefaultThreadPeek) + `}`
 
 	postText(t, ts, "/a2a", body)
 	got := <-seen
 	require.Equal(t, livekit.AgentHttp_AEK_TEXT, got.pre.GetKind())
-	require.Equal(t, "AT_1", got.pre.GetSessionId())
+	require.Equal(t, "AT_1", got.pre.GetThreadId())
 	require.Equal(t, body, got.body, "the peek must not take bytes from the worker")
 
 	c := <-calls
@@ -522,29 +522,29 @@ func TestFrontSessionResolverPlacesTextRoute(t *testing.T) {
 	postText(t, ts, "/plain", body)
 	got = <-seen
 	require.Equal(t, livekit.AgentHttp_AEK_HTTP, got.pre.GetKind())
-	require.Empty(t, got.pre.GetSessionId())
+	require.Empty(t, got.pre.GetThreadId())
 	require.Equal(t, body, got.body)
 	require.Empty(t, calls)
 }
 
-// a resolver that places nothing leaves the session to the worker.
-func TestFrontSessionResolverDeclines(t *testing.T) {
+// a resolver that places nothing leaves the thread to the worker.
+func TestFrontThreadResolverDeclines(t *testing.T) {
 	ts, seen := textFront(t, FrontParams{
-		ResolveSession: func(*http.Request, string, string, []byte) (string, bool) { return "", false },
+		ResolveThread: func(*http.Request, string, string, []byte) (string, bool) { return "", false },
 	})
 	postText(t, ts, "/a2a", "hello")
 	got := <-seen
 	require.Equal(t, livekit.AgentHttp_AEK_TEXT, got.pre.GetKind())
-	require.Empty(t, got.pre.GetSessionId())
+	require.Empty(t, got.pre.GetThreadId())
 	require.Equal(t, "hello", got.body)
 }
 
 // a body shorter than the peek bound is handed over whole, with nothing lost to
 // the read-ahead; an absent body peeks nothing.
-func TestFrontSessionPeekShortBody(t *testing.T) {
+func TestFrontThreadPeekShortBody(t *testing.T) {
 	peeks := make(chan string, 1)
 	ts, seen := textFront(t, FrontParams{
-		ResolveSession: func(_ *http.Request, _, _ string, body []byte) (string, bool) {
+		ResolveThread: func(_ *http.Request, _, _ string, body []byte) (string, bool) {
 			peeks <- string(body)
 			return "AT_2", true
 		},
@@ -553,13 +553,13 @@ func TestFrontSessionPeekShortBody(t *testing.T) {
 	postText(t, ts, "/a2a", "short")
 	require.Equal(t, "short", <-peeks)
 	got := <-seen
-	require.Equal(t, "AT_2", got.pre.GetSessionId())
+	require.Equal(t, "AT_2", got.pre.GetThreadId())
 	require.Equal(t, "short", got.body)
 
 	postText(t, ts, "/a2a", "")
 	require.Empty(t, <-peeks)
 	got = <-seen
-	require.Equal(t, "AT_2", got.pre.GetSessionId())
+	require.Equal(t, "AT_2", got.pre.GetThreadId())
 	require.Empty(t, got.body)
 }
 
@@ -568,7 +568,7 @@ const testAPIKey, testAPISecret = "APIfront", "front-secret-front-secret-front-s
 // verifiedAccess resolves access the way the server's middleware and resolver
 // do together: a LiveKit token from the bearer, or from access_token on the
 // query when there is no bearer, is verified and its claims decide the level
-// and the session grant. No token is anonymous.
+// and the thread grant. No token is anonymous.
 func verifiedAccess(agentName, deployment string) AccessResolver {
 	return func(r *http.Request, _, _ string) (Access, bool) {
 		raw := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")
@@ -590,7 +590,7 @@ func verifiedAccess(agentName, deployment string) AccessResolver {
 		if claims.AgentEndpoint.Allows(agentName, deployment) {
 			level = AccessGranted
 		}
-		return Access{Level: level, SessionGrant: claims.AgentSession}, true
+		return Access{Level: level, ThreadGrant: claims.AgentThread}, true
 	}
 }
 
@@ -603,23 +603,23 @@ func withScope(resolve AccessResolver, scope *Scope) AccessResolver {
 	}
 }
 
-func mintToken(t *testing.T, grant *auth.AgentSessionGrant) string {
+func mintToken(t *testing.T, grant *auth.AgentThreadGrant) string {
 	t.Helper()
 	tok, err := auth.NewAccessToken(testAPIKey, testAPISecret).
 		SetIdentity("caller").
 		SetAgentEndpointGrant(&auth.AgentEndpointGrant{Call: true}).
-		SetAgentSessionGrant(grant).
+		SetAgentThreadGrant(grant).
 		ToJWT()
 	require.NoError(t, err)
 	return tok
 }
 
-// the verified token's agentSession claim reaches the worker in the preamble, on
+// the verified token's agentThread claim reaches the worker in the preamble, on
 // a text route and a plain one alike, while the token itself does not.
-func TestFrontSessionGrantTravelsInPreamble(t *testing.T) {
+func TestFrontThreadGrantTravelsInPreamble(t *testing.T) {
 	ts, seen := textFront(t, FrontParams{ResolveAccess: verifiedAccess("a", "d")})
-	grant := &auth.AgentSessionGrant{
-		AgentName: "a", Sub: "demo-user", SessionIDs: []string{"AT_1", "AT_2"},
+	grant := &auth.AgentThreadGrant{
+		AgentName: "a", Sub: "demo-user", ThreadIDs: []string{"AT_1", "AT_2"},
 		List: true, Create: true, Send: true, Read: true,
 	}
 	tok := mintToken(t, grant)
@@ -636,23 +636,23 @@ func TestFrontSessionGrantTravelsInPreamble(t *testing.T) {
 		got := <-seen
 		require.True(t, got.pre.GetAuthorized())
 		require.Empty(t, got.header.Get("Authorization"), "the token must never reach the worker")
-		require.Equal(t, "a", got.pre.GetSessionGrant().GetAgentName())
-		require.Equal(t, "demo-user", got.pre.GetSessionGrant().GetSub())
-		require.Equal(t, []string{"AT_1", "AT_2"}, got.pre.GetSessionGrant().GetSessionIds())
-		require.True(t, got.pre.GetSessionGrant().GetList())
-		require.True(t, got.pre.GetSessionGrant().GetCreate())
-		require.True(t, got.pre.GetSessionGrant().GetSend())
-		require.True(t, got.pre.GetSessionGrant().GetRead())
-		require.False(t, got.pre.GetSessionGrant().GetDelete())
+		require.Equal(t, "a", got.pre.GetThreadGrant().GetAgentName())
+		require.Equal(t, "demo-user", got.pre.GetThreadGrant().GetSub())
+		require.Equal(t, []string{"AT_1", "AT_2"}, got.pre.GetThreadGrant().GetThreadIds())
+		require.True(t, got.pre.GetThreadGrant().GetList())
+		require.True(t, got.pre.GetThreadGrant().GetCreate())
+		require.True(t, got.pre.GetThreadGrant().GetSend())
+		require.True(t, got.pre.GetThreadGrant().GetRead())
+		require.False(t, got.pre.GetThreadGrant().GetDelete())
 	}
 }
 
 // a browser WebSocket cannot set headers, so the token may ride the query as
 // access_token: it is verified the same way, the grant travels in the preamble,
 // and the query the worker sees no longer carries it.
-func TestFrontSessionGrantFromQueryToken(t *testing.T) {
+func TestFrontThreadGrantFromQueryToken(t *testing.T) {
 	ts, seen := textFront(t, FrontParams{ResolveAccess: verifiedAccess("a", "d")})
-	tok := mintToken(t, &auth.AgentSessionGrant{AgentName: "a", Sub: "demo-user", Send: true})
+	tok := mintToken(t, &auth.AgentThreadGrant{AgentName: "a", Sub: "demo-user", Send: true})
 
 	req, err := http.NewRequest(http.MethodPost, ts.URL+PathPrefix+"a/d/a2a?access_token="+tok+"&keep=1", strings.NewReader("hello"))
 	require.NoError(t, err)
@@ -662,23 +662,23 @@ func TestFrontSessionGrantFromQueryToken(t *testing.T) {
 	require.Equal(t, http.StatusNoContent, resp.StatusCode)
 
 	got := <-seen
-	require.Equal(t, "demo-user", got.pre.GetSessionGrant().GetSub())
-	require.Equal(t, "a", got.pre.GetSessionGrant().GetAgentName())
-	require.True(t, got.pre.GetSessionGrant().GetSend())
+	require.Equal(t, "demo-user", got.pre.GetThreadGrant().GetSub())
+	require.Equal(t, "a", got.pre.GetThreadGrant().GetAgentName())
+	require.True(t, got.pre.GetThreadGrant().GetSend())
 	require.Empty(t, got.header.Get("Authorization"))
 	require.Equal(t, "/a2a?keep=1", got.target, "the query token is stripped like the bearer")
 }
 
-// the grant is the node's to assert: a client-set X-LiveKit-Session-Grant is
-// dropped, and a token with no agentSession claim leaves the field unset.
-func TestFrontDropsForgedSessionGrantHeader(t *testing.T) {
+// the grant is the node's to assert: a client-set X-LiveKit-Thread-Grant is
+// dropped, and a token with no agentThread claim leaves the field unset.
+func TestFrontDropsForgedThreadGrantHeader(t *testing.T) {
 	ts, seen := textFront(t, FrontParams{ResolveAccess: verifiedAccess("a", "d")})
 	tok := mintToken(t, nil)
 
 	req, err := http.NewRequest(http.MethodPost, ts.URL+PathPrefix+"a/d/a2a", strings.NewReader("hello"))
 	require.NoError(t, err)
 	req.Header.Set("Authorization", "Bearer "+tok)
-	req.Header.Set(sessionGrantHeader, `{"agentName":"a","sub":"someone-else","send":true}`)
+	req.Header.Set(threadGrantHeader, `{"agentName":"a","sub":"someone-else","send":true}`)
 	req.Header.Set("X-Keep", "1")
 	resp, err := http.DefaultClient.Do(req)
 	require.NoError(t, err)
@@ -687,8 +687,31 @@ func TestFrontDropsForgedSessionGrantHeader(t *testing.T) {
 
 	got := <-seen
 	require.True(t, got.pre.GetAuthorized())
-	require.Nil(t, got.pre.GetSessionGrant(), "a project token with no agentSession claim grants no session")
-	require.Empty(t, got.header.Values(sessionGrantHeader), "a client cannot forge the grant")
+	require.Nil(t, got.pre.GetThreadGrant(), "a project token with no agentThread claim grants no thread")
+	require.Empty(t, got.header.Values(threadGrantHeader), "a client cannot forge the grant")
 	require.Equal(t, "1", got.header.Get("X-Keep"), "other headers still pass")
 	require.Empty(t, got.header.Get("Authorization"))
 }
+
+// a grant that lists endpoints opens only those: the first path segment under
+// the deployment is the endpoint, and one outside the list is forbidden before
+// any worker sees the request. A grant with no endpoints opens every one.
+func TestFrontThreadGrantEndpointsScopeTheRoute(t *testing.T) {
+	ts, seen := textFront(t, FrontParams{ResolveAccess: verifiedAccess("a", "d")})
+	scoped := mintToken(t, &auth.AgentThreadGrant{AgentName: "a", Sub: "demo-user", Send: true, Endpoints: []string{"front-desk"}})
+	open := mintToken(t, &auth.AgentThreadGrant{AgentName: "a", Sub: "demo-user", Send: true})
+
+	post := func(tok string) int {
+		req, err := http.NewRequest(http.MethodPost, ts.URL+PathPrefix+"a/d/a2a", strings.NewReader("hello"))
+		require.NoError(t, err)
+		req.Header.Set("Authorization", "Bearer "+tok)
+		resp, err := http.DefaultClient.Do(req)
+		require.NoError(t, err)
+		resp.Body.Close()
+		return resp.StatusCode
+	}
+	require.Equal(t, http.StatusForbidden, post(scoped), "a2a is not among the grant's endpoints")
+	require.Equal(t, http.StatusNoContent, post(open))
+	got := <-seen
+	require.Empty(t, got.pre.GetThreadGrant().GetEndpoints())
+}

pkg/service/agentendpoint.go

@@ -17,6 +17,9 @@ package service
 import (
 	"net/http"
 
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
 	"github.com/livekit/livekit-server/pkg/agent/endpoint"
 )
 
@@ -38,9 +41,9 @@ func NewAgentEndpointService(h *AgentHandler, scopes *EndpointScopes) *AgentEndp
 						level = endpoint.AccessGranted
 					}
 					return endpoint.Access{
-						Scope:        scopes.Scope(GetAPIKey(r.Context()), agentName, deployment),
-						Level:        level,
-						SessionGrant: claims.AgentSession,
+						Scope:       scopes.Scope(GetAPIKey(r.Context()), agentName, deployment),
+						Level:       level,
+						ThreadGrant: claims.AgentThread,
 					}, true
 				}
 				// unauthenticated: one configured key makes the api key
@@ -61,6 +64,23 @@ func NewAgentEndpointService(h *AgentHandler, scopes *EndpointScopes) *AgentEndp
 			Logger:              h.logger,
 			MaxTunnelsPerWorker: int(h.endpointsConfig.MaxTunnels),
 			TunnelDrainTimeout:  h.endpointsConfig.TunnelDrainTimeout,
+			ResolveThread:       threadResolver(h),
 		}),
 	}
 }
+
+// threadResolver is the front's thread lookup when the server is configured
+// with an AgentThreads address; nil otherwise, and a text route is forwarded
+// unchanged.
+func threadResolver(h *AgentHandler) endpoint.ThreadResolver {
+	address := h.endpointsConfig.ThreadsAddress
+	if address == "" {
+		return nil
+	}
+	conn, err := grpc.NewClient(address, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		h.logger.Warnw("agent threads address is not dialable; threads are left to the workers", err, "address", address)
+		return nil
+	}
+	return endpoint.NewThreadResolver(conn, h.endpointsConfig.ThreadLookupTimeout, h.logger)
+}