feat(workflow): propagate TypeInfo through workflows

Author: THardy98

packages/client/src/workflow-client.ts

@@ -11,6 +11,8 @@ import type {
   WorkflowResultType,
   WorkflowIdConflictPolicy,
   WorkflowSerializationContext,
+  WorkflowTypeOptions,
+  PayloadTypeInfo,
 } 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';
@@ -360,6 +362,16 @@ export interface WorkflowResultOptions {
    * @default true
    */
   followRuns?: boolean;
+
+  /**
+   * Type information used to decode the Workflow result.
+   *
+   * This is only needed when getting a result from an existing Workflow handle or
+   * when the Workflow definition is not available to this client.
+   *
+   * @experimental
+   */
+  typeInfo?: PayloadTypeInfo;
 }
 
 /**
@@ -550,13 +562,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,
+      typeInfo: workflowTypeOptions.typeInfo,
+    };
+    const compiledOptions = compileWorkflowOptions(ensureArgs(workflowOptions));
     const adaptedInterceptors = interceptors.map((i) => adaptWorkflowClientInterceptor(i));
 
     const startWithDetails = composeInterceptors(
@@ -568,7 +583,7 @@ export class WorkflowClient extends BaseClient {
     return startWithDetails({
       options: compiledOptions,
       headers: {},
-      workflowType,
+      workflowType: workflowTypeOptions.type,
     });
   }
 
@@ -577,10 +592,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, typeInfo } = extractWorkflowTypeAndConfig(workflowTypeOrFunc, rest.typeInfo);
     assertRequiredWorkflowOptions(rest);
-    const compiledOptions = compileWorkflowOptions(ensureArgs(rest));
+    const workflowOptions = {
+      ...rest,
+      typeInfo,
+    };
+    const compiledOptions = compileWorkflowOptions(ensureArgs(workflowOptions));
     const signalWithStart = composeInterceptors(
       interceptors,
       'signalWithStart',
@@ -607,7 +626,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.typeInfo);
+    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({
@@ -617,6 +637,7 @@ export class WorkflowClient extends BaseClient {
       runIdForResult: wfStartOutput.runId,
       interceptors,
       followRuns: options.followRuns ?? true,
+      typeInfo: workflowTypeOptions.typeInfo,
     });
     return {
       ...baseHandle,
@@ -642,6 +663,7 @@ export class WorkflowClient extends BaseClient {
   ): Promise<WorkflowHandleWithSignaledRunId<WorkflowFn>> {
     const { workflowId } = options;
     const interceptors = this.getOrMakeInterceptors(workflowId);
+    const workflowTypeOptions = extractWorkflowTypeAndConfig(workflowTypeOrFunc, options.typeInfo);
     const runId = await this._signalWithStart(workflowTypeOrFunc, options, 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.
@@ -652,6 +674,7 @@ export class WorkflowClient extends BaseClient {
       runIdForResult: runId,
       interceptors,
       followRuns: options.followRuns ?? true,
+      typeInfo: workflowTypeOptions.typeInfo,
     }) as WorkflowHandleWithSignaledRunId<WorkflowFn>; // Cast is safe because we know we add the signaledRunId below
     (handle as any) /* readonly */.signaledRunId = runId;
     return handle;
@@ -736,11 +759,16 @@ export class WorkflowClient extends BaseClient {
       throw new Error('This WithStartWorkflowOperation instance has already been executed.');
     }
     startWorkflowOperation[withStartWorkflowOperationUsed] = true;
+    const { type: workflowType, typeInfo } = extractWorkflowTypeAndConfig(workflowTypeOrFunc, workflowOptions.typeInfo);
     assertRequiredWorkflowOptions(workflowOptions);
 
+    const resolvedWorkflowOptions = {
+      ...workflowOptions,
+      typeInfo,
+    };
     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 ?? [],
@@ -757,6 +785,7 @@ export class WorkflowClient extends BaseClient {
           firstExecutionRunId: startResponse.runId ?? undefined,
           interceptors,
           followRuns: workflowOptions.followRuns ?? true,
+          typeInfo,
         })
       );
 
@@ -798,10 +827,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.typeInfo);
+    await this._start(workflowTypeOptions, options, interceptors);
     return await this.result(workflowId, undefined, {
       ...options,
       followRuns: options.followRuns ?? true,
+      typeInfo: workflowTypeOptions.typeInfo,
     });
   }
 
@@ -857,12 +888,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?.typeInfo?.outputType
         );
