feat(store): TREE subscriptions and delivery over derived participants (RIG-4189)

go/internal/delivery/pre_settle_closure_pgtest_test.go

@@ -119,7 +119,7 @@ func owedTotal(t *testing.T, ctx context.Context, s *store.Store, agent store.Ac
 // deadline — the durable barrier for the crash/agent-authored legs, where the
 // owed row (not a dispatch) is the recovery scan's observable effect. Polls the
 // real store, never a sleep-as-synchronization.
-func waitOwed(t *testing.T, ctx context.Context, s *store.Store, agent store.AccountID, want int) { //nolint:unparam // read-clarity signature: want states the expected owed-row count at each call site; currently constant, not dead code.
+func waitOwed(t *testing.T, ctx context.Context, s *store.Store, agent store.AccountID, want int) {
 	t.Helper()
 	deadline := time.After(testTimeout)
 	tick := time.NewTicker(time.Millisecond)

go/internal/delivery/tree_mention_pgtest_test.go

@@ -0,0 +1,54 @@
+//go:build pgtest && unix
+
+package delivery
+
+import (
+	"context"
+	"testing"
+
+	"github.com/RigelBuild/compass/go/internal/store"
+)
+
+// A TREE channel has no member rows, so mention routing only works if
+// ChannelAgentMembers returns the anchor's subtree. The owed row is the durable
+// effect of a mention routed to an offline agent.
+func TestTreeChannelMentionsRouteToDerivedAgents(t *testing.T) {
+	ctx := context.Background()
+	s := openDeliveryStore(t)
+	owner := mustOwner(t, ctx, s)
+	root := mustAgentAcct(t, ctx, s, owner.ID, "root")
+	mid, err := s.CreateAgent(ctx, owner.ID, store.NewAgent{Handle: "mid", DisplayName: "mid", ParentAgentID: root.ID})
+	if err != nil {
+		t.Fatalf("CreateAgent(mid): %v", err)
+	}
+	leaf, err := s.CreateAgent(ctx, owner.ID, store.NewAgent{Handle: "leaf", DisplayName: "leaf", ParentAgentID: mid.ID})
+	if err != nil {
+		t.Fatalf("CreateAgent(leaf): %v", err)
+	}
+	outsider := mustAgentAcct(t, ctx, s, owner.ID, "outsider")
+	ch, err := s.CreateChannel(ctx, owner.ID, store.NewChannel{
+		Name: "tree-room", Kind: store.ChannelKindChannel,
+		ParentAgentID: root.ID, MembershipMode: store.ChannelMembershipModeTree,
+	})
+	if err != nil {
+		t.Fatalf("CreateChannel(TREE): %v", err)
+	}
+
+	handleMention := postThroughStore(t, ctx, s, ch.ID, root.ID, "@leaf please look")
+	agentsMention := postThroughStore(t, ctx, s, ch.ID, root.ID, "@agents standup")
+
+	c, _, _ := newPgConsumer(t, s)
+	startConsumer(t, c)
+	waitMarked(t, ctx, s, string(handleMention.ID))
+	waitMarked(t, ctx, s, string(agentsMention.ID))
+
+	// leaf: its @handle plus @agents; mid: @agents only; the author and an
+	// agent outside the subtree get nothing.
+	waitOwed(t, ctx, s, leaf.ID, 2)
+	waitOwed(t, ctx, s, mid.ID, 1)
+	for _, a := range []store.Account{root, outsider} {
+		if n := owedTotal(t, ctx, s, a.ID); n != 0 {
+			t.Fatalf("owed mentions for %s = %d; want 0", a.Handle, n)
+		}
+	}
+}

go/internal/store/channel_tree_delivery_pgtest_test.go

@@ -0,0 +1,161 @@
+//go:build pgtest
+
+package store
+
+import (
+	"slices"
+	"strings"
+	"testing"
+)
+
+// subscribeTree toggles account's subscription on a TREE channel, acting as
+// the account itself (any participant may toggle its own override).
+func subscribeTree(t *testing.T, s *Store, account AccountID, channel ChannelID, subscribed bool) {
+	t.Helper()
+	u := MemberUpdate{AccountID: account, Subscribed: subscribed, Unsubscribe: !subscribed}
+	if _, _, err := s.UpdateChannelMembers(t.Context(), account, channel, []MemberUpdate{u}, MemberUpdatesOptions{}); err != nil {
+		t.Fatalf("UpdateChannelMembers(subscribe %v, %s): %v", subscribed, account, err)
+	}
+}
+
+// deliverySites reports, for one agent and channel, the answer of each of
+// the four sites that gate delivery.
+type deliverySites struct {
+	subscribed, sweep, undelivered, inSweep bool
+}
+
+func readDeliverySites(t *testing.T, s *Store, agent AccountID, channel ChannelID, author AccountID) deliverySites {
+	t.Helper()
+	subscribers, err := s.SubscribedAgents(t.Context(), channel, author)
+	if err != nil {
+		t.Fatalf("SubscribedAgents: %v", err)
+	}
+	sweep, err := s.SweepChannels(t.Context(), agent)
+	if err != nil {
+		t.Fatalf("SweepChannels: %v", err)
+	}
+	undelivered, err := s.UndeliveredMessages(t.Context(), agent)
+	if err != nil {
+		t.Fatalf("UndeliveredMessages: %v", err)
+	}
+	inSweep, err := s.InSweepSet(t.Context(), agent, channel)
+	if err != nil {
+		t.Fatalf("InSweepSet: %v", err)
+	}
+	return deliverySites{
+		subscribed:  slices.Contains(subscribers, agent),
+		sweep:       slices.Contains(sweep, channel),
+		undelivered: len(undelivered[channel]) > 0,
+		inSweep:     inSweep,
+	}
+}
+
+func TestChannelTreeSubscriptionDrivesEveryDeliverySite(t *testing.T) {
+	s := newTestStore(t)
+	f := newTreeFixture(t, s)
+
+	if got := readDeliverySites(t, s, f.leaf.ID, f.channel.ID, f.root.ID); got != (deliverySites{}) {
+		t.Fatalf("unsubscribed derived participant sites = %+v; want none (COALESCE default)", got)
+	}
+
+	subscribeTree(t, s, f.leaf.ID, f.channel.ID, true)
+	if _, err := appendAsParticipant(t, s, f.root.ID, f.channel.ID, "after subscribe"); err != nil {
+		t.Fatalf("AppendMessage by anchor: %v", err)
+	}
+	want := deliverySites{subscribed: true, sweep: true, undelivered: true, inSweep: true}
+	if got := readDeliverySites(t, s, f.leaf.ID, f.channel.ID, f.root.ID); got != want {
+		t.Fatalf("subscribed derived participant sites = %+v; want %+v", got, want)
+	}
+	// The anchor and the middle agent never toggled, so the leaf's override
+	// must not leak to them.
+	if got := readDeliverySites(t, s, f.mid.ID, f.channel.ID, f.root.ID); got != (deliverySites{}) {
+		t.Fatalf("untoggled mid agent sites = %+v; want none", got)
+	}
+
+	subscribeTree(t, s, f.leaf.ID, f.channel.ID, false)
+	if got := readDeliverySites(t, s, f.leaf.ID, f.channel.ID, f.root.ID); got != (deliverySites{}) {
+		t.Fatalf("unsubscribed-again participant sites = %+v; want none", got)
+	}
+}
+
+func TestChannelTreeReparentOutMakesOverrideInert(t *testing.T) {
+	s := newTestStore(t)
+	f := newTreeFixture(t, s)
+	subscribeTree(t, s, f.leaf.ID, f.channel.ID, true)
+
+	if _, err := s.ReparentAgent(t.Context(), f.owner.ID, f.leaf.ID, f.sibling.ID); err != nil {
+		t.Fatalf("ReparentAgent(leaf → sibling): %v", err)
+	}
+	if _, err := appendAsParticipant(t, s, f.root.ID, f.channel.ID, "after move"); err != nil {
+		t.Fatalf("AppendMessage by anchor: %v", err)
+	}
+	if got := readDeliverySites(t, s, f.leaf.ID, f.channel.ID, f.root.ID); got != (deliverySites{}) {
+		t.Fatalf("reparented-out agent sites = %+v; want none", got)
+	}
+	members, err := s.ChannelAgentMembers(t.Context(), f.channel.ID, f.root.ID)
+	if err != nil {
+		t.Fatalf("ChannelAgentMembers: %v", err)
+	}
+	if slices.ContainsFunc(members, func(member ChannelAgentMember) bool { return member.ID == f.leaf.ID }) {
+		t.Fatalf("ChannelAgentMembers = %v; want reparented leaf absent", members)
+	}
+	var overrides int
+	if err := s.scopedPool().QueryRow(t.Context(),
+		`SELECT count(*) FROM channel_subscriptions WHERE channel_id = $1 AND account_id = $2`,
+		string(f.channel.ID), string(f.leaf.ID)).Scan(&overrides); err != nil {
+		t.Fatalf("count overrides: %v", err)
+	}
+	if overrides != 1 {
+		t.Fatalf("override rows after reparent = %d; want the row left in place", overrides)
+	}
+}
+
+// ChannelAgentMembers backs both @handle resolution and @agents expansion;
+// only the anchor-to-subtree descent can produce these rows.
+func TestChannelTreeAgentMembersAreTheSubtree(t *testing.T) {
+	s := newTestStore(t)
+	f := newTreeFixture(t, s)
+
+	members, err := s.ChannelAgentMembers(t.Context(), f.channel.ID, f.root.ID)
+	if err != nil {
+		t.Fatalf("ChannelAgentMembers: %v", err)
+	}
+	want := []AccountID{f.mid.ID, f.leaf.ID}
+	slices.Sort(want)
+	got := make([]AccountID, 0, len(members))
+	for _, member := range members {
+		got = append(got, member.ID)
+	}
+	slices.Sort(got)
+	if !slices.Equal(got, want) {
+		t.Fatalf("ChannelAgentMembers(author=root) = %v; want subtree %v without author, owner, or sibling", got, want)
+	}
+}
+
+func TestChannelTreeMembershipWritesRefused(t *testing.T) {
+	s := newTestStore(t)
+	f := newTreeFixture(t, s)
+
+	for name, u := range map[string]MemberUpdate{
+		"add":    {AccountID: f.sibling.ID},
+		"remove": {AccountID: f.leaf.ID, Remove: true},
+	} {
+		_, _, err := s.UpdateChannelMembers(t.Context(), f.owner.ID, f.channel.ID, []MemberUpdate{u}, MemberUpdatesOptions{})
+		sentinelIs(t, err, ErrInvalidArgument, "TREE "+name)
+	}
+	_, _, err := s.UpdateChannelMembers(t.Context(), f.owner.ID, f.channel.ID,
+		[]MemberUpdate{{AccountID: f.sibling.ID, Subscribed: true}}, MemberUpdatesOptions{})
+	sentinelIs(t, err, ErrNotFound, "TREE subscribe of a non-participant")
+
+	_, err = s.SetChannelPolicy(t.Context(), f.owner.ID, f.channel.ID, ChannelPolicy{MandatorySubscription: true})
+	sentinelIs(t, err, ErrInvalidArgument, "TREE mandatory flip")
+	_, err = s.SetChannelPolicy(t.Context(), f.owner.ID, f.channel.ID, ChannelPolicy{
+		PostPolicy: ChannelPostPolicyOwnerOnly, OwnerAccountID: f.owner.ID,
+	})
+	sentinelIs(t, err, ErrInvalidArgument, "TREE OWNER_ONLY")
+	// The owner-is-member check also refuses here, since TREE has no member
+	// rows; the message pins that the TREE guard is the one that fired.
+	if !strings.Contains(err.Error(), "TREE") {
+		t.Fatalf("TREE OWNER_ONLY refused by %v; want the TREE guard", err)
+	}
+}

go/internal/store/channels.go

@@ -688,6 +688,7 @@ func (s *Store) ChannelByNameForViewer(ctx context.Context, viewer AccountID, na
 // Returns the channel with its updated member set, plus the accounts a removal
 // actually deleted (a remove of a non-member deletes nothing and owes no event)
 // so the stream can deliver each departed member its one final ChannelChanged.
+// On a TREE channel only subscription changes are accepted (updateTreeSubscriptions).
 // D9 write-authz is enforced here in the store: the actor must be a member of
 // the channel to mutate it, so an unknown channel and a non-member both return
 // ErrNotFound (the not-found/forbidden merge).
@@ -716,6 +717,16 @@ func (s *Store) UpdateChannelMembers(ctx context.Context, actor AccountID, chann
 	}
 	mandatory := lock.MandatorySubscription
 	kind := ChannelKind(lock.Kind)
+	if ChannelMembershipMode(lock.MembershipMode) == ChannelMembershipModeTree {
+		if err := updateTreeSubscriptions(ctx, tx, channelID, updates); err != nil {
+			return Channel{}, nil, err
+		}
+		if err := tx.Commit(ctx); err != nil {
+			return Channel{}, nil, fmt.Errorf("store: commit tree subscriptions: %w", err)
+		}
+		ch, err := s.getChannel(ctx, channelID)
+		return ch, nil, err
+	}
 	// The unsubscribe guard reads the PRE-convert mandatory state: a DM is
 	// born-mandatory, so an unsubscribe batched with a genuine convert-add is
 	// rejected even though the post-convert channel would permit it. Not
@@ -776,6 +787,43 @@ func (s *Store) UpdateChannelMembers(ctx context.Context, actor AccountID, chann
 	return ch, removed, nil
 }
 
+// updateTreeSubscriptions applies a batch to a TREE channel, whose membership is
+// derived: add/remove are refused, and a subscribe toggle writes the
+// account's override row. Subscribe seeds the D2 cursor in the same tx, as the
+// explicit path does; an account outside the derived set is ErrNotFound.
+func updateTreeSubscriptions(ctx context.Context, tx pgx.Tx, channelID ChannelID, updates []MemberUpdate) error {
+	for _, u := range updates {
+		if u.AccountID == "" {
+			return fmt.Errorf("%w: member update missing account id", ErrInvalidArgument)
+		}
+		if u.Remove || (!u.Subscribed && !u.Unsubscribe) {
+			return fmt.Errorf("%w: TREE channel membership follows the agent tree; only subscription may change", ErrInvalidArgument)
+		}
+	}
+	for _, u := range updates {
+		participant, err := isChannelMember(ctx, tx, u.AccountID, channelID)
+		if err != nil {
+			return err
+		}
+		if !participant {
+			return fmt.Errorf("%w: account %q is not a participant of channel %q", ErrNotFound, u.AccountID, channelID)
+		}
+		if err := db.New(tx).UpsertChannelSubscription(ctx, db.UpsertChannelSubscriptionParams{
+			ChannelID:  string(channelID),
+			AccountID:  string(u.AccountID),
+			Subscribed: u.Subscribed,
+		}); err != nil {
+			return fmt.Errorf("store: upsert channel subscription: %w", err)
+		}
+		if u.Subscribed {
+			if err := seedDeliveryCursor(ctx, tx, u.AccountID, channelID); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 // maybeConvertDM applies the R4 DM-to-CHANNEL conversion when a genuine member
 // ADD targets a kind=DM channel, and returns the channel's kind afterward. A
 // genuine add on a DM MUST supply opts.ConvertChannelName (else
@@ -1020,6 +1068,16 @@ func (s *Store) SetChannelPolicy(ctx context.Context, actor AccountID, channelID
 	}
 	wasMandatory := lock.MandatorySubscription
 	currentOwner := lock.OwnerAccountID
+	// TREE delivery and the owner coherence check are defined over member rows,
+	// which a TREE channel never has; see SeedChannelDeliveryCursors.
+	if ChannelMembershipMode(lock.MembershipMode) == ChannelMembershipModeTree {
+		if p.MandatorySubscription != wasMandatory {
+			return Channel{}, fmt.Errorf("%w: mandatory subscription is not supported on a TREE channel", ErrInvalidArgument)
+		}
+		if p.PostPolicy == ChannelPostPolicyOwnerOnly {
+			return Channel{}, fmt.Errorf("%w: OWNER_ONLY is not supported on a TREE channel", ErrInvalidArgument)
+		}
+	}
 
 	// T4 owner-only policy gate. On an ownerless channel any member may set the
 	// first owner/policy. Once an owner exists, only that owner may change