agent endpoint: accept the text route kind and carry it in the preamble

ParseManifest admits AEK_TEXT next to AEK_HTTP and still refuses any other
kind. The kind is recorded on the parsed Endpoint and on each worker's
declaration, and the preamble carries the picked declaration's kind in
place of the AEK_HTTP constant, alongside its spelling of the route.
This commit is contained in:
Théo Monnom committed 2026-10-06 01:46:21 -07:00
1 parent b2d8d27a11
commit d3a7c1e3e6
5 files changed
+35 -11

No files matched your search

+11 -2
View File
@@ -86,7 +86,8 @@ func (a *attempt) retryable(err error) bool {
}
// newPreamble carries what the worker cannot derive from the HTTP bytes and
// must not infer from them - authenticated above all.
// must not infer from them - authenticated above all. What depends on the
// worker picked, the route and its kind, is set per attempt by setRoute.
func (a *attempt) newPreamble() *livekit.AgentHttp_StreamPreamble {
scheme := "http"
if a.req.TLS != nil {
@@ -95,7 +96,6 @@ func (a *attempt) newPreamble() *livekit.AgentHttp_StreamPreamble {
clientAddr, _, _ := net.SplitHostPort(a.req.RemoteAddr)
p := &livekit.AgentHttp_StreamPreamble{
Kind: livekit.AgentHttp_AEK_HTTP,
RequestId: a.requestID,
Authorized: a.granted,
ClientAddr: clientAddr,
@@ -104,6 +104,15 @@ func (a *attempt) newPreamble() *livekit.AgentHttp_StreamPreamble {
return p
}
// setRoute restates the declaration this attempt is placed against. Declarers
// of one route may spell it, and kind it, differently, so the worker is told
// its own. The preamble is re-serialized per attempt, so this reaches only the
// worker it was set for.
func (a *attempt) setRoute(w routeWorker) {
a.preamble.Route = w.raw
a.preamble.Kind = w.kind
}
// requestHeader builds the header set the worker sees.
func (a *attempt) requestHeader() http.Header {
r := a.req
+1 -3
View File
@@ -294,9 +294,7 @@ func (f *Front) ServeHTTP(w http.ResponseWriter, r *http.Request) {
a.before = bodyConsumed.Load()
a.refreshTimeout()
// the preamble is re-serialized per attempt, so this reaches only the
// worker it was set for
a.preamble.Route = picked.raw
a.setRoute(*picked)
switch f.bridge(w, a, picked.reg) {
case bridgeDone:
return
+8 -2
View File
@@ -42,6 +42,10 @@ type Endpoint struct {
Template *router.Template
Mask router.Mask
Public bool
// Kind is the route's declared kind. A text route is HTTP on the wire like
// any other; the kind tells the front it may read the body to place the
// request in an agent session.
Kind livekit.AgentHttp_AgentEndpointKind
}
// Ambiguous returns the declared templates whose shape forces the matcher to
@@ -129,7 +133,9 @@ func ParseManifest(endpoints []*livekit.AgentHttp_AgentEndpoint) (*Manifest, err
if err != nil {
return nil, err
}
if ep.GetKind() != livekit.AgentHttp_AEK_HTTP {
switch ep.GetKind() {
case livekit.AgentHttp_AEK_HTTP, livekit.AgentHttp_AEK_TEXT:
default:
return nil, fmt.Errorf("endpoint %q has unsupported kind %s", ep.GetPath(), ep.GetKind())
}
if len(ep.GetMethods()) == 0 {
@@ -150,7 +156,7 @@ func ParseManifest(endpoints []*livekit.AgentHttp_AgentEndpoint) (*Manifest, err
if tpl.Ambiguous() {
m.ambiguous = append(m.ambiguous, tpl.String())
}
m.Endpoints = append(m.Endpoints, Endpoint{Template: tpl, Mask: mask, Public: ep.GetPublic()})
m.Endpoints = append(m.Endpoints, Endpoint{Template: tpl, Mask: mask, Public: ep.GetPublic(), Kind: ep.GetKind()})
}
return m, nil
}
+12 -3
View File
@@ -113,8 +113,17 @@ func TestManifestValidation(t *testing.T) {
})
require.NoError(t, err, "all FastAPI verbs accepted")
_, err = ParseManifest([]*livekit.AgentHttp_AgentEndpoint{
{Path: "/ws", Kind: livekit.AgentHttp_AEK_TEXT, Methods: []string{"GET"}},
// a text route is HTTP on the wire; the kind is kept for the front
m, err := ParseManifest([]*livekit.AgentHttp_AgentEndpoint{
{Path: "/a2a", Kind: livekit.AgentHttp_AEK_TEXT, Methods: []string{"POST"}},
{Path: "/x", Methods: []string{"GET"}},
})
require.Error(t, err, "unsupported endpoint kind")
require.NoError(t, err, "text endpoint kind accepted")
require.Equal(t, livekit.AgentHttp_AEK_TEXT, m.Endpoints[0].Kind)
require.Equal(t, livekit.AgentHttp_AEK_HTTP, m.Endpoints[1].Kind)
_, err = ParseManifest([]*livekit.AgentHttp_AgentEndpoint{
{Path: "/x", Kind: livekit.AgentHttp_AgentEndpointKind(99), Methods: []string{"GET"}},
})
require.Error(t, err, "unknown endpoint kind")
}
+3 -1
View File
@@ -20,6 +20,7 @@ import (
"sync"
"sync/atomic"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
"github.com/livekit/livekit-server/pkg/agent/endpoint/router"
@@ -54,6 +55,7 @@ type routeWorker struct {
reg *Registration
raw string // the template as this worker declared it
public bool
kind livekit.AgentHttp_AgentEndpointKind
idx uint32 // its position in this worker's manifest
}
@@ -175,7 +177,7 @@ func (t *routeTable) declareLocked(reg *Registration, touched map[*Route]struct{
r = &Route{Template: ep.Template, Mask: key.method, key: key, ord: noOrder}
t.routes[key] = r
}
r.declare(routeWorker{reg: reg, raw: ep.Template.String(), public: ep.Public, idx: uint32(i)})
r.declare(routeWorker{reg: reg, raw: ep.Template.String(), public: ep.Public, kind: ep.Kind, idx: uint32(i)})
touched[r] = struct{}{}
}
}