fix(runnerhub): scope Runner-originated binding calls to the session's tenant (RIG-4146)

go/internal/runnerhub/binding_cache_test.go

@@ -151,6 +151,15 @@ func (f *fakeBindingStore) EffectiveTenant(context.Context) store.TenantID {
 const testRunnerID = "runner-1"
 
 // seed inserts a binding directly (test setup), bypassing the displacement path.
+func (f *fakeBindingStore) SessionBindingTenant(_ context.Context, sessionID, runnerID string) (store.TenantID, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if b, ok := f.bindings[sessionID]; ok && b.RunnerID == runnerID {
+		return f.tenant, nil
+	}
+	return "", store.ErrNotFound
+}
+
 func (f *fakeBindingStore) seed(sessionID string) {
 	f.mu.Lock()
 	defer f.mu.Unlock()

go/internal/runnerhub/hub.go

@@ -216,6 +216,9 @@ type SessionBindingStore interface {
 	// EffectiveTenant names the tenant a request-scoped call resolves against,
 	// so a post-write BindingChange publishes on that tenant's routing subject.
 	EffectiveTenant(ctx context.Context) store.TenantID
+	// SessionBindingTenant names the tenant binding sessionID to runnerID. The hub
+	// calls it under the system role: the Runner door carries no tenant.
+	SessionBindingTenant(ctx context.Context, sessionID, runnerID string) (store.TenantID, error)
 }
 
 // RoutingFabric is the binding-cache invalidation seam (RIG-3108 / §T4): a
@@ -811,6 +814,11 @@ func (h *Hub) deliverAck(ctx context.Context, ev RunnerEvent, ack *compassv1inte
 	// BEFORE the system-role escalation below. The binding read is single-valued only
 	// because RLS narrows it to one tenant; a BYPASSRLS read could return a row from an
 	// ARBITRARY tenant. Resolving here keeps it tenant-scoped and fail-closed.
+	ctx, scoped := h.runnerSessionCtx(ctx, ev.RunnerID, ev.SessionID)
+	if !scoped {
+		h.countDroppedAck(ev, "acking session is bound in several tenants")
+		return
+	}
 	agent, ok := h.accountForRunnerSession(ctx, ev.RunnerID, ev.SessionID)
 	if !ok {
 		h.countDroppedAck(ev, "no agent account bound to the acking session")
@@ -860,6 +868,11 @@ func (h *Hub) forgeNotificationAck(ctx context.Context, ev RunnerEvent, ack *com
 	// account on the REQUEST ctx, BEFORE the system-role escalation, so the
 	// binding read stays tenant-scoped and cannot return a foreign tenant's row
 	// under BYPASSRLS. Only the cursor advance below runs under the system role.
+	ctx, scoped := h.runnerSessionCtx(ctx, ev.RunnerID, ev.SessionID)
+	if !scoped {
+		h.countDroppedAck(ev, "acking session is bound in several tenants")
+		return
+	}
 	agent, ok := h.accountForRunnerSession(ctx, ev.RunnerID, ev.SessionID)
 	if !ok {
 		h.countDroppedAck(ev, "no agent account bound to the acking session")

go/internal/runnerhub/relay_comms.go

@@ -866,6 +866,10 @@ func commsCallError(err error) *compassv1internal.CommsCallError {
 // unknown, then reports the account so it can be woken. Only the owning Runner's
 // refusal counts, so a foreign Runner cannot unbind another's session.
 func (h *Hub) dropLostSession(ctx context.Context, runnerID, sessionID string) {
+	ctx, scoped := h.runnerSessionCtx(ctx, runnerID, sessionID)
+	if !scoped {
+		return
+	}
 	account, ok := h.accountForRunnerSession(ctx, runnerID, sessionID)
 	if !ok {
 		return
@@ -880,3 +884,34 @@ func (h *Hub) dropLostSession(ctx context.Context, runnerID, sessionID string) {
 		lost.OnSessionLost(sessionID, account)
 	}
 }
+
+// runnerSessionCtx scopes a Runner-originated ctx to the tenant that binds
+// sessionID to runnerID. Runners are shared across tenants, so the door sets none;
+// without this, binding reads and deletes run under the bootstrap tenant. The
+// tenant is read under the system role; everything after runs tenant-scoped.
+// An ambiguous session fails closed. A miss or a store fault keeps ctx, whose
+// own binding read then fails closed for a non-bootstrap session.
+func (h *Hub) runnerSessionCtx(ctx context.Context, runnerID, sessionID string) (context.Context, bool) {
+	if _, ok := store.TenantFromContext(ctx); ok {
+		return ctx, true
+	}
+	h.mu.Lock()
+	bindings := h.bindings
+	h.mu.Unlock()
+	if bindings == nil || runnerID == "" || sessionID == "" {
+		return ctx, true
+	}
+	tenant, err := bindings.SessionBindingTenant(store.WithSystemRole(ctx), sessionID, runnerID)
+	switch {
+	case err == nil:
+		return store.WithTenant(ctx, tenant), true
+	case errors.Is(err, store.ErrConflict):
+		h.log.Error("session bound in several tenants; refusing the runner call",
+			"session_id", sessionID, "runner_id", runnerID)
+		return ctx, false
+	case !errors.Is(err, store.ErrNotFound):
+		h.log.Warn("resolve session tenant failed; continuing unscoped",
+			"session_id", sessionID, "error", err)
+	}
+	return ctx, true
+}

go/internal/runnerhub/runner_tenant_pgtest_test.go

@@ -0,0 +1,84 @@
+//go:build pgtest && unix
+
+package runnerhub
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+
+	compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1"
+	"github.com/RigelBuild/compass/go/internal/pgtest"
+	"github.com/RigelBuild/compass/go/internal/store"
+)
+
+// A Runner is shared across tenants and its door carries none. A lost session of a
+// non-bootstrap tenant, read cold from the table, must still be released durably and
+// woken: the hub resolves the session's tenant before it reads or deletes.
+func TestDropLostSessionScopesToTheSessionTenant(t *testing.T) {
+	ctx := context.Background()
+	dsn := pgtest.RequireDSN(t)
+	st, err := store.Open(ctx, dsn)
+	if err != nil {
+		t.Fatalf("store.Open: %v", err)
+	}
+	t.Cleanup(st.Close)
+	if _, err := st.BootstrapAdmin(ctx, store.NewUser{Handle: "admin", DisplayName: "admin"}); err != nil {
+		t.Fatalf("BootstrapAdmin: %v", err)
+	}
+
+	tenantB := seedTenantRow(t, ctx, dsn, "tenant-b")
+	ctxB := store.WithTenant(ctx, tenantB)
+	owner, err := st.CreateUser(ctxB, store.NewUser{Handle: "owner-b", DisplayName: "owner-b"})
+	if err != nil {
+		t.Fatalf("CreateUser: %v", err)
+	}
+	agent, err := st.CreateAgent(ctxB, owner.ID, store.NewAgent{Handle: "agent-b", DisplayName: "agent-b"})
+	if err != nil {
+		t.Fatalf("CreateAgent: %v", err)
+	}
+	if _, err := st.RecordSessionBinding(ctxB, "sess-b", agent.ID, "runner-1"); err != nil {
+		t.Fatalf("RecordSessionBinding: %v", err)
+	}
+
+	// A fresh hub: its cache is cold, as after a Server restart.
+	hub := newHubOnly()
+	hub.SetSessionBindingStore(st)
+	sink := &recordingLostSink{}
+	hub.SetSessionLostSink(sink)
+	hub.enroll(ctx, "runner-1", runnerSubject(), compassv1.RuntimeTier_RUNTIME_TIER_UNSPECIFIED, compassv1.EgressPosture_EGRESS_POSTURE_UNSPECIFIED)
+	// enroll's reap ran unscoped and cannot see tenant B's row; re-record it so the
+	// test does not depend on that.
+	if _, err := st.RecordSessionBinding(ctxB, "sess-b", agent.ID, "runner-1"); err != nil {
+		t.Fatalf("RecordSessionBinding after enroll: %v", err)
+	}
+
+	hub.dropLostSession(ctx, "runner-1", "sess-b")
+
+	if len(sink.lost) != 1 || sink.lost[0] != agent.ID {
+		t.Fatalf("lost = %v, want [%s]: the tenant-B session was not resolved, so no wake", sink.lost, agent.ID)
+	}
+	if _, _, err := st.ResolveSessionBinding(ctxB, "sess-b"); err == nil {
+		t.Fatal("tenant B's durable binding survived the drop; a cache miss would resurrect it")
+	}
+}
+
+// seedTenantRow inserts a non-bootstrap tenant. tenants is RLS-exempt, so a plain
+// connection may write it.
+func seedTenantRow(t *testing.T, ctx context.Context, dsn, slug string) store.TenantID {
+	t.Helper()
+	conn, err := pgx.Connect(ctx, dsn)
+	if err != nil {
+		t.Fatalf("pgx.Connect: %v", err)
+	}
+	defer func() { _ = conn.Close(ctx) }() // test cleanup; nothing actionable remains
+	id := slug + "-id"
+	if _, err := conn.Exec(ctx,
+		"INSERT INTO tenants (id, slug, display_name, created_at_unix_ms) VALUES ($1, $2, $3, $4)",
+		id, slug, slug, time.Now().UnixMilli()); err != nil {
+		t.Fatalf("seed tenant %q: %v", slug, err)
+	}
+	return store.TenantID(id)
+}

go/internal/store/queries/session_bindings.sql

@@ -140,3 +140,9 @@ DELETE FROM session_bindings WHERE session_id = $1;
 DELETE FROM session_bindings
  WHERE runner_id = $1
 RETURNING session_id, agent_account_id;
+
+-- The one query here meant for the system role: a Runner-originated call carries
+-- no tenant, so the hub reads the session's tenant cross-tenant, then acts under it.
+-- :many so a session id minted in two tenants is refused, not resolved arbitrarily.
+-- name: SessionBindingTenants :many
+SELECT tenant_id FROM session_bindings WHERE session_id = $1 AND runner_id = $2;

go/internal/store/session_bindings.go

@@ -276,3 +276,26 @@ func (s *Store) DeleteSessionBindingsForRunner(ctx context.Context, runnerID str
 	})
 	return bindings, nil
 }
+
+// SessionBindingTenant names the tenant whose binding maps sessionID to runnerID.
+// It is the system-role read a Runner-originated call needs first: the Runner
+// door carries no tenant, so the hub resolves it here, then acts under WithTenant.
+// ErrNotFound when no tenant binds the pair; ErrConflict when several do, so an
+// ambiguous session never resolves to an arbitrary tenant.
+func (s *Store) SessionBindingTenant(ctx context.Context, sessionID, runnerID string) (TenantID, error) {
+	if sessionID == "" || runnerID == "" {
+		return "", fmt.Errorf("%w: session id and runner id are required", ErrInvalidArgument)
+	}
+	tenants, err := s.q.SessionBindingTenants(ctx, db.SessionBindingTenantsParams{SessionID: sessionID, RunnerID: runnerID})
+	if err != nil {
+		return "", fmt.Errorf("store: resolve session binding tenant: %w", err)
+	}
+	switch len(tenants) {
+	case 0:
+		return "", fmt.Errorf("%w: session %q is not bound to runner %q", ErrNotFound, sessionID, runnerID)
+	case 1:
+		return TenantID(tenants[0]), nil
+	default:
+		return "", fmt.Errorf("%w: session %q is bound in %d tenants", ErrConflict, sessionID, len(tenants))
+	}
+}

go/internal/store/session_bindings_pgtest_test.go

@@ -1116,3 +1116,32 @@ func TestRecordSessionBindingReportsDisplacedExactlyOnceAgainstARunnerSweep(t *t
 	}
 	t.Logf("%d iterations: the displaced session was reported to exactly one caller every time", iterations)
 }
+
+// TestSessionBindingTenantResolvesAcrossTenantsUnderSystemRole pins the read a
+// Runner-originated call scopes itself with: under the system role it names the
+// binding's tenant, a session minted in two tenants on one runner is a conflict,
+// and a pair nobody binds is not-found.
+func TestSessionBindingTenantResolvesAcrossTenantsUnderSystemRole(t *testing.T) {
+	s := newTestStore(t)
+	tenantB := seedTenant(t, s, "tenant-b")
+	ctxA := context.Background()
+	ctxB := WithTenant(context.Background(), tenantB)
+	sys := WithSystemRole(context.Background())
+
+	agentA := mustAgent(t, s, mustUser(t, s, "owner-a").ID, "agent-a")
+	agentB := seedTenantAgent(t, ctxB, s, "agent-b")
+	mustBind(t, ctxB, s, "sess-b", agentB.ID, "runner-1")
+
+	got, err := s.SessionBindingTenant(sys, "sess-b", "runner-1")
+	if err != nil || got != tenantB {
+		t.Fatalf("SessionBindingTenant(sess-b) = (%q, %v), want (%q, nil)", got, err, tenantB)
+	}
+	if _, err := s.SessionBindingTenant(sys, "sess-b", "runner-2"); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("SessionBindingTenant(sess-b, foreign runner) err = %v, want ErrNotFound", err)
+	}
+
+	mustBind(t, ctxA, s, "sess-b", agentA.ID, "runner-1")
+	if _, err := s.SessionBindingTenant(sys, "sess-b", "runner-1"); !errors.Is(err, ErrConflict) {
+		t.Fatalf("SessionBindingTenant(sess-b bound in two tenants) err = %v, want ErrConflict", err)
+	}
+}