agent threads: thread messages and grant, agent-db service to gRPC

.changeset/agent-threads.md

@@ -0,0 +1,8 @@
+---
+"github.com/livekit/protocol": minor
+"@livekit/protocol": minor
+---
+
+Agent threads, the durable conversation an agent has with a person across channels (`agent/livekit_agent_thread.proto`): `AgentThread` (the thread, its items as `ChatContext.ChatItem`) and `AgentTask` (background work an agent runs in it); the conversation tables in `livekit/agent/thread_schema_v1.sql`; `auth.AgentThreadGrant` under the `agentThread` claim (list and create by scope, read, write and delete by thread id, or every thread of the project with `all`); the `AT_` (thread) and `AST_` (agent stream) id prefixes in `utils/guid`.
+
+The agent-db and thread services are internal gRPC in cloud-protocol, so `AgentDBService` and its Twirp code are removed; the public API is public-api-server. A database lives in its project's data region: `AgentDB.CreateRequest.region` is removed, `DumpRequest`/`DumpResponse` are now `ExportRequest`/`ExportResponse`, and AgentDB times are `google.protobuf.Timestamp` / `Duration`. A batch answers one result per statement, tagged with its 0-based `statement` index on `Columns`, `ColumnBatch` and `ExecResult`.

auth/accesstoken.go

@@ -114,6 +114,11 @@ func (t *AccessToken) SetAgentEndpointGrant(grant *AgentEndpointGrant) *AccessTo
 	return t
 }
 
