diff --git a/agent/environment.go b/agent/environment.go index 24acbb323..c34bfd86d 100644 --- a/agent/environment.go +++ b/agent/environment.go @@ -14,10 +14,13 @@ func ValidateDeployment(deployment string) error { for i := 0; i < len(deployment); i++ { c := deployment[i] switch { + case c >= 'a' && c <= 'z', c >= 'A' && c <= 'Z', c >= '0' && c <= '9', c == '-', c == '.': case c == '_': return fmt.Errorf("deployment contains reserved character %q", c) - case c <= ' ' || c == 0x7f: - return fmt.Errorf("deployment contains whitespace or control byte at position %d", i) + default: + // deployments appear as a single URL path segment (/agents/{deployment}/...), + // so only alphanumerics, '-' and '.' are accepted + return fmt.Errorf("deployment contains invalid character %q at position %d", c, i) } } return nil diff --git a/livekit/livekit_agent.pb.go b/livekit/livekit_agent.pb.go index a1052dee1..025cdaf72 100644 --- a/livekit/livekit_agent.pb.go +++ b/livekit/livekit_agent.pb.go @@ -183,6 +183,112 @@ func (JobStatus) EnumDescriptor() ([]byte, []int) { return file_livekit_agent_proto_rawDescGZIP(), []int{2} } +type AgentHttp_AgentEndpointKind int32 + +const ( + AgentHttp_AEK_HTTP AgentHttp_AgentEndpointKind = 0 + // a client-initiated WebSocket route; the stream carries the upgrade exchange + // and then raw bidirectional bytes + AgentHttp_AEK_WEBSOCKET AgentHttp_AgentEndpointKind = 1 + // reserved for future text-mode endpoints + AgentHttp_AEK_TEXT AgentHttp_AgentEndpointKind = 2 +) + +// Enum value maps for AgentHttp_AgentEndpointKind. +var ( + AgentHttp_AgentEndpointKind_name = map[int32]string{ + 0: "AEK_HTTP", + 1: "AEK_WEBSOCKET", + 2: "AEK_TEXT", + } + AgentHttp_AgentEndpointKind_value = map[string]int32{ + "AEK_HTTP": 0, + "AEK_WEBSOCKET": 1, + "AEK_TEXT": 2, + } +) + +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} +} + +type AgentHttp_HttpStreamResetCode int32 + +const ( + AgentHttp_HSR_INTERNAL AgentHttp_HttpStreamResetCode = 0 + AgentHttp_HSR_CANCEL AgentHttp_HttpStreamResetCode = 1 + // the worker refused the stream before dispatching it to the application; + // the request was never applied and is safe to retry elsewhere + AgentHttp_HSR_REFUSED AgentHttp_HttpStreamResetCode = 2 + AgentHttp_HSR_PROTOCOL AgentHttp_HttpStreamResetCode = 3 +) + +// Enum value maps for AgentHttp_HttpStreamResetCode. +var ( + AgentHttp_HttpStreamResetCode_name = map[int32]string{ + 0: "HSR_INTERNAL", + 1: "HSR_CANCEL", + 2: "HSR_REFUSED", + 3: "HSR_PROTOCOL", + } + AgentHttp_HttpStreamResetCode_value = map[string]int32{ + "HSR_INTERNAL": 0, + "HSR_CANCEL": 1, + "HSR_REFUSED": 2, + "HSR_PROTOCOL": 3, + } +) + +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 +708,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 +796,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 +831,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 +847,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 +1006,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 +1105,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 +1180,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 +1449,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 +1513,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,226 +1631,1035 @@ func (x *JobTermination) GetJobId() string { return "" } -var File_livekit_agent_proto protoreflect.FileDescriptor - -const file_livekit_agent_proto_rawDesc = "" + - "\n" + - "\x13livekit_agent.proto\x12\alivekit\x1a\x14livekit_models.proto\x1a\x14logger/options.proto\"\xe2\x04\n" + - "\x03Job\x12\x0e\n" + - "\x02id\x18\x01 \x01(\tR\x02id\x12.\n" + - "\vdispatch_id\x18\t \x01(\tB\r\xbaP\n" + - "dispatchIDR\n" + - "dispatchId\x12$\n" + - "\x04type\x18\x02 \x01(\x0e2\x10.livekit.JobTypeR\x04type\x12!\n" + - "\x04room\x18\x03 \x01(\v2\r.livekit.RoomR\x04room\x12?\n" + - "\vparticipant\x18\x04 \x01(\v2\x18.livekit.ParticipantInfoH\x00R\vparticipant\x88\x01\x01\x12 \n" + - "\tnamespace\x18\x05 \x01(\tB\x02\x18\x01R\tnamespace\x12\x1f\n" + - "\bmetadata\x18\x06 \x01(\tB\x03\xc0P\x01R\bmetadata\x12\x1d\n" + - "\n" + - "agent_name\x18\a \x01(\tR\tagentName\x12'\n" + - "\x05state\x18\b \x01(\v2\x11.livekit.JobStateR\x05state\x12)\n" + - "\x10enable_recording\x18\n" + - " \x01(\bR\x0fenableRecording\x12\x1e\n" + - "\n" + - "deployment\x18\v \x01(\tR\n" + - "deployment\x12A\n" + - "\n" + - "attributes\x18\f \x03(\v2\x1c.livekit.Job.AttributesEntryB\x03\xc0P\x01R\n" + - "attributes\x12)\n" + - "\x10enable_redaction\x18\r \x01(\bR\x0fenableRedaction\x1a=\n" + - "\x0fAttributesEntry\x12\x10\n" + - "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + - "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01B\x0e\n" + - "\f_participant\"\xa9\x02\n" + - "\bJobState\x12*\n" + - "\x06status\x18\x01 \x01(\x0e2\x12.livekit.JobStatusR\x06status\x12\x14\n" + - "\x05error\x18\x02 \x01(\tR\x05error\x12\x1d\n" + - "\n" + - "started_at\x18\x03 \x01(\x03R\tstartedAt\x12\x19\n" + - "\bended_at\x18\x04 \x01(\x03R\aendedAt\x12\x1d\n" + - "\n" + - "updated_at\x18\x05 \x01(\x03R\tupdatedAt\x121\n" + - "\x14participant_identity\x18\x06 \x01(\tR\x13participantIdentity\x12(\n" + - "\tworker_id\x18\a \x01(\tB\v\xbaP\bworkerIDR\bworkerId\x12%\n" + - "\bagent_id\x18\b \x01(\tB\n" + - "\xbaP\aagentIDR\aagentId\"\xc8\x03\n" + - "\rWorkerMessage\x12<\n" + - "\bregister\x18\x01 \x01(\v2\x1e.livekit.RegisterWorkerRequestH\x00R\bregister\x12C\n" + - "\favailability\x18\x02 \x01(\v2\x1d.livekit.AvailabilityResponseH\x00R\favailability\x12B\n" + - "\rupdate_worker\x18\x03 \x01(\v2\x1b.livekit.UpdateWorkerStatusH\x00R\fupdateWorker\x129\n" + - "\n" + - "update_job\x18\x04 \x01(\v2\x18.livekit.UpdateJobStatusH\x00R\tupdateJob\x12)\n" + - "\x04ping\x18\x05 \x01(\v2\x13.livekit.WorkerPingH\x00R\x04ping\x12@\n" + - "\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" + - "\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" + - "\n" + - "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" + - "\amessage\"\x99\x01\n" + - "\x12SimulateJobRequest\x12$\n" + - "\x04type\x18\x01 \x01(\x0e2\x10.livekit.JobTypeR\x04type\x12!\n" + - "\x04room\x18\x02 \x01(\v2\r.livekit.RoomR\x04room\x12:\n" + - "\vparticipant\x18\x03 \x01(\v2\x18.livekit.ParticipantInfoR\vparticipant\"*\n" + - "\n" + - "WorkerPing\x12\x1c\n" + - "\ttimestamp\x18\x01 \x01(\x03R\ttimestamp\"Q\n" + - "\n" + - "WorkerPong\x12%\n" + - "\x0elast_timestamp\x18\x01 \x01(\x03R\rlastTimestamp\x12\x1c\n" + - "\ttimestamp\x18\x02 \x01(\x03R\ttimestamp\"\xbd\x02\n" + - "\x15RegisterWorkerRequest\x12$\n" + - "\x04type\x18\x01 \x01(\x0e2\x10.livekit.JobTypeR\x04type\x12\x1d\n" + - "\n" + - "agent_name\x18\b \x01(\tR\tagentName\x12\x18\n" + - "\aversion\x18\x03 \x01(\tR\aversion\x12#\n" + - "\rping_interval\x18\x05 \x01(\rR\fpingInterval\x12!\n" + - "\tnamespace\x18\x06 \x01(\tH\x00R\tnamespace\x88\x01\x01\x12O\n" + - "\x13allowed_permissions\x18\a \x01(\v2\x1e.livekit.ParticipantPermissionR\x12allowedPermissions\x12\x1e\n" + - "\n" + - "deployment\x18\t \x01(\tR\n" + - "deploymentB\f\n" + - "\n" + - "_namespace\"x\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" + - "\x11MigrateJobRequest\x12\x17\n" + - "\ajob_ids\x18\x02 \x03(\tR\x06jobIds\"Q\n" + - "\x13AvailabilityRequest\x12\x1e\n" + - "\x03job\x18\x01 \x01(\v2\f.livekit.JobR\x03job\x12\x1a\n" + - "\bresuming\x18\x02 \x01(\bR\bresuming\"\xe8\x03\n" + - "\x14AvailabilityResponse\x12\x1f\n" + - "\x06job_id\x18\x01 \x01(\tB\b\xbaP\x05jobIDR\x05jobId\x12\x1c\n" + - "\tavailable\x18\x02 \x01(\bR\tavailable\x12'\n" + - "\x0fsupports_resume\x18\x03 \x01(\bR\x0esupportsResume\x12\x1c\n" + - "\tterminate\x18\b \x01(\bR\tterminate\x12)\n" + - "\x10participant_name\x18\x04 \x01(\tR\x0fparticipantName\x121\n" + - "\x14participant_identity\x18\x05 \x01(\tR\x13participantIdentity\x121\n" + - "\x14participant_metadata\x18\x06 \x01(\tR\x13participantMetadata\x12o\n" + - "\x16participant_attributes\x18\a \x03(\v28.livekit.AvailabilityResponse.ParticipantAttributesEntryR\x15participantAttributes\x1aH\n" + - "\x1aParticipantAttributesEntry\x12\x10\n" + - "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + - "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"t\n" + - "\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" + - "\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" + - "\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" + - "\aJobType\x12\v\n" + - "\aJT_ROOM\x10\x00\x12\x10\n" + - "\fJT_PUBLISHER\x10\x01\x12\x12\n" + - "\x0eJT_PARTICIPANT\x10\x02*-\n" + - "\fWorkerStatus\x12\x10\n" + - "\fWS_AVAILABLE\x10\x00\x12\v\n" + - "\aWS_FULL\x10\x01*J\n" + - "\tJobStatus\x12\x0e\n" + - "\n" + - "JS_PENDING\x10\x00\x12\x0e\n" + - "\n" + - "JS_RUNNING\x10\x01\x12\x0e\n" + - "\n" + - "JS_SUCCESS\x10\x02\x12\r\n" + - "\tJS_FAILED\x10\x03BFZ#github.com/livekit/protocol/livekit\xaa\x02\rLiveKit.Proto\xea\x02\x0eLiveKit::Protob\x06proto3" +// ----- agent HTTP endpoints data plane ----- +// +// AgentHttp namespaces the data plane: workers expose HTTP endpoints served at +// /agents/{deployment}/{path} without binding any local listener. The worker +// dials a fixed pool of websocket wires; the server opens multiplexed streams +// over them and each stream carries one opaque HTTP/1.1 exchange, bridged into +// the worker's application in-process. Wires speak AgentHttp.Frame exclusively +// (one frame per websocket binary message); the control connection (the +// existing /agent registration socket) carries the manifest, load and drain +// state, never frames. Frame envelope adapted from livekit/protocol#1725. +type AgentHttp struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} -var ( - file_livekit_agent_proto_rawDescOnce sync.Once - file_livekit_agent_proto_rawDescData []byte -) +func (x *AgentHttp) Reset() { + *x = AgentHttp{} + mi := &file_livekit_agent_proto_msgTypes[16] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} -func file_livekit_agent_proto_rawDescGZIP() []byte { - file_livekit_agent_proto_rawDescOnce.Do(func() { - file_livekit_agent_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_livekit_agent_proto_rawDesc), len(file_livekit_agent_proto_rawDesc))) - }) - return file_livekit_agent_proto_rawDescData +func (x *AgentHttp) String() string { + return protoimpl.X.MessageStringOf(x) } -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_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 +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) } -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 - 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 + +// Deprecated: Use AgentHttp.ProtoReflect.Descriptor instead. +func (*AgentHttp) Descriptor() ([]byte, []int) { + return file_livekit_agent_proto_rawDescGZIP(), []int{16} } -func init() { file_livekit_agent_proto_init() } -func file_livekit_agent_proto_init() { +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); empty for AEK_WEBSOCKET + 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. Wire-level parameters (credit windows, frame +// size, stream caps) are set per wire by the node adopting it, in +// AttachDataConnectionResponse. +type AgentHttp_AgentEndpointSettings struct { + state protoimpl.MessageState `protogen:"open.v1"` + // negotiated data-plane protocol version + Protocol uint32 `protobuf:"varint,1,opt,name=protocol,proto3" json:"protocol,omitempty"` + // one token per registration, bound to (worker_id, instance_id); reused for + // all wires and across redials, expires with the epoch + AttachToken string `protobuf:"bytes,2,opt,name=attach_token,json=attachToken,proto3" json:"attach_token,omitempty"` + // fixed wire pool size the worker must dial + DataConnectionCount uint32 `protobuf:"varint,3,opt,name=data_connection_count,json=dataConnectionCount,proto3" json:"data_connection_count,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 +} + +func (x *AgentHttp_AgentEndpointSettings) GetAttachToken() string { + if x != nil { + return x.AttachToken + } + return "" +} + +func (x *AgentHttp_AgentEndpointSettings) GetDataConnectionCount() uint32 { + if x != nil { + return x.DataConnectionCount + } + return 0 +} + +// first frame on a wire (stream_id 0), sent by the worker after the upgrade. +// Wires may land on any node behind a load balancer, so they self-identify; +// the adopting node validates against the registration wherever it lives. +type AgentHttp_AttachDataConnection struct { + state protoimpl.MessageState `protogen:"open.v1"` + WorkerId string `protobuf:"bytes,1,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty"` + InstanceId string `protobuf:"bytes,2,opt,name=instance_id,json=instanceId,proto3" json:"instance_id,omitempty"` + AttachToken string `protobuf:"bytes,3,opt,name=attach_token,json=attachToken,proto3" json:"attach_token,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AgentHttp_AttachDataConnection) Reset() { + *x = AgentHttp_AttachDataConnection{} + mi := &file_livekit_agent_proto_msgTypes[21] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AgentHttp_AttachDataConnection) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AgentHttp_AttachDataConnection) ProtoMessage() {} + +func (x *AgentHttp_AttachDataConnection) 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_AttachDataConnection.ProtoReflect.Descriptor instead. +func (*AgentHttp_AttachDataConnection) Descriptor() ([]byte, []int) { + return file_livekit_agent_proto_rawDescGZIP(), []int{16, 2} +} + +func (x *AgentHttp_AttachDataConnection) GetWorkerId() string { + if x != nil { + return x.WorkerId + } + return "" +} + +func (x *AgentHttp_AttachDataConnection) GetInstanceId() string { + if x != nil { + return x.InstanceId + } + return "" +} + +func (x *AgentHttp_AttachDataConnection) GetAttachToken() string { + if x != nil { + return x.AttachToken + } + return "" +} + +// the adopting node's answer (stream_id 0); its wire parameters govern this +// wire only +type AgentHttp_AttachDataConnectionResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + // empty means attached; otherwise the reason the attach was rejected and the + // wire will be closed + Error string `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"` + // per-stream credit window in bytes + CreditWindow uint32 `protobuf:"varint,2,opt,name=credit_window,json=creditWindow,proto3" json:"credit_window,omitempty"` + // shared connection-level window in bytes, refilled by credit on stream 0 + ConnectionWindow uint32 `protobuf:"varint,3,opt,name=connection_window,json=connectionWindow,proto3" json:"connection_window,omitempty"` + // max Frame.data payload bytes per frame + MaxFrameSize uint32 `protobuf:"varint,4,opt,name=max_frame_size,json=maxFrameSize,proto3" json:"max_frame_size,omitempty"` + // hard cap on open streams on this wire (active + parked) + MaxStreamsPerConn uint32 `protobuf:"varint,5,opt,name=max_streams_per_conn,json=maxStreamsPerConn,proto3" json:"max_streams_per_conn,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AgentHttp_AttachDataConnectionResponse) Reset() { + *x = AgentHttp_AttachDataConnectionResponse{} + mi := &file_livekit_agent_proto_msgTypes[22] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AgentHttp_AttachDataConnectionResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AgentHttp_AttachDataConnectionResponse) ProtoMessage() {} + +func (x *AgentHttp_AttachDataConnectionResponse) 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_AttachDataConnectionResponse.ProtoReflect.Descriptor instead. +func (*AgentHttp_AttachDataConnectionResponse) Descriptor() ([]byte, []int) { + return file_livekit_agent_proto_rawDescGZIP(), []int{16, 3} +} + +func (x *AgentHttp_AttachDataConnectionResponse) GetError() string { + if x != nil { + return x.Error + } + return "" +} + +func (x *AgentHttp_AttachDataConnectionResponse) GetCreditWindow() uint32 { + if x != nil { + return x.CreditWindow + } + return 0 +} + +func (x *AgentHttp_AttachDataConnectionResponse) GetConnectionWindow() uint32 { + if x != nil { + return x.ConnectionWindow + } + return 0 +} + +func (x *AgentHttp_AttachDataConnectionResponse) GetMaxFrameSize() uint32 { + if x != nil { + return x.MaxFrameSize + } + return 0 +} + +func (x *AgentHttp_AttachDataConnectionResponse) GetMaxStreamsPerConn() uint32 { + if x != nil { + return x.MaxStreamsPerConn + } + return 0 +} + +// opens a stream carrying what the bytes themselves do not say +type AgentHttp_HttpStreamOpen struct { + state protoimpl.MessageState `protogen:"open.v1"` + RequestId string `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` + // remote address of the end client, informational + ClientAddr string `protobuf:"bytes,2,opt,name=client_addr,json=clientAddr,proto3" json:"client_addr,omitempty"` + // reserved for forward-compatible request metadata + 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 *AgentHttp_HttpStreamOpen) Reset() { + *x = AgentHttp_HttpStreamOpen{} + mi := &file_livekit_agent_proto_msgTypes[23] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AgentHttp_HttpStreamOpen) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AgentHttp_HttpStreamOpen) ProtoMessage() {} + +func (x *AgentHttp_HttpStreamOpen) ProtoReflect() protoreflect.Message { + mi := &file_livekit_agent_proto_msgTypes[23] + 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_HttpStreamOpen.ProtoReflect.Descriptor instead. +func (*AgentHttp_HttpStreamOpen) Descriptor() ([]byte, []int) { + return file_livekit_agent_proto_rawDescGZIP(), []int{16, 4} +} + +func (x *AgentHttp_HttpStreamOpen) GetRequestId() string { + if x != nil { + return x.RequestId + } + return "" +} + +func (x *AgentHttp_HttpStreamOpen) GetClientAddr() string { + if x != nil { + return x.ClientAddr + } + return "" +} + +func (x *AgentHttp_HttpStreamOpen) GetAttributes() map[string]string { + if x != nil { + return x.Attributes + } + return nil +} + +// half-close: the sender is done writing but keeps reading +type AgentHttp_HttpStreamEof struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AgentHttp_HttpStreamEof) Reset() { + *x = AgentHttp_HttpStreamEof{} + mi := &file_livekit_agent_proto_msgTypes[24] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AgentHttp_HttpStreamEof) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AgentHttp_HttpStreamEof) ProtoMessage() {} + +func (x *AgentHttp_HttpStreamEof) ProtoReflect() protoreflect.Message { + mi := &file_livekit_agent_proto_msgTypes[24] + 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_HttpStreamEof.ProtoReflect.Descriptor instead. +func (*AgentHttp_HttpStreamEof) Descriptor() ([]byte, []int) { + return file_livekit_agent_proto_rawDescGZIP(), []int{16, 5} +} + +type AgentHttp_HttpStreamReset struct { + state protoimpl.MessageState `protogen:"open.v1"` + Code AgentHttp_HttpStreamResetCode `protobuf:"varint,1,opt,name=code,proto3,enum=livekit.AgentHttp_HttpStreamResetCode" json:"code,omitempty"` + Error string `protobuf:"bytes,2,opt,name=error,proto3" json:"error,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AgentHttp_HttpStreamReset) Reset() { + *x = AgentHttp_HttpStreamReset{} + mi := &file_livekit_agent_proto_msgTypes[25] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AgentHttp_HttpStreamReset) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AgentHttp_HttpStreamReset) ProtoMessage() {} + +func (x *AgentHttp_HttpStreamReset) ProtoReflect() protoreflect.Message { + mi := &file_livekit_agent_proto_msgTypes[25] + 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_HttpStreamReset.ProtoReflect.Descriptor instead. +func (*AgentHttp_HttpStreamReset) Descriptor() ([]byte, []int) { + return file_livekit_agent_proto_rawDescGZIP(), []int{16, 6} +} + +func (x *AgentHttp_HttpStreamReset) GetCode() AgentHttp_HttpStreamResetCode { + if x != nil { + return x.Code + } + return AgentHttp_HSR_INTERNAL +} + +func (x *AgentHttp_HttpStreamReset) GetError() string { + if x != nil { + return x.Error + } + return "" +} + +// One multiplexed frame, carried as a single websocket binary message. +// Receivers MUST ignore frames for unknown stream ids: a reset can cross in +// flight with data already queued behind it, and ids are never reused within +// a wire. +type AgentHttp_Frame struct { + state protoimpl.MessageState `protogen:"open.v1"` + // the stream this frame addresses. 0 is the wire itself: the attach + // handshake rides it, and credit on it refills the shared connection + // window. The node opens streams and takes odd ids; even ids are reserved + // for worker-opened streams. + 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: + // + // *AgentHttp_Frame_Attach + // *AgentHttp_Frame_AttachResponse + // *AgentHttp_Frame_Open + // *AgentHttp_Frame_Data + // *AgentHttp_Frame_Eof + // *AgentHttp_Frame_Reset_ + // *AgentHttp_Frame_Credit + Message isAgentHttp_Frame_Message `protobuf_oneof:"message"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AgentHttp_Frame) Reset() { + *x = AgentHttp_Frame{} + mi := &file_livekit_agent_proto_msgTypes[26] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AgentHttp_Frame) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AgentHttp_Frame) ProtoMessage() {} + +func (x *AgentHttp_Frame) ProtoReflect() protoreflect.Message { + mi := &file_livekit_agent_proto_msgTypes[26] + 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_Frame.ProtoReflect.Descriptor instead. +func (*AgentHttp_Frame) Descriptor() ([]byte, []int) { + return file_livekit_agent_proto_rawDescGZIP(), []int{16, 7} +} + +func (x *AgentHttp_Frame) GetStreamId() uint32 { + if x != nil { + return x.StreamId + } + return 0 +} + +func (x *AgentHttp_Frame) GetMessage() isAgentHttp_Frame_Message { + if x != nil { + return x.Message + } + return nil +} + +func (x *AgentHttp_Frame) GetAttach() *AgentHttp_AttachDataConnection { + if x != nil { + if x, ok := x.Message.(*AgentHttp_Frame_Attach); ok { + return x.Attach + } + } + return nil +} + +func (x *AgentHttp_Frame) GetAttachResponse() *AgentHttp_AttachDataConnectionResponse { + if x != nil { + if x, ok := x.Message.(*AgentHttp_Frame_AttachResponse); ok { + return x.AttachResponse + } + } + return nil +} + +func (x *AgentHttp_Frame) GetOpen() *AgentHttp_HttpStreamOpen { + if x != nil { + if x, ok := x.Message.(*AgentHttp_Frame_Open); ok { + return x.Open + } + } + return nil +} + +func (x *AgentHttp_Frame) GetData() []byte { + if x != nil { + if x, ok := x.Message.(*AgentHttp_Frame_Data); ok { + return x.Data + } + } + return nil +} + +func (x *AgentHttp_Frame) GetEof() *AgentHttp_HttpStreamEof { + if x != nil { + if x, ok := x.Message.(*AgentHttp_Frame_Eof); ok { + return x.Eof + } + } + return nil +} + +func (x *AgentHttp_Frame) GetReset_() *AgentHttp_HttpStreamReset { + if x != nil { + if x, ok := x.Message.(*AgentHttp_Frame_Reset_); ok { + return x.Reset_ + } + } + return nil +} + +func (x *AgentHttp_Frame) GetCredit() uint32 { + if x != nil { + if x, ok := x.Message.(*AgentHttp_Frame_Credit); ok { + return x.Credit + } + } + return 0 +} + +type isAgentHttp_Frame_Message interface { + isAgentHttp_Frame_Message() +} + +type AgentHttp_Frame_Attach struct { + Attach *AgentHttp_AttachDataConnection `protobuf:"bytes,2,opt,name=attach,proto3,oneof"` +} + +type AgentHttp_Frame_AttachResponse struct { + AttachResponse *AgentHttp_AttachDataConnectionResponse `protobuf:"bytes,3,opt,name=attach_response,json=attachResponse,proto3,oneof"` +} + +type AgentHttp_Frame_Open struct { + Open *AgentHttp_HttpStreamOpen `protobuf:"bytes,4,opt,name=open,proto3,oneof"` +} + +type AgentHttp_Frame_Data struct { + // opaque HTTP/1.1 bytes: bare so the hot path costs one parse and no + // nested message + Data []byte `protobuf:"bytes,5,opt,name=data,proto3,oneof"` +} + +type AgentHttp_Frame_Eof struct { + Eof *AgentHttp_HttpStreamEof `protobuf:"bytes,6,opt,name=eof,proto3,oneof"` +} + +type AgentHttp_Frame_Reset_ struct { + Reset_ *AgentHttp_HttpStreamReset `protobuf:"bytes,7,opt,name=reset,proto3,oneof"` +} + +type AgentHttp_Frame_Credit struct { + // the sender may put this many more bytes in flight (stream 0: on the + // whole wire; otherwise: on this stream). Replenished on consumption, + // not receipt. + Credit uint32 `protobuf:"varint,8,opt,name=credit,proto3,oneof"` +} + +func (*AgentHttp_Frame_Attach) isAgentHttp_Frame_Message() {} + +func (*AgentHttp_Frame_AttachResponse) isAgentHttp_Frame_Message() {} + +func (*AgentHttp_Frame_Open) isAgentHttp_Frame_Message() {} + +func (*AgentHttp_Frame_Data) isAgentHttp_Frame_Message() {} + +func (*AgentHttp_Frame_Eof) isAgentHttp_Frame_Message() {} + +func (*AgentHttp_Frame_Reset_) isAgentHttp_Frame_Message() {} + +func (*AgentHttp_Frame_Credit) isAgentHttp_Frame_Message() {} + +// the server is draining: the worker should re-register elsewhere; existing +// streams 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[27] + 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[27] + 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, 8} +} + +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 = "" + + "\n" + + "\x13livekit_agent.proto\x12\alivekit\x1a\x14livekit_models.proto\x1a\x14logger/options.proto\"\xe2\x04\n" + + "\x03Job\x12\x0e\n" + + "\x02id\x18\x01 \x01(\tR\x02id\x12.\n" + + "\vdispatch_id\x18\t \x01(\tB\r\xbaP\n" + + "dispatchIDR\n" + + "dispatchId\x12$\n" + + "\x04type\x18\x02 \x01(\x0e2\x10.livekit.JobTypeR\x04type\x12!\n" + + "\x04room\x18\x03 \x01(\v2\r.livekit.RoomR\x04room\x12?\n" + + "\vparticipant\x18\x04 \x01(\v2\x18.livekit.ParticipantInfoH\x00R\vparticipant\x88\x01\x01\x12 \n" + + "\tnamespace\x18\x05 \x01(\tB\x02\x18\x01R\tnamespace\x12\x1f\n" + + "\bmetadata\x18\x06 \x01(\tB\x03\xc0P\x01R\bmetadata\x12\x1d\n" + + "\n" + + "agent_name\x18\a \x01(\tR\tagentName\x12'\n" + + "\x05state\x18\b \x01(\v2\x11.livekit.JobStateR\x05state\x12)\n" + + "\x10enable_recording\x18\n" + + " \x01(\bR\x0fenableRecording\x12\x1e\n" + + "\n" + + "deployment\x18\v \x01(\tR\n" + + "deployment\x12A\n" + + "\n" + + "attributes\x18\f \x03(\v2\x1c.livekit.Job.AttributesEntryB\x03\xc0P\x01R\n" + + "attributes\x12)\n" + + "\x10enable_redaction\x18\r \x01(\bR\x0fenableRedaction\x1a=\n" + + "\x0fAttributesEntry\x12\x10\n" + + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01B\x0e\n" + + "\f_participant\"\xa9\x02\n" + + "\bJobState\x12*\n" + + "\x06status\x18\x01 \x01(\x0e2\x12.livekit.JobStatusR\x06status\x12\x14\n" + + "\x05error\x18\x02 \x01(\tR\x05error\x12\x1d\n" + + "\n" + + "started_at\x18\x03 \x01(\x03R\tstartedAt\x12\x19\n" + + "\bended_at\x18\x04 \x01(\x03R\aendedAt\x12\x1d\n" + + "\n" + + "updated_at\x18\x05 \x01(\x03R\tupdatedAt\x121\n" + + "\x14participant_identity\x18\x06 \x01(\tR\x13participantIdentity\x12(\n" + + "\tworker_id\x18\a \x01(\tB\v\xbaP\bworkerIDR\bworkerId\x12%\n" + + "\bagent_id\x18\b \x01(\tB\n" + + "\xbaP\aagentIDR\aagentId\"\xc8\x03\n" + + "\rWorkerMessage\x12<\n" + + "\bregister\x18\x01 \x01(\v2\x1e.livekit.RegisterWorkerRequestH\x00R\bregister\x12C\n" + + "\favailability\x18\x02 \x01(\v2\x1d.livekit.AvailabilityResponseH\x00R\favailability\x12B\n" + + "\rupdate_worker\x18\x03 \x01(\v2\x1b.livekit.UpdateWorkerStatusH\x00R\fupdateWorker\x129\n" + + "\n" + + "update_job\x18\x04 \x01(\v2\x18.livekit.UpdateJobStatusH\x00R\tupdateJob\x12)\n" + + "\x04ping\x18\x05 \x01(\v2\x13.livekit.WorkerPingH\x00R\x04ping\x12@\n" + + "\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\"\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" + + "\n" + + "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\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" + + "\x04room\x18\x02 \x01(\v2\r.livekit.RoomR\x04room\x12:\n" + + "\vparticipant\x18\x03 \x01(\v2\x18.livekit.ParticipantInfoR\vparticipant\"*\n" + + "\n" + + "WorkerPing\x12\x1c\n" + + "\ttimestamp\x18\x01 \x01(\x03R\ttimestamp\"Q\n" + + "\n" + + "WorkerPong\x12%\n" + + "\x0elast_timestamp\x18\x01 \x01(\x03R\rlastTimestamp\x12\x1c\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" + + "agent_name\x18\b \x01(\tR\tagentName\x12\x18\n" + + "\aversion\x18\x03 \x01(\tR\aversion\x12#\n" + + "\rping_interval\x18\x05 \x01(\rR\fpingInterval\x12!\n" + + "\tnamespace\x18\x06 \x01(\tH\x00R\tnamespace\x88\x01\x01\x12O\n" + + "\x13allowed_permissions\x18\a \x01(\v2\x1e.livekit.ParticipantPermissionR\x12allowedPermissions\x12\x1e\n" + + "\n" + + "deployment\x18\t \x01(\tR\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" + + "_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\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" + + "\x03job\x18\x01 \x01(\v2\f.livekit.JobR\x03job\x12\x1a\n" + + "\bresuming\x18\x02 \x01(\bR\bresuming\"\xe8\x03\n" + + "\x14AvailabilityResponse\x12\x1f\n" + + "\x06job_id\x18\x01 \x01(\tB\b\xbaP\x05jobIDR\x05jobId\x12\x1c\n" + + "\tavailable\x18\x02 \x01(\bR\tavailable\x12'\n" + + "\x0fsupports_resume\x18\x03 \x01(\bR\x0esupportsResume\x12\x1c\n" + + "\tterminate\x18\b \x01(\bR\tterminate\x12)\n" + + "\x10participant_name\x18\x04 \x01(\tR\x0fparticipantName\x121\n" + + "\x14participant_identity\x18\x05 \x01(\tR\x13participantIdentity\x121\n" + + "\x14participant_metadata\x18\x06 \x01(\tR\x13participantMetadata\x12o\n" + + "\x16participant_attributes\x18\a \x03(\v28.livekit.AvailabilityResponse.ParticipantAttributesEntryR\x15participantAttributes\x1aH\n" + + "\x1aParticipantAttributesEntry\x12\x10\n" + + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"t\n" + + "\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\"\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\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\"\x84\r\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\x8a\x01\n" + + "\x15AgentEndpointSettings\x12\x1a\n" + + "\bprotocol\x18\x01 \x01(\rR\bprotocol\x12!\n" + + "\fattach_token\x18\x02 \x01(\tR\vattachToken\x122\n" + + "\x15data_connection_count\x18\x03 \x01(\rR\x13dataConnectionCount\x1a\x93\x01\n" + + "\x14AttachDataConnection\x12(\n" + + "\tworker_id\x18\x01 \x01(\tB\v\xbaP\bworkerIDR\bworkerId\x12.\n" + + "\vinstance_id\x18\x02 \x01(\tB\r\xbaP\n" + + "instanceIDR\n" + + "instanceId\x12!\n" + + "\fattach_token\x18\x03 \x01(\tR\vattachToken\x1a\xdd\x01\n" + + "\x1cAttachDataConnectionResponse\x12\x14\n" + + "\x05error\x18\x01 \x01(\tR\x05error\x12#\n" + + "\rcredit_window\x18\x02 \x01(\rR\fcreditWindow\x12+\n" + + "\x11connection_window\x18\x03 \x01(\rR\x10connectionWindow\x12$\n" + + "\x0emax_frame_size\x18\x04 \x01(\rR\fmaxFrameSize\x12/\n" + + "\x14max_streams_per_conn\x18\x05 \x01(\rR\x11maxStreamsPerConn\x1a\xf0\x01\n" + + "\x0eHttpStreamOpen\x12+\n" + + "\n" + + "request_id\x18\x01 \x01(\tB\f\xbaP\trequestIDR\trequestId\x12\x1f\n" + + "\vclient_addr\x18\x02 \x01(\tR\n" + + "clientAddr\x12Q\n" + + "\n" + + "attributes\x18\x03 \x03(\v21.livekit.AgentHttp.HttpStreamOpen.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\x1a\x0f\n" + + "\rHttpStreamEof\x1ac\n" + + "\x0fHttpStreamReset\x12:\n" + + "\x04code\x18\x01 \x01(\x0e2&.livekit.AgentHttp.HttpStreamResetCodeR\x04code\x12\x14\n" + + "\x05error\x18\x02 \x01(\tR\x05error\x1a\xb6\x03\n" + + "\x05Frame\x12(\n" + + "\tstream_id\x18\x01 \x01(\rB\v\xbaP\bstreamIDR\bstreamId\x12A\n" + + "\x06attach\x18\x02 \x01(\v2'.livekit.AgentHttp.AttachDataConnectionH\x00R\x06attach\x12Z\n" + + "\x0fattach_response\x18\x03 \x01(\v2/.livekit.AgentHttp.AttachDataConnectionResponseH\x00R\x0eattachResponse\x127\n" + + "\x04open\x18\x04 \x01(\v2!.livekit.AgentHttp.HttpStreamOpenH\x00R\x04open\x12\x14\n" + + "\x04data\x18\x05 \x01(\fH\x00R\x04data\x124\n" + + "\x03eof\x18\x06 \x01(\v2 .livekit.AgentHttp.HttpStreamEofH\x00R\x03eof\x12:\n" + + "\x05reset\x18\a \x01(\v2\".livekit.AgentHttp.HttpStreamResetH\x00R\x05reset\x12\x18\n" + + "\x06credit\x18\b \x01(\rH\x00R\x06creditB\t\n" + + "\amessage\x1a \n" + + "\x06GoAway\x12\x16\n" + + "\x06reason\x18\x01 \x01(\tR\x06reason\"B\n" + + "\x11AgentEndpointKind\x12\f\n" + + "\bAEK_HTTP\x10\x00\x12\x11\n" + + "\rAEK_WEBSOCKET\x10\x01\x12\f\n" + + "\bAEK_TEXT\x10\x02\"Z\n" + + "\x13HttpStreamResetCode\x12\x10\n" + + "\fHSR_INTERNAL\x10\x00\x12\x0e\n" + + "\n" + + "HSR_CANCEL\x10\x01\x12\x0f\n" + + "\vHSR_REFUSED\x10\x02\x12\x10\n" + + "\fHSR_PROTOCOL\x10\x03*<\n" + + "\aJobType\x12\v\n" + + "\aJT_ROOM\x10\x00\x12\x10\n" + + "\fJT_PUBLISHER\x10\x01\x12\x12\n" + + "\x0eJT_PARTICIPANT\x10\x02*-\n" + + "\fWorkerStatus\x12\x10\n" + + "\fWS_AVAILABLE\x10\x00\x12\v\n" + + "\aWS_FULL\x10\x01*J\n" + + "\tJobStatus\x12\x0e\n" + + "\n" + + "JS_PENDING\x10\x00\x12\x0e\n" + + "\n" + + "JS_RUNNING\x10\x01\x12\x0e\n" + + "\n" + + "JS_SUCCESS\x10\x02\x12\r\n" + + "\tJS_FAILED\x10\x03BFZ#github.com/livekit/protocol/livekit\xaa\x02\rLiveKit.Proto\xea\x02\x0eLiveKit::Protob\x06proto3" + +var ( + file_livekit_agent_proto_rawDescOnce sync.Once + file_livekit_agent_proto_rawDescData []byte +) + +func file_livekit_agent_proto_rawDescGZIP() []byte { + file_livekit_agent_proto_rawDescOnce.Do(func() { + file_livekit_agent_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_livekit_agent_proto_rawDesc), len(file_livekit_agent_proto_rawDesc))) + }) + return file_livekit_agent_proto_rawDescData +} + +var file_livekit_agent_proto_enumTypes = make([]protoimpl.EnumInfo, 5) +var file_livekit_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 29) +var file_livekit_agent_proto_goTypes = []any{ + (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_AttachDataConnection)(nil), // 26: livekit.AgentHttp.AttachDataConnection + (*AgentHttp_AttachDataConnectionResponse)(nil), // 27: livekit.AgentHttp.AttachDataConnectionResponse + (*AgentHttp_HttpStreamOpen)(nil), // 28: livekit.AgentHttp.HttpStreamOpen + (*AgentHttp_HttpStreamEof)(nil), // 29: livekit.AgentHttp.HttpStreamEof + (*AgentHttp_HttpStreamReset)(nil), // 30: livekit.AgentHttp.HttpStreamReset + (*AgentHttp_Frame)(nil), // 31: livekit.AgentHttp.Frame + (*AgentHttp_GoAway)(nil), // 32: livekit.AgentHttp.GoAway + nil, // 33: livekit.AgentHttp.HttpStreamOpen.AttributesEntry + (*Room)(nil), // 34: livekit.Room + (*ParticipantInfo)(nil), // 35: livekit.ParticipantInfo + (*ParticipantPermission)(nil), // 36: livekit.ParticipantPermission + (*ServerInfo)(nil), // 37: livekit.ServerInfo +} +var file_livekit_agent_proto_depIdxs = []int32{ + 0, // 0: livekit.Job.type:type_name -> livekit.JobType + 34, // 1: livekit.Job.room:type_name -> livekit.Room + 35, // 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 + 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 + 32, // 18: livekit.ServerMessage.go_away:type_name -> livekit.AgentHttp.GoAway + 0, // 19: livekit.SimulateJobRequest.type:type_name -> livekit.JobType + 34, // 20: livekit.SimulateJobRequest.room:type_name -> livekit.Room + 35, // 21: livekit.SimulateJobRequest.participant:type_name -> livekit.ParticipantInfo + 0, // 22: livekit.RegisterWorkerRequest.type:type_name -> livekit.JobType + 36, // 23: livekit.RegisterWorkerRequest.allowed_permissions:type_name -> livekit.ParticipantPermission + 24, // 24: livekit.RegisterWorkerRequest.endpoints:type_name -> livekit.AgentHttp.AgentEndpoint + 37, // 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 + 33, // 33: livekit.AgentHttp.HttpStreamOpen.attributes:type_name -> livekit.AgentHttp.HttpStreamOpen.AttributesEntry + 4, // 34: livekit.AgentHttp.HttpStreamReset.code:type_name -> livekit.AgentHttp.HttpStreamResetCode + 26, // 35: livekit.AgentHttp.Frame.attach:type_name -> livekit.AgentHttp.AttachDataConnection + 27, // 36: livekit.AgentHttp.Frame.attach_response:type_name -> livekit.AgentHttp.AttachDataConnectionResponse + 28, // 37: livekit.AgentHttp.Frame.open:type_name -> livekit.AgentHttp.HttpStreamOpen + 29, // 38: livekit.AgentHttp.Frame.eof:type_name -> livekit.AgentHttp.HttpStreamEof + 30, // 39: livekit.AgentHttp.Frame.reset:type_name -> livekit.AgentHttp.HttpStreamReset + 40, // [40:40] is the sub-list for method output_type + 40, // [40:40] is the sub-list for method input_type + 40, // [40:40] is the sub-list for extension type_name + 40, // [40:40] is the sub-list for extension extendee + 0, // [0:40] is the sub-list for field type_name +} + +func init() { file_livekit_agent_proto_init() } +func file_livekit_agent_proto_init() { if File_livekit_agent_proto != nil { return } @@ -1693,17 +2680,27 @@ 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{} file_livekit_agent_proto_msgTypes[14].OneofWrappers = []any{} + file_livekit_agent_proto_msgTypes[26].OneofWrappers = []any{ + (*AgentHttp_Frame_Attach)(nil), + (*AgentHttp_Frame_AttachResponse)(nil), + (*AgentHttp_Frame_Open)(nil), + (*AgentHttp_Frame_Data)(nil), + (*AgentHttp_Frame_Eof)(nil), + (*AgentHttp_Frame_Reset_)(nil), + (*AgentHttp_Frame_Credit)(nil), + } type x struct{} out := protoimpl.TypeBuilder{ 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: 29, NumExtensions: 0, NumServices: 0, }, diff --git a/protobufs/livekit_agent.proto b/protobufs/livekit_agent.proto index c18ecf203..f28ca5c55 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,136 @@ 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/{deployment}/{path} without binding any local listener. The worker +// dials a fixed pool of websocket wires; the server opens multiplexed streams +// over them and each stream carries one opaque HTTP/1.1 exchange, bridged into +// the worker's application in-process. Wires speak AgentHttp.Frame exclusively +// (one frame per websocket binary message); the control connection (the +// existing /agent registration socket) carries the manifest, load and drain +// state, never frames. Frame envelope adapted from livekit/protocol#1725. +message AgentHttp { + enum AgentEndpointKind { + AEK_HTTP = 0; + // a client-initiated WebSocket route; the stream carries the upgrade exchange + // and then raw bidirectional bytes + AEK_WEBSOCKET = 1; + // reserved for future text-mode endpoints + AEK_TEXT = 2; + } + + 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); empty for AEK_WEBSOCKET + 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. Wire-level parameters (credit windows, frame + // size, stream caps) are set per wire by the node adopting it, in + // AttachDataConnectionResponse. + message AgentEndpointSettings { + // negotiated data-plane protocol version + uint32 protocol = 1; + // one token per registration, bound to (worker_id, instance_id); reused for + // all wires and across redials, expires with the epoch + string attach_token = 2; + // fixed wire pool size the worker must dial + uint32 data_connection_count = 3; + } + + // first frame on a wire (stream_id 0), sent by the worker after the upgrade. + // Wires may land on any node behind a load balancer, so they self-identify; + // the adopting node validates against the registration wherever it lives. + message AttachDataConnection { + string worker_id = 1 [(logger.name) = "workerID"]; + string instance_id = 2 [(logger.name) = "instanceID"]; + string attach_token = 3; + } + + // the adopting node's answer (stream_id 0); its wire parameters govern this + // wire only + message AttachDataConnectionResponse { + // empty means attached; otherwise the reason the attach was rejected and the + // wire will be closed + string error = 1; + // per-stream credit window in bytes + uint32 credit_window = 2; + // shared connection-level window in bytes, refilled by credit on stream 0 + uint32 connection_window = 3; + // max Frame.data payload bytes per frame + uint32 max_frame_size = 4; + // hard cap on open streams on this wire (active + parked) + uint32 max_streams_per_conn = 5; + } + + enum HttpStreamResetCode { + HSR_INTERNAL = 0; + HSR_CANCEL = 1; + // the worker refused the stream before dispatching it to the application; + // the request was never applied and is safe to retry elsewhere + HSR_REFUSED = 2; + HSR_PROTOCOL = 3; + } + + // opens a stream carrying what the bytes themselves do not say + message HttpStreamOpen { + string request_id = 1 [(logger.name) = "requestID"]; + // remote address of the end client, informational + string client_addr = 2; + // reserved for forward-compatible request metadata + map attributes = 3; + } + + // half-close: the sender is done writing but keeps reading + message HttpStreamEof { + } + + message HttpStreamReset { + HttpStreamResetCode code = 1; + string error = 2; + } + + // One multiplexed frame, carried as a single websocket binary message. + // Receivers MUST ignore frames for unknown stream ids: a reset can cross in + // flight with data already queued behind it, and ids are never reused within + // a wire. + message Frame { + // the stream this frame addresses. 0 is the wire itself: the attach + // handshake rides it, and credit on it refills the shared connection + // window. The node opens streams and takes odd ids; even ids are reserved + // for worker-opened streams. + uint32 stream_id = 1 [(logger.name) = "streamID"]; + + oneof message { + AttachDataConnection attach = 2; + AttachDataConnectionResponse attach_response = 3; + HttpStreamOpen open = 4; + // opaque HTTP/1.1 bytes: bare so the hot path costs one parse and no + // nested message + bytes data = 5; + HttpStreamEof eof = 6; + HttpStreamReset reset = 7; + // the sender may put this many more bytes in flight (stream 0: on the + // whole wire; otherwise: on this stream). Replenished on consumption, + // not receipt. + uint32 credit = 8; + } + } + + // the server is draining: the worker should re-register elsewhere; existing + // streams run to completion + message GoAway { + string reason = 1; + } +}