From d3a7c1e3e67aa4edf5b73fd8ad6057804cde0f26 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Tue, 6 Oct 2026 01:46:21 -0700 Subject: [PATCH] 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. --- pkg/agent/endpoint/attempt.go | 13 +++++++++++-- pkg/agent/endpoint/front.go | 4 +--- pkg/agent/endpoint/manifest.go | 10 ++++++++-- pkg/agent/endpoint/manifest_test.go | 15 ++++++++++++--- pkg/agent/endpoint/routes.go | 4 +++- 5 files changed, 35 insertions(+), 11 deletions(-) diff --git a/pkg/agent/endpoint/attempt.go b/pkg/agent/endpoint/attempt.go index 6c3c64cef..9f799ce75 100644 --- a/pkg/agent/endpoint/attempt.go +++ b/pkg/agent/endpoint/attempt.go @@ -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 diff --git a/pkg/agent/endpoint/front.go b/pkg/agent/endpoint/front.go index ca0dc51a8..3f8d6c22c 100644 --- a/pkg/agent/endpoint/front.go +++ b/pkg/agent/endpoint/front.go @@ -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 diff --git a/pkg/agent/endpoint/manifest.go b/pkg/agent/endpoint/manifest.go index 2946f986e..0549a073a 100644 --- a/pkg/agent/endpoint/manifest.go +++ b/pkg/agent/endpoint/manifest.go @@ -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 } diff --git a/pkg/agent/endpoint/manifest_test.go b/pkg/agent/endpoint/manifest_test.go index d81922437..3dabda5ef 100644 --- a/pkg/agent/endpoint/manifest_test.go +++ b/pkg/agent/endpoint/manifest_test.go @@ -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") } diff --git a/pkg/agent/endpoint/routes.go b/pkg/agent/endpoint/routes.go index 759af4315..9b4a11b73 100644 --- a/pkg/agent/endpoint/routes.go +++ b/pkg/agent/endpoint/routes.go @@ -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{}{} } }