-        return result as any;
+        return result;
       } else if (ev.workflowExecutionFailedEventAttributes) {
         if (followRuns && ev.workflowExecutionFailedEventAttributes.newExecutionRunId) {
           execution.runId = ev.workflowExecutionFailedEventAttributes.newExecutionRunId;
@@ -1335,7 +1368,9 @@ 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.typeInfo?.inputTypes),
+      },
       signalName,
       signalInput: { payloads: await encodeToPayloadsWithContext(dataConverter, context, signalArgs) },
       taskQueue: {
@@ -1465,7 +1500,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.typeInfo?.inputTypes),
+      },
       taskQueue: {
         kind: temporal.api.enums.v1.TaskQueueKind.TASK_QUEUE_KIND_NORMAL,
         name: opts.taskQueue,
@@ -1752,6 +1789,7 @@ export class WorkflowClient extends BaseClient {
       runIdForResult: runId ?? options?.firstExecutionRunId,
       interceptors,
       followRuns: options?.followRuns ?? true,
+      typeInfo: options?.typeInfo,
     });
   }
 

packages/common/src/__tests__/test-workflow-type-info.ts

@@ -0,0 +1,47 @@
+import test from 'ava';
+import type { PayloadTypeInfo } from '../type-info';
+import { extractWorkflowType, extractWorkflowTypeAndConfig } from '../workflow-options';
+
+const typeInfo: PayloadTypeInfo = {
+  inputTypes: [{ hint: { converter: 'json' } }],
+  outputType: { hint: { converter: 'json' } },
+};
+
+test('resolves call-site type information for a string Workflow type', (t) => {
+  t.deepEqual(extractWorkflowTypeAndConfig('workflow', typeInfo), {
+    type: 'workflow',
+    typeInfo,
+  });
+});
+
+test('resolves definition-supplied type information for a Workflow function', (t) => {
+  const workflow = Object.assign(async function workflow(): Promise<void> {}, {
+    staticOptions: { typeInfo },
+  });
+
+  t.deepEqual(extractWorkflowTypeAndConfig(workflow), {
+    type: 'workflow',
+    typeInfo,
+  });
+  t.is(extractWorkflowType(workflow), 'workflow');
+});
+
+test('rejects call-site type information for a Workflow function', (t) => {
+  async function workflow(): Promise<void> {}
+
+  t.throws(() => extractWorkflowTypeAndConfig(workflow, typeInfo), {
+    instanceOf: TypeError,
+    message: /Workflow type information cannot be supplied at the call site/,
+  });
+});
+
+test('ignores explicitly undefined static options', (t) => {
+  const workflow = Object.assign(async function workflow(): Promise<void> {}, {
+    staticOptions: undefined,
+  });
+
+  t.deepEqual(extractWorkflowTypeAndConfig(workflow), {
+    type: 'workflow',
+    typeInfo: undefined,
+  });
+});

packages/common/src/workflow-definition-options.ts

@@ -1,5 +1,18 @@
+import type { PayloadTypeInfo } from './type-info';
 import type { VersioningBehavior } from './worker-deployments';
 
+/**
+ * Options that can be attached to a Workflow function at definition time.
+ *
+ * @experimental
+ */
+export interface WorkflowDefinitionConfig {
+  /** Options that may be evaluated for each Workflow Execution. */
+  workflowDefinitionOptions?: WorkflowDefinitionOptionsOrGetter;
+  /** Static metadata that must be available before Workflow input decoding. */
+  staticOptions?: WorkflowStaticOptions;
+}
+
 /**
  * Options that can be used when defining a workflow via {@link setWorkflowOptions}.
  */
@@ -30,6 +43,15 @@ export interface WorkflowDefinitionOptions {
   failureExceptionTypes?: Array<new (...args: any[]) => Error>;
 }
 
+export interface WorkflowStaticOptions {
+  /**
+   * Type information used to encode and decode Workflow input and output.
+   *
+   * @experimental
+   */
+  typeInfo?: PayloadTypeInfo;
+}
+
 type AsyncFunction<Args extends any[], ReturnType> = (...args: Args) => Promise<ReturnType>;
 export type WorkflowDefinitionOptionsOrGetter = WorkflowDefinitionOptions | (() => WorkflowDefinitionOptions);
 
