From f23ccecd07ffd404ace799fa75bf8299c2f47206 Mon Sep 17 00:00:00 2001 From: mintaka Date: Fri, 11 Sep 2026 15:15:42 -0400 Subject: [PATCH] test(comms): deepen supervisor orchestration coverage (RIG-3533) The orchestration file held one case: a supervisor assigning to two workers. Three contracts around it were unasserted, so a regression in any of them would have landed green. Per-topic thread isolation posts four workers' assignments interleaved across topics rather than grouped, because grouped posts read back correctly even when topic scoping is broken. Upward reports pin the author to the worker, not the supervisor or the owner. The membership-scoped audit asserts an outsider gets exactly zero hits, which is the no-existence-oracle contract rather than a cosmetic count. Each case was verified by mutating production and watching it fail: drop the ListMessages topic filter, credit a different valid channel member as author, and widen the search membership join past the caller. The audit case is mutated in the leak direction specifically, since a mutation that returns nothing exercises the positive assertion instead of the negative. The record's fourth case, reparent reflected in coordination membership, is deliberately omitted: TestReparentInEmitsMembershipMove already drives the identical setup and the same both-halves membership assertions, and also asserts the removed_account_ids event, so it is strictly stronger. Co-authored-by: Matt Wilkinson --- .../supervisor_orchestration_pgtest_test.go | 225 ++++++++++++++++++ 1 file changed, 225 insertions(+) diff --git a/go/internal/comms/supervisor_orchestration_pgtest_test.go b/go/internal/comms/supervisor_orchestration_pgtest_test.go index 8bc8d8a28..b456f34ff 100644 --- a/go/internal/comms/supervisor_orchestration_pgtest_test.go +++ b/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 +}