Allow handover and redirection interceptors to cover additional service prefixes (#10963)

## What changed?
`NamespaceHandoverInterceptor` and `Redirection` short-circuit on the
`WorkflowService` prefix,
so requests to other frontend gRPC services skip handover gating and
cross-cell redirection. Add
two opt-in, default-no-op seams (WorkflowService behavior unchanged):
- `NamespaceHandoverInterceptor.WithAdditionalServicePrefixes(...)` —
gate additional service prefixes.
- `Redirection.WithRedirectResponses(...)` — register methods as
redirectable, keyed by full gRPC
method; namespace resolved via the existing
`NamespaceIDGetter`/`NamespaceNameGetter` path, fails closed.

Neither references any embedding service; callers inject the
prefix/response map.

## Why?
Workflow-scoped requests on other frontend services must observe the
same handover wait and
active-cluster redirection as `WorkflowService`, or they can be served
on a cell that no longer
owns the partition during/after a handover. The prefix short-circuit
made the interceptors no-ops
for anything but `WorkflowService`.

## How did you test it?
- [x] built
- [x] covered by existing tests
- [ ] run locally and tested manually
- [ ] added new unit test(s)
- [ ] added new functional test(s)

Existing tests confirm the default (no-op) behavior is unchanged; the
opt-in path is exercised by a downstream functional test.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: jiechenz

common/rpc/interceptor/namespace_handover.go

@@ -34,6 +34,12 @@ type (
 		logger                                 log.Logger
 		requestErrorHandler                    ErrorHandler
 		additionalAllowedMethodsDuringHandover map[string]struct{}
+		// additionalServicePrefixes are gRPC service prefixes (besides WorkflowService) whose
+		// methods the handover gate also applies to. Empty by default; embedders set these via
+		// WithAdditionalServicePrefixes so the gate can cover other transports (e.g. a proxy
+		// service) whose requests already expose their namespace, without this package knowing
+		// about them.
+		additionalServicePrefixes []string
 	}
 )
 
@@ -64,6 +70,29 @@ func NewNamespaceHandoverInterceptor(
 	}
 }
 
+// WithAdditionalServicePrefixes returns a copy of the interceptor whose handover gate also applies
+// to methods under the given gRPC service prefixes (besides WorkflowService). Embedders use this to
+// extend the gate to other transports without this package referencing them.
+func (i *NamespaceHandoverInterceptor) WithAdditionalServicePrefixes(prefixes ...string) *NamespaceHandoverInterceptor {
+	clone := *i
+	clone.additionalServicePrefixes = append(append([]string{}, i.additionalServicePrefixes...), prefixes...)
+	return &clone
+}
+
+// handlesMethod reports whether the handover gate applies to fullMethod: always for WorkflowService,
+// plus any embedder-configured service prefixes.
+func (i *NamespaceHandoverInterceptor) handlesMethod(fullMethod string) bool {
+	if strings.HasPrefix(fullMethod, api.WorkflowServicePrefix) {
+		return true
+	}
+	for _, prefix := range i.additionalServicePrefixes {
+		if strings.HasPrefix(fullMethod, prefix) {
+			return true
+		}
+	}
+	return false
+}
+
 func (i *NamespaceHandoverInterceptor) Intercept(
 	ctx context.Context,
 	req any,
@@ -72,7 +101,7 @@ func (i *NamespaceHandoverInterceptor) Intercept(
 ) (_ any, retError error) {
 	defer log.CapturePanic(i.logger, &retError)
 
-	if !strings.HasPrefix(info.FullMethod, api.WorkflowServicePrefix) {
+	if !i.handlesMethod(info.FullMethod) {
 		return handler(ctx, req)
 	}
 

common/rpc/interceptor/redirection.go

@@ -181,6 +181,10 @@ type (
 		clientBean         client.Bean
 		metricsHandler     metrics.Handler
 		timeSource         clock.TimeSource
+		// redirectResponsesByFullMethod registers embedder methods (keyed by full gRPC method, so
+		// they don't collide with the bareredirectResponsesByFullMethod-name maps) as globally-redirectable, each mapped to its
+		// response constructor. Nil by default, preserving the WorkflowService-only behavior.
+		redirectResponsesByFullMethod map[string]responseConstructorFn
 	}
 )
 
@@ -215,6 +219,19 @@ func NewRedirection(
 	}
 }
 
+// WithRedirectResponses returns a copy of the interceptor that treats the given fullMethod ->
+// response-constructor entries as globally-redirectable APIs, keyed by full gRPC method so they
+// don't collide with the bare-method-name maps. The registered requests must expose their
+// namespace (via NamespaceNameGetter/NamespaceIDGetter) so it can be resolved for redirection.
+func (i *Redirection) WithRedirectResponses(responses map[string]func() any) *Redirection {
+	clone := *i
+	clone.redirectResponsesByFullMethod = make(map[string]responseConstructorFn, len(responses))
+	for fullMethod, ctor := range responses {
+		clone.redirectResponsesByFullMethod[fullMethod] = ctor
+	}
+	return &clone
+}
+
 var _ grpc.UnaryServerInterceptor = (*Redirection)(nil).Intercept
 
 func (i *Redirection) Intercept(
@@ -224,6 +241,19 @@ func (i *Redirection) Intercept(
 	handler grpc.UnaryHandler,
 ) (_ any, retError error) {
 	defer log.CapturePanic(i.logger, &retError)
+	if raFn, ok := i.redirectResponsesByFullMethod[info.FullMethod]; ok {
+		if !i.RedirectionAllowed(ctx) {
+			return handler(ctx, req)
+		}
+		// Resolve the namespace exactly like the WorkflowService global path below; the registered
+		// request must expose it via NamespaceNameGetter/NamespaceIDGetter. Fails closed (returns
+		// the error) rather than running a global-namespace request on a possibly non-owning cell.
+		namespaceName, err := GetNamespaceName(i.namespaceCache, req)
+		if err != nil {
+			return nil, err
+		}
+		return i.handleRedirectAPIInvocation(ctx, req, info, handler, api.MethodName(info.FullMethod), raFn, namespaceName)
+	}
 
 	if !strings.HasPrefix(info.FullMethod, api.WorkflowServicePrefix) {
 		return handler(ctx, req)