refactor(comms): set-based delivery-cursor seed for mandatory channels

go/internal/store/channel_policy_pgtest_test.go

@@ -197,21 +197,66 @@ func TestSetChannelPolicySeedsCursorsForNewlyMandatory(t *testing.T) {
 	}
 }
 
+// TestSetChannelPolicyMandatoryDoesNotSeedHumanMember pins the set-based seed's
+// agent-only guard (the JOIN agent_accounts): flipping mandatory on a channel
+// with both a human and an agent member seeds ONLY the agent — the human member
+// gets no cursor row. This fails if the JOIN were dropped or loosened to seed
+// every member.
+func TestSetChannelPolicyMandatoryDoesNotSeedHumanMember(t *testing.T) {
+	ctx := context.Background()
+	s := newTestStore(t)
+	owner := mustUser(t, s, "owner")
+	agent := mustAgent(t, s, owner.ID, "a1")
+	human := mustUser(t, s, "human")
+	// A non-mandatory channel with one agent and one human member, neither
+	// subscribed — so neither has a delivery cursor row yet.
+	ch := mustPolicyChannel(t, s, owner.ID, "coord", ChannelPolicy{}, agent.ID, human.ID)
+	unsubscribeMember(t, s, ch.ID, agent.ID)
+	unsubscribeMember(t, s, ch.ID, human.ID)
+
+	// Precondition: no cursor rows exist for either member on this channel.
+	for _, a := range []AccountID{agent.ID, human.ID} {
+		if _, _, ok := readCursor(t, s, a, ch.ID); ok {
+			t.Fatalf("precondition: member %s already has a cursor on %s", a, ch.ID)
+		}
+	}
+
+	if _, err := s.SetChannelPolicy(ctx, owner.ID, ch.ID, ChannelPolicy{
+		MandatorySubscription: true,
+	}); err != nil {
+		t.Fatalf("SetChannelPolicy: %v", err)
+	}
+
+	// The agent member is seeded (a delivery target); the human member is NOT —
+	// the agent-only JOIN admits no human_account row.
+	if _, _, ok := readCursor(t, s, agent.ID, ch.ID); !ok {
+		t.Fatalf("agent %s has no cursor after mandatory flip — an un-seeded delivery target", agent.ID)
+	}
+	if _, _, ok := readCursor(t, s, human.ID, ch.ID); ok {
+		t.Fatalf("human %s got a cursor after mandatory flip — the agent-only JOIN was bypassed", human.ID)
+	}
+}
+
 // TestCreateChannelBornMandatorySeedsCursors pins the create-with-policy path's
 // D2-hazard closure: a channel created with Policy.MandatorySubscription=true
 // makes every member a delivery target (D1 disjunct), so CreateChannel MUST seed
 // each agent member's cursor in the create txn — else the channel is born with
-// un-seeded delivery targets. Symmetric with the SetChannelPolicy flip.
+// un-seeded delivery targets — and a human member gets none (the seed is
+// agent-only). Symmetric with the SetChannelPolicy flip and its human test.
 func TestCreateChannelBornMandatorySeedsCursors(t *testing.T) {
 	ctx := context.Background()
 	s := newTestStore(t)
 	owner := mustUser(t, s, "owner")
 	a1 := mustAgent(t, s, owner.ID, "a1")
 	a2 := mustAgent(t, s, owner.ID, "a2")
+	// A human member proves the born-mandatory seed is agent-only on the create
+	// path too (symmetric with the SetChannelPolicy human test): the JOIN
+	// agent_accounts admits no human row.
+	human := mustUser(t, s, "human")
 
 	ch := mustPolicyChannel(t, s, owner.ID, "coord", ChannelPolicy{
 		MandatorySubscription: true,
-	}, a1.ID, a2.ID)
+	}, a1.ID, a2.ID, human.ID)
 
 	// Every agent member has a seeded cursor from birth (at head 0, an empty
 	// channel) and is caught-up (owed nothing) — no un-seeded delivery target,
@@ -233,6 +278,12 @@ func TestCreateChannelBornMandatorySeedsCursors(t *testing.T) {
 			t.Fatalf("agent %s owed %d messages after born-mandatory seed, want 0", a, len(owed[ch.ID]))
 		}
 	}
+
+	// The human member is NOT a delivery target — the agent-only JOIN admits no
+	// human row, so it has no cursor (symmetric with the SetChannelPolicy path).
+	if _, _, ok := readCursor(t, s, human.ID, ch.ID); ok {
+		t.Fatalf("human %s got a cursor after born-mandatory create — the agent-only JOIN was bypassed", human.ID)
+	}
 }
 
 // TestSetChannelPolicyNonOwnerOnOwnedChannelIsNotFound pins Matt's owner-only

go/internal/store/channels.go

