[DO NOT MERGE] POC for workflow type hints

packages/client/src/schedule-helpers.ts

@@ -5,7 +5,7 @@ import {
   compileRetryPolicy,
   decodePriority,
   decompileRetryPolicy,
-  extractWorkflowType,
+  extractWorkflowTypeAndConfig,
 } from '@temporalio/common';
 import { encodeUserMetadata, decodeUserMetadata } from '@temporalio/common/lib/internal-non-workflow/codec-helpers';
 import {
@@ -198,14 +198,16 @@ export function decodeOptionalStructuredCalendarSpecs(
 }
 
 export function compileScheduleOptions(options: ScheduleOptions): CompiledScheduleOptions {
-  const workflowType = extractWorkflowType(options.action.workflowType);
+  const { action } = options;
+  const { type: workflowType, typeHints } = extractWorkflowTypeAndConfig(action.workflowType, action.typeHints);
   return {
     ...options,
     action: {
-      ...options.action,
-      workflowId: options.action.workflowId ?? `${options.scheduleId}-workflow`,
+      ...action,
+      workflowId: action.workflowId ?? `${options.scheduleId}-workflow`,
       workflowType,
-      args: (options.action.args ?? []) as unknown[],
+      args: (action.args ?? []) as unknown[],
+      typeHints,
     },
   };
 }
@@ -214,15 +216,16 @@ export function compileUpdatedScheduleOptions(
   scheduleId: string,
   options: ScheduleUpdateOptions
 ): CompiledScheduleUpdateOptions {
-  const workflowTypeOrFunc = options.action.workflowType;
-  const workflowType = extractWorkflowType(workflowTypeOrFunc);
+  const { action } = options;
+  const { type: workflowType, typeHints } = extractWorkflowTypeAndConfig(action.workflowType, action.typeHints);
   return {
     ...options,
     action: {
-      ...options.action,
-      workflowId: options.action.workflowId ?? `${scheduleId}-workflow`,
+      ...action,
+      workflowId: action.workflowId ?? `${scheduleId}-workflow`,
       workflowType,
-      args: (options.action.args ?? []) as unknown[],
+      args: (action.args ?? []) as unknown[],
+      typeHints,
     },
   };
 }
@@ -264,7 +267,7 @@ export async function encodeScheduleAction(
       workflowType: {
         name: action.workflowType,
       },
-      input: { payloads: await encodeToPayloadsWithContext(dataConverter, context, action.args) },
+      input: { payloads: await encodeToPayloadsWithContext(dataConverter, context, action.args, action.typeHints?.inputTypes) },
       taskQueue: {
         kind: temporal.api.enums.v1.TaskQueueKind.TASK_QUEUE_KIND_NORMAL,
         name: action.taskQueue,
@@ -273,6 +276,7 @@ export async function encodeScheduleAction(
       workflowRunTimeout: msOptionalToTs(action.workflowRunTimeout),
       workflowTaskTimeout: msOptionalToTs(action.workflowTaskTimeout),
       retryPolicy: action.retry ? compileRetryPolicy(action.retry) : undefined,
+      // THOMAS - no support for memo yet
       memo: action.memo ? { fields: await encodeMapToPayloads(dataConverter, action.memo, context) } : undefined,
       searchAttributes:
         action.searchAttributes || action.typedSearchAttributes

packages/client/src/schedule-types.ts

@@ -6,6 +6,7 @@ import type {
   Workflow,
   TypedSearchAttributes,
   SearchAttributePair,
+  PayloadTypeHints,
 } from '@temporalio/common';
 import { makeProtoEnumConverters } from '@temporalio/common/lib/internal-workflow';
 import type { temporal } from '@temporalio/proto';
@@ -792,6 +793,7 @@ export type ScheduleOptionsStartWorkflowAction<W extends Workflow> = {
   | 'workflowTaskTimeout'
   | 'staticDetails'
   | 'staticSummary'
+  | 'typeHints'
 > & {
     /**
      * Workflow id to use when starting. Assign a meaningful business id.
@@ -838,7 +840,11 @@ export type CompiledScheduleAction = Replace<
     workflowType: string;
     args: unknown[];
   }
->;
+  > & {
+  // THOMAS - a bit hacky, needed due to CompiledScheduleAction being a derived
+  // type of ScheduleDescriptionAction for some reason
+  typeHints?: PayloadTypeHints;
+};
 
 /**
  * Policy for overlapping Actions.

packages/client/src/workflow-client.ts

@@ -11,6 +11,8 @@ import type {
   WorkflowResultType,
   WorkflowIdConflictPolicy,
   WorkflowSerializationContext,
+  WorkflowTypeOptions,
+  TypeHint,
 } from '@temporalio/common';
 import {
   CancelledFailure,
@@ -21,11 +23,11 @@ import {
   TimeoutType,
   WorkflowExecutionAlreadyStartedError,
   WorkflowNotFoundError,
-  extractWorkflowType,
   encodeWorkflowIdReusePolicy,
   decodeRetryState,
   encodeWorkflowIdConflictPolicy,
   compilePriority,
+  extractWorkflowTypeAndConfig,
 } from '@temporalio/common';
 import { encodeUserMetadata } from '@temporalio/common/lib/internal-non-workflow/codec-helpers';
 import { encodeUnifiedSearchAttributes } from '@temporalio/common/lib/converter/payload-search-attributes';
@@ -343,6 +345,16 @@ export interface WorkflowResultOptions {
    * @default true
    */
   followRuns?: boolean;
+
+  /**
+   * Type hint used to decode the Workflow result.
+   *
+   * This is only needed when getting a result from an existing Workflow handle or
+   * when overriding definition-supplied hints.
+   *
+   * @experimental
+   */
+  typeHint?: TypeHint;
 }
 
 /**
@@ -533,13 +545,16 @@ export class WorkflowClient extends BaseClient {
   }
 
   protected async _start<T extends Workflow>(
-    workflowTypeOrFunc: string | T,
+    workflowTypeOptions: WorkflowTypeOptions,
     options: WorkflowStartOptions<T>,
     interceptors: WorkflowClientInterceptor[]
   ): Promise<WorkflowStartOutput> {
-    const workflowType = extractWorkflowType(workflowTypeOrFunc);
     assertRequiredWorkflowOptions(options);
-    const compiledOptions = compileWorkflowOptions(ensureArgs(options));
+    const workflowOptions = {
+      ...options,
+      typeHints: workflowTypeOptions.typeHints,
+    };
+    const compiledOptions = compileWorkflowOptions(ensureArgs(workflowOptions));
     const adaptedInterceptors = interceptors.map((i) => adaptWorkflowClientInterceptor(i));
 
     const startWithDetails = composeInterceptors(
@@ -551,7 +566,7 @@ export class WorkflowClient extends BaseClient {
     return startWithDetails({
       options: compiledOptions,
       headers: {},
-      workflowType,
+      workflowType: workflowTypeOptions.type,
     });
   }
 
@@ -560,10 +575,14 @@ export class WorkflowClient extends BaseClient {
     options: WithWorkflowArgs<T, WorkflowSignalWithStartOptions<SA>>,
     interceptors: WorkflowClientInterceptor[]
   ): Promise<string> {
-    const workflowType = extractWorkflowType(workflowTypeOrFunc);
     const { signal, signalArgs, ...rest } = options;
+    const { type: workflowType, typeHints } = extractWorkflowTypeAndConfig(workflowTypeOrFunc, rest.typeHints);
     assertRequiredWorkflowOptions(rest);
-    const compiledOptions = compileWorkflowOptions(ensureArgs(rest));
+    const workflowOptions = {
+      ...rest,
+      typeHints,
+    };
+    const compiledOptions = compileWorkflowOptions(ensureArgs(workflowOptions));
     const signalWithStart = composeInterceptors(
       interceptors,
       'signalWithStart',
@@ -590,7 +609,8 @@ export class WorkflowClient extends BaseClient {
   ): Promise<WorkflowHandleWithStartDetails<T>> {
     const { workflowId } = options;
     const interceptors = this.getOrMakeInterceptors(workflowId);
-    const wfStartOutput = await this._start(workflowTypeOrFunc, { ...options, workflowId }, interceptors);
+    const workflowTypeOptions = extractWorkflowTypeAndConfig(workflowTypeOrFunc, options.typeHints);
+    const wfStartOutput = await this._start(workflowTypeOptions, { ...options, workflowId }, interceptors);
     // runId is not used in handles created with `start*` calls because these
     // handles should allow interacting with the workflow if it continues as new.
     const baseHandle = this._createWorkflowHandle({
@@ -600,6 +620,7 @@ export class WorkflowClient extends BaseClient {
       runIdForResult: wfStartOutput.runId,
       interceptors,
       followRuns: options.followRuns ?? true,
+      typeHint: workflowTypeOptions.typeHints?.outputType,
     });
     return {
       ...baseHandle,
@@ -719,11 +740,19 @@ export class WorkflowClient extends BaseClient {
       throw new Error('This WithStartWorkflowOperation instance has already been executed.');
     }
     startWorkflowOperation[withStartWorkflowOperationUsed] = true;
+    const { type: workflowType, typeHints } = extractWorkflowTypeAndConfig(
+      workflowTypeOrFunc,
+      workflowOptions.typeHints
+    );
     assertRequiredWorkflowOptions(workflowOptions);
 
+    const resolvedWorkflowOptions = {
+      ...workflowOptions,
+      typeHints,
+    };
     const startUpdateWithStartInput: WorkflowStartUpdateWithStartInput = {
-      workflowType: extractWorkflowType(workflowTypeOrFunc),
-      workflowStartOptions: compileWorkflowOptions(ensureArgs(workflowOptions)),
+      workflowType,
+      workflowStartOptions: compileWorkflowOptions(ensureArgs(resolvedWorkflowOptions)),
       workflowStartHeaders: {},
       updateName: typeof updateDef === 'string' ? updateDef : updateDef.name,
       updateArgs: args ?? [],
@@ -781,10 +810,12 @@ export class WorkflowClient extends BaseClient {
   ): Promise<WorkflowResultType<T>> {
     const { workflowId } = options;
     const interceptors = this.getOrMakeInterceptors(workflowId);
-    await this._start(workflowTypeOrFunc, options, interceptors);
+    const workflowTypeOptions = extractWorkflowTypeAndConfig(workflowTypeOrFunc, options.typeHints);
+    await this._start(workflowTypeOptions, options, interceptors);
     return await this.result(workflowId, undefined, {
       ...options,
       followRuns: options.followRuns ?? true,
+      typeHint: workflowTypeOptions.typeHints?.outputType,
     });
   }
 
@@ -838,12 +869,14 @@ export class WorkflowClient extends BaseClient {
         }
         // Note that we can only return one value from our workflow function in JS.
         // Ignore any other payloads in result
-        const [result] = await decodeArrayFromPayloads(
+        const result = await decodeFromPayloadsAtIndex<WorkflowResultType<T>>(
           dataConverter,
+          0,
           ev.workflowExecutionCompletedEventAttributes.result?.payloads,
-          context
+          context,
+          opts?.typeHint
         );
-        return result as any;
+        return result;
       } else if (ev.workflowExecutionFailedEventAttributes) {
         if (followRuns && ev.workflowExecutionFailedEventAttributes.newExecutionRunId) {
           execution.runId = ev.workflowExecutionFailedEventAttributes.newExecutionRunId;
@@ -1246,7 +1279,7 @@ export class WorkflowClient extends BaseClient {
       workflowIdReusePolicy: encodeWorkflowIdReusePolicy(options.workflowIdReusePolicy),
       workflowIdConflictPolicy: encodeWorkflowIdConflictPolicy(options.workflowIdConflictPolicy),
       workflowType: { name: workflowType },
-      input: { payloads: await encodeToPayloadsWithContext(dataConverter, context, options.args) },
+      input: { payloads: await encodeToPayloadsWithContext(dataConverter, context, options.args, options.typeHints?.inputTypes) },
       signalName,
       signalInput: { payloads: await encodeToPayloadsWithContext(dataConverter, context, signalArgs) },
       taskQueue: {
@@ -1258,6 +1291,7 @@ export class WorkflowClient extends BaseClient {
       workflowTaskTimeout: options.workflowTaskTimeout,
       workflowStartDelay: options.startDelay,
       retryPolicy: options.retry ? compileRetryPolicy(options.retry) : undefined,
+      // THOMAS - memo type hints not supported yet
       memo: options.memo ? { fields: await encodeMapToPayloads(dataConverter, options.memo, context) } : undefined,
       searchAttributes:
         options.searchAttributes || options.typedSearchAttributes
@@ -1352,7 +1386,9 @@ export class WorkflowClient extends BaseClient {
       workflowIdReusePolicy: encodeWorkflowIdReusePolicy(opts.workflowIdReusePolicy),
       workflowIdConflictPolicy: encodeWorkflowIdConflictPolicy(opts.workflowIdConflictPolicy),
       workflowType: { name: workflowType },
-      input: { payloads: await encodeToPayloadsWithContext(dataConverter, context, opts.args) },
+      input: {
+        payloads: await encodeToPayloadsWithContext(dataConverter, context, opts.args, opts.typeHints?.inputTypes),
+      },
       taskQueue: {
         kind: temporal.api.enums.v1.TaskQueueKind.TASK_QUEUE_KIND_NORMAL,
         name: opts.taskQueue,
@@ -1362,6 +1398,8 @@ export class WorkflowClient extends BaseClient {
       workflowTaskTimeout: opts.workflowTaskTimeout,
       workflowStartDelay: opts.startDelay,
       retryPolicy: opts.retry ? compileRetryPolicy(opts.retry) : undefined,
+      // TYPE HINTS: skipping memo for now
+      // (can support by adjusting typeHints field in BaseWorkflowOptions and WorkflowDefinitionConfig)
       memo: opts.memo ? { fields: await encodeMapToPayloads(dataConverter, opts.memo, context) } : undefined,
       searchAttributes:
         opts.searchAttributes || opts.typedSearchAttributes
@@ -1618,6 +1656,7 @@ export class WorkflowClient extends BaseClient {
       runIdForResult: runId ?? options?.firstExecutionRunId,
       interceptors,
       followRuns: options?.followRuns ?? true,
+      typeHint: options?.typeHint,
     });
   }