test(comms): deepen supervisor orchestration coverage (RIG-3533)

go/internal/comms/supervisor_orchestration_pgtest_test.go

@@ -27,6 +27,8 @@ package comms
 
 import (
 	"context"
+	"fmt"
+	"slices"
 	"testing"
 
 	"connectrpc.com/connect"
@@ -182,3 +184,226 @@ func assignmentTexts(msgs []*compassv1.Message, wantAuthor string) map[string]bo
 	}
 	return out
 }
+
+// TestSupervisorPerTopicThreadIsolation drives the many-conversations shape: one
+// supervisor fans out to four workers over per-worker topics in ONE channel,
+// posting INTERLEAVED across topics. A topic-scoped ListMessages must return
+// exactly that worker's thread, in order — the interleaving is what makes it
+// bite, since grouped posts would pass even if topic routing were broken.
+func TestSupervisorPerTopicThreadIsolation(t *testing.T) {
+	svc, st := newHandler(t)
+	ctx := context.Background()
+
+	owner := mustUser(t, st, "owner")
+	supervisor := mustAgent(t, st, owner.ID, "supervisor")
+	workers := make([]store.Account, 4)
+	handles := make([]string, 4)
+	for i := range workers {
+		workers[i] = mustAgent(t, st, owner.ID, fmt.Sprintf("worker-%d", i+1))
+		handles[i] = workers[i].Handle
+	}
+
+	created, err := svc.CreateChannel(WithActor(ctx, owner.ID), connect.NewRequest(&compassv1.CreateChannelRequest{
+		Name:          "coordination",
+		Kind:          compassv1.ChannelKind_CHANNEL_KIND_CHANNEL,
+		MemberHandles: append([]string{supervisor.Handle}, handles...),
+	}))
+	if err != nil {
+		t.Fatalf("CreateChannel(coordination): %v", err)
+	}
+	chID := created.Msg.GetChannel().GetId()
+
+	// One topic per worker; capture the server-assigned topic id so the later
+	// ListMessages scopes by id, not by re-resolving a name.
+	topicIDs := make([]string, 4)
+	topicNames := make([]string, 4)
+	for i := range workers {
+		topicNames[i] = fmt.Sprintf("topic-%d", i+1)
+	}
+
+	// Three interleaved rounds: post to topic 1, then 2, 3, 4, then back to 1…
+	// Each worker's thread is its three posts in send order. wantTexts records
+	// the expected per-topic thread as it is posted.
+	const rounds = 3
+	wantTexts := make([][]string, 4)
+	for r := range rounds {
+		for i := range workers {
+			text := fmt.Sprintf("assignment for worker %d round %d", i+1, r+1)
+			posted, err := svc.PostMessage(WithActor(ctx, supervisor.ID), connect.NewRequest(&compassv1.PostMessageRequest{
+				Container:   &compassv1.PostMessageRequest_ChannelId{ChannelId: chID},
+				Topic:       &compassv1.PostMessageRequest_TopicName{TopicName: topicNames[i]},
+				CreateTopic: true,
+				Blocks:      []*compassv1.MessageBlock{{Block: &compassv1.MessageBlock_Text{Text: text}}},
+			}))
+			if err != nil {
+				t.Fatalf("PostMessage(worker %d round %d): %v", i+1, r+1, err)
+			}
+			if topicIDs[i] == "" {
+				topicIDs[i] = posted.Msg.GetMessage().GetTopicId()
+			}
+			wantTexts[i] = append(wantTexts[i], text)
+		}
+	}
+
+	for i, w := range workers {
+		listed, err := svc.ListMessages(WithActor(ctx, w.ID), connect.NewRequest(&compassv1.ListMessagesRequest{
+			Container: &compassv1.ListMessagesRequest_ChannelId{ChannelId: chID},
+			TopicId:   topicIDs[i],
+		}))
+		if err != nil {
+			t.Fatalf("ListMessages(worker %d, topic-scoped): %v", i+1, err)
+		}
+		// ListMessages pages newest-first (messages.sql.go ORDER BY seq DESC), so
+		// reverse the expected thread to send order for the equality check.
+		got := textsInOrder(listed.Msg.GetMessages())
+		want := reversed(wantTexts[i])
+		// mutation: a topic filter that leaks another worker's post (or drops the
+		// TopicId clause) reddens this — the thread would carry a foreign or extra
+		// message, so the exact-slice compare fails.
+		if !slices.Equal(got, want) {
+			t.Fatalf("worker %d topic thread = %v, want exactly its own %v", i+1, got, want)
+		}
+	}
+}
+
+// TestSupervisorUpwardReportKeepsAuthor pins the upward-report direction: a
+// worker posts a report on its own topic, and the report lands on that topic
+// with the WORKER as author — not the supervisor and not the owner. The audit
+// trail must attribute a report to who wrote it.
+func TestSupervisorUpwardReportKeepsAuthor(t *testing.T) {
+	svc, st := newHandler(t)
+	ctx := context.Background()
+
+	owner := mustUser(t, st, "owner")
+	supervisor := mustAgent(t, st, owner.ID, "supervisor")
+	worker := mustAgent(t, st, owner.ID, "worker-1")
+
+	created, err := svc.CreateChannel(WithActor(ctx, owner.ID), connect.NewRequest(&compassv1.CreateChannelRequest{
+		Name:          "coordination",
+		Kind:          compassv1.ChannelKind_CHANNEL_KIND_CHANNEL,
+		MemberHandles: []string{supervisor.Handle, worker.Handle},
+	}))
+	if err != nil {
+		t.Fatalf("CreateChannel(coordination): %v", err)
+	}
+	chID := created.Msg.GetChannel().GetId()
+
+	const report = "worker-1 report: parser landed"
+	posted, err := svc.PostMessage(WithActor(ctx, worker.ID), connect.NewRequest(&compassv1.PostMessageRequest{
+		Container:   &compassv1.PostMessageRequest_ChannelId{ChannelId: chID},
+		Topic:       &compassv1.PostMessageRequest_TopicName{TopicName: "topic-1"},
+		CreateTopic: true,
+		Blocks:      []*compassv1.MessageBlock{{Block: &compassv1.MessageBlock_Text{Text: report}}},
+	}))
+	if err != nil {
+		t.Fatalf("PostMessage(worker report): %v", err)
+	}
+	topicID := posted.Msg.GetMessage().GetTopicId()
+
+	listed, err := svc.ListMessages(WithActor(ctx, supervisor.ID), connect.NewRequest(&compassv1.ListMessagesRequest{
+		Container: &compassv1.ListMessagesRequest_ChannelId{ChannelId: chID},
+		TopicId:   topicID,
+	}))
+	if err != nil {
+		t.Fatalf("ListMessages(supervisor, report topic): %v", err)
+	}
+	if got := len(listed.Msg.GetMessages()); got != 1 {
+		t.Fatalf("report topic carries %d messages, want the single report (1)", got)
+	}
+	msg := listed.Msg.GetMessages()[0]
+	// mutation: PostMessage stamping the actor wrong (e.g. the channel owner or a
+	// hardcoded supervisor) reddens this — a report must be attributed to the
+	// worker that authored it, or the audit trail lies about who reported.
+	if got := msg.GetAuthorAccountId(); got != string(worker.ID) {
+		t.Fatalf("report author = %q, want the worker %q (not supervisor %q / owner %q)", got, worker.ID, supervisor.ID, owner.ID)
+	}
+}
+
+// TestSupervisorAuditScopedAtFourWorkers extends the membership-scoped audit to
+// the four-worker shape: the owner's authorized SearchMessages finds the
+// coordination traffic, while an account NOT in the channel gets ZERO hits for
+// the same term. Zero, not "fewer" — a leaked count would be an existence
+// oracle (D9).
+func TestSupervisorAuditScopedAtFourWorkers(t *testing.T) {
+	svc, st := newHandler(t)
+	ctx := context.Background()
+
+	owner := mustUser(t, st, "owner")
+	supervisor := mustAgent(t, st, owner.ID, "supervisor")
+	handles := make([]string, 4)
+	for i := range handles {
+		handles[i] = mustAgent(t, st, owner.ID, fmt.Sprintf("worker-%d", i+1)).Handle
+	}
+
+	created, err := svc.CreateChannel(WithActor(ctx, owner.ID), connect.NewRequest(&compassv1.CreateChannelRequest{
+		Name:          "coordination",
+		Kind:          compassv1.ChannelKind_CHANNEL_KIND_CHANNEL,
+		MemberHandles: append([]string{supervisor.Handle}, handles...),
+	}))
+	if err != nil {
+		t.Fatalf("CreateChannel(coordination): %v", err)
+	}
+	chID := created.Msg.GetChannel().GetId()
+
+	// One assignment per worker, all sharing the audit word "dispatch".
+	for i := range handles {
+		if _, err := svc.PostMessage(WithActor(ctx, supervisor.ID), connect.NewRequest(&compassv1.PostMessageRequest{
+			Container:   &compassv1.PostMessageRequest_ChannelId{ChannelId: chID},
+			Topic:       &compassv1.PostMessageRequest_TopicName{TopicName: fmt.Sprintf("topic-%d", i+1)},
+			CreateTopic: true,
+			Blocks:      []*compassv1.MessageBlock{{Block: &compassv1.MessageBlock_Text{Text: fmt.Sprintf("dispatch task %d", i+1)}}},
+		})); err != nil {
+			t.Fatalf("PostMessage(assign %d): %v", i+1, err)
+		}
+	}
+
+	auditHits, err := svc.SearchMessages(WithActor(ctx, owner.ID), connect.NewRequest(&compassv1.SearchMessagesRequest{
+		Query: "dispatch",
+	}))
+	if err != nil {
+		t.Fatalf("SearchMessages(owner): %v", err)
+	}
+	// mutation: a search that under-scopes the owner's authorized view drops
+	// coordination traffic the owner is entitled to audit.
+	if got := len(auditHits.Msg.GetMessages()); got != len(handles) {
+		t.Fatalf("owner audit found %d messages, want all four assignments (%d)", got, len(handles))
+	}
+
+	outsider := mustUser(t, st, "outsider")
+	outsiderHits, err := svc.SearchMessages(WithActor(ctx, outsider.ID), connect.NewRequest(&compassv1.SearchMessagesRequest{
+		Query: "dispatch",
+	}))
+	if err != nil {
+		t.Fatalf("SearchMessages(outsider): %v", err)
+	}
+	// mutation: a search that scopes by the query alone instead of the caller's
+	// membership turns the audit into an existence oracle — a non-member would
+	// learn the coordination traffic exists. Must be exactly zero.
+	if got := len(outsiderHits.Msg.GetMessages()); got != 0 {
+		t.Fatalf("outsider audited %d messages in a channel it cannot see, want 0", got)
+	}
+}
+
+// textsInOrder collects the text-block string of each message, in the slice's
+// order, so a caller can assert an exact topic thread.
+func textsInOrder(msgs []*compassv1.Message) []string {
+	out := make([]string, 0, len(msgs))
+	for _, m := range msgs {
+		for _, b := range m.GetBlocks() {
+			if txt, ok := b.GetBlock().(*compassv1.MessageBlock_Text); ok {
+				out = append(out, txt.Text)
+			}
+		}
+	}
+	return out
+}
+
+// reversed returns a copy of in with the order flipped (send order ⇄ the
+// newest-first order ListMessages returns).
+func reversed(in []string) []string {
+	out := make([]string, len(in))
+	for i, s := range in {
+		out[len(in)-1-i] = s
+	}
+	return out
+}