unwrap ops and add --reverse testing

Author: spkane31

cliext/client.go

@@ -29,6 +29,11 @@ type ClientOptionsBuilder struct {
 	// Logger is the slog logger to use for the client. If set, it will be
 	// wrapped with the SDK's structured logger adapter.
 	Logger *slog.Logger
+
+	// PayloadCodec is populated by Build when a remote payload codec is
+	// configured. Callers can use it to decode payloads outside the gRPC
+	// interceptor chain (e.g. payloads nested inside opaque proto bytes).
+	PayloadCodec converter.PayloadCodec
 }
 
 type oauthCredentials struct {
@@ -248,13 +253,18 @@ func (b *ClientOptionsBuilder) Build(ctx context.Context) (client.Options, error
 		if err != nil {
 			return client.Options{}, fmt.Errorf("invalid codec headers: %w", err)
 		}
-		interceptor, err := newPayloadCodecInterceptor(
+		payloadCodec := newRemotePayloadCodec(
 			profile.Namespace, profile.Codec.Endpoint, profile.Codec.Auth, codecHeaders)
+		interceptor, err := converter.NewPayloadCodecGRPCClientInterceptor(
+			converter.PayloadCodecGRPCClientInterceptorOptions{
+				Codecs: []converter.PayloadCodec{payloadCodec},
+			})
 		if err != nil {
 			return client.Options{}, fmt.Errorf("failed creating payload codec interceptor: %w", err)
 		}
 		clientOpts.ConnectionOptions.DialOptions = append(
 			clientOpts.ConnectionOptions.DialOptions, grpc.WithChainUnaryInterceptor(interceptor))
+		b.PayloadCodec = payloadCodec
 	}
 
 	// Set connect timeout for GetSystemInfo if provided.
@@ -278,16 +288,17 @@ func parseKeyValuePairs(pairs []string) (map[string]string, error) {
 	return result, nil
 }
 
-// newPayloadCodecInterceptor creates a gRPC interceptor for remote payload codec.
-func newPayloadCodecInterceptor(
+// newRemotePayloadCodec constructs a remote payload codec from the configured endpoint,
+// auth, and headers. The returned codec can be used both inside a gRPC interceptor and
+// to decode payloads nested inside opaque proto bytes (e.g. system Nexus operation inputs).
+func newRemotePayloadCodec(
 	namespace string,
 	codecEndpoint string,
 	codecAuth string,
 	codecHeaders map[string]string,
-) (grpc.UnaryClientInterceptor, error) {
+) converter.PayloadCodec {
 	codecEndpoint = strings.ReplaceAll(codecEndpoint, "{namespace}", namespace)
-
-	payloadCodec := converter.NewRemotePayloadCodec(
+	return converter.NewRemotePayloadCodec(
 		converter.RemotePayloadCodecOptions{
 			Endpoint: codecEndpoint,
 			ModifyRequest: func(req *http.Request) error {
@@ -302,11 +313,6 @@ func newPayloadCodecInterceptor(
 			},
 		},
 	)
-	return converter.NewPayloadCodecGRPCClientInterceptor(
-		converter.PayloadCodecGRPCClientInterceptorOptions{
-			Codecs: []converter.PayloadCodec{payloadCodec},
-		},
-	)
 }
 
 func (c *oauthCredentials) getToken(ctx context.Context) (string, error) {

internal/temporalcli/client.go

@@ -21,8 +21,17 @@ import (
 // so often used by callers after this call to know the currently configured
 // namespace.
 func dialClient(cctx *CommandContext, c *cliext.ClientOptions) (client.Client, error) {
+	cl, _, err := dialClientWithCodec(cctx, c)
+	return cl, err
+}
+
+// dialClientWithCodec is like [dialClient] but also returns the configured remote
+// payload codec, or nil if no codec is configured. The codec is the same instance
+// used by the gRPC interceptor; callers can use it to decode payloads nested inside
+// opaque proto bytes (e.g. the request/response of a system Nexus operation).
+func dialClientWithCodec(cctx *CommandContext, c *cliext.ClientOptions) (client.Client, converter.PayloadCodec, error) {
 	if cctx.RootCommand == nil {
-		return nil, fmt.Errorf("root command unexpectedly missing when dialing client")
+		return nil, nil, fmt.Errorf("root command unexpectedly missing when dialing client")
 	}
 
 	// Set default identity if not provided
@@ -47,7 +56,7 @@ func dialClient(cctx *CommandContext, c *cliext.ClientOptions) (client.Client, e
 	}
 	clientOpts, err := builder.Build(cctx)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	// We do not put codec on data converter here, it is applied via
@@ -78,14 +87,14 @@ func dialClient(cctx *CommandContext, c *cliext.ClientOptions) (client.Client, e
 
 	cl, err := client.DialContext(dialCtx, clientOpts)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	// Since this namespace value is used by many commands after this call,
 	// we are mutating it to be the derived one
 	c.Namespace = clientOpts.Namespace
 
-	return cl, nil
+	return cl, builder.PayloadCodec, nil
 }
 
 func fixedHeaderOverrideInterceptor(

internal/temporalcli/commands.system_nexus.go

@@ -0,0 +1,48 @@
+package temporalcli
+
+import (
+	"context"
+
+	commonpb "go.temporal.io/api/common/v1"
+	"go.temporal.io/api/proxy"
+	"go.temporal.io/api/workflowservice/v1"
+	"go.temporal.io/sdk/converter"
+	"google.golang.org/protobuf/proto"
+)
+
+// systemNexusOpKey identifies a system Nexus operation by its (endpoint, operation) pair.
+type systemNexusOpKey struct {
+	Endpoint  string
+	Operation string
+}
+
+// systemNexusOpTypes maps a system Nexus operation to the proto request and response types
+// whose bytes are serialized in NexusOperationScheduled.Input and NexusOperationCompleted.Result.
+type systemNexusOpTypes struct {
+	// NewRequest returns a fresh, zero-valued instance of the request proto.
+	NewRequest func() proto.Message
+	// NewResponse returns a fresh, zero-valued instance of the response proto.
+	NewResponse func() proto.Message
+}
+
+// systemNexusOps is the global registry of known system Nexus operations on the
+// __temporal_system endpoint. Add new entries here as the server adds support for more
+// system operations. The keys' Operation values must match what the server records in
+// NexusOperationScheduledEventAttributes.Operation.
+var systemNexusOps = map[systemNexusOpKey]systemNexusOpTypes{
+	{Endpoint: temporalSystemNexusEndpoint, Operation: "SignalWithStartWorkflowExecution"}: {
+		NewRequest:  func() proto.Message { return &workflowservice.SignalWithStartWorkflowExecutionRequest{} },
+		NewResponse: func() proto.Message { return &workflowservice.SignalWithStartWorkflowExecutionResponse{} },
+	},
+}
+
+// decodePayloadsInProto walks a proto message and applies codec.Decode to every Payload
+// found inside it (including nested messages). The message is mutated in place.
+func decodePayloadsInProto(ctx context.Context, msg proto.Message, codec converter.PayloadCodec) error {
+	return proxy.VisitPayloads(ctx, msg, proxy.VisitPayloadsOptions{
+		SkipSearchAttributes: true,
+		Visitor: func(_ *proxy.VisitPayloadsContext, payloads []*commonpb.Payload) ([]*commonpb.Payload, error) {
+			return codec.Decode(payloads)
+		},
+	})
+}