From 5a0a934f29cd316232d282373b0e0ba9eda94343 Mon Sep 17 00:00:00 2001 From: Long Chen Date: Wed, 19 Aug 2026 16:34:48 +0800 Subject: [PATCH 1/4] Add the agent HTTP proxy protocol A worker does not accept inbound connections, so it dials a pool of websockets to the edge and the edge multiplexes requests over them. A stream carries opaque HTTP bytes both ways, so request streaming, SSE and websocket upgrades work without either side parsing HTTP. The worker names itself and announces its endpoints in the first frame of every wire. Registration reaches whichever node the load balancer picked, which is not the node the wires land on, so it cannot carry this. --- magefile.go | 1 + protobufs/livekit_agent_proxy.proto | 80 +++++++++++++++++++++++++++++ 2 files changed, 81 insertions(+) create mode 100644 protobufs/livekit_agent_proxy.proto diff --git a/magefile.go b/magefile.go index 9a5783d47..a5a8d82f8 100644 --- a/magefile.go +++ b/magefile.go @@ -67,6 +67,7 @@ func Proto() error { protoFiles := []string{ "livekit_agent.proto", + "livekit_agent_proxy.proto", "livekit_analytics.proto", "livekit_internal.proto", "livekit_models.proto", diff --git a/protobufs/livekit_agent_proxy.proto b/protobufs/livekit_agent_proxy.proto new file mode 100644 index 000000000..f2b8fd497 --- /dev/null +++ b/protobufs/livekit_agent_proxy.proto @@ -0,0 +1,80 @@ +// Copyright 2023 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +syntax = "proto3"; + +package livekit; +option go_package = "github.com/livekit/protocol/livekit"; +option csharp_namespace = "LiveKit.Proto"; +option ruby_package = "LiveKit::Proto"; + +import "logger/options.proto"; + +// Reaching a worker's HTTP endpoints from LiveKit Cloud. +// +// A worker cannot accept inbound connections, so it dials a pool of websockets to the +// edge, which multiplexes requests over them. A stream carries opaque HTTP bytes both +// ways, so streaming, SSE and websocket upgrades work without either side parsing HTTP. + +// One multiplexed frame, carried as a single websocket binary message. +message AgentHttpFrame { + // The stream this frame belongs to; the edge opens them all and takes odd ids. + // + // 0 is the connection itself rather than a stream on it, which only credit uses: every + // stream draws from a window the whole wire shares, and that window is refilled here. + uint32 stream_id = 1; + + oneof message { + AgentHttpRegistration registration = 2; + AgentHttpOpen open = 3; + // bare bytes, so a frame costs one copy and not a nested parse + bytes data = 4; + // half-close: nothing more this way, the other direction keeps going + AgentHttpEof eof = 5; + // the stream is over and whatever was in flight is lost + AgentHttpReset reset = 6; + // the sender may put this many more bytes in flight + uint32 credit = 7; + }; +} + +// Who the worker is and what it serves. The first frame of every wire. +// +// Registration lands on whichever node the load balancer picked, which is not the node +// the wires land on, so it cannot carry this. +message AgentHttpRegistration { + // Identifies the worker process, so it must be stable for its life and unique between + // workers. Not the id RegisterWorkerResponse gives: a wire is dialled before then. + string worker_id = 1 [(logger.name) = "workerID"]; + + // The first path segment of a request: "get_order" takes /get_order and all below it. + // The edge compares that one segment and forwards the path untouched, so the worker's + // own router answers 404 and 405 itself. + repeated string endpoints = 2; + + // Requests the edge may have in flight against this worker at once. 0 means no limit. + uint32 max_concurrent_requests = 3; +} + +// What the edge knows about a request that the bytes themselves do not carry. +message AgentHttpOpen { + string request_id = 1 [(logger.name) = "requestID"]; + // client address as the edge saw it, before any X-Forwarded-For it injects + string peer_addr = 2; + map attributes = 3; +} + +message AgentHttpEof {} + +message AgentHttpReset {} From c6b7e64da64c659dca4edc5c120189e58e215a81 Mon Sep 17 00:00:00 2001 From: github-actions <41898282+github-actions[bot]@users.noreply.github.com> Date: Wed, 19 Aug 2026 08:36:33 +0000 Subject: [PATCH 2/4] generated protobuf --- livekit/livekit_agent_proxy.pb.go | 501 ++++++++++++++++++++++++++++++ 1 file changed, 501 insertions(+) create mode 100644 livekit/livekit_agent_proxy.pb.go diff --git a/livekit/livekit_agent_proxy.pb.go b/livekit/livekit_agent_proxy.pb.go new file mode 100644 index 000000000..a8066d407 --- /dev/null +++ b/livekit/livekit_agent_proxy.pb.go @@ -0,0 +1,501 @@ +// Copyright 2023 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.12 +// protoc v7.35.1 +// source: livekit_agent_proxy.proto + +package livekit + +import ( + _ "github.com/livekit/protocol/livekit/logger" + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// One multiplexed frame, carried as a single websocket binary message. +type AgentHttpFrame struct { + state protoimpl.MessageState `protogen:"open.v1"` + // The stream this frame belongs to; the edge opens them all and takes odd ids. + // + // 0 is the connection itself rather than a stream on it, which only credit uses: every + // stream draws from a window the whole wire shares, and that window is refilled here. + StreamId uint32 `protobuf:"varint,1,opt,name=stream_id,json=streamId,proto3" json:"stream_id,omitempty"` + // Types that are valid to be assigned to Message: + // + // *AgentHttpFrame_Registration + // *AgentHttpFrame_Open + // *AgentHttpFrame_Data + // *AgentHttpFrame_Eof + // *AgentHttpFrame_Reset_ + // *AgentHttpFrame_Credit + Message isAgentHttpFrame_Message `protobuf_oneof:"message"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AgentHttpFrame) Reset() { + *x = AgentHttpFrame{} + mi := &file_livekit_agent_proxy_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AgentHttpFrame) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AgentHttpFrame) ProtoMessage() {} + +func (x *AgentHttpFrame) ProtoReflect() protoreflect.Message { + mi := &file_livekit_agent_proxy_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use AgentHttpFrame.ProtoReflect.Descriptor instead. +func (*AgentHttpFrame) Descriptor() ([]byte, []int) { + return file_livekit_agent_proxy_proto_rawDescGZIP(), []int{0} +} + +func (x *AgentHttpFrame) GetStreamId() uint32 { + if x != nil { + return x.StreamId + } + return 0 +} + +func (x *AgentHttpFrame) GetMessage() isAgentHttpFrame_Message { + if x != nil { + return x.Message + } + return nil +} + +func (x *AgentHttpFrame) GetRegistration() *AgentHttpRegistration { + if x != nil { + if x, ok := x.Message.(*AgentHttpFrame_Registration); ok { + return x.Registration + } + } + return nil +} + +func (x *AgentHttpFrame) GetOpen() *AgentHttpOpen { + if x != nil { + if x, ok := x.Message.(*AgentHttpFrame_Open); ok { + return x.Open + } + } + return nil +} + +func (x *AgentHttpFrame) GetData() []byte { + if x != nil { + if x, ok := x.Message.(*AgentHttpFrame_Data); ok { + return x.Data + } + } + return nil +} + +func (x *AgentHttpFrame) GetEof() *AgentHttpEof { + if x != nil { + if x, ok := x.Message.(*AgentHttpFrame_Eof); ok { + return x.Eof + } + } + return nil +} + +func (x *AgentHttpFrame) GetReset_() *AgentHttpReset { + if x != nil { + if x, ok := x.Message.(*AgentHttpFrame_Reset_); ok { + return x.Reset_ + } + } + return nil +} + +func (x *AgentHttpFrame) GetCredit() uint32 { + if x != nil { + if x, ok := x.Message.(*AgentHttpFrame_Credit); ok { + return x.Credit + } + } + return 0 +} + +type isAgentHttpFrame_Message interface { + isAgentHttpFrame_Message() +} + +type AgentHttpFrame_Registration struct { + Registration *AgentHttpRegistration `protobuf:"bytes,2,opt,name=registration,proto3,oneof"` +} + +type AgentHttpFrame_Open struct { + Open *AgentHttpOpen `protobuf:"bytes,3,opt,name=open,proto3,oneof"` +} + +type AgentHttpFrame_Data struct { + // bare bytes, so a frame costs one copy and not a nested parse + Data []byte `protobuf:"bytes,4,opt,name=data,proto3,oneof"` +} + +type AgentHttpFrame_Eof struct { + // half-close: nothing more this way, the other direction keeps going + Eof *AgentHttpEof `protobuf:"bytes,5,opt,name=eof,proto3,oneof"` +} + +type AgentHttpFrame_Reset_ struct { + // the stream is over and whatever was in flight is lost + Reset_ *AgentHttpReset `protobuf:"bytes,6,opt,name=reset,proto3,oneof"` +} + +type AgentHttpFrame_Credit struct { + // the sender may put this many more bytes in flight + Credit uint32 `protobuf:"varint,7,opt,name=credit,proto3,oneof"` +} + +func (*AgentHttpFrame_Registration) isAgentHttpFrame_Message() {} + +func (*AgentHttpFrame_Open) isAgentHttpFrame_Message() {} + +func (*AgentHttpFrame_Data) isAgentHttpFrame_Message() {} + +func (*AgentHttpFrame_Eof) isAgentHttpFrame_Message() {} + +func (*AgentHttpFrame_Reset_) isAgentHttpFrame_Message() {} + +func (*AgentHttpFrame_Credit) isAgentHttpFrame_Message() {} + +// Who the worker is and what it serves. The first frame of every wire. +// +// Registration lands on whichever node the load balancer picked, which is not the node +// the wires land on, so it cannot carry this. +type AgentHttpRegistration struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Identifies the worker process, so it must be stable for its life and unique between + // workers. Not the id RegisterWorkerResponse gives: a wire is dialled before then. + WorkerId string `protobuf:"bytes,1,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty"` + // The first path segment of a request: "get_order" takes /get_order and all below it. + // The edge compares that one segment and forwards the path untouched, so the worker's + // own router answers 404 and 405 itself. + Endpoints []string `protobuf:"bytes,2,rep,name=endpoints,proto3" json:"endpoints,omitempty"` + // Requests the edge may have in flight against this worker at once. 0 means no limit. + MaxConcurrentRequests uint32 `protobuf:"varint,3,opt,name=max_concurrent_requests,json=maxConcurrentRequests,proto3" json:"max_concurrent_requests,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AgentHttpRegistration) Reset() { + *x = AgentHttpRegistration{} + mi := &file_livekit_agent_proxy_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AgentHttpRegistration) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AgentHttpRegistration) ProtoMessage() {} + +func (x *AgentHttpRegistration) ProtoReflect() protoreflect.Message { + mi := &file_livekit_agent_proxy_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use AgentHttpRegistration.ProtoReflect.Descriptor instead. +func (*AgentHttpRegistration) Descriptor() ([]byte, []int) { + return file_livekit_agent_proxy_proto_rawDescGZIP(), []int{1} +} + +func (x *AgentHttpRegistration) GetWorkerId() string { + if x != nil { + return x.WorkerId + } + return "" +} + +func (x *AgentHttpRegistration) GetEndpoints() []string { + if x != nil { + return x.Endpoints + } + return nil +} + +func (x *AgentHttpRegistration) GetMaxConcurrentRequests() uint32 { + if x != nil { + return x.MaxConcurrentRequests + } + return 0 +} + +// What the edge knows about a request that the bytes themselves do not carry. +type AgentHttpOpen struct { + state protoimpl.MessageState `protogen:"open.v1"` + RequestId string `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` + // client address as the edge saw it, before any X-Forwarded-For it injects + PeerAddr string `protobuf:"bytes,2,opt,name=peer_addr,json=peerAddr,proto3" json:"peer_addr,omitempty"` + Attributes map[string]string `protobuf:"bytes,3,rep,name=attributes,proto3" json:"attributes,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AgentHttpOpen) Reset() { + *x = AgentHttpOpen{} + mi := &file_livekit_agent_proxy_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AgentHttpOpen) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AgentHttpOpen) ProtoMessage() {} + +func (x *AgentHttpOpen) ProtoReflect() protoreflect.Message { + mi := &file_livekit_agent_proxy_proto_msgTypes[2] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use AgentHttpOpen.ProtoReflect.Descriptor instead. +func (*AgentHttpOpen) Descriptor() ([]byte, []int) { + return file_livekit_agent_proxy_proto_rawDescGZIP(), []int{2} +} + +func (x *AgentHttpOpen) GetRequestId() string { + if x != nil { + return x.RequestId + } + return "" +} + +func (x *AgentHttpOpen) GetPeerAddr() string { + if x != nil { + return x.PeerAddr + } + return "" +} + +func (x *AgentHttpOpen) GetAttributes() map[string]string { + if x != nil { + return x.Attributes + } + return nil +} + +type AgentHttpEof struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AgentHttpEof) Reset() { + *x = AgentHttpEof{} + mi := &file_livekit_agent_proxy_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AgentHttpEof) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AgentHttpEof) ProtoMessage() {} + +func (x *AgentHttpEof) ProtoReflect() protoreflect.Message { + mi := &file_livekit_agent_proxy_proto_msgTypes[3] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use AgentHttpEof.ProtoReflect.Descriptor instead. +func (*AgentHttpEof) Descriptor() ([]byte, []int) { + return file_livekit_agent_proxy_proto_rawDescGZIP(), []int{3} +} + +type AgentHttpReset struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AgentHttpReset) Reset() { + *x = AgentHttpReset{} + mi := &file_livekit_agent_proxy_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AgentHttpReset) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AgentHttpReset) ProtoMessage() {} + +func (x *AgentHttpReset) ProtoReflect() protoreflect.Message { + mi := &file_livekit_agent_proxy_proto_msgTypes[4] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use AgentHttpReset.ProtoReflect.Descriptor instead. +func (*AgentHttpReset) Descriptor() ([]byte, []int) { + return file_livekit_agent_proxy_proto_rawDescGZIP(), []int{4} +} + +var File_livekit_agent_proxy_proto protoreflect.FileDescriptor + +const file_livekit_agent_proxy_proto_rawDesc = "" + + "\n" + + "\x19livekit_agent_proxy.proto\x12\alivekit\x1a\x14logger/options.proto\"\xb8\x02\n" + + "\x0eAgentHttpFrame\x12\x1b\n" + + "\tstream_id\x18\x01 \x01(\rR\bstreamId\x12D\n" + + "\fregistration\x18\x02 \x01(\v2\x1e.livekit.AgentHttpRegistrationH\x00R\fregistration\x12,\n" + + "\x04open\x18\x03 \x01(\v2\x16.livekit.AgentHttpOpenH\x00R\x04open\x12\x14\n" + + "\x04data\x18\x04 \x01(\fH\x00R\x04data\x12)\n" + + "\x03eof\x18\x05 \x01(\v2\x15.livekit.AgentHttpEofH\x00R\x03eof\x12/\n" + + "\x05reset\x18\x06 \x01(\v2\x17.livekit.AgentHttpResetH\x00R\x05reset\x12\x18\n" + + "\x06credit\x18\a \x01(\rH\x00R\x06creditB\t\n" + + "\amessage\"\x97\x01\n" + + "\x15AgentHttpRegistration\x12(\n" + + "\tworker_id\x18\x01 \x01(\tB\v\xbaP\bworkerIDR\bworkerId\x12\x1c\n" + + "\tendpoints\x18\x02 \x03(\tR\tendpoints\x126\n" + + "\x17max_concurrent_requests\x18\x03 \x01(\rR\x15maxConcurrentRequests\"\xe0\x01\n" + + "\rAgentHttpOpen\x12+\n" + + "\n" + + "request_id\x18\x01 \x01(\tB\f\xbaP\trequestIDR\trequestId\x12\x1b\n" + + "\tpeer_addr\x18\x02 \x01(\tR\bpeerAddr\x12F\n" + + "\n" + + "attributes\x18\x03 \x03(\v2&.livekit.AgentHttpOpen.AttributesEntryR\n" + + "attributes\x1a=\n" + + "\x0fAttributesEntry\x12\x10\n" + + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\x0e\n" + + "\fAgentHttpEof\"\x10\n" + + "\x0eAgentHttpResetBFZ#github.com/livekit/protocol/livekit\xaa\x02\rLiveKit.Proto\xea\x02\x0eLiveKit::Protob\x06proto3" + +var ( + file_livekit_agent_proxy_proto_rawDescOnce sync.Once + file_livekit_agent_proxy_proto_rawDescData []byte +) + +func file_livekit_agent_proxy_proto_rawDescGZIP() []byte { + file_livekit_agent_proxy_proto_rawDescOnce.Do(func() { + file_livekit_agent_proxy_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_livekit_agent_proxy_proto_rawDesc), len(file_livekit_agent_proxy_proto_rawDesc))) + }) + return file_livekit_agent_proxy_proto_rawDescData +} + +var file_livekit_agent_proxy_proto_msgTypes = make([]protoimpl.MessageInfo, 6) +var file_livekit_agent_proxy_proto_goTypes = []any{ + (*AgentHttpFrame)(nil), // 0: livekit.AgentHttpFrame + (*AgentHttpRegistration)(nil), // 1: livekit.AgentHttpRegistration + (*AgentHttpOpen)(nil), // 2: livekit.AgentHttpOpen + (*AgentHttpEof)(nil), // 3: livekit.AgentHttpEof + (*AgentHttpReset)(nil), // 4: livekit.AgentHttpReset + nil, // 5: livekit.AgentHttpOpen.AttributesEntry +} +var file_livekit_agent_proxy_proto_depIdxs = []int32{ + 1, // 0: livekit.AgentHttpFrame.registration:type_name -> livekit.AgentHttpRegistration + 2, // 1: livekit.AgentHttpFrame.open:type_name -> livekit.AgentHttpOpen + 3, // 2: livekit.AgentHttpFrame.eof:type_name -> livekit.AgentHttpEof + 4, // 3: livekit.AgentHttpFrame.reset:type_name -> livekit.AgentHttpReset + 5, // 4: livekit.AgentHttpOpen.attributes:type_name -> livekit.AgentHttpOpen.AttributesEntry + 5, // [5:5] is the sub-list for method output_type + 5, // [5:5] is the sub-list for method input_type + 5, // [5:5] is the sub-list for extension type_name + 5, // [5:5] is the sub-list for extension extendee + 0, // [0:5] is the sub-list for field type_name +} + +func init() { file_livekit_agent_proxy_proto_init() } +func file_livekit_agent_proxy_proto_init() { + if File_livekit_agent_proxy_proto != nil { + return + } + file_livekit_agent_proxy_proto_msgTypes[0].OneofWrappers = []any{ + (*AgentHttpFrame_Registration)(nil), + (*AgentHttpFrame_Open)(nil), + (*AgentHttpFrame_Data)(nil), + (*AgentHttpFrame_Eof)(nil), + (*AgentHttpFrame_Reset_)(nil), + (*AgentHttpFrame_Credit)(nil), + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_livekit_agent_proxy_proto_rawDesc), len(file_livekit_agent_proxy_proto_rawDesc)), + NumEnums: 0, + NumMessages: 6, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_livekit_agent_proxy_proto_goTypes, + DependencyIndexes: file_livekit_agent_proxy_proto_depIdxs, + MessageInfos: file_livekit_agent_proxy_proto_msgTypes, + }.Build() + File_livekit_agent_proxy_proto = out.File + file_livekit_agent_proxy_proto_goTypes = nil + file_livekit_agent_proxy_proto_depIdxs = nil +} From 28cfe408e5e986a6b19c7d5135607211cd44f46f Mon Sep 17 00:00:00 2001 From: Long Chen Date: Wed, 19 Aug 2026 17:24:28 +0800 Subject: [PATCH 3/4] Say controller, not edge The cloud side of this is the controller: the node a worker dials and that proxies requests to it. Nothing else in the tree calls it an edge. --- protobufs/livekit_agent_proxy.proto | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/protobufs/livekit_agent_proxy.proto b/protobufs/livekit_agent_proxy.proto index f2b8fd497..d8812ae4c 100644 --- a/protobufs/livekit_agent_proxy.proto +++ b/protobufs/livekit_agent_proxy.proto @@ -24,12 +24,12 @@ import "logger/options.proto"; // Reaching a worker's HTTP endpoints from LiveKit Cloud. // // A worker cannot accept inbound connections, so it dials a pool of websockets to the -// edge, which multiplexes requests over them. A stream carries opaque HTTP bytes both -// ways, so streaming, SSE and websocket upgrades work without either side parsing HTTP. +// cloud controller, which multiplexes requests over them. A stream carries opaque HTTP +// bytes both ways, so streaming, SSE and websocket upgrades work without HTTP parsing. // One multiplexed frame, carried as a single websocket binary message. message AgentHttpFrame { - // The stream this frame belongs to; the edge opens them all and takes odd ids. + // The stream this frame belongs to; the controller opens them all and takes odd ids. // // 0 is the connection itself rather than a stream on it, which only credit uses: every // stream draws from a window the whole wire shares, and that window is refilled here. @@ -59,18 +59,18 @@ message AgentHttpRegistration { string worker_id = 1 [(logger.name) = "workerID"]; // The first path segment of a request: "get_order" takes /get_order and all below it. - // The edge compares that one segment and forwards the path untouched, so the worker's - // own router answers 404 and 405 itself. + // The controller compares that one segment and forwards the path untouched, so the + // worker's own router answers 404 and 405 itself. repeated string endpoints = 2; - // Requests the edge may have in flight against this worker at once. 0 means no limit. + // Requests the controller may have in flight against this worker. 0 means no limit. uint32 max_concurrent_requests = 3; } -// What the edge knows about a request that the bytes themselves do not carry. +// What the controller knows about a request that the bytes themselves do not carry. message AgentHttpOpen { string request_id = 1 [(logger.name) = "requestID"]; - // client address as the edge saw it, before any X-Forwarded-For it injects + // client address as the controller saw it, before any X-Forwarded-For it injects string peer_addr = 2; map attributes = 3; } From 71ae3dd5dd66332a004073dccd6f4c5a424e482a Mon Sep 17 00:00:00 2001 From: github-actions <41898282+github-actions[bot]@users.noreply.github.com> Date: Wed, 19 Aug 2026 09:26:33 +0000 Subject: [PATCH 4/4] generated protobuf --- livekit/livekit_agent_proxy.pb.go | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/livekit/livekit_agent_proxy.pb.go b/livekit/livekit_agent_proxy.pb.go index a8066d407..dc4cb6686 100644 --- a/livekit/livekit_agent_proxy.pb.go +++ b/livekit/livekit_agent_proxy.pb.go @@ -39,7 +39,7 @@ const ( // One multiplexed frame, carried as a single websocket binary message. type AgentHttpFrame struct { state protoimpl.MessageState `protogen:"open.v1"` - // The stream this frame belongs to; the edge opens them all and takes odd ids. + // The stream this frame belongs to; the controller opens them all and takes odd ids. // // 0 is the connection itself rather than a stream on it, which only credit uses: every // stream draws from a window the whole wire shares, and that window is refilled here. @@ -209,10 +209,10 @@ type AgentHttpRegistration struct { // workers. Not the id RegisterWorkerResponse gives: a wire is dialled before then. WorkerId string `protobuf:"bytes,1,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty"` // The first path segment of a request: "get_order" takes /get_order and all below it. - // The edge compares that one segment and forwards the path untouched, so the worker's - // own router answers 404 and 405 itself. + // The controller compares that one segment and forwards the path untouched, so the + // worker's own router answers 404 and 405 itself. Endpoints []string `protobuf:"bytes,2,rep,name=endpoints,proto3" json:"endpoints,omitempty"` - // Requests the edge may have in flight against this worker at once. 0 means no limit. + // Requests the controller may have in flight against this worker. 0 means no limit. MaxConcurrentRequests uint32 `protobuf:"varint,3,opt,name=max_concurrent_requests,json=maxConcurrentRequests,proto3" json:"max_concurrent_requests,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -269,11 +269,11 @@ func (x *AgentHttpRegistration) GetMaxConcurrentRequests() uint32 { return 0 } -// What the edge knows about a request that the bytes themselves do not carry. +// What the controller knows about a request that the bytes themselves do not carry. type AgentHttpOpen struct { state protoimpl.MessageState `protogen:"open.v1"` RequestId string `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` - // client address as the edge saw it, before any X-Forwarded-For it injects + // client address as the controller saw it, before any X-Forwarded-For it injects PeerAddr string `protobuf:"bytes,2,opt,name=peer_addr,json=peerAddr,proto3" json:"peer_addr,omitempty"` Attributes map[string]string `protobuf:"bytes,3,rep,name=attributes,proto3" json:"attributes,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` unknownFields protoimpl.UnknownFields