@@ -41,3 +63,31 @@ export type WorkflowDefinitionOptionsOrGetter = WorkflowDefinitionOptions | (()
 export interface WorkflowFunctionWithOptions<Args extends any[], ReturnType> extends AsyncFunction<Args, ReturnType> {
   workflowDefinitionOptions: WorkflowDefinitionOptionsOrGetter;
 }
+
+/** @internal */
+export interface WorkflowFunctionWithStaticOptions<Args extends any[], ReturnType>
+  extends AsyncFunction<Args, ReturnType> {
+  staticOptions: WorkflowStaticOptions;
+}
+
+const workflowDefinitionOptionsProperty = 'workflowDefinitionOptions' satisfies keyof WorkflowFunctionWithOptions<
+  any[],
+  any
+>;
+const workflowStaticOptionsProperty = 'staticOptions' satisfies keyof WorkflowFunctionWithStaticOptions<any[], any>;
+
+/** @internal */
+export function isWorkflowFunctionWithOptions(obj: unknown): obj is WorkflowFunctionWithOptions<any[], any> {
+  return typeof obj === 'function' && Object.hasOwn(obj, workflowDefinitionOptionsProperty);
+}
+
+/** @internal */
+export function isWorkflowFunctionWithStaticOptions(
+  obj: unknown
+): obj is WorkflowFunctionWithStaticOptions<any[], any> {
+  if (typeof obj !== 'function' || !Object.hasOwn(obj, workflowStaticOptionsProperty)) {
+    return false;
+  }
+  const { staticOptions } = obj as { staticOptions?: unknown };
+  return typeof staticOptions === 'object' && staticOptions !== null;
+}

packages/common/src/workflow-options.ts

@@ -5,8 +5,8 @@ import type { Duration } from './time';
 import { makeProtoEnumConverters } from './internal-workflow';
 import type { SearchAttributePair, SearchAttributes, TypedSearchAttributes } from './search-attributes';
 import type { Priority } from './priority';
-import type { WorkflowFunctionWithOptions } from './workflow-definition-options';
-
+import { isWorkflowFunctionWithStaticOptions } from './workflow-definition-options';
+import type { PayloadTypeInfo } from './type-info';
 /**
  * Defines what happens when trying to start a Workflow with the same ID as a *Closed* Workflow.
  *
@@ -208,6 +208,13 @@ export interface BaseWorkflowOptions {
    * Priority of a workflow
    */
   priority?: Priority;
+
+  /**
+   * Type information used to encode and decode Workflow input and output.
+   *
+   * @experimental
+   */
+  typeInfo?: PayloadTypeInfo;
 }
 
 export type WithWorkflowArgs<W extends Workflow, T> = T &
@@ -255,15 +262,42 @@ export interface WorkflowDurationOptions {
 
 export type CommonWorkflowOptions = BaseWorkflowOptions & WorkflowDurationOptions;
 
-export function extractWorkflowType<T extends Workflow>(
-  workflowTypeOrFunc: string | T | WorkflowFunctionWithOptions<any[], any>
-): string {
-  if (typeof workflowTypeOrFunc === 'string') return workflowTypeOrFunc as string;
+export interface WorkflowTypeOptions {
+  type: string;
+  typeInfo?: PayloadTypeInfo;
+}
+
+export function extractWorkflowTypeAndConfig<T extends Workflow>(
+  workflowTypeOrFunc: string | T,
+  callSiteTypeInfo?: PayloadTypeInfo
+): WorkflowTypeOptions {
+  if (typeof workflowTypeOrFunc === 'string') {
+    return { type: workflowTypeOrFunc, typeInfo: callSiteTypeInfo };
+  }
   if (typeof workflowTypeOrFunc === 'function') {
-    if (workflowTypeOrFunc?.name) return workflowTypeOrFunc.name;
-    throw new TypeError('Invalid workflow type: the workflow function is anonymous');
+    if (!workflowTypeOrFunc.name) {
+      throw new TypeError('Invalid workflow type: the workflow function is anonymous');
+    }
+    if (callSiteTypeInfo !== undefined) {
+      throw new TypeError(
+        'Workflow type information cannot be supplied at the call site when using a workflow function. ' +
+          'Use defineWorkflowOptions(..., { staticOptions: { typeInfo } }) on the workflow function instead, ' +
+          'or pass the workflow type as a string.'
+      );
+    }
+    const definitionTypeInfo = isWorkflowFunctionWithStaticOptions(workflowTypeOrFunc)
+      ? workflowTypeOrFunc.staticOptions.typeInfo
+      : undefined;
+    return {
+      type: workflowTypeOrFunc.name,
+      typeInfo: definitionTypeInfo,
+    };
   }
   throw new TypeError(
     `Invalid workflow type: expected either a string or a function, got '${typeof workflowTypeOrFunc}'`
   );
 }
+
+export function extractWorkflowType<T extends Workflow>(workflowTypeOrFunc: string | T): string {
+  return extractWorkflowTypeAndConfig(workflowTypeOrFunc).type;
+}