agent threads: thread id and grant on jobs and the HTTP preamble

.changeset/agent-threads-runtime.md

@@ -0,0 +1,6 @@
+---
+"github.com/livekit/protocol": minor
+"@livekit/protocol": minor
+---
+
+Agent threads on the agent runtime: `Job.thread_id` and `AgentHttp.StreamPreamble.thread_id` / `thread_grant` (`AgentHttp.AgentThreadGrant`). `auth.AgentThreadGrant.ToProto` converts the claim for the preamble.

auth/grants.go

@@ -779,6 +779,23 @@ func (s *AgentThreadGrant) Clone() *AgentThreadGrant {
 	return &clone
 }
 
+// ToProto is the grant as the node forwards it to the worker in
+// AgentHttp.StreamPreamble.thread_grant; nil stays nil.
+func (s *AgentThreadGrant) ToProto() *livekit.AgentHttp_AgentThreadGrant {
+	if s == nil {
+		return nil
+	}
+	return &livekit.AgentHttp_AgentThreadGrant{
+		Scope:     s.Scope,
+		ThreadIds: append([]string{}, s.ThreadIDs...),
+		List:      s.List,
+		Create:    s.Create,
+		Write:     s.Write,
+		Read:      s.Read,
+		Delete:    s.Delete,
+	}
+}
+
 func (s *AgentThreadGrant) MarshalLogObject(e zapcore.ObjectEncoder) error {
 	if s == nil {
 		return nil

protobufs/livekit_agent.proto

@@ -36,6 +36,8 @@ message Job {
   string deployment = 11;
   map<string, string> attributes = 12 [(logger.sensitivity) = SENSITIVITY_PII];
   bool enable_redaction = 13;
+  // the thread ("AT_...") this job continues; empty when dispatch made none
+  string thread_id = 14 [(logger.name) = "threadID"];
 }
 
 message JobState {
@@ -297,7 +299,26 @@ message AgentHttp {
     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
+    // the thread ("AT_...") the node resolved for this request; empty when
+    // none, and the worker resolves it from the body
+    string thread_id = 8 [(logger.name) = "threadID"];
+    // the verified agentThread claim of the caller's token, which never
+    // reaches the worker; the worker enforces it on the addressed thread
+    AgentThreadGrant thread_grant = 9;
+    // NEXT_ID: 10
+  }
+
+  // auth.AgentThreadGrant on the wire: threads by scope or by id, with
+  // per-action flags
+  message AgentThreadGrant {
+    reserved 1, 9;
+    string scope = 2;
+    repeated string thread_ids = 3 [(logger.name) = "threadIDs"];
+    bool list = 4;
+    bool create = 5;
+    bool write = 6;
+    bool read = 7;
+    bool delete = 8;
   }
 
   // why a stream was reset before any HTTP bytes flowed. This travels as the