+func (t *AccessToken) SetAgentThreadGrant(grant *AgentThreadGrant) *AccessToken {
+	t.grant.AgentThread = grant
+	return t
+}
+
 func (t *AccessToken) SetMetadata(md string) *AccessToken {
 	t.grant.Metadata = md
 	return t

auth/accesstoken_test.go

@@ -245,6 +245,30 @@ func TestAgentEndpointGrantRoundTrip(t *testing.T) {
 	require.Equal(t, grant, decoded.AgentEndpoint)
 }
 
+func TestAgentThreadGrantRoundTrip(t *testing.T) {
+	t.Parallel()
+
+	apiKey, secret := apiKeypair()
+	grant := &AgentThreadGrant{Scope: "user-1", ThreadIDs: []string{"AT_1"}, Read: true, Write: true}
+	raw, err := NewAccessToken(apiKey, secret).
+		SetAgentThreadGrant(grant).
+		SetValidFor(time.Minute).
+		ToJWT()
+	require.NoError(t, err)
+
+	// the claim keys are camelCase
+	claims := decodeClaims(t, raw)
+	require.Contains(t, claims, `"agentThread"`)
+	require.Contains(t, claims, `"scope"`)
+	require.Contains(t, claims, `"threadIds"`)
+
+	v, err := ParseAPIToken(raw)
+	require.NoError(t, err)
+	_, decoded, err := v.Verify(secret)
+	require.NoError(t, err)
+	require.Equal(t, grant, decoded.AgentThread)
+}
+
 func decodeClaims(t *testing.T, raw string) string {
 	t.Helper()
 	parts := strings.Split(raw, ".")

auth/grants.go

@@ -169,6 +169,7 @@ type ClaimGrants struct {
 	Inference     *InferenceGrant     `json:"inference,omitempty"`
 	Observability *ObservabilityGrant `json:"observability,omitempty"`
 	AgentEndpoint *AgentEndpointGrant `json:"agentEndpoint,omitempty"`
+	AgentThread   *AgentThreadGrant   `json:"agentThread,omitempty"`
 	// Room configuration to use if this participant initiates the room
 	RoomConfig *RoomConfiguration `json:"roomConfig,omitempty"`
 	// Cloud-only, config preset to use
@@ -216,6 +217,7 @@ func (c *ClaimGrants) Clone() *ClaimGrants {
 	clone.Inference = c.Inference.Clone()
 	clone.Observability = c.Observability.Clone()
 	clone.AgentEndpoint = c.AgentEndpoint.Clone()
+	clone.AgentThread = c.AgentThread.Clone()
 	clone.Attributes = maps.Clone(c.Attributes)
 	clone.RoomConfig = c.RoomConfig.Clone()
 	if len(c.KindDetails) > 0 {
@@ -239,6 +241,7 @@ func (c *ClaimGrants) MarshalLogObject(e zapcore.ObjectEncoder) error {
 	e.AddObject("Inference", c.Inference)
 	e.AddObject("Observability", c.Observability)
 	e.AddObject("AgentEndpoint", c.AgentEndpoint)
+	e.AddObject("AgentThread", c.AgentThread)
 	e.AddObject("RoomConfig", logger.Proto((*livekit.RoomConfiguration)(c.RoomConfig)))
 	e.AddString("RoomPreset", c.RoomPreset)
 	return nil
@@ -712,6 +715,93 @@ func (s *AgentEndpointGrant) MarshalLogObject(e zapcore.ObjectEncoder) error {
 
 // ------------------------------------------------------------------
 
+// AgentThreadGrant gives a token access to threads, across the project's
+// agents: list and create within Scope, and read, write and delete the threads
+// in ThreadIDs. All extends it to every thread and scope of the project; the
+// action flags still apply. It grants thread data only; which endpoints a token
+// may call is agentEndpoint's.
+type AgentThreadGrant struct {
+	All       bool     `json:"all,omitempty"`
+	Scope     string   `json:"scope,omitempty"`
+	ThreadIDs []string `json:"threadIds,omitempty"`
+	List      bool     `json:"list,omitempty"`
+	Create    bool     `json:"create,omitempty"`
+	Write     bool     `json:"write,omitempty"`
+	Read      bool     `json:"read,omitempty"`
+	Delete    bool     `json:"delete,omitempty"`
+}
+
+// CanCreate reports whether the holder may create threads in scope. A scoped
+// grant creates only in its own scope.
+func (s *AgentThreadGrant) CanCreate(scope string) bool {
+	return s != nil && s.Create && s.inScope(scope)
+}
+
+// CanList reports whether the holder may list the threads of scope.
+func (s *AgentThreadGrant) CanList(scope string) bool {
+	return s != nil && s.List && s.inScope(scope)
+}
+
+// CanUpdate reports whether the holder may change any thread's attributes and
+// idle TTL, which only a project-wide grant may.
+func (s *AgentThreadGrant) CanUpdate() bool {
+	return s != nil && s.All && s.Write
+}
+
+// CanRead reports whether the holder may read threadID.
+func (s *AgentThreadGrant) CanRead(threadID string) bool {
+	return s != nil && s.Read && s.includes(threadID)
+}
+
+// CanWrite reports whether the holder may send to threadID.
+func (s *AgentThreadGrant) CanWrite(threadID string) bool {
+	return s != nil && s.Write && s.includes(threadID)
+}
+
+// CanDelete reports whether the holder may delete threadID.
+func (s *AgentThreadGrant) CanDelete(threadID string) bool {
+	return s != nil && s.Delete && s.includes(threadID)
+}
+
+func (s *AgentThreadGrant) inScope(scope string) bool {
+	return s.All || (s.Scope != "" && s.Scope == scope)
+}
+
+func (s *AgentThreadGrant) includes(threadID string) bool {
+	return threadID != "" && (s.All || slices.Contains(s.ThreadIDs, threadID))
+}
+
+func (s *AgentThreadGrant) Clone() *AgentThreadGrant {
+	if s == nil {
+		return nil
+	}
+
+	clone := *s
+	if len(s.ThreadIDs) > 0 {
+		clone.ThreadIDs = append([]string{}, s.ThreadIDs...)
+	}
+
+	return &clone
+}
+
+func (s *AgentThreadGrant) MarshalLogObject(e zapcore.ObjectEncoder) error {
+	if s == nil {
+		return nil
+	}
+
+	e.AddBool("All", s.All)
+	e.AddString("Scope", s.Scope)
+	zap.Strings("ThreadIDs", s.ThreadIDs).AddTo(e)
+	e.AddBool("List", s.List)
+	e.AddBool("Create", s.Create)
+	e.AddBool("Write", s.Write)
+	e.AddBool("Read", s.Read)
+	e.AddBool("Delete", s.Delete)
+	return nil
+}
+
+// ------------------------------------------------------------------
+
 func sourceToString(source livekit.TrackSource) string {
 	return strings.ToLower(source.String())
 }

auth/grants_test.go

@@ -36,6 +36,7 @@ func TestGrants(t *testing.T) {
 		require.Same(t, grants.Inference, clone.Inference)
 		require.Same(t, grants.SIP, clone.SIP)
 		require.Same(t, grants.AgentEndpoint, clone.AgentEndpoint)
+		require.Same(t, grants.AgentThread, clone.AgentThread)
 		require.True(t, reflect.DeepEqual(grants, clone))
 		require.True(t, reflect.DeepEqual(grants.Video, clone.Video))
 	})
@@ -65,6 +66,9 @@ func TestGrants(t *testing.T) {
 		// require AgentEndpoint
 		require.Same(t, grants.AgentEndpoint, clone.AgentEndpoint)
 		require.True(t, reflect.DeepEqual(grants.AgentEndpoint, clone.AgentEndpoint))
+		// require AgentThread
+		require.Same(t, grants.AgentThread, clone.AgentThread)
+		require.True(t, reflect.DeepEqual(grants.AgentThread, clone.AgentThread))
 	})
 
 	t.Run("clone with video", func(t *testing.T) {
@@ -524,3 +528,55 @@ func TestAgentEndpointGrantCloneIndependent(t *testing.T) {
 	require.True(t, grants.AgentEndpoint.Call)
 	require.Equal(t, "a", grants.AgentEndpoint.AgentName)
 }
+
+func TestAgentThreadGrantChecks(t *testing.T) {
+	t.Parallel()
+
+	// Scope covers list and create only; read, write and delete need the id.
+	g := &AgentThreadGrant{Scope: "u1", ThreadIDs: []string{"AT_1"}, List: true, Create: true, Read: true}
+	require.True(t, g.CanRead("AT_1"))
+	require.False(t, g.CanRead("AT_2"))
+	require.False(t, g.CanRead(""))
+	require.False(t, g.CanWrite("AT_1"))
+	require.False(t, g.CanDelete("AT_1"))
+	require.True(t, g.CanList("u1"))
+	require.False(t, g.CanList("u2"))
+	require.False(t, g.CanList(""))
+	require.True(t, g.CanCreate("u1"))
+	require.False(t, g.CanCreate("u2"))
+	require.False(t, g.CanUpdate())
+
+	idsOnly := &AgentThreadGrant{ThreadIDs: []string{"AT_4"}, Create: true, Write: true}
+	require.True(t, idsOnly.CanWrite("AT_4"))
+	require.False(t, idsOnly.CanCreate(""))
+
+	// All covers every thread and scope, still gated by the action flags.
+	all := &AgentThreadGrant{All: true, Read: true, List: true}
+	require.True(t, all.CanRead("AT_9"))
+	require.True(t, all.CanList("any"))
+	require.False(t, all.CanRead(""))
+	require.False(t, all.CanDelete("AT_9"))
+	require.False(t, all.CanCreate("any"))
+	require.False(t, all.CanUpdate())
+	require.True(t, (&AgentThreadGrant{All: true, Write: true}).CanUpdate())
+
+	var none *AgentThreadGrant
+	require.False(t, none.CanRead("AT_1"))
+	require.False(t, none.CanCreate("u1"))
+	require.False(t, none.CanUpdate())
+}
+
+func TestAgentThreadGrantCloneIndependent(t *testing.T) {
+	t.Parallel()
+
+	grants := &ClaimGrants{AgentThread: &AgentThreadGrant{Scope: "u1", ThreadIDs: []string{"AT_1"}, Write: true}}
+	clone := grants.Clone()
+	require.NotSame(t, grants.AgentThread, clone.AgentThread)
+	require.True(t, reflect.DeepEqual(grants.AgentThread, clone.AgentThread))
+
+	clone.AgentThread.Write = false
+	clone.AgentThread.ThreadIDs[0] = "AT_2"
+	clone.AgentThread.ThreadIDs = append(clone.AgentThread.ThreadIDs, "AT_3")
+	require.True(t, grants.AgentThread.Write)
+	require.Equal(t, []string{"AT_1"}, grants.AgentThread.ThreadIDs)
+}