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{}{} } }