feat(workflow): support TypeInfo for workflow transitions

Author: THardy98

packages/test/src/test-type-info.ts

@@ -1,6 +1,6 @@
 import { randomUUID } from 'crypto';
 import type { ExecutionContext } from 'ava';
-import { Client, WithStartWorkflowOperation } from '@temporalio/client';
+import { Client, WithStartWorkflowOperation, WorkflowFailedError } from '@temporalio/client';
 import { workflowInterceptorModules } from '@temporalio/testing';
 import { bundleWorkflowCode } from '@temporalio/worker';
 import type { TestWorkflowEnvironment } from './helpers';
@@ -12,9 +12,13 @@ import {
   makeConfigurableEnvironmentTestFn,
 } from './helpers-integration';
 import {
+  parentWorkflowChildDefinition,
+  parentWorkflowChildDefinitionInvalidCallSiteTypeInfo,
+  continueAsNewToWorkflowWithTypeInfo,
   finishSignal,
   finishUpdate,
   Order,
+  parentWorkflowChildString,
   Receipt,
   workflowTypeInfo,
   workflowWithSignalStart,
@@ -158,6 +162,38 @@ test('detached workflow handle uses call-site output type information', async (t
   });
 });
 
+test('same-type continue-as-new reuses definition-supplied input and output type information', async (t) => {
+  const h = configurableHelpers(t, t.context.workflowBundle, t.context.env);
+  const client = makeClient(t.context.env);
+  const worker = await h.createWorker({ dataConverter });
+
+  await worker.runUntil(async () => {
+    const result = await client.workflow.execute(workflowWithTypeInfo, {
+      workflowId: `wf-${randomUUID()}`,
+      taskQueue: h.taskQueue,
+      args: [new Order('order-1', 12345n, 1)],
+    });
+
+    assertReceipt(t, result);
+  });
+});
+
+test('continue-as-new to a different workflow uses explicit input type information', async (t) => {
+  const h = configurableHelpers(t, t.context.workflowBundle, t.context.env);
+  const client = makeClient(t.context.env);
+  const worker = await h.createWorker({ dataConverter });
+
+  await worker.runUntil(async () => {
+    const result = await client.workflow.execute(continueAsNewToWorkflowWithTypeInfo, {
+      workflowId: `wf-${randomUUID()}`,
+      taskQueue: h.taskQueue,
+      args: [new Order('order-1', 12345n)],
+    });
+
+    assertReceipt(t, result);
+  });
+});
+
 test('signal-with-start carries definition-supplied workflow type information', async (t) => {
   const h = configurableHelpers(t, t.context.workflowBundle, t.context.env);
   const client = makeClient(t.context.env);
@@ -195,3 +231,54 @@ test('update-with-start carries definition-supplied workflow type information',
     assertReceipt(t, await (await startOperation.workflowHandle()).result());
   });
 });
+
+test('child workflow uses definition-supplied input and output type information', async (t) => {
+  const h = configurableHelpers(t, t.context.workflowBundle, t.context.env);
+  const client = makeClient(t.context.env);
+  const worker = await h.createWorker({ dataConverter });
+
+  await worker.runUntil(async () => {
+    const result = await client.workflow.execute(parentWorkflowChildDefinition, {
+      workflowId: `wf-${randomUUID()}`,
+      taskQueue: h.taskQueue,
+      args: [new Order('order-1', 12345n)],
+    });
+
+    assertReceipt(t, result);
+  });
+});
+
+test('child workflow uses call-site input and output type information for string workflow type', async (t) => {
+  const h = configurableHelpers(t, t.context.workflowBundle, t.context.env);
+  const client = makeClient(t.context.env);
+  const worker = await h.createWorker({ dataConverter });
+
+  await worker.runUntil(async () => {
+    const result = await client.workflow.execute(parentWorkflowChildString, {
+      workflowId: `wf-${randomUUID()}`,
+      taskQueue: h.taskQueue,
+      args: [new Order('order-1', 12345n)],
+    });
+
+    assertReceipt(t, result);
+  });
+});
+
+test('child workflow definition with call-site type information is invalid', async (t) => {
+  const h = configurableHelpers(t, t.context.workflowBundle, t.context.env);
+  const client = makeClient(t.context.env);
+  const worker = await h.createWorker({ dataConverter });
+
+  await worker.runUntil(async () => {
+    const err = await t.throwsAsync(
+      client.workflow.execute(parentWorkflowChildDefinitionInvalidCallSiteTypeInfo, {
+        workflowId: `wf-${randomUUID()}`,
+        taskQueue: h.taskQueue,
+        args: [new Order('order-1', 12345n)],
+      }),
+      { instanceOf: WorkflowFailedError }
+    );
+
+    t.regex(err?.cause?.message ?? '', /Workflow type information cannot be supplied at the call site/);
+  });
+});

