diff --git a/config/config.go b/config/config.go
index f23d0a3a..16902e85 100644
--- a/config/config.go
+++ b/config/config.go
@@ -2,6 +2,7 @@ package config
 
 import (
 	"bytes"
+	"fmt"
 	"maps"
 	"os"
 
@@ -242,6 +243,35 @@ func (s *SATranslationConfig) IsEnabled() bool {
 	return len(s.NamespaceMappings) > 0
 }
 
+// Validate reports the first reason the configured namespace mappings cannot be applied per
+// namespace: search attributes are translated by namespaceId, so every mapping needs an id that
+// is present and unique.
+func (s *SATranslationConfig) Validate() error {
+	namesByNamespaceId := make(map[string]string, len(s.NamespaceMappings))
+	seenNames := make(map[string]struct{}, len(s.NamespaceMappings))
+	for _, m := range s.NamespaceMappings {
+		// Reported before the duplicate check so that a config with several mappings and no ids
+		// gets the actionable message rather than `duplicate namespaceId ""`.
+		if m.NamespaceId == "" {
+			return fmt.Errorf("searchAttributeTranslation: namespaceMappings[name=%q] has no namespaceId; search attributes are translated per namespace so namespaceId is required", m.Name)
+		}
+		if existing, found := namesByNamespaceId[m.NamespaceId]; found {
+			return fmt.Errorf("searchAttributeTranslation: namespaceMappings[name=%q] and namespaceMappings[name=%q] have duplicate namespaceId %q", existing, m.Name, m.NamespaceId)
+		}
+		namesByNamespaceId[m.NamespaceId] = m.Name
+
+		// The name is informational, so only reject duplicates of a name that was actually set.
+		if m.Name == "" {
+			continue
+		}
+		if _, found := seenNames[m.Name]; found {
+			return fmt.Errorf("searchAttributeTranslation: namespaceMappings has duplicate name %q", m.Name)
+		}
+		seenNames[m.Name] = struct{}{}
+	}
+	return nil
+}
+
 // ToMaps returns request and response mappings.
 func (s *SATranslationConfig) ToMaps(inBound bool) (map[string]map[string]string, map[string]map[string]string) {
 	reqMap := make(map[string]map[string]string)
@@ -330,12 +360,15 @@ func (s *SATranslationConfig) AsLocalToRemoteSATranslation() (SearchAttributeTra
 	if s.cachedBiMap.inner != nil {
 		return s.cachedBiMap, nil
 	}
+	// This is the only path that builds the translation, so it is where the config is checked.
+	if err := s.Validate(); err != nil {
+		return SearchAttributeTranslation{}, err
+	}
 	saTranslation := SearchAttributeTranslation{
 		inner: make(map[string]collect.StaticBiMap[string, string], len(s.NamespaceMappings)),
 	}
 	for _, mapping := range s.NamespaceMappings {
-		var err error
-		saTranslation.inner[mapping.NamespaceId], err = collect.NewStaticBiMap(func(yield func(string, string) bool) {
+		nsBiMap, err := collect.NewStaticBiMap(func(yield func(string, string) bool) {
 			for _, attrPair := range mapping.Mappings {
 				if !yield(attrPair.LocalName, attrPair.RemoteName) {
 					return
@@ -343,8 +376,11 @@ func (s *SATranslationConfig) AsLocalToRemoteSATranslation() (SearchAttributeTra
 			}
 		}, len(mapping.Mappings))
 		if err != nil {
-			return SearchAttributeTranslation{}, err
+			return SearchAttributeTranslation{}, fmt.Errorf(
+				"searchAttributeTranslation: namespaceMappings[name=%q namespaceId=%q]: %w",
+				mapping.Name, mapping.NamespaceId, err)
 		}
+		saTranslation.inner[mapping.NamespaceId] = nsBiMap
 	}
 	s.cachedBiMap = saTranslation
 	return saTranslation, nil
diff --git a/config/config_test.go b/config/config_test.go
index f42ce17d..af22b06e 100644
--- a/config/config_test.go
+++ b/config/config_test.go
@@ -322,3 +322,230 @@ func TestExampleChart(t *testing.T) {
 	require.Equal(t, ConnectionType("mux-client"), cc.Remote.ConnectionType)
 	require.Equal(t, "s2s-proxy-sample.example.tmprl.cloud:8233", cc.Remote.MuxAddressInfo.ConnectionString)
 }
+
+func TestSATranslationConfigValidate(t *testing.T) {
+	cases := []struct {
+		name string
+		cfg  SATranslationConfig
+		// wantValidateErr lists substrings the Validate error must contain. Empty means the
+		// config is valid.
+		wantValidateErr []string
+		// wantValidateErrExcludes lists substrings the Validate error must not contain.
+		wantValidateErrExcludes []string
+		// wantTranslationErr lists substrings the AsLocalToRemoteSATranslation error must
+		// contain. Leave nil when the translation fails for the same reason as Validate.
+		wantTranslationErr []string
+		// verify runs against the built translation when it is expected to succeed.
+		verify func(t *testing.T, saTranslation SearchAttributeTranslation)
+	}{
+		{
+			name: "no namespace mappings",
+			cfg:  SATranslationConfig{},
+			verify: func(t *testing.T, saTranslation SearchAttributeTranslation) {
+				require.Equal(t, 0, saTranslation.LenNamespaces())
+			},
+		},
+		{
+			// Two namespaces sharing an id used to silently overwrite each other, leaving one
+			// namespace translated with the other namespace's mappings.
+			name: "duplicate namespaceId",
+			cfg: SATranslationConfig{
+				NamespaceMappings: []SANamespaceMapping{
+					{
+						Name:        "namespace1",
+						NamespaceId: "namespace-id-1",
+						Mappings:    []SAMapping{{LocalName: "localOne", RemoteName: "remoteOne"}},
+					},
+					{
+						Name:        "namespace2",
+						NamespaceId: "namespace-id-1",
+						Mappings:    []SAMapping{{LocalName: "localTwo", RemoteName: "remoteTwo"}},
+					},
+				},
+			},
+			wantValidateErr: []string{
+				`namespaceMappings[name="namespace1"]`,
+				`namespaceMappings[name="namespace2"]`,
+				`duplicate namespaceId "namespace-id-1"`,
+			},
+		},
+		{
+			name: "empty namespaceId alongside another namespace",
+			cfg: SATranslationConfig{
+				NamespaceMappings: []SANamespaceMapping{
+					{
+						Name:     "legacyNamespace",
+						Mappings: []SAMapping{{LocalName: "localOne", RemoteName: "remoteOne"}},
+					},
+					{
+						Name:        "namespace2",
+						NamespaceId: "namespace-id-2",
+						Mappings:    []SAMapping{{LocalName: "localTwo", RemoteName: "remoteTwo"}},
+					},
+				},
+			},
+			wantValidateErr: []string{
+				`namespaceMappings[name="legacyNamespace"]`,
+				"has no namespaceId",
+			},
+		},
+		{
+			// The missing id is the actionable problem, so it is reported ahead of the duplicate
+			// id the empty values also form.
+			name: "every mapping missing its namespaceId",
+			cfg: SATranslationConfig{
+				NamespaceMappings: []SANamespaceMapping{
+					{
+						Name:     "namespace1",
+						Mappings: []SAMapping{{LocalName: "localOne", RemoteName: "remoteOne"}},
+					},
+					{
+						Name:     "namespace2",
+						Mappings: []SAMapping{{LocalName: "localTwo", RemoteName: "remoteTwo"}},
+					},
+				},
+			},
+			wantValidateErr: []string{
+				`namespaceMappings[name="namespace1"]`,
+				"has no namespaceId",
+			},
+			wantValidateErrExcludes: []string{"duplicate namespaceId"},
+		},
+		{
+			// namespaceId is required even for one namespace: there is no mapping that applies
+			// to every namespace, so an omitted id has nothing to match against.
+			name: "single mapping with empty namespaceId",
+			cfg: SATranslationConfig{
+				NamespaceMappings: []SANamespaceMapping{
+					{
+						Mappings: []SAMapping{{LocalName: "localOne", RemoteName: "remoteOne"}},
+					},
+				},
+			},
+			wantValidateErr: []string{"has no namespaceId", "namespaceId is required"},
+		},
+		{
+			// The shape the migration tooling emits today: the namespace is named but the
+			// namespaceId is blank. It must fail at startup naming the entry, rather than
+			// translating some arbitrary namespace.
+			name: "named mapping with empty namespaceId",
+			cfg: SATranslationConfig{
+				NamespaceMappings: []SANamespaceMapping{
+					{
+						Name: "migration-namespace",
+						Mappings: []SAMapping{
+							{LocalName: "CustomKeywordField", RemoteName: "Keyword01"},
+							{LocalName: "CustomStringField", RemoteName: "Text01"},
+						},
+					},
+				},
+			},
+			wantValidateErr: []string{
+				`namespaceMappings[name="migration-namespace"]`,
+				"has no namespaceId",
+			},
+		},
+		{
+			// The namespace is well formed, the mappings inside it are not: the bimap rejects
+			// them and the error must say which namespace it came from.
+			name: "duplicate localFieldName within one namespace",
+			cfg: SATranslationConfig{
+				NamespaceMappings: []SANamespaceMapping{
+					{
+						Name:        "namespace1",
+						NamespaceId: "namespace-id-1",
+						Mappings: []SAMapping{
+							{LocalName: "localOne", RemoteName: "remoteOne"},
+							{LocalName: "localOne", RemoteName: "remoteTwo"},
+						},
+					},
+				},
+			},
+			wantTranslationErr: []string{
+				`namespaceMappings[name="namespace1" namespaceId="namespace-id-1"]`,
+			},
+		},
+		{
+			name: "duplicate name across namespaces",
+			cfg: SATranslationConfig{
+				NamespaceMappings: []SANamespaceMapping{
+					{
+						Name:        "namespace1",
+						NamespaceId: "namespace-id-1",
+						Mappings:    []SAMapping{{LocalName: "localOne", RemoteName: "remoteOne"}},
+					},
+					{
+						Name:        "namespace1",
+						NamespaceId: "namespace-id-2",
+						Mappings:    []SAMapping{{LocalName: "localTwo", RemoteName: "remoteTwo"}},
+					},
+				},
+			},
+			wantValidateErr: []string{`duplicate name "namespace1"`},
+		},
+		{
+			name: "distinct namespaceIds translate independently",
+			cfg: SATranslationConfig{
+				NamespaceMappings: []SANamespaceMapping{
+					{
+						Name:        "namespace1",
+						NamespaceId: "namespace-id-1",
+						Mappings:    []SAMapping{{LocalName: "localOne", RemoteName: "remoteOne"}},
+					},
+					{
+						Name:        "namespace2",
+						NamespaceId: "namespace-id-2",
+						Mappings:    []SAMapping{{LocalName: "localTwo", RemoteName: "remoteTwo"}},
+					},
+				},
+			},
+			verify: func(t *testing.T, saTranslation SearchAttributeTranslation) {
+				require.Equal(t, 2, saTranslation.LenNamespaces())
+				require.Equal(t, "remoteOne", saTranslation.Get("namespace-id-1", "localOne"))
+				require.Equal(t, "remoteTwo", saTranslation.Get("namespace-id-2", "localTwo"))
+				// Each namespace only knows its own attributes.
+				require.Equal(t, "", saTranslation.Get("namespace-id-1", "localTwo"))
+				require.Equal(t, "", saTranslation.Get("namespace-id-2", "localOne"))
+				require.Equal(t, NewTuple("", false), NewTuple(saTranslation.GetExists("", "localOne")))
+				require.Equal(t, "localOne", saTranslation.Inverse().Get("namespace-id-1", "remoteOne"))
+				require.Equal(t, "localTwo", saTranslation.Inverse().Get("namespace-id-2", "remoteTwo"))
+			},
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			cfg := c.cfg
+			err := cfg.Validate()
+			if len(c.wantValidateErr) == 0 {
+				require.NoError(t, err)
+			} else {
+				require.Error(t, err)
+				for _, want := range c.wantValidateErr {
+					require.Contains(t, err.Error(), want)
+				}
+				for _, unwanted := range c.wantValidateErrExcludes {
+					require.NotContains(t, err.Error(), unwanted)
+				}
+			}
+
+			// Validate runs inside AsLocalToRemoteSATranslation, so an invalid config fails
+			// there for the same reason unless the case says otherwise.
+			wantTranslationErr := c.wantTranslationErr
+			if wantTranslationErr == nil {
+				wantTranslationErr = c.wantValidateErr
+			}
+			saTranslation, err := cfg.AsLocalToRemoteSATranslation()
+			if len(wantTranslationErr) == 0 {
+				require.NoError(t, err)
+				c.verify(t, saTranslation)
+				return
+			}
+			require.Error(t, err)
+			for _, want := range wantTranslationErr {
+				require.Contains(t, err.Error(), want)
+			}
+			require.Equal(t, 0, saTranslation.LenNamespaces())
+		})
+	}
+}
diff --git a/interceptor/reflection.go b/interceptor/reflection.go
index 9f836fa4..35d619a9 100644
--- a/interceptor/reflection.go
+++ b/interceptor/reflection.go
@@ -10,6 +10,8 @@ import (
 	"go.temporal.io/api/history/v1"
 	"go.temporal.io/api/namespace/v1"
 	"go.temporal.io/api/workflowservice/v1"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
+	replicationspb "go.temporal.io/server/api/replication/v1"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/persistence/serialization"
@@ -128,6 +130,22 @@ var (
 	}
 )
 
+const namespaceIDFieldName = "NamespaceId"
+
+// namespaceIDOwners maps struct types that OWN a namespace to their NamespaceId field index.
+// Deliberately an allowlist, not a name match: other types have a NamespaceId naming a
+// DIFFERENT namespace -- notably history.StartChildWorkflowExecutionInitiatedEventAttributes,
+// whose NamespaceId is the *child's* and which also carries SearchAttributes. Matching on the
+// field name would translate a parent's history event with the child's mapping.
+//
+// Built once at init and never mutated afterwards, so reads need no lock.
+var namespaceIDOwners = mustBuildNamespaceIDOwners(
+	reflect.TypeFor[replicationspb.HistoryTaskAttributes](),
+	reflect.TypeFor[replicationspb.BackfillHistoryTaskAttributes](),
+	reflect.TypeFor[replicationspb.SyncVersionedTransitionTaskAttributes](),
+	reflect.TypeFor[persistencespb.WorkflowExecutionInfo](),
+)
+
 // stringMatcher returns 2 values:
 //  1. new name. If there is no change, new name equals to input name
 //  2. whether or not the input name matches the defined rule(s).
@@ -137,6 +155,15 @@ type stringMatcher func(name string) (string, bool)
 // It returns whether anything was matched and any error it encountered.
 type visitor func(logger log.Logger, obj any, match stringMatcher) (bool, error)
 
+// saMatcherResolver returns the search attribute matcher configured for a namespace id.
+// false means the namespace has no mapping and its search attributes must be left untouched.
+type saMatcherResolver func(namespaceID string) (stringMatcher, bool)
+
+// blobVisitor visits the history events deserialized from a data blob. Callers close over
+// whatever matching state they need, since a data blob starts a fresh traversal that cannot
+// see the enclosing message.
+type blobVisitor func(events []*history.HistoryEvent) (bool, error)
+
 // visitNamespace uses reflection to recursively visit all fields
 // in the given object. When it finds namespace string fields, it invokes
 // the provided match function.
@@ -180,7 +207,9 @@ func visitNamespace(logger log.Logger, obj any, match stringMatcher) (bool, erro
 			}
 			return visit.Skip, nil
 		} else if dataBlobFieldNames[fieldType.Name] {
-			changed, err := visitDataBlobs(logger, vwp, match, visitNamespace)
+			changed, err := visitDataBlobs(logger, vwp, func(events []*history.HistoryEvent) (bool, error) {
+				return visitNamespace(logger, events, match)
+			})
 			matched = matched || changed
 			if err != nil {
 				return visit.Stop, err
@@ -207,10 +236,12 @@ func visitNamespace(logger log.Logger, obj any, match stringMatcher) (bool, erro
 	return matched, err
 }
 
-// visitSearchAttributes uses reflection to recursively visit all fields
-// in the given object. When it finds namespace string fields, it invokes
-// the provided match function.
-func visitSearchAttributes(logger log.Logger, obj any, match stringMatcher) (bool, error) {
+// visitSearchAttributes translates the search attributes in obj using the mapping configured
+// for whichever namespace owns them.
+//
+// fallbackNamespaceID is used when the parent chain reaches no namespace owner. See
+// resolveNamespaceID.
+func visitSearchAttributes(logger log.Logger, obj any, resolve saMatcherResolver, fallbackNamespaceID string) (bool, error) {
 	var matched bool
 
 	// The visitor function can return Skip, Stop, or Continue to control recursion.
@@ -225,12 +256,26 @@ func visitSearchAttributes(logger log.Logger, obj any, match stringMatcher) (boo
 			return action, nil
 		}
 		if dataBlobFieldNames[fieldType.Name] {
-			changed, err := visitDataBlobs(logger, vwp, match, visitSearchAttributes)
+			// Resolve once, here at the boundary: the parent chain is still intact, whereas the
+			// events inside the blob are visited in a fresh traversal that cannot see the
+			// enclosing namespace owner. Descend even when the namespace is unresolved, since
+			// visitDataBlobs also repairs invalid UTF-8 independently of any translation.
+			nsID := resolveNamespaceID(vwp, fallbackNamespaceID)
+			changed, err := visitDataBlobs(logger, vwp, func(events []*history.HistoryEvent) (bool, error) {
+				return visitSearchAttributes(logger, events, resolve, nsID)
+			})
 			matched = matched || changed
 			if err != nil {
 				return visit.Stop, err
 			}
 		} else if searchAttributeFieldNames[fieldType.Name] {
+			nsID := resolveNamespaceID(vwp, fallbackNamespaceID)
+			match, ok := resolve(nsID)
+			if !ok {
+				logSkippedSearchAttributes(logger, obj, nsID)
+				return visit.Continue, nil
+			}
+
 			// This could be *common.SearchAttributes, or it could be map[string]*common.Payload (indexed fields)
 			var changed bool
 			switch attrs := vwp.Interface().(type) {
@@ -244,7 +289,14 @@ func visitSearchAttributes(logger log.Logger, obj any, match stringMatcher) (boo
 					}
 				}
 			default:
-				return visit.Stop, fmt.Errorf("unhandled search attribute type: %T", attrs)
+				// Reachable: Add- and RemoveSearchAttributesRequest name their fields
+				// SearchAttributes too, but they hold a map[string]enums.IndexedValueType and a
+				// []string. Skip them instead of aborting translation of the whole message.
+				logger.Warn("unhandled search attribute type",
+					tag.NewStringTag("type", fmt.Sprintf("%T", attrs)))
+				metrics.SearchAttrTranslationSkipped.WithLabelValues(
+					metrics.SkipReasonUnsupportedType, metrics.SanitizedTypeName(obj)).Inc()
+				return visit.Continue, nil
 			}
 			matched = matched || changed
 
@@ -257,6 +309,24 @@ func visitSearchAttributes(logger log.Logger, obj any, match stringMatcher) (boo
 	return matched, err
 }
 
+func logSkippedSearchAttributes(logger log.Logger, obj any, nsID string) {
+	msgType := metrics.SanitizedTypeName(obj)
+	if nsID == "" {
+		// No enclosing namespace owner and nothing seeded a fallback. That is a gap in
+		// namespaceIDOwners or an unpaired response, not a configuration decision.
+		logger.Warn("could not resolve namespace for search attributes",
+			tag.NewStringTag("type", msgType))
+		metrics.SearchAttrTranslationSkipped.WithLabelValues(
+			metrics.SkipReasonUnresolvedNamespace, msgType).Inc()
+		return
+	}
+
+	// A namespace with no configured mapping is the expected steady state for every namespace
+	// that is not being migrated, so it is not counted.
+	logger.Debug("no search attribute mapping configured for namespace",
+		tag.NewStringTag("namespace-id", nsID), tag.NewStringTag("type", msgType))
+}
+
 func translateIndexedFields(fields map[string]*common.Payload, match stringMatcher) (map[string]*common.Payload, bool) {
 	if fields == nil {
 		return fields, false
@@ -287,10 +357,56 @@ func getParentFieldType(vwp visit.ValueWithParent) (result reflect.StructField,
 	return fieldType, action
 }
 
-func visitDataBlobs(logger log.Logger, vwp visit.ValueWithParent, match stringMatcher, visitor visitor) (bool, error) {
+// mustBuildNamespaceIDOwners resolves each owner's NamespaceId to a field index up front, so
+// that resolveNamespaceID never calls reflect.Type.FieldByName on the hot path: that linear
+// scans the 20+ fields of a generated proto struct on every hop. The owner set is fixed at
+// compile time, so an absent or retyped field is a programmer error (or an upstream proto
+// rename) and panics here rather than silently disabling translation.
+func mustBuildNamespaceIDOwners(ownerTypes ...reflect.Type) map[reflect.Type]int {
+	owners := make(map[reflect.Type]int, len(ownerTypes))
+	for _, ownerType := range ownerTypes {
+		field, ok := ownerType.FieldByName(namespaceIDFieldName)
+		if !ok {
+			panic(fmt.Sprintf("namespace owner %v has no %s field", ownerType, namespaceIDFieldName))
+		}
+		if len(field.Index) != 1 || field.Type.Kind() != reflect.String {
+			panic(fmt.Sprintf("namespace owner %v has a %s field that is not a direct string: %v",
+				ownerType, namespaceIDFieldName, field.Type))
+		}
+		owners[ownerType] = field.Index[0]
+	}
+	return owners
+}
+
+// resolveNamespaceID walks UP from vwp to the nearest enclosing namespace owner and returns its
+// NamespaceId. Walking only upward keeps the result independent of traversal order, which is
+// unspecified: visit.ValuesUnsafe pops the front of its worklist but swaps in the last element,
+// so it is neither breadth- nor depth-first. Tracking the most recent NamespaceId seen while
+// descending would be non-deterministic.
+//
+// fallback carries context across a boundary the parent chain cannot cross: a data blob, whose
+// events are visited in a fresh traversal, or a unary response whose paired request holds the
+// only namespace id.
+func resolveNamespaceID(vwp visit.ValueWithParent, fallback string) string {
+	for p := vwp.Parent; p != nil; p = p.Parent {
+		if p.Kind() != reflect.Struct {
+			continue
+		}
+		fieldIdx, ok := namespaceIDOwners[p.Type()]
+		if !ok {
+			continue
+		}
+		if nsID := p.Field(fieldIdx).String(); nsID != "" {
+			return nsID
+		}
+	}
+	return fallback
+}
+
+func visitDataBlobs(logger log.Logger, vwp visit.ValueWithParent, bv blobVisitor) (bool, error) {
 	switch evt := vwp.Interface().(type) {
 	case []*common.DataBlob:
-		newEvts, matched, changed, err := translateDataBlobs(logger, match, visitor, evt...)
+		newEvts, matched, changed, err := translateDataBlobs(logger, bv, evt...)
 		if err != nil {
 			return matched, err
 		}
@@ -301,7 +417,7 @@ func visitDataBlobs(logger log.Logger, vwp visit.ValueWithParent, match stringMa
 		}
 		return matched, nil
 	case *common.DataBlob:
-		newEvt, matched, changed, err := translateOneDataBlob(logger, match, visitor, evt)
+		newEvt, matched, changed, err := translateOneDataBlob(logger, bv, evt)
 		if err != nil {
 			return matched, err
 		}
@@ -316,9 +432,9 @@ func visitDataBlobs(logger log.Logger, vwp visit.ValueWithParent, match stringMa
 	}
 }
 
-func translateDataBlobs(logger log.Logger, match stringMatcher, visitor visitor, blobs ...*common.DataBlob) (result []*common.DataBlob, anyMatched, anyChanged bool, retErr error) {
+func translateDataBlobs(logger log.Logger, bv blobVisitor, blobs ...*common.DataBlob) (result []*common.DataBlob, anyMatched, anyChanged bool, retErr error) {
 	for i, blob := range blobs {
-		newBlob, matched, changed, err := translateOneDataBlob(logger, match, visitor, blob)
+		newBlob, matched, changed, err := translateOneDataBlob(logger, bv, blob)
 		anyChanged = anyChanged || changed
 		anyMatched = anyMatched || matched
 		if err != nil {
@@ -329,7 +445,7 @@ func translateDataBlobs(logger log.Logger, match stringMatcher, visitor visitor,
 	return blobs, anyMatched, anyChanged, nil
 }
 
-func translateOneDataBlob(logger log.Logger, match stringMatcher, visitor visitor, blob *common.DataBlob) (result *common.DataBlob, matched, changed bool, retErr error) {
+func translateOneDataBlob(logger log.Logger, bv blobVisitor, blob *common.DataBlob) (result *common.DataBlob, matched, changed bool, retErr error) {
 	if blob == nil || len(blob.Data) == 0 {
 		return blob, matched, changed, nil
 	}
@@ -356,7 +472,7 @@ func translateOneDataBlob(logger log.Logger, match stringMatcher, visitor visito
 		}
 	}
 
-	m, err := visitor(logger, events, match)
+	m, err := bv(events)
 	matched = matched || m
 	if err != nil {
 		return blob, matched, changed, err
diff --git a/interceptor/reflection_test.go b/interceptor/reflection_test.go
index d9bce11a..ed5f0e3a 100644
--- a/interceptor/reflection_test.go
+++ b/interceptor/reflection_test.go
@@ -1,11 +1,43 @@
 package interceptor
 
 import (
+	"reflect"
 	"testing"
 
+	"github.com/stretchr/testify/require"
+	"go.temporal.io/api/history/v1"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
+	replicationspb "go.temporal.io/server/api/replication/v1"
 	"go.temporal.io/server/common/log"
 )
 
+// TestNamespaceIDOwnersAreValid guards the allowlist against an upstream proto rename and,
+// more importantly, against anyone replacing it with a NamespaceId field-name match.
+func TestNamespaceIDOwnersAreValid(t *testing.T) {
+	expectedOwners := []reflect.Type{
+		reflect.TypeFor[replicationspb.HistoryTaskAttributes](),
+		reflect.TypeFor[replicationspb.BackfillHistoryTaskAttributes](),
+		reflect.TypeFor[replicationspb.SyncVersionedTransitionTaskAttributes](),
+		reflect.TypeFor[persistencespb.WorkflowExecutionInfo](),
+	}
+	require.Len(t, namespaceIDOwners, len(expectedOwners))
+
+	for _, ownerType := range expectedOwners {
+		fieldIdx, ok := namespaceIDOwners[ownerType]
+		require.True(t, ok, "%v is missing from namespaceIDOwners", ownerType)
+
+		field := ownerType.Field(fieldIdx)
+		require.Equal(t, namespaceIDFieldName, field.Name, "owner %v", ownerType)
+		require.Equal(t, reflect.String, field.Type.Kind(), "owner %v", ownerType)
+	}
+
+	// StartChildWorkflowExecutionInitiatedEventAttributes carries the *child's* NamespaceId
+	// alongside its own SearchAttributes. Treating it as an owner would translate a parent's
+	// history event with the child's mapping.
+	require.NotContains(t, namespaceIDOwners,
+		reflect.TypeFor[history.StartChildWorkflowExecutionInitiatedEventAttributes]())
+}
+
 func BenchmarkVisitNamespace(b *testing.B) {
 	variants := []struct {
 		testName    string
@@ -43,3 +75,49 @@ func BenchmarkVisitNamespace(b *testing.B) {
 		})
 	}
 }
+
+func BenchmarkVisitSearchAttributes(b *testing.B) {
+	variants := []struct {
+		testName    string
+		inputSAName string
+		mapping     map[string]string
+	}{
+		{
+			testName:    "name changed",
+			inputSAName: "orig",
+			mapping:     map[string]string{"orig": "orig.cloud"},
+		},
+		{
+			testName:    "name unchanged",
+			inputSAName: "orig",
+			mapping:     map[string]string{"other": "other.cloud"},
+		},
+	}
+	// Includes the deeply nested SyncVersionedTransitionTaskAttributes case, where the namespace
+	// owner sits several hops above the search attributes.
+	cases := generateSearchAttributeObjs()
+
+	logger := log.NewTestLogger()
+	for _, c := range cases {
+		b.Run(c.objName, func(b *testing.B) {
+			for _, variant := range variants {
+				resolve := constMatcherResolver(createStringMatcher(variant.mapping))
+				b.Run(variant.testName, func(b *testing.B) {
+					for i := 0; i < b.N; i++ {
+						b.StopTimer()
+						input := c.makeType(variant.inputSAName)
+
+						b.StartTimer()
+						_, _ = visitSearchAttributes(logger, input, resolve, "")
+					}
+				})
+			}
+		})
+	}
+}
+
+// constMatcherResolver applies one matcher to every namespace, for tests that are not
+// exercising per-namespace resolution.
+func constMatcherResolver(match stringMatcher) saMatcherResolver {
+	return func(string) (stringMatcher, bool) { return match, true }
+}
diff --git a/interceptor/search_attribute_translator.go b/interceptor/search_attribute_translator.go
index cae88ae7..df6e661a 100644
--- a/interceptor/search_attribute_translator.go
+++ b/interceptor/search_attribute_translator.go
@@ -3,6 +3,7 @@ package interceptor
 import (
 	"strings"
 
+	"go.temporal.io/server/api/adminservice/v1"
 	"go.temporal.io/server/common/api"
 	"go.temporal.io/server/common/log"
 
@@ -13,8 +14,8 @@ type (
 	saTranslator struct {
 		logger      log.Logger
 		matchMethod func(string) bool
-		reqMap      map[string]stringMatcher
-		respMap     map[string]stringMatcher
+		resolveReq  saMatcherResolver
+		resolveResp saMatcherResolver
 	}
 )
 
@@ -26,8 +27,10 @@ func NewSearchAttributeTranslator(logger log.Logger, reqMap, respMap map[string]
 			// We should never translate these responses to the search attribute's indexed field.
 			return !strings.HasPrefix(method, api.WorkflowServicePrefix)
 		},
-		reqMap:  createStringMatchers(reqMap),
-		respMap: createStringMatchers(respMap),
+		// The resolvers are built once here and hold no per-message state, so a single
+		// translator serves every concurrent stream.
+		resolveReq:  newSAMatcherResolver(reqMap),
+		resolveResp: newSAMatcherResolver(respMap),
 	}
 }
 
@@ -40,27 +43,33 @@ func (s *saTranslator) MatchMethod(m string) bool {
 }
 
 func (s *saTranslator) TranslateRequest(req any) (bool, error) {
-	return visitSearchAttributes(s.logger, req, s.getNamespaceReqMatcher(""))
+	return visitSearchAttributes(s.logger, req, s.resolveReq, "")
 }
 
-func (s *saTranslator) TranslateResponse(resp any) (bool, error) {
-	return visitSearchAttributes(s.logger, resp, s.getNamespaceRespMatcher(""))
-}
-
-func (s *saTranslator) getNamespaceReqMatcher(namespaceId string) stringMatcher {
-	// Placeholder: Just return the first one (only support one namespace mapping)
-	for _, matcher := range s.reqMap {
-		return matcher
+// TranslateResponse translates the search attributes in resp. Some admin service responses
+// carry history blobs but no namespace field of their own, so the paired request is used to
+// seed the namespace. req is nil for streams, which have no request to pair with.
+//
+// This relies on NamespaceId surviving TranslateRequest: the namespace name translator rewrites
+// Namespace, never NamespaceId. Adding namespace id translation later would silently break it.
+func (s *saTranslator) TranslateResponse(req, resp any) (bool, error) {
+	var fallbackNamespaceID string
+	switch r := req.(type) {
+	case *adminservice.GetWorkflowExecutionRawHistoryV2Request:
+		fallbackNamespaceID = r.NamespaceId
+	case *adminservice.GetWorkflowExecutionRawHistoryRequest:
+		fallbackNamespaceID = r.NamespaceId
 	}
-	return createStringMatcher(nil)
+	return visitSearchAttributes(s.logger, resp, s.resolveResp, fallbackNamespaceID)
 }
 
-func (s *saTranslator) getNamespaceRespMatcher(namespaceId string) stringMatcher {
-	// Placeholder: Just return the first one (only support one namespace mappping)
-	for _, matcher := range s.respMap {
-		return matcher
+func newSAMatcherResolver(nsMappings map[string]map[string]string) saMatcherResolver {
+	matchers := createStringMatchers(nsMappings)
+
+	return func(nsID string) (stringMatcher, bool) {
+		match, ok := matchers[nsID]
+		return match, ok
 	}
-	return createStringMatcher(nil)
 }
 
 func createStringMatchers(nsMappings map[string]map[string]string) map[string]stringMatcher {
diff --git a/interceptor/search_attribute_translator_test.go b/interceptor/search_attribute_translator_test.go
index f33b7a83..c9ec083f 100644
--- a/interceptor/search_attribute_translator_test.go
+++ b/interceptor/search_attribute_translator_test.go
@@ -1,6 +1,7 @@
 package interceptor
 
 import (
+	"sort"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -10,6 +11,7 @@ import (
 	"go.temporal.io/server/api/adminservice/v1"
 	"go.temporal.io/server/api/persistence/v1"
 	replicationspb "go.temporal.io/server/api/replication/v1"
+	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/persistence/serialization"
 )
 
@@ -45,7 +47,11 @@ type (
 )
 
 func TestTranslateSearchAttribute(t *testing.T) {
-	testTranslateObj(t, visitSearchAttributes, generateSearchAttributeObjs(), require.EqualExportedValues)
+	// These cases exercise a single namespace, so one matcher applies everywhere.
+	adapter := func(l log.Logger, obj any, m stringMatcher) (bool, error) {
+		return visitSearchAttributes(l, obj, constMatcherResolver(m), "")
+	}
+	testTranslateObj(t, adapter, generateSearchAttributeObjs(), require.EqualExportedValues)
 }
 
 func generateSearchAttributeObjs() []objCase {
@@ -233,3 +239,356 @@ func makeTestIndexedFieldMap(name string) map[string]*common.Payload {
 		},
 	}
 }
+
+const (
+	testNsA = "ns-a"
+	testNsB = "ns-b"
+	testNsC = "ns-c"
+
+	testSAName = "TestSA"
+	keywordA   = "Keyword01"
+	keywordB   = "Keyword02"
+)
+
+// testSAMappings maps the same search attribute to a different indexed field per namespace,
+// so a translation applied with the wrong namespace's mapping is visible in assertions.
+func testSAMappings() map[string]map[string]string {
+	return map[string]map[string]string{
+		testNsA: {testSAName: keywordA},
+		testNsB: {testSAName: keywordB},
+	}
+}
+
+func newTestSATranslator(t *testing.T, nsMappings map[string]map[string]string) Translator {
+	t.Helper()
+	return NewSearchAttributeTranslator(log.NewTestLogger(), nsMappings, nsMappings)
+}
+
+// makeMultiNamespaceFrame builds one replication frame carrying four tasks spanning three
+// namespaces, which is what the wire actually looks like: resolution has to happen per subtree.
+//
+// Every blob and IndexedFields map is built fresh. visit.Values keeps a set of pointers it has
+// already seen and skips repeats, so a hoisted, shared blob would leave the second subtree
+// untranslated and could still read as a pass.
+func makeMultiNamespaceFrame(saName string) *adminservice.StreamWorkflowReplicationMessagesResponse {
+	return &adminservice.StreamWorkflowReplicationMessagesResponse{
+		Attributes: &adminservice.StreamWorkflowReplicationMessagesResponse_Messages{
+			Messages: &replicationspb.WorkflowReplicationMessages{
+				ReplicationTasks: []*replicationspb.ReplicationTask{
+					{
+						Attributes: &replicationspb.ReplicationTask_HistoryTaskAttributes{
+							HistoryTaskAttributes: &replicationspb.HistoryTaskAttributes{
+								NamespaceId:  testNsA,
+								WorkflowId:   "wf-a",
+								Events:       makeHistoryEventsBlobWithSearchAttribute(saName),
+								NewRunEvents: makeHistoryEventsBlobWithSearchAttribute(saName),
+							},
+						},
+					},
+					{
+						Attributes: &replicationspb.ReplicationTask_HistoryTaskAttributes{
+							HistoryTaskAttributes: &replicationspb.HistoryTaskAttributes{
+								NamespaceId: testNsB,
+								WorkflowId:  "wf-b",
+								Events:      makeHistoryEventsBlobWithSearchAttribute(saName),
+							},
+						},
+					},
+					{
+						// WorkflowExecutionInfo owns its namespace directly: no blob involved.
+						Attributes: &replicationspb.ReplicationTask_SyncWorkflowStateTaskAttributes{
+							SyncWorkflowStateTaskAttributes: &replicationspb.SyncWorkflowStateTaskAttributes{
+								WorkflowState: &persistence.WorkflowMutableState{
+									ExecutionInfo: &persistence.WorkflowExecutionInfo{
+										NamespaceId:      testNsA,
+										WorkflowId:       "wf-a-state",
+										SearchAttributes: makeTestIndexedFieldMap(saName),
+										// Memo is the same type as SearchAttributes. It must not be rewritten.
+										Memo: makeTestIndexedFieldMap(saName),
+									},
+								},
+							},
+						},
+					},
+					{
+						Attributes: &replicationspb.ReplicationTask_HistoryTaskAttributes{
+							HistoryTaskAttributes: &replicationspb.HistoryTaskAttributes{
+								NamespaceId: testNsC,
+								WorkflowId:  "wf-c",
+								Events:      makeHistoryEventsBlobWithSearchAttribute(saName),
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
+// makeHistoryTaskFrame wraps the given events in a blob owned by nsID.
+func makeHistoryTaskFrame(nsID string, events ...*history.HistoryEvent) *adminservice.StreamWorkflowReplicationMessagesResponse {
+	blob, err := serialization.NewSerializer().SerializeEvents(events)
+	if err != nil {
+		panic(err)
+	}
+	return &adminservice.StreamWorkflowReplicationMessagesResponse{
+		Attributes: &adminservice.StreamWorkflowReplicationMessagesResponse_Messages{
+			Messages: &replicationspb.WorkflowReplicationMessages{
+				ReplicationTasks: []*replicationspb.ReplicationTask{
+					{
+						Attributes: &replicationspb.ReplicationTask_HistoryTaskAttributes{
+							HistoryTaskAttributes: &replicationspb.HistoryTaskAttributes{
+								NamespaceId: nsID,
+								Events:      blob,
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
+func firstTaskEvents(resp *adminservice.StreamWorkflowReplicationMessagesResponse) *common.DataBlob {
+	return resp.GetMessages().GetReplicationTasks()[0].GetHistoryTaskAttributes().GetEvents()
+}
+
+// blobSAKeys returns every search attribute key in the blob, sorted. Duplicates are kept so
+// that a partial translation (one event rewritten, another missed) fails the assertion.
+func blobSAKeys(t *testing.T, blob *common.DataBlob) []string {
+	t.Helper()
+	events, err := serialization.NewSerializer().DeserializeEvents(blob)
+	require.NoError(t, err)
+
+	var keys []string
+	for _, evt := range events {
+		keys = append(keys, mapKeys(evt.GetWorkflowExecutionStartedEventAttributes().GetSearchAttributes().GetIndexedFields())...)
+		keys = append(keys, mapKeys(evt.GetStartChildWorkflowExecutionInitiatedEventAttributes().GetSearchAttributes().GetIndexedFields())...)
+	}
+	sort.Strings(keys)
+	return keys
+}
+
+func mapKeys(fields map[string]*common.Payload) []string {
+	keys := make([]string, 0, len(fields))
+	for key := range fields {
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+	return keys
+}
+
+func TestTranslateSearchAttributePerNamespace(t *testing.T) {
+	tr := newTestSATranslator(t, testSAMappings())
+
+	// A single pass can pass by luck: visit.ValuesUnsafe pops the front of its worklist but
+	// swaps in the last element, so the order in which the four tasks are reached is
+	// unspecified. Rebuild the frame every iteration so no subtree is ever seen pre-translated.
+	for i := 0; i < 25; i++ {
+		frame := makeMultiNamespaceFrame(testSAName)
+
+		changed, err := tr.TranslateResponse(nil, frame)
+		require.NoError(t, err)
+		require.True(t, changed)
+
+		tasks := frame.GetMessages().GetReplicationTasks()
+		require.Len(t, tasks, 4)
+
+		nsATask := tasks[0].GetHistoryTaskAttributes()
+		require.Equal(t, []string{keywordA, keywordA}, blobSAKeys(t, nsATask.GetEvents()))
+		require.Equal(t, []string{keywordA, keywordA}, blobSAKeys(t, nsATask.GetNewRunEvents()))
+
+		require.Equal(t, []string{keywordB, keywordB}, blobSAKeys(t, tasks[1].GetHistoryTaskAttributes().GetEvents()),
+			"ns-b must get its own mapping, not ns-a's")
+
+		execInfo := tasks[2].GetSyncWorkflowStateTaskAttributes().GetWorkflowState().GetExecutionInfo()
+		require.Equal(t, []string{keywordA}, mapKeys(execInfo.GetSearchAttributes()))
+		require.Equal(t, []string{testSAName}, mapKeys(execInfo.GetMemo()), "Memo must not be rewritten")
+
+		require.Equal(t, []string{testSAName, testSAName}, blobSAKeys(t, tasks[3].GetHistoryTaskAttributes().GetEvents()),
+			"ns-c has no mapping and must be left untouched")
+	}
+}
+
+func TestTranslateSearchAttributeInBlobUsesEnclosingNamespace(t *testing.T) {
+	tr := newTestSATranslator(t, testSAMappings())
+
+	// The event names a different namespace as its parent workflow's. The enclosing
+	// HistoryTaskAttributes is what owns the namespace, so ns-a's mapping must win.
+	frame := makeHistoryTaskFrame(testNsA, &history.HistoryEvent{
+		EventId:   1,
+		EventType: enums.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED,
+		Attributes: &history.HistoryEvent_WorkflowExecutionStartedEventAttributes{
+			WorkflowExecutionStartedEventAttributes: &history.WorkflowExecutionStartedEventAttributes{
+				ParentWorkflowNamespaceId: testNsB,
+				SearchAttributes: &common.SearchAttributes{
+					IndexedFields: makeTestIndexedFieldMap(testSAName),
+				},
+			},
+		},
+	})
+
+	changed, err := tr.TranslateResponse(nil, frame)
+	require.NoError(t, err)
+	require.True(t, changed)
+	require.Equal(t, []string{keywordA}, blobSAKeys(t, firstTaskEvents(frame)))
+}
+
+func TestTranslateSearchAttributeIgnoresChildNamespaceId(t *testing.T) {
+	tr := newTestSATranslator(t, testSAMappings())
+
+	// StartChildWorkflowExecutionInitiatedEventAttributes has both a NamespaceId (the child's)
+	// and SearchAttributes (the parent's). Resolving by nearest struct with a NamespaceId field
+	// would translate this event with ns-b's mapping. This test fails if the owner allowlist is
+	// ever replaced with a field-name match.
+	frame := makeHistoryTaskFrame(testNsA, &history.HistoryEvent{
+		EventId:   1,
+		EventType: enums.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_INITIATED,
+		Attributes: &history.HistoryEvent_StartChildWorkflowExecutionInitiatedEventAttributes{
+			StartChildWorkflowExecutionInitiatedEventAttributes: &history.StartChildWorkflowExecutionInitiatedEventAttributes{
+				Namespace:   "child-ns",
+				NamespaceId: testNsB,
+				SearchAttributes: &common.SearchAttributes{
+					IndexedFields: makeTestIndexedFieldMap(testSAName),
+				},
+			},
+		},
+	})
+
+	changed, err := tr.TranslateResponse(nil, frame)
+	require.NoError(t, err)
+	require.True(t, changed)
+	require.Equal(t, []string{keywordA}, blobSAKeys(t, firstTaskEvents(frame)))
+}
+
+func TestTranslateSearchAttributeIgnoresParentNamespaceId(t *testing.T) {
+	tr := newTestSATranslator(t, testSAMappings())
+
+	// WorkflowExecutionInfo carries both NamespaceId and ParentNamespaceId. Only the former
+	// owns the search attributes.
+	execInfo := &persistence.WorkflowExecutionInfo{
+		NamespaceId:       testNsA,
+		ParentNamespaceId: testNsB,
+		SearchAttributes:  makeTestIndexedFieldMap(testSAName),
+	}
+
+	changed, err := tr.TranslateResponse(nil, execInfo)
+	require.NoError(t, err)
+	require.True(t, changed)
+	require.Equal(t, []string{keywordA}, mapKeys(execInfo.GetSearchAttributes()))
+}
+
+func TestTranslateSearchAttributeRawHistoryUsesPairedRequest(t *testing.T) {
+	tr := newTestSATranslator(t, testSAMappings())
+
+	// These responses carry history blobs but no namespace field of their own. Without the
+	// paired request there is nothing to resolve, and the response must be left alone rather
+	// than erroring or being translated with an arbitrary namespace's mapping.
+	newV2Resp := func() *adminservice.GetWorkflowExecutionRawHistoryV2Response {
+		return &adminservice.GetWorkflowExecutionRawHistoryV2Response{
+			HistoryBatches: []*common.DataBlob{makeHistoryEventsBlobWithSearchAttribute(testSAName)},
+		}
+	}
+	newResp := func() *adminservice.GetWorkflowExecutionRawHistoryResponse {
+		return &adminservice.GetWorkflowExecutionRawHistoryResponse{
+			HistoryBatches: []*common.DataBlob{makeHistoryEventsBlobWithSearchAttribute(testSAName)},
+		}
+	}
+
+	t.Run("V2 unpaired", func(t *testing.T) {
+		resp := newV2Resp()
+		changed, err := tr.TranslateResponse(nil, resp)
+		require.NoError(t, err)
+		require.False(t, changed)
+		require.Equal(t, []string{testSAName, testSAName}, blobSAKeys(t, resp.HistoryBatches[0]))
+	})
+
+	t.Run("V2 paired", func(t *testing.T) {
+		resp := newV2Resp()
+		req := &adminservice.GetWorkflowExecutionRawHistoryV2Request{NamespaceId: testNsB}
+		changed, err := tr.TranslateResponse(req, resp)
+		require.NoError(t, err)
+		require.True(t, changed)
+		require.Equal(t, []string{keywordB, keywordB}, blobSAKeys(t, resp.HistoryBatches[0]))
+	})
+
+	t.Run("unpaired", func(t *testing.T) {
+		resp := newResp()
+		changed, err := tr.TranslateResponse(nil, resp)
+		require.NoError(t, err)
+		require.False(t, changed)
+		require.Equal(t, []string{testSAName, testSAName}, blobSAKeys(t, resp.HistoryBatches[0]))
+	})
+
+	t.Run("paired", func(t *testing.T) {
+		resp := newResp()
+		req := &adminservice.GetWorkflowExecutionRawHistoryRequest{NamespaceId: testNsB}
+		changed, err := tr.TranslateResponse(req, resp)
+		require.NoError(t, err)
+		require.True(t, changed)
+		require.Equal(t, []string{keywordB, keywordB}, blobSAKeys(t, resp.HistoryBatches[0]))
+	})
+}
+
+func TestTranslateSearchAttributeEmptyNamespaceIdMatchesNothing(t *testing.T) {
+	// There is no mapping that applies to every namespace: an entry keyed by an empty namespace
+	// id matches only an unresolved namespace, which no well formed replication task produces.
+	// config.SATranslationConfig.Validate rejects such a config before it gets this far; this is
+	// the second line of defence, and it fails if a wildcard fallback is ever reintroduced.
+	tr := newTestSATranslator(t, map[string]map[string]string{"": {testSAName: keywordA}})
+
+	frame := makeMultiNamespaceFrame(testSAName)
+	changed, err := tr.TranslateResponse(nil, frame)
+	require.NoError(t, err)
+	require.False(t, changed)
+
+	tasks := frame.GetMessages().GetReplicationTasks()
+	require.Len(t, tasks, 4)
+	for i, task := range tasks {
+		if hta := task.GetHistoryTaskAttributes(); hta != nil {
+			require.Equal(t, []string{testSAName, testSAName}, blobSAKeys(t, hta.GetEvents()),
+				"task %d must be untouched", i)
+			continue
+		}
+		execInfo := task.GetSyncWorkflowStateTaskAttributes().GetWorkflowState().GetExecutionInfo()
+		require.Equal(t, []string{testSAName}, mapKeys(execInfo.GetSearchAttributes()),
+			"task %d must be untouched", i)
+	}
+}
+func TestTranslateSearchAttributeUnsupportedFieldTypes(t *testing.T) {
+	// Add- and RemoveSearchAttributesRequest name their fields SearchAttributes too, but hold a
+	// map[string]enums.IndexedValueType and a []string, so neither must be treated as an error
+	// that aborts translation of the whole message.
+	//
+	// Neither request type is enclosed by a namespace owner, so the namespace resolves to "",
+	// no matcher resolves, and the field is skipped before the type switch runs. That makes
+	// visitSearchAttributes' unsupported-type branch defensive rather than load-bearing: it
+	// only matters if a future message carries an odd SearchAttributes type *inside* a
+	// namespace owner. Skipping is still the right outcome there, which is why it warns and
+	// continues instead of returning visit.Stop.
+	{
+		t.Run("namespace keyed", func(t *testing.T) {
+			tr := newTestSATranslator(t, testSAMappings())
+
+			addReq := &adminservice.AddSearchAttributesRequest{
+				SearchAttributes: map[string]enums.IndexedValueType{
+					testSAName: enums.INDEXED_VALUE_TYPE_KEYWORD,
+				},
+			}
+			changed, err := tr.TranslateRequest(addReq)
+			require.NoError(t, err)
+			require.False(t, changed)
+			require.Equal(t, map[string]enums.IndexedValueType{
+				testSAName: enums.INDEXED_VALUE_TYPE_KEYWORD,
+			}, addReq.SearchAttributes)
+
+			removeReq := &adminservice.RemoveSearchAttributesRequest{
+				SearchAttributes: []string{testSAName},
+			}
+			changed, err = tr.TranslateRequest(removeReq)
+			require.NoError(t, err)
+			require.False(t, changed)
+			require.Equal(t, []string{testSAName}, removeReq.SearchAttributes)
+		})
+	}
+}
diff --git a/interceptor/translation_interceptor.go b/interceptor/translation_interceptor.go
index f06b8c4f..f59b85bc 100644
--- a/interceptor/translation_interceptor.go
+++ b/interceptor/translation_interceptor.go
@@ -61,7 +61,7 @@ func (i *TranslationInterceptor) Intercept(
 	for _, tr := range i.translators {
 		if tr.MatchMethod(info.FullMethod) {
 			start := time.Now()
-			changed, trErr := tr.TranslateResponse(resp)
+			changed, trErr := tr.TranslateResponse(req, resp)
 			logTranslateResult(tr, i.logger, changed, trErr, methodName+"Response", resp, time.Since(start))
 		}
 	}
@@ -106,7 +106,8 @@ func (w *streamTranslator) RecvMsg(m any) error {
 func (w *streamTranslator) SendMsg(m any) error {
 	for _, tr := range w.translators {
 		start := time.Now()
-		changed, trErr := tr.TranslateResponse(m)
+		// Streams have no request to pair with this message.
+		changed, trErr := tr.TranslateResponse(nil, m)
 		logTranslateResult(tr, w.logger, changed, trErr, "SendMsg", m, time.Since(start))
 	}
 	return w.ServerStream.SendMsg(m)
diff --git a/interceptor/translation_interceptor_test.go b/interceptor/translation_interceptor_test.go
index e67b4787..fae7e679 100644
--- a/interceptor/translation_interceptor_test.go
+++ b/interceptor/translation_interceptor_test.go
@@ -20,13 +20,16 @@ type spyTranslator struct {
 	matchCalls         int
 	translateReqCalls  int
 	translateRespCalls int
+	// pairedReqs records the request passed alongside each response, in call order.
+	pairedReqs []any
 }
 
 func (s *spyTranslator) Kind() string                       { return "spy" }
 func (s *spyTranslator) MatchMethod(string) bool            { s.matchCalls++; return true }
 func (s *spyTranslator) TranslateRequest(any) (bool, error) { s.translateReqCalls++; return false, nil }
-func (s *spyTranslator) TranslateResponse(any) (bool, error) {
+func (s *spyTranslator) TranslateResponse(req, _ any) (bool, error) {
 	s.translateRespCalls++
+	s.pairedReqs = append(s.pairedReqs, req)
 	return false, nil
 }
 
@@ -73,12 +76,41 @@ func TestTranslationInterceptor(t *testing.T) {
 			if tc.incomingHeaders != nil {
 				ctx = metadata.NewIncomingContext(ctx, metadata.New(tc.incomingHeaders))
 			}
-			_, err := ti.Intercept(ctx, &workflowservice.DescribeWorkflowExecutionRequest{}, info, handler)
+			req := &workflowservice.DescribeWorkflowExecutionRequest{}
+			_, err := ti.Intercept(ctx, req, info, handler)
 			require.NoError(t, err)
 
 			require.Equal(t, tc.expectedMatchCalls, spy.matchCalls, "MatchMethod call count")
 			require.Equal(t, tc.expectedReqCalls, spy.translateReqCalls, "TranslateRequest call count")
 			require.Equal(t, tc.expectedRespCalls, spy.translateRespCalls, "TranslateResponse call count")
+			if tc.expectedRespCalls > 0 {
+				require.Equal(t, []any{req}, spy.pairedReqs,
+					"unary responses must be translated alongside their request")
+			}
 		})
 	}
 }
+
+// fakeServerStream captures the messages a streamTranslator forwards.
+type fakeServerStream struct {
+	grpc.ServerStream
+	sent []any
+}
+
+func (f *fakeServerStream) SendMsg(m any) error {
+	f.sent = append(f.sent, m)
+	return nil
+}
+
+func TestStreamTranslatorSendMsgHasNoPairedRequest(t *testing.T) {
+	spy := &spyTranslator{}
+	inner := &fakeServerStream{}
+	st := newStreamTranslator(inner, log.NewTestLogger(), []Translator{spy})
+
+	msg := &workflowservice.DescribeWorkflowExecutionResponse{}
+	require.NoError(t, st.SendMsg(msg))
+
+	require.Equal(t, 1, spy.translateRespCalls)
+	require.Equal(t, []any{nil}, spy.pairedReqs, "stream messages have no request to pair with")
+	require.Equal(t, []any{msg}, inner.sent)
+}
diff --git a/interceptor/translator.go b/interceptor/translator.go
index eb6f8704..5d432ab2 100644
--- a/interceptor/translator.go
+++ b/interceptor/translator.go
@@ -10,7 +10,10 @@ type (
 	Translator interface {
 		MatchMethod(string) bool
 		TranslateRequest(any) (bool, error)
-		TranslateResponse(any) (bool, error)
+		// TranslateResponse receives the request paired with resp, so that a translator can
+		// take context from it that the response itself does not carry. req is nil for
+		// streams, whose messages have no request to pair with.
+		TranslateResponse(req, resp any) (bool, error)
 		Kind() string
 	}
 
@@ -47,7 +50,7 @@ func (n *translatorImpl) TranslateRequest(req any) (bool, error) {
 	return n.visitor(n.logger, req, n.matchReq)
 }
 
-func (n *translatorImpl) TranslateResponse(resp any) (bool, error) {
+func (n *translatorImpl) TranslateResponse(_, resp any) (bool, error) {
 	return n.visitor(n.logger, resp, n.matchResp)
 }
 
diff --git a/metrics/prometheus_defs.go b/metrics/prometheus_defs.go
index fbb26e04..57779a33 100644
--- a/metrics/prometheus_defs.go
+++ b/metrics/prometheus_defs.go
@@ -75,10 +75,20 @@ var (
 	TranslationErrors  = DefaultCounterVec("translation_error", "Count of message translation errors", translationLabels...)
 	TranslationLatency = DefaultHistogramVec("translation_latency", "Latency of message translations", translationLabels...)
 
+	// SearchAttrTranslationSkipped counts search attribute sites the translator deliberately
+	// left alone. It is not emitted for the expected steady state where a namespace resolves
+	// but has no configured mapping, which is true of every namespace not being migrated.
+	// The namespace id is logged rather than labelled, since its cardinality is unbounded.
+	SearchAttrTranslationSkipped = DefaultCounterVec("search_attribute_translation_skipped",
+		"Count of search attribute sites left untranslated", "reason", "message_type")
+
 	UTF8RepairTranslationKind = "utf8repair"
 	NamespaceTranslationKind  = "namespace"
 	SearchAttrTranslationKind = "search-attribute"
 	HistoryBlobMessageType    = "HistoryEventBlob"
+
+	SkipReasonUnresolvedNamespace = "unresolved_namespace"
+	SkipReasonUnsupportedType     = "unsupported_type"
 )
 
 // GetGRPCClientMetrics helps the GRPC client metrics objects feel more like the server one
@@ -139,4 +149,5 @@ func init() {
 	prometheus.MustRegister(TranslationCount)
 	prometheus.MustRegister(TranslationErrors)
 	prometheus.MustRegister(TranslationLatency)
+	prometheus.MustRegister(SearchAttrTranslationSkipped)
 }
diff --git a/proxy/cluster_connection.go b/proxy/cluster_connection.go
index f93a2da6..04b858c7 100644
--- a/proxy/cluster_connection.go
+++ b/proxy/cluster_connection.go
@@ -388,9 +388,6 @@ func makeServerOptions(c serverConfiguration, tlsConfig encryption.TLSConfig) ([
 
 	if c.saTranslations.LenNamespaces() > 0 {
 		c.loggers.Get(LogClusterConnection).Info("search attribute translation enabled", tag.NewAnyTag("mappings", c.saTranslations))
-		if c.saTranslations.LenNamespaces() > 1 {
-			panic("multiple namespace search attribute mappings are not supported")
-		}
 		translators = append(translators, interceptor.NewSearchAttributeTranslator(c.loggers.Get(LogInterceptor),
 			c.saTranslations.FlattenMaps(), c.saTranslations.Inverse().FlattenMaps()))
 	}
diff --git a/proxy/cluster_connection_test.go b/proxy/cluster_connection_test.go
index fcc52484..3ad8dd7a 100644
--- a/proxy/cluster_connection_test.go
+++ b/proxy/cluster_connection_test.go
@@ -15,7 +15,9 @@ import (
 	"go.temporal.io/server/common/log/tag"
 	"google.golang.org/grpc"
 
+	"github.com/temporalio/s2s-proxy/collect"
 	"github.com/temporalio/s2s-proxy/config"
+	"github.com/temporalio/s2s-proxy/encryption"
 	"github.com/temporalio/s2s-proxy/endtoendtest/testservices"
 	"github.com/temporalio/s2s-proxy/logging"
 	"github.com/temporalio/s2s-proxy/metrics"
@@ -361,3 +363,65 @@ func TestTCPListenerIsReleasedWhenNeverStarted(t *testing.T) {
 	}
 	runtime.KeepAlive(cc)
 }
+
+// TestMakeServerOptionsMultiNamespaceSATranslation guards the removal of the former
+// panic("multiple namespace search attribute mappings are not supported"), which fired on the
+// startup path and so crash-looped the whole proxy -- including namespaces that used no custom
+// search attributes.
+func TestMakeServerOptionsMultiNamespaceSATranslation(t *testing.T) {
+	saMappings := func(pairs ...config.SANamespaceMapping) config.SearchAttributeTranslation {
+		t.Helper()
+		translation, err := (&config.SATranslationConfig{NamespaceMappings: pairs}).AsLocalToRemoteSATranslation()
+		require.NoError(t, err)
+		return translation
+	}
+
+	for _, tc := range []struct {
+		name         string
+		saTranslator config.SearchAttributeTranslation
+	}{
+		{
+			name: "two namespaces",
+			saTranslator: saMappings(
+				config.SANamespaceMapping{
+					Name:        "ns-a",
+					NamespaceId: "11111111-1111-1111-1111-111111111111",
+					Mappings:    []config.SAMapping{{LocalName: "Keyword01", RemoteName: "TestSA"}},
+				},
+				config.SANamespaceMapping{
+					Name:        "ns-b",
+					NamespaceId: "22222222-2222-2222-2222-222222222222",
+					Mappings:    []config.SAMapping{{LocalName: "Keyword02", RemoteName: "TestSA"}},
+				},
+			),
+		},
+		{
+			name: "one namespace",
+			saTranslator: saMappings(config.SANamespaceMapping{
+				Name:        "ns-a",
+				NamespaceId: "11111111-1111-1111-1111-111111111111",
+				Mappings:    []config.SAMapping{{LocalName: "Keyword01", RemoteName: "TestSA"}},
+			}),
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			// nsTranslations is an interface, so it must be non-nil: makeServerOptions calls
+			// Len() on it unconditionally.
+			emptyNsTranslations, err := collect.NewStaticBiMap(func(func(string, string) bool) {}, 0)
+			require.NoError(t, err)
+
+			cfg := serverConfiguration{
+				directionLabel: "outbound",
+				nsTranslations: emptyNsTranslations,
+				saTranslations: tc.saTranslator,
+				loggers:        logging.NewLoggerProvider(log.NewTestLogger(), config.NewMockConfigProvider(config.S2SProxyConfig{})),
+			}
+
+			require.NotPanics(t, func() {
+				opts, err := makeServerOptions(cfg, encryption.TLSConfig{})
+				require.NoError(t, err)
+				require.NotEmpty(t, opts)
+			})
+		})
+	}
+}