feat(store): RLS enforcement + per-transaction tenant scoping (RIG-3106)

go/internal/delivery/consumer.go

@@ -305,6 +305,15 @@ func (c *Consumer) SetAgentWaker(w AgentWaker) {
 // bus shutdown (end silently). ctx threads from the serve group into every store
 // read and dispatch below; the loop never re-roots it.
 func (c *Consumer) Run(ctx context.Context) error {
+	// N5/OQ-4: the fan-out consumer is a cross-tenant background loop (it tails
+	// EVERY tenant's posted messages, sweeps EVERY agent's owed set, and scans
+	// the whole messages table). It must run under the BYPASSRLS system role, not
+	// the tenant-scoped request path — a request-path (fail-closed) scope would
+	// see zero rows and halt delivery fleet-wide. Marking the root ctx here
+	// propagates the system role into every store call the loop makes (replay,
+	// live dispatch, the settle/start drains, sweeps, and scanMissedMentions),
+	// since the loop threads this ctx and never re-roots it.
+	ctx = store.WithSystemRole(ctx)
 	sub, err := c.bus.Subscribe(0, c.bus.InstanceEpoch())
 	if err != nil {
 		// A fresh subscription at since_seq=0 on a live bus cannot underflow; any

go/internal/runnerhub/hub.go

@@ -746,6 +746,11 @@ func (h *Hub) deliverSession(sessionID string, sf *compassv1internal.SessionFram
 // ack must not kill the Runner's whole event stream. A nil delivery store (a
 // Deliver-only hub) drops the ack silently: no cursor exists to advance.
 func (h *Hub) deliverAck(ctx context.Context, ev RunnerEvent, ack *compassv1internal.DeliveryAck) {
+	// N5/OQ-4: the delivery-ack cursor advance is a cross-tenant system path —
+	// the ack resolves the acking agent's tenant only implicitly, and the cursor
+	// tables are RLS-policied, so this runs under the BYPASSRLS system role
+	// rather than a fail-closed request scope that would drop every ack.
+	ctx = store.WithSystemRole(ctx)
 	h.mu.Lock()
 	delivery := h.delivery
 	h.mu.Unlock()
@@ -802,6 +807,11 @@ func (h *Hub) deliverAck(ctx context.Context, ev RunnerEvent, ack *compassv1inte
 // from the durable gap. A nil delivery store (a Deliver-only hub) drops the ack
 // silently: no cursor exists to advance.
 func (h *Hub) forgeNotificationAck(ctx context.Context, ev RunnerEvent, ack *compassv1internal.ForgeNotificationAck) {
+	// N5/OQ-4: like deliverAck, the forge-delivery cursor advance is a
+	// cross-tenant system path — RLS-policied agent_forge_subscriptions rows
+	// advanced from a Runner event whose tenant is only implicit. Run under the
+	// BYPASSRLS system role rather than a fail-closed request scope.
+	ctx = store.WithSystemRole(ctx)
 	h.mu.Lock()
 	delivery := h.delivery
 	h.mu.Unlock()

go/internal/store/accounts.go

@@ -84,7 +84,7 @@ func (s *Store) CreateUser(ctx context.Context, u NewUser) (Account, error) {
 	}
 
 	id := newID()
-	tx, err := s.pool.Begin(ctx)
+	tx, err := s.beginTenantTx(ctx)
 	if err != nil {
 		return Account{}, fmt.Errorf("store: begin create user: %w", err)
 	}
@@ -143,7 +143,7 @@ func (s *Store) BootstrapAdmin(ctx context.Context, u NewUser) (Account, error)
 	}
 
 	id := newID()
-	tx, err := s.pool.Begin(ctx)
+	tx, err := s.beginTenantTx(ctx)
 	if err != nil {
 		return Account{}, fmt.Errorf("store: begin bootstrap admin: %w", err)
 	}
@@ -238,7 +238,7 @@ func (s *Store) EnsureLinearBridgeAccount(ctx context.Context) (Account, error)
 // fails startup — never silent adoption, mirroring adminByHandle's posture.
 func (s *Store) ensureSystemSubtypeAccount(ctx context.Context, handle, displayName string) (Account, error) {
 	id := newID()
-	tx, err := s.pool.Begin(ctx)
+	tx, err := s.beginTenantTx(ctx)
 	if err != nil {
 		return Account{}, fmt.Errorf("store: begin ensure system account: %w", err)
 	}
@@ -315,7 +315,7 @@ func (s *Store) CreateAgent(ctx context.Context, ownerUserID AccountID, a NewAge
 
 	accountID := newID()
 	channelID := newID()
-	tx, err := s.pool.Begin(ctx)
+	tx, err := s.beginTenantTx(ctx)
 	if err != nil {
 		return Account{}, fmt.Errorf("store: begin create agent: %w", err)
 	}
@@ -542,7 +542,7 @@ func (s *Store) ReparentAgent(ctx context.Context, caller, agentAccountID, newPa
 		return Account{}, fmt.Errorf("%w: agent account id is required", ErrInvalidArgument)
 	}
 
-	tx, err := s.pool.Begin(ctx)
+	tx, err := s.beginTenantTx(ctx)
 	if err != nil {
 		return Account{}, fmt.Errorf("store: begin reparent agent: %w", err)
 	}

go/internal/store/accounts_test.go

@@ -770,15 +770,15 @@ func TestEnsureSystemAccountWrongShapeSquatterConflicts(t *testing.T) {
 			t.Fatalf("insert squatter account: %v", err)
 		}
 		if _, err := s.pool.Exec(ctx,
-			"INSERT INTO user_accounts (account_id, role) VALUES ($1, $2)", id, int32(UserRoleMember),
+			"INSERT INTO user_accounts (account_id, role, tenant_id) VALUES ($1, $2, $3)", id, int32(UserRoleMember), string(s.resolveTenant(ctx)),
 		); err != nil {
 			t.Fatalf("insert squatter user_account: %v", err)
 		}
 		// Plant the reserved handle in the GLOBAL index (owner_user_id NULL): the
 		// squatter owns the resolution key the system seeder contends for.
 		if _, err := s.pool.Exec(ctx,
-			"INSERT INTO account_handles (account_id, handle, owner_user_id) VALUES ($1, $2, NULL)",
-			id, SystemAccountHandle,
+			"INSERT INTO account_handles (account_id, handle, owner_user_id, tenant_id) VALUES ($1, $2, NULL, $3)",
+			id, SystemAccountHandle, string(s.resolveTenant(ctx)),
 		); err != nil {
 			t.Fatalf("insert squatter handle row: %v", err)
 		}
@@ -797,16 +797,16 @@ func TestEnsureSystemAccountWrongShapeSquatterConflicts(t *testing.T) {
 			t.Fatalf("insert squatter account: %v", err)
 		}
 		if _, err := s.pool.Exec(ctx,
-			"INSERT INTO agent_accounts (account_id, owner_user_id) VALUES ($1, $2)", id, string(owner.ID),
+			"INSERT INTO agent_accounts (account_id, owner_user_id, tenant_id) VALUES ($1, $2, $3)", id, string(owner.ID), string(s.resolveTenant(ctx)),
 		); err != nil {
 			t.Fatalf("insert squatter agent_account: %v", err)
 		}
 		// Plant the reserved handle in the GLOBAL index (owner_user_id NULL) so
 		// the squatter actually owns the resolution key the system seeder
 		// contends for — the wrong-shape-row-in-the-global-index threat.
 		if _, err := s.pool.Exec(ctx,
-			"INSERT INTO account_handles (account_id, handle, owner_user_id) VALUES ($1, $2, NULL)",
-			id, SystemAccountHandle,
+			"INSERT INTO account_handles (account_id, handle, owner_user_id, tenant_id) VALUES ($1, $2, NULL, $3)",
+			id, SystemAccountHandle, string(s.resolveTenant(ctx)),
 		); err != nil {
 			t.Fatalf("insert squatter handle row: %v", err)
 		}

go/internal/store/agent_transcripts.go

@@ -326,6 +326,13 @@ func (s *Store) SessionResumeSnapshot(ctx context.Context, sessionID string) ([]
 	// rollback-after-commit convention); on any early return it aborts the tx.
 	defer func() { _ = tx.Rollback(ctx) }()
 
+	// Arm the read-only tx with the tenant scoping (SET LOCAL ROLE + GUC) before
+	// any policied read, exactly as beginTenantTx does for the read-write paths
+	// — a resume snapshot is a request-path read and must be tenant-scoped.
+	if err := armTx(ctx, s, tx); err != nil {
+		return nil, nil, err
+	}
+
 	qtx := s.q.WithTx(tx)
 
 	tailRows, err := qtx.SessionTranscript(ctx, sessionID)
@@ -522,7 +529,7 @@ func (s *Store) flushUpto(ctx context.Context, sessionID string, fromEntrySeq, u
 		return fmt.Errorf("store: flush put segment: %w", err)
 	}
 
-	tx, err := s.pool.Begin(ctx)
+	tx, err := s.beginTenantTx(ctx)
 	if err != nil {
 		return fmt.Errorf("store: begin flush: %w", err)
 	}

go/internal/store/authz.go

@@ -43,7 +43,7 @@ func requireChannelMember(ctx context.Context, q db.DBTX, actor AccountID, chann
 // into an error — the D9 discipline extended from the read RPCs to the live
 // stream (design.md:446-447: the fan-out is visibility-scoped).
 func (s *Store) IsChannelMember(ctx context.Context, actor AccountID, channelID ChannelID) (bool, error) {
-	return isChannelMember(ctx, s.pool, actor, channelID)
+	return isChannelMember(ctx, s.scopedPool(), actor, channelID)
 }
 
 // isChannelMember reports whether actor is a member of channelID (the
@@ -107,7 +107,7 @@ func requireGroupCreateAuthz(ctx context.Context, q db.DBTX, actor AccountID, gr
 // stream edge to filter AgentWorkspaceChanged events, mirroring the
 // OpenAgentWorkspace read gate. An unknown agent yields false (not visible).
 func (s *Store) IsAgentWorkspaceVisible(ctx context.Context, actor AccountID, agentAccountID AccountID) (bool, error) {
-	return isAgentWorkspaceVisible(ctx, s.pool, actor, agentAccountID)
+	return isAgentWorkspaceVisible(ctx, s.scopedPool(), actor, agentAccountID)
 }
 
 // isAgentWorkspaceVisible is the querier-based form IsAgentWorkspaceVisible

go/internal/store/channel_pins.go

@@ -56,7 +56,7 @@ type PinnedEntry struct {
 // A repoint whose msg is already pinned (a duplicate board entry) surfaces the
 // primary-key conflict as ErrConflict.
 func (s *Store) PinMessage(ctx context.Context, ch ChannelID, msg MessageID, replace MessageID, by AccountID) ([]PinnedEntry, error) {
-	tx, err := s.pool.Begin(ctx)
+	tx, err := s.beginTenantTx(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("store: begin pin message: %w", err)
 	}
@@ -111,7 +111,7 @@ func (s *Store) PinMessage(ctx context.Context, ch ChannelID, msg MessageID, rep
 // a no-op: the delete affects zero rows and the unchanged board is returned
 // (never an error), so a duplicate or racing unpin is benign.
 func (s *Store) UnpinMessage(ctx context.Context, ch ChannelID, msg MessageID, by AccountID) ([]PinnedEntry, error) {
-	tx, err := s.pool.Begin(ctx)
+	tx, err := s.beginTenantTx(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("store: begin unpin message: %w", err)
 	}
@@ -149,7 +149,7 @@ func (s *Store) UnpinMessage(ctx context.Context, ch ChannelID, msg MessageID, b
 // view and takes no lock — a snapshot of the board as committed, safe to run
 // against the pool directly.
 func (s *Store) PinnedEntries(ctx context.Context, ch ChannelID) ([]PinnedEntry, error) {
-	return pinnedEntriesTx(ctx, s.pool, ch)
+	return pinnedEntriesTx(ctx, s.scopedPool(), ch)
 }
 
 // lockChannelForPins takes the channels-row FOR UPDATE lock that serializes every

go/internal/store/channels.go

@@ -25,7 +25,7 @@ func (s *Store) CreateChannelGroup(ctx context.Context, ownerUserID AccountID, g
 		return ChannelGroup{}, fmt.Errorf("%w: group name is required", ErrInvalidArgument)
 	}
 
-	tx, err := s.pool.Begin(ctx)
+	tx, err := s.beginTenantTx(ctx)
 	if err != nil {
 		return ChannelGroup{}, fmt.Errorf("store: begin create group: %w", err)
 	}
@@ -104,7 +104,7 @@ func (s *Store) CreateChannel(ctx context.Context, actor AccountID, c NewChannel
 	}
 
 	id := newID()
-	tx, err := s.pool.Begin(ctx)
+	tx, err := s.beginTenantTx(ctx)
 	if err != nil {
 		return Channel{}, fmt.Errorf("store: begin create channel: %w", err)
 	}
@@ -293,7 +293,7 @@ func (s *Store) ListChannels(ctx context.Context, visibleTo AccountID) ([]Channe
 	for _, row := range rows {
 		channels = append(channels, channelFromRow(row.ID, row.Name, row.GroupID, row.Kind, row.PostPolicy, row.OwnerAccountID, row.MandatorySubscription))
 	}
-	if err := loadChannelMembers(ctx, s.pool, channels); err != nil {
+	if err := loadChannelMembers(ctx, s.scopedPool(), channels); err != nil {
 		return nil, err
 	}
 	return channels, nil
@@ -345,7 +345,7 @@ func (s *Store) ChannelByNameForViewer(ctx context.Context, viewer AccountID, na
 	for _, row := range rows {
 		channels = append(channels, channelFromRow(row.ID, row.Name, row.GroupID, row.Kind, row.PostPolicy, row.OwnerAccountID, row.MandatorySubscription))
 	}
-	if err := loadChannelMembers(ctx, s.pool, channels); err != nil {
+	if err := loadChannelMembers(ctx, s.scopedPool(), channels); err != nil {
 		return Channel{}, err
 	}
 	switch len(channels) {
@@ -372,7 +372,7 @@ func (s *Store) ChannelByNameForViewer(ctx context.Context, viewer AccountID, na
 // the channel to mutate it, so an unknown channel and a non-member both return
 // ErrNotFound (the not-found/forbidden merge).
 func (s *Store) UpdateChannelMembers(ctx context.Context, actor AccountID, channelID ChannelID, updates []MemberUpdate, opts MemberUpdatesOptions) (Channel, []AccountID, error) {
-	tx, err := s.pool.Begin(ctx)
+	tx, err := s.beginTenantTx(ctx)
 	if err != nil {
 		return Channel{}, nil, fmt.Errorf("store: begin update members: %w", err)
 	}
@@ -691,7 +691,7 @@ func upsertMemberErr(err error, m AccountID) error {
 //     already-subscribed member whose cursor exists is a no-op and a human
 //     member yields no row.
 func (s *Store) SetChannelPolicy(ctx context.Context, actor AccountID, channelID ChannelID, p ChannelPolicy) (Channel, error) {
-	tx, err := s.pool.Begin(ctx)
+	tx, err := s.beginTenantTx(ctx)
 	if err != nil {
 		return Channel{}, fmt.Errorf("store: begin set channel policy: %w", err)
 	}
@@ -826,7 +826,7 @@ func (s *Store) getChannel(ctx context.Context, id ChannelID) (Channel, error) {
 		return Channel{}, fmt.Errorf("store: get channel: %w", err)
 	}
 	channels := []Channel{channelFromRow(row.ID, row.Name, row.GroupID, row.Kind, row.PostPolicy, row.OwnerAccountID, row.MandatorySubscription)}
-	if err := loadChannelMembers(ctx, s.pool, channels); err != nil {
+	if err := loadChannelMembers(ctx, s.scopedPool(), channels); err != nil {
 		return Channel{}, err
 	}
 	return channels[0], nil
@@ -916,7 +916,7 @@ func (s *Store) OpenAgentWorkspace(ctx context.Context, actor AccountID, agentAc
 		return AgentWorkspace{}, fmt.Errorf("%w: agent account id is required", ErrInvalidArgument)
 	}
 
-	tx, err := s.pool.Begin(ctx)
+	tx, err := s.beginTenantTx(ctx)
 	if err != nil {
 		return AgentWorkspace{}, fmt.Errorf("store: begin open workspace: %w", err)
 	}

go/internal/store/coordination.go

@@ -366,7 +366,7 @@ func LockOwnerCoordinationTx(ctx context.Context, tx pgx.Tx, ownerUserID Account
 // touches the pgx pool directly. fn must confine all its reads/writes to the
 // passed tx.
 func (s *Store) WithTx(ctx context.Context, fn func(pgx.Tx) error) error {
-	tx, err := s.pool.Begin(ctx)
+	tx, err := s.beginTenantTx(ctx)
 	if err != nil {
 		return fmt.Errorf("store: begin tx: %w", err)
 	}

go/internal/store/coordination_pgtest_test.go

@@ -459,10 +459,10 @@ func TestUpsertConcurrentUserInsertSuffixesWithoutWedge(t *testing.T) {
 	}
 	userChID := newID()
 	if _, err := txUser.Exec(ctx,
-		`INSERT INTO channels (id, name, group_id, kind, post_policy, owner_account_id, mandatory_subscription) `+
-			`VALUES ($1, $2, $3, $4, $5, $6, $7)`,
+		`INSERT INTO channels (id, name, group_id, kind, post_policy, owner_account_id, mandatory_subscription, tenant_id) `+
+			`VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`,
 		userChID, "manager-coordination", string(groupID), int32(ChannelKindChannel),
-		int32(ChannelPostPolicyOpen), string(owner.ID), false,
+		int32(ChannelPostPolicyOpen), string(owner.ID), false, string(s.resolveTenant(ctx)),
 	); err != nil {
 		t.Fatalf("user tx insert channel: %v", err)
 	}