packages/test/src/workflows/type-info.ts

@@ -1,5 +1,14 @@
 import type { PayloadTypeInfo, TypeInfo } from '@temporalio/common';
-import { condition, defineSignal, defineUpdate, defineWorkflowOptions, setHandler } from '@temporalio/workflow';
+import {
+  condition,
+  continueAsNew,
+  defineSignal,
+  defineUpdate,
+  defineWorkflowOptions,
+  executeChild,
+  makeContinueAsNewFunc,
+  setHandler,
+} from '@temporalio/workflow';
 
 export class Order {
   constructor(
@@ -75,14 +84,72 @@ function assertOrder(order: Order): void {
   }
 }
 
+function assertReceipt(receipt: Receipt): void {
+  if (!(receipt instanceof Receipt)) {
+    throw new Error('Expected Receipt result');
+  }
+  if (typeof receipt.totalCents !== 'bigint') {
+    throw new Error('Expected Receipt.totalCents to be a bigint');
+  }
+}
+
 defineWorkflowOptions(workflowWithTypeInfo, {
   staticOptions: { typeInfo: workflowTypeInfo },
 });
 export async function workflowWithTypeInfo(order: Order): Promise<Receipt> {
   assertOrder(order);
+  if (order.remainingRuns > 0) {
+    await continueAsNew(new Order(order.id, order.totalCents, order.remainingRuns - 1));
+  }
   return new Receipt(order.id, order.totalCents);
 }
 
+export async function parentWorkflowChildDefinition(order: Order): Promise<Receipt> {
+  assertOrder(order);
+  const receipt = await executeChild(workflowWithTypeInfo, { args: [order] });
+  assertReceipt(receipt);
+  return receipt;
+}
+defineWorkflowOptions(parentWorkflowChildDefinition, {
+  staticOptions: { typeInfo: workflowTypeInfo },
+});
+
+export async function parentWorkflowChildString(order: Order): Promise<Receipt> {
+  assertOrder(order);
+  const receipt = await executeChild('workflowWithTypeInfo', {
+    args: [order],
+    typeInfo: workflowTypeInfo,
+  });
+  assertReceipt(receipt);
+  return receipt;
+}
+defineWorkflowOptions(parentWorkflowChildString, {
+  staticOptions: { typeInfo: workflowTypeInfo },
+});
+
+export async function parentWorkflowChildDefinitionInvalidCallSiteTypeInfo(order: Order): Promise<void> {
+  await executeChild(workflowWithTypeInfo, {
+    args: [order],
+    typeInfo: workflowTypeInfo,
+  });
+}
+defineWorkflowOptions(parentWorkflowChildDefinitionInvalidCallSiteTypeInfo, {
+  workflowDefinitionOptions: { failureExceptionTypes: [TypeError] },
+  staticOptions: { typeInfo: workflowTypeInfo },
+});
+
+export async function continueAsNewToWorkflowWithTypeInfo(order: Order): Promise<Receipt> {
+  assertOrder(order);
+  const continueAsTypedWorkflow = makeContinueAsNewFunc<typeof workflowWithTypeInfo>({
+    workflowType: 'workflowWithTypeInfo',
+    typeInfo: { inputTypes: workflowTypeInfo.inputTypes },
+  });
+  return await continueAsTypedWorkflow(order);
+}
+defineWorkflowOptions(continueAsNewToWorkflowWithTypeInfo, {
+  staticOptions: { typeInfo: workflowTypeInfo },
+});
+
 export const finishSignal = defineSignal('finish');
 
 export async function workflowWithSignalStart(order: Order): Promise<Receipt> {

packages/workflow/src/interfaces.ts

@@ -16,6 +16,7 @@ import type {
   VersioningBehavior,
   InitialVersioningBehavior,
   SuggestContinueAsNewReason,
+  PayloadTypeInfo,
 } from '@temporalio/common';
 import { SymbolBasedInstanceOfError } from '@temporalio/common/lib/type-helpers';
 import { makeProtoEnumConverters } from '@temporalio/common/lib/internal-workflow/enums-helpers';
@@ -398,6 +399,17 @@ export interface ContinueAsNewOptions {
    * @experimental Versioning semantics with continue-as-new are experimental and may change in the future.
    */
   initialVersioningBehavior?: InitialVersioningBehavior;
+
+  /**
+   * Input type information for the next Workflow run.
+   *
+   * When continuing as new to the same Workflow type, its definition-supplied input type information is reused.
+   * The next Workflow must preserve the current Workflow's output contract because clients retain the original
+   * handle's output type information across the run chain.
+   *
+   * @experimental
+   */
+  typeInfo?: Pick<PayloadTypeInfo, 'inputTypes'>;
 }
 
 /**
@@ -554,6 +566,13 @@ export const [encodeParentClosePolicy, decodeParentClosePolicy] = makeProtoEnumC
 );
 
 export interface ChildWorkflowOptions extends Omit<CommonWorkflowOptions, 'workflowIdConflictPolicy'> {
+  /**
+   * Type information used to encode and decode Child Workflow input and output.
+   *
+   * @experimental
+   */
+  typeInfo?: PayloadTypeInfo;
+
   /**
    * Workflow id to use when starting. If not specified a UUID is generated. Note that it is
    * dangerous as in case of client side retries no deduplication will happen based on the

packages/workflow/src/internals.ts

@@ -16,6 +16,7 @@ import type {
   WorkflowDefinitionOptions,
   WorkflowSerializationContext,
   PayloadTypeInfo,
+  TypeInfo,
 } from '@temporalio/common';
 import {
   defaultFailureConverter,
@@ -116,6 +117,7 @@ export interface Completion<Success, Context = never> {
   resolve(val: Success): void;
   reject(reason: Error): void;
   context?: Context;
+  outputTypeInfo?: TypeInfo;
 }
 
 export interface Condition {
@@ -755,10 +757,15 @@ export class Activator implements ActivationHandler {
     if (!activation.result) {
       throw new TypeError('Got ResolveChildWorkflowExecution activation with no result');
     }
-    const { resolve, reject, context } = this.consumeCompletion('childWorkflowComplete', getSeq(activation));
+    const { resolve, reject, context, outputTypeInfo } = this.consumeCompletion(
+      'childWorkflowComplete',
+      getSeq(activation)
+    );
     if (activation.result.completed) {
       const completed = activation.result.completed;
-      const result = completed.result ? this.payloadConverter.fromPayload(completed.result, context) : undefined;
+      const result = completed.result
+        ? fromPayloadsAtIndex(this.payloadConverter, 0, [completed.result], context, outputTypeInfo)
+        : undefined;
       resolve(result);
     } else if (activation.result.failed) {
       const { failure } = activation.result.failed;

packages/workflow/src/workflow.ts

@@ -24,7 +24,7 @@ import {
   compilePriority,
   encodeActivityCancellationType,
   encodeWorkflowIdReusePolicy,
-  extractWorkflowType,
+  extractWorkflowTypeAndConfig,
   HandlerUnfinishedPolicy,
   mapToPayloads,
   encodeInitialVersioningBehavior,
@@ -435,7 +435,7 @@ function startChildWorkflowExecutionNextHandler({
         seq,
         workflowId,
         workflowType,
-        input: toPayloadsWithContext(activator.payloadConverter, context, options.args),
+        input: toPayloadsWithContext(activator.payloadConverter, context, options.args, options.typeInfo?.inputTypes),
         retryPolicy: options.retry ? compileRetryPolicy(options.retry) : undefined,
         taskQueue: options.taskQueue || activator.info.taskQueue,
         workflowExecutionTimeout: msOptionalToTs(options.workflowExecutionTimeout),
@@ -478,6 +478,7 @@ function startChildWorkflowExecutionNextHandler({
       resolve,
       reject,
       context,
+      outputTypeInfo: options.typeInfo?.outputType,
     });
   });
   untrackPromise(startPromise);
@@ -875,22 +876,29 @@ export async function startChild<T extends Workflow>(
     'Workflow.startChild(...) may only be used from a Workflow Execution. Consider using Client.workflow.start(...) instead.)'
   );
   const optionsWithDefaults = addDefaultWorkflowOptions(options ?? ({} as any));
-  const workflowType = extractWorkflowType(workflowTypeOrFunc);
+  const { type: workflowType, typeInfo } = extractWorkflowTypeAndConfig(
+    workflowTypeOrFunc,
+    optionsWithDefaults.typeInfo
+  );
+  const workflowOptions = {
+    ...optionsWithDefaults,
+    typeInfo,
+  };
   const execute = composeInterceptors(
     activator.interceptors.outbound,
     'startChildWorkflowExecution',
     startChildWorkflowExecutionNextHandler
   );
   const [started, completed] = await execute({
     seq: activator.nextSeqs.childWorkflow++,
-    options: optionsWithDefaults,
+    options: workflowOptions,
     headers: {},
     workflowType,
   });
   const firstExecutionRunId = await started;
 
   return {
-    workflowId: optionsWithDefaults.workflowId,
+    workflowId: workflowOptions.workflowId,
     firstExecutionRunId,
     async result(): Promise<WorkflowResultType<T>> {
       return (await completed) as any;
@@ -906,7 +914,7 @@ export async function startChild<T extends Workflow>(
         args,
         target: {
           type: 'child',
-          childWorkflowId: optionsWithDefaults.workflowId,
+          childWorkflowId: workflowOptions.workflowId,
         },
         headers: {},
       });
@@ -976,15 +984,22 @@ export async function executeChild<T extends Workflow>(
     'Workflow.executeChild(...) may only be used from a Workflow Execution. Consider using Client.workflow.execute(...) instead.'
   );
   const optionsWithDefaults = addDefaultWorkflowOptions(options ?? ({} as any));
-  const workflowType = extractWorkflowType(workflowTypeOrFunc);
+  const { type: workflowType, typeInfo } = extractWorkflowTypeAndConfig(
+    workflowTypeOrFunc,
+    optionsWithDefaults.typeInfo
+  );
+  const workflowOptions = {
+    ...optionsWithDefaults,
+    typeInfo,
+  };
   const execute = composeInterceptors(
     activator.interceptors.outbound,
     'startChildWorkflowExecution',
     startChildWorkflowExecutionNextHandler
   );
   const execPromise = execute({
     seq: activator.nextSeqs.childWorkflow++,
-    options: optionsWithDefaults,
+    options: workflowOptions,
     headers: {},
     workflowType,
   });
@@ -1068,13 +1083,19 @@ export function makeContinueAsNewFunc<F extends Workflow>(
     ...rest,
   };
 
+  let resolvedTypeInfo = options?.typeInfo;
+  // Reuse current workflow type information only when continuing as new to the same Workflow type.
+  if (resolvedTypeInfo == null && requiredOptions.workflowType === info.workflowType) {
+    resolvedTypeInfo = activator.typeInfo;
+  }
+
   return (...args: Parameters<F>): Promise<never> => {
     const context = currentWorkflowSerializationContext(info);
     const fn = composeInterceptors(activator.interceptors.outbound, 'continueAsNew', async (input) => {
       const { headers, args, options } = input;
       throw new ContinueAsNew({
         workflowType: options.workflowType,
-        arguments: toPayloadsWithContext(activator.payloadConverter, context, args),
+        arguments: toPayloadsWithContext(activator.payloadConverter, context, args, resolvedTypeInfo?.inputTypes),
         headers,
         taskQueue: options.taskQueue,
         memo: options.memo && mapToPayloads(activator.payloadConverter, options.memo, context),