package message import ( "context" "strings" "testing" "time" "github.com/stretchr/testify/require" "github.com/tidbcloud/AgentTeam/internal/coredb" ) func TestScopeAPIHistoryReadAndMemberDisplayUseVisibility(t *testing.T) { t.Parallel() ctx := context.Background() db := startMessagingTestPool(t, ctx) repos := NewScopeRepositories(db) now := scopeAPINowForTest() workspaceID := "wsp_scope_api_visibility" ownerUserID := "usr_scope_api_owner" ownerMemberID := "mem_scope_api_owner" strangerUserID := "usr_scope_api_stranger" strangerMemberID := "mem_scope_api_stranger" privateMemberID := "mem_scope_api_private" scopeID := "scp_scope_api_visibility" insertMessagingWorkspace(t, ctx, db, workspaceID, now) insertMessagingUser(t, ctx, db, ownerUserID, "scope-api-owner@example.com", now) insertMessagingHumanMember(t, ctx, db, ownerMemberID, workspaceID, ownerUserID, now) insertMessagingUser(t, ctx, db, strangerUserID, "scope-api-stranger@example.com", now) insertMessagingHumanMember(t, ctx, db, strangerMemberID, workspaceID, strangerUserID, now) insertMessagingAgentMember(t, ctx, db, privateMemberID, workspaceID, now) require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{ ID: scopeID, WorkspaceID: workspaceID, Type: "channel", IdentityKey: "channel:scope-api-visibility", LatestSeq: 3, LatestGlobalSeq: 30, LatestMessageAt: &now, Status: "active", CreatedByMemberID: ownerMemberID, CreatedAt: now, UpdatedAt: now, })) require.NoError(t, repos.Scopes.UpsertChannelProfile(ctx, ScopeChannelProfileRecord{ ScopeID: scopeID, WorkspaceID: workspaceID, Slug: "scope-api", DisplayName: "Scope API", Visibility: "private", CreatedAt: now, UpdatedAt: now, })) for _, membership := range []ScopeMembershipRecord{ {ScopeID: scopeID, MemberID: ownerMemberID, Visibility: "public", CreatedAt: now, UpdatedAt: now}, {ScopeID: scopeID, MemberID: strangerMemberID, Visibility: "public", CreatedAt: now, UpdatedAt: now}, {ScopeID: scopeID, MemberID: privateMemberID, Visibility: "private", OwnerMemberID: ownerMemberID, CreatedAt: now, UpdatedAt: now}, } { require.NoError(t, repos.Memberships.Upsert(ctx, membership)) } for _, record := range []ScopeMessageRecord{ { ID: "msg_scope_api_owner", WorkspaceID: workspaceID, ScopeID: scopeID, Seq: 1, WorkspaceGlobalSeq: 10, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "owner apple", Metadata: []byte(`{"source":"owner"}`), CreatedAt: now.Add(time.Second), UpdatedAt: now.Add(time.Second), }, { ID: "msg_scope_api_hidden", WorkspaceID: workspaceID, ScopeID: scopeID, Seq: 2, WorkspaceGlobalSeq: 20, AuthorMemberID: strangerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "hidden apple", CreatedAt: now.Add(2 * time.Second), UpdatedAt: now.Add(2 * time.Second), }, { ID: "msg_scope_api_private", WorkspaceID: workspaceID, ScopeID: scopeID, Seq: 3, WorkspaceGlobalSeq: 30, AuthorMemberID: privateMemberID, AuthorType: "agent", MessageType: "text", Content: "private apple", CreatedAt: now.Add(3 * time.Second), UpdatedAt: now.Add(3 * time.Second), }, } { require.NoError(t, repos.Messages.Insert(ctx, record)) require.NoError(t, repos.Search.InsertDoc(ctx, ScopeSearchDocRecord{ MessageID: record.ID, WorkspaceID: record.WorkspaceID, ScopeID: record.ScopeID, GlobalSeq: record.WorkspaceGlobalSeq, Content: record.Content, CreatedAt: record.CreatedAt, })) } api := NewScopeAPIService(New(db, nil, 0, Options{})) history, err := api.ListHistory(ctx, ScopeAPIHistoryRequest{ WorkspaceID: workspaceID, ScopeID: scopeID, ViewerMemberID: privateMemberID, ViewerType: ScopeActorPrivateAgent, RequestKind: ScopeVisibilityRequestHistory, Limit: 10, }) require.NoError(t, err) require.Equal(t, scopeID, history.TargetKey) require.Equal(t, int64(3), history.CoversThroughSeq) require.Len(t, history.Messages, 2) require.Equal(t, "msg_scope_api_private", history.Messages[0].ID) require.Equal(t, "msg_scope_api_owner", history.Messages[1].ID) require.Equal(t, "owner", history.Messages[1].Metadata["source"]) allowed, err := api.CanReadMessage(ctx, ScopeAPIReadCheckRequest{ WorkspaceID: workspaceID, MessageID: "msg_scope_api_owner", ViewerMemberID: privateMemberID, ViewerType: ScopeActorPrivateAgent, RequestKind: ScopeVisibilityRequestAttachment, }) require.NoError(t, err) require.True(t, allowed.Allowed) require.Equal(t, ScopeVisibilityReasonAuthorInterval, allowed.Reason) denied, err := api.CanReadMessage(ctx, ScopeAPIReadCheckRequest{ WorkspaceID: workspaceID, MessageID: "msg_scope_api_hidden", ViewerMemberID: privateMemberID, ViewerType: ScopeActorPrivateAgent, RequestKind: ScopeVisibilityRequestDiagnostics, }) require.NoError(t, err) require.False(t, denied.Allowed) ownerRows, err := api.ListMemberDisplayRows(ctx, workspaceID, scopeID, ownerMemberID) require.NoError(t, err) require.Len(t, ownerRows, 3) var privateDisplay *ScopeMemberDisplayRow for i := range ownerRows { if ownerRows[i].MemberID == privateMemberID { privateDisplay = &ownerRows[i] } } require.NotNil(t, privateDisplay) require.Equal(t, "private", privateDisplay.Visibility) require.Equal(t, ownerMemberID, privateDisplay.OwnerMemberID) strangerRows, err := api.ListMemberDisplayRows(ctx, workspaceID, scopeID, strangerMemberID) require.NoError(t, err) require.Len(t, strangerRows, 2) for _, row := range strangerRows { require.NotEqual(t, privateMemberID, row.MemberID) require.Equal(t, "public", row.Visibility) } crossWorkspaceRows, err := api.ListMemberDisplayRows(ctx, "wsp_scope_api_other", scopeID, ownerMemberID) require.NoError(t, err) require.Empty(t, crossWorkspaceRows) nonMemberUserID := "usr_scope_api_nonmember" nonMemberID := "mem_scope_api_nonmember" insertMessagingUser(t, ctx, db, nonMemberUserID, "scope-api-nonmember@example.com", now) insertMessagingHumanMember(t, ctx, db, nonMemberID, workspaceID, nonMemberUserID, now) nonMemberRows, err := api.ListMemberDisplayRows(ctx, workspaceID, scopeID, nonMemberID) require.NoError(t, err) require.Empty(t, nonMemberRows) privateRows, err := api.ListMemberDisplayRows(ctx, workspaceID, scopeID, privateMemberID) require.NoError(t, err) require.Len(t, privateRows, 3) require.Contains(t, memberIDsForDisplayRows(privateRows), privateMemberID) } func TestScopeAPIHistoryAgentAuthorUsesHandleTargetAndDisplayName(t *testing.T) { t.Parallel() ctx := context.Background() db := startMessagingTestPool(t, ctx) repos := NewScopeRepositories(db) now := scopeAPINowForTest() const ( workspaceID = "wsp_scope_api_agent_author" ownerUserID = "usr_scope_api_agent_author" ownerMemberID = "mem_scope_api_agent_author_owner" scopeID = "scp_scope_api_agent_author" machineID = "mach_scope_api_agent_author" agentID = "agt_scope_api_agent_author" agentMemberID = "mem_scope_api_agent_author" agentHandle = "agent-after-rename" agentLabel = "Agent After Rename" messageID = "msg_scope_api_agent_author" ) insertMessagingWorkspace(t, ctx, db, workspaceID, now) insertMessagingUser(t, ctx, db, ownerUserID, "scope-api-agent-author@example.com", now) insertMessagingHumanMember(t, ctx, db, ownerMemberID, workspaceID, ownerUserID, now) insertMessagingAgentMember(t, ctx, db, agentMemberID, workspaceID, now) insertMessagingMachine(t, ctx, db, machineID, workspaceID, ownerUserID, now) insertMessagingAgent(t, ctx, db, agentID, workspaceID, machineID, agentMemberID, agentHandle, now) mustExecMessaging(t, ctx, db, ` UPDATE members SET display_name = $2 WHERE id = $1 `, agentMemberID, agentLabel) require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{ ID: scopeID, WorkspaceID: workspaceID, Type: "channel", IdentityKey: "channel:scope-api-agent-author", LatestSeq: 1, LatestGlobalSeq: 1, LatestMessageAt: &now, Status: "active", CreatedByMemberID: ownerMemberID, CreatedAt: now, UpdatedAt: now, })) require.NoError(t, repos.Scopes.UpsertChannelProfile(ctx, ScopeChannelProfileRecord{ ScopeID: scopeID, WorkspaceID: workspaceID, Slug: "scope-api-agent-author", DisplayName: "Scope API Agent Author", Visibility: "public", CreatedAt: now, UpdatedAt: now, })) require.NoError(t, repos.Memberships.Upsert(ctx, ScopeMembershipRecord{ ScopeID: scopeID, MemberID: ownerMemberID, Visibility: "public", CreatedAt: now, UpdatedAt: now, })) require.NoError(t, repos.Memberships.Upsert(ctx, ScopeMembershipRecord{ ScopeID: scopeID, MemberID: agentMemberID, Visibility: "public", CreatedAt: now, UpdatedAt: now, })) require.NoError(t, repos.Messages.Insert(ctx, ScopeMessageRecord{ ID: messageID, WorkspaceID: workspaceID, ScopeID: scopeID, Seq: 1, WorkspaceGlobalSeq: 1, AuthorMemberID: agentMemberID, AuthorType: ScopeActorAgent, MessageType: "text", Content: "agent message after rename", CreatedAt: now, UpdatedAt: now, })) api := NewScopeAPIService(New(db, nil, 0, Options{})) history, err := api.ListHistory(ctx, ScopeAPIHistoryRequest{ WorkspaceID: workspaceID, ScopeID: scopeID, ViewerMemberID: ownerMemberID, ViewerType: ScopeActorHuman, RequestKind: ScopeVisibilityRequestHistory, Limit: 10, }) require.NoError(t, err) require.Len(t, history.Messages, 1) require.Equal(t, agentHandle, history.Messages[0].AuthorProfile.TargetID) require.Equal(t, agentLabel, history.Messages[0].AuthorProfile.DisplayName) } func memberIDsForDisplayRows(rows []ScopeMemberDisplayRow) []string { ids := make([]string, 0, len(rows)) for _, row := range rows { ids = append(ids, row.MemberID) } return ids } func TestScopeAPIAgentHistoryInjectsThreadRootAndAdvancesThreadReadState(t *testing.T) { t.Parallel() ctx := context.Background() db := startMessagingTestPool(t, ctx) repos := NewScopeRepositories(db) now := scopeAPINowForTest() workspaceID := "wsp_scope_api_thread_root_read_state" ownerUserID := "usr_scope_api_thread_root_owner" ownerMemberID := "mem_scope_api_thread_root_owner" privateMemberID := "mem_scope_api_thread_root_private" parentScopeID := "scp_scope_api_thread_root_parent" threadScopeID := "thr_scope_api_thread_root_child" rootMessageID := "msg_scope_api_thread_root_parent" threadOldMessageID := "msg_scope_api_thread_root_old" threadNewMessageID := "msg_scope_api_thread_root_new" insertMessagingWorkspace(t, ctx, db, workspaceID, now) insertMessagingUser(t, ctx, db, ownerUserID, "scope-api-thread-root-owner@example.com", now) insertMessagingHumanMember(t, ctx, db, ownerMemberID, workspaceID, ownerUserID, now) insertMessagingAgentMember(t, ctx, db, privateMemberID, workspaceID, now) require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{ ID: parentScopeID, WorkspaceID: workspaceID, Type: "channel", IdentityKey: "channel:scope-api-thread-root-parent", LatestSeq: 5, LatestGlobalSeq: 70, LatestMessageAt: timePtr(now.Add(time.Second)), Status: "active", CreatedByMemberID: ownerMemberID, CreatedAt: now, UpdatedAt: now, })) require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{ ID: threadScopeID, WorkspaceID: workspaceID, Type: "thread", ParentScopeID: parentScopeID, RootMessageID: rootMessageID, RootMessageShortID: "root", LatestSeq: 3, LatestGlobalSeq: 30, LatestMessageAt: timePtr(now.Add(3 * time.Second)), Status: "active", CreatedByMemberID: ownerMemberID, CreatedAt: now, UpdatedAt: now, })) for _, membership := range []ScopeMembershipRecord{ {ScopeID: parentScopeID, MemberID: ownerMemberID, Visibility: "public", CreatedAt: now, UpdatedAt: now}, {ScopeID: parentScopeID, MemberID: privateMemberID, Visibility: "private", OwnerMemberID: ownerMemberID, CreatedAt: now, UpdatedAt: now}, {ScopeID: threadScopeID, MemberID: ownerMemberID, Visibility: "public", CreatedAt: now, UpdatedAt: now}, {ScopeID: threadScopeID, MemberID: privateMemberID, Visibility: "private", OwnerMemberID: ownerMemberID, LastReadSeq: 1, LastReadGlobalSeq: 10, CreatedAt: now, UpdatedAt: now}, } { require.NoError(t, repos.Memberships.Upsert(ctx, membership)) } for _, record := range []ScopeMessageRecord{ { ID: rootMessageID, WorkspaceID: workspaceID, ScopeID: parentScopeID, Seq: 5, WorkspaceGlobalSeq: 70, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "thread root context", CreatedAt: now.Add(time.Second), UpdatedAt: now.Add(time.Second), }, { ID: threadOldMessageID, WorkspaceID: workspaceID, ScopeID: threadScopeID, Seq: 2, WorkspaceGlobalSeq: 20, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "old thread message", CreatedAt: now.Add(2 * time.Second), UpdatedAt: now.Add(2 * time.Second), }, { ID: threadNewMessageID, WorkspaceID: workspaceID, ScopeID: threadScopeID, Seq: 3, WorkspaceGlobalSeq: 30, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "new thread message", CreatedAt: now.Add(3 * time.Second), UpdatedAt: now.Add(3 * time.Second), }, } { require.NoError(t, repos.Messages.Insert(ctx, record)) } api := NewScopeAPIService(New(db, nil, 0, Options{})) page, err := api.ListHistory(ctx, ScopeAPIHistoryRequest{ WorkspaceID: workspaceID, ScopeID: threadScopeID, ViewerMemberID: privateMemberID, ViewerType: ScopeActorPrivateAgent, RequestKind: ScopeVisibilityRequestHistory, Limit: 2, IncludeReplyThreadRoot: true, AdvanceReadState: true, }) require.NoError(t, err) require.Equal(t, threadScopeID, page.TargetKey) require.Equal(t, int64(3), page.CoversThroughSeq) require.Equal(t, int64(1), page.LastReadSeq) require.Equal(t, []string{rootMessageID, threadNewMessageID, threadOldMessageID}, historyMessageIDs(page.Messages)) require.Equal(t, []int64{5, 3, 2}, historyMessageSeqs(page.Messages)) require.Equal(t, int64(2), page.NextCursor) require.True(t, page.HasMore) var lastReadSeq, lastReadGlobalSeq int64 require.NoError(t, db.QueryRow(ctx, ` SELECT last_read_seq, last_read_global_seq FROM scope_memberships WHERE scope_id = $1 AND member_id = $2 `, threadScopeID, privateMemberID).Scan(&lastReadSeq, &lastReadGlobalSeq)) require.Equal(t, int64(3), lastReadSeq) require.Equal(t, int64(30), lastReadGlobalSeq) cursorPage, err := api.ListHistory(ctx, ScopeAPIHistoryRequest{ WorkspaceID: workspaceID, ScopeID: threadScopeID, ViewerMemberID: privateMemberID, ViewerType: ScopeActorPrivateAgent, RequestKind: ScopeVisibilityRequestHistory, BeforeSeq: 3, Limit: 10, IncludeReplyThreadRoot: true, AdvanceReadState: true, }) require.NoError(t, err) require.Equal(t, threadScopeID, cursorPage.TargetKey) require.Zero(t, cursorPage.CoversThroughSeq) require.Equal(t, int64(3), cursorPage.LastReadSeq) require.Equal(t, []string{threadOldMessageID}, historyMessageIDs(cursorPage.Messages)) require.NoError(t, db.QueryRow(ctx, ` SELECT last_read_seq, last_read_global_seq FROM scope_memberships WHERE scope_id = $1 AND member_id = $2 `, threadScopeID, privateMemberID).Scan(&lastReadSeq, &lastReadGlobalSeq)) require.Equal(t, int64(3), lastReadSeq) require.Equal(t, int64(30), lastReadGlobalSeq) } func TestScopeAPISearchRequiresReadReadyAndContinuesAfterInvisibleCandidates(t *testing.T) { t.Parallel() ctx := context.Background() db := startMessagingTestPool(t, ctx) repos := NewScopeRepositories(db) now := scopeAPINowForTest() workspaceID := "wsp_scope_api_search_ready" ownerUserID := "usr_scope_api_search_owner" ownerMemberID := "mem_scope_api_search_owner" hiddenUserID := "usr_scope_api_search_hidden" hiddenMemberID := "mem_scope_api_search_hidden" privateMemberID := "mem_scope_api_search_private" scopeID := "scp_scope_api_search_ready" insertMessagingWorkspace(t, ctx, db, workspaceID, now) insertMessagingUser(t, ctx, db, ownerUserID, "scope-api-search-owner@example.com", now) insertMessagingHumanMember(t, ctx, db, ownerMemberID, workspaceID, ownerUserID, now) insertMessagingUser(t, ctx, db, hiddenUserID, "scope-api-search-hidden@example.com", now) insertMessagingHumanMember(t, ctx, db, hiddenMemberID, workspaceID, hiddenUserID, now) insertMessagingAgentMember(t, ctx, db, privateMemberID, workspaceID, now) require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{ ID: scopeID, WorkspaceID: workspaceID, Type: "channel", IdentityKey: "channel:scope-api-search-ready", Status: "active", CreatedByMemberID: ownerMemberID, CreatedAt: now, UpdatedAt: now, })) require.NoError(t, repos.Memberships.Upsert(ctx, ScopeMembershipRecord{ ScopeID: scopeID, MemberID: privateMemberID, Visibility: "private", OwnerMemberID: ownerMemberID, CreatedAt: now, UpdatedAt: now, })) records := []ScopeMessageRecord{ { ID: "msg_scope_api_search_hidden_newer", WorkspaceID: workspaceID, ScopeID: scopeID, Seq: 3, WorkspaceGlobalSeq: 30, AuthorMemberID: hiddenMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "needle hidden newest", CreatedAt: now.Add(3 * time.Second), UpdatedAt: now.Add(3 * time.Second), }, { ID: "msg_scope_api_search_hidden_mid", WorkspaceID: workspaceID, ScopeID: scopeID, Seq: 2, WorkspaceGlobalSeq: 20, AuthorMemberID: hiddenMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "needle hidden mid", CreatedAt: now.Add(2 * time.Second), UpdatedAt: now.Add(2 * time.Second), }, { ID: "msg_scope_api_search_owner_visible", WorkspaceID: workspaceID, ScopeID: scopeID, Seq: 1, WorkspaceGlobalSeq: 10, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "needle owner visible", CreatedAt: now.Add(time.Second), UpdatedAt: now.Add(time.Second), }, } for _, record := range records { require.NoError(t, repos.Messages.Insert(ctx, record)) require.NoError(t, repos.Search.InsertDoc(ctx, ScopeSearchDocRecord{ MessageID: record.ID, WorkspaceID: record.WorkspaceID, ScopeID: record.ScopeID, GlobalSeq: record.WorkspaceGlobalSeq, Content: record.Content, CreatedAt: record.CreatedAt, })) } api := NewScopeAPIService(New(db, nil, 0, Options{})) _, err := api.SearchMessages(ctx, ScopeAPISearchRequest{ WorkspaceID: workspaceID, ScopeID: scopeID, ViewerMemberID: privateMemberID, ViewerType: ScopeActorPrivateAgent, Query: "needle", Limit: 1, CandidateBatchLimit: 2, MaxCandidateBatches: 1, }) require.ErrorIs(t, err, ErrMessageSearchUnavailable) candidatesByCursor := map[MessageSearchCursor][]ScopeMessageVisibilityCandidate{ {}: { scopeSearchCandidate(records[0], ScopeMembershipRecord{ScopeID: scopeID, MemberID: privateMemberID, Visibility: "private", OwnerMemberID: ownerMemberID}), scopeSearchCandidate(records[1], ScopeMembershipRecord{ScopeID: scopeID, MemberID: privateMemberID, Visibility: "private", OwnerMemberID: ownerMemberID}), }, {ScopeID: scopeID, Seq: 2, MessageID: records[1].ID}: { scopeSearchCandidate(records[2], ScopeMembershipRecord{ScopeID: scopeID, MemberID: privateMemberID, Visibility: "private", OwnerMemberID: ownerMemberID}), }, } firstPage, err := searchVisibleWithLoader(ctx, NewVisibilityFilter(), ScopeAPISearchRequest{ WorkspaceID: workspaceID, ScopeID: scopeID, ViewerMemberID: privateMemberID, ViewerType: ScopeActorPrivateAgent, Query: "needle", Limit: 1, CandidateBatchLimit: 2, MaxCandidateBatches: 1, }, 1, func(_ context.Context, cursor MessageSearchCursor, _ int) ([]ScopeMessageVisibilityCandidate, error) { return candidatesByCursor[cursor], nil }) require.NoError(t, err) require.Empty(t, firstPage.Messages) require.True(t, firstPage.HasMore) require.True(t, firstPage.CandidateBudgetExhausted) require.Equal(t, int64(2), firstPage.NextCursor) require.Equal(t, MessageSearchCursor{ScopeID: scopeID, Seq: 2, MessageID: records[1].ID}, firstPage.NextSearchCursor) secondPage, err := searchVisibleWithLoader(ctx, NewVisibilityFilter(), ScopeAPISearchRequest{ WorkspaceID: workspaceID, ScopeID: scopeID, ViewerMemberID: privateMemberID, ViewerType: ScopeActorPrivateAgent, Query: "needle", BeforeCursor: firstPage.NextSearchCursor, Limit: 1, CandidateBatchLimit: 2, MaxCandidateBatches: 1, }, 1, func(_ context.Context, cursor MessageSearchCursor, _ int) ([]ScopeMessageVisibilityCandidate, error) { return candidatesByCursor[cursor], nil }) require.NoError(t, err) require.Len(t, secondPage.Messages, 1) require.Equal(t, "msg_scope_api_search_owner_visible", secondPage.Messages[0].Message.ID) } func scopeSearchCandidate(message ScopeMessageRecord, membership ScopeMembershipRecord) ScopeMessageVisibilityCandidate { reason := "" switch strings.TrimSpace(membership.Visibility) { case "public": reason = ScopeVisibilityReasonMembership case "private": switch { case strings.TrimSpace(message.AuthorMemberID) == strings.TrimSpace(membership.MemberID): reason = ScopeVisibilityReasonPrivateOutput case strings.TrimSpace(membership.OwnerMemberID) != "" && strings.TrimSpace(message.AuthorMemberID) == strings.TrimSpace(membership.OwnerMemberID): reason = ScopeVisibilityReasonAuthorInterval } } return ScopeMessageVisibilityCandidate{ Message: message, Membership: membership, Reason: reason, } } func TestScopeAPISearchUsesScopeLocalSeqOrderingAndSearchSeqs(t *testing.T) { t.Parallel() ctx := context.Background() db := startMessagingTestPool(t, ctx) repos := NewScopeRepositories(db) now := scopeAPINowForTest() workspaceID := "wsp_scope_api_search_scope_seq" ownerUserID := "usr_scope_api_search_scope_seq_owner" ownerMemberID := "mem_scope_api_search_scope_seq_owner" scopeID := "scp_scope_api_search_scope_seq" insertMessagingWorkspace(t, ctx, db, workspaceID, now) insertMessagingUser(t, ctx, db, ownerUserID, "scope-api-search-scope-seq-owner@example.com", now) insertMessagingHumanMember(t, ctx, db, ownerMemberID, workspaceID, ownerUserID, now) require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{ ID: scopeID, WorkspaceID: workspaceID, Type: "channel", IdentityKey: "channel:scope-api-search-scope-seq", LatestSeq: 3, LatestGlobalSeq: 200, LatestMessageAt: timePtr(now.Add(3 * time.Second)), Status: "active", CreatedByMemberID: ownerMemberID, CreatedAt: now, UpdatedAt: now, })) require.NoError(t, repos.Memberships.Upsert(ctx, ScopeMembershipRecord{ ScopeID: scopeID, MemberID: ownerMemberID, Visibility: "public", CreatedAt: now, UpdatedAt: now, })) records := []ScopeMessageRecord{ { ID: "msg_scope_api_search_scope_seq_old", WorkspaceID: workspaceID, ScopeID: scopeID, Seq: 1, WorkspaceGlobalSeq: 100, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "needle oldest", CreatedAt: now.Add(time.Second), UpdatedAt: now.Add(time.Second), }, { ID: "msg_scope_api_search_scope_seq_mid", WorkspaceID: workspaceID, ScopeID: scopeID, Seq: 2, WorkspaceGlobalSeq: 200, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "needle middle", CreatedAt: now.Add(2 * time.Second), UpdatedAt: now.Add(2 * time.Second), }, { ID: "msg_scope_api_search_scope_seq_new", WorkspaceID: workspaceID, ScopeID: scopeID, Seq: 3, WorkspaceGlobalSeq: 1, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "needle newest", CreatedAt: now.Add(3 * time.Second), UpdatedAt: now.Add(3 * time.Second), }, } for _, record := range records { require.NoError(t, repos.Messages.Insert(ctx, record)) require.NoError(t, repos.Search.InsertDoc(ctx, ScopeSearchDocRecord{ MessageID: record.ID, WorkspaceID: record.WorkspaceID, ScopeID: record.ScopeID, GlobalSeq: record.WorkspaceGlobalSeq, Content: record.Content, CreatedAt: record.CreatedAt, })) } firstPage, err := searchVisibleWithLoader(ctx, NewVisibilityFilter(), ScopeAPISearchRequest{ WorkspaceID: workspaceID, ScopeID: scopeID, ViewerMemberID: ownerMemberID, ViewerType: ScopeActorHuman, Query: "needle", Limit: 2, CandidateBatchLimit: 3, MaxCandidateBatches: 1, }, 2, func(_ context.Context, cursor MessageSearchCursor, _ int) ([]ScopeMessageVisibilityCandidate, error) { switch cursor { case MessageSearchCursor{}: return []ScopeMessageVisibilityCandidate{ scopeSearchCandidate(records[1], ScopeMembershipRecord{ScopeID: scopeID, MemberID: ownerMemberID, Visibility: "public"}), scopeSearchCandidate(records[0], ScopeMembershipRecord{ScopeID: scopeID, MemberID: ownerMemberID, Visibility: "public"}), scopeSearchCandidate(records[2], ScopeMembershipRecord{ScopeID: scopeID, MemberID: ownerMemberID, Visibility: "public"}), }, nil case MessageSearchCursor{ScopeID: scopeID, Seq: 2, MessageID: records[1].ID}: return []ScopeMessageVisibilityCandidate{ scopeSearchCandidate(records[0], ScopeMembershipRecord{ScopeID: scopeID, MemberID: ownerMemberID, Visibility: "public"}), }, nil default: return nil, nil } }) require.NoError(t, err) require.Equal(t, []string{ "msg_scope_api_search_scope_seq_new", "msg_scope_api_search_scope_seq_mid", }, visibleMessageRefIDs(firstPage.Messages)) require.Equal(t, int64(2), firstPage.NextCursor) require.Equal(t, MessageSearchCursor{ScopeID: scopeID, Seq: 2, MessageID: records[1].ID}, firstPage.NextSearchCursor) require.True(t, firstPage.HasMore) secondPage, err := searchVisibleWithLoader(ctx, NewVisibilityFilter(), ScopeAPISearchRequest{ WorkspaceID: workspaceID, ScopeID: scopeID, ViewerMemberID: ownerMemberID, ViewerType: ScopeActorHuman, Query: "needle", BeforeCursor: firstPage.NextSearchCursor, Limit: 2, CandidateBatchLimit: 3, MaxCandidateBatches: 1, }, 2, func(_ context.Context, cursor MessageSearchCursor, _ int) ([]ScopeMessageVisibilityCandidate, error) { switch cursor { case MessageSearchCursor{}: return []ScopeMessageVisibilityCandidate{ scopeSearchCandidate(records[1], ScopeMembershipRecord{ScopeID: scopeID, MemberID: ownerMemberID, Visibility: "public"}), scopeSearchCandidate(records[0], ScopeMembershipRecord{ScopeID: scopeID, MemberID: ownerMemberID, Visibility: "public"}), scopeSearchCandidate(records[2], ScopeMembershipRecord{ScopeID: scopeID, MemberID: ownerMemberID, Visibility: "public"}), }, nil case MessageSearchCursor{ScopeID: scopeID, Seq: 2, MessageID: records[1].ID}: return []ScopeMessageVisibilityCandidate{ scopeSearchCandidate(records[0], ScopeMembershipRecord{ScopeID: scopeID, MemberID: ownerMemberID, Visibility: "public"}), }, nil default: return nil, nil } }) require.NoError(t, err) require.Equal(t, []string{"msg_scope_api_search_scope_seq_old"}, visibleMessageRefIDs(secondPage.Messages)) require.Equal(t, int64(1), secondPage.NextCursor) require.False(t, secondPage.HasMore) api := NewScopeAPIService(New(db, nil, 0, Options{})) historyPage, err := api.historyMessagesFromVisibleRefs(ctx, workspaceID, firstPage.Messages) require.NoError(t, err) require.Equal(t, []string{ "msg_scope_api_search_scope_seq_new", "msg_scope_api_search_scope_seq_mid", }, historyMessageIDs(historyPage)) require.Equal(t, []int64{3, 2}, historyMessageSeqs(historyPage)) appPage, err := api.appMessagesFromVisibleRefs(ctx, workspaceID, firstPage.Messages) require.NoError(t, err) require.Equal(t, []string{ "msg_scope_api_search_scope_seq_new", "msg_scope_api_search_scope_seq_mid", }, appMessageIDs(appPage)) require.Equal(t, []int64{3, 2}, appMessageSeqs(appPage)) } func TestScopeAPISearchWorkspaceWideUsesGlobalOrderAndLocalSeqs(t *testing.T) { t.Parallel() ctx := context.Background() db := startMessagingTestPool(t, ctx) repos := NewScopeRepositories(db) now := scopeAPINowForTest() workspaceID := "wsp_scope_api_search_workspace_order" ownerUserID := "usr_scope_api_search_workspace_owner" ownerMemberID := "mem_scope_api_search_workspace_owner" scopeAlphaID := "scp_scope_api_search_workspace_alpha" scopeBetaID := "scp_scope_api_search_workspace_beta" scopeGammaID := "scp_scope_api_search_workspace_gamma" insertMessagingWorkspace(t, ctx, db, workspaceID, now) insertMessagingUser(t, ctx, db, ownerUserID, "scope-api-search-workspace-owner@example.com", now) insertMessagingHumanMember(t, ctx, db, ownerMemberID, workspaceID, ownerUserID, now) for _, scopeRecord := range []ScopeRecord{ { ID: scopeAlphaID, WorkspaceID: workspaceID, Type: "channel", IdentityKey: "channel:scope-api-search-workspace-alpha", LatestSeq: 2, LatestGlobalSeq: 200, LatestMessageAt: timePtr(now.Add(2 * time.Second)), Status: "active", CreatedByMemberID: ownerMemberID, CreatedAt: now, UpdatedAt: now, }, { ID: scopeBetaID, WorkspaceID: workspaceID, Type: "channel", IdentityKey: "channel:scope-api-search-workspace-beta", LatestSeq: 1, LatestGlobalSeq: 1, LatestMessageAt: timePtr(now.Add(4 * time.Second)), Status: "active", CreatedByMemberID: ownerMemberID, CreatedAt: now, UpdatedAt: now, }, { ID: scopeGammaID, WorkspaceID: workspaceID, Type: "channel", IdentityKey: "channel:scope-api-search-workspace-gamma", LatestSeq: 1, LatestGlobalSeq: 150, LatestMessageAt: timePtr(now.Add(3 * time.Second)), Status: "active", CreatedByMemberID: ownerMemberID, CreatedAt: now, UpdatedAt: now, }, } { require.NoError(t, repos.Scopes.UpsertScope(ctx, scopeRecord)) require.NoError(t, repos.Memberships.Upsert(ctx, ScopeMembershipRecord{ ScopeID: scopeRecord.ID, MemberID: ownerMemberID, Visibility: "public", EffectiveFromSeq: 1, CreatedAt: now, UpdatedAt: now, })) } records := []ScopeMessageRecord{ { ID: "msg_scope_api_search_workspace_old", WorkspaceID: workspaceID, ScopeID: scopeAlphaID, Seq: 1, WorkspaceGlobalSeq: 100, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "needle alpha old", CreatedAt: now.Add(time.Second), UpdatedAt: now.Add(time.Second), }, { ID: "msg_scope_api_search_workspace_mid", WorkspaceID: workspaceID, ScopeID: scopeAlphaID, Seq: 2, WorkspaceGlobalSeq: 200, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "needle alpha mid", CreatedAt: now.Add(2 * time.Second), UpdatedAt: now.Add(2 * time.Second), }, { ID: "msg_scope_api_search_workspace_newer", WorkspaceID: workspaceID, ScopeID: scopeGammaID, Seq: 1, WorkspaceGlobalSeq: 150, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "needle gamma newer", CreatedAt: now.Add(3 * time.Second), UpdatedAt: now.Add(3 * time.Second), }, { ID: "msg_scope_api_search_workspace_newest", WorkspaceID: workspaceID, ScopeID: scopeBetaID, Seq: 1, WorkspaceGlobalSeq: 1, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "needle beta newest", CreatedAt: now.Add(4 * time.Second), UpdatedAt: now.Add(4 * time.Second), }, } for _, record := range records { require.NoError(t, repos.Messages.Insert(ctx, record)) require.NoError(t, repos.Search.InsertDoc(ctx, ScopeSearchDocRecord{ MessageID: record.ID, WorkspaceID: record.WorkspaceID, ScopeID: record.ScopeID, GlobalSeq: record.WorkspaceGlobalSeq, Content: record.Content, CreatedAt: record.CreatedAt, })) } page, err := searchVisibleWithLoader(ctx, NewVisibilityFilter(), ScopeAPISearchRequest{ WorkspaceID: workspaceID, ViewerMemberID: ownerMemberID, ViewerType: ScopeActorHuman, Query: "needle", Limit: 4, CandidateBatchLimit: 4, MaxCandidateBatches: 1, }, 4, func(_ context.Context, cursor MessageSearchCursor, _ int) ([]ScopeMessageVisibilityCandidate, error) { require.True(t, cursor.IsZero()) return []ScopeMessageVisibilityCandidate{ scopeSearchCandidate(records[1], ScopeMembershipRecord{ScopeID: scopeAlphaID, MemberID: ownerMemberID, Visibility: "public"}), scopeSearchCandidate(records[2], ScopeMembershipRecord{ScopeID: scopeGammaID, MemberID: ownerMemberID, Visibility: "public"}), scopeSearchCandidate(records[0], ScopeMembershipRecord{ScopeID: scopeAlphaID, MemberID: ownerMemberID, Visibility: "public"}), scopeSearchCandidate(records[3], ScopeMembershipRecord{ScopeID: scopeBetaID, MemberID: ownerMemberID, Visibility: "public"}), }, nil }) require.NoError(t, err) require.Equal(t, []string{ "msg_scope_api_search_workspace_mid", "msg_scope_api_search_workspace_newer", "msg_scope_api_search_workspace_old", "msg_scope_api_search_workspace_newest", }, visibleMessageRefIDs(page.Messages)) api := NewScopeAPIService(New(db, nil, 0, Options{})) historyPage, err := api.historyMessagesFromVisibleRefs(ctx, workspaceID, page.Messages) require.NoError(t, err) require.Equal(t, []string{ "msg_scope_api_search_workspace_mid", "msg_scope_api_search_workspace_newer", "msg_scope_api_search_workspace_old", "msg_scope_api_search_workspace_newest", }, historyMessageIDs(historyPage)) require.Equal(t, []int64{2, 1, 1, 1}, historyMessageSeqs(historyPage)) appPage, err := api.appMessagesFromVisibleRefs(ctx, workspaceID, page.Messages) require.NoError(t, err) require.Equal(t, []string{ "msg_scope_api_search_workspace_mid", "msg_scope_api_search_workspace_newer", "msg_scope_api_search_workspace_old", "msg_scope_api_search_workspace_newest", }, appMessageIDs(appPage)) require.Equal(t, []int64{2, 1, 1, 1}, appMessageSeqs(appPage)) require.Equal(t, []string{scopeAlphaID, scopeGammaID, scopeAlphaID, scopeBetaID}, appMessageChannelIDs(appPage)) } func TestScopeAPISearchWorkspaceWideCursorContinuesSameGlobalSeqAcrossScopes(t *testing.T) { t.Parallel() ctx := context.Background() now := scopeAPINowForTest() workspaceID := "wsp_scope_api_search_same_global" viewerMemberID := "mem_scope_api_search_same_global" scopeAlphaID := "scp_scope_api_search_same_global_alpha" scopeBetaID := "scp_scope_api_search_same_global_beta" first := ScopeMessageRecord{ ID: "msg_scope_api_search_same_global_alpha", WorkspaceID: workspaceID, ScopeID: scopeAlphaID, Seq: 7, WorkspaceGlobalSeq: 7, AuthorMemberID: viewerMemberID, AuthorType: ScopeActorHuman, Content: "needle alpha", CreatedAt: now, } second := ScopeMessageRecord{ ID: "msg_scope_api_search_same_global_beta", WorkspaceID: workspaceID, ScopeID: scopeBetaID, Seq: 7, WorkspaceGlobalSeq: 7, AuthorMemberID: viewerMemberID, AuthorType: ScopeActorHuman, Content: "needle beta", CreatedAt: now.Add(time.Second), } firstCursor := MessageSearchCursor{GlobalSeq: 7, ScopeID: scopeAlphaID, Seq: 7, MessageID: first.ID} load := func(_ context.Context, cursor MessageSearchCursor, _ int) ([]ScopeMessageVisibilityCandidate, error) { switch cursor { case MessageSearchCursor{}: return []ScopeMessageVisibilityCandidate{ scopeSearchCandidate(first, ScopeMembershipRecord{ScopeID: scopeAlphaID, MemberID: viewerMemberID, Visibility: "public"}), scopeSearchCandidate(second, ScopeMembershipRecord{ScopeID: scopeBetaID, MemberID: viewerMemberID, Visibility: "public"}), }, nil case firstCursor: return []ScopeMessageVisibilityCandidate{ scopeSearchCandidate(second, ScopeMembershipRecord{ScopeID: scopeBetaID, MemberID: viewerMemberID, Visibility: "public"}), }, nil default: return nil, nil } } firstPage, err := searchVisibleWithLoader(ctx, NewVisibilityFilter(), ScopeAPISearchRequest{ WorkspaceID: workspaceID, ViewerMemberID: viewerMemberID, ViewerType: ScopeActorHuman, Query: "needle", Limit: 1, CandidateBatchLimit: 2, MaxCandidateBatches: 1, }, 1, load) require.NoError(t, err) require.Equal(t, []string{first.ID}, visibleMessageRefIDs(firstPage.Messages)) require.True(t, firstPage.HasMore) require.Equal(t, firstCursor, firstPage.NextSearchCursor) secondPage, err := searchVisibleWithLoader(ctx, NewVisibilityFilter(), ScopeAPISearchRequest{ WorkspaceID: workspaceID, ViewerMemberID: viewerMemberID, ViewerType: ScopeActorHuman, Query: "needle", BeforeCursor: firstPage.NextSearchCursor, Limit: 1, CandidateBatchLimit: 2, MaxCandidateBatches: 1, }, 1, load) require.NoError(t, err) require.Equal(t, []string{second.ID}, visibleMessageRefIDs(secondPage.Messages)) } func TestScopeAPIPresentationsUseScopeMembershipForDMLabels(t *testing.T) { t.Parallel() query := scopePresentationsByIDQuery.SQL require.Contains(t, query, "dm_scope_human_members AS") require.Contains(t, query, "FROM scope_memberships sm") require.Contains(t, query, "sm.archived_at IS NULL") require.Contains(t, query, "sm.effective_to_seq IS NULL") require.NotContains(t, query, legacyTableNameForMessageTest("dm", "threads")) ctx := context.Background() db := startMessagingTestPool(t, ctx) repos := NewScopeRepositories(db) now := scopeAPINowForTest() workspaceID := "wsp_scope_api_dm_presentation" aliceUserID := "usr_scope_api_dm_alice" bobUserID := "usr_scope_api_dm_bob" closedUserID := "usr_scope_api_dm_closed" aliceMemberID := "mem_scope_api_dm_1_alice" bobMemberID := "mem_scope_api_dm_2_bob" closedMemberID := "mem_scope_api_dm_0_closed" dmScopeID := "dm_scope_api_presentation" threadScopeID := "thr_scope_api_parent_dm" closedSeq := int64(3) insertMessagingWorkspace(t, ctx, db, workspaceID, now) insertMessagingUser(t, ctx, db, aliceUserID, "scope-api-dm-alice@example.com", now) insertMessagingHumanMember(t, ctx, db, aliceMemberID, workspaceID, aliceUserID, now) insertMessagingUser(t, ctx, db, bobUserID, "scope-api-dm-bob@example.com", now) insertMessagingHumanMember(t, ctx, db, bobMemberID, workspaceID, bobUserID, now) insertMessagingUser(t, ctx, db, closedUserID, "scope-api-dm-closed@example.com", now) insertMessagingHumanMember(t, ctx, db, closedMemberID, workspaceID, closedUserID, now) require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{ ID: dmScopeID, WorkspaceID: workspaceID, Type: "dm", IdentityKey: "dm:" + dmScopeID, Status: "active", CreatedAt: now, UpdatedAt: now, })) require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{ ID: threadScopeID, WorkspaceID: workspaceID, Type: "thread", ParentScopeID: dmScopeID, RootMessageID: "msg_scope_api_parent_dm_root", RootMessageShortID: "rootdm", IdentityKey: "thread:" + threadScopeID, Status: "active", CreatedAt: now, UpdatedAt: now, })) for _, membership := range []ScopeMembershipRecord{ {ScopeID: dmScopeID, MemberID: aliceMemberID, Visibility: "public", EffectiveFromSeq: 0, CreatedAt: now, UpdatedAt: now}, {ScopeID: dmScopeID, MemberID: bobMemberID, Visibility: "public", EffectiveFromSeq: 0, CreatedAt: now, UpdatedAt: now}, {ScopeID: dmScopeID, MemberID: closedMemberID, Visibility: "public", EffectiveFromSeq: 0, EffectiveToSeq: &closedSeq, CreatedAt: now, UpdatedAt: now}, } { require.NoError(t, repos.Memberships.Upsert(ctx, membership)) } presentations, err := loadScopePresentations(ctx, db, workspaceID, []string{dmScopeID, threadScopeID}) require.NoError(t, err) require.Equal(t, "dm", presentations[dmScopeID].Type) require.Equal(t, aliceMemberID, presentations[dmScopeID].Name) require.Equal(t, aliceMemberID, presentations[dmScopeID].DisplayName) require.Equal(t, "thread", presentations[threadScopeID].Type) require.Equal(t, dmScopeID, presentations[threadScopeID].ParentID) require.Equal(t, "dm", presentations[threadScopeID].ParentType) require.Equal(t, aliceMemberID, presentations[threadScopeID].ParentName) require.Equal(t, aliceMemberID, presentations[threadScopeID].ParentDisplayName) } func historyMessageIDs(messages []HistoryMessage) []string { ids := make([]string, 0, len(messages)) for _, message := range messages { ids = append(ids, message.ID) } return ids } func visibleMessageRefIDs(messages []VisibleMessageRef) []string { ids := make([]string, 0, len(messages)) for _, message := range messages { ids = append(ids, message.Message.ID) } return ids } func historyMessageSeqs(messages []HistoryMessage) []int64 { seqs := make([]int64, 0, len(messages)) for _, message := range messages { seqs = append(seqs, message.Seq) } return seqs } func appMessageIDs(messages []AppMessage) []string { ids := make([]string, 0, len(messages)) for _, message := range messages { ids = append(ids, message.ID) } return ids } func appMessageSeqs(messages []AppMessage) []int64 { seqs := make([]int64, 0, len(messages)) for _, message := range messages { seqs = append(seqs, message.Seq) } return seqs } func appMessageChannelIDs(messages []AppMessage) []string { ids := make([]string, 0, len(messages)) for _, message := range messages { ids = append(ids, message.ChannelID) } return ids } func timePtr(value time.Time) *time.Time { return &value } func scopeAPINowForTest() time.Time { return time.Date(2026, 5, 13, 6, 0, 0, 0, time.UTC) } // seedScopeAPIReplyThreadRootSource binds a workflow_run or schedule_run source // ledger to a reply-thread-root message so the contribution source resolver can // resolve a typed workflow_run/schedule_run source for the message. This mirrors // the production write path that binds scope_messages to the native ledgers. func seedScopeAPIReplyThreadRootSource(t *testing.T, ctx context.Context, pool coredb.QueryExecer, workspaceID string, messageID string, deliveryScopeID string, kind string) { t.Helper() switch kind { case "workflow": runID := "wri_scope_api_thread_root_workflow" effectID := "wrf_scope_api_thread_root_workflow" invocationID := "wiv_scope_api_thread_root_workflow" planID := "wfp_scope_api_thread_root_workflow" workflowName := "Scope API Workflow" now := scopeAPINowForTest() mustExecMessaging(t, ctx, pool, ` INSERT INTO workflow_invocations ( id,workspace_id,plan_id,plan_version_id,workflow_name_snapshot,source_kind,source_occurrence_ref,source_occurred_at, input_json,input_digest,source_provenance_json, owner_user_id_snapshot,owner_member_id_snapshot,principal_member_id_snapshot,author_member_id_snapshot, invoked_by_member_id,invoked_by_actor_type,invoked_by_actor_id,idempotency_key,request_digest,accepted_at ) VALUES ($1,$2,$3,'plv_scope_api_thread_root',$4,'manual','manual:scope-api-thread-root',$5,'{}','digest','{}','usr','mem','mem','mem','usr','human','usr','scope-api-root',$6,$5) `, invocationID, workspaceID, planID, workflowName, now, strings.Repeat("0", 64)) mustExecMessaging(t, ctx, pool, ` INSERT INTO workflow_run_instances ( id,workspace_id,invocation_id,plan_id,plan_version_id,workflow_name_snapshot,status, input_json,input_digest,state_json,owner_user_id_snapshot,owner_member_id_snapshot,principal_member_id_snapshot,author_member_id_snapshot, queued_at,revision,last_event_seq,runnable_at,created_at,updated_at ) VALUES ($1,$2,$3,$4,'plv_scope_api_thread_root',$5,'waiting','{}','digest','{}','usr','mem','mem','mem',$6,1,1,$6,$6,$6) `, runID, workspaceID, invocationID, planID, workflowName, now) mustExecMessaging(t, ctx, pool, ` INSERT INTO workflow_run_activations ( id,workspace_id,run_id,logical_key,node_id,opcode,status,input_json,scope_context_json,attempt_count,created_at,updated_at ) VALUES ('wra_scope_api_thread_root',$2,$3,'scope-api-thread-root','message','send_message.callback','waiting','{}','{}',1,$4,$4) `, effectID, workspaceID, runID, now) mustExecMessaging(t, ctx, pool, ` INSERT INTO workflow_run_effects ( id,workspace_id,run_id,activation_id,effect_kind,idempotency_key,status,request_json,attempt_count,message_id,created_at,updated_at ) VALUES ($1,$2,$3,'wra_scope_api_thread_root','send_message',$1,'succeeded','{}',1,$4,$5,$5) `, effectID, workspaceID, runID, messageID, now) case "schedule": runID := "atr_scope_api_thread_root_schedule" automationID := "sat_scope_api_thread_root_schedule" now := scopeAPINowForTest() mustExecMessaging(t, ctx, pool, ` INSERT INTO automation_runs ( id,workspace_id,automation_kind,automation_id,idempotency_key,source_occurrence_ref,source_kind,source_type, status,scheduled_for,attempt_count,effect_kind,message_id,error_code,error_detail, target_kind,target_ref,delivery_scope_id,permission_check_scope_id, owner_member_id_snapshot,principal_member_id_snapshot,author_member_id_snapshot,automation_name_snapshot, notification_mode_snapshot,message_template_snapshot,rendered_content_preview,created_at,updated_at ) VALUES ($1,$2,'schedule',$3,$1,'occurrence:scope-api-root','time','time_trigger','succeeded',$4,1,'message',$5,'','','channel','#scope-api',$6,$6,'mem','mem','mem','Scope API Schedule','none','{}','',$4,$4) `, runID, workspaceID, automationID, now, messageID, deliveryScopeID) default: t.Fatalf("unsupported reply-thread-root source kind %q", kind) } } func TestScopeAPIReplyThreadRootMetadataWorkflowServiceAuthor(t *testing.T) { t.Parallel() ctx := context.Background() db := startMessagingTestPool(t, ctx) repos := NewScopeRepositories(db) now := scopeAPINowForTest() workspaceID := "wsp_scope_api_root_workflow" ownerUserID := "usr_scope_api_root_workflow" ownerMemberID := "mem_scope_api_root_workflow" parentScopeID := "scp_scope_api_root_workflow_parent" rootMessageID := "msg_scope_api_root_workflow" insertMessagingWorkspace(t, ctx, db, workspaceID, now) insertMessagingUser(t, ctx, db, ownerUserID, "scope-api-root-workflow@example.com", now) insertMessagingHumanMember(t, ctx, db, ownerMemberID, workspaceID, ownerUserID, now) require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{ ID: parentScopeID, WorkspaceID: workspaceID, Type: "channel", IdentityKey: "channel:scope-api-root-workflow-parent", LatestSeq: 5, LatestGlobalSeq: 70, LatestMessageAt: timePtr(now.Add(time.Second)), Status: "active", CreatedByMemberID: ownerMemberID, CreatedAt: now, UpdatedAt: now, })) require.NoError(t, repos.Messages.Insert(ctx, ScopeMessageRecord{ ID: rootMessageID, WorkspaceID: workspaceID, ScopeID: parentScopeID, Seq: 5, WorkspaceGlobalSeq: 70, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "workflow thread root", CreatedAt: now.Add(time.Second), UpdatedAt: now.Add(time.Second), })) seedScopeAPIReplyThreadRootSource(t, ctx, db, workspaceID, rootMessageID, parentScopeID, "workflow") api := NewScopeAPIService(New(db, nil, 0, Options{})) metadata, err := api.replyThreadRootMetadataFromVisibleRef(ctx, workspaceID, VisibleMessageRef{ Message: ScopeMessageRecord{ ID: rootMessageID, WorkspaceID: workspaceID, ScopeID: parentScopeID, Seq: 5, WorkspaceGlobalSeq: 70, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "workflow thread root", CreatedAt: now.Add(time.Second), UpdatedAt: now.Add(time.Second), }, }) require.NoError(t, err) require.Equal(t, rootMessageID, metadata["messageId"]) require.Equal(t, "service", metadata["authorType"], "Workflow root must not serialize the Human author type") require.Equal(t, "Scope API Workflow", metadata["authorName"]) require.Equal(t, "Scope API Workflow", metadata["senderName"]) profile, ok := metadata["authorProfile"].(map[string]any) require.True(t, ok) require.Equal(t, "service", profile["actorType"]) require.Equal(t, AuthorServiceKindWorkflow, profile["serviceKind"]) require.Equal(t, "Scope API Workflow", profile["displayName"]) source, ok := metadata["source"].(map[string]any) require.True(t, ok, "reply-thread-root metadata must carry the typed source") require.Equal(t, ContributionSourceKindWorkflowRun, source["kind"]) require.Equal(t, ContributionSourceRefTypeWorkflowRun, source["refType"]) require.NotContains(t, ownerMemberID, metadata["authorName"]) } func TestScopeAPIReplyThreadRootMetadataScheduleServiceAuthor(t *testing.T) { t.Parallel() ctx := context.Background() db := startMessagingTestPool(t, ctx) repos := NewScopeRepositories(db) now := scopeAPINowForTest() workspaceID := "wsp_scope_api_root_schedule" ownerUserID := "usr_scope_api_root_schedule" ownerMemberID := "mem_scope_api_root_schedule" parentScopeID := "scp_scope_api_root_schedule_parent" rootMessageID := "msg_scope_api_root_schedule" insertMessagingWorkspace(t, ctx, db, workspaceID, now) insertMessagingUser(t, ctx, db, ownerUserID, "scope-api-root-schedule@example.com", now) insertMessagingHumanMember(t, ctx, db, ownerMemberID, workspaceID, ownerUserID, now) require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{ ID: parentScopeID, WorkspaceID: workspaceID, Type: "channel", IdentityKey: "channel:scope-api-root-schedule-parent", LatestSeq: 5, LatestGlobalSeq: 70, LatestMessageAt: timePtr(now.Add(time.Second)), Status: "active", CreatedByMemberID: ownerMemberID, CreatedAt: now, UpdatedAt: now, })) require.NoError(t, repos.Messages.Insert(ctx, ScopeMessageRecord{ ID: rootMessageID, WorkspaceID: workspaceID, ScopeID: parentScopeID, Seq: 5, WorkspaceGlobalSeq: 70, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "schedule thread root", CreatedAt: now.Add(time.Second), UpdatedAt: now.Add(time.Second), })) seedScopeAPIReplyThreadRootSource(t, ctx, db, workspaceID, rootMessageID, parentScopeID, "schedule") api := NewScopeAPIService(New(db, nil, 0, Options{})) metadata, err := api.replyThreadRootMetadataFromVisibleRef(ctx, workspaceID, VisibleMessageRef{ Message: ScopeMessageRecord{ ID: rootMessageID, WorkspaceID: workspaceID, ScopeID: parentScopeID, Seq: 5, WorkspaceGlobalSeq: 70, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "schedule thread root", CreatedAt: now.Add(time.Second), UpdatedAt: now.Add(time.Second), }, }) require.NoError(t, err) require.Equal(t, rootMessageID, metadata["messageId"]) require.Equal(t, "service", metadata["authorType"], "Schedule root must not serialize the Human author type") require.Equal(t, "Scope API Schedule", metadata["authorName"]) require.Equal(t, "Scope API Schedule", metadata["senderName"]) profile, ok := metadata["authorProfile"].(map[string]any) require.True(t, ok) require.Equal(t, "service", profile["actorType"]) require.Equal(t, AuthorServiceKindSchedule, profile["serviceKind"]) require.Equal(t, "Scope API Schedule", profile["displayName"]) source, ok := metadata["source"].(map[string]any) require.True(t, ok, "reply-thread-root metadata must carry the typed source") require.Equal(t, ContributionSourceKindScheduleRun, source["kind"]) require.Equal(t, ContributionSourceRefTypeAutomationRun, source["refType"]) require.NotContains(t, ownerMemberID, metadata["authorName"]) } func TestScopeAPIReplyThreadRootMetadataOrdinaryAndMalformedSourcesRemainHuman(t *testing.T) { t.Parallel() ctx := context.Background() db := startMessagingTestPool(t, ctx) repos := NewScopeRepositories(db) now := scopeAPINowForTest() workspaceID := "wsp_scope_api_root_ordinary" ownerUserID := "usr_scope_api_root_ordinary" ownerMemberID := "mem_scope_api_root_ordinary" parentScopeID := "scp_scope_api_root_ordinary_parent" rootMessageID := "msg_scope_api_root_ordinary" insertMessagingWorkspace(t, ctx, db, workspaceID, now) insertMessagingUser(t, ctx, db, ownerUserID, "scope-api-root-ordinary@example.com", now) insertMessagingHumanMember(t, ctx, db, ownerMemberID, workspaceID, ownerUserID, now) require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{ ID: parentScopeID, WorkspaceID: workspaceID, Type: "channel", IdentityKey: "channel:scope-api-root-ordinary-parent", LatestSeq: 5, LatestGlobalSeq: 70, LatestMessageAt: timePtr(now.Add(time.Second)), Status: "active", CreatedByMemberID: ownerMemberID, CreatedAt: now, UpdatedAt: now, })) require.NoError(t, repos.Messages.Insert(ctx, ScopeMessageRecord{ ID: rootMessageID, WorkspaceID: workspaceID, ScopeID: parentScopeID, Seq: 5, WorkspaceGlobalSeq: 70, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "ordinary thread root", CreatedAt: now.Add(time.Second), UpdatedAt: now.Add(time.Second), })) api := NewScopeAPIService(New(db, nil, 0, Options{})) // Ordinary Human-owned root without a source ledger keeps its Human author. ordinary, err := api.replyThreadRootMetadataFromVisibleRef(ctx, workspaceID, VisibleMessageRef{ Message: ScopeMessageRecord{ ID: rootMessageID, WorkspaceID: workspaceID, ScopeID: parentScopeID, Seq: 5, WorkspaceGlobalSeq: 70, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "ordinary thread root", CreatedAt: now.Add(time.Second), UpdatedAt: now.Add(time.Second), }, }) require.NoError(t, err) require.Equal(t, ScopeActorHuman, ordinary["authorType"], "ordinary root must keep the Human author type") require.Equal(t, ownerMemberID, ordinary["authorName"], "ordinary root keeps the Human member as author name") if source, ok := ordinary["source"].(map[string]any); ok { require.NotEqual(t, ContributionSourceKindWorkflowRun, source["kind"]) require.NotEqual(t, ContributionSourceKindScheduleRun, source["kind"]) } else { t.Fatalf("ordinary root must carry the default author-based source") } require.NotContains(t, ordinary["authorProfile"], "workflow") // A malformed workflow source (kind workflow_run, wrong ref type) must fail // closed to the original Human author instead of manufacturing automation. require.NoError(t, repos.Messages.Insert(ctx, ScopeMessageRecord{ ID: "msg_scope_api_root_malformed", WorkspaceID: workspaceID, ScopeID: parentScopeID, Seq: 6, WorkspaceGlobalSeq: 71, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "malformed thread root", CreatedAt: now.Add(2 * time.Second), UpdatedAt: now.Add(2 * time.Second), })) // No source ledger is bound; the resolver returns no automation source, so // the overlay must leave the Human author intact (fail closed). malformed, err := api.replyThreadRootMetadataFromVisibleRef(ctx, workspaceID, VisibleMessageRef{ Message: ScopeMessageRecord{ ID: "msg_scope_api_root_malformed", WorkspaceID: workspaceID, ScopeID: parentScopeID, Seq: 6, WorkspaceGlobalSeq: 71, AuthorMemberID: ownerMemberID, AuthorType: ScopeActorHuman, MessageType: "text", Content: "malformed thread root", CreatedAt: now.Add(2 * time.Second), UpdatedAt: now.Add(2 * time.Second), }, }) require.NoError(t, err) require.Equal(t, ScopeActorHuman, malformed["authorType"], "malformed-source root must fail closed to the Human author") require.Equal(t, ownerMemberID, malformed["authorName"]) if source, ok := malformed["source"].(map[string]any); ok { require.NotEqual(t, ContributionSourceKindWorkflowRun, source["kind"], "malformed-source root must not manufacture a workflow source") require.NotEqual(t, ContributionSourceKindScheduleRun, source["kind"]) } else { t.Fatalf("malformed-source root must carry the default author-based source") } }