@@ -165,16 +165,16 @@ func (s *Store) CreateChannel(ctx context.Context, actor AccountID, c NewChannel
 	// agent member's delivery cursor MUST be seeded in this same tx — an
 	// un-seeded delivery target is the fail-DANGEROUS D2 hazard
 	// (compass-notification-delivery/design.md:293-311). Symmetric with
-	// SetChannelPolicy's newly-mandatory seed. seedDeliveryCursor is
-	// self-guarding (agent-only) and idempotent, so seeding every member is safe
-	// (a human member is a no-op). A non-mandatory channel seeds nothing here —
-	// its members seed at subscribe time (addOrUpdateMember), the pre-substrate
-	// behavior.
+	// SetChannelPolicy's newly-mandatory seed. One set-based statement seeds
+	// every agent member of the channel; it is self-guarding (agent-only) and
+	// idempotent, so a human member is a no-op. The member INSERTs above have
+	// already landed in this tx's snapshot, so the statement's channel_members
+	// read sees exactly this channel's member set. A non-mandatory channel seeds
+	// nothing here — its members seed at subscribe time (addOrUpdateMember), the
+	// pre-substrate behavior.
 	if c.Policy.MandatorySubscription {
-		for _, m := range members {
-			if err := seedDeliveryCursor(ctx, tx, m, ChannelID(id)); err != nil {
-				return Channel{}, err
-			}
+		if err := seedChannelDeliveryCursors(ctx, tx, ChannelID(id)); err != nil {
+			return Channel{}, err
 		}
 	}
 	if err := tx.Commit(ctx); err != nil {
@@ -692,32 +692,13 @@ func (s *Store) SetChannelPolicy(ctx context.Context, actor AccountID, channelID
 	}
 
 	// Newly-mandatory: every member becomes a delivery target, so seed each
-	// agent member's cursor in this same txn. seedDeliveryCursor is self-guarding
-	// (agent-only) and idempotent, so seeding every member is safe.
+	// agent member's cursor in this same txn — an un-seeded delivery target is
+	// the fail-DANGEROUS D2 hazard. One set-based statement seeds every agent
+	// member of the channel; it is self-guarding (agent-only) and idempotent, so
+	// seeding across the whole member set is safe (a human member is a no-op).
 	if p.MandatorySubscription && !wasMandatory {
-		rows, err := tx.Query(ctx,
-			"SELECT account_id FROM channel_members WHERE channel_id = $1", string(channelID))
-		if err != nil {
-			return Channel{}, fmt.Errorf("store: list channel members for seed: %w", err)
-		}
-		var members []AccountID
-		for rows.Next() {
-			var m string
-			if err := rows.Scan(&m); err != nil {
-				rows.Close()
-				return Channel{}, fmt.Errorf("store: scan member for seed: %w", err)
-			}
-			members = append(members, AccountID(m))
-		}
-		if err := rows.Err(); err != nil {
-			rows.Close()
-			return Channel{}, fmt.Errorf("store: iterate members for seed: %w", err)
-		}
-		rows.Close()
-		for _, m := range members {
-			if err := seedDeliveryCursor(ctx, tx, m, channelID); err != nil {
-				return Channel{}, err
-			}
+		if err := seedChannelDeliveryCursors(ctx, tx, channelID); err != nil {
+			return Channel{}, err
 		}
 	}
 

go/internal/store/delivery_cursors.go

@@ -35,6 +35,36 @@ func seedDeliveryCursor(ctx context.Context, tx pgx.Tx, agent AccountID, channel
 	return nil
 }
 
+// seedChannelDeliveryCursorsSQL seeds EVERY agent member of the channel to the
+// current channel head in one statement — MAX(seq) over the channel's messages,
+// 0 if empty, a same-channel constant across all seeded rows — with NO history
+// replay (design record D2). The JOIN agent_accounts is the agent-only guard
+// (the set form of the per-row WHERE EXISTS in seedDeliveryCursorSQL): a human
+// member has no agent_accounts row and so is a silent no-op rather than an FK
+// violation. ON CONFLICT DO NOTHING keeps a re-subscribe / re-run idempotent —
+// it never resets an existing cursor. $1 is the channel id.
+const seedChannelDeliveryCursorsSQL = `
+	INSERT INTO agent_delivery_cursors (agent_account_id, channel_id, acked_seq)
+	SELECT cm.account_id, $1,
+	       COALESCE((SELECT MAX(m.seq) FROM messages m JOIN topics t ON t.id = m.topic_id WHERE t.channel_id = $1), 0)
+	FROM channel_members cm
+	JOIN agent_accounts aa ON aa.account_id = cm.account_id
+	WHERE cm.channel_id = $1
+	ON CONFLICT (agent_account_id, channel_id) DO NOTHING`
+
+// seedChannelDeliveryCursors is the set-based counterpart to seedDeliveryCursor:
+// it seeds all agent members of the channel in a single statement (collapsing the
+// per-member seed loop), riding the caller's existing transaction so a missed
+// seed is a loud failure in that same commit. Self-guarding (agent-only, see
+// seedChannelDeliveryCursorsSQL) and idempotent, so it is safe to call for a
+// channel whose member set includes humans.
+func seedChannelDeliveryCursors(ctx context.Context, tx pgx.Tx, channel ChannelID) error {
+	if _, err := tx.Exec(ctx, seedChannelDeliveryCursorsSQL, string(channel)); err != nil {
+		return fmt.Errorf("store: seed channel delivery cursors: %w", err)
+	}
+	return nil
+}
+
 // SeedDeliveryCursor seeds acked_seq to the current channel head (MAX(seq) over
 // the channel's messages, 0 if empty) — NO history replay. It MUST be called in
 // the SAME txn as the channel_members row insert (the seed rides that commit, so