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 +}