diff --git a/.changeset/agent-endpoint-grant.md b/.changeset/agent-endpoint-grant.md new file mode 100644 index 000000000..95f277e59 --- /dev/null +++ b/.changeset/agent-endpoint-grant.md @@ -0,0 +1,6 @@ +--- +"github.com/livekit/protocol": patch +"@livekit/protocol": patch +--- + +Add `AgentEndpointGrant`, scoping calls to an agent's non-public HTTP endpoints diff --git a/agent/environment_test.go b/agent/environment_test.go new file mode 100644 index 000000000..5fcd9e130 --- /dev/null +++ b/agent/environment_test.go @@ -0,0 +1,43 @@ +package agent + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestValidateDeployment(t *testing.T) { + cases := []struct { + name string + deployment string + valid bool + }{ + {"empty", "", true}, + {"alphanumeric", "production", true}, + {"hyphen and dot", "prod-us.v2", true}, + {"colon", "prod:us", true}, + {"slash", "a/b", true}, + {"non-ascii", "prodüction", true}, + {"max length", strings.Repeat("a", MaxDeploymentLength), true}, + + {"underscore reserved", "prod_us", false}, + {"space", "prod us", false}, + {"tab", "prod\tus", false}, + {"newline", "prod\nus", false}, + {"del", "prod\x7fus", false}, + {"nul", "prod\x00us", false}, + {"too long", strings.Repeat("a", MaxDeploymentLength+1), false}, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + err := ValidateDeployment(c.deployment) + if c.valid { + require.NoError(t, err) + } else { + require.Error(t, err) + } + }) + } +} diff --git a/auth/accesstoken.go b/auth/accesstoken.go index d483bf821..5cc7de68d 100644 --- a/auth/accesstoken.go +++ b/auth/accesstoken.go @@ -109,6 +109,11 @@ func (t *AccessToken) SetObservabilityGrant(grant *ObservabilityGrant) *AccessTo return t } +func (t *AccessToken) SetAgentEndpointGrant(grant *AgentEndpointGrant) *AccessToken { + t.grant.AgentEndpoint = grant + return t +} + func (t *AccessToken) SetMetadata(md string) *AccessToken { t.grant.Metadata = md return t diff --git a/auth/accesstoken_test.go b/auth/accesstoken_test.go index 2cad1993d..4a2e7ec45 100644 --- a/auth/accesstoken_test.go +++ b/auth/accesstoken_test.go @@ -223,3 +223,33 @@ func TestAccessToken(t *testing.T) { func apiKeypair() (string, string) { return guid.New(utils.APIKeyPrefix), utils.RandomSecret() } + +func TestAgentEndpointGrantRoundTrip(t *testing.T) { + t.Parallel() + + apiKey, secret := apiKeypair() + grant := &AgentEndpointGrant{Call: true, AgentName: "my-agent", Deployment: "prod"} + raw, err := NewAccessToken(apiKey, secret). + SetAgentEndpointGrant(grant). + SetValidFor(time.Minute). + ToJWT() + require.NoError(t, err) + + // the claim key is camelCase + require.Contains(t, decodeClaims(t, raw), `"agentEndpoint"`) + + v, err := ParseAPIToken(raw) + require.NoError(t, err) + _, decoded, err := v.Verify(secret) + require.NoError(t, err) + require.Equal(t, grant, decoded.AgentEndpoint) +} + +func decodeClaims(t *testing.T, raw string) string { + t.Helper() + parts := strings.Split(raw, ".") + require.Len(t, parts, 3) + body, err := base64.RawURLEncoding.DecodeString(parts[1]) + require.NoError(t, err) + return string(body) +} diff --git a/auth/grants.go b/auth/grants.go index 11c27cdd7..4f56ad57a 100644 --- a/auth/grants.go +++ b/auth/grants.go @@ -168,6 +168,7 @@ type ClaimGrants struct { Agent *AgentGrant `json:"agent,omitempty"` Inference *InferenceGrant `json:"inference,omitempty"` Observability *ObservabilityGrant `json:"observability,omitempty"` + AgentEndpoint *AgentEndpointGrant `json:"agentEndpoint,omitempty"` // Room configuration to use if this participant initiates the room RoomConfig *RoomConfiguration `json:"roomConfig,omitempty"` // Cloud-only, config preset to use @@ -214,6 +215,7 @@ func (c *ClaimGrants) Clone() *ClaimGrants { clone.Agent = c.Agent.Clone() clone.Inference = c.Inference.Clone() clone.Observability = c.Observability.Clone() + clone.AgentEndpoint = c.AgentEndpoint.Clone() clone.Attributes = maps.Clone(c.Attributes) clone.RoomConfig = c.RoomConfig.Clone() if len(c.KindDetails) > 0 { @@ -236,6 +238,7 @@ func (c *ClaimGrants) MarshalLogObject(e zapcore.ObjectEncoder) error { e.AddObject("Agent", c.Agent) e.AddObject("Inference", c.Inference) e.AddObject("Observability", c.Observability) + e.AddObject("AgentEndpoint", c.AgentEndpoint) e.AddObject("RoomConfig", logger.Proto((*livekit.RoomConfiguration)(c.RoomConfig))) e.AddString("RoomPreset", c.RoomPreset) return nil @@ -445,14 +448,14 @@ func (v *VideoGrant) UpdateFromPermission(permission *livekit.ParticipantPermiss func (v *VideoGrant) ToPermission() *livekit.ParticipantPermission { return &livekit.ParticipantPermission{ - CanPublish: v.GetCanPublish(), - CanPublishData: v.GetCanPublishData(), - CanSubscribe: v.GetCanSubscribe(), - CanPublishSources: v.GetCanPublishSources(), - CanUpdateMetadata: v.GetCanUpdateOwnMetadata(), - Hidden: v.Hidden, - Recorder: v.Recorder, - Agent: v.Agent, + CanPublish: v.GetCanPublish(), + CanPublishData: v.GetCanPublishData(), + CanSubscribe: v.GetCanSubscribe(), + CanPublishSources: v.GetCanPublishSources(), + CanUpdateMetadata: v.GetCanUpdateOwnMetadata(), + Hidden: v.Hidden, + Recorder: v.Recorder, + Agent: v.Agent, CanSubscribeMetrics: v.GetCanSubscribeMetrics(), CanManageAgentSession: v.GetCanManageAgentSession(), } @@ -663,6 +666,52 @@ func (s *ObservabilityGrant) MarshalLogObject(e zapcore.ObjectEncoder) error { // ------------------------------------------------------------------ +type AgentEndpointGrant struct { + // Call grants to invoke an agent's non-public HTTP endpoints. + Call bool `json:"call,omitempty"` + // AgentName restricts the grant to one agent; empty grants every agent in the project. + AgentName string `json:"agentName,omitempty"` + // Deployment restricts the grant to one deployment; empty grants every + // deployment. A worker registered without one is addressed as "default". + Deployment string `json:"deployment,omitempty"` +} + +// Allows reports whether the grant authorizes calling non-public endpoints of +// (agentName, deployment). Matching is exact and case-sensitive; an empty scope +// field matches any value. +func (s *AgentEndpointGrant) Allows(agentName, deployment string) bool { + if s == nil || !s.Call { + return false + } + if s.AgentName != "" && s.AgentName != agentName { + return false + } + return s.Deployment == "" || s.Deployment == deployment +} + +func (s *AgentEndpointGrant) Clone() *AgentEndpointGrant { + if s == nil { + return nil + } + + clone := *s + + return &clone +} + +func (s *AgentEndpointGrant) MarshalLogObject(e zapcore.ObjectEncoder) error { + if s == nil { + return nil + } + + e.AddBool("Call", s.Call) + e.AddString("AgentName", s.AgentName) + e.AddString("Deployment", s.Deployment) + return nil +} + +// ------------------------------------------------------------------ + func sourceToString(source livekit.TrackSource) string { return strings.ToLower(source.String()) } diff --git a/auth/grants_test.go b/auth/grants_test.go index 042013088..8755a3c2e 100644 --- a/auth/grants_test.go +++ b/auth/grants_test.go @@ -35,6 +35,7 @@ func TestGrants(t *testing.T) { require.Same(t, grants.Agent, clone.Agent) require.Same(t, grants.Inference, clone.Inference) require.Same(t, grants.SIP, clone.SIP) + require.Same(t, grants.AgentEndpoint, clone.AgentEndpoint) require.True(t, reflect.DeepEqual(grants, clone)) require.True(t, reflect.DeepEqual(grants.Video, clone.Video)) }) @@ -61,6 +62,9 @@ func TestGrants(t *testing.T) { // require Inference require.Same(t, grants.Inference, clone.Inference) require.True(t, reflect.DeepEqual(grants.Inference, clone.Inference)) + // require AgentEndpoint + require.Same(t, grants.AgentEndpoint, clone.AgentEndpoint) + require.True(t, reflect.DeepEqual(grants.AgentEndpoint, clone.AgentEndpoint)) }) t.Run("clone with video", func(t *testing.T) { @@ -475,3 +479,48 @@ func TestRoomConfiguration_CheckCredentials(t *testing.T) { require.NoError(t, config.CheckCredentials()) }) } + +func TestAgentEndpointGrantAllows(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + grant *AgentEndpointGrant + agentName string + deployment string + want bool + }{ + {"nil grant", nil, "a", "prod", false}, + {"call unset with scope", &AgentEndpointGrant{AgentName: "a", Deployment: "prod"}, "a", "prod", false}, + {"wildcard", &AgentEndpointGrant{Call: true}, "a", "prod", true}, + {"agent match", &AgentEndpointGrant{Call: true, AgentName: "a"}, "a", "prod", true}, + {"agent mismatch", &AgentEndpointGrant{Call: true, AgentName: "a"}, "b", "prod", false}, + {"deployment match", &AgentEndpointGrant{Call: true, Deployment: "prod"}, "a", "prod", true}, + {"deployment mismatch", &AgentEndpointGrant{Call: true, Deployment: "prod"}, "a", "staging", false}, + {"both match", &AgentEndpointGrant{Call: true, AgentName: "a", Deployment: "prod"}, "a", "prod", true}, + {"both set, deployment differs", &AgentEndpointGrant{Call: true, AgentName: "a", Deployment: "prod"}, "a", "staging", false}, + {"agent case differs", &AgentEndpointGrant{Call: true, AgentName: "Agent"}, "agent", "prod", false}, + {"deployment case differs", &AgentEndpointGrant{Call: true, Deployment: "Prod"}, "a", "prod", false}, + {"empty deployment is wildcard", &AgentEndpointGrant{Call: true}, "a", "production", true}, + {"default pinned by literal", &AgentEndpointGrant{Call: true, Deployment: "default"}, "a", "default", true}, + {"default pin rejects others", &AgentEndpointGrant{Call: true, Deployment: "default"}, "a", "prod", false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + require.Equal(t, c.want, c.grant.Allows(c.agentName, c.deployment)) + }) + } +} + +func TestAgentEndpointGrantCloneIndependent(t *testing.T) { + t.Parallel() + + grants := &ClaimGrants{AgentEndpoint: &AgentEndpointGrant{Call: true, AgentName: "a", Deployment: "prod"}} + clone := grants.Clone() + require.NotSame(t, grants.AgentEndpoint, clone.AgentEndpoint) + + clone.AgentEndpoint.Call = false + clone.AgentEndpoint.AgentName = "b" + require.True(t, grants.AgentEndpoint.Call) + require.Equal(t, "a", grants.AgentEndpoint.AgentName) +} diff --git a/livekit/livekit_agent.pb.go b/livekit/livekit_agent.pb.go index a1052dee1..7ce1c942c 100644 --- a/livekit/livekit_agent.pb.go +++ b/livekit/livekit_agent.pb.go @@ -183,6 +183,124 @@ func (JobStatus) EnumDescriptor() ([]byte, []int) { return file_livekit_agent_proto_rawDescGZIP(), []int{2} } +type AgentHttp_AgentEndpointKind int32 + +const ( + AgentHttp_AEK_HTTP AgentHttp_AgentEndpointKind = 0 + // reserved for future text-mode endpoints + AgentHttp_AEK_TEXT AgentHttp_AgentEndpointKind = 1 +) + +// Enum value maps for AgentHttp_AgentEndpointKind. +var ( + AgentHttp_AgentEndpointKind_name = map[int32]string{ + 0: "AEK_HTTP", + 1: "AEK_TEXT", + } + AgentHttp_AgentEndpointKind_value = map[string]int32{ + "AEK_HTTP": 0, + "AEK_TEXT": 1, + } +) + +func (x AgentHttp_AgentEndpointKind) Enum() *AgentHttp_AgentEndpointKind { + p := new(AgentHttp_AgentEndpointKind) + *p = x + return p +} + +func (x AgentHttp_AgentEndpointKind) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (AgentHttp_AgentEndpointKind) Descriptor() protoreflect.EnumDescriptor { + return file_livekit_agent_proto_enumTypes[3].Descriptor() +} + +func (AgentHttp_AgentEndpointKind) Type() protoreflect.EnumType { + return &file_livekit_agent_proto_enumTypes[3] +} + +func (x AgentHttp_AgentEndpointKind) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use AgentHttp_AgentEndpointKind.Descriptor instead. +func (AgentHttp_AgentEndpointKind) EnumDescriptor() ([]byte, []int) { + return file_livekit_agent_proto_rawDescGZIP(), []int{16, 0} +} + +// why a stream was reset before any HTTP bytes flowed. This travels as the +// QUIC RESET_STREAM / STOP_SENDING error code, so it is a number with no room +// for detail: the reason string is logged by the side that reset and joined +// to this by request_id. An outcome after bytes have flowed travels in +// trailers instead, where it cannot race them. +type AgentHttp_HttpStreamResetCode int32 + +const ( + // no information. The ordinary cancel code, sent whenever a stream is torn + // down without a specific outcome, so it must stay the zero value and must + // not imply that anything was or was not applied. + AgentHttp_HSR_ABORT AgentHttp_HttpStreamResetCode = 0 + // the worker aborted before any application code observed the request, so + // nothing was applied and the exchange is safe to retry. Requires that the + // application was never entered; a 404 it returned is an ordinary response. + AgentHttp_HSR_REFUSED AgentHttp_HttpStreamResetCode = 1 + // the application was entered and then failed before producing a response + // head; side effects may already have happened, so this is not safe to + // retry. + AgentHttp_HSR_INTERNAL AgentHttp_HttpStreamResetCode = 2 + // the deadline elapsed before a response head was produced + AgentHttp_HSR_TIMEOUT AgentHttp_HttpStreamResetCode = 3 + // the peer's bytes were not valid HTTP/1.1 + AgentHttp_HSR_PROTOCOL AgentHttp_HttpStreamResetCode = 4 +) + +// Enum value maps for AgentHttp_HttpStreamResetCode. +var ( + AgentHttp_HttpStreamResetCode_name = map[int32]string{ + 0: "HSR_ABORT", + 1: "HSR_REFUSED", + 2: "HSR_INTERNAL", + 3: "HSR_TIMEOUT", + 4: "HSR_PROTOCOL", + } + AgentHttp_HttpStreamResetCode_value = map[string]int32{ + "HSR_ABORT": 0, + "HSR_REFUSED": 1, + "HSR_INTERNAL": 2, + "HSR_TIMEOUT": 3, + "HSR_PROTOCOL": 4, + } +) + +func (x AgentHttp_HttpStreamResetCode) Enum() *AgentHttp_HttpStreamResetCode { + p := new(AgentHttp_HttpStreamResetCode) + *p = x + return p +} + +func (x AgentHttp_HttpStreamResetCode) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (AgentHttp_HttpStreamResetCode) Descriptor() protoreflect.EnumDescriptor { + return file_livekit_agent_proto_enumTypes[4].Descriptor() +} + +func (AgentHttp_HttpStreamResetCode) Type() protoreflect.EnumType { + return &file_livekit_agent_proto_enumTypes[4] +} + +func (x AgentHttp_HttpStreamResetCode) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use AgentHttp_HttpStreamResetCode.Descriptor instead. +func (AgentHttp_HttpStreamResetCode) EnumDescriptor() ([]byte, []int) { + return file_livekit_agent_proto_rawDescGZIP(), []int{16, 1} +} + type Job struct { state protoimpl.MessageState `protogen:"open.v1"` Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` @@ -602,6 +720,7 @@ type ServerMessage struct { // *ServerMessage_Assignment // *ServerMessage_Termination // *ServerMessage_Pong + // *ServerMessage_GoAway Message isServerMessage_Message `protobuf_oneof:"message"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -689,6 +808,15 @@ func (x *ServerMessage) GetPong() *WorkerPong { return nil } +func (x *ServerMessage) GetGoAway() *AgentHttp_GoAway { + if x != nil { + if x, ok := x.Message.(*ServerMessage_GoAway); ok { + return x.GoAway + } + } + return nil +} + type isServerMessage_Message interface { isServerMessage_Message() } @@ -715,6 +843,12 @@ type ServerMessage_Pong struct { Pong *WorkerPong `protobuf:"bytes,4,opt,name=pong,proto3,oneof"` } +type ServerMessage_GoAway struct { + // the server is draining this control connection: re-register elsewhere, + // in-flight agent HTTP streams run to completion + GoAway *AgentHttp_GoAway `protobuf:"bytes,6,opt,name=go_away,json=goAway,proto3,oneof"` +} + func (*ServerMessage_Register) isServerMessage_Message() {} func (*ServerMessage_Availability) isServerMessage_Message() {} @@ -725,6 +859,8 @@ func (*ServerMessage_Termination) isServerMessage_Message() {} func (*ServerMessage_Pong) isServerMessage_Message() {} +func (*ServerMessage_GoAway) isServerMessage_Message() {} + type SimulateJobRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Type JobType `protobuf:"varint,1,opt,name=type,proto3,enum=livekit.JobType" json:"type,omitempty"` @@ -882,18 +1018,24 @@ func (x *WorkerPong) GetTimestamp() int64 { } type RegisterWorkerRequest struct { - state protoimpl.MessageState `protogen:"open.v1"` - Type JobType `protobuf:"varint,1,opt,name=type,proto3,enum=livekit.JobType" json:"type,omitempty"` - AgentName string `protobuf:"bytes,8,opt,name=agent_name,json=agentName,proto3" json:"agent_name,omitempty"` - // string worker_id = 2; - Version string `protobuf:"bytes,3,opt,name=version,proto3" json:"version,omitempty"` - // string name = 4 [deprecated = true]; + state protoimpl.MessageState `protogen:"open.v1"` + Type JobType `protobuf:"varint,1,opt,name=type,proto3,enum=livekit.JobType" json:"type,omitempty"` + AgentName string `protobuf:"bytes,8,opt,name=agent_name,json=agentName,proto3" json:"agent_name,omitempty"` + Version string `protobuf:"bytes,3,opt,name=version,proto3" json:"version,omitempty"` PingInterval uint32 `protobuf:"varint,5,opt,name=ping_interval,json=pingInterval,proto3" json:"ping_interval,omitempty"` Namespace *string `protobuf:"bytes,6,opt,name=namespace,proto3,oneof" json:"namespace,omitempty"` AllowedPermissions *ParticipantPermission `protobuf:"bytes,7,opt,name=allowed_permissions,json=allowedPermissions,proto3" json:"allowed_permissions,omitempty"` Deployment string `protobuf:"bytes,9,opt,name=deployment,proto3" json:"deployment,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // HTTP endpoints this worker serves through the data plane, in route order. + // Empty means the worker exposes no endpoints and needs no data connections. + Endpoints []*AgentHttp_AgentEndpoint `protobuf:"bytes,10,rep,name=endpoints,proto3" json:"endpoints,omitempty"` + // random per process; a re-registration with a new instance_id supersedes the + // previous epoch and closes its remaining connections + InstanceId string `protobuf:"bytes,11,opt,name=instance_id,json=instanceId,proto3" json:"instance_id,omitempty"` + // highest data-plane protocol version the worker supports; 0 = unsupported + EndpointProtocol uint32 `protobuf:"varint,12,opt,name=endpoint_protocol,json=endpointProtocol,proto3" json:"endpoint_protocol,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *RegisterWorkerRequest) Reset() { @@ -975,12 +1117,35 @@ func (x *RegisterWorkerRequest) GetDeployment() string { return "" } +func (x *RegisterWorkerRequest) GetEndpoints() []*AgentHttp_AgentEndpoint { + if x != nil { + return x.Endpoints + } + return nil +} + +func (x *RegisterWorkerRequest) GetInstanceId() string { + if x != nil { + return x.InstanceId + } + return "" +} + +func (x *RegisterWorkerRequest) GetEndpointProtocol() uint32 { + if x != nil { + return x.EndpointProtocol + } + return 0 +} + type RegisterWorkerResponse struct { - state protoimpl.MessageState `protogen:"open.v1"` - WorkerId string `protobuf:"bytes,1,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty"` - ServerInfo *ServerInfo `protobuf:"bytes,3,opt,name=server_info,json=serverInfo,proto3" json:"server_info,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + WorkerId string `protobuf:"bytes,1,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty"` + ServerInfo *ServerInfo `protobuf:"bytes,3,opt,name=server_info,json=serverInfo,proto3" json:"server_info,omitempty"` + // present iff the registration carried endpoints and the server supports the data plane + EndpointSettings *AgentHttp_AgentEndpointSettings `protobuf:"bytes,4,opt,name=endpoint_settings,json=endpointSettings,proto3" json:"endpoint_settings,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *RegisterWorkerResponse) Reset() { @@ -1027,6 +1192,13 @@ func (x *RegisterWorkerResponse) GetServerInfo() *ServerInfo { return nil } +func (x *RegisterWorkerResponse) GetEndpointSettings() *AgentHttp_AgentEndpointSettings { + if x != nil { + return x.EndpointSettings + } + return nil +} + type MigrateJobRequest struct { state protoimpl.MessageState `protogen:"open.v1"` // string job_id = 1 [deprecated = true]; @@ -1289,11 +1461,15 @@ func (x *UpdateJobStatus) GetError() string { } type UpdateWorkerStatus struct { - state protoimpl.MessageState `protogen:"open.v1"` - Status *WorkerStatus `protobuf:"varint,1,opt,name=status,proto3,enum=livekit.WorkerStatus,oneof" json:"status,omitempty"` - // optional string metadata = 2 [deprecated=true]; - Load float32 `protobuf:"fixed32,3,opt,name=load,proto3" json:"load,omitempty"` - JobCount uint32 `protobuf:"varint,4,opt,name=job_count,json=jobCount,proto3" json:"job_count,omitempty"` + state protoimpl.MessageState `protogen:"open.v1"` + Status *WorkerStatus `protobuf:"varint,1,opt,name=status,proto3,enum=livekit.WorkerStatus,oneof" json:"status,omitempty"` + Load float32 `protobuf:"fixed32,3,opt,name=load,proto3" json:"load,omitempty"` + JobCount uint32 `protobuf:"varint,4,opt,name=job_count,json=jobCount,proto3" json:"job_count,omitempty"` + // worker wants no new streams; existing streams run to completion + Draining bool `protobuf:"varint,5,opt,name=draining,proto3" json:"draining,omitempty"` + // monotonic per registration; status updates may interleave across connections + // and a stale report must not regress newer state + Seq uint64 `protobuf:"varint,6,opt,name=seq,proto3" json:"seq,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1349,6 +1525,20 @@ func (x *UpdateWorkerStatus) GetJobCount() uint32 { return 0 } +func (x *UpdateWorkerStatus) GetDraining() bool { + if x != nil { + return x.Draining + } + return false +} + +func (x *UpdateWorkerStatus) GetSeq() uint64 { + if x != nil { + return x.Seq + } + return 0 +} + type JobAssignment struct { state protoimpl.MessageState `protogen:"open.v1"` Job *Job `protobuf:"bytes,1,opt,name=job,proto3" json:"job,omitempty"` @@ -1453,6 +1643,357 @@ func (x *JobTermination) GetJobId() string { return "" } +// ----- agent HTTP endpoints data plane ----- +// +// AgentHttp namespaces the data plane: workers expose HTTP endpoints served at +// /agents/{agent_name}/{deployment}/{path} without binding any local listener. +// The worker opens ONE WebTransport (QUIC) session to /agent that carries both +// its control stream (the same WorkerMessage/ServerMessage exchange as the +// WebSocket control connection, length-delimited) and every HTTP exchange: the +// node opens one bidirectional QUIC stream per request. QUIC provides the +// multiplexing and per-stream flow control, so there is no credit accounting +// and no attach handshake. +// +// Each exchange stream is: +// +// [len:u32be][StreamPreamble] the only LiveKit framing +// [ ... opaque bytes ... ] one HTTP/1.1 exchange, to FIN or RESET_STREAM +// +// After the preamble the stream is byte-transparent: the node writes a +// canonical HTTP/1.1 request and the worker replies with an HTTP/1.1 response, +// each parsed by whatever HTTP implementation the side already has. A WebSocket +// upgrade is therefore an ordinary request whose response is a 101, after which +// the stream is a byte pipe. The protocol evolves by version negotiation at +// registration, so the preamble carries no version of its own. +// +// Bodies use ordinary HTTP/1.1 framing, Content-Length or chunked, and are +// unbounded: a body may be many gigabytes and must be streamed. Request and +// response size limits are policy for the layer above. +// +// How one direction ends: +// +// success end of message per the body's own framing, then FIN. +// failed mid-body chunked trailer fields x-lk-completion and x-lk-error, +// in band, so the outcome cannot race the bytes it +// describes. +// failed before any RESET_STREAM carrying an HttpStreamResetCode, sound +// bytes only here, where nothing is in flight. +// +// FIN before the body's own framing says it is complete is truncation, and a +// receiver must surface it as such. +// +// The x-lk- header prefix is reserved for this signalling in both directions. +// A node strips it from client-supplied request headers, so a client cannot +// forge one, and strips it from responses, so it never reaches the end client. +type AgentHttp struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AgentHttp) Reset() { + *x = AgentHttp{} + mi := &file_livekit_agent_proto_msgTypes[16] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AgentHttp) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AgentHttp) ProtoMessage() {} + +func (x *AgentHttp) ProtoReflect() protoreflect.Message { + mi := &file_livekit_agent_proto_msgTypes[16] + 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 AgentHttp.ProtoReflect.Descriptor instead. +func (*AgentHttp) Descriptor() ([]byte, []int) { + return file_livekit_agent_proto_rawDescGZIP(), []int{16} +} + +type AgentHttp_AgentEndpoint struct { + state protoimpl.MessageState `protogen:"open.v1"` + // starlette-style path template rooted at '/': {name} (one segment, default), + // {name:int}, {name:float}, {name:uuid} (constrained single segments), + // {name:path} (multi-segment). Custom converters are rejected. + Path string `protobuf:"bytes,1,opt,name=path,proto3" json:"path,omitempty"` + // uppercase, exactly as the application routes them (FastAPI does not imply + // HEAD from GET) + Methods []string `protobuf:"bytes,2,rep,name=methods,proto3" json:"methods,omitempty"` + Kind AgentHttp_AgentEndpointKind `protobuf:"varint,3,opt,name=kind,proto3,enum=livekit.AgentHttp_AgentEndpointKind" json:"kind,omitempty"` + // reachable without a project token + Public bool `protobuf:"varint,4,opt,name=public,proto3" json:"public,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AgentHttp_AgentEndpoint) Reset() { + *x = AgentHttp_AgentEndpoint{} + mi := &file_livekit_agent_proto_msgTypes[19] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AgentHttp_AgentEndpoint) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AgentHttp_AgentEndpoint) ProtoMessage() {} + +func (x *AgentHttp_AgentEndpoint) ProtoReflect() protoreflect.Message { + mi := &file_livekit_agent_proto_msgTypes[19] + 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 AgentHttp_AgentEndpoint.ProtoReflect.Descriptor instead. +func (*AgentHttp_AgentEndpoint) Descriptor() ([]byte, []int) { + return file_livekit_agent_proto_rawDescGZIP(), []int{16, 0} +} + +func (x *AgentHttp_AgentEndpoint) GetPath() string { + if x != nil { + return x.Path + } + return "" +} + +func (x *AgentHttp_AgentEndpoint) GetMethods() []string { + if x != nil { + return x.Methods + } + return nil +} + +func (x *AgentHttp_AgentEndpoint) GetKind() AgentHttp_AgentEndpointKind { + if x != nil { + return x.Kind + } + return AgentHttp_AEK_HTTP +} + +func (x *AgentHttp_AgentEndpoint) GetPublic() bool { + if x != nil { + return x.Public + } + return false +} + +// registration-level settings negotiated on the control connection, returned +// in RegisterWorkerResponse when the worker declares endpoints and the server +// supports the data plane. +type AgentHttp_AgentEndpointSettings struct { + state protoimpl.MessageState `protogen:"open.v1"` + // the negotiated data-plane protocol version; the worker speaks exactly this + // or closes the session + Protocol uint32 `protobuf:"varint,1,opt,name=protocol,proto3" json:"protocol,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AgentHttp_AgentEndpointSettings) Reset() { + *x = AgentHttp_AgentEndpointSettings{} + mi := &file_livekit_agent_proto_msgTypes[20] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AgentHttp_AgentEndpointSettings) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AgentHttp_AgentEndpointSettings) ProtoMessage() {} + +func (x *AgentHttp_AgentEndpointSettings) ProtoReflect() protoreflect.Message { + mi := &file_livekit_agent_proto_msgTypes[20] + 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 AgentHttp_AgentEndpointSettings.ProtoReflect.Descriptor instead. +func (*AgentHttp_AgentEndpointSettings) Descriptor() ([]byte, []int) { + return file_livekit_agent_proto_rawDescGZIP(), []int{16, 1} +} + +func (x *AgentHttp_AgentEndpointSettings) GetProtocol() uint32 { + if x != nil { + return x.Protocol + } + return 0 +} + +// the first thing on every exchange stream, node -> worker. It carries only +// what the worker cannot derive from the HTTP bytes that follow and what it +// must not infer from client-supplied headers - identity above all. +type AgentHttp_StreamPreamble struct { + state protoimpl.MessageState `protogen:"open.v1"` + // what the bytes after this preamble are. A field rather than a framing + // tag, so a new stream semantic costs no new mechanism. + Kind AgentHttp_AgentEndpointKind `protobuf:"varint,1,opt,name=kind,proto3,enum=livekit.AgentHttp_AgentEndpointKind" json:"kind,omitempty"` + RequestId string `protobuf:"bytes,2,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` + // the caller holds agent-endpoint call permission for this agent and + // deployment. Always true on a non-public route. + Authorized bool `protobuf:"varint,3,opt,name=authorized,proto3" json:"authorized,omitempty"` + // the manifest template that matched, e.g. "/items/{id}" + Route string `protobuf:"bytes,4,opt,name=route,proto3" json:"route,omitempty"` + // relative, because node and worker clocks are not synchronized. 0 imposes + // no deadline. + TimeoutMs uint32 `protobuf:"varint,5,opt,name=timeout_ms,json=timeoutMs,proto3" json:"timeout_ms,omitempty"` + // end client address, host only + ClientAddr string `protobuf:"bytes,6,opt,name=client_addr,json=clientAddr,proto3" json:"client_addr,omitempty"` + // "http" or "https" as the end client saw it; not derivable worker-side + Scheme string `protobuf:"bytes,7,opt,name=scheme,proto3" json:"scheme,omitempty"` // NEXT_ID: 8 + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AgentHttp_StreamPreamble) Reset() { + *x = AgentHttp_StreamPreamble{} + mi := &file_livekit_agent_proto_msgTypes[21] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AgentHttp_StreamPreamble) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AgentHttp_StreamPreamble) ProtoMessage() {} + +func (x *AgentHttp_StreamPreamble) ProtoReflect() protoreflect.Message { + mi := &file_livekit_agent_proto_msgTypes[21] + 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 AgentHttp_StreamPreamble.ProtoReflect.Descriptor instead. +func (*AgentHttp_StreamPreamble) Descriptor() ([]byte, []int) { + return file_livekit_agent_proto_rawDescGZIP(), []int{16, 2} +} + +func (x *AgentHttp_StreamPreamble) GetKind() AgentHttp_AgentEndpointKind { + if x != nil { + return x.Kind + } + return AgentHttp_AEK_HTTP +} + +func (x *AgentHttp_StreamPreamble) GetRequestId() string { + if x != nil { + return x.RequestId + } + return "" +} + +func (x *AgentHttp_StreamPreamble) GetAuthorized() bool { + if x != nil { + return x.Authorized + } + return false +} + +func (x *AgentHttp_StreamPreamble) GetRoute() string { + if x != nil { + return x.Route + } + return "" +} + +func (x *AgentHttp_StreamPreamble) GetTimeoutMs() uint32 { + if x != nil { + return x.TimeoutMs + } + return 0 +} + +func (x *AgentHttp_StreamPreamble) GetClientAddr() string { + if x != nil { + return x.ClientAddr + } + return "" +} + +func (x *AgentHttp_StreamPreamble) GetScheme() string { + if x != nil { + return x.Scheme + } + return "" +} + +// the server is draining: the worker should re-register elsewhere; in-flight +// HTTP exchanges run to completion +type AgentHttp_GoAway struct { + state protoimpl.MessageState `protogen:"open.v1"` + Reason string `protobuf:"bytes,1,opt,name=reason,proto3" json:"reason,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AgentHttp_GoAway) Reset() { + *x = AgentHttp_GoAway{} + mi := &file_livekit_agent_proto_msgTypes[22] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AgentHttp_GoAway) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AgentHttp_GoAway) ProtoMessage() {} + +func (x *AgentHttp_GoAway) ProtoReflect() protoreflect.Message { + mi := &file_livekit_agent_proto_msgTypes[22] + 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 AgentHttp_GoAway.ProtoReflect.Descriptor instead. +func (*AgentHttp_GoAway) Descriptor() ([]byte, []int) { + return file_livekit_agent_proto_rawDescGZIP(), []int{16, 3} +} + +func (x *AgentHttp_GoAway) GetReason() string { + if x != nil { + return x.Reason + } + return "" +} + var File_livekit_agent_proto protoreflect.FileDescriptor const file_livekit_agent_proto_rawDesc = "" + @@ -1506,7 +2047,7 @@ const file_livekit_agent_proto_rawDesc = "" + "\fsimulate_job\x18\x06 \x01(\v2\x1b.livekit.SimulateJobRequestH\x00R\vsimulateJob\x12=\n" + "\vmigrate_job\x18\a \x01(\v2\x1a.livekit.MigrateJobRequestH\x00R\n" + "migrateJobB\t\n" + - "\amessage\"\xbf\x02\n" + + "\amessage\"\xf5\x02\n" + "\rServerMessage\x12=\n" + "\bregister\x18\x01 \x01(\v2\x1f.livekit.RegisterWorkerResponseH\x00R\bregister\x12B\n" + "\favailability\x18\x02 \x01(\v2\x1c.livekit.AvailabilityRequestH\x00R\favailability\x128\n" + @@ -1514,7 +2055,8 @@ const file_livekit_agent_proto_rawDesc = "" + "assignment\x18\x03 \x01(\v2\x16.livekit.JobAssignmentH\x00R\n" + "assignment\x12;\n" + "\vtermination\x18\x05 \x01(\v2\x17.livekit.JobTerminationH\x00R\vtermination\x12)\n" + - "\x04pong\x18\x04 \x01(\v2\x13.livekit.WorkerPongH\x00R\x04pongB\t\n" + + "\x04pong\x18\x04 \x01(\v2\x13.livekit.WorkerPongH\x00R\x04pong\x124\n" + + "\ago_away\x18\x06 \x01(\v2\x19.livekit.AgentHttp.GoAwayH\x00R\x06goAwayB\t\n" + "\amessage\"\x99\x01\n" + "\x12SimulateJobRequest\x12$\n" + "\x04type\x18\x01 \x01(\x0e2\x10.livekit.JobTypeR\x04type\x12!\n" + @@ -1526,7 +2068,7 @@ const file_livekit_agent_proto_rawDesc = "" + "\n" + "WorkerPong\x12%\n" + "\x0elast_timestamp\x18\x01 \x01(\x03R\rlastTimestamp\x12\x1c\n" + - "\ttimestamp\x18\x02 \x01(\x03R\ttimestamp\"\xbd\x02\n" + + "\ttimestamp\x18\x02 \x01(\x03R\ttimestamp\"\xe6\x03\n" + "\x15RegisterWorkerRequest\x12$\n" + "\x04type\x18\x01 \x01(\x0e2\x10.livekit.JobTypeR\x04type\x12\x1d\n" + "\n" + @@ -1537,13 +2079,20 @@ const file_livekit_agent_proto_rawDesc = "" + "\x13allowed_permissions\x18\a \x01(\v2\x1e.livekit.ParticipantPermissionR\x12allowedPermissions\x12\x1e\n" + "\n" + "deployment\x18\t \x01(\tR\n" + - "deploymentB\f\n" + + "deployment\x12>\n" + + "\tendpoints\x18\n" + + " \x03(\v2 .livekit.AgentHttp.AgentEndpointR\tendpoints\x12.\n" + + "\vinstance_id\x18\v \x01(\tB\r\xbaP\n" + + "instanceIDR\n" + + "instanceId\x12+\n" + + "\x11endpoint_protocol\x18\f \x01(\rR\x10endpointProtocolB\f\n" + "\n" + - "_namespace\"x\n" + + "_namespaceJ\x04\b\x02\x10\x03J\x04\b\x04\x10\x05\"\xd5\x01\n" + "\x16RegisterWorkerResponse\x12(\n" + "\tworker_id\x18\x01 \x01(\tB\v\xbaP\bworkerIDR\bworkerId\x124\n" + "\vserver_info\x18\x03 \x01(\v2\x13.livekit.ServerInfoR\n" + - "serverInfo\",\n" + + "serverInfo\x12U\n" + + "\x11endpoint_settings\x18\x04 \x01(\v2(.livekit.AgentHttp.AgentEndpointSettingsR\x10endpointSettingsJ\x04\b\x02\x10\x03\",\n" + "\x11MigrateJobRequest\x12\x17\n" + "\ajob_ids\x18\x02 \x03(\tR\x06jobIds\"Q\n" + "\x13AvailabilityRequest\x12\x1e\n" + @@ -1564,19 +2113,53 @@ const file_livekit_agent_proto_rawDesc = "" + "\x0fUpdateJobStatus\x12\x1f\n" + "\x06job_id\x18\x01 \x01(\tB\b\xbaP\x05jobIDR\x05jobId\x12*\n" + "\x06status\x18\x02 \x01(\x0e2\x12.livekit.JobStatusR\x06status\x12\x14\n" + - "\x05error\x18\x03 \x01(\tR\x05error\"\x84\x01\n" + + "\x05error\x18\x03 \x01(\tR\x05error\"\xb8\x01\n" + "\x12UpdateWorkerStatus\x122\n" + "\x06status\x18\x01 \x01(\x0e2\x15.livekit.WorkerStatusH\x00R\x06status\x88\x01\x01\x12\x12\n" + "\x04load\x18\x03 \x01(\x02R\x04load\x12\x1b\n" + - "\tjob_count\x18\x04 \x01(\rR\bjobCountB\t\n" + - "\a_status\"d\n" + + "\tjob_count\x18\x04 \x01(\rR\bjobCount\x12\x1a\n" + + "\bdraining\x18\x05 \x01(\bR\bdraining\x12\x10\n" + + "\x03seq\x18\x06 \x01(\x04R\x03seqB\t\n" + + "\a_statusJ\x04\b\x02\x10\x03\"d\n" + "\rJobAssignment\x12\x1e\n" + "\x03job\x18\x01 \x01(\v2\f.livekit.JobR\x03job\x12\x15\n" + "\x03url\x18\x02 \x01(\tH\x00R\x03url\x88\x01\x01\x12\x14\n" + "\x05token\x18\x03 \x01(\tR\x05tokenB\x06\n" + "\x04_url\"1\n" + "\x0eJobTermination\x12\x1f\n" + - "\x06job_id\x18\x01 \x01(\tB\b\xbaP\x05jobIDR\x05jobId*<\n" + + "\x06job_id\x18\x01 \x01(\tB\b\xbaP\x05jobIDR\x05jobId\"\xaa\x05\n" + + "\tAgentHttp\x1a\x8f\x01\n" + + "\rAgentEndpoint\x12\x12\n" + + "\x04path\x18\x01 \x01(\tR\x04path\x12\x18\n" + + "\amethods\x18\x02 \x03(\tR\amethods\x128\n" + + "\x04kind\x18\x03 \x01(\x0e2$.livekit.AgentHttp.AgentEndpointKindR\x04kind\x12\x16\n" + + "\x06public\x18\x04 \x01(\bR\x06public\x1a?\n" + + "\x15AgentEndpointSettings\x12\x1a\n" + + "\bprotocol\x18\x01 \x01(\rR\bprotocolJ\x04\b\x02\x10\x03J\x04\b\x03\x10\x04\x1a\x8a\x02\n" + + "\x0eStreamPreamble\x128\n" + + "\x04kind\x18\x01 \x01(\x0e2$.livekit.AgentHttp.AgentEndpointKindR\x04kind\x12+\n" + + "\n" + + "request_id\x18\x02 \x01(\tB\f\xbaP\trequestIDR\trequestId\x12\x1e\n" + + "\n" + + "authorized\x18\x03 \x01(\bR\n" + + "authorized\x12\x14\n" + + "\x05route\x18\x04 \x01(\tR\x05route\x12\x1d\n" + + "\n" + + "timeout_ms\x18\x05 \x01(\rR\ttimeoutMs\x12$\n" + + "\vclient_addr\x18\x06 \x01(\tB\x03\xc0P\x01R\n" + + "clientAddr\x12\x16\n" + + "\x06scheme\x18\a \x01(\tR\x06scheme\x1a \n" + + "\x06GoAway\x12\x16\n" + + "\x06reason\x18\x01 \x01(\tR\x06reason\"/\n" + + "\x11AgentEndpointKind\x12\f\n" + + "\bAEK_HTTP\x10\x00\x12\f\n" + + "\bAEK_TEXT\x10\x01\"j\n" + + "\x13HttpStreamResetCode\x12\r\n" + + "\tHSR_ABORT\x10\x00\x12\x0f\n" + + "\vHSR_REFUSED\x10\x01\x12\x10\n" + + "\fHSR_INTERNAL\x10\x02\x12\x0f\n" + + "\vHSR_TIMEOUT\x10\x03\x12\x10\n" + + "\fHSR_PROTOCOL\x10\x04*<\n" + "\aJobType\x12\v\n" + "\aJT_ROOM\x10\x00\x12\x10\n" + "\fJT_PUBLISHER\x10\x01\x12\x12\n" + @@ -1605,70 +2188,82 @@ func file_livekit_agent_proto_rawDescGZIP() []byte { return file_livekit_agent_proto_rawDescData } -var file_livekit_agent_proto_enumTypes = make([]protoimpl.EnumInfo, 3) -var file_livekit_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 18) +var file_livekit_agent_proto_enumTypes = make([]protoimpl.EnumInfo, 5) +var file_livekit_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 23) var file_livekit_agent_proto_goTypes = []any{ - (JobType)(0), // 0: livekit.JobType - (WorkerStatus)(0), // 1: livekit.WorkerStatus - (JobStatus)(0), // 2: livekit.JobStatus - (*Job)(nil), // 3: livekit.Job - (*JobState)(nil), // 4: livekit.JobState - (*WorkerMessage)(nil), // 5: livekit.WorkerMessage - (*ServerMessage)(nil), // 6: livekit.ServerMessage - (*SimulateJobRequest)(nil), // 7: livekit.SimulateJobRequest - (*WorkerPing)(nil), // 8: livekit.WorkerPing - (*WorkerPong)(nil), // 9: livekit.WorkerPong - (*RegisterWorkerRequest)(nil), // 10: livekit.RegisterWorkerRequest - (*RegisterWorkerResponse)(nil), // 11: livekit.RegisterWorkerResponse - (*MigrateJobRequest)(nil), // 12: livekit.MigrateJobRequest - (*AvailabilityRequest)(nil), // 13: livekit.AvailabilityRequest - (*AvailabilityResponse)(nil), // 14: livekit.AvailabilityResponse - (*UpdateJobStatus)(nil), // 15: livekit.UpdateJobStatus - (*UpdateWorkerStatus)(nil), // 16: livekit.UpdateWorkerStatus - (*JobAssignment)(nil), // 17: livekit.JobAssignment - (*JobTermination)(nil), // 18: livekit.JobTermination - nil, // 19: livekit.Job.AttributesEntry - nil, // 20: livekit.AvailabilityResponse.ParticipantAttributesEntry - (*Room)(nil), // 21: livekit.Room - (*ParticipantInfo)(nil), // 22: livekit.ParticipantInfo - (*ParticipantPermission)(nil), // 23: livekit.ParticipantPermission - (*ServerInfo)(nil), // 24: livekit.ServerInfo + (JobType)(0), // 0: livekit.JobType + (WorkerStatus)(0), // 1: livekit.WorkerStatus + (JobStatus)(0), // 2: livekit.JobStatus + (AgentHttp_AgentEndpointKind)(0), // 3: livekit.AgentHttp.AgentEndpointKind + (AgentHttp_HttpStreamResetCode)(0), // 4: livekit.AgentHttp.HttpStreamResetCode + (*Job)(nil), // 5: livekit.Job + (*JobState)(nil), // 6: livekit.JobState + (*WorkerMessage)(nil), // 7: livekit.WorkerMessage + (*ServerMessage)(nil), // 8: livekit.ServerMessage + (*SimulateJobRequest)(nil), // 9: livekit.SimulateJobRequest + (*WorkerPing)(nil), // 10: livekit.WorkerPing + (*WorkerPong)(nil), // 11: livekit.WorkerPong + (*RegisterWorkerRequest)(nil), // 12: livekit.RegisterWorkerRequest + (*RegisterWorkerResponse)(nil), // 13: livekit.RegisterWorkerResponse + (*MigrateJobRequest)(nil), // 14: livekit.MigrateJobRequest + (*AvailabilityRequest)(nil), // 15: livekit.AvailabilityRequest + (*AvailabilityResponse)(nil), // 16: livekit.AvailabilityResponse + (*UpdateJobStatus)(nil), // 17: livekit.UpdateJobStatus + (*UpdateWorkerStatus)(nil), // 18: livekit.UpdateWorkerStatus + (*JobAssignment)(nil), // 19: livekit.JobAssignment + (*JobTermination)(nil), // 20: livekit.JobTermination + (*AgentHttp)(nil), // 21: livekit.AgentHttp + nil, // 22: livekit.Job.AttributesEntry + nil, // 23: livekit.AvailabilityResponse.ParticipantAttributesEntry + (*AgentHttp_AgentEndpoint)(nil), // 24: livekit.AgentHttp.AgentEndpoint + (*AgentHttp_AgentEndpointSettings)(nil), // 25: livekit.AgentHttp.AgentEndpointSettings + (*AgentHttp_StreamPreamble)(nil), // 26: livekit.AgentHttp.StreamPreamble + (*AgentHttp_GoAway)(nil), // 27: livekit.AgentHttp.GoAway + (*Room)(nil), // 28: livekit.Room + (*ParticipantInfo)(nil), // 29: livekit.ParticipantInfo + (*ParticipantPermission)(nil), // 30: livekit.ParticipantPermission + (*ServerInfo)(nil), // 31: livekit.ServerInfo } var file_livekit_agent_proto_depIdxs = []int32{ 0, // 0: livekit.Job.type:type_name -> livekit.JobType - 21, // 1: livekit.Job.room:type_name -> livekit.Room - 22, // 2: livekit.Job.participant:type_name -> livekit.ParticipantInfo - 4, // 3: livekit.Job.state:type_name -> livekit.JobState - 19, // 4: livekit.Job.attributes:type_name -> livekit.Job.AttributesEntry + 28, // 1: livekit.Job.room:type_name -> livekit.Room + 29, // 2: livekit.Job.participant:type_name -> livekit.ParticipantInfo + 6, // 3: livekit.Job.state:type_name -> livekit.JobState + 22, // 4: livekit.Job.attributes:type_name -> livekit.Job.AttributesEntry 2, // 5: livekit.JobState.status:type_name -> livekit.JobStatus - 10, // 6: livekit.WorkerMessage.register:type_name -> livekit.RegisterWorkerRequest - 14, // 7: livekit.WorkerMessage.availability:type_name -> livekit.AvailabilityResponse - 16, // 8: livekit.WorkerMessage.update_worker:type_name -> livekit.UpdateWorkerStatus - 15, // 9: livekit.WorkerMessage.update_job:type_name -> livekit.UpdateJobStatus - 8, // 10: livekit.WorkerMessage.ping:type_name -> livekit.WorkerPing - 7, // 11: livekit.WorkerMessage.simulate_job:type_name -> livekit.SimulateJobRequest - 12, // 12: livekit.WorkerMessage.migrate_job:type_name -> livekit.MigrateJobRequest - 11, // 13: livekit.ServerMessage.register:type_name -> livekit.RegisterWorkerResponse - 13, // 14: livekit.ServerMessage.availability:type_name -> livekit.AvailabilityRequest - 17, // 15: livekit.ServerMessage.assignment:type_name -> livekit.JobAssignment - 18, // 16: livekit.ServerMessage.termination:type_name -> livekit.JobTermination - 9, // 17: livekit.ServerMessage.pong:type_name -> livekit.WorkerPong - 0, // 18: livekit.SimulateJobRequest.type:type_name -> livekit.JobType - 21, // 19: livekit.SimulateJobRequest.room:type_name -> livekit.Room - 22, // 20: livekit.SimulateJobRequest.participant:type_name -> livekit.ParticipantInfo - 0, // 21: livekit.RegisterWorkerRequest.type:type_name -> livekit.JobType - 23, // 22: livekit.RegisterWorkerRequest.allowed_permissions:type_name -> livekit.ParticipantPermission - 24, // 23: livekit.RegisterWorkerResponse.server_info:type_name -> livekit.ServerInfo - 3, // 24: livekit.AvailabilityRequest.job:type_name -> livekit.Job - 20, // 25: livekit.AvailabilityResponse.participant_attributes:type_name -> livekit.AvailabilityResponse.ParticipantAttributesEntry - 2, // 26: livekit.UpdateJobStatus.status:type_name -> livekit.JobStatus - 1, // 27: livekit.UpdateWorkerStatus.status:type_name -> livekit.WorkerStatus - 3, // 28: livekit.JobAssignment.job:type_name -> livekit.Job - 29, // [29:29] is the sub-list for method output_type - 29, // [29:29] is the sub-list for method input_type - 29, // [29:29] is the sub-list for extension type_name - 29, // [29:29] is the sub-list for extension extendee - 0, // [0:29] is the sub-list for field type_name + 12, // 6: livekit.WorkerMessage.register:type_name -> livekit.RegisterWorkerRequest + 16, // 7: livekit.WorkerMessage.availability:type_name -> livekit.AvailabilityResponse + 18, // 8: livekit.WorkerMessage.update_worker:type_name -> livekit.UpdateWorkerStatus + 17, // 9: livekit.WorkerMessage.update_job:type_name -> livekit.UpdateJobStatus + 10, // 10: livekit.WorkerMessage.ping:type_name -> livekit.WorkerPing + 9, // 11: livekit.WorkerMessage.simulate_job:type_name -> livekit.SimulateJobRequest + 14, // 12: livekit.WorkerMessage.migrate_job:type_name -> livekit.MigrateJobRequest + 13, // 13: livekit.ServerMessage.register:type_name -> livekit.RegisterWorkerResponse + 15, // 14: livekit.ServerMessage.availability:type_name -> livekit.AvailabilityRequest + 19, // 15: livekit.ServerMessage.assignment:type_name -> livekit.JobAssignment + 20, // 16: livekit.ServerMessage.termination:type_name -> livekit.JobTermination + 11, // 17: livekit.ServerMessage.pong:type_name -> livekit.WorkerPong + 27, // 18: livekit.ServerMessage.go_away:type_name -> livekit.AgentHttp.GoAway + 0, // 19: livekit.SimulateJobRequest.type:type_name -> livekit.JobType + 28, // 20: livekit.SimulateJobRequest.room:type_name -> livekit.Room + 29, // 21: livekit.SimulateJobRequest.participant:type_name -> livekit.ParticipantInfo + 0, // 22: livekit.RegisterWorkerRequest.type:type_name -> livekit.JobType + 30, // 23: livekit.RegisterWorkerRequest.allowed_permissions:type_name -> livekit.ParticipantPermission + 24, // 24: livekit.RegisterWorkerRequest.endpoints:type_name -> livekit.AgentHttp.AgentEndpoint + 31, // 25: livekit.RegisterWorkerResponse.server_info:type_name -> livekit.ServerInfo + 25, // 26: livekit.RegisterWorkerResponse.endpoint_settings:type_name -> livekit.AgentHttp.AgentEndpointSettings + 5, // 27: livekit.AvailabilityRequest.job:type_name -> livekit.Job + 23, // 28: livekit.AvailabilityResponse.participant_attributes:type_name -> livekit.AvailabilityResponse.ParticipantAttributesEntry + 2, // 29: livekit.UpdateJobStatus.status:type_name -> livekit.JobStatus + 1, // 30: livekit.UpdateWorkerStatus.status:type_name -> livekit.WorkerStatus + 5, // 31: livekit.JobAssignment.job:type_name -> livekit.Job + 3, // 32: livekit.AgentHttp.AgentEndpoint.kind:type_name -> livekit.AgentHttp.AgentEndpointKind + 3, // 33: livekit.AgentHttp.StreamPreamble.kind:type_name -> livekit.AgentHttp.AgentEndpointKind + 34, // [34:34] is the sub-list for method output_type + 34, // [34:34] is the sub-list for method input_type + 34, // [34:34] is the sub-list for extension type_name + 34, // [34:34] is the sub-list for extension extendee + 0, // [0:34] is the sub-list for field type_name } func init() { file_livekit_agent_proto_init() } @@ -1693,6 +2288,7 @@ func file_livekit_agent_proto_init() { (*ServerMessage_Assignment)(nil), (*ServerMessage_Termination)(nil), (*ServerMessage_Pong)(nil), + (*ServerMessage_GoAway)(nil), } file_livekit_agent_proto_msgTypes[7].OneofWrappers = []any{} file_livekit_agent_proto_msgTypes[13].OneofWrappers = []any{} @@ -1702,8 +2298,8 @@ func file_livekit_agent_proto_init() { File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_livekit_agent_proto_rawDesc), len(file_livekit_agent_proto_rawDesc)), - NumEnums: 3, - NumMessages: 18, + NumEnums: 5, + NumMessages: 23, NumExtensions: 0, NumServices: 0, }, diff --git a/protobufs/livekit_agent.proto b/protobufs/livekit_agent.proto index c18ecf203..4747feb11 100644 --- a/protobufs/livekit_agent.proto +++ b/protobufs/livekit_agent.proto @@ -78,6 +78,10 @@ message ServerMessage { JobAssignment assignment = 3; JobTermination termination = 5; WorkerPong pong = 4; + + // the server is draining this control connection: re-register elsewhere, + // in-flight agent HTTP streams run to completion + AgentHttp.GoAway go_away = 6; } } @@ -115,20 +119,31 @@ message WorkerPong { } message RegisterWorkerRequest { + reserved 2, 4; // worker_id, name (retired) JobType type = 1; string agent_name = 8; - // string worker_id = 2; string version = 3; - // string name = 4 [deprecated = true]; uint32 ping_interval = 5; optional string namespace = 6; ParticipantPermission allowed_permissions = 7; string deployment = 9; + + // HTTP endpoints this worker serves through the data plane, in route order. + // Empty means the worker exposes no endpoints and needs no data connections. + repeated AgentHttp.AgentEndpoint endpoints = 10; + // random per process; a re-registration with a new instance_id supersedes the + // previous epoch and closes its remaining connections + string instance_id = 11 [(logger.name) = "instanceID"]; + // highest data-plane protocol version the worker supports; 0 = unsupported + uint32 endpoint_protocol = 12; } message RegisterWorkerResponse { + reserved 2; string worker_id = 1 [(logger.name) = "workerID"]; ServerInfo server_info = 3; + // present iff the registration carried endpoints and the server supports the data plane + AgentHttp.AgentEndpointSettings endpoint_settings = 4; } message MigrateJobRequest { @@ -168,10 +183,16 @@ message UpdateJobStatus { } message UpdateWorkerStatus { + reserved 2; // metadata (retired) optional WorkerStatus status = 1; - // optional string metadata = 2 [deprecated=true]; float load = 3; uint32 job_count = 4; + + // worker wants no new streams; existing streams run to completion + bool draining = 5; + // monotonic per registration; status updates may interleave across connections + // and a stale report must not regress newer state + uint64 seq = 6; } message JobAssignment { @@ -183,3 +204,129 @@ message JobAssignment { message JobTermination { string job_id = 1 [(logger.name) = "jobID"]; } + +// ----- agent HTTP endpoints data plane ----- +// +// AgentHttp namespaces the data plane: workers expose HTTP endpoints served at +// /agents/{agent_name}/{deployment}/{path} without binding any local listener. +// The worker opens ONE WebTransport (QUIC) session to /agent that carries both +// its control stream (the same WorkerMessage/ServerMessage exchange as the +// WebSocket control connection, length-delimited) and every HTTP exchange: the +// node opens one bidirectional QUIC stream per request. QUIC provides the +// multiplexing and per-stream flow control, so there is no credit accounting +// and no attach handshake. +// +// Each exchange stream is: +// +// [len:u32be][StreamPreamble] the only LiveKit framing +// [ ... opaque bytes ... ] one HTTP/1.1 exchange, to FIN or RESET_STREAM +// +// After the preamble the stream is byte-transparent: the node writes a +// canonical HTTP/1.1 request and the worker replies with an HTTP/1.1 response, +// each parsed by whatever HTTP implementation the side already has. A WebSocket +// upgrade is therefore an ordinary request whose response is a 101, after which +// the stream is a byte pipe. The protocol evolves by version negotiation at +// registration, so the preamble carries no version of its own. +// +// Bodies use ordinary HTTP/1.1 framing, Content-Length or chunked, and are +// unbounded: a body may be many gigabytes and must be streamed. Request and +// response size limits are policy for the layer above. +// +// How one direction ends: +// +// success end of message per the body's own framing, then FIN. +// failed mid-body chunked trailer fields x-lk-completion and x-lk-error, +// in band, so the outcome cannot race the bytes it +// describes. +// failed before any RESET_STREAM carrying an HttpStreamResetCode, sound +// bytes only here, where nothing is in flight. +// +// FIN before the body's own framing says it is complete is truncation, and a +// receiver must surface it as such. +// +// The x-lk- header prefix is reserved for this signalling in both directions. +// A node strips it from client-supplied request headers, so a client cannot +// forge one, and strips it from responses, so it never reaches the end client. +message AgentHttp { + enum AgentEndpointKind { + AEK_HTTP = 0; + // reserved for future text-mode endpoints + AEK_TEXT = 1; + } + + message AgentEndpoint { + // starlette-style path template rooted at '/': {name} (one segment, default), + // {name:int}, {name:float}, {name:uuid} (constrained single segments), + // {name:path} (multi-segment). Custom converters are rejected. + string path = 1; + // uppercase, exactly as the application routes them (FastAPI does not imply + // HEAD from GET) + repeated string methods = 2; + AgentEndpointKind kind = 3; + // reachable without a project token + bool public = 4; + } + + // registration-level settings negotiated on the control connection, returned + // in RegisterWorkerResponse when the worker declares endpoints and the server + // supports the data plane. + message AgentEndpointSettings { + reserved 2, 3; // attach_token, data_connection_count (retired with the wire pool) + // the negotiated data-plane protocol version; the worker speaks exactly this + // or closes the session + uint32 protocol = 1; + } + + // the first thing on every exchange stream, node -> worker. It carries only + // what the worker cannot derive from the HTTP bytes that follow and what it + // must not infer from client-supplied headers - identity above all. + message StreamPreamble { + // what the bytes after this preamble are. A field rather than a framing + // tag, so a new stream semantic costs no new mechanism. + AgentEndpointKind kind = 1; + string request_id = 2 [(logger.name) = "requestID"]; + // the caller holds agent-endpoint call permission for this agent and + // deployment. Always true on a non-public route. + bool authorized = 3; + // the manifest template that matched, e.g. "/items/{id}" + string route = 4; + // relative, because node and worker clocks are not synchronized. 0 imposes + // no deadline. + uint32 timeout_ms = 5; + // end client address, host only + string client_addr = 6 [(logger.sensitivity) = SENSITIVITY_PII]; + // "http" or "https" as the end client saw it; not derivable worker-side + string scheme = 7; + // NEXT_ID: 8 + } + + // why a stream was reset before any HTTP bytes flowed. This travels as the + // QUIC RESET_STREAM / STOP_SENDING error code, so it is a number with no room + // for detail: the reason string is logged by the side that reset and joined + // to this by request_id. An outcome after bytes have flowed travels in + // trailers instead, where it cannot race them. + enum HttpStreamResetCode { + // no information. The ordinary cancel code, sent whenever a stream is torn + // down without a specific outcome, so it must stay the zero value and must + // not imply that anything was or was not applied. + HSR_ABORT = 0; + // the worker aborted before any application code observed the request, so + // nothing was applied and the exchange is safe to retry. Requires that the + // application was never entered; a 404 it returned is an ordinary response. + HSR_REFUSED = 1; + // the application was entered and then failed before producing a response + // head; side effects may already have happened, so this is not safe to + // retry. + HSR_INTERNAL = 2; + // the deadline elapsed before a response head was produced + HSR_TIMEOUT = 3; + // the peer's bytes were not valid HTTP/1.1 + HSR_PROTOCOL = 4; + } + + // the server is draining: the worker should re-register elsewhere; in-flight + // HTTP exchanges run to completion + message GoAway { + string reason = 1; + } +}